让 LangGraph 不再神秘:AI 工作流为什么需要一张能「掉头」的图
想象你在维护一个客服机器人。用户发来一句「我的航班延误了,能改签吗?」最初你写的流程很死板:先查订单,再查航班动态,然后触发改签工具,最后拼回复。用 LangChain 的 Chain 来做,每一步的输出固定喂给下一步,整条流水线一个工位咬着一个工位,顺序不能乱。大多数顺风顺水的对话里这招够用,直到用户补了一句「我订的是商务舱,退改政策也适用吗?」——这时候流程早就走到「调用改签工具」了,想回头,没门。
真正的麻烦出在中间判断上。查完航班动态,发现大面积延误,系统其实应该建议免费退票而不是改签——可流程已经焊死,非要调改签工具不可。更头疼的是,用户中途补了舱位信息,你可能得折回去重查一遍退改政策。在 Chain 的世界里,流程一旦启动就只能单向推进。想根据中间结果调整方向,要么把所有可能路径提前用 if-else 枚举出来,要么把判断规则塞进提示词,让模型自己「猜」。前者分支一多就爆炸,后者等于把方向盘交给模型——它完全可能猜错。你要的不是一条更长的链,而是一种能「走两步、看一眼结果、再决定下一步」的结构。
一、从一次「答非所问」说起:AI 工作流不能只会直行
先看看原来的方案到底卡在哪。LangChain 的 Chain 本质上就是把多个步骤串成一条直线:A 的输出喂给 B,B 的输出喂给 C。这种设计处理「固定顺序、无分支」的任务很顺手,比如「读邮件 → 提炼要点 → 起草回复」。可一旦中间步骤的结果会影响后续走向,直线就绷不住了。
你当然可以用 if-else 在代码里把每个分支都提前写好。但 AI 应用的中间结果往往没法穷举——航班可能延误、可能取消、可能因舱位不同政策各异,每个判断点都枚举一遍,代码会膨胀成一棵难以维护的分支树。另一种常见招数是把判断规则写进提示词,让模型在生成回复时「顺手」决定流程怎么走。听着灵活,实则危险:提示词里的指令对模型来说只是「建议」,它可能理解偏、可能无视、可能自由发挥过头。你没法保证一段自然语言描述的规则被百分百执行。
这两条路暴露的是同一个问题:流程的控制逻辑没有被独立出来。要么焊死在代码结构里(if-else 树),要么混进模型的内容生成里(提示词规则)。前者不灵活,后者不可控。你需要的是一种把「流程走向」当作头等公民来建模的方式——它能根据中间结果改变路径,而且这种改变是确定性的,不靠模型临场发挥。这正是 LangGraph 的出发点。
二、把工作流画成图:状态、节点和会拐弯的边
LangGraph 的思路很直白:把工作流画成一张有向图。图由两类东西构成:节点和边。节点是图里的一个执行步骤,在代码里就是个普通函数;边是节点之间的箭头,表示「执行完这个节点,下一步去执行那个节点」。图论里管这叫有向图,意思是每条边都有方向,你不能逆着箭头走。LangGraph 官方文档里,核心类 StateGraph 就是用来表示这张图的——你往里注册节点、添加边,最后编译成一个可执行的工作流。
但光有节点和边还不够,还得解决「节点之间怎么传信息」的问题。LangGraph 的做法是引入一个全局的状态对象。你可以把它想成一块所有节点共用的白板:每个节点执行前读白板上自己需要的内容,执行完把自己的结果写回白板。在代码里,这个状态就是一个 TypedDict——一个带类型声明的字典,声明了白板上会有哪些字段。节点函数接收整个状态作为参数,返回一个字典,返回的键值对会更新到状态里。这样一来,节点之间不需要知道彼此的存在,只需约定「我在白板的哪个位置写什么」。比如查航班节点往状态里写 flight_status: 延误,判断节点读这个字段来决定下一步。这种设计让节点高度解耦:你可以在不碰其他节点的情况下,替换或新增一个步骤。
到这里,图和 Chain 的区别还不明显——如果所有边都是「执行完 A 必定执行 B」,那图就是一条拉直了的链。LangGraph 真正的杀手锏是条件边:一条边上挂一个函数,函数读当前状态,返回「下一步去哪个节点」。这就是图能拐弯的原因。回到客服的例子:查完航班动态后,你挂一条条件边,如果状态里 flight_status 是「正常」,走改签流程;如果是「大面积延误」,走退票建议流程。这个判断是确定的逻辑,不是让模型猜——模型只负责它擅长的理解用户意图,流程走向由代码控制。
最后一块拼图是循环。有向图里,边可以指回前面的节点。查完退改政策发现用户是商务舱、政策不一样,你可以让流程折返回去重新查——不需要新建节点,只要画一条从「政策查询」指回「订单查询」的边。循环让图具备了 Chain 没有的「重来」能力。LangGraph 的文档里明确区分了三种边:起始边把入口连到第一个节点,普通边表示「执行完 A 必定执行 B」,条件边用函数决定下一步去哪。循环就是普通边或条件边指回了前面的节点,仅此而已。
为什么这么设计而不是把逻辑写进提示词?因为提示词里的指令是「建议」,模型可能不遵守;图里的边是「规定」,代码保证它一定执行。LangGraph 的定位就是把「流程怎么走」从模型手里拿回来,交给开发者用代码明确控制。模型负责生成内容,你负责决定流程——各管各的。这也是为什么 LangGraph 常被称作「状态机工具」:它定义了一系列步骤(节点)、每步的状态转换逻辑(边),以及支撑循环的状态持久化机制,让工作流可以停下来、折返、再继续。
三、动手搭一个「查了不对就重查」的检索流程
前面讲的机制,用一个可运行的例子落地验证。我们做一个文档问答助手:用户提问后,先从文档库检索相关内容,再让模型判断检索结果够不够回答。不够就改写问题重新检索,最多重试两次。这个「不够就重来」的需求,正好用上循环和条件边。
第一步:定义状态
状态就是一个 TypedDict,声明这个工作流白板上会有哪些字段:
from typing import TypedDict
class GraphState(TypedDict):
question: str
documents: list
retries: int
question 存用户原始问题,documents 存检索结果,retries 记录重试了几次。LangGraph 的节点函数接收这个状态、返回一个字典,返回的键值对会更新到状态里。这里要注意:LangGraph 更新状态有两种方式,默认是覆盖——你返回 {"documents": docs},状态里原来的 documents 就被整个替换掉。这在大多数场景下够用了。
第二步:写节点函数
写两个节点,一个是检索,一个是判断:
def retrieve(state):
docs = search(state["question"])
return {"documents": docs}
def check(state):
if len(state["documents"]) < 3 and state["retries"] < 2:
return "rephrase"
return "generate"
retrieve 接收当前状态,读取 question 字段,调用 search 函数(这里省略了实现,它可以是向量数据库查询或关键词搜索),把检索结果写进状态的 documents 字段。check 不返回新数据,只返回一个字符串——这个字符串就是条件边的「路标」,告诉 LangGraph 下一步该走哪条边。注意 check 里读了 retries 字段来防止无限循环,但这个字段目前还没人更新,下一步我们会补上。
第三步:把图画出来
from langgraph.graph import StateGraph
graph = StateGraph(GraphState)
graph.add_node("retrieve", retrieve)
graph.add_node("check", check)
graph.add_node("rephrase", rewrite_question)
graph.add_node("generate", generate_answer)
graph.set_entry_point("retrieve")
graph.add_edge("retrieve", "check")
graph.add_conditional_edges("check", check, {
"rephrase": "rephrase",
"generate": "generate"
})
graph.add_edge("rephrase", "retrieve")
graph.add_edge("generate", "__end__")
先创建 StateGraph 实例,传入我们定义的状态类型。然后用 add_node 注册四个节点,每个节点关联一个函数——rewrite_question 和 generate_answer 的实现省略了,前者改写问题并返回 {"question": new_question, "retries": state["retries"] + 1},后者调用 LLM 生成最终回答。set_entry_point 指定入口节点是 retrieve,工作流从这里开始。add_edge("retrieve", "check") 是普通边,表示检索完必定执行判断。add_conditional_edges 是关键:它把 check 函数的返回值映射到具体节点——返回 "rephrase" 就走改写节点,返回 "generate" 就走生成节点。
最后两行是关键中的关键:add_edge("rephrase", "retrieve") 把改写问题的节点指回检索节点,形成循环——这就是「重来一次」的实现;add_edge("generate", "__end__") 里的 __end__ 是 LangGraph 内置的结束标记,走到这里整个流程终止。
跑起来后,如果第一次检索只返回 1 段文档,check 判断不够,流程会走 rephrase 改写问题,然后折返回 retrieve 重新检索。你会发现 retrieve 节点被执行了两次——这在 Chain 里做不到。第二次检索后,check 再次判断,如果文档够了就走 generate 结束;如果还不够但 retries 已经达到 2,也会走 generate,避免无限循环。整个过程中,retries 字段防止死循环,条件边保证「够不够」由代码判断,不由模型心情决定。你可以在 retrieve 里加一行 print(state["question"]),运行后会看到问题被改写后重新检索的完整轨迹。
四、图不是万能药:什么时候别用 LangGraph
先泼一盆冷水:如果你只需要「按固定顺序调两三个 API」,用 Chain 或直接写代码更省事。LangGraph 的图结构、状态管理、节点注册,这些全是额外的心智负担。为一个不需要折返的流程引入图,就像为了切一个苹果买了整套刀具——能用,但没必要。判断标准很简单:你的流程里有没有「根据中间结果改变后续走向」的需求?没有,就别上。
第二个坑是状态管理会随复杂度失控。所有节点共享一个全局状态,节点一多,谁写了哪个字段、哪个节点该读哪个字段,很快就会变成一团乱麻。你可能会遇到「A 节点写的字段被 B 节点意外覆盖」这种排查起来很痛苦的问题。LangGraph 提供了状态字段的读写权限控制(比如某些字段只允许特定节点写入),但配置这些本身又是一层学习成本。如果工作流超过十几个节点,先想想是不是该拆成多个子图,每个子图管好自己的状态,再用一个大图把它们串起来。
第三个坑和调试体验有关。图是隐式控制流——代码里看不到「先执行谁再执行谁」,执行顺序藏在边的定义里。普通代码出问题,你打日志、看调用栈就能定位;图出问题,你面对的是一个分散在多个节点和边定义里的逻辑,调用栈是看不见的。LangGraph 有可视化工具可以把图画出来,但定位问题的思路和普通代码完全不同,需要适应期。建议在开发阶段给每个节点加日志,输出它读写的关键字段,否则出 bug 时你只能靠猜。
最后一点:条件边适合确定性判断,不适合开放性决策。「根据文档质量决定重查还是生成」这种规则,代码能写清楚。但「根据用户情绪决定安抚还是转人工」这种模糊判断,硬写条件边会变成一堆脆弱的启发式规则——你怎么定义「情绪不好」?用关键词匹配?用情感分析分数阈值?这些规则在真实对话里很快就会漏。这种场景不如把决策交给模型——当然,代价是你得接受模型偶尔判断错。LangGraph 的灵活之处在于它允许你混合:确定性的规则用条件边写死,模糊的判断让模型通过函数调用或结构化输出来决定,然后你再根据模型的输出走不同的边。
小结
回到开头的客服机器人。你真正需要的不是更长的链,而是一张能折返的图:状态是节点间共用的白板,节点是白板上的操作,条件边决定下一步走哪条路,循环让流程能回头重来。判断逻辑用代码写死,生成内容交给模型——这是 LangGraph 的核心分工。它解决的不是「让 AI 更聪明」,而是「让 AI 应用的流程可控」。LangChain 的 Chain 适合直线流程,AgentExecutor 把控制权全交给模型导致黑盒,LangGraph 站在中间:流程走向由你定,内容生成交给模型。
记住三句话:流程会折返才需要它;确定性分支用条件边,模糊决策留给模型;节点超过十几个,先考虑拆分。下次遇到「答非所问」的 AI 工作流,先问自己:问题出在流程不会回头,还是出在模型本身?如果是前者,画一张图试试。你手头有没有一个「走到一半发现需要重来」的流程?它现在是怎么处理的——写死在 if-else 里,还是指望模型自己拐弯?
素材来源
DAMO开发者矩阵,由阿里巴巴达摩院和中国互联网协会联合发起,致力于探讨最前沿的技术趋势与应用成果,搭建高质量的交流与分享平台,推动技术创新与产业应用链接,围绕“人工智能与新型计算”构建开放共享的开发者生态。
更多推荐
所有评论(0)