☰
LangGraph实战:从状态机到图编排的LLM工作流设计
2026/10/23 5:17:10 网站建设 项目流程

做AI应用开发的朋友,应该都遇到过这个尴尬:大模型接口能说会道,但真要把一堆任务串联起来——先分析需求,再调用工具,然后写报告,中间还要根据结果决定分支——你很快就发现代码里全是全局变量、if-else套娃和状态标记位。我早先接手一个内部知识库问答项目时就是这样,LLM调用逻辑还算清晰,但一旦加上“检索不到就换种问法”“回答置信度低就让用户追问澄清”这类规则,整个流程的真实复杂度立刻从一条直线变成了蜘蛛网。那阵子我最大的感慨是:LLM应用缺的不是“调用能力”,而是“流程编排能力”。

后来接触了LangGraph,算是把这口气捋顺了。它本质上是一个专门给AI应用用的工作流编排框架,核心模型是“把整个业务流程画成一张图”:每个处理步骤是一个节点,节点之间用边连接,边可以是固定跳转,也可以是依据状态内容动态选择的条件路由。这样一来,你说的那些分支、循环、暂停、恢复都不用在业务代码里硬堆,而是变成图结构上自然的表达。这篇文章我会从框架的核心设计思路讲起,结合一个我实际搭过的“AI自动成稿”工作流,把节点、状态、条件边、持久化这些概念一个一个拆开讲明白,再分享一些踩过的坑。适合正在用LLM做应用、想把流程做复杂但又不希望代码烂掉的开发者,也适合对这种新范式感兴趣、想找一份系统性入门的读者。

1. 先搞清楚:LangGraph到底解决了什么问题

1.1 从“单次对话”到“可控工作流”

先退一步看传统LLM应用的做法。通常你写一个函数,组装好prompt,调用模型API,拿到结果再决定下一步。简单场景没问题,但真实业务很少只有一步。比如一个客服机器人,它可能先要判断用户意图,然后查订单、查物流、查退款规则,再综合起来生成回答。这些步骤之间还有依赖和条件,有些步骤可能要重复执行几次才能得到稳定结果,有些步骤则需要人工确认才能往下走。

在没有专门编排框架之前,我见过不少项目用“手写状态机”来实现。代码里定义一堆枚举状态,每个分支用switch-case处理,状态字段散落在各种类里,调试时在日志里打点确认当前走到哪一步。这能做,但维护成本很高。加一个新步骤就要动好几个函数,而且整个流程不可视化、不可暂停、销毁后没法从某个中间节点恢复。

LangGraph把这一切抽象成了“图”。节点就是函数,状态就是一张需要各节点共享的“黑板”,边就是执行流的规则。你不需要再自己维护状态机和跳转逻辑,框架会按图的结构自动驱动整个流程。最有价值的是它还内建了持久化和断点恢复能力,你甚至可以在流程中间停下来,等人工确认后再继续跑。这不是传统工作流引擎套了个AI壳,而是专门冲着LLM应用的特点(不确定、需要反馈、需要多轮修正)设计的。

1.2 图结构带来的三个关键红利

使用图结构编排AI流程,对比传统线性代码有三个非常直观的好处:

第一,可视化一切执行路径。画出来的图便是流程本身,开发讨论时对着图讲,比对着代码说“这个函数里调那个函数”高效太多。LangGraph甚至允许你导出图结构做可视化,团队过方案时直接投屏看节点关系。

第二,运行时状态透明。所有节点的输入输出都走同一个State对象,你可以在任意节点打印状态快照,排查“哪一步出了问题”非常方便。以前是看日志猜流程,现在直接看状态流转。

第三,自然的循环与分支表达。LLM返回结果往往带有不确定性,比如“生成的内容不满意就重新生成”,这在传统代码里是一个while循环,但在LangGraph里就是一条从节点指回它自己的边。图结构的表达能力和代码语义天然对齐,因为图本身就能表达环。

我还有一个体会,图结构对“人机协同”非常有价值。以前想做一个“AI初筛,人做决策,AI继续执行”的流程,得自己设计一套状态持久化方案,把中间结果存数据库,再写恢复逻辑。LangGraph自带的checkpoint机制直接支持这类需求,节点运行前可以设置中断点,中断后程序退出,下次拿到历史状态和用户输入继续跑。这个能力在真实业务场景里几乎是刚需。

2. 核心概念拆解:状态、节点、边与运行时

2.1 状态(State):所有节点共享的那张“黑板”

在LangGraph里,状态是所有节点的共享数据容器。你可以把它理解为每个节点都能读、能写的一份公共数据。节点函数接收当前状态作为输入,返回一个字典,框架会把返回的键值合并到状态里。

定义状态有两种主流方式:一种是Python的TypedDict,声明键和值的类型;另一种是用Pydantic模型,额外获得数据校验能力。我自己的习惯是,团队项目用Pydantic,因为它能保证节点传入传出数据符合预期,调试时能更早发现类型问题;个人实验和脚本就用TypedDict,写起来轻量。

这里必须强调一个很多新手容易忽略的细节:状态字段的“更新语义”。默认情况下,节点返回某个键的值,是整体覆盖还是追加合并?LangGraph允许你通过Annotated[类型, 归约器]来指定。最常用的归约器是operator.add,它表示这个字段的更新采取追加模式。比如你想在状态里维护一个历史消息列表,每个节点往列表里追加一条,那就应该定义成:

from typing import Annotated, TypedDict import operator class State(TypedDict): messages: Annotated[list, operator.add] topic: str article: str

这样一来,每个节点返回{"messages": ["新的记录"]}时,框架不会覆盖原来的列表,而是把新元素追加进去。别小看这个机制,它直接决定了你的节点到底该返回“增量”还是“全量”,写错的话要么数据丢失,要么节点之间互相污染。

2.2 节点(Node)与边(Edge):函数的流转规则

在LangGraph里,节点就是一个普通的Python函数。它接收state,返回一个字典(或者返回一个包含state的元组)。框架不要求你写类、写装饰器,这很符合Python社区的习惯,也让单测变得非常容易。

边则定义了节点的执行顺序和流转关系。常见的边有三种:

  • 普通边:从节点A无条件跳到节点B。
  • 条件边:根据状态内容算出一个结果,动态决定下一步去哪。这是LangGraph的灵魂,因为LLM应用里到处是“如果……就……否则……”。
  • 入口边和出口边:确定从哪个节点开始(set_entry_point),以及到哪个节点结束(到达END)。

条件边的写法也很简洁,就是一个纯函数形式的映射。比如“根据review结果判断是继续修改还是进入下一阶段”,你可以定义一个路由函数,然后把它传给add_conditional_edges。路由函数只读状态并返回一个字符串,框架根据返回值去匹配不同的后续节点。

2.3 运行时:图怎么被执行

图定义好之后,调用.compile()会返回一个可执行对象。你既可以同步调用.invoke(state)跑完整流程得到最终状态,也可以用.stream(state)逐步产生中间状态,从而在前端实现流式输出。

我最常用的是.stream()。举个例子,假设图里有三个节点,我希望用户看到“大纲生成中…”“正文写作中…”这样的进度提示,那就可以在stream模式下逐节点获取状态变化,并实时推送到前端。如果还是老式做法,自己写流程控制,这种逐阶段反馈就得手动敲事件回调,代码量差很多。

LangGraph还有上百种细节特性的组合,但核心心智模型始终就三个词:状态、节点、边。把这个搞透,剩下的都能渐进掌握。

3. 实操:做一个“技术文章自动生成工作流”

3.1 场景需求与图结构设计

为了讲清楚LangGraph的实际用法,我带大家完整搭一个“AI自动成稿”工作流。需求不复杂:用户输入一个主题,AI先写大纲,然后根据内部规则评审大纲,如果大纲质量不达标就最多自动修改两次,然后进入正文写作,写完再做一个终稿检查,输出最终结果。

你看这个需求里已经包含了顺序执行、条件分支、循环、状态共享,非常适合用图来建模。对应到图上就是5个节点:generate_outline(生成大纲)、review_outline(评审大纲)、rewrite_outline(改写大纲)、generate_article(生成正文)、final_check(终稿检查)。加上一个代表流程终止的END。

这里的设计思路是:把“改写的次数控制”放在状态里而不是放在循环条件里。具体来说,状态中维护一个revision_count字段,每进入一次rewrite_outline节点就加1,当评审节点发现改写次数已经超过上限,条件路由就引导流程进入generate_article,否则回到rewrite_outline。这样循环次数可控,不会因为LLM始终“不满意”而跑成死循环。

3.2 定义状态与辅助函数

第一步,定义状态:

from typing import TypedDict, Optional class State(TypedDict): topic: str outline: str outline_feedback: str revision_count: int article: str final_result: str

这里没有用Annotated + operator.add,因为每个字段在当前业务里都是“覆盖式”更新。revision_count作为计数,则需要节点返回时在原值基础上加1,这属于节点内部逻辑。

第二步,定义通用的大模型调用函数。为了演示方便,我用一个模拟函数替代真实API调用,实际项目里替换成LLM SDK即可:

def call_llm(prompt: str) -> str: # 实际项目中在这里调用大模型API # 这里用一个模拟函数占位 return f"基于主题生成的虚拟内容: {prompt[:50]}"

3.3 逐步构建在线流程

接着定义每个节点的处理函数:

def generate_outline(state: State): prompt = f"为文章主题《{state['topic']}》生成一份结构清晰的大纲" outline = call_llm(prompt) return {"outline": outline} def review_outline(state: State): # 模拟评审:这里你可以接入规则引擎,比如检查大纲是否包含“引言”和“总结” if "引言" not in state["outline"] or "总结" not in state["outline"]: feedback = "大纲缺少引言或总结部分,请补充" revision_count = state.get("revision_count", 0) + 1 return {"outline_feedback": feedback, "revision_count": revision_count} return {"outline_feedback": "大纲合格"} def rewrite_outline(state: State): prompt = (f"根据反馈修改大纲:{state['outline_feedback']}\n" f"当前大纲:{state['outline']}") new_outline = call_llm(prompt) return {"outline": new_outline} def generate_article(state: State): prompt = f"根据大纲写一篇完整文章:{state['outline']}" article = call_llm(prompt) return {"article": article} def final_check(state: State): # 模拟终稿检查 if len(state["article"]) < 20: return {"final_result": "终稿过短,需要人工复核"} return {"final_result": "终稿通过"}

节点函数有两个约定俗成的规矩:不要修改传入的state对象本身,而是返回一个包含更新字段的新字典;不要在节点内部实现流程控制,那是边的职责。很多初学者会把“if条件决定是否跳到另一个节点”写在节点函数里,这样虽然也能跑,但图结构就失去了意义,调试时也没法从图层面看到分支逻辑。

一切准备就绪,接下来把节点和边拼起来:

from langgraph.graph import StateGraph, END builder = StateGraph(State) # 添加节点 builder.add_node("generate_outline", generate_outline) builder.add_node("review_outline", review_outline) builder.add_node("rewrite_outline", rewrite_outline) builder.add_node("generate_article", generate_article) builder.add_node("final_check", final_check) # 设置入口 builder.set_entry_point("generate_outline") # 顺序边 builder.add_edge("generate_outline", "review_outline") builder.add_edge("rewrite_outline", "review_outline") builder.add_edge("generate_article", "final_check") builder.add_edge("final_check", END) # 条件边:评审通过才进入正文生成 def route_after_review(state: State): if state["outline_feedback"] == "大纲合格": return "generate_article" # 如果改写次数超过2次,强制进入正文生成 if state["revision_count"] >= 2: return "generate_article" return "rewrite_outline" builder.add_conditional_edges( "review_outline", route_after_review, { "generate_article": "generate_article", "rewrite_outline": "rewrite_outline", }, ) graph = builder.compile()

请注意条件边里返回的字符串,必须出现在映射字典的键中。如果映射关系缺失,框架会直接抛异常。这是一个挺常见的低级错误,在一张节点很多的图里排查起来很费劲,所以我建议把路由函数和映射字典写在一起,保持图结构的可读性。

3.4 运行效果与状态流转分析

执行流程直接调用:

result = graph.invoke({"topic": "LangGraph入门指南", "revision_count": 0}) print(result["final_result"]) print(result["outline"]) print(result["article"])

我用模拟数据测试时,第一次大纲缺少“引言”,所以流程进入rewrite_outline,然后回到review_outline,第二次通过评审,进入正文生成和终稿检查,最终输出。整个过程中你可以打印每个节点的返回结果,观察状态如何被一步一步填充。

这里我想强调一点:可观测性是这个框架带给你的最大红利。代码里每个节点函数第一行加上print(f"[执行节点] 当前状态: {state}"),一次运行下来,你就能完整看到AI工作流的状态流转。这在传统手写流程里很难做到这么清爽——每个节点都是同一套结构,日志也整齐。

4. 进阶特性:持久化、流式输出与多智能体协作

4.1 检查点机制:让工作流能暂停、能恢复

真实业务中,AI流程往往不是一把梭跑到底的。最典型的是生成结果需要人工审核——AI先起草,人点头,AI再做下一个动作。LangGraph的持久化机制为此提供了开箱即用的支持。

实现思路并不复杂:你在编译图的时候传入一个CheckpointSaver,框架会在每个节点运行后把完整状态保存起来。之后你可以在图的任意一条边上配置interrupt_before,比如在进入final_check之前停下来。当流程跑到中断点时,invoke会返回一个“需要继续”的信号,你不急着给数据,程序可以先退出。等人工在界面上点了“确认通过”之后,再从历史检查点恢复执行。

我第一次用这个功能时,最大的感受是省掉了自己设计“中间状态表”的工作。以前做类似功能,我得把大纲、正文、历史消息序列化成JSON存在数据库里,再写一套续跑函数去反序列化并手动调用下一步逻辑。现在这些都被框架接管了,状态schema定义好之后,恢复逻辑是通用的,开发效率提升非常明显。

4.2 流式输出:让用户看到AI“正在进行”

另一个非常影响用户体验的特性是流式输出。熟悉现代AI应用的朋友应该都习惯了大模型逐字打印内容的交互方式。如果工作流有多个节点,每个节点都会调用一次模型,那“流式输出”就不只是单个模型生成时的逐字流式,而应该是“整个工作流的过程流式”——用户看到的不只是文本在生成,而是流程在推进。

LangGraph的.stream()会把图中每个节点的输出逐步传给外部。你可以在一个循环里拿到状态快照,然后结合前端技术把这些信息用SSE或WebSocket推给浏览器。举个例子,当图执行到generate_article节点时,状态字段里会带上一个“当前生成的文本片段”,你可以实时渲染到界面上。这种做法比“整个流程结束再一次性返回”更适合LLM应用场景,因为大模型生成本来就有延迟,一次性等待会让用户以为页面卡死了。

4.3 多智能体协作:当多个AI角色同时出现在一张图里

LangGraph最被频繁讨论的进阶能力是多智能体协作。它的做法并不玄乎:把每个Agent本身也建模成一个节点(或者一个子图),多个Agent节点之间通过图结构协调。两类常见模式:

  • 监督者模式:一个“主管Agent”节点负责读取当前状态,决定把任务交给哪个“下属Agent”节点,下属执行完再回传结果给主管决定下一步。
  • 顺序流水线模式:每个Agent负责一个阶段,比如“研究员Agent”负责查资料,“写手Agent”负责写文章,“编辑Agent”负责修改润色,三个节点串成一线。

实现上,每个Agent节点内部可以是一个完整的LangGraph子图,这样你既能享受Agent内部的循环机制,又能在更高的图层面编排它们的协作关系。我建议在实际项目中先把单个Agent内部的流程画清楚,再用组合的方式把它嵌入更大的图。如果一开始就把多个Agent混在一个巨大的图里,节点数量和边关系会膨胀到很难维护。

从实用角度看,多智能体虽然听起来炫酷,但并非所有场景都需要。如果一个流程能用一个图节点加两次循环解决,就不要硬拆成三个Agent,每个Agent增加的大模型调用次数和延迟都是实打实的成本。

5. 避坑指南与常见问题实录

5.1 状态字段到底该怎么划分

我见过不少项目在状态设计上走极端。一种是把所有临时变量都塞进状态,导致状态对象越来越庞大,每个节点都读一大堆无关字段;另一种是什么都只放在节点局部变量里,结果节点之间共享数据时就缺字段,只能临时往状态里加键。

我的经验是,遵循一个简单的标准:只要有两个以上节点需要读写的字段,就放状态里;如果一个字段只属于单个节点的内部计算,就别放进状态。另外要避免直接修改状态里的嵌套结构。尽管LangGraph允许你返回新值覆盖,但如果你同时在节点内部原地改了Python字典的某个子字段,可能会产生难以察觉的共享引用问题。所以我的习惯是,节点函数里所有需要更新的内容都显式构造新值返回,不做原地修改。

5.2 条件路由导致死循环怎么办

LLM应用里的循环和传统程序不太一样。传统循环的终止条件往往很明确,但LLM节点的输入输出充满不确定性,很容易出现“生成的答案始终不满足条件,流程一直在重试”的情况。

我的解决套路是给所有循环分支设置“熔断计数器”。正如前面示例里的revision_count,它保证循环最多执行两次。更复杂的场景还可以设置一个全局的“最大步骤数”,在条件路由函数开头检查“总执行步骤数是否已经超过阈值”,超过就直接跳到终止节点。这属于一种朴素的兜底策略,但它能避免生产环境里的死循环事故。

5.3 多个分支并行时的共享状态问题

LangGraph支持扇出结构,一个节点的输出可以同时传给后续多个节点并行执行。但并行节点之间如果有多余的依赖关系,就会出错。典型错误是:两个并行节点同时想往状态里的同一个列表字段追加数据,这个在归约器是operator.add时是安全的,框架会合并结果;但如果一个节点需要读取另一个并行节点刚写好的字段,那就不行了,因为并行节点的执行顺序不受你控制。

所以我建议在设计时明确区分“并行只读共享数据”和“串行写入新数据”。如果是需要多个模型结果汇总后才能进入下一步,那就让它们各自把结果写入不同字段,再由下游节点负责读取和汇总,而不是指望另一个并行节点把结果“提前”准备好。

5.4 调试技巧与日志规划

调试LangGraph程序其实有一套很高效的方法论。最基础的是在每个节点入口打一行日志,包含节点名和关键状态字段。再进阶一点,你可以把State里每个字段的变更历史记录下来,每次进入和离开节点时做一个快照,分析执行路径时非常直观。

我还有一个习惯:把图本身打印或导出出来看一眼。LangGraph支持以文本形式打印图结构,你可以在写完图定义后先执行一下,确认边和节点的连接关系与设计一致。特别是条件边的映射表,打印出来检查keys和返回值是否匹配,这一步能提前发现自己看漏的路径。

5.5 性能与成本控制在真实业务中的考量

图编排带来了灵活,也带来了代价。每多一个节点,就可能多一次大模型调用,延迟和成本线性增加。对于线上服务,务必为每个节点设置超时时间,避免某个模型调用卡死拖垮整个流程。另外可以考虑对中间结果做缓存——同一个主题生成的大纲如果在历史状态里已经存在,可以直接复用,没必要再调一次模型。

我在实际部署时,还会把整张图封装成独立的服务接口,流程内部的状态对象不直接暴露给外部调用方,只暴露必要的输入和输出字段。这样即使内部要增加一个节点、调整一条条件边,外部API也不受影响。

6. 我的实际感受与选型建议

用LangGraph做了几个项目之后,我对“什么时候该用它”有了一点自己的判断。如果你的AI应用只是简单的“接收一个输入,调用一次模型,返回结果”,那完全没必要引入这套复杂度,直接写一个函数就够了。但一旦你开始遇到这些信号——流程里出现多个模型调用节点、有判断分支、需要反复重试、需要人工介入、希望工作流状态能序列化和恢复——那LangGraph能帮你省下的代码量和心智负担就会非常明显。

对于团队协作,图结构的另一个隐形好处是统一的流程描述语言。产品经理讲“先生成方案,方案不好就改,改完再执行”,开发说的“建一个生成节点,条件路由看到不合格就回到上一个节点”,两者在图上能对得很齐。我甚至见过团队直接用流程图文档来评审代码结构,这在纯代码时代是做不到的。

最后分享一个小技巧:不要在项目一开始就把图设计得特别复杂。我踩过坑,第一版就把十来个节点和多层条件路由全画出来,结果跑起来翻车后根本分不清是哪个环节的问题。更稳妥的做法是先用两三个节点跑通主链路,确认状态定义没问题后,再逐步把条件分支、循环和并行节点加进来。LangGraph的图结构本身是支持增量演进的,把一张小图改大,比在庞杂流程里修bug要轻松得多。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询