🤝 构建 Agent 实践指南 — 多 Agent 协作

消息传递 · 状态共享 · 任务编排 — 从 AutoGen/LangGraph/CrewAI/MetaGPT/CAMEL 源码提炼的多 Agent 协作方法论

1 为什么需要多 Agent？单 Agent 的极限

从 20 个 Agent 项目的源码分析中，我们发现单 Agent 系统面临三个根本性瓶颈：

2 协作模式全景图：6 种编排范式

从源码中提炼出 6 种多 Agent 编排模式，每种都有其最佳适用场景：

3 消息传递：多 Agent 的血液循环

多 Agent 系统的核心是 消息传递。不同框架的消息模型差异巨大，直接决定了系统的表达能力上限。

3.1 AutoGen：Actor Model 双模式通信

AutoGen Core 实现了两种消息传递模式，覆盖从简单到复杂的所有场景：

# AutoGen Core: RPC 通信
result = await runtime.send_message(
    message=TextMessage(content="分析这段代码", source="reviewer"),
    recipient=AgentId("code_analyst")
)

# AutoGen Core: 发布/订阅
await runtime.publish_message(
    message=EventMessage(event="task_completed", result=summary),
    topic_id=TopicId("notifications")
)

3.2 MetaGPT：Watch-Publish 精确路由

MetaGPT 的消息路由基于 cause_by 字段——每条消息携带"由哪个 Action 产生"的标记，Role 通过 watch 列表声明关心哪些 Action 的输出：

# MetaGPT: 消息路由核心
class Role(BaseRole):
    rc: RoleContext  # 包含 watch: set[str]

    async def _observe(self) -> int:
        news = self.rc.msg_buffer.pop_all()
        # 只保留 cause_by 在 watch 列表中的消息
        self.rc.news = [
            n for n in news
            if n.cause_by in self.rc.watch or self.name in n.send_to
        ]
        return len(self.rc.news)

# Environment 路由：根据 member_addrs 和 is_send_to() 推送到对应 Role 的 buffer
# Role 的 msg_buffer 是私有的，避免全局广播噪音

3.3 LangGraph：Channel + Reducer 状态通信

LangGraph 不使用"消息"概念——Agent 通过读写 Channel（共享状态键）通信：

# LangGraph: 通过共享 State 通信
class State(TypedDict):
    messages: Annotated[list, add_messages]  # Reducer: 追加而非覆盖
    next_agent: str
    research_results: Annotated[list, operator.add]  # 并行结果合并

# 并行 Worker 写入 research_results
# Reducer (operator.add) 自动合并两个 Worker 的输出
# 下游节点读到的就是合并后的完整结果

4 状态共享：从共享内存到事件溯源

多 Agent 协作的另一个核心问题：状态存储在哪里？谁来读写？

4.1 LangGraph 的检查点：状态时间旅行

LangGraph 拥有所有框架中最强的状态管理——检查点系统允许回溯任意历史状态：

# LangGraph: 检查点回溯
from langgraph.checkpoint.sqlite import SqliteSaver

checkpointer = SqliteSaver.from_conn_string("db.sqlite")
graph = builder.compile(checkpointer=checkpointer)

# 执行并自动保存检查点
result = graph.invoke({"messages": [...]}, config={"configurable": {"thread_id": "1"}})

# 查看所有历史状态
states = list(graph.get_state_history(config))
for state in states:
    print(f"Step {state.metadata['step']}: next={state.next}")

# 回滚到特定检查点
graph.invoke(None, config, checkpoint_id=states[3].config["configurable"]["checkpoint_id"])

# 修改状态后继续
graph.invoke(Command(update={"approved": True}), config)

4.2 AutoGen MagenticOne：Ledger 状态模型

MagenticOne 用结构化的 Ledger（账本） 管理协作状态，是最接近"项目管理"的编排：

# MagenticOne 双账本
class LedgerEntry(BaseModel):
    is_request_completed: bool
    is_in_progress: bool
    is_step_complete: bool
    next_speaker: str | None
    instruction_to_next_speaker: str | None

# 编排循环：
# 1. 初始化 Task Ledger (facts + plan)
# 2. 每轮：
#    a. 更新 Facts Ledger（基于新消息）
#    b. 更新 Plan Ledger（基于新事实）
#    c. 查询 Progress Ledger → 确定 next_speaker + instruction
#    d. 路由消息给 next_speaker
#    e. 检查 stall → 连续 stall >= max_stalls → 重新规划
#    f. 检查完成 → is_request_completed → 生成最终答案

5 任务编排：5 大框架的编排实现对比

5.1 AutoGen：5 种群组编排模式

AutoGen 提供最丰富的编排选项——从最简单到最复杂，渐进式覆盖：

# AutoGen: SelectorGroupChat 三层覆盖
team = SelectorGroupChat(
    participants=[writer, reviewer, editor],
    model_client=openai_client,
    selector_func=lambda messages: "reviewer" if "审查" in messages[-1].content else None,
    # selector_func 返回非 None → 覆盖 LLM 选择（确定性规则）
    # selector_func 返回 None → 走 LLM 选择
    candidate_func=lambda messages: ["writer", "editor"],  # 过滤候选
    allow_repeated_speaker=False,  # 不允许连续同一发言者
    max_selector_attempts=3,  # LLM 选择失败重试次数
)

5.2 LangGraph：StateGraph 图执行

LangGraph 的编排基于 Pregel 超步模型——所有节点在同一个超步并行执行，超步结束时通过 Reducer 合并结果：

# LangGraph: Supervisor 编排
from langgraph.graph import StateGraph, START, END

builder = StateGraph(State)

# 添加节点
builder.add_node("supervisor", supervisor_node)
builder.add_node("researcher", researcher_node)
builder.add_node("coder", coder_node)

# 添加边
builder.add_edge(START, "supervisor")
builder.add_conditional_edges(
    "supervisor",
    lambda state: state["next_agent"],  # 动态路由
    {"researcher": "researcher", "coder": "coder"}
)
builder.add_edge("researcher", "supervisor")
builder.add_edge("coder", "supervisor")
builder.add_conditional_edges("supervisor", should_end, {True: END, False: "supervisor"})

graph = builder.compile(checkpointer=SqliteSaver.from_conn_string("db.sqlite"))

5.3 CrewAI：Process-Driven + Flow

CrewAI 的编排分两层：Process 枚举控制基本模式，Flow 装饰器处理复杂流程：

# CrewAI: 基本模式
crew = Crew(
    agents=[researcher, writer, reviewer],
    tasks=[research_task, write_task, review_task],
    process=Process.sequential,  # 或 Process.hierarchical
)

# Hierarchical 模式：自动创建 Manager Agent
# Manager 获得 DelegateWorkTool + AskQuestionTool
# Manager 决定谁做什么

# Flow 系统：高级编排
class ResearchFlow(Flow):
    @start()
    async def plan(self, ctx):
        return {"plan": await planner.run(ctx)}

    @listen("plan")
    async def research(self, ctx):
        return {"research": await researcher.run(ctx)}

    @listen("research")
    async def write(self, ctx):
        return {"article": await writer.run(ctx)}

    @router("plan")
    async def route(self, ctx):
        if ctx.data["plan"]["needs_research"]:
            return "research"
        return "write"

5.4 MetaGPT：SOP 流水线

MetaGPT 的编排是 最不灵活但最可靠 的——SOP 一旦定义，角色就按流水线自动推进：

# MetaGPT: 软件公司 SOP
team = Team()
team.hire(
    ProductManager(),    # watch: {} → 触发 WritePRD
    Architect(),         # watch: {WritePRD} → 触发 WriteDesign
    ProjectManager(),    # watch: {WriteDesign} → 触发 WriteTasks
    Engineer(),          # watch: {WriteTasks} → 触发 WriteCode
    QaEngineer(),        # watch: {WriteCode} → 触发 WriteTest
)
team.run_project("构建一个博客系统")
# 自动按 SOP 推进: PRD → 设计 → 任务 → 代码 → 测试

5.5 CAMEL：RolePlaying + Workforce

CAMEL 提供两种多 Agent 模式：

# CAMEL RolePlaying: 两个 Agent 对话
from camel.societies import RolePlaying

role_play = RolePlaying(
    assistant_role_name="Python 程序员",
    user_role_name="代码审查员",
    assistant_agent_kwargs={"model": "gpt-4"},
    user_agent_kwargs={"model": "gpt-4"},
)
# 自动生成角色提示词，两 Agent 交替对话

# CAMEL Workforce: 层级任务分解
from camel.societies import Workforce
workforce = Workforce("研究团队")
workforce.add_agent(researcher)
workforce.add_agent(writer)
result = await workforce.process_task(complex_task)
# 自动分解 → 分配 → 执行 → 汇总

6 Agent 切换：Handoff vs Router vs Watch

"下一个该谁说话？" 是多 Agent 编排的核心问题。三种主流方案：

6.1 AutoGen Handoff 深度解析

Handoff 将 Agent 切换建模为 工具调用——这是 AutoGen v0.5 最创新的设计：

# AutoGen: Handoff 使用
analyst = AssistantAgent(
    "analyst",
    model_client=openai_client,
    tools=[search_tool],
    handoffs=["writer", "reviewer"],  # 声明可切换到谁
)

# 模型推理时自动看到 transfer_to_writer 和 transfer_to_reviewer 工具
# 模型决定切换时，HandoffMessage 携带当前上下文给下一个 Agent

7 嵌套与层级：子 Agent 的组合艺术

7.1 AutoGen SocietyOfMind：Team-in-Agent

SocietyOfMindAgent 将一个 Team 包装为单个 Agent——实现两层信息压缩：

# AutoGen: SocietyOfMindAgent
inner_team = RoundRobinGroupChat([writer, reviewer])

writer_team = SocietyOfMindAgent(
    "writer_team",
    team=inner_team,
    model_client=openai_client,
    # 内层 Team 完成多轮写作-审稿
    # SocietyOfMind 用 LLM 总结为单条响应
)

# 外层 Team 只看到总结，不感知内部细节
outer_team = RoundRobinGroupChat([writer_team, translator])

7.2 LangGraph 子图组合

# LangGraph: 编译后的图作为节点
research_subgraph = research_builder.compile(checkpointer=...)
coding_subgraph = coding_builder.compile()

parent = StateGraph(ParentState)
parent.add_node("research", research_subgraph)  # 子图作为节点
parent.add_node("coding", coding_subgraph)
parent.add_edge("research", "coding")

# 子图内部状态通过 Command(graph=Command.PARENT) 上报
def research_done(state):
    return Command(
        update={"research_results": state["results"]},
        graph=Command.PARENT  # 上报给父图
    )

7.3 OpenClaw Sub-Agent：spawn/steer/kill/yield

OpenClaw 的子代理模型最接近操作系统进程管理：

# OpenClaw: 子代理生命周期
sessions_spawn(
    task="分析这份报告",
    runtime="subagent",
    context="isolated",  # 或 "fork" 继承父会话上下文
    mode="run",          # 后台执行
)

# 管理操作
subagents(action="list")    # 查看所有子代理
subagents(action="steer", target="sub-123", message="加上对比分析")  # 动态干预
subagents(action="kill", target="sub-123")  # 终止子代理
sessions_yield()            # 等待子代理结果

8 并行执行：Map-Reduce 与异步扇出

8.1 LangGraph Send API：动态并行扇出

LangGraph 的 Send 是所有框架中唯一的 运行时动态并行 机制：

# LangGraph: Send 动态并行
from langgraph.types import Send

def route_to_workers(state):
    """运行时才知道有多少任务——动态扇出"""
    return [
        Send("worker_node", {"task": task})
        for task in state["tasks"]  # tasks 数量在运行时确定
    ]

builder.add_conditional_edges("planner", route_to_workers)

# 所有 worker 完成后，结果通过 Reducer (operator.add) 合并
# 然后继续执行后续节点

8.2 AutoGen 并行：异步 Task + Team

# AutoGen: 异步任务 (CrewAI 也有类似设计)
# 多个 Team 可以并发运行
import asyncio

results = await asyncio.gather(
    research_team.run(task="研究市场趋势"),
    coding_team.run(task="开发原型"),
    design_team.run(task="设计 UI"),
)
# 三个 Team 完全独立运行

8.3 GPT Researcher：并行子话题研究

GPT Researcher 的多 Agent 模式用 LangGraph 实现并行研究：

# GPT Researcher: ChiefEditorAgent 编排
# 1. Editor 将查询拆分为子话题
# 2. 每个子话题分配给独立的 Researcher Agent
# 3. Researcher 并行搜索、抓取、摘要
# 4. Reviewer 检查质量 → Reviser 修订
# 5. 最终汇总为完整报告

9 终止与收敛：何时结束协作？

多 Agent 协作最容易被忽视的问题：怎么知道做完了？

# AutoGen: 组合终止条件
from autogen_agentchat.conditions import (
    TextMentionTermination,
    MaxMessageTermination,
    TimeoutTermination,
    FunctionalTermination,
)

termination = TextMentionTermination("TERMINATE") | MaxMessageTermination(20)
# 任一条件满足即终止

# MagenticOne: stall 检测
# 连续 max_stalls 轮没有进展 → 重新规划
# 重新规划也 stall → 放弃

10 实战选型指南：场景 → 框架映射

10.1 实战代码：用 LangGraph 构建研究团队

# 完整示例: LangGraph Supervisor + Send 并行研究
from langgraph.graph import StateGraph, START, END
from langgraph.types import Send, Command
from langgraph.checkpoint.sqlite import SqliteSaver
from typing import Annotated
import operator

class State(TypedDict):
    query: str
    subtopics: list[str]
    research_results: Annotated[list, operator.add]  # Reducer: 并行合并
    final_report: str

def supervisor(state):
    """分解查询为子话题"""
    subtopics = llm.invoke(f"将以下查询分解为3个子话题: {state['query']}")
    return {"subtopics": subtopics}

def route_to_researchers(state):
    """动态并行扇出"""
    return [Send("researcher", {"subtopic": t}) for t in state["subtopics"]]

def researcher(state):
    """单个子话题研究"""
    result = llm.invoke(f"研究: {state['subtopic']}")
    return {"research_results": [result]}

def synthesizer(state):
    """汇总所有研究结果"""
    report = llm.invoke(f"基于以下研究结果生成报告: {state['research_results']}")
    return {"final_report": report}

builder = StateGraph(State)
builder.add_node("supervisor", supervisor)
builder.add_node("researcher", researcher)
builder.add_node("synthesizer", synthesizer)

builder.add_edge(START, "supervisor")
builder.add_conditional_edges("supervisor", route_to_researchers)
builder.add_edge("researcher", "synthesizer")
builder.add_edge("synthesizer", END)

graph = builder.compile(checkpointer=SqliteSaver.from_conn_string("research.db"))
result = graph.invoke({"query": "2026年AI Agent技术趋势"})

10.2 实战代码：用 AutoGen Swarm 构建动态协作

# 完整示例: AutoGen Swarm 动态切换
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import Swarm
from autogen_agentchat.conditions import MaxMessageTermination, TextMentionTermination
from autogen_ext.models.openai import OpenAIChatCompletionClient

model_client = OpenAIChatCompletionClient(model="gpt-4")

# 定义 Agent + Handoff
researcher = AssistantAgent(
    "researcher",
    model_client=model_client,
    tools=[search_tool, scrape_tool],
    handoffs=["writer", "reviewer"],  # 可切换到 writer 或 reviewer
)

writer = AssistantAgent(
    "writer",
    model_client=model_client,
    handoffs=["reviewer"],  # 写完给 reviewer 审
)

reviewer = AssistantAgent(
    "reviewer",
    model_client=model_client,
    handoffs=["writer", "researcher"],  # 可打回给 writer 或补充研究
)

# Swarm 编排
team = Swarm(
    [researcher, writer, reviewer],
    termination_condition=TextMentionTermination("APPROVED") | MaxMessageTermination(15),
)

result = await team.run(task="研究 AI Agent 框架对比，写一篇 2000 字的综述文章")

11 反模式与常见坑

12 决策树：我该用多 Agent 吗？

🔗 相关链接

• AutoGen 深度源码分析 — Actor Model + 5 种群组编排 + Handoff + SocietyOfMind
• LangGraph 深度分析 — Pregel 执行模型 + Checkpoint + Send API + Command
• CrewAI 深度分析 — Process-Driven 编排 + Flow 装饰器
• MetaGPT 深度分析 — SOP Watch-Publish + ActionNode Review-Revise
• CAMEL 深度分析 — RolePlaying + Workforce
• GPT Researcher 深度分析 — 多 Agent 研究编排
• 主循环指南 — 单 Agent 内部循环 vs 多 Agent 外部编排
• 记忆系统指南 — 多 Agent 间的记忆共享与隔离
• 提示词指南 — 多 Agent 提示词组装与上下文管理

生成时间：2026-05-17 | 数据来源：AutoGen v0.5, LangGraph v0.3, CrewAI v0.8, MetaGPT v0.8, CAMEL v0.2, GPT Researcher, OpenClaw 源码分析