Checkpoint、人工中断与重复执行
暂停不会冻结 Python 栈。LangGraph 保存状态，恢复时重新进入包含 interrupt 的节点，已提供的恢复值成为 interrupt 的返回值。因此中断前的副作用可能再次发生，不能把“有 checkpoint”理解为“每行只执行一次”。
观察节点重新进入
复用第一章环境，完整实验用列表记录节点进入次数，没有外部写入：
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt, Command
class State(TypedDict):
approved: bool
visits = []
def approve(state):
visits.append("entered")
decision = interrupt({"action":"create draft", "version":1})
return {"approved":decision is True}
b = StateGraph(State)
b.add_node("approve",approve)
b.add_edge(START,"approve"); b.add_edge("approve",END)
g = b.compile(checkpointer=InMemorySaver())
config = {"configurable":{"thread_id":"tenant-a:task-7"}}
first = g.invoke({"approved":False},config)
assert "__interrupt__" in first and len(visits) == 1
last = g.invoke(Command(resume=True),config)
assert last["approved"] is True and len(visits) == 2
assert g.get_state(config).next == ()
print(len(visits),last)第一次进入节点后暂停，第二次从节点顶部运行，故 visits 为2。将发信放在 append 的位置会重复发送。应把审批节点与效果节点分开，效果节点另有业务幂等键；即使节点拆开，崩溃发生在外部成功但 checkpoint 尚未提交时，仍需幂等保护。
thread_id、存储与访问权
thread_id 定位一组检查点，不是鉴权令牌。服务端应验证用户属于任务，再构造 config；不能允许客户端任意读取别人的 thread。InMemorySaver 在进程退出后丢失，不能演示重启恢复；生产选择数据库 saver，配置连接、迁移、保留删除与加密，并实测跨进程恢复。
恢复前重新检查动作摘要、审批有效期、资源版本和当前权限。Command(resume=False) 是拒绝，不能因为返回值存在就当同意；输入值也需要严格类型与 schema。普通失败后的继续执行和人工 interrupt 的 resume 是不同入口，不要把 invoke(None) 当作一切暂停的通用批准。
演进、回放与部署
旧教程的 MemorySaver 名称与新版 InMemorySaver 存在命名演进，关键仍是内存生命周期；换名字不会获得持久化。部署新图前保存 graph_version 与 state schema 版本，旧 checkpoint 的节点名被删除时应显式迁移或继续旧 worker。回放可能重做外部调用，诊断优先使用只读或记录结果替身。
测试覆盖拒绝、批准、错误 thread、重复恢复、节点错误和取消；trace 关联 thread、checkpoint、node、tool call 与任务状态。图流式事件只作进度，最终状态由 get_state/业务账本确认，断连后用同任务身份重新订阅。数据库、队列、worker 和并发租约属于部署层，图本身不替你解决所有冲突。
把检查点交给下一进程
内存检查点跟随进程消失。持久恢复要同时固定数据库路径、thread_id、图节点名称与状态schema；新进程重新构造图，再用相同线程身份发送Command。SQLite适合这个单机学习实验，不等于多主机生产数据库。
在本课程原环境追加安装 python -m pip install langgraph-checkpoint-sqlite==3.1.1（本轮实际对应langgraph-checkpoint 4.2.0）。保存下面完整文件为 persist.py 并运行 python persist.py。父进程启动两个短命子进程：第一个停在interrupt后退出，第二个重新打开数据库完成审批。每个子进程重新import并构图，不共享内存。
import sys, json, subprocess
from pathlib import Path
from tempfile import TemporaryDirectory
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.types import interrupt, Command
from langgraph.checkpoint.sqlite import SqliteSaver
class State(TypedDict):
text: str
approved: bool
def approve(s):
# 恢复会从本节点头重跑：这里不能先发送邮件/扣款。
decision=interrupt({'preview':s['text']})
return {'approved':decision is True}
def run(mode, db):
g=StateGraph(State);g.add_node('approve',approve)
g.add_edge(START,'approve');g.add_edge('approve',END)
with SqliteSaver.from_conn_string(db) as saver:
graph=g.compile(checkpointer=saver)
cfg={'configurable':{'thread_id':'review-1'}}
value={'text':'draft'} if mode=='start' else Command(resume=True)
result=graph.invoke(value,cfg,durability='sync')
if mode=='start':
assert result['__interrupt__']
else:
assert result['approved'] is True
print(mode)
if __name__=='__main__':
if len(sys.argv)==3:
run(sys.argv[1],sys.argv[2])
else:
with TemporaryDirectory() as folder:
db=str(Path(folder)/'state.sqlite')
for mode in ['start','resume']:
subprocess.run([sys.executable,__file__,mode,db],check=True)
print('two distinct processes persisted and resumed')预期依次输出start、resume和两进程成功说明。第一次invoke把待审批状态与暂停位置同步写盘；第二次不是从入口重复业务流程，而是从检查点恢复，但中断节点本身会从函数开头重入。把发送邮件放在interrupt之前仍可能重复，所以提交副作用应在批准后的独立节点，并使用业务幂等键。
部署新代码不能随意重命名被暂停节点或删除状态字段；先枚举旧线程版本，保留对应旧图工厂，或显式迁移状态并重验恢复点。状态存储不自动迁移外部副作用。
练习
持久化变体：将第二进程的批准值改为False，必须恢复为approved=False；把thread_id改成另一个值，不得错误批准原线程。解释两个结果为何不同。
用新的thread_id走拒绝路径，结果approved=False，visits总数增加2。证明两个thread不会混用审批结果。