Agent 交接时上下文丢失了

多 Agent 流水线中,子 Agent 接手任务后丢失关键背景,导致重复询问或输出偏离预期。本文逐步拆解丢失原因并给出可落地的修复方案。

你在 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,丢失点一目了然。

相关阅读

标签: #AI 编程 #Agents #排查