课程首页 › Module 4 · 多 Agent 协作 › L6

L6 多 Agent 系统：从单兵作战到团队协作

课程目标：理解为什么复杂任务必须靠多 Agent 分工，掌握三种主流多 Agent 架构模式，能为真实业务场景设计多 Agent 系统。

一、为什么需要多 Agent？#

L1-L5 我们一直在讲单 Agent——一个大脑决定一切。但现实中有些任务天然就需要分工：

你不会让一个人干所有这些事——人类靠分工提升效率，Agent 也是。

但多 Agent 不是越多越好。我们先看反例。

二、反模式：Agent 越多越乱#

2024 年 AutoGPT 火的时候，很多人尝试堆 Agent：

Planner → Researcher → Coder → Tester → Debugger → Deployer → Monitor → Optimizer

结果：每个 Agent 的输出都要塞进下一个 Agent 的 prompt，token 爆炸、上下文污染、决策链脱节。

学界的总结（来自 LangGraph 团队的 benchmark 报告）：

多 Agent 系统比单 Agent baseline 性能低 50% 是常见情况——除非 Supervisor 不改写 sub-agent 的输出。

结论：多 Agent 是把双刃剑。用得对，分工提效；用得乱，不如单 Agent。

三、用一个比喻：医院 vs 全科医生#

把单 Agent 比作全科医生：

把多 Agent 比作医院：

关键洞察：

四、三大主流多 Agent 模式#

实战中你会遇到的多 Agent 架构，可以归为三类：

模式 1：Manager-Worker（管理者—工人）#

            ┌─→ Researcher（查资料）
            │
Supervisor ─┼─→ Analyst（分析数据）
            │
            └─→ Writer（写报告）

模式 2：Group Chat（群聊讨论）#

            Speaker A
               ↓
Speaker B ← Manager → Speaker C
               ↓
            Speaker D

模式 3：Debate（辩论投票）#

Proposer A → 答案 A ─┐
                       ├─→ Aggregator → 最终答案
Proposer B → 答案 B ─┘

五、模式 1 实战：Manager-Worker 写市场分析报告#

我们用 LangGraph 完整实现一个 Manager-Worker 系统：用户说"分析一下 2026 年 AI Agent 行业"，Manager 自动拆任务，调度三个 Worker。

完整代码#

from typing import Annotated, Literal
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
from langchain_tavily import TavilySearch

# ============ 1. 定义 State ============

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]  # 对话历史
    task_plan: str                          # Manager 拆分的任务清单
    research_data: str                      # Researcher 收集的数据
    analysis: str                           # Analyst 的分析结果
    final_report: str                       # Writer 写的最终报告

# ============ 2. 准备工具 ============

search_tool = TavilySearch(max_results=3)
llm = ChatOpenAI(model="gpt-4o-mini")

# ============ 3. 定义 Worker 节点 ============

def researcher_node(state: AgentState):
    """Worker 1：研究员，收集数据"""
    task = state["task_plan"]
    prompt = f"""你是研究员。请基于以下任务搜索相关数据：
    任务：{task}
    
    用 search_web 工具搜索，整理出 3-5 个关键事实。"""
    
    llm_with_search = llm.bind_tools([search_tool])
    response = llm_with_search.invoke(prompt)
    
    # 简化：真实场景需要 ToolNode 执行工具调用
    return {"research_data": response.content or "（研究数据待补充）"}

def analyst_node(state: AgentState):
    """Worker 2：分析师，解读数据"""
    data = state["research_data"]
    prompt = f"""你是数据分析师。请基于以下研究数据进行分析：
    数据：{data}
    
    输出 3 个关键洞察和潜在的市场机会/风险。"""
    
    response = llm.invoke(prompt)
    return {"analysis": response.content}

def writer_node(state: AgentState):
    """Worker 3：写手，整合成报告"""
    data = state["research_data"]
    analysis = state["analysis"]
    prompt = f"""你是专业写手。请基于以下内容写一份结构化的市场分析报告：
    
    研究数据：{data}
    分析洞察：{analysis}
    
    报告应包含：摘要、关键发现、市场机会、风险提示、结论。"""
    
    response = llm.invoke(prompt)
    return {"final_report": response.content}

# ============ 4. 定义 Manager 节点 ============

def manager_node(state: AgentState):
    """Supervisor：决定下一步调哪个 Worker"""
    messages = state["messages"]
    last_msg = messages[-1].content if messages else ""
    
    prompt = f"""你是 Manager。请基于当前状态决定下一步：
    
    用户原始请求：{last_msg}
    当前进度：
    - 任务拆分：{'已完成' if state.get('task_plan') else '未完成'}
    - 研究数据：{'已收集' if state.get('research_data') else '未收集'}
    - 分析结果：{'已完成' if state.get('analysis') else '未完成'}
    - 最终报告：{'已完成' if state.get('final_report') else '未完成'}
    
    下一步应该调用哪个 Worker？只能选一个：
    - "plan"（还没拆任务时）
    - "researcher"（缺数据时）
    - "analyst"（缺分析时）
    - "writer"（缺报告时）
    - "finish"（全部完成时）
    
    只返回一个词，不要其他内容。"""
    
    decision = llm.invoke(prompt).content.strip().lower()
    return {"next_worker": decision}  # 这个字段用于条件边

# ============ 5. 路由函数 ============

def route_decision(state: AgentState) -> Literal["plan", "researcher", "analyst", "writer", "finish"]:
    """根据 manager 的决定路由"""
    decision = state.get("next_worker", "finish")
    if decision not in ["plan", "researcher", "analyst", "writer"]:
        return "finish"
    return decision

def planner_node(state: AgentState):
    """第一步：拆任务"""
    messages = state["messages"]
    user_request = messages[-1].content if messages else ""
    
    prompt = f"""请把以下任务拆分成 3 个研究子任务：
    任务：{user_request}
    
    输出格式：1. xxx 2. xxx 3. xxx"""
    
    response = llm.invoke(prompt)
    return {"task_plan": response.content}

# ============ 6. 装配图 ============

graph_builder = StateGraph(AgentState)

# 加节点
graph_builder.add_node("manager", manager_node)
graph_builder.add_node("plan", planner_node)
graph_builder.add_node("researcher", researcher_node)
graph_builder.add_node("analyst", analyst_node)
graph_builder.add_node("writer", writer_node)

# 加边
graph_builder.add_edge(START, "manager")  # 开始 → Manager 决策
graph_builder.add_conditional_edges(
    "manager",
    route_decision,
    {
        "plan": "plan",
        "researcher": "researcher",
        "analyst": "analyst",
        "writer": "writer",
        "finish": END
    }
)

# 每个 Worker 完成后回到 Manager
for worker in ["plan", "researcher", "analyst", "writer"]:
    graph_builder.add_edge(worker, "manager")

memory = MemorySaver()
graph = graph_builder.compile(checkpointer=memory)

# ============ 7. 运行 ============

config = {"configurable": {"thread_id": "report-001"}}
result = graph.invoke(
    {"messages": [{"role": "user", "content": "分析 2026 年 AI Agent 行业的市场前景"}]},
    config=config
)

print(result.get("final_report", "报告生成失败"))

运行过程（简化）#

[START] → manager: 决策 → plan
[plan] → 任务拆分完成 → manager
[manager]: 决策 → researcher
[researcher] → 数据收集完成 → manager
[manager]: 决策 → analyst
[analyst] → 分析完成 → manager
[manager]: 决策 → writer
[writer] → 报告完成 → manager
[manager]: 决策 → finish → END

这就是 Manager-Worker 模式的完整流程。

六、模式 2 实战：Group Chat 评审产品需求#

Group Chat 模式的核心是让多个 Agent 轮流发言，模拟真实会议。我们用 AutoGen 风格的简化实现：

# 简化版 Group Chat：让 ProductManager、Engineer、Designer 三方讨论一个需求
# 完整 AutoGen GroupChat 实现见 autogen 库

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

class ChatState(TypedDict):
    messages: Annotated[list, add_messages]
    speakers: list[str]  # 发言顺序

def pm_node(state: ChatState):
    """产品经理：提需求"""
    prompt = f"你是产品经理。请基于用户需求提出 MVP 功能列表：\n需求：{state['messages'][-1].content}"
    return {"messages": [llm.invoke(prompt)]}

def engineer_node(state: ChatState):
    """工程师：评估技术可行性"""
    last = state["messages"][-1].content
    prompt = f"你是工程师。请评估以下方案的技术可行性，给出实现建议：\n方案：{last}"
    return {"messages": [llm.invoke(prompt)]}

def designer_node(state: ChatState):
    """设计师：评估用户体验"""
    last = state["messages"][-1].content
    prompt = f"你是设计师。请评估以下方案的 UX，给出界面建议：\n方案：{last}"
    return {"messages": [llm.invoke(prompt)]}

# 装配图：PM → Engineer → Designer → END（每人一轮）
graph_builder = StateGraph(ChatState)
graph_builder.add_node("pm", pm_node)
graph_builder.add_node("engineer", engineer_node)
graph_builder.add_node("designer", designer_node)

graph_builder.add_edge(START, "pm")
graph_builder.add_edge("pm", "engineer")
graph_builder.add_edge("engineer", "designer")
graph_builder.add_edge("designer", END)

graph = graph_builder.compile()

Group Chat 模式实战更推荐用 AutoGen 库——它内置 GroupChatManager 处理发言顺序、轮次限制、冲突检测。我们这里只是演示原理。

七、模式 3 实战：Debate 模式做医学诊断#

Debate 模式让多个独立 Agent 给出答案，然后投票或聚合：

# 简化版 Debate：3 个 Agent 独立诊断，1 个 Aggregator 综合
# 适用于追求准确率的场景

def diagnostic_agent(agent_id: int):
    """生成一个诊断 Agent"""
    def node(state: ChatState):
        prompt = f"""你是第 {agent_id} 号诊断医生。请基于以下症状独立给出诊断：
        症状：{state['messages'][-1].content}
        
        要求：
        1. 独立思考，不参考其他医生
        2. 给出最可能的诊断 + 置信度（0-100%）
        3. 如果不确定，明确说不确定"""
        return {"messages": [llm.invoke(prompt)]}
    return node

def aggregator_node(state: ChatState):
    """聚合器：综合多个医生的诊断"""
    all_diagnoses = state["messages"][-3:]  # 取最后 3 条
    diagnoses_text = "\n\n".join([m.content for m in all_diagnoses])
    
    prompt = f"""你是主任医师。请综合以下 3 位医生的诊断：
    {diagnoses_text}
    
    综合考虑：
    1. 如果 3 人一致 → 直接给出最终诊断
    2. 如果 2:1 → 采用多数意见，但说明分歧
    3. 如果 3:0:0 全不同 → 要求进一步检查
    
    输出最终诊断 + 置信度。"""
    
    return {"messages": [llm.invoke(prompt)]}

# 装配图
graph_builder = StateGraph(ChatState)
graph_builder.add_node("doc1", diagnostic_agent(1))
graph_builder.add_node("doc2", diagnostic_agent(2))
graph_builder.add_node("doc3", diagnostic_agent(3))
graph_builder.add_node("aggregator", aggregator_node)

graph_builder.add_edge(START, "doc1")
graph_builder.add_edge("doc1", "doc2")
graph_builder.add_edge("doc2", "doc3")
graph_builder.add_edge("doc3", "aggregator")
graph_builder.add_edge("aggregator", END)

为什么 Debate 模式有效？

研究表明，多个独立判断 + 聚合的准确率通常高于单个最强 Agent。这是因为：

八、模式选择决策树#

实际项目中怎么选？给你一个决策树：

你的任务能清晰拆分成独立子任务吗？
├── 是 → Manager-Worker（首选）
│         - 子任务超过 5 个？ → 加层级（Manager of Managers）
│         - 子任务需要按顺序执行？ → 加 Chain
│         - 子任务可并行？ → 加 Parallel（LangGraph 支持 Send API）
│
└── 否 → 需要多视角碰撞吗？
          ├── 是 → Group Chat（适合协商、评审、头脑风暴）
          └── 否 → 需要高准确率吗？
                    ├── 是 → Debate（适合医疗、法律、金融）
                    └── 否 → 回到 Manager-Worker 或单 Agent

经验法则：

九、多 Agent 系统的四大陷阱#

陷阱 1：Supervisor 瓶颈#

Manager 看过所有消息，prompt 越来越长，token 爆炸。

解法：

陷阱 2：Token 浪费#

每个 Agent 都看到完整对话历史，N 个 Agent = N 倍 token。

解法：

陷阱 3：Sycophancy（谄媚效应）#

LLM Agent 倾向于"同意上一个 Agent 的观点"，而不是真正批评。结果是 Group Chat 看起来热闹但没产出。

解法：

陷阱 4：错误传播#

Worker 1 错了 → Worker 2 基于错的数据错得更多 → Supervisor 看到错的结果做错决策。

解法：

十、和 L1-L5 的连接#

回顾整个课程：

课主题在多 Agent 系统中的位置
L1Agent 是什么多 Agent 系统由多个 L1 定义的 Agent 组成
L2ReAct 循环每个 Agent 内部依然跑 ReAct
L3smolagentsWorker 的一种实现
L4Tool UseManager 和 Worker 都依赖工具调用
L5LangGraph多 Agent 编排的事实标准——状态图天然适合建模多 Agent
L6多 Agent把 L1-L5 的能力扩展到团队协作

完整心智模型：多 Agent = 多张 LangGraph 图 + 一张协调图 = 多组 Worker + 一个 Supervisor。

十一、动手时间：30 分钟设计你的多 Agent 系统#

请你为一个真实业务场景设计多 Agent 架构。

场景 A：自动代码审查系统#

输入：开发者提交一段代码 PR 输出：审查报告（包含 bug 风险、性能问题、安全问题、风格建议）

请设计：

  1. 用哪种多 Agent 模式？（Manager-Worker / Group Chat / Debate？）

  2. 分几个 Worker？每个 Worker 的职责？

  3. Supervisor 怎么决策？

场景 B：自动投资研究报告#

输入：用户问"分析一下英伟达这只股票" 输出：投资研究报告（基本面、技术面、风险、行业对比）

请设计：

  1. 用哪种多 Agent 模式？

  2. 分几个 Worker？每个 Worker 的输入是什么、输出是什么？

  3. 怎么避免"幻觉"——LLM 编造财务数据？

场景 C：自动教学内容生成#

输入：用户给一个 AI Agents 的学习主题 输出：完整课程章节（含概念解释、代码示例、练习题）

请设计：

  1. 用哪种多 Agent 模式？

  2. 分几个 Worker？每个 Worker 负责课程的哪个部分？

  3. 怎么保证多 Worker 输出的内容风格一致？

📋 场景 A 参考设计（点击展开）

场景 A：自动代码审查#

推荐模式：Manager-Worker

Worker 设计：

Supervisor 决策逻辑：

收到 PR → 触发所有 Worker 并行扫描
       → 收集四份报告
       → Aggregator 整合去重
       → 输出统一审查报告

为什么不用 Group Chat？ 代码审查是结构化分析，不需要多视角碰撞，Manager-Worker 更高效。

为什么不用 Debate？ 成本太高，代码审查不需要"投票"。

避免幻觉：所有 Worker 必须基于静态分析工具的真实输出（lint、bandit、pylint），不能凭空说"这里有 bug"。

十二、小结#

知识点一句话回顾
多 Agent 不是越多越好多 Agent 必须比单 Agent baseline 显著更好才值得
三大主流模式Manager-Worker（80% 场景）/ Group Chat（15%）/ Debate（5%）
Manager-WorkerSupervisor 拆任务 + Worker 独立执行，最适合可清晰拆分的任务
Group Chat多 Agent 轮流发言，适合协商、评审、头脑风暴
Debate多 Agent 独立判断 + 投票聚合，追求最高准确率
四大陷阱Supervisor 瓶颈 / Token 浪费 / Sycophancy / 错误传播

十三、课程完结 & 下一步#

🎉 恭喜你完成了全部 6 节课！ 你已经系统掌握了：

能力对应课程
解释 Agent vs Chatbot 的本质区别L1
描述 ReAct 循环的工作机制L2
用 smolagents 跑通第一个 AgentL3
手写 Function Calling Agent + 工具设计原则L4
用 LangGraph 构建有记忆的多步骤 AgentL5
设计和实现多 Agent 协作系统L6

接下来可以深入的方向#

  1. 生产级工程化：MCP 协议、AgentOps 监控、成本优化

  2. 更复杂的框架：AutoGen、CrewAI、OpenAI Agents SDK

  3. 特定领域应用：代码 Agent（Devin）、研究 Agent（Deep Research）、GUI Agent

  4. 底层原理：LLM 训练的 SFT/RLHF 如何影响 Agent 能力

  5. 学术前沿：Voyager（游戏 Agent）、Toolformer（工具学习）、AgentVerse（多 Agent 涌现）

推荐实战项目#

十四、Sources#

💬 有疑问？ 试试问我："我的项目该选哪种多 Agent 模式？"、"Supervisor 瓶颈怎么解决？"、"GAIA benchmark 怎么参加？"——我会用代码或比喻为你解答。

🎁 彩蛋：把 6 节课的任意内容抛给我，比如"用 L2 学过的 ReAct 帮我理解 LangGraph 的状态图"、"用 L4 的工具设计法则帮我设计一个 XX 工具"，我会把跨课知识串起来。

自测 · 检验是否真的掌握了

1. 为什么把 Agent 越堆越多反而可能不如单个 Agent？

堆叠 Agent 的常见失败是输出在层层转述中失真、管理节点改写工人结果导致决策链脱节，benchmark 显示这种情况性能可能低于单 Agent baseline。真正要解决的是分工边界与上下文隔离，而不是数量或并发速度，模型密钥也可以共用。

2. 要写一份需要多方评审、必须协商达成一致的需求文档，该选哪种模式？

需要多视角碰撞与协商的任务适合 Group Chat，让不同角色轮流发言、观点相互碰撞。Debate 追求的是多个独立判断投票后的准确率，适合医疗、法律这类关键决策；Manager-Worker 适合可清晰拆分的独立子任务，缺少协商环节。

3. 多 Agent 系统中的谄媚效应指的是什么问题？

谄媚指 LLM 天然倾向于同意前一个 Agent 的说法，于是群聊看起来热闹却没有真正的批判。对策包括强制每个角色至少指出一个风险、引入唱反调的魔鬼代言人角色等。上下文膨胀是 token 浪费问题，错误传播是另一个陷阱，都不是谄媚。
有疑问？ 这一课里任何看不懂的地方、想动手试的代码，都可以直接向你的 AI teacher 提问。