Instructor worked edition · 90 minutes of dedicated work · 2026-09-29
Runs on Google Colab or any Python 3.12+ Jupyter kernel.
This edition contains the worked cached_prefix_tokens, stable_first and transfer solution, instructor explanations and hidden checks. Learners should attempt the student edition first.
| Minutes | Dedicated work | Saved evidence |
|---|---|---|
| 0–10 | Predict what a model keeps when it summarizes Lucy's day | Written prediction |
| 10–30 | Check the real summary; repair with pinned records | Fact table |
| 30–55 | Construct cached_prefix_tokens and stable_first; pass the visible cases | Learner code and grade table |
| 55–70 | Replay ten turns in both layouts and compare with the measured prefill | Cache table |
| 70–85 | Changed-constraint task: a worst-case token count | Transfer results |
| 85–90 | Explain records, layout and margins; save evidence | Retained submission |
Scroll horizontally to view columns.
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:
- Unit A's parts:
estimate_tokens,message_tokens,Layers,budget_report,clear_old_tool_resultsandcompact, as the worked Unit A builds them. - Lucy's day:
lucy_day(), the pinned recordsPINNED, andsurvives(layers, needle). - Recorded measurements from the chapter's experiment with qwen2.5:1.5b:
REAL_SUMMARY, the summary the model wrote of the older part of Lucy's day;MEASURED, token counts by kind of text; andMEASURED_PREFILL, the median prefill per turn in each layout.
Run setup on every fresh kernel. Your saved work lives in practical-work/ch21-b.
Supplied setup, Unit A's parts and the recorded measurements
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 failures
In the chapter's experiment, qwen2.5:1.5b was asked to summarize the older part of Lucy's day: "keep every rule, decision, name and number Lucy stated; where she changed something, keep only the latest value." Before you read its summary, write down:
- which of Lucy's rules you expect it to keep;
- whether it keeps the closing time and the chocolate approval at their changed values;
- whether it can introduce something that was never said.
prediction_notes = {
"prediction": "Write which facts the summary keeps, and whether it keeps the latest values.",
"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.",
}Start from Unit A's handoff
This unit uses the budgets and pinned records you tested in Unit A. To use your own, replace None with the path to your practical-work/ch21-a/ch21-unit-a-handoff-v1.json. Leave it as None to start from the supplied reference, which is the handoff the worked Unit A produces. Your submission records which you chose.
LEARNER_HANDOFF = Noneimport shutil
COURSE_INPUT = COURSE_WORK / "ch21-unit-a-handoff-v1.json"
if LEARNER_HANDOFF is not None:
learner_input = Path(LEARNER_HANDOFF).expanduser().resolve()
if not learner_input.is_file():
raise FileNotFoundError("The selected learner handoff does not exist")
if learner_input != COURSE_INPUT.resolve():
shutil.copy2(learner_input, COURSE_INPUT)
HANDOFF_ORIGIN = "LEARNER_SELECTED"
else:
reference_encoded = (
"c-q}m%Wm5+5WM><7QD1Y?ZC1V$IUIM(^J|6h|x<C1hllV2vej$QZkev|K6n@l<l~O0x3}BVvGCY?952d0|2%vk~K"
"QZF5nj(0VdP=Y*;@<llioHIvtPc|K1;1E|kI~l(S`uU{jrguy9?()dvfAvDUdgVNEmpfh8KyCZXSqtF?wlqciwsv"
"~^M_424Ub@CsZvFqYb)1@7P+@(e!4DK%r#bk8XCDRf@uxzQyi7Z9C+D^?ctBgCnWwd4*D?mL(iP5L=Mxe`)xze|m"
"`tGT5)1{PeHsz*yWQ{>X6Ae7B9cG~#H1y>wyj1x~MEU;!BcZDaJeZBd9dwsRMTCgTjWEnT&35`!~5c3CBWGghb%b"
"QY^TvGPAxtGKUhoB!tC_e0h8_ZphsE&9tg<69X8G4kSQa0Sl2eNp}rLSt)+%2r-5o<Lna*_$r!CI3_t426S^s%=1"
"KEQE$gUWCJWm<<`hGCV@!B2m?QRfpfoZ=boVr*PS_*HMcPRd||z3tzMow7AhtTQ6z&^w1b{=qkIMt5{J+v7WyNKm"
"wr)lt6t7ydmGZk^#<da+Ce*6DZyTkh{Y=I?*53lY61%r?~$fxdo6b?HCJ?@^Zor66z>T#ioX(RnS)Wc+I2e<YvRq"
"mggWO2`T&yeh=cBpPuxcp3Z#_qq#4"
)
COURSE_INPUT.write_bytes(zlib.decompress(base64.b85decode(reference_encoded)))
HANDOFF_ORIGIN = "SUPPLIED_REFERENCE"
print("Starting evidence:", HANDOFF_ORIGIN)handoff = json.loads(COURSE_INPUT.read_text(encoding="utf-8"))
completed = handoff.get("status") == "COMPLETED" and handoff.get("unit") == "ch21-a"
handoff_status = "VERIFIED" if completed else "INVALID"
TIGHT_BUDGET = min(handoff["budgets"])
HANDOFF_PINNED = handoff["pinned"]
print(
"UNIT_A_HANDOFF",
handoff_status,
"| tightest budget",
TIGHT_BUDGET,
"| pinned records",
len(HANDOFF_PINNED),
)1. Check a summary a model wrote
The facts below are the ones the summary had to keep. For each, the check looks for the current value and for the value Lucy later replaced. A summary can keep a fact, lose it, or keep the old value as if it were current.
print(REAL_SUMMARY)
print("-" * 60)
FACTS = [
("the shop's name", "Lucy's Scoops", None),
("mango is never ordered from Frostline", "mango", None),
("closing time", "7 pm", "6 pm"),
("chocolate approved", "15", "12"),
("vanilla supplier", "Polar Dairy", None),
("health inspector", "Dana Ortiz", None),
]
summary_check = {}
for fact, current, old in FACTS:
text = REAL_SUMMARY.lower()
has_current = current.lower() in text
has_old = old is not None and old.lower() in text
status = (
"kept"
if has_current and not has_old
else "kept, with the old value too"
if has_current
else "OLD VALUE"
if has_old
else "LOST"
)
summary_check[fact] = status
print(f"{fact:40} {status}")Read what the model wrote, not only the table. The table says the mango rule was kept, because the word "mango" is there. The summary says "updates on orders from Frostline (mango)": it kept the words and turned the rule into its opposite. A keyword check cannot see that, and neither can compaction. A summary is text the model produced, and it can state things that were never said. It is also the only memory the compacted conversation keeps of the older turns. Now repair it the way Unit A did: move the rules into pinned records, which compaction never touches, and compact the same day with the same real summary.
real_summarizer = lambda older: REAL_SUMMARY # noqa: E731
SUMMARY_BUDGET = TIGHT_BUDGET + estimate_tokens(REAL_SUMMARY)
without = compact(Layers(SYSTEM, [], lucy_day()), SUMMARY_BUDGET, real_summarizer)
with_pinned = compact(
Layers(SYSTEM, HANDOFF_PINNED, lucy_day()), SUMMARY_BUDGET + 120, real_summarizer
)
for fact, current, _ in FACTS:
plain = "visible" if survives(without, current) else "missing"
pinned = "visible" if survives(with_pinned, current) else "missing"
print(f"{fact:40} without pinned: {plain:8} with pinned: {pinned}")With pinned records, every rule is visible whatever the summary says. That does not make the summary right: where it disagrees with a record, the model now sees both. The record should win, and the system text can say so, but the cleaner fix is to keep facts that have a record out of the summary's job altogether.
2. Construct cached_prefix_tokens
A prefix cache reuses the work done on the start of the previous request, up to the first difference. Everything after the first difference is read again. At the level of messages, cached_prefix_tokens(previous, current) returns the sum of message_tokens over the leading messages of current that are identical, in role and content, to the messages at the same positions in previous. It stops at the first difference. Do not change the inputs.
The starter assumes the cache reuses every message the two requests have in common, anywhere.
def cached_prefix_tokens(previous, current):
"""Tokens at the start of `current` identical to `previous`, up to the first difference."""
total = 0
for before, now in zip(previous, current, strict=False):
if before.get("role") != now.get("role") or before.get("content") != now.get("content"):
break
total += message_tokens(now)
return total3. Construct stable_first
stable_first(stable, snapshot, history, question) builds one request so that the prefix cache can do its job. It returns a new list:
- a system message whose content is exactly
stable; - every message of
history, unchanged and in order; - one user message whose content is
snapshot + "\n" + question.
The snapshot, meaning the time and the freezer readings, changes every turn, so it goes last. Do not change history.
The starter puts the snapshot at the top of the system message, where it changes the first message of every request.
def stable_first(stable, snapshot, history, question):
"""One request: stable system text, then history, then the volatile snapshot and question."""
return [
{"role": "system", "content": stable},
*[dict(m) for m in history],
{"role": "user", "content": snapshot + "\n" + question},
]Hint 1: "up to the first difference"
Walk both lists in step with zip. The first position where role or content differ ends the reuse, even if later messages match again. The cache matches a prefix, not a set.
Hint 2: what changes every turn
The rules and the tool list change almost never. The conversation only grows at its end. The time and the freezer snapshot change every turn. Put them last, and every earlier message stays the same from one turn to the next.
import copy
A = {"role": "system", "content": "rules " * 50}
B = {"role": "user", "content": "q1"}
C = {"role": "assistant", "content": "a1"}
D = {"role": "user", "content": "q2"}
def grade_unit(cached, ordered):
rows = []
def row(case, passed, observed):
rows.append({"case": case, "observed": observed, "status": "PASS" if passed else "FAIL"})
previous, current = [A, B, C], [A, B, C, D]
saved = copy.deepcopy((previous, current))
got = cached(previous, current)
row(
"an identical start is reused",
got == sum(map(message_tokens, [A, B, C])) and (previous, current) == saved,
got,
)
changed = {"role": "system", "content": "10:05 " + A["content"]}
got = cached([A, B, C], [changed, B, C, D])
row("a changed first message reuses nothing", got == 0, got)
got = cached([A, B, C], [A, {"role": "user", "content": "other"}, C, D])
row("reuse stops at the first difference", got == message_tokens(A), got)
got = cached([], [A])
row("nothing before, nothing reused", got == 0, got)
history = [B, C]
saved = copy.deepcopy(history)
request = ordered("RULES", "10:05 vanilla 12", history, "Anything low?")
row(
"stable text is the whole system message",
request[0] == {"role": "system", "content": "RULES"},
request[0]["content"][:20],
)
row("history follows unchanged", request[1:3] == [B, C] and history == saved, len(request))
row(
"snapshot and question come last",
request[-1] == {"role": "user", "content": "10:05 vanilla 12\nAnything low?"},
request[-1]["content"][:30],
)
return rows
visible_results = grade_unit(cached_prefix_tokens, stable_first)
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")4. Replay ten turns in both layouts
Once the visible cases pass, the cell below replays ten turns of Lucy's day twice. The first time, the snapshot sits at the top of the system message. The second time it sits in the last message, via stable_first. For each turn it counts the tokens a prefix cache could reuse from the previous request. Then it prints the prefill the chapter measured on qwen2.5:1.5b for the same two layouts.
Predict the reuse in each layout before running it.
replay = None
if VISIBLE_PASSED:
stable = "You are the assistant for Lucy's ice cream shop.\n" + "\n".join(
f"Rule {i}: confirm every order with Lucy before it is placed." for i in range(40)
)
rng = random.Random(3)
snapshots = [
f"Current time 10:{t:02d}. Freezer: "
+ ", ".join(f"{f} {rng.randint(0, 40)}" for f in FLAVORS)
for t in range(10)
]
replay = {}
for layout in ("volatile-first", "stable-first"):
history, previous, reused = [], [], []
for turn in range(10):
question = f"Question {turn}: is any flavor below 5 tubs?"
if layout == "volatile-first":
request = [
{"role": "system", "content": snapshots[turn] + "\n\n" + stable},
*history,
{"role": "user", "content": question},
]
else:
request = stable_first(stable, snapshots[turn], history, question)
total = sum(map(message_tokens, request))
reused.append(round(cached_prefix_tokens(previous, request) / total, 2))
history += [request[-1], {"role": "assistant", "content": "Checked."}]
previous = request
replay[layout] = reused
print(f"{layout:15} share of each request reusable: {reused}")
print("measured median prefill per turn after the first (ms):", MEASURED_PREFILL)
assert all(share == 0 for share in replay["volatile-first"])
assert all(share > 0.8 for share in replay["stable-first"][1:])
else:
print("REPLAY_NOT_READY — repair cached_prefix_tokens and stable_first, then run again.")With the snapshot first, nothing is reusable on any turn, because the first message changes every time. With the snapshot last, almost every token of each request was already read on the previous turn. The measured prefill shows what that costs on a real model.
An estimate that undercounts
Unit A's budgets counted four characters per token. The recorded measurements show how far that is from a real tokenizer, by kind of text. A budget of 4,096 estimated tokens, filled with the numbers-heavy text a sales log contains, needs this many real tokens:
for row in MEASURED:
ratio = row["chars"] / row["actual"]
real = math.ceil(4096 * 4 / ratio)
print(f"{row['kind']:8} {ratio:.2f} chars per token: 4,096 estimated = about {real:,} real")A context that fits by the estimate can overflow the model's window. The chapter measured what one model server does then: it cut the prompt silently and answered from what was left. Budget with the ratio of the text you actually send, or with the worst one, and keep a margin.
exercise_report = {
"unit": "ch21-b",
"attempted": 1,
"completed": int(VISIBLE_PASSED),
"failed": int(not VISIBLE_PASSED),
"skipped": 0,
"replay": "PASSED" if replay else "NOT_READY",
"handoff": handoff_status,
"summary_check": summary_check,
}
print("EXERCISE_REPORT=" + json.dumps(exercise_report, sort_keys=True))Changed-constraint construction: a worst-case token count
Allow fifteen minutes: three to predict, eight to implement and trace, and four for a case of your own.
worst_case_tokens(text, measured) counts text pessimistically. measured is a list of rows like MEASURED, each with chars and actual. Divide the length of text by the smallest ratio of chars / actual among the rows, and round up with math.ceil. With no rows, fall back to one character per token, which is len(text). Do not change the inputs.
Write your expected values before you run the table.
def worst_case_tokens(text, measured):
if not measured:
return len(text)
ratio = min(row["chars"] / row["actual"] for row in measured)
return math.ceil(len(text) / ratio)ROWS = [
{"kind": "prose", "chars": 500, "actual": 100},
{"kind": "numbers", "chars": 300, "actual": 200},
]
TRANSFER_CASES = [
("the smallest ratio decides", ["x" * 30, ROWS], 20),
("rounds up", ["x" * 31, ROWS], 21),
("one row", ["x" * 10, ROWS[:1]], 2),
("no rows: one character per token", ["x" * 7, []], 7),
("empty text", ["", ROWS], 0),
]
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(worst_case_tokens, 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 say what a worst-case count costs: for Lucy's prose, how much of the budget would it leave unused?
Instructor explanation and additional transfer cases
Three misconceptions come up.
"The model summarized its own conversation, so the summary is faithful." A summary is generated text, with every failure Chapter 1 described. The recorded summary is a real example: check which facts it kept, dropped or changed. Anything the application must rely on belongs in a record the summary cannot change.
"The cache reuses whatever the two requests share." It reuses a prefix. One changed character in the first message makes everything after it new. Timestamps and snapshots at the top of a system prompt cost a full read on every turn.
"Four characters per token is close enough." For prose it overcounts; for numbers and JSON it undercounts, by more than half in the chapter's measurement. A budget near the window's edge needs the real ratio or a margin.
INSTRUCTOR_TRANSFER_CASES = [
("equal ratios", ["x" * 12, [{"chars": 4, "actual": 1}, {"chars": 8, "actual": 2}]], 3),
("a ratio below one", ["xx", [{"chars": 1, "actual": 2}]], 4),
]
instructor_observations = run_transfer(worst_case_tokens, INSTRUCTOR_TRANSFER_CASES)
assert TRANSFER_PASSED and all(r["passed"] for r in instructor_observations)assert cached_prefix_tokens([A], [A]) == message_tokens(A)
assert cached_prefix_tokens([A, B], [A]) == message_tokens(A)
assert cached_prefix_tokens([{"role": "user", "content": "rules " * 50}], [A]) == 0, "role matters"
assert stable_first("S", "snap", [], "q") == [
{"role": "system", "content": "S"},
{"role": "user", "content": "snap\nq"},
]
print("HOLDOUT_RESULT=" + json.dumps({"status": "PASSED", "unit": "ch21-b"}, sort_keys=True))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.
The cache replay counts reusable messages; it does not time a model. The measured prefill comes from one model on one machine. Real servers cache at the token level and evict caches over time, so the share reused is an upper bound.
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-b",
"planned_minutes": 90,
"starting_evidence": HANDOFF_ORIGIN,
"prediction": prediction_notes,
"explanation": explanation_notes,
"core_report": exercise_report,
"transfer": transfer_observations,
"explanation_review": "HUMAN_REVIEW_REQUIRED",
}
submission_path = COURSE_WORK / "ch21-b-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-b",
"transfer_passed": TRANSFER_PASSED,
"starting_evidence": course_submission["starting_evidence"],
"edition": "instructor",
},
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.