AI Agent 上下文爆了怎么办？从 3 小时崩溃到 8 小时稳定运行的上下文工程实战
·
目录
五、实现一：Checkpoint 持久化——让 Agent 能断点续跑
5.1 LangGraph Checkpoint 的工作原理
八、实现四：渐进式工具加载——告别 200K Token 的 Tool Schema
一、问题背景：一个跑了 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——因为每一步都要背负之前所有步骤的上下文。
二、环境说明与前置依赖
| 组件 | 版本 | 说明 |
|---|---|---|
| Python | 3.12 | 推荐 3.11+ |
| langgraph | 1.2.0 | 含 DeltaChannel 支持（v1.2 新增） |
| langgraph-checkpoint-postgres | 2.0.17 | 生产环境 Checkpoint 后端 |
| langmem | 0.1.0 | 语义记忆管理（LangChain 出品） |
| anthropic | 0.54.0 | Claude API SDK |
| PostgreSQL | 16 | 需要 pgvector 扩展（语义检索） |
| 模型 | Claude Sonnet 4 | 200K 上下文窗口 |
# 安装依赖
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 步的典型任务，结果很扎眼：
| Token 来源 | 消耗占比 | 问题本质 |
|---|---|---|
| MCP 工具 Schema 定义 | 72% (144K) | 28 个工具的全量 JSON Schema 一次性注入 |
| 工具调用原始输出 | 18% (36K) | API 返回、文件内容全部堆积在上下文中 |
| 推理链 + 对话历史 | 7% (14K) | 只有这部分是"有效上下文" |
| System Prompt | 3% (6K) | 固定开销，基本可控 |
画成架构图更直观：
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 倍。
| 模式 | 200 步存储量 | 恢复延迟 | 适用场景 |
|---|---|---|---|
| 全量快照 | 5.3 GB | ~300ms | 短任务 (< 20 步) |
| DeltaChannel | 129 MB | ~80ms (K=50) | 长任务 (50+ 步) |
六、实现二：分层记忆架构——热/温/冷三层管理
6.1 为什么分三层
不分层的记忆有一个致命问题：要么信息太多模型处理不过来，要么信息太少模型丢了关键上下文。三层结构的核心思想是按"访问热度"梯度存储，在 Token 预算和召回精度之间取得平衡。
| 层级 | 范围 | 存储形式 | Token 预算 | 检索精度 |
|---|---|---|---|---|
| 🔥 热层 | 最近 10 轮对话 | 完整原始内容 | ~8K | 100%（在窗口中） |
| 🌡 温层 | 第 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_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 驱动的摘要压缩：
- 工具输出截断（零成本）——工具返回后立即清除原始输出，只保留结构化结论
- 观察屏蔽（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": 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 效果验证
| 策略 | Token 节省 | 实现成本 | 副作用风险 |
|---|---|---|---|
| 工具输出截断 | ~30% | 极低（代码层面） | 低：丢失细节但保留结论 |
| 观察屏蔽 | ~52% | 低（结构化替换） | 中：需要按需检索回原始输出 |
| 锚定迭代摘要 | ~60% | 中（额外 LLM 调用） | 中：摘要可能丢失关键信息 |
| 三条策略叠加 | ~80% | 中 | 可控：分层回退机制兜底 |
八、实现四：渐进式工具加载——告别 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）跑了三次实验，每次从零开始：
| 指标 | 优化前（基线） | 优化后 | 变化 |
|---|---|---|---|
| 最长连续运行时间 | 3h 12m（崩溃） | 8h+（32h 压测无崩溃） | +167% |
| 单任务平均 Token 消耗 | 380K | 72K | -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.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 的查询词和匹配结果，定期审计"模型搜了但没找到"的模式，补充索引。
十一、总结与完整代码仓库
核心要点
- 上下文腐烂是 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-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 工程师
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
8
0- 0
已为社区贡献2条内容
所有评论(0)