尧图网页设计 
如果你最近在写 Agent 应用八成已经碰到过这种场面单个大模型调用写得特别顺一问一答都很正常可一旦让多个 Agent 协作代码就变得像一团乱麻。你在 A 节点里调 B 节点的输出在 C 节点里又改回了 A 的状态prompt 越拼越长工具列表越挂越多最后连自己都说不清一次任务到底走到了哪一步。这个问题的本质不是大模型不够聪明而是我们没有给 Agent 提供一个可靠的“运行环境”。LangGraph 之所以在 AI 大模型应用开发里越来越火不是因为多了一个包装大模型的库而是它把 Agent 编排这件事从“写流程脚本”变成了“构建状态图”。你可以像设计一个有向无环图一样设计任务流程让每个 Agent 只负责自己那一段由框架来管理状态传递、条件分支、并行执行、循环终止和持久化记忆。这篇文章我会从多智能体架构设计的角度出发完整拆解 LangGraph 的核心组件然后用三个可以直接运行的代码实战带你从零搭出一个具备条件路由、子图复用、并行分发、Supervisor 主管编排的多智能体系统。最后再给出常见问题和工程化建议。整篇文章不需要你预先有很深的多智能体基础但建议你已经会写简单的 Python并且知道 ChatOpenAI 这类模型接口的基本用法。1. 多智能体开发为什么需要编排框架先看一个真实场景。假设你要做一个内容生成系统需要“选题 Agent”“写作 Agent”“审核 Agent”三个角色协作。如果不用编排框架你会怎么做大概率是手写三个函数A 函数返回结果B 函数接收并加工C 函数再审核。过程中只要有一个环节需要人工确认、或者需要根据上一轮结果动态决定下一步派给谁你的 if-else 就会开始失控。多智能体的价值从来不是“人多力量大”而是“分工明确 上下文隔离 职责可复用”。选题 Agent 只需要维护选题相关的工具和 prompt写作 Agent 只需要关心写作规范审核 Agent 只需要掌握审核标准。每个 Agent 自身的复杂度下降了但 Agent 之间的调度复杂度却上升了。调度复杂度体现在三个地方状态不知道在哪。上一次任务跑到了哪一步A 产生了什么中间结果这些结果怎么传给 B分支逻辑写死在业务代码里。条件一多if-else 嵌套流程根本没法维护。没有统一的观测和恢复机制。出错了不知道在哪一步出错想从中间恢复更是难上加难。LangGraph 解决的就是这三个问题。它允许你把多智能体系统建模为一个图节点是 Agent 或普通函数边是流转规则State 是贯穿全程的共享状态Conditional Edges 是动态分支Checkpointer 是每一步的持久化快照。换句话说LangGraph 让你从“写一段会调用模型的脚本”升级为“设计一个可测试、可控制、可回滚的 Agent 执行引擎”。这是多智能体走向生产环境时最需要补上的一课。2. LangGraph 与 LangChain 的关系和核心概念很多人会把 LangGraph 和 LangChain 混在一起甚至认为 LangGraph 就是 LangChain 的最新版本。实际上两者定位不同关系可以这样理解LangChain 是组件库负责提供模型封装、Prompt 模板、向量库、工具调用这些基础能力。LangGraph 是编排运行时负责把这些组件组织成一个可执行的图管理状态流转和分支控制。如果做一个类比LangChain 像是你采购的“零件”LangGraph 像是把这些零件组装成机器的“流水线控制系统”。你可以只用 LangGraph 不依赖 LangChain 的很多高级封装也可以只使用 LangChain 快速实现 RAG 检索但要让多个 Agent 稳定协作LangGraph 是最直接的解决方案。对比维度LangChainLangGraph定位组件库 / 开发框架编排运行时核心抽象Chain、AgentExecutorStateGraph、State状态管理隐式传递显式 State可持久化流程控制顺序、简单分支条件路由、循环、并行、子图适合场景RAG 管道、快速原型复杂多智能体、生产级工作流接下来LangGraph 的几个核心概念必须理解清楚后面所有代码都会围绕它们展开。State全局共享的状态对象。每个节点都可以读取 State也可以返回一个 dict 来更新 State。Node一个节点就是一个 Python 函数。函数接收当前 State返回部分更新。Edge从一个节点到另一个节点的确定连线。Conditional Edge根据 State 内容动态决定下一个节点是分支控制的灵魂。Checkpointer检查点机制保存每个步骤的执行状态支持断点、恢复、记忆。把这几个概念串起来你定义一个 State 数据结构然后往 StateGraph 里注册多个 Node用 Edge 或 Conditional Edge 把它们连接起来最后编译成一个 Graph 对象。运行阶段你调用graph.invoke()传入初始 State框架会按照图结构自动调度节点执行直到遇到 END。3. 环境准备与安装先搭好环境。本文示例基于 Python 3.10 及以上版本这是 LangGraph 目前比较稳妥的运行环境。建议创建独立虚拟环境避免污染其他项目python -m venv .venv source .venv/bin/activate # Windows 使用 .venv\Scripts\activate安装 LangGraph 及相关依赖pip install -U langgraph langchain-core langchain-openai这里说明一下langchain-openai是 LangChain 对 OpenAI 兼容接口的封装后面调用大模型时用它。如果你用的是本地部署模型如 vLLM 或 Ollama也可以通过base_url参数接入 OpenAI 兼容接口。安装完成后验证环境是否正常python -c from langgraph.graph import StateGraph; print(StateGraph)如果成功打印出类信息说明环境安装成功。大模型的 API Key 建议放在环境变量里不要直接写进代码。示例export OPENAI_API_KEYsk-xxxx在正式代码中我会用os.getenv(OPENAI_API_KEY)来读取。如果你没有 OpenAI Key也可以用本地模型服务把base_url指向本地地址例如from langchain_openai import ChatOpenAI model ChatOpenAI( modelqwen2.5:7b, api_keyEMPTY, base_urlhttp://localhost:8000/v1, )这样就把模型调用替换成本地服务了。不过为了演示流程我会先给出一个不依赖真实大模型的版本确保你把多智能体编排本身跑通。4. LangGraph 核心组件详解4.1 State所有节点共享的“可编程内存”State 是 LangGraph 的灵魂。它本质上是一个 TypedDict 或 Pydantic 模型定义了这次任务运行中所有节点都能访问的数据结构。每个节点函数接收当前 State然后返回一个 dictLangGraph 会自动把返回的字段合并进 State。这里容易踩一个坑如果你返回的是列表字段默认行为是“覆盖”而不是“追加”。需要把列表字段标记为Annotated[List, operator.add]LangGraph 才会用 reducer 做追加合并。后面并行分发实战里会演示。State 的设计原则是只放节点之间需要共享的信息不要把所有中间变量都塞进去。State 越大图越难调试。4.2 Node最普通的 Python 函数Node 没有黑魔法就是一个普通函数。它的输入是 State输出是一个 dict表示对 State 的部分更新。def my_node(state: MyState) - dict: # 读取 state query state[query] # 做一些处理 result f处理结果: {query} # 返回部分更新 return {result: result}函数不需要写副作用不需要自己维护全局变量所有数据都通过 State 流转。这个约束让每个节点都可以独立测试也方便后续加日志和监控。4.3 Edge 和 Conditional Edge从“写死流程”到“动态路由”普通 Edge 表示顺序执行比如A - B - C。Conditional Edge 则定义了一个路由函数函数根据当前 State 的内容返回下一个要执行节点的名字。这是多智能体系统里最重要的设计之一。因为 Agent 系统的特点就是流程不确定到底该让谁先处理取决于用户问题、中间结果、甚至模型判断。条件路由把这种不确定性显式建模出来不再是散落在各个函数里的 if-else。def route_by_category(state: MyState) - str: if state[category] tech: return tech_node return other_node builder.add_conditional_edges(classify, route_by_category)路由函数返回的字符串必须是在图上注册过的节点名或者 END。很多初学者在这里出错返回了一个没注册的名字LangGraph 运行时会直接报错。4.4 Checkpointer让 Agent 拥有“记忆”和“恢复能力”Checkpointer 负责在图的每个节点执行前后保存 State 快照。有了检查点你可以实现中断和恢复一次长任务运行到一半进程挂了重启后从上次检查点继续。对话记忆把每一步的消息列表保留下来作为下一次 invoke 的上下文。多轮交互通过thread_id把不同轮次的对话关联到同一个检查点。最简单的演示级 Checkpointer 是内存版数据只存在于当前进程重启即丢失from langgraph.checkpoint.memory import InMemorySaver saver InMemorySaver() graph builder.compile(checkpointersaver) result graph.invoke( {messages: []}, config{configurable: {thread_id: demo-thread}} )关于“怎么使用 InMemorySaver 中的内容构建传入大模型的上下文”需要理解检查点保存的是完整的 State 快照而不是自动拼接好的 prompt。当你恢复某个 thread 时LangGraph 会从检查点读出状态继续执行图的后续节点。如果你想把这些历史消息喂给大模型只需要从 State 里读取消息列表字段传给模型即可。它解决的是“持久化”问题不是“prompt 拼接”问题。4.5 Tool 集成一切业务能力都可以变成 Agent 的工具在 LangGraph 里一个节点内部可以调用任何东西包括普通函数、API、数据库查询、LangChain 的 Tool、MCP 协议暴露的工具等。最近有个很火的梗是“如何将小龙虾或者爱马仕集成到多智能体系统中”。其实剥开来看答案很简单在多智能体系统中任何能被 Python 函数包装的外部能力都可以成为 Agent 的工具。比如你有一个“小龙虾养殖池塘环境监测”函数把它用tool装饰一下注册给对应的 Agent这个 Agent 就能调度它。爱马仕如果指的是商品库存查询服务也是同样的接入思路。from langchain_core.tools import tool def query_pond(pond_id: str) - str: # 实际项目中这里会读数据库或传感器 return f池塘 {pond_id}: 水温 26℃溶氧 5.2mg/L状态正常 tool def shrimp_pond_tool(pond_id: str) - str: 查询小龙虾养殖池塘的实时环境数据。 return query_pond(pond_id)MCP 在这个体系里扮演的是“工具协议标准化”的角色。工具提供方用一个统一的 MCP Server 暴露能力Agent 端通过 MCP Client 发现和调用工具。LangGraph 不限制工具来源你可以直接绑定 LangChain Tool也可以接入自定义工具函数还可以集成 MCP 工具。5. 多智能体架构设计与四种常见交互模式多智能体的架构设计没有唯一的官方标准但业界常见的归纳方式可以把多智能体交互模式分为四类分别是中心化编排、流水线、并行分发、层次化。5.1 中心化编排模式Supervisor / Orchestrator-Worker一个主管 Agent 负责理解用户问题决定调哪个专家 Agent然后根据专家输出继续决策直到认为任务完成。这是生产中最常用、最稳妥的模式适合有明确任务划分的场景。用户输入 - Supervisor - Worker A、Worker B、Worker C - Supervisor - 最终输出LLM 主管是最典型的形式。Supervisor 本身也可以是一个不调用大模型的路由决策器直接根据规则选择专家。这个模式可控性强但主管 Agent 容易成为瓶颈因为所有消息都经过它。5.2 流水线模式Pipeline固定顺序执行前一个 Agent 的输出作为后一个 Agent 的输入。适合数据加工、多阶段生成的场景比如“内容选题 - 内容写作 - 内容审核”。优点是流程简单清晰缺点是扩展性和适应性差流程一旦变化就要改图结构。5.3 并行分发模式Fan-out / Parallel一个任务拆分成多个子任务同时分发给多个 worker 执行最后汇总结果。适合多路召回、多角色评审、批量数据处理。LangGraph 里用SendAPI 实现扇出用带 reducer 的字段实现结果聚合。5.4 层次化模式Hierarchical多级 Supervisor 嵌套每个主管管理一组专家 Agent上层主管只和下层主管通信。适合组织复杂、专家数量多的项目。优点是扩展性强缺点是通信成本和实现复杂度高。模式核心特征优点缺点典型场景中心化编排主管统一调度可控性强、好理解主管可能成为瓶颈客服、代码生成、内容创作流水线固定顺序执行简单直观不灵活内容加工、多阶段质检并行分发子任务同时执行吞吐高需要聚合逻辑多路检索、批量审核层次化多级主管嵌套扩展性强实现复杂大型组织流程、复杂业务选型建议刚开始做多智能体优先从中心化编排模式开始。先用一个 Supervisor 两个专家跑通流程再根据业务需要增加并行分发和子图复用。不要一上来就做层次化否则状态管理和调试难度会直线上升。6. 代码实战一条件路由与分支控制第一个实战我们不依赖真实大模型用规则做分类目标是把条件路由流程彻底跑通。这也回答了开发中最常用的需求用户请求进来后如何动态分配给不同的专家节点。场景根据用户 query 中的关键字把请求分流到编程专家、文案专家、兜底处理三种节点。# demo_router.py from typing import TypedDict, Literal from langgraph.graph import StateGraph, START, END class RouterState(TypedDict): query: str category: str answer: str def classify(state: RouterState) - dict: 简单规则分类。实际项目中这里可以换成 LLM 分类。 q state.get(query, ) if any(k in q for k in [bug, 代码, 程序, 报错]): category programmer elif any(k in q for k in [标题, 文案, 文章, 推广]): category copywriter else: category unknown return {category: category} def route_after_classify(state: RouterState) - Literal[programmer, copywriter, unknown]: return state[category] def programmer_node(state: RouterState) - dict: return {answer: f[编程专家] 收到任务{state[query]}} def copywriter_node(state: RouterState) - dict: return {answer: f[文案专家] 收到任务{state[query]}} def unknown_node(state: RouterState) - dict: return {answer: f[兜底处理] 无法分类的任务{state[query]}} builder StateGraph(RouterState) builder.add_node(classify, classify) builder.add_node(programmer, programmer_node) builder.add_node(copywriter, copywriter_node) builder.add_node(unknown, unknown_node) builder.add_edge(START, classify) builder.add_conditional_edges(classify, route_after_classify) builder.add_edge(programmer, END) builder.add_edge(copywriter, END) builder.add_edge(unknown, END) graph builder.compile() if __name__ __main__: for q in [帮我修一个 bug, 写一条推广文案, 明天天气如何]: result graph.invoke({query: q}) print(fQ: {q}) print(fA: {result[answer]}) print(- * 40)这段代码的核心在add_conditional_edges。它的第一个参数是源节点第二个参数是路由函数。路由函数接收当前 State返回目标节点的名字。你必须保证返回值是已经通过add_node注册过的节点名或者是 END否则运行时直接报“未知节点”错误。运行python demo_router.py预期输出Q: 帮我修一个 bug A: [编程专家] 收到任务帮我修一个 bug ---------------------------------------- Q: 写一条推广文案 A: [文案专家] 收到任务写一条推广文案 ---------------------------------------- Q: 明天天气如何 A: [兜底处理] 无法分类的任务明天天气如何 ----------------------------------------条件路由的本质不是“大模型在做判断”而是“基于状态执行一个决策函数”。决策函数里可以放规则也可以调用大模型还可以组合外部服务。关键点是决策逻辑必须独立、可测试这样整个图的流程才是可预测的。这里还要提一下循环控制。假设路由函数不小心把某个节点指回起点图会一直循环。LangGraph 默认有递归深度限制超出后会抛错。你可以通过graph.invoke(..., config{recursion_limit: 25})调整限制但在生产环境更推荐直接设计好终止条件不要依赖递归上限兜底。7. 代码实战二子图与并行分支7.1 子图把完整流程当做一个节点复用当你发现同一个 Agent 流程在多处都要使用时最直接的办法是把这段流程定义成子图然后在主图里作为一个节点使用。子图的输入输出通过主图节点的返回值做显式映射。下面示例把第 6 节的路由图当作子图嵌入一个新的主图# subgraph_demo.py from typing import TypedDict from langgraph.graph import StateGraph, START, END from demo_router import graph as router_graph class MainState(TypedDict): raw_query: str result: str def call_router_inner(state: MainState) - dict: # 调用子图传入子图需要的字段 sub_result router_graph.invoke({query: state[raw_query]}) # 把子图输出映射到主图 State return {result: sub_result[answer]} main_builder StateGraph(MainState) main_builder.add_node(call_router, call_router_inner) main_builder.add_edge(START, call_router) main_builder.add_edge(call_router, END) main_graph main_builder.compile() if __name__ __main__: out main_graph.invoke({raw_query: 帮我修一个 bug}) print(out[result])子图的优势是复用和解耦。如果你的团队已经沉淀了一个“代码审查 Agent”子图其他流程只要把这个子图注册成节点再处理好字段映射即可不需要知道子图内部是如何实现的。7.2 使用 Send 做并行分支并行分发是提高多智能体吞吐量的核心手段。LangGraph 的SendAPI 允许把一个节点扇出到多个相同节点的并行执行。比如批量对多个任务并行处理并汇总结果# parallel_demo.py import operator from typing import Annotated, TypedDict, List from langgraph.graph import StateGraph, START, END from langgraph.constants import Send class EvalState(TypedDict): tasks: List[str] results: Annotated[List[str], operator.add] def passthrough(state: EvalState) - dict: # 不需要做实际处理这个节点只是扇出起点 return {} def fan_out(state: EvalState) - List[Send]: # 给每个 task 发一个 worker 实例 return [Send(worker, {task: t}) for t in state[tasks]] def worker(state: dict) - dict: # 注意这里接收的是 Send 分发进来的单独 task return {results: [f已完成: {state[task]}]} eval_builder StateGraph(EvalState) eval_builder.add_node(fan_out, passthrough) eval_builder.add_node(worker, worker) eval_builder.add_edge(START, fan_out) eval_builder.add_conditional_edges(fan_out, fan_out) eval_builder.add_edge(worker, END) eval_graph eval_builder.compile() if __name__ __main__: out eval_graph.invoke({tasks: [需求分析, 代码编写, 测试执行]}) for r in out[results]: print(r)这里最关键的是results: Annotated[List[str], operator.add]。没有这个 reducer多个 worker 返回的结果会互相覆盖最后只剩最后一个结果。加了 reducer 之后所有 worker 返回的列表会被自动拼接起来。运行python parallel_demo.py预期输出已完成: 需求分析 已完成: 代码编写 已完成: 测试执行并行分支最常见的报错就是“为什么我只拿到了一个 worker 的结果”十有八九是 State 里聚合字段没有配置 reducer。7.3 循环检测的实践提醒LangGraph 支持循环图比如 Supervisor 模式里主管节点和专家节点之间就是循环边。循环本身不是问题但循环必须有一个明确的终止条件。推荐做法在 Supervisor 节点里显式判断“是否已经有专家回复过”有则输出 END。给图设置一个合理的recursion_limit防止异常情况下的死循环。在关键节点加日志观察每次循环时 State 的变化。8. 代码实战三Supervisor 多智能体编排完整实战第三个实战我们实现一个“主管 专家”的多智能体系统。用户输入问题Supervisor 先判断要交给哪个专家专家处理完成后回到 SupervisorSupervisor 再决定是继续派单还是结束。为了稳定运行我加了一个简单的终止条件只要消息里已经出现专家回复Supervisor 直接输出 END。# supervisor_demo.py import os from typing import TypedDict from langgraph.graph import StateGraph, START, END from langchain_openai import ChatOpenAI class AgentState(TypedDict): messages: list next: str class SupervisorDecision(BaseModel): next: str Field(descriptioncode_expert / copy_expert / END)先等等这里需要导入 pydantic 的 BaseModel 和 Field。在文件顶部补充from pydantic import BaseModel, Field完整代码如下# supervisor_demo.py import os from typing import TypedDict from pydantic import BaseModel, Field from langgraph.graph import StateGraph, START, END from langchain_openai import ChatOpenAI class AgentState(TypedDict): messages: list next: str class SupervisorDecision(BaseModel): next: str Field(descriptioncode_expert / copy_expert / END) def supervisor_node(state: AgentState) - dict: # 如果已经有过专家回复就直接结束 for m in state[messages]: if 专家] in m.get(content, ): return {next: END} prompt ( 你是多智能体主管。根据用户最新消息判断下一步派给谁 code_expert 处理编程问题copy_expert 处理文案问题。 如果判断任务已完成则输出 END。 ) model ChatOpenAI( modelgpt-4o-mini, api_keyos.getenv(OPENAI_API_KEY), ) structured_model model.with_structured_output(SupervisorDecision) decision structured_model.invoke( [{role: system, content: prompt}] state[messages] ) return {next: decision.next} def route_after_supervisor(state: AgentState) - str: return state.get(next, END) def code_expert_node(state: AgentState) - dict: answer [代码专家] 已根据需求生成 Python 代码请查收。 new_messages state[messages] [{role: assistant, content: answer}] return {messages: new_messages} def copy_expert_node(state: AgentState) - dict: answer [文案专家] 已输出一版推广文案请查收。 new_messages state[messages] [{role: assistant, content: answer}] return {messages: new_messages} builder StateGraph(AgentState) builder.add_node(supervisor, supervisor_node) builder.add_node(code_expert, code_expert_node) builder.add_node(copy_expert, copy_expert_node) builder.add_edge(START, supervisor) builder.add_conditional_edges( supervisor, route_after_supervisor, { code_expert: code_expert, copy_expert: copy_expert, END: END, }, ) builder.add_edge(code_expert, supervisor) builder.add_edge(copy_expert, supervisor) graph builder.compile() if __name__ __main__: messages [{role: user, content: 请帮我写一段 Python 排序代码}] result graph.invoke({messages: messages, next: }) for m in result[messages]: print(f{m[role]}: {m[content]})如果你没有配置OPENAI_API_KEY这个示例会卡在模型调用上。可以把 Supervisor 决策逻辑临时换成规则分类用bug或代码之类的关键字做路由这样没有 Key 也能跑通整个循环图结构。实际项目中Supervisor 不只是做路由它还需要承担上下文汇总、任务拆解、质量检查等职责。它的 prompt 设计应该明确说明有哪些专家 Agent 可用。每个专家 Agent 擅长什么。什么情况下判定任务已经完成。如果专家 Agent 返回结果不满足要求是否要重新派单。这才是 Supervisor 模式真正要解决的工程问题。9. 运行结果验证与常见问题排查运行本文所有示例按顺序执行python demo_router.py python subgraph_demo.py python parallel_demo.py python supervisor_demo.py前三个示例不依赖外部模型服务正常情况可以直接看到预期输出。supervisor_demo.py需要先配置好模型服务地址和 Key否则会超时或报认证错误。在生产或调试过程中最常遇到的问题集中在以下几个方面问题现象可能原因排查方式解决方案报错找不到MemorySaver/InMemorySaverLangGraph 版本不同类名或导入路径不同执行pip show langgraph查看版本根据版本选择langgraph.checkpoint.memory.InMemorySaver老版本可能是MemorySaver条件路由报“未知节点”路由函数返回了未注册的节点名打印路由函数的 LangGraph多智能体编排实战：从条件路由到Supervisor主管系统
如果你最近在写 Agent 应用八成已经碰到过这种场面单个大模型调用写得特别顺一问一答都很正常可一旦让多个 Agent 协作代码就变得像一团乱麻。你在 A 节点里调 B 节点的输出在 C 节点里又改回了 A 的状态prompt 越拼越长工具列表越挂越多最后连自己都说不清一次任务到底走到了哪一步。这个问题的本质不是大模型不够聪明而是我们没有给 Agent 提供一个可靠的“运行环境”。LangGraph 之所以在 AI 大模型应用开发里越来越火不是因为多了一个包装大模型的库而是它把 Agent 编排这件事从“写流程脚本”变成了“构建状态图”。你可以像设计一个有向无环图一样设计任务流程让每个 Agent 只负责自己那一段由框架来管理状态传递、条件分支、并行执行、循环终止和持久化记忆。这篇文章我会从多智能体架构设计的角度出发完整拆解 LangGraph 的核心组件然后用三个可以直接运行的代码实战带你从零搭出一个具备条件路由、子图复用、并行分发、Supervisor 主管编排的多智能体系统。最后再给出常见问题和工程化建议。整篇文章不需要你预先有很深的多智能体基础但建议你已经会写简单的 Python并且知道 ChatOpenAI 这类模型接口的基本用法。1. 多智能体开发为什么需要编排框架先看一个真实场景。假设你要做一个内容生成系统需要“选题 Agent”“写作 Agent”“审核 Agent”三个角色协作。如果不用编排框架你会怎么做大概率是手写三个函数A 函数返回结果B 函数接收并加工C 函数再审核。过程中只要有一个环节需要人工确认、或者需要根据上一轮结果动态决定下一步派给谁你的 if-else 就会开始失控。多智能体的价值从来不是“人多力量大”而是“分工明确 上下文隔离 职责可复用”。选题 Agent 只需要维护选题相关的工具和 prompt写作 Agent 只需要关心写作规范审核 Agent 只需要掌握审核标准。每个 Agent 自身的复杂度下降了但 Agent 之间的调度复杂度却上升了。调度复杂度体现在三个地方状态不知道在哪。上一次任务跑到了哪一步A 产生了什么中间结果这些结果怎么传给 B分支逻辑写死在业务代码里。条件一多if-else 嵌套流程根本没法维护。没有统一的观测和恢复机制。出错了不知道在哪一步出错想从中间恢复更是难上加难。LangGraph 解决的就是这三个问题。它允许你把多智能体系统建模为一个图节点是 Agent 或普通函数边是流转规则State 是贯穿全程的共享状态Conditional Edges 是动态分支Checkpointer 是每一步的持久化快照。换句话说LangGraph 让你从“写一段会调用模型的脚本”升级为“设计一个可测试、可控制、可回滚的 Agent 执行引擎”。这是多智能体走向生产环境时最需要补上的一课。2. LangGraph 与 LangChain 的关系和核心概念很多人会把 LangGraph 和 LangChain 混在一起甚至认为 LangGraph 就是 LangChain 的最新版本。实际上两者定位不同关系可以这样理解LangChain 是组件库负责提供模型封装、Prompt 模板、向量库、工具调用这些基础能力。LangGraph 是编排运行时负责把这些组件组织成一个可执行的图管理状态流转和分支控制。如果做一个类比LangChain 像是你采购的“零件”LangGraph 像是把这些零件组装成机器的“流水线控制系统”。你可以只用 LangGraph 不依赖 LangChain 的很多高级封装也可以只使用 LangChain 快速实现 RAG 检索但要让多个 Agent 稳定协作LangGraph 是最直接的解决方案。对比维度LangChainLangGraph定位组件库 / 开发框架编排运行时核心抽象Chain、AgentExecutorStateGraph、State状态管理隐式传递显式 State可持久化流程控制顺序、简单分支条件路由、循环、并行、子图适合场景RAG 管道、快速原型复杂多智能体、生产级工作流接下来LangGraph 的几个核心概念必须理解清楚后面所有代码都会围绕它们展开。State全局共享的状态对象。每个节点都可以读取 State也可以返回一个 dict 来更新 State。Node一个节点就是一个 Python 函数。函数接收当前 State返回部分更新。Edge从一个节点到另一个节点的确定连线。Conditional Edge根据 State 内容动态决定下一个节点是分支控制的灵魂。Checkpointer检查点机制保存每个步骤的执行状态支持断点、恢复、记忆。把这几个概念串起来你定义一个 State 数据结构然后往 StateGraph 里注册多个 Node用 Edge 或 Conditional Edge 把它们连接起来最后编译成一个 Graph 对象。运行阶段你调用graph.invoke()传入初始 State框架会按照图结构自动调度节点执行直到遇到 END。3. 环境准备与安装先搭好环境。本文示例基于 Python 3.10 及以上版本这是 LangGraph 目前比较稳妥的运行环境。建议创建独立虚拟环境避免污染其他项目python -m venv .venv source .venv/bin/activate # Windows 使用 .venv\Scripts\activate安装 LangGraph 及相关依赖pip install -U langgraph langchain-core langchain-openai这里说明一下langchain-openai是 LangChain 对 OpenAI 兼容接口的封装后面调用大模型时用它。如果你用的是本地部署模型如 vLLM 或 Ollama也可以通过base_url参数接入 OpenAI 兼容接口。安装完成后验证环境是否正常python -c from langgraph.graph import StateGraph; print(StateGraph)如果成功打印出类信息说明环境安装成功。大模型的 API Key 建议放在环境变量里不要直接写进代码。示例export OPENAI_API_KEYsk-xxxx在正式代码中我会用os.getenv(OPENAI_API_KEY)来读取。如果你没有 OpenAI Key也可以用本地模型服务把base_url指向本地地址例如from langchain_openai import ChatOpenAI model ChatOpenAI( modelqwen2.5:7b, api_keyEMPTY, base_urlhttp://localhost:8000/v1, )这样就把模型调用替换成本地服务了。不过为了演示流程我会先给出一个不依赖真实大模型的版本确保你把多智能体编排本身跑通。4. LangGraph 核心组件详解4.1 State所有节点共享的“可编程内存”State 是 LangGraph 的灵魂。它本质上是一个 TypedDict 或 Pydantic 模型定义了这次任务运行中所有节点都能访问的数据结构。每个节点函数接收当前 State然后返回一个 dictLangGraph 会自动把返回的字段合并进 State。这里容易踩一个坑如果你返回的是列表字段默认行为是“覆盖”而不是“追加”。需要把列表字段标记为Annotated[List, operator.add]LangGraph 才会用 reducer 做追加合并。后面并行分发实战里会演示。State 的设计原则是只放节点之间需要共享的信息不要把所有中间变量都塞进去。State 越大图越难调试。4.2 Node最普通的 Python 函数Node 没有黑魔法就是一个普通函数。它的输入是 State输出是一个 dict表示对 State 的部分更新。def my_node(state: MyState) - dict: # 读取 state query state[query] # 做一些处理 result f处理结果: {query} # 返回部分更新 return {result: result}函数不需要写副作用不需要自己维护全局变量所有数据都通过 State 流转。这个约束让每个节点都可以独立测试也方便后续加日志和监控。4.3 Edge 和 Conditional Edge从“写死流程”到“动态路由”普通 Edge 表示顺序执行比如A - B - C。Conditional Edge 则定义了一个路由函数函数根据当前 State 的内容返回下一个要执行节点的名字。这是多智能体系统里最重要的设计之一。因为 Agent 系统的特点就是流程不确定到底该让谁先处理取决于用户问题、中间结果、甚至模型判断。条件路由把这种不确定性显式建模出来不再是散落在各个函数里的 if-else。def route_by_category(state: MyState) - str: if state[category] tech: return tech_node return other_node builder.add_conditional_edges(classify, route_by_category)路由函数返回的字符串必须是在图上注册过的节点名或者 END。很多初学者在这里出错返回了一个没注册的名字LangGraph 运行时会直接报错。4.4 Checkpointer让 Agent 拥有“记忆”和“恢复能力”Checkpointer 负责在图的每个节点执行前后保存 State 快照。有了检查点你可以实现中断和恢复一次长任务运行到一半进程挂了重启后从上次检查点继续。对话记忆把每一步的消息列表保留下来作为下一次 invoke 的上下文。多轮交互通过thread_id把不同轮次的对话关联到同一个检查点。最简单的演示级 Checkpointer 是内存版数据只存在于当前进程重启即丢失from langgraph.checkpoint.memory import InMemorySaver saver InMemorySaver() graph builder.compile(checkpointersaver) result graph.invoke( {messages: []}, config{configurable: {thread_id: demo-thread}} )关于“怎么使用 InMemorySaver 中的内容构建传入大模型的上下文”需要理解检查点保存的是完整的 State 快照而不是自动拼接好的 prompt。当你恢复某个 thread 时LangGraph 会从检查点读出状态继续执行图的后续节点。如果你想把这些历史消息喂给大模型只需要从 State 里读取消息列表字段传给模型即可。它解决的是“持久化”问题不是“prompt 拼接”问题。4.5 Tool 集成一切业务能力都可以变成 Agent 的工具在 LangGraph 里一个节点内部可以调用任何东西包括普通函数、API、数据库查询、LangChain 的 Tool、MCP 协议暴露的工具等。最近有个很火的梗是“如何将小龙虾或者爱马仕集成到多智能体系统中”。其实剥开来看答案很简单在多智能体系统中任何能被 Python 函数包装的外部能力都可以成为 Agent 的工具。比如你有一个“小龙虾养殖池塘环境监测”函数把它用tool装饰一下注册给对应的 Agent这个 Agent 就能调度它。爱马仕如果指的是商品库存查询服务也是同样的接入思路。from langchain_core.tools import tool def query_pond(pond_id: str) - str: # 实际项目中这里会读数据库或传感器 return f池塘 {pond_id}: 水温 26℃溶氧 5.2mg/L状态正常 tool def shrimp_pond_tool(pond_id: str) - str: 查询小龙虾养殖池塘的实时环境数据。 return query_pond(pond_id)MCP 在这个体系里扮演的是“工具协议标准化”的角色。工具提供方用一个统一的 MCP Server 暴露能力Agent 端通过 MCP Client 发现和调用工具。LangGraph 不限制工具来源你可以直接绑定 LangChain Tool也可以接入自定义工具函数还可以集成 MCP 工具。5. 多智能体架构设计与四种常见交互模式多智能体的架构设计没有唯一的官方标准但业界常见的归纳方式可以把多智能体交互模式分为四类分别是中心化编排、流水线、并行分发、层次化。5.1 中心化编排模式Supervisor / Orchestrator-Worker一个主管 Agent 负责理解用户问题决定调哪个专家 Agent然后根据专家输出继续决策直到认为任务完成。这是生产中最常用、最稳妥的模式适合有明确任务划分的场景。用户输入 - Supervisor - Worker A、Worker B、Worker C - Supervisor - 最终输出LLM 主管是最典型的形式。Supervisor 本身也可以是一个不调用大模型的路由决策器直接根据规则选择专家。这个模式可控性强但主管 Agent 容易成为瓶颈因为所有消息都经过它。5.2 流水线模式Pipeline固定顺序执行前一个 Agent 的输出作为后一个 Agent 的输入。适合数据加工、多阶段生成的场景比如“内容选题 - 内容写作 - 内容审核”。优点是流程简单清晰缺点是扩展性和适应性差流程一旦变化就要改图结构。5.3 并行分发模式Fan-out / Parallel一个任务拆分成多个子任务同时分发给多个 worker 执行最后汇总结果。适合多路召回、多角色评审、批量数据处理。LangGraph 里用SendAPI 实现扇出用带 reducer 的字段实现结果聚合。5.4 层次化模式Hierarchical多级 Supervisor 嵌套每个主管管理一组专家 Agent上层主管只和下层主管通信。适合组织复杂、专家数量多的项目。优点是扩展性强缺点是通信成本和实现复杂度高。模式核心特征优点缺点典型场景中心化编排主管统一调度可控性强、好理解主管可能成为瓶颈客服、代码生成、内容创作流水线固定顺序执行简单直观不灵活内容加工、多阶段质检并行分发子任务同时执行吞吐高需要聚合逻辑多路检索、批量审核层次化多级主管嵌套扩展性强实现复杂大型组织流程、复杂业务选型建议刚开始做多智能体优先从中心化编排模式开始。先用一个 Supervisor 两个专家跑通流程再根据业务需要增加并行分发和子图复用。不要一上来就做层次化否则状态管理和调试难度会直线上升。6. 代码实战一条件路由与分支控制第一个实战我们不依赖真实大模型用规则做分类目标是把条件路由流程彻底跑通。这也回答了开发中最常用的需求用户请求进来后如何动态分配给不同的专家节点。场景根据用户 query 中的关键字把请求分流到编程专家、文案专家、兜底处理三种节点。# demo_router.py from typing import TypedDict, Literal from langgraph.graph import StateGraph, START, END class RouterState(TypedDict): query: str category: str answer: str def classify(state: RouterState) - dict: 简单规则分类。实际项目中这里可以换成 LLM 分类。 q state.get(query, ) if any(k in q for k in [bug, 代码, 程序, 报错]): category programmer elif any(k in q for k in [标题, 文案, 文章, 推广]): category copywriter else: category unknown return {category: category} def route_after_classify(state: RouterState) - Literal[programmer, copywriter, unknown]: return state[category] def programmer_node(state: RouterState) - dict: return {answer: f[编程专家] 收到任务{state[query]}} def copywriter_node(state: RouterState) - dict: return {answer: f[文案专家] 收到任务{state[query]}} def unknown_node(state: RouterState) - dict: return {answer: f[兜底处理] 无法分类的任务{state[query]}} builder StateGraph(RouterState) builder.add_node(classify, classify) builder.add_node(programmer, programmer_node) builder.add_node(copywriter, copywriter_node) builder.add_node(unknown, unknown_node) builder.add_edge(START, classify) builder.add_conditional_edges(classify, route_after_classify) builder.add_edge(programmer, END) builder.add_edge(copywriter, END) builder.add_edge(unknown, END) graph builder.compile() if __name__ __main__: for q in [帮我修一个 bug, 写一条推广文案, 明天天气如何]: result graph.invoke({query: q}) print(fQ: {q}) print(fA: {result[answer]}) print(- * 40)这段代码的核心在add_conditional_edges。它的第一个参数是源节点第二个参数是路由函数。路由函数接收当前 State返回目标节点的名字。你必须保证返回值是已经通过add_node注册过的节点名或者是 END否则运行时直接报“未知节点”错误。运行python demo_router.py预期输出Q: 帮我修一个 bug A: [编程专家] 收到任务帮我修一个 bug ---------------------------------------- Q: 写一条推广文案 A: [文案专家] 收到任务写一条推广文案 ---------------------------------------- Q: 明天天气如何 A: [兜底处理] 无法分类的任务明天天气如何 ----------------------------------------条件路由的本质不是“大模型在做判断”而是“基于状态执行一个决策函数”。决策函数里可以放规则也可以调用大模型还可以组合外部服务。关键点是决策逻辑必须独立、可测试这样整个图的流程才是可预测的。这里还要提一下循环控制。假设路由函数不小心把某个节点指回起点图会一直循环。LangGraph 默认有递归深度限制超出后会抛错。你可以通过graph.invoke(..., config{recursion_limit: 25})调整限制但在生产环境更推荐直接设计好终止条件不要依赖递归上限兜底。7. 代码实战二子图与并行分支7.1 子图把完整流程当做一个节点复用当你发现同一个 Agent 流程在多处都要使用时最直接的办法是把这段流程定义成子图然后在主图里作为一个节点使用。子图的输入输出通过主图节点的返回值做显式映射。下面示例把第 6 节的路由图当作子图嵌入一个新的主图# subgraph_demo.py from typing import TypedDict from langgraph.graph import StateGraph, START, END from demo_router import graph as router_graph class MainState(TypedDict): raw_query: str result: str def call_router_inner(state: MainState) - dict: # 调用子图传入子图需要的字段 sub_result router_graph.invoke({query: state[raw_query]}) # 把子图输出映射到主图 State return {result: sub_result[answer]} main_builder StateGraph(MainState) main_builder.add_node(call_router, call_router_inner) main_builder.add_edge(START, call_router) main_builder.add_edge(call_router, END) main_graph main_builder.compile() if __name__ __main__: out main_graph.invoke({raw_query: 帮我修一个 bug}) print(out[result])子图的优势是复用和解耦。如果你的团队已经沉淀了一个“代码审查 Agent”子图其他流程只要把这个子图注册成节点再处理好字段映射即可不需要知道子图内部是如何实现的。7.2 使用 Send 做并行分支并行分发是提高多智能体吞吐量的核心手段。LangGraph 的SendAPI 允许把一个节点扇出到多个相同节点的并行执行。比如批量对多个任务并行处理并汇总结果# parallel_demo.py import operator from typing import Annotated, TypedDict, List from langgraph.graph import StateGraph, START, END from langgraph.constants import Send class EvalState(TypedDict): tasks: List[str] results: Annotated[List[str], operator.add] def passthrough(state: EvalState) - dict: # 不需要做实际处理这个节点只是扇出起点 return {} def fan_out(state: EvalState) - List[Send]: # 给每个 task 发一个 worker 实例 return [Send(worker, {task: t}) for t in state[tasks]] def worker(state: dict) - dict: # 注意这里接收的是 Send 分发进来的单独 task return {results: [f已完成: {state[task]}]} eval_builder StateGraph(EvalState) eval_builder.add_node(fan_out, passthrough) eval_builder.add_node(worker, worker) eval_builder.add_edge(START, fan_out) eval_builder.add_conditional_edges(fan_out, fan_out) eval_builder.add_edge(worker, END) eval_graph eval_builder.compile() if __name__ __main__: out eval_graph.invoke({tasks: [需求分析, 代码编写, 测试执行]}) for r in out[results]: print(r)这里最关键的是results: Annotated[List[str], operator.add]。没有这个 reducer多个 worker 返回的结果会互相覆盖最后只剩最后一个结果。加了 reducer 之后所有 worker 返回的列表会被自动拼接起来。运行python parallel_demo.py预期输出已完成: 需求分析 已完成: 代码编写 已完成: 测试执行并行分支最常见的报错就是“为什么我只拿到了一个 worker 的结果”十有八九是 State 里聚合字段没有配置 reducer。7.3 循环检测的实践提醒LangGraph 支持循环图比如 Supervisor 模式里主管节点和专家节点之间就是循环边。循环本身不是问题但循环必须有一个明确的终止条件。推荐做法在 Supervisor 节点里显式判断“是否已经有专家回复过”有则输出 END。给图设置一个合理的recursion_limit防止异常情况下的死循环。在关键节点加日志观察每次循环时 State 的变化。8. 代码实战三Supervisor 多智能体编排完整实战第三个实战我们实现一个“主管 专家”的多智能体系统。用户输入问题Supervisor 先判断要交给哪个专家专家处理完成后回到 SupervisorSupervisor 再决定是继续派单还是结束。为了稳定运行我加了一个简单的终止条件只要消息里已经出现专家回复Supervisor 直接输出 END。# supervisor_demo.py import os from typing import TypedDict from langgraph.graph import StateGraph, START, END from langchain_openai import ChatOpenAI class AgentState(TypedDict): messages: list next: str class SupervisorDecision(BaseModel): next: str Field(descriptioncode_expert / copy_expert / END)先等等这里需要导入 pydantic 的 BaseModel 和 Field。在文件顶部补充from pydantic import BaseModel, Field完整代码如下# supervisor_demo.py import os from typing import TypedDict from pydantic import BaseModel, Field from langgraph.graph import StateGraph, START, END from langchain_openai import ChatOpenAI class AgentState(TypedDict): messages: list next: str class SupervisorDecision(BaseModel): next: str Field(descriptioncode_expert / copy_expert / END) def supervisor_node(state: AgentState) - dict: # 如果已经有过专家回复就直接结束 for m in state[messages]: if 专家] in m.get(content, ): return {next: END} prompt ( 你是多智能体主管。根据用户最新消息判断下一步派给谁 code_expert 处理编程问题copy_expert 处理文案问题。 如果判断任务已完成则输出 END。 ) model ChatOpenAI( modelgpt-4o-mini, api_keyos.getenv(OPENAI_API_KEY), ) structured_model model.with_structured_output(SupervisorDecision) decision structured_model.invoke( [{role: system, content: prompt}] state[messages] ) return {next: decision.next} def route_after_supervisor(state: AgentState) - str: return state.get(next, END) def code_expert_node(state: AgentState) - dict: answer [代码专家] 已根据需求生成 Python 代码请查收。 new_messages state[messages] [{role: assistant, content: answer}] return {messages: new_messages} def copy_expert_node(state: AgentState) - dict: answer [文案专家] 已输出一版推广文案请查收。 new_messages state[messages] [{role: assistant, content: answer}] return {messages: new_messages} builder StateGraph(AgentState) builder.add_node(supervisor, supervisor_node) builder.add_node(code_expert, code_expert_node) builder.add_node(copy_expert, copy_expert_node) builder.add_edge(START, supervisor) builder.add_conditional_edges( supervisor, route_after_supervisor, { code_expert: code_expert, copy_expert: copy_expert, END: END, }, ) builder.add_edge(code_expert, supervisor) builder.add_edge(copy_expert, supervisor) graph builder.compile() if __name__ __main__: messages [{role: user, content: 请帮我写一段 Python 排序代码}] result graph.invoke({messages: messages, next: }) for m in result[messages]: print(f{m[role]}: {m[content]})如果你没有配置OPENAI_API_KEY这个示例会卡在模型调用上。可以把 Supervisor 决策逻辑临时换成规则分类用bug或代码之类的关键字做路由这样没有 Key 也能跑通整个循环图结构。实际项目中Supervisor 不只是做路由它还需要承担上下文汇总、任务拆解、质量检查等职责。它的 prompt 设计应该明确说明有哪些专家 Agent 可用。每个专家 Agent 擅长什么。什么情况下判定任务已经完成。如果专家 Agent 返回结果不满足要求是否要重新派单。这才是 Supervisor 模式真正要解决的工程问题。9. 运行结果验证与常见问题排查运行本文所有示例按顺序执行python demo_router.py python subgraph_demo.py python parallel_demo.py python supervisor_demo.py前三个示例不依赖外部模型服务正常情况可以直接看到预期输出。supervisor_demo.py需要先配置好模型服务地址和 Key否则会超时或报认证错误。在生产或调试过程中最常遇到的问题集中在以下几个方面问题现象可能原因排查方式解决方案报错找不到MemorySaver/InMemorySaverLangGraph 版本不同类名或导入路径不同执行pip show langgraph查看版本根据版本选择langgraph.checkpoint.memory.InMemorySaver老版本可能是MemorySaver条件路由报“未知节点”路由函数返回了未注册的节点名打印路由函数的