LangGraph 多智能体协作模式：从 Supervisor 到层级化团队设计

发表于2026-02-25|更新于2026-04-29|视界AI研究

|浏览量:

当单个 AI Agent 的能力边界开始显现，多智能体协作成为突破瓶颈的关键路径。LangGraph 作为 LangChain 生态中构建复杂 Agent 工作流的核心框架，提供了一套基于状态机的多智能体编排方案。本文将深入解析三种主流的多智能体协作模式，并探讨如何在实际项目中做出技术选型。

一、为什么需要多智能体架构

1.1 单 Agent 的能力瓶颈

在实际工程中，单个 Agent 面临的典型困境包括：

工具选择的认知负荷

当可用工具超过 15-20 个时，Agent 的工具选择准确率会显著下降。模型需要在每一步从庞大的工具集中做出决策，这不仅增加了延迟，还提高了错误调用的概率。

提示词复杂度的边际递减

试图在单个系统提示词中覆盖所有场景，往往导致指令互相干扰。开发者不断追加边界情况和示例，反而让模型更难抓住核心逻辑。

故障定位的困难

单 Agent 架构中，一次失败可能源于提示词、工具实现、模型能力或上下文管理的任何一个环节，排查成本极高。

1.2 多智能体的核心优势

LangGraph 团队总结的实践经验表明，多智能体设计在以下方面具有显著优势：

维度	单 Agent	多智能体
工具管理	集中式，选择复杂	分布式，各司其职
提示词设计	大而全，容易冲突	小而精，目标明确
故障隔离	牵一发而动全身	局部问题局部修复
可测试性	端到端黑盒	单元化可独立验证
扩展性	追加成本高	新增 Agent 低侵入

关键洞察在于：多智能体不是为了让更多 Agent 参与工作，而是为了让每个 Agent 专注于更少的事情。

二、LangGraph 多智能体的三种架构模式

LangGraph 将多智能体工作流抽象为图结构：Agent 是节点，连接关系是边，共享状态是图的载体。基于这种抽象，衍生出三种典型的协作模式。

2.1 模式一：协作式多智能体（Collaboration）

这是最简单的多智能体形态，核心特征是共享记忆空间。

架构设计

┌─────────────────────────────────────────┐
│           Shared State (Messages)       │
│  ┌─────────┐    ┌─────────┐            │
│  │ Agent A │◄──►│ Agent B │            │
│  │ (Coder) │    │(Reviewer)│           │
│  └────┬────┘    └────┬────┘            │
│       │              │                  │
│       └──────┬───────┘                  │
│              ▼                          │
│         Router Node                     │
│    (Decide next agent)                  │
└─────────────────────────────────────────┘

两个或多个 Agent 操作同一个消息列表（scratchpad），每个 Agent 的思考和输出对其他 Agent 完全可见。

典型场景

结对编程：Coder Agent 写代码，Reviewer Agent 实时审查
文档协作：Writer Agent 起草，Editor Agent 润色
头脑风暴：多个角色从不同角度分析同一问题

代码骨架

from langgraph.graph import StateGraph, MessagesState
from langchain_core.messages import SystemMessage

# 定义两个 Agent 的系统提示
coder_prompt = """你是一个资深 Python 工程师。根据需求编写代码，
如果代码已完成，请回复 FINAL ANSWER。"""

reviewer_prompt = """你是一个代码审查专家。检查代码的质量和潜在问题，
如果代码合格，请回复 FINAL ANSWER。"""

def coder_node(state: MessagesState):
    """Coder Agent 节点"""
    messages = state["messages"]
    # 添加系统提示
    full_messages = [SystemMessage(content=coder_prompt)] + messages
    # 调用 LLM
    response = llm.invoke(full_messages)
    return {"messages": [response]}

def reviewer_node(state: MessagesState):
    """Reviewer Agent 节点"""
    messages = state["messages"]
    full_messages = [SystemMessage(content=reviewer_prompt)] + messages
    response = llm.invoke(full_messages)
    return {"messages": [response]}

def router(state: MessagesState):
    """路由决策：根据最后一条消息决定下一步"""
    last_message = state["messages"][-1].content
    
    if "FINAL ANSWER" in last_message:
        return END
    # 简单轮换策略
    elif state["messages"][-1].name == "coder":
        return "reviewer"
    else:
        return "coder"

# 构建图
builder = StateGraph(MessagesState)
builder.add_node("coder", coder_node)
builder.add_node("reviewer", reviewer_node)

builder.add_conditional_edges("coder", router)
builder.add_conditional_edges("reviewer", router)
builder.set_entry_point("coder")

graph = builder.compile()

优缺点分析

优点：

实现简单，无需复杂的状态同步机制
信息透明，便于调试和追踪

缺点：

上下文膨胀：所有历史消息被反复传递
缺乏隐私性：Agent 无法拥有”内部思考”空间
扩展性受限：Agent 数量增加后状态管理混乱

2.2 模式二：Supervisor 模式

Supervisor 模式引入了一个中央协调者，核心特征是独立工作空间 + 统一调度。

架构设计

                    ┌─────────────────┐
                    │   Supervisor    │
                    │  (Router Agent) │
                    └────────┬────────┘
                             │
        ┌────────────────────┼────────────────────┐
        │                    │                    │
        ▼                    ▼                    ▼
┌───────────────┐   ┌───────────────┐   ┌───────────────┐
│   Research    │   │    Writer     │   │   Coder       │
│    Agent      │   │    Agent      │   │   Agent       │
│               │   │               │   │               │
│ Own scratchpad│   │ Own scratchpad│   │ Own scratchpad│
│ Own tools     │   │ Own tools     │   │ Own tools     │
└───────┬───────┘   └───────┬───────┘   └───────┬───────┘
        │                    │                    │
        └────────────────────┼────────────────────┘
                             ▼
                    ┌─────────────────┐
                    │  Shared State   │
                    │ (Final outputs) │
                    └─────────────────┘

每个 Worker Agent 拥有独立的提示词、工具和内部状态，仅将最终输出提交到共享空间。Supervisor 负责决策下一个由哪个 Agent 执行。

Supervisor 的本质

Supervisor 本身也是一个 Agent，只是它的”工具”是其他 Agent。这种设计体现了分层抽象的思想：

# Supervisor 的工具定义
worker_tools = [
    Tool(name="research", func=research_agent.invoke, 
         description="搜索和整理信息"),
    Tool(name="write", func=writer_agent.invoke,
         description="撰写和编辑内容"),
    Tool(name="code", func=coder_agent.invoke,
         description="编写和测试代码")
]

supervisor = create_react_agent(llm, worker_tools)

典型场景

内容生产流水线：Research → Write → Edit → Publish
软件开发流程：PM → Architect → Coder → Reviewer → Tester
客户服务： triage → technical → billing → escalation

代码实现要点

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

class TeamState(TypedDict):
    """团队共享状态"""
    messages: Annotated[list, add_messages]
    next_agent: str  # Supervisor 决策结果
    task_count: int  # 防止无限循环

def supervisor_node(state: TeamState):
    """Supervisor Agent 决定下一步"""
    members = ["researcher", "writer", "coder"]
    
    system_prompt = f"""你是团队主管，负责协调以下成员：{members}。
    根据当前任务状态，选择最合适的成员执行下一步工作。
    如果任务完成，回复 FINISH。"""
    
    messages = [SystemMessage(content=system_prompt)] + state["messages"]
    response = llm.with_structured_output(RouterDecision).invoke(messages)
    
    return {"next_agent": response.next}

def agent_node(state: TeamState, agent_name: str, agent_runnable):
    """通用 Agent 节点包装器"""
    # 只传递相关消息给子 Agent
    agent_output = agent_runnable.invoke(state["messages"])
    
    return {
        "messages": [{
            "role": "assistant",
            "content": agent_output,
            "name": agent_name
        }],
        "task_count": state.get("task_count", 0) + 1
    }

# 构建工作流
builder = StateGraph(TeamState)
builder.add_node("supervisor", supervisor_node)
builder.add_node("researcher", lambda s: agent_node(s, "researcher", research_agent))
builder.add_node("writer", lambda s: agent_node(s, "writer", writer_agent))
builder.add_node("coder", lambda s: agent_node(s, "coder", coder_agent))

# Supervisor 路由
builder.add_conditional_edges(
    "supervisor",
    lambda x: x["next_agent"],
    {"researcher": "researcher", "writer": "writer", 
     "coder": "coder", "FINISH": END}
)

# Worker 完成后返回 Supervisor
for worker in ["researcher", "writer", "coder"]:
    builder.add_edge(worker, "supervisor")

builder.set_entry_point("supervisor")

优缺点分析

优点：

清晰的职责边界，每个 Agent 专注特定领域
可独立优化和测试每个 Agent
Supervisor 可以基于任务状态做智能决策

缺点：

需要设计 Supervisor 的决策逻辑
Worker Agent 之间缺乏直接通信
可能存在 Supervisor 单点瓶颈

2.3 模式三：层级化团队（Hierarchical Teams）

层级化团队是 Supervisor 模式的扩展，支持嵌套子图和递归委托。

架构设计

┌─────────────────────────────────────────┐
│          Top-Level Supervisor           │
│         (Strategic Coordinator)         │
└─────────────────┬───────────────────────┘
                  │
    ┌─────────────┼─────────────┐
    │             │             │
    ▼             ▼             ▼
┌────────┐  ┌────────────┐  ┌──────────┐
│  Data  │  │  Writing   │  │  Coding  │
│  Team  │  │   Team     │  │   Team   │
│ (Graph)│  │  (Graph)   │  │  (Graph) │
└────┬───┘  └─────┬──────┘  └────┬─────┘
     │            │              │
  ┌──┴──┐    ┌────┴────┐    ┌────┴────┐
  │W1 W2│    │ W1  W2  │    │ W1  W2  │
  └─────┘    └─────────┘    └─────────┘

每个”团队”本身就是一个完整的 LangGraph，拥有自己的 Supervisor 和 Workers。顶层 Supervisor 不需要关心子团队的内部实现。

核心创新：子图作为节点

from langgraph.graph import StateGraph

# 先构建一个子团队（写作团队）
writing_builder = StateGraph(TeamState)
writing_builder.add_node("writer", writer_agent)
writing_builder.add_node("editor", editor_agent)
writing_builder.add_node("proofreader", proofreader_agent)
# ... 配置内部边 ...
writing_team = writing_builder.compile()

# 构建另一个子团队（研发团队）
coding_builder = StateGraph(TeamState)
coding_builder.add_node("architect", architect_agent)
coding_builder.add_node("coder", coder_agent)
coding_builder.add_node("tester", tester_agent)
# ... 配置内部边 ...
coding_team = coding_builder.compile()

# 顶层图将子图作为普通节点使用
top_builder = StateGraph(State)
top_builder.add_node("orchestrator", orchestrator_node)
top_builder.add_node("writing_team", writing_team)  # 子图作为节点
top_builder.add_node("coding_team", coding_team)    # 子图作为节点

# 配置顶层路由
top_builder.add_conditional_edges(
    "orchestrator",
    route_to_team,
    {"writing": "writing_team", "coding": "coding_team", "FINISH": END}
)

典型场景

大型项目交付：每个职能团队（产品、设计、开发、测试）是一个子图
多领域研究：科学计算、文献调研、实验设计各自由专业团队处理
企业自动化：HR、财务、IT 等部门作为独立子图，由中央协调

层级化的价值

复杂性封装：顶层只需要知道”写作团队能产出文档”，无需了解内部流程
递归委托：子团队的 Supervisor 可以继续向下委托
独立演进：子团队可以独立优化，不影响其他团队
可复用性：写作团队可以被多个上层流程复用

三、模式选型决策框架

3.1 决策树

开始
│
├─ Agent 数量 ≤ 2，且需要紧密协作？
│   └─ YES → 协作式（Collaboration）
│
├─ 任务有明确的阶段划分？
│   └─ YES → Supervisor 模式
│
├─ 涉及多个专业领域，需要领域专家？
│   └─ YES → 层级化团队（Hierarchical）
│
└─ 小规模 PoC，快速验证？
    └─ YES → 协作式

3.2 量化评估指标

评估维度	协作式	Supervisor	层级化
实现复杂度	⭐⭐	⭐⭐⭐	⭐⭐⭐⭐⭐
可扩展性	⭐⭐	⭐⭐⭐⭐	⭐⭐⭐⭐⭐
调试难度	⭐⭐⭐	⭐⭐	⭐⭐⭐
团队并行度	⭐⭐⭐⭐	⭐⭐⭐	⭐⭐⭐⭐⭐
容错能力	⭐⭐	⭐⭐⭐	⭐⭐⭐⭐

3.3 实际选型建议

选择协作式模式当：

快速原型验证
Agent 之间确实需要高频信息交换
上下文长度可控（<4K tokens）

选择 Supervisor 模式当：

任务流程相对固定
需要清晰的阶段边界
团队规模 3-5 个 Agent

选择层级化团队当：

涉及多个专业领域
需要团队级别的复用
项目规模大，需要分层治理

四、工程实践要点

4.1 状态设计最佳实践

多智能体系统的状态设计是关键架构决策：

# 推荐：分层状态设计
class GlobalState(TypedDict):
    """全局共享状态"""
    messages: Annotated[list, add_messages]  # 所有 Agent 可见
    task_history: list  # 执行记录
    final_output: str   # 最终产出

class AgentInternalState(TypedDict):
    """Agent 内部状态（不共享）"""
    scratchpad: list    # 内部思考过程
    tool_calls: list    # 工具调用记录
    retry_count: int    # 重试计数

4.2 循环控制与终止条件

多智能体系统容易产生无限循环，必须设计可靠的终止机制：

from langgraph.graph import END

def check_termination(state: State):
    """多重终止条件检查"""
    
    # 条件1：任务完成标记
    if state.get("is_complete"):
        return END
    
    # 条件2：最大轮数限制
    if state.get("iteration_count", 0) > 10:
        state["termination_reason"] = "MAX_ITERATION"
        return END
    
    # 条件3：无进展检测
    if detect_no_progress(state):
        state["termination_reason"] = "NO_PROGRESS"
        return END
    
    # 继续执行
    return "continue"

4.3 错误处理与降级策略

class AgentExecutionError(Exception):
    """Agent 执行错误"""
    pass

def agent_node_with_fallback(state: State, agent_name: str):
    """带降级策略的 Agent 节点"""
    try:
        result = agent_runnable.invoke(state)
        return {"messages": [result], "error": None}
    except ToolExecutionError as e:
        # 工具执行失败：记录错误，返回给 Supervisor 决策
        return {
            "messages": [],
            "error": {"type": "tool_error", "agent": agent_name, "detail": str(e)}
        }
    except LLMRateLimitError as e:
        # 速率限制：触发指数退避重试
        return {
            "messages": [],
            "error": {"type": "rate_limit", "retry_after": e.retry_after}
        }
    except Exception as e:
        # 未知错误：优雅降级
        logger.error(f"Agent {agent_name} failed: {e}")
        return {
            "messages": [{"role": "assistant", "content": "任务执行遇到技术问题，请稍后重试。"}],
            "error": {"type": "unknown", "detail": str(e)}
        }

4.4 可观测性设计

from langsmith import traceable

@traceable(name="multi_agent_orchestrator")
def run_multi_agent_workflow(input_query: str):
    """可追踪的多智能体工作流执行"""
    
    # 记录工作流启动
    logger.info(f"Starting workflow for: {input_query}")
    
    initial_state = {"messages": [{"role": "user", "content": input_query}]}
    
    # 执行图
    for event in graph.stream(initial_state):
        # 实时输出中间状态（用于调试和监控）
        node_name = list(event.keys())[0]
        node_output = event[node_name]
        
        logger.info(f"Node {node_name} completed", extra={
            "node": node_name,
            "output_length": len(str(node_output)),
            "timestamp": time.time()
        })
        
        # 可选择在特定节点触发人工审核
        if node_name == "critical_decision":
            human_approval = request_human_approval(node_output)
            if not human_approval:
                break
    
    return event

五、从理论到实践：完整示例

以下是一个基于 Supervisor 模式的完整研究助手实现：

"""
多智能体研究助手：Researcher + Writer + FactChecker
"""

from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langchain_core.messages import SystemMessage, HumanMessage
from langchain_openai import ChatOpenAI
from langchain_community.tools import DuckDuckGoSearchRun

# ========== 状态定义 ==========
class ResearchState(TypedDict):
    messages: Annotated[list, add_messages]
    research_notes: str
    draft: str
    final_report: str
    next: Literal["researcher", "writer", "fact_checker", "FINISH"]

# ========== 工具定义 ==========
search_tool = DuckDuckGoSearchRun()

# ========== Agent 定义 ==========
llm = ChatOpenAI(model="gpt-4")

def researcher_node(state: ResearchState):
    """研究 Agent：搜索并整理信息"""
    prompt = """你是一个专业的研究助理。根据用户的问题，搜索相关信息并整理成结构化的笔记。
    重点关注：核心事实、数据来源、不同观点。
    使用搜索工具获取最新信息。"""
    
    messages = [SystemMessage(content=prompt)] + state["messages"]
    
    # 绑定搜索工具的 LLM
    research_llm = llm.bind_tools([search_tool])
    response = research_llm.invoke(messages)
    
    # 如果调用了工具，执行搜索
    if response.tool_calls:
        search_results = search_tool.invoke(response.tool_calls[0]["args"])
        # 让 LLM 整理搜索结果
        summary_prompt = f"基于以下搜索结果，整理研究笔记：\n{search_results}"
        summary = llm.invoke([HumanMessage(content=summary_prompt)])
        return {
            "research_notes": summary.content,
            "messages": [{"role": "assistant", "content": summary.content}],
            "next": "writer"
        }
    
    return {
        "research_notes": response.content,
        "messages": [response],
        "next": "writer"
    }

def writer_node(state: ResearchState):
    """写作 Agent：基于研究结果撰写报告"""
    prompt = f"""你是一个技术报告撰写专家。基于以下研究笔记，撰写一份清晰的分析报告。
    
研究笔记：
{state.get('research_notes', '')}

要求：
- 结构清晰（背景、分析、结论）
- 语言简洁专业
- 保持客观中立"""

    messages = [SystemMessage(content=prompt)] + state["messages"]
    response = llm.invoke(messages)
    
    return {
        "draft": response.content,
        "messages": [response],
        "next": "fact_checker"
    }

def fact_checker_node(state: ResearchState):
    """事实核查 Agent：检查报告中的事实准确性"""
    prompt = f"""你是一个严谨的事实核查员。检查以下报告中的关键事实声明，
    标记任何需要验证或存疑的内容。

报告内容：
{state.get('draft', '')}

如果发现重大问题，回复 "NEEDS_REVISION:<具体问题>"；
如果基本准确，回复 "APPROVED"。"""

    response = llm.invoke([HumanMessage(content=prompt)])
    content = response.content
    
    if content.startswith("NEEDS_REVISION"):
        return {
            "messages": [response],
            "next": "writer"  # 返回修改
        }
    else:
        return {
            "final_report": state["draft"],
            "messages": [response],
            "next": "FINISH"
        }

def supervisor_node(state: ResearchState):
    """Supervisor：简单路由，实际可由 LLM 决策"""
    return {"next": state.get("next", "researcher")}

# ========== 构建工作流 ==========
builder = StateGraph(ResearchState)

# 添加节点
builder.add_node("supervisor", supervisor_node)
builder.add_node("researcher", researcher_node)
builder.add_node("writer", writer_node)
builder.add_node("fact_checker", fact_checker_node)

# Supervisor 路由
builder.add_conditional_edges(
    "supervisor",
    lambda x: x["next"],
    {
        "researcher": "researcher",
        "writer": "writer",
        "fact_checker": "fact_checker",
        "FINISH": END
    }
)

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

builder.set_entry_point("supervisor")

# 编译
app = builder.compile()

# ========== 运行 ==========
if __name__ == "__main__":
    result = app.invoke({
        "messages": [{"role": "user", "content": "分析 LangGraph 在多智能体系统中的应用场景"}],
        "research_notes": "",
        "draft": "",
        "final_report": ""
    })
    
    print("=== 最终报告 ===")
    print(result["final_report"])