目录

一、问题背景：一个跑了 3 小时就崩溃的 Agent

二、环境说明与前置依赖

三、根因分析：上下文是怎么被"撑爆"的

3.1 三条"账单"——谁在吃 Token

3.2 三个核心根因

四、方案设计：三层记忆 + 主动压缩 + 工具门控

五、实现一：Checkpoint 持久化——让 Agent 能断点续跑

5.1 LangGraph Checkpoint 的工作原理

5.2 生产环境配置（PostgreSQL）

5.3 为什么用 DeltaChannel 而不是全量快照

六、实现二：分层记忆架构——热/温/冷三层管理

6.1 为什么分三层

6.2 LangMem 实现三温层

6.3 在 Agent 节点中集成记忆

七、实现三：上下文压缩策略——Token 消费直降 80%

7.1 三条压缩策略

7.2 实现代码

7.3 效果验证

八、实现四：渐进式工具加载——告别 200K Token 的 Tool Schema

8.1 问题

8.2 方案：Deferred Tool Registry

8.3 收益

九、性能对比：优化前后的硬数据

十、生产环境注意事项与风险提示

十一、总结与完整代码仓库

核心要点

适用边界

参考资源

一、问题背景：一个跑了 3 小时就崩溃的 Agent

2026 年初，我们团队上线了一个内部用的代码审查 Agent。它的工作流程不复杂：拉取 PR、逐文件分析代码质量、生成审查报告、提交评论。

Demo 跑得顺风顺水。上线第一天就翻车了。

这个 Agent 跑的是 ReAct 模式（Reason → Act → Observe → 循环），每次工具调用都会把工具的输入和输出原封不动塞进上下文。一个普通 PR 大概触发 40-60 次工具调用——读文件、跑 lint、查数据库、生成建议。跑到第 3 个小时，上下文窗口里塞了满打满算的180K Token，然后就开始出各种怪事：

• 模型开始"忘记"一小时前分析过的文件，反复重新读取
• 生成的审查建议越来越短，从 200 字缩水到 20 字
• 偶尔会在两个工具调用之间原地死循环——不断重新请求同一个文件
• 最离谱的一次，它对着一个已经分析完的文件又生成了 3 份重复报告

这个问题在业内有个名字：上下文腐烂（Context Rot）。Transformer 的注意力机制会产生 \( O(n^2) \) 的配对关系——上下文翻倍，模型需要解析的关联量变四倍。这不是换个更大的模型窗口就能解决的事。

关键数据：据 Factory AI 对 36000 条真实工程会话的评估，Agent 任务执行的前 30% 步骤只消耗约 20% 的 Token，但最后 30% 的步骤消耗了将近50%的 Token——因为每一步都要背负之前所有步骤的上下文。

二、环境说明与前置依赖

# 安装依赖 pip install "langgraph>=1.2" \ "langgraph-checkpoint-postgres>=2.0" \ "langmem>=0.1" \ "anthropic>=0.54" \ "psycopg[binary,pool]>=3.2" \ "pgvector>=0.3"

三、根因分析：上下文是怎么被"撑爆"的

3.1 三条"账单"——谁在吃 Token

我们给 Agent 加了 Token 埋点，跑了一个 60 步的典型任务，结果很扎眼：

画成架构图更直观：

3.2 三个核心根因

根因一：工具 Schema 膨胀。Agent 能力越强，注册的工具越多。28 个工具的 JSON Schema 定义（包含参数、类型、描述）就占了 144K Token。这是固定开销，不管任务需不需要这些工具。

根因二：工具输出"僵尸数据"。文件读完之后，8000 Token 的文件内容就不再需要了——但 ReAct 循环里它不会被自动清除。到第 60 步时，前 40 步的工具输出基本是"僵尸 Token"，只占空间不提供价值。

根因三："Lost in the Middle"效应。这是 Transformer 注意力机制的已知问题——模型对上下文窗口中间的 Token 召回率显著低于开头和结尾。早期步骤的推理结论，在长上下文中基本等于丢了。

四、方案设计：三层记忆 + 主动压缩 + 工具门控

针对上面三个根因，我们设计了一套组合方案：

下面逐个实现。

五、实现一：Checkpoint 持久化——让 Agent 能断点续跑

这是整个方案的地基。没有 Checkpoint，Agent 崩溃后从零开始，之前消耗的 Token 全白费。

5.1 LangGraph Checkpoint 的工作原理

LangGraph 在 Agent 执行的每个步骤（node）之后自动保存一个状态快照（checkpoint）。通过thread_id区分不同会话。

从 v1.2 开始，LangGraph 引入了DeltaChannel机制——不再每个步骤都保存完整状态快照，而是只保存增量（delta）。对长任务 Agent 来说，这是质的飞跃。

5.2 生产环境配置（PostgreSQL）

"""checkpoint_setup.py —— 生产级 Checkpoint 配置""" import psycopg from langgraph.checkpoint.postgres import PostgresSaver from langgraph.graph import StateGraph, MessagesState, START, END from langchain_anthropic import ChatAnthropic # ===== 1. 建立数据库连接 ===== DB_URI = ( "postgresql://agent:secure_pass@localhost:5432/agent_db" "?sslmode=require" "&application_name=code_review_agent" ) conn = psycopg.connect(DB_URI) # ===== 2. 初始化 Checkpoint 表（仅首次） ===== checkpointer = PostgresSaver(conn) checkpointer.setup() # 自动创建 checkpoints 和 writes 表 # ===== 3. 编译 Graph ===== llm = ChatAnthropic(model="claude-sonnet-4-20250514") def call_model(state: MessagesState): response = llm.invoke(state["messages"]) return {"messages": [response]} builder = StateGraph(MessagesState) builder.add_node("model", call_model) builder.add_edge(START, "model") builder.add_edge("model", END) graph = builder.compile(checkpointer=checkpointer) # ===== 4. 使用 thread_id 管理会话 ===== config = {"configurable": {"thread_id": "pr-review-1842"}} # 首轮调用 result = graph.invoke( {"messages": [{"role": "user", "content": "请审查 PR #1842"}]}, config=config ) # Agent 崩溃或超时后，同 thread_id 自动恢复 result = graph.invoke( {"messages": [{"role": "user", "content": "继续上次的审查"}]}, config=config # 同一个 thread_id，自动加载上次 checkpoint )

5.3 为什么用 DeltaChannel 而不是全量快照

默认的全量快照模式下，检查点存储增长是 \( O(n^2) \)——一个 200 步的 coding agent，序列化到检查点的数据量达到5.3 GB。DeltaChannel 把这一步降到129 MB，降幅超过 40 倍。

六、实现二：分层记忆架构——热/温/冷三层管理

6.1 为什么分三层

不分层的记忆有一个致命问题：要么信息太多模型处理不过来，要么信息太少模型丢了关键上下文。三层结构的核心思想是按"访问热度"梯度存储，在 Token 预算和召回精度之间取得平衡。

6.2 LangMem 实现三温层

LangMem 是 LangChain 在 2026 年 6 月发布的记忆管理库，它把记忆的"存储-检索-更新"封装成了标准化的 Manager。下面是生产配置：

"""memory_layers.py —— 三层记忆架构的 LangMem 实现""" from langmem import ( create_memory_store_manager, create_prompt_optimizer, ) from langgraph.store.postgres import PostgresStore from langgraph.checkpoint.postgres import PostgresSaver import psycopg # ===== 数据库连接 ===== DB_URI = "postgresql://agent:secure_pass@localhost:5432/agent_db?sslmode=require" conn = psycopg.connect(DB_URI) # ===== 初始化 Store（长期记忆）和 Checkpointer（短期记忆） ===== store = PostgresStore(conn) store.setup() # 创建 store 表结构 checkpointer = PostgresSaver(conn) # ===== 记忆管理器（自动管理温层和冷层） ===== memory_manager = create_memory_store_manager( model="claude-sonnet-4-20250514", namespace=("code_review_agent", "{thread_id}"), schemas=[ { "name": "code_analysis_result", "description": "对单个文件的分析结论，包含评估等级、发现的问题和建议", "fields": { "file_path": "文件路径", "risk_level": "风险等级: low/medium/high/critical", "issues_found": "发现的具体问题列表", "key_suggestion": "核心改进建议", "analysis_round": "第几轮分析" } }, { "name": "user_preference", "description": "用户的审查偏好和风格要求", "fields": { "preference_type": "偏好类型", "description": "具体偏好描述" } } ], instructions=( "从代码审查对话中提取关键信息。" "对于已分析完的文件，只保留结论，删除原始代码内容。" "对于用户反馈，提取可跨会话复用的偏好设置。" "相同文件的新分析结果应覆盖旧结果。" ), ) # ===== Prompt 优化器（热层：最近上下文自动注入） ===== prompt_optimizer = create_prompt_optimizer( model="claude-sonnet-4-20250514", kind="gradient", # 梯度优化模式：少量示例 → 持续改进 max_recent_messages=10, # 热层：保留最近 10 轮完整对话 )

6.3 在 Agent 节点中集成记忆

"""agent_with_memory.py —— 在 Agent Graph 节点中使用记忆""" from dataclasses import dataclass from langgraph.graph import StateGraph, MessagesState, START from langgraph.runtime import Runtime from langchain_anthropic import ChatAnthropic import uuid @dataclass class AgentContext: pr_id: str thread_id: str llm = ChatAnthropic(model="claude-sonnet-4-20250514") async def analyze_file_node( state: MessagesState, runtime: Runtime[AgentContext], memory_manager, ): """分析文件的 Agent 节点，每次执行前检索相关记忆""" pr_id = runtime.context.pr_id user_message = state["messages"][-1].content # ----- Step 1: 检索温层 + 冷层记忆 ----- # 查找之前分析过的同一 PR 下文件 memories = await memory_manager.store.asearch( ("code_review_agent", pr_id), query=user_message, limit=5, filter={"schema_name": "code_analysis_result"} ) # 查找用户偏好（跨会话） preferences = await memory_manager.store.asearch( ("code_review_agent", "preferences"), query="code review style requirements", limit=3, filter={"schema_name": "user_preference"} ) # ----- Step 2: 构建上下文注入 ----- memory_context_parts = [] if memories: analyzed_files = [m.value for m in memories] memory_context_parts.append( "## 之前已分析的文件（只包含结论，不含原始代码）：\n" + "\n".join([ f"- {f['file_path']}: 风险={f['risk_level']}, " f"建议={f['key_suggestion']}" for f in analyzed_files ]) ) if preferences: memory_context_parts.append( "## 用户审查偏好：\n" + "\n".join([f"- {p.value['description']}" for p in preferences]) ) memory_context = "\n\n".join(memory_context_parts) # ----- Step 3: 调用 LLM（带记忆上下文） ----- system_msg = ( "你是一个代码审查 Agent。" "当前上下文包含了之前分析过的文件摘要和用户偏好。\n\n" + memory_context ) response = await llm.ainvoke([ {"role": "system", "content": system_msg}, *state["messages"] ]) # ----- Step 4: 写入新的分析结果到记忆 ----- file_path = extract_file_path(user_message) risk = assess_risk_level(response.content) await memory_manager.store.aput( ("code_review_agent", pr_id), str(uuid.uuid4()), { "schema_name": "code_analysis_result", "file_path": file_path, "risk_level": risk, "issues_found": extract_issues(response.content), "key_suggestion": extract_suggestions(response.content), "analysis_round": count_existing_memories(memories) + 1, } ) return {"messages": [response]}

⚠️ 注意：不要为每一次 LLM 调用都写入记忆。只写入有跨步复用价值的结论。一个简单判断标准：这条信息在 5 步之后还会被用到吗？如果不会，就不要写。

七、实现三：上下文压缩策略——Token 消费直降 80%

7.1 三条压缩策略

我们落地的压缩策略是"先硬件，再软件"——先做不依赖 LLM 的零成本清理，再做LLM 驱动的摘要压缩：

1. 工具输出截断（零成本）——工具返回后立即清除原始输出，只保留结构化结论
2. 观察屏蔽（Observation Masking）——JetBrains 2025 年的研究表明，把旧工具输出替换为结构化占位符，可以实现52% 的 Token 成本降低，且不损失任务完成率
3. 锚定迭代摘要（LLM 驱动）——接近窗口阈值时触发压缩，将对话历史压缩为结构化摘要，采用增量合并而非全量重建

7.2 实现代码

"""context_compressor.py —— 上下文压缩工具""" from typing import TypedDict, Sequence from langgraph.graph import StateGraph, MessagesState from langgraph.prebuilt import ToolNode from langgraph.checkpoint.memory import MemorySaver import json # ===== 策略一：工具输出自动截断 ===== TOOL_OUTPUT_MAX_TOKENS = 800 # 每个工具输出上限 class TruncatedToolNode(ToolNode): """包装 ToolNode，自动截断过长输出""" def _truncate_output(self, content: str, tool_name: str) -> str: estimated_tokens = len(content) // 4 # 粗略估算 if estimated_tokens <= TOOL_OUTPUT_MAX_TOKENS: return content # 保留头部和尾部，中间截断 head_chars = int(TOOL_OUTPUT_MAX_TOKENS * 2.5) tail_chars = int(TOOL_OUTPUT_MAX_TOKENS * 1.5) return ( content[:head_chars] + f"\n\n[... 中间 {estimated_tokens - TOOL_OUTPUT_MAX_TOKENS} tokens 已截断 ...]\n\n" + content[-tail_chars:] + f"\n\n[截断标记] {tool_name} 原始输出 {estimated_tokens} tokens → " + f"保留 {TOOL_OUTPUT_MAX_TOKENS} tokens" ) def _run_one(self, *args, **kwargs): result = super()._run_one(*args, **kwargs) if hasattr(result, 'content'): result.content = self._truncate_output( result.content, kwargs.get('name', 'unknown_tool') ) return result # ===== 策略二：观察屏蔽（Observation Masking） ===== OBSERVATION_MASK_THRESHOLD = 5 # 超过 5 轮的工具输出开始屏蔽 class ObservationMaskingState(TypedDict): messages: Sequence[dict] tool_outputs: list[dict] # 记录每条工具输出的元信息 masked_count: int def apply_observation_masking(state: ObservationMaskingState) -> dict: """将旧工具输出替换为结构化占位符""" messages = list(state["messages"]) tool_outputs = state.get("tool_outputs", []) masked = 0 for i, msg in enumerate(messages): if msg.get("role") != "tool": continue # 判断这个工具输出是否足够"旧" distance_from_end = len(messages) - i - 1 if distance_from_end <= OBSERVATION_MASK_THRESHOLD: continue # 记录元信息并替换为占位符 tool_name = msg.get("name", "unknown_tool") original_len = len(msg.get("content", "")) tool_outputs.append({ "step": i, "tool": tool_name, "original_tokens": original_len // 4, "masked_at_step": len(messages), }) messages[i] = { **msg, "content": json.dumps({ "_masked": True, "tool": tool_name, "summary": f"[第 {i} 步的输出，{original_len} 字符，" f"已在第 {len(messages)} 步被屏蔽]", "_retrieve_key": f"step_{i}_{tool_name}" }, ensure_ascii=False), } masked += 1 return { "messages": messages, "tool_outputs": tool_outputs, "masked_count": state.get("masked_count", 0) + masked, } # ===== 策略三：锚定迭代摘要 ===== from langchain_core.messages import SystemMessage, HumanMessage from langchain_anthropic import ChatAnthropic SUMMARY_TRIGGER_RATIO = 0.70 # 上下文使用 70% 时触发压缩 class ContextSummaryState(TypedDict): messages: Sequence[dict] persistent_summary: str # 锚定的持久摘要，增量更新 summary_generation_count: int summary_llm = ChatAnthropic(model="claude-haiku-4-5-20251001") # 用轻量模型做摘要 async def anchored_incremental_summary(state: ContextSummaryState) -> dict: """锚定迭代摘要：增量合并新内容到持久摘要""" existing_summary = state.get("persistent_summary", "") messages = state["messages"] # 估算 Token 使用量 total_tokens = sum(len(str(m.get("content", ""))) for m in messages) // 4 max_context = 200000 # Claude Sonnet 4 的窗口 if total_tokens < max_context * SUMMARY_TRIGGER_RATIO: return {} # 还没到阈值，不触发压缩 # 只取新增的消息（上一份摘要覆盖范围之后的） last_summarized_index = state.get("summary_generation_count", 0) * 20 new_messages = messages[last_summarized_index:] new_content = "\n".join([ f"[{m.get('role', 'unknown')}]: {str(m.get('content', ''))[:500]}" for m in new_messages ]) # 锚定合并：不是全量重建，而是把新内容合并到已有摘要 merge_prompt = f"""你是上下文摘要引擎。将新的 Agent 交互内容合并到已有摘要中。 已有摘要： {existing_summary if existing_summary else "（尚无摘要）"} 新的交互内容： {new_content} 请输出合并后的摘要，格式： ## 会话目标 [一句话描述任务目标] ## 已完成步骤 - [步骤1描述] - [步骤2描述] ## 关键发现/结论 - [重要发现] ## 当前状态 [当前进度和下一步] ## 待处理事项 - [待办1]""" response = await summary_llm.ainvoke([HumanMessage(content=merge_prompt)]) return { "persistent_summary": response.content, "summary_generation_count": state.get("summary_generation_count", 0) + 1, }

7.3 效果验证

八、实现四：渐进式工具加载——告别 200K Token 的 Tool Schema

8.1 问题

28 个工具的 JSON Schema 占了 144K Token——相当于上下文的 72%。更麻烦的是，大量无关的工具定义会触发Lost in the Middle效应，模型在选工具时反而更容易出错。

8.2 方案：Deferred Tool Registry

"""deferred_tools.py —— 渐进式工具加载""" from typing import Callable, Optional class DeferredToolRegistry: """ 渐进式工具注册表。 工作流程： 1. 启动时只注册 5 个核心工具（完整 Schema） 2. 长尾工具只保留极简描述（name + 一句话说明） 3. 模型需要时，通过 tool_search 工具按需唤醒 """ def __init__(self): self._active_tools: dict[str, Callable] = {} # 当前激活的工具 self._deferred_tools: dict[str, dict] = {} # 延迟加载的工具 self._tool_index: dict[str, str] = {} # 精简描述索引 def register_active(self, name: str, tool_fn: Callable): """注册为启动时即激活的核心工具""" self._active_tools[name] = tool_fn def register_deferred(self, name: str, tool_fn: Callable, short_desc: str): """注册为延迟加载的长尾工具（只记录极简描述）""" self._deferred_tools[name] = tool_fn self._tool_index[name] = short_desc def activate(self, name: str) -> Optional[Callable]: """按需激活一个延迟工具""" if name in self._active_tools: return self._active_tools[name] if name in self._deferred_tools: print(f"[Tool Registry] 唤醒长尾工具: {name}") self._active_tools[name] = self._deferred_tools.pop(name) return self._active_tools[name] return None def get_tool_index_prompt(self) -> str: """生成精简的工具索引（约 600 tokens，而非 144K）""" lines = ["## 可用工具索引（使用 tool_search 唤醒长尾工具）："] for name in self._active_tools: lines.append(f"- {name}: 已激活，可直接调用") for name, desc in self._tool_index.items(): lines.append(f"- {name}: {desc} (使用 tool_search 唤醒)") return "\n".join(lines) # ===== 使用示例 ===== registry = DeferredToolRegistry() # 核心工具：启动时即激活（约 5 个，8K Token） registry.register_active("read_file", read_file_fn) registry.register_active("analyze_code", analyze_code_fn) registry.register_active("post_review", post_review_fn) registry.register_active("tool_search", tool_search_fn) # 工具本身 registry.register_active("get_checkpoint", get_checkpoint_fn) # 长尾工具：延迟加载（保留极简描述） registry.register_deferred( "query_codebase", query_codebase_fn, "在代码库中搜索特定模式或调用关系" ) registry.register_deferred( "run_security_scan", run_security_scan_fn, "对指定文件执行安全漏洞扫描" ) registry.register_deferred( "generate_test", generate_test_fn, "为指定函数生成单元测试代码" ) # ... 其余 20 个长尾工具类似 # 实现 tool_search 工具——模型调用它来唤醒长尾工具 def tool_search_fn(tool_description: str) -> str: """ 模型按需搜索工具时触发。 输入：对所需工具的描述 输出：匹配到的工具名称和完整 Schema """ # 简化版：关键词匹配。生产环境可以用语义检索 for name, desc in registry._tool_index.items(): if any(kw in tool_description.lower() for kw in desc.lower().split()): registry.activate(name) return f"工具 '{name}' 已激活: {desc}" return f"未找到匹配 '{tool_description}' 的工具，" + \ f"当前可唤醒工具: {list(registry._tool_index.keys())}"

8.3 收益

这套方案在易点天下的 K8s 运维 Agent 中验证过：工具 Schema Token 从144K → ~8K，同时工具调用准确率从70% → 90%——因为模型面对的工具选择更聚焦了。

九、性能对比：优化前后的硬数据

我们用同一个 PR（涉及 23 个文件变更，约 4500 行 diff）跑了三次实验，每次从零开始：

数据说明：实验环境为 AWS m6i.xlarge（4 vCPU, 16GB RAM），PostgreSQL 16（RDS db.t3.medium，100GB SSD）。成本基于 Claude Sonnet 4 API 定价（$3/$15 per 1M input/output tokens）。压测为 32 小时连续运行，处理 50 个随机 PR。

十、生产环境注意事项与风险提示

⚠️ 风险一：压缩摘要的信息丢失
锚定迭代摘要虽然比全量重建可靠（Factory 评估显示在准确性、完整性和任务连续性三个维度上持续优于全量重建），但它仍然是有损压缩。建议：（1）为关键操作保留独立的结构化日志（如"任务完成了哪些步骤"），不依赖摘要来重建；（2）摘要中使用明确的结构化字段（目标、进度、产出物、下一步），而非自由文本。

⚠️ 风险二：上下文污染的连锁效应
如果摘要中混入了错误信息（模型幻觉生成的内容），后续步骤会基于错误信息推理。建议：对每次摘要注入做新鲜度检查——确认摘要中的事实在最新的对话中仍然有效。

⚠️ 风险三：DeltaChannel 的生产限制
DeltaChannel 在 langgraph 1.2 中仍处于 beta 阶段。如果你的 Graph 节点中有非累加类型的 state 字段（如一个会被覆盖的配置值），DeltaChannel 的语义可能与全量快照不同。建议先在 staging 环境充分测试，确认 state 重建一致性。

⚠️ 风险四：渐进式工具加载的覆盖盲区
tool_search 的关键词匹配在生产环境中可能漏掉模型真正需要的工具。建议在日志中记录每次 tool_search 的查询词和匹配结果，定期审计"模型搜了但没找到"的模式，补充索引。

十一、总结与完整代码仓库

核心要点

1. 上下文腐烂是 Agent 生产化的第一杀手——不是模型不够聪明，是它的工作记忆被垃圾信息淹没了。
2. Checkpoint 是地基。没有它，Agent 崩溃后所有 Token 白费。用 DeltaChannel 可以把持久化成本从 O(n²) 降到接近 O(n)。
3. 三层记忆 + 上下文压缩 + 工具门控，三条腿一起走才有显著效果。单独用任何一条都能省 30-50% Token，但组合之后才能达到 80%。
4. 压缩先硬件后软件——先做不依赖 LLM 的工具输出截断和观察屏蔽（零成本），再做 LLM 驱动的摘要压缩（有成本但更精准）。
5. 越压缩越要审。压缩后的上下文是"二手信息"，需要额外的校验机制防止污染传播。

适用边界

本文方案适用于：

• 适用 运行超过 30 步的 Agent（代码审查、长文档分析、自动化测试）
• 适用 工具数量超过 10 个的 Agent
• 适用 使用 LangGraph + Claude/GPT 的 Agent 架构
• 部分适用 短任务 Agent（< 10 步）——只需 Checkpoint，不需要全套压缩
• 不适用 需要保留所有工具输出精确内容的任务（如全文翻译）——观察屏蔽会破坏任务语义

参考资源

• LangGraph Memory 官方文档：https://docs.langchain.com/oss/python/langgraph/add-memory
• LangGraph DeltaChannel 设计文档：https://blog.langchain.com/delta-channels-evolving-agent-runtime
• LangMem GitHub 仓库：https://github.com/langchain-ai/langmem
• JetBrains Observation Masking 论文（2025.12）
• Factory AI / ZenML 上下文压缩评估（2026.03）
• Zylos Research: Context Engineering for Long-Running Agents（2026.06）

如有疑问欢迎在评论区交流。这套方案在我们的生产环境跑了两个月，不是纸上谈兵。如果你在落地过程中遇到具体问题，评论区描述你的场景，我会尽量回复。

—— 一个持续跟上下文腐烂斗争的 Agent 工程师