PostgresSaver: production-grade relational checkpoints

源码版本1.2.9

Responsibilities

PostgresSaver is the BaseCheckpointSaver implementation on Postgres and is LangGraph's officially recommended production saver. The code lives in libs/checkpoint-postgres/langgraph/checkpoint/postgres/; the sync entry point is PostgresSaver in libs/checkpoint-postgres/langgraph/checkpoint/postgres/__init__.py (PostgresSaver:40), the async version is AsyncPostgresSaver in aio.py (AsyncPostgresSaver:40), and they share the BasePostgresSaver base class (BasePostgresSaver:297), which holds the SQL templates and migration scripts.

Unlike SQLite's single-table + primary-key dedup approach, PostgresSaver splits the data into three tables: checkpoints (main table, JSONB), checkpoint_blobs (channel value binary), and checkpoint_writes (task incremental writes) (MIGRATIONS:43-91). This split lets reads fetch main + blobs + writes in one SELECT with lateral subqueries (SELECT_SQL:93-118), avoiding the N+1 queries SQLite suffers from. At the same time, the blob column is a standalone BYTEA that does not participate in JSONB deserialization, so large objects do not slow down main-table reads.

PostgresSaver additionally supports psycopg3's Pipeline mode (supports_pipeline:60) — multiple statements are batched and sent down, with the server not returning results immediately. This fits the executemany(blob) + execute(checkpoint) combination inside a single put.

Design motivation

• Three-table split, JSONB for the body: the checkpoints main table uses a JSONB column to hold the Checkpoint TypedDict; checkpoint_blobs stores large objects separately (checkpoint_blobs:57-65). This avoids serializing the entire channel_values into JSONB — an approach that falls over when there are many channels or large values.
• One SELECT assembles the full tuple: SELECT_SQL uses two lateral subqueries (array_agg over checkpoint_blobs + array_agg over checkpoint_writes) to fetch everything in one round trip (lateral subqueries:101-117); _load_checkpoint_tuple just unpacks the arrays on the Python side, no second query needed.
• UPSERT uses ON CONFLICT DO UPDATE/NOTHING to distinguish semantics: UPSERT_CHECKPOINT_BLOBS_SQL uses DO NOTHING (UPSERT_BLOBS:131-135), UPSERT_CHECKPOINTS_SQL uses DO UPDATE (UPSERT_CHECKPOINTS:137-144), UPSERT_CHECKPOINT_WRITES_SQL uses DO UPDATE, and INSERT_CHECKPOINT_WRITES_SQL uses DO NOTHING (INSERT_WRITES:155-159). Control-channel writes need to overwrite the latest state (UPDATE); ordinary business writes that collide are ignored (NOTHING) — mirroring InMemory's if inner_key[1] >= 0 ... continue.
• Migration list appends version numbers: MIGRATIONS is a list (MIGRATIONS:43-91); setup() reads the maximum v from the checkpoint_migrations table and runs only the migrations after it (setup:85-110). Adding a column, an index, or changing a constraint is done by appending a migration string — existing migrations are never edited.
• Pipeline mode gives true concurrency on a single connection: when pipeline=True, sync PostgresSaver._cursor acquires a conn.pipeline() context (pipeline branch:424-432); multiple cursors on the same connection do not block each other. The pipe mode (from from_conn_string(pipeline=True)) lets one connection be reused across threads (pipe branch:413-423), with self.lock serializing cursor acquisition but not statement execution.

Key files

• MIGRATIONS:43-91 — CREATE statements for the three tables + WAL/index/task_path migrations.
• SELECT_SQL:93-118 — main query; lateral subqueries array_agg blobs and writes together.
• SELECT_PENDING_SENDS_SQL:120-129 — fetches pending PUSH task writes, ordered by task_path / task_id / idx.
• UPSERT / INSERT SQL:131-159 — four write statements, split between DO UPDATE and DO NOTHING.
• BasePostgresSaver:297 — shared base class, holds the SQL template constants.
• PostgresSaver class:40-60 — sync implementation; takes a Connection or ConnectionPool, with a threading.Lock.
• from_conn_string:62-83 — context manager factory; pipeline=True uses conn.pipeline().
• setup:85-110 — reads the migration version, runs unapplied migrations in order, records v.
• put:263-345 — moves _DeltaSnapshot / non-primitive values into blob_values, writes the body into JSONB.
• _cursor:405-442 — picks among three context modes based on pipeline / pipe.
• AsyncPostgresSaver:40 — async version; aput / aget_tuple / alist are natively async.

Data flow

PostgresSaver.put is a good example of how a saver decomposes the Checkpoint TypedDict — it first decides which channel values are inlined into the JSONB body and which are moved to checkpoint_blobs, then runs executemany(blob) + execute(checkpoint) inside a single cursor:

# inline primitive values in checkpoint table
# others are stored in blobs table
blob_values = {}
for k, v in checkpoint["channel_values"].items():
    if isinstance(v, _DeltaSnapshot):
        blob_values[k] = copy["channel_values"].pop(k)
        copy["channel_values"][k] = True
    elif v is None or isinstance(v, (str, int, float, bool)):
        pass
    else:
        blob_values[k] = copy["channel_values"].pop(k)

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)),
        ),
    )

(put body:309-344) Note the special path for _DeltaSnapshot — the channel value in the main table is just a True placeholder, while the real data goes into the blob. This is what DeltaChannel walks the ancestor chain to reconstruct on read-back. On read-back, get_tuple runs SELECT_SQL; psycopg converts JSONB to dict and BYTEA to bytes automatically, and _load_checkpoint_tuple assembles the CheckpointTuple on the Python side (_load_checkpoint_tuple:552).

Boundaries and failure modes

• from_conn_string(pipeline=True) cannot be combined with a ConnectionPool: __init__ explicitly raises (pool vs pipe check:52-55); pipeline is a single-connection mode, while a pool is already multi-connection.
• setup() must be called explicitly once: the docstring says "MUST be called directly by the user the first time checkpointer is used" (setup docstring:85-91). It creates tables and runs migrations; the database user must have the necessary permissions before it runs.
• Reordering MIGRATIONS breaks things: migrations use list position as the version number (MIGRATIONS:43-91); a published migration string cannot be edited, or the v in checkpoint_migrations will be out of sync with the code's position.
• CREATE INDEX CONCURRENTLY fails inside a transaction: the two concurrent index creation statements in MIGRATIONS (CONCURRENTLY indexes:82-89) run in the same setup() context as other migrations; if the connection is not in autocommit mode they will fail. from_conn_string explicitly sets autocommit=True for this reason (autocommit:76-78).
• supports_pipeline is detected at runtime: Capabilities().has_pipeline() (has_pipeline:60) is queried in __init__ for both the psycopg version and server-side capability. Old psycopg or certain managed Postgres offerings fall back to plain transaction mode (transaction fallback:434-439).
• DeltaChannel has a dedicated fast path: get_delta_channel_history is overridden in PostgresSaver (get_delta_channel_history:444-463) to run a two-stage query (Stage 1 paginates metadata + Stage 2 UNION ALL fetches writes), bypassing the base class default. A custom saver must implement this interface too if it wants to support DeltaChannel.

Summary

PostgresSaver is LangGraph's recommended production saver. Its three-table split lets "main-table JSONB + large-object BYTEA + incremental writes" each flow through the most suitable storage. Pipeline mode and the native async implementation deliver production-scale throughput. For the interface shape see BaseCheckpointSaver; for lightweight scenarios see InMemorySaver + SqliteSaver.

See official docs: LangGraph 文档 · README.