在之前的多个项目里我一直想用 LangGraph 把“多个大模型角色协作”这件事做成可控、可观测、能上线的工程方案但早期资料零散网上教程大多停在“用 LangChain 调一次模型”的程度真正把多智能体架构、核心组件、分支路由、状态持久化讲透的内容很少。这篇文章我会从多智能体的基本概念讲起逐个拆解 LangGraph 的 State、Node、Edge、Conditional Edge、Send、Checkpoint 等核心组件然后给出一套可以本地运行的多智能体协作实战项目最后整理开发中常见的问题排查思路和工程化建议适合正在学习 AI 大模型应用开发、准备上手 LangGraph 多智能体的开发者。1. 多智能体与 LangGraph 背景与核心概念1.1 什么是多智能体系统先来理解“多智能体”这个概念。在 AI 大模型领域单个智能体通常是指“一个大模型 若干工具 一套提示词”组成的最小执行单元。你可以把它理解成一个能独立完成某类任务的数字员工。多智能体系统就是多个这样的数字员工组成一个团队每个员工负责一个独立角色比如数据分析、文案撰写、代码审查、质量检查等然后通过消息传递、状态共享和路由调度协同完成一个更复杂的任务。一个最简单的多智能体协作流程是用户提出一个复杂需求。调度者分析需求决定先让哪个角色处理。角色 A 处理完后把结果写入共享状态。调度者根据结果决定是否交给角色 B。所有角色完成后汇总输出最终答案。这种模式的好处是任务分工明确、每一步都能观察和干预、出现问题容易定位。比如“先生成报告再检查报告质量质量不合格就退回重写”这种流程用单个大模型调用做出来比较笨拙但用多智能体架构却很自然。1.2 LangGraph 是什么LangGraph 是 LangChain 社区推出的一个面向 LLM 应用的状态化编排框架它的核心思路是把智能体的执行过程建模成一张“图”。在这张图里节点表示一次计算比如调用大模型、执行一个函数、调用一个工具。边表示节点之间的流转关系比如“A 执行完必须执行 B”。状态则是整个图的共享数据每个节点都能读取和修改状态下一个节点能看到上一个节点的修改结果。相比直接用 Python 代码写 if-else 来回调用大模型LangGraph 有几点很关键天然的循环支持智能体经常会遇到“工具调用 - 返回结果 - 再调用 - 再返回”这本质上是一个循环LangGraph 对循环的支持非常自然。可观测性图执行过程中的每一步状态变化都能被追踪。可持久化通过 Checkpoint 机制把图的状态保存下来支持记忆和断点恢复。有向图带来的可控性业务怎么流转图就怎么画不依赖模型自由发挥。1.3 LangGraph 与 LangChain 的区别很多刚开始接触大模型开发的同学会混淆 LangChain 和 LangGraph。从定位上讲LangChain 的核心是“链”也就是把一次提示词调用、一个工具调用、一次输出解析串成一条顺序执行的管道适合做相对固定的流程。LangGraph 的核心是“图”支持分支、循环、并行、条件路由适合做动态、多角色、有反馈闭环的复杂编排。可以这样简单理解如果业务是固定的三步走用 LangChain 就够如果业务是“根据中间结果决定下一步去哪里”或者需要多个智能体来回协作LangGraph 更合适。需要注意的是LangGraph 并不是 LangChain 的替代品两者可以共存。LangGraph 节点内部依然可以使用 LangChain 的模型封装、Prompt 模板、输出解析器等能力。1.4 为什么需要 LangGraph 来构建多智能体应用单智能体应用的逻辑通常比较简单就是“问题 - 模型 - 答案”。一旦到了多智能体场景问题就变复杂了多个智能体之间如何传递数据某个智能体失败了是由另一个智能体重试还是直接退出是顺序执行还是并行执行如何让模型来动态决定下一步交给哪个智能体用户多轮对话时智能体协作的中间状态怎么保存这些问题本质上都是“流程控制”问题而流程控制是图最擅长的事。LangGraph 把状态管理、路由、循环、并行、持久化这些能力都内置了让我们可以专注于业务而不是自己写一套状态机。2. 环境准备与版本说明2.1 安装依赖在开始写代码之前先准备 Python 环境。建议使用 Python 3.10 及以上版本避免旧版本对类型注解和异步语法支持不完整。创建虚拟环境后安装以下核心依赖pip install langgraph langchain-core langchain-openai如果你的环境需要访问 OpenAI 接口还需要配置 API Keyexport OPENAI_API_KEY你的 API Key如果使用的是国内模型服务或者其他兼容 OpenAI 协议的接口可以通过 base_url 参数指定服务地址比如from langchain_openai import ChatOpenAI llm ChatOpenAI( model你的模型名称, api_key你的 API Key, base_url你的接口地址 )为了文章示例的安全性和可复制性我会先用模拟函数扮演大模型跑通完整的图结构在第四节会给出如何替换成真实大模型调用的方式。2.2 版本差异提醒LangGraph 目前迭代速度比较快不同小版本之间接口细节可能不同。本文示例基于常见稳定版本的常用 API核心组件如StateGraph、TypedDict、add_messages、conditional_edges、Send都属于主接口在大版本内基本保持一致。如果你用的版本较旧部分命名可能不同比如早期版本中图的入口方法叫set_entry_point后续版本仍然兼容而一些新增的快捷方法则可能只在较新版本里可用。建议安装时不要锁定过旧版本直接安装最新版即可。pip install --upgrade langgraph安装完成后可以验证版本python -c import langgraph; print(langgraph.__version__)如果执行报错说明当前版本可能还没有暴露这个包入口可以参考官方文档确认导入方式。2.3 项目结构本文的实战项目是一个多智能体协作的“情报简报生成系统”项目结构如下langgraph-multi-agent-demo/ ├── main.py # 入口文件构建图并执行 ├── state.py # 定义共享状态 ├── agents.py # 定义各个智能体节点 ├── tools.py # 定义工具函数 └── requirements.txt # 依赖清单实际开发中建议把状态定义、节点函数、图构建逻辑拆分成不同模块避免把所有代码堆在一个文件里。3. LangGraph 核心组件拆解3.1 State共享状态在 LangGraph 中State 就是图的全局状态。每个节点执行后可以返回一个字典LangGraph 会把字典里的字段更新到全局状态里。State 通常用 TypedDict 或 Pydantic 模型定义。下面是一个示例from typing import TypedDict, Annotated from langgraph.graph.message import add_messages class AgentState(TypedDict): # 使用 add_messages reducer使每次写入 messages 都是追加而不是覆盖 messages: Annotated[list, add_messages] # 用户原始问题 query: str # 分析结果 analysis: str # 写作文案 draft: str # 审核意见 feedback: strAnnotated[list, add_messages]是 LangGraph 里一个很有用的写法。默认情况下多个节点如果返回同一个字段后执行的节点会直接覆盖前一个节点的值。但加上add_messages后每次写入messages字段都会自动合并到已有列表里适合保存多轮对话记录。理解这一点很重要State 不只是简单字典字段的合并策略是通过 reducer 来控制的。3.2 Node节点节点就是普通 Python 函数接收整个 State 作为参数返回一个字典把需要更新的字段返回给图。def analyze_node(state: AgentState) - dict: query state[query] # 这里是模拟大模型分析 analysis f针对问题「{query}」的分析结果这是一个 LangGraph 多智能体示例。 return {analysis: analysis}节点的编写规范函数参数是 state类型是整个 AgentState。返回值必须是字典里面包含要更新的字段。不要直接修改入参 state而是返回新字典。LangGraph 内部会负责合并状态。在图中注册节点使用add_nodefrom langgraph.graph import StateGraph graph StateGraph(AgentState) graph.add_node(analyze, analyze_node)3.3 Edge普通边和条件边普通边表示无条件流转。比如分析结束后必须写文案graph.add_edge(analyze, write)条件边表示根据当前状态动态决定下一步进入哪个节点。条件边由一个路由函数返回目标节点名称def route_after_write(state: AgentState) - str: if 需要修改 in state.get(feedback, ): return write # 回到写作文案节点重写 return finish # 质量合格进入结束节点注册条件边graph.add_conditional_edges( write, route_after_write, { write: write, finish: END } )这里要注意条件边函数返回的字符串必须是第三个参数映射表里的 key否则运行时会报错。这个映射表的作用是把路由函数返回值映射到真实的节点名。3.4 编译图与执行所有节点和边都定义好后需要调用compile()方法把图编译成可执行对象app graph.compile() result app.invoke({ query: 帮我总结 LangGraph 的核心组件, messages: [] })invoke的入参是初始状态字典返回值是最终状态字典。整个过程可以用stream方法逐步观察for chunk in app.stream({query: 你好}): print(chunk)stream每次输出一个节点执行后的状态变化非常适合调试。3.5 Checkpoint持久化与记忆Checkpoint 是 LangGraph 中实现对话记忆和断点恢复的机制。在 compile 时传入一个 CheckpointSaver图每次执行后都会保存当前状态下次执行时可以从某个历史状态继续。from langgraph.checkpoint.memory import InMemorySaver memory InMemorySaver() app graph.compile(checkpointermemory) config {configurable: {thread_id: user-session-001}} result app.invoke( {query: 第二轮对话}, configconfig )通过thread_id区分不同会话同一个会话的多次执行会共享历史状态。需要注意InMemorySaver只适合开发和测试使用生产环境建议使用langgraph-checkpoint-postgres或langgraph-checkpoint-dynamodb等持久化存储。3.6 Send动态并行分支在多智能体场景中经常需要把一个任务拆分成多个子任务并行执行。比如要分析 5 份报告每个报告交给同一个分析智能体并行处理。这时可以使用Send。Send的作用是在节点里返回多个Send对象每个 Send 对象指定目标节点和该节点要接收的状态。from langgraph.types import Send def dispatch_node(state: AgentState): topics [LangGraph, 多智能体, AI Agent, 状态编排, Checkpoint] return [ Send(analyze_topic, {topic: t}) for t in topics ]这里每个Send(analyze_topic, {topic: t})表示把{topic: t}作为初始状态交给analyze_topic节点执行。LangGraph 会并行创建多个analyze_topic的执行分支。Send适合任务数量在运行时才能确定的场景比如读取了一堆文档、生成了多个子任务和普通一次性 add_node 完全不同。3.7 常见理解误区节点返回的状态只会更新它返回的字段不会清空其他字段。如果不加 reducer后写入的字段会覆盖前值。想追加就用Annotated搭配 reducer。END不是一个节点它表示图的终点不能给它 add_node。条件边的映射表 key 要和路由函数返回值完全一致大小写都不能错。4. 完整实战案例多智能体协作系统4.1 需求与架构设计接下来我们实现一个“情报简报生成系统”。系统的目标是根据用户提出的主题生成一份结构完整、质量合格的技术简报。我们设计三个智能体角色分析师ResearchAgent负责整理主题相关的分析要点。撰稿人WriterAgent负责把分析要点扩展成完整简报。审核员CriticAgent负责检查简报质量决定是否通过。完整流程如下用户输入主题。分析师写出分析要点。撰稿人基于分析要点生成简报。审核员检查简报。如果简报不合格回到撰稿人节点重写。如果合格结束流程并输出最终简报。这个案例虽然简单但已经覆盖了 LangGraph 的 State、Node、Edge、Conditional Edge、循环、Compile、Stream 这些核心能力。4.2 定义共享状态文件state.pyfrom typing import TypedDict, Annotated from langgraph.graph.message import add_messages class BriefState(TypedDict): # 用户输入的主题 topic: str # 分析师的分析要点 analysis: str # 撰稿人生成的简报 brief: str # 审核员的反馈意见 feedback: str # 重写次数避免无限循环 rewrite_count: int # 消息记录 messages: Annotated[list, add_messages]rewrite_count字段很关键多智能体循环中必须要有循环终止条件否则可能陷入死循环。4.3 模拟大模型调用的工具函数文件tools.py先定义几个模拟大模型的函数。后面要接真实大模型时只需要替换这些函数内部实现。import random def mock_llm_analyze(topic: str) - str: 模拟分析师对主题进行要点拆解。 return ( f针对《{topic}》的分析要点如下\n f1. 大模型应用正在从单模型调用走向多智能体协作。\n f2. LangGraph 是当前主流的状态化编排框架。\n f3. 工程落地的关键在于状态管理、路由设计和可观测性。\n f4. 建议先以图思维拆解业务流程再实现代码。 ) def mock_llm_write(analysis: str) - str: 模拟撰稿人根据分析要点生成完整简报。 return ( f【技术简报】\n f一、背景\n{analysis}\n f二、核心挑战\n f多智能体场景下角色分工、数据传递、流程控制都是必须解决的问题。\n f三、LangGraph 方案\n f使用状态图编排多个智能体利用条件边实现质量反馈闭环。\n f四、结论\n fLangGraph 将复杂流程显式建模显著提高了多智能体应用的稳定性。 ) def mock_llm_critic(brief: str) - tuple[str, str]: 模拟审核员检查简报质量。 返回元组(是否通过, 审核反馈) # 为了让示例更容易看到“重写”效果这里随机返回一次不合格 if LangGraph in brief and random.random() 0.3: return pass, 内容完整结构清晰审核通过。 return fail, 缺少实际代码示例需要补充技术细节后重写。这里我用随机数让审核有一定概率失败从而演示循环重写机制。实际项目中应该用真实大模型判断或者接入规则引擎。4.4 实现智能体节点文件agents.pyfrom state import BriefState from tools import mock_llm_analyze, mock_llm_write, mock_llm_critic def analysis_node(state: BriefState) - dict: 分析师节点对主题进行要点拆解。 topic state[topic] analysis mock_llm_analyze(topic) return { analysis: analysis, messages: [{role: assistant, content: f分析师已完成要点拆解{analysis}}] } def write_node(state: BriefState) - dict: 撰稿人节点基于分析要点生成简报。 analysis state.get(analysis, ) brief mock_llm_write(analysis) return { brief: brief, rewrite_count: state.get(rewrite_count, 0) 1, messages: [{role: assistant, content: f撰稿人已完成第 {state.get(rewrite_count, 0) 1} 版简报。}] } def critic_node(state: BriefState) - dict: 审核员节点检查简报质量。 brief state.get(brief, ) verdict, feedback mock_llm_critic(brief) return { feedback: feedback, messages: [{role: assistant, content: f审核结果{verdict}{feedback}}] } def route_after_critic(state: BriefState) - str: 审核后路由 如果审核未通过且重写次数未超过上限回到撰稿人节点重写 否则进入结束节点。 feedback state.get(feedback, ) rewrite_count state.get(rewrite_count, 0) if 通过 in feedback: return finish if rewrite_count 3: return finish return rewrite注意route_after_critic返回的是映射表的 key不是直接返回节点名。上一节我们提过这个细节这里再次验证。4.5 构建图文件main.pyfrom langgraph.graph import StateGraph, END from langgraph.checkpoint.memory import InMemorySaver from state import BriefState from agents import analysis_node, write_node, critic_node, route_after_critic def build_graph(): graph StateGraph(BriefState) # 注册节点 graph.add_node(analysis, analysis_node) graph.add_node(write, write_node) graph.add_node(critic, critic_node) # 设置入口节点 graph.set_entry_point(analysis) # 固定顺序边 graph.add_edge(analysis, write) graph.add_edge(write, critic) # 条件边审核员决定是重写还是结束 graph.add_conditional_edges( critic, route_after_critic, { rewrite: write, finish: END, } ) return graph.compile(checkpointerInMemorySaver()) if __name__ __main__: app build_graph() config {configurable: {thread_id: demo-001}} result app.invoke( { topic: LangGraph 多智能体实战, analysis: , brief: , feedback: , rewrite_count: 0, messages: [] }, configconfig ) print( * 50) print(最终简报) print(result[brief]) print( * 50) print(审核反馈) print(result[feedback])4.6 运行与验证保存所有文件后执行python main.py预期输出会包含分析师要点拆解、撰稿人简报生成、审核员的反馈。如果审核返回“不通过”图会自动回到 write 节点重新生成最多重写 3 次后强制结束。这里有一个值得观察的点因为用了add_messagesmessages 字段会累积所有节点的日志你可以通过打印result[messages]查看完整的执行轨迹。如果运行时报错expected value of type str for key topic...之类的类型异常多半是初始 state 里字段缺失或类型不对对照状态定义补齐字段即可。4.7 接入真实大模型上面示例中的mock_llm_*函数返回的是固定文本实际项目中需要替换成真实模型调用。以 OpenAI 风格接口为例可以这样实现分析师节点from langchain_openai import ChatOpenAI llm ChatOpenAI( modelgpt-4o-mini, temperature0.2 ) def real_analysis_node(state: BriefState) - dict: topic state[topic] prompt f你是技术分析师请针对《{topic}》输出 4 条分析要点。 response llm.invoke(prompt) analysis response.content return { analysis: analysis, messages: [{role: assistant, content: f分析师完成{analysis}}] }同理撰稿人和审核员节点也可以改成提示词模板 模型调用的方式。审核员节点可以要求模型返回固定格式方便代码解析prompt f 请作为审核员检查以下简报是否合格。 简报内容 {brief} 如果合格请只回复pass 如果不合格请只回复fail:需要修改的原因 通过设置强约束提示词将模型输出保持为机器可解析的格式这是工程化的常用做法。5. 多智能体架构模式总结5.1 顺序执行模式与上面 4.5 的案例类似多个节点按固定顺序执行适合流程特别明确的业务。优点是简单、稳定缺点是模型没有决策空间。适用场景固定流水线任务比如数据清洗 - 特征提取 - 写报告。5.2 并行扇出模式使用Send将一个任务拆分成多个子任务并行执行。适合“处理多份文档”“批量分析多个维度”等场景。参考结构def fan_out_node(state): items state[items] return [Send(process_item, {item: item}) for item in items] def process_item_node(state): item state[item] result do_something(item) return {results: [result]} # 需要配合 reducer 合并结果并行子任务的结果合并需要额外处理。如果多个子节点返回同一个字段必须定义 reducer比如from typing import Annotated from langgraph.graph.message import add_messages class ParallelState(TypedDict): results: Annotated[list, lambda a, b: a b] # 自定义 reducer 合并列表5.3 监督者模式监督者模式是多智能体架构中最常见的模式之一。核心思路是一个 Supervisor Agent 负责理解用户意图动态决定下一步把任务交给哪个专业智能体。在 LangGraph 中Supervisor 就是路由节点用大模型来替换条件边里的规则判断。这样系统的流程控制不再是死板的 if-else而是模型的动态决策。5.4 分层模式当智能体数量比较多时可以把一部分智能体封装成一个子图再把子图作为整体嵌入到更大的图里。这种模式和软件工程里的模块化思想一致。LangGraph 允许直接在一个图里编译另一个子图并作为节点添加适合大型复杂业务。5.5 架构选型建议如果智能体角色少、流程固定顺序模式最简单。如果子任务相互独立优先并行缩短耗时。如果流程会随着用户意图变化监督者模式更合适。如果智能体数量超过 5 个优先考虑分层分组降低单张图的复杂度。6. 常见问题与排查思路以下是我在开发 LangGraph 多智能体应用时遇到的典型问题整理成表格方便查阅问题现象常见原因解决思路条件边报错 KeyError路由函数返回的 key 不在映射表中检查返回值和映射表的 key 是否完全一致状态字段被覆盖没有使用 reducer将需要追加的字段声明为Annotated[list, add_messages]或自定义 reducer图不执行某节点入口节点设置错误确认set_entry_point指向的节点已注册死循环缺少循环终止条件添加重试次数统计节点达到上限强制结束并行子任务结果丢失多节点同时返回同名字段没有合并给目标字段配置自定义 reducer提示词模型输出无法解析模型没有按格式返回使用输出解析器或要求模型返回 JSON并做异常兜底调用模型报 401API Key 错误或未设置检查环境变量和实例化参数调用模型报 429请求频率超过限制降低并发、增加重试退避除了表格里的问题还有两个经常踩的坑。第一个坑是版本差异。LangGraph 迭代很快网上资料可能是几个月前的接口命名可能已经变化。遇到接口报错优先去官方文档核对当前版本的StateGraphAPI不要盲目照抄老教程。第二个坑是状态字典里塞了太多数据。有人会把大文本、图片、临时计算结果全部都塞进 State导致每次状态传递都开销很大调试也不方便。建议 State 只保存核心流程数据需要隔离的大对象存入外部存储用 id 引用。还有一个关于执行方式的排查建议如果你使用invoke发现结果和预期不符可以把程序改成stream逐节点观察for event in app.stream(initial_state, configconfig): for key, value in event.items(): print(f节点: {key}) print(value)这样能清楚看到每个节点到底返回了什么、路由函数走了哪条边比直接观察最终结果直观得多。7. 最佳实践与工程建议7.1 状态设计要克制State 是图的公共黑板但不是垃圾桶。在设计状态时可以问自己三个问题这个字段会被两个以上节点用到吗如果只在一个节点内部使用就不应该放进 State。这个字段会持续增长吗如果是消息列表要考虑裁剪旧消息避免上下文过长。这个字段是临时中间结果吗如果是建议在一个节点内部处理完再输出最终结果。7.2 节点函数保持单一职责每个节点只做一件事情节点命名要能直接反映职责。比如不要写一个process_all_node里面既调模型又解析输出还写日志。应该拆成format_input_node、call_model_node、parse_output_node。这样拆分的好处是调试方便、复用方便、测试方便。模型调用失败时你只需要替换call_model_node的实现。7.3 循环控制必须要有硬上限多智能体应用的运行时间模型和普通接口完全不同。普通接口通常毫秒级响应多智能体可能涉及多轮模型调用和工具调用耗时可能突破几十秒甚至几分钟。所以必须设置节点重试次数上限。智能体重写轮次上限。总执行步数上限。在上面的案例中rewrite_count 3就是循环终止条件。实际项目中可以在路由函数里增加这种判断也可以在invoke前后用超时机制兜底。7.4 日志与可观测性生产环境一定要记录足够详细的日志。每次模型调用要记录modelxxx prompt_tokensxxx completion_tokensxxx latency_msxxx agent_namexxx thread_idxxx这样不仅能排查问题还能做成本分析。建议在节点执行入口和出口各加一行日志输出节点名称和关键状态字段的变化。7.5 工具调用权限与安全边界多智能体通常伴随工具调用。凡是给智能体开放的工具都要遵守最小权限原则只开放业务必需的工具。工具内部做参数校验和越权检查。对可能产生外部副作用的工具增加人工确认环节。敏感信息不要出现在提示词里。例如一个“发送邮件”工具不应该允许智能体直接向任意地址发送邮件至少要校验收件人是否在白名单内。7.6 成本控制多智能体的 token 消耗通常是单次模型调用的数倍。控制成本可以从以下几方面入手缓存高频问题。多智能体之间传递内容时只传摘要而不是全文。对同一文本内容避免多个智能体重复读取完整原文。设置最大执行步数防止循环导致 token 浪费。7.7 异步与并发部署invoke是同步执行高并发场景下会阻塞线程。生产环境推荐使用异步接口result await app.ainvoke(initial_state, configconfig) async for event in app.astream(initial_state, configconfig): # 处理事件 pass如果你通过 FastAPI 暴露接口在async def端点里调用ainvoke更合适避免阻塞事件循环。8. 总结与下一步学习路线写到这里我们已经完成了一套从零开始的 LangGraph 多智能体实战理解了多智能体的概念以及 LangGraph 在其中的定位。掌握了 State、Node、Edge、Conditional Edge、Send、Checkpoint 等核心组件。实现了一个“分析师 撰稿人 审核员”的三智能体协作系统包含循环和条件路由。整理了多智能体架构的几种常见模式。梳理了开发中的高频问题和高阶工程建议。如果你准备继续深入学习建议按下面的路线走精读 LangGraph 官方文档重点关注 Concepts 部分对 State、Graph、Checkpoint 的说明。在本地跑一个基于真实模型的监督者模式案例体会 Supervisor 动态路由。研究Send的并行扇出机制尝试构建一个批量处理的智能体。阅读 LangGraph 源码中关于 reducer 的合并逻辑加深对状态管理的理解。尝试连接 Postgres 或 Redis 作为 Checkpoint 存储实现会话级记忆持久化。最后给你一个实操建议不要在项目一开始就追求复杂的智能体架构。先用模拟数据把图的结构、路由逻辑、状态流转跑通再逐步把模拟节点替换成真实模型调用。这种“先搭骨架再填肉”的方式能帮你更快定位问题是出在提示词、状态还是流程控制上。如果这篇文章对你理解 LangGraph 多智能体有帮助可以收藏备用也欢迎动手把案例跑起来遇到问题可以在评论区交流。