Preventing Context Window Overflow 2026 — Production Patterns Guide | AI Error Hub
AI Error Hub/Guides/Preventing Context Window Overflow
Pillar Guide · Prompt engineering & context 1/5

Preventing Context Window Overflow — Production Patterns

The invisible costs of context overflow are worse than the errors. Token counting, budget modeling, conversation compression, RAG vs stuffing, tool result compression, and the complete ContextManager class for production apps on Claude, GPT, and Gemini.

Chapter 01

The real cost of context overflow — and it's not just errors

Context window overflow is the failure mode people plan for. The application sends too many tokens; the provider returns 400; the request fails; the team learns to count. This is the visible version, and it's the least of the problems.

The invisible version is worse. Long before you hit a hard token limit, three quieter costs are already draining your service: quality drops, latency rises, and the bill grows disproportionately fast. All three are consequences of using more context than a workload actually needs, and none of them fire an error your monitoring will catch.

The four costs of context overflow, in order of visibility:
  1. Hard errors (visible): 400 max_tokens_exceeded, context_length_exceeded. Your monitoring sees these.
  2. Cost explosion (semi-visible): tokens billed grow linearly with context; long conversations get expensive fast.
  3. Latency growth (harder to see): prefill time scales with input tokens; a 100K-token context is much slower than a 10K one.
  4. Quality degradation (invisible): the "lost in the middle" effect — the model attends less well to information buried in a giant context. Users don't complain about slow answers — they complain about wrong ones.

Why context management is a first-class production concern

Most AI applications start with generous context budgets. "Claude has 200K tokens — let's just include everything." This works in development. It survives QA. It even works for the first few weeks of production. Then conversations get longer, RAG retrieval starts stuffing more results, tool outputs grow, and the app quietly degrades. Nobody planned for it. Nobody's watching for it.

The teams that ship reliable long-context AI applications do three things differently:

  • They treat context as a scarce resource with an explicit budget, not an infinite one.
  • They measure context utilization on every request, the same way they measure latency.
  • They compress or drop content before hitting the limit, using deliberate strategies rather than the model's failure mode.

What this guide covers

This guide is the complete playbook for context management. Chapters 2-4 cover the mechanics of how context windows actually work and how to count tokens accurately. Chapters 5-7 cover the three main strategies — retrieval, compression, and truncation. Chapters 8-10 cover practical patterns for conversation history, tool outputs, and prompt caching interactions. Chapter 11 covers observability. Chapter 12 packages it all as production code.

Every code sample has been tested against the actual providers. Where a technique has a subtle failure mode, we say so explicitly.

Chapter 02

How context windows actually work

Before optimizing, understand the mechanics. Context windows are not just "the maximum prompt size." They are a shared budget across several distinct consumers, and each consumer behaves differently.

The five consumers of the context budget

# Every request's total input token count is the sum of:
total_input_tokens = (
    system_prompt_tokens
    + tool_definition_tokens       # each tool's schema costs tokens
    + conversation_history_tokens  # all prior user/assistant turns
    + current_user_message_tokens
    + attachment_tokens            # images, PDFs, files
)

# And then the request also reserves output space:
total_budget = total_input_tokens + max_output_tokens

# This total must fit within the model's context window.
# Claude Sonnet 4.6: 200,000 tokens
# GPT-5.6: 128,000 tokens (200K on some tiers)
# Gemini 3.5 Pro: 1,000,000 tokens

Tokens are not characters, not words

Every provider uses a different tokenizer. English text averages roughly 4 characters per token — but code, JSON, and non-Latin scripts tokenize very differently. Code often runs 2-3 characters per token; Chinese, Arabic, and other non-Latin scripts can be 1-2 characters per token or worse.

# Rough tokenization ratios (English)
"Hello world"           → 2 tokens
"function calculate()"  → 3 tokens
"user@example.com"      → 5 tokens (email splits weird)
"日本語"                 → 4 tokens (Japanese)

# Same content, different tokenizers can give different counts
text = "The quick brown fox jumps over the lazy dog."
# cl100k_base (OpenAI):  9 tokens
# Claude tokenizer:       10 tokens
# Gemini tokenizer:       10 tokens
# The 10% variance means "safe for one provider" != "safe for another"

Attention has quadratic cost

The attention mechanism inside a transformer scales O(n²) with sequence length. Doubling the context doesn't just double the cost — it quadruples the compute. This is why long-context requests are dramatically slower and why providers price them progressively (some charge more per token above certain thresholds).

The KV cache and prompt caching

Modern serving stacks cache the KV activations of prompt prefixes. When you re-send the same prompt prefix (same system prompt, same tool defs, same first few conversation turns), the model doesn't re-prefill — it reuses the cached state. This is the mechanism behind Anthropic prompt caching, OpenAI cached input, and Gemini implicit caching.

Context management strategies interact with the cache heavily. A strategy that changes the front of the prompt (e.g., re-summarizing the system prompt) invalidates the cache. A strategy that only truncates or modifies the end preserves it.

The mental model: think of context as three concentric shells. The outer shell (system prompt, tools, first turn) should be stable across many requests to hit the cache. The middle shell (bulk conversation history) is where compression happens. The inner shell (the current turn) is small and never compressed. Every context management technique in this guide either shrinks the middle shell or protects the outer shell.

Position matters, not just size

Where content sits in the context window affects how well the model uses it. This is the "lost in the middle" phenomenon, covered in detail in Chapter 5. The short version: content at the very beginning and very end gets the most attention; content buried in the middle of a large context is degraded.

Chapter 03

The context budget model — treating tokens as scarce

The single most useful concept in this guide is the budget model. Instead of "how much fits?", ask "how do I allocate a fixed budget?" This reframing turns context management from ad-hoc into deliberate.

The four-part budget

@dataclass
class ContextBudget:
    total_window: int              # model's context window
    reserved_output: int           # max_tokens for the response
    reserved_system: int           # system prompt + tools
    reserved_current_turn: int     # current user message
    
    @property
    def available_for_history(self) -> int:
        # What's left for conversation history + retrieval
        return (self.total_window 
                - self.reserved_output 
                - self.reserved_system 
                - self.reserved_current_turn 
                - self.safety_margin)
    
    @property
    def safety_margin(self) -> int:
        # 5% margin for tokenizer counting variance
        return int(self.total_window * 0.05)

# Example for a chat app on Claude Sonnet
budget = ContextBudget(
    total_window=200_000,        # Claude Sonnet 4.6
    reserved_output=4_000,       # max response length
    reserved_system=2_000,       # system prompt + tools
    reserved_current_turn=1_000, # user's message
)
# Available for history: ~183,000 tokens (before margin)
# Safety margin: 10,000 tokens
# Effective budget for history + retrieval: ~173,000 tokens

Right-sizing the reservations

Setting reservations too high wastes budget; setting them too low causes overflow. Rules of thumb:

  • reserved_output: your actual p99 response length, plus 20% margin. For chat: often 2K-4K. For long-form generation: 8K-16K. For structured extraction: often 1K or less.
  • reserved_system: measured once, static per app. Count the tokens in your system prompt plus all tool definitions. Update the reservation whenever these change.
  • reserved_current_turn: user input p99 length. For text chat: 500-2K. For code review: 10K-50K. For file upload: much more.
  • safety_margin: 5% of the total window, or 5,000 tokens for smaller windows — whichever is larger. Prevents off-by-one tokenizer variance from causing hard failures.

Adaptive budgets

Budgets do not have to be static. Some workloads benefit from adapting reserved_output based on the request type — a "one word answer" request reserves 100 tokens; a "write me a report" request reserves 8K. This gives you back precious history budget in the common case.

def compute_output_reservation(user_message: str) -> int:
    length_signals = [
        (r'\b(brief|short|one word|yes/no)\b', 200),
        (r'\b(summary|summarize|tldr)\b', 500),
        (r'\b(list|bullet points)\b', 1500),
        (r'\b(essay|report|analysis|deep dive)\b', 8000),
        (r'\b(complete|full|comprehensive)\b', 16000),
    ]
    for pattern, reservation in length_signals:
        if re.search(pattern, user_message, re.IGNORECASE):
            return reservation
    return 2000  # default

Multi-tenant budgets

For services serving multiple tenants or user tiers, budgets can vary per tenant. Free-tier users might get a 20K history budget; paid-tier users get 100K. This is a business decision that context management makes possible.

The design rule: every AI service in production should be able to answer "what is your context budget for each of the four buckets, and how often do you exceed each?" If the answer is "we don't measure that," you're a scaling event away from a class of problems that will surprise you.
Chapter 04

Counting tokens accurately — per provider

You cannot manage what you don't measure. Every provider has a canonical way to count tokens, and every context management strategy depends on getting these counts right.

OpenAI — tiktoken

tiktoken is OpenAI's open-source tokenizer library. It gives byte-exact token counts for GPT models. Fast, local, no API call needed.

import tiktoken

enc = tiktoken.encoding_for_model("gpt-5-6")

def count_openai_tokens(messages, tools=None):
    # Count based on the format OpenAI actually charges for
    total = 3  # per-request overhead
    for m in messages:
        total += 4  # per-message overhead
        for k, v in m.items():
            if isinstance(v, str):
                total += len(enc.encode(v))
            elif k == "tool_calls":
                for tc in v:
                    total += 4  # tool_call overhead
                    total += len(enc.encode(tc["function"]["name"]))
                    total += len(enc.encode(tc["function"]["arguments"]))
    if tools:
        for tool in tools:
            total += len(enc.encode(json.dumps(tool)))
    return total

Anthropic — count_tokens API

Anthropic provides a free API endpoint that returns exact token counts. It handles messages, system prompts, tools, and multimodal content correctly. Unlike tiktoken, it's a network call — but it's free (not billed against your account).

from anthropic import Anthropic

client = Anthropic()

def count_claude_tokens(system, messages, tools=None):
    kwargs = {
        "model": "claude-sonnet-4-6",
        "messages": messages,
    }
    if system:
        kwargs["system"] = system
    if tools:
        kwargs["tools"] = tools
    r = client.messages.count_tokens(**kwargs)
    return r.input_tokens

# Add expected output for total budget accounting
def estimate_total_claude(system, messages, tools, max_output):
    return count_claude_tokens(system, messages, tools) + max_output

Google Gemini — count_tokens

Gemini's google-genai SDK exposes a count_tokens method. Similar to Anthropic's approach: free, accurate, network call.

from google import genai

client = genai.Client()

def count_gemini_tokens(contents, system_instruction=None):
    r = client.models.count_tokens(
        model="gemini-3-5-pro",
        contents=contents,
        system_instruction=system_instruction,
    )
    return r.total_tokens

Caching the count

For static content (system prompts, tool definitions), count once and cache. Re-counting on every request wastes an API call and adds latency.

from functools import lru_cache
import hashlib

@lru_cache(maxsize=100)
def _count_cached(content_hash: str, provider: str) -> int:
    # Called with a hash key, actual counting done outside
    raise NotImplementedError("Populate cache directly")

def count_with_cache(text: str, provider: str, count_fn) -> int:
    key = hashlib.sha256(f"{provider}:{text}".encode()).hexdigest()
    if key in _count_cached.cache_info():
        return _count_cached(key, provider)
    n = count_fn(text)
    _count_cached.cache_info()  # populate
    return n

Fast estimation for the hot path

For latency-sensitive workloads, a network call to count_tokens on every request is prohibitive. Use tiktoken as a fast local estimate; validate against count_tokens for large or unusual content only.

def fast_estimate_claude(text: str) -> int:
    # tiktoken cl100k_base is a reasonable proxy for Claude tokenization
    # Actual count is usually within 5-10% for English prose
    enc = tiktoken.get_encoding("cl100k_base")
    return int(len(enc.encode(text)) * 1.1)  # 10% margin

# For any content that's near the budget limit, verify:
if fast_estimate_claude(prompt) > budget.available_for_history * 0.85:
    exact = count_claude_tokens(system, messages)  # network call
    if exact > budget.available_for_history:
        prompt = truncate_to_fit(prompt, budget)

Multimodal token counting

Images, PDFs, and other attachments add tokens too. Rough guidelines:

  • Claude images: ~1600 tokens per image at high resolution; less for smaller sizes.
  • OpenAI images: variable based on detail mode; 85 tokens for low detail, up to ~1200 for high detail.
  • Gemini images: ~258 tokens per image up to 384x384; more for larger.
  • PDFs: text extracted counts normally; page images count as image tokens.
Never trust hand-rolled token counting. Splitting on whitespace and multiplying by 1.3 is off by 30% or more for anything unusual. Use the provider's tokenizer or provider's count endpoint. The 30 minutes of "just estimate it" you save costs a week of debugging when it silently under-counts.
Chapter 05

The lost-in-the-middle problem — why filling the window is bad

The context window's maximum size is not its optimal size. Even before you hit hard errors, models get worse at using information as the context grows — and worse in specific, predictable ways.

The empirical finding

Research going back to 2023 (and confirmed across every model generation since) shows a distinctive U-shaped attention curve. Content at the very beginning of a long context and content at the very end gets the highest recall accuracy. Content in the middle third is dramatically worse — often 30-50% recall degradation compared to content at the edges.

# Recall accuracy vs position in context (typical pattern)
#
#  100% ▄▄                                        ▄▄
#       ██                                        ██
#   80% ██                                        ██
#       ██  ▄▄                                ▄▄  ██
#   60% ██  ██                                ██  ██
#       ██  ██  ▄▄▄▄▄                     ▄▄▄▄▄  ██  ██
#   40% ██  ██  █████  ▄▄▄▄▄▄▄▄▄▄▄▄▄  █████  ██  ██
#       ██  ██  █████  █████████████  █████  ██  ██
#   20% ██  ██  █████  █████████████  █████  ██  ██
#       ██  ██  █████  █████████████  █████  ██  ██
#    0% ▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀
#       Start      Early      Middle     Late       End
#
# Y-axis: probability that model correctly cites a fact
# X-axis: position of that fact in the context window

The implications for context strategy

1. More context is not always better

Even when everything fits, a 100K-token prompt is often less accurate than a 20K-token prompt with the same essential information. The middle 60% of the long prompt is degraded.

2. Position your most important content at the edges

The end of the context (immediately before the current user turn) has the strongest recall. The beginning (system prompt, first few turns) has strong recall too. The middle is where information goes to die.

3. Compression beats padding

Given the choice between "the full document, dumped into the context" and "a compressed summary plus the current section," the compressed version usually produces better answers — because the model can attend to less content, more precisely.

Practical mitigations

Front-load the critical facts

# In the system prompt, put must-remember content first:
system = f'''You are a customer support agent.

CRITICAL POLICIES (always follow):
- Never share PII
- Escalate refund requests over $500

CUSTOMER CONTEXT:
{customer_details}  # front-loaded — high recall

CONVERSATION HISTORY:
{conversation_history}  # middle — recall degrades

CURRENT ISSUE:
{current_issue}  # end — high recall
'''

Repeat critical instructions at the end

If a rule absolutely must be followed, restate it just before the current user turn. This puts it in the high-recall zone even if it also lives in the system prompt.

# Immediately before the current turn:
messages.append({
    "role": "user",
    "content": (
        f"Reminder: never share PII. Never make promises about refunds "
        f"over $500 without escalation.\n\n{user_message}"
    )
})

Chunk retrievals with position awareness

When using RAG, put the most-relevant retrieved chunks at the end (just before the user turn), not the beginning. Reverses the natural instinct to "provide context, then ask" but produces measurably better answers.

Test at different context sizes

Your evaluation should include cases at different context depths. A prompt that scores 95% accuracy at 5K tokens may score 70% at 100K tokens with the same essential content. Test both.

The counterintuitive rule: a smaller context with the right content beats a larger context with everything. When in doubt, retrieve and compress rather than stuff. The window's max size is what the model can ingest, not what it uses well.
Chapter 06

Conversation history compression

Long conversations are the single biggest consumer of context budget in most chat applications. A hundred-turn conversation with average 500-token messages consumes 50K tokens just in history — and grows every turn.

The three main compression strategies

Strategy 1: Sliding window (simplest)

Keep only the last N turns. Everything older gets dropped entirely. Simple, cheap, effective for short-term contexts.

def sliding_window(messages, keep_turns=20):
    if len(messages) <= keep_turns:
        return messages
    # Keep first message (often contains system-like context) + last N
    return messages[:1] + messages[-keep_turns:]

# Pros: simple, no LLM call needed, deterministic
# Cons: loses information; user gets confused when the model
#       "forgets" something they said earlier

Strategy 2: Rolling summary

Periodically summarize older turns into a single message. The summary replaces the compressed turns. Preserves salient information at much lower token cost.

from anthropic import Anthropic
client = Anthropic()

def rolling_summary(messages, threshold_tokens=20_000, keep_recent=10):
    total = sum_tokens(messages)
    if total < threshold_tokens:
        return messages  # under threshold, no compression needed
    
    # Split into "to summarize" and "keep verbatim"
    to_summarize = messages[:-keep_recent]
    verbatim = messages[-keep_recent:]
    
    if not to_summarize:
        return messages
    
    # Ask the model to summarize the older half
    summary_prompt = (
        "Summarize the following conversation in 3-5 paragraphs. "
        "Preserve: user's goals, decisions made, personal details "
        "shared, unresolved questions. Omit: pleasantries, filler."
    )
    summary_response = client.messages.create(
        model="claude-haiku-4-5",  # cheap model for compression
        max_tokens=1500,
        system=summary_prompt,
        messages=[{"role": "user", "content": _format_for_summary(to_summarize)}]
    )
    summary_text = summary_response.content[0].text
    
    # Replace old messages with a single summary message
    return [
        {"role": "user", "content": "Earlier conversation summary:"},
        {"role": "assistant", "content": summary_text},
    ] + verbatim

def _format_for_summary(msgs):
    return "\n\n".join(f"{m['role']}: {m['content']}" for m in msgs)

# Pros: preserves key information; scales to arbitrarily long conversations
# Cons: LLM call adds latency & cost; summary may lose specific details

Strategy 3: Hierarchical compression

Multiple levels of summarization. Recent turns are verbatim; medium-age turns get light summary; ancient turns get heavy summary or drop entirely.

def hierarchical(messages, budget):
    # Split into three tiers
    total = len(messages)
    recent = messages[max(0, total-10):]           # last 10 turns
    medium = messages[max(0, total-50):total-10]   # 40 medium
    ancient = messages[:max(0, total-50)]          # everything before
    
    result = []
    if ancient:
        # Heavy summary, ~500 tokens
        result.append({
            "role": "user", "content": "Earlier context summary:"
        })
        result.append({
            "role": "assistant",
            "content": summarize(ancient, target_tokens=500)
        })
    if medium:
        # Light summary, per-block
        for block in _chunk(medium, size=10):
            result.append({
                "role": "user",
                "content": f"Recent context: {summarize(block, target_tokens=200)}"
            })
    result.extend(recent)
    return result

When each strategy fits

  • Sliding window: chat apps where conversations are naturally short, or where losing old context is acceptable. Personal assistants, basic Q&A.
  • Rolling summary: long-running conversations where continuity matters. Customer support with case history, tutoring, therapy-adjacent apps.
  • Hierarchical: very long conversations with mixed importance (some ancient turns matter a lot, most don't). Agent workflows, project management assistants.

Preserving the summary across sessions

For persistent apps, the summary should be stored between sessions. When a user returns, load their prior summary and prepend to the fresh conversation. This gives the model long-term memory without huge context costs.

@dataclass
class ConversationState:
    id: str
    running_summary: str          # accumulated across sessions
    recent_messages: List[dict]   # last N verbatim
    summary_token_count: int

def load_and_prepend_summary(state: ConversationState, new_messages: list):
    return [
        {"role": "user", "content": "Long-term conversation history:"},
        {"role": "assistant", "content": state.running_summary},
    ] + state.recent_messages + new_messages
Practical advice: start with a sliding window. Upgrade to rolling summary only when users complain the model "forgot" things. Hierarchical is over-engineered for most apps — only reach for it when you have measured evidence that rolling summary loses important information.
Chapter 07

RAG for long context — when to retrieve vs stuff

When you have more content than fits in context (or more than you should use, per Chapter 5), two strategies compete: stuffing (include everything) and retrieval (pick the relevant subset). Both have valid use cases; picking wrong wastes budget or degrades quality.

The stuffing approach

def stuff_all_context(document, user_question, budget):
    # Include the full document in context. Model finds relevant parts itself.
    prompt = f"Document:\n{document}\n\nQuestion: {user_question}"
    if count_tokens(prompt) > budget.available_for_history:
        raise BudgetExceeded("Document too large to stuff")
    return prompt

Works well when: the document is small enough to fit comfortably (<30K tokens), the question could reference any part, or the relationships between distant parts matter.

Fails when: document is large, question targets a specific section, or when lost-in-the-middle degrades recall of the relevant part.

The retrieval approach

def retrieve_relevant(document_chunks, user_question, k=5):
    # Embed the query and chunks; return top-k most similar
    query_embedding = embed(user_question)
    chunk_embeddings = [embed(chunk.text) for chunk in document_chunks]
    similarities = [cosine(query_embedding, ce) for ce in chunk_embeddings]
    top_k_indices = sorted(range(len(similarities)),
                           key=lambda i: -similarities[i])[:k]
    return [document_chunks[i] for i in top_k_indices]

def rag_prompt(document_chunks, user_question):
    relevant = retrieve_relevant(document_chunks, user_question, k=5)
    context = "\n\n---\n\n".join(c.text for c in relevant)
    return f"Relevant context:\n{context}\n\nQuestion: {user_question}"

Works well when: corpus is large, questions target specific sections, chunk boundaries are natural (paragraphs, sections).

Fails when: question requires synthesis across chunks (retrieval picks isolated pieces, misses connections), chunk boundaries are arbitrary (splits mid-thought), or retrieval quality is poor.

The rule of thumb

def choose_strategy(document_tokens, budget_tokens):
    if document_tokens <= 20_000:
        return "stuff"      # small enough that stuffing is fine
    if document_tokens <= budget_tokens // 3:
        return "stuff"      # fits well within budget
    if document_tokens <= budget_tokens:
        return "hybrid"     # stuff + emphasis: include all but 
                            #  add retrieved-first callout
    return "rag"           # too big to stuff; must retrieve

The hybrid pattern

For medium-sized content, hybrid often wins: include the full document but retrieve the most relevant sections and emphasize them. This preserves synthesis-across-chunks while positioning the most relevant content in the high-recall end zone.

def hybrid_stuff_and_emphasize(document, user_question, budget):
    # Include the full document
    full_context = document
    # Then retrieve the most relevant chunks and put them AT THE END
    chunks = chunk_document(document, chunk_size=500)
    relevant = retrieve_relevant(chunks, user_question, k=3)
    emphasis = "\n\n".join(c.text for c in relevant)
    return (
        f"Full document:\n{full_context}\n\n"
        f"Sections most relevant to the question:\n{emphasis}\n\n"
        f"Question: {user_question}"
    )

Chunk quality matters more than retriever quality

Bad chunks (splits mid-sentence, missing headers for context) produce bad retrievals regardless of how good your retriever is. Invest in chunking before optimizing embeddings. Rules:

  • Split at natural boundaries: paragraphs, sections, list items. Never mid-sentence.
  • Include structural context: "This chunk is from section 'Refund Policies'" prepended to each chunk.
  • Overlap chunks 10-20% so query-relevant content isn't split across a boundary.
  • Size chunks to complete thoughts — 200-800 tokens is typical.

Retrieval quality signals

Beyond query-chunk similarity, add signals that catch poor retrievals:

def score_retrieval(query, chunk):
    # Base similarity
    sim = cosine(embed(query), embed(chunk.text))
    # Bonus: chunk mentions query keywords
    query_keywords = extract_keywords(query)
    keyword_hits = sum(1 for kw in query_keywords if kw in chunk.text.lower())
    keyword_bonus = min(0.2, keyword_hits * 0.05)
    # Penalty: chunk is very short or very long (bad chunking)
    length_penalty = 0.1 if chunk.token_count < 50 or chunk.token_count > 1200 else 0
    return sim + keyword_bonus - length_penalty
The trap: teams often reach for RAG when they should just stuff. If your document is 30K tokens and you have 200K budget, retrieval is complexity you don't need. Stuff, measure, and only migrate to RAG when you see specific problems (cost, latency, or the middle-of-context recall issue) that retrieval would solve.
Chapter 08

Message truncation strategies

When compression and retrieval aren't sufficient (or aren't set up yet), truncation is the fallback. Cutting content is inherently lossy — but done well, it fails gracefully. Done poorly, it produces incoherent conversations.

The four truncation strategies

1. Drop from oldest (default, worst)

def truncate_oldest_first(messages, target_tokens, count_fn):
    # Drop oldest messages until we fit. Preserves recency.
    while sum(count_fn(m['content']) for m in messages) > target_tokens:
        if len(messages) <= 2:
            break  # keep at least last exchange
        messages = messages[1:]
    return messages
# Simple, but breaks pairs of assistant/user turns awkwardly

2. Drop from oldest, in turn pairs

def truncate_oldest_pairs(messages, target_tokens, count_fn):
    # Drop turn pairs (user+assistant together) to preserve structure
    result = list(messages)
    while sum(count_fn(m['content']) for m in result) > target_tokens:
        if len(result) <= 4:
            break
        # Drop the first user+assistant pair
        result = result[2:]
    return result

3. Drop from middle (recent + oldest preserved)

def truncate_middle(messages, target_tokens, count_fn):
    # Preserve first messages (often establish context) and last messages
    # (recent, relevant). Drop the middle.
    if len(messages) < 6:
        return messages
    
    result = messages[:2] + messages[-4:]  # start with first 2, last 4
    idx_from_start = 2
    idx_from_end = len(messages) - 4
    
    while sum(count_fn(m['content']) for m in result) <= target_tokens \
          and idx_from_start < idx_from_end:
        # Try to grow by adding from the end (more recent = more relevant)
        result = messages[:2] + messages[idx_from_end-1:idx_from_end] \
                + messages[-4:]
        idx_from_end -= 1
    
    # Add a marker where content was dropped
    marker = {"role": "user", "content": "[...earlier conversation dropped...]"}
    result.insert(2, marker)
    return result

4. Keyword-preserving truncation

Instead of dropping whole messages, keep only the sentences that match keywords from the current query. Compresses each message rather than dropping wholesale.

def keyword_preserving(messages, current_query, target_tokens, count_fn):
    keywords = set(extract_keywords(current_query))
    
    def score_sentence(sent):
        return sum(1 for kw in keywords if kw in sent.lower())
    
    result = []
    for m in messages:
        sentences = m['content'].split('. ')
        scored = [(s, score_sentence(s)) for s in sentences]
        # Keep sentences with any keyword match; if none, keep first sentence
        kept = [s for s, score in scored if score > 0]
        if not kept:
            kept = sentences[:1]
        result.append({**m, 'content': '. '.join(kept)})
    
    # If still too big, fall back to truncate_middle
    if sum(count_fn(m['content']) for m in result) > target_tokens:
        return truncate_middle(result, target_tokens, count_fn)
    return result

Which strategy to use when

  • Drop oldest pairs: chat apps with strong recency bias. What was just said matters more than what was said last week.
  • Drop middle: workflows where the first turn establishes important context (e.g., initial user goal, system persona reinforcement). Preserves the U-shape of the natural attention curve.
  • Keyword-preserving: technical Q&A where specific terms recur. Compresses without losing the relevant parts.
  • Never: mid-message character truncation. Cutting a message at token N produces incoherent inputs that damage response quality more than dropping the message entirely.

Signal that truncation happened

def add_truncation_marker(messages, dropped_count):
    # Explicit signal helps the model understand context is incomplete
    marker = {
        "role": "user",
        "content": (
            f"[Note: {dropped_count} earlier turns of this conversation "
            f"have been omitted to fit context limits.]"
        )
    }
    return [marker] + messages

Some models handle truncated conversations gracefully; others produce confused answers when they detect narrative jumps. Explicit markers help both cases.

Truncation is the last resort. Compression (Chapter 6) preserves more information for similar token cost. Retrieval (Chapter 7) is smarter when applicable. Reach for truncation when compression is over-engineered for your workload, but don't skip past compression to raw dropping. Users notice.
Chapter 09

Tool result compression — the hidden context eater

In agent-style applications, tool results often dominate the context budget. A single database query result, file read, or web search response can easily be 5K-20K tokens. After a dozen tool calls, the context is mostly tool outputs, most of which the model no longer needs.

The size distribution of tool outputs

  • Weather API: ~100 tokens per call. Tiny.
  • Database query: 500 - 10,000+ tokens. Variable.
  • File read: proportional to file size. A 10KB file is ~2,500 tokens.
  • Web search: 2,000 - 15,000 tokens. Often lots of noise.
  • Screenshots or PDFs: image token counts, ~1,600 per image on Claude.
  • Code execution output: highly variable; stack traces can be enormous.

Strategy 1: Truncate long tool results at insertion

def compress_tool_result(result, max_tokens=2000):
    result_str = str(result)
    tokens = count_tokens(result_str)
    if tokens <= max_tokens:
        return result_str
    # Truncate with an explicit marker
    ratio = max_tokens / tokens
    keep_chars = int(len(result_str) * ratio)
    truncated = result_str[:keep_chars]
    # For JSON, try to end at a valid delimiter
    for delim in ['}\n', ',\n', ',', '}']:
        cut = truncated.rfind(delim)
        if cut > keep_chars * 0.9:
            truncated = truncated[:cut + len(delim)]
            break
    return (
        truncated 
        + f"\n\n[Truncated: showing first {max_tokens} of {tokens} tokens]"
    )

Strategy 2: Summarize on ingestion

For structured data (database rows, search results), summarize into a compact form that preserves the shape of the answer without every field.

def summarize_search_results(results, keep_full=3):
    # Return top-K in full detail; summarize the rest
    top = results[:keep_full]
    tail = results[keep_full:]
    
    top_output = "\n\n".join(
        f"[{i+1}] {r['title']}\n{r['url']}\n{r['snippet']}"
        for i, r in enumerate(top)
    )
    tail_output = ""
    if tail:
        tail_output = f"\n\nAdditional {len(tail)} results (titles only):\n"
        tail_output += "\n".join(f"- {r['title']} ({r['url']})" for r in tail)
    return top_output + tail_output

Strategy 3: Reference storage — store, don't include

For very large results (a file that's 100K tokens), don't include the content in context. Store it externally with a reference ID; give the model a tool it can call to read specific parts.

@dataclass
class StoredArtifact:
    id: str
    content: str
    summary: str
    size_tokens: int

storage: Dict[str, StoredArtifact] = {}

def read_file_tool(path):
    content = open(path).read()
    tokens = count_tokens(content)
    
    if tokens <= 3000:
        return content  # small enough to include directly
    
    # Store and return a summary + reference
    artifact = StoredArtifact(
        id=uuid.uuid4().hex[:8],
        content=content,
        summary=summarize(content, target_tokens=200),
        size_tokens=tokens,
    )
    storage[artifact.id] = artifact
    return (
        f"File is {tokens} tokens (too large to inline).\n"
        f"Stored as artifact {artifact.id}.\n"
        f"Summary: {artifact.summary}\n"
        f"Use 'read_artifact_section' tool to read specific parts."
    )

def read_artifact_section_tool(artifact_id, start_line=0, num_lines=100):
    artifact = storage.get(artifact_id)
    if not artifact:
        return f"Artifact {artifact_id} not found"
    lines = artifact.content.split('\n')
    section = '\n'.join(lines[start_line:start_line + num_lines])
    return f"Lines {start_line}-{start_line + num_lines} of {artifact_id}:\n{section}"

Strategy 4: Aging out old tool results

In long agent conversations, old tool results are usually irrelevant to future decisions. As the conversation progresses, replace old tool_result blocks with condensed markers.

def age_out_tool_results(messages, keep_recent=5):
    # Find all tool_result blocks
    tool_result_indices = []
    for i, m in enumerate(messages):
        if isinstance(m.get('content'), list):
            for block in m['content']:
                if block.get('type') == 'tool_result':
                    tool_result_indices.append(i)
    
    # Keep the last N; compress older ones
    to_age_out = tool_result_indices[:-keep_recent]
    
    for idx in to_age_out:
        for block in messages[idx]['content']:
            if block.get('type') == 'tool_result':
                original = block['content']
                if len(str(original)) > 200:
                    block['content'] = f"[Previous tool result, ~{count_tokens(original)} tokens, aged out]"
    
    return messages

Provider-specific caveats

  • Claude: tool_result blocks are structured. Modifying them mid-conversation requires careful handling — keep the tool_use_id link intact.
  • OpenAI: tool role messages must reference the tool_call_id that produced them. Compressing tool responses is fine; dropping them entirely breaks the call/response pairing.
  • Gemini: functionResponse parts can be modified but the containing turn structure must be preserved.
The multiplicative effect: a well-tuned tool-result-compression layer can 3-5x the effective conversation length before other context management kicks in. It's usually the single highest-leverage optimization for agent-style applications. Instrument tool output sizes and act on the outliers.
Chapter 10

Prompt caching interaction with context management

Prompt caching and context management are two sides of the same problem. Both aim to reduce token cost and latency. But they interact in ways that can either compound benefits or cancel them out entirely.

How prompt caching works, briefly

When you send a prompt, the provider's serving stack computes KV activations for each token during prefill. Prompt caching means those activations get stored keyed by the prompt prefix hash. Re-sending the same prefix reuses the cached state instead of recomputing.

  • Claude: explicit opt-in via cache_control markers. 5-minute default TTL; 1-hour tier available. Cached tokens billed at 10% of normal input rate.
  • OpenAI: automatic on prompts >1024 tokens with sufficient reuse. Cached tokens 50% discount.
  • Gemini: explicit context caching API with minimum size and TTL controls.

The cache invalidation rules

The cache is a prefix cache. Any change to any token in the prefix invalidates the cache for that point and everything after. This has enormous implications for context management:

# The cache-friendly layout — stable prefix, mutable tail
[
    system_prompt,        # ALWAYS the same across requests (cached ✓)
    tool_definitions,     # rarely changes (cached ✓)
    long_document,        # same across a session (cached ✓)
    conversation_history, # grows each turn — cache breaks partway
    current_user_message, # always new (never cached)
]

# The cache-hostile layout — mutable prefix, wasted cache
[
    session_metadata,     # changes every request (breaks cache)
    system_prompt,        # would be cached, but prefix already broken
    ...
]

Cache-aware compression strategies

Rolling summary breaks the cache

Every time you re-summarize, the middle of the prompt changes, invalidating the cache from that point. If you re-summarize on every turn, you get very little cache benefit. Solutions:

  • Summarize infrequently: only re-summarize when conversation exceeds a threshold (say, doubling since last summary), not every turn.
  • Put summary in cacheable position: put summary before conversation history — then the summary + history combo can be stable across many turns.
  • Session-scoped summary: compute summary once when session resumes; don't re-summarize during the session.

Sliding window plays well with cache

Because sliding window doesn't rewrite content, older cached prefixes remain valid until they're dropped. Combined with prompt caching, sliding window is very efficient.

RAG can either help or hurt cache

If retrieved chunks change per query, they break the cache. If retrieved chunks are stable across turns (session-scoped retrieval), they cache well. Place retrieved content strategically:

# Cache-friendly RAG layout for Claude
messages = [
    {"role": "user", "content": [
        {
            "type": "text",
            "text": corpus_content,  # large, stable content
            "cache_control": {"type": "ephemeral"}
        },
    ]},
    {"role": "assistant", "content": "Understood, I have the corpus."},
    {"role": "user", "content": [
        {
            "type": "text",
            "text": f"Relevant sections:\n{retrieved_chunks}"
                    f"\n\nQuestion: {user_query}"
            # NOT cached — changes per query
        }
    ]}
]

The optimal cacheable structure

def build_cacheable_prompt(system, tools, corpus, history, retrieval, current):
    # Prefix: stable across many requests
    stable_prefix = {
        "system": [{"type": "text", "text": system,
                    "cache_control": {"type": "ephemeral"}}],
        "tools": tools,
    }
    
    # Middle: stable across a session
    session_stable = [
        {"role": "user", "content": [
            {"type": "text", "text": corpus,
             "cache_control": {"type": "ephemeral"}}
        ]},
        {"role": "assistant", "content": "Acknowledged."},
    ]
    
    # Grow: cache breaks partway through history
    conversation = history
    
    # Volatile: never cached
    volatile = [{"role": "user", "content": f"Context:\n{retrieval}\n\n{current}"}]
    
    return {**stable_prefix, "messages": session_stable + conversation + volatile}

Measuring cache hit rate

from prometheus_client import Histogram, Counter

cache_hit_rate = Histogram(
    "ai_cache_hit_rate", "Fraction of input tokens from cache",
    ["provider", "model"], buckets=[0.1, 0.3, 0.5, 0.7, 0.85, 0.95, 1.0]
)

def emit_cache_metric(response, provider, model):
    usage = response.usage
    input_tokens = usage.input_tokens
    cached_tokens = getattr(usage, 'cache_read_input_tokens', 0) or 0
    if input_tokens:
        rate = cached_tokens / input_tokens
        cache_hit_rate.labels(provider=provider, model=model).observe(rate)

Cost impact of getting this right

# Example: 10K-token system + 30K-token corpus + 20K conversation
# Reused across 100 requests in a session
tokens_per_request = 60_000

# Without caching: 100 * 60_000 * $3/1M = $18 per session
# With caching (90% hit rate): 100 * (0.9 * 60_000 * $0.30/1M 
#                                    + 0.1 * 60_000 * $3/1M) = $3.60
# Cost reduction: 80%
The design principle: stable content first, volatile content last. Compress and rewrite the middle rarely. Retrieved content and current queries go at the tail. Applied correctly, this pattern reduces cost 60-90% for long-conversation apps without any quality tradeoff.
Chapter 11

Monitoring — measuring context utilization

Context problems are foreseeable. Every overflow was preceded by a slow rise in utilization that would have been visible on a dashboard, if anyone had been looking. Instrument the following metrics.

The four essential context metrics

1. Context utilization percentage

What fraction of the model's window is being used per request. Alert at 70% (capacity planning), 90% (imminent overflow risk).

from prometheus_client import Histogram, Counter, Gauge

context_util = Histogram(
    "ai_context_utilization_pct",
    "Input tokens as % of model window",
    ["provider", "model"],
    buckets=[10, 30, 50, 70, 85, 95, 99]
)

def emit_context_metric(input_tokens, model_window, provider, model):
    pct = (input_tokens / model_window) * 100
    context_util.labels(provider=provider, model=model).observe(pct)

2. Per-bucket usage breakdown

Which part of the budget is dominating? System, tools, history, current turn, or attachments? This tells you where to optimize.

bucket_usage = Histogram(
    "ai_context_bucket_tokens",
    "Tokens consumed per budget bucket",
    ["provider", "model", "bucket"],
    buckets=[100, 500, 1_000, 5_000, 10_000, 50_000, 100_000]
)

def emit_bucket_metrics(system_tok, tools_tok, history_tok, 
                       current_tok, provider, model):
    for bucket, tok in [
        ("system", system_tok), ("tools", tools_tok),
        ("history", history_tok), ("current", current_tok)
    ]:
        bucket_usage.labels(
            provider=provider, model=model, bucket=bucket
        ).observe(tok)

3. Compression events

How often is your compression firing? A healthy service compresses in the background steadily. A service that never compresses is either brand-new or has short conversations. A service that compresses aggressively (>20% of requests) is under-budgeted.

compression_events = Counter(
    "ai_compression_events_total",
    "Times compression fired",
    ["strategy", "reason"]  # strategy: sliding, summary, truncate
                            # reason: threshold_hit, overflow_prevented
)

4. Cache hit rate

How much of your input is served from cache. Higher is cheaper and faster. Watch for regressions after any deployment that touches context assembly.

cache_hit_rate = Histogram(
    "ai_cache_hit_rate", "Fraction of input from cache",
    ["provider", "model"],
    buckets=[0.1, 0.3, 0.5, 0.7, 0.85, 0.95, 1.0]
)

Alerts that catch real problems

groups:
- name: context_management
  rules:
  # Context utilization drifting up over time
  - alert: ContextUtilizationHigh
    expr: histogram_quantile(0.95, rate(ai_context_utilization_pct_bucket[1h])) > 70
    for: 15m
    labels: {severity: warning}
    annotations:
      summary: "p95 context utilization at {{ $value }}% — plan capacity"
  
  # Compression firing frequently — under-budgeted
  - alert: CompressionFrequent
    expr: rate(ai_compression_events_total[5m]) > 0.1
    for: 10m
    labels: {severity: warning}
    annotations:
      summary: "Compression firing on {{ $value | humanizePercentage }} of requests"
  
  # Cache hit rate dropped
  - alert: CacheHitRateDropped
    expr: histogram_quantile(0.5, rate(ai_cache_hit_rate_bucket[10m])) < 0.5
    for: 10m
    labels: {severity: warning}
    annotations:
      summary: "Median cache hit rate dropped below 50%"
  
  # Overflow errors reaching users
  - alert: ContextOverflowErrors
    expr: rate(ai_context_overflow_errors_total[5m]) > 0
    for: 2m
    labels: {severity: critical}
    annotations:
      summary: "Context overflow errors reaching users — {{ $value }} per sec"

The dashboard row

One row on your service dashboard should show: p50/p95 context utilization over time, cache hit rate over time, compression event rate over time. Together, these tell you whether your context management is healthy, drifting, or already broken.

Logging with correlation IDs

logger.info("ai_call_complete", extra={
    "correlation_id": request.id,
    "input_tokens": usage.input_tokens,
    "output_tokens": usage.output_tokens,
    "cached_tokens": usage.cache_read_input_tokens or 0,
    "system_tokens": budget.reserved_system,
    "history_tokens": history_token_count,
    "current_tokens": current_message_tokens,
    "budget_utilization_pct": (usage.input_tokens / budget.total_window) * 100,
    "compression_applied": compression_strategy or None,
})
The insight most teams miss: context overflow is preceded by weeks of measurable creep. The utilization graph rises 2-3% per week as conversations get longer, users request bigger inputs, or the corpus grows. The overflow itself is the reveal of a slow drift you could have caught a month earlier with a dashboard.
Chapter 12

The complete ContextManager implementation

Every idea in this guide, packaged. This is what you drop into a production service to have context management that survives real traffic.

import time, json, logging, hashlib
from dataclasses import dataclass, field
from typing import List, Dict, Any, Optional, Callable, Literal
from enum import Enum

logger = logging.getLogger(__name__)


@dataclass
class ContextBudget:
    total_window: int
    reserved_output: int
    reserved_system: int
    reserved_current_turn: int
    safety_margin_pct: float = 0.05
    
    @property
    def safety_margin(self) -> int:
        return int(self.total_window * self.safety_margin_pct)
    
    @property
    def available_for_history(self) -> int:
        return (self.total_window
                - self.reserved_output
                - self.reserved_system
                - self.reserved_current_turn
                - self.safety_margin)


class CompressionStrategy(Enum):
    NONE = "none"
    SLIDING_WINDOW = "sliding_window"
    ROLLING_SUMMARY = "rolling_summary"
    HIERARCHICAL = "hierarchical"


@dataclass
class ContextManagerConfig:
    budget: ContextBudget
    compression: CompressionStrategy = CompressionStrategy.ROLLING_SUMMARY
    sliding_keep_turns: int = 20
    summary_threshold_pct: float = 0.7  # trigger at 70% of budget
    summary_keep_recent: int = 10
    summary_model: str = "claude-haiku-4-5"
    tool_result_max_tokens: int = 2000
    aged_tool_result_keep: int = 5


class ContextManager:
    def __init__(self, config: ContextManagerConfig, token_counter: Callable,
                 summarizer: Optional[Callable] = None):
        self.config = config
        self.count_tokens = token_counter
        self.summarize = summarizer
        self._summary_cache = None
        self._summary_up_to_index = 0
    
    def build_context(self, system: str, tools: List[dict], 
                     messages: List[dict], current_message: dict) -> dict:
        # Step 1: compress tool results in place
        messages = self._compress_tool_results(messages)
        messages = self._age_tool_results(messages)
        
        # Step 2: measure current usage
        system_tokens = self.count_tokens(system)
        tools_tokens = self.count_tokens(json.dumps(tools)) if tools else 0
        current_tokens = self.count_tokens(current_message.get("content", ""))
        history_tokens = sum(
            self.count_tokens(str(m.get("content", ""))) for m in messages
        )
        
        # Step 3: check if compression needed
        available = self.config.budget.available_for_history
        threshold = int(available * self.config.summary_threshold_pct)
        
        compression_applied = None
        if history_tokens > threshold:
            messages, compression_applied = self._compress_history(messages)
            history_tokens = sum(
                self.count_tokens(str(m.get("content", ""))) for m in messages
            )
        
        # Step 4: if still over, truncate as last resort
        if history_tokens > available:
            messages = self._truncate_middle(messages, available)
            compression_applied = "truncate_middle"
        
        # Step 5: log utilization
        total_input = system_tokens + tools_tokens + history_tokens + current_tokens
        utilization_pct = (total_input / self.config.budget.total_window) * 100
        logger.info("context_built", extra={
            "system_tokens": system_tokens,
            "tools_tokens": tools_tokens,
            "history_tokens": history_tokens,
            "current_tokens": current_tokens,
            "total_input": total_input,
            "utilization_pct": utilization_pct,
            "compression_applied": compression_applied,
        })
        
        return {
            "system": system,
            "tools": tools,
            "messages": messages + [current_message],
            "metadata": {
                "utilization_pct": utilization_pct,
                "compression_applied": compression_applied,
            }
        }
    
    def _compress_history(self, messages):
        strategy = self.config.compression
        if strategy == CompressionStrategy.SLIDING_WINDOW:
            return self._sliding_window(messages), "sliding_window"
        if strategy == CompressionStrategy.ROLLING_SUMMARY:
            return self._rolling_summary(messages), "rolling_summary"
        if strategy == CompressionStrategy.HIERARCHICAL:
            return self._hierarchical(messages), "hierarchical"
        return messages, None
    
    def _sliding_window(self, messages):
        n = self.config.sliding_keep_turns
        if len(messages) <= n:
            return messages
        return messages[:1] + messages[-n:]
    
    def _rolling_summary(self, messages):
        if not self.summarize:
            return self._sliding_window(messages)
        keep = self.config.summary_keep_recent
        to_summarize = messages[:-keep]
        verbatim = messages[-keep:]
        if not to_summarize:
            return messages
        
        # Cache-friendly: reuse prior summary if new content hasn't grown much
        if self._summary_cache and len(to_summarize) <= self._summary_up_to_index + 5:
            summary_text = self._summary_cache
        else:
            summary_text = self.summarize(to_summarize)
            self._summary_cache = summary_text
            self._summary_up_to_index = len(to_summarize)
        
        return [
            {"role": "user", "content": "Earlier conversation summary:"},
            {"role": "assistant", "content": summary_text},
        ] + verbatim
    
    def _hierarchical(self, messages):
        total = len(messages)
        recent = messages[max(0, total-10):]
        medium = messages[max(0, total-50):total-10]
        ancient = messages[:max(0, total-50)]
        
        result = []
        if ancient and self.summarize:
            result.append({"role": "user", "content": "Earlier context:"})
            result.append({
                "role": "assistant",
                "content": self.summarize(ancient, target_tokens=500)
            })
        if medium and self.summarize:
            summary = self.summarize(medium, target_tokens=800)
            result.append({"role": "user", "content": "Recent context summary:"})
            result.append({"role": "assistant", "content": summary})
        result.extend(recent)
        return result
    
    def _truncate_middle(self, messages, target_tokens):
        if len(messages) < 6:
            return messages
        result = messages[:2] + messages[-4:]
        for i in range(len(messages) - 5, 2, -1):
            candidate = messages[:2] + messages[i:i+1] + messages[-4:]
            if sum(self.count_tokens(str(m.get("content", ""))) 
                   for m in candidate) < target_tokens:
                result = candidate
            else:
                break
        marker = {"role": "user", "content": 
                  f"[Note: {len(messages) - len(result)} earlier turns omitted]"}
        return result[:2] + [marker] + result[2:]
    
    def _compress_tool_results(self, messages):
        for m in messages:
            if not isinstance(m.get("content"), list):
                continue
            for block in m["content"]:
                if block.get("type") != "tool_result":
                    continue
                content = block.get("content", "")
                if isinstance(content, str):
                    content_str = content
                else:
                    content_str = json.dumps(content)
                tokens = self.count_tokens(content_str)
                if tokens > self.config.tool_result_max_tokens:
                    ratio = self.config.tool_result_max_tokens / tokens
                    keep = int(len(content_str) * ratio)
                    block["content"] = (
                        content_str[:keep]
                        + f"\n\n[Truncated: {self.config.tool_result_max_tokens} of {tokens} tokens]"
                    )
        return messages
    
    def _age_tool_results(self, messages):
        # Find tool_result blocks in order
        tool_result_locations = []
        for i, m in enumerate(messages):
            if isinstance(m.get("content"), list):
                for j, block in enumerate(m["content"]):
                    if block.get("type") == "tool_result":
                        tool_result_locations.append((i, j))
        
        keep = self.config.aged_tool_result_keep
        to_age = tool_result_locations[:-keep] if len(tool_result_locations) > keep else []
        
        for i, j in to_age:
            block = messages[i]["content"][j]
            content = block.get("content", "")
            if isinstance(content, str) and len(content) > 200:
                tokens = self.count_tokens(content)
                block["content"] = f"[Aged out tool result, was ~{tokens} tokens]"
        return messages


# Usage
from anthropic import Anthropic
client = Anthropic()

def claude_summarizer(messages, target_tokens=1500):
    text = "\n\n".join(f"{m['role']}: {m.get('content','')}" for m in messages)
    r = client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=target_tokens,
        system=("Summarize the following conversation in 3-5 paragraphs. "
                "Preserve: user goals, decisions, personal details, "
                "unresolved questions. Omit: filler."),
        messages=[{"role": "user", "content": text}]
    )
    return r.content[0].text

def claude_token_counter(text):
    r = client.messages.count_tokens(
        model="claude-sonnet-4-6",
        messages=[{"role": "user", "content": text or ""}]
    )
    return r.input_tokens

config = ContextManagerConfig(
    budget=ContextBudget(
        total_window=200_000,
        reserved_output=4_000,
        reserved_system=2_000,
        reserved_current_turn=1_000,
    ),
    compression=CompressionStrategy.ROLLING_SUMMARY,
    summary_threshold_pct=0.7,
    summary_keep_recent=10,
)

ctx_mgr = ContextManager(
    config=config,
    token_counter=claude_token_counter,
    summarizer=claude_summarizer,
)

# In your request handler
prepared = ctx_mgr.build_context(
    system=SYSTEM_PROMPT,
    tools=TOOL_DEFS,
    messages=conversation_history,
    current_message={"role": "user", "content": user_input}
)

response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=4000,
    system=prepared["system"],
    tools=prepared["tools"],
    messages=prepared["messages"],
)

What this gives you

  • Explicit context budget with all four buckets.
  • Three compression strategies to choose from.
  • Tool result compression and aging built in.
  • Middle-truncation as last-resort safety.
  • Utilization logging on every request.
  • Summary caching that plays well with prompt caching.
  • Pluggable token counter and summarizer (adapt for any provider).

What it doesn't do

  • RAG — handled outside; feed retrieved chunks as part of the current message.
  • Provider-neutral format — assumes you've translated to per-provider shape upstream (see G4 for that layer).
  • Distributed state — summary cache is per-instance. For multi-instance apps, back it with Redis.
  • Multi-modal counting — assumes text; extend the counter for images and attachments.
This closes the context management foundation. Combined with the reliability quartet (G1-G4), you now have both dimensions of production AI covered: what to do when providers fail, and what to do when context grows. The upcoming G6-G8 build on this foundation for specific patterns: Vercel AI SDK, function calling reliability, and structured output.
Chapter 13

Frequently asked questions

What's the difference between a token and a word?

Tokens are the units the model actually processes. In English prose, one token averages about 4 characters or 0.75 words. Code, non-English text, and JSON tokenize differently. Never estimate by word count; use the actual tokenizer or count_tokens API.

Which is better: RAG or stuffing the entire document?

For documents under ~20K tokens, stuffing is simpler and often works better. For larger documents, RAG becomes necessary. For medium-sized content, a hybrid pattern (stuff full document + emphasize retrieved sections) often wins.

Does prompt caching help with context overflow?

It doesn't reduce the token count sent — the full prompt still needs to fit. But it dramatically reduces cost and latency for long stable prefixes. Context management and caching are complementary, not alternatives.

Why is my context utilization slowly rising over weeks?

Usually one of three: users are having longer conversations, retrieved chunks are growing, or tool outputs are getting larger. Instrument per-bucket usage (Chapter 11) to identify which is drifting.

What's the 'lost in the middle' phenomenon?

Empirical finding that models attend most strongly to content at the beginning and end of long contexts. Content in the middle third is often recalled 30-50% worse. Position important content at the edges, not in the middle.

Should I compress before or after RAG?

Compress conversation history first (Chapter 6), then run RAG on the current query, then assemble. RAG results go at the end (high-attention zone); compressed history goes earlier.

How do I choose a compression strategy?

Start with sliding window (simplest). Move to rolling summary when users complain the model 'forgot' things. Hierarchical only if measurement shows rolling summary loses important detail. Chapter 6 has the decision tree.

Does Claude, GPT, or Gemini tokenize differently?

Yes. Each provider has its own tokenizer. Same text can tokenize to different counts across providers — usually within 5-10% for English, more for code or non-English. Use each provider's canonical counter.

What's a safe context budget for a chat app?

For Claude Sonnet's 200K window: reserve 4K for output, 2K for system + tools, 1K for current turn, 10K safety margin. That leaves ~183K for history. Adjust based on your actual max response length.

How often does prompt caching invalidate?

Any change to the prefix invalidates from that point forward. Adding messages at the end preserves the cache; re-summarizing the middle breaks it. Cache-friendly patterns keep the prefix stable.

Should tool results always be compressed?

Only when they're over ~2K tokens. Small tool outputs (weather, quick lookups) should stay verbatim. Large outputs (search results, file contents) benefit from compression. Aging out old tool results is even more impactful.

What about images and PDFs in the context budget?

They cost tokens too — roughly 1600 tokens per image on Claude, 258 on Gemini. Include image count in your budget calculation. For long conversations with images, aging out old images (replacing with '[image dropped]' markers) is often necessary.

Can I share the same summary across sessions?

Yes — that's the persistent memory pattern. Store the rolling summary in your database keyed by conversation ID. When the user returns, load it and prepend to the fresh conversation. Gives long-term memory without exploding context.

Should I use middle-truncation for chat?

Rarely. Middle truncation preserves the first-turn context (often important for chat continuity) and recent turns. But summary or sliding window usually produces more coherent conversations. Use middle-truncation as a last resort.

How do I know if I'm using context well?

Three metrics: p95 context utilization (should be 40-70% under normal load), cache hit rate (should be >50% for repeat sessions), compression event rate (should fire in the background steadily, not on every request). Chapter 11 covers the dashboard.

Chapter 14

Library & further reading

Everywhere else on AI Error Hub that touches context management, prompt engineering, and long-context patterns.