BaseCheckpointSaver

​

Overview

• Persist state within and across interactions
• Resume from interrupts
• Enable time-travel debugging
• Support conversational memory

​

Class Definition

from langgraph.checkpoint.base import BaseCheckpointSaver

class BaseCheckpointSaver(Generic[V]):
    """Base class for creating a graph checkpointer."""
    
    serde: SerializerProtocol = JsonPlusSerializer()

​

Constructor

def __init__(
    self,
    *,
    serde: SerializerProtocol | None = None,
) -> None

​

Parameters

• serde (SerializerProtocol, optional): Serializer for encoding/decoding checkpoints. Defaults to JsonPlusSerializer().

​

Key Concepts

​

Thread ID

config = {"configurable": {"thread_id": "my-thread"}}
graph.invoke(inputs, config)

​

Usage Patterns

• Single-shot workflows: Use a unique ID (e.g., uuid4) for each independent run
• Conversational memory: Reuse the same thread_id to accumulate state across invocations

​

Core Methods

​

Synchronous Methods

​

get

def get(self, config: RunnableConfig) -> Checkpoint | None

• config (RunnableConfig): Configuration specifying which checkpoint to retrieve

• Checkpoint | None: The requested checkpoint, or None if not found

​

get_tuple

def get_tuple(self, config: RunnableConfig) -> CheckpointTuple | None

• config (RunnableConfig): Configuration specifying which checkpoint to retrieve

• CheckpointTuple | None: The requested checkpoint tuple, or None if not found

​

list

def list(
    self,
    config: RunnableConfig | None,
    *,
    filter: dict[str, Any] | None = None,
    before: RunnableConfig | None = None,
    limit: int | None = None,
) -> Iterator[CheckpointTuple]

• config (RunnableConfig | None): Base configuration for filtering checkpoints
• filter (dict[str, Any] | None): Additional filtering criteria for metadata
• before (RunnableConfig | None): List checkpoints created before this configuration
• limit (int | None): Maximum number of checkpoints to return

• Iterator[CheckpointTuple]: Iterator of matching checkpoint tuples

​

put

def put(
    self,
    config: RunnableConfig,
    checkpoint: Checkpoint,
    metadata: CheckpointMetadata,
    new_versions: ChannelVersions,
) -> RunnableConfig

• config (RunnableConfig): Configuration for the checkpoint
• checkpoint (Checkpoint): The checkpoint to store
• metadata (CheckpointMetadata): Additional metadata for the checkpoint
• new_versions (ChannelVersions): New channel versions as of this write

• RunnableConfig: Updated configuration after storing the checkpoint

​

put_writes

def put_writes(
    self,
    config: RunnableConfig,
    writes: Sequence[tuple[str, Any]],
    task_id: str,
    task_path: str = "",
) -> None

• config (RunnableConfig): Configuration of the related checkpoint
• writes (Sequence[tuple[str, Any]]): List of writes to store
• task_id (str): Identifier for the task creating the writes
• task_path (str): Path of the task creating the writes (default: "")

​

delete_thread

def delete_thread(self, thread_id: str) -> None

• thread_id (str): The thread ID whose checkpoints should be deleted

​

delete_for_runs

def delete_for_runs(self, run_ids: Sequence[str]) -> None

• run_ids (Sequence[str]): The run IDs whose checkpoints should be deleted

​

copy_thread

def copy_thread(
    self,
    source_thread_id: str,
    target_thread_id: str,
) -> None

• source_thread_id (str): The thread ID to copy from
• target_thread_id (str): The thread ID to copy to

​

prune

def prune(
    self,
    thread_ids: Sequence[str],
    *,
    strategy: str = "keep_latest",
) -> None

• thread_ids (Sequence[str]): The thread IDs to prune
• strategy (str): The pruning strategy. Options:
  • "keep_latest": Retains only the most recent checkpoint per namespace
  • "delete": Removes all checkpoints

​

Asynchronous Methods

• aget(config) - Async version of get()
• aget_tuple(config) - Async version of get_tuple()
• alist(config, *, filter, before, limit) - Async version of list()
• aput(config, checkpoint, metadata, new_versions) - Async version of put()
• aput_writes(config, writes, task_id, task_path) - Async version of put_writes()
• adelete_thread(thread_id) - Async version of delete_thread()
• adelete_for_runs(run_ids) - Async version of delete_for_runs()
• acopy_thread(source_thread_id, target_thread_id) - Async version of copy_thread()
• aprune(thread_ids, *, strategy) - Async version of prune()

​

Utility Methods

​

get_next_version

def get_next_version(self, current: V | None, channel: None) -> V

• current (V | None): The current version identifier (int, float, or str)
• channel (None): Deprecated argument, kept for backwards compatibility

• V: The next version identifier, which must be monotonically increasing

​

Data Types

​

Checkpoint

class Checkpoint(TypedDict):
    """State snapshot at a given point in time."""
    
    v: int  # Checkpoint format version (currently 1)
    id: str  # Unique, monotonically increasing ID
    ts: str  # Timestamp in ISO 8601 format
    channel_values: dict[str, Any]  # Channel snapshots
    channel_versions: ChannelVersions  # Channel version strings
    versions_seen: dict[str, ChannelVersions]  # Node-channel version tracking
    updated_channels: list[str] | None  # Channels updated in this checkpoint

​

CheckpointMetadata

class CheckpointMetadata(TypedDict, total=False):
    """Metadata associated with a checkpoint."""
    
    source: Literal["input", "loop", "update", "fork"]  # Checkpoint source
    step: int  # Step number (-1 for first input, 0+ for loop steps)
    parents: dict[str, str]  # Parent checkpoint IDs by namespace
    run_id: str  # ID of the run that created this checkpoint

​

CheckpointTuple

class CheckpointTuple(NamedTuple):
    """A tuple containing a checkpoint and its associated data."""
    
    config: RunnableConfig
    checkpoint: Checkpoint
    metadata: CheckpointMetadata
    parent_config: RunnableConfig | None = None
    pending_writes: list[PendingWrite] | None = None

​

Implementation Notes

1. Inherit from BaseCheckpointSaver: Extend this base class
2. Implement required methods: At minimum, implement:
   • get_tuple()
   • list()
   • put()
   • put_writes()
3. Consider async support: Implement async versions to avoid blocking
4. Handle serialization: Use the provided serde for encoding/decoding
5. Thread safety: Ensure your implementation is thread-safe if needed

​

See Also

• SqliteSaver - SQLite implementation
• PostgresSaver - Postgres implementation
• MemorySaver - In-memory implementation