前面两篇讨论了两个问题：什么情况下需要多个 Agent，以及任务执行过程中由谁决定下一步。

如果继续往前深入 Multi-Agent 的设计，需要解决一个新的问题：一个 Agent 完成工作后，应该把什么交给下一个 Agent？

最简单的方式，是把前一个 Agent 的回答直接放进下一个 Agent 的上下文。

Agent 数量少、任务简单时，这种方式通常还能工作。

随着任务变长，问题会逐渐显现：前一个 Agent 的结果是否已经完成？哪些结论已经验证？事实来源还能不能追溯？如果只完成了一部分，后续 Agent 应该继续、重试还是等待补充？

如果结果本身很大，后面的 Agent 是否还要把所有内容完整带进上下文？

这些问题最终都会落到同一件事上：Agent 之间需要明确的工作契约。

一、消息与接口的职责边界

讨论 Agent 之间的数据交接，首先要区分三个概念：

对象作用典型内容
Message沟通请求、说明、追问、进度
Task表达一项工作任务 ID、状态、生命周期
Artifact保存任务成果报告、代码、数据、图片、结构化结果

A2A（Agent2Agent Protocol）规范明确区分了 Message、Task 和 Artifact。

Task 表示一项有状态的工作，会经历自己的生命周期；Message 表示通信中的一次交互；Artifact 则表示任务执行过程中产生的结果。A2A 当前规范还明确建议，处于 completed 状态的 Task 应使用 Artifact 返回生成结果。

例如 Research Agent 完成研究后，可以形成这样的交接：

DiagramMermaid
正在生成图表

这里的三部分承担着不同职责。

research completed 是一条通知，告诉 Writer 研究阶段已经结束。

completed 是任务状态，程序可以据此知道这项工作已经进入完成状态。

真正供 Writer 使用的研究成果，则放在 ResearchBundle 这个 Artifact 中。

这种区分很重要。

Message 更适合承载交流过程中的信息，例如：

代码片段Text
正在查询第二个数据源。

或者：

代码片段Text
目前缺少 2025 年的数据，是否继续？

其价值主要体现在交互过程。

正式的研究报告、代码文件、结构化数据等结果，则更适合放入 Artifact。

此外，还有一个工程上的原因。

A2A 中的 Task.history 本身就是可选的，客户端获取 Task 时也不应假设所有历史消息都会被永久、完整保存。实时流式连接发生中断时，当前连接上的增量更新同样可能停止接收。相比之下，Artifact 属于 Task 的输出。只要服务端仍然保留对应 Task，客户端就可以重新获取任务，并读取其中已经生成的 Artifact。

上述三者的职责可概括为：

代码片段Text
Message 负责交流过程
Task 负责工作生命周期
Artifact 负责工作成果

因此，需要继续处理、重复读取或者作为正式成果保存的数据，不应该只存在于聊天消息里。

二、Agent 之间的工作契约

假设系统中有这样一条流程：

代码片段Text
Research → Writer → Reviewer

这里暂时只讨论 Research 和 Writer 之间如何交接。

一种常见写法是：

代码片段Text
Research Agent:

我查到了这些资料……

Writer Agent:

请根据 Research Agent 的结果写文章。

这种设计的问题在于，Writer 还要重新解释上一段自然语言。

它需要自己判断：

代码片段Text
哪些内容属于事实？
哪些属于 Research Agent 的分析？
哪些信息有可靠来源？
哪些问题还没有解决？
研究是否已经完整结束？

每经过一个 Agent，都要重新做一次语义判断。

Agent 越多，这些隐含信息越容易在传递过程中发生变化。

更稳定的方式，是让 Research 返回一个固定结构：

代码片段JSON
{
  "schema_version": "1.0",
  "task_id": "research-42",
  "result_status": "completed",
  "facts": [
    {
      "claim": "...",
      "source_id": "src-7",
      "confidence": 0.94
    }
  ],
  "sources": [
    "src-7",
    "src-9"
  ],
  "open_questions": [
    "..."
  ],
  "artifact_refs": [
    "artifact://research/raw-notes-42"
  ]
}

这里可以把整个对象定义为：

代码片段Text
ResearchBundle

Bundle 表示一组已经组织好的相关数据。

Research 和 Writer 可以提前约定各个字段的含义。比如：

facts 保存可以交给后续阶段使用的事实。

sources 保存资料来源。

open_questions 记录当前没有解决的问题。

artifact_refs 指向完整研究笔记、原始数据或者其他体积较大的结果。

schema_version 则表示当前采用哪一版接口结构。

有了这些约定，Writer 不再需要从几千字自然语言中猜测哪些内容可以直接使用。

Agent 之间开始拥有类似普通软件模块之间的接口边界。

所谓 Agent 的工作契约，本质上就是提前约定：

代码片段Text
我要给你什么？
每个字段是什么意思？
哪些字段必须存在？
出现异常时怎样表达？
哪些结果可以继续使用？
大数据放在哪里？

Prompt 解决 Agent 如何完成工作。

契约解决 Agent 如何把工作交出去。

三、结构化数据的适用边界

有了契约，并不意味着所有内容都应该变成严格字段。

实际开发中，更适合结构化的通常是程序需要直接判断的数据。

例如：

代码片段JSON
{
  "result_status": "partial",
  "error_type": "SOURCE_UNAVAILABLE",
  "retryable": true
}

程序可能根据这些字段决定是否继续执行，因此状态最好使用固定值。

相比下面这句话：

代码片段Text
资料好像没有完全找齐，之后可能需要再尝试一次。

结构化数据更容易直接在代码控制逻辑中使用。例如：

代码片段Python
if result["result_status"] == "partial" and result["retryable"]:
    retry_failed_sources()

这里不需要模型再次理解“好像没有完全找齐”到底代表什么。

类似的还有：

代码片段Text
task_id
schema_version
source_id
artifact_id
timestamp
error_code
retryable

它们都比较适合采用固定格式。

自然语言则适合表达真正需要语义的信息：

代码片段JSON
{
  "claim": "...",
  "summary": "...",
  "reason": "..."
}

例如 summary 很难通过枚举表示，因为它本身就是一段需要模型理解和生成的内容。

因此，我们记住一个原则：程序需要判断的内容尽量结构化，模型负责理解和生成的内容保留自然语言。

这和 Tool 的设计思路很接近。

例如一个搜索工具可以规定输入：

代码片段JSON
{
  "type": "object",
  "properties": {
    "query": {
      "type": "string"
    },
    "depth": {
      "type": "string",
      "enum": [
        "quick",
        "standard",
        "deep"
      ]
    }
  },
  "required": [
    "query",
    "depth"
  ]
}

这里：

代码片段Text
required

表示必须提供哪些字段。

代码片段Text
enum

限制字段只能采用指定值。

收到数据以后，程序可以按照 Schema 检查字段是否存在、类型是否正确、值是否合法。

Agent 之间的数据交接同样适合采用这种思路。

四、事实来源的追溯链路

Multi-Agent 系统还有一个很实际的问题：信息会被反复改写。

Research Agent 找到的数据可能是：

代码片段Text
某项指标同比增长 37%。

Writer 为了让文章更自然，写成：

代码片段Text
该指标出现明显增长。

Reviewer 又调整成：

代码片段Text
该领域正在快速增长。

文字经过几次加工以后，原来的“同比增长 37%”已经变成了范围更大的判断。

这时再问：

代码片段Text
这个结论来自哪里？

系统可能已经回答不出来。

因此，用于 Agent 协作的数据结构中，最好同时保存 Provenance。

Provenance 可以理解为数据的来源关系以及产生过程。

它主要解决几个问题：

代码片段Text
这条信息来自哪里？
什么时候获取？
由哪个 Agent 产生？
对应哪一份原始材料？

例如：

代码片段JSON
{
  "id": "claim-12",
  "claim": "某项指标同比增长 37%",
  "source_id": "src-7",
  "source_uri": "https://example.com/report",
  "retrieved_at": "2026-09-23T08:00:00Z",
  "producer": "research-agent",
  "confidence": 0.94
}

Writer 可以重新组织文字，但仍然保留：

代码片段Text
claim-12 → src-7

Reviewer 对某个结论有疑问时，可以沿着 source_id 找回原始资料。

这样就不需要重新执行整轮 Research。

confidence 则需要谨慎处理。

如果 0.94 只是模型根据感觉生成的数字，就不能把它理解成统计意义上的“94% 正确概率”。

更合理的方式，是提前定义置信度规则，例如：

代码片段Text
high
medium
low

或者根据来源数量、来源质量、交叉验证结果计算。

真正重要的并不是这个数字本身。重要的是：事实经过多个 Agent 以后，来源关系仍然存在。

五、异常状态的契约表达

真实系统里，一个 Agent 很可能只能完成部分工作。

Research Agent 计划查询 10 个来源，其中两个网站超时。

Coding Agent 完成了主要功能，但还有三个测试没有通过。

数据 Agent 缺少部分字段，只生成了当前能够计算的结果。

这时，仅仅使用：

代码片段Text
success
failed

往往不够。

应用自己的结果契约可以进一步描述：

代码片段JSON
{
  "result_status": "partial",
  "retryable": true,
  "missing_inputs": [],
  "errors": [
    {
      "code": "SOURCE_TIMEOUT",
      "source_id": "src-9"
    }
  ],
  "artifact_refs": [
    "artifact://research/partial-result-42"
  ]
}

这里有一个概念需要区分。

partial 是我们自己的业务结果状态，表示：

代码片段Text
已经产生有效结果，但结果并不完整。

它并不是 A2A 的标准 TaskState。

A2A 当前的任务状态包括：

代码片段Text
submitted
working
input-required
auth-required
completed
failed
canceled
rejected

以及用于未知状态的表示。

其中：

working 表示任务仍在处理。

input-required 表示需要额外输入才能继续。

auth-required 表示需要完成认证。

completed 表示任务正常结束。

failed 表示任务因错误结束。

业务结果和协议状态属于两个层次，不需要强行做成同一套枚举。

例如：

代码片段JSON
{
  "task_state": "completed",
  "result_status": "partial"
}

可以表达：

任务本身已经结束，但最终只拿到了部分可用结果。

具体是否允许这种组合，应由应用自己的语义决定。

关键在于，部分完成、错误原因以及是否能够重试，都应该成为契约的一部分。

如果成功时返回 JSON：

代码片段JSON
{
  "status": "completed"
}

失败时却只返回一句：

代码片段Text
这次好像出了点问题。

后面的 Agent 依然很难稳定处理。

六、大型结果的引用传递

另一个容易被忽略的问题是上下文长度。

假设 Research Agent 产生了一万字研究资料：

代码片段Text
Research
   ↓
10,000 字资料
   ↓
Writer
   ↓
10,000 字资料
+ 5,000 字草稿
   ↓
Reviewer

随着 Agent 数量增加，后面的上下文会越来越长。

如果每个 Agent 还继续总结前一个 Agent 的结果，又会出现另一种问题：

代码片段Text
原始资料
   ↓
Research Summary
   ↓
Writer Summary
   ↓
Reviewer Summary

每经过一层，都可能损失一些信息。

更适合工程系统的方式，是把体积较大的结果保存起来，只传轻量引用。

例如完整研究报告可以放入：

代码片段Text
文件系统
对象存储
数据库
Artifact Store

Agent 之间只传：

代码片段JSON
{
  "artifact_id": "research-42",
  "uri": "artifact://research/report-42",
  "media_type": "text/markdown",
  "sha256": "a83f...",
  "schema_version": "1.0"
}

这里：

artifact_id 用于识别结果。

uri 表示内容在哪里。

media_type 表示数据格式。

sha256 可以用于确认读取到的内容是不是预期版本。

这样，无论 Artifact 是几 KB 的 Markdown，还是几百 MB 的数据文件，Agent 之间都只需要传递一个很小的引用对象。

还有一个直接收益：并不是所有 Agent 都需要读取完整结果。

例如：

代码片段Text
Research
   ↓
ResearchBundle
   ├── facts
   ├── sources
   └── raw_notes_ref

Writer 可能只读取 facts 和 sources。

只有发现信息不足时，才加载 raw_notes_ref 对应的完整笔记。

这比一开始就把所有内容塞进上下文更合理。

七、协作链路中的数据交接

现在重新看：

代码片段Text
Research → Writer → Reviewer

Research 完成工作后，可以先形成：

代码片段Text
Task
state = completed

同时产生：

代码片段Text
ResearchBundle v1

例如：

代码片段JSON
{
  "schema_version": "1.0",
  "task_id": "research-42",
  "result_status": "completed",
  "facts": [
    {
      "id": "claim-1",
      "claim": "...",
      "source_id": "src-7"
    }
  ],
  "sources": [
    "src-7",
    "src-9"
  ],
  "open_questions": [],
  "artifact_refs": [
    "artifact://research/raw-notes-42"
  ]
}

Writer 收到后，先检查：

代码片段Text
schema_version 是否支持

必要字段是否存在

result_status 是否满足写作条件

Writer 自己的消费契约还可以规定：

代码片段Text
只有带 source_id 的 facts 可以直接作为事实使用。

open_questions 中的信息仍然属于待确认内容，Writer 不能自行补全。

Writer 完成以后，再输出：

代码片段Text
DraftArticle v1

Reviewer 读取 DraftArticle，以及必要的 ResearchBundle。

它不需要重新接收 Research 和 Writer 之间的所有聊天内容。

审校完成后可以返回：

代码片段JSON
{
  "accepted_claims": [
    "claim-1",
    "claim-3"
  ],
  "rejected_claims": [
    "claim-2"
  ],
  "required_changes": [
    {
      "claim_id": "claim-2",
      "reason": "source does not support this conclusion"
    }
  ]
}

三个 Agent 仍然可以通过 Message 交流。

但真正推动任务向前发展的，已经是一组明确的数据对象。

这时系统能够回答很多具体问题：

代码片段Text
这一阶段是否完成？
正式结果在哪里？
这条事实是谁产生的？
来源是什么？
当前结果是否完整？
失败部分能不能重试？
现在使用哪个 Schema 版本？

如果 Agent 之间只传一段不断增长的 Conversation History，这些问题通常很难准确回答。

Task、Artifact、Schema 和 Provenance 的价值，就体现在这里。

八、LangGraph 中的契约实现

前面的设计并不依赖某一种 Agent 框架。

只要多个 Agent 之间存在长期、稳定的数据交接，都可以建立类似的契约。

放到 LangGraph 中，这种关系尤其直观。

LangGraph 的 StateGraph 围绕 State 运行：节点读取当前 State，完成处理以后返回需要更新的数据。

因此，可以直接把 Agent 之间的工作契约放进 State。

仍然使用这个示例：

代码片段Text
Research → Writer → Reviewer

首先定义 Research Agent 的结果模型。

这里使用 Pydantic：

代码片段Python
from typing import Literal, TypedDict
 
from pydantic import BaseModel, Field
 
 
class Fact(BaseModel):
    id: str
    claim: str
    source_id: str
 
    # 如果使用置信度，需要给它明确的业务定义
    confidence: float | None = Field(
        default=None,
        ge=0,
        le=1,
    )
 
 
class ArtifactRef(BaseModel):
    artifact_id: str
    uri: str
    media_type: str
    sha256: str
 
 
class ResearchBundle(BaseModel):
    schema_version: Literal["1.0"] = "1.0"
 
    task_id: str
 
    # 这是应用自己的结果状态，
    # 不等同于 A2A TaskState
    result_status: Literal[
        "completed",
        "partial",
        "failed",
    ]
 
    facts: list[Fact]
    sources: list[str]
    open_questions: list[str]
    artifact_refs: list[ArtifactRef]

这里的 ResearchBundle 就是 Research 和 Writer 之间的接口。

它规定 Research 必须交出哪些内容。

工作流 State 可以继续定义为：

代码片段Python
class WorkflowState(TypedDict, total=False):
    research: dict
    draft: str
    review: dict

三个字段分别对应不同阶段的结果：

代码片段Text
research
    Research Agent 的结构化结果

draft
    Writer Agent 生成的文章

review
    Reviewer Agent 的审校结果

接下来实现 Research 节点：

代码片段Python
def research_agent(state: WorkflowState):
    bundle = ResearchBundle(
        task_id="research-42",
        result_status="completed",
        facts=[
            Fact(
                id="claim-1",
                claim=(
                    "A2A 将 Task、Message 和 Artifact "
                    "定义为不同的数据对象"
                ),
                source_id="src-a2a-1",
                confidence=0.95,
            )
        ],
        sources=[
            "src-a2a-1"
        ],
        open_questions=[],
        artifact_refs=[
            ArtifactRef(
                artifact_id="raw-notes-42",
                uri="artifact://research/raw-notes-42",
                media_type="text/markdown",
                sha256="a83f...",
            )
        ],
    )
 
    return {
        "research": bundle.model_dump()
    }

这里发生了一个很重要的变化。

Research Agent 不再返回：

代码片段Text
我研究完了，下面是完整结果……

然后附上一大段自然语言。

它返回的是一个符合 ResearchBundle 契约的数据对象。

完整研究笔记也没有直接放进 State。

State 中只保存：

代码片段Text
artifact://research/raw-notes-42

这样的引用。

Writer 节点收到数据以后，可以先验证契约：

代码片段Python
def writer_agent(state: WorkflowState):
    research = ResearchBundle.model_validate(
        state["research"]
    )
 
    if research.result_status != "completed":
        raise ValueError(
            "research result is not completed"
        )
 
    usable_facts = [
        fact
        for fact in research.facts
        if fact.source_id
    ]
 
    draft = "\n".join(
        fact.claim
        for fact in usable_facts
    )
 
    return {
        "draft": draft
    }

model_validate() 是 Pydantic 提供的模型验证方法。

它会检查输入是否符合 ResearchBundle。

例如：

代码片段Text
task_id 是否存在
result_status 是否属于允许值
facts 是否符合 Fact 结构
confidence 是否处于允许范围

如果上游输出已经发生结构变化，可以在 Writer 执行主要逻辑之前发现。

Writer 还实现了一条自己的消费规则：

代码片段Python
if fact.source_id

只有存在来源的事实才能进入后续处理。

原本只能写在 Prompt 里的约束：

代码片段Text
请只使用有来源的信息。

现在有一部分已经进入程序逻辑。

Reviewer 也可以采用同样的方式：

代码片段Python
def reviewer_agent(state: WorkflowState):
    research = ResearchBundle.model_validate(
        state["research"]
    )
 
    draft = state["draft"]
 
    source_ids = {
        fact.source_id
        for fact in research.facts
    }
 
    return {
        "review": {
            "result_status": "completed",
            "checked_sources": sorted(source_ids),
            "draft_length": len(draft),
            "required_changes": [],
        }
    }

最后建立 LangGraph：

代码片段Python
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
 
 
builder = StateGraph(WorkflowState)
 
builder.add_node(
    "research",
    research_agent,
)
 
builder.add_node(
    "writer",
    writer_agent,
)
 
builder.add_node(
    "reviewer",
    reviewer_agent,
)
 
builder.add_edge(
    START,
    "research",
)
 
builder.add_edge(
    "research",
    "writer",
)
 
builder.add_edge(
    "writer",
    "reviewer",
)
 
builder.add_edge(
    "reviewer",
    END,
)
 
graph = builder.compile(
    checkpointer=InMemorySaver()
)

调用时，再为这次工作流指定 thread_id：

代码片段Python
result = graph.invoke(
    {},
    {
        "configurable": {
            "thread_id": "article-42"
        }
    },
)

LangGraph 的 Checkpointer 会保存工作流 State 的 checkpoint。

checkpoint 按 thread 组织，并在 graph 的 super-step 边界保存 State 快照，可用于状态恢复、Human-in-the-loop、调试和故障恢复。

示例使用 InMemorySaver，它把 checkpoint 放在内存里，进程重启以后数据会丢失。生产环境需要跨进程保存状态时，应使用 PostgreSQL、SQLite 等持久化 Checkpointer。

现在整个工作流变成：

代码片段Text
Research Agent
      │
      │ ResearchBundle
      ▼
Writer Agent
      │
      │ Draft
      ▼
Reviewer Agent
      │
      │ ReviewResult
      ▼
     END

节点之间讨论的问题也发生了变化。

不再只是：

代码片段Text
上一个 Agent 说了什么？

而是：

代码片段Text
上一个阶段产生了什么数据？
这些数据符合哪个 Schema？
哪些字段可以直接消费？
哪些结果仍然不完整？
大结果保存在哪里？
来源能不能追溯？
当前工作流能不能恢复？

这也是 LangGraph State 很适合承担的一项职责。

State 不必只存 Conversation History。

它同样可以成为多个 Agent 之间的数据接口。

不过，LangGraph State 和 A2A Artifact 仍然属于不同层次。

LangGraph State 解决的是：

代码片段Text
一个工作流内部怎样传递和保存状态？

A2A 解决的是：

代码片段Text
独立 Agent 或 Agent 服务之间怎样交换
Message、Task 和 Artifact？

如果 Research、Writer 和 Reviewer 都运行在同一个 LangGraph 应用中，可以直接通过 State 传递 ResearchBundle。

如果 Research Agent 是独立服务，而 Writer Agent 运行在另一套系统里，那么 ResearchBundle 可以作为 Artifact，通过 A2A 交给 Writer。

框架负责运行 Agent。

契约负责让 Agent 之间稳定交接工作。

总结

本文讨论了 Multi-Agent 系统中一个很基础、也很容易被忽略的问题：Agent 之间究竟应该如何交接数据。

Message 适合沟通和状态通知，Task 用来描述工作的生命周期，Artifact 承载正式结果；在此基础上，再通过 Schema 约束结构、Provenance 保留来源、Error Contract 表达异常，并用 Artifact Reference 传递较大的结果。

当这些信息形成明确的工作契约后，后续 Agent 就不必反复解释上一轮自然语言。系统也能更清楚地判断任务是否完成、结果是否可靠、错误能否恢复，以及数据是否可以继续使用。对于长期运行的 Multi-Agent 系统，这些接口设计和 Prompt 本身同样重要。