PostgresSaver: Produktionsfähiger relationaler Checkpoint

源码版本1.2.9

Verantwortung

PostgresSaver ist die BaseCheckpointSaver-Implementierung für Postgres und der von LangGraph offiziell empfohlene Saver für den Produktiveinsatz. Der Code liegt in libs/checkpoint-postgres/langgraph/checkpoint/postgres/, der synchrone Einstieg ist PostgresSaver in libs/checkpoint-postgres/langgraph/checkpoint/postgres/__init__.py (PostgresSaver:40), die asynchrone Variante in aio.py als AsyncPostgresSaver (AsyncPostgresSaver:40); beide teilen sich die Basisklasse BasePostgresSaver (BasePostgresSaver:297), die für die SQL-Templates und Migrations-Skripte zuständig ist.

Anders als bei SQLite, das mit einer einzigen Tabelle und Primärschlüssel-Deduplizierung arbeitet, zerlegt PostgresSaver die Daten in drei Tabellen: checkpoints (Haupttabelle, JSONB), checkpoint_blobs (binäre Channel-Werte) und checkpoint_writes (inkrementelle Task-Schreibvorgänge) (MIGRATIONS:43-91). Diese Aufteilung erlaubt es, beim Lesen mit einem einzigen SELECT inklusive lateraler Unterabfragen (lateral subqueries) die Haupttabelle, Blobs und Writes in einem Rutsch zurückzuholen (SELECT_SQL:93-118), was N+1-Queries wie bei SQLite vermeidet; gleichzeitig werden Blobs separat als BYTEA gespeichert und nicht in die JSONB-Deserialisierung einbezogen, sodass große Objekte das Lesen der Haupttabelle nicht ausbremsen.

PostgresSaver unterstützt zusätzlich den Pipeline-Modus von psycopg3 (supports_pipeline:60) – mehrere Statements werden gebündelt abgeschickt, ohne dass der Server sofort Ergebnisse zurückgibt; ideal für Kombinationen wie executemany(blob) + execute(checkpoint) innerhalb eines einzelnen put.

Entwurfsmotivation

• Drei-Tabellen-Trennung, JSONB für den Hauptkörper: Die checkpoints-Haupttabelle speichert das Checkpoint-TypedDict in einer JSONB-Spalte, während checkpoint_blobs große Objekte separat vorhält (checkpoint_blobs:57-65). Das vermeidet es, die gesamten channel_values in die JSONB-Zeile zu quetschen – ein Ansatz, der bei vielen Kanälen oder großen Werten (channel) in die Knie geht.
• Ein einziger SELECT setzt das Tuple zusammen: SELECT_SQL nutzt zwei laterale Unterabfragen (array_agg über checkpoint_blobs + array_agg über checkpoint_writes), um alle Daten in einem Durchlauf zu holen (lateral subqueries:101-117); _load_checkpoint_tuple muss auf Python-Seite nur noch das Array entpacken, ohne zweite Abfragen.
• UPSERT mit ON CONFLICT DO UPDATE/NOTHING für unterschiedliche Semantik: UPSERT_CHECKPOINT_BLOBS_SQL nutzt DO NOTHING (UPSERT_BLOBS:131-135), UPSERT_CHECKPOINTS_SQL nutzt DO UPDATE (UPSERT_CHECKPOINTS:137-144), UPSERT_CHECKPOINT_WRITES_SQL nutzt DO UPDATE, und INSERT_CHECKPOINT_WRITES_SQL nutzt DO NOTHING (INSERT_WRITES:155-159). Kontroll-Channel-Schreibungen (writes) überschreiben den neuesten Zustand (UPDATE), normale Schreibungen werden bei Konflikt ignoriert (NOTHING) – die Semantik entspricht dem if inner_key[1] >= 0 ... continue aus InMemory.
• Migrationsliste wird per Versionsnummer ergänzt: MIGRATIONS ist eine Liste (MIGRATIONS:43-91); setup() liest die maximale Version aus der Tabelle checkpoint_migrations und führt nur die darauffolgenden Migrationen aus (setup:85-110). Neue Spalten, Indizes oder Constraints werden durch Anhängen eines Migrations-Strings ergänzt, ohne bestehende Migrationen zu verändern.
• Pipeline-Modus für echte Nebenläufigkeit auf einer Verbindung: Bei pipeline=True holt sich der synchrone PostgresSaver._cursor den Kontext conn.pipeline() (pipeline branch:424-432), sodass mehrere Cursor auf derselben Verbindung sich nicht gegenseitig blockieren; der pipe-Modus (über from_conn_string(pipeline=True)) erlaubt es einer Verbindung, unter mehreren Threads wiederverwendet zu werden (pipe branch:413-423), kombiniert mit self.lock für serielle Cursor, aber nicht serielle Statements.

Schlüsseldateien

• MIGRATIONS:43-91 — CREATE-Statements der drei Tabellen + WAL-/Index-/task_path-Migrationen.
• SELECT_SQL:93-118 — Hauptabfrage, laterale Unterabfragen fassen blobs und writes via array_agg zusammen.
• SELECT_PENDING_SENDS_SQL:120-129 — Liest ausstehende PUSH-Task-Writes, sortiert nach task_path / task_id / idx.
• UPSERT / INSERT SQL:131-159 — Vier Schreib-Statements, unterteilt in DO UPDATE / DO NOTHING.
• BasePostgresSaver:297 — Gemeinsame Basisklasse, hält die SQL-Templatе-Konstanten.
• PostgresSaver Klasse:40-60 — Synchrone Implementierung, nimmt Connection oder ConnectionPool, mit threading.Lock.
• from_conn_string:62-83 — Context-Manager-Factory, bei pipeline=True wird conn.pipeline() verwendet.
• setup:85-110 — Liest die Migrationsversion, führt noch nicht angewandte Migrationen in Reihenfolge aus und notiert die Version.
• put:263-345 — Verschiebt _DeltaSnapshot / nicht-primitive Werte nach blob_values, der Hauptkörper geht ins JSONB.
• _cursor:405-442 — Wählt je nach Pipeline / Pipe einen von drei Kontexten aus.
• AsyncPostgresSaver:40 — Asynchrone Variante, aput / aget_tuple / alist sind nativ async.

Datenfluss

PostgresSaver.put ist ein gutes Beispiel dafür, wie ein Saver das Checkpoint-TypedDict zerlegt – es entscheidet zuerst, welche Channel-Werte inline in den JSONB-Hauptkörper wandern und welche nach checkpoint_blobs verschoben werden, und führt dann innerhalb eines Cursors executemany(blob) + execute(checkpoint) aus:

# 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) Beachten Sie, dass _DeltaSnapshot einen Sonderweg geht – der Channel-Wert wird in der Haupttabelle nur durch ein True als Platzhalter repräsentiert, die echten Daten landen im Blob; das ist die Grundlage, auf der DeltaChannel beim Wiederaufbau rückwärts durch die Ahnenkette läuft. Beim Zurücklesen führt get_tuple das SELECT_SQL aus; psycopg wandelt JSONB automatisch in ein dict und BYTEA in bytes um, und auf Python-Seite wird _load_checkpoint_tuple aufgerufen, um das CheckpointTuple zusammenzusetzen (_load_checkpoint_tuple:552).

Grenzen und Fehler

• from_conn_string(pipeline=True) lässt sich nicht mit ConnectionPool kombinieren: __init__ wirft explizit eine Ausnahme (pool vs pipe check:52-55); Pipeline ist ein Einzelverbindungsmodus, ein Pool ist bereits verbindungsübergreifend.
• setup() muss explizit einmal aufgerufen werden: Der Docstring hält fest: «MUST be called directly by the user the first time checkpointer is used» (setup docstring:85-91). Es legt Tabellen an und führt Migrationen aus; vorher müssen die Datenbankberechtigungen stehen.
• MIGRATIONS-Reihenfolge darf nicht geändert werden: Migrationen verwenden die Listenposition als Versionsnummer (MIGRATIONS:43-91), bereits veröffentlichte Migrationsstrings dürfen nicht modifiziert werden; sonst verschiben sich die in checkpoint_migrations eingetragenen Versionen relativ zu den Code-Positionen.
• CREATE INDEX CONCURRENTLY scheitert innerhalb einer Transaktion: Die beiden nebenläufigen Index-Erzeugungen in MIGRATIONS (CONCURRENTLY indexes:82-89) werden im selben setup()-Kontext wie die anderen Migrationen ausgeführt und schlagen fehl, wenn die Verbindung nicht im autocommit-Modus ist; darum setzt from_conn_string explizit autocommit=True (autocommit:76-78).
• supports_pipeline wird zur Laufzeit geprüft: Capabilities().has_pipeline() (has_pipeline:60) ermittelt in __init__ die psycopg-Version und serverseitige Fähigkeiten; bei älterem psycopg oder manchen Managed-Postgres-Anbietern fällt es auf den normalen Transaktionsmodus zurück (transaction fallback:434-439).
• DeltaChannel hat einen eigenen Fast-Path: get_delta_channel_history ist in PostgresSaver überschrieben (get_delta_channel_history:444-463) und läuft über eine zweistufige Abfrage (Stage 1: Metadata-Paging + Stage 2: UNION ALL für Writes) statt über die Standardimplementierung der Basisklasse; ein eigener Saver muss diese Schnittstelle ebenfalls implementieren, um DeltaChannel zu unterstützen.

Zusammenfassung

PostgresSaver ist der von LangGraph empfohlene Saver für den Produktiveinsatz; die Drei-Tabellen-Aufteilung lässt «JSONB-Haupttabelle + BYTEA für große Objekte + inkrementelle writes» jeweils in den passenden Speicher fließen. Der Pipeline-Modus und die native Async-Implementierung liefern den Durchsatz für produktive Skalierung. Die Schnittstellenform siehe BaseCheckpointSaver, für leichtgewichtige Szenarien siehe InMemorySaver + SqliteSaver.

Siehe offizielle Dokumentation: LangGraph 文档 · README.