如果你的多步骤Agent还在靠一串串if-else硬撑你迟早会遇到这个坎每加一个分支代码就膨胀一轮想回退上一步重试得手写一整套状态保存更别说把人工审核插到流程中间——那基本等于重构。我去年用LangChain搭一个带工具调用的Agent时就被这些事反复折腾。后来切到LangGraph思维方式换成状态图之后很多问题一下子就顺了节点就是执行动作边就是流转规则全局状态就是一张数据快照整个工作流变成了一张可以随时查看、随时中断、随时恢复的有向图。这篇内容适合两类人看一类是把LangChain的链式调用写到没法维护、想换编排方案的人另一类是听说过状态图但不知道怎么下手、想快速跑通一个能落地的LangGraph工作流的初学者。我会把核心概念、完整示例、条件路由、人工介入以及我在实际项目中踩过的坑一次性讲清楚。1. 为什么是状态图从链式调用到有向图1.1 链式调用的天花板LangChain早期的核心抽象是Runnable把提示词、模型调用、输出解析串成一条链A接BB接C像流水线一样直着走下去。单一顺序还好一旦出现判断一下再选择走哪条路这步结果不满意就退回上一步重来需要把中间状态存起来供后续查询这类需求链式结构就开始别扭了。我记得很清楚当时为了做一个先收集资料→判断资料是否充足→不充足就再收集一轮→充足后写摘要的流程我在链外部套了好几层循环和条件判断还要自己维护一个dict来存中间结果。流程跑起来没问题可一旦出错你想定位现在走到哪一步了、中间状态是什么就非常痛苦因为你根本没有执行状态这个概念只有一条已经跑完或跑到一半的链。这就是链式调用的天花板它适合无状态、无分支、无回退的线性流程。而真实工作流几乎都有分支、循环、回退、暂停。1.2 状态图到底在表达什么LangGraph的解法很朴素把工作流建模成一张有向图图的节点是你要执行的每一步动作边则定义动作之间的流转关系。每一次执行整张图上共享一个State对象它就像一只在中转站之间传递的快递箱每个节点取走需要的、放回产生的然后按边去往下一个节点。这个思想其实源自有限状态机。计算机科学里的状态机画出来就是圆圈箭头圆圈是状态箭头是触发转移的事件。LangGraph把它工程化成了节点边全局状态的执行编排模型只不过这里的节点不只是状态更像是带着数据和逻辑的动作单元。为什么这种建模方式适合复杂工作流三个原因分支和循环是图结构的天然表达不用绕道if-else全局状态让每一步都有迹可循出问题能定位到具体节点图是可视化结构画出来就能讲清楚设计阶段方便评审。我后来发现一个窍门先用纸笔画你的工作流画完基本就是StateGraph的骨架。节点画成方框条件判断画成菱形箭头标上数据流转。搬进代码只是把这张图翻译成add_node和add_edge调用而已。1.3 和低代码工作流平台的关系很多人在搜LangGraph的时候同时会看到dify、coze、n8n这些工作流平台容易搞混。它们解决的是同一类问题——把AI应用的执行过程编排成流程——但抽象层级完全不同。dify、coze、n8n是低代码/可视化平台你在画布上拖节点、连线、配置参数平台负责执行。好处是上手快、非技术同事也能用坏处是灵活性被平台限定自定义逻辑和嵌入自家系统都受限。LangGraph是代码级的工作流SDK图的逻辑全部由代码定义想在哪个环节插一段自己的Python逻辑、接私有的存储系统、做精细化控制完全不受限。我现在的分工是给客户做快速原型用coze/dify交付正式系统用LangGraph。前者帮我对齐需求后者保证可控性和可维护性。如果你要问LangGraph和dify那种商业平台谁更好答案取决于你的约束条件要快、不要代码选平台要深度定制、要嵌进自己产品选LangGraph。2. LangGraph的三块基石State、Node、Edge2.1 State流动的数据载体State是LangGraph工作流的数据中枢。它可以是TypedDict、dataclass或者Pydantic模型定义一个工作流里需要传递的全部字段。官方文档里最常见的是TypedDict简单直观。from typing import TypedDict class BlogState(TypedDict): topic: str # 输入技术选题 insight: str # 内容简报 outline: str # 文章大纲 draft: str # 初稿 validated: bool # 质量是否达标写State的时候有两个容易忽略的点。第一State的字段最好只表达这一轮工作流需要的数据不要把临时变量、调试信息全塞进去。它不只是内存对象在启用断点续跑之后会被序列化保存字段越多恢复成本越高。第二节点函数返回的是对State的增量更新不是让你直接修改传入的state对象。LangGraph约定每个节点把新数据以dict形式返回框架负责把返回值合并进全局State。这种设计是为了支持节点并行执行和状态快照恢复也是新手最容易写错的地方——后面第6章我会细说。2.2 Node最小执行单元Node就是一个普通Python函数输入State输出一个dict用于更新State。函数内部做什么都不限制调大模型、跑SQL、查接口、发邮件、算数都行。def outline_node(state: BlogState): # 这里调用大模型生成大纲 prompt f请为《{state[topic]}》写一份三级大纲 outline model.invoke(prompt).content return {outline: outline}节点函数建议保持单一职责一个节点只干一件事。以前在链式代码里我习惯把分析判断生成写进一个大函数换成LangGraph之后我会拆成分析节点判断路由生成节点三个独立部分。拆细的好处是执行历史清晰、单点好调试、条件边可以插在任何两个节点之间。图结构允许你以很细的粒度编排流程拆太粗反而浪费了这套抽象。2.3 Edge与Graph装配从零散函数到可执行图节点定义完需要用StateGraph把它们装起来。Edge分两种普通边和条件边。普通边表示无条件从A走到B条件边表示走哪条路由函数说了算。from langgraph.graph import StateGraph, START, END graph StateGraph(BlogState) graph.add_node(outline, outline_node) graph.add_node(draft, draft_node) graph.add_edge(START, outline) graph.add_edge(outline, draft) graph.add_edge(draft, END) app graph.compile()这里START和END是两个特殊节点代表流程的起点和终点。compile之后得到的app就是一个可执行对象用app.invoke({topic: ...})传入初始State就能跑。我第一眼看到这套API的时候觉得这不像写代码更像画流程图。你的思维方式要从调用一个函数得到结果切换成定义一张图然后让数据在上面流动。一旦切换过去编排复杂流程就变成了搭积木。注意LangGraph版本迭代很快早期某些写法比如用add_edge而不显式使用START/END在新版本里已经变了。你在网上搜到两年前的代码跑不通是正常的建议以官方文档的最新示例为准。3. 第一个可以跑的工作流技术选题生成器3.1 需求设计与状态定义说了这么多理论直接上一个能跑通的Demo。我选了一个我自己每天都在用的场景生成技术博客初稿。流程设计成四步分析选题生成内容简报检查这份简报判断是否值得写条件路由生成大纲写初稿并做基础质量校验不合格就回炉重写。状态定义沿用上一节的BlogState但需要加一个校验用的计数器防止死循环from typing import TypedDict, Optional class BlogState(TypedDict): topic: str insight: Optional[str] outline: Optional[str] draft: Optional[str] retry_count: int为什么字段要标Optional因为节点是按顺序执行的在流程早期outline和draft还不存在。运行时如果直接取state[outline]拿不存在的Key会报错。TypedDict的Optional只是类型提示但配合代码里的get方法能让流程前的字段访问更安全。3.2 核心节点实现这里我用一个mock_model代替真实的大模型调用方便你直接跑通逻辑。把它换成你的模型客户端代码骨架完全不用改。def mock_model(prompt: str) - str: # 模拟返回真实场景替换为 chat_model.invoke(prompt).content return f[模拟输出] {prompt[:30]}... def insight_node(state: BlogState): prompt f分析选题的读者群体、核心看点和潜在难点{state[topic]} insight mock_model(prompt) return {insight: insight} def outline_node(state: BlogState): prompt f基于以下内容简报生成大纲{state[insight]} outline mock_model(prompt) return {outline: outline} def draft_node(state: BlogState): prompt f基于以下大纲写初稿{state[outline]} draft mock_model(prompt) return {draft: draft}每个节点做的事都很纯粹读自己需要的字段处理返回要写入的字段。节点与节点之间通过State解耦谁在上游谁在下游只取决于图结构不取决于代码调用顺序。这意味着你想在中间插一个新的处理节点不影响已有节点。3.3 路由、循环与编译执行接下来加条件路由和质量校验。路由函数的输入是State输出是目标节点的名字。框架根据返回值决定下一步走到哪。def should_write(state: BlogState) - str: insight state.get(insight, ) if 值得写 in insight or len(insight) 50: return outline return rewrite def validate_draft(state: BlogState) - str: draft state.get(draft, ) # 简单规则内容太短或者质量不达标就重写 if len(draft) 100 and state.get(retry_count, 0) 2: return draft return end装配阶段把条件边接上from langgraph.graph import StateGraph, START, END graph StateGraph(BlogState) graph.add_node(insight, insight_node) graph.add_node(rewrite, insight_node) graph.add_node(outline, outline_node) graph.add_node(draft, draft_node) graph.add_edge(START, insight) graph.add_conditional_edges(insight, should_write, {outline: outline, rewrite: rewrite}) graph.add_edge(rewrite, insight) graph.add_edge(outline, draft) graph.add_conditional_edges(draft, validate_draft, {draft: draft, end: END}) app graph.compile() result app.invoke({topic: LangGraph入门, retry_count: 0})注意到graph.add_edge(rewrite, insight)这条边了吗它让流程能从rewrite节点回到insight节点形成一条回环——这是链式结构里最麻烦、图结构里最自然的操作。再加一个计数器上限就能控制回环次数state[retry_count] 1写在函数返回前。这个Demo虽然简单但它涵盖了状态图工作流的全部基本动作顺序执行insight→outline→draft、条件分支should_write、循环回退draft不达标→回到draft、终止条件validate_draft返回end。把这四个动作熟练掌握你已经能表达绝大多数工作流了。注意路由函数里返回的字符串必须是add_node时注册过的节点名或END。写错名字不会在编译时报错直到运行到这一步才抛异常排查起来要多看一眼日志。4. 条件路由和循环的精髓让工作流自主决策4.1 条件边路由函数决定下一步条件边是LangGraph最值钱的设计。它把决策逻辑从节点内部抽出来变成图上一个独立层次。怎么理解这个好处以前在节点内部写if-else决策逻辑是隐藏的外人看不到你的流程到底有哪些可能路径。把决策抽成路由函数之后图结构上一个节点可能通向的所有路径都变成显式可见的边。画图评审的时候每个分支一目了然。这不仅仅是代码风格问题它直接影响了多人协作时大家对这个系统到底会怎么走的理解成本。路由函数里可以写任意逻辑不限于规则判断。实际生产里我经常用大模型来判断调用模型返回走A分支还是B分支再映射成节点名。这让决策本身也可以被AI驱动。def route_by_model(state: BlogState) - str: prompt f阅读当前简报判断应当生成大纲还是补充资料。只回答一个词outline 或 research。\n{state[insight]} decision mock_model(prompt).strip().lower() if decision research: return research return outline把决策逻辑放在独立路由函数的另一个好处是路由可以被复用。同一个质量判断逻辑可以用在不同的子图之间而不用复制到各个节点里。4.2 循环、回退与终止条件循环是Agent类应用的核心能力。ReAct模式的本质就是思考→行动→观察→再思考的循环直到模型认为找到答案才退出。LangGraph里循环就是一个条件边把终点指回起点。循环设计时容易忽略的是终止条件。没有终止条件的循环在运行时可能无限跑下去大模型场景尤其危险——模型输出不稳定可能反复做同一个动作。我的经验是至少加两道保险设置最大迭代次数比如retry_count大于3不管结果如何直接走向END在路由函数里对每次返回做归一化和校验防止模型输出不可识别内容导致路由永远走同一分支。def max_retry_check(state: BlogState) - str: if state[retry_count] 3: return end return retry4.3 规划模式先列计划再执行最近常被问到的planning模式本质上也是状态图的一种编排方式第一个节点让大模型输出一份执行计划后续节点按照计划逐步执行每执行一步就把实际结果写回State最后再有一个节点校验计划完成度。这种模式和直接一步步硬编码的最大区别是计划不是固定在代码里的而是模型根据当前任务动态生成的。适合需求内容本身不确定、需要有探索余地的场景比如研究型任务、资料综述、复杂多步骤问题。在LangGraph里实现起来就是add一个plan节点 → 它调用模型输出计划的结构化文本 → 条件路由根据计划内容分发给各个执行节点。没什么黑魔法但表达力很强。5. 把人加进去断点、续跑与人工审核5.1 checkpointer与thread_id工作流不可能永远全自动尤其是面向真实业务的时候总得有个人工确认环节AI生成了初稿需要编辑看完确认才发出去。LangGraph的处理方式是给工作流加一个checkpointer检查点配合thread_id实现执行状态的持久化和恢复。from langgraph.checkpoint.memory import MemorySaver checkpointer MemorySaver() app graph.compile(checkpointercheckpointer) config {configurable: {thread_id: blog-001}} result app.invoke({topic: LangGraph入门, retry_count: 0}, configconfig)这里的thread_id相当于一次业务流程的身份证。同一份状态、同一个执行进度都会被记录在这个thread下。下次你再带着同一个thread_id调用框架能拿回之前的执行状态而不是从零开始。MemorySaver只把状态存在内存里进程重启就丢生产环境建议换成持久化存储的checkpointer比如用SQLite或PostgreSQL实现的版本。数据量一大或者要做跨实例断点恢复内存版肯定是不够的。5.2 interrupt实现暂停等待如果你希望在某个节点执行前暂停等待外部输入LangGraph提供了一个interrupt机制。它会在指定节点前停下把控制权交回给调用方等调用方传入补充数据后再继续。举个例子生成大纲之后我想让用户确认一下再写初稿。我只需在compile之后的执行配置里指定要中断的位置app graph.compile(checkpointercheckpointer) config {configurable: {thread_id: blog-001}} # 第一次调用执行到 outline 节点前暂停 graph.invoke({topic: ...}, config) # 拿到当前状态人工审核大纲内容 snapshot app.get_state(config) print(snapshot.values[outline]) # 确认没问题继续执行到完成 result app.invoke(None, config)这套机制在业务流程中非常有用比如财务审批、内容审核、人工纠偏。它在代码层面解决的是机器执行到一半怎么停下来问人的问题而不需要你手动把中间结果存到数据库、再另起一个进程恢复。以往我自己实现这种暂停等待要写不少胶水代码有了interrupt之后工作量少了一个数量级。5.3 恢复到断点的方法恢复的核心就是Thread checkpoint。注意几个细节invoke第二次时输入数据可以传None因为状态已经从断点恢复了不需要从头传入你可以用get_state查看当前节点、待处理节点、State数据快照配合update_state可以在恢复前修正某个字段如果你用的是内存checkpointer服务一重启所有中途停下来的流程都找不回来了。真实项目里至少用文件级或数据库级持久化别图省事。我目前的习惯是所有涉及人工审核的节点都放在需要中断的位置然后配合thread_id把每个流程串起来。前端展示的执行进度就直接读get_state的返回值后端恢复执行就再invoke一次两边各干各的互不干扰。6. 实战中的坑和选型建议6.1 五个我亲自踩过的坑这五个坑是我从初学到落地过程中真实遇到过的希望能帮后面的人少走弯路。第一个坑节点函数原地修改State。我一开始在节点里直接写state[draft] xxx结果下游节点拿不到更新。因为LangGraph的State默认是不可变更新的节点返回值才用于生成新的State版本。正确写法始终是return {draft: xxx}。第二个坑路由函数的返回值和add_node注册名不一致。注册的节点名叫generate_draft路由函数里手滑写成了draft_gen编译不报错跑到一半报错。排查这类问题没有捷径建议把所有节点名集中在一个常量文件里别散落在各个函数里用魔法字符串。第三个坑忽略checkpointer状态序列化。State里放了不可序列化的对象比如数据库连接、OpenAI客户端对象一旦启用checkpointer就会报错。凡是需要跨节点共享的重资源要么惰性初始化要么只存配置信息不存实例。第四个坑条件边过多导致图结构难维护。这是我自己在项目里切身体会的——图能表达一切但不代表你应该什么都用条件边。曾把十来个业务分支全部路由到一个父节点后来发现排查路径极难。设计阶段优先级应该是线性边优先于条件边条件边优先于在节点里写大if-else。第五个坑版本API变化。LangGraph从0.1到0.4断点API、checkpointer的接口有过不少调整。很多网上的教程代码直接跑会报错。看文档一定选latest版本跑示例代码遇到异常优先查Release Notes而不是改环境。6.2 LangGraph和LangChain怎么分工这两兄弟经常被放在一起问面试题。一句话结论LangChain负责模型和工具的能力层LangGraph负责流程和状态的执行层。你在LangChain里定义用什么模型、怎么加载文档、怎么调用工具、怎么解析输出在LangGraph里则专注于编排几步、什么顺序、什么条件下走哪条分支。实际项目里两者配合使用LangGraph的节点函数内部调用LangChain的ChatOpenAI、Tool、DocumentLoader等等。LangGraph也提供了预置的Agent构建器比如create_react_agent适合想要快速跑一个ReAct Agent、不想手动写循环逻辑的人。但我自己更推荐从StateGraph手动定义因为可控制性更强且你一旦理解了状态图手动定义的成本其实不高。6.3 什么场景别用LangGraph有句话叫拿着锤子看什么都是钉子写技术方案也一样。不是所有工作流都适合LangGraph。如果你的流程就是一两步调用用LangChain的Runnable链就够了引入图结构反而笨重。如果你们团队全是非技术背景你只是想搭一个可视化流程给业务同事用那dify、coze、n8n更合适。如果流程极其简单且不需要状态持久化和人工介入硬上LangGraph就是过度设计。我判断是否用LangGraph的标准只有三条有没有分支循环需不需要中途暂停和状态恢复流程是不是会随着业务演进不断变复杂三条里占两条就值得用不然就别给自己找事。个人体会是状态图这个抽象并不复杂复杂的是你愿不愿意换个思路组织代码。当你真正把一个多分支、多回退、带人工审核的流程画成一张图并跑起来的时候会明显感觉到代码可控性上了一个台阶。接下来你可以试着把自己手头最乱的那条流程画出来从纸笔到代码跑通后还会想重构更多流程。
