LangGraph多智能体实战：状态管理与条件路由全解析

先说一个很现实的困惑很多同学学 LangGraph跟着官方文档把 ChatModel、ToolNode、StateGraph 都跑通了但一旦要做一个真正需要“多个角色协作”的复杂需求比如一个写代码专员、一个代码审查员、一个测试执行员协同完成任务立刻就不知道该怎么设计了。单个图会画多个节点叠加也会写但智能体之间怎么分工、怎么传递状态、怎么让一个智能体的输出成为另一个智能体的输入这些关键问题卡住了一大批人。这篇文章不打算重复官方文档的结构也不打算给你堆概念。我会用一套可运行的多智能体实战代码把 LangGraph 里最容易踩坑的架构设计、状态管理、节点编排、条件路由和并行分发讲清楚。读完以后你能独立设计一个多智能体协作系统而不是只会跑 demo。1. 多智能体为什么值得学先说一个判断多智能体解决的不是“模型能力”问题而是“流程确定性”问题。单 Agent 应用里大模型一般是“一个节点干到底”用户给需求模型直接产出结果。这种方式适合简单问答、单轮工具调用。但真实业务很少这么简单。举例来说软件开发流程里你希望 AI 先写方案、再写代码、再跑测试、再出重构建议每一步都有明确的输入输出和验收标准或者你做内容生产希望 AI 先生成大纲、再分别写不同章节、最后统一风格校对。这种流程用单 Agent 硬写提示词会变得极其臃肿而且一旦中间某一步出错整个会话状态很难回退和重试。多智能体架构把一个大任务拆成多个职责单一的子智能体每个子智能体有自己的 Prompt、自己的工具、自己的状态读写权限。LangGraph 的价值在于它把这种编排用图结构固化下来让开发者能精确控制每个节点的进入条件、执行顺序、失败分支和状态回流。相比 LangChain 时代的链式调用LangGraph 更接近“有状态的工作流引擎”这也是它适合做生产级 AI 应用的原因。从学习成本看多智能体比单智能体多出来的核心不是模型调用次数而是“状态如何共享”和“节点之间如何通信”。把这两个问题解决多智能体系统基本就通了。2. LangGraph 核心概念与架构认知2.1 LangGraph 与 LangChain 的区别LangChain 最初给人的印象是把各种模型、工具、向量库串成一条链。链是线性的适合顺序执行但遇到分支、循环、人工审批、回滚这类真实工作流就力不从心。LangGraph 在 LangChain 生态里补的正是图编排能力节点是函数边是路由规则状态在图中流转图本身可编译、可保存、可恢复。从架构上看LangGraph 更像是一个专门为大模型工作流设计的“有限状态机 图执行引擎”。它强调三件事所有节点共享一个状态对象但每个节点可以只读写自己的字段边的类型分普通边和条件边条件边由函数返回值决定下一步走向图可以被持久化Checkpoint 机制支持断点续跑、人工介入和时间旅行。2.2 StateGraph 的工作原理StateGraph 是 LangGraph 的核心类。它接收一个状态 SchemaSchema 决定了整个图里流转的数据结构。节点函数的形式是(state) - dict返回值会按照 reducer 规则合并进全局状态。这里很多人第一次会懵的地方是节点函数返回的 dict 覆盖的是整个 state还是只合并局部字段LangGraph 默认行为是合并到顶层字段如果你操作的是嵌套结构需要额外定义 reducer 或使用专门的更新方式。这个细节后面代码实战部分会单独演示。2.3 核心组件速览组件作用类比State / Schema全局数据模型所有节点共享项目组共享的白板Node一个执行单元通常是函数或 Runnable一个执行具体任务的员工Edge普通边无条件执行下一节点流水线上的固定传送带Conditional Edge根据返回结果动态选择下一节点带分拣开关的传送带Checkpoint保存图执行的完整快照游戏存档Send API动态并行分发到多个节点实例项目经理把任务批量派发给多个并行小组理解这六个组件基本就能读懂 80% 的 LangGraph 代码。3. 环境准备与前置条件3.1 运行环境本教程代码使用 Python 3.10 以上版本。推荐使用虚拟环境隔离依赖避免污染系统 Python。建议环境配置Python 3.10langgraph 0.2 以上版本langchain-core 0.2 以上版本langchain-openai 0.1 以上版本或你使用的模型厂商 SDKpython-dotenv 用于加载 API Key版本说明LangGraph 迭代速度较快2025 年后 API 有一些调整。本文示例不依赖过于冷门的特性升级到最新版也能运行。如果遇到 API 变动以你安装版本的官方文档为准。3.2 安装依赖pip install langgraph langchain-core langchain-openai python-dotenv如果你想用本地模型或 Ollama可以不装 langchain-openai改成对应的适配包。本文后续示例使用 OpenAI 兼容接口方便对接不同模型服务。3.3 配置 API Key在项目根目录创建.env文件OPENAI_API_KEYsk-xxxxxx BASE_URLhttps://api.example.com/v1 MODEL_NAMEgpt-4o-mini读取环境变量from dotenv import load_dotenv import os load_dotenv() OPENAI_API_KEY os.getenv(OPENAI_API_KEY) BASE_URL os.getenv(BASE_URL) MODEL_NAME os.getenv(MODEL_NAME)这里强调一个工程习惯永远不要把 API Key 直接写死在代码里。生产中建议使用密钥管理服务或环境变量注入。4. 多智能体架构设计4.1 架构选择的三个问题动手写代码之前先回答三个问题任务可以拆成几个互相独立、职责清晰的步骤吗步骤之间有明确的输入输出边界吗是否需要人工参与某个步骤的审批如果三个都是“是”那么多智能体架构就是合适的。如果任务高度耦合、步骤边界模糊强行拆智能体只会增加复杂度和 token 开销。4.2 一个真实的业务场景我们以“AI 软件开发助手”为例。这个助手接收用户需求完成三件事生成技术方案和代码对代码进行审查并给出修改建议根据审查意见修复代码最终产出完整的代码块。传统单 Agent 实现会把三件事塞进一个 Prompt模型需要自己切换角色。这种方式的问题在于方案阶段产生的中间思考会被代码阶段的上下文稀释审查阶段的规则容易和开发阶段提示词冲突状态不可控失败时难以定位是哪一步出的问题。多智能体方案定义三个子智能体architect_agent负责理解需求、输出技术方案和代码reviewer_agent以资深代码审查员身份审查代码输出问题列表fixer_agent根据审查意见修复代码输出最终版本。三者通过共享状态协作流程由 LangGraph 保证。接下来进入核心代码实战。5. 完整示例与代码实现5.1 定义状态 Schema状态 Schema 是多智能体系统的数据契约。字段越明确节点之间越不容易出现隐式依赖。from typing import TypedDict, List, Annotated import operator class AgentState(TypedDict): user_request: str # 用户原始需求 tech_scheme: str # 技术方案 generated_code: str # 生成的代码 review_comments: List[str] # 审查意见列表 fix_round: int # 修复轮次 final_code: str # 最终代码 messages: Annotated[List[str], operator.add] # 全局消息日志注意messages字段使用了Annotated[List[str], operator.add]这表示该字段的每次更新采用追加模式而不是覆盖模式。如果不声明这个 reducerLangGraph 默认是覆盖模式容易造成多智能体过程中日志丢失。这是 LangGraph 状态管理最容易踩的坑。5.2 初始化大模型为了演示方便这里封装一个统一获取模型的方法from langchain_openai import ChatOpenAI def get_llm(): return ChatOpenAI( modelMODEL_NAME, api_keyOPENAI_API_KEY, base_urlBASE_URL, temperature0.2, )temperature在代码生成和审查场景建议设置低一点减少随机性。5.3 定义节点函数节点函数的特点是输入整个 state输出一个 partial dict。每个节点不应该直接修改外部变量而是通过返回值声明自己的更新。5.3.1 架构师节点def architect_node(state: AgentState) - dict: llm get_llm() prompt f 你是一名资深软件架构师。请根据用户需求设计方案并输出代码。 需求{state[user_request]} 要求 1. 先说明技术选型理由 2. 再给出完整代码 3. 代码必须放在代码块中。 resp llm.invoke(prompt) content resp.content # 简单拆分假设技术方案和代码都在一个文本里 # 实际项目可以要求模型按 JSON 返回再解析 return { tech_scheme: content, generated_code: content, messages: [farchitect 完成: {content[:50]}...], }这个节点做的事情很集中把用户需求包装成架构师 Prompt让模型返回方案和代码。为了演示清晰方案和代码先不拆分实际项目建议让模型返回结构化 JSON。5.3.2 审查员节点def reviewer_node(state: AgentState) - dict: llm get_llm() prompt f 你是一名苛刻的代码审查专家。请审查以下代码指出存在的问题。 代码 {state[generated_code]} 要求 1. 只输出问题列表编号排列 2. 对每个问题标注严重级别高/中/低 3. 如果没有问题输出 APPROVED。 resp llm.invoke(prompt) content resp.content if APPROVED in content: return {review_comments: [], messages: [reviewer 通过审查]} return {review_comments: [content], messages: [reviewer 提出修改意见]}审查节点是典型的质量闸门它不直接修改代码只产生意见列表。这种职责划分让后续路由逻辑非常清晰。5.3.3 修复员节点def fixer_node(state: AgentState) - dict: llm get_llm() comments \n.join(state[review_comments]) prompt f 你是一名高级开发工程师。请根据审查意见修复代码。 原始需求{state[user_request]} 当前代码{state[generated_code]} 审查意见{comments} 要求 1. 输出修复后的完整代码 2. 简要说明你做了哪些修改。 resp llm.invoke(prompt) return { generated_code: resp.content, final_code: resp.content, fix_round: state[fix_round] 1, review_comments: [], messages: [fixer 完成修复], }修复节点会清空审查意见避免下一次审查继续看到旧问题。这里也使用return {generated_code: resp.content}来更新字段关键点是它把修复后的代码覆盖到generated_code实现循环迭代。5.4 条件路由函数条件路由是多智能体系统灵活性的关键。路由函数根据当前状态决定下一步进入哪个节点。def should_continue(state: AgentState) - str: if state[review_comments] and state[fix_round] 3: return fix if state[review_comments]: return max_round_exceeded return end逻辑不难有审查意见就回去修复最多修三轮防止死循环没有审查意见则结束流程。这是一个非常实用的“多轮修复”模式。5.5 构建和编译图from langgraph.graph import StateGraph, START, END def build_graph(): g StateGraph(AgentState) # 添加节点 g.add_node(architect, architect_node) g.add_node(reviewer, reviewer_node) g.add_node(fixer, fixer_node) # 添加边 g.add_edge(START, architect) g.add_edge(architect, reviewer) # 条件边审查后走路由 g.add_conditional_edges( reviewer, should_continue, { fix: fixer, end: END, max_round_exceeded: END, } ) # 修复后回到审查 g.add_edge(fixer, reviewer) return g.compile()这个流程已经具备真实业务闭环的基本形态起始节点 → 架构师 → 审查员 → 条件判断 → 修复员 → 回到审查员 → 最终结束。5.6 让多个修复任务动态并行上面的循环是串行循环。还有一种常见场景架构师把任务拆成两个独立模块两个模块的代码可以并行生成。LangGraph 的SendAPI 就是专门处理这种动态并行分发的。from langgraph.types import Send def continue_to_fix(state: AgentState) - List[Send]: # 假设有多个代码片段需要分别修复 tasks [] for idx, code in enumerate(state[generated_code]): tasks.append(Send(fixer, {generated_code: code, task_id: idx})) return tasks这里注意Send的第一个参数是目标节点名第二个参数是传递给该节点的“局部状态”。目标节点收到的是该子状态而不是完整状态。这种并行模式适合批量代码修复、多文件审查、多路内容生成等场景。6. 运行结果与效果验证6.1 运行代码if __name__ __main__: graph build_graph() initial_state { user_request: 写一个 Python 函数判断一个字符串是否是回文并包含单元测试, tech_scheme: , generated_code: , review_comments: [], fix_round: 0, final_code: , messages: [], } result graph.invoke(initial_state) print( 运行结束 ) print(消息记录, result[messages]) print(修复轮次, result[fix_round]) print(最终代码) print(result[final_code] or result[generated_code])6.2 判断运行成功成功运行的标志有两个流程正常到达END没有死循环result[messages]中能看到architect 完成、reviewer 提出修改意见或reviewer 通过审查等记录。如果配置正确你会在输出里看到类似记录消息记录 [architect 完成: 技术方案..., reviewer 提出修改意见, fixer 完成修复, reviewer 通过审查] 修复轮次 1 最终代码 ...修复后的完整代码...如果运行时出现Could not identify a node or edge, 基本可以确定是节点名或路由映射表里的 key 不一致。6.3 通过可视化验证流程LangGraph 编译后的图对象可以直接输出图片适合检查路由逻辑是否符合预期from IPython.display import Image, display graph build_graph() display(Image(graph.get_graph().draw_mermaid_png()))如果你不用 Jupyter也可以用graph.get_graph().to_json()查看节点和边的 JSON 结构来验证建模是否正确。7. 常见问题与排查思路多智能体开发比单智能体多了很多不确定性下面列出我见过的高频问题。7.1 状态覆盖导致日志丢失问题现象可能原因排查方式解决方案多个节点的返回值只有最后一个生效没有为 list 字段声明 reducer打印每次状态更新后的字段内容使用Annotated[List[str], operator.add]状态字段被节点意外修改节点函数内部直接操作了输入对象检查节点函数是否返回新 dict严格遵循函数式写法内部深拷贝再更新7.2 图进入死循环问题现象可能原因排查方式解决方案程序没有正常结束条件路由缺少退出机制检查路由函数所有分支是否都映射到节点或 END增加最大轮数限制超过阈值强制 END修复后重新进入审查但审查永远不通过审查标准过于苛刻或模型自身返回不稳定打印审查意见内容判断是模型问题还是 Prompt 问题降低审查要求或限制最大审查轮数7.3 模型调用报错或超时问题现象可能原因排查方式解决方案Rate limit 报错API 并发限制查看模型服务的配额和负载增加重试机制或使用max_concurrency控制并行数超时大模型生成时间过长检查输入 token 长度优化 Prompt减少历史消息累加7.4 了解LangGraph 中的 Send到底怎么用很多初学者看文档时会卡在Send上。Send不是把整个 state 发过去而是告诉图引擎“请用一份新的子状态去执行某个节点”。它适合批量并行任务分发。核心用法是放在一个函数里返回Send对象列表这个函数通常挂在某个节点之后作为动态路由。常见错误是把Send当成普通调用来执行一旦理解成“异步分发到节点的消息”这个 API 就不难了。7.5 LangGraph 与 LangChain 的关系很多人会混淆这两个项目。一句话总结LangChain 是模型调用和工具封装的工具库而 LangGraph 是流程编排引擎。LangGraph 可以完全不依赖 LangChain 使用但两者配合最方便。8. 最佳实践与工程建议多智能体项目能否从 demo 走向生产看的不是某次代码能不能跑通而是工程化细节是否有保障。8.1 状态设计先行先定义好完整的State再写节点逻辑。尤其要想清楚哪些字段是要累积的、哪些字段是覆盖的、哪些字段只读。好的状态设计能让多智能体协作变得清晰糟糕的状态设计会在后期产生大量互相覆盖的 bug。8.2 为每个智能体设置独立的 Prompt这里的独立不只是“不同字符串”还包括角色边界明确架构师不负责“修复”审查员不负责“编写新功能”输出格式明确最好要求 JSON 结构化输出然后用 Pydantic 校验输出约束明确限制代码长度、禁止某些遗留做法。8.3 必要的地方加人工审批生产环境里AI 生成的代码直接进入代码仓库极具风险。LangGraph 的interrupt_before参数可以在指定节点前暂停执行把控制权交给人。graph build_graph() # 在审查节点执行前暂停等待人工确认 graph graph.compile(interrupt_before[reviewer])这种“人机协同”模式比让 AI 全自动完成更符合真实工程需要尤其在涉及关键业务逻辑和权限操作时必须设置防线。遇到类似需求建议优先考虑用这个能力做安全边界。8.4 重视日志和可观测性多智能体系统里模型调用次数多、状态变化频繁必须从第一行代码开始就打日志。LangGraph 的messages累加字段不只是给模型看的更是定位问题的核心线索。生产环境可以接入 LangSmith 或自建日志中心记录每次节点执行的输入输出、token 消耗和耗时。8.5 控制并发与成本并行节点能提升速度但会放大成本。核心建议在Send分发前先判断并行任务数量是否合理对每个节点设置最大调用次数对不重要的节点使用更小的模型对最终成品节点使用强模型。8.6 多智能体不一定越多越好最后给一个偏经验的判断一个通用单智能体能解决的需求不要强行拆成多智能体。多智能体带来的是可控性和分工代价是复杂度、token 消耗、延迟和调试成本。只有流程边界清晰、需要不同角色扮演或者需要并行处理时多智能体架构才是更优解。从这个角度看很多团队在业务里真正需要的不是“N 个智能体”而是一个以流程为核心、允许局部并行和条件分发的“图谱化助手”。LangGraph 恰好同时支持这两种形态这也是它值得深入学习的原因。9. 总结与后续学习方向这篇文章从实际痛点出发梳理了 LangGraph 多智能体开发的核心路径状态 Schema 定义、节点职责拆分、条件路由、动态并行分发和人工审批。顺着这个路径你已经能搭建一个具备“开发→审查→修复→再审查”闭环的多智能体系统。接下来的学习建议是分三步走第一步把本文代码跑通理解should_continue路由和review_comments字段的流转方式第二步引入结构化输出Pydantic和 Checkpoint 持久化观察图恢复执行的效果第三步尝试用Send实现多文件并行修复再接入你的真实业务场景。由于 LangGraph 迭代较快配置 API 或版本兼容问题请以官方文档为准但核心设计理念和流程图思路具有较高的稳定性值得投入时间掌握。建议收藏本文遇到状态更新、路由死循环或并行分发的问题时回来看排查表能节省不少时间。