Student edition · 90 minutes of dedicated work · 2026-09-29
Runs on Google Colab or any Python 3.12+ Jupyter kernel. It needs no packages beyond the standard library, no model and no network.
This is the first of Chapter 21's two practical units, each a complete ninety-minute session.
- Unit A accounts for every token Lucy's agent sends, and builds the compaction that fits a long day into a budget without losing what must not be lost.
- Unit B breaks the context on purpose: a summary that forgets, a layout that defeats the cache, an estimate that undercounts. Then it repairs each one.
You bring Python lists and dictionaries, and the idea of a model call from Chapter 1. Tokens, the context window, layers and compaction are introduced below, before you build anything.
By the end you should be able to:
- Split a request into layers, stable first, and account for every token to one of them.
- Estimate tokens from characters, and say when the estimate is wrong.
- Implement
clear_old_tool_resultsandcompact, which fit a conversation into a budget in two steps and never touch the application's pinned records. - Show that a rule kept only in the transcript can be lost by compaction, and the same rule kept as a record cannot.
| Minutes | Dedicated work | Saved evidence |
|---|---|---|
| 0–10 | Predict which of Lucy's facts survive when the day no longer fits | Written prediction |
| 10–25 | Layers, the budget report, estimates and both cuts of a long tool result | Observed outputs |
| 25–55 | Construct clear_old_tool_results and compact; pass the visible cases | Learner code and grade table |
| 55–70 | Compact Lucy's day at three budgets and check what survived | Compaction reports |
| 70–85 | Changed-constraint task: keep only the lines that answer a question | Transfer results |
| 85–90 | Explain records versus transcript and save evidence | Retained submission |
Scroll horizontally to view columns.
These times are planning estimates, not measured completion times. Run All only checks that the notebook executes; the unfinished student functions deliberately report NEEDS_WORK. Keep your first attempt before opening the worked edition.
Run the self-contained setup
The unit needs only Python's standard library. The collapsed cell below creates your work folder and defines the supplied parts of the unit:
- Counting:
estimate_tokens, about four characters per token, andmessage_tokensfor one chat message. - Layers:
Layers(system, pinned, conversation)holds what the model will see, stable first.budget_reportcounts the tokens in each layer, as/contextdoes in a coding agent. - Cutting a tool result:
headandhead_and_tail. - Lucy's day:
lucy_day()returns a long conversation. Early on, Lucy states rules and decisions and later changes some of them, and bulky tool results sit in between.QUESTIONSpairs each fact with where it lives. - Summarizers:
keep_numbers, which keeps only lines with a digit in them, andforget_everything, which keeps nothing. Both are deterministic stand-ins for a model writing a summary.
Run setup on every fresh kernel. Your saved work lives in practical-work/ch21-a. Restarting a kernel clears variables, not saved files.
Supplied setup, layers, Lucy's day and summarizers
The complete supplied setup cell is in the downloadable notebook. Run it there before attempting the cells below. This web edition omits only that compressed setup payload.
Commit to a prediction before the examples
Lucy's day is a long conversation: rules and decisions at the start, some changed later, and bulky freezer and sales tables in between. It no longer fits the budget, so the oldest part must go. Before you run anything below, write down which of these facts you expect the model can still see after the day is cut to fit:
- the shop's name, and the rule about mango and Frostline;
- the closing time, which Lucy changed from 6 pm to 7 pm;
- delivery A-4817's 20 tubs, which appeared only in a tool result;
- where the sign-up sheet goes, from the last minute of the day.
prediction_notes = {
"prediction": "Write which of the four facts survive a cut to the budget.",
"reason": "Name the rule behind that prediction.",
"falsifier": "Name an observation that would prove the explanation wrong.",
"revision": "After execution, explain what changed in your understanding.",
}Layers, and a budget for each
A model sees one request: a sequence of messages. It is useful to split that request into layers by how often they change. The system text changes almost never. The pinned records, facts the application itself keeps (Lucy's rules, approvals and suppliers), change rarely. The conversation grows every turn. Sending the stable layers first matters for cost (Chapter 9's prefix cache), and splitting them lets you see where the budget goes.
day = lucy_day()
print(len(day), "messages")
print(json.dumps(budget_report(Layers(SYSTEM, [], day)), indent=1))
print(json.dumps(budget_report(Layers(SYSTEM, PINNED, day)), indent=1))
biggest = max(day, key=message_tokens)
print("largest message:", biggest["role"], message_tokens(biggest), "tokens")Most of the day's tokens are tool results that nobody will read again: freezer tables and sales logs. The rules Lucy stated cost almost nothing. That imbalance is why compaction starts with tool results.
An estimate, and when it is wrong
estimate_tokens assumes four characters per token. A tokenizer splits digits, punctuation and JSON keys into more pieces than English words, so the right ratio depends on the kind of text. The chapter counted real tokens with the qwen2.5 tokenizer; the cell below compares the estimate with those counts.
MEASURED = [
{"kind": "prose", "chars": 1775, "estimate": 444, "actual": 349},
{"kind": "json", "chars": 2573, "estimate": 644, "actual": 1048},
{"kind": "numbers", "chars": 2016, "estimate": 504, "actual": 1187},
{"kind": "python", "chars": 2400, "estimate": 600, "actual": 540},
]
for row in MEASURED:
error = row["estimate"] / row["actual"] - 1
counts = f"estimate {row['estimate']:5}, tokenizer {row['actual']:5} ({error:+.0%})"
print(f"{row['kind']:8} {row['chars']:5} characters: {counts}")Two cuts of a long tool result
When one tool result is too long, keep part of it. head keeps the start, which loses the most recent rows of a log. head_and_tail keeps both ends and says how many lines it left out, so the model at least knows something is missing.
log = "\n".join(
f"{10 + i // 12}:{(i * 5) % 60:02d} | {FLAVORS[i % 10]} | {1 + i % 3} tubs" for i in range(80)
)
print(head(log, 60))
print("---")
print(head_and_tail(log, 60))1. Construct clear_old_tool_results
clear_old_tool_results(messages, keep_last) returns a new list in which every tool result except the last keep_last has its content replaced by a short marker:
[tool result cleared to save context: N tokens]N is estimate_tokens of the old content. Keep every other message unchanged, keep the order, and do not change the input list or its dictionaries. The call that produced each result stays in the conversation, so the agent can see that it looked something up, and can look it up again.
The starter clears nothing.
def clear_old_tool_results(messages, keep_last):
"""All but the last `keep_last` tool results replaced by a marker; the input is unchanged."""
return [dict(m) for m in messages]2. Construct compact
compact(layers, budget, summarize) returns new Layers that fit in budget tokens, according to budget_report(...)["total"], in at most two steps:
- If the layers already fit, return a copy unchanged.
- Clear old tool results, keeping the last two. If that fits, stop.
- Otherwise replace every message before the latest user message with one user message whose content is
"Summary of the earlier conversation:\n" + summarize(older). Keep the latest user message and everything after it.
Never change system or pinned. If system and pinned alone exceed the budget, if there is nothing older to summarize, or if the result still does not fit, raise ContextOverflowError. Never return a context over the budget.
The starter drops the oldest messages until the rest fits, which is what many agents do.
def compact(layers, budget, summarize):
"""Fit `layers` into `budget`: clear old tool results, then summarize the oldest turns."""
conversation = list(layers.conversation)
while (
conversation
and budget_report(Layers(layers.system, layers.pinned, conversation))["total"] > budget
):
conversation.pop(0)
return Layers(layers.system, list(layers.pinned), conversation)Hint 1: why the tool results go first
The worked example showed where the tokens are: freezer tables and sales logs. Clearing them loses facts that appeared only in a tool result, but the agent can call the tool again. A rule Lucy said once cannot be fetched again.
Hint 2: what "the latest user message" protects
The latest user message is the question being answered now. Summarizing it away would answer a question nobody asked. Find its index with the last position whose role is user.
Hint 3: why raise instead of cutting
A context over the budget is either rejected by the model API or cut somewhere you did not choose. An exception is a failure someone can see and fix; a silent cut is a wrong answer later.
import copy
TOOL = tool_message("stock_level", {"rows": ["vanilla 12"] * 40})
SMALL = [{"role": "user", "content": "hi"}, {"role": "assistant", "content": "hello"}]
LONG = [
m
for i in range(8)
for m in ({"role": "user", "content": f"q{i}"}, TOOL, {"role": "assistant", "content": f"a{i}"})
]
def cleared_count(messages):
return sum("cleared to save context" in str(m["content"]) for m in messages)
def run_compact(candidate, layers, budget, summarize):
try:
out = candidate(layers, budget, summarize)
return out, None
except ContextOverflowError:
return None, "ContextOverflowError"
except Exception as error:
return None, type(error).__name__
def grade_unit(clear, compactor):
rows = []
def row(case, passed, observed):
rows.append({"case": case, "observed": observed, "status": "PASS" if passed else "FAIL"})
original = copy.deepcopy(LONG)
out = clear(LONG, 2)
row(
"clears all but the last two",
cleared_count(out) == 6 and LONG == original,
cleared_count(out),
)
row(
"keeps every message and the order",
[m["role"] for m in out] == [m["role"] for m in LONG],
len(out),
)
row(
"keep_last beyond the count clears nothing",
cleared_count(clear(LONG, 99)) == 0,
cleared_count(clear(LONG, 99)),
)
fits, error = run_compact(compactor, Layers("s", ["rule"], SMALL), 10_000, forget_everything)
row(
"a context that fits is unchanged",
error is None and fits.conversation == SMALL,
error or "unchanged",
)
layers = Layers("s", ["Never order mango from Frostline."], LONG)
after_clear = budget_report(Layers("s", layers.pinned, clear_old_tool_results_reference(LONG)))[
"total"
]
out, error = run_compact(compactor, layers, after_clear, forget_everything)
step_one = error is None and out is not None and len(out.conversation) == len(LONG)
row(
"step one alone when it is enough",
step_one and cleared_count(out.conversation) == 6,
error or len(LONG),
)
out, error = run_compact(compactor, layers, 250, forget_everything)
ok = (
error is None
and out is not None
and len(out.conversation) >= 3
and out.conversation[-3]["content"] == "q7"
and budget_report(out)["total"] <= 250
)
row(
"summarizes up to the latest question",
ok,
error or (out and out.conversation[0]["content"][:30]),
)
row(
"pinned records are untouched",
error is None and out is not None and out.pinned == layers.pinned,
error or "pinned kept",
)
_, error = run_compact(compactor, Layers("s", ["x" * 4000], SMALL), 100, forget_everything)
row("pinned records over the budget raise", error == "ContextOverflowError", error)
_, error = run_compact(
compactor,
Layers("s", [], [{"role": "user", "content": "y" * 4000}]),
100,
forget_everything,
)
row("a latest turn over the budget raises", error == "ContextOverflowError", error)
return rows
def clear_old_tool_results_reference(messages):
positions = [i for i, m in enumerate(messages) if m.get("role") == "tool"]
old = set(positions[:-2])
marker = "[tool result cleared to save context: {} tokens]"
return [
{**m, "content": marker.format(estimate_tokens(m["content"]))} if i in old else dict(m)
for i, m in enumerate(messages)
]
visible_results = grade_unit(clear_old_tool_results, compact)
VISIBLE_PASSED = all(r["status"] == "PASS" for r in visible_results)
for visible_row in visible_results:
print(visible_row["status"], visible_row["case"], "->", visible_row["observed"])
print("VISIBLE_CONTRACT", "PASSED" if VISIBLE_PASSED else "NEEDS_WORK")3. Compact Lucy's day at three budgets
Once the visible cases pass, the cell below compacts Lucy's day three times, with the keep_numbers summarizer:
- at the whole day's size, where nothing needs to change;
- at the size the day has after clearing its old tool results, where step one is enough;
- at 400 tokens, where the summary must do the rest.
Each time it checks which of the facts are still anywhere the model can see. Then it repeats 400 tokens with the rules moved into pinned records.
Before running it, predict which facts survive each budget, and which survive 400 tokens with pinned records.
survival = None
if VISIBLE_PASSED:
survival = {}
whole = budget_report(Layers(SYSTEM, [], lucy_day()))["total"]
after_clearing = budget_report(Layers(SYSTEM, [], clear_old_tool_results(lucy_day(), 2)))[
"total"
]
print(
f"the whole day: {whole} tokens; after clearing old tool results: {after_clearing} tokens"
)
plans = (
("whole day", [], whole),
("clear only", [], after_clearing),
("400", [], 400),
("400+pinned", PINNED, 400),
)
for label, pinned, budget in plans:
layers = compact(Layers(SYSTEM, pinned, lucy_day()), budget, keep_numbers)
survival[label] = {
"budget": budget,
"tokens": budget_report(layers)["total"],
"kept": [fact for fact, needle, _ in QUESTIONS if survives(layers, needle)],
}
kept, tokens = len(survival[label]["kept"]), survival[label]["tokens"]
print(f"{label:11} {tokens:5} tokens, {kept} of {len(QUESTIONS)} facts visible")
for fact, _, where in QUESTIONS:
status = "kept" if fact in survival[label]["kept"] else "LOST"
print(" ", status, f"{fact} ({where})")
assert len(survival["whole day"]["kept"]) == len(QUESTIONS)
assert "delivery A-4817 was 20 tubs" not in survival["clear only"]["kept"]
assert "mango is never ordered from Frostline" in survival["clear only"]["kept"]
assert "mango is never ordered from Frostline" not in survival["400"]["kept"]
assert "mango is never ordered from Frostline" in survival["400+pinned"]["kept"]
else:
print("COMPACTION_NOT_READY — repair clear_old_tool_results and compact, then run again.")Read the lists before moving on.
- The whole day fits its own size, so nothing changed.
- At the size after clearing, step one was enough: every rule stayed. The facts that lived only in a tool result went: delivery A-4817 and the strawberry count. The agent could fetch those again.
- At 400 tokens the summary kept only the lines with digits in them. That included both closing times and both chocolate approvals, old and new, and none of Lucy's rules, which contain no digits.
- With the rules pinned, the same 400 tokens kept every rule, because pinned records are never compacted.
A summary is a guess about what will matter later. A record is the application's decision.
4. Save the handoff
Unit B breaks compaction on purpose. The handoff records your budgets and what survived at each.
ARTIFACT_PATH = Path("ch21-unit-a-handoff-v1.json")
artifact_status = "NOT_WRITTEN"
if survival is not None:
handoff = {
"unit": "ch21-a",
"status": "COMPLETED",
"budgets": [survival[label]["budget"] for label in ("whole day", "clear only", "400")],
"summarizer": "keep_numbers",
"pinned": PINNED,
"survival": survival,
}
ARTIFACT_PATH.write_text(json.dumps(handoff, indent=2, sort_keys=True) + "\n", encoding="utf-8")
artifact_status = "WRITTEN"
print(ARTIFACT_PATH)
else:
print("HANDOFF_NOT_WRITTEN")Exit ticket
Answer three questions:
- Why does compaction clear tool results before it summarizes anything?
- Which of Lucy's facts could the agent recover after compaction without asking her, and how?
- Why does
compactraise an error instead of returning a context a little over the budget?
exercise_report = {
"unit": "ch21-a",
"attempted": 1,
"completed": int(VISIBLE_PASSED),
"failed": int(not VISIBLE_PASSED),
"skipped": 0,
"compaction": "PASSED" if survival else "NOT_READY",
"handoff": artifact_status,
}
print("EXERCISE_REPORT=" + json.dumps(exercise_report, sort_keys=True))Changed-constraint construction: keep only the lines that answer a question
Allow fifteen minutes: three to predict, eight to implement and trace, and four for a case of your own.
Cutting a long tool result by position keeps lines that may not matter. When the agent knows what it is looking for, it can keep the lines that match. transfer_filter(text, needle, budget) returns the lines of text that contain needle, compared case-insensitively. It keeps the most recent matching lines, meaning those nearest the end, within budget tokens by estimate_tokens(line + "\n"), in their original order. After them comes one final line:
[kept K of M matching lines; T lines in total]With no matching line, return just the final line with K and M both 0. Do not change the input.
Write your expected values before you run the table.
def transfer_filter(text, needle, budget):
raise NotImplementedError(
"Keep the most recent matching lines within the budget, then the count line"
)SALES = "\n".join(
[
"10:02 | vanilla | 2 tubs",
"10:05 | Pistachio | 1 tubs",
"10:09 | mint | 3 tubs",
"10:15 | pistachio | 2 tubs",
"10:20 | pistachio | 3 tubs",
]
)
TRANSFER_CASES = [
(
"every match fits",
[SALES, "pistachio", 100],
"\n".join(
[
"10:05 | Pistachio | 1 tubs",
"10:15 | pistachio | 2 tubs",
"10:20 | pistachio | 3 tubs",
"[kept 3 of 3 matching lines; 5 lines in total]",
]
),
),
(
"only the most recent fit",
[SALES, "pistachio", 14],
"\n".join(
[
"10:15 | pistachio | 2 tubs",
"10:20 | pistachio | 3 tubs",
"[kept 2 of 3 matching lines; 5 lines in total]",
]
),
),
("no match", [SALES, "mango", 100], "[kept 0 of 0 matching lines; 5 lines in total]"),
(
"a zero budget keeps none",
[SALES, "mint", 0],
"[kept 0 of 1 matching lines; 5 lines in total]",
),
]
def run_transfer(candidate, cases):
observations = []
for label, arguments, expected in cases:
supplied = copy.deepcopy(arguments)
try:
actual = candidate(*supplied)
except NotImplementedError:
actual = {"unfinished": True}
except Exception as error:
actual = {"raises": type(error).__name__}
passed = actual == expected and supplied == arguments
observations.append(
{"case": label, "expected": expected, "observed": actual, "passed": passed}
)
print("PASS" if passed else "NEEDS_WORK", label)
return observations
transfer_observations = run_transfer(transfer_filter, TRANSFER_CASES)
TRANSFER_PASSED = all(r["passed"] for r in transfer_observations)
print("TRANSFER_STATUS", "PASS" if TRANSFER_PASSED else "NEEDS_WORK")Design a counterexample and retrieve the mechanism
Add one case with an expected outcome you worked out independently, and rerun the driver. Then ask which question about the sales log this filter cannot answer however large its budget, and what the agent would need instead.
Save your evidence and explain the result
Fill in the prediction notes and your explanation before saving. Include:
- the exact observed value, and the input that caused it;
- your code's invocation point;
- one failed hypothesis;
- the strongest claim the evidence still cannot support.
This unit measures what is visible to the model, not what the model does with it. A fact in the context can still be misread; the chapter's experiment measures that on real models.
explanation_notes = {
"causal_trace": "Explain the input, learner invocation and observed result.",
"failed_hypothesis": "Describe a prediction the evidence changed.",
"remaining_limit": "Name the guarantee not established by this experiment.",
}
course_submission = {
"unit": "ch21-a",
"planned_minutes": 90,
"starting_evidence": globals().get("HANDOFF_ORIGIN", "INDEPENDENT_UNIT_A"),
"prediction": prediction_notes,
"explanation": explanation_notes,
"core_report": exercise_report,
"transfer": transfer_observations,
"explanation_review": "HUMAN_REVIEW_REQUIRED",
}
submission_path = COURSE_WORK / "ch21-a-submission-v1.json"
submission_path.write_text(
json.dumps(course_submission, indent=2, sort_keys=True), encoding="utf-8"
)
print("Saved evidence:", submission_path)
print(
"COURSE_REPORT="
+ json.dumps(
{
"unit": "ch21-a",
"transfer_passed": TRANSFER_PASSED,
"starting_evidence": course_submission["starting_evidence"],
"edition": "student",
},
sort_keys=True,
)
)Keep building with Prof Rod
Found this material through a colleague, classroom or shared download? Get the complete book at profrod.ai/book and join the Prof Rod learner community. Bring one result, one question or one failure you learned from. Share this resource with another learner and keep its source links with it so they can find the full course and future updates.