服务内部的 Agent2Agent 怎么交互？AgentScope vs LangChain 实战对比

Google 在 2025 年 4 月把 A2A（Agent2Agent）协议推成了开放标准，很多人就开始用「A2A」这个词代指所有多 Agent 协作。
但实际上 —— A2A 协议解决的是一个特定问题，而「服务内部多 Agent 怎么协作」是另一个完全不同的问题。
这篇文章把这两件事拆开讲清楚。

⚠️ 代码声明：本文所有代码示例按公开文档写法，不能保证在最新版本下能直接跑通。两个框架都在快速迭代，API 路径和函数签名可能已变。建议把代码当伪代码 / 设计参考，实际落地以官方最新文档为准。跑不通时不要慌，对照报错查文档改 1-2 行就行。

读完你应该能：

• 区分网络级的 A2A 协议和进程内的多 Agent 协作
• 说出服务内部 Agent2Agent 的三种主流模式
• 用 AgentScope 或 LangGraph 写一个能跑的多 Agent 系统
• 理解 A2A 协议是怎么把不同框架的 Agent 串起来的

---

一、先澄清：A2A 协议 ≠ 多 Agent 协作

1.1 A2A 协议是什么

A2A（Agent2Agent）是 Google 在 2025 年 4 月联合 50 多家公司推出的网络级开放协议，目的是让 不同厂商、不同语言、不同进程写的 Agent 能互相调用。

类比一下理解：

协议	解决什么	对应网络层
HTTP	浏览器和服务端怎么对话	应用层
SMTP	不同邮件服务商怎么互发邮件	应用层
A2A	不同 Agent 服务怎么互相调用	应用层

A2A 的核心机制：

// 每个 Agent 暴露一份「名片」（类似 OpenAPI）
GET https://agent-a.example.com/.well-known/agent.json
{
  "name": "Research Agent",
  "skills": [
    {"id": "web-search", "description": "联网检索资料"},
    {"id": "summarize",  "description": "长文摘要"}
  ],
  "auth": "bearer"
}

调用方发 tasks/send，服务端通过 SSE（Server-Sent Events）流式返回进度和结果。完整流程：

Client                              A2A Server (Agent B)
  │                                       │
  │─── GET /.well-known/agent.json ──────▶│   1. 拿 Agent Card
  │◀──── 200 OK {name, skills...} ────────│
  │                                       │
  │─── POST /tasks/send ────────────────▶│   2. 提交任务
  │      { "skill": "web-search",        │
  │        "input": "A2A 协议" }          │
  │                                       │
  │◀─── SSE: data: {state: working} ─────│   3. 流式进度
  │◀─── SSE: data: {state: working} ─────│
  │◀─── SSE: data: {state: completed,    │
  │                artifacts: [...]} ────│   4. 最终结果
  │                                       │

1.2 服务内部的 Agent2Agent 是另一回事

同一个进程、同一份代码里，多个 Agent 之间怎么协作 —— 这不是协议问题，是架构模式问题。

为什么需要拆开讲？因为你搜「A2A」搜到的内容，90% 是讲上面那个网络协议，但真正写代码时遇到的 90% 问题，是进程内多 Agent 怎么组织。

本文主角：服务内部的 Agent2Agent（多 Agent 协作模式）
次要话题：A2A 协议（只在最后一节用来串联两个框架的 Agent）

---

二、服务内多 Agent 协作的三种主流模式

不管用 AgentScope、LangGraph 还是自研框架，多 Agent 协作逃不出这三种模式：

2.1 Orchestrator 编排（最常见）

     ┌─────────────────┐
     │  Orchestrator   │ ← 主管 Agent（往往是 LLM 驱动的 ReAct）
     │  （主管）        │
     └────────┬────────┘
              │ 调用
    ┌─────────┼─────────┐
    ▼         ▼         ▼
┌───────┐ ┌───────┐ ┌───────┐
│Agent A│ │Agent B│ │Agent C│ ← 下属 Agent（也可能是 LLM，也可能是函数）
└───────┘ └───────┘ └───────┘

• 主管 Agent 通过 LLM 决策「下一步让谁干」
• 下属 Agent 对主管来说就是「一个 Tool」
• 典型代表：AutoGPT、MetaGPT、LangGraph Supervisor

2.2 Pipeline 串行（流水线）

Input → Agent A → Agent B → Agent C → Output

• 每个 Agent 是确定性的处理节点
• 上游输出直接喂下游，不需要 LLM 决策
• 适合结构化数据处理：解析 → 分析 → 写报告

2.3 Peer-to-Peer（消息总线）

┌──────┐  ↔  ┌──────┐  ↔  ┌──────┐
│Agent A│    │Agent B│    │Agent C│
└──────┘     └──────┘     └──────┘
     ↘          ↓           ↙
          ┌─────────┐
          │ MsgHub  │ ← 共享消息池
          └─────────┘

• Agent 之间对等通信，无中心
• 通过消息总线（如 AgentScope 的 MsgHub）广播或点对点
• 适合多 Agent 共同决策的场景：协商、辩论、投票

---

三、AgentScope 实战：消息驱动的多 Agent

AgentScope （阿里达摩院出品）走的是消息驱动路线 —— Agent 之间靠 Msg 对象通信，对等且解耦。

3.1 模式 1：Orchestrator（Agent 当 Tool）

from agentscope.agent import ReActAgent
from agentscope.tool import Toolkit
from agentscope.model import DashScopeChatModel

# 两个子 Agent
researcher = ReActAgent(
    name="researcher",
    sys_prompt="你是一个研究员，擅长联网调研和资料整理。",
    model=DashScopeChatModel("qwen-max"),
)
coder = ReActAgent(
    name="coder",
    sys_prompt="你是一个工程师，擅长写 Python 代码。",
    model=DashScopeChatModel("qwen-max"),
)

# 关键：把 Agent 注册成 Tool
toolkit = Toolkit()
toolkit.register_agent_tool(researcher)
toolkit.register_agent_tool(coder)

# 主管 Agent
orchestrator = ReActAgent(
    name="orchestrator",
    sys_prompt=(
        "你是项目主管。复杂问题先调研（researcher），"
        "再让 coder 实现，最后整合输出。"
    ),
    model=DashScopeChatModel("qwen-max"),
    toolkit=toolkit,
)

# 跑起来
msg = Msg("user", "调研 A2A 协议并写一个最小示例", role="user")
result = await orchestrator(msg)
print(result.content)

关键 API：toolkit.register_agent_tool(agent) —— 这一行把整个 Agent 包成了一个可调用的 Tool。

3.2 模式 2：Pipeline（流水线）

from agentscope.pipeline import sequential_pipeline

# 三个 Agent 串成流水线
result = await sequential_pipeline(
    agents=[parser_agent, analyst_agent, writer_agent],
    msg=Msg("user", "分析上季度销售数据并写报告", role="user"),
)
# data flow: parser → analyst → writer

sequential_pipeline 把上游输出直接喂下游，不经过 LLM 决策 —— 适合流程固定的场景。

3.3 模式 3：MsgHub（对等多 Agent 协作）

这是 AgentScope 最独特的设计 —— 多个 Agent 进同一个聊天室，共享上下文：

from agentscope.hub import MsgHub
from agentscope.pipeline import fanout_pipeline

# 三个 Agent 加入同一个 MsgHub
hub = MsgHub(
    participants=[planner_agent, critic_agent, executor_agent],
    announcement=Msg(
        "system",
        "用户需求：设计一个分布式任务调度系统。",
        role="system",
    ),
)

# fanout：让多个 Agent 并行独立响应同一个 msg
await fanout_pipeline(
    agents=[planner_agent, critic_agent, executor_agent],
    msg=Msg("user", "请各自给出方案", role="user"),
    hub=hub,
)

MsgHub 的妙处在于：每个 Agent 看到的对话历史是共享的（其他 Agent 说的话它也能看到），这天然适合辩论、协商场景：

# 典型用法：让两个 Agent 来回辩论 N 轮
async with MsgHub(participants=[debater_a, debater_b]) as hub:
    for round in range(3):
        await debater_a(hub.last_msg)  # A 先说
        await debater_b(hub.last_msg)  # B 反驳
    # 结束，用 judge_agent 给最终结论

---

四、LangChain/LangGraph 实战：图编排

LangGraph 是 LangChain 团队对多 Agent 协作的官方解法 —— 把协作建模成有向状态图。

4.1 核心抽象：StateGraph

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator

class State(TypedDict):
    messages: Annotated[list, operator.add]   # 累加
    draft: str
    revision_count: int

def researcher_node(state: State):
    """节点 1：研究员"""
    research = do_research(state["messages"][-1].content)
    return {
        "messages": [("ai", f"调研结果：{research}")],
        "revision_count": state.get("revision_count", 0) + 1,
    }

def writer_node(state: State):
    """节点 2：写作者"""
    draft = write_article(state["messages"])
    return {"draft": draft}

def should_continue(state: State) -> str:
    """条件边：决定下一步走哪个节点"""
    if state["revision_count"] >= 2:
        return "end"
    return "writer"

# 组装图
graph = StateGraph(State)
graph.add_node("researcher", researcher_node)
graph.add_node("writer", writer_node)

graph.add_edge("researcher", "writer")
graph.add_conditional_edges(
    "writer",
    should_continue,
    {"end": END, "writer": "writer"},   # 循环
)
graph.set_entry_point("researcher")

app = graph.compile()

# 跑起来
result = app.invoke({"messages": [("user", "写一篇 A2A 文章")]})

核心概念：

• State：所有节点共享的字典（带类型注解，LangGraph 自动合并）
• Node：一个函数，接收 State、返回 State 的部分字段
• Edge：节点之间的连接，可以是固定边也可以是条件边

4.2 LangGraph 的杀手锏：可视化 + Checkpoint

# 1. 调试时画图
from langgraph.visualization import draw_mermaid
print(draw_mermaid(app.get_graph()))

# 2. 关键能力：Human-in-the-loop（人工介入）
from langgraph.checkpoint.memory import MemorySaver

app = graph.compile(
    checkpointer=MemorySaver(),
    interrupt_before=["writer"],   # ← 在 writer 之前暂停，等人审批
)

config = {"configurable": {"thread_id": "1"}}
result = app.invoke({"messages": [("user", "写")]}, config=config)

# 此时 result 是空的，程序停了；人在 UI 上点确认后：
app.invoke(None, config=config)  # 继续

4.3 Supervisor 模式（一行起手式）

如果你就想用主管模式，不想手画图：

from langgraph.prebuilt import create_supervisor

supervisor = create_supervisor(
    agents=[research_agent, code_agent],
    model=llm,
    prompt="你是主管，分派任务给 researcher 或 coder",
)
app = supervisor.compile()
result = app.invoke({"messages": [("user", "调研 A2A 并写 demo")]})

create_supervisor 内部就是个有向图：supervisor 节点 + N 个 worker 节点 + 条件边。

---

五、横向对比

维度	AgentScope	LangGraph
核心抽象	Agent + Msg + MsgHub	Node + Edge + State
多 Agent 通信方式	直接传 Msg 对象；MsgHub 共享上下文	共享 State dict（自动合并）
编排方式	Pipeline / ReAct / A2A 服务	StateGraph（图）
可视化	内置 log 打印	LangGraph Studio（强）
Human-in-the-loop	自己实现	一行配置 interrupt_before
长任务 / 异步	原生 A2A 协议支持 task 生命周期	需要 checkpoint + 外部存储自己接
跨服务 / 跨进程	内置 A2A Server / Client	自己实现；LangGraph Platform 提供托管
中文友好度	高（阿里达摩院 + 中文文档）	一般（英文为主）
学习曲线	平（API 直观）	陡（图 + State 设计需要经验）
适合场景	中文业务、多 Agent 协商、想快速上手	复杂工作流、需要可视化、需要 HITL

我的选型经验：

• 场景简单 / 想快跑通：AgentScope，MsgHub 写起来真的很爽
• 流程复杂 / 要可视化 / 要人工介入：LangGraph，图的表达力强很多
• 跨语言 / 跨团队 / 跨厂商：A2A 协议（两个框架都支持或可以接入）

---

六、进阶：跨框架互操作

最后讲一个真实场景 —— 怎么把 LangChain Agent 接进 AgentScope 系统（或反过来）。

答案就是 A2A 协议：把其中一个框架的 Agent 暴露成 A2A 服务，另一个框架当 A2A 客户端调用。

6.1 AgentScope 暴露 A2A 服务

from agentscope.a2a import A2AServer

server = A2AServer(
    agent=researcher,
    port=8001,
    agent_card={
        "name": "Researcher Agent",
        "description": "调研专家",
        "skills": [{"id": "web-search", "description": "联网调研"}],
    },
)
await server.start()

6.2 LangGraph 当 A2A 客户端调用

import httpx
import json

async def call_researcher_agent(query: str) -> str:
    """通过 A2A 协议调用 AgentScope 的 researcher"""
    async with httpx.AsyncClient() as client:
        # 1. 拿 Agent Card（可选，缓存也行）
        card = await client.get("http://localhost:8001/.well-known/agent.json")

        # 2. 提交 task
        resp = await client.post(
            "http://localhost:8001/tasks/send",
            json={
                "skill": "web-search",
                "input": query,
            },
        )
        task_id = resp.json()["task_id"]

        # 3. SSE 流式拿结果
        async with client.stream(
            "GET", f"http://localhost:8001/tasks/{task_id}/events"
        ) as stream:
            final = ""
            async for line in stream.aiter_lines():
                if line.startswith("data: "):
                    event = json.loads(line[6:])
                    if event["state"] == "completed":
                        final = event["artifacts"][0]["text"]
            return final

# 包成 LangGraph 节点
def researcher_node(state):
    result = asyncio.run(call_researcher_agent(state["query"]))
    return {"research": result}

这样 LangGraph 的工作流就能无缝调用 AgentScope 的 Agent，反之亦然。

---

七、几个反直觉的提醒

最后是几个容易踩的坑：

7.1 多 Agent ≠ 更好的结果

多数情况下，一个强 Agent + 好工具，比 3 个弱 Agent 协作更靠谱。

多 Agent 引入额外的 LLM 调用（主管决策 + 通信开销），token 成本可能涨 2-3 倍，但结果质量往往提升不到 10%。只有当任务天然可拆（调研 vs 写代码），多 Agent 才有意义。

7.2 MsgHub 不是银弹

MsgHub 的「共享上下文」听起来很美，但所有 Agent 都要把整段历史塞进 prompt。3 个 Agent 还行，10 个 Agent 就爆 token 了。生产环境要么做上下文摘要，要么用 LangGraph 的显式 State 传递。

7.3 StateGraph 的 State 设计是核心

LangGraph 看上去是画图，实际上难点在State schema 设计。State 字段太多 → 每次 LLM 调用都浪费；State 字段太少 → 节点之间传不动。建议从 3-5 个字段开始，按需演进。

7.4 不要为了 A2A 而 A2A

A2A 协议适合跨组织、跨语言、跨厂商的 Agent 互操作。如果你的所有 Agent 都是同一团队、同一技术栈、同一进程写的，上 A2A 反而增加复杂度（多一层网络调用、多一份序列化成本）。先评估是否真的需要跨进程。

---

八、总结

层级	问题	解法
跨服务 / 跨语言	不同团队写的 Agent 怎么互相调用	A2A 协议（HTTP + JSON-RPC + SSE）
进程内多 Agent 协作	同一份代码里多个 Agent 怎么组织	三种模式：Orchestrator / Pipeline / Peer-to-Peer
特定框架落地	用哪个框架	AgentScope（消息驱动）/ LangGraph（图编排）

一句话总结：A2A 是网络层的协议，服务内部 Agent 协作是架构层的模式。两者正交 —— 一个项目可以同时用 LangGraph 编排内部流程，又把 LangGraph 暴露成 A2A 服务给其他团队用。

---

参考

• Google A2A Protocol Spec
• AgentScope 官方文档
• LangGraph 官方文档
• LangGraph Multi-Agent Patterns

---

评论区聊聊你的多 Agent 实践 —— 你用哪个框架？踩过什么坑？