你在 LangGraph 里把一个「收集需求」Agent 的输出交给「生成代码」Agent,结果后者问了一遍「请问需求是什么?」——明明前一步已经梳理清楚了。或者在 CrewAI 里,Researcher Agent 搜集了 20 条参考资料,交给 Writer Agent 时只剩下一句空洞的摘要。这类问题在 AutoGen、OpenAI Swarm、Temporal Workflow 里同样频繁出现。根本症状是:接收方 Agent 的初始 prompt 里缺少上游产出的关键信息,导致它从零开始推断或直接忽略前置工作。
常见原因
1. Handoff payload 只传了摘要而不是全文
编排框架默认把上一步的 output 字段截断到固定长度,或者开发者在组装 handoff message 时手写了一个「summarize then pass」的逻辑,把重要细节丢掉了。接收 Agent 拿到的是「任务完成」的结论,而不是实际内容。
怎么判断:在接收 Agent 的系统提示或第一条 user 消息里搜索上游的关键词(如具体文件名、数字、错误码)。如果找不到,payload 就是被截断了。
2. 状态对象的 schema 不匹配
发送方写入 state["research_notes"],接收方读的是 state["notes"]。LangGraph 的 TypedDict 如果没有严格校验,这个错误不会抛异常,接收方只会拿到 None。
怎么判断:打印接收 Agent 实际收到的 state 字典,检查期望字段是否存在、值是否为空。
3. 异步交接时消息乱序
在并发度大于 1 的流水线里(例如 AutoGen 的 GroupChat 开了多个 speaker),两条消息同时到达时,框架的消息队列可能把摘要消息排在正文消息之前投递给下一 Agent,导致它先看到「任务完成」再看到内容,但已经开始生成了。
怎么判断:查看 Trace 里各消息的 timestamp,确认接收方处理消息的顺序是否与发送顺序一致。
4. 长上下文被 context window 截断
上游 Agent 输出了 8000 token 的调研报告,整个对话历史加起来已经逼近模型限制,框架的滑动窗口把最早(也是最重要)的那部分截掉了。
怎么判断:用 tiktoken 或框架自带的 token 计数器测量拼接后的 prompt 长度,与模型 context limit 对比。如果超过 80%,就是截断风险区。
5. Handoff 走了错误的通道
在 Temporal 或 Inngest 这类工作流引擎里,Agent A 把结果写进了 activity 的 local state,而不是 workflow 的全局 state。Agent B 启动时拿不到那段 local state,因为它跑在不同的 worker 实例上。
怎么判断:检查上游 activity 的返回值是否被 workflow 显式传入下游 activity 的输入参数。如果是隐式依赖 worker 内存,就会在分布式部署时失效。
6. System prompt 里遗漏了「你是流水线的一部分」这一上下文
接收 Agent 的 system prompt 把自己定义为「独立助手」,没有告知它前一步已经做了什么,导致它重新开始而不是接续工作。
怎么判断:读接收 Agent 的 system prompt,看是否有对流水线位置的说明,例如「你是第二步,第一步已完成 X」。
最短修复路径
Step 1:在 handoff payload 里传完整结构化对象
不要传字符串摘要,改传 JSON 对象:
# LangGraph 示例
def handoff_node(state: WorkflowState) -> WorkflowState:
# 错误:只传摘要
# state["handoff"] = summarize(state["research_output"])
# 正确:传完整结构
state["handoff"] = {
"research_notes": state["research_output"],
"source_urls": state["source_urls"],
"key_entities": state["key_entities"],
"step": "research_complete"
}
return state
Step 2:统一 state schema 并加断言
from typing import TypedDict, Required
class WorkflowState(TypedDict, total=False):
research_notes: Required[str] # 必填字段
source_urls: list[str]
handoff: dict
# 在接收节点入口加校验
def writer_node(state: WorkflowState) -> WorkflowState:
assert state.get("research_notes"), "handoff 缺少 research_notes"
...
Step 3:在接收 Agent 的 system prompt 里注入上下文摘要
system_prompt = f"""你是写作 Agent,负责流水线的第二步。
第一步(Research Agent)已完成,产出摘要如下:
{state['research_notes'][:2000]}
请在此基础上继续,不要重复收集资料。"""
Step 4:对长输出做显式压缩而不是隐式截断
def compress_for_handoff(text: str, max_tokens: int = 3000) -> str:
"""用模型自身把长输出压缩成结构化摘要,保留关键细节。"""
if token_count(text) <= max_tokens:
return text
return llm.invoke(f"把下面内容压缩成不超过 {max_tokens} token 的结构化摘要,保留所有数字、名称和代码片段:\n\n{text}")
Step 5:在 Temporal/Inngest 里显式传递返回值
# Temporal workflow 示例
@workflow.defn
class ResearchWriteWorkflow:
@workflow.run
async def run(self, input: str) -> str:
# 显式把 activity 返回值传入下一 activity
research_result = await workflow.execute_activity(
research_activity, input, schedule_to_close_timeout=timedelta(minutes=5)
)
# research_result 现在是 write_activity 的显式输入
return await workflow.execute_activity(
write_activity, research_result, schedule_to_close_timeout=timedelta(minutes=5)
)
Step 6:加端到端集成测试验证 handoff 完整性
# 在 CI 里跑 handoff smoke test
python -m pytest tests/test_handoff_integrity.py -v
# 断言接收 Agent 的第一条消息包含上游关键词
预防建议
- 把所有 handoff payload 定义为 Pydantic model,字段缺失时直接抛
ValidationError,不要用裸字典 - 在每个节点入口记录「我收到了哪些字段 + 各字段 token 数」的结构化日志
- 为「接收方能看到上游关键字段」写专项集成测试,纳入 CI
- 规定 handoff payload 的最大 token 预算(例如 4000 token),超出时触发显式压缩而不是静默截断
- 在 system prompt 里为每个 Agent 标注流水线位置编号(如「步骤 2/4」)和前驱步骤的产出类型
- 对分布式 worker 场景(Temporal/Inngest),所有跨 activity 的数据必须通过 workflow 返回值显式传递,禁止依赖 worker 内存
- 定期在 staging 环境跑全链路 trace 检查,确认每个 handoff 节点的 input token 数符合预期
常见问答 (FAQ)
Q: 为什么本地测试没问题,上线后才丢失上下文? A: 本地通常单进程运行,状态在内存里共享。上线后多 worker 部署,不同 activity 跑在不同进程甚至不同机器上,隐式的内存共享就会失效。必须通过显式返回值传递所有跨步骤数据。
Q: LangGraph 的 MemorySaver 能解决这个问题吗?
A: 能部分缓解——MemorySaver 会持久化 graph state,跨节点共享。但如果 schema 字段名拼错,或者并发节点写入同一字段产生竞态,依然会丢数据。推荐把 MemorySaver 和字段断言结合使用。
Q: 上游输出太长,压缩摘要又会丢细节,有没有更好的办法?
A: 用「结构化抽取」代替自由文本压缩:让上游 Agent 输出固定 schema 的 JSON(如 {"entities": [...], "conclusions": [...], "raw_ref": "..."}),下游只读它需要的字段。这样既控制了 token 数,又保留了精确细节。
Q: 如何在不改框架代码的情况下快速排查是哪一步丢失了上下文?
A: 在每个节点的入口和出口各打一行日志,内容是 "node=X fields=Y token_count=Z"。把这些日志收进 Langfuse 或 LangSmith,用 trace 视图逐节点对比 input/output,丢失点一目了然。