LangGraph06：检查点与持久化-平芜编程栈

LangGraph 检查点与持久化：MemorySaver / SqliteSaver / PostgresSaver 怎么选

这是本系列的收尾篇。状态管理解决了"怎么改才安全"，检查点解决"改好的状态怎么存才不丢"——本篇拆解 Checkpoint 的数据结构、三种存储后端的实现差异与选型决策，最后用"状态管理 + 检查点"的协作闭环，把 LangGraph 的执行机制串起来收个尾。

一个二小时任务的悲剧

你的 Agent 正在执行一个长任务：查了四十分钟资料、正在生成最终报告。突然节点 OOM，服务重启。任务从头再来。

问题不在模型，在执行过程没有被存档。LangGraph 的答案是检查点（Checkpointer）：在流程的关键节点把状态保存下来，重启后从最近的存档接着跑。

一、一次"存档"里装了什么

检查点的数据结构可以简化为一个数据类：

@dataclassclassCheckpoint:checkpoint_id:str# 本次存档的唯一编号thread_id:str# 属于哪条会话/任务线state:Dict[str,Any]# 序列化后的完整状态（核心）created_at:datetime# 存档时间metadata:Dict[str,Any]# 附加信息：模型、耗时、触发者等parent_checkpoint_id:Optional[str]# 上一次存档的编号（链式关系）checkpoint_version:int# 版本号，防并发覆盖

每个字段都有明确的工程职责：

字段	职责
`checkpoint_id`	唯一标识一次存档，防止重复与混淆
`thread_id`	归属哪个会话，支撑上一篇讲的记忆隔离
`state`	核心数据——序列化后的完整状态，恢复执行的依据
`created_at`	时间戳，支持过期清理与历史查询
`metadata`	监控调试用的附加信息
`parent_checkpoint_id`	把存档连成链，支持回滚到任意历史版本
`checkpoint_version`	递增版本号，并发写入时不互相踩踏

parent_checkpoint_id让检查点不只是"最新快照"，而是一条可回溯的版本链——类似代码仓库的提交历史，你可以回到任意一步查看当时的完整状态。

二、三个后端逐个拆

框架统一了BaseCheckpointSaver接口，业务代码不变，只换存储实现。

1. MemorySaver：进程内存

classMemorySaver(BaseCheckpointSaver):def__init__(self):self.storage:Dict[str,List[Checkpoint]]={}

实现就是一个字典，按thread_id存检查点列表。读写速度最快，但进程一结束全部消失。定位：开发、测试、跑 demo。

2. SqliteSaver：本地文件

classSqliteSaver(BaseCheckpointSaver):def__init__(self,conn:sqlite3.Connection):self.conn=conn

把检查点存进 SQLite 文件，无需部署数据库服务器，重启不丢数据。定位：单机小流量的轻量生产环境。

3. PostgresSaver：企业级仓库

面向分布式生产的设计，三个关键点值得细看：

连接方式灵活——优先复用外部传入的连接池，其次按参数自建连接，二者必选其一：

classPostgresSaver(BaseCheckpointSaver):def__init__(self,conn=None,conn_kwargs=None,table_name="langgraph_checkpoints"):ifconn:self.conn=conn# 复用连接池（推荐）elifconn_kwargs:self.conn=psycopg2.connect(**conn_kwargs)else:raiseValueError("必须提供 conn 或 conn_kwargs")self.table_name=table_name self._init_table()# 自动建表，杜绝"表不存在"

表结构有讲究——状态存JSONB而非纯文本，支持对状态内部字段做索引查询；parent_checkpoint_id加自引用外键保证存档链完整；thread_id、created_at建索引避免全表扫描：

CREATETABLEIFNOTEXISTSlanggraph_checkpoints(checkpoint_idVARCHAR(64)PRIMARYKEY,thread_idVARCHAR(64)NOTNULL,state JSONBNOTNULL,created_atTIMESTAMPNOTNULLDEFAULTCURRENT_TIMESTAMP,metadata JSONB,parent_checkpoint_idVARCHAR(64),checkpoint_versionINTNOTNULLDEFAULT1);CREATEINDEXIFNOTEXISTSidx_thread_idONlanggraph_checkpoints(thread_id);CREATEINDEXIFNOTEXISTSidx_created_atONlanggraph_checkpoints(created_at);

写入链路带防护——参数化查询防注入；版本号按thread_id查询最大值再 +1，防止并发写入互相覆盖；写入失败回滚事务，不产生脏数据。

定位：多实例 Agent 集群、跨机房部署的企业级场景。

三、选型决策

维度	MemorySaver	SqliteSaver	PostgresSaver
持久化	进程结束即丢	本地文件持久	数据库持久 + 容灾备份
并发能力	单进程	弱（文件锁）	强（行级锁 + 连接池）
查询能力	无	需全表扫描	JSONB + 索引，高效检索
部署成本	零	极低	需要运维数据库
适用环境	开发 / 测试	单机生产	分布式高并发生产

决策一句话：开发测试无脑MemorySaver，单机小流量上SqliteSaver，多实例分布式必选PostgresSaver。

四、StateManager 与 Checkpointer 的协作闭环

状态管理和检查点各管一段：StateManager 保证"改得安全"，Checkpointer 保证"存得不丢"。二者交互构成完整的生命周期：

三个关键环节：

先改后存：状态必须经过 StateManager 的规则（深度合并、预留字段保护）生成合法状态，检查点只负责保存，不掺和修改逻辑——单一职责；
恢复即续跑：重启后从存储后端取最新检查点，反序列化为状态对象交还 StateManager，工作流从断点继续；
存档时机：在"节点执行完成""分支判断前"等关键节点存档，而不是每行改动都存——平衡性能与可恢复性。

用一段伪代码串起"更新 → 保存 → 恢复"的完整链路：

# 状态管理：不可变更新new_state=state_manager.update(current_state,{"final_answer":"CPU 使用率 80%"})# 检查点：序列化保存checkpointer.put({"thread_id":"user_123"},new_state.model_dump())# —— 模拟重启 ——# 检查点：读取最新存档，反序列化restored=checkpointer.get({"thread_id":"user_123"})# 状态管理：继续在恢复的状态上修改state=state_manager.assign(restored,{"final_answer":"CPU 使用率 82%"})

五、收尾：执行机制全景回顾

把六篇的内容串成一条主线，LangGraph 的完整执行机制是"定义 → 编译 → 执行"三阶段：

执行层的循环每轮只做三件事：执行当前节点 → 把返回的部分更新按规则合并进全局状态 → 查边的路由规则决定下一站，直到走到 END。沿途由 StateManager 保证每次更新安全、由 Checkpointer 保证每个关键节点有存档——这就是 LangGraph 能从"测试玩具"升级为"生产级工作流框架"的全部秘密。