第5章多智能体编排¶

学习目标¶

理解用图模型表达多 Agent 协作的优势
掌握 Supervisor、Hierarchical、协作网络三种主流编排模式
能够实现任务路由与 Agent 间消息传递
学会把子图作为节点组合进父图
构建一个企业级多 Agent 系统

问题导入¶

单个 Agent 再强，也有能力边界：一个 Agent 同时懂写代码、查文献、做财务分析，往往每样都不精。而且 Prompt 越长、工具越多，模型越容易"分心"。解决方案是把大 Agent 拆成多个专职 Agent，各司其职、协同完成任务。但多 Agent 引入新问题：谁负责分派任务？Agent 之间怎么传递上下文？出现冲突怎么裁决？LangGraph 用图模型给出了优雅答案——把每个 Agent 当作节点，把分派/裁决当作边，整张图就是协作流程。

核心概念¶

1. 为什么用图编排多 Agent¶

其他多 Agent 框架（如 AutoGen）多以"对话"为协作媒介：Agent 们在群聊里发言、互相 @。这种方式直观但难控制——谁该发言、何时结束、如何插入审批都不好约束。LangGraph 把协作抽象为图：

每个 Agent 是一个节点（本身可以是子图）。
分派、裁决、汇总也是节点。
边定义"谁之后轮到谁"，条件边实现动态路由。

好处是：协作流程显式、可观测、可暂停，且复用前几章的状态、检查点、人在回路能力。

2. Supervisor 模式¶

最常用的多 Agent 模式。一个 Supervisor Agent 充当"调度员"，负责：

接收用户请求。
决定交给哪个专职 Agent 处理（基于 LLM 判断或规则）。
收集该 Agent 的结果，决定是否还需要其他 Agent 介入。
满足要求后汇总输出。

拓扑上是一个星形：Supervisor 居中，N 个 Worker 围绕。Supervisor 与 Worker 之间用条件边往返。优势是中心化控制、职责清晰；劣势是 Supervisor 可能成为瓶颈。

LangGraph 官方提供 langgraph-supervisor 包，用 create_supervisor 快速搭建。其本质仍是图：Supervisor 节点 + 多个 Agent 节点 + 条件边路由。

3. Hierarchical 模式¶

当 Worker 数量多、Supervisor 单层难以管理时，采用层级结构：顶层 Supervisor 管几个中层 Supervisor，每个中层 Supervisor 再管若干底层 Agent。例如：

顶层：总调度
中层 A：负责"研发"线，下辖代码 Agent、测试 Agent
中层 B：负责"运营"线，下辖文案 Agent、数据 Agent

实现上，每个中层是一个子图（本身是 Supervisor 图），作为节点接入父图。LangGraph 的子图机制让层级嵌套自然落地。

4. 协作网络模式¶

去中心化模式：没有固定 Supervisor，Agent 之间按需直接通信。适合"辩论""投票""接力"等场景。例如三个 Agent 分别从乐观、悲观、中立视角分析一只股票，最后由一个汇总节点综合。

实现上是一张网状图：Agent 节点之间互相连边，或都连到一个共享"黑板"节点读写中间结论。要小心避免无限循环，建议设 recursion_limit 与明确的终止条件。

5. 任务路由¶

无论哪种模式，"把任务交给谁"都是核心问题。两种路由策略：

LLM 路由：Supervisor 用 LLM 根据任务描述选择 Agent。灵活但慢且贵。
规则路由：基于关键词、分类标签用条件边硬编码。快且便宜，但需预先枚举。

实践中常混合：先用规则做粗分类（如含"代码"走研发线），再用 LLM 在子图内细分配。

6. 子图与状态隔离¶

把一个 Agent 实现为子图后，它有自己的内部 State。父图调用子图时：

共享 State：子图直接读写父图 State（字段需兼容）。
隔离 State：子图有独立 State，父图通过输入输出映射与它交互。

隔离 State 让 Agent 可独立开发、测试、复用，是构建企业级 Agent 库的关键。

7. Agent 间消息传递¶

多 Agent 通信的两种风格：

消息列表共享：所有 Agent 读写同一个 messages，用 name 字段区分发言者。适合"群聊"式协作。
任务字段传递：Supervisor 写 task 字段，Worker 读 task、写 result。适合"派单-交付"式协作。

后者更结构化、更易追踪，是企业系统的推荐做法。

原理图¶

flowchart TB
    USER([用户输入]) --> SUPER[Supervisor<br/>任务路由]

    subgraph 研发线
        CODE[代码 Agent]
        TEST[测试 Agent]
    end
    subgraph 运营线
        COPY[文案 Agent]
        DATA[数据 Agent]
    end
    subgraph 质量线
        REVIEW[评审 Agent]
    end

    SUPER -- "研发任务" --> CODE
    SUPER -- "运营任务" --> COPY
    SUPER -- "数据任务" --> DATA
    CODE --> TEST
    TEST --> REVIEW
    COPY --> REVIEW
    DATA --> REVIEW
    REVIEW -- "需返工" --> SUPER
    REVIEW -- "通过" --> END([最终输出])

图：混合编排示例。Supervisor 做任务路由，研发线内部是 Hierarchical（代码→测试→评审），评审 Agent 可触发返工回到 Supervisor，形成闭环。

示例代码¶

下面用 langgraph-supervisor 构建"研发 + 运营"双线 Supervisor 多 Agent 系统。

import os
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent
from langgraph_supervisor import create_supervisor

# 1. 共享模型
llm = ChatOpenAI(model="gpt-4o-mini", api_key=os.getenv("OPENAI_API_KEY"))

# 2. 定义各专职 Agent 的工具
@tool
def write_code(spec: str) -> str:
    """根据需求规格编写代码。"""
    return f"已实现：{spec}\n```python\ndef solution(): pass\n```"

@tool
def run_tests(code: str) -> str:
    """运行测试并返回结果。"""
    return "测试结果：5 passed, 0 failed"

@tool
def write_copy(topic: str) -> str:
    """撰写营销文案。"""
    return f"【爆款】{topic}，限时优惠！"

@tool
def query_metrics(metric: str) -> str:
    """查询业务指标。"""
    return f"{metric} = 1280（环比 +12%）"

# 3. 构建专职 Agent（每个都是一个独立的 ReAct 子图）
researcher = create_react_agent(
    model=llm,
    tools=[write_code, run_tests],
    name="researcher",   # name 用于 Supervisor 路由
    prompt="你是研发 Agent，负责写代码与测试。任务完成后简要汇报。",
)

copywriter = create_react_agent(
    model=llm,
    tools=[write_copy, query_metrics],
    name="copywriter",
    prompt="你是运营 Agent，负责文案撰写与数据查询。任务完成后简要汇报。",
)

# 4. 用 Supervisor 把它们编排成多 Agent 图
team = create_supervisor(
    model=llm,
    agents=[researcher, copywriter],
    prompt=(
        "你是团队主管。根据用户请求把任务分派给合适的 Agent：\n"
        "- researcher：代码、测试、技术问题\n"
        "- copywriter：文案、营销数据、运营问题\n"
        "可分步分派多个 Agent。所有子任务完成后，汇总结果回复用户。"
    ),
).compile()

# 5. 运行
if __name__ == "__main__":
    result = team.invoke({
        "messages": [{"role": "user",
            "content": "帮我写一个加法函数并测试，再写一句推广文案并查本月活跃用户数。"}],
    })
    # 打印所有 Agent 的发言
    for msg in result["messages"]:
        name = getattr(msg, "name", msg.__class__.__name__)
        print(f"[{name}] {msg.content[:120]}")

运行后，Supervisor 会先把"写加法函数并测试"分派给 researcher，再把"写文案 + 查活跃用户"分派给 copywriter，最后汇总回复。每个 Agent 各自用工具完成任务，互不干扰。整个协作流程在 messages 中清晰可查。

手写 Hierarchical 示例（片段）¶

若不用 langgraph-supervisor 包，层级结构也可手写：

from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from typing import TypedDict, Annotated

class State(TypedDict):
    messages: Annotated[list, add_messages]
    next: str   # Supervisor 决定的下一个 Agent

def supervisor(state: State) -> dict:
    # 用 LLM 决定下一个 Agent，或直接返回 END
    decision = llm.invoke(
        f"根据对话决定下一步交给 researcher / copywriter / END。对话：{state['messages'][-3:]}"
    )
    return {"next": decision.content.strip()}

def researcher_node(state: State) -> dict:
    # 调用 researcher 子图，传入当前消息
    res = researcher.invoke({"messages": state["messages"]})
    return {"messages": res["messages"]}

builder = StateGraph(State)
builder.add_node("supervisor", supervisor)
builder.add_node("researcher", researcher_node)
builder.add_edge(START, "supervisor")
builder.add_conditional_edges("supervisor", lambda s: s["next"],
                              {"researcher": "researcher", "END": END, "researcher": "researcher"})
builder.add_edge("researcher", "supervisor")  # Agent 完成后回到 supervisor
# graph = builder.compile()

手写版控制更细，适合需要自定义路由逻辑、共享状态字段、人在回路审批的场景。

企业案例¶

案例 1：企业级多 Agent 研发助手¶

某互联网公司构建研发助手，采用 Hierarchical 模式：

顶层 Supervisor：接收需求，分流到"前端""后端""测试"三条线。
后端线 Supervisor：下辖"接口设计 Agent""代码 Agent""SQL Agent"。
测试线：下辖"用例生成 Agent""执行 Agent""缺陷分析 Agent"。

每条线是独立子图，各自有检查点与人在回路（如接口设计后需架构师审批）。系统支持并发：前端与后端可并行启动。上线后，一个中等复杂度需求从"提需求"到"出可运行代码"平均 15 分钟，比单 Agent 快 3 倍且质量更稳。

案例 2：投研多 Agent 协作网络¶

投研团队用协作网络模式做股票分析：三个 Agent 分别扮演"多头分析师""空头分析师""风险中性观察者"，各自检索资料、形成观点，写入共享 blackboard 字段；第四个"首席"Agent 读取三方观点综合出投资建议。网络拓扑让"辩论"自然发生，且每位 Agent 的 Prompt 短而专注，质量高于让一个 Agent"装作三人"。

案例 3：客服 Supervisor + 转人工¶

客服系统用 Supervisor 模式：Supervisor 路由到"订单 Agent""退款 Agent""FAQ Agent"。当三者都无法解决时，Supervisor 走条件边到"人工座席节点"——这其实是一个 interrupt_before 暂停点，座席接手后用 update_state 注入回复再恢复。多 Agent 与人在回路无缝结合。

常见问题¶

Q1：Supervisor 模式和 Hierarchical 模式如何选？ Agent 数量少（<5）且任务扁平，用 Supervisor；Agent 多、可按业务线分组、单层 Supervisor 决策压力大时，用 Hierarchical 把管理分摊到多级。

Q2：多 Agent 会不会比单 Agent 慢？ 若串行分派，延迟会累加。优化方式：Supervisor 一次分派多个并行 Agent（fan-out）、给 Agent 设较短 max_output_tokens、用更快的模型做路由。并行后总延迟接近最慢的那个 Agent。

Q3：Agent 之间会"打架"吗？ 会。例如两个 Agent 给出矛盾结论。解决：引入"裁决 Agent"做冲突检测与投票；或在 State 中加 conflicts 字段，Supervisor 检测到后触发复核。

Q4：子图能否独立部署？ 能。子图编译后是独立 Runnable，可单独暴露为 API。父图通过 HTTP 调用而非内存引用，实现跨服务协作。注意状态序列化与超时处理。

Q5：多 Agent 系统怎么调试？ 善用检查点：每个 Agent 执行后都有快照，用 get_state_history 回放"Supervisor 为什么分派给 A 而非 B"。也可让 Supervisor 在路由时输出"理由"字段便于审计。

面试题¶

1. 用图编排多 Agent 相比"群聊对话"模式有何优势？ 答：协作流程显式可观测、可暂停、可恢复；路由逻辑用条件边精确控制而非依赖模型自发 @；复用检查点与人在回路；支持并行与层级嵌套。

2. Supervisor 模式与 Hierarchical 模式的区别？ 答：Supervisor 是单层星形，中心调度员直接管所有 Worker；Hierarchical 是多层树，中层 Supervisor 各管一条业务线，适合 Agent 数量多、需分组的场景。

3. 多 Agent 通信的两种风格及适用场景？ 答：消息列表共享（群聊式，适合开放协作）与任务字段传递（派单-交付式，结构化、易追踪，企业系统推荐）。

4. 如何避免多 Agent 死循环？ 答：设 recursion_limit、明确终止条件（Supervisor 判断"已完成"走 END）、引入裁决 Agent 处理冲突、为每条边设最大往返次数。

Q5：子图的状态如何与父图隔离？ 答：子图定义独立 State 类型，父图调用时通过输入输出映射转换；或用 add_node 时指定子图的 input / output schema 做字段过滤。

练习题¶

在示例基础上增加一个 "reviewer" Agent，对所有 Agent 的产出做质量评审，未通过则回到 Supervisor 重新分派。
把示例改为 Hierarchical：顶层 Supervisor 分流到"研发线"和"运营线"两个子 Supervisor，每个子 Supervisor 再管自己的 Agent。
实现一个协作网络：三个 Agent 分别从"技术""市场""财务"视角分析一个创业项目，写入共享 blackboard 字段，最后由汇总节点综合。设 recursion_limit=10 防止循环。

小结¶

本章把单 Agent 图扩展到多 Agent 协作。Supervisor 模式适合中小规模中心化调度，Hierarchical 适合大规模分组管理，协作网络适合辩论投票场景。任务路由可 LLM 可规则，子图机制让 Agent 可独立开发与复用。结合检查点与人在回路，多 Agent 系统同样可暂停、可恢复、可回放。至此你已完成 LangGraph 卷的学习，下一卷将进入更宏观的多智能体系统设计，与本章的图编排能力相辅相成。

第5章 多智能体编排¶