第17章 多 Agent 模式实战

17.1 引言

单个 Agent 能力有限——它只有一组工具、一种提示、一个执行循环。当任务复杂度上升时（如"帮我分析这家公司的财报，然后写一份投资建议书"），单个 Agent 要么因为工具太多导致 LLM 选择困难，要么因为系统提示过长影响推理质量。多 Agent 系统通过分工协作解决这个问题：不同的 Agent 负责不同的能力域，通过明确的通信机制协同完成复杂任务。

LangGraph 的底层基础设施——StateGraph、Send、Command、子图——天然支持多 Agent 模式的构建。图的节点可以是子图（即嵌套的 Agent），条件边可以实现动态路由，Send 支持一对多的任务分发，Command 支持跨图的状态更新和导航。

本章将系统介绍基于 LangGraph 实现的四种典型多 Agent 架构：Supervisor（监督者）、Swarm（蜂群）、分层 Agent 和协作 Agent。每种模式都有其适用场景和权衡，我们会结合源码和实战代码深入分析它们的设计原理。

本章要点

1. Supervisor 模式——中央调度者分配任务给专业 Agent
2. Swarm 模式——Agent 之间点对点的控制权交接
3. 分层 Agent——多层级的 Supervisor 树形结构
4. 协作 Agent——共享状态的对等协作
5. 基于 LangGraph 原语实现各种模式的核心技术

17.2 多 Agent 的核心抽象

17.2.1 Agent 作为子图

在 LangGraph 中，一个 Agent 本质上就是一个 CompiledStateGraph。多 Agent 系统就是一个图中嵌套了多个子图：

# 每个 Agent 是一个独立的图
research_agent = create_react_agent(model, research_tools, name="researcher")
writer_agent = create_react_agent(model, writing_tools, name="writer")

# 多 Agent 系统是一个包含子图的父图
builder = StateGraph(TeamState)
builder.add_node("researcher", research_agent)
builder.add_node("writer", writer_agent)

17.2.2 通信机制的三种模式

17.3 Supervisor 模式

17.3.1 架构概述

Supervisor 模式是最直观的多 Agent 架构：一个"主管"Agent 接收用户请求，分析任务需求，然后将子任务分配给专业 Agent。每个专业 Agent 完成后将结果返回给 Supervisor，Supervisor 决定是否需要更多工作。

17.3.2 实现方式

from typing import Annotated, Literal, TypedDict
from langchain_core.messages import BaseMessage, HumanMessage
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import create_react_agent
from langgraph.types import Command

class SupervisorState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    next_agent: str

# 创建专业 Agent
research_agent = create_react_agent(model, research_tools, name="researcher")
writer_agent = create_react_agent(model, writing_tools, name="writer")

def supervisor_node(state: SupervisorState) -> Command:
    """Supervisor 决定下一步交给哪个 Agent"""
    response = supervisor_llm.invoke([
        SystemMessage(content="""你是一个任务分配主管。
        分析用户请求，决定交给哪个 Agent：
        - researcher: 负责信息检索和数据分析
        - writer: 负责内容创作和文档编写
        - FINISH: 任务完成，返回最终结果"""),
        *state["messages"]
    ])

    # 解析 LLM 的决策
    next_agent = parse_decision(response.content)

    if next_agent == "FINISH":
        return Command(goto=END, update={"messages": [response]})
    else:
        return Command(
            goto=next_agent,
            update={"messages": [response]}
        )

# 构建 Supervisor 图
builder = StateGraph(SupervisorState)
builder.add_node("supervisor", supervisor_node)
builder.add_node("researcher", research_agent)
builder.add_node("writer", writer_agent)

builder.add_edge(START, "supervisor")
builder.add_edge("researcher", "supervisor")  # 完成后返回 Supervisor
builder.add_edge("writer", "supervisor")

graph = builder.compile()

17.3.3 Supervisor 的关键设计

1. 中心化决策：Supervisor 是唯一的决策者，专业 Agent 只负责执行
2. 循环结构：Agent 完成后总是返回 Supervisor，由 Supervisor 决定下一步
3. Command 导航：使用 Command(goto=next_agent) 实现动态路由

17.3.4 适用场景与局限

适用场景：

• 子任务之间有明确的先后依赖
• 需要中央协调来决定任务顺序
• Agent 数量较少（3-5 个）

局限：

• Supervisor 是单点瓶颈
• 每次决策都需要额外的 LLM 调用
• 子任务之间无法直接通信

17.4 Swarm 模式

17.4.1 架构概述

Swarm 模式（蜂群模式）没有中央调度者。每个 Agent 在完成自己的工作后，可以直接将控制权交给另一个 Agent。这类似于客服系统中的"转接"——当前 Agent 判断用户的需求超出自己的能力范围时，主动转接给更合适的 Agent。

17.4.2 实现方式

Swarm 的核心是"handoff"工具——每个 Agent 有一个特殊工具用于将控制权转交给其他 Agent：

from langchain_core.tools import tool
from langgraph.types import Command

# 创建 handoff 工具
def create_handoff_tool(target_agent: str, description: str):
    @tool(name=f"transfer_to_{target_agent}")
    def handoff() -> Command:
        f"""将对话转交给 {target_agent}。{description}"""
        return Command(goto=target_agent)
    return handoff

# 定义 Agent 和它们的 handoff 能力
greeting_tools = [
    create_handoff_tool("tech_support", "当用户有技术问题时转交"),
    create_handoff_tool("billing", "当用户有账单问题时转交"),
]

tech_tools = [
    search_docs_tool,
    create_handoff_tool("greeting", "当问题解决后转回"),
]

billing_tools = [
    check_balance_tool,
    process_refund_tool,
    create_handoff_tool("greeting", "当问题解决后转回"),
]

# 创建各 Agent
greeting_agent = create_react_agent(model, greeting_tools, name="greeting")
tech_agent = create_react_agent(model, tech_tools, name="tech_support")
billing_agent = create_react_agent(model, billing_tools, name="billing")

# 构建 Swarm 图
builder = StateGraph(SwarmState)
builder.add_node("greeting", greeting_agent)
builder.add_node("tech_support", tech_agent)
builder.add_node("billing", billing_agent)
builder.add_edge(START, "greeting")  # 入口始终是接待 Agent

# 每个 Agent 可以通过 Command 跳转到任意其他 Agent
# 或者到达 END 结束对话
graph = builder.compile()