做Agent开发的人大概率都遇到过同一个烦恼业务逻辑越写越复杂if-else越叠越深模型调用、工具调用、条件判断全搅在一起。一开始还能靠一个十几行的chain撑住等需求一多代码就开始变成意大利面条。我自己的转折点是从硬编码流程切到LangGraph用状态图的方式重新组织整个工作流结构才真正清晰起来。这篇文章就来聊聊怎么从零开始用LangGraph基于状态图把工作流搭起来。LangGraph到底解决什么问题简单说它是一个面向LLM应用的工作流编排框架把复杂的Agent流程建模成一张有向图。图上有节点、有边、有状态流程怎么走、数据怎么传、什么时候该停下等人介入全都一目了然。如果你已经在用LangChain写链式调用或者觉得Dify、Coze这类图形化平台不够灵活想用代码精确控制每个环节LangGraph就是更适合的工具。1. 为什么用LangGraph而不是硬编码工作流1.1 LangGraph与LangChain的区别很多人第一次接触LangGraph时都会困惑它和LangChain到底是什么关系。我的理解是这样的LangChain是组件库提供各种大模型封装、提示词模板、检索器、工具调用等基础模块而LangGraph是编排引擎解决的是这些模块怎么串成一个复杂流程的问题。如果把Agent比作一家公司LangChain是各个业务部门的职能组件LangGraph就是公司的组织架构和业务流程——谁先干活、谁后干活、什么条件走哪条路。更本质的区别在数据流上。LangChain早期的LCELLangChain Expression Language是一条单向流水线A结果传给BB结果传给C没法往回走也没法根据中间结果反复循环。而LangGraph是基于状态图的节点之间可以有分支、可以有循环甚至可以根据运行中的结果动态决定下一步去哪个节点。Agentic应用里最常见的ReAct模式——思考、行动、观察、再思考——就天然需要一个循环结构用LCEL硬写会非常别扭用LangGraph则顺理成章。顺带说一句如果你用过Dify或者Coze这类低代码平台它们的工作流画布也是同一个思路拖拽节点、连线、配条件。LangGraph相当于把这个能力下沉到代码层面好处是版本可控、逻辑可测、可以嵌入你自己的工程体系。坏处嘛就是你得自己画图、自己排查没有图形界面那种所见即所得。1.2 状态图的思维转变初次上手LangGraph最需要转变的一个认知是不要再用一段从上到下执行的代码来思考流程而是用一张图加一份全局状态来思考流程。所有被建模为工作流的问题都可以拆成三个要素状态State整个工作流共享的数据结构比如输入文本、中间结果、最终答案都挂在状态上。每个节点执行完就更新状态下一个节点读到的就是最新版。节点Node一个普通的Python函数接收当前状态处理后返回新的字段更新。节点只负责一件事输入状态、输出增量。边Edge节点之间的连接关系。普通边是无条件的顺序传递条件边则根据节点返回值或者状态字段动态决定下一步去哪个节点。我用一个日常例子说明。快递分拣系统拿到一个包裹状态里的输入数据先扫描条码读取目的地节点A然后根据地区判断路由条件边同城送对应分拣口跨省送对应分拣口特殊件走人工通道节点B/C。这个流程如果写成if-else当然也能跑但每加一种包裹类型就要改一改主流程代码。用状态图建模新增的只是节点和一条边主流程不变系统整体可维护性完全不同。这背后其实是一个工程哲学的问题显式建模控制流而不是把控制流隐式地埋在代码分支里。状态图把你所有的业务分支、循环、终态都摆在明面上天然适合做可视化、权限控制和断点续跑这也是它被越来越多Agent项目选中的根本原因。2. 环境准备与状态设计2.1 安装LangGraph与基础配置安装非常简单直接用pip装就好。我建议顺手把langchain-openai也装上因为LangGraph生态里很多模型调用都是基于LangChain的ChatModel抽象。如果后面要接真实的大模型还需要准备对应的API Key如果暂时不打算花钱调API完全可以先用一个模拟模型跑流程下面的示例也是这么设计的。建议的Python版本是3.9以上太老的版本在类型标注和语法上可能会有兼容问题。pip install langgraph langchain-openai装完可以验证一下版本LangGraph目前的版本迭代比较快API有细微调整遇到网上教程和你的版本对不上时第一反应应该是查官方文档而不是怀疑自己写错了。2.2 设计工作流的状态Schema状态设计是整个工作流的灵魂几乎所有后续踩的坑都源自早期状态设计不够清楚。在LangGraph里状态通常用TypedDict定义每个字段代表工作流中的一个数据项。以我下面要做的内容审核与优化工作流为例一个完整的状态Schema大概长这样from typing import TypedDict class ReviewState(TypedDict): article: str # 待审核的文章原文 score: int # 质量评分0-100 category: str # 内容分类 needs_human: bool # 是否需要人工介入 final_version: str # 最终输出版本每个字段都要想清楚两个问题这个数据从哪里来会被哪个节点消费。比如score字段初始值可能不存在是check_quality节点算出来再写进去的。LangGraph允许节点只更新部分字段所以不需要在一开始就填满所有值。但有一个坑要注意如果你的节点返回了Schema里不存在的keyLangGraph会直接报schema validation错误。这个约束其实很友好它逼着你在动手写代码之前就把数据结构定好省得后面流程跑通了才发现字段名字不统一。2.3 没有大模型Key也能跑的模拟方案我不建议一上来就急着接真实的大模型因为工作流本身的调试已经够复杂如果再叠加API限流、Key失效、网络波动这些外部因素排查问题会非常痛苦。先写一个模拟的质量检查器返回值完全由规则决定这样整个图的逻辑可以先跑通后面随时可以替换成真正的模型调用。def check_quality(state: ReviewState) - dict: article state[article] length_score min(len(article) // 50, 50) # 长度评分满分50 keyword_score 30 if LangGraph in article else 10 # 关键词评分 score length_score keyword_score category tech if LangGraph in article else general needs_human 30 score 60 return { score: score, category: category, needs_human: needs_human, }这种工作流逻辑与模型能力解耦的做法是我实测下来最稳妥的入门路径。先保证图的拓扑没问题再逐步把规则函数换成真实模型调用每一步都能定位到具体问题而不是一锅粥。3. 构建第一个状态图工作流内容审核与优化3.1 核心节点的定义与职责划分节点就是一个普通函数输入是当前完整状态输出是一个字典字典里的key会合并进状态。理解这个合并机制很重要节点不需要返回所有字段只返回自己负责更新的那部分即可。在我这个内容审核工作流里一共设计三个核心节点def rewrite(state: ReviewState) - dict: # 模拟借助模型优化内容 improved state[article].replace(LangGraph, LangGraph状态图工作流) return {final_version: improved \n已完成自动优化} def human_review(state: ReviewState) - dict: # 模拟人工审核实际项目中这块会配合中断机制 return {final_version: state[article] \n已通过人工审核}每个节点职责单一。check_quality只管打分和分类rewrite只管优化文本不要在一个节点里既做判断又做改写否则后面排查问题的时候会很痛苦。这也是状态图工作流最核心的设计纪律节点要小职责要纯边上的路由逻辑才清晰。3.2 组装状态图与条件路由节点定义好之后就可以开始搭图了。这里我用到条件路由的核心APIadd_conditional_edges。它接收三个参数起始节点名、路由函数、路径映射表。路由函数根据当前状态返回一个字符串key这个key决定去哪个节点。from langgraph.graph import StateGraph, START, END def route_after_check(state: ReviewState) - str: if state[score] 60: return publish elif state[needs_human]: return review else: return rewrite builder StateGraph(ReviewState) builder.add_node(check_quality, check_quality) builder.add_node(rewrite, rewrite) builder.add_node(human_review, human_review) builder.add_edge(START, check_quality) builder.add_conditional_edges( check_quality, route_after_check, {publish: END, review: human_review, rewrite: rewrite}, ) builder.add_edge(rewrite, END) builder.add_edge(human_review, END) graph builder.compile()这里面的publish映射到END表示流程结束review和rewrite分别指向人工审核和自动优化节点。整个图的逻辑一眼就能看懂进来先检查质量高分直接发布中等分走人工低分自动优化后就结束。这个映射表我建议显式写清楚因为LangGraph在运行时如果路由函数返回了映射表里没有的key会直接报错。所以路由函数返回什么、映射表里配了什么必须一一对应少一个都不行。3.3 运行工作流与查看执行结果图编译好之后用invoke方法传入初始状态就能跑起来。注意初始状态不需要填满所有字段只需要填入你有的数据节点里会自动把计算出来的字段合并进去。result graph.invoke({ article: LangGraph是一个基于状态图的工作流编排框架适合构建复杂的AI Agent流程。, score: 0, category: , needs_human: False, final_version: , }) print(result[final_version])第一次跑通的时候建议去打印完整的result看看每个字段最终的值是什么。我踩过的坑是因为只关注final_version忽略了score和needs_human导致排查问题时没法判断分支到底有没有按预期走。完整打印能帮你确认这条输入到底走了哪个分支哪个节点改了哪些字段状态最终长什么样。如果想让调试更直观可以改用stream方法逐步观察for event in graph.stream({ article: LangGraph是一个基于状态图的工作流编排框架。, score: 0, category: , needs_human: False, final_version: , }): print(event)stream的输出会按超步分组你可以清楚地看到每个节点执行完后的状态快照。这在排查哪个节点没按预期更新状态时几乎是神器。4. 状态持久化、多轮记忆与人工介入4.1 用Checkpointer实现跨轮状态保存上面搭建的图每次invoke都是独立运行的状态用完即弃。但在很多真实业务里工作流需要跨多轮交互保留记忆或者跑了一半需要暂停等人确认之后再继续。这就轮到LangGraph的持久化机制登场了。核心概念是Checkpointer负责把每一步的状态快照保存下来。配合thread_id可以让多个invoke调用共享同一份状态组成一个完整的会话from langgraph.checkpoint.memory import InMemorySaver checkpointer InMemorySaver() graph builder.compile(checkpointercheckpointer) config {configurable: {thread_id: review-0001}} result1 graph.invoke({article: LangGraph入门指南基于状态图构建工作流。, score: 0, category: , needs_human: False, final_version: }, config) result2 graph.invoke({article: 这是第二轮补充输入。, score: 0, category: , needs_human: False, final_version: }, config)这里有两个细节。第一thread_id是一个纯字符串你可以用它关联业务实体编号比如订单号、用户ID第二InMemorySaver是内存保存进程重启就没了生产环境需要用持久化存储的实现比如SqliteSaver或者PostgresSaver。第一次用checkpointer时我犯过一个错明明加了checkpointer第二次invoke还在问为什么状态没保存结果发现是忘了传config。记住加了checkpointer之后每次invoke都必须带上config否则每轮仍然是独立状态。4.2 用interrupt实现人工审核与断点恢复如果说条件路由是工作流自动流转的发动机那interrupt就是人工介入的刹车踏板。很多业务场景都要求流程中途停下来等一个人拍板——比如内容审核发现敏感内容需要人工确认、订单金额异常需要人工审批、AI生成的代码需要开发人员Review。from langgraph.types import interrupt from langgraph.types import Command def human_review_node(state: ReviewState) - dict: decision interrupt({ article: state[article], score: state[score], }) return {final_version: decision.get(final_version, state[article])}工作流执行到这个节点时会暂停并抛出一个中断信号把数据暴露给外部系统。外部处理完成后用Command(resume...)把结果塞回去工作流接着往下跑# 外部系统拿到中断数据后人工确认结果 result graph.invoke(Command(resume{ final_version: state_article \n人工已确认发布 }), config)这个模式在LangGraph里叫Human-in-the-loop在内网审批流程、内容审核、客服工单流转这些场景里特别香。比起自己写一套存草稿、开个HTTP接口等人回调的方案用interrupt省掉了大量胶水代码而且状态天然连续不会出现流程走一半就丢了上下文的情况。4.3 状态Reducer并行节点与消息累加的基石工作流一旦复杂起来就会遇到一个基础问题多个节点往同一个状态字段写值到底以谁为准LangGraph的答案是Reducer——每个字段都可以配置一个自定义合并策略。LangGraph内置了一个很实用的消息Reduceradd_messages。它会把两条消息按时间顺序追加成一个消息列表而不是简单覆盖。这在多轮对话Agent里是刚需因为状态里的messages字段需要累积每一轮的提问和回答。from typing import Annotated from langgraph.graph.message import add_messages class ChatState(TypedDict): messages: Annotated[list, add_messages]如果你有自己特定的合并逻辑也可以写自定义Reducer。比如要记录所有节点产生的事件日志def merge_events(left: list, right: list) - list: return left right class WorkflowState(TypedDict): article: str events: Annotated[list, merge_events]没有配置Reducer的普通字段默认行为是后写的覆盖先写的。这在顺序流程里没问题但一旦遇到并行节点两个节点同时往同一个key写值没有Reducer就会直接报错因为LangGraph无法决定以谁为准。这是从入门到进阶最常见的坎之一理解了Reducer你对状态如何被更新的理解就真正到位了。5. 从入门到落地调试技巧与常见问题排查5.1 学会观察工作流的每一步调试LangGraph工作流最大的难点在于它不是一个直线函数调用链而是一张图。你看得见的只是最终的invoke返回值看不见中间的流转过程。所以我强烈建议从一开始就养成用stream和get_state调试的习惯。除了前面提到的stream还有一个隐藏利器是graph.get_state_history它可以回放当前thread的全部历史状态快照适合回答这个流程到底执行了多少步这类问题。配合结构化日志比如每个节点进来时打印状态关键字段、出去时打印修改内容定位问题会快很多。提示实际项目中别用print调试尤其在异步或多线程环境下输出顺序会乱。更稳的做法是日志加上thread_id和节点名这样排查问题时能按线程维度过滤。5.2 高频报错场景与排查速查下面这张表是入门阶段几乎必遇问题的汇总我整理成了速查格式希望对你有帮助。报错信息或现象常见原因排查与解决RecursionLimitExceeded条件路由配置成了环且没有恰当的终止条件检查条件路由的映射关系确保最终态能到达ENDInvalidUpdateError节点返回了Schema中不存在的key逐字段比对TypedDict定义与节点返回值控制流没按预期走分支路由函数的返回值与映射表key不匹配完整打印状态确认路由函数拿到的字段值是否符合预期并行节点写同一key报错该字段没有配置Reducer合并策略给这个key配置自定义Reducer或者拆分成不同key分别返回多次invoke后状态不连续忘记传thread_id对应的config确认每次调用都带上同一个config对象节点一直走不到END某个分支漏了add_edge到END画一张节点的转移图逐条核对边的连接关系5.3 一个真实的踩坑案例我自己在把LangGraph应用到一个简历筛选工作流时踩过一个特别容易复现的坑。初始版本的条件路由设计成初筛通过 - 进入深度评估 - 深度评估通过 - 发送通知看起来没什么问题。结果跑了一批样本之后发现有相当比例的简历在深度评估阶段反复绕圈代码报RecursionLimitExceeded。排查半天才发现问题出在深度评估节点里有个小bug有些边缘case没有更新状态的evaluation_result字段导致同一个结果反复从条件路由走回评估节点形成了死循环。这暴露了一个状态图工作流中特别底层的问题——条件路由的终止条件本质上依赖某个状态字段的值如果这个字段没有被按预期更新图就会在环里出不来。后来我的解决方案是在每个节点入口做一个断言式校验比如评估节点进入前先检查该简历是否已经在评估中、是否有过初筛记录如果状态异常就直接路由到人工处理节点不让它无限循环。这也是工作流工程比普通脚本工程更需要注意的一点加防御式检查宁可让流程提前终态也不要在环里空转。写在最后的实践心得整套LangGraph用下来我的最大感受是它逼着你在写业务之前先把流程想清楚。状态Schema是什么、节点有哪些、分支怎么走、什么时候需要人工介入这些在绘制状态图的时候就必须敲定而不是边写代码边拍脑袋。这看起来增加了前期设计成本但带来的回报是后期几乎不用重构流程骨架新需求进来只需要加节点和边。另外就是节奏问题。我第一次接触LangGraph时贪多一上来就想写一个带并行分支、带多Agent协作的大图结果光排查循环和状态覆盖就花了两天。后来调整策略每次都从最小闭环跑起——单个节点、一条链路、一个分支跑通了再往上加复杂功能。有条不紊地增量比一次性搭建大而全要高效得多。LangGraph目前还在快速迭代中如果你看完这篇准备上手我的建议很直接把你手头最简单的一个自动化流程拿过来改造哪怕是读一封邮件 - 提取要点 - 生成回复这种三步流程用状态图画一遍收获都会来得非常快。后面如果要往更复杂的方向走并行分支、子图复用、多Agent协作这些都是在这个基础上自然延伸出来的能力。