前四篇搓出来的 agent 有一个隐藏的定时炸弹:上下文窗口是有限的。
对话越长,消息历史越大。工具调用的结果尤其占空间——一次 run_bash 可能返回几千行日志,一次 run_read_file 可能读回一整个文件。聊着聊着,token 数就逼近模型的上下文窗口上限了。
到了上限会怎样?要么 API 直接报错,要么模型开始"忘事"——早期的对话内容被挤出注意力范围,模型的回答开始前后矛盾。
你用 Claude Code 时可能注意过,长时间工作后它会提示"context compressed"之类的信息。这不是 bug,是它在主动压缩上下文,腾出空间继续工作。
这篇来实现这个机制。我们要解决两个层面的问题:工具结果的即时压缩,和整体上下文的全量压缩。
第一层:压缩旧的工具结果
先处理最容易膨胀的部分。每次工具调用都会产生一个 ToolMessage,里面装着工具的返回值。这些返回值在刚产生时很有用——模型需要看到结果才能决定下一步。但几轮之后,早期的工具结果就没什么参考价值了。
思路很直接:只保留最近几条 ToolMessage 的完整内容,更早的压缩成 [compressed]。
KEEP_RECENT = 3
def compact_tool_result(messages: list[BaseMessage]) -> list[BaseMessage]:
"""压缩旧的 ToolMessage,只保留最近 KEEP_RECENT 条的完整内容"""
tool_indices = [
i for i, msg in enumerate(messages)
if isinstance(msg, ToolMessage) and len(msg.content) > 100
]
if len(tool_indices) <= KEEP_RECENT:
return messages
for i in tool_indices[:-KEEP_RECENT]:
messages[i] = ToolMessage(
content="[compressed]",
tool_call_id=messages[i].tool_call_id,
)
return messages
几个细节。
只压缩内容超过 100 字符的 ToolMessage。短的结果(比如 (no output) 或 Edited file.py)本身就不占多少空间,压缩了反而丢失信息。
不能直接删除旧的 ToolMessage。因为模型通过 tool_call_id 把工具调用和结果配对。如果结果消失了,消息历史的结构就断了,模型可能会困惑。所以用 [compressed] 替换内容,保留消息本身和 id。
KEEP_RECENT = 3 意味着最近 3 条工具结果保持完整。这个数字是个平衡——太小了模型可能丢失正在进行的工作的上下文,太大了压缩效果不明显。
这个函数在 agent 循环里被频繁调用:每次模型回复后、每次工具执行后都会跑一遍。保持消息列表始终处于"瘦身"状态。
第二层:全量上下文压缩
工具结果压缩能缓解问题,但治不了根。如果用户和 agent 聊了几十轮,光是 HumanMessage 和 AIMessage 的累积就够呛了。
这时候需要更激进的手段:把整个对话历史总结成一段摘要,然后用摘要替换原始历史。
先估算当前 token 数:
TOKEN_LIMIT = 30000
def calculation_token(messages: list) -> int:
"""估算 token 数量"""
return len(str(messages)) // 4
用字符数除以 4 来估算 token 数,粗糙但够用。精确计算需要用 tiktoken 之类的库,对于我们这个 demo 没必要。
每次用户输入后,先检查 token 数。超过 30000 就触发压缩:
if messages:
token_count = calculation_token(messages)
if token_count > TOKEN_LIMIT:
messages = compact_context(messages)
compact_context 是核心函数,做三件事。
第一,把完整的对话历史保存到本地文件。这是保险措施——压缩是有损的,万一需要回溯,原始记录还在:
MEMORY_CACHE_DIR = WORKDIR / ".memory"
memory_path = MEMORY_CACHE_DIR / f"memory_{int(time.time())}.jsonl"
with open(memory_path, "w") as f:
for msg in messages:
f.write(json.dumps(msg, default=str) + "\n")
用时间戳命名文件,每次压缩都保留一份快照。JSONL 格式,一行一条消息,方便后续检索。
第二,调用一个独立的 LLM 实例来总结对话:
_llm = ChatOpenAI(
model="deepseek-chat",
base_url="https://api.deepseek.com/v1",
api_key=os.getenv("DEEPSEEK_API_KEY"),
max_tokens=2000
)
_prompt = ChatPromptTemplate.from_messages([
("human", "Summarize this conversation for continuity. Include: "
"1) What was accomplished, 2) Current state, 3) Key decisions made. "
"Be concise but preserve critical details.\n\n{full_context}")
])
_response = (_prompt | _llm).invoke({"full_context": full_context})
注意 max_tokens=2000。总结不需要太长,2000 token 足够概括一段很长的对话了。prompt 里明确要求总结三个方面:做了什么、当前状态、关键决策。这三个维度能让模型在压缩后继续工作而不丢失方向。
第三,用总结替换原始消息历史:
return [
SystemMessage(content=SYSTEM_PROMPT),
HumanMessage(content=f"[Context has been compressed. Memory: {memory_path}]\n\n{summary}"),
AIMessage(content="Understood. I have the context from the summary. Continuing.")
]
压缩后的消息列表只有三条:system prompt、总结内容、一条确认消息。从几十条消息缩减到三条,token 数断崖式下降。
HumanMessage 里带上了 memory 文件的路径。如果模型后续需要回溯细节,理论上可以用 run_read_file 去读原始记录。
那条 AIMessage("Understood...") 看起来像是多余的,其实不是。它让模型"认领"了这段总结,后续对话时模型会把总结当作自己已知的上下文,而不是一段需要回应的用户输入。
两层压缩的配合
在 agent 循环里,两层压缩各司其职:
# 用户输入后,检查是否需要全量压缩
if token_count > TOKEN_LIMIT:
messages = compact_context(messages)
# 每次工具调用后,压缩旧的工具结果
messages = compact_tool_result(messages)
compact_tool_result 是持续性的"减肥",每轮都跑,控制工具结果的膨胀速度。compact_context 是偶发性的"大手术",只在 token 数超限时触发,把整个历史压缩成摘要。
大部分时候只有第一层在工作。只有长时间的连续对话才会触发第二层。
压缩的代价
上下文压缩不是免费的。
总结是有损的。模型在总结时会丢弃它认为不重要的细节,但这个判断不一定准确。有时候一个早期的小决策会影响后续的工作,被总结掉之后模型就不知道了。
总结本身要花一次 API 调用。虽然用了 max_tokens=2000 控制成本,但在 token 已经快爆的时候再多一次调用,时机上有点尴尬。
压缩后模型的"记忆"会有断层感。它知道之前做了什么(通过总结),但不记得具体的对话语气和细节。用户可能会觉得 agent 突然变得"生疏"了。
这些都是工程上的取舍。真正的 Claude Code 在这方面做了更多优化——比如分层压缩、选择性保留关键消息、更精确的 token 计算。但核心思路是一样的:在有限的窗口里,尽可能保留最有价值的信息。
回头看看
五篇写下来,我们从零搭建了一个 coding agent 的完整骨架:
- 第一篇:agent 循环和工具调用——让模型能动手
- 第二篇:任务规划——让模型能拆解复杂任务
- 第三篇:技能加载——让模型能按需补充专业知识
- 第四篇:sub-agent——让模型能分派子任务
- 第五篇:上下文压缩——让模型能长时间工作
这五个能力组合在一起,就是一个 Claude Code 的简化版。代码总共不到一千行,但覆盖了 coding agent 最核心的几个机制。