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.connconn把检查点存进 SQLite 文件无需部署数据库服务器重启不丢数据。定位单机小流量的轻量生产环境。3. PostgresSaver企业级仓库面向分布式生产的设计三个关键点值得细看连接方式灵活——优先复用外部传入的连接池其次按参数自建连接二者必选其一classPostgresSaver(BaseCheckpointSaver):def__init__(self,connNone,conn_kwargsNone,table_namelanggraph_checkpoints):ifconn:self.connconn# 复用连接池推荐elifconn_kwargs:self.connpsycopg2.connect(**conn_kwargs)else:raiseValueError(必须提供 conn 或 conn_kwargs)self.table_nametable_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 集群、跨机房部署的企业级场景。三、选型决策维度MemorySaverSqliteSaverPostgresSaver持久化进程结束即丢本地文件持久数据库持久 容灾备份并发能力单进程弱文件锁强行级锁 连接池查询能力无需全表扫描JSONB 索引高效检索部署成本零极低需要运维数据库适用环境开发 / 测试单机生产分布式高并发生产决策一句话开发测试无脑MemorySaver单机小流量上SqliteSaver多实例分布式必选PostgresSaver。四、StateManager 与 Checkpointer 的协作闭环状态管理和检查点各管一段StateManager 保证改得安全Checkpointer 保证存得不丢。二者交互构成完整的生命周期反序列化恢复状态节点执行产生状态更新StateManagerupdate / assign / delete生成合规的新状态Checkpointer序列化并持久化重启 / 扩容 / 断点续跑读取最新检查点三个关键环节先改后存状态必须经过 StateManager 的规则深度合并、预留字段保护生成合法状态检查点只负责保存不掺和修改逻辑——单一职责恢复即续跑重启后从存储后端取最新检查点反序列化为状态对象交还 StateManager工作流从断点继续存档时机在节点执行完成分支判断前等关键节点存档而不是每行改动都存——平衡性能与可恢复性。用一段伪代码串起更新 → 保存 → 恢复的完整链路# 状态管理不可变更新new_statestate_manager.update(current_state,{final_answer:CPU 使用率 80%})# 检查点序列化保存checkpointer.put({thread_id:user_123},new_state.model_dump())# —— 模拟重启 ——# 检查点读取最新存档反序列化restoredcheckpointer.get({thread_id:user_123})# 状态管理继续在恢复的状态上修改statestate_manager.assign(restored,{final_answer:CPU 使用率 82%})五、收尾执行机制全景回顾把六篇的内容串成一条主线LangGraph 的完整执行机制是定义 → 编译 → 执行三阶段定义层StateGraph add_node add_edge画好蓝图编译层compile 校验 封装执行器生成 CompiledGraph执行层状态驱动的循环:运行节点 → 合并状态 → 查边路由直到 END执行层的循环每轮只做三件事执行当前节点 → 把返回的部分更新按规则合并进全局状态 → 查边的路由规则决定下一站直到走到 END。沿途由 StateManager 保证每次更新安全、由 Checkpointer 保证每个关键节点有存档——这就是 LangGraph 能从测试玩具升级为生产级工作流框架的全部秘密。小结本系列六篇的脉络是什么图三要素节点/边/状态 五大能力把 AI 应用变成持续运行的系统上手add_messages累加语义 StateGraph四步搭图工具调用定义 → 挂载 → 决策 → 路由 → 执行回流的闭环记忆checkpointer 存档 thread_id分区隔离状态管理TypedDict/Pydantic 双模式 update/assign/delete的不可变设计持久化三种存储后端选型 状态管理与检查点的协作闭环。记住最后一句话节点干活、边定路线、状态存记忆、检查点保平安——掌握了这四件事LangGraph 的复杂应用就都只是在这张图上继续加积木而已。
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.connconn把检查点存进 SQLite 文件无需部署数据库服务器重启不丢数据。定位单机小流量的轻量生产环境。3. PostgresSaver企业级仓库面向分布式生产的设计三个关键点值得细看连接方式灵活——优先复用外部传入的连接池其次按参数自建连接二者必选其一classPostgresSaver(BaseCheckpointSaver):def__init__(self,connNone,conn_kwargsNone,table_namelanggraph_checkpoints):ifconn:self.connconn# 复用连接池推荐elifconn_kwargs:self.connpsycopg2.connect(**conn_kwargs)else:raiseValueError(必须提供 conn 或 conn_kwargs)self.table_nametable_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 集群、跨机房部署的企业级场景。三、选型决策维度MemorySaverSqliteSaverPostgresSaver持久化进程结束即丢本地文件持久数据库持久 容灾备份并发能力单进程弱文件锁强行级锁 连接池查询能力无需全表扫描JSONB 索引高效检索部署成本零极低需要运维数据库适用环境开发 / 测试单机生产分布式高并发生产决策一句话开发测试无脑MemorySaver单机小流量上SqliteSaver多实例分布式必选PostgresSaver。四、StateManager 与 Checkpointer 的协作闭环状态管理和检查点各管一段StateManager 保证改得安全Checkpointer 保证存得不丢。二者交互构成完整的生命周期反序列化恢复状态节点执行产生状态更新StateManagerupdate / assign / delete生成合规的新状态Checkpointer序列化并持久化重启 / 扩容 / 断点续跑读取最新检查点三个关键环节先改后存状态必须经过 StateManager 的规则深度合并、预留字段保护生成合法状态检查点只负责保存不掺和修改逻辑——单一职责恢复即续跑重启后从存储后端取最新检查点反序列化为状态对象交还 StateManager工作流从断点继续存档时机在节点执行完成分支判断前等关键节点存档而不是每行改动都存——平衡性能与可恢复性。用一段伪代码串起更新 → 保存 → 恢复的完整链路# 状态管理不可变更新new_statestate_manager.update(current_state,{final_answer:CPU 使用率 80%})# 检查点序列化保存checkpointer.put({thread_id:user_123},new_state.model_dump())# —— 模拟重启 ——# 检查点读取最新存档反序列化restoredcheckpointer.get({thread_id:user_123})# 状态管理继续在恢复的状态上修改statestate_manager.assign(restored,{final_answer:CPU 使用率 82%})五、收尾执行机制全景回顾把六篇的内容串成一条主线LangGraph 的完整执行机制是定义 → 编译 → 执行三阶段定义层StateGraph add_node add_edge画好蓝图编译层compile 校验 封装执行器生成 CompiledGraph执行层状态驱动的循环:运行节点 → 合并状态 → 查边路由直到 END执行层的循环每轮只做三件事执行当前节点 → 把返回的部分更新按规则合并进全局状态 → 查边的路由规则决定下一站直到走到 END。沿途由 StateManager 保证每次更新安全、由 Checkpointer 保证每个关键节点有存档——这就是 LangGraph 能从测试玩具升级为生产级工作流框架的全部秘密。小结本系列六篇的脉络是什么图三要素节点/边/状态 五大能力把 AI 应用变成持续运行的系统上手add_messages累加语义 StateGraph四步搭图工具调用定义 → 挂载 → 决策 → 路由 → 执行回流的闭环记忆checkpointer 存档 thread_id分区隔离状态管理TypedDict/Pydantic 双模式 update/assign/delete的不可变设计持久化三种存储后端选型 状态管理与检查点的协作闭环。记住最后一句话节点干活、边定路线、状态存记忆、检查点保平安——掌握了这四件事LangGraph 的复杂应用就都只是在这张图上继续加积木而已。
