上下文工程工具
管理长对话上下文的工程化方法
🌐 查看英文原文 | 源码 Notebook
AI Agent 的上下文工程：Memory vs. Compaction vs. Tool Clearing
简介
构建长周期 Agent 时，一个常见的挑战是管理上下文。工具结果、模型的自身推理和用户消息都会累积，最终要么达到 token 限制，要么开始为不再有帮助的上下文付费。针对针插稻草堆式基准测试的研究揭示了 上下文腐化 的概念：随着上下文窗口中 token 数量的增加，模型从该上下文中准确回忆信息的能力会降低。因此，甚至在达到硬上下文限制之前，Agent 可能从每个 token 中获得的收益就会减少。
我们关于 为 AI Agent 有效的上下文工程 的工程博客将此视为一个资源问题：上下文是有限的，边际收益递减，核心学科是找到一组最小的、高信号 token，以最大化你期望的结果的可能性。为此有几个杠杆：子 Agent 在自己的上下文中隔离工作，程序性工具调用将大结果完全排除在窗口之外，以及其他方法。 This cookbook focuses on three: 压缩、工具结果清理 and 内存. All three are effective strategies for context engineering, but since they all operate to make the context window more efficient in different ways, they can be hard to distinguish. Understanding those distinctions is what lets you map each tool to the part of your workload it actually helps with. Alongside other core context management strategies like utilizing subagents, these three are crucial for teams building long-running agents to understand. They also all have first-party API support, so you can adopt them without building orchestration infrastructure.
- 压缩 distills the contents of a context window into a high-fidelity summary, letting the agent continue with minimal performance degradation when the conversation gets long.
- 工具结果清理 addresses the bloat from tool use itself. As an agent pulls in tools and calls them, the results pile up, and deciding how much of that tool output to keep becomes an increasingly important part of managing context. Clearing drops old, re-fetchable results while keeping the record that the call happened.
- 内存 is structured note-taking: the agent writes to persistent external storage so it can track progress across tasks and sessions without keeping everything in active context. Claude Code在生产中采用了多种策略：用于长对话的压缩和两个互补的记忆系统以实现跨会话持久化。我们的API提供了所有三种策略的第一方实现：服务器端压缩、上下文编辑（包括清除工具结果）和记忆工具。本食谱将介绍如何使用它们进行设计：何时适用每个策略，如何配置它们，单独使用与一起使用时会有什么变化，以及不同组合的示例用例。
示例集中在长期运行的研究代理上：它读取文档集合，做笔记，并在多个会话中基于其发现进行构建。这是一个有用的测试案例，因为它自然地触及了所有三个问题：庞大的文档读取（清除）、长分析对话（压缩）以及需要在会话之间持续存在的知识（记忆）。
你将学到什么
- 当代理的上下文主要由大型的可重新获取的工具结果（如文件读取和API响应）主导时，如何使用
clear_tool_uses来限制会话中的token增长 - 如何使用服务器端压缩保持长对话的进行，包括如何序列化
compaction块并检查哪些内容在摘要中幸存 - 如何通过实现模型自行驱动的基于文件的内存处理程序来在会话之间持久化代理知识，以便会话2从会话1离开的地方继续
- 如何最有效地实现每个原语，替换默认的压缩提示以保留代理需要的知识，指导代理写入
/memories的内容，并针对自己的工作负载的工具使用模式测试清除配置 - 如何诊断工作负载实际存在的上下文问题部分，并选择针对该部分的原始方法，以及一个将工作负载特征映射到正确工具的框架
前置条件
运行此笔记本，您需要：
- Anthropic API密钥设置为环境中的
ANTHROPIC_API_KEY或.env文件（在此处获取） - Python 3.11+，并安装了
anthropic、python-dotenv和matplotlib包 - 与此笔记本一起的
research_corpus.py（包含在存储库中）。它定义了CORPUS，一个包含八个关于衰老研究模型生物的合成评论文档的字典（每个文档约40K个token，总计约320K个token），以及后来用于测试压缩后存活的探测问题。您可以通过替换字典来替换自己的文档。
从cookbooks存储库运行？在运行笔记本之前，请确保您的当前工作目录是
tool_use/context_engineering。
步骤 0：环境配置
在此目录中创建一个.env文件，并添加您的Anthropic API密钥：ANTHROPIC_API_KEY=your-key-here
%%capture
%pip install anthropic python-dotenv matplotlibimport json
import os
import tempfile
from collections import namedtuple
from pathlib import Path
import anthropic
import matplotlib.pyplot as plt
from dotenv import load_dotenv
load_dotenv()
if not os.environ.get("ANTHROPIC_API_KEY"):
raise ValueError("ANTHROPIC_API_KEY not set. Add it to a .env file or export it.")
CORPUS_PATH = Path("research_corpus.py")
assert CORPUS_PATH.exists(), (
f"research_corpus.py not found in {Path.cwd()}. It should be alongside this notebook."
)
client = anthropic.Anthropic()
MODEL = "claude-sonnet-4-6"
print(f"anthropic SDK {anthropic.__version__}, model {MODEL}")anthropic SDK 0.84.0, model claude-sonnet-4-6# Force-reload the corpus module in case this kernel has a stale cached version
import importlib
import research_corpus
importlib.reload(research_corpus)
from research_corpus import COMPACTION_PROBES, CORPUS
# Count tokens via the API (cached in-memory so repeat kernel runs are fast).
_token_cache: dict[str, int] = {}
def count_tokens(text: str) -> int:
if text not in _token_cache:
_token_cache[text] = client.messages.count_tokens(
model=MODEL, messages=[{"role": "user", "content": text}]
).input_tokens
return _token_cache[text]
print(f"CORPUS is a dict of {len(CORPUS)} synthetic documents held in Python memory.")
print("When the agent calls read_file, the content is served from this dict and")
print("lands directly in the agent's context window — no disk I/O involved.\n")
_total_tokens = 0
for path, content in CORPUS.items():
n_tok = count_tokens(content)
_total_tokens += n_tok
# Strip the virtual directory prefix for display clarity
display_name = path.removeprefix("/research/")
print(f" {display_name:<26} ~{n_tok:>6,} tokens")
print(f"\n Total corpus: ~{_total_tokens:,} tokens")
# Sanity check: the corpus should be large enough to push context well past
# 200K in a single read-pass. If this fires, research_corpus.py may be stale.
assert _total_tokens > 250_000, (
f"Corpus is only {_total_tokens:,} tokens; expected >250K. "
"Restart the kernel and re-run, or verify research_corpus.py is current."
)CORPUS is a dict of 8 synthetic documents held in Python memory.
When the agent calls read_file, the content is served from this dict and
lands directly in the agent's context window — no disk I/O involved.
celegans_review.md ~41,260 tokens
drosophila_review.md ~41,200 tokens
mouse_review.md ~41,155 tokens
zebrafish_review.md ~41,095 tokens
killifish_review.md ~41,115 tokens
yeast_review.md ~41,101 tokens
nmr_review.md ~41,017 tokens
rhesus_review.md ~41,012 tokens
Total corpus: ~328,955 tokens
```## 问题：长时间运行的研究 Agent
在这个食谱中，该 Agent 扮演着一位生物学研究者的角色，为衰老和长寿研究撰写模型生物比较综述。这项任务足够现实，值得注意：它涉及阅读一系列综述文档（每个生物一个），提取可比事实（寿命、遗传可操作性、转化相关性），做结构化笔记，并综合阅读到的所有发现。
这种工作正是上下文管理开始发挥作用的地方。每份文档大约有 40K 个 token（叙述加上大量干预数据附录表格），任务要求 Agent 分两批阅读：首先阅读四个高吞吐量生物（秀丽线虫、果蝇、酵母、杀鲶鱼），然后阅读四个低吞吐量生物（小鼠、斑马鱼、裸鼹鼠、恒河猴）。两批结构是本食谱的实验设计选择：它产生一个上下文轨迹，超过第一批的压缩触发点，并在第二批超过 200K 参考线，因此每个原始操作对轨迹的影响在同一运行中都是可见的。如果没有上下文管理，Agent 的上下文在任务中途会增长到数十万个 token。而且由于工作跨越会话，即使是完成的一次运行，在下一会话开始时也没有记住学到了什么。
### 研究任务
Agent 的具体任务是：在三个维度上比较 `/research/` 中的模型生物（寿命和实验吞吐量、遗传可操作性、与人类衰老的转化相关性），分两批阅读八篇综述文档并边读边做笔记，然后撰写比较综述。
## 三个 API 如何映射到问题
每个 API 针对不同的上下文增长类型。了解你面临的是哪种类型是选择正确工具的第一步。
### 概念上
::: callout-conceptually
:::
**压缩**是将接近上下文窗口限制的对话内容进行总结，并以此总结重新开始的做法。它的目的是以高保真度的方式提炼上下文窗口，以便代理可以以最小的性能下降继续进行。压缩的艺术在于保留什么与丢弃什么：过于激进的压缩可能会丢失微妙但至关重要的上下文，其重要性只有在之后才会变得明显。总结保留了架构决策、未解决的问题和关键事实，同时丢弃了冗余内容；它是有损的，但处理了所有上下文增长，而不仅仅是工具结果。压缩是一个**完整转录**操作：用户消息、助手消息、工具调用、工具结果，甚至之前的压缩块都被压缩到总结中。
相比之下，**工具结果清除**是一个**子转录**操作。它遍历消息列表，并外科手术般地替换`tool_result`内容块，而其他所有内容——用户消息、助手推理、`tool_use`记录——则保持不变。当代理调用工具时，结果成为对话历史的一部分，并在每个后续回合中计入上下文预算。其中大部分内容是可以重新获取的：代理可以重新读取的文件内容，它可以重新请求的API响应。清除用简短的占位符替换旧的`tool_result`块，保留`tool_use`记录，以便模型仍然知道它已经进行了调用，但丢弃了庞大的有效载荷。一旦工具在消息历史中调用得很深，代理很少需要再次看到原始结果；清除是恢复空间最安全、最轻触的方式之一。如果代理需要数据，它只需再次调用工具即可。
::: callout-注意
请注意，清除操作不会影响对话的上下文，它只是从对话历史中移除了工具结果。
:::
**记忆**，或结构化笔记，是一种技术，其中代理定期将笔记持久化存储在上下文窗口之外，然后在稍后时间将其拉回。这提供了具有最小开销的持久记忆：代理跟踪复杂任务的进度，保持否则会在数十次工具调用或上下文重置中丢失的关键上下文。在重置（新会话或压缩后）之后，代理读取自己的笔记并继续。您实现存储后端，因此您控制存储的内容和时长。
除了启用这些基本功能之外，了解如何最有效地实现它们也很重要：默认行为可以帮助您开始，但压缩摘要的质量和存储在内存中的内容的实用性都取决于您提供的指导。以下每个基本功能的章节都包括一个关于有效实现的子章节。
### 战术上
| API | Identifier | Beta header | Triggered by | Configurable knobs |
|---|---|---|---|---|
| Compaction | `compact_20260112` | `compact-2026-01-12` | Token threshold (server-side, min 50K) | `trigger` (default 150K), `instructions`, `pause_after_compaction` |
| Tool clearing | `clear_tool_uses_20250919` | `context-management-2025-06-27` | Token threshold (server-side) | `trigger` (default 100K), `keep` (default 3 tool uses), `clear_at_least`, `exclude_tools`, `clear_tool_inputs` |
| Memory tool | `memory_20250818` | none (standalone) | The model (it's a tool call) | Client implements: `view`, `create`, `str_replace`, `insert`, `delete`, `rename` |
### 映射到研究 Agent
对于研究代理而言，三个问题可以清晰地对应：
- 代理的运行评论（“C. elegans 的生命周期为 18 天，全基因组 RNAi，小鼠为 30 个月，但每个队列的成本为 10 万美元...”）和用户的后续问题累积成一个长对话。这是一个 **压缩** 问题。
- 阅读八个 ~40K-token 的评论文档会产生大约 320K 个 token 的工具结果量，这显著进入了模型性能因上下文退化而下降的范围。其中大部分是代理可以按需重新阅读的。这是一个 **清除** 问题。
- 工作跨越多个会话。如果第 1 次会话确定斗鱼是寿命最短的脊椎动物（4-6 个月），我们希望第 2 次会话保留这一发现并在此基础上进行扩展，而不是从头开始重新发现。这是一个 **记忆** 问题。
一个粗略的心智模型用于优先级排序：当窗口增长过大时，压缩会压缩整个窗口，清除会清除窗口内过时的可重新获取的数据，而记忆会将信息移出窗口，以便在会话之间存活。每一层都增加了配置以调整和交互以理解，因此最好从与您实际观察到的瓶颈相匹配的那一层开始。
## 研究 Agent
在探索每个基本元素之前，我们首先设置代理本身：工具模式、工具执行和可以带或不带任何上下文管理配置运行的代理循环。所有内容都是内联的，这样您可以看到完整的循环。
```python
def agent_loop():
# 代理循环代码
pass请注意，代理循环代码的具体实现将取决于您的具体需求和上下文。 :::```python # ── Tool schemas ───────────────────────────────────────────────────────── RESEARCH_TASK = “““Compare the model organisms in /research/ on three dimensions: 1. Lifespan and experimental throughput (how fast can you get a lifespan result?) 2. Genetic tractability (what tools exist for manipulation and screening?) 3. Translational relevance (how well do findings transfer to human aging?)
Work through the organisms in two batches, recording a note after each batch before moving on:
BATCH 1 (high-throughput models): Read celegans_review.md, drosophila_review.md, yeast_review.md, and killifish_review.md. Record a single detailed note covering all four organisms on all three dimensions. Do not begin Batch 2 until this note is recorded.
BATCH 2 (low-throughput models): Read mouse_review.md, zebrafish_review.md, nmr_review.md, and rhesus_review.md. Record a single detailed note covering all four organisms on all three dimensions.
After both batches, write a comprehensive comparative synthesis contrasting the high-throughput and low-throughput models.”“”
BASE_TOOLS: list[dict] = [ { “name”: “search_files”, “description”: “Search document filenames and contents for a keyword. Returns matching paths with a short excerpt.”, “input_schema”: { “type”: “object”, “properties”: {“query”: {“type”: “string”, “description”: “Search term”}}, “required”: [“query”], }, }, { “name”: “read_file”, “description”: “Read the full content of a document by path.”, “input_schema”: { “type”: “object”, “properties”: { “path”: {“type”: “string”, “description”: “e.g. /research/celegans_review.md”} }, “required”: [“path”], }, }, { “name”: “record_finding”, “description”: “Record a research finding (freeform text). Findings are held for the duration of this session only; they are not persisted across sessions.”, “input_schema”: { “type”: “object”, “properties”: {“finding”: {“type”: “string”}}, “required”: [“finding”], }, },]
The memory tool spec is defined here but only added to the agent’s tool
list when a memory_handler is passed to run_research_session. Baseline
and clearing/compaction runs do NOT get the memory tool.
MEMORY_TOOL_SPEC = {“type”: “memory_20250818”, “name”: “memory”}
SYSTEM_PROMPT = “““You are a biology research analyst writing a comparative review of model organisms for aging research.
Work systematically: search and read review documents, take notes on key facts (lifespans, genetic tools, translational strengths and caveats), and build a comparative picture.
Be concise in your reasoning text; the goal is notes, not essays.”“”
```python
# ── Tool execution ───────────────────────────────────────────────────────
def execute_research_tool(name: str, tool_input: dict, notes: list[str]) -> str:
if name == "search_files":
query = tool_input.get("query", "").lower()
if not query:
return "Error: query is required."
hits: list[str] = []
for path, content in CORPUS.items():
if query in path.lower() or query in content.lower():
for line in content.split("\n"):
if query in line.lower():
hits.append(f" {path}\n → {line.strip()[:120]}")
break
else:
hits.append(f" {path}")
return (
f"Found {len(hits)} match(es):\n" + "\n".join(hits)
if hits
else f"No matches for '{query}'."
)
if name == "read_file":
path = tool_input.get("path", "")
content = CORPUS.get(path)
if content is None:
return f"Error: '{path}' not found. Available: {', '.join(CORPUS.keys())}"
return content
if name == "record_finding":
finding = tool_input.get("finding", "")
notes.append(finding)
return f"Finding #{len(notes)} recorded (session-local)."
return f"Error: unknown tool '{name}'"# ── Session result container ─────────────────────────────────────────────
SessionResult = namedtuple(
"SessionResult",
[
"messages", # final message list
"notes", # notes taken this session
"token_trajectory", # list of (turn, total_context_tokens)
"events", # list of dicts describing compaction / clearing events
"tool_counts", # dict of tool_name -> call count
"file_reads", # list of (turn, path) for each read_file call
"hit_limit", # True if the session stopped because it hit the context window
"final_text", # final assistant text (convenience)
],
)
def _format_tool_arg(name: str, tool_input: dict) -> str:
"""Human-readable one-line summary of a tool call's arguments for verbose output."""
if name == "search_files":
return repr(tool_input.get("query", ""))
if name == "read_file":
return tool_input.get("path", "?")
if name == "record_finding":
finding = tool_input.get("finding", "")
preview = finding[:50].replace("\n", " ")
return f'"{preview}{"..." if len(finding) > 50 else ""}"'
if name == "memory":
cmd = tool_input.get("command", "?")
path = tool_input.get("path", tool_input.get("old_path", ""))
return f"{cmd} {path}"
return str(tool_input)[:60]
# ── Agent loop ───────────────────────────────────────────────────────────
def run_research_session(
initial_prompt: str,
*,
context_management: dict | None = None,
betas: list[str] | None = None,
memory_handler=None,
max_turns: int = 12,
label: str = "session",
verbose: bool = True,
) -> SessionResult:
"""Run the research agent. Catches context-window overflow gracefully."""
tools = list(BASE_TOOLS)
if memory_handler is not None:
tools.append(MEMORY_TOOL_SPEC)
messages: list[dict] = [{"role": "user", "content": initial_prompt}]
notes: list[str] = []
token_trajectory: list[tuple[int, int]] = []
events: list[dict] = []
tool_counts: dict[str, int] = {}
file_reads: list[tuple[int, str]] = [] # (turn, path) for each read_file
hit_limit = False
final_text = ""
if verbose:
print(f"┌─ [{label}]")
for turn in range(1, max_turns + 1):
kwargs: dict = dict(
model=MODEL,
max_tokens=4096,
system=SYSTEM_PROMPT,
tools=tools,
messages=messages,
)
if context_management:
kwargs["context_management"] = context_management
if betas:
kwargs["betas"] = betas
# Call the API; catch context-window overflow
try:
if betas:
response = client.beta.messages.create(**kwargs)
else:
response = client.messages.create(**kwargs)
except anthropic.BadRequestError as e:
# Context window exceeded (or similar input-too-large error)
hit_limit = True
if verbose:
print(f"│ ⚠ CONTEXT WINDOW LIMIT REACHED at turn {turn} (API rejected)")
print(f"│ {str(e)[:200]}")
break
# Track TOTAL context size: uncached + cache-read + cache-created.
# usage.input_tokens alone excludes cached tokens, which makes the
# plot show dips that are just cache hits, not context management.
u = response.usage
total_in = (
u.input_tokens
+ (getattr(u, "cache_read_input_tokens", None) or 0)
+ (getattr(u, "cache_creation_input_tokens", None) or 0)
)
token_trajectory.append((turn, total_in))
# Surface context-management events on their own prominent lines
cm = getattr(response, "context_management", None)
if cm is not None and getattr(cm, "applied_edits", None):
for edit in cm.applied_edits:
cleared_uses = getattr(edit, "cleared_tool_uses", None)
cleared_toks = getattr(edit, "cleared_input_tokens", None)
events.append(
{
"turn": turn,
"kind": "clearing",
"cleared_tool_uses": cleared_uses,
"cleared_input_tokens": cleared_toks,
}
)
if verbose:
print(
f"│ ✂ CLEARING (turn {turn}): {cleared_uses or '?'} tool results cleared, "
f"~{cleared_toks:,} tokens freed"
if cleared_toks
else f"│ ✂ CLEARING (turn {turn}): applied"
)
# Serialize assistant content, track tool calls and compaction
serialized: list[dict] = []
tool_calls: list[dict] = []
turn_tool_calls: list[tuple[str, dict]] = [] # (name, input) for verbose display
compaction_this_turn = False
for block in response.content:
if block.type == "text":
serialized.append({"type": "text", "text": block.text})
if block.text.strip():
final_text = block.text
elif block.type == "tool_use":
serialized.append(
{
"type": "tool_use",
"id": block.id,
"name": block.name,
"input": block.input,
}
)
tool_calls.append({"id": block.id, "name": block.name, "input": block.input})
turn_tool_calls.append((block.name, block.input))
tool_counts[block.name] = tool_counts.get(block.name, 0) + 1
# Track file reads specifically so we can show what clearing drops
if block.name == "read_file":
file_reads.append((turn, block.input.get("path", "?")))
elif block.type == "thinking":
serialized.append(
{"type": "thinking", "thinking": block.thinking, "signature": block.signature}
)
elif block.type == "compaction":
serialized.append({"type": "compaction", "content": block.content})
events.append({"turn": turn, "kind": "compaction", "summary": block.content})
compaction_this_turn = True
if verbose:
print(
f"│ ⊟ COMPACTION (turn {turn}): "
f"~{count_tokens(block.content):,}-token summary replaces prior turns"
)
messages.append({"role": "assistant", "content": serialized})
if not tool_calls:
if verbose and not compaction_this_turn:
print(f"│ turn {turn:2d} ctx={total_in:>7,} (final answer)")
break
# Execute tools, collecting result sizes so the verbose print
# can show how much each call added to context
tool_results: list[dict] = []
result_sizes: list[int] = []
for call in tool_calls:
if call["name"] == "memory" and memory_handler is not None:
result = memory_handler.handle(call["input"])
else:
result = execute_research_tool(call["name"], call["input"], notes)
tool_results.append(
{
"type": "tool_result",
"tool_use_id": call["id"],
"content": result,
}
)
result_sizes.append(len(result) if isinstance(result, str) else 0)
messages.append({"role": "user", "content": tool_results})
if verbose and not compaction_this_turn:
# Header with context size, then one line per tool call with
# its argument and approximate result size (so you can see
# which calls are responsible for the next turn's ctx jump)
print(f"│ turn {turn:2d} ctx={total_in:>7,}")
for (name, tinput), rsize in zip(turn_tool_calls, result_sizes, strict=False):
size_note = f" → ~{rsize // 4:,} tok" if rsize > 200 else ""
print(f"│ {name:<16} {_format_tool_arg(name, tinput)}{size_note}")
if verbose:
if token_trajectory:
peak = max(t for _, t in token_trajectory)
status = "⚠ HIT CONTEXT LIMIT" if hit_limit else "completed"
print(
f"└─ {status}: {len(token_trajectory)} turns, peak ctx {peak:,}, "
f"final ctx {token_trajectory[-1][1]:,}, {len(events)} context event(s)\n"
)
else:
print("└─ HIT CONTEXT LIMIT on first turn: 0 turns completed\n")
return SessionResult(
messages, notes, token_trajectory, events, tool_counts, file_reads, hit_limit, final_text
)
def show_cleared_reads(result: SessionResult, keep: int):
"""Show which file reads are no longer in context after clearing.
Clearing replaces tool results older than the last `keep` tool uses
with placeholders. We reconstruct the tool-use order and mark any
read that falls outside the surviving `keep`-window as cleared.
Note: if clearing fires multiple times, only the last event's
boundary is considered and earlier cleared-then-re-read files may
be misclassified.
"""
if not result.file_reads:
print("No file reads in this session.")
return
clearing_events = [e for e in result.events if e["kind"] == "clearing"]
if not clearing_events:
print("Clearing never fired; all file reads remain in context.")
return
# Walk messages in order to reconstruct the sequence of tool_use blocks
# and which turn each one came from. The last `keep` of these survive
# the most recent clearing; earlier ones are cleared.
tool_use_seq: list[tuple[int, str, str]] = [] # (turn, name, path-if-read)
turn = 0
for msg in result.messages:
if msg["role"] == "assistant" and isinstance(msg["content"], list):
turn += 1
for block in msg["content"]:
if block.get("type") == "tool_use":
name = block.get("name", "?")
path = block.get("input", {}).get("path", "") if name == "read_file" else ""
tool_use_seq.append((turn, name, path))
last_clear_turn = clearing_events[-1]["turn"]
seq_before = [t for t in tool_use_seq if t[0] < last_clear_turn]
cleared_reads = [(t, p) for (t, n, p) in seq_before[:-keep] if n == "read_file"]
surviving_reads = [
(t, p) for tu in seq_before[-keep:] for (t, n, p) in [tu] if n == "read_file"
]
# Reads at or after the last clearing turn are untouched by it.
surviving_reads += [
(t, p) for (t, n, p) in tool_use_seq if t >= last_clear_turn and n == "read_file"
]
total = len(result.file_reads)
print(f"Total file reads across session: {total}")
print(f"Last clearing event fired at turn {last_clear_turn} (keep={keep})")
print(f"\nReads cleared from context: {len(cleared_reads)}")
for turn, path in cleared_reads[:12]:
print(f" ✗ turn {turn:2d}: {path}")
if len(cleared_reads) > 12:
print(f" ... and {len(cleared_reads) - 12} more")
print(
f"\nReads still in context (within the keep={keep} window or after "
f"the last clearing): {len(surviving_reads)}"
)
for turn, path in surviving_reads[:6]:
print(f" ✓ turn {turn:2d}: {path}")
if len(surviving_reads) > 6:
print(f" ... and {len(surviving_reads) - 6} more")
# ── Plot helpers ─────────────────────────────────────────────────────────
def plot_trajectories(
results: dict[str, SessionResult],
title: str = "Context size per turn",
triggers: dict[str, int] | None = None,
project_growth_for: str | None = None,
):
"""Line plot of context tokens per turn.
Vertical dashed lines mark turns where clearing fired; dash-dot lines
mark compaction. Horizontal dotted lines mark configured trigger
thresholds (pass triggers={"clearing": 20000, "compaction": 50000} etc).
project_growth_for: label of a run to extrapolate. Fits a line to the
last 5 points and draws a dotted segment forward ~8 turns to show where
unmanaged growth is heading.
"""
fig, ax = plt.subplots(figsize=(10, 4.5))
colors = plt.rcParams["axes.prop_cycle"].by_key()["color"]
for i, (label, res) in enumerate(results.items()):
turns = [t for t, _ in res.token_trajectory]
tokens = [tok for _, tok in res.token_trajectory]
color = colors[i % len(colors)]
ax.plot(turns, tokens, marker="o", label=label, markersize=4, color=color)
# Mark events on the x-axis
for ev in res.events:
style = "--" if ev["kind"] == "clearing" else "-."
ax.axvline(x=ev["turn"], color=color, linestyle=style, alpha=0.25, linewidth=1)
# Dotted growth projection for the named run, capped at 1M
if label == project_growth_for and len(turns) >= 3:
HARD_LIMIT = 1_000_000
fit_n = min(5, len(turns))
xs, ys = turns[-fit_n:], tokens[-fit_n:]
n = len(xs)
sx, sy = sum(xs), sum(ys)
slope = (n * sum(x * y for x, y in zip(xs, ys, strict=False)) - sx * sy) / (
n * sum(x * x for x in xs) - sx * sx
)
intercept = (sy - slope * sx) / n
proj_x, proj_y = [], []
for x in range(turns[-1], turns[-1] + 9):
y = slope * x + intercept
if y > HARD_LIMIT:
# Clip the last segment to the 1M ceiling and stop
if proj_y and slope > 0:
proj_x.append(proj_x[-1] + (HARD_LIMIT - proj_y[-1]) / slope)
proj_y.append(HARD_LIMIT)
break
proj_x.append(x)
proj_y.append(y)
if proj_y:
ax.plot(proj_x, proj_y, linestyle=":", color=color, alpha=0.6, linewidth=1.5)
# Horizontal reference lines for trigger thresholds
if triggers:
for name, value in triggers.items():
ax.axhline(y=value, color="gray", linestyle=":", alpha=0.6, linewidth=1)
ax.annotate(
f"{name} trigger: {value:,}",
xy=(ax.get_xlim()[1], value),
xytext=(-5, 3),
textcoords="offset points",
ha="right",
va="bottom",
fontsize=8,
color="gray",
)
# 200K reference: earlier models cap here and would hard-stop
if ax.get_ylim()[1] > 30_000:
ax.axhline(y=200_000, color="gray", linestyle="--", alpha=0.5, linewidth=1)
ax.annotate(
"200K: earlier models stop here",
xy=(ax.get_xlim()[0], 200_000),
xytext=(5, -12),
textcoords="offset points",
ha="left",
va="top",
fontsize=8,
color="gray",
alpha=0.8,
)
ax.set_xlabel("Turn")
ax.set_ylabel("Context tokens (incl. cached)")
ax.set_title(title)
ax.legend(loc="best")
ax.grid(alpha=0.3)
plt.tight_layout()
plt.show()
def plot_summary_bars(results: dict[str, SessionResult], title: str = "Session outcomes"):
"""Side-by-side bars: final context, file reads."""
labels = list(results.keys())
final_ctx = [r.token_trajectory[-1][1] for r in results.values()]
reads = [r.tool_counts.get("read_file", 0) for r in results.values()]
fig, axes = plt.subplots(1, 2, figsize=(9, 3.5))
fig.suptitle(title, y=1.02)
for ax, values, ylabel in zip(
axes,
[final_ctx, reads],
["Final context (tokens)", "File reads"],
strict=False,
):
bars = ax.bar(range(len(labels)), values, color=plt.cm.Set2(range(len(labels))))
ax.set_ylabel(ylabel)
ax.set_xticks(range(len(labels)))
ax.set_xticklabels(labels, rotation=30, ha="right", fontsize=8)
ax.grid(axis="y", alpha=0.3)
for bar, val in zip(bars, values, strict=False):
ax.text(
bar.get_x() + bar.get_width() / 2,
bar.get_height(),
f"{val:,}" if val > 1000 else str(val),
ha="center",
va="bottom",
fontsize=7,
)
plt.tight_layout()
plt.show()
```### 基线：无上下文管理
首先，我们以无上下文管理配置运行代理。由于语料库庞大（每份文档及其附录表格约为40K个token），上下文积累很快。我们将从两个角度来分析这次运行：在1M-token窗口上发生的情况，以及在200K窗口上可能发生的情况。
#### 第一部分：在 1M token 窗口上
Claude Sonnet 4.6 和 Claude Opus 4.6 都提供了 [1M-token上下文窗口](https://platform.claude.com/docs/en/build-with-claude/context-windows)。对于这个任务，基线的总输入保持在那个限制之下：代理读取整个语料库并综合生成，而不会遇到硬性限制。下方的轨迹显示了运行过程攀升至数十万个token，虚线表示以相同的速率持续增长。
```plaintext
Claude Sonnet 4.6 and Claude Opus 4.6 both provide a [1M-token context window](https://platform.claude.com/docs/en/build-with-claude/context-windows). For this task, the baseline's total input stays under that limit: the agent reads the full corpus and synthesizes without hitting a hard wall. The trajectory below shows the run climbing to hundreds of thousands of tokens, with the dotted line projecting continued growth at the same rate.
``````python
baseline = run_research_session(RESEARCH_TASK, label="baseline", max_turns=12)
plot_trajectories(
{"baseline": baseline},
"Baseline on a 1M window: context grows past 200K",
project_growth_for="baseline",
)
peak = max(t for _, t in baseline.token_trajectory)
n_turns = len(baseline.token_trajectory)
print(f"Peak context: {peak:,} tokens across {n_turns} turns")
print(f"File reads: {baseline.tool_counts.get('read_file', 0)}, Notes: {len(baseline.notes)}")
if baseline.hit_limit:
print("\n⚠ This run hit the 1M hard limit.")
else:
print(
"\nThe run stayed within the 1M window. The dotted line projects "
"where continued growth at the same rate would land, capped at 1M."
)┌─ [baseline]
│ turn 1 ctx= 1,058
│ read_file /research/celegans_review.md → ~27,465 tok
│ read_file /research/drosophila_review.md → ~27,435 tok
│ read_file /research/yeast_review.md → ~27,320 tok
│ read_file /research/killifish_review.md → ~27,349 tok
│ turn 2 ctx=166,043
│ record_finding "## BATCH 1 NOTES: High-Throughput Model Organisms ..."
│ turn 3 ctx=168,242
│ read_file /research/mouse_review.md → ~27,390 tok
│ read_file /research/zebrafish_review.md → ~27,363 tok
│ read_file /research/nmr_review.md → ~27,287 tok
│ read_file /research/rhesus_review.md → ~27,285 tok
│ turn 4 ctx=332,821
│ record_finding "## BATCH 2 NOTES: Low-Throughput Model Organisms —..."
│ turn 5 ctx=335,279 (final answer)
└─ completed: 5 turns, peak ctx 335,279, final ctx 335,279, 0 context event(s)<Figure size 1000x450 with 1 Axes>Peak context: 335,279 tokens across 5 turns
File reads: 8, Notes: 2
The run stayed within the 1M window. The dotted line projects where continued growth at the same rate would land, capped at 1M.# ── What's actually sitting in the context window at the end of the run ──
# Walk the message list and bucket everything by type. This is what the
# model has to attend to on every turn: no hard wall stopped it, but the
# pile keeps growing and every fact from early turns is buried deeper.
def _bucket_context(messages: list[dict]) -> dict[str, tuple[int, int]]:
"""Categorize message content into (count, approx_tokens) per bucket."""
buckets: dict[str, list[int]] = {
"File-read results": [],
"Other tool results": [],
"Agent reasoning text": [],
"Tool-call records": [],
"User/task prompts": [],
}
for msg in messages:
content = msg.get("content")
if isinstance(content, str):
buckets["User/task prompts"].append(len(content))
elif isinstance(content, list):
for block in content:
bt = block.get("type")
if bt == "tool_result":
body = block.get("content", "")
body = body if isinstance(body, str) else json.dumps(body)
# Heuristic: large tool results are file reads
key = "File-read results" if len(body) > 5000 else "Other tool results"
buckets[key].append(len(body))
elif bt == "text":
buckets["Agent reasoning text"].append(len(block.get("text", "")))
elif bt == "tool_use":
buckets["Tool-call records"].append(len(json.dumps(block.get("input", {}))))
# chars // 4 as a rough token estimate; the exact number isn't the point
return {k: (len(v), sum(v) // 4) for k, v in buckets.items() if v}
print("=" * 72)
print("WHAT THE MODEL IS ATTENDING TO at the end of the baseline run")
print("=" * 72)
breakdown = _bucket_context(baseline.messages)
total_est = sum(tok for _, tok in breakdown.values())
# Scale the char-based estimates so the total matches the API's reported
# context size. This keeps the proportions meaningful and the absolute
# numbers consistent with the usage figures shown elsewhere.
api_total = baseline.token_trajectory[-1][1]
scale = api_total / total_est if total_est else 1.0
print(f"\nContext breakdown ({api_total:,} tokens total per API usage):\n")
for label, (count, tokens) in sorted(breakdown.items(), key=lambda kv: -kv[1][1]):
scaled = int(tokens * scale)
pct = 100 * tokens / total_est if total_est else 0
print(f" {label:<26} ~{scaled:>8,} tokens ({count:>3} blocks, {pct:4.1f}%)")
if baseline.file_reads:
first_turn, first_path = baseline.file_reads[0]
doc_name = first_path.split("/")[-1]
print(f"\nFirst document read: {doc_name} at turn {first_turn}.")========================================================================
WHAT THE MODEL IS ATTENDING TO at the end of the baseline run
========================================================================
Context breakdown (335,279 tokens total per API usage):
File-read results ~ 322,946 tokens ( 8 blocks, 96.3%)
Tool-call records ~ 6,287 tokens ( 10 blocks, 1.9%)
Agent reasoning text ~ 5,660 tokens ( 5 blocks, 1.7%)
User/task prompts ~ 357 tokens ( 1 blocks, 0.1%)
Other tool results ~ 26 tokens ( 2 blocks, 0.0%)
First document read: celegans_review.md at turn 1.
```以上分解使规模具体化。在每一步中，模型都携带数十万个文件内容的token，其中大部分是代理已经处理并做了笔记的文档。第一份读取的文档仍然在窗口中，但到运行结束时，它后面跟着数十万个其他工具结果的token以及代理的所有推理和笔记。它没有被移除；它与其他所有内容争夺注意力。这就是上下文退化出现的地方：随着窗口的填充，从那个深度回忆细节的能力会下降，尽管内容在技术上仍然存在。并且预填充延迟与上下文长度成比例，所以每一步都值得处理整个堆栈。
#### 第二部分：在 200K token 窗口上
早期模型的上限是 200K tokens。在这些模型上，相同的基线运行会遇到一个硬墙：一旦上下文超过限制，API 就会拒绝下一个请求，任务就会在运行中途停止。
下表找到了基线首次超过 200K 的回合，并展示了从 200K 模型的角度来看运行看起来是什么样子：到那个点相同的轨迹，然后突然停止。
| 回合 | 模型视角 |
| --- | --- |
| ... | ... |
| 200,000 | 硬停 |```python
# No new agent run here. We reuse the Part-1 baseline's token_trajectory
# and clip it at the turn where context first exceeded 200K, to show what
# the same run would look like from a 200K model's perspective.
cross_turn = next((t for t, tok in baseline.token_trajectory if tok > 200_000), None)
if cross_turn is None:
print(
"Baseline never exceeded 200K tokens this run; a 200K model "
"would have seen the same result."
)
else:
# Plot the turns that fit under 200K, plus a dotted segment to the
# rejected turn and a red cutoff marker between them.
clipped_traj = [(t, tok) for t, tok in baseline.token_trajectory if t < cross_turn]
turns_c = [t for t, _ in clipped_traj]
toks_c = [tok for _, tok in clipped_traj]
rejected_ctx = dict(baseline.token_trajectory).get(cross_turn, 200_000)
fig, ax = plt.subplots(figsize=(10, 4.5))
ax.plot(turns_c, toks_c, marker="o", markersize=4, label="baseline")
# Dotted segment from last successful turn to the rejected turn
ax.plot(
[turns_c[-1], cross_turn],
[toks_c[-1], min(rejected_ctx, 200_000)],
linestyle=":",
color="#1f77b4",
alpha=0.6,
linewidth=1.5,
)
# Hard red vertical line marking the cutoff between the last
# successful turn and the rejected one
cutoff_x = (turns_c[-1] + cross_turn) / 2
ax.axvline(x=cutoff_x, color="#d62728", linewidth=2)
ax.set_xlim(0.5, cross_turn + 0.5)
ax.set_ylim(0, 205_000)
ax.text(
0.02,
0.96,
"context window: 200,000",
transform=ax.transAxes,
ha="left",
va="top",
fontsize=9,
color="#d62728",
)
ax.set_xlabel("Turn")
ax.set_ylabel("Context tokens (incl. cached)")
ax.set_title("Baseline on a 200K window: agent stopped mid-task")
ax.legend(loc="lower right")
ax.grid(alpha=0.3)
plt.tight_layout()
plt.show()
last_turn = clipped_traj[-1][0]
last_ctx = clipped_traj[-1][1]
reads_done = sum(1 for t, _ in baseline.file_reads if t < cross_turn)
notes_at_stop = sum(
1
for m in baseline.messages[: 2 * cross_turn]
if isinstance(m.get("content"), list)
for b in m["content"]
if b.get("type") == "tool_use" and b.get("name") == "record_finding"
)
print("⚠ Baseline HIT THE CONTEXT WINDOW LIMIT.")
print(f" Completed {last_turn} turns before the API rejected the next request.")
print(f" Last successful context: {last_ctx:,} tokens")
print(f" File reads attempted: {reads_done}")
print(f" Notes taken: {notes_at_stop}")
print()
print(
" The agent stopped mid-task. Without context management, it "
"cannot continue past this point."
)<Figure size 1000x450 with 1 Axes>⚠ Baseline HIT THE CONTEXT WINDOW LIMIT.
Completed 3 turns before the API rejected the next request.
Last successful context: 168,242 tokens
File reads attempted: 8
Notes taken: 2
The agent stopped mid-task. Without context management, it cannot continue past this point.
```两种故障模式都源于同一个根本问题：上下文窗口充满了数十万个文件内容的标记，其中大部分已经被处理并记录。不同的是故障的表现形式。在20万窗口中，这是一个硬停止：API拒绝下一个请求，任务在中间阶段结束。在100万窗口中，代理继续运行，但随着窗口的填充，上下文退化开始出现：早期文档的读取仍然技术上存在，但到运行结束时，它被之前读取的所有内容所淹没，模型回忆其细节的能力下降。代理完成了，但合成的质量取决于与那堆内容作斗争的回忆。预填充延迟也随着它而变化：每一轮都需要处理完整上下文，无论其中有多少仍然有用。
下面的原语通过保持工作集足够小来解决这个问题，这样两种故障模式都不会发生：窗口不会填满，因此较小的模型不会停止，较大的模型也不会退化。图表包括虚线200K参考线，以便您可以看到早期模型会被切断的位置。
***
## 压缩（Compaction）
[压缩](https://platform.claude.com/docs/en/build-with-claude/compaction) 是管理长时间对话上下文的有用策略：它将接近上下文窗口限制的对话内容进行总结，并使用该总结重新启动。这解决了代理自己的推理文本、用户之间的来回对话以及会话过程中做出的决策。具体操作序列和早期轮次的准确措辞将不会被保留，但代理做出的目标、决策和重大发现将被总结——总结保留的内容取决于您的压缩提示，我们将在下面进行介绍。
在其核心，压缩以高保真度提炼上下文窗口的内容，使代理能够以最小的性能下降继续进行。权衡在于选择摘要必须保留的内容与可以安全丢弃的内容：过于激进的压缩可能会丢失微妙但关键的上下文，其重要性只有在之后才会变得明显。摘要保留了关键决策和事实，但可能会丢弃特定的数字或确切的措辞。这会消耗推理（摘要模型运行），但处理所有上下文增长，而不仅仅是工具结果。
### 底层工作原理
以下是一个压缩的最小示例实现。我们的第一方API提供了一个健壮、经过测试的版本（在标记阈值自动触发、一个可以本地嵌入对话的文本内容块、正确的工具使用配对），但下面这个 ~25 行版本使机制具体化：将对话渲染为文本，要求模型对其进行摘要，用那个摘要替换旧消息。
```python
def compact_context(context):
# 代码实现
pass请注意，这里的代码仅为示例，实际应用中可能需要更复杂的逻辑。 :::```python def demo_compact(client, messages: list[dict], model: str) -> list[dict]: “““Summarize the conversation and return a fresh message list containing only the summary.”“” # Render to a plain-text transcript lines: list[str] = [] for msg in messages: role = msg[“role”] content = msg.get(“content”, ““) if isinstance(content, str): lines.append(f”[{role}] {content}“) elif isinstance(content, list): for block in content: bt = block.get(”type”, “?”) if bt == “text”: lines.append(f”[{role}] {block[‘text’]}“) elif bt ==”tool_use”: lines.append(f”[{role}→tool] {block[‘name’]}({block.get(‘input’, {})})“) elif bt ==”tool_result”: body = block.get(“content”, ““) body = body if isinstance(body, str) else json.dumps(body) lines.append(f”[tool_result] {body[:200]}{‘…’ if len(body) > 200 else ’’}“) transcript =”“.join(lines)
resp = client.messages.create(
model=model,
max_tokens=1024,
messages=[
{
"role": "user",
"content": (
"Summarize the following agent transcript, preserving key "
"decisions, facts learned, and open threads. Be concise.\n\n"
f"<transcript>\n{transcript}\n</transcript>"
),
}
],
)
summary = resp.content[0].text
return [{"role": "user", "content": f"[conversation summary]\n{summary}"}]
```python
# Demonstrate on a longer synthetic conversation so the summary is meaningfully shorter
demo_convo = []
organisms = [
(
"C. elegans",
"18-day median lifespan at 20°C. Survival curve takes 4-6 weeks. Automated platforms handle thousands of animals. Genome-wide RNAi by feeding, fast CRISPR knock-ins. ~40% human orthology, no adaptive immunity, post-mitotic soma.",
),
(
"Drosophila",
"60-80 day median lifespan at 25°C. Survival curves take 3-4 months. GAL4/UAS tissue-specific expression, FLP/FRT mosaic analysis, mature CRISPR. ~60% human disease gene orthology. Good for cardiac aging and neurodegeneration models.",
),
(
"Mouse",
"~24-30 month median lifespan. Full study is 3-year commitment, $1-2/day/animal, $100K+ per 100-mouse cohort. Cre-lox conditionals, inbred strain panels, humanized models. ~85% orthology, best translational predictor.",
),
(
"Killifish",
"4-6 month median lifespan (GRZ strain), shortest-lived vertebrate in the lab. Survival curves under a year. CRISPR works, toolkit still thin. ~70% orthology. Rapid vertebrate aging phenotypes on a compressed timeline.",
),
(
"Yeast",
"Replicative lifespan ~25 divisions, ~1-2 weeks to result. Genome-wide deletion collection, trivial CRISPR, pooled competitive assays. ~30% orthology. Great for mechanism, not for translation.",
),
(
"Naked mole-rat",
"Max lifespan >37 years, negligible senescence. No controlled genetics (eusocial breeding). Cell-line studies only. Near-zero cancer incidence. Comparative biology resource, not a screening organism.",
),
]
for name, facts in organisms:
demo_convo.append({"role": "user", "content": f"Tell me about {name} for aging research."})
demo_convo.append({"role": "assistant", "content": f"{name}: {facts}"})
demo_convo.append({"role": "user", "content": "What are the key tradeoffs across all of these?"})
demo_convo.append(
{
"role": "assistant",
"content": "The core tradeoff is throughput vs. translational relevance. Short-lived invertebrates (worm, fly, yeast) let you screen fast but translate weakly. Mouse is the translational gatekeeper but slow and expensive. Killifish tries to split the difference. The mole-rat is a comparative outlier you study to understand exceptional longevity, not to screen interventions.",
}
)
before_tokens = count_tokens(json.dumps(demo_convo))
compacted = demo_compact(client, demo_convo, MODEL)
after_tokens = count_tokens(json.dumps(compacted))
print(
f"Conversation size: ~{before_tokens:,} → ~{after_tokens:,} tokens "
f"({(before_tokens - after_tokens) / before_tokens:.0%} reduction)"
if after_tokens < before_tokens
else f"Conversation size: ~{before_tokens:,} → ~{after_tokens:,} tokens"
)
print(f"\nSummary produced:\n{compacted[0]['content'][:800]}")
if len(compacted[0]["content"]) > 800:
print("...")Conversation size: ~741 → ~449 tokens (39% reduction)
Summary produced:
[conversation summary]
## Summary: Model Organisms for Aging Research
### Key Facts Learned
| Organism | Lifespan | Study Duration | Human Orthology | Standout Feature |
|---|---|---|---|---|
| *C. elegans* | 18-day median | 4–6 weeks | ~40% | Genome-wide RNAi; high throughput |
| *Drosophila* | 60–80 days | 3–4 months | ~60% | Tissue-specific genetics; cardiac/neuro models |
| Mouse | 24–30 months | ~3 years | ~85% | Best translational predictor; expensive ($100K+/cohort) |
| Killifish | 4–6 months | <1 year | ~70% | Shortest-lived vertebrate; thin toolkit |
| Yeast | ~25 divisions | 1–2 weeks | ~30% | Mechanistic screening; poor translation |
| Naked mole-rat | >37 years | N/A | — | Negligible senescence; no genetic control |
### Core Tradeoff
**Throughput ↔ Translational relevance** —
...
```以下示例演示了机制：模型生成一个简化的对话版本，代理可以从该版本继续。
### 使用 API
我们的API以`compact_20260112`上下文编辑的形式原生提供此功能。它在达到标记阈值（最小50K）时自动触发，返回一个类型为`compaction`的内容块，该块可以原生地嵌入到对话中，并处理跨摘要边界的工具使用配对。当压缩触发时，您将压缩块序列化回(`{"type": "compaction", "content": block.content}`)并在下一个请求中丢弃它之前的一切。
**API 文档：** [压缩 — platform.claude.com](https://platform.claude.com/docs/en/build-with-claude/compaction)
以下是配置了压缩的运行研究代理。我们将触发器设置为180K，以便第一批读取（约165K）保持在它之下：压缩轨迹跟踪基线通过该批次，然后在第二个批次将上下文推过触发器时发生分歧。注意输出中的`⊟ COMPACTION`行以及图表上摘要替换早期对话时的下降。
::: callout-注意
请注意，压缩功能在达到一定标记阈值时自动触发，并且压缩块在API请求中会被序列化并丢弃。
:::```python
COMPACTION_TRIGGER = 180_000 # above batch-1's ~165K so the rise is visible before compaction fires
compaction_run = run_research_session(
RESEARCH_TASK,
betas=["compact-2026-01-12"],
context_management={
"edits": [
{
"type": "compact_20260112",
"trigger": {"type": "input_tokens", "value": COMPACTION_TRIGGER},
}
]
},
max_turns=12,
label="compaction",
)
plot_trajectories(
{"baseline": baseline, "compaction": compaction_run},
"Compaction: tracks baseline through batch 1, then summarizes and continues",
triggers={"compaction": COMPACTION_TRIGGER},
)
compaction_events = [e for e in compaction_run.events if e["kind"] == "compaction"]
# Outcome comparison
print(
f"Baseline: {'HIT LIMIT at' if baseline.hit_limit else 'completed'} "
f"turn {len(baseline.token_trajectory)}, "
f"peak {max(t for _, t in baseline.token_trajectory):,} tokens"
)
print(
f"Compaction: {'HIT LIMIT at' if compaction_run.hit_limit else 'completed'} "
f"turn {len(compaction_run.token_trajectory)}, "
f"peak {max(t for _, t in compaction_run.token_trajectory):,} tokens, "
f"{len(compaction_events)} compaction event(s)"
)
if compaction_events:
for i, ev in enumerate(compaction_events, 1):
print(
f" Compaction #{i} at turn {ev['turn']}: ~{count_tokens(ev['summary']):,}-token summary"
)
baseline_peak = max(t for _, t in baseline.token_trajectory)
compaction_peak = max(t for _, t in compaction_run.token_trajectory)
print(
f"\n→ Baseline climbed to {baseline_peak:,} tokens; compaction kept "
f"the peak at {compaction_peak:,}. On a smaller context window the "
f"baseline would have been cut off mid-task; here it completed, but "
f"every turn paid to process the full pile."
)┌─ [compaction]
│ turn 1 ctx= 1,129
│ read_file /research/celegans_review.md → ~27,465 tok
│ read_file /research/drosophila_review.md → ~27,435 tok
│ read_file /research/yeast_review.md → ~27,320 tok
│ read_file /research/killifish_review.md → ~27,349 tok
│ turn 2 ctx=166,103
│ record_finding "## BATCH 1 DETAILED NOTE: High-Throughput Model Or..."
│ turn 3 ctx=169,164
│ read_file /research/mouse_review.md → ~27,390 tok
│ read_file /research/zebrafish_review.md → ~27,363 tok
│ read_file /research/nmr_review.md → ~27,287 tok
│ read_file /research/rhesus_review.md → ~27,285 tok
│ ⊟ COMPACTION (turn 4): ~2,783-token summary replaces prior turns
│ turn 5 ctx= 5,635
│ search_files 'comparative aging model organisms synthesis pipeline translational'
│ turn 6 ctx= 5,751
│ search_files 'aging model organism review'
│ turn 7 ctx= 5,829 (final answer)
└─ completed: 7 turns, peak ctx 169,164, final ctx 5,829, 1 context event(s)<Figure size 1000x450 with 1 Axes>Baseline: completed turn 5, peak 335,279 tokens
Compaction: completed turn 7, peak 169,164 tokens, 1 compaction event(s)
Compaction #1 at turn 4: ~2,783-token summary
→ Baseline climbed to 335,279 tokens; compaction kept the peak at 169,164. On a smaller context window the baseline would have been cut off mid-task; here it completed, but every turn paid to process the full pile.# What compaction costs: check the summary text directly for a MIX of
# details. We inspect what the summary PRESERVED, not what the agent
# happens to have in context at end-of-run (which could include reads
# that happened after compaction fired).
compaction_events = [e for e in compaction_run.events if e["kind"] == "compaction"]
print("=" * 70)
print("WHAT COMPACTION COSTS: what the summary preserved vs. dropped")
print("=" * 70)
if not compaction_events:
print("(Compaction did not fire this run; probe skipped)")
else:
# Use the last compaction summary — it covers the most history.
last_ev = compaction_events[-1]
summary_text = last_ev["summary"]
n_tok = count_tokens(summary_text)
print(
f"Last compaction at turn {last_ev['turn']} produced a "
f"~{n_tok:,}-token summary. The file content read before "
f"that point was distilled into this.\n"
)
print(
"Checking the summary text directly for six details: three "
"HIGH-LEVEL facts (central to the task, likely summarized) and "
"three OBSCURE specifics (appendix table cells, unlikely to make "
"the cut).\n"
)
results = []
for question, answer_key, expected_survive in COMPACTION_PROBES:
survived = answer_key.lower() in summary_text.lower()
results.append((question, answer_key, expected_survive, survived))
for label, expect in [
("HIGH-LEVEL FACTS (expected to survive)", True),
("OBSCURE SPECIFICS (expected to be lost)", False),
]:
print(f"── {label} ──")
for q, key, exp, surv in results:
if exp != expect:
continue
status = "✓ in summary" if surv else "✗ not in summary"
print(f" {status:<18} looking for '{key}' — {q}")
print()
hl_survived = sum(1 for _, _, e, s in results if e and s)
hl_total = sum(1 for _, _, e, _ in results if e)
ob_survived = sum(1 for _, _, e, s in results if not e and s)
ob_total = sum(1 for _, _, e, _ in results if not e)
print("=" * 70)
print(
f"RESULT: high-level {hl_survived}/{hl_total} preserved, "
f"obscure {ob_survived}/{ob_total} preserved"
)
print("=" * 70)
# Show a representative excerpt of the summary so the reader can see
# the compression directly.
excerpt = summary_text.strip()
if len(excerpt) > 600:
excerpt = excerpt[:600] + "…"
print("\nExcerpt from the summary (first ~600 chars):\n")
for line in excerpt.splitlines():
print(f" {line}")======================================================================
WHAT COMPACTION COSTS: what the summary preserved vs. dropped
======================================================================
Last compaction at turn 4 produced a ~2,783-token summary. The file content read before that point was distilled into this.
Checking the summary text directly for six details: three HIGH-LEVEL facts (central to the task, likely summarized) and three OBSCURE specifics (appendix table cells, unlikely to make the cut).
── HIGH-LEVEL FACTS (expected to survive) ──
✓ in summary looking for '18' — What is the approximate median lifespan of C. elegans at 20°C?
✓ in summary looking for 'killifish' — Which model organism is the shortest-lived vertebrate used in aging research?
✓ in summary looking for '60' — Roughly what percentage of human disease genes have Drosophila orthologs?
── OBSCURE SPECIFICS (expected to be lost) ──
✗ not in summary looking for '61' — In the appendix Table A5, what is the I-squared heterogeneity value for the NAD+ precursor intervention?
✗ not in summary looking for '55' — In appendix Table A2, what was the effect magnitude for the IIS reduction intervention in cohort 2?
✗ not in summary looking for '0.72' — In appendix Table A7, what is the PhenoAge-like epigenetic clock acceleration ratio under DR for liver tissue?
======================================================================
RESULT: high-level 3/3 preserved, obscure 0/3 preserved
======================================================================
Excerpt from the summary (first ~600 chars):
## Task Overview
I am a biology research analyst writing a comparative review of model organisms for aging research. The task has three phases:
1. **BATCH 1 (DONE):** Read and record a note on four high-throughput models (C. elegans, Drosophila, yeast, killifish)
2. **BATCH 2 (IN PROGRESS):** Read four low-throughput models (mouse, zebrafish, NMR, rhesus) and record a single note — **FILES HAVE BEEN READ, NOTE NOT YET RECORDED**
3. **FINAL:** Write a comprehensive comparative synthesis contrasting both batches
---
## Current State
### Batch 1 — COMPLETE
- All four files read: `celegans_revi…
```### 分析
基线持续上升，直到它触及上下文窗口限制（在小窗口上是一个硬停止）或积累足够的标记，以至于上下文旋转导致召回率明显下降。压缩解决了这两个问题：当上下文越过触发点时，较老的对话被模型生成的摘要所取代，上下文急剧下降。代理继续在一个精简的窗口中进行，而不是一个不断增长的窗口。
上面的探针直接检查摘要文本中的*混合*细节。通常出现的模式是：与任务相关的关键事实（代理注意到的寿命数字、生物体身份、主要比较）通常在摘要中幸存。晦涩的细节（附录表中的一个单独的细胞、异质性统计数据）通常不会。这与工具结果清理有明显的区别：清理会大量删除工具结果，因此内容直到重新检索才会消失，而压缩则保持压缩形式的内容，但失去了逐字逐句的细节。
压缩为你提供的是一种通用方法来保持窗口精简：它同时处理对话和工具结果，重要内容以摘要形式幸存，代理在它本应被切断或淹没的条件下继续工作。它不能为你提供具体细节的逐字逐句的准确性，或跨会话的持久性。如果你的上下文膨胀主要是可重新检索的工具输出，清理更便宜且无损失（代理可以再次调用工具）。如果是无法重新检索的对话和推理，压缩是合适的解决方案。
### 有效地实现压缩
`instructions`参数允许你完全替换默认的摘要提示。[压缩文档](https://platform.claude.com/docs/en/build-with-claude/compaction#custom-summarization-instructions)给出了默认提示的逐字逐句内容：
> 您已经为上述初始任务编写了部分转录文本。请编写该转录文本的摘要。此摘要的目的是提供连续性，以便您可以在未来的情境中继续解决问题，届时上述原始历史可能无法访问，并将被此摘要所取代。写下任何可能有助于您的内容，包括状态、下一步行动、学习经验等。您必须将您的摘要包裹在 `<summary></summary>` 块中。
这有助于您开始。然而，自定义 `instructions` 并不能补充此提示——它们完全取代了它。所以如果您提供了自己的，您将负责完整的框架。文档中关于编码情境的示例是：“关注保留代码片段、变量名和技术决策。”
对于这本食谱的研究代理，您可能需要编写一些内容，以命名上述探测显示可能丢失的具体细节：
::: callout-attention
请注意，以下内容是翻译结果，保留了原文的Markdown格式、技术术语、代码引用、变量名、函数名、HTML标签和callout语法，以及段落结构和换行。
:::```python
context_management={
"edits": [{
"type": "compact_20260112",
"trigger": {"type": "input_tokens", "value": 150_000},
"instructions": (
"Summarize this research agent transcript. Preserve every "
"quantitative lifespan figure and effect size with its "
"source organism, and note which documents have been read "
"and which remain. Wrap the summary in <summary></summary>."
),
}]
}
```***
## 工具结果清除（Tool-Result Clearing）
当代理调用工具时，每个结果都会以 `tool_result` 块的形式附加到对话中（[上下文编辑文档](https://platform.claude.com/docs/en/build-with-claude/context-editing)）。这些块在后续的每个回合中都会计入输入令牌预算，即使代理已经处理了内容并继续进行也是如此。对于可重新调用的工具（文件读取、API查询、搜索），携带原始结果通常是不必要的；如果需要，代理可以再次调用该工具。
清除操作用简短的占位符字符串替换旧的 `tool_result` 块。先前的 `tool_use` 块仍然保留，因此模型保留了一个记录，表明它已经进行了调用（以及输入了什么），但庞大的响应体已经消失。这是三种原语中最便宜的一种：没有推理成本，只需对消息列表进行机械编辑。
### 底层工作原理
为了使机制具体化，这里是一个工具结果清除的最小示例实现。我们的第一方API提供了一个健壮、经过测试的版本（自动触发、正确的块配对不变性、工具排除等），但看到 ~15 行版本可以使核心操作具体化：遍历消息列表，找到 `tool_result` 块，将除最近几个之外的所有内容替换为占位符。
***
```python
def clear_tool_results(messages):
# ... 实现细节 ...请注意，此代码示例仅为说明目的，并非完整实现。 :::``python def demo_clear_tool_results(messages: list[dict], keep: int = 3) -> int: """Replace the content of all but the lastkeep` tool_result blocks. Returns the number of blocks cleared.”“” positions: list[tuple[int, int]] = [] for mi, msg in enumerate(messages): if msg.get(“role”) != “user” or not isinstance(msg.get(“content”), list): continue for bi, block in enumerate(msg[“content”]): if isinstance(block, dict) and block.get(“type”) == “tool_result”: positions.append((mi, bi))
to_clear = positions[:-keep] if keep > 0 else positions
for mi, bi in to_clear:
messages[mi]["content"][bi]["content"] = "[cleared to save context]"
return len(to_clear)
```python
# Demonstrate on a small message list using the actual research corpus
sample_paths = list(CORPUS.keys())[:3] # first three docs
demo = [{"role": "user", "content": "Read three files and summarize."}]
for i, path in enumerate(sample_paths):
demo.append(
{
"role": "assistant",
"content": [
{"type": "tool_use", "id": f"t{i}", "name": "read_file", "input": {"path": path}},
],
}
)
demo.append(
{
"role": "user",
"content": [
{"type": "tool_result", "tool_use_id": f"t{i}", "content": CORPUS[path]},
],
}
)
before_tokens = count_tokens(json.dumps(demo))
cleared = demo_clear_tool_results(demo, keep=1)
after_tokens = count_tokens(json.dumps(demo))
print(f"Cleared {cleared} of 3 tool results (keep=1 leaves the most recent)")
print(
f"Message-list size: ~{before_tokens:,} → ~{after_tokens:,} tokens "
f"({(before_tokens - after_tokens) / before_tokens:.0%} reduction)"
)
print()
print("What each tool_result looks like now:")
for _i, msg in enumerate(demo):
if not isinstance(msg.get("content"), list):
continue
for block in msg["content"]:
if block.get("type") == "tool_use":
print(f"\n [tool_use] read_file({block['input']['path']})")
elif block.get("type") == "tool_result":
body = block["content"]
if body == "[cleared to save context]":
print(f" [tool_result] {body!r} (← cleared)")
else:
preview = body[:80].replace("\n", " ")
print(f" [tool_result] {preview!r}... (~{count_tokens(body):,} tokens retained)")Cleared 2 of 3 tool results (keep=1 leaves the most recent)
Message-list size: ~128,740 → ~43,060 tokens (67% reduction)
What each tool_result looks like now:
[tool_use] read_file(/research/celegans_review.md)
[tool_result] '[cleared to save context]' (← cleared)
[tool_use] read_file(/research/drosophila_review.md)
[tool_result] '[cleared to save context]' (← cleared)
[tool_use] read_file(/research/mouse_review.md)
[tool_result] '# Model Organism Review: Mus musculus in Aging Research The laboratory mouse is'... (~41,155 tokens retained)
```以上示例展示了机制。其中缺失的部分包括：token计数和自动触发、正确的`tool_use`/`tool_result`配对不变性、工具特定的排除以及模型端对清除发生的意识。
### 使用 API
我们的API以`clear_tool_uses_20250919`上下文编辑的形式原生提供此功能。它处理token计数和触发服务器端，保留块配对，并允许您免除特定工具的清除（当内存工具也处于活动状态时很有用，我们稍后会看到）。清除触发时，响应包括`context_management.applied_edits`，其中包含已清除的工具使用数量和释放的token数量。
> 此外，还有`clear_thinking_20251015`用于扩展思考块。相同的配置形状，不同的`type`。如果您同时使用两者，它必须是`edits`数组中的第一个条目。
**API 文档：** [上下文编辑 — platform.claude.com](https://platform.claude.com/docs/en/build-with-claude/context-editing)
以下是启用清除功能运行的研究代理。基线上下文随着每个文件的读取而上升；清除通过在上下文超过触发器时丢弃旧的工具结果来保持此运行的范围。注意输出中的`✂ 清除`行；图上的虚线垂直线标记了每次触发。
::: callout-注意
请注意，清除操作可能会影响模型的性能和输出质量。
:::```python
CLEARING_KEEP = 4 # How many most-recent tool results survive each clearing
clearing_run = run_research_session(
RESEARCH_TASK,
betas=["context-management-2025-06-27"],
context_management={
"edits": [
{
"type": "clear_tool_uses_20250919",
"trigger": {"type": "input_tokens", "value": 30_000},
"keep": {"type": "tool_uses", "value": CLEARING_KEEP},
"clear_at_least": {"type": "input_tokens", "value": 10_000},
}
]
},
label="clearing",
max_turns=12,
)
plot_trajectories(
{"baseline": baseline, "clearing": clearing_run},
"Clearing: stays bounded while the baseline climbs",
triggers={"clearing": 30_000},
)
# Show outcome comparison
baseline_turns = len(baseline.token_trajectory)
clearing_turns = len(clearing_run.token_trajectory)
print(
f"Baseline: {'HIT LIMIT at' if baseline.hit_limit else 'completed'} "
f"turn {baseline_turns}, peak {max(t for _, t in baseline.token_trajectory):,} tokens"
)
print(
f"Clearing: {'HIT LIMIT at' if clearing_run.hit_limit else 'completed'} "
f"turn {clearing_turns}, peak {max(t for _, t in clearing_run.token_trajectory):,} tokens"
)
baseline_peak = max(t for _, t in baseline.token_trajectory)
clearing_peak = max(t for _, t in clearing_run.token_trajectory)
print(
f"\n→ Baseline climbed to {baseline_peak:,} tokens; clearing kept "
f"the peak at {clearing_peak:,}. On a smaller context window the "
f"baseline would have been cut off; here both completed, but the "
f"baseline was attending to far more on every turn."
)
print(f"\nClearing events: {len([e for e in clearing_run.events if e['kind'] == 'clearing'])}")
print(
f"File reads: baseline={baseline.tool_counts.get('read_file', 0)}, "
f"clearing={clearing_run.tool_counts.get('read_file', 0)}"
)┌─ [clearing]
│ turn 1 ctx= 1,058
│ read_file /research/celegans_review.md → ~27,465 tok
│ read_file /research/drosophila_review.md → ~27,435 tok
│ read_file /research/yeast_review.md → ~27,320 tok
│ read_file /research/killifish_review.md → ~27,349 tok
│ turn 2 ctx=166,043
│ record_finding "## BATCH 1 NOTES: High-Throughput Model Organisms ..."
│ turn 3 ctx=169,211
│ read_file /research/mouse_review.md → ~27,390 tok
│ read_file /research/zebrafish_review.md → ~27,363 tok
│ read_file /research/nmr_review.md → ~27,287 tok
│ read_file /research/rhesus_review.md → ~27,285 tok
│ ✂ CLEARING (turn 4): 4 tool results cleared, ~163,817 tokens freed
│ turn 4 ctx=169,174
│ record_finding ""
│ ✂ CLEARING (turn 5): 4 tool results cleared, ~163,811 tokens freed
│ turn 5 ctx=169,255
│ record_finding ""
│ ✂ CLEARING (turn 6): 4 tool results cleared, ~163,807 tokens freed
│ turn 6 ctx=169,313
│ record_finding "## BATCH 2 NOTES: Low-Throughput Model Organisms (..."
│ ✂ CLEARING (turn 7): 4 tool results cleared, ~162,930 tokens freed
│ turn 7 ctx=173,137 (final answer)
└─ completed: 7 turns, peak ctx 173,137, final ctx 173,137, 4 context event(s)<Figure size 1000x450 with 1 Axes>Baseline: completed turn 5, peak 335,279 tokens
Clearing: completed turn 7, peak 173,137 tokens
→ Baseline climbed to 335,279 tokens; clearing kept the peak at 173,137. On a smaller context window the baseline would have been cut off; here both completed, but the baseline was attending to far more on every turn.
Clearing events: 4
File reads: baseline=8, clearing=8# What's lost: which file reads are no longer in the clearing run's context?
print("=" * 60)
print("WHAT CLEARING COSTS: reads no longer in context")
print("=" * 60)
show_cleared_reads(clearing_run, keep=CLEARING_KEEP)
print(
"\nThe cleared reads above are gone from the conversation. If the "
"agent needs that content again, it must call read_file again; the "
"information is re-fetchable, but the original read is no longer "
"visible in context."
)============================================================
WHAT CLEARING COSTS: reads no longer in context
============================================================
Total file reads across session: 8
Last clearing event fired at turn 7 (keep=4)
Reads cleared from context: 7
✗ turn 1: /research/celegans_review.md
✗ turn 1: /research/drosophila_review.md
✗ turn 1: /research/yeast_review.md
✗ turn 1: /research/killifish_review.md
✗ turn 3: /research/mouse_review.md
✗ turn 3: /research/zebrafish_review.md
✗ turn 3: /research/nmr_review.md
Reads still in context (within the keep=4 window or after the last clearing): 1
✓ turn 3: /research/rhesus_review.md
The cleared reads above are gone from the conversation. If the agent needs that content again, it must call read_file again; the information is re-fetchable, but the original read is no longer visible in context.
```### 分析
基线持续上升；清除运行保持有界。一旦上下文超过触发点（此处为30K）并且记录中的`keep`工具使用次数超过，清除服务器端：比最近`keep`工具结果更早的工具结果被替换为占位符，上下文降回。图上的虚线标记了每次触发。这个有界窗口意味着运行不会触及较小模型上的硬限制，并且不会积累到上下文退化导致召回率下降的范围。
上方的第二个单元格显示了这会带来什么成本。除了最近几个文件外，所有读取的文件都从上下文中消失。当代理达到综合阶段时，它有两个选择。它可以从自己的笔记以及任何最近读取并幸存下来的内容开始工作：如果笔记详尽，这很好；如果笔记稀疏，综合过程会遗漏代理看到但没有记录的细节。或者它可以再次调用`read_file`来重新获取清除的内容：清除运行可能显示比基线更多的文件读取次数，因为有些读取在代理完成之前就已经被清除。第二条路径的成本取决于你的工具：重新读取本地文件几乎免费，但重新调用速率限制或慢速API则不是。调整`keep`和`trigger`会改变代理在这两条路径之间的着陆点。
清除为你提供的是一个无推理成本的有限窗口，避免了硬限制截止和伴随大量累积上下文而来的召回率退化。但它不能帮助你处理不是工具结果的内容（代理自己的推理、用户消息）或任何会话间的持久性。
### 有效地实现清除
与压缩和内存不同，清除没有调整的提示，所有旋钮都是数字的（`trigger`、`keep`、`clear_at_least`）或基于列表的（`exclude_tools`）。一个需要理解的交易：清除使缓存的提示前缀无效。为了解决这个问题，清除足够的标记，使缓存失效变得值得；`clear_at_least`参数确保每次清除至少清除一定数量的标记。每次清除触发时，您将承担缓存写入成本，但后续请求可以重用新缓存的提示前缀。
`trigger`和`keep`的正确值取决于您的代理如何使用工具结果：它们有多大，代理多久会重新访问它们，重新获取是否便宜。上面的清除运行使用了trigger=30K和keep=4；稍后使用的所有三个运行具有更高的trigger和keep=6，以便清除和压缩分担工作。针对您自己的代理的工作负载测试几个配置：每个响应中的`context_management.applied_edits`字段显示了清除了多少工具使用和标记，这使得每个配置的效果可以直接观察到。
***
## 内存工具
[内存工具](https://platform.claude.com/docs/en/agents-and-tools/tool-use/memory-tool)使Claude能够通过内存文件目录在对话中存储和检索信息。Claude可以创建、读取、更新和删除在会话之间持久存在的文件，从而使其能够在不将所有内容都保留在上下文窗口的情况下随着时间的推移构建知识。
这是即时上下文检索的关键原语：而不是一开始就加载所有相关信息，代理将他们学到的存储在内存中，并在需要时拉回。这使活动上下文专注于当前相关的信息，这对于一次性加载所有内容会压倒窗口的长运行工作流程至关重要。清除和压缩都作用于当前上下文；当新会话开始且窗口为空时，它们都不起作用。内存解决了这个问题。
***
## Memory 工具
The [memory tool](https://platform.claude.com/docs/en/agents-and-tools/tool-use/memory-tool) enables Claude to store and retrieve information across conversations through a memory file directory. Claude can create, read, update, and delete files that persist between sessions, allowing it to build knowledge over time without keeping everything in the context window.
This is the key primitive for just-in-time context retrieval: rather than loading all relevant information upfront, agents store what they learn in memory and pull it back on demand. This keeps the active context focused on what's currently relevant, which is critical for long-running workflows where loading everything at once would overwhelm the window. Clearing and compaction both operate on the current context; neither helps when a new session starts and the window is empty. Memory solves that problem.
内存工具在客户端运行：Claude通过工具调用执行内存操作，而您的应用程序在本地执行这些操作。这使您能够完全控制数据存储的位置和方式。API提供了工具协议，并自动注入一个系统提示以建立内存检查行为；您实现存储后端。
### 内部工作原理
以下是一个最小化示例实现：在会话后写入并在下一次会话前读取的键值存储。我们的第一方API提供了稳健版本（模型决定其推理过程中保存什么和何时保存，完整的文件操作，自动注入的协议提示），但这个约10行的版本使核心模式具体化。
```html
::: callout-notice
请注意，以下代码示例仅供参考，实际应用中可能需要根据具体情况进行调整。
:::
```python
# 假设的内存操作函数
def memory_operation(key, value):
# 在这里执行内存操作
pass
# 假设的存储后端实现
class StorageBackend:
def __init__(self):
# 初始化存储后端
pass
def save(self, key, value):
# 保存数据到后端
pass
def load(self, key):
# 从后端加载数据
pass{
"key": "value"
}CREATE TABLE memory (
key VARCHAR(255),
value TEXT
);
``````python
class SimpleMemoryStore:
def __init__(self):
self._store: dict[str, str] = {}
def save(self, key: str, value: str) -> None:
self._store[key] = value
def load(self, key: str) -> str | None:
return self._store.get(key)
def keys(self) -> list[str]:
return list(self._store)
# Demonstrate the pattern manually
store = SimpleMemoryStore()
store.save(
"lifespans",
"C. elegans: ~18 days. Mouse: ~24-30 months. Killifish: 4-6 months (shortest vertebrate).",
)
store.save(
"tractability",
"Worm has genome-wide RNAi by feeding. Mouse has Cre-lox conditionals. Killifish toolkit still thin.",
)
# In "session 2", you'd inject these into the first prompt:
preamble = "\n".join(f"- {k}: {store.load(k)}" for k in store.keys())
print(f"Session-2 opening prompt would include:\n\nPrior research notes:\n{preamble}")Session-2 opening prompt would include:
Prior research notes:
- lifespans: C. elegans: ~18 days. Mouse: ~24-30 months. Killifish: 4-6 months (shortest vertebrate).
- tractability: Worm has genome-wide RNAi by feeding. Mouse has Cre-lox conditionals. Killifish toolkit still thin.
```以下示例展示了模式，但它让您负责决定保存什么以及何时加载。这正是模型更有能力去做的工作：它在推理过程中知道哪些事实重要，以及何时需要回忆它们。
### 使用API
我们的API以`memory_20250818`工具的形式原生提供此功能。模型决定保存什么以及何时保存，作为其工具使用循环的一部分，一个自动注入的系统提示建立了协议（“在做任何其他事情之前，始终查看您的内存目录”），并且该工具提供完整的文件操作，而不是键值操作。这是一个客户端工具：API提供协议，您实现文件后端。
**API文档：** [内存工具 — platform.claude.com](https://platform.claude.com/docs/en/agents-and-tools/tool-use/memory-tool)
以下是一个实现所有六个命令的完整处理程序。
```python
def handle_command(command):
# 处理命令的代码
pass{
"command": "save_memory",
"data": {
"memory": "重要信息"
}
}SELECT * FROM memory WHERE id = 1;
``````python
class MemoryToolHandler:
"""Client-side backend for the memory_20250818 tool.
Implements view, create, str_replace, insert, delete, rename.
File-backed (tempdir) so writes persist across notebook cells.
"""
def __init__(self, base_dir: str | None = None):
self.base = Path(base_dir or tempfile.mkdtemp(prefix="memory_"))
self.base.mkdir(parents=True, exist_ok=True)
def _resolve(self, path: str) -> Path:
rel = path.removeprefix("/memories").removeprefix("/") or "."
resolved = (self.base / rel).resolve()
if not str(resolved).startswith(str(self.base.resolve())):
raise ValueError(f"path escapes memory directory: {path}")
return resolved
def handle(self, tool_input: dict) -> str:
cmd = tool_input.get("command")
try:
if cmd == "view":
return self._view(tool_input.get("path", "/memories"))
if cmd == "create":
return self._create(tool_input["path"], tool_input["file_text"])
if cmd == "str_replace":
return self._str_replace(
tool_input["path"], tool_input["old_str"], tool_input["new_str"]
)
if cmd == "insert":
return self._insert(
tool_input["path"], tool_input["insert_line"], tool_input["insert_text"]
)
if cmd == "delete":
return self._delete(tool_input["path"])
if cmd == "rename":
return self._rename(tool_input["old_path"], tool_input["new_path"])
return f"Error: unknown command '{cmd}'"
except Exception as e:
return f"Error: {e}"
def _view(self, path: str) -> str:
p = self._resolve(path)
if not p.exists():
return (
"Directory /memories is empty."
if path in ("/memories", "/memories/")
else f"Error: '{path}' does not exist."
)
if p.is_dir():
entries = sorted(p.iterdir())
if not entries:
return "(empty directory)"
return "\n".join((f"{e.name}/" if e.is_dir() else e.name) for e in entries)
content = p.read_text()
lines = content.split("\n")
return "\n".join(f"{i + 1:4d}\t{line}" for i, line in enumerate(lines))
def _create(self, path: str, file_text: str) -> str:
p = self._resolve(path)
p.parent.mkdir(parents=True, exist_ok=True)
p.write_text(file_text)
return f"Created {path}"
def _str_replace(self, path: str, old_str: str, new_str: str) -> str:
p = self._resolve(path)
content = p.read_text()
if old_str not in content:
return f"Error: old_str not found in {path}"
if content.count(old_str) > 1:
return f"Error: old_str appears {content.count(old_str)} times; be more specific."
p.write_text(content.replace(old_str, new_str, 1))
return f"Replaced in {path}"
def _insert(self, path: str, insert_line: int, insert_text: str) -> str:
p = self._resolve(path)
lines = p.read_text().split("\n")
lines.insert(insert_line, insert_text)
p.write_text("\n".join(lines))
return f"Inserted at line {insert_line} in {path}"
def _delete(self, path: str) -> str:
p = self._resolve(path)
if p.is_dir():
import shutil
shutil.rmtree(p)
else:
p.unlink()
return f"Deleted {path}"
def _rename(self, old_path: str, new_path: str) -> str:
old_p, new_p = self._resolve(old_path), self._resolve(new_path)
new_p.parent.mkdir(parents=True, exist_ok=True)
old_p.rename(new_p)
return f"Renamed {old_path} → {new_path}"
```> 安全提示：`_resolve` 方法防止路径遍历（`../../etc/passwd`）。在生产环境中，您还希望限制文件大小和总目录大小。请参阅[内存食谱](https://platform.claude.com/cookbook/tool-use-memory-cookbook)以了解更深入的内存模式。
### 展示优势
为了具体地看到效果，我们在三个会话中运行代理：
1. **会话 1** 进行初始研究并通过 `/memories` 写入其发现。
2. **会话 2（无内存）** 在一个空的内存目录中运行后续任务。它必须从头开始重新发现一切。
3. **会话 2（有内存）** 运行相同的后续任务，但可以访问会话 1 保存的文件。它首先读取这些文件，并在其基础上构建，而不是重新进行研究。
两个会话 2 运行的比较是内存优势变得明显的地方。
::: callout-注意
请注意，这里使用了Markdown格式，包括标题、列表、链接、图片等，并且保留了技术术语和代码引用。
:::```python
# Session 1: initial research pass, memory tool enabled
memory = MemoryToolHandler()
# Focused task for S1: read a handful of docs, save the comparative findings
S1_TASK = """Compare four model organisms for aging research: C. elegans, mouse,
killifish, and yeast. For each one, summarize lifespan, genetic tractability,
and translational relevance. Read the review documents and save your
comparative findings to /memories so a follow-up session can build on them."""
memory_s1 = run_research_session(
S1_TASK,
memory_handler=memory,
max_turns=15,
label="memory/s1",
)
print("=" * 60)
print("What Session 1 wrote to /memories:")
print("=" * 60)
for f in sorted(memory.base.iterdir()):
content = f.read_text()
print(f"\n▸ {f.name} (~{count_tokens(content):,} tokens)")
for line in content.split("\n")[:10]:
print(f" {line}")
if len(content.split("\n")) > 10:
print(" ...")┌─ [memory/s1]
│ turn 1 ctx= 1,941
│ memory view /memories
│ turn 2 ctx= 2,025
│ search_files 'C. elegans aging lifespan'
│ search_files 'mouse aging model organism'
│ search_files 'killifish aging'
│ search_files 'yeast aging replicative chronological'
│ turn 3 ctx= 2,386
│ search_files 'aging model organism review'
│ search_files 'lifespan genetics'
│ turn 4 ctx= 2,576
│ search_files 'review' → ~205 tok
│ search_files 'organism' → ~205 tok
│ turn 5 ctx= 3,296
│ read_file /research/celegans_review.md → ~27,465 tok
│ read_file /research/mouse_review.md → ~27,390 tok
│ read_file /research/killifish_review.md → ~27,349 tok
│ read_file /research/yeast_review.md → ~27,320 tok
│ turn 6 ctx=168,218
│ memory create /memories/aging_model_organisms_comparison.md
│ turn 7 ctx=171,354
│ record_finding "Four aging model organism reviews fully read and s..."
│ turn 8 ctx=171,935 (final answer)
└─ completed: 8 turns, peak ctx 171,935, final ctx 171,935, 0 context event(s)
============================================================
What Session 1 wrote to /memories:
============================================================
▸ aging_model_organisms_comparison.md (~2,999 tokens)
# Comparative Notes: Four Model Organisms in Aging Research
## Source files read: /research/celegans_review.md, mouse_review.md, killifish_review.md, yeast_review.md
## Session status: COMPLETE — all four reviews read, notes synthesised
---
## 1. C. elegans (Caenorhabditis elegans)
### Lifespan
- Wild-type median: ~18 days at 20°C (N2 strain)
...# Session 2 WITHOUT memory: follow-up task with empty /memories
# This shows what happens when the agent has no prior knowledge to draw on.
S2_TASK = (
"Write a final synthesis comparing all eight model organisms for "
"aging research (C. elegans, Drosophila, yeast, killifish, mouse, "
"zebrafish, naked mole-rat, rhesus), focusing on which organisms "
"are strongest for which kinds of questions. Use information from "
"/memories for organisms already covered there; read source "
"documents for any organism not yet captured."
)
empty_memory = MemoryToolHandler() # fresh, empty /memories
memory_s2_without = run_research_session(
S2_TASK,
memory_handler=empty_memory,
max_turns=12,
label="memory/s2-without",
)┌─ [memory/s2-without]
│ turn 1 ctx= 1,952
│ memory view /memories
│ turn 2 ctx= 2,056
│ search_files 'C. elegans aging model organism'
│ search_files 'Drosophila aging model organism'
│ search_files 'yeast aging model organism'
│ search_files 'killifish aging model organism'
│ turn 3 ctx= 2,415
│ search_files 'aging model organism lifespan'
│ search_files 'mouse aging research'
│ search_files 'naked mole-rat longevity'
│ search_files 'rhesus macaque aging'
│ turn 4 ctx= 2,755
│ search_files 'model organism' → ~205 tok
│ search_files 'aging research review'
│ search_files 'lifespan genetics'
│ turn 5 ctx= 3,277
│ read_file /research/celegans_review.md → ~27,465 tok
│ read_file /research/drosophila_review.md → ~27,435 tok
│ read_file /research/yeast_review.md → ~27,320 tok
│ read_file /research/killifish_review.md → ~27,349 tok
│ turn 6 ctx=168,250
│ read_file /research/mouse_review.md → ~27,390 tok
│ read_file /research/zebrafish_review.md → ~27,363 tok
│ read_file /research/nmr_review.md → ~27,287 tok
│ read_file /research/rhesus_review.md → ~27,285 tok
│ turn 7 ctx=332,824
│ memory create /memories/organism_notes.md
│ turn 8 ctx=333,872
│ record_finding "All 8 organism review documents read and key facts..."
│ turn 9 ctx=333,977 (final answer)
└─ completed: 9 turns, peak ctx 333,977, final ctx 333,977, 0 context event(s)# Session 2 WITH memory: same follow-up task, but with access to Session 1's files
# The agent should read /memories first and build on it instead of re-researching.
memory_s2 = run_research_session(
S2_TASK,
memory_handler=memory, # SAME handler as Session 1
max_turns=12,
label="memory/s2-with",
)
# Show what Session 2 read from /memories in its opening turns
print("=" * 60)
print("What Session 2 (with memory) read from /memories:")
print("=" * 60)
for msg_i, msg in enumerate(memory_s2.messages[:8]):
# Look for memory tool calls in assistant turns
if msg["role"] == "assistant" and isinstance(msg["content"], list):
for block in msg["content"]:
if block.get("type") == "tool_use" and block.get("name") == "memory":
inp = block.get("input", {})
print(f"\n[memory {inp.get('command', '?')}] {inp.get('path', '')}")
# Look for the corresponding tool results (what was read)
if msg["role"] == "user" and isinstance(msg["content"], list) and msg_i <= 6:
for block in msg["content"]:
if block.get("type") == "tool_result":
result = block.get("content", "")
if isinstance(result, str) and len(result) > 20:
# Show a preview of what was read
preview = result[:350].replace("\n", "\n ")
print(f" └ {preview}{'...' if len(result) > 350 else ''}")┌─ [memory/s2-with]
│ turn 1 ctx= 1,952
│ memory view /memories
│ turn 2 ctx= 2,043
│ memory view /memories/aging_model_organisms_comparison.md → ~2,707 tok
│ turn 3 ctx= 5,737
│ read_file /research/drosophila_review.md → ~27,435 tok
│ read_file /research/zebrafish_review.md → ~27,363 tok
│ read_file /research/nmr_review.md → ~27,287 tok
│ read_file /research/rhesus_review.md → ~27,285 tok
│ turn 4 ctx=170,401
│ memory str_replace /memories/aging_model_organisms_comparison.md
│ turn 5 ctx=172,415
│ record_finding "All eight model organism reviews fully read and sy..."
│ turn 6 ctx=172,623 (final answer)
└─ completed: 6 turns, peak ctx 172,623, final ctx 172,623, 0 context event(s)
============================================================
What Session 2 (with memory) read from /memories:
============================================================
[memory view] /memories
└ aging_model_organisms_comparison.md
[memory view] /memories/aging_model_organisms_comparison.md
└ 1 # Comparative Notes: Four Model Organisms in Aging Research
2 ## Source files read: /research/celegans_review.md, mouse_review.md, killifish_review.md, yeast_review.md
3 ## Session status: COMPLETE — all four reviews read, notes synthesised
4
5 ---
6
7 ## 1. C. elegans (Caenorhabditis elegans)
8
9 ### Lifespan
10 -...
└ # Model Organism Review: Drosophila melanogaster in Aging Research
Drosophila melanogaster, the common fruit fly, has been a genetic model for
over a century and an aging model since Pearl's demographic work in the
1920s. It occupies a useful middle ground: more complex than C. elegans
(tissue diversity, a functional heart, behavioral repertoire) ...
└ # Model Organism Review: Danio rerio (Zebrafish) in Aging Research
Zebrafish are a vertebrate model with particular strengths in developmental
biology and regeneration that have been increasingly adopted for aging
research over the past decade. Their transparent larvae and high fecundity
make them competitive with invertebrates for some screening ...
└ # Model Organism Review: Heterocephalus glaber (Naked Mole-Rat)
The naked mole-rat is an eusocial rodent from East African burrows with a
maximum lifespan exceeding 30 years, roughly ten times that of a similarly
sized mouse. It is studied as a model of exceptional longevity rather than
as a general aging model: the question is what makes it long-...
└ # Model Organism Review: Macaca mulatta (Rhesus Macaque)
Rhesus macaques are the primary non-human primate model for aging research.
With a maximum lifespan of ~40 years and physiology closely mirroring human,
they represent the closest experimentally accessible approximation to human
aging, at correspondingly high cost.
## Lifespan and throughpu...
[memory str_replace] /memories/aging_model_organisms_comparison.md# Compare: the bar chart shows final context and file reads side by side.
plot_summary_bars(
{
"S2 without memory": memory_s2_without,
"S2 with memory": memory_s2,
},
"Memory benefit: Session 2 with vs. without Session 1's saved findings",
)
s2_with_reads = memory_s2.tool_counts.get("read_file", 0)
s2_without_reads = memory_s2_without.tool_counts.get("read_file", 0)
s2_with_mem = memory_s2.tool_counts.get("memory", 0)
s2_without_mem = memory_s2_without.tool_counts.get("memory", 0)
print("Session 2 comparison:")
print(
f" Without memory: {s2_without_reads} file reads, {s2_without_mem} memory ops, "
f"peak ctx {max(t for _, t in memory_s2_without.token_trajectory):,}"
)
print(
f" With memory: {s2_with_reads} file reads, {s2_with_mem} memory ops, "
f"peak ctx {max(t for _, t in memory_s2.token_trajectory):,}"
)
if s2_with_reads < s2_without_reads:
saved = s2_without_reads - s2_with_reads
print(
f"\n→ Memory saved {saved} file read(s): the agent pulled Session 1's "
"findings from /memories instead of re-reading the source documents."
)
elif s2_with_reads == 0 and s2_with_mem > 0:
print(
"\n→ Session 2 with memory completed using only /memories; "
"no source-document re-reads needed."
)
else:
print(
"\n→ Read counts are similar this run. The benefit depends on how "
"comprehensive Session 1's notes were and how the agent chose to use them."
)
print(
"\nThis comparison shows memory working well because Session 1's notes "
"were comprehensive. If Session 1 had saved sparse or poorly organized "
"notes, Session 2 would fall back to re-reading source documents. "
"Memory's value depends on the agent's judgment about what to write."
)<Figure size 900x350 with 2 Axes>Session 2 comparison:
Without memory: 8 file reads, 2 memory ops, peak ctx 333,977
With memory: 4 file reads, 3 memory ops, peak ctx 172,623
→ Memory saved 4 file read(s): the agent pulled Session 1's findings from /memories instead of re-reading the source documents.
This comparison shows memory working well because Session 1's notes were comprehensive. If Session 1 had saved sparse or poorly organized notes, Session 2 would fall back to re-reading source documents. Memory's value depends on the agent's judgment about what to write.
```### 分析
比较使好处具体化。没有记忆的Session 2没有任何东西可以借鉴；`/memories`为空，因此它必须回到源文档中重新发现相同的事实。带有记忆的Session 2通过读取`/memories`（自动注入的协议使其成为默认的第一步），找到Session 1保存的发现，并可以从这些发现中构建综合，而不是重新阅读每个源文档。
这实际上是即时检索：不是将所有先前知识加载到第一个提示中，而是代理根据需要从记忆中提取相关部分。文件读取次数和条形图中的最终上下文直接量化了差异。
记忆能给你的是跨会话持久性，对代理选择保存的任何内容都能保持无损的保真度。它不能给你的是任何帮助会话上下文增长（Session 1的峰值上下文仍然很高），并且每次读取和写入都会增加工具调用的开销。记忆解决了跨会话问题；清除和压缩解决了会话内的问题。
### 有效地实现 Memory
`memory_20250818` 工具自动注入一个系统提示，建立检查内存优先的协议和假设中断的心态（“在做任何其他事情之前...始终查看您的内存目录...您的上下文窗口可能随时被重置”）。这处理了基本机制。除此之外，[记忆工具文档](https://platform.claude.com/docs/en/agents-and-tools/tool-use/memory-tool#prompting-guidance)描述了多种塑造模型保存内容的方法：
**主题指导。**您可以使用简单的系统提示指令来引导写入的内容：“仅在您的记忆系统中写下与<主题>相关的信息。”对于这本烹饪书的研发代理，这可能意味着“保存比较发现和关键数据，而不是原始文档内容。”
```::: callout-attention
请注意，以下内容为Markdown格式，请确保在翻译时保留格式。
```::: callout-attention
Please note that the following content is in Markdown format, and ensure that the format is retained in the translation.
```::: callout-attention
以下代码块中的内容为示例代码，请确保在翻译时保留代码格式。
```::: callout-code
def example_function():
# 示例代码
print("Hello, World!")保持 /memories 有序。 如果您观察到模型创建了杂乱的内存文件，尝试添加以下内容：“在编辑您的内存文件夹时，始终尝试保持其内容最新、一致和有序。您可以重命名或删除不再相关的文件。除非必要，否则不要创建新文件。” 这可以防止目录在会话之间积累半重叠的笔记。
初始化会话结构。 对于多会话工作，尝试运行一个专门的第一会话，在实质性工作开始之前设置内存工件：进度日志、功能清单、任何设置脚本的引用。随后的会话通过读取这些工件来恢复状态。以这种方式预先填充 /memories 为后续会话提供了一个一致的结构来工作，而不是每个会话都发明自己的组织。
存储卫生。 在客户端，您还可以跟踪文件大小以防止无限制增长，考虑清除长时间未访问的内存文件，并验证路径遍历。
总结：每个原语的作用
| 原语 | 操作于 | 付出的代价 | 解决 |
|---|---|---|---|
| 压缩 | 当前窗口中的整个对话 | 将逐字细节压缩成摘要；可能丢失模糊的具体信息 | 会话内所有增长 |
| 清除 | 当前窗口中的工具结果 | 旧工具结果从上下文中消失（如果需要再次获取，必须重新获取） | 工具结果膨胀 |
| 内存 | 外部存储，跨窗口 | 工具调用开销；仅与代理选择保存的内容一样好 | 跨会话持久性 |
下表将三个单独的运行并排展示，加上基线。请注意，内存的会话2是不同的任务（后续综合），因此绝对数字不能直接与其他人比较；对于内存来说，重要的是上面显示的 S2-with 与 S2-without 的比较。```python # Side-by-side view of each primitive’s solo run solo_results = { “baseline”: baseline, “clearing”: clearing_run, “compaction”: compaction_run, “memory (S2 with)”: memory_s2, } plot_summary_bars(solo_results, “Each primitive alone: session outcomes”)
Completion status and events per run
print(f”{‘Run’:<20} {‘Status’:<18} {‘Context events’}“) print(”-” * 60) for name, res in solo_results.items(): status = “HIT LIMIT” if res.hit_limit else “completed” n_clear = sum(1 for e in res.events if e[“kind”] == “clearing”) n_compact = sum(1 for e in res.events if e[“kind”] == “compaction”) ev_parts = [] if n_clear: ev_parts.append(f”{n_clear} clearing”) if n_compact: ev_parts.append(f”{n_compact} compaction”) print(f” {name:<18} {status:<18} {‘,’.join(ev_parts) if ev_parts else ‘none’}“)
<Figure size 900x350 with 2 Axes>
Run Status Context events
baseline completed none clearing completed 4 clearing compaction completed 1 compaction memory (S2 with) completed none ``三个原语针对上下文问题的不同部分，因此它们可以分层使用。[Claude Code](https://code.claude.com/docs/zh/memory) 是一个实际应用的例子，它使用了压缩功能以及 [两个互补的内存系统](https://code.claude.com/docs/zh/memory#claudemd-vs-auto-memory)：CLAUDE.md` 文件包含用户定义的指令和规则（编码标准、项目架构、工作流程），这些是由开发者编写并提交到源代码控制中的；自动内存则包含 Claude 自己编写的知识和模式（构建命令、调试洞察、从更正中发现的偏好）。这两种都是 Claude Code 有用的记忆形式。
Claude Code 的设计表明，相同的代理可以有不同的内存形态；一种是用户编写的，另一种是模型编写的。同样适用于压缩和清除：每个都有配置旋钮（触发阈值、自定义指令、排除哪些工具），这让你可以根据自己的用例调整行为。这就是为什么上面“有效实施”部分中的提示和配置指南很重要：默认行为是一个起点，但正确的设置取决于你的代理实际上做什么。
组合使用
这三个原语针对上下文问题的不同部分，因此它们可以分层。Claude Code 是一个真实世界的例子，它使用了压缩功能以及 两个互补的内存系统：CLAUDE.md 文件保存用户定义的指令和规则（编码标准、项目架构、工作流程），这些是由开发者编写的并检查到源代码控制中；自动内存保存 Claude 自己编写的知识和模式（构建命令、调试洞察、从更正中发现的偏好）。这两种都是 Claude Code 有用的记忆形式。
Claude Code 的设计表明，相同的代理可以有不同的内存形态；一种是用户编写的，另一种是模型编写的。同样适用于压缩和清除：每个都有配置旋钮（触发阈值、自定义指令、排除哪些工具），这让你可以根据自己的用例调整行为。这就是为什么上面“有效实施”部分中的提示和配置指南很重要：默认行为是一个起点，但正确的设置取决于你的代理实际上做什么。
请注意，以下代码示例中的 print 函数用于输出结果。
def print_result(result):
print(result)关于
exclude_tools的说明：当将清除操作与内存工具结合使用时，exclude_tools: ["memory"]设置（如下配置所示）可以防止代理的内存读写被清除。如果没有这个设置，代理可能会丢失它刚刚保存的跟踪。在内存工具文档中，建议在分层使用这两个工具时明确使用此设置。
以下我们同时激活所有三个原语运行研究代理，并追踪会话过程中每个原语所执行的操作。
关于配置的说明：两个触发器都设置在第一个批次大小之上（约167K），因此轨迹跟踪基线通过批次1。当批次2将上下文推送到约330K时，清除操作首先触发（keep=6丢弃最早的读取，留下约210K），然后压缩操作在清除留下的内容上触发。内存在整个过程中都是活跃的。这种调整是为了演示，以便在单次运行中激活所有三个原语。生产配置取决于您的代理特定的上下文增长模式。
# 以下代码展示了如何使用RAG进行文本嵌入
def embed_text(text):
# 将文本转换为嵌入
embedding = rag.embed(text)
return embedding
# 以下代码展示了如何使用token进行文本分割
def tokenize_text(text):
# 将文本分割为token
tokens = rag.tokenize(text)
return tokens
# 以下代码展示了如何使用prompt进行文本生成
def generate_text(prompt):
# 使用prompt生成文本
generated_text = rag.generate(prompt)
return generated_text
# 以下代码展示了如何使用Agent进行API调用
def call_api(agent, url, data):
# 使用Agent调用API
response = agent.call(url, data)
return response
# 以下代码展示了如何使用Python处理JSON数据
def process_json(json_data):
# 处理JSON数据
data = json.loads(json_data)
return data
# 以下代码展示了如何使用SQL查询数据库
def query_database(sql):
# 查询数据库
results = sql.query(sql)
return results
``````python
# Both triggers sit above batch-1's ~167K so the rise is visible. At
# batch-2 (~330K) clearing fires first (keep=6 drops the earliest 3-4
# reads, leaving ~210K) and compaction then fires on what clearing
# left. Tuned for demo visibility, not a production recommendation.
combo_memory = MemoryToolHandler()
COMBO_CLEARING_TRIGGER = 200_000
COMBO_COMPACTION_TRIGGER = 180_000
combo_run = run_research_session(
"""Compare the model organisms in /research/ on lifespan, genetic
tractability, and translational relevance.
Work in two batches, recording a note after each batch before moving on.
Batch 1 (high-throughput): read celegans_review.md, drosophila_review.md,
yeast_review.md, killifish_review.md, then record a note. Batch 2
(low-throughput): read mouse_review.md, zebrafish_review.md,
nmr_review.md, rhesus_review.md, then record a note. Save a comparative
summary to /memories so future sessions can build on it. Finish with a
detailed synthesis.""",
betas=["context-management-2025-06-27", "compact-2026-01-12"],
context_management={
"edits": [
{
"type": "clear_tool_uses_20250919",
"trigger": {"type": "input_tokens", "value": COMBO_CLEARING_TRIGGER},
"keep": {"type": "tool_uses", "value": 6},
"clear_at_least": {"type": "input_tokens", "value": 30_000},
"exclude_tools": ["memory"],
},
{
"type": "compact_20260112",
"trigger": {"type": "input_tokens", "value": COMBO_COMPACTION_TRIGGER},
},
],
},
memory_handler=combo_memory,
max_turns=20,
label="all-three",
)┌─ [all-three]
│ turn 1 ctx= 2,106
│ memory view /memories
│ turn 2 ctx= 2,206
│ read_file /research/celegans_review.md → ~27,465 tok
│ read_file /research/drosophila_review.md → ~27,435 tok
│ read_file /research/yeast_review.md → ~27,320 tok
│ read_file /research/killifish_review.md → ~27,349 tok
│ turn 3 ctx=167,185
│ memory create /memories/batch1_notes.md
│ turn 4 ctx=169,938
│ read_file /research/mouse_review.md → ~27,390 tok
│ read_file /research/zebrafish_review.md → ~27,363 tok
│ read_file /research/nmr_review.md → ~27,287 tok
│ read_file /research/rhesus_review.md → ~27,285 tok
│ ✂ CLEARING (turn 5): 2 tool results cleared, ~81,993 tokens freed
│ ⊟ COMPACTION (turn 5): ~2,839-token summary replaces prior turns
│ ✂ CLEARING (turn 6): 2 tool results cleared, ~81,810 tokens freed
│ turn 6 ctx= 4,901
│ memory view /memories/batch1_notes.md → ~2,370 tok
│ ✂ CLEARING (turn 7): 2 tool results cleared, ~81,683 tokens freed
│ turn 7 ctx= 8,121
│ memory create /memories/batch2_notes.md
│ ✂ CLEARING (turn 8): 2 tool results cleared, ~81,508 tokens freed
│ turn 8 ctx= 10,524
│ memory create /memories/comparative_summary.md
│ ✂ CLEARING (turn 9): 2 tool results cleared, ~81,413 tokens freed
│ turn 9 ctx= 12,799
│ search_files 'translational relevance aging pipeline'
│ search_files 'lifespan maximum cohort size intervention'
│ ✂ CLEARING (turn 10): 4 tool results cleared, ~162,379 tokens freed
│ turn 10 ctx= 13,096
│ search_files 'yeast replicative lifespan'
│ search_files 'rhesus macaque caloric restriction'
│ ✂ CLEARING (turn 11): 4 tool results cleared, ~162,371 tokens freed
│ turn 11 ctx= 13,290
│ record_finding "All 8 model organisms reviewed and compared. Notes..."
│ ✂ CLEARING (turn 12): 4 tool results cleared, ~162,337 tokens freed
│ turn 12 ctx= 13,461
│ memory str_replace /memories/comparative_summary.md
│ ✂ CLEARING (turn 13): 4 tool results cleared, ~162,331 tokens freed
│ turn 13 ctx= 13,749 (final answer)
└─ completed: 13 turns, peak ctx 169,938, final ctx 13,749, 10 context event(s)# Build a timeline showing when each primitive activated over the session
print("=" * 70)
print("SESSION TIMELINE: all three primitives active")
print("=" * 70)
# Build turn-indexed event markers
clear_turns = {e["turn"] for e in combo_run.events if e["kind"] == "clearing"}
compact_turns = {e["turn"] for e in combo_run.events if e["kind"] == "compaction"}
# Find per-turn memory tool calls by walking messages
mem_call_turns = set()
turn = 0
for msg in combo_run.messages:
if msg["role"] == "assistant" and isinstance(msg["content"], list):
turn += 1
for block in msg["content"]:
if block.get("type") == "tool_use" and block.get("name") == "memory":
mem_call_turns.add(turn)
# Print turn-by-turn with markers
for t, ctx in combo_run.token_trajectory:
markers = []
if t in clear_turns:
markers.append("✂ CLEARING")
if t in compact_turns:
markers.append("⊟ COMPACTION")
if t in mem_call_turns:
markers.append("◇ memory")
marker_str = " " + " ".join(markers) if markers else ""
print(f" turn {t:2d} ctx={ctx:>7,}{marker_str}")
# Summary of what each primitive did
n_clearing = len(clear_turns)
n_compaction = len(compact_turns)
n_memory_calls = combo_run.tool_counts.get("memory", 0)
mem_files = sorted(combo_memory.base.iterdir())
print("\n" + "─" * 70)
print("What each primitive did this session:")
print(f"\n ✂ Tool-result clearing fired {n_clearing} time(s)")
print(
f" → Dropped old tool results to keep context capped at ~{max(t for _, t in combo_run.token_trajectory):,} tokens"
)
print(
f" → {combo_run.tool_counts.get('read_file', 0)} file reads total "
f"(vs baseline's {baseline.tool_counts.get('read_file', 0)}); some were re-fetches after context was reset"
)
if n_compaction > 0:
print(f"\n ⊟ Compaction fired {n_compaction} time(s)")
for ev in combo_run.events:
if ev["kind"] == "compaction":
print(
f" → At turn {ev['turn']}: replaced prior conversation with ~{count_tokens(ev['summary']):,}-token summary"
)
else:
print(
f"\n ⊟ Compaction did not fire (clearing kept context below the {COMBO_COMPACTION_TRIGGER:,} trigger)"
)
print(
f"\n ◇ Memory tool called {n_memory_calls} time(s), wrote {len(mem_files)} file(s) to /memories"
)
for f in mem_files:
content = f.read_text()
print(f" → {f.name} (~{count_tokens(content):,} tokens)")
lines = content.split("\n")
for line in lines[:6]:
print(f" {line}")
if len(lines) > 6:
print(" ...")
print("\n" + "─" * 70)
print(
f"Peak context: {max(t for _, t in combo_run.token_trajectory):,} tokens "
f"(vs baseline's {max(t for _, t in baseline.token_trajectory):,})"
)
print(f"Final context: {combo_run.token_trajectory[-1][1]:,} tokens")
print(
f"Session {'completed' if not combo_run.hit_limit else 'HIT LIMIT'} "
f"in {len(combo_run.token_trajectory)} turns"
)======================================================================
SESSION TIMELINE: all three primitives active
======================================================================
turn 1 ctx= 2,106 ◇ memory
turn 2 ctx= 2,206
turn 3 ctx=167,185 ◇ memory
turn 4 ctx=169,938
turn 5 ctx= 4,798 ✂ CLEARING ⊟ COMPACTION ◇ memory
turn 6 ctx= 4,901 ✂ CLEARING ◇ memory
turn 7 ctx= 8,121 ✂ CLEARING ◇ memory
turn 8 ctx= 10,524 ✂ CLEARING ◇ memory
turn 9 ctx= 12,799 ✂ CLEARING
turn 10 ctx= 13,096 ✂ CLEARING
turn 11 ctx= 13,290 ✂ CLEARING
turn 12 ctx= 13,461 ✂ CLEARING ◇ memory
turn 13 ctx= 13,749 ✂ CLEARING
──────────────────────────────────────────────────────────────────────
What each primitive did this session:
✂ Tool-result clearing fired 9 time(s)
→ Dropped old tool results to keep context capped at ~169,938 tokens
→ 8 file reads total (vs baseline's 8); some were re-fetches after context was reset
⊟ Compaction fired 1 time(s)
→ At turn 5: replaced prior conversation with ~2,839-token summary
◇ Memory tool called 7 time(s), wrote 3 file(s) to /memories
→ batch1_notes.md (~2,610 tokens)
# Batch 1: High-Throughput Model Organisms — Notes
## 1. C. elegans (Caenorhabditis elegans)
### Lifespan
- Wild-type median: ~18 days at 20°C (N2 strain)
...
→ batch2_notes.md (~2,254 tokens)
# Batch 2: Vertebrate & Exceptional Longevity Models — Notes
## 5. Mus musculus (Mouse)
### Lifespan
- Median: ~24–30 months (C57BL/6J); max ~4.5 yrs (Ames dwarf + CR)
...
→ comparative_summary.md (~2,173 tokens)
# Comparative Summary: Model Organisms for Aging Research
## Quick Reference Table
| Organism | Median Lifespan | Time to Result | Human Orthology | Genetic Tractability | Translational Relevance | Primary Role |
|---|---|---|---|---|---|---|
...
──────────────────────────────────────────────────────────────────────
Peak context: 169,938 tokens (vs baseline's 335,279)
Final context: 13,749 tokens
Session completed in 13 turnsplot_trajectories(
{"baseline": baseline, "all three active": combo_run},
"All three primitives active: context stays bounded while the session continues",
triggers={"clearing": COMBO_CLEARING_TRIGGER, "compaction": COMBO_COMPACTION_TRIGGER},
)<Figure size 1000x450 with 1 Axes>### 时间线显示的内容
With all three primitives active, each one activated for its own reason during the session. The trajectory tracked the baseline through batch 1: both triggers sit above the first batch’s size, so neither edit fired until batch 2 pushed context past ~330K. At that point clearing dropped the earliest reads and compaction summarized what remained, letting the agent continue. Memory was active throughout, with the agent checking /memories at the start and saving its comparative notes for future sessions. The timeline above shows all three cooperating across one session.
Getting the primitives to split the work usefully takes some tuning; plan to experiment with the values against your own workload.
The point isn’t that running all three produces the “best” numbers; it’s that they each handle a different part of the problem when that problem actually arises. The useful question isn’t “should I use all three?” but “which of the three problems does my workload actually have?”
何时可能不需要某个原语
Not every workload needs every tool. A few cases where you’d deliberately leave one out:
- Skip memory if you want each session to start fresh. A user-facing chatbot where every conversation should be independent doesn’t need cross-session persistence; adding memory would carry state you don’t want.
- Skip compaction if your sessions are short enough to stay under the context limit naturally. Compaction is lossy (specific details get summarized away), so if you don’t need the headroom, you’re paying fidelity for nothing.
- Skip clearing if the agent genuinely needs to see past tool results in full. An agent doing cross-document analysis where it compares passages side by side can’t re-fetch its way back to a cleared result fast enough; clearing would force redundant reads.
要点和后续步骤
实验中的经验教训
Running the research agent under these different configurations surfaces a few practical lessons: 轨迹的形状反映了每个工具的作用。 当清除火灾时，你会在转弯处看到下降，这是旧工具结果被移除的地方；在较长的会话中，这可能会随着上下文再次超过触发器而重复。压缩每次触发时都会产生更大的下降，因为摘要替换了整个运行周期，而不仅仅是其中的工具结果。本食谱中的图表旨在使这些效果可见，这样你可以具体地看到当你转动旋钮时发生了什么。哪种工具适合你的工作负载是一个单独的问题，由代理需要做什么来驱动。
损失性是一个连续体，而不是二元的。 只要工具可以重新调用，清除就是无损的。压缩以受控的方式损失：摘要提示（默认或自定义）决定了什么可以幸存。内存对保存的内容是无损的，但它的好坏取决于代理对保存内容的判断。每个原语以不同的方式权衡保真度。
分层在能力和复杂性上增加相同程度的量。 使用这三个工具可以覆盖更多的上下文问题，但也意味着更多的旋钮需要调整和更多的交互需要追踪。在添加原语之前，有用的问题是它解决你工作负载中的哪个具体问题。
在更大的上下文窗口中。 随着 Sonnet 4.6 和 Opus 4.6 提供 1M-token 的上下文，这个空间是有用的：更多的逐字细节可以保留，损失性操作可以分散进行。但是，正如基线上下文分解所显示的，1M 模型的工作集会像 200K 模型一样快速地充满过时的工具结果；区别在于硬限制在哪里，而不是上下文积累的速度。上下文腐烂和预填充延迟与窗口中的内容量成比例，而不是与窗口的限制成比例，因此即使硬限制很远，保持工作集精简仍然值得做。
思考你的工作负载
| 如果你的工作负载有… | 首先尝试的原始方法 | 注意事项 |
|---|---|---|
| 持续数天或数周会话 | 内存工具 | 工具调用开销；如果事实发生变化，则内存过时 |
| 应该在会话之间学习用户偏好的代理 | 内存工具 | 个人信息/敏感数据策略；如果用户改变主意，则偏好过时 |
| 大型、可重新获取的工具结果 | 清除 | 代理重新读取刚刚清除的内容；调整 keep 和触发 |
| 对话作为主要上下文 | 压缩 | 特定数字被总结掉 |
| 难以重新获取的工具结果（临时API、上传） | 在清除之上压缩 | 那些特定结果上的摘要保真度 |
| 每个会话都应该从头开始 | 跳过内存 | 你不想要的跨会话状态 |
| 会话保持在窗口之下 | 跳过压缩 | 你不需要的损失性 |
本教程未涵盖的内容
超越基础调优。 上面的“有效实现”部分为每个原始方法提供了起点。下一步是实验：不同的用例将根据参数和提示从相同的原始方法中获得不同的价值。一个编码代理和一个研究代理都可能使用压缩，但适用于一个的 instructions 字符串不适用于另一个；清除阈值和指导模型写入 /memories 的内容也是如此。
设置测试工具包在这里很有帮助。以一个简单的例子，这个食谱中的代理循环（run_research_session）返回 token_trajectory、events 和 tool_counts：你可以在少量配置下运行你的代理，并并排绘制轨迹，并衡量对你来说重要的事情（任务质量、延迟、令牌消耗）。
超越基础调优
超越基础调优。 上面的“有效实现”部分为每个原始方法提供了起点。下一步是实验：不同的用例将根据参数和提示从相同的原始方法中获得不同的价值。一个编码代理和一个研究代理都可能使用压缩，但适用于一个的 instructions 字符串不适用于另一个；清除阈值和指导模型写入 /memories 的内容也是如此。
设置测试工具包在这里很有帮助。以一个简单的例子，这个食谱中的代理循环（run_research_session）返回 token_trajectory、events 和 tool_counts：你可以在少量配置下运行你的代理，并并排绘制轨迹，并衡量对你来说重要的事情（任务质量、延迟、令牌消耗）。 相邻特征。 程序化工具调用（PTC） 通过在由模型编写的程序内部运行工具，防止大量结果进入上下文，这是一种针对工具膨胀问题的不同方法。工具搜索 当你有许多工具时，会修剪工具定义的膨胀。
相关阅读
《内存食谱》 深入探讨了使用代码审查代理的内存模式，而 《压缩食谱》 则单独覆盖了压缩。关于多会话软件代理中上下文管理技术的详细案例研究，请参阅 《为长时间运行的代理的有效约束》。
结论
这个笔记本介绍了为长时间运行的代理进行上下文管理的三个基本原理：压缩以压缩对话历史，工具结果清除以丢弃可重新获取的工具输出，以及内存工具以在会话之间持久化知识。每个原理都针对上下文问题的不同部分。
对于你的代理来说，哪些原理很重要取决于其上下文增长的实际来源。本食谱中的配置和代理循环是在不同配置下运行您自己的工作负载的起点，并查看发生了什么变化。
Tip
中文读者提示：本章节的代码和输出为英文原文。如需理解具体实现细节，可参考上方中文导读和代码注释。如有疑问，欢迎在 GitHub Issues 讨论。