4.3 多Agent协作与LangGraph编排

4.3 多Agent协作与LangGraph编排 — LangChain框架精通 多智能体实战

本节导读：学完本节，你将理解为什么需要多 Agent 协作，掌握用 LangGraph StateGraph 编排多个 Agent 的方法，并能够构建一个"主调度 Agent + 多个专家 Agent"的典型多智能体系统。

学习目标

• 理解单 Agent 的能力边界和多 Agent 协作的适用场景
• 掌握 LangGraph StateGraph 的基本用法，能定义节点、边和条件路由
• 能够构建"调度器 + 专家池"模式的多 Agent 系统
• 理解 Agent 间通信的两种模式：消息传递和共享状态

核心概念

为什么需要多 Agent？

当你用 create_agent 构建了一个工具丰富的 Agent 后，很快会碰到一个天花板：工具越多，模型决策越不准。

这不是模型的错。当一个 Agent 有 20 个工具时，模型需要从 20 个工具的描述中判断该用哪个、传什么参数，这本身就是一个复杂的分类问题。工具越多，选错工具的概率越高。更糟糕的是，不同领域的工具混在一起，模型容易"串台"——用计算器的逻辑去调数据库查询。

多 Agent 协作的核心思路是分而治之：不是给一个 Agent 装所有工具，而是把工具按领域分组，每个 Agent 只管自己领域的工具，再用一个"调度器"决定把任务分给谁。

三种典型的多 Agent 架构模式

在实际项目中，多 Agent 协作有三种最常见的架构模式，适用不同的场景：

模式一：扇出分发（Fan-out）——一个调度器把任务分给多个 Agent，每个 Agent 独立处理后结果汇总。适合"一个问题需要多个角度回答"的场景，比如用户问了一个既涉及技术又涉及业务的问题。

模式二：顺序流水线（Pipeline）——Agent 按固定顺序依次处理，前一个的输出是后一个的输入。适合"有明确阶段划分"的任务，比如"需求分析 → 代码生成 → 代码审查 → 测试"。

模式三：迭代协作（Iterative）——Agent 之间来回传递，直到满足某个终止条件。适合"需要反复打磨"的任务，比如"代码生成 → 测试 → 修改 → 再测试"循环。

本节会依次实现这三种模式。实际项目中，往往是三种模式的混合——一个复杂系统可能既有分发又有流水线还有迭代。

LangGraph：Agent 编排框架

LangGraph 是 LangChain 生态中专门用于编排工作流的框架。它用有向图（StateGraph）来定义 Agent 的执行流程——节点是执行单元，边定义流转方向，条件边实现动态路由。

为什么不用 Python 的 if/while 来编排？因为真实的 Agent 系统需要：持久化断点恢复、人机协作中断点、可视化调试、并行执行。这些需求用普通代码实现非常痛苦，而 LangGraph 原生支持。

StateGraph 的核心设计哲学是"状态驱动"：每个节点接收当前状态、执行逻辑、返回状态更新。图的执行过程就是状态在节点之间流转的过程。这种设计让执行过程天然可序列化（方便持久化和调试），也让节点之间完全解耦（节点只关心状态，不关心上下游是谁）。

环境准备 / 前置知识

# 安装 LangGraph
pip install -qU langgraph langchain "langchain[openai]"
export OPENAI_API_KEY="your-api-key-here"
​

前置要求：

• 已掌握本教程 4.1 节的 create_agent 基础用法
• 了解 Python 的函数式编程概念（高阶函数、闭包）
• 理解图的基本概念（节点、边、有向图）

分步实战

步骤 1：用 StateGraph 构建最简工作流

在构建多 Agent 系统之前，先理解 StateGraph 的基本机制：

from langgraph.graph import StateGraph, START, END
from typing import TypedDict

# 1. 定义状态结构
class WorkflowState(TypedDict):
    query: str              # 用户查询
    category: str           # 分类结果
    response: str           # 最终回复

# 2. 定义节点函数
def classify_query(state: WorkflowState) -> dict:
    """对用户查询进行分类。"""
    query = state["query"]
    
    # 简单的关键词分类（实际场景用 LLM）
    if any(w in query for w in ["代码", "bug", "函数", "API"]):
        category = "coding"
    elif any(w in query for w in ["数据", "分析", "报表", "统计"]):
        category = "data"
    else:
        category = "general"
    
    return {"category": category}

def handle_coding(state: WorkflowState) -> dict:
    """处理编码相关问题。"""
    return {"response": f"[代码专家] 收到编码问题：{state['query']}。建议检查相关函数的参数类型和返回值。"}

def handle_data(state: WorkflowState) -> dict:
    """处理数据分析问题。"""
    return {"response": f"[数据分析师] 收到数据分析请求：{state['query']}。正在生成分析报表..."}

def handle_general(state: WorkflowState) -> dict:
    """处理通用问题。"""
    return {"response": f"[通用助手] 收到问题：{state['query']}。这是我的回答..."}

# 3. 定义路由函数
def route_by_category(state: WorkflowState) -> str:
    """根据分类结果路由到对应的处理节点。"""
    return state["category"]

# 4. 构建图
graph = StateGraph(WorkflowState)

# 添加节点
graph.add_node("classify", classify_query)
graph.add_node("coding_expert", handle_coding)
graph.add_node("data_analyst", handle_data)
graph.add_node("general_assistant", handle_general)

# 添加边
graph.add_edge(START, "classify")
graph.add_conditional_edges(
    "classify",
    route_by_category,
    {
        "coding": "coding_expert",
        "data": "data_analyst",
        "general": "general_assistant",
    }
)
graph.add_edge("coding_expert", END)
graph.add_edge("data_analyst", END)
graph.add_edge("general_assistant", END)

# 5. 编译并运行
app = graph.compile()

result = app.invoke({"query": "帮我看看这个函数为什么报 bug"})
print(result["response"])
# 输出：[代码专家] 收到编码问题：帮我看看这个函数为什么报 bug...
​

这段代码展示了 StateGraph 的五个核心操作：定义状态、定义节点、定义路由、构建图、编译运行。注意每个节点函数接收 state 并返回一个字典——这个字典会自动合并到 state 中（只更新你返回的字段）。

理解节点函数的返回值语义：节点函数不需要返回完整的 state，只需要返回你想要更新的字段。LangGraph 会自动把返回值合并到当前 state 中。这意味着多个节点可以各自更新不同的字段，互不干扰。

步骤 2：构建调度器 + 专家 Agent 系统

现在把 StateGraph 和 create_agent 结合起来，构建真正的多 Agent 系统。这是生产环境中最常见的多 Agent 架构——"调度器 + 专家池"模式：

from langgraph.graph import StateGraph, START, END
from langchain.agents import create_agent
from langchain.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from typing import TypedDict

# ========== 状态定义 ==========
class MultiAgentState(TypedDict):
    messages: list
    assigned_agent: str
    final_response: str

# ========== 专家 Agent 定义 ==========

@tool
def execute_code(code: str) -> str:
    """执行 Python 代码并返回输出。"""
    try:
        local_vars = {}
        exec(code, {"__builtins__": __builtins__}, local_vars)
        return str(local_vars.get("result", "代码执行完成，无输出"))
    except Exception as e:
        return f"执行错误：{e}"

@tool
def read_file(filepath: str) -> str:
    """读取文件内容。"""
    try:
        with open(filepath, 'r') as f:
            return f.read()[:2000]
    except FileNotFoundError:
        return f"文件 {filepath} 不存在"

coding_agent = create_agent(
    model="openai:gpt-4o-mini",
    tools=[execute_code, read_file],
    system_prompt="你是代码专家。只回答编程相关问题，给出可运行的代码示例。",
)

@tool
def query_sql(sql: str) -> str:
    """执行 SQL 查询并返回结果。"""
    return f"SQL 查询结果：[{sql}] 返回 3 行数据（模拟）"

@tool
def generate_chart(chart_type: str, data: str) -> str:
    """生成数据可视化图表。"""
    return f"已生成 {chart_type} 类型图表（模拟）"

data_agent = create_agent(
    model="openai:gpt-4o-mini",
    tools=[query_sql, generate_chart],
    system_prompt="你是数据分析师。只回答数据查询和分析相关问题。",
)

# ========== 调度器节点 ==========
def router_node(state: MultiAgentState) -> dict:
    """分析用户意图，决定分发给哪个专家 Agent。"""
    router = create_agent(
        model="openai:gpt-4o-mini",
        tools=[],
        system_prompt="""你是一个任务路由器。根据用户消息判断应该分发给哪个专家：
- coding: 编程、代码调试、API 使用、文件操作
- data: 数据查询、统计分析、报表生成
- general: 其他问题

只输出一个词：coding / data / general""",
    )
    
    result = router.invoke({"messages": state["messages"]})
    agent_name = result["messages"][-1].content.strip().lower()
    
    if agent_name not in ("coding", "data", "general"):
        agent_name = "general"
    
    return {"assigned_agent": agent_name}

def coding_expert_node(state: MultiAgentState) -> dict:
    result = coding_agent.invoke({"messages": state["messages"]})
    return {"final_response": result["messages"][-1].content}

def data_expert_node(state: MultiAgentState) -> dict:
    result = data_agent.invoke({"messages": state["messages"]})
    return {"final_response": result["messages"][-1].content}

def general_assistant_node(state: MultiAgentState) -> dict:
    assistant = create_agent(
        model="openai:gpt-4o-mini", tools=[],
        system_prompt="你是一个通用助手，回答用户的日常问题。",
    )
    result = assistant.invoke({"messages": state["messages"]})
    return {"final_response": result["messages"][-1].content}

def route_to_expert(state: MultiAgentState) -> str:
    return state["assigned_agent"]

# ========== 构建图 ==========
graph = StateGraph(MultiAgentState)
graph.add_node("router", router_node)
graph.add_node("coding_expert", coding_expert_node)
graph.add_node("data_expert", data_expert_node)
graph.add_node("general_assistant", general_assistant_node)

graph.add_edge(START, "router")
graph.add_conditional_edges("router", route_to_expert, {
    "coding": "coding_expert",
    "data": "data_expert",
    "general": "general_assistant",
})
graph.add_edge("coding_expert", END)
graph.add_edge("data_expert", END)
graph.add_edge("general_assistant", END)

app = graph.compile(checkpointer=InMemorySaver())

from langchain_core.utils.uuid import uuid7
config = {"configurable": {"thread_id": str(uuid7())}}
result = app.invoke(
    {"messages": [{"role": "user", "content": "帮我写一个快速排序"}]},
    config=config,
)
print(result["final_response"])
​

这个架构的关键设计决策：

1. 调度器用 LLM 而非规则来做分类——因为用户的表达方式千变万化，规则很难覆盖
2. 调度器本身不携带工具——它的唯一职责是分类，工具只会干扰它的判断
3. 每个 Agent 的系统提示词中明确限定领域——防止"越界"回答
4. 所有 Agent 共享同一个 messages 列表——确保每个专家都能看到完整的对话上下文

步骤 3：迭代协作模式——代码生成与测试循环

上面的例子是"单次分发"模式。但很多场景需要 Agent 之间来回通信——比如代码专家写完代码后，需要测试专家来验证，不通过则修改：

class CollaborativeState(TypedDict):
    messages: list
    code: str
    test_result: str
    needs_revision: bool
    iteration: int

def coder_node(state: CollaborativeState) -> dict:
    coder = create_agent(
        model="openai:gpt-4o-mini",
        tools=[execute_code],
        system_prompt="你是高级开发工程师。根据需求编写 Python 代码。如果 test_result 包含错误，修改代码修复。",
    )
    
    prompt = state["messages"][-1]["content"]
    if state.get("test_result"):
        prompt += f"\n\n之前的测试结果：{state['test_result']}\n请根据测试结果修改代码。"
    
    result = coder.invoke({"messages": [{"role": "user", "content": prompt}]})
    return {"code": result["messages"][-1].content, "iteration": state.get("iteration", 0) + 1}

def tester_node(state: CollaborativeState) -> dict:
    tester = create_agent(
        model="openai:gpt-4o-mini",
        tools=[execute_code],
        system_prompt="你是测试工程师。运行给定的代码，检查是否有错误。输出格式：passed: true/false, error: 描述",
    )
    
    result = tester.invoke({
        "messages": [{"role": "user", "content": f"运行并测试以下代码：\n```python\n{state['code']}\n```"}]
    })
    test_output = result["messages"][-1].content
    passed = "passed" in test_output.lower() and "false" not in test_output.lower()
    
    return {"test_result": test_output, "needs_revision": not passed}

def should_continue(state: CollaborativeState) -> str:
    if state.get("needs_revision") and state.get("iteration", 0) < 3:
        return "revise"
    return "done"

# 构建协作图——注意这里形成了循环：coder → tester → coder
graph = StateGraph(CollaborativeState)
graph.add_node("coder", coder_node)
graph.add_node("tester", tester_node)

graph.add_edge(START, "coder")
graph.add_edge("coder", "tester")
graph.add_conditional_edges("tester", should_continue, {
    "revise": "coder",
    "done": END,
})

app = graph.compile()

result = app.invoke({
    "messages": [{"role": "user", "content": "写一个二分查找函数，处理边界情况"}],
    "iteration": 0,
})
print(f"最终代码：\n{result['code']}")
print(f"迭代次数：{result['iteration']}")
​

这个"代码 → 测试 → 修改 → 再测试"的循环是迭代协作模式的典型实现。关键点是设置最大迭代次数（这里是 3 次），否则如果代码一直有 bug，Agent 会无限循环下去，既浪费 Token 也浪费时间。

完整示例

下面是一个"技术支持多 Agent 系统"的完整示例，集成了分类路由、专家处理和工单升级三种能力：

from langgraph.graph import StateGraph, START, END
from langchain.agents import create_agent
from langchain.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from langchain_core.utils.uuid import uuid7
from typing import TypedDict
import json

class SupportState(TypedDict):
    messages: list
    category: str
    resolution: str

@tool
def search_docs(query: str) -> str:
    """搜索技术文档库。"""
    docs = {
        "安装": "请运行 pip install langchain langchain-openai 并设置 OPENAI_API_KEY 环境变量。",
        "超时": "默认超时为 30 秒。可通过 config={'timeout': 120} 调整。",
        "API Key": "API Key 通过环境变量或直接传入 init 参数设置。",
    }
    for key, val in docs.items():
        if key in query:
            return f"文档匹配：{val}"
    return "未找到相关文档。"

@tool
def check_system_status() -> str:
    """检查系统运行状态。"""
    return json.dumps({"api_status": "正常", "response_time": "120ms", "error_rate": "0.1%"}, ensure_ascii=False)

@tool
def escalate_ticket(issue: str, priority: str) -> str:
    """将问题升级为工单。"""
    ticket_id = f"TKT-{uuid7().hex[:6].upper()}"
    return f"工单已创建：{ticket_id}，优先级：{priority}"

docs_agent = create_agent(model="openai:gpt-4o-mini", tools=[search_docs],
    system_prompt="你是文档查询专家。只用 search_docs 工具查找答案。")
ops_agent = create_agent(model="openai:gpt-4o-mini", tools=[check_system_status],
    system_prompt="你是运维专家。用 check_system_status 检查系统状态。")
escalation_agent = create_agent(model="openai:gpt-4o-mini", tools=[escalate_ticket],
    system_prompt="你是升级处理专家。当自动解决失败时创建工单。")

def classify(state: SupportState) -> dict:
    router = create_agent(model="openai:gpt-4o-mini", tools=[],
        system_prompt="分类：docs（文档问题）/ ops（系统故障）/ escalation（需人工）。只输出一个词。")
    r = router.invoke({"messages": state["messages"]})
    cat = r["messages"][-1].content.strip().lower()
    return {"category": cat if cat in ("docs", "ops", "escalation") else "docs"}

def handle_docs(state: SupportState) -> dict:
    r = docs_agent.invoke({"messages": state["messages"]})
    return {"resolution": r["messages"][-1].content}

def handle_ops(state: SupportState) -> dict:
    r = ops_agent.invoke({"messages": state["messages"]})
    return {"resolution": r["messages"][-1].content}

def handle_escalation(state: SupportState) -> dict:
    r = escalation_agent.invoke({"messages": state["messages"]})
    return {"resolution": r["messages"][-1].content}

graph = StateGraph(SupportState)
graph.add_node("classify", classify)
graph.add_node("docs", handle_docs)
graph.add_node("ops", handle_ops)
graph.add_node("escalation", handle_escalation)
graph.add_edge(START, "classify")
graph.add_conditional_edges("classify", lambda s: s["category"], {
    "docs": "docs", "ops": "ops", "escalation": "escalation"})
graph.add_edge("docs", END)
graph.add_edge("ops", END)
graph.add_edge("escalation", END)

app = graph.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": str(uuid7())}}
result = app.invoke({"messages": [{"role": "user", "content": "API 调用超时怎么办？"}]})
print(result["resolution"])
​

常见问题 FAQ

Q1：多 Agent 和单 Agent 加很多工具有什么本质区别？

A：本质区别在于"决策边界"。单 Agent 的模型需要在一次推理中同时完成"选哪个工具"和"怎么用"两个决策；多 Agent 把"选哪个领域"和"怎么用工具"分成了两层。当工具数超过 10 个或工具跨度多个领域时，多 Agent 的准确率显著更高。

Q2：StateGraph 和 create_agent 能混用吗？

A：可以，而且这是推荐做法。create_agent 创建的 Agent 本质上就是一个编译好的 LangGraph 图。你可以在 StateGraph 的节点函数中调用 agent.invoke，就像上面的例子一样。create_agent 负责"单个 Agent 的内部循环"，StateGraph 负责"多个 Agent 之间的编排"。

Q3：多个 Agent 共享同一个 checkpointer 吗？

A：上面的例子中，每个专家 Agent 有自己的内部状态（通过 checkpointer），StateGraph 也有自己的全局状态。两者是独立的。如果需要专家 Agent 之间共享上下文，应该通过 StateGraph 的 state 来传递，而不是依赖各自的 checkpointer。

Q4：多 Agent 系统的成本怎么控制？

A：每个 Agent 调用都是独立的 API 请求。控制策略：调度器用最便宜的模型（如 gpt-4o-mini），专家 Agent 根据任务复杂度选模型；在 StateGraph 中设置最大迭代次数防止死循环；对高频简单问题用缓存或规则直接处理，不走 Agent。

最佳实践与避坑

实践 1：调度器用轻量模型

调度器只做分类，不需要强大的推理能力。用 gpt-4o-mini 或 gemini-flash 做调度，可以省 80% 以上的调度成本，而且分类准确率不会明显下降。

实践 2：给每个专家 Agent 设定明确的职责边界

在系统提示词中明确告诉专家 Agent："只回答 XX 领域的问题，其他问题请回复'超出我的职责范围'"。这防止专家"越界"回答自己不擅长的问题，也避免多个专家给出互相矛盾的答案。

实践 3：State 中保留原始消息，不要只传摘要

专家 Agent 需要看到用户的原始问题来做出准确判断。如果你在传递过程中对消息做了摘要，可能会丢失关键信息。

坑点 1：忘记给条件边设置所有可能的返回值

如果路由函数返回了一个条件边映射中没有的值，LangGraph 会报错。一定要在路由函数中做兜底处理，或者用默认边。

坑点 2：迭代循环没有设置终止条件

上面的代码生成循环中，我们设置了最大迭代次数 3 次。如果你忘记这个限制，代码有 bug 时 Agent 会无限循环。每个有循环的 StateGraph 都必须有终止条件。

坑点 3：多个节点同时更新同一个 State 字段

如果两个节点都返回了同一个字段的更新，后执行的会覆盖先执行的。需要合并时，使用 Annotated 和 reducer 函数来定义合并策略。

本节小结

本节我们从"为什么需要多 Agent"出发，通过 StateGraph 构建了三种典型的多 Agent 架构：单次分发模式（调度器 + 专家池）、迭代协作模式（代码专家与测试专家的循环）、技术支持系统（多领域分流处理）。核心收获是：create_agent 解决"单个 Agent 怎么干活"，StateGraph 解决"多个 Agent 怎么协作"。

至此，你已经掌握了 LangChain Agent 开发的核心能力。下一章我们将进入生产部署的话题——如何让 Agent 在真实环境中稳定运行、如何监控和调试 Agent 的行为。

延伸阅读

• 官方文档：LangGraph 概览页面（docs.langchain.com/oss/python/langgraph/overview）
• 官方文档：LangGraph StateGraph API 文档
• 相关章节：本教程 4.1 节介绍了 create_agent 基础，4.2 节介绍了工具进阶，本节在此基础上展开了多 Agent 协作
• 相关章节：本教程第 5 章将介绍生产部署和可观测性

关键词：LangChain框架精通, 多Agent, LangGraph, StateGraph, 智能体协作, 调度器, 教程, 实战, 最佳实践
难度：进阶
预计阅读：20 分钟