InMemorySaver + SqliteSaver: In-Process- und Datei-Level-Checkpoints

源码版本1.2.9

Verantwortung

BaseCheckpointSaver liefert die Schnittstellenform; die beiden am häufigsten verwendeten Saver, die sie tatsächlich implementieren, sind InMemorySaver und SqliteSaver – ersterer stopft den gesamten Zustand in ein prozessinternes defaultdict, letzterer zerlegt dieselben Felder in zwei SQLite-Tabellen checkpoints und writes. Beide decken zusammen die Szenarien «Entwicklung/Debugging» und «Ein-Maschine-Persistenz» ab; für produktive Skalierung wechselt man zu PostgresSaver.

InMemorySaver liegt in libs/checkpoint/langgraph/checkpoint/memory/__init__.py (InMemorySaver:33); intern hält es drei defaultdicts für drei Dinge: storage für Checkpoint-Hauptkörper + Metadaten + parent_id, writes für (thread_id, ns, checkpoint_id) → (task_id, idx) → Schreib-Datensatz, und blobs für (thread_id, ns, channel, version) → serialisierte Binärdaten (storage / writes / blobs:68-83). Diese Bucket-Struktur entspricht eins zu eins der Drei-Tabellen-Struktur von Postgres/SQLite, nur dass statt Tabellen dicts verwendet werden.

SqliteSaver liegt in libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py (SqliteSaver:45). Es erbt direkt von BaseCheckpointSaver[str], nimmt eine sqlite3.Connection entgegen und legt in setup() zwei Tabellen an (<SrcLink path="libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py" lines="129-166" label="setup"/>); put / put_writes nutzen Standard-SQL mit INSERT OR REPLACE / INSERT OR IGNORE. Es bringt ein eigenes threading.Lock mit, das die einzelne Verbindung serialisiert, weshalb check_same_thread=False sicher ist.

Zudem ist MemorySaver = InMemorySaver ein historischer Alias (MemorySaver alias:631); from langgraph.checkpoint.memory import MemorySaver aus älterem Code liefert dieselbe Klasse.

Entwurfsmotivation

Die drei Dict-Buckets von InMemory entsprechen drei Tabellen: Die drei Felder storage / writes / blobs (data members:68-83) entsprechen eins zu eins den SQLite-Tabellen checkpoints / writes und den Postgres-Tabellen checkpoints / checkpoint_writes / checkpoint_blobs. InMemory speichert Blobs separat, damit die Serialisierung eines Channel-Werts nur einmal beim Versionswechsel stattfindet und nicht bei jedem put des gesamten Checkpoints für alle Kanäle erneut erfolgt.
PersistentDict für optionales Persistieren: Das memory-Modul enthält außerdem ein PersistentDict(defaultdict) (PersistentDict:634), das das dict per pickle in eine Datei dump-t und beim Neustart wieder lädt. InMemorySaver.__init__ nimmt einen factory-Parameter entgegen (__init__:85-99), damit bei factory=PersistentDict automatisch ein Context-Manager angebunden wird – ein Kompromisspfad für den Fall «nur fürs Debugging, aber Zustand über Neustarts hinweg erhalten».
SQLite mit WAL + einem Lock: setup() setzt als Erstes PRAGMA journal_mode=WAL (WAL:141), sodass Lesezugriffe Schreibzugriffe nicht blockieren; Schreiben selbst wird aber über self.lock serialisiert, da sqlite3.Connection nicht threadsicher ist. Der Docstring sagt ausdrücklich «skaliert nicht auf mehrere Threads» (note:48-54); für Nebenläufigkeit muss auf Postgres gewechselt werden.
Schreib-Tabelle unterscheidet REPLACE und IGNORE: put_writes wählt in SQLite je nachdem, ob alle Writes in WRITES_IDX_MAP enthalten sind, zwischen INSERT OR REPLACE und INSERT OR IGNORE (put_writes query:462-466): Kontroll-Channel-Schreibungen (ERROR / INTERRUPT / RESUME) dürfen überschreiben, normale Business-Writes werden bei Konflikt ignoriert – die Semantik entspricht dem if inner_key[1] >= 0 ... continue in InMemory (InMemorySaver.put_writes:499-509).
InMemory verwendet max(checkpoints.keys()) als latest: Es gibt kein explizites ORDER BY; stattdessen wird auf die monoton steigende checkpoint_id-Zeichenkette vertraut (latest:282-283), daher muss checkpoint_id lexikographisch vergleichbar sein (UUID6/UUID7), nicht ein zufälliges UUID.

Schlüsseldateien

InMemorySaver Klassendefinition:33-83 — Klassendeklaration + die drei defaultdict-Felder storage / writes / blobs.
InMemorySaver.put:427-471 — Zerlegt channel_values in blobs, Checkpoint-Hauptkörper geht in storage, gibt neues Config zurück.
InMemorySaver.put_writes:473-509 — Überspringt bereits vorhandene negative Kontroll-Writes, normale Writes werden über (task_id, idx) indiziert.
InMemorySaver.get_tuple:236-316 — Entweder gezielte Abfrage über checkpoint_id oder max(keys()) für den neuesten, dann Zusammenbau des CheckpointTuple.
MemorySaver alias:631 — Historischer Alias, äquivalent zu InMemorySaver.
SqliteSaver Klasse:45-95 — Nimmt sqlite3.Connection, mit threading.Lock für serielle Schreibzugriffe.
SqliteSaver.setup:129-166 — Legt Tabellen an und aktiviert WAL.
SqliteSaver.get_tuple:191-263 — Zwei vorbereitete SELECTs: gezielte Abfrage über checkpoint_id oder ORDER BY checkpoint_id DESC LIMIT 1 für den neuesten, dann JOIN über writes.
SqliteSaver.put:387-443 — INSERT OR REPLACE INTO checkpoints; metadata wird als JSON zu bytes serialisiert.
AsyncSqliteSaver:38 — Asynchrone Variante, aput / aget_tuple / alist sind mit aiosqlite neu implementiert.

Datenfluss

InMemorySaver.put erledigt «channel_values in blobs zerlegen» und «Hauptkörper in storage» in einem Durchgang; hier lässt sich sauber erkennen, wie ein Saver ein Checkpoint-TypedDict in eine dreischichtige Struktur zerlegt:

python

c = checkpoint.copy()
thread_id = config["configurable"]["thread_id"]
checkpoint_ns = config["configurable"]["checkpoint_ns"]
values: dict[str, Any] = c.pop("channel_values")  # type: ignore[misc]
for k, v in new_versions.items():
    self.blobs[(thread_id, checkpoint_ns, k, v)] = (
        self.serde.dumps_typed(values[k]) if k in values else ("empty", b"")
    )
self.storage[thread_id][checkpoint_ns].update(
    {
        checkpoint["id"]: (
            self.serde.dumps_typed(c),
            self.serde.dumps_typed(get_checkpoint_metadata(config, metadata)),
            config["configurable"].get("checkpoint_id"),  # parent
        )
    }
)

(put body:448-464) Beachten Sie: blobs verwendet (thread_id, checkpoint_ns, channel, version) als Schlüssel, während storage (thread_id, checkpoint_ns, checkpoint_id) verwendet – die beiden Indizes überschneiden sich nicht; Channel-Wert und Checkpoint-Hauptkörper sind entkoppelt, und Kanäle mit unveränderter Version müssen nicht neu serialisiert werden. Beim Zurücklesen ruft get_tuple _load_blobs auf, um für jedes (channel, version)-Paar aus channel_versions den zugehörigen Blob zu deserialisieren und in channel_values einzusetzen (_load_blobs:259-265), und verpackt alles im CheckpointTuple.

SqliteSaver folgt derselben Logik, nur dass dicts durch SQL ersetzt werden und der Blob-Bucket durch die Spalte checkpoint BLOB in der Haupttabelle plus die Spalte value BLOB in der writes-Tabelle repräsentiert wird – es gibt keine separate Blob-Tabelle; alle Channel-Werte werden direkt per serde in das checkpoint-Blob-Feld serialisiert:

python

cur.execute(
    "INSERT OR REPLACE INTO checkpoints (thread_id, checkpoint_ns, checkpoint_id, parent_checkpoint_id, type, checkpoint, metadata) VALUES (?, ?, ?, ?, ?, ?, ?)",
    (
        str(config["configurable"]["thread_id"]),
        checkpoint_ns,
        checkpoint["id"],
        config["configurable"].get("checkpoint_id"),
        type_,
        serialized_checkpoint,
        serialized_metadata,
    ),
)

(SqliteSaver.put INSERT:425-436)

Grenzen und Fehler

InMemorySaver verliert alles beim Absturz des Prozesses: Der Docstring sagt ausdrücklich «nur InMemorySaver für Debugging oder Tests verwenden» (warning:40-44); für Persistenz muss auf Sqlite oder Postgres gewechselt werden.
InMemorySaver holt sich «latest» über die lexikographische Reihenfolge der checkpoint_id: max(checkpoints.keys()) (max:282-283) erwartet monoton steigende IDs; wenn ein eigener Saver nicht-monotone IDs verwendet, liefert get_tuple ohne checkpoint_id die falsche Zeile.
SqliteSaver: eine Verbindung, ein Lock: Der Context-Manager cursor() holt sich self.lock (cursor:181-189); nebenläufige Schreibzugriffe werden serialisiert, lange Transaktionen blockieren andere Lesezugriffe. Für Nebenläufigkeit bleibt nur der Pipeline-Modus von Postgres.
SqliteSaver hartcodiert ?-Platzhalter und Spaltenreihenfolge: Die Spaltenreihenfolge muss mit der Liste INSERT INTO checkpoints (...) übereinstimmen (INSERT columns:426); bei Migrationen mit neuen Spalten müssen beide Stellen synchron geändert werden. Die task_path-Spalte wurde nachträglich hinzugefügt, siehe das Ende von Postgres MIGRATIONS (task_path migration:90).
AsyncSqliteSaver ist nicht einfach to_thread gewrappt: AsyncSqliteSaver:38 ist eine eigenständige Klasse; aput / aget_tuple / alist sind mit aiosqlite vollständig neu geschrieben (aput:509), nicht das asyncio.to_thread-Forwarding der Basisklasse – der async-Pfad kann tatsächlich parallel laufen und wird nicht durch das globale Lock serialisiert.
Bei factory= mit PersistentDict muss beim Beenden close() aufgerufen werden: PersistentDict.sync() schreibt den pickle-Dump in die Datei (sync:657); ohne __exit__ wird nicht automatisch persistiert. with InMemorySaver(factory=PersistentDict, ...) ist die gebräuchliche Idiomatik.

Zusammenfassung

InMemorySaver ist die Referenzimplementierung mit «drei Dict-Buckets», SqliteSaver entfaltet dieselben drei Buckets in zwei SQLite-Tabellen mit WAL; zusammen decken sie Debugging und Ein-Maschine-Persistenz ab. Primärschlüssel-Design, Blob-Aufteilung, negative Indizes für Kontroll-Channels – all diese Konventionen haben in diesen beiden Implementierungen ihren Ursprung. Für produktive Skalierung siehe PostgresSaver, für die Schnittstellenform BaseCheckpointSaver.

Siehe offizielle Dokumentation: LangGraph 文档 · README.

InMemorySaver + SqliteSaver: In-Process- und Datei-Level-Checkpoints ​

Verantwortung ​

Entwurfsmotivation ​

Schlüsseldateien ​

Datenfluss ​

Grenzen und Fehler ​

Zusammenfassung ​