LangGraph 状态持久化实战：langgraph-checkpoint-sqlite 的 SqliteSaver 与 AsyncSqliteSaver 深度解析

LangGraph 的 checkpointer 机制负责把图（Graph）每一步执行后的状态写入存储，从而实现会话记忆、中断恢复与时间旅行。langgraph-checkpoint-sqlite 是官方提供的 SQLite 后端 checkpoint 保存器，同时提供同步（SqliteSaver）与异步（AsyncSqliteSaver）两种实现。本文基于仓库中 libs/checkpoint-sqlite/README.md 的完整用法展开，并结合源码梳理其表结构、序列化、过滤查询与安全配置，帮助你在本地开发、测试与轻量部署场景中正确落地 LangGraph 的状态持久化。

定位与安装

该库的定位很明确：为 LangGraph 提供基于 SQLite 的 checkpoint saver 实现，通过 aiosqlite 提供异步支持，适用于本地开发、测试或轻量部署场景（引自 README 中 “What is this?” 一节的原始描述）。

安装方式（README 推荐）：

uv add langgraph-checkpoint-sqlite

从 pyproject.toml 可以确认其运行前提与依赖：

项目	值	说明
包名	langgraph-checkpoint-sqlite	当前版本 3.1.1
Python 要求	>=3.10	低版本解释器不可用
核心依赖	langgraph-checkpoint>=4.1.0,<5.0.0	提供 BaseCheckpointSaver 与 serde 协议
异步驱动	aiosqlite>=0.20	AsyncSqliteSaver 必需
其他依赖	sqlite-vec>=0.1.6	随包声明的依赖项

需要说明的是，SQLite 方案在 AsyncSqliteSaver 的 docstring 中明确提示：由于 SQLite 写性能的限制，不推荐用于生产负载，生产环境应考虑 PostgreSQL 等更稳健的数据库（对应仓库中的 libs/checkpoint-postgres 包）。因此本包的最佳适用面是：单机脚本、Notebook 演示、单元测试与小型内部工具。

同步用法：SqliteSaver 的完整示例

README 给出了 SqliteSaver 的标准用法。写配置的 configurable 中需要 thread_id 与 checkpoint_ns，读配置只需要 thread_id：

from langgraph.checkpoint.sqlite import SqliteSaver

write_config = {"configurable": {"thread_id": "1", "checkpoint_ns": ""}}
read_config = {"configurable": {"thread_id": "1"}}

with SqliteSaver.from_conn_string(":memory:") as checkpointer:
    checkpoint = {
        "v": 4,
        "ts": "2024-07-31T20:14:19.804150+00:00",
        "id": "1ef4f797-8335-6428-8001-8a1503f9b875",
        "channel_values": {
            "my_key": "meow",
            "node": "node"
        },
        "channel_versions": {
            "__start__": 2,
            "my_key": 3,
            "start:node": 3,
            "node": 3
        },
        "versions_seen": {
            "__input__": {},
            "__start__": {
                "__start__": 1
            },
            "node": {
                "start:node": 2
            }
        },
    }

    # store checkpoint
    checkpointer.put(write_config, checkpoint, {}, {})

    # load checkpoint
    checkpointer.get(read_config)

    # list checkpoints
    list(checkpointer.list(read_config))

结合 源码，可以补充以下使用细节：

1. 两种构造方式：SqliteSaver(conn) 直接传入 sqlite3.Connection；from_conn_string(conn_string) 是上下文管理器，接受 ":memory:"（内存库，进程结束即消失）或文件路径（如 "checkpoints.sqlite"），退出 with 块时自动关闭连接。
2. 序列化器可注入：构造函数签名为 SqliteSaver(conn, *, serde=None)，默认使用 JsonPlusSerializer，即 libs/checkpoint 中定义的默认 serde 协议实现，可处理 LangChain/LangGraph 原语、datetime、枚举等类型。
3. 线程模型：from_conn_string 内部以 check_same_thread=False 建连，SqliteSaver 自身持有一把 threading.Lock，所有数据库访问都经过 cursor() 上下文管理器 加锁并在关闭时 commit。因此它是线程安全的，但 docstring 同时提醒：这是面向 demo 与小型项目的轻量实现，不能扩展到多线程高并发写入（“does not scale to multiple threads”）。

首次使用自动建表：schema 设计

SqliteSaver 首次执行游标操作时会调用 setup() 自动建表，脚本如下（源码 L139-L164）：

PRAGMA journal_mode=WAL;
CREATE TABLE IF NOT EXISTS checkpoints (
    thread_id TEXT NOT NULL,
    checkpoint_ns TEXT NOT NULL DEFAULT '',
    checkpoint_id TEXT NOT NULL,
    parent_checkpoint_id TEXT,
    type TEXT,
    checkpoint BLOB,
    metadata BLOB,
    PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id)
);
CREATE TABLE IF NOT EXISTS writes (
    thread_id TEXT NOT NULL,
    checkpoint_ns TEXT NOT NULL DEFAULT '',
    checkpoint_id TEXT NOT NULL,
    task_id TEXT NOT NULL,
    idx INTEGER NOT NULL,
    channel TEXT NOT NULL,
    type TEXT,
    value BLOB,
    PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id, task_id, idx)
);

这个 schema 值得逐列理解：

• checkpoints 表：每行是一个超步（superstep）后的完整状态快照。checkpoint 列为 serde 序列化后的二进制 BLOB，type 列记录序列化类型标签（loads_typed 时成对使用）；parent_checkpoint_id 构成状态链，是 get_tuple 返回 parent_config、支持时间旅行回退的关键。联合主键 (thread_id, checkpoint_ns, checkpoint_id) 保证线程/命名空间/快照三元组唯一。
• writes 表：存放挂起的中间写入（pending writes）。put_writes 在任务执行中先把 (channel, value) 写入此表，下一个 checkpoint 生成后才被“消费”。主键中的 task_id + idx 区分同一 checkpoint 下不同任务、同一任务内的多条写入；特殊通道（如 __interrupt__、__error__）会经过 WRITES_IDX_MAP 映射为固定 idx，见 源码。
• PRAGMA journal_mode=WAL：开启写前日志模式，让读操作不与写事务互斥，这对单进程内的并发读写是有实际收益的。
• 版本号生成：get_next_version() 生成形如 00000000000000000000000000000001.3928374628000000 的字符串（32 位递增整数 + 16 位随机小数）。SQLite 没有 Postgres 的数值版本号，采用字符串版本时以整数部分递增、小数部分随机的方式保证单调递增与唯一性——这也是为什么 ORDER BY checkpoint_id DESC 能够正确表达“从新到旧”。

get / put / list 的查询语义

• get_tuple(config)（源码 L191-L293）：若 config 带 checkpoint_id，按主键精确取一条；否则 ORDER BY checkpoint_id DESC LIMIT 1 取该 thread/namespace 的最新快照。取到快照后还会查 writes 表附带挂起写入，并组装 CheckpointTuple（含 config、反序列化后的 checkpoint、metadata、parent_config、pending writes）。
• put(config, checkpoint, metadata, new_versions)：用 INSERT OR REPLACE 写入 checkpoints 表，parent_checkpoint_id 取自 config 中的 checkpoint_id（即上一个快照），metadata 以 JSON 文本存储（ensure_ascii=False）。返回值是补齐了 checkpoint_id 的新 config。
• list(config, *, filter, before, limit)：按 checkpoint_id DESC 返回迭代器，支持 before（只返回更早的快照，游标翻页）与 limit。

异步用法：AsyncSqliteSaver

README 的 Async 示例与同步版一一对应，方法名以 a 前缀区分：

from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver

async with AsyncSqliteSaver.from_conn_string(":memory:") as checkpointer:
    checkpoint = {
        "v": 4,
        "ts": "2024-07-31T20:14:19.804150+00:00",
        "id": "1ef4f797-8335-6428-8001-8a1503f9b875",
        "channel_values": {
            "my_key": "meow",
            "node": "node"
        },
        "channel_versions": {
            "__start__": 2,
            "my_key": 3,
            "start:node": 3,
            "node": 3
        },
        "versions_seen": {
            "__input__": {},
            "__start__": {
                "__start__": 1
            },
            "node": {
                "start:node": 2
            }
        },
    }

    # store checkpoint
    await checkpointer.aput(write_config, checkpoint, {}, {})

    # load checkpoint
    await checkpointer.aget(read_config)

    # list checkpoints
    [c async for c in checkpointer.alist(read_config)]

从 aio.py 源码 看，有三个与使用强相关的设计点：

1. 必须用 async with 管理生命周期。from_conn_string 是 @asynccontextmanager（L131-L145），退出时关闭 aiosqlite 连接。docstring 特别警告：若忘记关闭连接，程序会“挂住”不退出（事件循环中仍有未释放的资源）。
2. 同步方法是从后台线程桥接进事件循环的。get_tuple / list / put 等同名同步方法内部使用 asyncio.run_coroutine_threadsafe(..., self.loop).result()（L147-L175）。其中有一道跨线程护栏：如果检测到当前调用发生在保存器所绑定的事件循环线程上，会直接抛出 asyncio.InvalidStateError，提示“主线程请使用异步接口，如 await checkpointer.aget_tuple(...) 或 await graph.ainvoke(...)”，避免同线程自死锁。换言之，异步保存器的同步方法只能从另一个线程（例如 LangGraph 的后台任务线程）调用。
3. 建表逻辑与同步版一致。await saver.setup() 执行同样的 PRAGMA journal_mode=WAL + 两张 CREATE TABLE IF NOT EXISTS 脚本（L305-L344），锁换成 asyncio.Lock。

因此推荐的异步用法是全程走 a* 方法，或者像 docstring 示例那样把整个图执行放进 async with 块内：

async with AsyncSqliteSaver.from_conn_string("checkpoints.sqlite") as saver:
    graph = builder.compile(checkpointer=saver)
    config = {"configurable": {"thread_id": "thread-1"}}
    async for event in graph.astream_events(..., config, version="v1"):
        print(event)

基于 metadata 的过滤查询：filter、before 与 SQL 注入防护

list / alist 接受 filter: dict 参数，可以按 checkpoint metadata 做等值过滤。其实现位于 utils.py，search_where(config, filter, before) 将三类条件拼成参数化 WHERE 子句：

• config 条件：thread_id = ?，可选 checkpoint_ns = ?、checkpoint_id = ?；
• filter 条件：对 metadata JSON 使用 SQLite 的 json_extract，形如 json_extract(CAST(metadata AS TEXT), '$.<key>') = ?；
• before 条件：checkpoint_id < ?，实现游标翻页。

过滤值的类型处理（_where_value）覆盖了几种边界：None 生成 IS ?；布尔值转为 1/0（SQLite 无独立 bool 类型）；dict/list 用无空格的 json.dumps 序列化后与 json_extract 的返回精确比较。

安全性方面，_validate_filter_key（L11-L28）用正则 ^[a-zA-Z0-9_.-]+$ 校验过滤键，拒绝任何可能构造 SQL 注入的字符。这个测试在 tests/test_sqlite.py 中也有对应覆盖（TestSqliteSaver 直接 import 了 _metadata_predicate 与 search_where 做单元验证）。

一个典型场景：给不同 thread 打上来源标签后按标签列出快照——

# 写入时 metadata 可包含任意可 JSON 序列化的字段
checkpointer.put(
    {"configurable": {"thread_id": "42", "checkpoint_ns": ""}},
    checkpoint,
    {"source": "loop", "user_tag": "vip"},  # metadata
    {},
)

# 之后按 metadata 过滤
snapshots = list(checkpointer.list(
    {"configurable": {"thread_id": "42"}},
    filter={"user_tag": "vip"},
    limit=10,
))

进阶能力：delete_thread 与 delta channel 历史

除核心三方法外，README 未展开但源码已实现的能力值得了解：

• delete_thread(thread_id)（同步 / 异步）：先删 checkpoints 后删 writes 中该 thread 的所有行，用于清理整条会话历史。
• get_delta_channel_history(config=..., channels=...)：SqliteSaver 对该方法做了 SQLite 专属的两段式快速路径重写（L503-L583），共享实现放在 _delta.py。设计注释说明了与 Postgres 版的结构性差异：SQLite 没有 JSONB，必须反序列化完整 checkpoint blob 才能检查 channel_values；因此阶段 1 按 checkpoint_id DESC 流式拉取祖先链（DELTA_STAGE1_SQL），逐行反序列化、用完即弃，把峰值内存控制在“同一时刻约一个反序列化 checkpoint”；阶段 2 按通道生成 UNION ALL 查询（SQLite 没有 Postgres 的数组绑定语法 = ANY(%s)，改为每个分支内联 IN (?, ?, ...) 占位符）。该方法服务 DeltaChannel（增量通道）的语义：沿祖先链找到每个通道值的种子（seed）并收集沿途 writes。对应的端到端冒烟测试见 tests/test_get_delta_channel_history.py，覆盖了空 channels、多快照线程下“旧到新”排序、找不到种子时省略 seed、以及 _DeltaSnapshot 作为种子返回等场景。

安全：限制 checkpoint 反序列化的类型面

README 的 Security 一节（对应 libs/checkpoint/README.md 的 serde 安全说明）给出了明确的部署建议：

Set LANGGRAPH_STRICT_MSGPACK=true or pass an explicit allowed_msgpack_modules list when creating your checkpointer. This restricts checkpoint deserialization to known-safe types, preventing code execution if the database is compromised.

即：JsonPlusSerializer 默认允许还原 checkpoint 数据中出现的任意 Python 类型；如果数据库文件被恶意篡改，反序列化时可能执行任意代码。新应用应当设置环境变量 LANGGRAPH_STRICT_MSGPACK=true，或在构造 saver 时通过 serde 参数传入显式限制过的序列化器（SqliteSaver(conn, serde=...) 支持自定义 SerializerProtocol 实现）。对 SQLite 这类常落在本地磁盘、可能被随意拷贝/共享的文件格式，这条建议尤其值得照做。

小结：选型与边界

维度	结论（依据仓库证据）
适用场景	本地开发、Notebook、单元测试、轻量单机部署（README 原文定位）
生产负载	不建议；AsyncSqliteSaver docstring 明确推荐生产环境使用 PostgreSQL 等数据库
并发	同步版持 threading.Lock 线程安全，但“does not scale to multiple threads”；异步版持 asyncio.Lock，且同步桥接方法不能从事件循环线程调用
Python	3.10+，依赖 langgraph-checkpoint>=4.1.0,<5.0.0 与 aiosqlite>=0.20
关键操作	put/get/list（及 a* 变体）、put_writes、delete_thread、get_delta_channel_history、filter/before/limit 查询
安全	建议 LANGGRAPH_STRICT_MSGPACK=true 或显式 allowed_msgpack_modules

如果你正在为 LangGraph 应用选择 checkpointer：先用 SqliteSaver.from_conn_string(":memory:") 跑通状态持久化逻辑，再切到文件路径验证跨进程恢复，最后在生产环境评估 libs/checkpoint-postgres 的 PostgresSaver——两者的 API 面（BaseCheckpointSaver 协议）是一致的，迁移成本主要在于连接管理而非代码改写。