Phase 12 · 多智能体协作（Agent-to-Agent, A2A）

为什么需要它（Why）

Phase 5 我们对比了「多 Agent 工作流」的几种范式；Phase 11 我们用 MCP 让一个智能体调用外部工具（agent ↔ tool）。

但真实生产里，常常不是一个智能体单打独斗，而是多个智能体互相配合——比如「调研 Agent」先收集事实，「写作 Agent」再落成成文，「主管 Agent」统筹调度。这就是 A2A（Agent-to-Agent，智能体互联）。

MCP 是 agent 接工具，A2A 是 agent 接 agent。两者是「智能体互联」这张大图里的两条主线。

本 Phase 用 LangGraph 原生多智能体（supervisor + worker），不引入任何额外协议 SDK，单机 make run 即可跑——正好呼应本教程主线：状态图（StateGraph）+ 节点（Node）+ 条件边（Conditional Edge）。

是什么（What）

一个由三张状态图拼成的协作系统：

• supervisor（主管）：一个 Agent，用 LLM 判断「下一步该交给谁」，输出 researcher / writer / FINISH。
• researcher（调研员）：一个 Agent，只负责列出关键要点。
• writer（写作者）：一个 Agent，基于要点写成短文。

三者共享同一份 messages 状态，互相「看得见」对方的产出；条件边根据主管的决策在 worker 之间路由。

        ┌─────────────┐
START → │ supervisor  │ ──researcher──→ ┌────────────┐
        └─────────────┘ ←──────────────│ researcher │
              │FINISH→END               └────────────┘
              └──writer──→ ┌──────────┐
                          │  writer   │ ──→ 回到 supervisor
                          └──────────┘

怎么写（How）

1. 共享状态

next 记录主管的路由决策；messages 用 add_messages 累加。

class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], add_messages]
    next: str

2. 两个 worker（都是 Agent）

用 create_react_agent 把角色提示包成独立 Agent。注意：角色提示只在本次调用内通过 SystemMessage 注入，不写进共享状态，避免污染对话轨迹。

async def call_researcher(state: AgentState) -> dict:
    agent = create_react_agent(get_chat_model(), tools=[])
    result = await agent.ainvoke(
        {"messages": [SystemMessage(content=RESEARCHER_PROMPT)] + list(state["messages"])}
    )
    answer = result["messages"][-1]
    return {"messages": [answer]}  # 只把最终回答追加进共享状态，避免重复

3. supervisor 路由节点

主管节点调用 LLM 做决策，并做容错（只取合法 token）：

async def supervisor(state: AgentState) -> dict:
    model = get_chat_model()
    decision = await model.ainvoke(
        [HumanMessage(content=ROUTING_INSTRUCTION)] + list(state["messages"])
    )
    text = decision.content.strip()
    for token in ("FINISH", "writer", "researcher"):
        if token in text:
            text = token
            break
    else:
        text = "FINISH"
    return {"messages": [AIMessage(content=f"（主管决定：{label}）")], "next": text}

4. 用条件边把三者拼成图

builder = StateGraph(AgentState)
builder.add_node("supervisor", supervisor)
builder.add_node("researcher", call_researcher)
builder.add_node("writer", call_writer)

builder.add_edge(START, "supervisor")
builder.add_conditional_edges(
    "supervisor", should_continue,
    {"researcher": "researcher", "writer": "writer", "FINISH": END},
)
builder.add_edge("researcher", "supervisor")
builder.add_edge("writer", "supervisor")
graph = builder.compile()

关键点：worker 干完活都回到 supervisor，由主管决定下一步——这就是「主管 + 工作」多智能体编排的本质。

跑一下（Run）

make run p=12
# 或 python -m examples.p12.hello_chain

示例会向用户提问「用三句话向不懂技术的人解释 Agent 是什么」，你会看到终端依次打印主管的路由决策、调研员的要点、写作者的短文，以及最终的完整 messages 轨迹。典型过程：

[supervisor] 路由决策 → 调研员
[worker] 调研员产出：
- Agent 是能感知环境并自主行动的软件
- 它用大语言模型做决策
- 可以调用工具完成任务
[supervisor] 路由决策 → 写作者
[worker] 写作者产出：
Agent 就像一个小助手……
[supervisor] 路由决策 → 结束

小结（Summary）

• A2A = 多个智能体互相配合；MCP = 智能体接工具。两者共同构成「智能体互联」。
• LangGraph 原生多智能体就是 状态图 + 节点 + 条件边：每个 Agent 是一个节点，主管的决策是条件边。
• 与 Phase 11 一样，worker 之间、主管与各 worker 之间共享 messages，但角色提示不进共享状态，保持轨迹干净。
• 进阶方向：把 worker 换成 MCP 工具服务器背后真正的子智能体、引入 langgraph-supervisor 库的标准 handoff 模式、或接入 Google A2A 协议做跨进程智能体互调（本教程聚焦原生范式，未展开协议标准）。