1. 先搞清楚Agent 跑一半突然挂了你的进度还剩多少我最早用 LangGraph 做多步骤 Agent 的时候踩过一个特别真实的坑一个包含信息收集 → 方案生成 → 人工确认 → 执行落地四步的自动化流程跑到第三步人工确认时服务进程因为内存问题崩了。重启之后发现整个图从头开始跑前面收集到的所有信息全部丢光。更尴尬的是那个人工确认节点前面已经调用过一次外部 API重新执行意味着同一条工单被重复提交了一次。这就是很多人刚接触 LangGraph 时没意识到的问题LangGraph 的图默认是一次性的。你用invoke跑一张图进程结束状态就没了。如果你的图只是简单的输入 → 输出那没问题但只要你开始做多轮对话、人工审批、异步任务、长时间运行的 Agent或者任何需要跑一半停下来等外部输入的场景没有持久化就等于裸奔。这篇文章要聊的就是把 LangGraph 的**断点恢复Checkpoint / Interrupt和幂等执行Idempotent Execution**这两件事串起来后的完整实战方案。适合谁看已经跑通了 LangGraph 基础流程、开始做真实业务 Agent 的开发者尤其是涉及人工介入、定时任务、失败重试这些场景的人。先说明一下LangGraph 是 LangChain 社区推出的图编排框架它和 LangChain 的区别在于LangChain 强调的是链——固定的顺序调用LangGraph 强调的是图——有分支、有条件跳转、有循环并且每一步都能被暂停和恢复。断点恢复这个能力恰恰是图和链拉开差距的关键点。如果你现在做的 Agent 还停留在一次性问答这篇文章里的内容可能暂时用不上但只要你的流程开始出现下面三个信号——执行过程超过 5 分钟、需要人工中途确认、失败后希望接着跑而不是重头跑——那断点恢复和幂等就是绕不开的两个课题。下文所有内容都是我实际在项目里验证过的不是抄官方 demo。2. 断点恢复的底层逻辑Checkpoint 机制到底存了什么2.1 一个关键概念图的状态State不是变量是存档要理解断点恢复先得理解 LangGraph 里 State 的本质。看下面这个最简单的图定义from typing import TypedDict, Annotated from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver class AgentState(TypedDict): user_request: str collected_info: dict approval_result: str | None def collect_info(state: AgentState): # 模拟外部 API 调用比如查库存 info {stock: 100, price: 299} return {collected_info: info} def final_answer(state: AgentState): return {approval_result: f已确认: {state[collected_info]}} builder StateGraph(AgentState) builder.add_node(collect, collect_info) builder.add_node(answer, final_answer) builder.add_edge(START, collect) builder.add_edge(collect, answer) builder.add_edge(answer, END)这段代码跑起来没问题但如果你不给compile()传 checkpointer图每次执行后 State 就消失了。State 不是你函数里的局部变量它是一份执行存档。每个节点函数接收 state、返回部分更新LangGraph 会把这些更新合并成新的状态传给下一个节点。这个存档如果只存在于内存中进程一结束全没如果落盘了图就能从任何一个节点继续往下走。LangGraph 的 checkpointer 就是负责做这件事的组件。它的工作机制是图每执行完一个节点更准确地说每完成一个 super-step就把当前状态和执行的进度信息写入存储后端。你想成打游戏自动存档就行——每过一个检查点就保存一次存档里记录了你在第几关、血量多少、背包里有什么。之后哪怕游戏崩溃、电脑重启读档之后还是接着上次的位置玩。2.2 Checkpointer 的几种存储后端别一上来就选内存LangGraph 里 checkpointer 是接口化的官方提供了多种实现后端存储位置适用场景我的评价MemorySaver进程内存单进程、演示重启全丢基本只适合开发调试SqliteSaverSQLite 文件单机持久化、中小规模实战首选轻量可靠AsyncSqliteSaverSQLite 文件异步FastAPI 等异步场景注意线程模型见后文避坑PostgresSaverPostgreSQL多实例、分布式部署需要单独建表适合生产集群我在实际项目里用的是SqliteSaver文件存本地既避开内存方案的重启即失忆又比上 Postgres 省掉一堆运维成本。初始化和编译的写法很固定from langgraph.checkpoint.sqlite import SqliteSaver # 注意SqliteSaver 需要传入一个连接对象 import sqlite3 conn sqlite3.connect(agent_state.db, check_same_threadFalse) checkpointer SqliteSaver(conn) graph builder.compile(checkpointercheckpointer)这里有一个我在官方文档里没看到但实际踩过的坑check_same_threadFalse这个参数一定要加上。原因很简单LangGraph 的图执行可能在不同的线程中被调用比如你在 FastAPI 的异步 handler 里调用图SQLite 默认不允许跨线程使用同一个连接不加这个参数跑着跑着就给你抛一个ProgrammingError: SQLite objects created in a thread can only be used in that same thread。还有一点关于 LangGraph 的版本演进早期大家熟悉的langgraph.checkpoint.sqlite.SqliteSaver在不同的版本中有过 API 调整部分版本引入了AsyncSqliteSaver专供异步环境。如果你用的是较新的版本需要严格区分同步调用和异步调用对应哪一种 saver别混用。2.3 Thread一张图可以开无数个存档槽位引入 checkpointer 之后你每次调用图就不再是简单的跑一次而是在某个存档槽位上跑一次。这个存档槽位就是thread_idconfig {configurable: {thread_id: order-review-001}} result graph.invoke( {user_request: 帮我生成采购单并等待审批}, configconfig )同一个thread_id下的多次调用共享同一个状态。这个设计的意义特别大你可以把一次完整的 Agent 执行拆成多次invoke调用每次调用之间可以间隔任意时间甚至跨进程、跨机器。这就是断点恢复的前提——图能认出这是我的旧档而不是每次都开新档。换个更直白的说法没有thread_id的调用就像在网吧临时上机人走了电脑一关记录清零有thread_id的调用就像有自己的会员档案什么时候回来登录之后还能从上次的下机位置继续玩。2.4 interrupt_before / interrupt_after在图的任意位置踩刹车有了存档机制还差一个刹车能力。LangGraph 的interrupt_before和interrupt_after就是干这个的。它们的语义是interrupt_before[node_name]在进入该节点之前暂停interrupt_after[node_name]在该节点执行完之后暂停实际效果是图运行到这个位置后不会继续往下走而是返回一个中断信号把当前状态和我停在哪了的信息都存下来然后等待你下次调用。graph builder.compile( checkpointercheckpointer, interrupt_before[human_approval], # 在人工审批节点之前停下来 )我第一次用这个功能时的感受是LangGraph 把Agent 停下来等人这件事变成了语言层面的能力而不是自己拿 while 循环 状态标记硬凑。它是图执行引擎自己认识的一种状态恢复的时候也是引擎自动接管从断点接着跑。3. 一个完整实战带人工审批的多步 Agent 加上断点恢复光说原理不够我直接把一个真实改过的项目简化成一个可复现的案例。场景是用户提交一个采购申请Agent 自动查库存、生成采购方案然后停下来等人工审批审批通过后再写数据库、发通知。3.1 先定义状态和节点from typing import TypedDict, Annotated, Literal from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.sqlite import SqliteSaver import sqlite3 class PurchaseState(TypedDict): item_id: str quantity: int stock_info: dict plan: dict approval: str | None order_id: str | None def check_stock(state: PurchaseState): # 模拟查库存真实环境这里可能是调用 ERP 接口 stock_info {item_id: state[item_id], available: 50} print(f[1/4] 查询库存: {stock_info}) return {stock_info: stock_info} def generate_plan(state: PurchaseState): qty state[quantity] available state[stock_info][available] if qty available: plan {status: insufficient, suggested_qty: available} else: plan {status: ok, suggested_qty: qty} print(f[2/4] 生成采购方案: {plan}) return {plan: plan} def human_approval(state: PurchaseState): # 正常情况下这个节点不会被真正执行因为我们在它前面踩了刹车 print([3/4] 人工审批通过) return {approval: approved} def execute_order(state: PurchaseState): # 模拟写数据库、发通知 print([4/4] 创建订单、发送通知) order_id fPO-{state[item_id]}-001 return {order_id: order_id} builder StateGraph(PurchaseState) builder.add_node(check_stock, check_stock) builder.add_node(generate_plan, generate_plan) builder.add_node(human_approval, human_approval) builder.add_node(execute_order, execute_order) builder.add_edge(START, check_stock) builder.add_edge(check_stock, generate_plan) builder.add_edge(generate_plan, human_approval) builder.add_edge(human_approval, execute_order) builder.add_edge(execute_order, END)这里我故意把human_approval写成普通节点函数因为它本质上就是一个人回来之后点击批准按钮这个动作的占位符。后面你会看到这个节点的真正逻辑不在函数里而在我们恢复执行时的调用方式里。3.2 编译时挂上 checkpointer 和中断点conn sqlite3.connect(purchase_agent.db, check_same_threadFalse) checkpointer SqliteSaver(conn) graph builder.compile( checkpointercheckpointer, interrupt_before[human_approval], ) config {configurable: {thread_id: purchase-2025-001}}interrupt_before[human_approval]的含义是当图执行到generate_plan并准备进入human_approval时立刻停下来。此时human_approval函数本身没有被调用图的状态停留在即将进入该节点的位置。3.3 第一次 invoke跑到断点停下result graph.invoke( { item_id: ITEM-042, quantity: 80, }, configconfig, ) print(返回结果:, result) print(图的执行状态:, result.get(__interrupt__))执行后控制台输出大致是[1/4] 查询库存: {item_id: ITEM-042, available: 50} [2/4] 生成采购方案: {status: insufficient, suggested_qty: 50} 返回结果: {item_id: ITEM-042, quantity: 80, stock_info: {...}, plan: {...}, approval: None, order_id: None, __interrupt__: (Interrupt(value()),)}关键点在这里前两个节点正常执行到了human_approval之前停了下来execute_order没有被执行数据库也没有被写入。返回的 dict 里多了一个__interrupt__字段它就是引擎告诉你我停在这了的信号。实际项目中你拿到这个结果后应该把它存到业务数据库里或者直接让 API 返回给前端前端展示等待审批中的状态。审批人点击通过后后端再继续调用图。3.4 恢复执行带着审批结果从断点继续跑现在人工审批通过了你要让图继续跑。注意恢复执行和首次执行用的是同一个thread_id但入参变成了None。这是 LangGraph 的固定用法——invoke(None)表示继续上次的执行不修改状态# 模拟审批人在页面上点了同意 resume_result graph.invoke(None, configconfig) print(恢复执行结果:, resume_result)如果在某些版本中你传入的不是None而是某个值这个值会作为Command(resumevalue)传给被中断的节点。比如你的断点设置在某个需要用户输入文本的节点上恢复时就可以传用户输入的内容。但在我们当前的场景里审批动作本身不需要给节点函数传参直接None就行。控制台输出[3/4] 人工审批通过 [4/4] 创建订单、发送通知 恢复执行结果: {item_id: ITEM-042, quantity: 80, stock_info: {...}, plan: {status: insufficient, suggested_qty: 50}, approval: approved, order_id: PO-ITEM-042-001, __interrupt__: ()}这时候你观察到了一个关键现象前两个节点没有重新执行。check_stock没有再次打印查询库存generate_plan也没有重新生成方案。图直接从human_approval后面开始执行execute_order。这就是 checkpointer 存档的价值——节点不会重复跑外部副作用不会因为恢复而翻倍。等下我这里要澄清一下节点不会重复跑是 LangGraph 引擎层面的保证但它只保证图的执行轨迹从断点继续不保证你的节点函数在过去某个时间点是否已经产生过副作用。这句话听着像绕口令其实是理解幂等性的钥匙下文第 4 节专门展开。3.5 状态更新的实际操作审批不通过怎么办真实业务里审批人可能点的是拒绝。如果直接invoke(None)图会假设审批通过并继续往下走。正确做法是先更新状态里的审批字段再恢复执行from langgraph.types import Command # 审批拒绝先把状态里的 approval 改成 denied再恢复 graph.update_state( config, {approval: denied}, ) resume_result graph.invoke(None, configconfig)update_state是 LangGraph 提供的一个作弊方法它允许你在两次执行之间手动修改存档中的状态字段。修改完之后再恢复human_approval节点看到的状态里就是approvaldenied你可以在节点函数里判断这个字段做不同处理。我在真实项目中通常这样设计审批节点节点函数只负责检查state[approval]字段如果等于approved就继续等于denied就直接走END或另一个通知驳回的节点。审批动作本身由外部触发后端 API 调用update_stateinvoke(None)节点函数永远不直接接触审批界面。4. 真正要命的问题幂等执行为什么还要你自己控制4.1 先说清楚断点恢复和幂等的关系很多人看完第 3 节会以为既然 LangGraph 能从断点恢复不会重头执行那不就不会重复执行副作用了吗这个理解只对了一半。LangGraph 的断点恢复保证的是图不会重新执行已经走过的节点——注意是已经走过的节点。这里有个隐蔽的问题如果图执行到节点 A 时A 函数内部调用了一个外部 APIAPI 调用已经发出去了但还没来得及把结果写入 state进程就崩了。重启后LangGraph 发现节点 A 没有完成没有写出返回值于是重新执行 A。这时候外部 API 会被调用第二次。你品一下这个过程状态没有更新成功 ≠ 副作用没有发生。外部系统数据库、第三方接口、消息队列的写入一旦发出就无法撤回而 LangGraph 判定节点是否完成依据的是状态是否更新。这两者之间的时间差就是幂等性问题埋下的雷。4.2 幂等三个层次你现在在哪一层我自己的理解是幂等性在 Agent 执行链路里分三个层次层次含义需要做的工作引擎层图不会重复执行已完成节点LangGraph 自动保证节点层同一个节点函数被调用两次时不产生重复副作用需要自己写代码外部系统层即使重复请求到达外部系统能识别并去重需要和上游/下游系统约定幂等键第 3 节里的例子因为节点都是模拟的print 赋值所以看不出问题。一旦你开始动真格的——真的往数据库里插了订单记录、真的调了第三方下单 API、真的往消息队列里发了通知——就必须考虑节点层的幂等。我自己做过的蠢事是一个生成订单节点里直接执行了INSERT INTO orders结果在某次恢复测试中同一个线程 ID 被打断后重复调用了两次invoke(None)数据库里凭空多了一条重复订单。事后查日志发现第一次确实只跑到了断点但第二次恢复执行时execute_order被调用后前面的human_approval状态更新和execute_order之间出了个异常框架重试了一次节点。4.3 在节点函数里做幂等用状态字段当已执行标记最简单粗暴的幂等方案是在状态里加一个字段充当这步已经做过了的标记节点函数开头先查标记已有标记就直接返回不再执行副作用代码。class PurchaseState(TypedDict): item_id: str quantity: int stock_info: dict plan: dict approval: str | None order_id: str | None order_created: bool # 幂等标记 def execute_order(state: PurchaseState): # 幂等检查如果已经创建过订单直接短路 if state.get(order_created): print([幂等] 订单已存在跳过重复创建) return {} # 模拟调用外部订单服务 写数据库 order_id create_order_in_db(state[item_id], state[quantity]) print(f[4/4] 创建订单: {order_id}) return {order_id: order_id, order_created: True}当节点被意外重跑时第一次跑已经写入了order_createdTrue第二次跑进来第一个 if 就拦住了。这个思路简单直白而且不依赖外部系统纯靠 LangGraph 的 state 合并机制就能实现。缺点是如果外部 API 调用发生在order_created状态写入之前仍然有重复调用的窗口。更稳妥的方式是利用业务幂等键在调用外部系统时生成一个idempotency_key通常就是thread_id 节点名 业务参数的哈希传给下游系统。下游系统如果支持幂等键比如 Stripe 支付、订单系统普遍支持重复请求会直接返回第一次的结果副作用不会重复。这个属于系统设计层面的兜底建议有条件的项目把节点标记和幂等键两层都做上。4.4 重试和幂等的配合不是所有重试都该重跑图还有一类常见陷阱是图级重试和节点级重试的混淆。有些人在invoke调用外包裹一层try...except失败就重新invoke同一 config。这个做法在LangGraph 引擎还没有执行任何节点时是安全的但如果图已经执行到一半外层重试等于强行把已经完成一半的图从头再跑一遍checkpointer 甚至可能报错——因为它发现这个thread_id已经有过执行记录、但当前传入的状态和新图定义不一致。正确做法图执行异常时不要重新invoke整个图而是根据异常发生的节点位置决定是从断点恢复、还是用update_state修正状态后继续。如果要重试的其实只是单个节点内的一次外部 API 调用比如网络超时那应该在该节点函数内部自己做好重试而不是上升到图层面。import time def execute_order_with_retry(state: PurchaseState): if state.get(order_created): return {} # 节点内部重试最多3次每次间隔2秒 for attempt in range(3): try: order_id create_order_in_db(state[item_id], state[quantity]) return {order_id: order_id, order_created: True} except TimeoutError: if attempt 2: raise time.sleep(2)节点内部重试和幂等标记要配合好。上面的代码里如果第一次尝试已经成功写库、但因为网络问题没拿到返回值第二次尝试会重复写库——所以幂等标记的写入时机比重试逻辑的编写时机更讲究。理想顺序是先给外部系统发一个幂等键 → 成功后再更新本地标记 → 如果中途失败本地标记没写入重试时用同一个幂等键继续请求外部系统外部系统会返回第一次的结果。5. 我在实际项目里踩过的坑恢复执行的前前后后5.1 SQLite 连接和线程模型的坑值得先说前面提过check_same_threadFalse这里再补充一个更隐蔽的问题如果你在 FastAPI 里用AsyncSqliteSaver但图的调用是同步函数会出现事件循环阻塞。我在一个内部工具里把SqliteSaver换成了AsyncSqliteSaver并在 async handler 里调用图结果一跑就卡住。原因在于AsyncSqliteSaver的设计是给异步图调用用的必须配合graph.ainvoke()注意是 ainvoke而不是invoke。正确的写法是async def handle_approval(): async with AsyncSqliteSaver.from_conn_string(agent_state.db) as saver: graph builder.compile(checkpointersaver, interrupt_before[human_approval]) result await graph.ainvoke(None, configconfig)如果你是同步 FastAPI 或者普通的 Django 环境老老实实用SqliteSaver不要为了异步而异步。5.2 恢复执行时传入None还是新状态很多人搞混有个朋友在恢复执行时传入了完整的新状态比如graph.invoke({...所有字段...}, configconfig)结果抛了一个错或者图从某个奇怪的地方开始跑。这里要记清楚invoke(None)是继续之前没跑完的图invoke(状态字典)是开一个新执行从 START 开始。如果你已经有过thread_id的执行记录再传状态字典LangGraph 会在同一个 thread 上开启新一轮执行而不是接着断点继续这会导致非常难查的 bug——你以为是在恢复其实是在重开。判断当前图是否处于暂停待恢复状态可以通过graph.get_state(config)来查state_snapshot graph.get_state(config) print(state_snapshot.next) # 如果是一个非空元组说明还有节点待执行 print(state_snapshot.values) # 当前存档里的状态值如果next返回的值正是你断点所在的节点说明图正停在断点上等待恢复。这时候再invoke(None)才是正确的恢复动作。5.3 恢复之后想改流程怎么办用update_state而不是改代码有一次我在中断之后发现某个节点里依赖的外部接口换了参数格式。按直觉想应该改节点函数代码再重新跑。但 LangGraph 的机制是节点函数在编译时已经固定图执行状态里不保存函数代码。你改了函数代码对已经存档的 thread 没有任何影响。这时候正确做法是用update_state提前把新格式的参数写进状态里或者如果改动太大直接开一个新的thread_id重新执行。在实际业务中已经跑到中途的流程如果改了代码我倾向于直接作废旧 thread 新开一个避免新旧代码逻辑混在一个存档里排查起来精神分裂。5.4 断点恢复和幂等标记一起做时Markdown 式的 Checklist最后分享一个我自己整理的项目落地检查表每次接入新流程时照着过一遍[ ] 是否给compile()传了 checkpointer没有这个后面全白谈[ ] 是否在需要人工介入或需要长时间等待的位置配置了interrupt_before或interrupt_after[ ] 每次invoke是否固定使用thread_id是否在业务数据库里保存了 thread_id 和业务单据号的映射[ ] 恢复执行时确认用的是invoke(None)而不是误传状态字典[ ] 所有涉及外部副作用写库、调 API、发消息的节点是否都做了幂等标记检查[ ] 外部系统调用是否带了幂等键下游是否支持幂等去重[ ] 节点内部的失败重试是否会与幂等标记产生竞态先重试还是先标记这套检查表看起来简单但帮我挡住了至少三次线上事故。一次是忘了加 checkpointer 还开着两个 uvicorn worker导致 thread_id 对应的存档在两个进程里各写各的恢复时读到的状态不一致另一次是断点恢复后同一个节点因为状态更新超时被框架重试幸亏有幂等标记才没造成重复订单。把机制理解透把这些细节写进流程LangGraph 才能真正扛得住真实业务。我个人现在做 Agent 项目的习惯是凡是涉及外部系统写入的节点默认按幂等去写宁可多写几行检查代码也绝不指望不会重跑这种侥幸。断点恢复解决的是流程能不能继续幂等解决的是重复执行会不会造成破坏两者缺一不可。你先把这两件事想明白LangGraph 在你手里才不是个玩具而是能上生产的编排框架。