Agent A 调研了一遍代码库、写出详细分析,交接给 Agent B 去落地改动。结果 Agent B 从头开始,把 Agent A 早就答过的问题又问一遍,输出还跟三步之前定下的决策互相矛盾。又或者在 AutoGen/AG2 的群聊里,编码助手无视了规划者选定的架构,因为消息线程被摘要过、关键约束被抹掉了。交接边界本质上是一个有损压缩点:只要你没有显式地把状态序列化下来,下游 Agent 就是在闭着眼睛干活。
最快的修复(先从这里下手):别再用一段散文式摘要来传交接数据。定义一个带类型的对象(一个 Pydantic model,或 LangGraph TypedDict 里的字段),把完整的决策、文件路径、约束都放进去,再通过一个显式的 {handoff_context} 占位槽注入到下一个 Agent 的 prompt 里。任何体积大的产物(文件转储、测试日志)写进共享存储,只传 key。仅这一处改动就能解决大多数交接丢失;本页剩下的内容是帮你判断自己究竟属于哪一类。
你属于哪一类?
| 你观察到的症状 | 最可能的原因 | 跳转 |
|---|---|---|
| Agent B 只拿到一句「任务完成」而非细节 | 传的是摘要而非结构化状态 | 原因 1 |
| 长文件/日志内容在下一条消息里中途断掉 | 没有共享存储;消息被截断 | 原因 2 |
| 下游 Agent 的 prompt 里根本没地方接收上下文 | prompt 模板缺少上下文槽 | 原因 3 |
| 本地正常,serverless/多 worker 上出问题 | 每次调用都是全新实例、无历史 | 原因 4 |
| 偶发丢失,只在高负载或并发时出现 | 竞态 / 上下文乱序到达 | 原因 5 |
| 字段在、但类型不对或为空 | 序列化不匹配或字段名写错 | 原因 6 |
常见原因
1. 传了截断后的摘要,而不是结构化状态
最常见的元凶。编排器为了塞进下一个模型的 context window,把 Agent A 的输出压缩了,摘要过程丢掉了具体信息:文件路径、选定的库、被否决的备选方案、错误信息。Agent B 拿到的是「分析认证问题」,而不是「src/auth/jwt.ts 第 47 行用了 HS256 加硬编码 salt,应改为 RS256 并从环境变量加载密钥」。
在 AutoGen/AG2 里这一步常常静默发生,因为群聊管理器的摘要方式默认走 LLM 反思(reflection_with_llm),它的提示词就是一句「Summarize the takeaway from the conversation.」——这条提示词本来就会主动丢掉精确字符串。
怎么判断:把 Agent A 的原始输出和 Agent B 第一条消息里实际收到的内容对比。如果 Agent A 输出的是结构化 JSON 或代码块,而交接消息变成了一段散文,那就是被压缩了。
2. 工具设计是无状态的,没有共享 memory 存储
在 CrewAI、AutoGen 这类框架里,Agent 默认通过聊天消息传数据。长工具输出(文件读取、测试日志、API 响应)超出一条消息能干净容纳的量,就会被截断,关键行从尾部掉出去。整个过程没有任何外部存储被写入。
怎么判断:在框架的消息列表里搜截断标记,例如 ... [truncated]、[output clipped],或者突然的静默断尾。数一数每条交接消息的字符数,跟模型的 context limit 对比。
3. prompt 模板里没有「前置决策」槽
下游 Agent 的 system prompt 没有接收累计上下文的占位符。编排器用 system_prompt.format(task=task) 调用它,从没注入过 prior_decisions、constraints 或 artifacts。这个 Agent 在设计上就是从零开始的。
怎么判断:打开流水线里每个 Agent 的 prompt 模板。如果没有一个引用了 context、prior_decisions 或 handoff 占位符,那就是完全没有做上下文注入。
4. 框架每次调用都重置对话历史
某些编排环境,尤其是无状态的 AWS Lambda 或 Cloud Run,每次调用都创建一个全新的 Agent 实例。每次调用的对话历史都是零。所有上下文都必须在输入 payload 里显式传入,没有任何隐式延续。这正是「本地能跑、上线就崩」这一类 bug:本地单进程共享内存,多 worker 不共享。
怎么判断:在每次 Agent 调用前打印消息历史长度(LangChain 里用 print(len(agent.memory.messages)),OpenAI Agents SDK 里看 trace 中的 run 输入)。如果总是打印 0 或 1,说明历史没有保留下来。
5. 异步流水线里的竞态:上下文乱序到达
在 Temporal workflow 或 Inngest 异步步骤里,Agent B 可能在 Agent A 的最终产物写入完成之前就开始执行了。它读到一个不完整或空的上下文存储,然后带着过期或空的上下文继续往下跑。
怎么判断:检查 workflow 的步骤依赖。如果 Agent B 的步骤把 Agent A 的步骤列为可选,或者没有显式 await 它,那么这层依赖就没有被强制执行。
6. Agent 之间序列化不匹配或字段名写错
Agent A 把上下文写成 Python dataclass 或 Pydantic model。编排器把它序列化成 JSON 时丢掉了不可 JSON 序列化的字段(datetime 变成字符串,enum 变成 int,嵌套对象被压平)。又或者写入方设了 state["research_notes"]、读取方查的是 state["notes"];在未做校验的 LangGraph TypedDict 里,这种错误不会抛异常,只会静默返回 None。
怎么判断:把 Agent A 写入的对象和 Agent B 读到的对象做 diff。任何类型变了、消失了、或回来是 None 的字段,都是序列化或字段名问题的受害者。
最短修复路径
Step 1:给每次交接加一个结构化上下文对象
用带类型的交接信封替代自由文本传递:
from dataclasses import dataclass, asdict
import json
@dataclass
class HandoffContext:
task_id: str
goal: str
decisions: list[dict] # [{"decision": "...", "rationale": "..."}]
artifacts: dict[str, str] # {"name": "store-key-or-path"}
constraints: list[str]
prior_errors: list[str]
payload = json.dumps(asdict(ctx), default=str) # default=str 保住 datetime
把它作为第一条 user 消息传过去,或者通过 {handoff_context} 槽注入到 system prompt。如果你希望这个边界在字段缺失时直接拒绝(抛 ValidationError)而不是把 None 传到下游,那就用 Pydantic 而不是裸 dataclass。
Step 2:大产物写进共享存储,只传引用
绝不要把文件内容直接 inline 进消息。写进共享存储,传 key:
import uuid, redis
r = redis.Redis()
def store_artifact(content: str) -> str:
key = f"artifact:{uuid.uuid4()}"
r.set(key, content, ex=3600) # 1 小时 TTL
return key
# Agent A 写入:
handoff.artifacts["analysis"] = store_artifact(analysis_text)
# Agent B 读取:
analysis = r.get(handoff.artifacts["analysis"]).decode()
Redis、S3,或共享卷上的临时文件都行。原则是:消息携带引用,不携带 payload 本体。
Step 3:审一遍每个 prompt 模板有没有上下文注入槽
# 列出所有「没有」上下文占位符的 Agent prompt 文件
grep -rL "context\|prior_decisions\|handoff" ./prompts/ ./agents/
对每个找出来的文件,补一个槽:
You are continuing work started by a prior agent. You are step 2 of 4.
Prior context (do not re-collect this):
{handoff_context}
Your task:
{task}
标注步骤编号(step 2 of 4)能可观地减少「让我重新开始」的行为,因为模型被明确告知自己处在流水线中段,而不是一个独立助手。
Step 4:用框架自带的交接原语,别手搓消息传递
截至 2026 年 6 月,主流框架各自都有一等公民式的方式来跨交接携带上下文。用它,别自己拼字符串。
LangGraph(用 Command,并配 checkpointer 做持久化):
from langgraph.types import Command
def agent_a(state):
# Command 把状态更新和路由目标一起带过去
return Command(
goto="agent_b",
update={"handoff_context": ctx}, # 结构化,不是摘要
)
# 跨 run 持久化。注意:MemorySaver 已更名为 InMemorySaver。
from langgraph.checkpoint.sqlite import SqliteSaver # 重启后仍在
graph = builder.compile(checkpointer=SqliteSaver.from_conn_string("state.db"))
InMemorySaver 仅供本地开发(重启即丢)。单机持久化用 SqliteSaver/AsyncSqliteSaver,多 worker 生产环境用 Postgres saver。
OpenAI Agents SDK(Swarm 已弃用,Agents SDK 是其继任者,截至 2026 年年中为 v0.17.x)。context 是你传给 Runner.run() 的一个带类型对象,它会到达每个 agent、tool 和 handoff:
from agents import Agent, Runner
result = await Runner.run(planner, input=task, context=my_typed_context)
CrewAI:强制上游输出结构化,并用 context 声明依赖关系:
research = Task(description="...", output_pydantic=ResearchNotes, agent=researcher)
write = Task(description="...", context=[research], agent=writer) # 拿到校验过的 JSON
Step 5:在编排层强制交接顺序
LangGraph 里用显式边,让 B 在 A 完成前无法启动:
graph.add_edge("agent_a", "agent_b")
# 不要用带默认 fallthrough 的 conditional edge,那会跳过 A
Temporal 里 await 每个 activity,并把它的返回值传入下一个(绝不要跨 activity 依赖共享的 worker 内存):
analysis = await workflow.execute_activity(
agent_a_activity, task, schedule_to_close_timeout=timedelta(minutes=5)
)
result = await workflow.execute_activity(
agent_b_activity, analysis, schedule_to_close_timeout=timedelta(minutes=5)
)
Step 6:每次运行都记录交接 payload
import logging
logger = logging.getLogger("handoff")
def handoff(ctx, next_agent: str):
logger.info("HANDOFF to %s: %s", next_agent, json.dumps(asdict(ctx), default=str))
这会建立一条可检索的审计链路。一旦发生上下文丢失,你就能 diff 发出去的和收到的。如果你跑着 tracer(LangSmith、Langfuse),给每个节点记一行摘要,比如 node=writer fields=4 in_tokens=1820,这样你扫一眼 trace 就能看到字段数在哪个边界掉下去。
如何确认已修复
- 在接收方 Agent 的入口加断言,确认必填字段存在且非空,例如
assert state.get("research_notes"), "handoff missing research_notes"。原本会静默丢失上下文的那次运行,现在会大声失败。 - 写一个集成测试:依次跑 A 再跑 B,断言 B 的第一条 prompt 里包含 Agent A 产出的某个已知字符串(一个具体文件路径或错误码),而不是只测每个 Agent 单独能跑。
- 重跑最初失败的那个场景。Agent B 应当直接基于前置决策行动、不再重复询问,也不应与早先的选择相矛盾。
预防建议
- 在写任何 Agent 之前先定义带类型的
HandoffContext(首选 Pydantic);把它当作 Agent 间的 API 契约,让校验在边界处直接拒绝缺失字段。 - 大产物存外部(Redis、S3、磁盘),消息里只传 key 或 URI,并设固定 token 预算(例如 4000 token),超出时显式压缩而不是静默截断。
- 给每个 Agent 的 system prompt 都加上
{handoff_context}槽,连第一个 Agent 也加,这样以后在上游补 Agent 时这个槽始终都在。给每一步标号(step 2 of 4),让模型知道自己处在流水线中段。 - 在编排层设显式步骤依赖;绝不依赖时序或「约定俗成」的顺序。
- 维护一份决策日志,每个 Agent 只追加不覆盖,让下游 Agent 看到此前的全部推理。
- 开发期跑逐节点的结构化日志或 trace;生产期保留一份精简的字段数摘要用于 diff。
常见问答 (FAQ)
Q:LangGraph 会自动处理上下文传递吗?
A:在单次 run 内会:LangGraph 在节点间贯穿传递整个 State 对象,你在 schema 里定义的任何字段都能存活。但它不会自动在多次 run 之间持久化,那得编译时配一个 checkpointer。注意更名:内存型 saver 现在叫 InMemorySaver(旧名 MemorySaver);任何需要重启后仍在的场景,用 SqliteSaver 或 Postgres saver。
Q:我们的 Agent 用了不同模型,这会导致上下文丢失吗? A:会。不同模型 context window 不同,过于「热心」的编排器会静默把交接裁剪到迁就较小的那个。截至 2026 年 6 月主流前沿模型都很宽裕(Claude Opus 4.7、Sonnet 4.6 和 Gemini 3.1 Pro 都是 1M token;GPT-5.5 也很大,但应用内上限随套餐而异),所以通常的解法是别再 inline 巨型产物,而不是去升级模型。务必按链路里最小的那个模型来给交接定大小。
Q:inline 交接消息多大算太大? A:inline 消息控制在大约 2000 token 以内。再大的就存外部、用 key 引用。这样交接消息查起来快、不会被截断,日志也好记录。
Q:OpenAI Swarm 不是做交接的标准方式吗?
A:已经不是了。Swarm 是一个教学/实验性项目、现已弃用;OpenAI 的 Agents SDK 是其生产级继任者(仓库会把你重定向过去)。新项目应当用 Agents SDK,那里 context 是传给 Runner.run() 的一个带类型对象,并会转发到每个 agent 和 handoff。
Q:可以用向量库存交接上下文吗? A:可以,但会引入检索的不确定性:Agent B 未必能召回 Agent A 设定的确切约束。对于硬约束(架构决策、被否决的选项、错误特征),用可靠键值存储里的结构化 JSON。向量检索只用于那种允许模糊召回的大型参考语料。
Q:为什么本地正常、上线后才丢上下文? A:本地通常是单进程共享内存;生产把 Agent 摊到多个 worker、容器或 serverless 调用上,它们之间什么都不共享。所有跨步骤数据都必须显式传递(作为返回值、checkpointer 条目,或共享存储的 key),绝不能假定它还在内存里。