12.1 多智能体工作流介绍
什么是多智能体系统？
想象一个公司里有不同部门：销售、技术、财务、客服。每个部门有专人负责，遇到问题时会转交给最合适的人处理。多智能体系统就是这样的工作方式——把复杂任务分配给多个专业化的 AI Agent，让它们协作完成。
为什么需要多智能体？
当你的 AI 应用遇到以下情况时，单个 Agent 就不够用了：
| 问题 | 单 Agent 的困境 | 多 Agent 的解决方案 |
|---|---|---|
| 工具太多 | Agent 不知道该选哪个工具 | 每个 Agent 只负责几个相关工具 |
| 上下文溢出 | 对话历史超出模型限制 | 每个 Agent 只保留相关上下文 |
| 任务复杂 | 一个模型难以胜任所有领域 | 专业 Agent 各司其职 |
一句话总结：与其让一个"全能选手"处理所有事情，不如让多个"专家"分工协作。
两种核心架构模式
LangGraph 提供了两种主要的多智能体协作模式：
1. Tool Calling 模式（集中式）
这是最常用的模式，就像一个项目经理统筹调度团队成员：
用户请求 → Supervisor（总调度）→ 选择合适的 Worker Agent → 执行任务 → 返回结果给 Supervisor → 汇总回复用户特点：
- ✅ 流程清晰可控
- ✅ 所有通信经过中心节点
- ✅ 易于调试和监控
- ❌ Supervisor 可能成为瓶颈
适用场景：
- 任务有明确的阶段划分
- 需要严格的执行顺序
- 多领域协作（日历、邮件、CRM 等）
完整实现：三层架构的 Supervisor 模式
Supervisor 模式的核心思想是分层处理：
┌─────────────────────────────────────────────────────────────┐
│ 第三层：Supervisor Agent │
│ - 理解用户自然语言请求 │
│ - 决定调用哪个专业 Agent │
│ - 汇总结果回复用户 │
├─────────────────────────────────────────────────────────────┤
│ 第二层：专业子 Agent（Calendar Agent、Email Agent） │
│ - 接收自然语言请求 │
│ - 转换为结构化的工具调用 │
│ - 处理领域特定逻辑 │
├─────────────────────────────────────────────────────────────┤
│ 第一层：底层 API 工具 │
│ - 需要精确格式的输入（ISO 日期、邮件地址等） │
│ - 执行实际操作 │
└─────────────────────────────────────────────────────────────┘Step 1：定义底层 API 工具
python
from langchain.tools import tool
# 日历相关的底层工具
@tool
def create_calendar_event(
title: str,
start_time: str, # 必须是 ISO 格式：2025-01-15T14:00:00
end_time: str,
attendees: list[str],
location: str = ""
) -> str:
"""创建日历事件，需要精确的 ISO 日期格式"""
# 实际项目中这里会调用 Google Calendar API
return f"✅ 事件已创建: {title}，时间: {start_time} - {end_time}"
@tool
def get_available_time_slots(
attendees: list[str],
date: str,
duration_minutes: int
) -> list[str]:
"""查询指定日期的可用时间段"""
# 实际项目中这里会查询日历系统
return ["09:00", "14:00", "16:00"]
# 邮件相关的底层工具
@tool
def send_email(
to: list[str],
subject: str,
body: str,
cc: list[str] = []
) -> str:
"""发送邮件，需要格式正确的邮箱地址"""
# 实际项目中这里会调用邮件 API
return f"✅ 邮件已发送给: {', '.join(to)}"Step 2：创建专业子 Agent
python
from langchain.agents import create_agent
# 日历专家 Agent
# 职责：将自然语言（"明天下午2点"）转换为 ISO 格式
calendar_agent = create_agent(
model="openai:gpt-4o",
tools=[create_calendar_event, get_available_time_slots],
system_prompt="""你是日历管理专家。
你的职责：
1. 将用户的自然语言日期（如"明天下午2点"）解析为 ISO 格式
2. 在创建事件前，先检查时间是否可用
3. 处理时区转换和冲突检测
始终使用工具完成任务，不要只是回复文字。"""
)
# 邮件专家 Agent
# 职责：撰写专业邮件，提取收件人信息
email_agent = create_agent(
model="openai:gpt-4o",
tools=[send_email],
system_prompt="""你是邮件撰写专家。
你的职责：
1. 根据用户需求撰写专业、得体的邮件
2. 从上下文中提取收件人邮箱地址
3. 生成合适的邮件主题
始终使用 send_email 工具发送邮件。"""
)Step 3：将子 Agent 包装为 Supervisor 的工具
python
@tool
def schedule_event(request: str) -> str:
"""
处理日程安排请求。
支持自然语言输入，如："安排明天下午2点与张三的会议"
"""
result = calendar_agent.invoke({
"messages": [{"role": "user", "content": request}]
})
return result["messages"][-1].text
@tool
def manage_email(request: str) -> str:
"""
处理邮件相关请求。
支持自然语言输入，如："给团队发邮件通知明天的会议"
"""
result = email_agent.invoke({
"messages": [{"role": "user", "content": request}]
})
return result["messages"][-1].textStep 4：创建 Supervisor Agent
python
supervisor_agent = create_agent(
model="openai:gpt-4o",
tools=[schedule_event, manage_email],
system_prompt="""你是智能个人助手，负责协调日程和邮件管理。
当用户提出请求时：
1. 分析请求涉及哪些领域（日历、邮件或两者）
2. 按正确顺序调用相应工具
3. 汇总结果，给用户清晰的反馈
对于复合请求（如"安排会议并发邮件通知"），请依次调用多个工具。"""
)Step 5：运行示例
python
# 单领域请求
result = supervisor_agent.invoke({
"messages": [{
"role": "user",
"content": "帮我安排明天下午2点的团队周会"
}]
})
# Supervisor → schedule_event → calendar_agent → create_calendar_event
# 多领域复合请求
result = supervisor_agent.invoke({
"messages": [{
"role": "user",
"content": "安排下周一上午10点与设计团队的评审会议，然后给他们发邮件提醒带上设计稿"
}]
})
# Supervisor → schedule_event → calendar_agent → create_calendar_event
# → manage_email → email_agent → send_email2. Handoffs 模式（去中心化）
这种模式像接力赛，Agent 之间直接传递"接力棒"：
用户请求 → Agent A（发现需要其他专业）→ 直接交接给 Agent B → Agent B 处理完成 → 回复用户特点：
- ✅ 没有单点瓶颈
- ✅ Agent 可以直接与用户交互
- ✅ 更灵活的对话流程
- ❌ 流程不如集中式清晰
适用场景：
- 客服系统（转接到不同部门）
- 多轮对话需要切换专家
- 需要 Agent 直接与用户交互
层级式多智能体团队
当系统变得更复杂时，可以建立多层级的 Agent 架构：
┌─────────────┐
│ 总 Supervisor │
└──────┬──────┘
┌───────────────┼───────────────┐
▼ ▼ ▼
┌────────────┐ ┌────────────┐ ┌────────────┐
│ 研究团队主管 │ │ 开发团队主管 │ │ 运营团队主管 │
└─────┬──────┘ └─────┬──────┘ └─────┬──────┘
│ │ │
┌─────┴─────┐ ┌─────┴─────┐ ┌─────┴─────┐
│ 搜索专家 │ │ 前端开发 │ │ 数据分析 │
│ 论文分析 │ │ 后端开发 │ │ 报告生成 │
└───────────┘ └───────────┘ └───────────┘三层架构设计：
| 层级 | 角色 | 职责 |
|---|---|---|
| 顶层 | 总 Supervisor | 理解用户需求，分配给合适的团队 |
| 中层 | 团队主管 | 协调团队内的专业 Agent |
| 底层 | 专业 Agent | 执行具体任务，使用专业工具 |
优势：
- 每层职责清晰，易于维护
- 添加新功能不影响其他模块
- 可以独立测试和优化每个层级
Human-in-the-Loop：关键操作的人工审批
在生产环境中，某些敏感操作（如发送邮件、创建日程、转账）需要人工确认后才能执行。LangGraph 提供了 HumanInTheLoopMiddleware 来实现这一功能。
基本配置
python
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.memory import InMemorySaver
# 为日历 Agent 添加人工审批
calendar_agent = create_agent(
model="openai:gpt-4o",
tools=[create_calendar_event, get_available_time_slots],
system_prompt=CALENDAR_AGENT_PROMPT,
middleware=[
HumanInTheLoopMiddleware(
interrupt_on={"create_calendar_event": True}, # 创建事件时暂停
description_prefix="📅 日历事件待审批"
)
]
)
# 为邮件 Agent 添加人工审批
email_agent = create_agent(
model="openai:gpt-4o",
tools=[send_email],
system_prompt=EMAIL_AGENT_PROMPT,
middleware=[
HumanInTheLoopMiddleware(
interrupt_on={"send_email": True}, # 发送邮件时暂停
description_prefix="📧 外发邮件待审批"
)
]
)
# Supervisor 需要配置 checkpointer 来支持暂停/恢复
supervisor_agent = create_agent(
model="openai:gpt-4o",
tools=[schedule_event, manage_email],
system_prompt=SUPERVISOR_PROMPT,
checkpointer=InMemorySaver() # 生产环境建议使用 PostgresSaver
)捕获和处理中断
python
config = {"configurable": {"thread_id": "user-session-123"}}
interrupts = []
# 流式执行，捕获中断点
for step in supervisor_agent.stream(
{"messages": [{"role": "user", "content": "安排明天的会议并发邮件通知"}]},
config
):
for update in step.values():
if isinstance(update, dict):
# 正常的消息更新
for message in update.get("messages", []):
message.pretty_print()
else:
# 这是一个中断，需要人工审批
interrupt_ = update[0]
interrupts.append(interrupt_)
print(f"⏸️ 等待审批: {interrupt_.value}")审批、编辑或拒绝
python
from langgraph.types import Command
resume = {}
for interrupt_ in interrupts:
# 显示待审批的操作详情
action = interrupt_.value["action_requests"][0]
print(f"操作: {action['tool']}")
print(f"参数: {action['arguments']}")
# 用户决策（实际项目中这里是 UI 交互）
user_decision = input("请选择 [approve/edit/reject]: ")
if user_decision == "approve":
# 直接批准
resume[interrupt_.id] = {"decisions": [{"type": "approve"}]}
elif user_decision == "edit":
# 修改后批准（例如修改邮件主题）
edited_action = action.copy()
edited_action["arguments"]["subject"] = "【重要】" + action["arguments"]["subject"]
resume[interrupt_.id] = {
"decisions": [{"type": "edit", "edited_action": edited_action}]
}
else:
# 拒绝执行
resume[interrupt_.id] = {"decisions": [{"type": "reject"}]}
# 恢复执行
for step in supervisor_agent.stream(Command(resume=resume), config):
for message in step.get("messages", []):
message.pretty_print()高级：信息流控制
多智能体系统的成功关键在于上下文工程（Context Engineering）——精确控制每个 Agent 看到什么信息、返回什么信息。
1. 控制子 Agent 的输入
方法一：传递完整对话上下文
有时子 Agent 需要理解完整的对话历史才能正确处理请求（例如用户说"和昨天一样的时间"）：
python
from langchain.tools import tool, ToolRuntime
@tool
def schedule_event(
request: str,
runtime: ToolRuntime # 注入运行时上下文
) -> str:
"""处理日程安排，可以理解对话上下文"""
# 获取原始用户消息
original_user_message = next(
msg for msg in runtime.state["messages"]
if msg.type == "human"
)
# 构建包含上下文的提示
prompt = f"""用户原始请求: {original_user_message.text}
当前子任务: {request}
请根据上下文理解用户意图。"""
result = calendar_agent.invoke({
"messages": [{"role": "user", "content": prompt}]
})
return result["messages"][-1].text方法二：传递自定义状态
python
from langchain.agents import AgentState
class CustomState(AgentState):
user_timezone: str
user_preferences: dict
@tool
def schedule_event(
request: str,
runtime: ToolRuntime[None, CustomState]
) -> str:
"""处理日程安排，使用用户偏好"""
result = calendar_agent.invoke({
"messages": [{"role": "user", "content": request}],
"user_timezone": runtime.state["user_timezone"],
"preferences": runtime.state["user_preferences"]
})
return result["messages"][-1].text2. 控制子 Agent 的输出
方法一：返回结构化数据
python
import json
@tool
def schedule_event(request: str) -> str:
"""返回结构化的执行结果"""
result = calendar_agent.invoke({
"messages": [{"role": "user", "content": request}]
})
# 返回结构化 JSON，方便 Supervisor 解析
return json.dumps({
"status": "success",
"event_id": "evt_abc123",
"event_title": "团队周会",
"event_time": "2025-01-15T14:00:00",
"summary": result["messages"][-1].text
}, ensure_ascii=False)方法二：使用 Command 更新共享状态
python
from typing import Annotated
from langchain.tools import InjectedToolCallId
from langchain_core.messages import ToolMessage
from langgraph.types import Command
@tool
def schedule_event(
request: str,
tool_call_id: Annotated[str, InjectedToolCallId]
) -> Command:
"""使用 Command 同时更新消息和共享状态"""
result = calendar_agent.invoke({
"messages": [{"role": "user", "content": request}]
})
return Command(update={
# 更新共享状态
"last_created_event": result.get("event_id"),
"pending_notifications": True,
# 返回工具消息
"messages": [
ToolMessage(
content=result["messages"][-1].text,
tool_call_id=tool_call_id
)
]
})关键设计原则
1. 工具命名和描述
子 Agent 的名称和描述直接影响 Supervisor 的路由决策：
python
# ❌ 不好的命名 —— Supervisor 不知道什么时候该用
@tool("agent1", description="处理一些事情")
def bad_agent(): ...
# ✅ 好的命名 —— 清晰说明能力和适用场景
@tool(
"financial_analyst",
description="分析财务报表、计算投资回报率、生成财务预测报告。适用于涉及财务数据分析的请求。"
)
def good_agent(): ...2. 何时使用多智能体
| 情况 | 推荐方案 |
|---|---|
| 工具少于 5 个，任务简单 | 单 Agent 即可 |
| 工具 5-15 个，有明确分类 | Supervisor + 2-3 个专业 Agent |
| 工具 15+ 个，多领域复杂任务 | 层级式多智能体架构 |
| 需要 Agent 直接与用户交互 | Handoffs 模式 |
| 敏感操作需要人工确认 | 添加 HumanInTheLoopMiddleware |
3. Tool Calling vs Handoffs 选择指南
| 考虑因素 | Tool Calling（Supervisor） | Handoffs（去中心化） |
|---|---|---|
| 控制流 | ✅ 集中控制，流程清晰 | ❌ 分散控制，较难追踪 |
| 用户交互 | ❌ 子 Agent 不直接与用户对话 | ✅ 当前 Agent 直接与用户交互 |
| 对话自然度 | ❌ 较机械 | ✅ 更像人类对话 |
| 调试难度 | ✅ 容易调试 | ❌ 需要追踪 Agent 切换 |
| 适用场景 | 任务编排、批处理 | 客服、多领域咨询 |
多智能体协作示例
这张图展示了一个典型的多智能体协作流程，多个专业 Agent 共同完成一个复杂任务。
延伸阅读
下一节预告：我们将通过具体的代码案例，手把手实现一个完整的多智能体系统。