Sqlite与Postgres

InMemorySaver 的三本账在进程 RAM 里，重启即失。上线把同一套 BaseCheckpointSaver 接口换成 SQLite（单机/边缘）或 PostgreSQL（多副本、备份）。图侧写法不变：compile(checkpointer=cp) + thread_id。

段末注释：BaseCheckpointSaver = get_tuple / put / put_writes / list 契约；换后端不改 Pregel 调用点。

社区方案：官方独立包 langgraph-checkpoint-sqlite / langgraph-checkpoint-postgres，不要自写 ORM 表。风险：漏 setup() 无表；SQLite 写锁扛不住多 worker；连接泄漏。

图 1　图只认 Saver 电话；左边 RAM、右边磁盘，信封仍是 (thread_id, ns, id)（对应 §4）

1. 定位

维度	内容
角色	生产级 checkpointer 后端
输入 → 输出	同一 `get_tuple` / `put` → 行落在 DB
核心 API	`SqliteSaver.from_conn_string`、`PostgresSaver`
依赖 LangChain	无

2. 图拓扑

节点表

节点名	职责	写
`tick`	计数	`n`

边表

源	目标	类型
START	tick	固定
tick	END	固定

3. invoke 生命周期

1. Saver.setup() 建 checkpoints / writes 表（首次）
2. compile(checkpointer=cp) 仍把 Saver 挂到 Pregel
3. invoke：调度前 get_tuple → SELECT 最新行
4. 每 superstep put / put_writes → INSERT
5. 进程重启后同 thread_id：get_tuple 从磁盘读回 n

4. 原理

4.1 图怎么调：与 InMemorySaver 同一套电话

Pregel 不区分 内存还是 Postgres。invoke / get_state / update_state 仍然只打：

时机	方法	SQL 近似
调度前、`get_state`	`get_tuple(config)`	`WHERE thread_id=? AND checkpoint_ns=?`；无 id 则 `ORDER BY checkpoint_id DESC LIMIT 1`
superstep 结束	`put(...)`	`INSERT` 到 `checkpoints`，channel 值进 blob 列
节点完成	`put_writes(...)`	`INSERT` 到 `writes`，挂同一三维键
`get_state_history`	`list(config)`	同 thread+ns，id 降序

复合主键仍是 (thread_id, checkpoint_ns, checkpoint_id)。换 Saver 不必改 cfg。

4.2 三本账如何落表

InMemorySaver 的 storage / blobs / writes 对应推荐 schema：

-- storage + blobs 的合体
PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id)  -- checkpoints

-- pending writes
PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id, task_id, task_path, idx)  -- writes

serde.dumps_typed 仍负责序列化；只是字节进 BYTEA / BLOB，不是进 Python dict。

4.3 包与连接

实现不在核心 langgraph，需额外 install。SQLite：sqlite:///checkpoints.db；Postgres：postgresql://user:pass@host/db。首次 setup()（或部分版本 from_conn_string 内建表）。用 with Saver... 管连接。

5. 最小可运行示例（SQLite）

import tempfile
from typing import TypedDict

from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.graph import StateGraph, START, END


class State(TypedDict):
    n: int


def tick(state: State) -> dict:
    return {"n": state["n"] + 1}  # 拓扑 §2 tick


builder = StateGraph(State)
builder.add_node("tick", tick)
builder.add_edge(START, "tick")
builder.add_edge("tick", END)

with tempfile.NamedTemporaryFile(suffix=".db") as f:
    conn = f"sqlite:///{f.name}"
    with SqliteSaver.from_conn_string(conn) as cp:
        graph = builder.compile(checkpointer=cp)
        cfg = {"configurable": {"thread_id": "sqlite-1"}}
        graph.invoke({"n": 0}, cfg)
        out = graph.invoke({"n": 0}, cfg)  # get_tuple 读到 n=1 再 +1
        print(out["n"])  # 2
        print(graph.get_state(cfg).values["n"])  # 2

5.1 Postgres（示意）

# pip install langgraph-checkpoint-postgres psycopg[binary]
# from langgraph.checkpoint.postgres import PostgresSaver
# with PostgresSaver.from_conn_string(DB_URI) as cp:
#     cp.setup()
#     graph = builder.compile(checkpointer=cp)

6. 执行追踪

调用	Saver 动作	`n`
第1次 invoke	`get_tuple` 空 → `put` n=1	1
进程仍在，第2次	`get_tuple` 读 n=1 → `put` n=2	2
换进程、同 db、同 thread	`get_tuple` 仍读到 n=2	续跑

InMemorySaver 在「换进程」这一行会变空。

重要配置参数

参数（API 名）	类型 / 默认值	功能说明	作用与影响	参考起点 / 常用范围	配置指导
`SqliteSaver.from_conn_string`	context mgr	打开本地 DB 并实现 Saver	文件锁、单机	`sqlite:///...`	高并发换 Postgres
`PostgresSaver.from_conn_string`	context mgr	打开集群 DB	多副本共享	连接串	`setup()` + 连接池
`cp.setup()`	DDL	建 checkpoints / writes	漏了则 INSERT 失败	首次部署	迁移走官方
`compile(checkpointer=cp)`	同内存篇	图仍只打 get_tuple/put	写法不变	任意 Saver	勿混用两个 Saver 实例写同一 thread
`configurable.thread_id`	str	行隔离第一维	须建索引	UUID	长度 < 255（Postgres 列限）
连接池 / 备份	运维	并发与合规	慢查、retention	QPS、GDPR	监控