12.2 多智能体工作流复杂案例
在上一节中，我们了解了多智能体系统的基本概念。本节将深入探讨 Supervisor 模式的高级用法，通过一个内容创作团队的完整案例，展示如何构建企业级的多智能体协作系统。
为什么需要复杂的多智能体架构？
当你的 AI 应用场景变得复杂时，简单的单 Agent 方案往往力不从心：
| 挑战 | 表现 | 解决思路 |
|---|---|---|
| 决策复杂度高 | 需要综合判断、权衡利弊 | 专业 Agent 各司其职，协同决策 |
| 规则维护困难 | 规则多且易冲突、更新成本高 | 分解到不同 Agent，降低耦合 |
| 依赖非结构化数据 | 需要理解文档、对话等 | 专门的信息提取与处理 Agent |
核心洞察：复杂任务的关键不在于单个模型有多强大，而在于如何让多个专业 Agent 高效协作。
Supervisor 模式核心机制
Supervisor 模式的核心思想是中央调度——由一个"总指挥"来协调各个专业 Agent 的工作。
运作流程
┌──────────────────────────────────────────────────────────┐
│ 用户请求 │
│ ↓ │
│ ┌─────────┐ │
│ │Supervisor│ ← 分析意图、分配任务 │
│ └────┬────┘ │
│ ┌─────────────┼─────────────┐ │
│ ↓ ↓ ↓ │
│ ┌─────────┐ ┌─────────┐ ┌─────────┐ │
│ │ Agent A │ │ Agent B │ │ Agent C │ │
│ └────┬────┘ └────┬────┘ └────┬────┘ │
│ ↓ ↓ ↓ │
│ ┌─────────┐ │
│ │Supervisor│ ← 汇总结果、反馈用户 │
│ └─────────┘ │
└──────────────────────────────────────────────────────────┘与 Swarm 模式的核心差异
| 特性 | Supervisor 模式 | Swarm 模式 |
|---|---|---|
| 控制方式 | 集中式，Supervisor 统筹 | 去中心化，Agent 自主交接 |
| 用户交互 | 主要通过 Supervisor | 可与任意 Agent 直接交互 |
| 执行流程 | 任务完成后返回 Supervisor | Agent 之间直接传递控制权 |
| 起始节点 | 固定为 Supervisor | 可配置默认 Agent |
| 通信路径 | 必经 Supervisor | Agent 点对点通信 |
| 最佳场景 | 流程明确、需要监控 | 灵活对话、专家轮换 |
完整实现：内容创作团队
下面我们构建一个智能内容创作系统，包含：
- 调研专员：负责搜索资料、收集信息
- 数据分析师：负责数据计算和统计分析
- 总编辑（Supervisor）：协调团队、整合输出
环境准备
bash
pip install langgraph-supervisor langchain-openai langchain-communitypython
import os
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
from langgraph_supervisor import create_supervisor
from langchain_community.tools.tavily_search import TavilySearchResults
# 配置 LLM
llm = ChatOpenAI(
model="gpt-4o",
api_key=os.getenv("OPENAI_API_KEY")
)定义工具集
python
# 数据分析工具
def calculate_sum(numbers: list[float]) -> float:
"""对一组数字求和"""
return sum(numbers)
def calculate_average(numbers: list[float]) -> float:
"""计算平均值"""
return sum(numbers) / len(numbers) if numbers else 0
def calculate_percentage(part: float, total: float) -> float:
"""计算百分比"""
return (part / total * 100) if total != 0 else 0
# 搜索工具
search_tool = TavilySearchResults(
max_results=3,
tavily_api_key=os.getenv("TAVILY_API_KEY")
)创建专业 Agent
python
# 调研专员 Agent
researcher = create_react_agent(
model=llm,
tools=[search_tool],
name="research_specialist",
prompt="""你是一名专业的信息调研员。
职责范围：
- 执行网络搜索，收集相关资料
- 整理并提炼关键信息
- 不要进行任何数学计算
工作规范：
- 完成调研后，将结果直接汇报给总编辑
- 只返回调研结果，不要添加无关内容"""
)
# 数据分析师 Agent
analyst = create_react_agent(
model=llm,
tools=[calculate_sum, calculate_average, calculate_percentage],
name="data_analyst",
prompt="""你是一名专业的数据分析师。
职责范围：
- 执行数学计算和统计分析
- 处理数值型数据
- 不要进行网络搜索
工作规范：
- 完成分析后，将结果直接汇报给总编辑
- 只返回计算结果和分析结论"""
)创建 Supervisor
python
# 总编辑（Supervisor）
editor_in_chief = create_supervisor(
model=llm,
agents=[researcher, analyst],
prompt="""你是内容创作团队的总编辑，负责协调以下成员：
- research_specialist：负责信息搜索和资料收集
- data_analyst：负责数据计算和统计分析
工作原则：
1. 分析用户请求，判断需要调用哪个专家
2. 一次只分配任务给一个专家，避免并行调用
3. 自己不执行具体任务，只负责协调和汇总
4. 复杂请求可能需要多轮协调（先调研、再分析）""",
add_handoff_back_messages=True,
output_mode="full_history"
).compile()测试运行
python
def run_creative_task(query: str):
"""执行创作任务"""
print(f"📝 任务: {query}")
print("-" * 60)
for chunk in editor_in_chief.stream(
{"messages": [{"role": "user", "content": query}]},
subgraphs=False
):
for node_name, node_data in chunk.items():
if "messages" in node_data:
last_msg = node_data["messages"][-1]
print(f"[{node_name}] {last_msg.content[:200]}...")
print("=" * 60)
# 示例：综合性请求
run_creative_task(
"帮我调研一下2024年中国新能源汽车市场的销量数据，"
"然后计算前三大品牌的市场份额占比"
)执行流程：
- 总编辑接收请求，判断需要先调研
- 分配给 research_specialist 搜索市场数据
- 调研完成后，任务返回总编辑
- 总编辑将数据分配给 data_analyst 计算份额
- 分析完成后，总编辑汇总并回复用户
消息历史管理策略
在多 Agent 协作中，如何管理对话历史至关重要。LangGraph 提供了两种策略：
完整历史模式
保留 Agent 执行过程中的所有消息，适合需要追溯和调试的场景：
python
supervisor = create_supervisor(
agents=[researcher, analyst],
output_mode="full_history" # 保留完整对话历史
)消息流示例：
Human → Supervisor → [Tool: transfer_to_researcher] → Researcher
→ [搜索过程...] → [Tool: transfer_back] → Supervisor
→ [Tool: transfer_to_analyst] → Analyst → [计算过程...]
→ [Tool: transfer_back] → Supervisor → 最终回复精简结果模式
只保留每个 Agent 的最终输出，适合需要简洁上下文的场景：
python
supervisor = create_supervisor(
agents=[researcher, analyst],
output_mode="last_message" # 只保留最后一条消息
)消息流示例：
Human → Supervisor → Researcher最终结果 → Analyst最终结果 → 最终回复如何选择？
| 场景 | 推荐模式 | 理由 |
|---|---|---|
| 开发调试阶段 | full_history | 便于追踪问题 |
| 上下文敏感任务 | full_history | Agent 需要了解前序工作 |
| 简单独立任务 | last_message | 减少 token 消耗 |
| 长对话场景 | last_message | 避免上下文溢出 |
多层级架构设计
当系统规模扩大时，可以构建层级式的 Supervisor 结构：
┌────────────────┐
│ 总编辑（顶层） │
└───────┬────────┘
┌─────────────┼─────────────┐
↓ ↓ ↓
┌───────────┐ ┌───────────┐ ┌───────────┐
│ 调研组长 │ │ 分析组长 │ │ 创作组长 │
└─────┬─────┘ └─────┬─────┘ └─────┬─────┘
│ │ │
┌─────┴─────┐ ┌─────┴─────┐ ┌─────┴─────┐
│ 搜索专员 │ │ 统计专员 │ │ 文案专员 │
│ 整理专员 │ │ 可视化专员│ │ 校对专员 │
└───────────┘ └───────────┘ └───────────┘实现多层级 Supervisor
python
# 第一层：创建各组的 Supervisor
research_team = create_supervisor(
[search_agent, organize_agent],
model=llm,
supervisor_name="research_lead"
).compile(name="research_team")
analysis_team = create_supervisor(
[stats_agent, viz_agent],
model=llm,
supervisor_name="analysis_lead"
).compile(name="analysis_team")
# 第二层：顶层 Supervisor 管理各组
top_supervisor = create_supervisor(
[research_team, analysis_team],
model=llm,
supervisor_name="editor_in_chief"
).compile(name="content_team")优势：
- ✅ 职责清晰，易于维护
- ✅ 可独立扩展各团队
- ✅ 支持更复杂的协作流程
自定义任务交接机制
默认的 handoff 工具可能无法满足特定需求，你可以创建自定义版本：
基础自定义
python
from langgraph_supervisor import create_handoff_tool
# 自定义工具名称和描述
custom_tools = [
create_handoff_tool(
agent_name="research_specialist",
name="assign_research_task",
description="将调研类任务分配给调研专员，包括信息搜索、资料整理等"
),
create_handoff_tool(
agent_name="data_analyst",
name="assign_analysis_task",
description="将数据分析任务分配给分析师，包括计算、统计、对比等"
)
]
supervisor = create_supervisor(
[researcher, analyst],
model=llm,
tools=custom_tools # 使用自定义工具
)高级自定义：带参数的交接
python
from typing import Annotated
from langchain_core.tools import tool, InjectedToolCallId
from langgraph.prebuilt import InjectedState
from langgraph.types import Command
from langgraph.graph import MessagesState
def create_task_handoff(agent_name: str, description: str):
"""创建带任务描述的交接工具"""
@tool(f"delegate_to_{agent_name}", description=description)
def handoff_with_context(
task_description: str, # LLM 填写的任务描述
priority: str = "normal", # 任务优先级
state: Annotated[MessagesState, InjectedState] = None,
tool_call_id: Annotated[str, InjectedToolCallId] = None
) -> Command:
"""
将任务交接给指定 Agent。
Args:
task_description: 具体任务说明
priority: 任务优先级 (high/normal/low)
"""
# 构建交接消息
handoff_msg = {
"role": "tool",
"content": f"任务已分配给 {agent_name}\n"
f"任务说明: {task_description}\n"
f"优先级: {priority}",
"name": f"delegate_to_{agent_name}",
"tool_call_id": tool_call_id
}
return Command(
goto=agent_name,
update={
**state,
"messages": state["messages"] + [handoff_msg],
"current_task": task_description,
"task_priority": priority
},
graph=Command.PARENT
)
return handoff_with_context控制交接消息显示
python
# 不显示交接过程的系统消息
supervisor = create_supervisor(
[researcher, analyst],
model=llm,
add_handoff_messages=False # 隐藏交接消息
)
# 自定义交接工具前缀
supervisor = create_supervisor(
[researcher, analyst],
model=llm,
handoff_tool_prefix="assign_to" # 工具名变为 assign_to_xxx
)消息转发优化
有时 Supervisor 不需要对 Agent 的回复做额外处理，可以直接转发给用户：
python
from langgraph_supervisor.handoff import create_forward_message_tool
# 创建转发工具
forward_tool = create_forward_message_tool("editor_in_chief")
supervisor = create_supervisor(
[researcher, analyst],
model=llm,
tools=[forward_tool] # 添加转发能力
)使用场景：
- Agent 输出已经足够完整，无需 Supervisor 润色
- 减少 Supervisor 的额外处理，降低延迟
- 避免信息在转述过程中失真
Swarm 模式对比实现
为了理解 Supervisor 与 Swarm 的区别，我们用 Swarm 模式实现同样的系统：
python
from langgraph_swarm import create_swarm, create_handoff_tool
# Swarm 模式下，每个 Agent 需要持有交接工具
handoff_to_analyst = create_handoff_tool(
agent_name="data_analyst",
description="当需要数据计算时，转交给数据分析师"
)
handoff_to_researcher = create_handoff_tool(
agent_name="research_specialist",
description="当需要搜索资料时，转交给调研专员"
)
# 重新创建带有交接能力的 Agent
researcher_swarm = create_react_agent(
model=llm,
tools=[search_tool, handoff_to_analyst], # 可以转交给分析师
name="research_specialist",
prompt="你是调研专员。完成调研后，如需数据分析，转交给分析师。"
)
analyst_swarm = create_react_agent(
model=llm,
tools=[calculate_sum, calculate_average, handoff_to_researcher],
name="data_analyst",
prompt="你是数据分析师。如需补充资料，转交给调研专员。"
)
# 创建 Swarm
swarm_team = create_swarm(
agents=[researcher_swarm, analyst_swarm],
default_active_agent="research_specialist" # 默认从调研开始
).compile()对比总结
| 维度 | Supervisor 实现 | Swarm 实现 |
|---|---|---|
| 代码复杂度 | 较简单，集中配置 | 需要为每个 Agent 配置交接工具 |
| 流程可控性 | 高，Supervisor 把关 | 低，Agent 自主决策 |
| 调试难度 | 低，有明确的中心节点 | 高，需要追踪 Agent 切换链 |
| 灵活性 | 中等 | 高，支持任意 Agent 间协作 |
| 适用场景 | 流程驱动的任务 | 对话驱动的交互 |
Agent 执行控制
调用模式
python
# 同步调用 - 等待完成
result = supervisor.invoke(
{"messages": [{"role": "user", "content": "查询今年销售数据"}]}
)
# 异步调用 - 非阻塞
result = await supervisor.ainvoke(
{"messages": [{"role": "user", "content": "查询今年销售数据"}]}
)
# 流式输出 - 实时反馈
for chunk in supervisor.stream(
{"messages": [{"role": "user", "content": "查询今年销售数据"}]},
stream_mode="updates"
):
print(chunk)迭代次数限制
防止 Agent 陷入无限循环：
python
from langgraph.errors import GraphRecursionError
max_rounds = 5
recursion_limit = 2 * max_rounds + 1
try:
result = supervisor.invoke(
{"messages": [{"role": "user", "content": query}]},
{"recursion_limit": recursion_limit}
)
except GraphRecursionError:
print("⚠️ 达到最大迭代次数，请简化任务或检查 Agent 配置")实战建议
1. Agent 命名与描述
python
# ❌ 模糊的描述
@tool("helper", description="帮助处理事情")
def bad_agent(): ...
# ✅ 清晰的描述
@tool(
"market_researcher",
description="专门负责市场调研，包括：竞品分析、行业趋势、用户画像。"
"适用于需要外部市场信息的请求。"
)
def good_agent(): ...2. 渐进式复杂度
| 阶段 | 建议方案 |
|---|---|
| 验证想法 | 单 Agent + 少量工具 |
| 初步生产 | Supervisor + 2-3 专业 Agent |
| 规模化 | 层级式多智能体架构 |
3. 常见问题排查
| 问题 | 可能原因 | 解决方案 |
|---|---|---|
| Supervisor 不调用 Agent | 工具描述不清晰 | 优化 Agent 工具的 description |
| Agent 执行错误任务 | 职责边界模糊 | 明确每个 Agent 的 prompt |
| 无限循环 | 缺少终止条件 | 设置 recursion_limit |
| 上下文丢失 | 使用了 last_message 模式 | 改用 full_history 模式 |
小结
本节我们深入学习了 Supervisor 模式的高级用法：
- ✅ 核心机制：中央调度、专业分工
- ✅ 消息管理：完整历史 vs 精简结果
- ✅ 层级架构：支持大规模团队协作
- ✅ 自定义交接：灵活控制任务流转
- ✅ 模式对比：Supervisor vs Swarm 的适用场景
核心要点：多智能体系统的关键在于职责清晰和协作顺畅。Supervisor 模式提供了一种结构化的方式来组织 Agent 团队，特别适合流程明确、需要监控的业务场景。
下一节预告：我们将探讨 Agentic RAG——如何让 Agent 更智能地检索和利用知识库。