LangGraph 这个系列写到第 7 篇按惯例应该聊点真正能拿去上生产的东西了。前面几篇我们把 StateGraph 的节点、边、状态传递都过了一遍也写过几个带条件分支的 Demo但说句实在话从「能跑」到「扛得住真实业务流量」之间还隔着一整层功夫状态怎么持久化、循环怎么防止失控、并行分支的结果怎么合并、人工审核环节怎么嵌入编排、出问题的时候怎么排查。这一篇我打算用一个完整的智能工单处理工作流作为主线把这些生产级问题一个个拆开讲清楚。这个案例我挑了客服工单场景是因为它天然包含了大模型工作流编排里几乎所有的典型结构单轮识别、多分支路由、循环自查、并行处理、人工介入、结果合并。你在本地复现完这个例子再回头去做内容审核流、数据分析流或者 Agent 工具调用流很多设计思路是直接通用的。1. 先把整体架构想明白为什么选 LangGraph1.1 大模型编排和传统流程引擎不是一回事很多从后端转过来的同学有个惯性工作流不就是状态机配合消息队列吗我用 Spring StateMachine 或者干脆写个状态表也能串起来。这话搁在规则明确的业务场景里没错但一旦节点逻辑变成「让模型判断一下该走哪条路」「让模型写一版初稿然后另一路模型做质检」传统流程引擎就开始别扭了。别扭的根源在于三点。第一传统流程的状态机是确定性的每到一个节点输入输出结构是提前定死的而大模型节点的输出天然带不确定性同一个 prompt 换一种说法可能就变了状态结构需要更灵活地容纳模型返回的中间结果。第二大模型应用里循环逻辑极其常见——生成结果不满意的自纠错、格式校验不通过的重试、多轮工具调用的继续判断——传统流程引擎要么不支持循环要么会把循环写成丑陋的回调嵌套。第三也是最重要的大模型工作流普遍需要「有人在场」审核节点要人点按钮敏感操作要人确认多轮对话要等用户回复。传统流程引擎靠消息做异步驱动做人工审批那一套流程非常重。LangGraph 把图结构、状态管理、持久化、人工中断这几件事揉在了一起而且是用写代码的方式定义图不是拖拽配置。这一点对程序员很友好因为图的逻辑可以进 Git可以做 Code Review可以写测试用例。1.2 状态管理是 LangGraph 区别于普通代码调用的核心如果你只是按顺序调用几个大模型 API用 Python 脚本就够根本不需要图编排框架。LangGraph 的价值在于它把「每一步产生的结果」沉淀成了一个显式的、可持久化的状态对象。举个例子一个最朴素的工单处理流程raw_text - 模型识别意图 - 模型生成回复 - 返回普通脚本写法就是函数套函数中间变量全在内存里一旦进程重启上下文全丢。而 LangGraph 的写法是把这个流程拆成节点每个节点接收整个 State 对象、修改 State 对象、返回更新后的字段。图执行完之后这个 State 可以被序列化存储下次继续跑的时候可以从任意断点恢复。这个设计对复杂工作流的意义怎么强调都不过分。因为大模型节点的执行时间很长、成本很高如果中途某个节点挂了或者人工审核需要等待整个 State 能落盘、能恢复就意味着不需要从头重跑一遍前面的模型调用。1.3 不同编排方案的取舍对照我把几种常见方案放在一起对比一下方便你判断自己的场景到底该不该上 LangGraph。方案状态管理循环控制人工介入可观测性适用场景纯 Python 代码靠函数参数传递难持久化自己写 while容易乱自己写接口等回调靠 print 和日志实验脚本、一次性分析传统工作流引擎强类型状态需提前建模不擅长循环有审批流支持依赖引擎自带监控企业存量流程改造LangGraph灵活 Schema Checkpointer条件边天然支持循环interrupt 机制原生支持可配合 LangSmith 或自建日志大模型驱动、需人工确认、状态需恢复的智能应用我的建议是节点少于五六个、没有循环和人工介入需求的别折腾框架直接写代码节点多、有路由分支、需要人工审核、希望出错后能从断点续跑的上 LangGraph 是合理选择。2. 生产级工作流的状态治理与图结构设计2.1 StateGraph 三个核心概念再捋一遍虽然前面几篇反复讲过但生产场景下还是要再确认一次理解因为很多坑就是从概念模糊开始的。节点Node就是一个普通的 Python 函数入参是当前 State 对象或其子集返回值是一个字典里面是要更新的字段。节点函数理论上可以做任何事调大模型、查数据库、发消息、执行代码。边Edge定义节点之间的流转关系。普通边表示顺序执行条件边Conditional Edge根据一个路由函数的结果决定下一步去哪个节点。条件边是实现分支和循环的关键。状态State图的全局数据中枢。LangGraph 的 State 默认是 TypedDict每个字段可以配置为「覆盖」还是「追加」两种 reducer。覆盖就是我们常用的赋值更新追加则是把新值拼到已有列表后面非常适合用来收集多轮模型输出。这张图你要是能画清楚代码基本就不会走样START - 意图识别 - 优先级判断条件边高优先级走加急处理普通走常规处理 - 常规路径方案生成 - 自检循环最多2次 - 合并 - 加急路径并行处理 A/B 两路 - 合并 - 人工审核interrupt 暂停等待人工确认 - 终稿生成 - END2.2 状态 Schema 设计最容易踩的坑状态字段设计是 LangGraph 应用里影响面最大、返工成本最高的决策。我看到不少新手的第一个版本就是把所有中间结果都塞进一个字段用的还是覆盖模式结果并行分支一跑就互相覆盖。这里有一个我在实际项目里总结出的设计原则模型每产出一个「独立语义单元」就给它单独开一个字段。拿工单工作流举例至少要划分这几类字段输入类raw_text原始工单内容、ticket_id、customer_info识别类intent问题类型、priority优先级、sentiment情绪分值生成类draft_reply初稿、revised_reply修订稿、parallel_results并行分支结果列表控制类review_decision人工审核意见、retry_count重试计数、status为什么不能一股脑扔到一个messages列表里因为不同的字段更新频率不同raw_text只在入口写入一次draft_reply可能被循环节点反复覆盖parallel_results需要追加而不是覆盖。用 TypedDict 分开定义每个字段单独控制 reducer后面写条件判断的代码才清晰。还有一个容易忽略的细节LangGraph 的 State 里不要放不可序列化的对象比如模型实例、文件句柄、数据库连接。我之前犯过这个错图能跑但一旦启用 Checkpointer 做持久化就报错因为状态序列化的时候遇到 object 直接失败。所有资源类的东西都要在节点函数内部获取或者用懒加载方式处理。2.3 条件边与循环控制的正确姿势条件边是 LangGraph 最灵活也最容易出问题的部分。一个核心认知是条件边返回的不一定是「节点名称」你也可以返回多条边的映射。在实际项目中我强烈建议给每个条件分支写一个独立的路由函数不要直接在add_conditional_edges里写 lambda。原因很简单路由函数是调试的切入点独立命名函数可以加日志、加单元测试lambda 一旦报错你连从哪查起都不知道。循环控制在工单工作流里最常见的场景是生成质量自查。设计上要注意三点第一循环次数必须有硬上限。不管模型判断「还需改进」多少次最多重试 N 次就强制放行或降级。没有上限的循环就是生产事故的定时炸弹。第二循环计数器要作为显式状态字段维护。每轮循环在节点里读retry_count、加一、写回。这样整个循环逻辑是可控、可观测的。第三循环退出条件要写在路由函数里并且路由函数要能同时读到「当前结果」和「重试计数」两个信息否则你没法判断到底是继续还是结束。3. 完整实操打造一个可上线的智能工单处理工作流3.1 场景定义与节点流转设计我虚构一个业务场景某电商平台收到了用户提交的售后工单系统需要自动完成以下流程——识别问题类型、判断优先级、生成处理方案、自检优化、根据优先级串行或并行处理、人工复核、输出最终答复。图结构拆解如下classify对原始工单做意图识别和优先级判断输出intent、priorityroute_by_priority条件路由返回handle_urgent或handle_normalhandle_normal串行调用模型生成初步处理方案handle_urgent并行拆成async_solution_a和async_solution_b两路分别生成方案和风险分析quality_check对方案做自检不合格且未超过最大重试次数则回到handle_normalmerge合并并行/串行结果汇成一份待审核的完整回复human_review通过interrupt()暂停图执行等待人工确认finalize根据人工意见生成终稿最后流转到 END这个流程覆盖了单步判定classify、条件路由route_by_priority、循环quality_check 回头、并行urgent 路径、合并merge、人工介入human_review、终态处理finalize。3.2 代码实现状态定义、节点函数、图编译完整的可运行示例代码我已经按生产要求整理好关键细节都在注释里。import operator from typing import Annotated, Dict, List, Optional, TypedDict from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import InMemorySaver class TicketState(TypedDict): 工单处理工作流的状态定义。 每个字段都是独立语义单元控制更新方式。 ticket_id: str raw_text: str intent: Optional[str] priority: Optional[str] draft_reply: Optional[str] revised_reply: Optional[str] # parallel_results 使用追加模式防止并行分支相互覆盖 parallel_results: Annotated[List[Dict], operator.add] merged_reply: Optional[str] review_decision: Optional[str] review_comment: Optional[str] final_reply: Optional[str] retry_count: int max_retry: int def classify(state: TicketState) - Dict: 意图识别 优先级判断。 真实项目里这里调用大模型我精简成一个关键词判断函数。 text state[raw_text] urgent_keywords [紧急, 退款, 投诉, 马上] priority urgent if any(k in text for k in urgent_keywords) else normal intent after_sales if 退 in text else consult return { intent: intent, priority: priority, } def route_by_priority(state: TicketState) - str: 条件路由函数根据 priority 字段返回下一个节点名称。 这里必须单独成函数便于后续加日志和单元测试。 if state[priority] urgent: return handle_urgent return handle_normal def handle_normal(state: TicketState) - Dict: 串行处理生成一版初步方案。 draft f针对[{state[intent]}]的初步处理方案核实订单信息安抚用户情绪给出补偿方案。 return {draft_reply: draft} def async_solution_a(state: TicketState) - Dict: 并行分支A生成解决方案。 solution_a {channel: solution_a, content: 优先退款并赠优惠券24小时内处理。} return {parallel_results: [solution_a]} def async_solution_b(state: TicketState) - Dict: 并行分支B分析潜在风险。 solution_b {channel: solution_b, content: 风险点退款周期超过预期可能引发二次投诉需客服同步跟进。} return {parallel_results: [solution_b]} def handle_urgent(state: TicketState) - Dict: 并行处理入口这里只是标记。 实际上并行分支通过 add_edge 同时连接两个节点实现。 return {} def quality_check(state: TicketState) - Dict: 质量自检 1. 如果初稿为空强制重写。 2. 如果包含“跟进”等风险词且未超过最大重试次数触发循环。 draft state.get(draft_reply, ) or state.get(merged_reply, ) retry_count state[retry_count] if not draft: return {retry_count: retry_count 1} if 跟进 in draft and retry_count state[max_retry]: return {retry_count: retry_count 1} return {retry_count: retry_count} def route_by_quality(state: TicketState) - str: 循环路由回退到 handle_normal 继续生成还是进入 merge。 if state[retry_count] 0 and state[retry_count] state[max_retry]: draft state.get(draft_reply, ) or state.get(merged_reply, ) if 跟进 in draft: return handle_normal return merge def merge(state: TicketState) - Dict: 合并节点将并行结果或串行结果汇成待审核文本。 parts [state.get(draft_reply) or ] for item in state.get(parallel_results, []): parts.append(f[{item[channel]}]: {item[content]}) merged \n.join(p for p in parts if p) return {merged_reply: merged} def human_review(state: TicketState) - Dict: 人工审核节点图执行到这里会暂停等外部人工输入。 # 这是参考 LangGraph 官方代码风格编写的示意片段 # 具体 interrupt 调用需要结合所使用的 langgraph 版本 API 为准 mock_review input_or_hook(state) decision mock_review.get(decision, approve) comment mock_review.get(comment, ) return {review_decision: decision, review_comment: comment} def input_or_hook(state: TicketState) - Dict: 这里在实际项目中是等待外部系统回调示例里直接给默认值。 return {decision: approve, comment: 符合流程要求} def finalize(state: TicketState) - Dict: 终稿生成按人工意见修订。 if state[review_decision] approve: return {final_reply: state[merged_reply]} return {final_reply: state[merged_reply] \n[已按人工意见修订]} def build_graph(): 组装图结构。 graph StateGraph(TicketState) graph.add_node(classify, classify) graph.add_node(handle_normal, handle_normal) graph.add_node(handle_urgent, handle_urgent) graph.add_node(async_solution_a, async_solution_a) graph.add_node(async_solution_b, async_solution_b) graph.add_node(quality_check, quality_check) graph.add_node(merge, merge) graph.add_node(human_review, human_review) graph.add_node(finalize, finalize) graph.add_edge(START, classify) graph.add_conditional_edges(classify, route_by_priority, { handle_normal: handle_normal, handle_urgent: handle_urgent, }) graph.add_edge(handle_normal, quality_check) graph.add_conditional_edges(quality_check, route_by_quality, { handle_normal: handle_normal, merge: merge, }) # 并行分支handle_urgent 同时触发两个子节点 graph.add_edge(handle_urgent, async_solution_a) graph.add_edge(handle_urgent, async_solution_b) graph.add_edge(async_solution_a, merge) graph.add_edge(async_solution_b, merge) graph.add_edge(merge, human_review) graph.add_edge(human_review, finalize) graph.add_edge(finalize, END) return graph.compile(checkpointerInMemorySaver()) if __name__ __main__: app build_graph() # 生产环境用 thread_id 区分会话便于 Checkpointer 恢复状态 config {configurable: {thread_id: ticket-10086}} result app.invoke( {ticket_id: 10086, raw_text: 我买的商品有质量问题要求立即退款, retry_count: 0, max_retry: 2}, config ) print(final_reply:, result.get(final_reply))这段代码你在本地跑起来不会有问题但我要强调几点生产化改造的方向。InMemorySaver只是让你本地调试方便的进程一重启状态就没了。生产环境至少换成 SQLite 或 Postgres 的 Checkpointer 实现或者对接 Redis 这一类外部存储。Checkpointer 的选型直接影响你能不能在多实例部署时共享会话状态。human_review节点里的input_or_hook函数是我为示例简化的真实项目里应该用interrupt()让图真正停下来把待审核内容推给前端或消息队列等人工处理完再通过app.invoke(None, config)或类似方法恢复执行。这个机制我在第 5 节问题排查部分还会展开讲。3.3 并行分支的两个关键细节handle_urgent 到 async_solution_a 和 async_solution_b 这两条边在 LangGraph 里就是这么朴素的写法从一个节点同时拉出两条普通边。但实际执行时你要确认两件事。第一两个并行分支是否真的并行执行LangGraph 的默认调度器会对互不依赖的节点做并发但如果你在节点函数里用了同一个全局变量或同一个连接池并发就会变成问题。我建议所有资源都在节点函数内部独立获取。第二两个分支往哪个节点汇合示例里是汇到同一个 merge 节点。这里有个状态更新顺序的问题parallel_results用的是追加 reducer两个分支返回的列表会被依次 append 进去所以永远不需要担心覆盖。你只要保证最终结果不依赖合并顺序结构上就是安全的。3.4 人工审核节点如何在真实环境里落地人工介入是大模型工作流上生产绕不开的一环——尤其是面向用户的业务机器直接给答复风险太高。LangGraph 处理这件事的思路是「把图执行挂起等待外部输入」。我实际做过的方案是这样在 human_review 节点内调用interrupt()函数把当前展示给审核人员的所有上下文作为中断值返回外部系统拿到这个中断值后推送到内部审核后台审核人员操作完成后后台调用一次基于同一thread_id的恢复接口LangGraph 会从断点接着跑下去。这个模式有两个好处。一是中间状态全部通过 Checkpointer 持久化即使审核人员几天后才处理状态也不会丢失。二是整个工作流的状态历史天然就是审计日志谁在什么时候审核了什么内容一查就清楚。4. 方案深度解析模型选择、参数调优与性能考量4.1 工作流里每个节点是不是都需要用最强模型这是我在方案评审阶段被问得最多的问题。很多团队的习惯是既然上了大模型那就所有节点都用同一个最强的模型。这其实是成本浪费和延迟放大。我的做法是按节点职责给模型分级识别判断类节点比如 classify 意图识别、优先级判断输出是结构化的几个枚举值用轻量模型就够了。这类任务 prompt 写得清楚模型能力要求不高用大模型属于杀鸡用牛刀。内容生成类节点比如 handle_normal、async_solution_a涉及商务文案、方案表达需要中上档模型确保语言质量。质量自检和终稿生成节点既要理解语义又要做细节修改可以考虑安排更强的模型或者把自检拆成「格式校验 语义评估」两个子节点格式校验用规则引擎语义评估用模型。这样分级下来整体成本可能只有全用强模型的 40% 到 60%而用户体验上几乎感知不到差异。关键在于你要给每个节点定义清晰的输出约束用 Pydantic 或 JSON Schema 把模型输出锁死弱模型也能稳定工作。4.2 超时、重试与降级的工程实践在 LangGraph 编排里模型 API 调用是外部依赖超时和重试策略必须提前设计好。我这里给出一套我在项目中验证过的参数基线连接超时3 秒。超过直接判失败不等待。读超时轻量模型 20 秒生成类节点 60 秒。再久说明模型服务异常等下去没有意义。重试次数2 次。第一次失败马上重试第二次失败走降级。降级策略分类节点降级为默认值normal/consult生成节点降级为预设模板审核节点强制人工介入。这套基线的好处是每个节点最坏情况下有明确的兜底行为整个图不会因为某一次模型调用超时就整体挂死。你可以在 LangGraph 的节点函数里做一层统一封装把所有模型调用包进一个带超时和重试的 wrapper。4.3 并行度和资源管理并行节点的数量并不是越多越好。模型 API 通常有速率限制你一个工作流里同时跑 10 个并行节点每个都要调模型下一秒就可能触发限流。我在实际项目中按槽位模型来控制并发给每个节点分配一个信号量全局限流数按账号配额的一半设置剩余配额留给低优先级批处理任务。这样并行分支真正跑起来的实际并发就有一个安全边界。另外如果你的图是部署在 Serverless 容器里的并行节点的内存峰值要比单节点高好几倍部署配置要充分留余量。我在项目里用并行分支时容器内存基准直接按单节点的 3 倍申请。5. 常见问题与排查技巧实录5.1 状态字段更新了但下游节点读到的还是旧值这个坑非常经典。LangGraph 的每个节点返回值是「增量更新」只修改你返回的字段没返回的字段保持原值。但很多人的错觉是节点返回一个完整的 State 就以为整个 State 被替换了其实不是。排查方法是在每个节点函数的入口打一行日志打印关键字段的前 50 个字符跑完一个流程就能看出来是哪个节点丢数据。另外注意区分「覆盖模式」和「追加模式」如果你发现某个列表字段的值被后一次结果顶掉了大概率是没用Annotated[..., operator.add]声明追加模式。5.2 循环节点死循环任务一直不结束条件边构成的循环里没有天然终止条件时图就会像写了个 while True 一样一直跑下去。我在代码里加了retry_count和max_retry就是为了防这个。排查和预防手段第一max_retry必须有这是底线保护第二在路由函数里加日志每路由一次就输出当前计数和走向第三在编译图时设置递归调用深度限制很多版本 LangGraph 支持recursion_limit超过直接抛异常比无限循环好养活。5.3 并行分支的结果合并时出现字段缺失如果你在 merge 节点里遍历parallel_results偶尔会发现列表里只有一条记录。这通常是因为并行分支里的某个节点执行失败了而你没有捕获异常。LangGraph 的单个节点抛错会中断整个 pragma 的执行。我的经验是并行节点内部一定要自己捕获异常并返回错误标记字段比如{parallel_results: [{channel: solution_b, error: timeout}]}merge 节点要做「有错误标记就走降级」的判断而不是假设所有分支都成功。5.4 启用了 Checkpointer 后状态恢复不对Checkpointer 是按thread_id区分会话的。如果你换了新的 thread_id图会被当作全新状态开始执行如果复用同一个 thread_id 重新调用会接着上次的断点继续。常见错误是你把用户 ID 直接当 thread_id 用结果用户上一次没跑完的流程被新请求接着跑了数据串得莫名其妙。正确做法是每次业务请求生成一个独立的 request_id 作为 thread_id和用户 ID 做一个映射表这样既能断点续跑又能区分不同请求。5.5 可观测性日志埋点与追踪生产环境里 LangGraph 应用排查问题光靠 print 绝对不够。我在所有节点函数的入口和出口都加结构化日志字段包含节点名、thread_id、关键状态字段哈希、耗时。如果你是自建日志系统建议把每个节点的输入输出大小、耗时、模型调用是否成功、重试次数都塞进一条 JSON 日志。配合 LangFuse 这一类可观测工具就能把一次完整执行链路渲染成 waterfall 视图哪个节点慢、哪个节点失败一眼定位。日志埋点还有一个额外好处可以用日志数据离线统计分析图执行的成功率、节点耗时分布、各路由分支的占比这些数据对后续优化 prompt 和模型选型非常有用。6. 最后补充一些我自己实践下来的体会这一篇的示例代码完全是「能跑」的最小骨架但我希望你在理解它之后做几件事把InMemorySaver换成真实持久化存储把human_review节点改成真正的interrupt()挂起模式给每个节点加上超时重试和结构化日志。这些改造做完你的 LangGraph 工作流才算真正有了生产的样子。另外有一个小小的建议调试图的时候不要一上来就跑全流程。先用只含两三个节点的最小图验证状态传递和条件边逻辑确认无误后再逐步把并行、循环、人工审核这些复杂结构加进去。模块化验证在图编排里节省的调试时间远比表面上看起来多。