【LangGraph 状态持久化（Checkpoint）详解】学习笔记

目录什么是状态持久化持久化方案对比内存持久化MemoryPersistenceSQLite 持久化SqlitePersistenceAgent 多轮对话持久化AgentPersistenceget_state 与 get_state_history 详解总结对比1. 什么是状态持久化LangGraph 默认情况下每次invoke()执行完毕后State 就销毁了。但很多场景需要跨调用保留状态场景需求多轮对话AI 记住上一轮聊了什么工作流断点续跑某步骤失败后从失败点继续状态审计查看每一步的中间状态回滚/分支回到历史状态重新执行LangGraph 通过Checkpointer检查点保存器实现状态持久化。核心 API# 编译时指定 checkpointer app graph.compile(checkpointerInMemorySaver()) ​ # 配置 thread_id会话 ID config {configurable: {thread_id: user-001}} ​ # 同 thread_id 多次 invoke状态自动恢复 result1 app.invoke({messages: []}, config) # 第1次 result2 app.invoke({messages: []}, config) # 第2次从第1次结果继续Checkpoint 的工作原理第1次 invoke() │ ├── 创建 checkpoint_1开始前状态 ├── 执行节点创建 checkpoint_2中间状态 ├── 执行完成创建 checkpoint_3最终状态 │ ▼ 持久化存储内存 / SQLite / Postgres ​ 第2次 invoke()同 thread_id │ ├── 从存储加载 checkpoint_3 作为初始状态 ├── 创建 checkpoint_4开始前状态 └── ...2. 持久化方案对比方案后端数据生命周期安装要求适用场景InMemorySaver内存程序关闭即丢失内置无需安装开发调试、单次 sessionSqliteSaverSQLite 文件持久化到文件pip install langgraph-checkpoint-sqlite本地实验、单机部署PostgresSaverPostgreSQL持久化到数据库pip install langgraph-checkpoint-postgres生产环境、多实例共享3. 内存持久化MemoryPersistence3.1 文件MemoryPersistence.py三步工作流step_one → step_two → step_three配合InMemorySaver演示状态的保存和读取。3.2 核心代码from langgraph.checkpoint.memory import InMemorySaver ​ # 编译图时传入 checkpointer app graph.compile(checkpointerInMemorySaver()) ​ # 每个 thread_id 独立存储状态 config {configurable: {thread_id: user_13811112222}} ​ # 第1次执行 result app.invoke({messages: [开始执行], step_count: 0}, config)3.3 测试输出首次执行结果: {messages: [开始执行, 执行了步骤 1, 执行了步骤 2, 执行了步骤 3], step_count: 3} ​ 检查存储的状态: {messages: [开始执行, 执行了步骤 1, 执行了步骤 2, 执行了步骤 3], step_count: 3} ​ 执行历史从最新到最旧共 6 个 checkpoint: checkpoint_1: {messages: [开始执行, ...步骤 1, 步骤 2, 步骤 3]} ← 最终 checkpoint_2: {messages: [开始执行, ...步骤 1, 步骤 2]} ← step_two 后 checkpoint_3: {messages: [开始执行, ...步骤 1]} ← step_one 后 checkpoint_4: {messages: [开始执行]} ← 初始 checkpoint_5: {messages: []} ← 系统内部3.4 关键行为持久化的核心是 thread_id同一个thread_id多次invoke()→ 状态自动叠加相当于继续对话不同thread_id→ 完全隔离恢复执行# 如果图已完成invoke(None) 直接返回已有结果不重新执行 result2 app.invoke(None, config)4. SQLite 持久化SqlitePersistence4.1 文件SqlitePersistence.py与内存持久化功能相同但数据存储在 SQLite 文件中程序重启后数据不丢失。4.2 安装依赖pip install langgraph-checkpoint-sqlite4.3 核心代码import sqlite3 from langgraph.checkpoint.sqlite import SqliteSaver ​ # 1. 建立 SQLite 连接 conn sqlite3.connect( databaseD:/hermes-project/project/sqlite_data.db, check_same_threadFalse ) ​ # 2. 创建 SqliteSaver sqliteDB SqliteSaver(connconn) ​ # 3. 编译图时传入 graph builder.compile(checkpointersqliteDB) ​ # 4. 执行同 thread_id 自动持久化 config {configurable: {thread_id: user-001}} result graph.invoke({messages: []}, config)4.4 数据库结构执行后 SQLite 文件中有两个核心表checkpoints 表每个状态变更的完整快照thread_idcheckpoint_idparent_checkpoint_idcheckpoint (BLOB)user-001uuid-1None初始状态空 messagesuser-001uuid-2uuid-1开始执行user-001uuid-3uuid-2node_1 执行后[abc,def]writes 表每个节点的写入明细thread_idcheckpoint_idchannelvalueuser-001uuid-1messages[]初始user-001uuid-3messages[abc,def]4.5 连续执行的 Checkpoint 链第1次执行: invoke({messages:[]}) checkpoint_1 (开始, messages[]) ↓ checkpoint_2 (node_1运行中) ↓ checkpoint_3 (完成, messages[abc,def]) ​ 第2次执行: invoke({messages:[]}) ← 从 checkpoint_3 继续 checkpoint_4 (开始, 恢复 [abc,def]) ↓ checkpoint_5 (node_1运行中) ↓ checkpoint_6 (完成, [abc,def,abc,def])每次invoke()生成3 个 checkpoint开始→中间→完成通过parent_checkpoint_id串联。4.6 查看 SQLite 数据库import sqlite3 conn sqlite3.connect(D:/hermes-project/project/sqlite_data.db) # 列出所有表 cur conn.execute(SELECT name FROM sqlite_master WHERE typetable) for t in cur.fetchall(): print(t[0]) # 查看 checkpoints rows conn.execute( SELECT thread_id, checkpoint_id, parent_checkpoint_id FROM checkpoints ORDER BY checkpoint_id ).fetchall()5. Agent 多轮对话持久化AgentPersistence5.1 文件AgentPersistence.py将 Checkpoint 与 LangChain 的create_agent()高级 API 结合实现带记忆的多轮对话。5.2 核心代码from langchain.agents import create_agent from langgraph.checkpoint.memory import InMemorySaver llm init_chat_model(modelqwen-plus, ...) checkpointer InMemorySaver() agent create_agent(modelllm, checkpointercheckpointer) config {configurable: {thread_id: user-001}} # 第一轮自我介绍 msg1 agent.invoke( {messages: [(user, 你好我叫张三喜欢足球)]}, config ) # 第二轮AI 记得上轮信息 msg2 agent.invoke( {messages: [(user, 我叫什么我喜欢做什么)]}, config ) # AI 回复你叫张三喜欢足球 ⚽5.3 测试输出AI: 你好张三很高兴认识你同为球迷——足球的魅力在于团队精神与激情活动 AI: 你叫张三喜欢足球关键第二轮 AI 准确回答了张三和足球——这是 InMemorySaver 在 agent 层面自动持久化了对话历史。5.4 AgentPersistence 的技术要点要点说明create_agent中的 checkpointerLangChain 1.2 的 Agent 高级 API 原生支持 checkpointer消息自动累加messages字段使用add_messagesreducer新消息追加到历史thread_id 隔离不同 thread_id 的对话历史互不干扰可替换后端InMemorySaver → SqliteSaver/PostgresSaver 无需改业务代码6. get_state 与 get_state_history 详解6.1 get_state获取当前状态# 获取指定 thread 的当前状态 saved_state app.get_state(config) print(saved_state.values) # 状态字典 print(saved_state.next) # 下一步要执行的节点空已完成输出{messages: [开始执行, 步骤1, 步骤2, 步骤3], step_count: 3} () ← 空元组表示图已执行完毕6.2 get_state_history获取完整执行历史# 获取指定 thread 的所有历史 checkpoint history app.get_state_history(config) for checkpoint in history: print(checkpoint.values) print(checkpoint.next) print(checkpoint.config[configurable][checkpoint_id]) print(---)输出按从最新到最旧排列# 最新最终状态 当前状态: {messages: [开始执行, 步骤1, 步骤2, 步骤3], step_count: 3} --- 当前状态: {messages: [开始执行, 步骤1, 步骤2], step_count: 2} --- 当前状态: {messages: [开始执行, 步骤1], step_count: 1} --- 当前状态: {messages: [开始执行], step_count: 0} --- 当前状态: {messages: [], step_count: 0} ← 初始状态6.3 checkpoint 的返回结构class StateSnapshot: values: dict # 该时刻的完整 State next: tuple # 下一步要执行的节点空已完成 config: dict # 包含 checkpoint_id 等配置 metadata: dict # source, step 等信息 created_at: datetime # 创建时间 parent_config: dict # 上一个 checkpoint 的配置 tasks: tuple # 待执行的任务7. 总结对比7.1 三个文件的定位文件后端用途核心 APIMemoryPersistence.pyInMemorySaver基础演示状态保存、历史查看compile(checkpointer),get_state(),get_state_history()SqlitePersistence.pySqliteSaver文件级持久化程序重启不丢失sqlite3.connect(),SqliteSaver(conn)AgentPersistence.pyInMemorySaverAgent 多轮对话记忆create_agent(checkpointer), 多次invoke()7.2 核心 API 速查操作API内存持久化compile(checkpointerInMemorySaver())SQLite持久化compile(checkpointerSqliteSaver(conn))配置线程config {configurable: {thread_id: xxx}}获取当前状态app.get_state(config)获取历史状态app.get_state_history(config)更新状态app.update_state(config, values{...})Agent 持久化create_agent(modelllm, checkpointer...)7.3 学习路线1. MemoryPersistence.py → 理解 checkpoint 是什么、get_state 怎么用 │ ▼ 2. SqlitePersistence.py → 数据持久化到文件查看数据库结构 │ ▼ 3. AgentPersistence.py → 实战多轮对话 AI 助手