C3-12 多Agent协作

🎯 本节目标:说清 5 种多 Agent 协作模式及适用场景,用代码实现 Supervisor 和 Swarm 模式,根据任务复杂度选择合适的协作模式。

为什么/什么时候需要多 Agent?

单个 Agent 的局限:

❌ 上下文窗口有限:装不下所有工具描述(10+ 工具就让 LLM 选择困难)
❌ 角色混淆:既当"研究员"又当"程序员"还当"审核员"
❌ 任务复杂时容易跑偏:长任务中途"忘记"目标

解决方案:专业化分工,多个 Agent 协作完成。

五种核心协作模式

Python
模式一:串行链式(Sequential Chain)
  Agent A → Agent B → Agent C → 输出
  (适合:流水线任务)

模式二:路由分发(Router)
  Router Agent → Agent A(技术问题)
                → Agent B(销售问题)
                → Agent C(售后问题)
  (适合:多意图分流)

模式三:并行处理(Parallel)
  任务 → Agent A + Agent B + Agent C → 合并结果
  (适合:多角度分析、竞速)

模式四:Supervisor(中央协调者)★ 最常用
  Supervisor → 派发任务给 A/B/C → 收集结果 → 输出
  (适合:子agent任务分配，类似Claude code的 subagents 模式)

模式五:Swarm(对等接力)
  Agent A ←→ Agent B ←→ Agent C(自主决定移交)
  (适合:动态、对话式场景，类似Claude code的 AgentTeams 模式)

模式一:串行链式(Sequential Chain)

原理

Plain Text
输入 → Agent A(预处理)→ Agent B(分析)→ Agent C(输出)→ 最终结果

代码示例(LangChain Expression Language)

Python
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# 三个 Agent(其实是三条 Prompt Chain)
prompt1 = ChatPromptTemplate.from_template("把这段话翻译成英文:\n{input}")
prompt2 = ChatPromptTemplate.from_template("检查这段英文的语法错误并修正:\n{english}")
prompt3 = ChatPromptTemplate.from_template("把修正后的英文写进一段优雅的邮件格式:\n{corrected}")

# 串联成 Chain
chain = (
    prompt1 | llm | StrOutputParser() |
    prompt2 | llm | StrOutputParser() |
    prompt3 | llm
)

result = chain.invoke({"input": "我想预约下周二下午 3 点的会议"})
print(result)

适用场景

场景	为什么适合
文本处理流水线	每步输出是下一步输入
RAG:检索 → 阅读 → 回答	天然串行
代码:生成 → 审查 → 测试	严格顺序依赖

模式二:路由分发(Router)

原理

Plain Text
用户输入
    ↓
Router Agent(分类)
    ├─ 技术问题 → Tech Agent
    ├─ 订单问题 → Order Agent
    └─ 售后问题 → Support Agent

代码示例(LangGraph 实现)

Python
from typing import TypedDict, Annotated
from operator import add
from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
import os, json

class RouterState(TypedDict):
    messages: Annotated[list, add]
    route: str   # 路由决定

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# Router 节点
def router_node(state):
    response = llm.invoke([
        {"role": "system", "content":
         "你是路由 Agent。根据用户输入,输出 JSON:"
         "{'route': 'tech' | 'order' | 'support'}。"},
        *state["messages"]
    ])
    route = json.loads(response.content)["route"]
    return {"route": route}

# 各 Specialist 节点
def tech_node(state):
    response = llm.invoke([
        {"role": "system", "content": "你是技术支持专家。"},
        *state["messages"]
    ])
    return {"messages": state["messages"] + [response]}

def order_node(state):
    response = llm.invoke([
        {"role": "system", "content": "你是订单处理专家。"},
        *state["messages"]
    ])
    return {"messages": state["messages"] + [response]}

# 构建路由图
graph = StateGraph(RouterState)
graph.add_node("router", router_node)
graph.add_node("tech", tech_node)
graph.add_node("order", order_node)
graph.set_entry_point("router")
graph.add_conditional_edges(
    "router",
    lambda s: s["route"],   # 返回值直接对应节点名
    {"tech": "tech", "order": "order", "support": END}
)
graph.add_edge("tech", END)
graph.add_edge("order", END)

app = graph.compile()
result = app.invoke({
    "messages": [("user", "我的 API 调用一直报 401 错误")],
    "route": ""
})

模式三:并行处理(Parallel)

原理

Plain Text
任务
    ├─ Agent A(角度 1 分析)─┐
    ├─ Agent B(角度 2 分析)─┼─ 合并 Agent → 最终输出
    └─ Agent C(角度 3 分析)─┘

代码示例(LangGraph 并行)

Python
from langgraph.graph import StateGraph, START, END

class ParallelState(TypedDict):
    messages: list
    perspectives: list   # 收集各 Agent 输出

def analyst_a(state):
    # 从技术角度分析
    return {"perspectives": [{"agent": "tech", "analysis": "..."}]}

def analyst_b(state):
    # 从商业角度分析
    return {"perspectives": [{"agent": "biz", "analysis": "..."}]}

def merger(state):
    # 合并所有视角
    merged = "\n".join(
        f"【{p['agent']}】{p['analysis']}" for p in state["perspectives"]
    )
    return {"messages": state["messages"] + [
        {"role": "assistant", "content": merged}
    ]}

graph = StateGraph(ParallelState)
graph.add_node("a", analyst_a)
graph.add_node("b", analyst_b)
graph.add_node("merge", merger)

graph.set_entry_point("a")
# 关键:a 和 b 都指向 START 后的同一个 fan-out
graph.add_edge("a", "merge")
graph.add_edge("b", "merge")
graph.add_edge("merge", END)

app = graph.compile()

💡 竞速技巧:让 3 个 Agent 并行跑同一个任务,取最快/最好的结果。

模式四:Supervisor(中央协调者)★ 最常用

原理

graph TD U[用户任务] --> S[Supervisor 中央协调者] S -->|派遣| R[Researcher 研究专家] S -->|派遣| W[Writer 写作专家] S -->|派遣| C[Critic 审核专家] R -->|汇报| S W -->|汇报| S C -->|汇报| S S --> F[最终输出] style S fill:#FFE4B5 style R fill:#87CEEB style W fill:#90EE90 style C fill:#FFB6C1

特点:

Supervisor 决定每一步该调用谁
子 Agent 之间不直接通信(全部经过 Supervisor)
流程可预测,适合生产环境

代码示例(完整可运行)

Plain Text
from typing import TypedDict, Annotated
from operator import add
from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
import os, json

# ========== 状态定义 ==========
class SupervisorState(TypedDict):
    messages: Annotated[list, add]
    next_agent: str          # Supervisor 决定的下一个 Agent
    research_output: str
    draft_output: str
    final_output: str

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# ========== Supervisor 节点 ==========
def supervisor_node(state):
    """中央协调者:决定下一步该谁干活"""
    response = llm.invoke([
        {"role": "system", "content": """你是 Supervisor,协调以下 Agent:
        - researcher:信息收集和数据分析
        - writer:内容创作和报告撰写
        - critic:审核和提出改进意见

        根据当前进度,输出 JSON:{"next": "researcher" | "writer" | "critic" | "FINISH"}
        如果任务已完成,输出 {"next": "FINISH"}。"""},
        {"role": "user", "content": f"当前状态:{json.dumps(state, ensure_ascii=False)}"}
    ])
    decision = json.loads(response.content)
    return {"next_agent": decision["next"]}

# ========== Worker 节点 ==========
def researcher_node(state):
    response = llm.invoke([
        {"role": "system", "content": "你是研究专家,收集和组织信息。"},
        *state["messages"]
    ])
    return {
        "research_output": response.content,
        "messages": state["messages"] + [response]
    }

def writer_node(state):
    response = llm.invoke([
        {"role": "system", "content": f"你是写作专家。基于以下研究结果写内容:\n{state.get('research_output', '')}"},
    ])
    return {
        "draft_output": response.content,
        "messages": state["messages"] + [response]
    }

def critic_node(state):
    response = llm.invoke([
        {"role": "system", "content": f"你是审核专家。审核以下内容:\n{state.get('draft_output', '')}"},
    ])
    return {"messages": state["messages"] + [response]}

# ========== 条件路由 ==========
def route_to_next(state):
    if state["next_agent"] == "FINISH":
        return END
    return state["next_agent"]

# ========== 构建图 ==========
graph = StateGraph(SupervisorState)
graph.add_node("supervisor", supervisor_node)
graph.add_node("researcher", researcher_node)
graph.add_node("writer", writer_node)
graph.add_node("critic", critic_node)

graph.set_entry_point("supervisor")
graph.add_conditional_edges(
    "supervisor", route_to_next,
    {"researcher": "researcher", "writer": "writer",
     "critic": "critic", END: END}
)
# Worker 完成后回到 Supervisor
graph.add_edge("researcher", "supervisor")
graph.add_edge("writer", "supervisor")
graph.add_edge("critic", "supervisor")

app = graph.compile()

# ========== 运行 ==========
result = app.invoke({
    "messages": [("user", "写一篇关于 Agent 协作模式的技术分析")],
    "next_agent": "",
    "research_output": "",
    "draft_output": "",
    "final_output": ""
})
print(result["messages"][-1].content)

Supervisor 模式优劣势

维度	优势	劣势
控制力	完全集中,流程可预测	Supervisor 成为瓶颈
扩展性	建议 3-8 个 Agent	超过 10 个后路由逻辑复杂
容错	单点故障	需要额外实现高可用

模式五:Swarm(对等接力)

原理

Python
用户 → Triage Agent(入口)
            ├─ 技术问题 → Tech Agent ──┐
            └─ 订单问题 → Order Agent ──┤
                                      ↓
                               最终 Agent 直接返回用户

特点:

没有中央协调者
每个 Agent 自主决定"移交给谁"
类似"击鼓传花",动态性最强

代码示例(OpenAI Swarm 风格)

Plain Text
# pip install swarm
from swarm import Swarm, Agent

client = Swarm()

# ========== 定义移交函数 ==========
def transfer_to_tech():
    """转交给技术支持 Agent"""
    return tech_agent

def transfer_to_order():
    """转交给订单处理 Agent"""
    return order_agent

def escalate_to_human():
    """升级到人工"""
    return "ESCALATE"

# ========== 定义 Agent ==========
triage_agent = Agent(
    name="Triage Agent",
    instructions="""你是客服分流 Agent。根据问题类型路由:
    - 技术问题(安装、配置、Bug)→ transfer_to_tech
    - 订单问题(退款、物流、支付)→ transfer_to_order
    - 无法分类 → escalate_to_human""",
    functions=[transfer_to_tech, transfer_to_order, escalate_to_human],
)

tech_agent = Agent(
    name="Tech Support",
    instructions="""你是技术支持专家。解决问题后直接回复用户。
    如果涉及退款/订单,调用 transfer_to_order。""",
    functions=[transfer_to_order, escalate_to_human],
)

order_agent = Agent(
    name="Order Agent",
    instructions="""你是订单处理专家。处理完后直接回复用户。
    如果需要技术排查,调用 transfer_to_tech。""",
    functions=[transfer_to_tech, escalate_to_human],
)

# ========== 运行 ==========
response = client.run(
    agent=triage_agent,
    messages=[{"role": "user",
               "content": "我的订单显示已发货,但 3 天了还没到,而且 App 一直闪退"}],
)
print(response.messages[-1]["content"])
# 智能体会自动在 Triage → Order → Tech 之间流转

Swarm 模式优劣势

维度	优势	劣势
灵活性	自主决定移交,适配动态场景	流转路径难以预测
容错	无单点故障	需设置移交次数限制防死循环
可观测	去中心化,Trace 难度大	需显式注入追踪逻辑

五种模式对比总结

🦄 也有一些不同叫法，比如Planner-Excutor/Orchestrator-worker模式等，实际原理都类似supervisor模式

模式	中心化程度	典型场景
串行链式	无中心	文本流水线、RAG
路由分发	弱中心(Router)	智能客服、工单分流
并行处理	无中心	多角度分析、竞速
Supervisor	强中心	报告生成、确定性工作流
Swarm	无中心	客户服务、多轮对话

根据实际场景进行模式组合举例：

当选择"智能客服/工单系统":

推荐 Router + Swarm 组合。Router 做第一层分流(技术/订单/售后),Swarm 做每层内的动态移交。

当选择"内容生成/报告撰写":

推荐 Supervisor 模式。Supervisor 协调 Researcher → Writer → Critic,流程可控,质量稳定。

当选择"代码生成/技术分析":

推荐 Supervisor + 并行 组合。多个 Agent 并行尝试不同实现,Supervisor 选最优结果。

当选择"最简单":

从 串行链式 开始,3 个 Agent 串起来,代码最少,概念最清晰。