LangGraph 多 Agent 协作实战：Supervisor 模式与任务分发

上个月我帮团队重构一个 Agent 系统，原本单个 Agent 挂着 12 个工具：搜索、代码执行、文档生成、邮件发送… 结果呢？LLM 经常在工具之间纠结，把代码执行器用在了应该搜索的场景里。调试时看着日志，根本搞不清是哪个工具调用出了问题——整个系统就是个黑盒。

后来我们把架构拆成了 Supervisor + Workers 模式：一个总控 Agent 负责路由，三个专家 Agent 各司其职。工具选择错误率直接降到了原来的三分之一，调试也清晰多了，每层都能逐级 trace。

说白了，当你的 Agent 持有超过 10 个工具时，单 Agent 架构就有问题了。这篇文章带你从 Supervisor 模式的架构原理，到 create_supervisor API 的完整用法，再到一个 Research + Writing 团队的实战案例。代码都能直接跑，GitHub 仓库链接也放后面了。

一、为什么需要多 Agent 系统？

单 Agent 的三条死穴

我踩过的坑，可能你也正在踩。单 Agent 系统看着简单，但实际上有三条致命问题：

第一条：工具太多，选择障碍。

这个不是夸张。当单个 Agent 持有工具超过 10 个时，LLM 的工具选择错误率会明显上升。你可能会说，现在模型不是越来越聪明吗？确实，但问题是——工具描述在 prompt 里堆成一团，模型需要在 10+ 个工具里挑对的那一个，这个认知负担不是小数目。

我之前那个系统，搜索工具和代码执行工具的功能描述有点重叠（都能”查找信息”），结果模型经常在两者之间反复横跳，白白浪费了好几轮对话。

第二条：上下文积累，窗口爆炸。

所有子任务的对话历史都压在一个 context window 里。跑几轮下来，window 被历史 tool call 淹没，核心指令被稀释得不成样子。模型开始”忘事”，甚至把用户最初的请求都搞丢了。

这个我深有体会。有次调试一个跑了 20 轮的 Agent，context 里塞满了各种函数调用记录，最后模型输出的内容跟用户最初的需求已经没啥关系了。

第三条：不可调试，排查困难。

出了问题，你根本不知道是哪个工具调用的锅。单 Agent 是个黑盒，日志里全是流水账式的 tool call 记录，想定位问题？得一行行翻。

Supervisor 模式就能解决这些问题：用一个”总控 Agent”协调多个”专家 Agent”，职责分离，各司其职。

Supervisor 模式的核心理念

说起来也简单——就是分工。

想象一个团队：有个项目经理负责统筹协调，底下有研究员专注调研，有工程师专注实现，有文档专家专注写报告。每个人都有自己的专长，项目经理不需要什么都懂，只需要知道”这个任务该派给谁”。

Supervisor 模式就是这个思路：

• Supervisor（总控 Agent）：不干具体活，只做路由、协调和结果整合
• Worker Agents（专家 Agent）：每人专注一个领域，工具集精简，职责清晰

这样有什么好处？

工具数量被分散到各个 Worker 里，每个 Agent 只需要在自己的工具集里做选择。上下文也分散了，每个 Agent 只维护自己那一小部分对话历史。调试时能逐层 trace，Supervisor 分配任务给哪个 Worker、Worker 执行了什么操作，一目了然。

二、Supervisor 模式架构原理

先看张架构图：

                    ┌─────────────────┐
                    │   用户请求       │
                    └────────┬────────┘
                             │
                             ▼
                    ┌─────────────────┐
                    │   Supervisor    │
                    │   (总控 Agent)   │
                    │                 │
                    │  路由 + 协调    │
                    │  + 结果整合     │
                    └────────┬────────┘
                             │
              ┌──────────────┼──────────────┐
              │              │              │
              ▼              ▼              ▼
       ┌──────────┐   ┌──────────┐   ┌──────────┐
       │ Research │   │   Math   │   │ Writing  │
       │  Agent   │   │  Agent   │   │  Agent   │
       │          │   │          │   │          │
       │ 搜索工具 │   │ 计算工具 │   │ 生成工具 │
       └────┬─────┘   └────┬─────┘   └────┬─────┘
            │              │              │
            │   Worker     │   Worker     │   Worker
            │   执行结果   │   执行结果   │   执行结果
            │              │              │
            └──────────────┴──────────────┘
                           │
                           ▼
                    ┌─────────────────┐
                    │   Supervisor    │
                    │   整合结果      │
                    └────────┬────────┘
                             │
                             ▼
                    ┌─────────────────┐
                    │   最终答案      │
                    └─────────────────┘

核心组件职责

Supervisor 做三件事：

1. 路由：分析用户请求，判断该派给哪个 Worker
2. 协调：管理 Worker 之间的任务流转
3. 整合：汇总各 Worker 的执行结果，输出最终答案

Worker Agent 各司其职：

每个 Worker 只有自己的专属工具集。比如 Research Agent 可能只有搜索工具和网页抓取工具，Math Agent 只有加减乘除计算器。工具少了，选择准确率自然就上去了。

消息传递机制

这里有个关键点：全局状态（Global Graph State）。

所有 Agent 共享同一个状态对象，Worker 执行完任务后，把结果 append 到状态的 messages 字段里。Supervisor 看到新的 message，再决定下一步派给谁。

这是个 append-only 的机制——消息只增不减，保证了对话历史的完整性。

Fan-out / Fan-in

复杂任务可能需要多个 Worker 并行执行。比如用户问”比较 A 和 B 两个产品的市场数据”，Supervisor 可以同时派发两个 Research 任务（一个查 A，一个查 B），这就是 fan-out。

等两个 Worker 都返回结果了，Supervisor 再把它们整合，这就是 fan-in。

LangGraph 支持这种并行模式，不过基础篇咱们先不展开，后面进阶技巧里再细说。

"LangGraph 提供了一种构建多 Agent 系统的方式，其中每个 Agent 都有自己的工具集和职责范围，通过 Supervisor 进行协调和任务分发。"

三、create_supervisor API 详解

理论讲完了，开始写代码。

安装和导入

pip install langgraph-supervisor langchain-openai

from langchain_openai import ChatOpenAI
from langgraph_supervisor import create_supervisor
from langgraph.prebuilt import create_react_agent

定义工具

先给各个 Worker 准备工具：

from typing import Annotated

# 数学计算工具
def add(
    a: Annotated[float, "第一个数字"],
    b: Annotated[float, "第二个数字"]
) -> float:
    """Add two numbers together."""
    return a + b

def multiply(
    a: Annotated[float, "第一个数字"],
    b: Annotated[float, "第二个数字"]
) -> float:
    """Multiply two numbers."""
    return a * b

# 搜索工具（模拟）
def web_search(query: str) -> str:
    """Search the web for information."""
    # 实际项目中可以接入 Tavily、Serper 等
    if "population" in query.lower():
        return "北京人口约 2189 万（2023 年数据）"
    elif "weather" in query.lower():
        return "北京今日晴，气温 15-25°C"
    else:
        return f"搜索结果：{query}"

这里用了 Python 的 Annotated 类型提示，让模型更清楚每个参数的含义。工具函数的 docstring 也很重要——模型会通过它来理解工具的功能。

创建 Worker Agents

model = ChatOpenAI(model="gpt-4o")

# 数学专家 Agent
math_agent = create_react_agent(
    model=model,
    tools=[add, multiply],
    name="math_expert",
    prompt="你是一个数学专家，专注于数值计算。当用户需要做数学运算时，使用你的工具来完成任务。"
)

# 研究专家 Agent
research_agent = create_react_agent(
    model=model,
    tools=[web_search],
    name="research_expert",
    prompt="你是一个资深研究员，擅长搜索和整理信息。当用户需要查询资料时，使用搜索工具来获取答案。"
)

这里有几个注意点：

1. name 字段很重要：Supervisor 会通过名字来识别和调用 Worker
2. prompt 定义角色：告诉这个 Agent 它的专长是什么
3. 工具集精简：每个 Agent 只有必要工具，不多不少

创建 Supervisor

# 创建 Supervisor 系统
supervisor = create_supervisor(
    agents=[math_agent, research_agent],
    model=model,
    prompt="""你是一个团队 Leader，负责协调各个专家 Agent。

根据用户请求，决定应该把任务派给谁：
- 需要数学计算 → math_expert
- 需要搜索资料 → research_expert
- 任务完成 → 直接回答用户

如果多个专家需要配合，按照合理的顺序依次调用。"""
)

# 编译成可执行的应用
app = supervisor.compile()

create_supervisor 接收三个核心参数：

• agents：Worker Agent 列表
• model：Supervisor 自己用的大模型
• prompt：告诉 Supervisor 怎么分配任务

运行示例

from langchain_core.messages import HumanMessage

# 测试数学问题
result = app.invoke({
    "messages": [HumanMessage(content="计算 123 加 456 等于多少")]
})
print(result["messages"][-1].content)
# 输出：123 加 456 等于 579

# 测试搜索问题
result = app.invoke({
    "messages": [HumanMessage(content="北京的人口是多少")]
})
print(result["messages"][-1].content)
# 输出：根据搜索结果，北京人口约 2189 万

Supervisor 会自动判断请求类型，然后路由到正确的 Worker。全程对用户透明，用户根本不需要知道后面有多个 Agent 在工作。

四、实战案例：构建 Research + Writing 团队

上面是最基础的示例。现在来构建一个更完整的系统：一个能自动调研并生成技术文章的团队。

场景说明

用户输入一个技术主题，系统自动完成：

1. 调研相关资料
2. 生成文章大纲
3. 撰写完整内容
4. 审核校对

这需要三个专业 Agent 配合工作。

定义完整工具集

from typing import TypedDict, List
import json

# 模拟搜索工具
def tech_search(query: str) -> str:
    """搜索技术资料和文档。"""
    # 实际项目中接入 Tavily 或 Serper
    database = {
        "langgraph": "LangGraph 是 LangChain 推出的 Agent 框架，支持状态管理和循环图结构。",
        "supervisor": "Supervisor 模式是 Multi-Agent 系统的核心架构，由一个总控 Agent 协调多个专家 Agent。",
        "multi-agent": "多 Agent 系统通过任务分发和协作，解决单 Agent 工具过多和上下文爆炸的问题。"
    }

    results = []
    for key, value in database.items():
        if key in query.lower():
            results.append(value)

    return json.dumps(results) if results else "未找到相关资料，建议扩展搜索范围"

# 大纲生成工具
def generate_outline(topic: str) -> str:
    """根据主题生成文章大纲。"""
    return json.dumps({
        "title": f"{topic} 完全指南",
        "sections": [
            "1. 概述与背景",
            "2. 核心概念",
            "3. 实战案例",
            "4. 最佳实践",
            "5. 总结"
        ]
    }, ensure_ascii=False)

# 内容生成工具
def write_section(section_title: str, context: str) -> str:
    """根据标题和上下文生成段落内容。"""
    # 实际项目中这里可以调用大模型
    return f"## {section_title}\n\n基于调研资料，{section_title} 的核心要点如下...\n\n"

# 审核工具
def review_content(content: str) -> str:
    """审核内容的准确性和可读性。"""
    issues = []
    if len(content) < 100:
        issues.append("内容过短，建议扩展")
    if "TODO" in content:
        issues.append("存在未完成的 TODO 标记")

    return json.dumps({
        "passed": len(issues) == 0,
        "issues": issues,
        "suggestion": "内容质量良好，可以发布" if not issues else "请根据问题修改后重新提交"
    }, ensure_ascii=False)

创建 Worker Agents

# 研究员 Agent
researcher = create_react_agent(
    model=model,
    tools=[tech_search],
    name="researcher",
    prompt="""你是一位资深技术研究员，擅长快速调研和理解新技术。

职责：
1. 接收调研主题
2. 使用搜索工具查找相关资料
3. 整理成结构化的调研报告

注意：只做调研，不做写作。把调研结果交给 writer。"""
)

# 作者 Agent
writer = create_react_agent(
    model=model,
    tools=[generate_outline, write_section],
    name="writer",
    prompt="""你是一位技术写作专家，擅长将复杂概念转化为清晰易懂的文章。

职责：
1. 接收调研报告
2. 生成文章大纲
3. 撰写各个章节的内容

注意：完成初稿后交给 reviewer 审核。"""
)

# 审稿人 Agent
reviewer = create_react_agent(
    model=model,
    tools=[review_content],
    name="reviewer",
    prompt="""你是一位严格的审稿人，确保文章的质量和准确性。

职责：
1. 检查内容的完整性和准确性
2. 评估文章的可读性和逻辑性
3. 提出修改建议或确认发布

注意：如果发现问题，退回给 writer 修改。"""
)

构建 Supervisor 逻辑

# 使用 StateGraph 构建自定义 Supervisor
from langgraph.graph import StateGraph, END
from typing import Annotated, Sequence
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages

# 定义状态
class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], add_messages]
    next_agent: str

# Supervisor 决策逻辑
def supervisor_node(state: AgentState) -> dict:
    """根据当前进度决定下一步执行哪个 Agent。"""
    messages = state["messages"]

    # 调用大模型做决策
    decision = model.invoke([
        {"role": "system", "content": """你是团队 Leader，根据对话历史决定下一步：

- 如果还没有调研资料 → 返回 'researcher'
- 如果有调研资料但还没有文章 → 返回 'writer'
- 如果有文章但还没审核 → 返回 'reviewer'
- 如果审核通过 → 返回 'FINISH'

只返回 Agent 名称，不要其他内容。"""},
        *messages
    ])

    next_agent = decision.content.strip()

    # 映射到正确的 Agent 名称
    agent_map = {
        "researcher": "researcher",
        "writer": "writer",
        "reviewer": "reviewer",
        "FINISH": END
    }

    return {"next_agent": agent_map.get(next_agent, "researcher")}

# 构建图
workflow = StateGraph(AgentState)

# 添加节点
workflow.add_node("supervisor", supervisor_node)
workflow.add_node("researcher", researcher)
workflow.add_node("writer", writer)
workflow.add_node("reviewer", reviewer)

# 添加条件边（Supervisor 的路由逻辑）
workflow.add_conditional_edges(
    "supervisor",
    lambda state: state["next_agent"],
    {
        "researcher": "researcher",
        "writer": "writer",
        "reviewer": "reviewer",
        END: END
    }
)

# 所有 Agent 完成后都回到 Supervisor
for agent in ["researcher", "writer", "reviewer"]:
    workflow.add_edge(agent, "supervisor")

# 设置入口点
workflow.set_entry_point("supervisor")

# 编译
app = workflow.compile()

这个架构有个循环机制：所有 Worker 完成任务后都回到 Supervisor，Supervisor 再决定下一步是派给另一个 Worker 还是结束。

执行流程

result = app.invoke({
    "messages": [HumanMessage(content="写一篇关于 LangGraph Supervisor 模式的技术文章")]
})

# 查看最终结果
print(result["messages"][-1].content)

# 查看执行轨迹
for i, msg in enumerate(result["messages"]):
    print(f"{i+1}. {msg.__class__.__name__}: {msg.content[:100]}...")

执行流程大致是这样的：

用户请求 → Supervisor 分析 → 派给 Researcher →
Researcher 调研 → 返回 Supervisor → 派给 Writer →
Writer 写作 → 返回 Supervisor → 派给 Reviewer →
Reviewer 审核 → 返回 Supervisor → 确认完成 → 输出结果

每一层都能 trace，调试起来清晰多了。

五、进阶技巧

消息转发优化：create_forward_message_tool

有个问题你可能已经发现了：Worker 执行完任务后，返回的消息会被 Supervisor 接收并可能被重述一遍。这浪费 Token，而且信息可能被稀释。

LangGraph 提供了 create_forward_message_tool 来解决这个问题：

from langgraph_supervisor.handoff import create_forward_message_tool

# 创建转发工具
forward_tool = create_forward_message_tool("supervisor")

# 在创建 Supervisor 时传入
supervisor = create_supervisor(
    agents=[researcher, writer, reviewer],
    model=model,
    tools=[forward_tool]  # 添加转发工具
)

这个工具让 Supervisor 可以直接把 Worker 的响应转发给用户，而不需要重新总结。省 Token，效率也更高。

层级团队架构

如果项目更复杂，还可以构建多层 Supervisor：

                    ┌──────────────┐
                    │ Top Supervisor│
                    └──────┬───────┘
                           │
           ┌───────────────┼───────────────┐
           │               │               │
           ▼               ▼               ▼
    ┌────────────┐  ┌────────────┐  ┌────────────┐
    │Research    │  │ Writing    │  │ QA         │
    │Team        │  │ Team       │  │ Team       │
    │Supervisor  │  │ Supervisor │  │ Supervisor │
    └─────┬──────┘  └─────┬──────┘  └─────┬──────┘
          │               │               │
     ┌────┼────┐    ┌────┼────┐    ┌────┼────┐
     │    │    │    │    │    │    │    │    │
   Web  Doc  API   Out  Cont Rev   Test Code Audit
 Search Scrp Parse line ent  iew

每个子团队有自己的 Supervisor，上面再有一个总 Supervisor 协调。这种架构适合大型项目，职责划分更细。

错误处理

Worker 执行失败怎么办？

from langgraph.pregel import RetryPolicy

# 配置重试策略
retry_policy = RetryPolicy(
    max_attempts=3,
    initial_interval=1.0,
    backoff_factor=2.0
)

app = workflow.compile(retry_policy=retry_policy)

还可以在 Supervisor 的 prompt 里加入错误处理逻辑：

如果某个 Agent 执行失败：
1. 记录错误信息
2. 尝试调用备用 Agent
3. 如果多次失败，向用户报告问题

状态持久化

多轮对话需要保存状态。LangGraph 提供了 Checkpointer 机制：

from langgraph.checkpoint.memory import MemorySaver

# 使用内存存储（开发环境）
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)

# 执行时指定 thread_id
config = {"configurable": {"thread_id": "user-123"}}
result = app.invoke({"messages": [HumanMessage(content="...")]}, config=config)

# 后续对话会保留上下文
result2 = app.invoke({"messages": [HumanMessage(content="继续上次的任务")]}, config=config)

生产环境可以用 Redis 或 PostgreSQL 作为 Checkpointer。

"Hierarchical Agent Teams 展示了如何构建多层 Supervisor 架构，实现更复杂的多 Agent 协作系统。"

六、生产部署建议

监控和调试

LangSmith 是 LangChain 官方的监控平台，能帮你追踪每一步的执行细节：

import os

os.environ["LANGSMITH_API_KEY"] = "your-api-key"
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_PROJECT"] = "multi-agent-project"

配置好之后，每次执行都会在 LangSmith 上留下完整轨迹，包括：

• 每个 Agent 的输入输出
• 工具调用的参数和返回值
• Token 消耗统计
• 执行耗时分析

调试时特别有用，不用再对着日志一行行翻。

Token 成本控制

Supervisor 模式虽然好，但多 Agent 会增加 Token 消耗。几个优化建议：

1. 精简 Supervisor 的 prompt：只包含必要的路由逻辑
2. 使用 forward_message_tool：避免重复总结
3. 合理分配工具：每个 Worker 只有必要工具
4. 控制对话轮数：设置最大轮数限制

# 设置最大轮数
app = workflow.compile(
    checkpointer=memory,
    interrupt_after=20  # 最多执行 20 步
)

与 AWS Bedrock 集成

如果你的项目在 AWS 上，可以用 Bedrock 的模型：

from langchain_aws import ChatBedrock

model = ChatBedrock(
    model_id="anthropic.claude-3-sonnet-20240229-v1:0",
    region_name="us-east-1"
)

# 其他代码不变，直接替换 model 即可
supervisor = create_supervisor(
    agents=[math_agent, research_agent],
    model=model
)

最佳实践总结

说了这么多，总结几条实战经验：

1. 从小规模开始：先从 2-3 个 Agent 开始，逐步增加
2. 职责要清晰：每个 Worker 的领域要明确，避免重叠
3. 用 LangSmith 监控：开发阶段就接入，方便调试
4. 关注 Token 成本：多 Agent 会放大消耗，需要优化
5. 善用 forward_message_tool：能省不少 Token
6. 参考官方仓库：langgraph-supervisor-py 有完整示例

总结

Supervisor 模式本质上就是分工协作——把大任务拆小，让每个 Agent 专注做自己擅长的事。

从单 Agent 的三条死穴（工具选择障碍、上下文爆炸、不可调试），到 Supervisor 模式的优雅解决方案，这篇文章带你走了完整的一遍。create_supervisor API 用起来其实不复杂，关键是理解背后的架构思想。

建议你从小项目开始练手。先搭建一个两 Agent 的系统（比如一个搜索 + 一个总结），跑通之后再逐步扩展。LangSmith 的监控一定要配上，调试时你会感谢它的。

完整代码示例在 GitHub 仓库里：langgraph-supervisor-py。官方教程也值得一读：Hierarchical Agent Teams。

有问题评论区留言，我看到都会回。

---

参考资料

• LangGraph Supervisor Reference
• Hierarchical Agent Teams Tutorial
• LangGraph Multi-Agent Workflows Blog
• GitHub - langgraph-supervisor-py
• Build Multi-Agent Systems with AWS Bedrock

常见问题

Supervisor 模式和普通 Multi-Agent 有什么区别？

普通 Multi-Agent 可能每个 Agent 都能直接响应用户，导致职责混乱。Supervisor 模式引入一个总控 Agent 专门做路由和协调，Worker Agent 只负责执行具体任务，职责更清晰。

什么时候应该用 Supervisor 模式？

当你的 Agent 工具超过 10 个、或者任务需要多个领域协作、或者调试困难时，就该考虑 Supervisor 模式了。简单任务单 Agent 足够。

Supervisor 模式会增加 Token 消耗吗？

会。多 Agent 意味着多次模型调用。但可以通过 forward_message_tool、精简 prompt、控制轮数来优化。总体上，职责分离带来的调试便利和准确性提升，通常比额外 Token 更有价值。

如何调试多 Agent 系统？

LangSmith 是首选。它能追踪每一步的输入输出、工具调用、Token 消耗。开发阶段就接入，比事后翻日志效率高很多。

create_supervisor 和 StateGraph 有什么关系？

create_supervisor 是高层 API，快速搭建简单 Supervisor 系统。StateGraph 是底层 API，适合构建自定义路由逻辑和复杂层级架构。两者可以组合使用。

Worker Agent 可以是另一个 Supervisor 吗？

可以。这就是层级团队架构。子团队有自己的 Supervisor，上面再有一个总 Supervisor 协调。适合大型复杂项目。