尧图网页设计 
近两年基于大模型的应用开发已经明显从“单模型调用”走向“多智能体协作”。很多开发者发现单纯调一次大模型接口很容易但真正要把复杂业务拆成多个角色、让不同 Agent 按流程协作、在关键节点做条件判断代码复杂度会立刻上升。LangGraph 正是为了解决这类问题出现的编排框架。这篇文章会从多智能体架构讲起拆解 LangGraph 的核心组件并用两个完整的代码实战演示从单 Agent 到多 Agent 协作的全过程同时整理了环境安装、常见报错和工程落地建议。1. 多智能体与 LangGraph 基础概念1.1 为什么需要多智能体架构在早期的 AI 应用开发中我们通常使用一个 Prompt 把所有要求写进去让大模型一次性完成从理解问题到生成答案的全流程。这种方式在简单场景下够用但业务复杂后会暴露几个明显问题单一任务里混合了太多职责Prompt 越长模型越容易丢失细节。很难对中间过程做控制例如校验用户输入、判断是否需要查数据库、决定是否调用外部工具。无法复用已经开发好的智能体每次新需求都要重新写一套逻辑。多智能体架构的核心思路是把一个复杂任务拆分成多个可以独立工作的 Agent每个 Agent 只负责一个领域或一个操作。比如客服系统里可以拆成意图识别 Agent、订单查询 Agent、售后处理 Agent 和人工转接 Agent。每个 Agent 单独维护、单独升级再由一个上层编排引擎按照流程和条件触发调度。多智能体的常见交互模式主要有四种单一决策者模式一个主 Agent 统一接收用户请求再决定调用哪个子 Agent。顺序协作模式多个 Agent 按固定流程依次执行前一个 Agent 的输出作为后一个 Agent 的输入。层级管理模式存在多个管理层级父 Agent 管理多个子 Agent子 Agent 之间不直接通信。自由协作模式Agent 之间可以互相传递消息共同完成一个复杂目标。LangGraph 对上述几种模式都有比较好的表达能力因为它本身就是有向图结构图中每个节点是一个执行单元每条边是节点之间的流转关系。1.2 LangGraph 是什么LangGraph 是 LangChain 社区推出的一个编排框架它借用图计算的思想来构建大模型应用。开发者把业务流程定义成一张图图的节点是函数、工具或 Agent图的边是状态转移逻辑。LangGraph 负责调度这些节点维护状态并在需要时支持循环、条件分支、并行执行和记忆管理。相比直接用 Python 写一个循环调用大模型LangGraph 的优势在于状态管理内置化整张图共享一个 State 对象节点之间通过 State 传递数据。支持条件路由根据大模型输出或者函数返回值动态决定下一个执行节点。支持循环执行便于实现“反复思考直到达到条件”的 Agent 行为。支持子图嵌套一个图可以作为一个节点嵌入到另一个图中。提供 Checkpointer可以把每一步的图状态保存下来实现断点续跑、人工审批和记忆持久化。LangGraph 和 LangChain 的关系也很容易被新手混淆。LangChain 偏重封装各种工具和大模型接口比如文档加载、向量检索、Prompt 模板、模型调用封装。LangGraph 则偏重流程编排和状态管理它可以独立使用也可以和 LangChain 配合使用。你可以只用 LangGraph自己写节点函数实现全部逻辑也可以把 LangChain 的 AgentExecutor 封装进某个节点。1.3 LangGraph 的适用场景LangGraph 比较适合需要复杂流程控制的 AI 应用典型场景包括企业内部知识库问答需要根据问题内容分流到不同检索通道。数据分析助手需要先判断用户意图再决定调用哪张表、哪个 SQL 模板。自动化报告生成多个 Agent 分别负责数据获取、数据清洗、文本生成和格式排版。客服工单系统需要按对话状态执行不同分支比如转人工、升级处理等。需要人工审批的流程例如 AI 生成文案后先暂停请求人工确认再继续执行。如果只是写一个简单的 Prompt 调用接口不涉及多步判断和状态流转LangGraph 可能偏重了。但如果业务中有多个工具、多个角色、多种条件路径LangGraph 的价值就会体现得很明显。2. 环境准备与安装在开始写代码之前先把运行环境准备好。LangGraph 目前以 Python 生态为主虽然社区也有人讨论 Rust 移植版本但官方推荐路径仍然是 Python。2.1 Python 环境建议建议使用 Python 3.10 及以上版本新版 LangGraph 对 Python 3.9 以下版本的支持已经逐渐弱化。如果你本机有多个 Python 版本建议为当前项目创建独立虚拟环境。创建虚拟环境python -m venv langgraph_env激活虚拟环境# Windows langgraph_env\Scripts\activate # macOS / Linux source langgraph_env/bin/activate激活成功后命令行前缀会变化说明当前已经进入虚拟环境。2.2 安装 LangGraph 及相关依赖安装 LangGraph 本体pip install langgraph如果需要配合 LangChain 使用建议同时安装pip install langchain langchain-core langchain-openai这里需要说明版本问题。LangGraph 更新比较快不同版本之间的 API 可能有细微调整。比如StateGraph的初始化方式、add_conditional_edges的写法在 0.1.x 和 0.2.x 中都有变化。本文示例以当前主流写法为准如果你安装的版本较新遇到 API 变化时可以优先查阅官方文档或者项目目录下的site-packages/langgraph源码。如果希望在本地浏览器中调试图和 Agent 运行过程可以安装 LangGraph Studio 对应的 CLI 工具pip install langgraph-cli[inmem]启动本地开发模式langgraph dev这个命令会启动一个本地服务可以在浏览器中可视化查看图结构、节点状态和运行轨迹。不过它是一个可选项不安装也不影响本文的代码示例。为了方便调试还可以安装 jieba 和中文字符串处理工具pip install jieba2.3 项目结构本文的实战案例会围绕一个简化版“智能客服 数据分析助手”展开项目结构如下langgraph-demo/ ├── .env ├── requirements.txt ├── agent_common.py ├── agent_router.py ├── agent_subgraph.py ├── main.pyagent_common.py存放公共状态定义和工具函数。agent_router.py单 Agent 示例演示基础状态流转。agent_subgraph.py多 Agent 示例演示条件路由和子图。main.py入口脚本运行多个示例。如果你的业务相对简单不拆分多个文件也可以但拆分文件更利于后续维护和扩展。2.4 配置大模型 APILangGraph 本身不包含大模型能力需要配合 OpenAI、通义千问、文心一言、DeepSeek、本地模型等推理服务使用。最通用的方式是配置环境变量。在项目根目录创建.env文件OPENAI_API_KEY你的API Key如果你使用的是国内大模型服务通常需要自定义 base_url。LangChain 的 OpenAI 兼容接口可以这样配置from langchain_openai import ChatOpenAI model ChatOpenAI( model你的模型名称, api_key你的API Key, base_urlhttps://你的服务地址/v1, temperature0 )需要注意不同平台的模型名称和接口地址不同本文示例只演示结构不绑定某一个具体平台。如果你使用本地部署的大模型也可以通过 Ollama 或 vLLM 暴露 OpenAI 兼容接口然后仍然使用ChatOpenAI接入。3. LangGraph 核心组件拆解3.1 核心组件有哪些LangGraph 的编程模型可以类比为一个有向图。理解它不需要太复杂的算法背景只需抓住以下六个核心概念StateGraph图的容器负责注册节点、添加边、编译图。State状态对象定义整张图共享的数据结构。Node节点每个节点是一个可调用函数。Edge边表示节点之间的流转关系。Conditional Edge条件边根据函数返回值动态决定下一个节点。Checkpointer检查点保存每一步的状态用于恢复和记忆。下面我们把每个概念拆开来看。3.2 StateGraph图的容器StateGraph是整个流程编排的入口。在创建图时需要传入状态的数据结构定义。通常我们定义一个TypedDict或 PydanticBaseModel用于描述整张图中流转的数据有哪些字段。from typing import TypedDict, Annotated import operator class State(TypedDict): query: str result: str messages: Annotated[list, operator.add]在这个状态定义中query是用户输入。result是最终输出。messages是对话过程中产生的消息列表。注意Annotated[list, operator.add]这种写法。它表示当多个节点同时写messages时不是覆盖而是将新消息追加到现有列表。这是 LangGraph 中常见的状态合并策略适合累积场景。当然你也可以不写这种高级注解直接用普通字典结构多个节点之间手动合并状态。但使用注解可以让 LangGraph 自动处理合并逻辑更省心。3.3 Node 与 Edge节点和边要给图添加一个节点需要指定节点名称和对应的处理函数。from langgraph.graph import StateGraph def my_node(state: dict) - dict: return {result: hello} graph StateGraph(State) graph.add_node(my_node, my_node)这里的my_node接收一个 State 字典返回一个部分更新的字典。LangGraph 会把这些字段合并进全局状态。节点定义好之后需要添加边来确定执行顺序。最简单的顺序边graph.add_edge(start_node, end_node)这表示start_node执行完成后会直接进入end_node。如果图只有一个入口可以设置起始节点graph.set_entry_point(start_node)也可以使用更通用的方式graph.add_edge(START, start_node)这里的START是 LangGraph 提供的特殊节点表示图处理的起点。3.4 Conditional Edge条件分支条件分支是 LangGraph 最核心的能力之一也是实现多智能体路由的基础。假设我们有两个下游节点agent_a和agent_b现在要根据某个函数返回值决定走哪条边from langgraph.graph import START, END def router_node(state: dict) - str: # 返回下一个节点的名称 if 天气 in state[query]: return agent_a return agent_b graph.add_conditional_edges( router_node, router_node, { agent_a: agent_a, agent_b: agent_b, } )add_conditional_edges接收三个参数第一个参数是源节点名称。第二个参数是路由函数输入 State输出某个标识。第三个参数是一个映射字典把标识映射到具体节点名称。这种设计把“判断逻辑”和“流程结构”解耦你可以在路由函数里写任意复杂的逻辑甚至调用另一个大模型来判断意图。3.5 循环与并行图结构和传统的线性流程最不一样的地方在于循环。我们可以让某个节点执行完成后重新回到之前的节点graph.add_edge(action_node, check_node) graph.add_conditional_edges( check_node, should_continue, { continue: action_node, finish: END, } )这个结构可以实现 Agent 反复执行“思考-行动-观察结果-再思考”的循环直到满足退出条件。并行分支则通过多个add_edge实现。例如一个节点执行完同时触发两个分支graph.add_edge(source_node, branch_a) graph.add_edge(source_node, branch_b) graph.add_edge(branch_a, merge_node) graph.add_edge(branch_b, merge_node)当merge_node被多个上游节点指向时LangGraph 会等所有上游执行完成后再执行merge_node。这种方式很适合“多个 Agent 并行检索数据最后统一汇总”的场景。3.6 子图 Subgraph子图其实就是把一个编译好的图当作另一个图的节点来使用。subgraph_app sub_graph.compile() def subgraph_node(state: dict) - dict: return subgraph_app.invoke({query: state[query]}) graph.add_node(subgraph_node, subgraph_node)通过这种方式你可以把复杂的业务封装成独立的子图在主图中只暴露一个节点。子图拥有自己独立的状态和流程便于团队分工开发也便于单元测试。3.7 Checkpointer 与记忆管理默认情况下每次调用图State 都是全新的。如果希望图具备记忆能力可以使用MemorySaver或SqliteSaver。from langgraph.checkpoint.memory import MemorySaver checkpointer MemorySaver() app graph.compile(checkpointercheckpointer) config {configurable: {thread_id: user_123}} app.invoke({query: 你好}, configconfig) app.invoke({query: 我刚刚说了什么}, configconfig)thread_id表示一个会话线程的标识。同一个线程内的多次调用可以共享历史状态。MemorySaver只是内存实现服务重启后数据会丢失生产环境建议使用SqliteSaver或 PostgreSQL 持久化方案。需要注意的是Checkpointer 保存的是图的状态快照和每一步的更新信息而不是通过外部数据库单独保存对话历史。它是 LangGraph 实现多轮对话记忆的底层机制。3.8 LangGraph 与 LangChain 的配合方式LangGraph 可以独立使用但在实际项目中大家经常和 LangChain 一起用。用 LangChain 的ChatOpenAI封装大模型接口。用 LangChain 的工具调用机制让大模型生成结构化工具参数。用 LangGraph 把这些能力编排成可控制的流程图。LangChain 的create_react_agent内部其实就是基于 LangGraph 实现的。我们可以在 LangGraph 的节点中直接调用 LangChain Agent也可以自己写节点逻辑调用大模型的bind_tools接口来实现工具调用。下面进入代码实战部分。4. 实战一构建一个带路由的智能体4.1 场景说明第一个实战案例实现一个简单的智能客服路由系统。用户输入一个问题系统先判断问题类型然后转发到对应 Agent如果问题中带有“订单”转订单处理 Agent。如果问题中带有“物流”转物流查询 Agent。其他问题转通用对话 Agent。这个例子虽然业务逻辑简单但完整演示了 LangGraph 的条件路由和节点调度是后续多智能体协作的基础。4.2 定义状态首先编写agent_router.py先导入必要模块并定义状态。# 文件路径langgraph-demo/agent_router.py from typing import TypedDict from langgraph.graph import StateGraph, START, END class RouterState(TypedDict): query: str response: str routed_to: str这里定义了三个字段query用户的输入。response每个 Agent 处理后的回答。routed_to记录路由到了哪个 Agent方便调试。4.3 定义 Agent 节点三个 Agent 节点这里为了演示简化成普通函数实际项目中可以在这里调用大模型。def order_agent(state: RouterState) - dict: return { response: f【订单Agent】已处理你的订单问题{state[query]}, routed_to: order_agent, } def logistics_agent(state: RouterState) - dict: return { response: f【物流Agent】已处理你的物流问题{state[query]}, routed_to: logistics_agent, } def general_agent(state: RouterState) - dict: return { response: f【通用Agent】已收到你的问题{state[query]}, routed_to: general_agent, }每个节点接收state返回一个部分状态字典。LangGraph 会把返回的字段合并到全局 State 中。4.4 定义路由节点路由节点的作用是根据用户输入判断应该走哪个分支。def router_node(state: RouterState) - str: query state[query] if 订单 in query: return order if 物流 in query: return logistics return general这个函数返回的是字符串标识不是字典。因为它是条件边的路由函数LangGraph 会拿这个返回值去映射字典中查找对应的目标节点。4.5 构建图接下来把节点注册到图中并连接条件边。def build_router_graph(): graph StateGraph(RouterState) graph.add_node(router, router_node) graph.add_node(order, order_agent) graph.add_node(logistics, logistics_agent) graph.add_node(general, general_agent) graph.add_edge(START, router) graph.add_conditional_edges( router, router_node, { order: order, logistics: logistics, general: general, } ) graph.add_edge(order, END) graph.add_edge(logistics, END) graph.add_edge(general, END) return graph.compile()这里有一点容易踩坑add_conditional_edges的第一个参数是源节点名称第二个参数是路由函数第三个参数是映射字典。如果你的路由函数返回了映射字典中不存在的 keyLangGraph 会直接报错。如果你希望这种情况下走默认分支可以给add_conditional_edges传入一个path_map并确保路由函数返回的每个值都在映射中或者传入None来表示忽略。通常建议映射完整不要依赖隐式默认。4.6 运行与验证在文件底部加入测试代码if __name__ __main__: app build_router_graph() test_queries [ 我想查询一下订单状态, 快递到哪里了, 今天天气怎么样, ] for q in test_queries: result app.invoke({query: q}) print(f用户问题{q}) print(f路由结果{result[routed_to]}) print(f回答内容{result[response]}) print( * 50)运行该脚本python agent_router.py预期输出类似用户问题我想查询一下订单状态 路由结果order 回答内容【订单Agent】已处理你的订单问题我想查询一下订单状态 用户问题快递到哪里了 路由结果logistics 回答内容【物流Agent】已处理你的物流问题快递到哪里了 用户问题今天天气怎么样 路由结果general 回答内容【通用Agent】已收到你的问题今天天气怎么样 到这里一个最简单但结构完整的 LangGraph 条件路由程序就跑通了。5. 实战二多智能体协作与子图结合5.1 场景说明第二个实战案例相对复杂演示多智能体协作。需求如下用户输入一个问题。主图首先判断问题是否与数据分析相关。如果相关进入数据分析子图。子图内部有两个并行节点一个负责汇总数据指标一个负责生成分析建议最后合并结果。如果不相关直接走通用对话节点。输出最终结果并在最终响应中带上来路信息。这个案例会用到Conditional Edge实现主路由。Subgraph封装数据分析流程。并行分支汇总。状态合并。5.2 定义状态先创建agent_subgraph.py定义主图和子图共享的状态。# 文件路径langgraph-demo/agent_subgraph.py from typing import TypedDict, Annotated, List import operator from langgraph.graph import StateGraph, START, END class AnalysisState(TypedDict): query: str metrics: str suggestion: str summary: str trace: Annotated[List[str], operator.add] class MainState(TypedDict): query: str final_answer: str trace: Annotated[List[str], operator.add]这里重点解释两个字段trace使用Annotated[List[str], operator.add]当多个节点并行写入时所有节点的贡献都会追加到同一个列表中不会互相覆盖。子图状态和主图状态分开定义子图不关心主图有哪些字段主图也不关注子图的内部字段只关心子图返回的结果。5.3 构建数据分析子图子图内部包含三个节点collect_metrics_node收集数据指标。generate_suggestion_node生成建议。merge_summary_node合并前两个节点的结果。def collect_metrics_node(state: AnalysisState) - dict: return { metrics: f指标汇总根据问题{state[query]}提取到核心指标A和指标B。, trace: [collect_metrics], } def generate_suggestion_node(state: AnalysisState) - dict: return { suggestion: f分析建议建议关注指标B的变化趋势并进一步拆分维度。, trace: [generate_suggestion], } def merge_summary_node(state: AnalysisState) - dict: merged f{state[metrics]}\n{state[suggestion]} return { summary: merged, trace: [merge_summary], }构建子图def build_analysis_subgraph(): sub_graph StateGraph(AnalysisState) sub_graph.add_node(collect_metrics, collect_metrics_node) sub_graph.add_node(generate_suggestion, generate_suggestion_node) sub_graph.add_node(merge_summary, merge_summary_node) sub_graph.add_edge(START, collect_metrics) sub_graph.add_edge(START, generate_suggestion) sub_graph.add_edge(collect_metrics, merge_summary) sub_graph.add_edge(generate_suggestion, merge_summary) sub_graph.add_edge(merge_summary, END) return sub_graph注意这里同时从START引出了两条边分别指向collect_metrics和generate_suggestion。LangGraph 会并行调度这两个节点并且等两个节点都执行完毕后再进入merge_summary。为什么要用并行而不是串行在这个场景里指标汇总和建议生成不互相依赖两者可以并行执行从而减少整体耗时。实际项目中如果两个节点都调用大模型并行会比串行明显更快。5.4 主图中接入子图主图需要定义一个子图包装节点。子图编译后其实就是一个可调用对象。def analysis_node(state: MainState) - dict: sub_graph build_analysis_subgraph().compile() sub_result sub_graph.invoke({query: state[query]}) return { final_answer: sub_result[summary], trace: [analysis_node], }这里把子图编译并调用的过程封装在一个普通函数里。因为子图有自己的内部状态所以主图不需要关心子图内部到底怎么流转。通用对话节点def general_node(state: MainState) - dict: return { final_answer: f通用回答你的问题是『{state[query]}』但目前只支持数据分析相关问题。, trace: [general_node], }路由函数def main_router(state: MainState) - str: if 分析 in state[query] or 数据 in state[query]: return analysis return general5.5 构建主图def build_main_graph(): main_graph StateGraph(MainState) main_graph.add_node(analysis, analysis_node) main_graph.add_node(general, general_node) main_graph.add_edge(START, analysis) main_graph.add_edge(START, general) main_graph.add_edge(analysis, END) main_graph.add_edge(general, END) app main_graph.compile() return app细心的话会发现这里我没有使用路由器节点而是直接从START引出两条边到两个节点并且两个节点都指向END。那么 LangGraph 会怎么处理呢答案是它会并行执行analysis和general两个节点然后把它们对final_answer的写入合并到最终状态中。当两个节点都往同一个字段写入字符串时后写入的值会覆盖先写入的值所以final_answer的最终值取决于节点执行顺序不一定符合预期。这正好引出 LangGraph 并行分支中一个非常容易踩的坑并行节点向同一个非合并字段写入会导致结果不确定性。为了稳妥我们把主图的逻辑改得更明确一点先经过一个路由节点再进入目标分支。修改如下def main_router(state: MainState) - str: if 分析 in state[query] or 数据 in state[query]: return analysis return general def build_main_graph(): main_graph StateGraph(MainState) main_graph.add_node(router, main_router) main_graph.add_node(analysis, analysis_node) main_graph.add_node(general, general_node) main_graph.add_edge(START, router) main_graph.add_conditional_edges( router, main_router, { analysis: analysis, general: general, } ) main_graph.add_edge(analysis, END) main_graph.add_edge(general, END) return main_graph.compile()这样同一个用户的请求只会进入一个分支不会出现并行写入冲突的问题。5.6 运行与验证if __name__ __main__: app build_main_graph() test_queries [ 请帮我分析一下上个月的销售数据, 你好你是谁, ] for q in test_queries: result app.invoke({query: q}) print(f用户问题{q}) print(f最终回答{result[final_answer]}) print(f执行轨迹{result[trace]}) print( * 50)运行python agent_subgraph.py预期输出用户问题请帮我分析一下上个月的销售数据 最终回答指标汇总根据问题请帮我分析一下上个月的销售数据提取到核心指标A和指标B。 分析建议建议关注指标B的变化趋势并进一步拆分维度。 执行轨迹[collect_metrics, generate_suggestion, analysis_node, merge_summary] 用户问题你好你是谁 最终回答通用回答你的问题是『你好你是谁』但目前只支持数据分析相关问题。 执行轨迹[general_node] 注意执行轨迹的顺序可能不完全一致因为并行节点的执行顺序和调度有关但merge_summary一定会在collect_metrics和generate_suggestion都执行完之后才执行。这是 LangGraph 图调度机制保证的。5.7 如何在节点里调用真实大模型上面两个实战为了聚焦流程Agent 节点内部都是普通逻辑。真实项目中节点里通常需要调用大模型。这里补充一个调用示例。from langchain_core.messages import HumanMessage, SystemMessage from langchain_openai import ChatOpenAI import os model ChatOpenAI( modelos.getenv(OPENAI_MODEL, gpt-4o-mini), api_keyos.getenv(OPENAI_API_KEY), base_urlos.getenv(OPENAI_BASE_URL), temperature0, ) def llm_node(state: MainState) - dict: messages [ SystemMessage(content你是一个智能助手请回答用户问题。), HumanMessage(contentstate[query]), ] response model.invoke(messages) return {final_answer: response.content} def build_llm_graph(): graph StateGraph(MainState) graph.add_node(llm, llm_node) graph.add_edge(START, llm) graph.add_edge(llm, END) return graph.compile() if __name__ __main__: from dotenv import load_dotenv load_dotenv() app build_llm_graph() result app.invoke({query: 你好简单介绍一下你自己}) print(result[final_answer])如果你没有安装python-dotenv可以用pip install python-dotenv安装。也可以直接在脚本里用os.environ设置环境变量但要注意不要把 API Key 提交到代码仓库。6. LangGraph 常见报错与排查思路LangGraph 开发过程中新手可能会遇到各种问题。下面整理一些比较常见的报错和排查思路。6.1 报错缺少graph_state或状态字段不存在错误现象KeyError: response可能原因节点函数返回的字段没有在 State 中定义。如果State是一个TypedDict节点返回的字段名称必须存在于TypedDict定义中否则 LangGraph 无法合并。排查步骤检查节点函数返回的字典 key 是否拼写正确。检查 State 定义是否遗漏了该字段。检查节点返回的是否是dict而不是字符串。解决方案在 State 中补充对应字段或者修改节点函数返回现有字段。6.2 条件边返回的 key 不在映射字典中错误现象ValueError: After conditional edge, expected node xxx not found in path map可能原因路由函数返回了某个标识但add_conditional_edges的第三个参数映射字典中不包含该标识。排查步骤打印路由函数的实际返回值。检查映射字典中是否包含所有可能返回的标识。解决方案把路由函数所有可能的输出值都写入映射字典或者确保路由函数只返回映射中存在的值。6.3 并行节点写入同一字段导致相互覆盖错误现象程序能运行但最终结果不稳定有时是 A 的结果有时是 B 的结果。可能原因两个并行节点都向final_answer字段写入字符串而该字段没有配置合并策略。排查步骤查看 State 定义中是否有Annotated[..., operator.add]。判断该字段是否适合追加而不是覆盖。解决方案方案一改用Annotated[List[str], operator.add]把每个节点的输出追加到列表。方案二增加一个合并节点等并行节点执行完后统一汇总。6.4 使用MemorySaver后状态不更新错误现象同一thread_id下多次调用模型看到的依然是第一次的状态。可能原因没有在compile时传入checkpointer或者每次调用时没有传入相同的config。排查步骤检查编译代码是graph.compile()还是graph.compile(checkpointercheckpointer)。检查每次invoke时是否传入了相同的thread_id。解决方案统一使用同一个checkpointer实例并在config中使用固定的thread_id。6.5 大模型调用超时或报 401错误现象AuthenticationError: Incorrect API key provided或APITimeoutError: Request timed out.可能原因API Key 配置不正确。base_url 或模型名称不符合服务方要求。网络环境不稳定。排查步骤先用 curl 或 Postman 直接测试接口连通性。检查环境变量读取是否成功。检查模型名称是否存在于服务方列表。解决方案确认 API Key 有效。设置外部服务时使用官方提供的 base_url。如果使用本地模型检查模型服务是否已启动端口是否正确。我整理了一份排查清单方便遇到问题时快速对照问题现象常见原因解决思路节点返回 KeyError返回字段不在 State 定义中检查字段名并同步补齐 State条件路由报错路由返回值不在映射字典中打印返回值并补全映射并行结果互相覆盖非合并字段多节点写入用 Annotated 合并或增加汇总节点多次调用状态不更新未传 checkpointer 或 thread_id 不一致统一 checkpointer 和 thread_id大模型调用 401API Key 或 base_url 错误直接测试接口核对配置LangGraph 版本 API 不一致框架更新导致写法变化查看当前版本源码和官方文档7. 多智能体工程落地建议看完上面的示例你已经能跑通一个多 Agent 流程。但在真实项目中还需要考虑更多工程问题。7.1 状态设计要克制很多初学者容易把 State 设计成一个大而全的字典所有中间结果都往里塞。这样会导致节点之间隐式耦合调式困难。建议只保留真正需要跨节点流转的数据。尽量让节点函数内部的数据不外泄。临时变量放在节点函数内部即可不需要写进 State。7.2 条件路由函数要有明确输出规范路由函数是整个图的关键路径必须保证输出稳定、可预期。建议路由函数内尽量用结构化逻辑判断不要依赖大模型自由文本输出。如果必须让大模型做路由让模型输出 JSON再解析后映射到节点名称。路由函数要有兜底返回避免出现未定义分支。7.3 日志和追踪要贯穿整个流程在生产环境中你是否能快速定位是哪一步出了问题建议在节点函数内部加日志记录输入输出摘要。使用 LangGraph 官方的回调机制或者langsmith做链路追踪。在 State 中保留 trace 字段记录节点执行顺序。7.4 错误处理要在节点内完成LangGraph 的图调度本身不会捕获节点内的异常。如果某个节点内部报错整张图会直接中断。建议在每个节点函数内部做 try-except并返回一个错误状态字段让上层节点根据状态决定是重试还是走兜底分支。示例def safe_node(state: dict) - dict: try: result do_something(state[query]) return {result: result, error: None} except Exception as e: return {result: , error: str(e)}主图中有错误字段时可以在路由函数里判断是否需要走重试或人工处理分支。7.5 安全与权限控制多智能体系统可能涉及数据库读取、文件访问、外部系统调用安全边界尤其重要。节点内部调用外部工具时要校验输入参数避免把用户输入直接拼进 SQL 或 Shell 命令。涉及用户私有数据时先做权限校验再进入子图处理。API Key 必须走环境变量或密钥管理服务不能硬编码在代码中。如果 Agent 需要操作生产环境建议增加人工确认节点先暂停流程等待审批。7.6 性能优化思路多智能体系统的主要性能瓶颈通常在大模型调用和外部工具调用。可以从几个方向优化并行分支替代串行调用。小模型做路由判断大模型做生成任务。对不频繁变化的数据做缓存。限制循环节点的最大执行次数避免死循环。使用流式输出时结合stream方法逐 token 返回给前端。7.7 从 LangGraph 0.2 到新版升级LangGraph 版本升级过程中最常遇到的变化包括StateGraph的导入路径可能从langgraph.graph调整到新模块。MemorySaver的导入路径已经多次变化。add_conditional_edges的签名增加了path_map和output_keys等参数。升级时建议先阅读官方 changelog运行现有测试用例不要盲目升级框架版本。8. 总结与学习路线本文围绕 LangGraph 多智能体实战展开详细拆解了多智能体架构的基本概念、LangGraph 的核心组件、条件路由、子图嵌套、并行分支、记忆管理等关键知识点并提供了两个可直接运行的实战示例。你已经掌握的技能包括理解 LangGraph 与 LangChain 的关系与区别。掌握 StateGraph、State、Node、Edge、条件路由和子图的使用方法。会构建一个简单的条件路由智能体。会构建一个带子图和并行分支的多 Agent 系统。知道常见报错的排查思路。了解多智能体系统在生产环境中的工程注意事项。接下来你可以继续学习的方向LangGraph 官方文档中关于Command、SendAPI 的用法用于动态创建多个任务分支。尝试把 LangGraph 与向量数据库结合实现 RAG 多跳检索。尝试把 LangGraph 接入 FastAPI对外提供 Agent 服务接口。研究 LangGraph Studio 的可视化调试能力提升开发效率。参考上海交大《动手学大模型》等开源教程进一步完善大模型应用的基本功。多智能体编排是一个实践性非常强的方向。建议你先把本文两个示例跑通然后尝试改造成你自己的业务场景比如知识库问答、数据分析助手、自动化客服。遇到问题不要怕把报错信息拆开看从 State 和 Edge 这两个角度去分析绝大部分问题都能找到答案。 LangGraph实战：从单Agent到多智能体协作
近两年基于大模型的应用开发已经明显从“单模型调用”走向“多智能体协作”。很多开发者发现单纯调一次大模型接口很容易但真正要把复杂业务拆成多个角色、让不同 Agent 按流程协作、在关键节点做条件判断代码复杂度会立刻上升。LangGraph 正是为了解决这类问题出现的编排框架。这篇文章会从多智能体架构讲起拆解 LangGraph 的核心组件并用两个完整的代码实战演示从单 Agent 到多 Agent 协作的全过程同时整理了环境安装、常见报错和工程落地建议。1. 多智能体与 LangGraph 基础概念1.1 为什么需要多智能体架构在早期的 AI 应用开发中我们通常使用一个 Prompt 把所有要求写进去让大模型一次性完成从理解问题到生成答案的全流程。这种方式在简单场景下够用但业务复杂后会暴露几个明显问题单一任务里混合了太多职责Prompt 越长模型越容易丢失细节。很难对中间过程做控制例如校验用户输入、判断是否需要查数据库、决定是否调用外部工具。无法复用已经开发好的智能体每次新需求都要重新写一套逻辑。多智能体架构的核心思路是把一个复杂任务拆分成多个可以独立工作的 Agent每个 Agent 只负责一个领域或一个操作。比如客服系统里可以拆成意图识别 Agent、订单查询 Agent、售后处理 Agent 和人工转接 Agent。每个 Agent 单独维护、单独升级再由一个上层编排引擎按照流程和条件触发调度。多智能体的常见交互模式主要有四种单一决策者模式一个主 Agent 统一接收用户请求再决定调用哪个子 Agent。顺序协作模式多个 Agent 按固定流程依次执行前一个 Agent 的输出作为后一个 Agent 的输入。层级管理模式存在多个管理层级父 Agent 管理多个子 Agent子 Agent 之间不直接通信。自由协作模式Agent 之间可以互相传递消息共同完成一个复杂目标。LangGraph 对上述几种模式都有比较好的表达能力因为它本身就是有向图结构图中每个节点是一个执行单元每条边是节点之间的流转关系。1.2 LangGraph 是什么LangGraph 是 LangChain 社区推出的一个编排框架它借用图计算的思想来构建大模型应用。开发者把业务流程定义成一张图图的节点是函数、工具或 Agent图的边是状态转移逻辑。LangGraph 负责调度这些节点维护状态并在需要时支持循环、条件分支、并行执行和记忆管理。相比直接用 Python 写一个循环调用大模型LangGraph 的优势在于状态管理内置化整张图共享一个 State 对象节点之间通过 State 传递数据。支持条件路由根据大模型输出或者函数返回值动态决定下一个执行节点。支持循环执行便于实现“反复思考直到达到条件”的 Agent 行为。支持子图嵌套一个图可以作为一个节点嵌入到另一个图中。提供 Checkpointer可以把每一步的图状态保存下来实现断点续跑、人工审批和记忆持久化。LangGraph 和 LangChain 的关系也很容易被新手混淆。LangChain 偏重封装各种工具和大模型接口比如文档加载、向量检索、Prompt 模板、模型调用封装。LangGraph 则偏重流程编排和状态管理它可以独立使用也可以和 LangChain 配合使用。你可以只用 LangGraph自己写节点函数实现全部逻辑也可以把 LangChain 的 AgentExecutor 封装进某个节点。1.3 LangGraph 的适用场景LangGraph 比较适合需要复杂流程控制的 AI 应用典型场景包括企业内部知识库问答需要根据问题内容分流到不同检索通道。数据分析助手需要先判断用户意图再决定调用哪张表、哪个 SQL 模板。自动化报告生成多个 Agent 分别负责数据获取、数据清洗、文本生成和格式排版。客服工单系统需要按对话状态执行不同分支比如转人工、升级处理等。需要人工审批的流程例如 AI 生成文案后先暂停请求人工确认再继续执行。如果只是写一个简单的 Prompt 调用接口不涉及多步判断和状态流转LangGraph 可能偏重了。但如果业务中有多个工具、多个角色、多种条件路径LangGraph 的价值就会体现得很明显。2. 环境准备与安装在开始写代码之前先把运行环境准备好。LangGraph 目前以 Python 生态为主虽然社区也有人讨论 Rust 移植版本但官方推荐路径仍然是 Python。2.1 Python 环境建议建议使用 Python 3.10 及以上版本新版 LangGraph 对 Python 3.9 以下版本的支持已经逐渐弱化。如果你本机有多个 Python 版本建议为当前项目创建独立虚拟环境。创建虚拟环境python -m venv langgraph_env激活虚拟环境# Windows langgraph_env\Scripts\activate # macOS / Linux source langgraph_env/bin/activate激活成功后命令行前缀会变化说明当前已经进入虚拟环境。2.2 安装 LangGraph 及相关依赖安装 LangGraph 本体pip install langgraph如果需要配合 LangChain 使用建议同时安装pip install langchain langchain-core langchain-openai这里需要说明版本问题。LangGraph 更新比较快不同版本之间的 API 可能有细微调整。比如StateGraph的初始化方式、add_conditional_edges的写法在 0.1.x 和 0.2.x 中都有变化。本文示例以当前主流写法为准如果你安装的版本较新遇到 API 变化时可以优先查阅官方文档或者项目目录下的site-packages/langgraph源码。如果希望在本地浏览器中调试图和 Agent 运行过程可以安装 LangGraph Studio 对应的 CLI 工具pip install langgraph-cli[inmem]启动本地开发模式langgraph dev这个命令会启动一个本地服务可以在浏览器中可视化查看图结构、节点状态和运行轨迹。不过它是一个可选项不安装也不影响本文的代码示例。为了方便调试还可以安装 jieba 和中文字符串处理工具pip install jieba2.3 项目结构本文的实战案例会围绕一个简化版“智能客服 数据分析助手”展开项目结构如下langgraph-demo/ ├── .env ├── requirements.txt ├── agent_common.py ├── agent_router.py ├── agent_subgraph.py ├── main.pyagent_common.py存放公共状态定义和工具函数。agent_router.py单 Agent 示例演示基础状态流转。agent_subgraph.py多 Agent 示例演示条件路由和子图。main.py入口脚本运行多个示例。如果你的业务相对简单不拆分多个文件也可以但拆分文件更利于后续维护和扩展。2.4 配置大模型 APILangGraph 本身不包含大模型能力需要配合 OpenAI、通义千问、文心一言、DeepSeek、本地模型等推理服务使用。最通用的方式是配置环境变量。在项目根目录创建.env文件OPENAI_API_KEY你的API Key如果你使用的是国内大模型服务通常需要自定义 base_url。LangChain 的 OpenAI 兼容接口可以这样配置from langchain_openai import ChatOpenAI model ChatOpenAI( model你的模型名称, api_key你的API Key, base_urlhttps://你的服务地址/v1, temperature0 )需要注意不同平台的模型名称和接口地址不同本文示例只演示结构不绑定某一个具体平台。如果你使用本地部署的大模型也可以通过 Ollama 或 vLLM 暴露 OpenAI 兼容接口然后仍然使用ChatOpenAI接入。3. LangGraph 核心组件拆解3.1 核心组件有哪些LangGraph 的编程模型可以类比为一个有向图。理解它不需要太复杂的算法背景只需抓住以下六个核心概念StateGraph图的容器负责注册节点、添加边、编译图。State状态对象定义整张图共享的数据结构。Node节点每个节点是一个可调用函数。Edge边表示节点之间的流转关系。Conditional Edge条件边根据函数返回值动态决定下一个节点。Checkpointer检查点保存每一步的状态用于恢复和记忆。下面我们把每个概念拆开来看。3.2 StateGraph图的容器StateGraph是整个流程编排的入口。在创建图时需要传入状态的数据结构定义。通常我们定义一个TypedDict或 PydanticBaseModel用于描述整张图中流转的数据有哪些字段。from typing import TypedDict, Annotated import operator class State(TypedDict): query: str result: str messages: Annotated[list, operator.add]在这个状态定义中query是用户输入。result是最终输出。messages是对话过程中产生的消息列表。注意Annotated[list, operator.add]这种写法。它表示当多个节点同时写messages时不是覆盖而是将新消息追加到现有列表。这是 LangGraph 中常见的状态合并策略适合累积场景。当然你也可以不写这种高级注解直接用普通字典结构多个节点之间手动合并状态。但使用注解可以让 LangGraph 自动处理合并逻辑更省心。3.3 Node 与 Edge节点和边要给图添加一个节点需要指定节点名称和对应的处理函数。from langgraph.graph import StateGraph def my_node(state: dict) - dict: return {result: hello} graph StateGraph(State) graph.add_node(my_node, my_node)这里的my_node接收一个 State 字典返回一个部分更新的字典。LangGraph 会把这些字段合并进全局状态。节点定义好之后需要添加边来确定执行顺序。最简单的顺序边graph.add_edge(start_node, end_node)这表示start_node执行完成后会直接进入end_node。如果图只有一个入口可以设置起始节点graph.set_entry_point(start_node)也可以使用更通用的方式graph.add_edge(START, start_node)这里的START是 LangGraph 提供的特殊节点表示图处理的起点。3.4 Conditional Edge条件分支条件分支是 LangGraph 最核心的能力之一也是实现多智能体路由的基础。假设我们有两个下游节点agent_a和agent_b现在要根据某个函数返回值决定走哪条边from langgraph.graph import START, END def router_node(state: dict) - str: # 返回下一个节点的名称 if 天气 in state[query]: return agent_a return agent_b graph.add_conditional_edges( router_node, router_node, { agent_a: agent_a, agent_b: agent_b, } )add_conditional_edges接收三个参数第一个参数是源节点名称。第二个参数是路由函数输入 State输出某个标识。第三个参数是一个映射字典把标识映射到具体节点名称。这种设计把“判断逻辑”和“流程结构”解耦你可以在路由函数里写任意复杂的逻辑甚至调用另一个大模型来判断意图。3.5 循环与并行图结构和传统的线性流程最不一样的地方在于循环。我们可以让某个节点执行完成后重新回到之前的节点graph.add_edge(action_node, check_node) graph.add_conditional_edges( check_node, should_continue, { continue: action_node, finish: END, } )这个结构可以实现 Agent 反复执行“思考-行动-观察结果-再思考”的循环直到满足退出条件。并行分支则通过多个add_edge实现。例如一个节点执行完同时触发两个分支graph.add_edge(source_node, branch_a) graph.add_edge(source_node, branch_b) graph.add_edge(branch_a, merge_node) graph.add_edge(branch_b, merge_node)当merge_node被多个上游节点指向时LangGraph 会等所有上游执行完成后再执行merge_node。这种方式很适合“多个 Agent 并行检索数据最后统一汇总”的场景。3.6 子图 Subgraph子图其实就是把一个编译好的图当作另一个图的节点来使用。subgraph_app sub_graph.compile() def subgraph_node(state: dict) - dict: return subgraph_app.invoke({query: state[query]}) graph.add_node(subgraph_node, subgraph_node)通过这种方式你可以把复杂的业务封装成独立的子图在主图中只暴露一个节点。子图拥有自己独立的状态和流程便于团队分工开发也便于单元测试。3.7 Checkpointer 与记忆管理默认情况下每次调用图State 都是全新的。如果希望图具备记忆能力可以使用MemorySaver或SqliteSaver。from langgraph.checkpoint.memory import MemorySaver checkpointer MemorySaver() app graph.compile(checkpointercheckpointer) config {configurable: {thread_id: user_123}} app.invoke({query: 你好}, configconfig) app.invoke({query: 我刚刚说了什么}, configconfig)thread_id表示一个会话线程的标识。同一个线程内的多次调用可以共享历史状态。MemorySaver只是内存实现服务重启后数据会丢失生产环境建议使用SqliteSaver或 PostgreSQL 持久化方案。需要注意的是Checkpointer 保存的是图的状态快照和每一步的更新信息而不是通过外部数据库单独保存对话历史。它是 LangGraph 实现多轮对话记忆的底层机制。3.8 LangGraph 与 LangChain 的配合方式LangGraph 可以独立使用但在实际项目中大家经常和 LangChain 一起用。用 LangChain 的ChatOpenAI封装大模型接口。用 LangChain 的工具调用机制让大模型生成结构化工具参数。用 LangGraph 把这些能力编排成可控制的流程图。LangChain 的create_react_agent内部其实就是基于 LangGraph 实现的。我们可以在 LangGraph 的节点中直接调用 LangChain Agent也可以自己写节点逻辑调用大模型的bind_tools接口来实现工具调用。下面进入代码实战部分。4. 实战一构建一个带路由的智能体4.1 场景说明第一个实战案例实现一个简单的智能客服路由系统。用户输入一个问题系统先判断问题类型然后转发到对应 Agent如果问题中带有“订单”转订单处理 Agent。如果问题中带有“物流”转物流查询 Agent。其他问题转通用对话 Agent。这个例子虽然业务逻辑简单但完整演示了 LangGraph 的条件路由和节点调度是后续多智能体协作的基础。4.2 定义状态首先编写agent_router.py先导入必要模块并定义状态。# 文件路径langgraph-demo/agent_router.py from typing import TypedDict from langgraph.graph import StateGraph, START, END class RouterState(TypedDict): query: str response: str routed_to: str这里定义了三个字段query用户的输入。response每个 Agent 处理后的回答。routed_to记录路由到了哪个 Agent方便调试。4.3 定义 Agent 节点三个 Agent 节点这里为了演示简化成普通函数实际项目中可以在这里调用大模型。def order_agent(state: RouterState) - dict: return { response: f【订单Agent】已处理你的订单问题{state[query]}, routed_to: order_agent, } def logistics_agent(state: RouterState) - dict: return { response: f【物流Agent】已处理你的物流问题{state[query]}, routed_to: logistics_agent, } def general_agent(state: RouterState) - dict: return { response: f【通用Agent】已收到你的问题{state[query]}, routed_to: general_agent, }每个节点接收state返回一个部分状态字典。LangGraph 会把返回的字段合并到全局 State 中。4.4 定义路由节点路由节点的作用是根据用户输入判断应该走哪个分支。def router_node(state: RouterState) - str: query state[query] if 订单 in query: return order if 物流 in query: return logistics return general这个函数返回的是字符串标识不是字典。因为它是条件边的路由函数LangGraph 会拿这个返回值去映射字典中查找对应的目标节点。4.5 构建图接下来把节点注册到图中并连接条件边。def build_router_graph(): graph StateGraph(RouterState) graph.add_node(router, router_node) graph.add_node(order, order_agent) graph.add_node(logistics, logistics_agent) graph.add_node(general, general_agent) graph.add_edge(START, router) graph.add_conditional_edges( router, router_node, { order: order, logistics: logistics, general: general, } ) graph.add_edge(order, END) graph.add_edge(logistics, END) graph.add_edge(general, END) return graph.compile()这里有一点容易踩坑add_conditional_edges的第一个参数是源节点名称第二个参数是路由函数第三个参数是映射字典。如果你的路由函数返回了映射字典中不存在的 keyLangGraph 会直接报错。如果你希望这种情况下走默认分支可以给add_conditional_edges传入一个path_map并确保路由函数返回的每个值都在映射中或者传入None来表示忽略。通常建议映射完整不要依赖隐式默认。4.6 运行与验证在文件底部加入测试代码if __name__ __main__: app build_router_graph() test_queries [ 我想查询一下订单状态, 快递到哪里了, 今天天气怎么样, ] for q in test_queries: result app.invoke({query: q}) print(f用户问题{q}) print(f路由结果{result[routed_to]}) print(f回答内容{result[response]}) print( * 50)运行该脚本python agent_router.py预期输出类似用户问题我想查询一下订单状态 路由结果order 回答内容【订单Agent】已处理你的订单问题我想查询一下订单状态 用户问题快递到哪里了 路由结果logistics 回答内容【物流Agent】已处理你的物流问题快递到哪里了 用户问题今天天气怎么样 路由结果general 回答内容【通用Agent】已收到你的问题今天天气怎么样 到这里一个最简单但结构完整的 LangGraph 条件路由程序就跑通了。5. 实战二多智能体协作与子图结合5.1 场景说明第二个实战案例相对复杂演示多智能体协作。需求如下用户输入一个问题。主图首先判断问题是否与数据分析相关。如果相关进入数据分析子图。子图内部有两个并行节点一个负责汇总数据指标一个负责生成分析建议最后合并结果。如果不相关直接走通用对话节点。输出最终结果并在最终响应中带上来路信息。这个案例会用到Conditional Edge实现主路由。Subgraph封装数据分析流程。并行分支汇总。状态合并。5.2 定义状态先创建agent_subgraph.py定义主图和子图共享的状态。# 文件路径langgraph-demo/agent_subgraph.py from typing import TypedDict, Annotated, List import operator from langgraph.graph import StateGraph, START, END class AnalysisState(TypedDict): query: str metrics: str suggestion: str summary: str trace: Annotated[List[str], operator.add] class MainState(TypedDict): query: str final_answer: str trace: Annotated[List[str], operator.add]这里重点解释两个字段trace使用Annotated[List[str], operator.add]当多个节点并行写入时所有节点的贡献都会追加到同一个列表中不会互相覆盖。子图状态和主图状态分开定义子图不关心主图有哪些字段主图也不关注子图的内部字段只关心子图返回的结果。5.3 构建数据分析子图子图内部包含三个节点collect_metrics_node收集数据指标。generate_suggestion_node生成建议。merge_summary_node合并前两个节点的结果。def collect_metrics_node(state: AnalysisState) - dict: return { metrics: f指标汇总根据问题{state[query]}提取到核心指标A和指标B。, trace: [collect_metrics], } def generate_suggestion_node(state: AnalysisState) - dict: return { suggestion: f分析建议建议关注指标B的变化趋势并进一步拆分维度。, trace: [generate_suggestion], } def merge_summary_node(state: AnalysisState) - dict: merged f{state[metrics]}\n{state[suggestion]} return { summary: merged, trace: [merge_summary], }构建子图def build_analysis_subgraph(): sub_graph StateGraph(AnalysisState) sub_graph.add_node(collect_metrics, collect_metrics_node) sub_graph.add_node(generate_suggestion, generate_suggestion_node) sub_graph.add_node(merge_summary, merge_summary_node) sub_graph.add_edge(START, collect_metrics) sub_graph.add_edge(START, generate_suggestion) sub_graph.add_edge(collect_metrics, merge_summary) sub_graph.add_edge(generate_suggestion, merge_summary) sub_graph.add_edge(merge_summary, END) return sub_graph注意这里同时从START引出了两条边分别指向collect_metrics和generate_suggestion。LangGraph 会并行调度这两个节点并且等两个节点都执行完毕后再进入merge_summary。为什么要用并行而不是串行在这个场景里指标汇总和建议生成不互相依赖两者可以并行执行从而减少整体耗时。实际项目中如果两个节点都调用大模型并行会比串行明显更快。5.4 主图中接入子图主图需要定义一个子图包装节点。子图编译后其实就是一个可调用对象。def analysis_node(state: MainState) - dict: sub_graph build_analysis_subgraph().compile() sub_result sub_graph.invoke({query: state[query]}) return { final_answer: sub_result[summary], trace: [analysis_node], }这里把子图编译并调用的过程封装在一个普通函数里。因为子图有自己的内部状态所以主图不需要关心子图内部到底怎么流转。通用对话节点def general_node(state: MainState) - dict: return { final_answer: f通用回答你的问题是『{state[query]}』但目前只支持数据分析相关问题。, trace: [general_node], }路由函数def main_router(state: MainState) - str: if 分析 in state[query] or 数据 in state[query]: return analysis return general5.5 构建主图def build_main_graph(): main_graph StateGraph(MainState) main_graph.add_node(analysis, analysis_node) main_graph.add_node(general, general_node) main_graph.add_edge(START, analysis) main_graph.add_edge(START, general) main_graph.add_edge(analysis, END) main_graph.add_edge(general, END) app main_graph.compile() return app细心的话会发现这里我没有使用路由器节点而是直接从START引出两条边到两个节点并且两个节点都指向END。那么 LangGraph 会怎么处理呢答案是它会并行执行analysis和general两个节点然后把它们对final_answer的写入合并到最终状态中。当两个节点都往同一个字段写入字符串时后写入的值会覆盖先写入的值所以final_answer的最终值取决于节点执行顺序不一定符合预期。这正好引出 LangGraph 并行分支中一个非常容易踩的坑并行节点向同一个非合并字段写入会导致结果不确定性。为了稳妥我们把主图的逻辑改得更明确一点先经过一个路由节点再进入目标分支。修改如下def main_router(state: MainState) - str: if 分析 in state[query] or 数据 in state[query]: return analysis return general def build_main_graph(): main_graph StateGraph(MainState) main_graph.add_node(router, main_router) main_graph.add_node(analysis, analysis_node) main_graph.add_node(general, general_node) main_graph.add_edge(START, router) main_graph.add_conditional_edges( router, main_router, { analysis: analysis, general: general, } ) main_graph.add_edge(analysis, END) main_graph.add_edge(general, END) return main_graph.compile()这样同一个用户的请求只会进入一个分支不会出现并行写入冲突的问题。5.6 运行与验证if __name__ __main__: app build_main_graph() test_queries [ 请帮我分析一下上个月的销售数据, 你好你是谁, ] for q in test_queries: result app.invoke({query: q}) print(f用户问题{q}) print(f最终回答{result[final_answer]}) print(f执行轨迹{result[trace]}) print( * 50)运行python agent_subgraph.py预期输出用户问题请帮我分析一下上个月的销售数据 最终回答指标汇总根据问题请帮我分析一下上个月的销售数据提取到核心指标A和指标B。 分析建议建议关注指标B的变化趋势并进一步拆分维度。 执行轨迹[collect_metrics, generate_suggestion, analysis_node, merge_summary] 用户问题你好你是谁 最终回答通用回答你的问题是『你好你是谁』但目前只支持数据分析相关问题。 执行轨迹[general_node] 注意执行轨迹的顺序可能不完全一致因为并行节点的执行顺序和调度有关但merge_summary一定会在collect_metrics和generate_suggestion都执行完之后才执行。这是 LangGraph 图调度机制保证的。5.7 如何在节点里调用真实大模型上面两个实战为了聚焦流程Agent 节点内部都是普通逻辑。真实项目中节点里通常需要调用大模型。这里补充一个调用示例。from langchain_core.messages import HumanMessage, SystemMessage from langchain_openai import ChatOpenAI import os model ChatOpenAI( modelos.getenv(OPENAI_MODEL, gpt-4o-mini), api_keyos.getenv(OPENAI_API_KEY), base_urlos.getenv(OPENAI_BASE_URL), temperature0, ) def llm_node(state: MainState) - dict: messages [ SystemMessage(content你是一个智能助手请回答用户问题。), HumanMessage(contentstate[query]), ] response model.invoke(messages) return {final_answer: response.content} def build_llm_graph(): graph StateGraph(MainState) graph.add_node(llm, llm_node) graph.add_edge(START, llm) graph.add_edge(llm, END) return graph.compile() if __name__ __main__: from dotenv import load_dotenv load_dotenv() app build_llm_graph() result app.invoke({query: 你好简单介绍一下你自己}) print(result[final_answer])如果你没有安装python-dotenv可以用pip install python-dotenv安装。也可以直接在脚本里用os.environ设置环境变量但要注意不要把 API Key 提交到代码仓库。6. LangGraph 常见报错与排查思路LangGraph 开发过程中新手可能会遇到各种问题。下面整理一些比较常见的报错和排查思路。6.1 报错缺少graph_state或状态字段不存在错误现象KeyError: response可能原因节点函数返回的字段没有在 State 中定义。如果State是一个TypedDict节点返回的字段名称必须存在于TypedDict定义中否则 LangGraph 无法合并。排查步骤检查节点函数返回的字典 key 是否拼写正确。检查 State 定义是否遗漏了该字段。检查节点返回的是否是dict而不是字符串。解决方案在 State 中补充对应字段或者修改节点函数返回现有字段。6.2 条件边返回的 key 不在映射字典中错误现象ValueError: After conditional edge, expected node xxx not found in path map可能原因路由函数返回了某个标识但add_conditional_edges的第三个参数映射字典中不包含该标识。排查步骤打印路由函数的实际返回值。检查映射字典中是否包含所有可能返回的标识。解决方案把路由函数所有可能的输出值都写入映射字典或者确保路由函数只返回映射中存在的值。6.3 并行节点写入同一字段导致相互覆盖错误现象程序能运行但最终结果不稳定有时是 A 的结果有时是 B 的结果。可能原因两个并行节点都向final_answer字段写入字符串而该字段没有配置合并策略。排查步骤查看 State 定义中是否有Annotated[..., operator.add]。判断该字段是否适合追加而不是覆盖。解决方案方案一改用Annotated[List[str], operator.add]把每个节点的输出追加到列表。方案二增加一个合并节点等并行节点执行完后统一汇总。6.4 使用MemorySaver后状态不更新错误现象同一thread_id下多次调用模型看到的依然是第一次的状态。可能原因没有在compile时传入checkpointer或者每次调用时没有传入相同的config。排查步骤检查编译代码是graph.compile()还是graph.compile(checkpointercheckpointer)。检查每次invoke时是否传入了相同的thread_id。解决方案统一使用同一个checkpointer实例并在config中使用固定的thread_id。6.5 大模型调用超时或报 401错误现象AuthenticationError: Incorrect API key provided或APITimeoutError: Request timed out.可能原因API Key 配置不正确。base_url 或模型名称不符合服务方要求。网络环境不稳定。排查步骤先用 curl 或 Postman 直接测试接口连通性。检查环境变量读取是否成功。检查模型名称是否存在于服务方列表。解决方案确认 API Key 有效。设置外部服务时使用官方提供的 base_url。如果使用本地模型检查模型服务是否已启动端口是否正确。我整理了一份排查清单方便遇到问题时快速对照问题现象常见原因解决思路节点返回 KeyError返回字段不在 State 定义中检查字段名并同步补齐 State条件路由报错路由返回值不在映射字典中打印返回值并补全映射并行结果互相覆盖非合并字段多节点写入用 Annotated 合并或增加汇总节点多次调用状态不更新未传 checkpointer 或 thread_id 不一致统一 checkpointer 和 thread_id大模型调用 401API Key 或 base_url 错误直接测试接口核对配置LangGraph 版本 API 不一致框架更新导致写法变化查看当前版本源码和官方文档7. 多智能体工程落地建议看完上面的示例你已经能跑通一个多 Agent 流程。但在真实项目中还需要考虑更多工程问题。7.1 状态设计要克制很多初学者容易把 State 设计成一个大而全的字典所有中间结果都往里塞。这样会导致节点之间隐式耦合调式困难。建议只保留真正需要跨节点流转的数据。尽量让节点函数内部的数据不外泄。临时变量放在节点函数内部即可不需要写进 State。7.2 条件路由函数要有明确输出规范路由函数是整个图的关键路径必须保证输出稳定、可预期。建议路由函数内尽量用结构化逻辑判断不要依赖大模型自由文本输出。如果必须让大模型做路由让模型输出 JSON再解析后映射到节点名称。路由函数要有兜底返回避免出现未定义分支。7.3 日志和追踪要贯穿整个流程在生产环境中你是否能快速定位是哪一步出了问题建议在节点函数内部加日志记录输入输出摘要。使用 LangGraph 官方的回调机制或者langsmith做链路追踪。在 State 中保留 trace 字段记录节点执行顺序。7.4 错误处理要在节点内完成LangGraph 的图调度本身不会捕获节点内的异常。如果某个节点内部报错整张图会直接中断。建议在每个节点函数内部做 try-except并返回一个错误状态字段让上层节点根据状态决定是重试还是走兜底分支。示例def safe_node(state: dict) - dict: try: result do_something(state[query]) return {result: result, error: None} except Exception as e: return {result: , error: str(e)}主图中有错误字段时可以在路由函数里判断是否需要走重试或人工处理分支。7.5 安全与权限控制多智能体系统可能涉及数据库读取、文件访问、外部系统调用安全边界尤其重要。节点内部调用外部工具时要校验输入参数避免把用户输入直接拼进 SQL 或 Shell 命令。涉及用户私有数据时先做权限校验再进入子图处理。API Key 必须走环境变量或密钥管理服务不能硬编码在代码中。如果 Agent 需要操作生产环境建议增加人工确认节点先暂停流程等待审批。7.6 性能优化思路多智能体系统的主要性能瓶颈通常在大模型调用和外部工具调用。可以从几个方向优化并行分支替代串行调用。小模型做路由判断大模型做生成任务。对不频繁变化的数据做缓存。限制循环节点的最大执行次数避免死循环。使用流式输出时结合stream方法逐 token 返回给前端。7.7 从 LangGraph 0.2 到新版升级LangGraph 版本升级过程中最常遇到的变化包括StateGraph的导入路径可能从langgraph.graph调整到新模块。MemorySaver的导入路径已经多次变化。add_conditional_edges的签名增加了path_map和output_keys等参数。升级时建议先阅读官方 changelog运行现有测试用例不要盲目升级框架版本。8. 总结与学习路线本文围绕 LangGraph 多智能体实战展开详细拆解了多智能体架构的基本概念、LangGraph 的核心组件、条件路由、子图嵌套、并行分支、记忆管理等关键知识点并提供了两个可直接运行的实战示例。你已经掌握的技能包括理解 LangGraph 与 LangChain 的关系与区别。掌握 StateGraph、State、Node、Edge、条件路由和子图的使用方法。会构建一个简单的条件路由智能体。会构建一个带子图和并行分支的多 Agent 系统。知道常见报错的排查思路。了解多智能体系统在生产环境中的工程注意事项。接下来你可以继续学习的方向LangGraph 官方文档中关于Command、SendAPI 的用法用于动态创建多个任务分支。尝试把 LangGraph 与向量数据库结合实现 RAG 多跳检索。尝试把 LangGraph 接入 FastAPI对外提供 Agent 服务接口。研究 LangGraph Studio 的可视化调试能力提升开发效率。参考上海交大《动手学大模型》等开源教程进一步完善大模型应用的基本功。多智能体编排是一个实践性非常强的方向。建议你先把本文两个示例跑通然后尝试改造成你自己的业务场景比如知识库问答、数据分析助手、自动化客服。遇到问题不要怕把报错信息拆开看从 State 和 Edge 这两个角度去分析绝大部分问题都能找到答案。