多智能体模式

前置阅读：多智能体概览 · 智能体 Agent

本章深入讲解三大核心多智能体模式：Subagents（子代理）、Handoffs（任务交接）、Router（路由分发）。每种模式都包含概念说明、适用场景和完整代码示例。

必须深刻理解，不能跳过：worker 与编排的分离。

本章所有"专家 Agent"都是用 from langchain.agents import create_agent 构建的 worker--它们本身是完整的 Agent Loop，但不知道彼此的存在。协作行为（谁先执行、谁交接给谁、如何汇总）由编排层（langgraph-supervisor / langgraph-swarm）定义。worker 只管"接到输入 -> 调工具 -> 返回输出"，编排层管"把输出送给下一个 worker"。

安装依赖

# worker 用 LangChain create_agent
pip install langchain langgraph

# 编排层二选一（或都装）
pip install langgraph-supervisor  # Supervisor / 层级 handoff
pip install langgraph-swarm       # Swarm / 平级 handoff / Router

Subagents 子代理

概念

Subagents 模式采用层级结构：一个父 Agent（Supervisor）负责理解用户意图、拆解任务，然后将子任务委派给专职的子 Agent 执行。子 Agent 完成后将结果返回给父 Agent，由父 Agent 汇总输出最终答案。

这是多智能体系统中最通用的模式--当你不确定选哪种模式时，Subagents 通常是安全的默认选择。

适用场景

• 研究报告生成：搜索子 Agent 收集资料 -> 分析子 Agent 提炼观点 -> 写作子 Agent 组织文章
• 数据处理流水线：提取子 Agent 解析数据 -> 转换子 Agent 清洗格式 -> 加载子 Agent 写入存储
• 项目管理：分解任务 -> 分配给不同专家 -> 汇总进度

代码示例：研究 Agent（langgraph-supervisor）

以下示例用 langgraph-supervisor 的 create_supervisor 编排两个 worker--搜索子 Agent 和分析子 Agent。每个 worker 用 create_agent 构建，create_supervisor 自动为它们生成 handoff 工具。

import os
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.tools import tool
from langgraph_supervisor import create_supervisor

# 模型配置来自环境变量；"openai:gpt-4o" 是官方示例名称，可能随时间变化
model = init_chat_model(os.environ["LLM_MODEL"])

# ============ 子 Agent 1：搜索（worker）============
@tool
def web_search(query: str) -> str:
    """搜索网页获取最新信息"""
    # 实际项目中接入 Tavily / SerpAPI 等
    return f"搜索 '{query}' 的结果：发现 3 篇相关论文和 5 篇行业报告"

@tool
def academic_search(query: str) -> str:
    """搜索学术论文"""
    return f"学术搜索 '{query}'：找到 2 篇高引用论文"

search_agent = create_agent(
    model,
    tools=[web_search, academic_search],
    system_prompt=(
        "你是专业的信息检索员。用户给你搜索关键词，你负责全面搜索并返回原始结果。"
        "不要分析，只搜索和汇总搜索结果。"
    ),
    name="search_expert",  # name 是 handoff 的目标标识，必填
)

# ============ 子 Agent 2：分析（worker）============
@tool
def extract_key_points(text: str) -> str:
    """从文本中提取关键观点"""
    return (
        "关键观点：1) AI 安全研究投入增长 200% "
        "2) 对齐问题成为核心议题 3) 多国出台监管框架"
    )

@tool
def generate_summary(points: str) -> str:
    """根据关键观点生成结构化摘要"""
    return f"结构化摘要：\n- 趋势：{points}\n- 影响：深远\n- 建议：持续关注"

analysis_agent = create_agent(
    model,
    tools=[extract_key_points, generate_summary],
    system_prompt="你是数据分析专家。接收原始资料后提取关键观点并生成结构化摘要。",
    name="analysis_expert",
)

# ============ 编排层：Supervisor ============
# create_supervisor 自动为每个 worker 生成 handoff 工具
# supervisor 通过调用 handoff 工具把控制权交给对应 worker
workflow = create_supervisor(
    agents=[search_agent, analysis_agent],
    model=model,
    prompt=(
        "你是研究项目管理者。收到用户的研究需求后："
        "1. 先委派搜索任务收集资料（search_expert）"
        "2. 再将搜索结果交给分析 Agent 提炼（analysis_expert）"
        "3. 最后综合所有结果，用清晰的格式回复用户"
    ),
    output_mode="last_message",  # 只把 worker 最后一条消息加入历史
)

# 编译并执行
app = workflow.compile()
result = app.invoke(
    {"messages": [{"role": "user", "content": "帮我研究 2024 年 AI 安全领域的最新进展"}]}
)
print(result["messages"][-1].content)

Supervisor 的委派机制

create_supervisor 内部做了什么：

1. 为每个 worker 生成一个 create_handoff_tool(agent_name="search_expert", description="...") 工具
2. Supervisor LLM 调用某个 handoff 工具时，工具内部执行 Command(goto="search_expert", update={...})
3. LangGraph 运行时跳转到 search_expert 节点，worker 执行完毕后控制权返回 supervisor
4. Supervisor 根据返回结果决定下一步（继续 handoff 给另一个 worker，或汇总输出）

关键设计原则：

1. worker 名称清晰：name= 参数是 handoff 的目标标识，supervisor 通过它选择 worker
2. worker 独立：子 Agent 不依赖父 Agent 的上下文，通过 state 中的 messages 接收全部所需信息
3. 结果精简：output_mode="last_message" 只把 worker 最后一条消息加入历史，避免 supervisor 上下文膨胀

# 好的做法：worker name 明确表达职责
search_agent = create_agent(
    model, tools=[...], system_prompt="...", name="search_expert"
)

# 不好的做法：name 模糊或缺失
search_agent = create_agent(model, tools=[...], system_prompt="...")  # name 缺失，handoff 无法定位

Handoffs 任务交接

概念

Handoffs 模式下，Agent 之间直接传递控制权。当前 Agent 判断任务超出自己的能力范围时，主动将对话连同上下文"交接"给更合适的 Agent。

与 Subagents 的关键区别：Subagents 有明确的上下级关系（supervisor -> worker），而 Handoffs 是平级 Agent 之间的协作--控制权可以在任意 Agent 间流转。

必须深刻理解，不能跳过：handoff = 控制权转移 + 状态更新。

create_handoff_tool 生成的工具，其内部实现是：

from langgraph.types import Command

def handoff_to_billing(**kwargs):
    return Command(
        goto="billing_agent",           # 控制权转移到哪个 worker
        update={"messages": [...]},      # 把当前上下文写入共享 state
    )

这意味着 handoff 不是"调用另一个 Agent 的函数"，而是"在 LangGraph 图中跳转到另一个节点，并更新共享状态"。这就是为什么 handoff 工具来自 langgraph-supervisor / langgraph-swarm，而不是 LangChain 核心--它本质是 LangGraph 的图跳转语义。

适用场景

• 客服系统：分诊 Agent 判断问题类型 -> 交接给账单 / 技术 / 售后专家
• 多步骤审批：初审 Agent -> 复审 Agent -> 终审 Agent
• 问题升级：一线 Agent 无法解决 -> 交接给高级 Agent

代码示例：客服交接系统（langgraph-swarm）

以下示例用 langgraph-swarm 的 create_swarm + create_handoff_tool 实现客服交接。每个专家 worker 用 create_agent 构建，并在工具列表中加入 create_handoff_tool 实现平级交接。

import os
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.tools import tool
from langgraph_swarm import create_swarm, create_handoff_tool

model = init_chat_model(os.environ["LLM_MODEL"])

# ============ 专家 Agent 定义（worker）============

# 账单专家
@tool
def check_billing(account_id: str) -> str:
    """查询账单信息"""
    return f"账户 {account_id}：当前余额 ¥128.50，上月消费 ¥89.00，无欠费"

@tool
def apply_refund(account_id: str, amount: float, reason: str) -> str:
    """申请退款"""
    return f"退款申请已提交：账户 {account_id}，金额 ¥{amount}，原因：{reason}"

billing_agent = create_agent(
    model,
    tools=[
        check_billing,
        apply_refund,
        # 账单专家可以交接给技术专家或通用客服
        create_handoff_tool(agent_name="tech_agent", description="转交技术支持专家"),
        create_handoff_tool(agent_name="general_agent", description="转交通用客服"),
    ],
    system_prompt=(
        "你是账单专家。只处理账单查询、费用争议、退款申请。"
        "如果用户问题与账单无关，使用交接工具转给对应专家。"
    ),
    name="billing_agent",
)

# 技术专家
@tool
def diagnose_network(issue: str) -> str:
    """诊断网络问题"""
    return f"网络诊断结果：{issue} - 检测到 DNS 解析延迟，建议更换 DNS 服务器"

@tool
def check_service_status(service: str) -> str:
    """检查服务状态"""
    return f"服务 {service} 状态：运行正常，延迟 23ms，可用率 99.97%"

tech_agent = create_agent(
    model,
    tools=[
        diagnose_network,
        check_service_status,
        create_handoff_tool(agent_name="billing_agent", description="转交账单专家"),
        create_handoff_tool(agent_name="general_agent", description="转交通用客服"),
    ],
    system_prompt="你是技术支持专家。只处理技术故障、网络问题、服务状态查询。",
    name="tech_agent",
)

# 通用客服（入口分诊）
@tool
def query_faq(question: str) -> str:
    """查询常见问题"""
    return f"FAQ 匹配结果：关于 '{question}' 的常见解答..."

general_agent = create_agent(
    model,
    tools=[
        query_faq,
        create_handoff_tool(agent_name="billing_agent", description="转交账单专家"),
        create_handoff_tool(agent_name="tech_agent", description="转交技术支持专家"),
    ],
    system_prompt=(
        "你是客服分诊 Agent。职责：1. 理解用户问题类别 "
        "2. 交接给合适的专家 3. 一般性问题自行用 FAQ 回答"
    ),
    name="general_agent",
)

# ============ 编排层：Swarm ============
# default_active_agent 指定入口 worker（充当分诊角色）
workflow = create_swarm(
    agents=[billing_agent, tech_agent, general_agent],
    default_active_agent="general_agent",  # 通用客服作为入口分诊
)
app = workflow.compile()

# 执行
result = app.invoke(
    {"messages": [{"role": "user", "content": "我的网络突然很慢，而且上个月的账单好像多扣了钱"}]},
    config={"configurable": {"thread_id": "session-1"}},
)
print(result["messages"][-1].content)

交接机制与状态传递

Handoffs 的关键在于上下文的完整传递。create_handoff_tool 内部通过 Command(goto=..., update={...}) 将当前对话历史写入共享 state，目标 Agent 接管后从 state 中读取完整上下文：

# create_handoff_tool 的等价手动实现（展示原理）
from langgraph.types import Command
from langchain.tools import tool

@tool
def handoff_to_billing(user_message: str) -> Command:
    """交接给账单专家。当用户询问账单、费用、退款相关问题时使用。"""
    return Command(
        goto="billing_agent",
        update={
            # 把交接原因和上下文写入 messages，目标 worker 能读到
            "messages": [
                {"role": "user", "content": f"[交接上下文] {user_message}"}
            ]
        },
    )

避免交接死循环：Swarm 通过 thread_id 维护状态，但 LLM 仍可能反复交接。建议：

1. 在 system_prompt 中明确"已交接过的问题不要再交回"
2. 使用 LangGraph 的 recursion_limit（在 invoke 时通过 config 传入，不是 create_agent 参数）限制图跳转次数

# 正确做法：在 invoke/stream 时通过 config 限制递归
result = app.invoke(
    {"messages": [...]},
    config={
        "configurable": {"thread_id": "session-1"},
        "recursion_limit": 20,  # 限制图跳转次数
    },
)

旧版（仅用于读旧项目）

旧教程可能用 create_react_agent(model, tools, prompt=...) 构建 worker，或手动把 worker 封装成 @tool 函数实现 Supervisor。这些方式在 LangChain 1.0 后已弃用：

• create_react_agent -> 改用 create_agent，参数 prompt= -> system_prompt=
• 手动 @tool 封装 worker -> 改用 langgraph-supervisor 的 create_supervisor 或 langgraph-swarm 的 create_swarm

Router 路由分发

概念

Router 模式是最简单的多智能体架构：一个中央路由器接收所有请求，根据分类规则将请求分发给对应的专家 Agent。路由器本身不处理业务逻辑，只负责"分类"和"转发"。

与 Handoffs 的区别：Router 是单跳分发（路由器 -> 专家），不存在专家之间的横向交接。在 langgraph-swarm 中，Router 是 Swarm 的特例--default_active_agent 充当路由器，专家 worker 不配置 create_handoff_tool（只能被路由到，不能主动交接）。

适用场景

• 知识问答：数学问题 -> 数学 Agent，历史问题 -> 历史 Agent
• 多语言支持：检测语言 -> 对应语言的 Agent
• 服务台：根据请求类型分发到不同的处理 Agent

代码示例：知识路由器（langgraph-swarm）

import os
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.tools import tool
from langgraph_swarm import create_swarm, create_handoff_tool

model = init_chat_model(os.environ["LLM_MODEL"])

# ============ 专家 Agent（worker，不配置出向 handoff）============

@tool
def solve_equation(equation: str) -> str:
    """求解数学方程"""
    return f"方程 {equation} 的解：x = 42"

@tool
def explain_theorem(name: str) -> str:
    """解释数学定理"""
    return f"{name} 定理：...详细解释..."

math_agent = create_agent(
    model,
    tools=[solve_equation, explain_theorem],
    system_prompt="你是数学教授。用严谨但易懂的方式解答数学问题，提供解题步骤。",
    name="math_agent",
)

@tool
def lookup_experiment(topic: str) -> str:
    """查找科学实验"""
    return f"关于 {topic} 的经典实验：...实验描述..."

@tool
def explain_phenomenon(phenomenon: str) -> str:
    """解释科学现象"""
    return f"{phenomenon} 的科学解释：...原理说明..."

science_agent = create_agent(
    model,
    tools=[lookup_experiment, explain_phenomenon],
    system_prompt="你是自然科学老师。用通俗的语言解释科学原理，配合实验例子。",
    name="science_agent",
)

@tool
def lookup_event(event: str) -> str:
    """查找历史事件"""
    return f"历史事件 {event}：...时间线和背景..."

@tool
def analyze_impact(event: str) -> str:
    """分析历史事件的影响"""
    return f"{event} 的影响：...社会、经济、文化层面分析..."

history_agent = create_agent(
    model,
    tools=[lookup_event, analyze_impact],
    system_prompt="你是历史学家。从多角度分析历史事件，注重因果关系和现代启示。",
    name="history_agent",
)

# ============ 路由器 Agent（入口，配置出向 handoff）============
router_agent = create_agent(
    model,
    tools=[
        create_handoff_tool(agent_name="math_agent", description="转交数学专家"),
        create_handoff_tool(agent_name="science_agent", description="转交科学专家"),
        create_handoff_tool(agent_name="history_agent", description="转交历史专家"),
    ],
    system_prompt=(
        "你是知识问答路由器。你的唯一任务是判断用户问题属于哪个领域，然后交接给对应专家。"
        "路由规则：数学/计算/方程 -> math_agent；物理/化学/生物 -> science_agent；"
        "历史/事件/人物 -> history_agent。不要自己回答，务必交接给专家。"
    ),
    name="router_agent",
)

# ============ 编排层：Swarm（Router 模式）============
workflow = create_swarm(
    agents=[router_agent, math_agent, science_agent, history_agent],
    default_active_agent="router_agent",  # 路由器作为入口
)
app = workflow.compile()

# 执行
result = app.invoke(
    {"messages": [{"role": "user", "content": "为什么苹果会从树上掉下来？牛顿是怎么发现万有引力的？"}]},
    config={"configurable": {"thread_id": "session-1"}},
)
print(result["messages"][-1].content)

路由决策逻辑

路由器的决策可以基于两种策略：

策略一：LLM 自然语言分类（上面示例的方式）

路由器本身是一个 worker，通过 system_prompt 和 handoff 工具描述让 LLM 判断路由目标。优点是灵活，缺点是有一定的推理成本。

策略二：嵌入向量匹配

预先为每个专家 Agent 创建领域描述的嵌入向量，用户问题通过向量相似度匹配最合适的专家。优点是快速且确定性高：

from langchain.embeddings import init_embeddings

embeddings = init_embeddings(os.environ["EMBEDDING_MODEL"])

# 预定义每个专家的领域描述
domain_descriptions = {
    "math_agent": "数学 方程 计算 几何 代数 微积分 概率 统计",
    "science_agent": "物理 化学 生物 天文 地理 自然科学 实验",
    "history_agent": "历史 事件 人物 朝代 文明 战争 革命",
}

# 预计算领域嵌入
domain_embeddings = {
    name: embeddings.embed_query(desc)
    for name, desc in domain_descriptions.items()
}

def route_by_embedding(question: str) -> str:
    """基于嵌入向量的路由"""
    q_embedding = embeddings.embed_query(question)
    # 计算余弦相似度，选择最匹配的领域
    scores = {
        name: cosine_similarity(q_embedding, emb)
        for name, emb in domain_embeddings.items()
    }
    return max(scores, key=scores.get)

三种模式对比

维度	Subagents	Handoffs	Router
拓扑结构	树形（supervisor -> worker）	网状（任意方向）	星形（router -> 专家）
控制权	Supervisor 始终持有	在 worker 间流转	路由器单次分发
落地包	langgraph-supervisor	langgraph-swarm / supervisor	langgraph-swarm
核心 API	create_supervisor	create_handoff_tool + create_swarm	create_swarm（专家无出向 handoff）
复杂度	中等	中等	低
灵活性	高（动态委派）	高（动态交接）	低（固定分发）
状态管理	Supervisor 维护全局状态	Swarm state 共享	无状态或极少状态
典型用例	研究、项目管理	客服、审批流	问答、分类分发

数据流对比

前端类比

三种模式在前端也有对应物：

• Subagents ≈ React 的组合组件模式：父组件负责状态管理和数据协调，子组件各自渲染独立区域
• Handoffs ≈ React Router 路由切换：页面之间传递 state，控制权从一个页面转移到另一个页面
• Router ≈ API Gateway / BFF：统一入口根据请求类型分发到不同的微服务

原生语义：Agent 的"分发"是基于语义理解而非 URL 匹配，因此分发结果具有概率性而非确定性。handoff 底层是 LangGraph 的 Command(goto=, update=) 图跳转，不是 HTTP 调用或函数调用。

EnviroNexus 场景映射

EnviroNexus 环保知识库的客服分诊场景是典型的 Handoffs/Router 模式：

用户咨询 -> general_agent（分诊）
  -> handoff 到 billing_agent（如果是会员/套餐问题）
  -> handoff 到 standard_agent（如果是标准号查询）
  -> handoff 到 factor_agent（如果是排放因子计算）
  -> handoff 到 regulation_agent（如果是法规解读）

每个专家 worker 用 create_agent 构建：
  - standard_agent：检索标准号 + MethodCard
  - factor_agent：factor_alias 匹配 + 排放因子计算
  - regulation_agent：法规正文检索 + 条款解读

编排层用 langgraph-swarm 的 create_swarm，
general_agent 作为 default_active_agent 充当分诊入口。

铁律：Retriever 找 Document 不回答；Tool 是模型能力入口、内部可调 Retriever；无 evidence_refs 不得出标准结论。

常见问题

Q：Handoffs 和 Router 看起来很像，怎么区分？

A：关键区别在于是否存在横向交接：

• Router：路由器 -> 专家，专家直接回复，不会再交给其他专家（专家没有出向 handoff 工具）
• Handoffs：Agent A -> Agent B -> Agent C，控制权可以多次转移（每个 Agent 都有 create_handoff_tool）

Q：Subagents 的 Supervisor 可以嵌套吗？

A：可以。Supervisor A 管理的 worker B 本身也可以是一个 create_supervisor 编排的子图，形成多级层次。但建议层级不超过 3 层，否则调试和错误追踪会变得困难。

Q：多个子 Agent 可以并行执行吗？

A：create_supervisor 支持 parallel_tool_calls=True（仅 OpenAI 和 Anthropic 模型），允许 supervisor 同时 handoff 给多个 worker。如果需要更精细的并行控制，应使用 LangGraph 的 并行节点 特性。

Q：路由器分错了怎么办？

A：几种应对策略：

1. 优化路由器的 system_prompt，增加分类示例
2. 使用嵌入向量匹配提高准确率
3. 在专家 worker 中添加"拒绝处理"逻辑--当发现问题不属于自己时 handoff 回路由器
4. 添加兜底 worker 处理无法分类的请求

Q：什么时候用 langgraph-supervisor，什么时候用 langgraph-swarm？

A：

• 层级管理、父 Agent 汇总结果 -> langgraph-supervisor（Supervisor 始终持有控制权）
• 平级交接、控制权在 Agent 间流转 -> langgraph-swarm（无固定父节点）
• 单跳路由分发 -> langgraph-swarm（default_active_agent 充当路由器，专家无出向 handoff）

下一步

• 高级多智能体 - Skills 技能模式（Deep Agents）、LangGraph 自定义工作流、内存共享
• 多智能体概览 - 回顾五种模式全景和选型决策
• Agents - 单 Agent 基础知识
• Deep Agents Subagents - 长周期研究、多文件多 Subagent 的高级 Harness

参考资源

• LangGraph Supervisor 文档
• LangGraph Swarm 文档
• create_supervisor API
• create_swarm API
• LangGraph Multi-Agent 概念