LangGraph实战指南：从状态机到Agent工作流编排

LangGraph 是当前构建 Agent、多轮对话和复杂工作流时绕不开的一个编排框架。它的核心不是“多了一个新模型”而是把任务拆成节点和边用一份共享的 State 把每一步串起来让你能精确控制分支、循环、并行和持久化。这篇文章适合两类人一类是刚接触 LangGraph、被各种术语绕晕的初学者另一类是已经会用 LangChain 写简单链条但遇到条件路由、循环和子图时不知道怎么落地的开发者。最值得先记住的一点是LangGraph 里你写的不是一段顺序执行代码而是一张图每个节点负责修改 State边决定下一步去哪。很多教程一上来就贴 ReAct Agent 的完整实现代码很长但最后你连 State 怎么更新、为什么走某个分支都没搞清楚。我的建议是先跑一个只有两个节点的最小图再慢慢加条件路由、子图和持久化。哪怕教程标题写着“2026 全套实战”也不要把它当成一个固定版本LangGraph 生态迭代很快落到本机时先确认依赖版本再往下走。1. LangGraph 到底解决什么问题和 LangChain 是什么关系1.1 一张图、一份 State、一组节点LangGraph 解决的是「流程控制」问题。传统 LangChain 的 Chain 是线性链写一个固定流程很容易但一旦你需要根据上一步结果决定下一步走哪个分支、同一个问题并发查多份资料再汇总、网络超时后重试并记录中间结果线性链就会变得很难维护。LangGraph 把流程抽象成一张图节点是实际执行逻辑比如调用一次大模型、查一次数据库、调用一个工具。边是流转条件决定当前节点执行完后下一步去哪。State 是跨节点共享的数据每执行完一个节点State 可能被更新。节点的输入是当前 State输出是一个 dict 或特殊对象图引擎会用输出去更新 State。这个过程看起来像状态机所以 LangGraph 也常被叫作基于图状态机的编排框架。1.2 LangGraph 与 LangChain 的分工很多人会把 LangChain 和 LangGraph 当成二选一实际这两个东西分工不同。LangChain 更像工具箱提供模型封装、Prompt 模板、工具调用协议、文档加载器等。LangGraph 更像调度系统负责决定流程怎么走。你完全可以在 LangGraph 的节点里使用langchain-openai的模型也可以在节点里直接调用 OpenAI SDK、本地 HTTP 服务或普通 Python 函数。LangGraph 本身不绑定某个模型也不要求你必须用 LangChain。判断标准很简单如果你的核心问题是“怎么让模型稳定调用工具”先处理模型接入和工具定义。如果你的核心问题是“多个步骤之间怎么分支、循环、重试、保存状态”才需要 LangGraph。如果只是两个固定步骤顺序执行用普通代码或者 LangChain 的链就够了不必引入图框架。1.3 直接学 LangGraph 前要有的基础建议先具备三样东西会写 Python 函数理解函数参数和返回值。知道TypedDict大概是什么它能约束字典的键和值类型。能调用大模型接口无论用 OpenAI SDK、LangChain 封装还是本地模型服务。不需要精通图论。图在 LangGraph 里更多是一个概念真正要理解的是节点如何返回增量、边如何决定流转、State 如何在多次执行中保存。把这三点想清楚后面学条件路由、子图、并行分支都会顺很多。2. 环境准备先把最小图跑通2.1 依赖安装与版本确认LangGraph 本身是 Python 库安装前先确认你的 Python 版本能在 3.9 以上推荐用 3.11 或 3.12。常见安装命令如下pip install -U langgraph langchain-openai如果你还需要持久化功能常见做法是额外安装 checkpoint 扩展pip install -U langgraph-checkpoint安装完不要急着写代码先确认版本pip show langgraph python -c import langgraph; print(langgraph.__version__)这一步可以避开很多“源码和文档对不上”的问题。LangGraph 的 API 在不同版本之间调整过比如某些常量、类型导入路径可能变了。你看官方文档时也要先确认文档对应的版本和你本机安装的版本是否一致不一致时以本机版本为准。2.2 模型接入的两种方式LangGraph 不内置模型模型在节点函数里被调用。常见有两种接入方式。第一种是使用 LangChain 的模型封装例如langchain-openaifrom langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4o-mini, temperature0)这种方式需要你配置 API Key比如设置环境变量OPENAI_API_KEY。没有 Key 时运行会报认证错误。第二种是直接调用本地模型服务。只要能通过 HTTP 接口返回文本LangGraph 节点里就能使用。比如本地部署了 Ollama节点里用对应的 Python 库去调用。这种方式适合本地测试因为不依赖外部 API速度也更容易控制。我一般建议初学者先用自己最熟悉的模型接入方式先把一个最小图跑通再考虑换成其他模型。不要在第一步同时引入 LangGraph、LangChain、本地模型部署三个新东西否则出问题时很难判断是哪一层的问题。2.3 最小 StateGraph 示例先用一个最简单的例子理解整体流程from typing import TypedDict from langgraph.graph import StateGraph, START, END class DemoState(TypedDict): step: int messages: list def first_node(state: DemoState) - dict: print(first_node 收到 state:, state) return {step: 1, messages: state[messages] [first 已执行]} def second_node(state: DemoState) - dict: print(second_node 收到 state:, state) return {step: 2, messages: state[messages] [second 已执行]} builder StateGraph(DemoState) builder.add_node(first, first_node) builder.add_node(second, second_node) builder.add_edge(START, first) builder.add_edge(first, second) builder.add_edge(second, END) graph builder.compile() result graph.invoke({step: 0, messages: []}) print(最终结果:, result)这段代码做了一件事从START进入first再进入second最后到END。每次节点返回一个 dictLangGraph 会用 dict 中的键去更新State。这里要注意first_node返回的messages不是简单覆盖而是手动把原来的state[messages]拼上了新内容。为什么这样做因为默认情况下LangGraph 对 State 中某个键的处理是“用返回值替换旧值”而不是自动追加。如果你希望多个节点不断往同一个列表里加内容需要自己把旧列表和新内容拼起来或者用后面讲的 reducer 机制。2.4 怎么判断第一次运行成功运行成功后控制台会打印两次“收到 state”最终result应该是{step: 2, messages: [first 已执行, second 已执行]}如果只打印了 first说明 second 节点没有被执行检查边是否写对。如果messages只有一条说明某个节点返回的列表拼接过有问题。如果报错找不到START或END多半是 langgraph 版本较旧导入方式不同。先跑最小图是因为你能用最少变量确认环境、依赖、模型链路都正常。如果这一步报错后面所有内容都没意义。3. 节点函数如何修改 State最容易错的一步3.1 定义 State 时要关注数据类型State 可以是你定义的任何结构但常用TypedDict来声明。例如class ChatState(TypedDict): messages: list step: int query: str定义 State 时建议只放真正需要跨节点共享的字段。如果某个字段只在一个节点内部使用没必要放进 State。State 字段越多追踪起来越麻烦。节点函数签名通常是def node_fn(state: ChatState) - dict: ...参数是当前 State返回值是更新 State 的部分内容。你可以只返回需要修改的字段不需要把整个 State 重新返回一遍。3.2 节点返回的是增量不是完整状态很多初学者会写成def first_node(state: ChatState) - ChatState: state[step] 1 return state这种写法在简单场景能跑但容易带来隐患。LangGraph 的处理方式是接收你返回的 dict把里面的键合并到当前 State 中。你返回完整state只是碰巧也能工作但它会让“本次节点到底改了哪个字段”变得不清晰。更推荐的写法def first_node(state: ChatState) - dict: return {step: 1}这样其他字段不会被改动调试时看返回值一目了然。返回增量而不是返回完整 State是 LangGraph 里最重要的习惯之一。3.3 reducer 与 Annotated追加还是覆盖如果两个节点都想往同一个messages列表里追加内容默认覆盖行为就很麻烦。LangGraph 提供 reducer 机制可以用类型注解指定“这个字段如何合并”from typing import TypedDict, Annotated import operator class ChatState(TypedDict): messages: Annotated[list, operator.add] step: int声明之后节点里返回{messages: [新的消息]}LangGraph 会自动把新列表追加到旧列表而不需要你手动拼接全部消息。为什么要有这个机制因为并行场景下多个节点可能同时返回同一个字段LangGraph 需要知道是覆盖、拼接还是按某种规则合并。operator.add对列表来说是拼接对数字来说是相加。如果你的字段想用其他合并规则也可以自己写一个 reducer 函数只要函数签名接收“当前值”和“新值”返回合并后的结果。要特别注意的是声明了 reducer 之后节点返回类型必须和 reducer 预期一致。比如Annotated[list, operator.add]的字段如果返回None就会触发类型或合并异常这也是常见报错来源之一。3.4 从输出和日志里确认状态更新跑完graph.invoke后直接打印结果可以确认最终 State。但更推荐在节点内部加日志观察每一步输入输出def node_fn(state: ChatState) - dict: print(当前 step:, state.get(step)) ...如果你发现节点返回了step: 2但后续节点拿到的还是 1优先怀疑两件事返回值的键名是不是写错了比如step拼成了stpe。是否声明了 reducer而 reducer 的合并结果不是你预期的那样。LangGraph 的错误信息有时很长但关键位置一般在报错前几行。先看是哪一步抛错再回到节点函数里核对返回值。4. 条件路由、循环检测与并行分支4.1 conditional_edge 的路径映射条件路由是 LangGraph 比较核心的能力。它解决的是“下一步去哪由当前状态决定”的问题。命令是add_conditional_edges。它接收三个参数当前节点名。路由函数输入当前 State返回一个字符串或Send对象。路径映射把路由函数返回的字符串映射到目标节点名。例如def route_by_step(state: DemoState) - str: if state[step] 2: return end return second builder.add_conditional_edges( first, route_by_step, {second: second, end: END} )这段逻辑是执行完first后检查State中的step如果大于等于 2 就结束否则去second。这里的要点是路由函数不要做真正的业务处理它只做判断。真正的业务处理放在目标节点里。如果把“业务处理”和“路由判断”写在同一个节点一旦判断条件复杂后面很难维护。4.2 循环依靠什么退出LangGraph 的图可以有环。比如一个节点处理完后如果问题还没解决可以路由回前面的节点重新处理。这就是循环。循环必须有一个退出条件否则会无限跑下去。LangGraph 用recursion_limit限制最大执行步数常见默认限制是 25 步具体以你安装版本的文档为准。在invoke时可以通过配置调整result graph.invoke( {step: 0, messages: []}, config{recursion_limit: 20} )当执行步数超过限制时会抛GraphRecursionError。遇到这个错误优先检查路由函数里的退出条件是否真的可能满足。我建议在循环相关节点里打印关键字段比如当前是第几步、本轮结果是什么。不要只在最后看报错否则你很难判断是哪一轮判断出了问题。4.3 并行分支与 Send有些场景需要把一个任务拆成多个子任务并行执行例如同时查多个关键词、同时翻译多段文本最后再汇总。LangGraph 里可以用条件路由函数返回Send对象来实现分发给多个节点。Send包含两个信息目标节点名以及传给该节点的 State 片段。from langgraph.types import Send def route_to_workers(state: DemoState): return [Send(worker, {item: item}) for item in state[items]] builder.add_node(worker, worker_node) builder.add_conditional_edges(parent_node, route_to_workers)这里把parent_node后的流程分发给多个worker实例。每个worker拿到不同的item执行完后再进入汇总节点。需要注意两点不同版本Send的导入路径可能不一样较新版本一般在langgraph.types里老版本可能在langgraph.constants。安装后先确认示例代码是从哪里导入的。LangGraph 的“并行分支”是指图结构上允许并行不保证一定启动线程或进程。真正的大模型并发调用还受你的 API 限流、线程池、机器资源影响。不能只靠图并行就认为并发一定上去了。4.4 分支太多时怎么设计节点条件路由一多图会变得复杂。建议遵循一个原则路由函数只返回“路由名”不要返回业务数据。反例def bad_route(state): if state[step] 1: state[plan] A return node_a return node_b把plan写在路由函数里会让调试变得困难因为路由函数改写 State 的行为不够直观。正例是让路由函数只做判断所有状态修改放在后续节点中def good_route(state): return node_a if state[step] 1 else node_b如果你的图分支超过 5 个可以先用一张表格列出“当前条件 → 进入节点 → 该节点负责什么”再写代码。这样避免边写边改最后连自己都分不清某个分支为什么存在。5. 子图、持久化与批量任务编排5.1 子图解决什么场景子图适合两类场景一段复杂流程会被多处复用。主图太大希望把某个模块拆出来单独维护。LangGraph 里常见的子图用法是把一个已编译的图当作一个节点来调用sub_builder StateGraph(SubState) sub_builder.add_node(a, sub_node_a) sub_builder.add_edge(START, a) sub_builder.add_edge(a, END) sub_graph sub_builder.compile() def parent_node(state: DemoState) - dict: sub_result sub_graph.invoke({input: state[query]}) return {sub_result: sub_result}然后把这个parent_node加入主图。这样做的好处是子图内部的路由、循环、状态更新都被封装起来主图只关心子图的输入和输出。使用子图时最要留意的是输入输出字段对齐。子图内部使用的 State 和主图不一定相同你要在调用时手动把主图字段映射成子图输入字段再把子图返回结果取出来放回主图 State。5.2 checkpointer 与断点续跑LangGraph 的持久化主要靠 checkpointer。它可以在每次节点执行后保存图的状态这样程序中断或进程重启后还能从某个检查点继续运行。常见的内存版写法from langgraph.checkpoint.memory import MemorySaver checkpointer MemorySaver() graph builder.compile(checkpointercheckpointer) config {configurable: {thread_id: thread-001}} graph.invoke({messages: []}, config)这里thread_id是区分多轮会话或者多个任务的关键。同一个thread_id对应一条独立的状态记录不同的任务使用不同thread_id避免状态串掉。MemorySaver适合学习和小型测试但它把状态存在内存里重启后就会丢失。生产环境建议使用支持落盘的持久化后端比如基于 SQLite 或 Postgres 的 checkpoint 方案。具体配置不同版本差异较大以官方文档示例为准。5.3 批量任务要盯住四个指标批量任务不能只看“能不能跑”。我建议至少盯四个指标指标判断标准成功率连续跑一批任务有多少任务能拿到完整结果。吞吐单位时间内能完成多少条任务观察排队时间是否不断增长。资源占用CPU、内存、磁盘、API 调用量是否在可控范围。可恢复性任务失败后能否重试丢一条队列后能否从断点继续。如果只是把一批输入循环调用graph.invoke问题不大但一旦某个任务中途失败你要能快速定位是哪条任务、在哪个节点失败、State 到了什么程度否则排查成本会很高。5.4 一个批量编排的通用思路批量处理的通用思路可以这样设计每条任务分配独立task_id作为thread_id。每条任务用独立输入字典避免多个任务共享同一个可变对象。记录每个任务的状态待处理、执行中、成功、失败。失败时捕获异常保存错误信息把任务状态改为“待重试”。设置最大重试次数防止一个坏任务无限重跑。控制并发数不要一上来就开最大并发先用 2 到 3 个并发跑一小批观察资源占用和 API 限流情况。如果输出需要落盘建议输出文件命名里带上任务 ID 和时间戳比如result_task_001_20260218.json。这样即使多个任务并发写文件也不会互相覆盖。6. 常见报错和排查顺序6.1 启动失败依赖、路径、权限看到启动失败、进程退出、报错码非 0先不要怀疑框架按这个顺序排查Python 环境是不是你安装依赖的那个环境。Windows 下容易装到一个 Python、运行用另一个 Python。依赖版本是否匹配。检查pip show langgraph和pip show langchain-openai。当前工作目录和代码文件路径是否包含中文或特殊字符某些库在 Windows 下处理这类路径会出问题。是否有权限写入输出目录。批量任务最容易遇到“日志目录不存在”或“磁盘写满”。如果你看到“退出码 2”或类似信息大概率是环境或路径问题不是 LangGraph 框架本身的问题。先把最小示例跑通再回到业务代码。6.2 节点不执行或状态没更新节点不执行最常见的原因是条件路由返回的节点名不在映射里。比如路由函数返回model_call但映射里写的是modelcallLangGraph 会提示找不到对应节点。状态没更新最常见的原因是节点返回的键名和 State 里的键名不一致。比如 State 里是steps节点返回step那steps永远不会变。排查时直接打印节点输入和输出def node_fn(state): print(输入 state 的 step:, state.get(step)) new_step state.get(step, 0) 1 print(即将返回 step:, new_step) return {step: new_step}节点不执行和状态没更新是两个不同问题先确认是“没进节点”还是“进了但更新失败”。6.3 循环和递归限制如果报错信息里有recursion或GraphRecursionError说明图执行超出了最大步数。优先检查循环里的退出条件。比如路由函数里写的是if state[step] 5: return retry return end但如果某个节点没有递增step这个循环永远不会退出。不要只调大recursion_limit那只是把问题往后推迟。正确做法是确认每一步循环里某个关键字段一定在变化并且最终能满足退出条件。6.4 流式输出与接口对接问题如果你用graph.stream查看节点输出默认可能按节点粒度输出而不是像直接调模型那样一 token 一 token 吐字。读者问“为什么运行没有流式效果”多半是因为没设置合适的stream_mode或者模型调用本身没有开流式。不同版本的stream参数不一样建议先查官方文档看当前版本支持哪些模式再决定用哪种。也可以先回到基础调用确认你的模型接口单独调用时能正常返回内容再考虑 LangGraph 的流式配置。模型本身不返回东西LangGraph 再怎么会编排也没用。6.5 资源占用和生产化边界LangGraph 本身占用的资源并不高真正吃资源的是你节点里的大模型推理、向量检索、文档解析等操作。低配置机器能跑通小示例是正常的但不代表可以批量跑高并发任务。生产化时要额外考虑API Key 的管理和限流避免并发一高就大量报 429。Checkpoint 的存储位置内存版不适合多实例部署。日志里是否能看到任务 ID、节点名、耗时和错误摘要。并发任务之间是否使用了同一个可修改的全局变量。LangGraph 的 State 虽然帮你管理图内状态但如果你在节点里操作全局列表或全局字典多线程下仍然可能出问题。7. 学习路线和落地建议怎么用课件代码少走弯路7.1 推荐执行顺序如果你的课件代码或仓库包含多个示例建议按下面的顺序执行而不是一开始就打开最后一个多智能体协作文件顺序练习目标验收标准1最小 StateGraph两个节点顺序执行结果正确2State 修改节点返回增量字段按预期更新3条件路由根据 State 进入不同分支4循环能正常退出不会无限递归5子图主图能调用子图输出字段对齐6持久化同一个线程 ID 能续跑7批量任务多任务独立执行失败能定位每练一个示例都要做一次“破坏性测试”。比如故意让路由函数返回一个不存在的节点名观察报错长什么样故意让 reducer 返回错误类型观察异常信息。见过了这些报错后面遇到时就不会慌。7.2 调试习惯最实用的调试习惯有三个第一先看打印再看异常。在节点函数里加打印确认当前输入和输出不要直接猜。第二先缩小范围再复现。把链路拆成最小可运行示例如果最小示例能跑说明问题出在业务逻辑而不是 LangGraph 本身。第三先确认输入格式再改参数。很多“运行不起来”的问题不是框架配置问题而是传入的输入字典少了某个键或者模型返回格式不符合后续节点预期。7.3 什么时候不该用 LangGraph不是所有流程都需要 LangGraph。如果任务只有两三个固定步骤没有分支、没有循环、不需要持久化直接用普通 Python 函数调用更简单。强行引入图框架只会增加理解成本和调试负担。反过来如果你的流程开始出现以下信号就该考虑 LangGraph下一步执行依赖上一步结果用普通 if else 写得很乱。同一个流程需要按不同条件反复运行。需要保存中间状态中断后还要能恢复。多个步骤并行执行最后再合并。7.4 最后说一点心态学习 LangGraph 时不要被“图”、“状态机”、“条件边”这些词吓住。它本质就是一个能记录状态的流程引擎。你写的每个节点只是普通 Python 函数边就是决定下一步走哪里的规则。先把最小图跑起来再一步一步加复杂度。我比较推荐的学习方式是每周只吃透一个能力点。第一周跑通单图和 State第二周写条件路由和循环第三周做子图和持久化第四周把之前的内容组合成一个批量处理任务。这样即使遇到报错你也清楚问题出在哪一块而不是把所有语法混在一起。课件代码只是一个起点真正有价值的是你把它改成自己的场景。比如把示例里的模型换成本地模型把示例里的假数据换成真实业务输入把单条 invoke 改成带任务 ID 的批量队列。改完这些LangGraph 对你来说就不是一个教程名词而是一个能落地的工具了。