★ FEATURED ARTICLE 前阵子帮某电商公司的客服团队做工单自动分拣需求一句话就能说清用LangGraph把AI对工单的分类结果自动发起成后续处理流程。听上去不复杂可真开始动手我却发现模型输出和流程走向之间隔着一条鸿沟——分类结果要驱动退款、技术、归档三套不同的处理链条中途还要停下来等人工审批进程挂了要还原现场运营同事还想随时知道每个工单到底卡在哪。这篇文章就是这次完整实战的复盘笔记写给准备用LangGraph做动态流程编排的开发者也写给那些正在犹豫要不要把流程逻辑交给图引擎的人。我会尽量把为什么这么设计、踩了哪些坑、上线前补了什么功课都写清楚。1. 从一个让我抓狂的需求说起为什么普通代码撑不住动态流程1.1 用if-else写LLM流程前三天很爽第四天开始想重构最早接到需求时我的第一反应是这不就是个分类接口加几个处理函数吗先把工单文本喂给大模型让它输出一个分类结果然后对着结果写一个if-else分发每种分类调用对应的处理函数完事。于是第一版代码大概是这样的逻辑先调LLM拿分类然后if category refund走退款流程elif category tech走技术支持流程其他走归档。前三天确实跑通了demo演示给运营看也很惊艳但第四天需求变了技术类工单里如果模型判断是硬件故障需要升级给二线工程师否则自动回复排查手册。我改了改if-else加了层嵌套又好了。第五天需求又变了退款金额超过1000块的工单不能自动退款要挂起等人工审批审批通过后再继续原流程。这时候我开始感觉不对劲。为了知道一个工单当前走到哪一步我得把状态字段传来传去为了让人工审批之后能继续流程我得把未完成的步骤和参数序列化存起来等审批接口回调时再恢复还要避免重复执行为了排查问题我只能靠猜因为流程是散落在各个函数里的隐式逻辑。这种写法最大的问题不在于代码丑而在于你无法回答三个问题当前这个工单到底在哪一步它下一步有几种可能去向如果中途中断了怎么恢复现场普通函数链天然没有流程视图状态被拆碎存在各种局部变量和数据库字段里。而LangGraph重新把这些问题变成了图结构接下来我详细说说它好在哪里。1.2 状态机思维把流程画成图让引擎替你做状态管理LangGraph的核心思路是不把你写的处理函数按顺序硬串起来而是允许你定义一张有向图节点是处理步骤边是流转条件所有节点共享一个全局状态对象。节点读状态、改状态边根据状态里的字段决定下一步去哪张节点。这个过程本质上就是一个显式的、可持久化的状态机。我个人的理解是普通代码像是打电话A打完给BB打完给C信息靠口头传递中间挂了就断了LangGraph像是写工单流转单每个节点往同一张单据上填内容填完之后系统根据单据当前内容决定下一个处理人整个过程每一步都有存档。用AI发起流程应用的时候这种设计尤其重要因为大模型的输出天然带有不确定性你需要让流程的下一步由模型输出动态决定而不是在代码里硬编码死。当然并不是所有项目都需要上LangGraph。如果你的场景就是单次调用大模型返回结果或者固定三步链式调用没有分支那用普通代码简单直接上状态图反而是过度设计。我踩过的判断标准是只要你的流程里出现模型输出决定下一步走向、中途要挂起等人确认、失败后要恢复现场这三件事中的任意一件就可以认真考虑用它了。2. 建图前必须想清楚的四个概念状态、节点、边和检查点LangGraph的API细节不算多但如果你没想清楚底层建模逻辑写起来容易到处碰壁。我按自己理解拆成四块State、Node、Edge、Checkpointer逐个说透。2.1 State状态所有节点的共享单据State就是一个普通的TypedDict所有节点共享同一个状态对象。每个节点函数接收当前整个State作为参数然后返回一个字典表示要更新哪些字段。LangGraph会把返回的字典合并回State传给下一个节点。这里有个核心概念叫Reducer归并器。默认情况下节点返回的字段是直接覆盖旧值的但有些字段我们希望是累加的比如对话消息列表。要给字段指定归并逻辑写法是用Annotated加一个归并函数比如Annotated[list, add_messages]表示新消息追加到旧消息后面而不是整体覆盖。我在工单系统里定义的状态大致长这样from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class TicketState(TypedDict): ticket_id: str # 工单号 text: str # 原始工单内容 category: str # 模型分类结果refund / tech / suggestion / other amount: float # 涉及金额用于退款审批 needs_escalation: bool # 是否升级二线 result: str # 最终处理结果 messages: Annotated[list, add_messages] # 过程中产生的所有消息设计State时有个很重要的约束尽量只放可JSON序列化的数据。因为State以后要进检查点保存、要跨请求恢复、要能打印出来调试如果塞一个自定义类对象进去到时候序列化你会非常痛苦。我一开始图省事把数据库连接对象放进了State结果检查点保存直接报错后来老老实实把连接对象拿出来只放连接ID和必要参数。2.2 节点Node一个节点只做一件能被单独测试的事节点就是一个普通函数签名统一为def node(state: TicketState) - dict:。它接收当前快照返回要更新的字段。函数内部做什么都行调大模型、查数据库、发消息、调用内部APILangGraph不管它只关心你返回了哪些字段。我踩过的颗粒度教训是一个节点里不要塞太多事。最初我把模型分类解析结果写日志全塞在一个节点里后来发现日志格式要改、模型解析逻辑要调改动全挤在一个函数里测试也不好写。后来强制自己按一次LLM调用、一个领域动作来划分节点比如类节点只负责调模型并解析出category退款节点只负责对接退款服务和写result。这样做的好处是每个节点可以单独传一个构造好的State去测出问题时定位也快。另外有一点要特别注意不要在节点里做跨节点共享的副作用操作。比如你在这个节点里写了一个数据库记录等下因为故障重放、图重新执行这个节点可能被执行两次副作用就会出现重复。我现在的做法是把外部系统调用的副作用尽量放到流程末端、或用专门的动作节点承载让重放时副作用是幂等的。2.3 边Edge条件边就是把模型判断翻译成流程走向边分成两种普通边和条件边。普通边就是add_edge(A, B)表示A跑完无条件去B。条件边用add_conditional_edge它接收当前State返回一个值然后根据返回值映射到不同的下游节点。我在工单系统里最核心的一段路由代码是这样写的builder.add_node(classify, classify_node) builder.add_node(refund, refund_node) builder.add_node(tech, tech_node) builder.add_node(suggestion, suggestion_node) builder.add_node(escalate, escalate_node) builder.add_node(human_review, human_review_node) builder.add_edge(START, classify) builder.add_conditional_edge(classify, route_after_classify, {...})路由函数本身特别简单就是读取State里的category字段然后返回def route_after_classify(state: TicketState) - str: return state[category]映射表把返回值对应到节点名{ refund: refund, tech: tech, suggestion: suggestion, other: suggestion }这段逻辑的关键在于大模型本身并不具备跳转图节点的能力它只是输出一个分类字符串真正控制流程走向的是这个路由函数。这句话我建议反复读几遍因为它就是整个LangGraph配LLM的思维核心模型负责判断和生成流程引擎负责把判断结果转换成确定性的分支跳转。这样设计的好处是即使模型抽风输出了一个不在枚举范围内的值你也能在路由函数里加兜底逻辑而不会让流程流到一个不存在的节点。2.4 检查点Checkpointer有了它流程才能中断恢复如果编译图时不传checkpointerLangGraph只会在一次invoke调用内部维持状态调用结束状态就没了。一旦你在生产里需要进程重启后还能接着跑、人工审批挂起后继续原流程就必须在编译时传入checkpointer并同时提供线程ID。线程ID是状态的隔离维度同一张图在处理不同工单时用不同thread_id状态互不干扰。我当时的做法是一个工单号作为一个thread_id这样每个工单就是一个独立的流程实例。内存型检查点适合本地开发和测试生产环境我建议至少用支持持久化存储的版本把检查点落库。有了检查点之后即使进程崩溃重启只要thread_id不变就能从最近保存的步骤继续执行。这是普通函数链完全不具备的能力也是LangGraph在AI发起流程应用这种场景里最值钱的部分之一。3. 实战拆解让大模型当路由的工单自动分诊系统概念讲完回到我的实际项目。这个系统做的事情很简单工单进来先让大模型做分类路由不同类型走不同流程涉及高金额或高风险的操作必须人工审批全程记录每个工单的状态流转。3.1 分类节点让模型输出结构化的枚举值而不是自由文本分类节点是整个路由的起点它做得对不对直接影响后面所有分支。我是这样实现它的from langchain_core.output_parsers import StrOutputParser classify_prompt 你是工单分类器只能从下面选项中选择一个分类 refund: 退款、退货、少发货、金额不对 tech: 软件报错、登录失败、功能不可用、设备故障 suggestion: 产品建议、功能期望、一般咨询 other: 其他 工单内容 {text} 只输出分类名称本身不要解释。 def classify_node(state: TicketState) - dict: response llm.invoke( classify_prompt.format(textstate[text]), temperature0 ) category response.content.strip().lower() if category not in {refund, tech, suggestion, other}: category other return {category: category, messages: [{role: assistant, content: f分类结果{category}}]}三个细节我特别说明一下。第一温度设置为0分类任务不需要创造性温度越高越容易乱选。第二我让模型只输出枚举值本身而不是一段解释这样解析逻辑简单可靠而不是在一个长文本里用正则抠字段。第三即使提示词已经限定了枚举我仍然在代码里做了二次校验非法值一律落到other保证路由函数拿到的值一定是合法的。后来模型服务调整过一次出现过category返回带标点的情况就是这个兜底救了我。3.2 退款链路与人工审批挂起退款类工单的处理流程是这样的进入退款节点先尝试自动校验订单状态和金额。如果金额超过1000块挂起等人工审批如果金额在限额内自动走退款服务返回结果到result字段。挂起功能我用的是LangGraph的interrupt机制。interrupt可以在节点中间暂停图的执行并把一个请求对象抛给外部调用方等外部注入继续信号后图从暂停位置继续跑。退款节点关键代码如下from langgraph.types import interrupt, Command def refund_node(state: TicketState) - dict: # 这里省略校验订单的步骤 if state[amount] 1000: approval interrupt({ type: refund_approval, ticket_id: state[ticket_id], amount: state[amount] }) if not approval.get(approved): return {result: refund_rejected} return {result: refund_approved} # 金额正常直接走自动退款 return {result: refund_auto_done}注意interrupt抛出的内容是一个请求描述图执行到这里会真正停下来外部系统比如一个审批后台可以通过thread_id查询到当前状态看到next字段是哪个节点在等待然后通过Command(resume...)把审批结果传回来。这个机制比我之前用数据库状态字段硬写恢复逻辑要干净得多因为它天然支持流程暂停在图的中间位置、接着往下走不需要我自己去设计什么恢复令牌。3.3 技术链路升级与建议归档技术类工单的处理相对简单进入tech节点后我让模型再做一次判断看是否属于需要二线介入的硬件故障输出needs_escalation布尔值。如果为真走escalate节点随便锁一个工单池否则返回一条排查手册给用户并把result写成auto_reply。这里我想强调一个设计取舍两次LLM调用我放在了两个独立节点里classify和tech而不是在classify里一次把所有字段都抽完。原因是第一次只需要分类分类的结果决定了是否走tech分支只有确定走tech分支后才值得花二次调用的成本去判断是否需要升级。如果一开始就让模型把什么金额、是否升级全抽一遍大部分工单根本用不上这些字段白白浪费token而且抽取字段越多越容易出错。建议类工单直接归档。归档节点做的事情非常简单就是把result字段设置为archived并拼一条记录到messages里。这个节点堪称全图最简单但最稳定的节点也是我说的少而简单原则的最好例子。3.4 编译完整图和一次真实运行所有节点和边定义完后编译并运行from langgraph.checkpoint.memory import MemorySaver builder StateGraph(TicketState) # ... 按上面方式 add_node / add_edge ... builder.add_edge(refund, human_review) builder.add_edge(tech, human_review) builder.add_edge(suggestion, END) builder.add_edge(escalate, human_review) builder.add_edge(human_review, END) graph builder.compile(checkpointerMemorySaver()) config {configurable: {thread_id: T20240101}} graph.invoke({ ticket_id: T20240101, text: 我买的鼠标用了三天就坏了登录键失灵麻烦看看怎么办, amount: 0.0 }, config)当我把这条鼠标坏了的工单跑进去时classify节点输出的category是tech路由函数把它导向tech节点tech节点再次判断后needs_escalation为true于是走向escalate节点最后停在human_review等人工确认。整个过程我在调用后用graph.get_state(config)去看能直接看到当前step的字段、所有节点历史记录以及next字段指向哪个节点这种透明可查的体验是之前写函数链从未有过的。4. 把Demo推向真实服务时踩过的坑并发、流式与续跑Demo跑通只是开始真正把它接到Web服务里、同时服务多个工单的时候我才遇到了一系列之前没想过的麻烦。这一节全是我自己踩出来的教训。4.1 并发场景下thread_id一定要隔离不要复用第一次接入FastAPI时我用了一个固定的thread_id想着所有工单共享一张图、无所谓。结果跑了一阵后发现不同工单的状态互相污染A工单的分类结果跑到了B工单的流程里。原因很简单LangGraph的状态隔离维度是thread_id同一个thread_id下无论多少张图实例最终读写的是同一份状态。两个请求同时进来都往同一个thread_id写后写的就把先写的覆盖了。修复方式也很简单每个工单请求动态生成独立thread_id比如直接用工单号、会话ID或UUID。我还额外做了一层保护同一个thread_id不允许并发invoke如果检测到已有未完成的执行返回处理中而不是强行再开一次。这个坑背后其实是一个通用规律把LangGraph当一个应用引擎用时你要像操作数据库一样认真对待状态隔离。thread_id就是状态的主键千万别图省事用一个固定值糊弄过去。4.2 流式输出节点内部直接stream图的外部是收不到增量的工单场景里有一部分处理结果是需要给用户端实时显示的我在做一个在线问答功能时特别需要看大模型的逐字输出。一开始我在节点函数内部直接对模型接口做stream把增量print到服务端日志里却发现无论我在外层用graph.stream()怎么调都只能拿到节点完成后的整体返回拿不到token级增量。原因是LangGraph的流式机制是图级别的它只负责在节点间流动状态如果你在节点内部消费了模型的stream图引擎根本不知道这些增量。后来我改用LangGraph提供的stream_modemessages让节点返回消息对象而不是纯文本字符串由引擎层统一做增量推送这才真正在客户端看到了逐字效果。所以使用前先想清楚你要的是节点级进度还是token级流式。前者用默认的values或updates模式就够后者要主动选择messages模式并且节点返回的消息类型要配合。4.3 检查点序列化State里塞自定义对象保存时直接炸这个问题我在第2章提了一句再展开细说一次。当时我用了一个自定义的OrderServiceClient对象放在State里想省去每次节点重新初始化的麻烦。结果编译时一切正常一执行到检查点保存直接报序列化错误整个请求失败。LangGraph的检查点机制需要把State快照序列化后存储自定义对象如果没实现序列化方法就会在这里爆雷。我的解决方案是State里只放字符串、数字、列表、字典这些JSON友好类型所有需要外部服务的客户端对象在建图时作为闭包或全局对象传入节点函数而不是放进State。这样不仅检查点能正常工作整个图状态也能随时打印出来调试对排查路由问题帮助极大。4.4 interrupt挂起后用Command(resume...)恢复而不是重新invoke人工审批功能做好后我第一次测试时用了一个很笨的办法审批通过后把整张图重新invoke一遍结果发现流程从START重新跑分类节点又执行了一次完全不是想要的从暂停处继续。正确做法是先通过graph.get_state(config)拿到当前状态确认next字段还是那个等待审批的节点然后调用graph.invoke(Command(resume{approved: True}), config)引擎会从interrupt被挂起的位置恢复执行把resume里传入的数据作为interrupt的返回值继续往下跑。这才是我想要的断点续跑语义。这个坑让我深刻理解了一点interrupt和resume是成对出现的机械操作interrupt让图停下来resume让图从停下来之前的地方接着走。它不是重跑一遍流程而是精确回到暂停点。理解了这个很多人工审批怎么做的困惑都会消失。4.5 条件边写错映射图在原地死循环靠recursion_limit兜底有段时间我在条件路由函数里写了一个错误映射把A节点配置成返回B又把B节点的条件边配置回A结果一个工单在A、B之间来回跳。日志刷得飞快最后被LangGraph的recursion_limit拦住报错。这个错误的隐蔽性在于没有语法错误也没有类型错误纯粹是路由逻辑写成了环。我后来养成了几个习惯一是在构图时先梳理清楚每个节点最多有几个下游出口、是否可能回到上游二是在State里加一个steps字段每次节点执行就累计步数超过预设阈值直接结束并标记异常三是把recursion_limit从默认值调低让死循环更早暴露。这些措施加进去之后即使路由写错了也只是单个工单走到异常终止不会拖垮整个服务。5. 上线前补课人工审核、超时兜底和路由质量监控功能跑通后离真正上生产还有一段距离。这一节我讲讲上线前必须处理的几类问题全是实战里很可能忽略但后果严重的事。5.1 人工审核节点的细节挂起不是终点要能感知、要能操作、要能超时interrupt把图挂起之后外部系统需要有一个发现挂起、处理挂起的完整闭环。我在服务里做了三件事第一轮询或监听所有进行中的thread一旦发现状态里有interrupt请求就生成一条待审批事件推给运营后台第二运营后台审批通过后调用Command(resume...)让图继续第三设置审批超时时间比如24小时没有处理的工单自动执行一个超时节点把工单标记为超时未处理并通知相关人员。很多初学者做到第二步就停了觉得能审批就完了但真实场景里没人审批的工单会一直占着资源所以超时兜底一定要有。我还在人工审核节点里额外加了日志记录审批人ID、审批动作、时间戳为后续审计留痕。5.2 调用大模型的失败兜底分类挂了不能让整条流程跟着死大模型接口不可能永远稳定超时、限流、返回格式异常都是常态。如果classify节点抛异常默认整张图会失败工单卡在原地这对生产是不能接受的。我的兜底策略是在每个调用大模型的节点里做异常捕获分类失败时返回other分类走兜底节点在result里标记分类失败已降级处理同时把异常信息写入messages。这样流程不会死运营还能从状态里看到分类失败的事实后续人工介入。我还给模型调用本身套了超时参数宁可这次判断返回慢也不能让整个图流程线程一直挂着。5.3 路由质量监控模型分类准不准用数据说话而不是感觉既然分类结果决定了流程走向那分类的准确率就决定了整个系统的质量。我上线后做了一套简单的质量监控每个节点执行时打结构化日志记录工单号、节点名、入参摘要、出参、耗时。然后聚合统计各分类路径的占比发现技术类占比异常高我就知道可能是提示词引导太激进或者工单文本分布发生了变化。除了统计占比我还做了一个离线回归测试集从历史工单里抽出200条有明确人工标签的样本每次调整提示词或模型参数后把这200条重新跑一遍图对比分类和路由结果与人工标签的差异。这个方法极其朴素但极其有效它让我在改提示词时有了可以量化的判断依据而不是感觉准了一点。5.4 别急着上多Agent先把单图层级玩透后面有同事建议我把系统改成多个Agent互相协作的形态比如专门的客服Agent、退款Agent、技术Agent让它们自己对话。我的判断是对工单分诊这个场景单一流程图加条件路由已经足够清晰强行上多Agent只会增加不可控性。LangGraph同样支持更复杂的子图和多Agent协作但它的底层仍然是状态、节点、边和条件路由这四个基础概念。先把单图建模建明白多Agent就是在这个基础上把某些节点替换成子图而已顺序搞反了会非常痛苦。写到最后的一点体会这套工单系统上线跑了两个多月最大的感触是LangGraph真正改变我工作习惯的不是那个图引擎本身而是它逼我在写代码之前必须画出一张状态转换表——哪个节点读哪些字段、改哪些字段、下一步可能去哪儿、异常了走哪条边。把这些想清楚之后后面的代码基本是顺水推舟。如果你现在正准备拿LangGraph接一个大模型驱动的流程应用我建议你也先花半小时把这几个问题写在纸上然后再写代码肯定能少走不少弯路。最后分享一个小技巧我每次给图新增节点时会在路由函数里同时加一个日志点把当前state摘要 路由选择结果打印出来跑一轮测试自然就能在日志里看到完整的路径链路。这个小习惯帮我抓到了很多条件分支配错的问题比对着文档检查配置高效得多。
阅读完成 · 觉得有帮助？
