Your agent handles a 10-turn task fine. At turn 40 it re-reads a file it already read at turn 12, forgets a constraint you put in the system prompt, and answers a question it answered twenty minutes ago. Nothing crashed. The context window is not full either. It is just crowded, and the model is spending attention on stale tool output instead of your instructions.
That is measurable, not folklore. Chroma tested 18 models across 8 input lengths and 11 needle positions and found accuracy dropping as input grows, unevenly, and dropping faster when the question and the answer do not share vocabulary. Their conclusion: models do not use context uniformly.
Anthropic frames the same effect as an attention budget. Every token you add depletes it, and a transformer builds n² pairwise token relationships, so focus stretches thin as the window fills. The fix is not a bigger window. It is deciding what the agent carries in context and what it leaves in a file.
Four moves do the work here. This walkthrough builds all four in one runnable Python file, with token counts to show what each one buys.
What you need
- Python 3.11 or newer
pip install anthropicANTHROPIC_API_KEYin your environment- a folder of markdown files to search, any docs directory will do
- budget for a few API calls per run
The examples use claude-sonnet-4-5. Swap in whatever model you pay for.
Step 1: measure before you change anything
Tools and a searchable corpus first.
# agent.py
import json
from pathlib import Path
from anthropic import Anthropic
MODEL = "claude-sonnet-4-5"
client = Anthropic()
CORPUS = Path("corpus")
def search_docs(query: str) -> str:
hits = []
for path in sorted(CORPUS.glob("*.md")):
text = path.read_text()
if query.lower() in text.lower():
hits.append(f"{path.name} :: {text[:200]}")
return "\n".join(hits[:10]) or "no matches"
def read_doc(name: str) -> str:
path = (CORPUS / name).resolve()
if CORPUS.resolve() not in path.parents:
return "refused: outside corpus"
return path.read_text()[:8000]
HANDLERS = {"search_docs": search_docs, "read_doc": read_doc}
TOOLS = [
{
"name": "search_docs",
"description": "Search corpus files. Returns file name plus a 200 character snippet.",
"input_schema": {
"type": "object",
"properties": {"query": {"type": "string"}},
"required": ["query"],
},
},
{
"name": "read_doc",
"description": "Read one corpus file by name, truncated to 8000 characters.",
"input_schema": {
"type": "object",
"properties": {"name": {"type": "string"}},
"required": ["name"],
},
},
]
Now the loop with token accounting attached.
SYSTEM = """You answer questions about the local corpus.
Cite the file each claim came from. Keep answers short."""
def run_tool(block):
return {
"type": "tool_result",
"tool_use_id": block.id,
"content": HANDLERS[block.name](**block.input),
}
def budget(messages) -> int:
return client.messages.count_tokens(
model=MODEL, messages=messages, tools=TOOLS
).input_tokens
def chat(messages, **kwargs):
response = client.messages.create(
model=MODEL, max_tokens=4096, system=SYSTEM,
messages=messages, tools=TOOLS, **kwargs
)
usage = response.usage
print(f" in={usage.input_tokens:,} out={usage.output_tokens:,}")
return response
def main(question: str) -> None:
messages = [{"role": "user", "content": question}]
for turn in range(40):
response = chat(messages)
messages.append({"role": "assistant", "content": response.content})
if response.stop_reason != "tool_use":
print(response.content[-1].text)
return
messages.append({
"role": "user",
"content": [run_tool(b) for b in response.content if b.type == "tool_use"],
})
Run one question from a fresh corpus and watch the in= column. It climbs every turn because every tool result stays in the payload, even the ones the agent finished with five turns ago. That number is your baseline, and everything below exists to bend it.
Step 2: write the things you cannot afford to lose into a file
Structured note-taking means the agent writes findings to storage outside the window and reads them back when it needs them. Notes survive compaction, restarts, and a killed process, and nothing else in this article does.
Two tools, backed by one file.
NOTES = Path("notes/STATE.md")
NOTES.parent.mkdir(exist_ok=True)
def write_notes(content: str) -> str:
with NOTES.open("a") as handle:
handle.write(f"\n{content}\n")
return f"appended {len(content)} chars to {NOTES}"
def read_notes() -> str:
if not NOTES.exists():
return "no notes yet"
return NOTES.read_text()[-4000:]
HANDLERS.update({"write_notes": write_notes, "read_notes": read_notes})
TOOLS.extend([
{
"name": "write_notes",
"description": (
"Append a durable finding to notes/STATE.md. Use it for decisions, "
"file paths, dead ends and open questions. Survives compaction."
),
"input_schema": {
"type": "object",
"properties": {"content": {"type": "string"}},
"required": ["content"],
},
},
{
"name": "read_notes",
"description": "Read notes/STATE.md before starting work.",
"input_schema": {"type": "object", "properties": {}},
},
])
The tools only matter if the system prompt tells the agent when to reach for them.
SYSTEM = """You answer questions about the local corpus.
Cite the file each claim came from. Keep answers short.
Before your first search, call read_notes.
After any finding that changes your plan, call write_notes with the file path
and what it means for the task."""
Why this holds up: when compaction later wipes the transcript, the agent reloads a two kilobyte file instead of re-reading twelve documents. Anthropic's memory tool works on the same idea, a client-side file store the model reads and writes, and enabling it injects a system instruction telling the model to check that store before it starts and to record progress as it goes.
Make the notes file boring and specific. Paths, decisions, error text, open questions. A notes file full of prose summaries is another context problem wearing a different hat.
Step 3: stop paying for tool results you already read
The lightest form of compaction is throwing away tool results. Once the agent has processed a file's contents, the raw text sits in every subsequent request for no reason.
This one is server-side, one config block, and it needs the beta header.
CONTEXT_EDITS = {
"edits": [
{
"type": "clear_tool_uses_20250919",
"trigger": {"type": "input_tokens", "value": 30_000},
"keep": {"type": "tool_uses", "value": 3},
"clear_at_least": {"type": "input_tokens", "value": 5_000},
"exclude_tools": ["read_notes"],
}
]
}
response = client.beta.messages.create(
model=MODEL,
max_tokens=4096,
system=SYSTEM,
messages=messages,
tools=TOOLS,
betas=["context-management-2025-06-27"],
context_management=CONTEXT_EDITS,
)
Four knobs, each with a job. trigger sets the token count where clearing starts. keep preserves the most recent tool pairs so the agent keeps its current working state. clear_at_least refuses to clear unless it frees enough tokens to be worth it, which matters because clearing invalidates your cached prompt prefix. exclude_tools names results that never get evicted; use memory there if you use the built-in memory tool.
Set clear_at_least high enough that each clear pays for the cache rewrite it causes. A clear that frees 800 tokens and resets a 40,000 token cache is a net loss.
Step 4: compact the transcript when the transcript is the bulk
Tool result clearing cannot help when the messages themselves are large: a support agent with 200 short exchanges, or a session where someone keeps refining the goal. That is when you summarize and restart.
COMPACT_AT = 60_000
KEEP_TAIL = 6
COMPACT_PROMPT = """You compress an agent transcript so the agent can continue
working without losing anything task-critical.
Preserve verbatim where possible:
- the original request and every constraint the user stated
- file paths, commands, function names, error text
- decisions made, and the reason given at the time
- what was tried and failed, so it is not retried
- open questions and the current next step
Drop raw tool output that is already summarized, restatements and narration.
Write in the language the user used. Output only the summary."""
def flatten(message) -> str:
if isinstance(message["content"], str):
return message["content"]
parts = []
for block in message["content"]:
if block.type == "text":
parts.append(block.text)
elif block.type == "tool_use":
parts.append(f"[tool_use {block.name} {json.dumps(block.input)}]")
else:
parts.append(f"[{block.type}]")
return "\n".join(parts)
def compact(messages):
old, tail = messages[:-KEEP_TAIL], messages[-KEEP_TAIL:]
transcript = "\n\n".join(f"{m['role']}: {flatten(m)}" for m in old)
summary = client.messages.create(
model=MODEL,
max_tokens=2000,
system=COMPACT_PROMPT,
messages=[{"role": "user", "content": transcript}],
).content[0].text
return [{
"role": "user",
"content": f"<summary_of_earlier_work>\n{summary}\n</summary_of_earlier_work>",
}] + tail
Gate it inside the loop, and check the count every few turns rather than every turn, since count_tokens costs a round trip.
if turn % 5 == 0 and budget(messages) > COMPACT_AT:
before = budget(messages)
messages = compact(messages)
print(f" compacted {before:,} -> {budget(messages):,} tokens")
Three details decide whether compaction helps or quietly ruins the run.
Keep the tail raw. The last six messages hold the immediate working state, including tool results the agent is still reasoning over. Summarizing them costs more than it saves.
Tune recall before precision. Anthropic's guidance is to write the compaction prompt against real traces from failing runs: first make it capture everything relevant, then cut what turned out to be superfluous. A prompt tuned on examples you invented will not match how your agent actually goes wrong.
Tag the summary. <summary_of_earlier_work> tells the model this is a compression rather than something the user said, and it gives you something greppable in logs when a run goes sideways.
If maintaining that is not worth it, the Claude Developer Platform ships summary-based compaction as a server-side beta strategy next to tool result clearing. Same trade-off, less code.
Step 5: isolate the work in subagents instead of compressing it
Compression is lossy by design. Isolation is not. A subagent gets a clean context window, can burn tens of thousands of tokens searching, and hands back a short summary. Anthropic puts that summary around 1,000 to 2,000 tokens, which is the whole point: the parent pays for the conclusion, not the exploration.
Add a task tool that the parent calls like any other.
WORKER_SYSTEM = """You are a research worker with read-only tools.
Work until you can answer, then stop and reply in this exact shape:
FINDINGS:
- one line per finding, each with the file or command it came from
UNCERTAIN:
- anything you could not confirm
At most 400 words. Never paste file contents. Never ask a question back."""
def run_subagent(spec: str) -> str:
messages = [{"role": "user", "content": spec}]
try:
for _ in range(12):
response = client.messages.create(
model=MODEL, max_tokens=2000, system=WORKER_SYSTEM,
messages=messages, tools=TOOLS,
)
messages.append({"role": "assistant", "content": response.content})
if response.stop_reason != "tool_use":
return "".join(b.text for b in response.content if b.type == "text")
messages.append({
"role": "user",
"content": [run_tool(b) for b in response.content if b.type == "tool_use"],
})
except Exception as exc:
return f"worker failed: {exc}"
return "worker hit its turn limit without an answer"
Register it and write a description that tells the parent how to brief it.
HANDLERS["task"] = run_subagent
TOOLS.append({
"name": "task",
"description": (
"Run a subagent in a clean context window and get back a short summary. "
"Use it for breadth-first search across many files."
),
"input_schema": {
"type": "object",
"properties": {
"spec": {
"type": "string",
"description": "Self-contained brief. The worker sees none of this conversation.",
}
},
"required": ["spec"],
},
})
Two rules keep this from turning into a mess.
The spec must be self-contained. If the worker needs the user's constraint, restate it in the spec. A subagent that has to ask a follow-up question is a bug, because it has no path back.
Failures must return strings, not raise. With parallel workers one exception can cancel its siblings, so returning worker failed: ... lets the parent decide whether the gap matters.
To run several at once, cap the width and use threads, since the sync SDK has no event loop to hang asyncio.gather on.
from concurrent.futures import ThreadPoolExecutor
SPECS = [
"how is authentication configured",
"which endpoints write to the database",
"what is rate limited and why",
]
with ThreadPoolExecutor(max_workers=4) as pool:
summaries = list(pool.map(run_subagent, SPECS))
Then add one line to the parent's system prompt: Delegate breadth-first search with task. Answer directly when one or two reads will do. Without a rule about when to delegate, models tend to either never touch it or call it for every question.
Inside Claude Code the same idea is a markdown file in .claude/agents/.
---
name: corpus-researcher
description: Searches the local corpus for one question and returns a short evidence-backed summary. Use when an answer needs more than two file reads.
tools: Read, Grep, Glob
model: haiku
---
You search the corpus and answer one question at a time.
Report findings as bullets with file paths. Never return file contents verbatim.
Claude Code picks the file up within a few seconds, runs it in its own context window with only those tools, and returns the result to the main conversation. Leaving Agent out of the tools list stops it from spawning further subagents, which is usually what you want when you are paying per token.
When each one is the wrong choice
| Situation | Reach for |
|---|---|
| Long back-and-forth, conversational flow matters | compaction |
| Iterative work with clear milestones | notes file |
| Breadth-first search across many sources | subagents |
| Subtasks depend on each other's output | neither, chain them in one context |
| One or two tool calls | nothing, just call the model |
Subagents are the expensive option. Anthropic reports agents using roughly 4× the tokens of a chat interaction, and their multi-agent research system around 15×. That is a fair trade when parallel search is your bottleneck and the task is worth the spend, and a waste when a single call would have finished the job.
Two failure modes to watch. Synthesis loss, where the parent compresses a worker's report and drops the one detail that mattered, which is why the worker should return evidence (paths, commands) rather than conclusions alone. And non-independence, where subtask two secretly needs subtask one's output. Run those at the same time and you get confident garbage. That case wants one context and sequential steps.
Make it prove itself
Log input_tokens per request to a CSV, run the same 20 questions with the mechanisms off and then on, and compare accuracy and total tokens. Compaction thresholds and subagent briefs are exactly the kind of thing that looks fine in a demo and misbehaves on the 200th turn of a real session, and one run cannot show you that. Keep the notes file in git too. It doubles as a readable trace of what the agent believed at each step, which is more useful than a log line when you are debugging why it went down the wrong path.
References
- Effective context engineering for AI agents (Anthropic Engineering)
- Context Rot: How Increasing Input Tokens Impacts LLM Performance (Chroma)
- Managing context on the Claude Developer Platform (context editing + memory tool)
- Context editing docs (Claude Developer Platform)
- Create custom subagents (Claude Code docs)