LangGraph多智能体实战：状态管理与条件路由编排复杂流程

1. 背景与核心概念：为什么需要 LangGraph 多智能体

1.1 从单智能体到多智能体

如果你接触过基于 LLM 的 Agent 开发，大概率经历过这样的过程：最初是让模型根据 Prompt 直接回答，后来发现一次调用不够用，便引入了 ReAct 模式的循环调用，让模型在“思考—工具调用—观察结果”之间反复迭代。这个阶段可以称为单智能体阶段。

单智能体的核心逻辑是一个循环：

text复制用户输入 -> 大模型分析 -> 决定是否调用工具 -> 调用工具 -> 将工具结果返回给模型 -> 生成最终回答

它的优点很直接：代码量少、逻辑简单、适合业务链路不复杂的场景。但随着业务复杂度提升，单智能体会暴露几个明显问题：

• 职责混乱：所有逻辑都塞进同一个 Prompt，导致模型经常在多个意图之间“打架”。
• 上下文膨胀：多轮工具调用会导致历史消息越来越大，既浪费 Token，又增加了模型注意力分散的概率。
• 难以编排复杂流程：线性循环只能处理“一问一答一工具”的流程，遇到需要并行处理、条件分支、多角色协作的场景就力不从心。
• 调试困难：所有逻辑在一个循环里，定位“哪个环节出了问题”全靠日志硬猜。

多智能体架构的出现，就是为了解决这些问题。

1.2 多智能体架构的本质

多智能体（Multi-Agent）并不是简单地把多个模型实例堆在一起，而是把一个大任务拆分成多个子任务，每个子任务由一个或多个智能体负责，然后通过明确的通信与调度机制让它们协作完成整体目标。

可以把多智能体理解成一个“项目团队”：

• 管理者（Supervisor / Orchestrator）：负责任务拆分、进度跟踪、结果汇总。
• 执行者（Worker）：负责具体领域的任务，比如代码生成、数据分析、SQL 查询、资料检索。
• 协作机制：智能体之间如何传递信息、如何决定下一步行动。

多智能体架构的优势在于：

1. 职责清晰：每个智能体只关注自己的领域，Prompt 可以写得更加聚焦。
2. 并行能力：多个独立子任务可以同时执行，提升整体效率。
3. 可扩展性：新增能力时，只需要新增一个智能体，而不需要修改整个系统。
4. 可观测性：每个智能体的输入输出都能独立追踪，调试效率明显提升。

1.3 LangGraph 是什么

LangGraph 是 LangChain 团队推出的一个用于构建有状态、可编排的 LLM 应用框架，它特别适合实现复杂的 Agent 流程。

如果只看名字，很容易误以为 LangGraph 是“用于做图数据库”或“用于画流程图”的库，实际上它是围绕“图结构”来编排应用逻辑：

• 节点（Node）：一个 Node 表示一个执行步骤，可以是一个 LLM 调用、一个工具函数、一个条件判断，也可以是任意自定义 Python 函数。
• 边（Edge）：定义节点之间的流转关系。
• 状态（State）：在整个图执行过程中共享的数据结构，相当于应用的内存。

这种设计让整个多智能体流程变得可视化、可控、可恢复。你可以像画流程图一样设计 Agent 行为，并且每一步的状态变化都可以被跟踪。

1.4 LangGraph 与 LangChain 的关系

LangGraph 基于 LangChain 生态，但定位不同：

• LangChain 提供的是各种组件的封装，比如模型调用、Prompt 模板、输出解析器、工具加载等。
• LangGraph 提供的是流程编排能力，解决“如何把多个组件串起来并让它们协同工作”的问题。

你完全可以单独使用 LangGraph，但在实际项目中，通常会结合 LangChain 的模型封装和工具生态一起使用。

简单总结：

text复制LangChain = 组件库
LangGraph = 流程编排器

如果你已经熟悉 LangChain，学习 LangGraph 会非常顺畅；如果你还不太熟悉 LangChain，也不要紧，LangGraph 本身的上手门槛并不高。

2. 环境准备与版本说明

2.1 运行环境

本文示例以 Python 环境为基础，操作系统可以是 Windows、macOS 或 Linux。建议使用 Python 3.9 及以上版本。如果你的环境是 Python 3.8，部分依赖可能无法安装，建议先升级 Python 版本。

bash复制python --version

输出示例：

text复制Python 3.11.8

2.2 安装 LangGraph 与依赖

LangGraph 是开源库，可以直接通过 pip 安装。不过需要注意，LangGraph 迭代速度较快，API 名称和版本兼容性会随时间变化。建议在项目初始化时使用虚拟环境隔离依赖。

bash复制# 创建虚拟环境（可选但推荐）
python -m venv langgraph-demo
source langgraph-demo/bin/activate  # Windows 下使用 langgraph-demo\Scripts\activate

# 安装核心库
pip install langgraph langchain langchain-openai

如果你使用其他模型供应商，可能需要额外安装对应的包：

bash复制# 使用 Anthropic Claude
pip install langchain-anthropic

# 使用 国内模型或兼容 OpenAI 接口的模型
pip install langchain-openai

版本方面，以安装完成后的实际版本为准。langgraph 目前仍处于快速迭代期，每次升级都可能导致 API 细节变化。本文示例的核心写法在较新的版本中均可运行，但如果你使用的是较老的版本，建议先升级到最新版再测试。

2.3 注册 API Key

LangGraph 本身是一个编排框架，并不内置大模型能力，你需要准备一个可用的 LLM API Key。

以 OpenAI 兼容接口为例，设置环境变量：

bash复制export OPENAI_API_KEY="sk-你的-key"

在 Windows PowerShell 下：

powershell复制$env:OPENAI_API_KEY="sk-你的-key"

出于安全和隐私考虑，不要把 API Key 直接写进代码文件，更不要提交到 Git 仓库。建议统一通过环境变量读取。

2.4 示例项目结构

为了后续实战部分更清晰，我们建议按下面的目录结构组织代码：

text复制langgraph-multi-agent-demo/
├── requirements.txt
├── .env
├── state.py          # 定义多智能体共享状态
├── agents.py         # 定义各智能体节点
├── graph.py          # 组装流程图
└── main.py           # 入口：执行一次多智能体任务

这样拆分的好处是：状态、节点、图定义各司其职，后续扩展新智能体时不需要改动整体结构。

3. 核心概念与组件拆解

多智能体架构在 LangGraph 中落地时，要重点掌握以下核心组件：State、Node、Edge、Graph、Checkpointer 和 Conditional Edge。

3.1 State：多智能体的共享状态

State 是 LangGraph 中最基础的概念，它定义了整个图中流转的数据结构。可以把它理解为一个全局的 Python 字典，但 LangGraph 对 State 做了一层状态管理封装。

一个典型的多智能体状态定义如下：

python复制# state.py
from typing import TypedDict, Annotated, List
import operator
from langgraph.graph.message import add_messages


class AgentState(TypedDict):
    # 用户最初的问题
    question: str
    # 当前任务执行阶段
    stage: str
    # 多智能体之间的消息历史
    messages: Annotated[list, add_messages]
    # 数据分析结果
    analysis_result: str
    # 最终汇总输出
    final_answer: str

这里解释几个关键点：

• TypedDict：是 Python 中对字典类型做静态类型标注的方式，LangGraph 会用它推导 State 结构。
• Annotated[list, add_messages]：表示 messages 字段是一个列表，并且新增消息时不是覆盖旧值，而是把新消息追加到列表中。add_messages 是 LangGraph 内置的 reducer 函数。
• reducer：State 字段可以配置特殊的合并规则。默认情况下，节点返回值会覆盖原有字段；而声明了 reducer 的字段会按照 reducer 逻辑合并。这多智能体对话场景中非常重要，因为每个智能体都可能往消息列表里追加内容。

3.2 Node：智能体执行单元

在 LangGraph 中，一个 Node 就是一个 Python 函数（也可以是 async 函数）。每个 Node 接收当前 State 作为输入，返回一个字典，字典中的键值会更新到 State 中。

python复制def analyst_node(state: AgentState):
    # 处理问题
    result = ...
    # 返回需要更新到 State 中的内容
    return {"analysis_result": result}

在多智能体中，每个智能体通常对应至少一个 Node，也可以在同一个 Node 里通过内部逻辑启动多个子智能体。

3.3 Edge 与 Conditional Edge

Edge 定义节点之间的固定流转路径。比如 A -> B 表示执行完节点 A 后，下一步执行节点 B。

Conditional Edge 则是带条件的路由，它根据当前 State 的内容动态决定下一步走向哪个节点。这个机制在多智能体中非常重要，因为它让“管理者智能体”可以根据任务内容选择不同的执行者。

python复制def router(state: AgentState):
    if state["stage"] == "analysis":
        return "analyst_agent"
    elif state["stage"] == "coding":
        return "coder_agent"
    else:
        return "final_node"

3.4 Graph：将组件组装成流程

在较新的 LangGraph 版本中，推荐使用 StateGraph 来构建图：

python复制from langgraph.graph import StateGraph, START, END

graph = StateGraph(AgentState)

graph.add_node("analyst_agent", analyst_node)
graph.add_node("coder_agent", coder_node)
graph.add_node("final_node", final_node)

graph.add_edge(START, "analyst_agent")
graph.add_conditional_edge("analyst_agent", router)
graph.add_edge("final_node", END)

其中：

• START 表示流程入口。
• END 表示流程结束。
• add_node 注册节点。
• add_edge 添加固定边。
• add_conditional_edge 添加条件路由。

构建完成后，通过 compile() 方法生成可执行对象：

python复制app = graph.compile()

3.5 Checkpointer：持久化与断点续跑

Checkpointer 是 LangGraph 的一大亮点。它可以把图的执行状态保存到存储后端（内存、SQLite、Redis 等），从而支持：

• 多轮对话的历史状态保存。
• 程序中断后恢复执行。
• 人工在中间节点介入审批。

多智能体场景下，Checkpointer 尤其适合“人工审批”流程：当一个智能体生成内容后，流程先暂停，等人工审核通过再继续。

python复制from langgraph.checkpoint.memory import InMemorySaver

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

正式环境可以使用持久化存储后端。

3.6 多智能体协作模式

在 LangGraph 中实现多智能体，通常有三种典型模式：

模式一：Supervisor 模式

一个“管理者智能体”负责接收用户请求，分析任务类型后，把子任务分配给不同的执行智能体。执行完成后，管理者再汇总结果。

常用场景：客服机器人、企业知识库问答、通用任务调度。这是最常使用的模式，本文实战部分主要围绕这种模式展开。

模式二：Pipeline 模式

多个智能体按照固定顺序依次执行，前一个智能体的输出作为后一个智能体的输入。比如：检索智能体 -> 分析智能体 -> 报告生成智能体。

常用场景：数据处理流水线、报告生成流水线。

模式三：Network 模式

智能体之间可以互相通信，协作关系更加复杂。这种模式实现难度高，调试成本也高，适合非常复杂的科研或仿真场景，一般业务项目不建议一上来就使用。

4. 完整实战：构建一个三智能体协作系统

接下来我们完成一个可直接运行的多智能体实战项目。这个项目的目标如下：

用户输入一个问题，由“分析智能体”划分任务类型：

• 如果涉及数据处理，交给“数据处理智能体”。
• 如果涉及代码编写，交给“代码生成智能体”。
• 两类任务都不涉及，则直接进入最终回答节点。

最后，由“汇总智能体”把各智能体的结果整合成一份完整的回答。

4.1 创建项目结构

在终端执行：

bash复制mkdir langgraph-multi-agent-demo
cd langgraph-multi-agent-demo

然后按照前文所述创建文件。

4.2 编写状态定义

首先定义多智能体共享的数据结构。

python复制# state.py
from typing import TypedDict, Annotated, List
import operator
from langgraph.graph.message import add_messages


class AgentState(TypedDict):
    """
    多智能体共享状态
    """
    # 用户原始问题
    question: str

    # 流程阶段，用于路由判断
    stage: str

    # 智能体之间的消息记录
    messages: Annotated[list, add_messages]

    # 数据分析结果
    analysis_result: str

    # 代码生成结果
    code_result: str

    # 最终回答
    final_answer: str

这里把 messages 设计为 Annotated[list, add_messages]，意味着所有智能体往 messages 字段追加内容时，LangGraph 会保留历史消息，而不是用最新消息覆盖。

4.3 编写智能体节点

每个智能体本质是一个函数，接收 AgentState，返回需要更新的字段。

python复制# agents.py
from state import AgentState


def llm_call(prompt: str) -> str:
    """
    模拟调用大模型。
    在实际项目中，这里替换为真实的模型调用。
    """
    return f"[模拟LLM输出] 处理内容：{prompt[:100]}..."


def route_agent(state: AgentState) -> dict:
    """
    调度智能体：判断问题类型，分发给对应执行者。
    """
    question = state.get("question", "")

    if any(keyword in question for keyword in ["数据", "统计", "分析", "csv", "excel"]):
        stage = "data"
    elif any(keyword in question for keyword in ["代码", "函数", "脚本", "python", "实现"]):
        stage = "code"
    else:
        stage = "final"

    return {"stage": stage, "messages": [{"role": "assistant", "content": f"调度结果: {stage}"}]}


def data_agent(state: AgentState) -> dict:
    """
    数据处理智能体
    """
    question = state.get("question", "")
    analysis = llm_call(f"数据分析任务: {question}")
    return {
        "analysis_result": analysis,
        "messages": [{"role": "assistant", "content": f"数据分析完成: {analysis}"}]
    }


def code_agent(state: AgentState) -> dict:
    """
    代码生成智能体
    """
    question = state.get("question", "")
    code = llm_call(f"代码生成任务: {question}")
    return {
        "code_result": code,
        "messages": [{"role": "assistant", "content": f"代码生成完成: {code}"}]
    }


def final_agent(state: AgentState) -> dict:
    """
    最终汇总智能体
    """
    parts = []
    if state.get("analysis_result"):
        parts.append(f"分析结果：{state['analysis_result']}")
    if state.get("code_result"):
        parts.append(f"代码结果：{state['code_result']}")

    if not parts:
        parts.append(llm_call(f"直接回答: {state.get('question', '')}"))

    final_answer = "\n".join(parts)

    return {
        "final_answer": final_answer,
        "messages": [{"role": "assistant", "content": f"最终回答: {final_answer}"}]
    }

说明一下，上述代码使用了一个模拟的 llm_call 函数，主要目的是先跑通多智能体流程。后续替换成真实模型调用即可。

4.4 组装图流程

接下来把节点和路由机制组装成 Graph。

python复制# graph.py
from langgraph.graph import StateGraph, START, END
from state import AgentState
from agents import route_agent, data_agent, code_agent, final_agent


def create_graph():
    graph = StateGraph(AgentState)

    # 注册节点
    graph.add_node("route_agent", route_agent)
    graph.add_node("data_agent", data_agent)
    graph.add_node("code_agent", code_agent)
    graph.add_node("final_agent", final_agent)

    # 入口 -> 调度智能体
    graph.add_edge(START, "route_agent")

    # 调度之后条件路由
    graph.add_conditional_edge(
        "route_agent",
        lambda state: state["stage"],
        {
            "data": "data_agent",
            "code": "code_agent",
            "final": "final_agent",
        },
    )

    # 执行者完成后交给最终汇总
    graph.add_edge("data_agent", "final_agent")
    graph.add_edge("code_agent", "final_agent")

    # 最终节点结束
    graph.add_edge("final_agent", END)

    return graph.compile()

这里条件路由的写法需要注意：lambda state: state["stage"] 返回一个字符串，然后映射到对应节点的名称。这种写法非常直观，很适合多智能体中的任务分发。

4.5 编写入口脚本

python复制# main.py
from graph import create_graph
import uuid


def run_task(question: str):
    # 每次执行使用不同的 conversation_id，避免状态混淆
    config = {"configurable": {"thread_id": str(uuid.uuid4())}}

    app = create_graph()

    result = app.invoke(
        input={
            "question": question,
            "stage": "",
            "messages": [],
            "analysis_result": "",
            "code_result": "",
            "final_answer": ""
        },
        config=config
    )

    print("=" * 50)
    print("用户问题:", question)
    print("-" * 50)
    print("最终结果:", result.get("final_answer", ""))
    print("=" * 50)


if __name__ == "__main__":
    run_task("帮我把这份销售数据做统计分析，并生成可视化方案")
    run_task("帮我写一段 Python 脚本，实现批量重命名文件")
    run_task("中国的首都是哪里？")

这里加了一个 thread_id 参数，是因为 LangGraph 在配置了 Checkpointer 时会根据 thread_id 区分不同的会话状态。这里即使没有显式使用 Checkpointer，也保留了标准调用的写法。

4.6 运行与预期结果

在项目根目录执行：

bash复制python main.py

预期输出大致如下：

text复制==================================================
用户问题: 帮我把这份销售数据做统计分析，并生成可视化方案
--------------------------------------------------
最终结果: 分析结果：[模拟LLM输出] 处理内容：数据分析任务...
==================================================
==================================================
用户问题: 帮我写一段 Python 脚本，实现批量重命名文件
--------------------------------------------------
最终结果: 代码结果：[模拟LLM输出] 处理内容：代码生成任务...
==================================================
==================================================
用户问题: 中国的首都是哪里？
--------------------------------------------------
最终结果: [模拟LLM输出] 处理内容：直接回答...
==================================================

从执行结果可以看到，同一个 Graph 根据用户问题的不同类型，自动选择了不同的执行路径，这就是多智能体路由调度的最核心效果。

4.7 替换为真实大模型调用

模拟 llm_call 的目的是验证流程逻辑。真实项目中，我们需要替换成可用的 LLM 调用。

下面给出基于 langchain-openai 的示例：

python复制# llm_client.py
from langchain_openai import ChatOpenAI

# 初始化模型
llm = ChatOpenAI(
    model="gpt-4o-mini",
    temperature=0.2,
    api_key="your-api-key",  # 建议通过环境变量读取
    base_url="https://api.openai.com/v1"  # 若使用代理或兼容接口，替换为实际地址
)


def call_llm(prompt: str) -> str:
    response = llm.invoke(prompt)
    return response.content

这样只要把 agents.py 中的 llm_call 替换为 call_llm，就完成了从模拟到真实的切换。

实际项目中，更推荐的做法是让每个智能体使用不同的 System Prompt，这样每个智能体只负责自己的领域。比如：

python复制def data_agent(state: AgentState) -> dict:
    system_prompt = "你是一名资深数据分析师，擅长清洗数据、统计分析和可视化方案设计。"
    user_prompt = f"任务：{state.get('question', '')}"
    messages = [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": user_prompt}
    ]
    response = llm.invoke(messages)
    return {
        "analysis_result": response.content,
        "messages": [{"role": "assistant", "content": f"数据分析完成: {response.content}"}]
    }

这样每个智能体的职责边界就更加清晰了。

5. 多智能体架构进阶：记忆、人工审批与并行执行

5.1 持久化记忆

上文提到了 Checkpointer，这里我们实际看一下。

如果希望多轮对话中，智能体能够记住上下文，就需要将执行状态保存下来。最简单的做法是使用内存存储。

python复制from langgraph.checkpoint.memory import InMemorySaver

memory = InMemorySaver()

app = create_graph_with_checkpointer()

config = {"configurable": {"thread_id": "conversation-1"}}

# 第一轮
result1 = app.invoke(
    {"question": "帮我分析销售数据", "messages": []},
    config=config
)

# 第二轮，同一个会话
result2 = app.invoke(
    {"question": "顺便帮我生成每周趋势图代码", "messages": []},
    config=config
)

在业务系统中，可以将会话 ID 与用户 ID 绑定。生产环境用 SQLite 或 Redis 作为后端存储。

5.2 人工审批节点

在不少企业中，AI 生成的内容不能直接执行，需要人工审批。LangGraph 的 interrupt 机制可以完美支持这种场景。

以代码智能体为例：当代码生成后，流程暂停，等人在审核通过后再继续执行最终汇总。

python复制from langgraph.types import interrupt
from langgraph.checkpoint.memory import InMemorySaver


def code_agent_with_review(state: AgentState) -> dict:
    # 生成代码
    code = llm_call(f"代码生成任务: {state.get('question', '')}")

    # 暂停等待人工审批
    human_review = interrupt({"code_preview": code})

    if human_review.get("approved"):
        return {"code_result": code}
    else:
        return {"code_result": "[人工未通过，已取消结果]"}

关于 interrupt 的具体参数与用法，不同 LangGraph 版本略有差异。它的核心思想是在执行到该节点时挂起流程，外部系统获取待办任务、人工处理后，再通过 app.invoke 或 app.resume 继续执行。

这种方式非常适合“AI 生成 -> 人工审批 -> 正式执行”的企业场景。

5.3 多智能体并行执行

有些任务可以并行处理。比如用户要求“同时分析销售数据和生成分析报告框架”，这两个任务互不依赖，可以并行执行。

LangGraph 可以通过 Fan-out 边实现并行。

python复制graph.add_edge("route_agent", "data_agent")
graph.add_edge("route_agent", "code_agent")

当 route_agent 执行完成后，data_agent 和 code_agent 会同时启动。LangGraph 内部通过并发执行加快整体速度。

不过需要注意：并行执行时，多个节点对 State 的更新可能产生竞争。因此对共享字段要根据业务语义决定是否使用 reducer。

5.4 多智能体性能与成本

在实际项目中，多智能体虽然灵活，但也带来了更高的成本。有多少个智能体参与，通常就会调用多少次模型，Prompt 设计也会更长。建议根据业务收益来判断是否真的需要多智能体，而不是为了技术炫技强行上多智能体架构。

如果任务链路只有两三个固定分支，用 LangGraph 的单图流程或 LangChain 的链式调用完全够用。

6. 常见问题与排查思路

6.1 常见错误与解决方案

6.2 多智能体调试建议

多智能体系统比单智能体更难调试，建议从三个层面入手：

1. 状态追踪：在关键节点打印 State 内容。
2. 流程追踪：给每个 Node 添加日志，记录节点名称、入参和出参。
3. 模型调用追踪：开启模型的详细日志，检查每个节点的 Prompt 是否合理。

也可以通过 LangGraph 官方的可视化工具查看图结构，确认路由关系是否符合预期。如果不方便使用在线工具，可以画一张简单的 ASCII 图：

text复制START -> route_agent
route_agent -> data_agent (包含"数据"关键词)
route_agent -> code_agent (包含"代码"关键词)
route_agent -> final_agent (其他情况)
data_agent -> final_agent
code_agent -> final_agent
final_agent -> END

6.3 状态管理注意事项

多智能体中，State 是共享的。这意味着一个节点写入的内容，其他节点可以读到。但这也会带来一个潜在问题：节点之间的隐式耦合。

比如 final_agent 读取了 analysis_result 和 code_result，那么如果 data_agent 节点未执行，final_agent 可能会因为字段缺失而报错。建议在所有读取字段时使用 state.get("字段名", 默认值) 来规避。

python复制def final_agent(state: AgentState) -> dict:
    analysis = state.get("analysis_result", "")
    code = state.get("code_result", "")
    ...

这样即使某个分支没有执行，流程也不会直接崩溃。

7. 最佳实践与工程建议

7.1 智能体职责边界设计

不要把多个职责塞进同一个智能体。建议遵循“一个智能体只做一件事”的原则。比如“数据分析智能体”不要同时负责“代码生成”，否则又会退化回单智能体的混乱状态。

在 Prompt 设计中，明确以下信息：

• 智能体的角色。
• 它负责什么任务。
• 它不应该处理什么任务。
• 它输出的格式要求。

7.2 配置管理与安全

• 所有 API Key 通过环境变量或密钥管理服务注入。
• 不要将密钥提交到代码仓库。
• 如果需要多环境隔离（开发、测试、生产），建议将模型配置、路由规则等做成配置项。
• 在涉及数据库、删除、文件覆盖等敏感操作时，必须有权限校验和人工审批机制。

7.3 错误处理与重试

模型调用可能因为限流、网络波动等原因失败，多智能体场景更容易出现这类问题，因为调用次数更多。

建议封装一个带重试机制的模型调用函数：

python复制import time
from functools import wraps


def retry(max_retries=3, delay=2):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    print(f"调用失败，第 {attempt + 1} 次重试: {e}")
                    time.sleep(delay)
            raise RuntimeError("模型调用多次失败")
        return wrapper
    return decorator


@retry(max_retries=3)
def call_llm(prompt: str) -> str:
    ...

7.4 可观测性

生产环境中的多智能体系统，必须关注日志和链路追踪。建议在每个智能体节点中记录结构化日志：

python复制import logging
logger = logging.getLogger(__name__)


def data_agent(state: AgentState) -> dict:
    logger.info(f"[data_agent] 开始执行，问题: {state.get('question', '')}")
    result = ...
    logger.info(f"[data_agent] 完成，结果长度: {len(result)}")
    return result

如果团队有链路追踪体系，可以在每个 Node 中埋入 trace_id，方便全链路排查。

7.5 测试策略

多智能体系统测试重点包括：

1. 路由测试：给定不同输入，验证是否进入正确分支。
2. 单节点测试：每个智能体节点单独测试。
3. 集成测试：验证完整流程能够串联并产出最终结果。
4. 回归测试：模型提示词改动后，验证关键场景未出现回退。

7.6 避免过度设计

多智能体方案虽然在宣传上很吸引人，但并非所有业务都需要。如果任务逻辑简单、分支固定，用普通函数流程反而更稳定、成本更低。

适合多智能体的场景通常具备以下特征：

• 业务链路长且包含多个独立子任务。
• 不同环节需要不同的 Prompt 和领域知识。
• 希望具备并行执行能力。
• 需要多智能体协作，而不是简单线性执行。

如果你的项目只是“调用一次模型、解析一次输出”，完全不需要引入 LangGraph。

8. 总结与下一步学习路线

通过本文，你已经搭建并运行了一个基本的 LangGraph 多智能体系统。回顾一下关键内容：

• 多智能体架构解决的是“单智能体难以编排复杂流程”的问题。
• LangGraph 通过 State、Node、Edge 和 Checkpointer 提供了完整的流程编排能力。
• Supervisor 模式是最通用的多智能体协作模式。
• 状态管理是多智能体开发的核心，务必厘清每个字段的归属与更新方式。
• 有条件路由（Conditional Edge）让“管理者智能体”能够根据任务内容动态选择执行者。
• 生产环境必须关注成本、日志、权限、人工审批与安全边界。

下一步可以按以下路线继续深入：

1. 把模拟 LLM 调用替换为真实模型，测试不同模型在实际业务中的表现。
2. 尝试在 LangGraph 中接入外部工具（如 SQL 工具、代码解释器、搜索接口），让智能体具备“行动能力”。
3. 学习 Checkpointer 的持久化配置，在业务系统中实现连续对话和断点恢复。
4. 设计一套完整的“AI 生成 -> 人工审批 -> 自动执行”的企业级流程。
5. 关注 LangGraph 官方文档和更新日志，及时适应版本 API 变化。

多智能体是一个实操性很强的领域，与其在概念里绕圈，不如先把一个最简单的三智能体流程跑起来，再逐步增加复杂度。如果你在运行本文代码时遇到任何问题，建议先从状态字段和路由映射入手排查，大部分新手问题都出在这两个环节。

动手改一改 agents.py 中的路由关键词，添加你自己的业务分支，很快就能感受到 LangGraph 在流程编排上的灵活度。