首页 / 文章

Claude Code的上下文与内存内部机制,作为LangGraph智能体重新构建

Claude Code 如何构建 API 调用、运行子智能体并压缩长会话,以及如何在 LangGraph 中借助检查点机制和内存文件重新实现这些功能。

发布者:REACTAPP.TOP阅读约 11 分钟2610 词

来源: Scaling Agent Harnesses for Long-Running Sessions

从泄露的 Claude Code v2.1.87 源代码中可以看出,该工具在长时间的编码过程中是如何处理上下文和内存的。它解决了常见的长时间运行代理所面临的问题:上下文超出模型处理范围、无法预测的工具输出、在恰当时机调出之前会话中的记忆,以及在总结后仍保持项目指令的完整性。下文将分析这一机制,然后使用 LangGraph 及其内置的检查点功能构建一个简化版本。

Claude Code 如何管理上下文

一个API请求由三部分组成。其中两部分——系统提示(身份信息及git上下文)和工具数组(Bash、Read、Edit、Skill、Agent和MCP的JSON架构)——位于消息列表之外,因此压缩操作不会涉及它们。其余部分则作为用户角色条目被封装在<system-reminder>标签中放入消息数组中:最上方是指令文件,中间是当前对话及工具执行结果,而在每轮工具执行之后则会添加附件条目(记忆内容、技能与Agent列表、MCP指令、钩子函数输出)。当窗口被填满后,压缩机制会简化每条消息,并从原始来源重新生成重要的附件。随后是查询循环、子Agent以及压缩处理。

查询循环

输入消息后会启动一个循环,每次迭代都对应一次 API 调用,该循环会持续进行,直到模型在没有调用任何工具的情况下给出答案。

在循环开始之前,后台的 Sonnet 会仅根据用户问题的文本来选取最多五个相关的记忆文件,因为发送整个对话内容会耗费过多资源。这些结果不会立即被使用,而是会在工具执行完成后才被使用,因此每一轮的第一次 API 调用都不会包含它们。

在将数据发送给 API 之前,每次迭代都会对内存中的消息数组进行压缩处理。首先,它会将列表截取到上一次压缩边界之后的内容,而磁盘上的完整 JSONL 文件则不会被修改。接着,它会将任何过大的工具输出结果写入单独的文件,并用 2 kb 的预览内容替代原内容。如果经过这些处理后,token 数量仍超过上下文窗口容量的约 93%,则会触发进一步压缩。

回复可以是纯文本,这样该轮对话即告结束;也可以是一个或多个工具调用,这些调用会被执行后让对话再次进入循环。这些工具可分为几类:通用工具,如Bash、Read、Edit和Grep;Skill工具,它会将完整的SKILL.md文件引入对话中;Agent工具,它会启动一个独立的子代理并返回单一结果;还有ToolSearch工具,用于获取下一次迭代所需的工具架构。每轮工具处理结束后,会为技能列表、代理列表、MCP指令以及所有已完成预取的记忆内容注入增量附件,仅会输出自上一次处理以来发生变动的内容。

子代理

调用 Agent 工具会促使 Claude Code 在同一进程内创建一个隔离的上下文。子代理以父代理对话记录的副本作为起点,获得属于自己的工具和系统提示词,以关闭思考功能的方式运行以降低输出token成本,并独立执行相同的查询循环代码。Exploration 和 Planning 类型的代理也会丢失该副本中的 CLAUDE.md 和 git status 内容,从而进一步节省token。

只有最终结果消息会反馈给父代理。子代理内的工具调用、推理过程以及任何压缩操作都会被记录到侧链转录文件 subagents/agent-{id}.jsonl 中,而父代理永远不会打开该文件。这样一来成本就能得到隔离:数十次内部工具调用最终只会在父代理的上下文中生成一条消息。

压缩

代码中存在一个微型压缩层,用于在完全压缩之前先缩小旧工具的生成结果,但此处该功能已被关闭且不在适用范围内。压缩操作本身有两种策略,第二种策略会在第一种不可用时作为备用方案。第一种策略会立即用准备好的会话内存文件替代内容,因此压缩过程无需单独发起 API 调用;而另一种策略则由后台代理通过多轮独立的调用逐步构建该文件,目前该路径因功能开关被禁用。当前所使用的第二种策略是完整的 LLM 压缩方式:一个分支代理会获取当前的消息以及结构化的压缩提示,然后将整个对话内容压缩至约五千个标记符。最新的消息会以原样形式保留在摘要旁边,以便模型仍能获取即时的工作上下文,而更早的所有内容则会被替换为该摘要。

接下来,原本存储在 messages[] 中并被汇总处理的附件会重新生成。代理列表、延迟执行的工具名称以及MCP指令会再次完整显示,会话启动钩子也会重新运行。之前被调用的技能内容会从容量约为25,000个标记的内存映射中恢复,而最近被修改过的五个文件则会从磁盘重新读取。系统提示、tools[]、Claude.md以及git上下文从来都不在 messages[] 中,因此压缩操作不会影响它们;这些内容的缓存会被直接清除,以便在下一轮处理时获得最新数据。

故意有一件事被省略了:技能目录并未重新构建,从而节省了大约4,000个标记。其副作用是,在压缩之后,模型无法找到那些尚未使用过的技能,除非Claude.md中列出了它们。压缩还会导致细节丢失,因为早期的摘要会与其他内容一起被汇总,经过第三次处理后,原始事实已被压缩了三倍。实际上并没有什么真正丢失,因为磁盘上的只读JSONL文件保存了全部历史记录,而压缩边界标记(parentUuid: null)则指示加载器从何处继续加载。

简化版的LangGraph

使用 LangGraph 可以构建相同架构的简化版本。它已提供用于持久化的检查点机制以及用于删除消息的 RemoveMessage 函数。跨会话内存的实现方式与 Claude Code 相同,即每轮都会读取磁盘上的 Markdown 文件。需要自行编写的是压缩逻辑。

在映射图中,每条箭头都将左侧的 Claude Code 概念与右侧对应的 LangGraph 概念相连。标有“改进点”的边表明 LangGraph 能够将指令保存在 messages[] 之外,因此压缩操作不会影响到这些指令。

该图由五个节点构成。call_model输出的虚线边为条件性路径,用于通过tools返回工具调用结果,再将文本回复发送至post_turn。在load_context阶段,内存文件(.agent/memory/*.md)会从磁盘读取并整合到call_model中的SystemMessage中。这样这些文件就不会出现在messages[]数组中,因而能保持原样不受影响。

图结构构建

这些节点按顺序执行,call_model之后会根据条件分支要么循环使用各种工具,要么进入post_turn阶段。SQLite检查点机制可确保进程重启后所有状态依然完整。

def build_graph():
    builder = StateGraph(AgentState)

    builder.add_node("load_context", load_context)
    builder.add_node("prune_and_compact", prune_and_compact)
    builder.add_node("call_model", call_model)
    builder.add_node("tools", ToolNode(TOOLS))
    builder.add_node("post_turn", post_turn)

    builder.add_edge(START, "load_context")
    builder.add_edge("load_context", "prune_and_compact")
    builder.add_edge("prune_and_compact", "call_model")
    builder.add_conditional_edges(
        "call_model", route_after_model,
        {"tools": "tools", "post_turn": "post_turn"},
    )
    builder.add_edge("tools", "call_model")
    builder.add_edge("post_turn", END)

    checkpointer = SqliteSaver(sqlite3.connect("checkpoints.sqlite"))
    return builder.compile(checkpointer=checkpointer)

# Subagent: separate StateGraph with own call_model + tools loop
subagent_builder = StateGraph(SubagentState)
subagent_builder.add_node("call_model", subagent_call_model)
subagent_builder.add_node("tools", ToolNode([bash, read_file, write_file]))
subagent_builder.add_edge(START, "call_model")
subagent_builder.add_conditional_edges("call_model", subagent_route,
    {"tools": "tools", "__end__": END})
subagent_builder.add_edge("tools", "call_model")
subagent_graph = subagent_builder.compile()

这两个图表之间的关联存在于run_subagent工具中:

@tool
def run_subagent(task: str) -> str:
    """Run an isolated subagent for a specific task."""
    result = subagent_graph.invoke({"messages": [HumanMessage(content=task)]})
    return result["messages"][-1].content  # only final result

一旦模型调用 run_subagent,主 tools 节点就会执行该函数,进而触发子代理图,后者会分别运行自己的 call_model 和 tools 循环。最终只有以字符串形式返回的答案会传回主图,因此主图会将其视为普通工具的输出。子代理没有指令、内存文件或检查点,因此其上下文较为简单,每次调用都是独立的。该示例中只有一种子代理类型。Claude Code 会在运行时从多种类型中选择(用于搜索的 Explore、用于规划的 Plan、拥有所有工具的通用类型,以及像 vercel:ai-architect 这样的插件代理),每种类型都有其自身的工具、权限和系统提示。

提示词

这些是 LangGraph 复制版本中使用的简化版,而 Claude Code 的实际提示词更为详细。

Claude Code自身的精简提示词结构要复杂得多,包含九个部分、一个<analysis>草稿区、一段“CRITICAL: 仅以文本形式回复”的前置说明,以及一个示例格式块。下方的简化版本保留了核心内容:

Your task is to create a detailed summary of the conversation so far,
paying close attention to the user's explicit requests and your previous actions.
This summary should capture technical details, code patterns, and decisions
that would be essential for continuing the work without losing context.

Before writing the summary, analyze the conversation chronologically in <analysis> tags.
Then provide the summary in <summary> tags with these sections:

1. Primary Request and Intent: The user's explicit requests in detail.
2. Key Technical Concepts: Important technologies, frameworks, and patterns discussed.
3. Files and Code Sections: Specific files examined, modified, or created, with paths
   and why each was important.
4. Errors and Fixes: Errors encountered, how they were fixed, and any user corrections
   or feedback (pay special attention to "don't do X" or "actually I meant Y").
5. Problem Solving: Problems solved and any ongoing troubleshooting.
6. Pending Tasks: Tasks explicitly requested but not yet completed.
7. Current Work: What was being worked on immediately before this summary,
   with specific file names and details from the most recent messages.

在代码中,首先会对对话内容进行分割,trim_messages函数用于确定需要保留的最新部分。只有较旧的内容会以该提示发送给模型,而模型会看到所有内容(用户消息、助手回复、工具结果)并对其进行整合。完整的实现方式见下文的压缩部分。

Claude Code中的会话内存更新提示采用包含十个部分的模板,每部分都有字符数限制,需要使用{{variableName}}作为占位符,并且有严格的关于保留各部分标题的规则。简化版本可以概括其核心思路:

Based on the conversation above, update the session notes.

Update each section below with specific, actionable details from the conversation.
Write info-dense content: file paths, function names, error messages, exact commands.
Skip a section if there are no new insights to add. Do not add filler.
Keep each section under 500 words. If approaching the limit, cycle out less important
details while preserving the most critical information.
Always update Current State to reflect the most recent work.

Sections:
# Session Title
# Current State
# Task Specification
# Files and Functions
# Errors and Corrections
# Key Results
# Next Steps

该生成模板包含了工作流、代码库文档、经验总结、关键成果以及工作日志等板块,每个板块的字符限制为2,000个标记,整个文件的字符限制为12,000个标记;同时会通过FileEdit功能让后台代理直接编辑相应板块,而不会重新生成整个文件。

状态:消息之外的指令

在每次API调用时,指令和记忆内容都会从磁盘加载并放入SystemMessage中,就像Claude Code中的prependUserContext()函数所做的那样。这些内容不会保存在messages[]数组中,因此压缩操作会忽略它们。而调用的技能内容及会话记忆则存储在不会被压缩的状态字段中:

class AgentState(MessagesState):
    summary: str                    # compaction summary
    instructions: str               # CLAUDE.md equivalent (never compacted)
    invoked_skills: dict[str, str]  # skill name → content (survives compaction)
    session_memory: str | None      # background summary for instant compaction

# Cross-session memory: .agent/memory/*.md files read from disk every turn.
# Survives compaction (re-read from disk, never in messages[]).
# Agent creates memories via write_file tool. For production at scale,
# use a lightweight LLM call to select relevant files (Claude Code's approach).

压缩处理:总结旧内容,保留最新内容

该设计遵循 Claude Code 的原则。最新消息保持完整,因为模型刚刚处理过这些消息,只有较旧的消息会被总结。

def compact(state: AgentState) -> dict:
    messages = state["messages"]

    # Keep recent messages (Claude Code uses calculateMessagesToKeepIndex)
    kept = trim_messages(
        messages, strategy="last",
        token_counter=count_tokens_approximately,
        max_tokens=COMPACT_THRESHOLD // 5,
        start_on="human", end_on=("human", "tool"),
    )
    kept_ids = {id(m) for m in kept}

    # Summarise only the OLDER messages
    to_summarise = [m for m in messages if id(m) not in kept_ids]
    if not to_summarise:
        return {}

    response = model.invoke(
        to_summarise + [HumanMessage(content=COMPACT_PROMPT)]
    )

    # Delete old, keep recent, inject summary + re-inject skills
    delete = [RemoveMessage(id=m.id) for m in to_summarise]
    summary = HumanMessage(content=f"[Summary]\n{response.content}")
    skill_msgs = [
        HumanMessage(content=f"[Skill: {n}]\n{c[:5000]}")
        for n, c in state.get("invoked_skills", {}).items()
    ]
    return {"messages": [*delete, summary, *skill_msgs], "summary": response.content}

输出顺序为摘要、近期保留的消息,以及重新注入的技能,而非仅用一个摘要替代所有内容。调用RemoveMessage后会生成一个不含旧消息的新检查点,而之前的检查点仍会保留在数据库中以便恢复,这与Claude Code的JSONL格式保留压缩前的记录方式相同。

内存:基于文件,每轮都会读取

该示例模仿了Claude Code的实际实现方式。内存文件存储在磁盘上的.agent/memory/*.md路径下,call_model会在每轮中读取这些文件,且它们永远不会被放入messages[]数组中。由于这些文件是从磁盘重新读取的,而非根据对话内容重新生成,因此压缩操作不会导致它们丢失。

# In call_model — read memory files from disk every turn
MEMORY_DIR = Path(".agent/memory")
if MEMORY_DIR.exists():
    for mem_file in MEMORY_DIR.glob("*.md"):
        content = mem_file.read_text()[:4096]
        system_parts.append(f"Memory ({mem_file.stem}):\n{content}")

# Agent creates memories by writing files:
# write_file(".agent/memory/user_prefs.md", "User prefers concise responses.")

Claude Code 每轮会单独进行一次 Sonnet 调用来选择相关的记忆文件,最多可从众多文件中挑选5个:快速模型会接收问题以及文件名和描述列表,然后返回最匹配的结果。示例中会加载所有文件,虽然简单但无法扩展。拥有大量记忆内容的实际应用环境应遵循原始方案:通过向轻量级 LLM 提供问题与记忆索引来决定加载哪些文件。

Skills:一种采用渐进式展示的工具

@tool
def load_skill(skill_name: str) -> str:
    """Load a specialised skill prompt by name."""
    return SKILL_REGISTRY.get(skill_name, f"Unknown skill: {skill_name}")

在指令文件中列出关键技能,以确保其在数据压缩后仍能保留。技能目录仅为便利工具,并非强制要求。

会话结束后会存储什么

假设用户提出多个问题,触发数据压缩,之后再次返回。此时智能体的存储数据如下所示:

.agent-data/
  checkpoints.sqlite       ~5 MB   (all state snapshots, grows with session)
  tool-results/
    bash-84ffc64a.txt      15 KB   (large bash output persisted with preview)

.agent/
  memory/
    project_goal.md        206 B   (cross-session, survives compaction)
    user_prefs.md          125 B   (read from disk every turn)

对于相同的工作量,Claude Code 能存储更多内容,因为它使用只读追加型的 JSONL 格式记录,这些记录永远不会被删除,并且会为每个子智能体以及压缩备份分别保存文件:

~/.claude/projects/{cwd}/
  {sessionId}.jsonl        22 MB   (full transcript, never deleted)
  {sessionId}/
    tool-results/          12 files
    subagents/             96 files (subagent transcripts + compact backups)
    session-memory/        (disabled)
  history.jsonl            13k lines (prompt recall, UI only)
  memory/                  4 files (auto-memory)

LangGraph的检查点机制使用状态快照来替代事件日志,因此占用空间更小。早期的检查点会保留下来,可在调试或恢复时通过graph.get_state_history(config)进行查看。

副本的功能范围

LangGraph副本负责处理任务的核心部分:从磁盘加载指令,保留最新消息的同时压缩旧消息,将大型工具的运行结果保存到文件中,以及每轮重新读取跨会话内存。生产环境还在其基础上增加了通过Sonnet接口实现的基于大语言模型的内存选择功能、涵盖技能与智能体的增量附件、独立的子智能体、插件钩子以及提示词缓存的处理功能。这样的设计为任何想要构建自有智能体的人提供了实用的参考。