Lesson 2: Context Window Budgeting
The budget equation
Every request must satisfy one inequality:
system + history + retrieved context + output reserve ≤ context window
The window is fixed by the model. The four terms on the left are yours to control, and if their sum exceeds the window the request fails, or worse, gets silently truncated.
@dataclass(frozen=True)
class ContextBudget:
window: int
system_tokens: int
output_reserve: int
@property
def available_for_content(self) -> int:
"""Tokens left for history and retrieved context."""
return self.window - self.system_tokens - self.output_reserve
The output reserve is the term people omit, and omitting it is the single most common production bug in this area.
The context window is shared between input and output. A model with a 200,000 token window and 199,000 tokens of input has 1,000 tokens left to answer in. It will either refuse or produce a truncated answer, and Module 6 established that truncation is not repairable by retrying.
Reserve deliberately:
budget = ContextBudget(
window=provider.capabilities.max_context_tokens,
system_tokens=counter.count_text(SYSTEM_PROMPT),
output_reserve=MAX_OUTPUT_TOKENS, # what you set as max_tokens
)
The reserve should equal the max_tokens you actually request, since that is the most the model can produce. Reserving less means the request can fit and the answer cannot.
Reasoning models need a larger reserve. Thinking tokens are output tokens and consume window space, so a model doing extended reasoning may need several times the reserve of a direct-answering model for the same visible answer length.
Pre-flight checks and failing fast
Module 5 established fail fast as a principle. Here it means checking the budget before sending rather than discovering the problem in an error response.
def preflight(
messages: list[Message], budget: ContextBudget, counter: TokenCounter
) -> None:
"""Raise before sending if the request cannot fit."""
count = counter.count_request(messages)
if count.total + budget.output_reserve > budget.window:
raise ContextOverflowError(
f"request needs {count.total} tokens plus {budget.output_reserve} "
f"reserved for output, exceeding the {budget.window} window",
token_count=count,
budget=budget,
)
Three reasons to check locally rather than letting the provider reject it. A rejected request still costs latency, and on some providers it costs money. The provider's error message tells you that you exceeded the limit, not which component was too large. And a local check lets you fix the request by trimming rather than failing it, which is what the rest of this lesson is about.
Carry the token count on the exception so the caller can act. Module 5's point about exceptions carrying data applies exactly here.
Truncation strategies
When content exceeds the budget, something must be removed. Four strategies, with different costs.
Drop oldest removes from the front of the history until it fits.
def drop_oldest(messages: list[Message], budget: int, counter: TokenCounter) -> list[Message]:
"""Remove the oldest turns, always keeping the system message and the last user turn."""
system = [m for m in messages if m.role == "system"]
rest = [m for m in messages if m.role != "system"]
while rest and counter.count_message_tokens(system + rest) > budget:
rest.pop(0)
return system + rest
Simple, cheap, and it silently forgets. A user who stated a constraint twenty turns ago will find the model has lost it, with no indication that anything was dropped.
Note the two things it never removes: the system message, which carries your instructions, and the most recent user turn, which is the actual question. Dropping either produces nonsense.
Sliding window keeps a fixed number of recent turns. The same behaviour with a simpler rule and less adaptivity.
Importance ranked keeps content by relevance rather than recency.
def keep_important(
messages: list[Message], budget: int, counter: TokenCounter, *, query: str
) -> list[Message]:
"""Keep the most relevant turns, preserving chronological order in the output."""
system = [m for m in messages if m.role == "system"]
recent = messages[-2:] # always keep the current exchange
candidates = [m for m in messages[:-2] if m.role != "system"]
scored = score_relevance(candidates, query) # embeddings, keywords, or heuristics
kept: list[Message] = []
used = counter.count_message_tokens(system + recent)
for message in sorted(scored, key=lambda s: s.score, reverse=True):
cost = counter.count_message_tokens([message.message])
if used + cost > budget:
continue
kept.append(message.message)
used += cost
kept.sort(key=lambda m: messages.index(m)) # restore chronological order
return system + kept + recent
Better retention of important content, at the cost of a relevance computation and the risk of dropping something that seemed irrelevant and was not. The chronological restore matters: presenting turns out of order confuses the model about sequence.
Summarize and compact replaces old turns with a generated summary.
async def compact(
messages: list[Message], *, keep_recent: int = 6
) -> list[Message]:
"""Replace older turns with a summary, keeping recent turns verbatim."""
system = [m for m in messages if m.role == "system"]
body = [m for m in messages if m.role != "system"]
if len(body) <= keep_recent:
return messages
older, recent = body[:-keep_recent], body[-keep_recent:]
summary = await provider.generate(
[Message(role="system", content=COMPACT_SYSTEM),
Message(role="user", content=format_for_summary(older))],
model=CHEAP_MODEL,
)
return system + [Message(role="assistant", content=f"[Earlier conversation]\n{summary.text}")] + recent
The best retention and the only strategy with a cost, since it requires a model call. Use a cheap model, since summarising a conversation is not a task needing your most expensive one.
History compaction and when to trigger it
Compaction is expensive enough that when you run it matters.
Trigger on a proportion of the budget, not on turn count. Ten turns of short exchanges and ten turns containing pasted documents are entirely different amounts of context.
COMPACT_THRESHOLD = 0.7 # compact when history exceeds 70% of its allowance
async def maybe_compact(
messages: list[Message], budget: ContextBudget, counter: TokenCounter
) -> list[Message]:
"""Compact before the budget is exhausted rather than at the moment of failure."""
history_tokens = counter.count_message_tokens(messages)
allowance = budget.available_for_content
if history_tokens < allowance * COMPACT_THRESHOLD:
return messages
return await compact(messages)
Compact early rather than at the limit. Waiting until the request would overflow means compaction sits directly in the latency path of a user request that was going to fail. Triggering at seventy percent means it happens with room to spare, and it can run between turns rather than during one.
Compact once, not repeatedly. Summarising a summary loses information at each pass and the degradation compounds. Keep the original messages if you can afford the storage, so a re-compaction works from the full history rather than from the previous summary.
Tell the user. A conversation that has been compacted behaves differently, and a small indication that earlier context has been summarised is more honest than silent forgetting.
Budgeting retrieved context
Module 9 left top_k as a decision to be measured. This is where the budget side of that decision is made.
def fit_chunks(
chunks: list[ScoredChunk], *, available: int, counter: TokenCounter
) -> tuple[list[ScoredChunk], int]:
"""Take chunks in ranked order until the budget is spent."""
kept: list[ScoredChunk] = []
used = 0
for scored in chunks:
cost = counter.count_text(scored.chunk.text) + CHUNK_WRAPPER_TOKENS
if used + cost > available:
continue # skip and try smaller ones behind it
kept.append(scored)
used += cost
return kept, used
continue rather than break is deliberate: a large chunk that does not fit should not prevent a smaller relevant one behind it from being included.
How many chunks fit is not the same as how many should be sent. Module 9 noted that beyond a point, extra chunks dilute rather than help. The budget is an upper bound, and the quality-optimal number is usually lower, which your evaluation set determines.
Which chunks to keep when they do not all fit. Ranked order is the default. Two refinements worth considering: prefer chunks from distinct documents when coverage matters more than depth, and prefer shorter chunks at equal relevance, since two shorter chunks may beat one long one for the same token spend.
Order of eviction across categories. When the whole request is over budget, decide in advance what gets cut first. A defensible default is retrieved context first, since it is the largest and most compressible term, then older history, then never the system prompt or the current question.
Graceful behaviour when input genuinely does not fit
Sometimes the input is simply too large: a user pastes a 400-page document into a 200,000 token window.
Do not silently truncate. Cutting the input and answering from the first portion produces an answer that appears complete and is based on part of the material, which is the worst outcome available.
Four honest options.
Reject with a specific message. "This document is roughly 340,000 tokens and the limit is 200,000. Try splitting it or asking about a specific section." That names the problem and suggests an action.
Route to a larger model if one is available and affordable, which the capability declaration from Module 7 makes checkable.
Process in parts and combine, meaning map over sections and reduce the results. This changes the answer's character, since a cross-document comparison is not reliably reconstructible from per-section summaries, so tell the user it happened.
Retrieve rather than stuff, which is the Module 9 answer: index the document and retrieve the relevant parts rather than sending all of it.
Whichever you choose, the response must say what was done. This is Module 5's honest degradation, applied to context.