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 | 进程内存 | 单进程、演示 | 重启全丢，基本只适合开发调试 |
SqliteSaver | SQLite 文件 | 单机持久化、中小规模 | 实战首选，轻量可靠 |
AsyncSqliteSaver | SQLite 文件（异步） | FastAPI 等异步场景 | 注意线程模型，见后文避坑 |
PostgresSaver | PostgreSQL | 多实例、分布式部署 | 需要单独建表，适合生产集群 |
我在实际项目里用的是SqliteSaver，文件存本地，既避开内存方案的"重启即失忆"，又比上 Postgres 省掉一堆运维成本。初始化和编译的写法很固定：
from langgraph.checkpoint.sqlite import SqliteSaver # 注意：SqliteSaver 需要传入一个连接对象 import sqlite3 conn = sqlite3.connect("agent_state.db", check_same_thread=False) checkpointer = SqliteSaver(conn) graph = builder.compile(checkpointer=checkpointer)这里有一个我在官方文档里没看到但实际踩过的坑：check_same_thread=False这个参数一定要加上。原因很简单，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_id：
config = {"configurable": {"thread_id": "order-review-001"}} result = graph.invoke( {"user_request": "帮我生成采购单并等待审批"}, config=config )同一个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( checkpointer=checkpointer, 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 = f"PO-{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_thread=False) checkpointer = SqliteSaver(conn) graph = builder.compile( checkpointer=checkpointer, 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, }, config=config, ) 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, config=config) print("恢复执行结果:", resume_result)如果在某些版本中你传入的不是None而是某个值，这个值会作为Command(resume=value)传给被中断的节点。比如你的断点设置在某个需要用户输入文本的节点上，恢复时就可以传用户输入的内容。但在我们当前的场景里，审批动作本身不需要给节点函数传参，直接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, config=config)update_state是 LangGraph 提供的一个"作弊"方法，它允许你在两次执行之间手动修改存档中的状态字段。修改完之后再恢复，human_approval节点看到的状态里就是approval="denied"，你可以在节点函数里判断这个字段做不同处理。
我在真实项目中通常这样设计审批节点：节点函数只负责检查state["approval"]字段，如果等于approved就继续，等于denied就直接走END或另一个"通知驳回"的节点。审批动作本身由外部触发（后端 API 调用update_state+invoke(None)），节点函数永远不直接接触审批界面。
4. 真正要命的问题：幂等执行为什么还要你自己控制
4.1 先说清楚"断点恢复"和"幂等"的关系
很多人看完第 3 节会以为：既然 LangGraph 能从断点恢复，不会重头执行，那不就不会重复执行副作用了吗？这个理解只对了一半。
LangGraph 的断点恢复保证的是"图不会重新执行已经走过的节点"——注意，是"已经走过的"节点。这里有个隐蔽的问题：如果图执行到节点 A 时，A 函数内部调用了一个外部 API，API 调用已经发出去了，但还没来得及把结果写入 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_created=True，第二次跑进来第一个 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_thread=False，这里再补充一个更隐蔽的问题：如果你在 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(checkpointer=saver, interrupt_before=["human_approval"]) result = await graph.ainvoke(None, config=config)如果你是同步 FastAPI 或者普通的 Django 环境，老老实实用SqliteSaver，不要为了"异步"而异步。
5.2 恢复执行时传入None还是新状态？很多人搞混
有个朋友在恢复执行时传入了完整的新状态，比如：
graph.invoke({...所有字段...}, config=config)结果抛了一个错，或者图从某个奇怪的地方开始跑。这里要记清楚：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 在你手里才不是个玩具，而是能上生产的编排框架。
