# sqlite/__init__.py:45
class SqliteSaver(BaseCheckpointSaver[str]):
    conn: sqlite3.Connection      # 数据库连接
    is_setup: bool                # 是否已建过表（懒建表用）

    def __init__(self, conn, *, serde=None) -> None:
        super().__init__(serde=serde)
        self.jsonplus_serde = JsonPlusSerializer()
        self.conn = conn
        self.is_setup = False
        self.lock = threading.Lock()   # ← 线程锁，保证并发安全

第一次操作前要先建表。setup 用一段 SQL 脚本一次建好（checkpoint/sqlite/__init__.py:129-166）：

-- sqlite/__init__.py:139  (self.conn.executescript 里)
PRAGMA journal_mode=WAL;                      -- ① 开启 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,                -- 父档 id，串起家谱链
    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)
);

所有 SQL 都通过一个统一的 cursor 上下文管理器执行（checkpoint/sqlite/__init__.py:168-189）：

# sqlite/__init__.py:168
@contextmanager
def cursor(self, transaction: bool = True) -> Iterator[sqlite3.Cursor]:
    with self.lock:                    # ① 先抢锁 —— 同一时刻只有一个线程能进
        self.setup()                   # ② 确保表已建（懒建表就发生在这）
        cur = self.conn.cursor()
        try:
            yield cur                  # ③ 把游标交给调用方执行 SQL
        finally:
            if transaction:
                self.conn.commit()     # ④ 默认提交事务（写操作要）
            cur.close()                # ⑤ 无论如何都关游标

存一份档案，核心就一条 SQL（checkpoint/sqlite/__init__.py:387,418-443）：

# sqlite/__init__.py:418
thread_id = config["configurable"]["thread_id"]
checkpoint_ns = config["configurable"]["checkpoint_ns"]
type_, serialized_checkpoint = self.serde.dumps_typed(checkpoint)   # ① 整包序列化档
serialized_metadata = json.dumps(
    get_checkpoint_metadata(config, metadata), ensure_ascii=False
).encode("utf-8", "ignore")                                          # ② metadata 存 JSON
with self.cursor() as cur:
    cur.execute(
        "INSERT OR REPLACE INTO checkpoints "
        "(thread_id, checkpoint_ns, checkpoint_id, parent_checkpoint_id, type, checkpoint, metadata) "
        "VALUES (?, ?, ?, ?, ?, ?, ?)",
        (
            str(thread_id), checkpoint_ns, checkpoint["id"],
            config["configurable"].get("checkpoint_id"),   # ③ 父 id = 当前 config 的 id
            type_, serialized_checkpoint, serialized_metadata,
        ),
    )
return {"configurable": {"thread_id": thread_id, "checkpoint_ns": checkpoint_ns,
                         "checkpoint_id": checkpoint["id"]}}          # ④ 返回新坐标

取档比存档多一步——档和它的 pending_writes 在两张表，要查两次（checkpoint/sqlite/__init__.py:191,226-293）：

# sqlite/__init__.py:226
checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
with self.cursor(transaction=False) as cur:            # 读操作，不开事务
    if checkpoint_id := get_checkpoint_id(config):     # 给了 id → 查指定档
        cur.execute("SELECT ... FROM checkpoints WHERE thread_id=? AND checkpoint_ns=? AND checkpoint_id=?", ...)
    else:                                               # 没给 id → 查最新档
        cur.execute("SELECT ... FROM checkpoints WHERE thread_id=? AND checkpoint_ns=? "
                    "ORDER BY checkpoint_id DESC LIMIT 1", ...)   # ← 按 id 降序取第一条
    if value := cur.fetchone():
        (thread_id, checkpoint_id, parent_checkpoint_id, type, checkpoint, metadata) = value
        # 第二次查询：捞这份档的 pending_writes
        cur.execute("SELECT task_id, channel, type, value FROM writes "
                    "WHERE thread_id=? AND checkpoint_ns=? AND checkpoint_id=? ORDER BY task_id, idx", ...)
        return CheckpointTuple(
            config,
            self.serde.loads_typed((type, checkpoint)),      # ① 反序列化档（通道值就在里面）
            json.loads(metadata) if metadata else {},        # ② metadata 反序列化
            ({...parent...} if parent_checkpoint_id else None),  # ③ 父坐标
            [(task_id, channel, self.serde.loads_typed((type, value)))    # ④ 半成品
             for task_id, channel, type, value in cur],
        )

存半成品写入时，它会根据"是不是特殊写"切换 SQL（checkpoint/sqlite/__init__.py:445,462-482）：

# sqlite/__init__.py:462
query = (
    "INSERT OR REPLACE INTO writes (...) VALUES (?, ?, ?, ?, ?, ?, ?, ?)"
    if all(w[0] in WRITES_IDX_MAP for w in writes)     # ① 全是特殊写 → 允许覆盖
    else "INSERT OR IGNORE INTO writes (...) VALUES (?, ?, ?, ?, ?, ?, ?, ?)"  # ② 含普通写 → 已存在则忽略
)
with self.cursor() as cur:
    cur.executemany(
        query,
        [
            (str(thread_id), str(checkpoint_ns), str(checkpoint_id),
             task_id,
             WRITES_IDX_MAP.get(channel, idx),          # ③ 特殊写用负数 idx，普通写用序号
             channel, *self.serde.dumps_typed(value))
            for idx, (channel, value) in enumerate(writes)
        ],
    )

还记得 Day 35 InMemory 的异步方法"直接调同步"吗？SqliteSaver 更干脆——直接不支持异步（checkpoint/sqlite/__init__.py:585-624）：

# sqlite/__init__.py:585
async def aget_tuple(self, config) -> CheckpointTuple | None:
    raise NotImplementedError(_AIO_ERROR_MSG)     # 抛异常，附带指路信息

# _AIO_ERROR_MSG（:34）大意：
# "SqliteSaver 不支持 async。请改用 AsyncSqliteSaver（需装 aiosqlite）：
#  from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver"

async def aput(self, config, checkpoint, metadata, new_versions):
    raise NotImplementedError(_AIO_ERROR_MSG)

✋ 10 分钟动手

# 1. 读 setup 建表 + cursor 锁
sed -n '129,189p' libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py

# 2. put / put_writes 两段对照读
sed -n '418,443p;462,482p' libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py

# 3. 亲手落盘一个对话，再用 sqlite3 命令行看两张表
python - <<'PY'
from langgraph.graph import StateGraph, START
from langgraph.checkpoint.sqlite import SqliteSaver
import sqlite3
g=StateGraph(dict); g.add_node("a",lambda s:{"n":s.get("n",0)+1}); g.add_edge(START,"a")
conn=sqlite3.connect("demo.sqlite", check_same_thread=False)
app=g.compile(checkpointer=SqliteSaver(conn))
app.invoke({"n":0}, {"configurable":{"thread_id":"t1"}})
print("表:", conn.execute("SELECT name FROM sqlite_master WHERE type='table'").fetchall())
print("档数:", conn.execute("SELECT count(*) FROM checkpoints").fetchone())
PY