
目录一、问题背景一个跑了 3 小时就崩溃的 Agent二、环境说明与前置依赖三、根因分析上下文是怎么被撑爆的3.1 三条账单——谁在吃 Token3.2 三个核心根因四、方案设计三层记忆 主动压缩 工具门控五、实现一Checkpoint 持久化——让 Agent 能断点续跑5.1 LangGraph Checkpoint 的工作原理5.2 生产环境配置PostgreSQL5.3 为什么用 DeltaChannel 而不是全量快照六、实现二分层记忆架构——热/温/冷三层管理6.1 为什么分三层6.2 LangMem 实现三温层6.3 在 Agent 节点中集成记忆七、实现三上下文压缩策略——Token 消费直降 80%7.1 三条压缩策略7.2 实现代码7.3 效果验证八、实现四渐进式工具加载——告别 200K Token 的 Tool Schema8.1 问题8.2 方案Deferred Tool Registry8.3 收益九、性能对比优化前后的硬数据十、生产环境注意事项与风险提示十一、总结与完整代码仓库核心要点适用边界参考资源一、问题背景一个跑了 3 小时就崩溃的 Agent2026 年初我们团队上线了一个内部用的代码审查 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——因为每一步都要背负之前所有步骤的上下文。二、环境说明与前置依赖组件版本说明Python3.12推荐 3.11langgraph1.2.0含 DeltaChannel 支持v1.2 新增langgraph-checkpoint-postgres2.0.17生产环境 Checkpoint 后端langmem0.1.0语义记忆管理LangChain 出品anthropic0.54.0Claude API SDKPostgreSQL16需要 pgvector 扩展语义检索模型Claude Sonnet 4200K 上下文窗口# 安装依赖 pip install langgraph1.2 \ langgraph-checkpoint-postgres2.0 \ langmem0.1 \ anthropic0.54 \ psycopg[binary,pool]3.2 \ pgvector0.3三、根因分析上下文是怎么被撑爆的3.1 三条账单——谁在吃 Token我们给 Agent 加了 Token 埋点跑了一个 60 步的典型任务结果很扎眼Token 来源消耗占比问题本质MCP 工具 Schema 定义72% (144K)28 个工具的全量 JSON Schema 一次性注入工具调用原始输出18% (36K)API 返回、文件内容全部堆积在上下文中推理链 对话历史7% (14K)只有这部分是有效上下文System Prompt3% (6K)固定开销基本可控画成架构图更直观3.2 三个核心根因根因一工具 Schema 膨胀。Agent 能力越强注册的工具越多。28 个工具的 JSON Schema 定义包含参数、类型、描述就占了 144K Token。这是固定开销不管任务需不需要这些工具。根因二工具输出僵尸数据。文件读完之后8000 Token 的文件内容就不再需要了——但 ReAct 循环里它不会被自动清除。到第 60 步时前 40 步的工具输出基本是僵尸 Token只占空间不提供价值。根因三Lost in the Middle效应。这是 Transformer 注意力机制的已知问题——模型对上下文窗口中间的 Token 召回率显著低于开头和结尾。早期步骤的推理结论在长上下文中基本等于丢了。四、方案设计三层记忆 主动压缩 工具门控针对上面三个根因我们设计了一套组合方案下面逐个实现。五、实现一Checkpoint 持久化——让 Agent 能断点续跑这是整个方案的地基。没有 CheckpointAgent 崩溃后从零开始之前消耗的 Token 全白费。5.1 LangGraph Checkpoint 的工作原理LangGraph 在 Agent 执行的每个步骤node之后自动保存一个状态快照checkpoint。通过thread_id区分不同会话。从 v1.2 开始LangGraph 引入了DeltaChannel机制——不再每个步骤都保存完整状态快照而是只保存增量delta。对长任务 Agent 来说这是质的飞跃。5.2 生产环境配置PostgreSQLcheckpoint_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_passlocalhost:5432/agent_db ?sslmoderequire application_namecode_review_agent ) conn psycopg.connect(DB_URI) # 2. 初始化 Checkpoint 表仅首次 checkpointer PostgresSaver(conn) checkpointer.setup() # 自动创建 checkpoints 和 writes 表 # 3. 编译 Graph llm ChatAnthropic(modelclaude-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(checkpointercheckpointer) # 4. 使用 thread_id 管理会话 config {configurable: {thread_id: pr-review-1842}} # 首轮调用 result graph.invoke( {messages: [{role: user, content: 请审查 PR #1842}]}, configconfig ) # Agent 崩溃或超时后同 thread_id 自动恢复 result graph.invoke( {messages: [{role: user, content: 继续上次的审查}]}, configconfig # 同一个 thread_id自动加载上次 checkpoint )5.3 为什么用 DeltaChannel 而不是全量快照默认的全量快照模式下检查点存储增长是 \( O(n^2) \)——一个 200 步的 coding agent序列化到检查点的数据量达到5.3 GB。DeltaChannel 把这一步降到129 MB降幅超过 40 倍。模式200 步存储量恢复延迟适用场景全量快照5.3 GB~300ms短任务 ( 20 步)DeltaChannel129 MB~80ms (K50)长任务 (50 步)六、实现二分层记忆架构——热/温/冷三层管理6.1 为什么分三层不分层的记忆有一个致命问题要么信息太多模型处理不过来要么信息太少模型丢了关键上下文。三层结构的核心思想是按访问热度梯度存储在 Token 预算和召回精度之间取得平衡。层级范围存储形式Token 预算检索精度 热层最近 10 轮对话完整原始内容~8K100%在窗口中 温层第 11–40 轮滚动结构化摘要~4K~85%锚定合并❄️ 冷层40 轮之前 跨会话语义索引 向量检索~3K (注入)~75%语义召回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_passlocalhost:5432/agent_db?sslmoderequire conn psycopg.connect(DB_URI) # 初始化 Store长期记忆和 Checkpointer短期记忆 store PostgresStore(conn) store.setup() # 创建 store 表结构 checkpointer PostgresSaver(conn) # 记忆管理器自动管理温层和冷层 memory_manager create_memory_store_manager( modelclaude-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( modelclaude-sonnet-4-20250514, kindgradient, # 梯度优化模式少量示例 → 持续改进 max_recent_messages10, # 热层保留最近 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(modelclaude-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), queryuser_message, limit5, filter{schema_name: code_analysis_result} ) # 查找用户偏好跨会话 preferences await memory_manager.store.asearch( (code_review_agent, preferences), querycode review style requirements, limit3, 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 驱动的摘要压缩工具输出截断零成本——工具返回后立即清除原始输出只保留结构化结论观察屏蔽Observation Masking——JetBrains 2025 年的研究表明把旧工具输出替换为结构化占位符可以实现52% 的 Token 成本降低且不损失任务完成率锚定迭代摘要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: fstep_{i}_{tool_name} }, ensure_asciiFalse), } 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(modelclaude-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(contentmerge_prompt)]) return { persistent_summary: response.content, summary_generation_count: state.get(summary_generation_count, 0) 1, }7.3 效果验证策略Token 节省实现成本副作用风险工具输出截断~30%极低代码层面低丢失细节但保留结论观察屏蔽~52%低结构化替换中需要按需检索回原始输出锚定迭代摘要~60%中额外 LLM 调用中摘要可能丢失关键信息三条策略叠加~80%中可控分层回退机制兜底八、实现四渐进式工具加载——告别 200K Token 的 Tool Schema8.1 问题28 个工具的 JSON Schema 占了 144K Token——相当于上下文的 72%。更麻烦的是大量无关的工具定义会触发Lost in the Middle效应模型在选工具时反而更容易出错。8.2 方案Deferred Tool Registrydeferred_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跑了三次实验每次从零开始指标优化前基线优化后变化最长连续运行时间3h 12m崩溃8h32h 压测无崩溃167%单任务平均 Token 消耗380K72K-81%工具调用准确率70%91%21pp重复读取文件次数平均 8.3 次/文件平均 1.1 次/文件-87%上下文窗口使用率均值86% (172K/200K)34% (68K/200K)-52pp单任务 API 成本~$14.20~$2.85-80%崩溃恢复时间重新开始3h 浪费恢复上次 checkpoint 5s从小时级 → 秒级审查报告完整性评分62/100后半段质量差91/100全程一致29数据说明实验环境为 AWS m6i.xlarge4 vCPU, 16GB RAMPostgreSQL 16RDS db.t3.medium100GB 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 的查询词和匹配结果定期审计模型搜了但没找到的模式补充索引。十一、总结与完整代码仓库核心要点上下文腐烂是 Agent 生产化的第一杀手——不是模型不够聪明是它的工作记忆被垃圾信息淹没了。Checkpoint 是地基。没有它Agent 崩溃后所有 Token 白费。用 DeltaChannel 可以把持久化成本从 O(n²) 降到接近 O(n)。三层记忆 上下文压缩 工具门控三条腿一起走才有显著效果。单独用任何一条都能省 30-50% Token但组合之后才能达到 80%。压缩先硬件后软件——先做不依赖 LLM 的工具输出截断和观察屏蔽零成本再做 LLM 驱动的摘要压缩有成本但更精准。越压缩越要审。压缩后的上下文是二手信息需要额外的校验机制防止污染传播。适用边界本文方案适用于适用 运行超过 30 步的 Agent代码审查、长文档分析、自动化测试适用 工具数量超过 10 个的 Agent适用 使用 LangGraph Claude/GPT 的 Agent 架构部分适用 短任务 Agent 10 步——只需 Checkpoint不需要全套压缩不适用 需要保留所有工具输出精确内容的任务如全文翻译——观察屏蔽会破坏任务语义参考资源LangGraph Memory 官方文档https://docs.langchain.com/oss/python/langgraph/add-memoryLangGraph DeltaChannel 设计文档https://blog.langchain.com/delta-channels-evolving-agent-runtimeLangMem GitHub 仓库https://github.com/langchain-ai/langmemJetBrains Observation Masking 论文2025.12Factory AI / ZenML 上下文压缩评估2026.03Zylos Research: Context Engineering for Long-Running Agents2026.06如有疑问欢迎在评论区交流。这套方案在我们的生产环境跑了两个月不是纸上谈兵。如果你在落地过程中遇到具体问题评论区描述你的场景我会尽量回复。—— 一个持续跟上下文腐烂斗争的 Agent 工程师