当前位置：首页 > news >正文

上下文撑破之前，Claude Code 如何“清理记忆“——源码精读（二）

news 2026/5/26 19:11:23

上下文撑破之前，Claude Code 如何"清理记忆"——源码精读（二）

核心摘要

一个 Agent 能处理的信息量，根本上受制于上下文窗口。当你要求它分析整个代码仓库——读 50 个文件、跑 30 条命令——上下文轻松突破 100k token，然后就 OOM 了。Claude Code 用三个机制来破这道墙：Subagent（把脏活甩给子进程，父节点保持干净）、Skill 系统（按需加载知识，不预先塞满 prompt）、三层上下文压缩（micro → auto → reactive，激进程度递增）。本文是源码精读第二篇，深入这三个机制的实现细节。

🎯 为什么上下文会"爆"？

先量化一下这个问题。

假设一次代码分析任务： - 读 10 个 Python 文件（每个 100 行） ≈ 40,000 token - 运行 20 条 bash 命令（每条输出 50 行） ≈ 20,000 token - 工具调用结构体本身 ≈ 5,000 token ────────────── 合计 ≈ 65,000 token

这还是单次任务。一个长期运行的 Agent，context 只会越来越大。更麻烦的是：上下文增大不只是花钱的问题，而是质量问题——模型的注意力是有限的，早期读的文件内容会被后来的工具结果覆盖，系统提示的影响力逐渐被稀释。

下面三个机制，分别从不同角度解决这个问题。

🔧 Subagent：用独立上下文隔离"脏活"

问题的本质

父 Agent 问：“这个项目用什么测试框架？”

要回答这个问题，可能需要读requirements.txt、setup.py、pytest.ini、tox.ini……读完之后，这些文件的内容全部永久留在父 Agent 的 messages 里——但父 Agent 只需要一个词的答案：“pytest”。

这就是上下文污染：子任务的中间过程消耗了父 Agent 宝贵的上下文空间。

Subagent 的解法

图4：父 Agent 通过 task 工具派发子任务，子 Agent 以空 messages=[] 启动，可能执行 30+ 次工具调用读取大量文件，但整个 sub_messages 执行完毕后直接丢弃，父 Agent 只收到最终摘要文本，上下文保持干净。

核心代码：

defrun_subagent(prompt:str)->str:# 子 Agent 用全新的空消息列表启动sub_messages=[{"role":"user","content":prompt}]for_inrange(30):# 安全限制：最多 30 轮response=client.messages.create(model=MODEL,system=SUBAGENT_SYSTEM,messages=sub_messages,tools=CHILD_TOOLS,max_tokens=8000,)sub_messages.append({"role":"assistant","content":response.content})ifresponse.stop_reason!="tool_use":break# 执行工具，追加结果...# 子 Agent 可能跑了 30 次工具调用# 整个 sub_messages 直接丢弃# 父 Agent 只收到这段摘要文本return"".join(b.textforbinresponse.contentifhasattr(b,"text"))or"(no summary)"

一个关键设计决策：子 Agent 不能使用task工具（禁止递归生成子 Agent），避免指数级的上下文爆炸。

PARENT_TOOLS=CHILD_TOOLS+[task_tool]# 父 Agent 有 task 工具# 子 Agent 只有 CHILD_TOOLS，不包含 task

这和人类分工一模一样：你委托同事做调研，最后他给你一份一页纸的报告，而不是把所有调研材料都扔给你。

🔧 Skill 系统：按需加载的领域知识

把知识全塞进 System Prompt 的代价

假设你想让 Agent 遵循特定的工作流：git 约定、测试模式、代码审查清单、PDF 处理流程……

如果把所有知识全塞进 system prompt：

10 个 Skill × 每个 2000 token = 20,000 token 固定开销

而且大部分跟当前任务毫无关系。你在处理 git 操作，PDF 处理知识就是在白白占位。

两层注入：廉价索引 + 按需内容

Skill 系统把知识分成两层：

第一层（System Prompt，始终存在）：

You are a coding agent. Skills available: - git: Git workflow helpers ← 每个 Skill ~100 token - test: Testing best practices - pdf: Process PDF files

第二层（模型主动调用load_skill("git")时按需注入）：

<skillname="git">完整 git 工作流说明... ← 2000 token，用时才加载 Step 1: ...</skill>

每个 Skill 是一个目录，包含一个带 YAML frontmatter 的 Markdown 文件：

skills/ git/ SKILL.md # ---\nname: git\ndescription: Git workflow\n---\n... pdf/ SKILL.md code-review/ SKILL.md

SkillLoader 扫描这些文件，自动构建索引：

classSkillLoader:def__init__(self,skills_dir:Path):self.skills={}forfinsorted(skills_dir.rglob("SKILL.md")):text=f.read_text()meta,body=self._parse_frontmatter(text)name=meta.get("name",f.parent.name)self.skills[name]={"meta":meta,"body":body}defget_content(self,name:str)->str:skill=self.skills.get(name)returnf'<skill name="{name}">\n{skill["body"]}\n</skill>'

注册为普通工具，加载逻辑不过是一次字典查找：

TOOL_HANDLERS={# ...其他工具..."load_skill":lambda**kw:SKILL_LOADER.get_content(kw["name"]),}

模型知道有哪些 Skill（廉价），需要时再加载完整内容（贵）。这是"延迟加载"思想在 Agent 设计中的直接应用。

🔧 三层上下文压缩：从微手术到全身麻醉

这是 Claude Code 中最精妙的工程设计之一。

图5：三层压缩的触发条件和激进程度依次递增。第一层静默替换旧工具结果（无信息损失）；第二层在 token 超阈值时用 LLM 做摘要（有信息损失但磁盘有完整备份）；第三层是 API 报错时的应急响应。三层配合实现"无限会话"。

第一层：micro_compact——悄无声息地清理旧垃圾

最轻量的一层。工具结果用完之后，它的历史价值其实已经不大了——"10 分钟前读的文件内容"大概率已经被处理完。

defmicro_compact(messages:list)->list:# 找到所有 tool_resulttool_results=[]fori,msginenumerate(messages):ifmsg["role"]=="user"andisinstance(msg.get("content"),list):forj,partinenumerate(msg["content"]):ifisinstance(part,dict)andpart.get("type")=="tool_result":tool_results.append((i,j,part))# 只保留最近 N 个，旧的替换为占位符iflen(tool_results)<=KEEP_RECENT:returnmessagesfor_,_,partintool_results[:-KEEP_RECENT]:iflen(part.get("content",""))>100:part["content"]="[Old tool result content cleared]"returnmessages

模型不知道发生了什么。它只是发现一些旧的工具结果变成了占位符——这是可接受的，因为那些内容它已经处理过了。

Anthropic 版 Claude Code 还有一个更精妙的变体：直接用 cache editing API 在 API 层删除，不修改本地消息，连模型看到的历史都不用动。

第二层：auto_compact——LLM 给自己写会议纪要

当 token 数超过阈值（context_window - 13,000 buffer），就要更激进的手段：

defauto_compact(messages:list)->list:# 1. 先把完整历史存磁盘（安全网）transcript_path=TRANSCRIPT_DIR/f"transcript_{int(time.time())}.jsonl"withopen(transcript_path,"w")asf:formsginmessages:f.write(json.dumps(msg,default=str)+"\n")# 2. 用 LLM 做摘要response=client.messages.create(model=MODEL,messages=[{"role":"user","content":"Summarize this conversation for continuity..."+json.dumps(messages,default=str)[:80000]}],max_tokens=2000,)# 3. 用一条摘要消息替换所有历史return[{"role":"user","content":f"[Compressed]\n\n{response.content[0].text}"},]