ARTICLE DETAIL

资讯详情

深耕网站建设、视觉设计与SEO优化的一线实战洞察。

LangGraph入门到实战:用状态机构建可控的LLM工作流

LangGraph入门到实战:用状态机构建可控的LLM工作流 像 LangGraph 这类工具最容易劝退人的地方不是它本身多难而是很多人一上来就搜“最全最细语言图教程”结果收藏了一堆资料却连一个最小可运行流程都没跑起来。LangGraph 本质上是一个用于构建 LLM 应用工作流的编排框架。它解决的真正问题是让多个 LLM 调用、工具调用、条件判断和循环逻辑以图结构的方式组织起来而不是在一个 Prompt 里硬塞所有逻辑。这篇文章适合三类人看已经接触过 LangChain、想进一步控制流程的用普通 Chain 写复杂业务时经常被分支和循环搞到崩溃的以及想在生产环境里让 LLM 应用可维护、可恢复、可追踪的人。最值得关注的能力我会先放在前面LangGraph 不是模型不是 Prompt 工具而是一套能让流程图真正落地的状态机框架。下面按我的实际踩坑顺序把入门到项目实战的路径拆开讲。1. 先搞清楚 LangGraph 和 LangChain 的区别避免方向跑偏1.1 不是又一个模型而是把流程变成可运行图很多刚接触 LangGraph 的人会误以为它像 Ollama 或 vLLM 一样是一个模型运行工具。实际上它不负责推理它负责的是“编排”。说得直白些你仍然需要调用某个大语言模型的接口或本地模型但多个调用之间怎么衔接、什么时候调用工具、调用结果往哪里放这些流程逻辑由 LangGraph 管理。它把一次任务拆成节点和边。节点可以是一个 LLM 调用、一个 Python 函数、一个数据库操作或者一个工具调用边就是节点之间的连接关系。这种设计最大的好处是流程不再是一长串隐式代码而是可以被观察、被暂停、被恢复、被分支、被循环的显式结构。在我实际使用中LangGraph 最值得先理解的概念有三个State、Node、Edge。State 是全局数据载体Node 是执行单元Edge 决定下一个执行单元。当你只是写单次 Prompt 调用时这三个概念完全用不上一旦你的应用需要多轮对话、需要判断后走不同分支、需要循环处理直到满足条件它们就是核心。1.2 LangChain 和 LangGraph 的关系链是图的一个特例LangChain 里最经典的概念是 Chain也就是把一个 Prompt 模板传给模型再把结果传下一个环节。这种串行结构在简单场景下很好用但复杂业务流程马上会遇到问题如果第二步结果不满足条件要回到第一步怎么写如果多个模型调用需要并行执行且最终汇合怎么写如果流程执行到一半崩溃了如何恢复LangGraph 提供的是更通用的图结构。你可以把一条 Chain 理解成没有分支、没有循环、只有一条路径的图。真正需要上 LangGraph 的项目通常至少满足以下一个特征流程有分支、流程有循环、需要持久化状态、需要人工介入节点、需要细粒度控制并行。我见过一个典型的错误案例项目中只有两个串行 LLM 调用却硬用 LangGraph 画了五个节点结果代码量没减少反而增加了状态管理的复杂度。串行任务用 Chain 或普通函数就能解决LangGraph 不会让简单链路变快它解决的是复杂链路的可控性。1.3 适合与不适合的场景先判断再选型不适合 LangGraph 的场景包括一次性文本生成、简单问答、单一 Prompt 调用、两个模型的固定串联。这些场景用普通函数或 LangChain 的 Chain 更直接不需要引入图结构。适合的场景则更偏向真实业务系统客服机器人需要多轮对话并要根据对话历史决定是否查询订单、是否转人工、是否推荐商品。智能写作工作流先写大纲、再写章节、然后检查合规、不合格要重写。数据处理流水线读取文件后判断类型不同类型走不同解析路径最后统一输出。Agent 应用需要循环调用工具直到获得最终答案工具调用失败要重试或换工具。我个人的判断标准是如果流程能用一张流程图完整画出来而且图中有菱形判断框或回路箭头LangGraph 就值得用如果只是从上到下三个方框就先别折腾了。2. 环境准备与最小 Demo先用最少代码跑通图结构2.1 安装依赖与版本确认LangGraph 的安装非常简单Python 3.9 以上环境执行pip install langgraph即可。如果你还需要调用 OpenAI、Anthropic 或其他模型需要一起安装对应 SDK。实际项目中还会用到 LangChain 的模型封装比如langchain-openai所以建议一次性安装下面这类依赖pip install langgraph langchain-core langchain-openai这里有两个容易踩坑的地方。第一langgraph 和 langchain 的版本不是完全绑定的新版本 langgraph 有时候会依赖更新的 langchain-core所以最好用虚拟环境安装避免和旧项目冲突。第二如果你只是学习不需要立刻配置 API Key可以先用一个返回固定文本的模拟节点跑通流程验证的是图结构本身而不是模型调用。2.2 最小图两个节点一条边我从一个最小图开始实验。第一个节点负责生成一个数字第二个节点负责把数字加一。虽然这个例子没有任何业务价值但它能帮助你理解 State 如何流动。from typing import TypedDict from langgraph.graph import StateGraph, END class MyState(TypedDict): count: int def increment(state: MyState) - dict: return {count: state[count] 1} graph StateGraph(MyState) graph.add_node(increment, increment) graph.set_entry_point(increment) graph.add_edge(increment, END) app graph.compile() result app.invoke({count: 0}) print(result)这段代码里最关键的部分是节点函数的返回值。注意节点函数返回的不是整个 State而是一个部分更新LangGraph 会自动把返回值合并到当前 State 中。执行结果是{count: 1}。我建议你把这段代码跑起来然后重复执行几次观察 State 的变化。一旦理解了“节点返回什么State 就更新什么”后面学条件路由和循环都会快很多。2.3 验证方法看 State、看结果、看流程图很多人跑完一次就结束了其实还应该做两个验证。第一看最终结果是否符合预期第二看这个图是不是按照你设计的顺序执行的。LangGraph 在编译后可以生成 Mermaid 图结构。虽然文章里不展开 Mermaid但你可以自己在 Notebook 里查看流程确认节点和边的连接关系。这个习惯非常重要因为随着图变大很多人忘了自己在哪一步加了边最终导致流程在中间卡住或循环。执行顺序的验证还可以通过一个简单技巧每个节点函数里打印一行日志。比如print(enter increment)。这种方式虽然原始但能很快判断图是不是按照预期拓扑顺序执行。3. 核心概念State、Node、Edge 和 conditional_edge3.1 State 是唯一的数据源不要偷偷存全局变量LangGraph 设计里State 就是整个流程的唯一事实来源。每个节点从 state 里读数据通过返回值更新 state。这个设计非常符合函数式编程的优点节点内部无副作用流程可重放、可恢复、可测试。但我在实际项目里常看到问题有些开发者在节点里定义全局变量或者把结果写入类属性以为这样更高效。短期看没报错一旦需要暂停恢复或者并发执行数据就乱了。LangGraph 的 checkpointer 会把 State 保存下来如果你的数据散落在全局变量里恢复后是拿不回来的。因此第一条规则是所有要跨节点使用的数据都必须放进 State。临时变量、中间结果、模型返回的 token 数、工具返回的错误信息只要下游节点要用就放进 State。3.2 Node 和 Edge 的关系Node 干活Edge 决定走向每个 Node 都是可重入函数接收当前 State返回部分更新。Edge 决定下一个节点是哪个。这种关系理解起来不复杂但很多人容易忽略一点Node 本身不应该关心下一个节点是谁。它只需要把结果写进 State具体下一步由边来决定。这样能保证节点可以被复用比如同一个“检查输入合法性”节点可以被三个不同前置节点连接。在实际项目中我发现一个很好的设计模式把判断逻辑写在节点函数里把路由决定写在边函数里。节点只负责计算结果比如返回一个quality_score: 0.85边函数读取这个分数决定是走“接受并结束”还是“打回重写”。这样改判断标准时只改边函数不需要动核心业务节点。3.3 conditional_edge 深度解析不是 if 语句是状态路由表条件路由可能是 LangGraph 里最容易被低估的功能。很多人把它理解成“if-else”但它的真实能力更像一个路由表可以根据当前 State 动态决定下一个节点。def quality_router(state: MyState) - str: if state[quality_score] 0.8: return accept return rewrite graph.add_conditional_edges( quality_check, quality_router, { accept: finalize, rewrite: rewrite_node, } )这里的quality_router函数返回的不是布尔值而是一个路由名称然后用路由名称去映射真实节点名。这个设计的好处是你不必在节点内部耦合“如果分数高就结束如果低就重写”这种业务逻辑而是把策略集中到边定义里。实际使用中需要留意两点。第一路由函数必须返回已注册的 key否则会报错第二条件路由可以指向自身这就构成了循环。循环是 LangGraph 的核心能力但也是新手最容易翻车的地方后面我会专门讲循环检测。3.4 在节点函数中修改 State 的注意事项热搜词里有一个问题很典型如何在节点函数改变 state 状态值。答案是直接返回一个新的字典对应要更新的字段即可。但这里有几个细节要记住。第一如果要删除某个字段只返回None是不够的通常需要在 Field 配置里设置删除策略或者使用专门的处理方式。第二不要试图在节点内部修改传入的 state 对象本身比如state[count] 1虽然有些情况能工作但可能破坏 LangGraph 的状态追踪机制。正确的做法是返回更新后的字段。第三节点函数可以是异步函数异步节点的返回方式与同步节点一致但调用方式要使用ainvoke或astream。async def async_increment(state: MyState) - dict: return {count: state[count] 2}这三点覆盖了我见过的大多数 State 修改问题。核心思路始终是State 是不可变更新不是原地修改。4. 条件路由、循环检测、子图与并行分支实战4.1 条件路由的实际场景用 AI 决定下一个节点上面讲的quality_router是普通函数做路由。实际项目中更常见的是让模型参与判断。比如在客服机器人里系统先判断用户意图再决定下一步走“查订单”“推荐商品”还是“转人工”。这个场景下你可以在一个节点里调用 LLM让它输出一个结构化意图标签然后边的路由函数读取这个标签。LangGraph 与 LangChain 的模型输出解析器配合很顺畅可以让模型输出 JSON再用 Pydantic 校验。校验失败时返回一个 retry 标签让流程回到重新解析节点而不是直接崩溃。这个模式非常强大。它把本来写在业务代码里的大量 if-else 交给模型判断同时保留代码层面的兜底逻辑。注意一个边界模型判断并不 100% 可靠所以 AI 路由节点后面一定要有校验节点校验不通过就进入重试或人工分支。4.2 循环检测避免流程卡死在死循环里LangGraph 支持循环路由因为很多 Agent 场景需要反复调用工具。但如果循环条件永远无法满足流程就会一直跑下去直到 token 用尽或超时。所以循环检测是生产级应用必须考虑的事。我常用的做法有两种。第一种在 State 里设置一个计数器字段每次进入循环节点就加一路由函数检查计数是否达到上限超限后强制走退出生分支。def loop_router(state: MyState) - str: if state[loop_count] 3: return exit return continue第二种使用 LangGraph 自带的递归限制参数比如在 invoke 时设置recursion_limit。这个参数会限制图中节点执行的最大次数防止意外死循环导致任务无限耗尽资源。注意不要只依赖 recursion_limit。大型图里正常执行也可能触发较高节点次数真正要做的还是把业务级的循环控制写进 State 和路由函数。4.3 子图把复杂流程拆成独立单元当图变大后所有节点塞在一张图里会变得难以阅读也难以复用。LangGraph 支持子图也就是把一个子流程图作为一个节点嵌入到父图里。我举一个实际例子。一个“内容发布”系统包含“写文章”“合规检查”“发布”三个大模块。其中“写文章”模块又包含“生成大纲”“写初稿”“改写润色”三个小节点。你可以把这三个小节点封装成一个子图然后在主图里只添加一个“写文章”节点。子图的好处不仅在于可读性还在于可以做局部调试。你可以单独编译并调用子图传入初始状态验证子图内部逻辑再接入主图。这样排查问题时不需要每次跑完整流程。子图的时间线控制也灵活一些。有时候你希望子图内部节点执行后只返回部分字段给父图这时候在子图编译时配置输入输出字段映射即可。但映射关系会增加复杂度我的建议是除非父图和子图的状态结构差异很大否则保持一致的 State 字段减少转换成本。4.4 并行分支什么时候用什么时候不用并行分支是 LangGraph 一个看起来很诱人的功能。实际构造也容易比如从一个节点出发连接两个不同的节点LangGraph 会自动并行执行。但并行不是免费的。第一它会同时发起多个模型调用Token 消耗会翻倍第二如果两个并行分支修改同一个 State 字段最后一个写入会覆盖前面一个容易产生竞态问题。我的建议是并行分支尽量让每个分支只负责不同的字段比如一个分支查询天气另一个分支查询交通最后合并到一个汇总节点。另外不是所有节点都适合异步。如果两个分支里有一个已经很快另一个很慢整体耗时其实取决于慢分支。并行优化不会让慢任务变快只能掩盖部分延迟。所以在设计并行分支前先问自己这两个任务真的能并发吗还只是我看起来该并发。5. 长期记忆与聊天记忆不重启就记住重启也能记住5.1 短期记忆用 checkpoint 保存状态LangGraph 有一个常被拿来说事的能力记忆。但记忆分两种很多人没有区分清楚。第一种是短期记忆也就是对话过程中需要记住前面几轮的上下文。LangGraph 通过 checkpointer 实现这个能力。你可以在编译图时传入一个 checkpointer 对象比如内存版或数据库版。每次调用 invoke 时都要传入一个线程 IDLangGraph 根据线程 ID 保存和恢复状态。from langgraph.checkpoint.memory import InMemorySaver memory InMemorySaver() app graph.compile(checkpointermemory) result1 app.invoke({messages: [(user, 我叫小李)]}, config{configurable: {thread_id: user_001}}) result2 app.invoke({messages: [(user, 我叫什么名字)]}, config{configurable: {thread_id: user_001}})第二次调用时LangGraph 会从 checkpointer 读取这个用户的旧状态所以模型能看到“我叫小李”这条历史消息。这里最容易踩的坑是忘记传thread_id导致每次调用都被看作新会话。我在学习时建议把 InMemorySaver 跑通即可它用于测试够用。生产环境则要根据吞吐和持久化要求换用数据库实现不展开细讲。5.2 长期记忆两种常见做法长期记忆解决的是跨会话记忆也就是用户下次登录系统仍然记得他的偏好。LangGraph 的长期记忆功能在不同版本里的接口差异不小所以我先不写具体代码只说思路。常见做法是把需要记住的信息写入外部存储比如数据库或向量库。在图的执行流程中增加一个“记忆提取”节点和一个“记忆更新”节点。开始时从存储中读取与该用户相关的记忆注入到 Prompt结束时把新出现的用户偏好更新回存储。这种做法的关键点是哪些信息值得长期记住我建议不要尝试把所有对话都存下来而是要显式提取。比如用户说“我喜欢简洁风格”这就是一条值得长期保存的偏好用户说“今天天气不错”这句话没有长期保存价值。用一个 LLM 节点专门做记忆抽取比直接存原始对话更可控。5.3 参数和配置需要注意的地方记忆相关配置里我遇到过两个典型问题。第一checkpointer 与 State 结构强相关如果改了 State 字段但旧数据还在数据库里加载新会话时可能出现字段缺失。所以生产环境升级时需要考虑 State 版本管理。第二通过线程 ID 恢复历史对话不代表所有历史消息都会自动进入模型上下文。你仍然需要在节点里决定如何组织消息列表比如只取最近十轮或摘要历史。LangGraph 提供了一些内存消息管理工具但核心逻辑还是要自己写清楚阈值和摘要策略。如果只是学习别一上来就上外部存储。先用 InMemorySaver 理解“线程 ID 状态恢复”这个机制再考虑跨进程持久化。6. 批量任务与生产化不能只看能不能跑6.1 单任务跑通之后先做三件事很多人把 Demo 跑通之后就直接开始写批量任务结果在批量场景遇到大量奇怪问题。我建议做三件准备工作。第一用同一个输入连续跑三次确认结果的一致性。LLM 本身有随机性但如果结果差异巨大说明流程中缺少温度参数控制或逻辑不够稳定。第二写一个最小失败样例比如故意传一个格式错误的数据确认图会报错还是返回一个错误标记。第三确认输出目录或返回结构是否能明确对应到输入任务。做完这三件事再开批量任务会顺很多。否则你排查的第一个问题往往不是算法或图结构而是“这条数据跑完输出到哪去了”。6.2 失败重试与任务队列先把失败当成正常行为真实生产环境里任务一定会失败。原因可能是 API 超时、模型返回格式错误、数据库连接中断、工具调用异常。如果你在批量任务里不做失败重试那么一个异常任务可能中断整个批次。我常用的做法是把批量逻辑放在图外面通过一个循环管理任务列表。每个任务独立调用图捕获异常并记录结果而不是让 LangGraph 本身承担繁重的重试逻辑。图内可以处理“单次运行中的小错误”比如路由失败时走重试节点但“整个图运行崩溃后重启任务”应由外部任务队列处理。tasks [{id: i, input: text} for i, text in enumerate(inputs)] for task in tasks: try: result app.invoke(task[input]) save_result(task[id], result) except Exception as e: save_error(task[id], e) retry_queue.put(task)批量任务时的另一个坑是输出命名。如果你的输入是多个文件或多条文本建议给每个输出带上唯一的任务 ID 或输入文件名避免结果覆盖。这个看似简单的问题在真实项目里出现频率非常高。6.3 日志、命名和资源监控批量跑之前必须检查如果你准备把 LangGraph 应用推向批量生产下面几项是必查项日志每个节点是否打印进入、退出、关键状态变化。至少要能通过日志还原一次执行路径。资源监控CPU、内存、显存、API 调用频率。如果同时跑 50 个图API 并发是否达到上限。超时设置对每个外部调用设置合理的超时时间避免单个节点无限等待。限流不是所有 API 都支持无限并发要配合限流模块保护上游服务。我见过最典型的问题是本地跑单条数据时很顺利但批量时速度骤降原因是 API 并发被限流大量请求排队。这种情况下加更多并发没有意义先看日志里的等待时间再调整并发数和重试策略。6.4 API 化和前端可视化LangGraph 与可视化面板的现实问题热搜词里提到“ECharts 与 LangGraph 能否实现节点可视化”。这个问题要看你想可视化什么。LangGraph 自身编译后的图可以显示节点和边的静态结构但很多人想要的是运行过程的动态追踪比如当前执行到哪个节点、每个节点耗时、输出实时变化。实现这种效果不简单。你可以把 LangGraph 运行日志通过 WebSocket 推送到前端再用 ECharts 的 graph 类型绘制有向图节点颜色随执行状态变化。LangGraph 本身的回调机制能帮你拿到节点开始和结束事件。这里的关键不是画图而是事件流如何和前端状态对应。如果你是初学者我建议先不要自己画可视化面板这不是 LangGraph 最核心的部分。先用日志和命令行输出理解执行流程等真正常态运行后再考虑可视化。画图是锦上添花不是核心能力。7. 常见报错与排查顺序先看日志再看参数最后才怀疑模型7.1 你可能会遇到的几类报错LangGraph 的报错有时非常难懂因为错误可能在图内部某个节点被包装后抛出。我先列几种我在实践里遇到的高频问题。第一节点不存在错误。通常是你把 conditional_edges 的映射值写错了路由返回的 key 没有在映射里找到对应节点。第二State 字段不存在。常见于节点函数读取了没有在 State 中定义的字段。第三递归深度超限。图内部循环太多触发了 recursion_limit。第四checkpointer 相关错误。通常与线程 ID 缺失或序列化问题有关。第五异步与同步混用错误。部分节点是 async其他是 sync需要确保调用方式正确。7.2 排查链路按顺序来不要乱猜我在排查一个 LangGraph 问题的时候一般按这个顺序走。第一步看现象。报错、卡住、最终输出为空、输出不稳定现象不同排查重点不同。第二步看输入。输入数据是否符合预期有没有空字段、错误类型、编码问题。第三步看节点日志。如果把日志打印清楚你很快能定位到最后一个成功执行的节点。第四步看 State。通过调试输出或 checkpointer 查看执行到中途时的 State 是什么。第五步看边和路由函数。重点检查 conditional_edges 的 key 是否和节点名一致。第六步看环境。版本是否兼容API 是否有效资源是否足够。这套链路里最常见的问题其实在第 5 步。很多看似“模型答案不对”的问题实际上是因为路由函数返回了错误 key导致图走了完全不同的分支。7.3 一些提高稳定性的建议如果你要让 LangGraph 应用长期稳定运行我有几条实际建议。第一给每个节点命名时直接用语义化名称比如extract_user_info、validate_address不要用node_1、node_2。否则图一旦大起来排查日志基本等于灾难。第二把路由策略单独写在边函数里不要嵌在节点内。第三为关键的 State 字段配置默认值。这样即使某些分支不更新字段下游节点也能安全读取。第四在入口处做输入校验而不是指望模型中转了之后再去兜底。最后说一个我自己的习惯我不会一上来就把所有能想到的节点、子图、并行、条件路由全部写上。先做一个最小可用版本比如只有一个节点、一个边、一个简单的条件分支跑起来后再逐层加入记忆、工具调用、并行分支。每个阶段都独立验证一轮再进入下一个阶段。这种节奏比收藏再多“全流程实战教程”都有效。LangGraph 的学习曲线并不在“语法”上而在“流程设计”上。把状态变化、路由逻辑、循环边界想清楚剩下的就是往图里填节点。如果你现在正准备上手一个带分支或循环的 LLM 应用建议先花半小时把最小 Demo 和条件路由都跑通再开始设计大图。真正踩过几次坑之后你会发现大部分问题不是工具不够强而是流程边界没有划清楚。
返回列表