1. 为什么现在还要回头看LangGraph多智能体并非锦上添花过去一两年里大模型应用从单轮问答快速演进到复杂任务编排单纯依赖一个LLM调用已经很难覆盖真实业务场景。比如你要做一个企业级的智能客服它既需要根据用户问题检索内部知识库又需要查订单状态、算价格、写工单这些能力如果都塞进一个agent里提示词会越写越臃肿函数调用列表动辄几十个模型开始频繁选错工具错误也很难定位。多智能体架构就是在这种背景下被推到前台的。它的核心思路不是让一个万能agent干所有事而是拆成多个各司其职的智能体每个智能体专注于某一类能力再由一个协调层比如Supervisor负责调度。这样每个agent的提示词都能保持精简、职责边界清晰、出现问题也能单独调试。而LangGraph恰恰是目前落地多智能体系统时我见过的最顺手的编排框架。它跟LangChain最大的区别在于LangChain的Chain是线性管道数据从一头进去、另一头出来中途很难绕回去重新处理LangGraph则把应用建模成一张有向图节点是处理逻辑边是流转关系天然支持循环、分支、状态持久化。用大白话说LangChain像是流水线LangGraph像是带回路的生产车间可以让任务回溯、分派、合并。这篇文章不会泛泛讲概念我会从架构设计、核心组件、真实可跑的代码、以及我在实际项目中踩过的坑这四个层面把LangGraph多智能体从原理到实战完整走一遍。无论你是初次接触多智能体还是已经在LangChain里写过Agent但总觉得别扭这篇的内容都能直接拿来参考。2. 多智能体架构的整体设计思路2.1 三种主流协作模式先想清楚再动手多智能体系统听起来高大上但落到架构层面其实就三种经典模式绝大多数业务场景都能套进去。第一种是Supervisor模式也是最常用的一种。系统里有一个主管智能体它负责理解用户需求、拆解任务然后分发给若干个专门的Worker智能体。Worker执行完把结果回报给Supervisor由主管决定是继续调用其他Worker、还是直接结束。这种模式的优点是控制集中、职责清晰适合任务类型比较多样、需要统一决策的场景比如一个AI内容运营系统主管收到写一篇新品推文的需求后会先调用市场分析智能体出话题方向再调用文案智能体写初稿最后调用编辑智能体做校验。第二种是协作模式多个智能体之间没有明显的上下级关系而是共享一个工作区各自负责一部分任务通过共同维护的状态来完成接力。这种模式适合任务可以被明确切分、每个智能体的产出是下一个智能体输入的流水线场景。第三种是层级模式在Supervisor之上再套一层Supervisor形成两级甚至多级管理。这种模式只有在智能体数量非常多、一个主管管不过来时才需要日常业务里能用Supervisor解决的不要硬上层级否则状态数据会变得极其复杂调试成本成倍上涨。我在实际项目里感受最深的一点多智能体方案的选择一定要先于代码编写否则写着写着就会发现这个流程用单agent加几个工具也能做为什么非要拆成多个agent。拆分的核心标准是有没有独立的状态生命周期和独立的工具集。如果两个任务共享同一套工具、同一段上下文那它们应该是一个agent内部的两步只有当一个任务需要完全不同的提示词、不同的可用工具、甚至不同的模型时才值得拆成独立智能体。2.2 为什么选LangGraph做编排层市面上的多智能体框架不少AutoGen、MetaGPT、AgentScope各有特点。但LangGraph有一个非常突出的优势它的编排逻辑足够底层给了开发者最大的可控性。很多框架把多智能体做成了开箱即用的高层抽象你只需要配置几个参数就能跑一个群聊式多智能体。听起来爽但一旦模型行为不符合预期你想在某个节点中间插入一个人工审核步骤、想在某条执行路径上增加重试逻辑、想精确控制状态里每个字段的读写权限就会发现高层抽象根本拦不住你改不了。LangGraph走的是另一条路。它只给你提供图和状态这两个最核心的抽象节点和边完全由你自己定义。你写一个Python函数它就是节点你决定它连向哪里它就是边。这种设计在初期确实比高级框架多写几行代码但它把所有控制权都还给了开发者。上面说的那些复杂逻辑——人工介入、条件分支、循环重试、状态回滚——在图模型里都变成了顺理成章的事。另外一点很实际LangGraph和LangChain生态无缝衔接。如果你已经在用LangChain的LCEL表达式、Tool类、ChatModel接口迁移到LangGraph几乎零成本。我见过很多人纠结LangChain和LangGraph到底学哪个其实它们不是替代关系LangChain提供组件库LangGraph提供编排框架两者组合才是完整应用。3. 核心组件拆解把LangGraph的底层逻辑吃透3.1 State一切流转的核心状态定义决定了系统复杂度在LangGraph中整个图的运行过程就是状态不断流转的过程。状态定义通常用TypedDict或Pydantic模型描述图中流动的数据结构例如消息列表、当前执行步骤、中间结果等。from typing import TypedDict, Annotated, List def merge_messages(left: List[dict], right: List[dict]) - List[dict]: return left right class AgentState(TypedDict): messages: Annotated[List[dict], merge_messages] next_agent: str intermediate_results: dict这里的Annotated标注了消息字段的归并函数这个细节非常关键。默认情况下LangGraph的节点返回更新时会覆盖原有字段值但通过归并函数可以自定义合并逻辑。messages用归并函数做累加才能让每个节点都能看到完整的对话历史intermediate_results用覆盖逻辑则保证每次只保留最新结果。从实际使用角度我强烈建议把状态设计得克制一些。状态字段越多图的可解释性越差排查问题时越难定位是哪个节点改了哪个字段。好的状态设计是messages只存模型交互记录业务上下文单独放一个字段每个节点的输入输出尽量明确。3.2 Node和Edge图的基本单元函数就是节点LangGraph里节点就是一个普通Python函数接收当前状态返回状态的部分更新。这个设计非常朴素却意味着你所有的业务逻辑都可以原封不动放进节点里不需要学习任何新的代码范式。def research_node(state: AgentState): query state[messages][-1][content] result search_tool.run(query) return {intermediate_results: {research: result}}这种节点函数有两个注意点。一是函数尽量要有明确的输入输出契约状态里写什么、返回什么更新应该清晰固定不能今天返回一个格式、明天换另一种二是节点内部不要做过于重的外部调用而不设超时否则整个图的执行时间会被一个异常节点拖垮。边的定义有两种。普通边固定连接两个节点条件边根据当前状态动态决定下一跳。条件边是实现多智能体路由的核心机制执行流程中的判断题全靠它完成。def route_decision(state: AgentState) - str: if state[next_agent] coder: return coder if state[next_agent] reviewer: return reviewer return __end__ builder.add_conditional_edges(supervisor, route_decision, {coder: coder, reviewer: reviewer, __end__: END})3.3 Conditional Edge与Checkpointer分支和记忆是杀手锏条件边的函数很纯粹输入状态返回一个字符串这个字符串对应某个下游节点。它让图具备了动态决策能力比如Supervisor节点让LLM输出一个JSON指明下一个行动者是谁条件边读取这个JSON决定走向。Checkpointer是LangGraph另一个容易被低估的能力。它负责把每一步的状态持久化到存储后端默认有内存版MemorySaver也有SQLite和Postgres这类持久化实现。Checkpointer解决的痛点是多智能体应用运行到一半如果某一步出错能否从最近的检查点恢复而不是从头再来以及用户关闭会话后系统能否记住之前的执行状态。from langgraph.checkpoint.memory import MemorySaver checkpointer MemorySaver() app builder.compile(checkpointercheckpointer)这里要提醒一点MemorySaver只适合开发调试重启进程状态就没了。生产环境至少要上SQLite数据量大了再考虑用外部存储。3.4 Tool与模型绑定多智能体能力的边界来源有了图结构每个节点内部还需要真正干活的工具和模型。在LangGraph里通常的做法是把工具列表与模型绑定让模型在节点内自动决定调用哪个工具。from langchain_openai import ChatOpenAI from langchain_core.tools import tool tool def search_knowledge_base(query: str) - str: 检索内部知识库 return 知识库中与{}相关的内容.format(query) llm ChatOpenAI(modelgpt-4o, temperature0) llm_with_tools llm.bind_tools([search_knowledge_base])在多智能体架构下每个Worker节点绑定自己的工具集而不是所有工具一股脑绑定给一个模型。工具少的模型在函数调用上明显更准确这也是拆分多智能体能提升整体稳定性的一个直接原因。4. 代码实战从零做一个可运行的Supervisor多智能体4.1 环境安装与前置准备先安装依赖建议在Python 3.11的虚拟环境里操作pip install langgraph langchain-openai这里有个版本提醒LangGraph的API迭代速度很快我写这篇文章时常用的是0.2.x和0.3.x版本如果你的版本更新个别导入路径可能变了以官方文档为准。遇到ModuleNotFoundError时不要慌优先去官方迁移文档里查导入路径。在环境层面我还建议安装langchain-community里面包含一些额外的工具实现虽然多智能体核心用不上但做节点内部的工具组合时可能用到。4.2 定义一个基础的多智能体状态图我们从最简结构开始先实现一个主管分派、两个工人干活的图不做太复杂的归并逻辑保证能跑通。from typing import TypedDict, Annotated, List from langgraph.graph import StateGraph, END from langgraph.checkpoint.memory import MemorySaver class SimpleState(TypedDict): messages: Annotated[List[dict], lambda l, r: l r] output: str def worker_a(state: SimpleState): return {output: A完成检索结果猫是猫科动物。} def worker_b(state: SimpleState): return {output: B完成总结猫属于猫科动物擅长捕猎。} def supervisor(state: SimpleState): # 实际项目中这里会让LLM做决策我们先写死路由逻辑 return {output: 分派给A和B} builder StateGraph(SimpleState) builder.add_node(supervisor, supervisor) builder.add_node(worker_a, worker_a) builder.add_node(worker_b, worker_b) builder.set_entry_point(supervisor) builder.add_edge(supervisor, worker_a) builder.add_edge(supervisor, worker_b) builder.add_edge(worker_a, END) builder.add_edge(worker_b, END) app builder.compile()跑这个图会发现一个很有意思的现象从supervisor出发有两条边分别连到worker_a和worker_b实际执行时LangGraph会并行执行这两个节点最终状态会被合并。这在多智能体协作里非常实用意味着你可以让多个Worker同时干活再汇总结果。4.3 实现条件路由让LLM做主分配上面的示例路由是写死的真实系统里通常要让Supervisor根据用户输入动态决定调用哪个Worker。这一步就要用到条件边。import json from typing import TypedDict, Annotated, List from langgraph.graph import StateGraph, END class AgentState(TypedDict): messages: Annotated[List[dict], lambda l, r: l r] next: str def supervisor_node(state: AgentState): prompt 你是任务主管。请判断用户需求应该交给哪个工作智能体。 用户需求{last_msg} 可选智能体researcher检索信息、calculator数学计算、finisher收尾总结 只输出一个JSON格式{{next: 智能体名称}} .format(last_msgstate[messages][-1][content]) response llm.invoke(prompt) parsed json.loads(response.content) return {next: parsed[next]} def researcher_node(state: AgentState): return {messages: [{role: assistant, content: 已完成资料检索}]} def calculator_node(state: AgentState): return {messages: [{role: assistant, content: 计算结果为42}]} def finisher_node(state: AgentState): return {messages: [{role: assistant, content: 已汇总所有结果生成最终答案。}]} def route_next(state: AgentState) - str: if state.get(next) researcher: return researcher elif state.get(next) calculator: return calculator elif state.get(next) finisher: return finisher else: return END builder StateGraph(AgentState) builder.add_node(supervisor, supervisor_node) builder.add_node(researcher, researcher_node) builder.add_node(calculator, calculator_node) builder.add_node(finisher, finisher_node) builder.set_entry_point(supervisor) builder.add_conditional_edges(supervisor, route_next, { researcher: researcher, calculator: calculator, finisher: finisher, END: END }) builder.add_edge(researcher, supervisor) builder.add_edge(calculator, supervisor) builder.add_edge(finisher, END) app builder.compile()注意这里我把worker执行完后的边又指回了supervisor形成循环。这是LangGraph区别于LangChain的关键能力任务不是一条直线跑完而是可以反复进入主管节点做下一轮判断直到finisher收尾为止。4.4 带记忆能力的完整实战持久化会话状态开发时我们直接编译就能跑但生产环境还需要能在两次会话之间保留历史这就用到了Checkpointer。创建一个带线程ID的会话配置每次运行时传入同一个thread_id图就会在该线程下累积状态相当于给多智能体系统装上了内存。from langgraph.checkpoint.memory import MemorySaver config {configurable: {thread_id: user_001}} app builder.compile(checkpointerMemorySaver()) # 第一次运行 result1 app.invoke( {messages: [{role: user, content: 帮我查一下猫科动物的特征}]}, config ) # 第二次运行模型能看到上一次的对话历史 result2 app.invoke( {messages: [{role: user, content: 那狗呢}]}, config )这个能力在实际落地中太关键了。多智能体每次执行之间如果可以保留上下文用户体验质变同时也让排查追踪有了线索。4.5 更多开箱即用的高层APILangGraph预构建Agent如果不想从零搭图LangGraph也封装了预构建的Agent结构比如create_react_agent它内部已经把工具调用、循环、状态更新都封装好了适合快速验证想法from langgraph.prebuilt import create_react_agent agent_executor create_react_agent( modelllm, tools[search_knowledge_base], state_modifier你是一个只能使用工具的智能助手。, )我的使用建议是 prototyping阶段用create_react_agent快速验证生产阶段还是自己用StateGraph搭因为预构建Agent的自定义空间有限有些边界情况比如某个节点要做额外校验、某条路径要跳过工具直接答复需要手动控制。5. 实操中的常见问题与排查经验这部分是我最想分享的全是实际踩坑记录。5.1 状态字段更新失效原因多半是归并函数缺失很多人刚开始用Annotated时习惯性地用lambda合并字段结果节点返回新消息时旧消息全没了或新消息无法累加。原因在于如果你定义的状态字段没有归并函数LangGraph默认采用覆盖更新一旦节点返回值是空列表就会把整段历史清空。我的建议是消息类字段一定写显式归并函数业务临时字段用覆盖策略并确保每个节点明确返回。排查时先打印节点返回值和图执行后的state多智能体系统出问题时第一步永远是看状态在哪个节点发生了预期外的变化。5.2 多智能体循环变死循环条件边必须兜底多智能体系统一旦引入循环最怕的是图永远跳不出循环。我遇到过一种典型场景Supervisor节点让LLM选下一个Worker模型反复输出同一个Worker名字执行路径就在这个Worker和Supervisor之间来回转直到达到递归上限报错。解决办法有两个。一是给条件路由函数加终止条件比如记录执行步数超过N步强制跳到END二是在LLM决策的提示词里强调可选的收尾选项比如如果任务已经完成直接输出finisher。经验是两条都做提示词做软约束、代码做硬兜底双保险。5.3 Checkpointer恢复时状态错乱注意线程隔离用Checkpointer后如果多个用户共用同一个thread_id状态会互相污染。生产环境必须保证每个会话上下文使用独立的thread_id一般直接用用户会话ID或请求ID。另一个常见坑是thread_id相同但图结构定义变了比如新增节点恢复老线程时可能报错。这种情况要么做数据迁移要么让旧线程直接失效重新开始我在项目里通常选择后者多智能体系统每次升级后最好不要依赖旧状态。5.4 多智能体并发执行的结果冲突当多个Worker并行执行后合并状态时如果两个节点同时更新同一个字段最终值取决于执行顺序可能不符合预期。解决思路是并发节点尽量写入不同的状态字段或者提供明确的归并函数处理冲突。比如多个Worker都产生理由用列表归并就比覆盖稳妥得多。这里做一个排查速查表方便快速定位现象可能原因排查方向节点未执行流程直接跳过条件边返回了错误目标打印路由函数的返回值状态里消息被清空字段覆盖更新或归并函数问题检查Annotated归并逻辑循环执行次数异常LLM路由决策未终止增加最大步数硬限制状态恢复后内容错乱thread_id冲突或图结构变化检查线程隔离与版本兼容模型工具调用频繁失效绑定工具太多或提示词冲突精简工具集、拆分Worker5.5 性能与成本控制经验多智能体系统的每次任务执行往往会调用多次LLM成本比单agent高不少。实测下来Supervisor每轮调度就是一次LLM调用Worker执行又是一次如果任务复杂循环四五轮一次用户请求就可能烧掉十几次模型调用。建议在节点函数里加缓存相同输入直接返回历史结果另外对不关键的任务用便宜的小模型做路由决策只有真正需要专业能力时才调用强模型。6. 进阶技巧与个人实践心得6.1 让多智能体系统保持稳定的三个小技巧第一每个智能体的系统提示词开头必须声明自己的职责边界明确你只负责X遇到Y任务必须说无法处理。我见过太多失败的多智能体案例问题不在架构而在子智能体缺乏边界感动不动越俎代庖。第二节点函数里一定要加异常捕获和降级逻辑。用一个try-except把工具调用包起来工具崩了就在态里记录错误、返回给上层而不是让整个图执行抛异常。多智能体系统越复杂单个节点的韧性就越重要。第三运行日志里带上thread_id和节点名。排查分布式问题不能靠print推测建议在进入和离开节点时输出结构化日志记录状态摘要。LangGraph自带的LangSmith服务可以做完整追踪个人开发者也可以用最简单的方式在每个节点函数里打一个包含状态关键字的日志。6.2 从LangGraph还能延伸出什么玩法LangGraph多智能体框架值得深入挖掘的方向不少。一个是人机协同审批流在某个节点前插入human-in-the-loop逻辑让图运行到关键节点时暂停等待人工确认LangGraph原生支持interrupt机制这能覆盖大量企业级需求。另一个是长时运行任务结合持久化存储让一个多智能体协作任务运行几小时甚至几天中途系统重启也能恢复。最后是事件驱动模式让多智能体架构从请求-响应变成事件驱动。配合消息队列使用业务系统产生事件时再唤醒相应的智能体节点整个系统的异步能力和可扩展性又会提升一个台阶。6.3 踩过几次坑之后的真心话如果你第一次接触LangGraph不要急着堆一个复杂的多智能体系统。先用最简单的两个节点把一个闭环跑通然后逐步增加条件分支、循环、状态持久化最后才考虑多智能体协作。我见过太多人一上来就模仿开源项目搭了一个五六层的图结果连路由都调不通。多智能体的复杂度是指数级上升的每一层抽象都需要你的设计足够清晰而不是靠堆功能掩盖思路混乱。根据我个人经验LangGraph的学习曲线比LangChain高一点但一旦跨过去你设计的系统会从提示词工程式的单次调用真正转变为可编排、可恢复、可观察的复杂应用。这才是多智能体技术带来真正的生产力提升所在。 LangGraph多智能体架构实战：从状态图到条件边与持久化
1. 为什么现在还要回头看LangGraph多智能体并非锦上添花过去一两年里大模型应用从单轮问答快速演进到复杂任务编排单纯依赖一个LLM调用已经很难覆盖真实业务场景。比如你要做一个企业级的智能客服它既需要根据用户问题检索内部知识库又需要查订单状态、算价格、写工单这些能力如果都塞进一个agent里提示词会越写越臃肿函数调用列表动辄几十个模型开始频繁选错工具错误也很难定位。多智能体架构就是在这种背景下被推到前台的。它的核心思路不是让一个万能agent干所有事而是拆成多个各司其职的智能体每个智能体专注于某一类能力再由一个协调层比如Supervisor负责调度。这样每个agent的提示词都能保持精简、职责边界清晰、出现问题也能单独调试。而LangGraph恰恰是目前落地多智能体系统时我见过的最顺手的编排框架。它跟LangChain最大的区别在于LangChain的Chain是线性管道数据从一头进去、另一头出来中途很难绕回去重新处理LangGraph则把应用建模成一张有向图节点是处理逻辑边是流转关系天然支持循环、分支、状态持久化。用大白话说LangChain像是流水线LangGraph像是带回路的生产车间可以让任务回溯、分派、合并。这篇文章不会泛泛讲概念我会从架构设计、核心组件、真实可跑的代码、以及我在实际项目中踩过的坑这四个层面把LangGraph多智能体从原理到实战完整走一遍。无论你是初次接触多智能体还是已经在LangChain里写过Agent但总觉得别扭这篇的内容都能直接拿来参考。2. 多智能体架构的整体设计思路2.1 三种主流协作模式先想清楚再动手多智能体系统听起来高大上但落到架构层面其实就三种经典模式绝大多数业务场景都能套进去。第一种是Supervisor模式也是最常用的一种。系统里有一个主管智能体它负责理解用户需求、拆解任务然后分发给若干个专门的Worker智能体。Worker执行完把结果回报给Supervisor由主管决定是继续调用其他Worker、还是直接结束。这种模式的优点是控制集中、职责清晰适合任务类型比较多样、需要统一决策的场景比如一个AI内容运营系统主管收到写一篇新品推文的需求后会先调用市场分析智能体出话题方向再调用文案智能体写初稿最后调用编辑智能体做校验。第二种是协作模式多个智能体之间没有明显的上下级关系而是共享一个工作区各自负责一部分任务通过共同维护的状态来完成接力。这种模式适合任务可以被明确切分、每个智能体的产出是下一个智能体输入的流水线场景。第三种是层级模式在Supervisor之上再套一层Supervisor形成两级甚至多级管理。这种模式只有在智能体数量非常多、一个主管管不过来时才需要日常业务里能用Supervisor解决的不要硬上层级否则状态数据会变得极其复杂调试成本成倍上涨。我在实际项目里感受最深的一点多智能体方案的选择一定要先于代码编写否则写着写着就会发现这个流程用单agent加几个工具也能做为什么非要拆成多个agent。拆分的核心标准是有没有独立的状态生命周期和独立的工具集。如果两个任务共享同一套工具、同一段上下文那它们应该是一个agent内部的两步只有当一个任务需要完全不同的提示词、不同的可用工具、甚至不同的模型时才值得拆成独立智能体。2.2 为什么选LangGraph做编排层市面上的多智能体框架不少AutoGen、MetaGPT、AgentScope各有特点。但LangGraph有一个非常突出的优势它的编排逻辑足够底层给了开发者最大的可控性。很多框架把多智能体做成了开箱即用的高层抽象你只需要配置几个参数就能跑一个群聊式多智能体。听起来爽但一旦模型行为不符合预期你想在某个节点中间插入一个人工审核步骤、想在某条执行路径上增加重试逻辑、想精确控制状态里每个字段的读写权限就会发现高层抽象根本拦不住你改不了。LangGraph走的是另一条路。它只给你提供图和状态这两个最核心的抽象节点和边完全由你自己定义。你写一个Python函数它就是节点你决定它连向哪里它就是边。这种设计在初期确实比高级框架多写几行代码但它把所有控制权都还给了开发者。上面说的那些复杂逻辑——人工介入、条件分支、循环重试、状态回滚——在图模型里都变成了顺理成章的事。另外一点很实际LangGraph和LangChain生态无缝衔接。如果你已经在用LangChain的LCEL表达式、Tool类、ChatModel接口迁移到LangGraph几乎零成本。我见过很多人纠结LangChain和LangGraph到底学哪个其实它们不是替代关系LangChain提供组件库LangGraph提供编排框架两者组合才是完整应用。3. 核心组件拆解把LangGraph的底层逻辑吃透3.1 State一切流转的核心状态定义决定了系统复杂度在LangGraph中整个图的运行过程就是状态不断流转的过程。状态定义通常用TypedDict或Pydantic模型描述图中流动的数据结构例如消息列表、当前执行步骤、中间结果等。from typing import TypedDict, Annotated, List def merge_messages(left: List[dict], right: List[dict]) - List[dict]: return left right class AgentState(TypedDict): messages: Annotated[List[dict], merge_messages] next_agent: str intermediate_results: dict这里的Annotated标注了消息字段的归并函数这个细节非常关键。默认情况下LangGraph的节点返回更新时会覆盖原有字段值但通过归并函数可以自定义合并逻辑。messages用归并函数做累加才能让每个节点都能看到完整的对话历史intermediate_results用覆盖逻辑则保证每次只保留最新结果。从实际使用角度我强烈建议把状态设计得克制一些。状态字段越多图的可解释性越差排查问题时越难定位是哪个节点改了哪个字段。好的状态设计是messages只存模型交互记录业务上下文单独放一个字段每个节点的输入输出尽量明确。3.2 Node和Edge图的基本单元函数就是节点LangGraph里节点就是一个普通Python函数接收当前状态返回状态的部分更新。这个设计非常朴素却意味着你所有的业务逻辑都可以原封不动放进节点里不需要学习任何新的代码范式。def research_node(state: AgentState): query state[messages][-1][content] result search_tool.run(query) return {intermediate_results: {research: result}}这种节点函数有两个注意点。一是函数尽量要有明确的输入输出契约状态里写什么、返回什么更新应该清晰固定不能今天返回一个格式、明天换另一种二是节点内部不要做过于重的外部调用而不设超时否则整个图的执行时间会被一个异常节点拖垮。边的定义有两种。普通边固定连接两个节点条件边根据当前状态动态决定下一跳。条件边是实现多智能体路由的核心机制执行流程中的判断题全靠它完成。def route_decision(state: AgentState) - str: if state[next_agent] coder: return coder if state[next_agent] reviewer: return reviewer return __end__ builder.add_conditional_edges(supervisor, route_decision, {coder: coder, reviewer: reviewer, __end__: END})3.3 Conditional Edge与Checkpointer分支和记忆是杀手锏条件边的函数很纯粹输入状态返回一个字符串这个字符串对应某个下游节点。它让图具备了动态决策能力比如Supervisor节点让LLM输出一个JSON指明下一个行动者是谁条件边读取这个JSON决定走向。Checkpointer是LangGraph另一个容易被低估的能力。它负责把每一步的状态持久化到存储后端默认有内存版MemorySaver也有SQLite和Postgres这类持久化实现。Checkpointer解决的痛点是多智能体应用运行到一半如果某一步出错能否从最近的检查点恢复而不是从头再来以及用户关闭会话后系统能否记住之前的执行状态。from langgraph.checkpoint.memory import MemorySaver checkpointer MemorySaver() app builder.compile(checkpointercheckpointer)这里要提醒一点MemorySaver只适合开发调试重启进程状态就没了。生产环境至少要上SQLite数据量大了再考虑用外部存储。3.4 Tool与模型绑定多智能体能力的边界来源有了图结构每个节点内部还需要真正干活的工具和模型。在LangGraph里通常的做法是把工具列表与模型绑定让模型在节点内自动决定调用哪个工具。from langchain_openai import ChatOpenAI from langchain_core.tools import tool tool def search_knowledge_base(query: str) - str: 检索内部知识库 return 知识库中与{}相关的内容.format(query) llm ChatOpenAI(modelgpt-4o, temperature0) llm_with_tools llm.bind_tools([search_knowledge_base])在多智能体架构下每个Worker节点绑定自己的工具集而不是所有工具一股脑绑定给一个模型。工具少的模型在函数调用上明显更准确这也是拆分多智能体能提升整体稳定性的一个直接原因。4. 代码实战从零做一个可运行的Supervisor多智能体4.1 环境安装与前置准备先安装依赖建议在Python 3.11的虚拟环境里操作pip install langgraph langchain-openai这里有个版本提醒LangGraph的API迭代速度很快我写这篇文章时常用的是0.2.x和0.3.x版本如果你的版本更新个别导入路径可能变了以官方文档为准。遇到ModuleNotFoundError时不要慌优先去官方迁移文档里查导入路径。在环境层面我还建议安装langchain-community里面包含一些额外的工具实现虽然多智能体核心用不上但做节点内部的工具组合时可能用到。4.2 定义一个基础的多智能体状态图我们从最简结构开始先实现一个主管分派、两个工人干活的图不做太复杂的归并逻辑保证能跑通。from typing import TypedDict, Annotated, List from langgraph.graph import StateGraph, END from langgraph.checkpoint.memory import MemorySaver class SimpleState(TypedDict): messages: Annotated[List[dict], lambda l, r: l r] output: str def worker_a(state: SimpleState): return {output: A完成检索结果猫是猫科动物。} def worker_b(state: SimpleState): return {output: B完成总结猫属于猫科动物擅长捕猎。} def supervisor(state: SimpleState): # 实际项目中这里会让LLM做决策我们先写死路由逻辑 return {output: 分派给A和B} builder StateGraph(SimpleState) builder.add_node(supervisor, supervisor) builder.add_node(worker_a, worker_a) builder.add_node(worker_b, worker_b) builder.set_entry_point(supervisor) builder.add_edge(supervisor, worker_a) builder.add_edge(supervisor, worker_b) builder.add_edge(worker_a, END) builder.add_edge(worker_b, END) app builder.compile()跑这个图会发现一个很有意思的现象从supervisor出发有两条边分别连到worker_a和worker_b实际执行时LangGraph会并行执行这两个节点最终状态会被合并。这在多智能体协作里非常实用意味着你可以让多个Worker同时干活再汇总结果。4.3 实现条件路由让LLM做主分配上面的示例路由是写死的真实系统里通常要让Supervisor根据用户输入动态决定调用哪个Worker。这一步就要用到条件边。import json from typing import TypedDict, Annotated, List from langgraph.graph import StateGraph, END class AgentState(TypedDict): messages: Annotated[List[dict], lambda l, r: l r] next: str def supervisor_node(state: AgentState): prompt 你是任务主管。请判断用户需求应该交给哪个工作智能体。 用户需求{last_msg} 可选智能体researcher检索信息、calculator数学计算、finisher收尾总结 只输出一个JSON格式{{next: 智能体名称}} .format(last_msgstate[messages][-1][content]) response llm.invoke(prompt) parsed json.loads(response.content) return {next: parsed[next]} def researcher_node(state: AgentState): return {messages: [{role: assistant, content: 已完成资料检索}]} def calculator_node(state: AgentState): return {messages: [{role: assistant, content: 计算结果为42}]} def finisher_node(state: AgentState): return {messages: [{role: assistant, content: 已汇总所有结果生成最终答案。}]} def route_next(state: AgentState) - str: if state.get(next) researcher: return researcher elif state.get(next) calculator: return calculator elif state.get(next) finisher: return finisher else: return END builder StateGraph(AgentState) builder.add_node(supervisor, supervisor_node) builder.add_node(researcher, researcher_node) builder.add_node(calculator, calculator_node) builder.add_node(finisher, finisher_node) builder.set_entry_point(supervisor) builder.add_conditional_edges(supervisor, route_next, { researcher: researcher, calculator: calculator, finisher: finisher, END: END }) builder.add_edge(researcher, supervisor) builder.add_edge(calculator, supervisor) builder.add_edge(finisher, END) app builder.compile()注意这里我把worker执行完后的边又指回了supervisor形成循环。这是LangGraph区别于LangChain的关键能力任务不是一条直线跑完而是可以反复进入主管节点做下一轮判断直到finisher收尾为止。4.4 带记忆能力的完整实战持久化会话状态开发时我们直接编译就能跑但生产环境还需要能在两次会话之间保留历史这就用到了Checkpointer。创建一个带线程ID的会话配置每次运行时传入同一个thread_id图就会在该线程下累积状态相当于给多智能体系统装上了内存。from langgraph.checkpoint.memory import MemorySaver config {configurable: {thread_id: user_001}} app builder.compile(checkpointerMemorySaver()) # 第一次运行 result1 app.invoke( {messages: [{role: user, content: 帮我查一下猫科动物的特征}]}, config ) # 第二次运行模型能看到上一次的对话历史 result2 app.invoke( {messages: [{role: user, content: 那狗呢}]}, config )这个能力在实际落地中太关键了。多智能体每次执行之间如果可以保留上下文用户体验质变同时也让排查追踪有了线索。4.5 更多开箱即用的高层APILangGraph预构建Agent如果不想从零搭图LangGraph也封装了预构建的Agent结构比如create_react_agent它内部已经把工具调用、循环、状态更新都封装好了适合快速验证想法from langgraph.prebuilt import create_react_agent agent_executor create_react_agent( modelllm, tools[search_knowledge_base], state_modifier你是一个只能使用工具的智能助手。, )我的使用建议是 prototyping阶段用create_react_agent快速验证生产阶段还是自己用StateGraph搭因为预构建Agent的自定义空间有限有些边界情况比如某个节点要做额外校验、某条路径要跳过工具直接答复需要手动控制。5. 实操中的常见问题与排查经验这部分是我最想分享的全是实际踩坑记录。5.1 状态字段更新失效原因多半是归并函数缺失很多人刚开始用Annotated时习惯性地用lambda合并字段结果节点返回新消息时旧消息全没了或新消息无法累加。原因在于如果你定义的状态字段没有归并函数LangGraph默认采用覆盖更新一旦节点返回值是空列表就会把整段历史清空。我的建议是消息类字段一定写显式归并函数业务临时字段用覆盖策略并确保每个节点明确返回。排查时先打印节点返回值和图执行后的state多智能体系统出问题时第一步永远是看状态在哪个节点发生了预期外的变化。5.2 多智能体循环变死循环条件边必须兜底多智能体系统一旦引入循环最怕的是图永远跳不出循环。我遇到过一种典型场景Supervisor节点让LLM选下一个Worker模型反复输出同一个Worker名字执行路径就在这个Worker和Supervisor之间来回转直到达到递归上限报错。解决办法有两个。一是给条件路由函数加终止条件比如记录执行步数超过N步强制跳到END二是在LLM决策的提示词里强调可选的收尾选项比如如果任务已经完成直接输出finisher。经验是两条都做提示词做软约束、代码做硬兜底双保险。5.3 Checkpointer恢复时状态错乱注意线程隔离用Checkpointer后如果多个用户共用同一个thread_id状态会互相污染。生产环境必须保证每个会话上下文使用独立的thread_id一般直接用用户会话ID或请求ID。另一个常见坑是thread_id相同但图结构定义变了比如新增节点恢复老线程时可能报错。这种情况要么做数据迁移要么让旧线程直接失效重新开始我在项目里通常选择后者多智能体系统每次升级后最好不要依赖旧状态。5.4 多智能体并发执行的结果冲突当多个Worker并行执行后合并状态时如果两个节点同时更新同一个字段最终值取决于执行顺序可能不符合预期。解决思路是并发节点尽量写入不同的状态字段或者提供明确的归并函数处理冲突。比如多个Worker都产生理由用列表归并就比覆盖稳妥得多。这里做一个排查速查表方便快速定位现象可能原因排查方向节点未执行流程直接跳过条件边返回了错误目标打印路由函数的返回值状态里消息被清空字段覆盖更新或归并函数问题检查Annotated归并逻辑循环执行次数异常LLM路由决策未终止增加最大步数硬限制状态恢复后内容错乱thread_id冲突或图结构变化检查线程隔离与版本兼容模型工具调用频繁失效绑定工具太多或提示词冲突精简工具集、拆分Worker5.5 性能与成本控制经验多智能体系统的每次任务执行往往会调用多次LLM成本比单agent高不少。实测下来Supervisor每轮调度就是一次LLM调用Worker执行又是一次如果任务复杂循环四五轮一次用户请求就可能烧掉十几次模型调用。建议在节点函数里加缓存相同输入直接返回历史结果另外对不关键的任务用便宜的小模型做路由决策只有真正需要专业能力时才调用强模型。6. 进阶技巧与个人实践心得6.1 让多智能体系统保持稳定的三个小技巧第一每个智能体的系统提示词开头必须声明自己的职责边界明确你只负责X遇到Y任务必须说无法处理。我见过太多失败的多智能体案例问题不在架构而在子智能体缺乏边界感动不动越俎代庖。第二节点函数里一定要加异常捕获和降级逻辑。用一个try-except把工具调用包起来工具崩了就在态里记录错误、返回给上层而不是让整个图执行抛异常。多智能体系统越复杂单个节点的韧性就越重要。第三运行日志里带上thread_id和节点名。排查分布式问题不能靠print推测建议在进入和离开节点时输出结构化日志记录状态摘要。LangGraph自带的LangSmith服务可以做完整追踪个人开发者也可以用最简单的方式在每个节点函数里打一个包含状态关键字的日志。6.2 从LangGraph还能延伸出什么玩法LangGraph多智能体框架值得深入挖掘的方向不少。一个是人机协同审批流在某个节点前插入human-in-the-loop逻辑让图运行到关键节点时暂停等待人工确认LangGraph原生支持interrupt机制这能覆盖大量企业级需求。另一个是长时运行任务结合持久化存储让一个多智能体协作任务运行几小时甚至几天中途系统重启也能恢复。最后是事件驱动模式让多智能体架构从请求-响应变成事件驱动。配合消息队列使用业务系统产生事件时再唤醒相应的智能体节点整个系统的异步能力和可扩展性又会提升一个台阶。6.3 踩过几次坑之后的真心话如果你第一次接触LangGraph不要急着堆一个复杂的多智能体系统。先用最简单的两个节点把一个闭环跑通然后逐步增加条件分支、循环、状态持久化最后才考虑多智能体协作。我见过太多人一上来就模仿开源项目搭了一个五六层的图结果连路由都调不通。多智能体的复杂度是指数级上升的每一层抽象都需要你的设计足够清晰而不是靠堆功能掩盖思路混乱。根据我个人经验LangGraph的学习曲线比LangChain高一点但一旦跨过去你设计的系统会从提示词工程式的单次调用真正转变为可编排、可恢复、可观察的复杂应用。这才是多智能体技术带来真正的生产力提升所在。