LangGraph 实战第 3 课：SqliteSaver 落盘，让 Agent 能倒带重来

生命不息，折腾不止。上一课给 Agent 装了「人工审批」闸门，但那个 checkpointer 是放在内存里的，服务一重启就全没了——这一课我们换成落盘的 SqliteSaver，再学会 get_state / update_state 两招「时间旅行」，跑错的步骤倒回去改完再重来。

一、内存 checkpointer 为什么撑不住

上一课编译图的时候，我们挂的是 InMemorySaver。它本质上就是进程内存里的一个字典：{thread_id: [checkpoint, checkpoint, ...]}。拿来演示、调 bug 很爽，但有两个致命伤：

一重启就蒸发：Agent 正暂停在「人工审批」那里等人拍板，服务器来一次重启，状态全没了，thread_id 那个「抽屉」被清空，人也恢复不了。
只能单进程用：内存里的东西没法共享给别的进程、别的机器，做不了多副本、也查不了历史。

所以真要上线，就得把 checkpoint 从内存挪到磁盘/数据库。LangGraph 官方把 checkpointer 抽象成了一个统一接口，换实现就是换一个类的事。最轻量的落盘方案就是 SQLite——单文件、零部署、开箱即用，正好接上一课的场景。

（先说清适用边界：SQLite 适合单机、并发不大的小项目；真要上生产扛并发，后面换成 PostgresSaver 也就改一行，本节末尾会点到。）

二、换上 SqliteSaver：暂停的执行落盘

先装包。注意 SqliteSaver 单独拆在 langgraph-checkpoint-sqlite 包里，不是 langgraph 主包自带：

1	pip install langgraph-checkpoint-sqlite

然后改两处：把 InMemorySaver() 换成 SqliteSaver，再把上一课搭好的图重新编译一次。图的节点、边、条件路由全都不动，只换 checkpointer：

import sqlite3
from langgraph.checkpoint.sqlite import SqliteSaver

# 建一个 SQLite 连接，check_same_thread=False 让多线程/Web 服务场景也能用
conn = sqlite3.connect("checkpoints.sqlite", check_same_thread=False)
checkpointer = SqliteSaver(conn)

# builder 就是上一课那个 StateGraph，节点/边/路由照旧
graph = builder.compile(checkpointer=checkpointer)

这么一改，每个 checkpoint 就被 SqliteSaver 序列化成字节存进 checkpoints.sqlite 文件里。服务重启、进程挂掉，只要这个文件还在，thread_id 对应的执行就还在，Command(resume=...) 照样能恢复。

模型这部分完全沿用上一课的配置（OpenAI 兼容协议接 GPT/Claude 都行；想省事可以指到 ai.aklibk.com 这种中转站，国内直连、免绑卡、人民币按量付费，不折腾海外卡）。这一课重点是持久化，模型不动。

想先冒烟测试一下 SqliteSaver 本身能不能跑，官方还有个「上下文管理器 + 内存库」的写法：

from langgraph.checkpoint.sqlite import SqliteSaver

with SqliteSaver.from_conn_string(":memory:") as checkpointer:
    graph = builder.compile(checkpointer=checkpointer)
    # 在这个 with 块里 invoke 图即可

但 :memory: 还是内存库，只适合测连通性；要真落盘，就用上面 sqlite3.connect("checkpoints.sqlite") 的写法。

三、get_state：随时看抽屉里的现状

持久化之后，我们多了一个「偷看现场」的能力。get_state(config) 会从 checkpointer 里把某个 session（thread_id）最新的那个 checkpoint 捞出来，返回一个快照（StateSnapshot）。最常用的几个字段：

snapshot.values —— 当前这张共享表的最终值（草稿、决定、状态分别是什么）
snapshot.next —— 接下来要跑哪些节点（空元组 = 已经跑完 / 卡住等待）
snapshot.config —— 恢复这次执行要用的 config，里面有 checkpoint_id

还是拿上一课的审批 Agent 举例，让它完整跑一遍「写草稿 → 审批 → 发布」，然后瞄一眼现场：

from langchain_core.messages import HumanMessage
from langgraph.types import Command

config = {"configurable": {"thread_id": "doc-20261016"}}
graph.invoke({"messages": [HumanMessage(content="写条开源发布推文")]}, config)
graph.invoke(Command(resume="approve"), config)   # 批准，流程跑完

snapshot = graph.get_state(config)
print(snapshot.values)          # draft / decision / status 各自的最终值
print(snapshot.next)            # -> () 空元组，说明已经跑完了
print(snapshot.config["configurable"]["checkpoint_id"])  # 当前所在检查点的 id

get_state 最经典的用途：当 Agent 卡在 interrupt 等人拍板时，服务可以拿它把「草稿是啥、卡在哪个节点」展示给人看。

四、get_state_history：像 git log 一样翻历史

比 get_state 更狠的是 get_state_history(config)——它把这个 session 历史上所有的 checkpoint 按「新 → 旧」倒序列出来。理解成 Git 的 git log 就对了：每跑完一个节点，LangGraph 就自动「提交」一份快照。

history = list(graph.get_state_history(config))

for s in history:
    print(s.config["configurable"]["checkpoint_id"], "->", s.next)

跑一次审批流程，历史从新到旧大致长这样（简化示意，实际每个节点边界都会存一份）：

(最新)  ... -> ()                 # publish 节点跑完，流程到终点
        ... -> ('publish',)       # 审批节点被 resume 后，decision=approve
        ... -> ...                # interrupt 卡住的那个检查点（正在等人）
        ... -> ('human_review',)  # write_draft 刚跑完，下一步进审批
(最旧)  ... -> ('write_draft',)   # 起点，还没开始写草稿

next 这个字段是关键：它告诉你「这个检查点之后下一步要跑谁」。想「回到过去」，就是先在这些历史快照里捞出你想要的、记下它的 config，再从那里接着跑——这就是时间旅行的起点。

五、时间旅行第一式：update_state 分叉，改完过去再重来

假设上一流程已经把草稿「发布」了，但老板看完说：草稿里有错别字，得改。正常思路是删了重发，但 Agent 的「过去」是可以改的。分三步。

第一步，找到想改的那个检查点。 我们要改草稿，草稿是 write_draft 节点产出的，所以找「write_draft 刚跑完、下一步要进 human_review」的那个快照：

1 2	history = list(graph.get_state_history(config)) before_review = next(s for s in history if s.next == ("human_review",))

第二步，用 update_state 在这个检查点上改值。 关键参数是 as_node="write_draft"：它告诉 LangGraph「你就当是 write_draft 节点重新产出了这份草稿」，这样新的执行会从 write_draft 的后继节点（也就是 human_review）接着往下走：

fork_config = graph.update_state(
    before_review.config,
    values={"draft": "改好的新草稿（去掉了错别字）"},
    as_node="write_draft",
)

第三步，从这个分叉点继续往下跑。 传 None 表示「不新增输入，就从该检查点接着跑」：

1 2	result = graph.invoke(None, fork_config) # human_review 会再次 interrupt 等你拍板，用上一课的 Command(resume=...) 恢复即可

这里有个一定要弄懂的点：update_state 不会改写历史，也不会回滚原线程。它是在你选的检查点旁边「分叉」出一条新时间线——原来的 before_review.config 和它之后那一串 checkpoint 原封不动，而 fork_config 是一个全新的分支，checkpoint_id 也不同。所以「改过去」其实是「岔出去一条新路」，旧路还在，随时能切回去看。

如果不显式指定 as_node，直接 graph.update_state(before_review.config, values={...}) 也能跑，值会按通道的 reducer 规则合并；但显式写 as_node 能更清楚地控制「从哪个节点的后继继续」，不容易踩到 resume 起点不对的坑。

六、时间旅行第二式：回放 replay

update_state 是「改完再跑」，还有一种更轻的玩法叫回放（replay）：不改任何值，直接从某个历史检查点再跑一遍后面的节点。

1 2	# 从 before_review 那个检查点原样再跑一遍（不 update，直接用旧 config） replay_result = graph.invoke(None, before_review.config)

和「分叉」最大的区别是——回放会真的重新执行节点，不是从缓存里读结果：

human_review 会重新 interrupt，write_draft 里如果有 LLM 调用，那条调用会重新打一次，可能返回不一样的话术。
所以回放适合「调试 / 复现 / 看 Agent 到底是怎么走到这步的」；想「改个变量重试」，用上一节的 update_state 分叉才对。

（官方文档明确写过：回放会重新执行节点，interrupt 和 API 调用都会再触发；从已经到终点的最后检查点回放是空操作 no-op。）

七、三个坑 + 生产建议 + 下一步

坑一：依赖包装错。 是 pip install langgraph-checkpoint-sqlite，不是 langgraph-checkpoint；装完 from langgraph.checkpoint.sqlite import SqliteSaver 才导入得进去。

坑二：check_same_thread。 在 Web 服务这种多线程场景下，直接用 sqlite3.connect("checkpoints.sqlite") 会报错，记得加 check_same_thread=False。

坑三：checkpoint 文件也要备份。 落盘不等于万无一失，那个 .sqlite 文件就是 Agent 的「记忆」，要跟着你的 321 备份一起备。官方还提醒：生产环境设 LANGGRAPH_STRICT_MSGPACK=true 能限制反序列化的类型，防恶意构造文件执行代码。

生产建议：单机小项目 SqliteSaver 够用；一旦要多大副本、高并发，就把 SqliteSaver 换成 PostgresSaver（pip install langgraph-checkpoint-postgres，from langgraph.checkpoint.postgres import PostgresSaver），图和代码都不用动，只换 checkpointer 那一行。

到这里，内存 checkpointer 的两大痛点——不能持久化、不能复查历史——都被解决了。Agent 从此有了「记忆」和「后悔药」两样东西。