1. 从一次多轮对话翻车说起为什么需要LangGraph去年底我接了一个智能客服的项目需求听起来很简单用户问问题系统查知识库回答如果答不上来就转人工。我一开始用传统的链式调用Chain搭了个原型前两轮对话跑得挺顺到第三轮就出问题了——用户说“那帮我改一下刚才那个订单的地址”系统完全不知道“刚才那个订单”指的是什么因为链式调用是无状态的每次请求都是全新的开始。这就是典型的对话状态管理缺失。传统Chain模式像一条单向流水线数据从A流到B再流到C中间没有记忆也没法根据情况拐弯。而真实业务里用户会追问、会改主意、会突然切换话题甚至会在关键节点需要人工审核。这些需求逼着我从Chain转向了LangGraph。LangGraph的核心思路是把对话流程建模成一张图Graph节点Node是处理步骤边Edge是流转方向而状态State贯穿整张图像一条河一样带着上下文从上游流到下游。更关键的是它支持条件路由——根据当前状态决定下一步走哪个分支支持人工介入——在关键节点暂停等人确认后再继续还支持Checkpointer——把每一步的状态持久化随时可以恢复。这篇文章我会用一个完整的案例把这些能力串起来一个带人工审核的智能工单处理系统。用户提交工单系统自动分类、检索知识库、生成回复草稿如果置信度低或者涉及敏感操作就暂停等人工审核审核通过后继续执行。整套流程涉及状态定义、条件分支、中断恢复、持久化存储我会把每一步的代码和踩过的坑都摊开讲。适合谁看如果你已经用过LangChain的基础Chain想进阶到有状态、可中断、可恢复的复杂流程这篇就是为你写的。如果你还没接触过LangGraph也没关系我会从核心概念讲起保证你能跟着跑通。2. 核心概念拆解State、Node、Edge、Checkpointer到底怎么配合2.1 State不是普通变量它是图的“共享内存”很多人第一次接触LangGraph的State会把它当成一个普通的字典传来传去。其实不是。State是整张图的共享内存每个节点都能读到它也能往里面写东西。LangGraph用了一个很巧妙的设计你定义一个TypedDict或者Pydantic模型来描述State的结构然后每个节点函数接收当前State返回一个增量更新partial update框架会自动帮你合并。举个例子我定义的工单State长这样from typing import TypedDict, Annotated from langgraph.graph import add_messages class TicketState(TypedDict): messages: Annotated[list, add_messages] ticket_id: str category: str confidence: float draft_reply: str needs_human: bool human_approved: bool这里有个关键点messages字段用了Annotated[list, add_messages]。这个add_messages是一个reducer函数它告诉LangGraph当多个节点都往messages里写东西时不要覆盖而是追加。默认情况下如果你不指定reducer后写的会覆盖先写的。这个设计在对话场景里特别重要因为每一轮对话都要保留历史。注意State的字段尽量用扁平结构不要嵌套太深。我试过嵌套三层字典结果调试时打印出来一团糟后来改成扁平结构加前缀命名清晰多了。2.2 Node是纯函数但别在里面做副作用操作Node就是图里的一个处理单元本质上是一个Python函数接收State返回更新。LangGraph的Node有一个很重要的约束尽量保持纯函数。什么意思就是同样的输入应该产生同样的输出不要在Node里面直接写数据库、发邮件、调外部API产生副作用。为什么因为LangGraph支持重放replay和恢复resume。如果Node里有副作用恢复时可能会重复执行。正确的做法是把副作用操作抽出来要么放在Node返回之后由外部处理要么用幂等设计。比如我一开始在“发送回复”这个Node里直接调了邮件API结果人工审核恢复后邮件被发了两次。后来改成Node只负责生成“待发送”标记真正的发送逻辑放在图执行完成之后统一处理。2.3 Edge分两种普通边和条件边普通边就是A节点执行完直接到B节点用add_edge(A, B)。条件边则是根据State里的某个值决定下一步去哪用add_conditional_edges。条件路由是LangGraph最实用的能力之一。我的工单系统里有三个分支如果分类置信度高于0.8且不涉及敏感操作直接自动回复如果置信度在0.5到0.8之间走人工审核如果低于0.5直接转人工并标记为“需人工处理”。def route_after_classify(state: TicketState): if state[confidence] 0.8 and not state[needs_human]: return auto_reply elif state[confidence] 0.5: return human_review else: return escalate这个路由函数返回的是目标节点的名字LangGraph会根据返回值把流程导向对应的节点。注意路由函数本身不修改State它只做判断。2.4 Checkpointer是状态持久化的关键Checkpointer是LangGraph里最容易被忽视但最重要的组件。它负责在每一步执行后把State保存下来这样当流程中断比如等人工审核后可以从断点恢复而不是从头再来。LangGraph提供了几种CheckpointerMemorySaver存在内存里适合开发和测试SqliteSaver存在本地文件适合单机部署PostgresSaver存在数据库适合生产环境。我一开始用MemorySaver重启服务后所有待审核的工单全丢了后来换成SqliteSaver才解决。Checkpointer的工作机制是每次图执行到一个“超级步”super-step结束时自动保存当前State的快照并关联一个thread_id。恢复时用同样的thread_id调用就能从上次中断的地方继续。实操心得thread_id的命名要有业务含义比如用工单号或者用户ID加时间戳。我试过用随机UUID结果排查问题时根本对不上号。3. 完整案例带人工审核的智能工单处理系统3.1 系统流程设计与状态定义先把这个系统的完整流程画清楚用文字描述不用图用户提交工单进入classify节点系统对工单内容做分类比如“退款”、“技术故障”、“咨询”并给出置信度。根据置信度和是否敏感路由到三个分支之一auto_reply、human_review、escalate。auto_reply节点生成回复草稿并直接发送。human_review节点生成草稿后中断等待人工审核。人工审核通过后继续到send_reply不通过则到revise节点重新生成。escalate节点直接标记为人工处理不生成自动回复。所有分支最终汇聚到finalize节点记录处理结果。State的定义我前面已经给了这里补充一下每个字段的用途字段类型用途messageslist对话历史用add_messages追加ticket_idstr工单唯一标识categorystr分类结果confidencefloat分类置信度draft_replystr生成的回复草稿needs_humanbool是否涉及敏感操作human_approvedbool人工审核是否通过3.2 节点实现分类、生成、审核、发送分类节点我用了一个简单的关键词匹配加LLM判断的混合策略。纯LLM分类有时候不稳定加上关键词规则可以兜底。def classify_node(state: TicketState): content state[messages][-1].content # 关键词规则 sensitive_keywords [退款, 注销, 投诉] needs_human any(kw in content for kw in sensitive_keywords) # LLM分类这里用伪代码表示 category, confidence llm_classify(content) return { category: category, confidence: confidence, needs_human: needs_human }生成回复草稿的节点def generate_draft_node(state: TicketState): prompt f工单分类{state[category]}\n用户问题{state[messages][-1].content}\n请生成回复草稿。 draft llm_generate(prompt) return {draft_reply: draft}人工审核节点本身不做事它只是一个中断点。LangGraph的中断机制是在编译图时指定interrupt_before或interrupt_aftergraph builder.compile( checkpointercheckpointer, interrupt_before[human_review] )这样当流程到达human_review节点之前会自动暂停把State保存到Checkpointer然后返回控制权给调用方。调用方拿到中断信号后可以展示草稿给人工审核审核完再调用graph.invoke(None, config)继续执行。发送节点def send_reply_node(state: TicketState): if state.get(human_approved): # 实际发送逻辑 send_email(state[draft_reply]) return {messages: [AIMessage(contentstate[draft_reply])]} else: return {messages: [AIMessage(content审核未通过需修改)]}3.3 条件路由的三种分支与优先级路由函数我前面给了简化版实际项目中要考虑更多情况。比如“退款”类工单即使置信度很高也必须人工审核因为涉及资金操作。所以路由逻辑要加一层敏感判断def route_after_classify(state: TicketState): if state[needs_human]: return human_review if state[confidence] 0.8: return auto_reply elif state[confidence] 0.5: return human_review else: return escalate这里有个优先级问题needs_human的判断放在最前面因为敏感操作优先级最高。我踩过的坑是先把置信度判断放前面结果一个“退款”工单因为置信度0.9直接自动回复了差点造成资金损失。注意条件路由的返回值必须是图中已定义的节点名否则运行时会报错。建议把节点名定义成常量避免拼写错误。3.4 Checkpointer配置与中断恢复实操Checkpointer的配置很简单但有几个细节要注意from langgraph.checkpoint.sqlite import SqliteSaver checkpointer SqliteSaver.from_conn_string(tickets.db) graph builder.compile( checkpointercheckpointer, interrupt_before[human_review] )调用时传入thread_idconfig {configurable: {thread_id: ticket_12345}} # 第一次调用会在human_review前中断 result graph.invoke({messages: [HumanMessage(content我要退款)]}, config) # 此时State已保存可以读取 state graph.get_state(config) print(state.values[draft_reply]) # 人工审核通过后更新State并继续 graph.update_state(config, {human_approved: True}) result graph.invoke(None, config)这里的关键是graph.update_state它允许你在中断后修改State。我一开始不知道这个API试图重新invoke整个图结果又从头跑了一遍。实操心得interrupt_before和interrupt_after的区别要搞清楚。interrupt_before是在节点执行前暂停interrupt_after是在节点执行后暂停。人工审核场景一般用interrupt_before因为审核前不需要执行审核节点本身。4. 踩坑实录状态管理、路由、中断恢复的常见问题4.1 状态覆盖问题为什么我的messages只剩最后一条这是新手最容易踩的坑。如果你定义State时没有给messages加reducer每次节点返回新的messages列表时会直接覆盖旧值。我一开始就是这么写的class State(TypedDict): messages: list # 没有reducer结果对话到第三轮历史全没了。正确的写法是from typing import Annotated from langgraph.graph import add_messages class State(TypedDict): messages: Annotated[list, add_messages]add_messages会自动把新消息追加到列表末尾而不是替换。如果你需要自定义合并逻辑也可以写自己的reducer函数。4.2 条件路由不生效检查返回值是否匹配节点名条件路由不生效通常有两个原因一是路由函数返回值拼写错误二是节点名和返回值不一致。LangGraph在编译时不会校验路由返回值只有运行时才会报错。我的建议是把节点名定义成常量NODE_CLASSIFY classify NODE_AUTO_REPLY auto_reply NODE_HUMAN_REVIEW human_review NODE_ESCALATE escalate路由函数返回这些常量添加边时也用常量这样拼写错误在IDE里就能发现。4.3 中断恢复后重复执行Checkpointer和幂等设计前面提到过如果Node里有副作用恢复时可能重复执行。除了把副作用抽出来还有一种方案是幂等设计。比如发送邮件时用工单ID作为幂等键发送前先查一下是否已发送过。def send_email_idempotent(ticket_id, content): if redis.get(fsent:{ticket_id}): return send_email(content) redis.set(fsent:{ticket_id}, 1, ex86400)这样即使重复执行也不会真的发两次。4.4 常见问题速查表问题现象可能原因解决方法messages只剩最后一条未加add_messages reducer用Annotated[list, add_messages]条件路由不跳转返回值与节点名不匹配用常量定义节点名中断后无法恢复未传thread_id或Checkpointer未配置检查config和checkpointer恢复后重复执行副作用Node内有非幂等操作抽离副作用或做幂等设计State更新不生效返回了完整State而非增量只返回需要更新的字段图编译报错节点名重复或边指向不存在的节点检查add_node和add_edge5. 进阶技巧让LangGraph流程更稳、更好维护5.1 用子图拆分复杂流程当流程节点超过10个时整张图会变得很难维护。LangGraph支持子图Subgraph可以把一组相关节点封装成一个子图然后像普通节点一样嵌入主图。比如我把“分类路由”封装成一个子图“生成审核发送”封装成另一个子图主图只负责编排。子图的State可以和主图不同通过输入输出映射来衔接。这个特性在大型项目里特别有用不同团队可以各自维护自己的子图。5.2 状态版本管理与迁移生产环境中State的结构可能会变化比如新增字段、修改字段类型。如果Checkpointer里存的是旧版本State恢复时会出错。LangGraph的Checkpointer支持版本迁移你可以在保存State时带上版本号恢复时根据版本号做转换。我目前的方案是在State里加一个schema_version字段每次结构变更时递增恢复时检查版本并做兼容处理。虽然有点土但很实用。5.3 可观测性日志、追踪与调试LangGraph的调试信息默认比较简略建议接入LangSmith或者自己打日志。我在每个节点入口和出口都加了日志import logging logger logging.getLogger(__name__) def classify_node(state: TicketState): logger.info(fclassify_node input: ticket_id{state[ticket_id]}) # ... 处理逻辑 logger.info(fclassify_node output: category{category}, confidence{confidence}) return {...}这样出问题时可以快速定位是哪个节点、哪一步出了偏差。另外graph.get_state(config)可以随时读取当前State配合日志使用效果很好。5.4 性能优化减少不必要的LLM调用LLM调用是整条链路里最慢也最贵的环节。我的优化策略是分类节点先用关键词规则过滤只有规则无法判断时才调LLM生成草稿节点缓存常见问题的回复模板命中模板直接返回不调LLM。实测下来整体响应时间从平均3.2秒降到了1.1秒成本也降了六成。实操心得LangGraph的节点是串行执行的如果某个节点特别慢可以考虑把它拆成多个并行节点用add_edge的并行能力加速。不过并行节点之间的State合并要小心确保reducer能正确处理并发写入。6. 我个人在实际操作中的几点体会这套工单系统上线跑了三个月处理了大概两万多个工单人工审核的介入率从最初的35%降到了12%自动回复的准确率稳定在91%左右。回过头看LangGraph最让我满意的不是它的功能有多强而是它的心智模型足够简单State是共享内存Node是处理单元Edge是流转规则Checkpointer是存档点。把这四个概念吃透剩下的就是业务逻辑的堆叠。如果让我给刚上手的人一条建议那就是先把State定义清楚再写节点。我见过太多人上来就写节点逻辑写到一半发现State不够用回头改State导致所有节点都要调整。State是整张图的契约契约定好了后面的实现就是水到渠成的事。另外Checkpointer千万别等到上线才配。开发阶段就用SqliteSaver养成每步持久化的习惯调试时会感谢自己的。MemorySaver只适合跑单元测试真实场景下服务一重启数据就没了这个坑我替你们踩过了。
