第3章：多 Agent 编排模式 —— Supervisor / Swarm / Hierarchical / Mesh / Pipeline

单 agent 能解决的问题有限——做客服需要”销售 + 售后 + 风控”配合,做代码需要”PM + 架构师 + 编码 + 测试”分工。多 agent 怎么协作?谁拍板、谁汇报、谁兜底?本章把 5 种主流编排模式(Supervisor / Swarm / Hierarchical / Mesh / Pipeline)拆开讲透,配 ASCII 拓扑图、适用场景对照、LangGraph 官方实现剖析,以及”50+ agent 时为什么必须层级化”的工程经验。

1. 多 Agent 系统的三个核心问题

在选编排模式前,先问自己:

问题	影响选型
谁决定下一步谁干?	中心化(Supervisor)还是去中心化(Swarm)
agent 数量?	< 5(任意)/ 5-20(Hierarchical)/ 50+(必须 Hierarchical)
协作模式?	顺序(Pipeline)/ 并行(Mesh)/ 自由(Swarm)
可观测性要求?	强审计→Supervisor;弱→Swarm
延迟敏感?	低延迟→Swarm(skip 中央)/ 高吞吐→Pipeline

🌟 行业事实:90% 生产系统用 Supervisor 或 Hierarchical,Swarm 主要在特殊场景(博弈、emergent behavior、对延迟极致敏感)。

2. Supervisor:中心化路由

2.1 拓扑

                ┌─────────────────┐
                │   Supervisor     │
                │  (LLM router)    │
                └─┬─────┬─────┬───┘
                  │     │     │
        ┌─────────┘     │     └─────────┐
        ▼               ▼               ▼
   ┌────────┐      ┌────────┐      ┌────────┐
   │Worker A│      │Worker B│      │Worker C│
   │(Sales) │      │ (CS)   │      │ (Risk) │
   └────┬───┘      └────┬───┘      └────┬───┘
        │               │               │
        └───────┬───────┴───────────────┘
                ▼
            返回 Supervisor → 决定下一步 / 结束

2.2 工作流程

用户消息到 Supervisor
Supervisor LLM 看 query + worker 列表 → 决定 route 给谁
Worker 执行并返回结果
Supervisor 看结果决定:再 route 一次 / 终止 / 补充信息

2.3 LangGraph 代码

LangChain 出了官方包 langgraph-supervisor:

from langgraph_supervisor import create_supervisor
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI

sales_agent = create_react_agent(
    ChatOpenAI(model="gpt-4o"),
    tools=[query_product, create_order],
    name="sales",
    prompt="你是销售 agent,处理产品咨询和下单",
)
cs_agent = create_react_agent(
    ChatOpenAI(model="gpt-4o"),
    tools=[query_order, refund],
    name="cs",
    prompt="你是售后 agent,处理订单查询和退款",
)
risk_agent = create_react_agent(
    ChatOpenAI(model="gpt-4o-mini"),
    tools=[fraud_check],
    name="risk",
    prompt="你是风控 agent,判断请求是否可疑",
)

supervisor = create_supervisor(
    [sales_agent, cs_agent, risk_agent],
    model=ChatOpenAI(model="gpt-4o"),
    prompt=(
        "你是主管,根据用户请求 route 给合适的 worker。"
        "如果涉及金额 >¥1000,先 route 给 risk,通过后再继续。"
    ),
)
app = supervisor.compile()

2.4 优劣

✅ 优势:

决策集中,可审计
worker 可独立开发、独立测试
安全策略(如 risk check)容易实现

❌ 劣势:

每次 route 多一次 LLM 调用 → 延迟翻倍
Supervisor 是单点(故障 / 偏见)
大规模(50+ worker)时 Supervisor prompt 爆炸

2.5 适用

客服 / 工单系统(强审计)
安全敏感场景(每次都要 risk check)
团队规模 < 20 个 worker

3. Swarm:去中心化 Handoff

3.1 拓扑

   ┌────────┐ handoff ┌────────┐ handoff ┌────────┐
   │Agent A │────────►│Agent B │────────►│Agent C │
   └────┬───┘         └────┬───┘         └────┬───┘
        │ handoff          │ handoff          │
        └────────┬─────────┘                  │
                 ▼                            ▼
            ┌────────┐               ┌────────┐
            │Agent D │               │Agent E │
            └────────┘               └────────┘

   没有中心,任意 agent 可以 handoff 给任意 agent

3.2 OpenAI Swarm / Agents SDK

OpenAI 提出 Swarm 范式,极简代码:

from agents import Agent, handoff, Runner

triage = Agent(
    name="Triage",
    instructions="判断用户意图,handoff 给对应 agent",
    handoffs=[],  # 后面填
)
sales = Agent(
    name="Sales",
    instructions="处理销售",
    handoffs=[],
)
cs = Agent(
    name="CS",
    instructions="处理售后",
    handoffs=[],
)

# Handoff 关系是有向图
triage.handoffs = [sales, cs]
sales.handoffs = [triage, cs]   # 销售完成可转给 CS
cs.handoffs = [sales]           # 售后发现问题可转回销售

result = Runner.run_sync(triage, "我要退订单 #123")

3.3 优劣

✅ 优势:

低延迟(skip 中央)
容易 scale(没有单点)
适合”涌现”行为研究

❌ 劣势:

难审计(谁决定 handoff 了?)
容易死循环(A→B→A→B…)
调试地狱

3.4 适用

实时博弈、群体智能仿真
简单 handoff 流(triage + 几个专家)
对延迟极致敏感

🌟 OpenAI Swarm 现已并入 Agents SDK(2025-Q1),作为推荐范式。

4. Hierarchical:层级分治

4.1 拓扑

               ┌────────────────┐
               │   CEO Agent     │
               └─┬────────────┬──┘
                 │            │
        ┌────────┘            └────────┐
        ▼                              ▼
   ┌─────────┐                    ┌─────────┐
   │ Manager │                    │ Manager │
   │   A     │                    │   B     │
   └─┬───┬───┘                    └─┬───┬───┘
     │   │                          │   │
   ┌─▼┐ ┌▼─┐                      ┌─▼┐ ┌▼─┐
   │W1│ │W2│                      │W3│ │W4│
   └──┘ └──┘                      └──┘ └──┘

4.2 思想

把 Supervisor 分层——CEO(总调度)→ Manager(领域专家)→ Worker(具体执行)。每层只跟相邻层通信。

4.3 何时必上 Hierarchical

IBM 研究:50+ agent 必须层级化,否则单 supervisor 的 prompt 容纳不下,LLM 路由准确率断崖式下降。

4.4 LangGraph 实现

LangGraph 支持 subgraph —— 上层 graph 把下层 graph 作为节点。每个 manager 是一个 subgraph:

manager_a_graph = StateGraph(ManagerAState)
# ... 添加 worker 1, worker 2 节点和 supervisor 路由 ...
manager_a_compiled = manager_a_graph.compile()

ceo_graph = StateGraph(CEOState)
ceo_graph.add_node("manager_a", manager_a_compiled)   # ← subgraph as node
ceo_graph.add_node("manager_b", manager_b_compiled)
ceo_graph.add_node("ceo_router", ceo_router_fn)
ceo_graph.set_entry_point("ceo_router")
ceo_graph.add_conditional_edges(
    "ceo_router",
    lambda s: s["next"],
    {"a": "manager_a", "b": "manager_b"},
)

4.5 优劣

✅ 优势:

唯一可 scale 到 50+ agent 的方案
每层 prompt 简洁(只看相邻层)
责任清晰,故障隔离

❌ 劣势:

层级 hop 增加延迟
跨层信息传递可能失真
设计难度高

4.6 适用

大型企业 agent 系统(销售 + 售后 + 风控 + 营销 + …)
复杂研发任务(架构 + 多个开发 + 测试 + 上线)
任何 worker 数量 ≥ 20 的系统

5. Mesh:对等网络

5.1 拓扑

   ┌─A─┐──┌─B─┐
   │   ╳   │   │     A↔B↔C↔D 任意两两可通信
   │   ╳   │   │     共享 blackboard / event bus
   ┌─C─┘──┌─D─┘

5.2 思想

所有 agent 对等,通过共享 blackboard / event bus 异步交换信息。每个 agent 监听感兴趣的事件,按自己的角色响应。

5.3 实现

import asyncio
from collections import defaultdict

class EventBus:
    def __init__(self):
        self.subscribers = defaultdict(list)
    
    def subscribe(self, event_type, agent):
        self.subscribers[event_type].append(agent)
    
    async def publish(self, event):
        await asyncio.gather(*[
            agent.handle(event) for agent in self.subscribers[event.type]
        ])

bus = EventBus()

class AnalystAgent:
    async def handle(self, event):
        if event.type == "raw_data":
            analysis = await llm.analyze(event.data)
            await bus.publish(Event("analysis_done", analysis))

class WriterAgent:
    async def handle(self, event):
        if event.type == "analysis_done":
            report = await llm.write_report(event.data)
            await bus.publish(Event("report_ready", report))

bus.subscribe("raw_data", AnalystAgent())
bus.subscribe("analysis_done", WriterAgent())

5.4 适用

事件驱动场景(实时数据流)
多个独立 agent 同时关注一个共享状态
通常和 Kafka / Redis Streams / NATS 集成

不适合”对话式” agent。

6. Pipeline:流水线

6.1 拓扑

   ┌─────┐    ┌─────┐    ┌─────┐    ┌─────┐
   │ A   │───►│ B   │───►│ C   │───►│ D   │
   │ETL  │    │Norm │    │LLM  │    │Save │
   └─────┘    └─────┘    └─────┘    └─────┘

6.2 思想

deterministic 顺序——每个 agent 处理上游产出,产出给下游。最像传统数据流水线。

6.3 适用

数据处理(ETL → 清洗 → 分析 → 入库)
报告生成(收集 → 分析 → 写作 → 校对)
内容审核(初审 → AI 检测 → 人工复核)

6.4 优劣

✅ 优势:可预测、易测试、易并行(同步多 batch) ❌ 劣势:不灵活,适应不了动态需求

6.5 LangGraph 实现

最简——线性 add_edge:

graph.add_edge("a", "b")
graph.add_edge("b", "c")
graph.add_edge("c", "d")

Pipeline 是 Graph 的特例(无分支)。

7. Agents-as-Tools 模式

7.1 思想

把一个 agent 包装成 tool,让另一个 agent 调用——既保留模块化,又比 Supervisor route 简单:

@tool
def consult_legal_agent(question: str) -> str:
    """咨询法务 agent 关于合规问题。"""
    legal_agent = create_react_agent(...)
    result = legal_agent.invoke({"messages": [("user", question)]})
    return result["messages"][-1].content

main_agent = create_react_agent(
    llm,
    tools=[search, calculator, consult_legal_agent],
)

7.2 vs Handoff(Swarm)/ Route(Supervisor)

模式	子 agent 是否独立?	控制权返回?
Tool	独立 sub-execution	✅ 立即返回
Handoff(Swarm)	接管整个会话	❌ 不返回
Route(Supervisor)	临时分派	✅ 返回 supervisor

🍎 AWS Prescriptive Guidance 把 Agents-as-Tools 列为”最简多 agent 模式”,非常推荐用于 < 5 agents 的简单场景。

8. 混合模式与选型

8.1 真实生产几乎都是混合

例子	混合模式
电商客服系统	Hierarchical(CEO=主管,Manager=销售/售后/物流,Worker=具体技能) + 每个 Worker 内部 ReAct + 关键节点用 Saga
代码 agent(Devin)	Pipeline(规划→编码→测试→部署) + 编码阶段用 ReAct + 测试失败 Reflexion
数据分析 agent	Mesh(分析师/可视化/报告各自订阅事件)+ 主流程 Pipeline
多 LLM router	Supervisor(选 LLM)+ 每个 LLM 内 Tool-as-Thought

8.2 选型决策树

agent 数量?
│
├─ < 3 个 → Tool 模式(Agents-as-Tools)就够
│
├─ 3-10 个 → Supervisor(LangGraph supervisor / OpenAI Triage)
│   └─ 强动态、低延迟? → Swarm(OpenAI Agents handoff)
│
├─ 10-50 个 → Hierarchical(2-3 层)
│
├─ 50+ → Hierarchical(3+ 层)+ Pipeline 子结构
│
└─ 事件驱动场景 → Mesh + 事件总线

8.3 设计 checklist

控制流是中心化还是去中心化?
谁来兜底”没有 agent 能处理”?(default fallback)
如何防死循环?(max depth / cycle detection)
跨 agent 状态如何共享?(LangGraph shared state / event bus)
失败如何传播?(Saga 第 6 章详讲)
怎么测试?(每个 agent 单元测 + 端到端集成)
怎么观测?(每个 handoff/route 都有 trace,第 8 章)

✅ 自我检验清单

5 种模式拓扑:能默写 Supervisor / Swarm / Hierarchical / Mesh / Pipeline 的 ASCII 图
3 个核心问题:能解释每种模式如何回答”谁拍板、agent 数量、协作模式”
Supervisor 优劣:能讲清”集中决策 vs 多一次 LLM 延迟”的权衡
Swarm 死循环:能解释为什么 Swarm 容易死循环,以及怎么防
50+ agent 必上 Hierarchical:能解释原因(prompt 容量、路由准确率)
Agents-as-Tools:能比较它和 Handoff、Route 的区别
混合模式:能为 3 个具体业务设计混合编排方案
LangGraph supervisor 代码:能不查资料写一个 3-worker supervisor
subgraph:能用 subgraph 实现 2 层 hierarchical 结构

📚 参考资料

框架与代码

LangGraph Multi-Agent:官方文档
langgraph-supervisor:github.com/langchain-ai/langgraph-supervisor-py
OpenAI Agents SDK / Swarm:openai.github.io/openai-agents-python
CrewAI:role-based 编排,github.com/joaomdmoura/crewAI
Microsoft AutoGen:conversation-based,microsoft.github.io/autogen
Swarms:enterprise-grade,github.com/kyegomez/swarms

模式综述

Agent Orchestration Patterns: Swarm vs Mesh vs Hierarchical vs Pipeline:GuruSup
Swarm vs. Supervisor: Multi-Agent Architecture Guide:Augment Code
Build Multi-Agent Systems Using the Agents-as-Tools Pattern:AWS DEV
Multi-Agent Orchestration in LangGraph: Supervisor vs Swarm:DEV

学术

MetaGPT —— 多 agent 软件公司模拟
CAMEL —— 角色扮演多 agent 协作
AgentVerse —— 多 agent 涌现行为研究

搜索