-- postgres/base.py:47  checkpoints：档本体（注意 checkpoint 列是 JSONB！）
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,
    checkpoint JSONB NOT NULL,            -- 档（不含大通道值），JSONB 可被 SQL 内部查询
    metadata JSONB NOT NULL DEFAULT '{}',
    PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id)
);
-- postgres/base.py:57  checkpoint_blobs：通道值单独存，按 (channel, version) 去重
CREATE TABLE IF NOT EXISTS checkpoint_blobs (
    thread_id TEXT NOT NULL, checkpoint_ns TEXT NOT NULL DEFAULT '',
    channel TEXT NOT NULL, version TEXT NOT NULL,
    type TEXT NOT NULL, blob BYTEA,
    PRIMARY KEY (thread_id, checkpoint_ns, channel, version)   -- ← 版本去重的关键
);
-- postgres/base.py:66  checkpoint_writes：半成品写入（同 Sqlite 的 writes）
-- postgres/base.py:44  checkpoint_migrations：只有一列 v，记录已迁移到第几版

Postgres 用一个字符串列表管理所有历史表结构变更（checkpoint/postgres/base.py:43-91）：

# postgres/base.py:40 注释：
#   "To add a new migration, add a new string to the MIGRATIONS list.
#    The position of the migration in the list is the version number."
MIGRATIONS = [
    "CREATE TABLE IF NOT EXISTS checkpoint_migrations (v INTEGER PRIMARY KEY);",   # v=0
    "CREATE TABLE IF NOT EXISTS checkpoints (...);",                              # v=1
    "CREATE TABLE IF NOT EXISTS checkpoint_blobs (...);",                         # v=2
    "CREATE TABLE IF NOT EXISTS checkpoint_writes (...);",                        # v=3
    "ALTER TABLE checkpoint_blobs ALTER COLUMN blob DROP not null;",              # v=4
    "SELECT 1;",   # v=5  故意的 no-op（历史遗留，占位保证后续版本号对齐）
    "CREATE INDEX CONCURRENTLY IF NOT EXISTS checkpoints_thread_id_idx ...;",      # v=6
    "... checkpoint_blobs_thread_id_idx ...",                                     # v=7
    "... checkpoint_writes_thread_id_idx ...",                                    # v=8
    "ALTER TABLE checkpoint_writes ADD COLUMN IF NOT EXISTS task_path TEXT ...;", # v=9
]

setup 的核心就是"查当前版本 → 把之后的迁移补跑"（checkpoint/postgres/__init__.py:85-110）：

# postgres/__init__.py:85
def setup(self) -> None:
    with self._cursor() as cur:
        cur.execute(self.MIGRATIONS[0])         # ① 先确保 migrations 表存在（第 0 条）
        results = cur.execute(
            "SELECT v FROM checkpoint_migrations ORDER BY v DESC LIMIT 1")
        row = results.fetchone()
        version = -1 if row is None else row["v"]   # ② 查"已迁到第几版"，没有则 -1
        for v, migration in zip(
            range(version + 1, len(self.MIGRATIONS)),      # ③ 从"下一版"到"最新版"
            self.MIGRATIONS[version + 1 :],
            strict=False,
        ):
            cur.execute(migration)                          # 跑这条迁移
            cur.execute("INSERT INTO checkpoint_migrations (v) VALUES (%s)", (v,))  # 记账
    if self.pipe:
        self.pipe.sync()

Sqlite 取档要查两次（档 + 半成品）并在 Python 里组装。Postgres 更狠——一条 SQL 用子查询把通道值和半成品全 JOIN 拼好（checkpoint/postgres/base.py:93-118）：

-- postgres/base.py:93
select
    thread_id, checkpoint, checkpoint_ns, checkpoint_id, parent_checkpoint_id, metadata,
    (   -- ① 子查询：按 channel_versions 把 blobs 表里的通道值 JOIN 出来
        select array_agg(array[bl.channel::bytea, bl.type::bytea, bl.blob])
        from jsonb_each_text(checkpoint -> 'channel_versions')      -- 展开档里的"通道→版本"
        inner join checkpoint_blobs bl
            on bl.thread_id = checkpoints.thread_id
            and bl.checkpoint_ns = checkpoints.checkpoint_ns
            and bl.channel = jsonb_each_text.key         -- 通道名对上
            and bl.version = jsonb_each_text.value       -- 版本号也对上
    ) as channel_values,
    (   -- ② 子查询：把这份档的 pending_writes 也一起捞出来
        select array_agg(array[cw.task_id::text::bytea, cw.channel::bytea, cw.type::bytea, cw.blob]
                         order by cw.task_id, cw.idx)
        from checkpoint_writes cw
        where cw.thread_id = checkpoints.thread_id and ... = checkpoints.checkpoint_id
    ) as pending_writes
from checkpoints

存档时它比 Sqlite 多一个聪明判断——不是所有通道值都拆 blob，小的基本类型直接内联进 JSONB（checkpoint/postgres/__init__.py:299-345）：

# postgres/__init__.py:299
copy = checkpoint.copy()
copy["channel_values"] = copy["channel_values"].copy()
blob_values = {}
for k, v in checkpoint["channel_values"].items():
    if isinstance(v, _DeltaSnapshot):                 # DeltaChannel 快照 → 拆 blob
        blob_values[k] = copy["channel_values"].pop(k)
        copy["channel_values"][k] = True              #   档里留个占位 True
    elif v is None or isinstance(v, (str, int, float, bool)):
        pass                                          # ← 小基本类型：留在档 JSONB 里，不拆
    else:
        blob_values[k] = copy["channel_values"].pop(k)   # 其他（大对象）→ 拆进 blobs
with self._cursor(pipeline=True) as cur:
    if blob_versions := {k: v for k, v in new_versions.items() if k in blob_values}:
        cur.executemany(self.UPSERT_CHECKPOINT_BLOBS_SQL,
                        self._dump_blobs(thread_id, checkpoint_ns, blob_values, blob_versions))
    cur.execute(self.UPSERT_CHECKPOINTS_SQL,
                (thread_id, checkpoint_ns, checkpoint["id"], checkpoint_id,
                 Jsonb(copy), Jsonb(get_serializable_checkpoint_metadata(config, metadata))))

Sqlite 用"一把锁串行化"，Postgres 用真正的并发机制。核心在 _cursor（checkpoint/postgres/__init__.py:404-443）：

# postgres/__init__.py:404
@contextmanager
def _cursor(self, *, pipeline: bool = False):
    with self.lock, _internal.get_connection(self.conn) as conn:   # 从连接池取连接
        if self.pipe:
            # 连接已处于 pipeline 模式：可被多协程并发使用，但一次一个游标
            try:
                with conn.cursor(binary=True, row_factory=dict_row) as cur:
                    yield cur
            finally:
                if pipeline:
                    self.pipe.sync()             # 一次性把批量攒的命令发出去
        elif pipeline:
            if self.supports_pipeline:
                with conn.pipeline(), conn.cursor(binary=True, row_factory=dict_row) as cur:
                    yield cur                    # 临时开 pipeline，批量发送减少往返
            else:
                with conn.transaction(), conn.cursor(...) as cur:   # 不支持则退回普通事务
                    yield cur
        else:
            with conn.cursor(binary=True, row_factory=dict_row) as cur:
                yield cur

最后看幂等——Postgres 用标准 ON CONFLICT（Sqlite 是 INSERT OR REPLACE/IGNORE）（checkpoint/postgres/base.py:131-159）：

-- postgres/base.py:131  通道值：主键冲突 → 什么都不做（版本相同即同值，无需覆盖）
INSERT INTO checkpoint_blobs (...) VALUES (...)
ON CONFLICT (thread_id, checkpoint_ns, channel, version) DO NOTHING;

-- postgres/base.py:137  档：主键冲突 → 更新（同一 checkpoint_id 重存，覆盖）
INSERT INTO checkpoints (...) VALUES (...)
ON CONFLICT (thread_id, checkpoint_ns, checkpoint_id)
DO UPDATE SET checkpoint = EXCLUDED.checkpoint, metadata = EXCLUDED.metadata;

-- put_writes 两条（:146 UPSERT / :155 INSERT ... DO NOTHING）
-- 全特殊写 → DO UPDATE（允许覆盖）；含普通写 → DO NOTHING（幂等去重）

✋ 10 分钟动手

# 1. 四张表定义 + MIGRATIONS 列表
sed -n '43,91p' libs/checkpoint-postgres/langgraph/checkpoint/postgres/base.py

# 2. 那条"JOIN 拼通道值"的 SELECT_SQL
sed -n '93,118p' libs/checkpoint-postgres/langgraph/checkpoint/postgres/base.py

# 3. setup 断点续跑 + put 小值内联
sed -n '85,110p' libs/checkpoint-postgres/langgraph/checkpoint/postgres/__init__.py
sed -n '308,320p' libs/checkpoint-postgres/langgraph/checkpoint/postgres/__init__.py

# 4. 对照 Sqlite/Postgres 的幂等语法差异
grep -n "OR IGNORE\|OR REPLACE" libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py
grep -n "ON CONFLICT" libs/checkpoint-postgres/langgraph/checkpoint/postgres/base.py