BaseCheckpointSaver: shape of the persistence interface

源码版本1.2.9

Responsibilities

In the LangGraph runtime, every time the Pregel engine finishes a superstep it must persist the current values of all channels along with the writes produced by that step. That way, the next time the same thread_id comes in, the run can continue from the previous state; it also enables resuming after an interrupt, replaying history, and time-travel debugging. BaseCheckpointSaver is the abstract interface for this "persistence" layer (BaseCheckpointSaver:176). It does not care where you store the data — an in-memory dict, SQLite, Postgres, Redis all work — it only mandates a handful of methods the saver must implement: put / get / get_tuple / list / put_writes (put / get / list / put_writes:227-318).

It also pins the shape of the "checkpoint" concept as the Checkpoint TypedDict: Checkpoint TypedDict:92-123, carrying channel_values, channel_versions, versions_seen, and id / ts / v. What put receives is not just a Checkpoint — it also takes metadata and new_versions, and the saver decides which fields go into the main table and which into a blob table. put_writes separately stores "the writes this task produced in this step", decoupled from the checkpoint, so writes can be flushed incrementally without rewriting the entire snapshot on every write.

In other words, BaseCheckpointSaver is the contract between PregelLoop and the concrete storage backend: the engine only talks to this interface, so swapping the backend implementation does not require touching the engine; a backend writing its own saver only needs to fill in the five methods, and the engine automatically inherits that backend's concurrency and persistence characteristics.

Design motivation

• State and writes stored separately: put stores the full snapshot; put_writes stores the incremental writes a task produces within a step (put_writes:300-318). The split lets resume-after-interrupt reconstruct precisely "which task has already written in this step, which has not", rather than only being able to roll back to the previous checkpoint wholesale.
• thread_id as primary key: the config returned by put carries only thread_id / checkpoint_ns / checkpoint_id (put:277-298); these three fields locate a checkpoint across calls and across processes. checkpoint_ns is the namespace used by subgraphs; an empty string means the top-level graph.
• Sync + async dual interfaces: every write method has an async counterpart aput / aget / alist / aput_writes (async methods:417-509) whose default implementation raises NotImplementedError. A saver subclass can implement only the sync side and let the async side forward through asyncio.to_thread, or write a true async path using a native driver.
• Metadata carries a source label: the source field in CheckpointMetadata takes input / loop / update / fork (source:41-48). Both list and filtering rely on metadata, so the saver must persist metadata alongside the checkpoint — it cannot drop it.
• DeltaChannel bypass: the docstrings of prune / delete_for_runs / copy_thread repeatedly warn (DeltaChannel-aware ops:320-415) that if the graph uses DeltaChannel, a custom saver cannot just delete the latest checkpoint — it must also preserve the checkpoint_writes along the ancestor chain, otherwise delta reconstruction will silently return empty.

Key files

• BaseCheckpointSaver class:176 — the abstract base class itself, defining the put/get/list/put_writes interface.
• Checkpoint TypedDict:92-123 — the field shape of one checkpoint: channel_values/channel_versions/versions_seen/id/ts.
• CheckpointTuple:139-146 — return value of get_tuple, packaging checkpoint, metadata, parent_config, and pending_writes into a NamedTuple.
• CheckpointMetadata:38-86 — source / step / parents / run_id / and counters_since_delta_snapshot for DeltaChannel.
• get method:227-237 — get delegates to get_tuple by default, returning only the checkpoint field.
• put method:277-298 — stores a full checkpoint + metadata, returns the updated config.
• put_writes method:300-318 — stores the incremental write list for a task under a checkpoint.
• delete_thread / delete_for_runs:320-348 — thread-level cleanup, with DeltaChannel warnings.
• WRITES_IDX_MAP:795 — the four control channels ERROR / SCHEDULED / INTERRUPT / RESUME occupy negative indices to avoid primary-key collisions with real task writes.

Data flow

After each superstep, PregelLoop calls the saver to persist the current values of all channels and the new version numbers from this step. The default signature of put and the InMemorySaver implementation make this clear — the engine calls put, the saver splits channel_values into blobs keyed by (thread_id, ns, channel, version), writes the main body to the main table, and returns a new config pointing at the just-written checkpoint_id.

def put(
    self,
    config: RunnableConfig,
    checkpoint: Checkpoint,
    metadata: CheckpointMetadata,
    new_versions: ChannelVersions,
) -> RunnableConfig:
    """Store a checkpoint with its configuration and metadata."""
    raise NotImplementedError

This is the pure interface in the base class (put method:277-298), with no default implementation. Each subclass decides for itself which fields go into the main table, which go into blobs, whether to use ON CONFLICT DO UPDATE, and so on. On read-back, get_tuple is responsible for deserializing the blobs back into channel_values and bundling the pending_writes for the same checkpoint into the CheckpointTuple (get_tuple:239-251).

Boundaries and failure modes

• put_writes reserves negative indices for control channels: WRITES_IDX_MAP maps ERROR / SCHEDULED / INTERRUPT / RESUME to -1 / -2 / -3 / -4 (WRITES_IDX_MAP:795). A custom saver's primary key (task_id, idx) must accept negative values, otherwise control writes will hit primary-key conflicts.
• get defaults to calling get_tuple: a subclass only needs to implement get_tuple; the default get will fetch .checkpoint for you (get:227-237). But get_tuple itself is raise NotImplementedError if not overridden — it cannot silently return None.
• async defaults to NotImplementedError: when a sync-only saver does not implement aput, the default async path does not auto-convert to a thread — either the subclass writes its own asyncio.to_thread forwarding, or the caller invokes the sync path (async defaults:468-509).
• metadata cannot be dropped: the filter argument of list filters by metadata fields (list method:253-275). If a saver implementation skips or corrupts metadata, both filtering and pagination will silently break.
• No naive deletion under DeltaChannel: the docstring of the base prune explicitly states (prune:374-415) — keeping only the latest checkpoint will silently empty delta-channel reconstruction without raising. Custom saver implementations must walk the ancestor chain before deleting.
• checkpoint_ns defaults to empty string: subgraph invocations carry a non-empty ns; top-level invocations use "". The saver's primary key must include ns, otherwise checkpoints of different subgraphs with the same name will overwrite each other.

Summary

BaseCheckpointSaver pins "how a checkpoint is stored, read, listed, and incrementally written" into five methods + one TypedDict shape. The Pregel engine only talks to the interface, so backend implementations are freely swappable. For two reference implementations see the sibling pages InMemorySaver + SqliteSaver and PostgresSaver. For how the runtime decides the flushing cadence see Pregel engine and PregelLoop.

See official docs: LangGraph 文档 · README.