Checkpointers¶

Classes:

Name	Description
`CheckpointMetadata`	Metadata associated with a checkpoint.
`Checkpoint`	State snapshot at a given point in time.
`BaseCheckpointSaver`	Base class for creating a graph checkpointer.

Functions:

Name	Description
`create_checkpoint`	Create a checkpoint for the given channels.

CheckpointMetadata ¶

Bases: TypedDict

Metadata associated with a checkpoint.

Attributes:

Name	Type	Description
`source`	`Literal['input', 'loop', 'update', 'fork']`	The source of the checkpoint.
`step`	`int`	The step number of the checkpoint.
`parents`	`dict[str, str]`	The IDs of the parent checkpoints.

source `instance-attribute` ¶

source: Literal['input', 'loop', 'update', 'fork']

The source of the checkpoint.

"input": The checkpoint was created from an input to invoke/stream/batch.
"loop": The checkpoint was created from inside the pregel loop.
"update": The checkpoint was created from a manual state update.
"fork": The checkpoint was created as a copy of another checkpoint.

step `instance-attribute` ¶

step: int

The step number of the checkpoint.

-1 for the first "input" checkpoint. 0 for the first "loop" checkpoint. ... for the nth checkpoint afterwards.

parents `instance-attribute` ¶

parents: dict[str, str]

The IDs of the parent checkpoints.

Mapping from checkpoint namespace to checkpoint ID.

Checkpoint ¶

Bases: TypedDict

State snapshot at a given point in time.

Attributes:

Name	Type	Description
`v`	`int`	The version of the checkpoint format. Currently 1.
`id`	`str`	The ID of the checkpoint. This is both unique and monotonically
`ts`	`str`	The timestamp of the checkpoint in ISO 8601 format.
`channel_values`	`dict[str, Any]`	The values of the channels at the time of the checkpoint.
`channel_versions`	`ChannelVersions`	The versions of the channels at the time of the checkpoint.
`versions_seen`	`dict[str, ChannelVersions]`	Map from node ID to map from channel name to version seen.
`updated_channels`	`list[str] \| None`	The channels that were updated in this checkpoint.

v `instance-attribute` ¶

v: int

The version of the checkpoint format. Currently 1.

id `instance-attribute` ¶

id: str

The ID of the checkpoint. This is both unique and monotonically increasing, so can be used for sorting checkpoints from first to last.

ts `instance-attribute` ¶

ts: str

The timestamp of the checkpoint in ISO 8601 format.

channel_values `instance-attribute` ¶

channel_values: dict[str, Any]

The values of the channels at the time of the checkpoint. Mapping from channel name to deserialized channel snapshot value.

channel_versions `instance-attribute` ¶

channel_versions: ChannelVersions

The versions of the channels at the time of the checkpoint. The keys are channel names and the values are monotonically increasing version strings for each channel.

versions_seen `instance-attribute` ¶

versions_seen: dict[str, ChannelVersions]

Map from node ID to map from channel name to version seen. This keeps track of the versions of the channels that each node has seen. Used to determine which nodes to execute next.

updated_channels `instance-attribute` ¶

updated_channels: list[str] | None

The channels that were updated in this checkpoint.

BaseCheckpointSaver ¶

Bases: Generic[V]

Base class for creating a graph checkpointer.

Checkpointers allow LangGraph agents to persist their state within and across multiple interactions.

Attributes:

Name	Type	Description
`serde`	`SerializerProtocol`	Serializer for encoding/decoding checkpoints.

Note

When creating a custom checkpoint saver, consider implementing async versions to avoid blocking the main thread.

Methods:

Name	Description
`get`	Fetch a checkpoint using the given configuration.
`get_tuple`	Fetch a checkpoint tuple using the given configuration.
`list`	List checkpoints that match the given criteria.
`put`	Store a checkpoint with its configuration and metadata.
`put_writes`	Store intermediate writes linked to a checkpoint.
`delete_thread`	Delete all checkpoints and writes associated with a specific thread ID.
`aget`	Asynchronously fetch a checkpoint using the given configuration.
`aget_tuple`	Asynchronously fetch a checkpoint tuple using the given configuration.
`alist`	Asynchronously list checkpoints that match the given criteria.
`aput`	Asynchronously store a checkpoint with its configuration and metadata.
`aput_writes`	Asynchronously store intermediate writes linked to a checkpoint.
`adelete_thread`	Delete all checkpoints and writes associated with a specific thread ID.
`get_next_version`	Generate the next version ID for a channel.

config_specs `property` ¶

config_specs: list

Define the configuration options for the checkpoint saver.

Returns:

Name	Type	Description
`list`	`list`	List of configuration field specs.

get ¶

get(config: RunnableConfig) -> Checkpoint | None

Fetch a checkpoint using the given configuration.

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig`	Configuration specifying which checkpoint to retrieve.	required

Returns:

Type	Description
`Checkpoint \| None`	Optional[Checkpoint]: The requested checkpoint, or None if not found.

get_tuple ¶

get_tuple(config: RunnableConfig) -> CheckpointTuple | None

Fetch a checkpoint tuple using the given configuration.

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig`	Configuration specifying which checkpoint to retrieve.	required

Returns:

Type	Description
`CheckpointTuple \| None`	Optional[CheckpointTuple]: The requested checkpoint tuple, or None if not found.

Raises:

Type	Description
`NotImplementedError`	Implement this method in your custom checkpoint saver.

list ¶

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

List checkpoints that match the given criteria.

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig \| None`	Base configuration for filtering checkpoints.	required
`filter`	`dict[str, Any] \| None`	Additional filtering criteria.	`None`
`before`	`RunnableConfig \| None`	List checkpoints created before this configuration.	`None`
`limit`	`int \| None`	Maximum number of checkpoints to return.	`None`

Returns:

Type	Description
`Iterator[CheckpointTuple]`	Iterator[CheckpointTuple]: Iterator of matching checkpoint tuples.

Raises:

Type	Description
`NotImplementedError`	Implement this method in your custom checkpoint saver.

put ¶

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

Store a checkpoint with its configuration and metadata.

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig`	Configuration for the checkpoint.	required
`checkpoint`	`Checkpoint`	The checkpoint to store.	required
`metadata`	`CheckpointMetadata`	Additional metadata for the checkpoint.	required
`new_versions`	`ChannelVersions`	New channel versions as of this write.	required

Returns:

Name	Type	Description
`RunnableConfig`	`RunnableConfig`	Updated configuration after storing the checkpoint.

Raises:

Type	Description
`NotImplementedError`	Implement this method in your custom checkpoint saver.

put_writes ¶

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

Store intermediate writes linked to a checkpoint.

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig`	Configuration of the related checkpoint.	required
`writes`	`Sequence[tuple[str, Any]]`	List of writes to store.	required
`task_id`	`str`	Identifier for the task creating the writes.	required
`task_path`	`str`	Path of the task creating the writes.	`''`

Raises:

Type	Description
`NotImplementedError`	Implement this method in your custom checkpoint saver.

delete_thread ¶

delete_thread(thread_id: str) -> None

Delete all checkpoints and writes associated with a specific thread ID.

Parameters:

Name	Type	Description	Default
`thread_id`	`str`	The thread ID whose checkpoints should be deleted.	required

aget `async` ¶

aget(config: RunnableConfig) -> Checkpoint | None

Asynchronously fetch a checkpoint using the given configuration.

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig`	Configuration specifying which checkpoint to retrieve.	required

Returns:

Type	Description
`Checkpoint \| None`	Optional[Checkpoint]: The requested checkpoint, or None if not found.

aget_tuple `async` ¶

aget_tuple(
    config: RunnableConfig,
) -> CheckpointTuple | None

Asynchronously fetch a checkpoint tuple using the given configuration.

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig`	Configuration specifying which checkpoint to retrieve.	required

Returns:

Type	Description
`CheckpointTuple \| None`	Optional[CheckpointTuple]: The requested checkpoint tuple, or None if not found.

Raises:

Type	Description
`NotImplementedError`	Implement this method in your custom checkpoint saver.

alist `async` ¶

alist(
    config: RunnableConfig | None,
    *,
    filter: dict[str, Any] | None = None,
    before: RunnableConfig | None = None,
    limit: int | None = None
) -> AsyncIterator[CheckpointTuple]

Asynchronously list checkpoints that match the given criteria.

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig \| None`	Base configuration for filtering checkpoints.	required
`filter`	`dict[str, Any] \| None`	Additional filtering criteria for metadata.	`None`
`before`	`RunnableConfig \| None`	List checkpoints created before this configuration.	`None`
`limit`	`int \| None`	Maximum number of checkpoints to return.	`None`

Returns:

Type	Description
`AsyncIterator[CheckpointTuple]`	AsyncIterator[CheckpointTuple]: Async iterator of matching checkpoint tuples.

Raises:

Type	Description
`NotImplementedError`	Implement this method in your custom checkpoint saver.

aput `async` ¶

aput(
    config: RunnableConfig,
    checkpoint: Checkpoint,
    metadata: CheckpointMetadata,
    new_versions: ChannelVersions,
) -> RunnableConfig

Asynchronously store a checkpoint with its configuration and metadata.

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig`	Configuration for the checkpoint.	required
`checkpoint`	`Checkpoint`	The checkpoint to store.	required
`metadata`	`CheckpointMetadata`	Additional metadata for the checkpoint.	required
`new_versions`	`ChannelVersions`	New channel versions as of this write.	required

Returns:

Name	Type	Description
`RunnableConfig`	`RunnableConfig`	Updated configuration after storing the checkpoint.

Raises:

Type	Description
`NotImplementedError`	Implement this method in your custom checkpoint saver.

aput_writes `async` ¶

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

Asynchronously store intermediate writes linked to a checkpoint.

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig`	Configuration of the related checkpoint.	required
`writes`	`Sequence[tuple[str, Any]]`	List of writes to store.	required
`task_id`	`str`	Identifier for the task creating the writes.	required
`task_path`	`str`	Path of the task creating the writes.	`''`

Raises:

Type	Description
`NotImplementedError`	Implement this method in your custom checkpoint saver.

adelete_thread `async` ¶

adelete_thread(thread_id: str) -> None

Delete all checkpoints and writes associated with a specific thread ID.

Parameters:

Name	Type	Description	Default
`thread_id`	`str`	The thread ID whose checkpoints should be deleted.	required

get_next_version ¶

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

Generate the next version ID for a channel.

Default is to use integer versions, incrementing by 1. If you override, you can use str/int/float versions, as long as they are monotonically increasing.

Parameters:

Name	Type	Description	Default
`current`	`V \| None`	The current version identifier (int, float, or str).	required
`channel`	`None`	Deprecated argument, kept for backwards compatibility.	required

Returns:

Name	Type	Description
`V`	`V`	The next version identifier, which must be increasing.

create_checkpoint ¶

create_checkpoint(
    checkpoint: Checkpoint,
    channels: Mapping[str, ChannelProtocol] | None,
    step: int,
    *,
    id: str | None = None
) -> Checkpoint

Create a checkpoint for the given channels.

Classes:

Name	Description
`SerializerProtocol`	Protocol for serialization and deserialization of objects.
`CipherProtocol`	Protocol for encryption and decryption of data.

SerializerProtocol ¶

Bases: UntypedSerializerProtocol, Protocol

Protocol for serialization and deserialization of objects.

dumps: Serialize an object to bytes.
dumps_typed: Serialize an object to a tuple (type, bytes).
loads: Deserialize an object from bytes.
loads_typed: Deserialize an object from a tuple (type, bytes).

Valid implementations include the pickle, json and orjson modules.

CipherProtocol ¶

Bases: Protocol

Protocol for encryption and decryption of data. - encrypt: Encrypt plaintext. - decrypt: Decrypt ciphertext.

Methods:

Name	Description
`encrypt`	Encrypt plaintext. Returns a tuple (cipher name, ciphertext).
`decrypt`	Decrypt ciphertext. Returns the plaintext.

encrypt ¶

encrypt(plaintext: bytes) -> tuple[str, bytes]

Encrypt plaintext. Returns a tuple (cipher name, ciphertext).

decrypt ¶

decrypt(ciphername: str, ciphertext: bytes) -> bytes

Decrypt ciphertext. Returns the plaintext.

Classes:

Name	Description
`JsonPlusSerializer`	Serializer that uses ormsgpack, with a fallback to extended JSON serializer.

JsonPlusSerializer ¶

Bases: SerializerProtocol

Serializer that uses ormsgpack, with a fallback to extended JSON serializer.

Classes:

Name	Description
`EncryptedSerializer`	Serializer that encrypts and decrypts data using an encryption protocol.

EncryptedSerializer ¶

Bases: SerializerProtocol

Serializer that encrypts and decrypts data using an encryption protocol.

Methods:

Name	Description
`dumps_typed`	Serialize an object to a tuple (type, bytes) and encrypt the bytes.
`from_pycryptodome_aes`	Create an EncryptedSerializer using AES encryption.

dumps_typed ¶

dumps_typed(obj: Any) -> tuple[str, bytes]

Serialize an object to a tuple (type, bytes) and encrypt the bytes.

from_pycryptodome_aes `classmethod` ¶

from_pycryptodome_aes(
    serde: SerializerProtocol = JsonPlusSerializer(),
    **kwargs: Any
) -> EncryptedSerializer

Create an EncryptedSerializer using AES encryption.

Classes:

Name	Description
`InMemorySaver`	An in-memory checkpoint saver.
`PersistentDict`	Persistent dictionary with an API compatible with shelve and anydbm.

InMemorySaver ¶

Bases: BaseCheckpointSaver[str], AbstractContextManager, AbstractAsyncContextManager

An in-memory checkpoint saver.

This checkpoint saver stores checkpoints in memory using a defaultdict.

Note

Only use InMemorySaver for debugging or testing purposes. For production use cases we recommend installing langgraph-checkpoint-postgres and using PostgresSaver / AsyncPostgresSaver.

If you are using the LangGraph Platform, no checkpointer needs to be specified. The correct managed checkpointer will be used automatically.

Parameters:

Name	Type	Description	Default
`serde`	`SerializerProtocol \| None`	The serializer to use for serializing and deserializing checkpoints. Defaults to None.	`None`

Examples:

    import asyncio

    from langgraph.checkpoint.memory import InMemorySaver
    from langgraph.graph import StateGraph

    builder = StateGraph(int)
    builder.add_node("add_one", lambda x: x + 1)
    builder.set_entry_point("add_one")
    builder.set_finish_point("add_one")

    memory = InMemorySaver()
    graph = builder.compile(checkpointer=memory)
    coro = graph.ainvoke(1, {"configurable": {"thread_id": "thread-1"}})
    asyncio.run(coro)  # Output: 2

Methods:

Name	Description
`get_tuple`	Get a checkpoint tuple from the in-memory storage.
`list`	List checkpoints from the in-memory storage.
`put`	Save a checkpoint to the in-memory storage.
`put_writes`	Save a list of writes to the in-memory storage.
`delete_thread`	Delete all checkpoints and writes associated with a thread ID.
`aget_tuple`	Asynchronous version of get_tuple.
`alist`	Asynchronous version of list.
`aput`	Asynchronous version of put.
`aput_writes`	Asynchronous version of put_writes.
`adelete_thread`	Delete all checkpoints and writes associated with a thread ID.
`get`	Fetch a checkpoint using the given configuration.
`aget`	Asynchronously fetch a checkpoint using the given configuration.

Attributes:

Name	Type	Description
`config_specs`	`list`	Define the configuration options for the checkpoint saver.

config_specs `property` ¶

config_specs: list

Define the configuration options for the checkpoint saver.

Returns:

Name	Type	Description
`list`	`list`	List of configuration field specs.

get_tuple ¶

get_tuple(config: RunnableConfig) -> CheckpointTuple | None

Get a checkpoint tuple from the in-memory storage.

This method retrieves a checkpoint tuple from the in-memory storage based on the provided config. If the config contains a "checkpoint_id" key, the checkpoint with the matching thread ID and timestamp is retrieved. Otherwise, the latest checkpoint for the given thread ID is retrieved.

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig`	The config to use for retrieving the checkpoint.	required

Returns:

Type	Description
`CheckpointTuple \| None`	Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.

list ¶

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

List checkpoints from the in-memory storage.

This method retrieves a list of checkpoint tuples from the in-memory storage based on the provided criteria.

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig \| None`	Base configuration for filtering checkpoints.	required
`filter`	`dict[str, Any] \| None`	Additional filtering criteria for metadata.	`None`
`before`	`RunnableConfig \| None`	List checkpoints created before this configuration.	`None`
`limit`	`int \| None`	Maximum number of checkpoints to return.	`None`

Yields:

Type	Description
`CheckpointTuple`	Iterator[CheckpointTuple]: An iterator of matching checkpoint tuples.

put ¶

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

Save a checkpoint to the in-memory storage.

This method saves a checkpoint to the in-memory storage. The checkpoint is associated with the provided config.

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig`	The config to associate with the checkpoint.	required
`checkpoint`	`Checkpoint`	The checkpoint to save.	required
`metadata`	`CheckpointMetadata`	Additional metadata to save with the checkpoint.	required
`new_versions`	`ChannelVersions`	New versions as of this write	required

Returns:

Name	Type	Description
`RunnableConfig`	`RunnableConfig`	The updated config containing the saved checkpoint's timestamp.

put_writes ¶

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

Save a list of writes to the in-memory storage.

This method saves a list of writes to the in-memory storage. The writes are associated with the provided config.

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig`	The config to associate with the writes.	required
`writes`	`Sequence[tuple[str, Any]]`	The writes to save.	required
`task_id`	`str`	Identifier for the task creating the writes.	required
`task_path`	`str`	Path of the task creating the writes.	`''`

Returns:

Name	Type	Description
`RunnableConfig`	`None`	The updated config containing the saved writes' timestamp.

delete_thread ¶

delete_thread(thread_id: str) -> None

Delete all checkpoints and writes associated with a thread ID.

Parameters:

Name	Type	Description	Default
`thread_id`	`str`	The thread ID to delete.	required

Returns:

Type	Description
`None`	None

aget_tuple `async` ¶

aget_tuple(
    config: RunnableConfig,
) -> CheckpointTuple | None

Asynchronous version of get_tuple.

This method is an asynchronous wrapper around get_tuple that runs the synchronous method in a separate thread using asyncio.

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig`	The config to use for retrieving the checkpoint.	required

Returns:

Type	Description
`CheckpointTuple \| None`	Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.

alist `async` ¶

alist(
    config: RunnableConfig | None,
    *,
    filter: dict[str, Any] | None = None,
    before: RunnableConfig | None = None,
    limit: int | None = None
) -> AsyncIterator[CheckpointTuple]

Asynchronous version of list.

This method is an asynchronous wrapper around list that runs the synchronous method in a separate thread using asyncio.

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig \| None`	The config to use for listing the checkpoints.	required

Yields:

Type	Description
`AsyncIterator[CheckpointTuple]`	AsyncIterator[CheckpointTuple]: An asynchronous iterator of checkpoint tuples.

aput `async` ¶

aput(
    config: RunnableConfig,
    checkpoint: Checkpoint,
    metadata: CheckpointMetadata,
    new_versions: ChannelVersions,
) -> RunnableConfig

Asynchronous version of put.

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig`	The config to associate with the checkpoint.	required
`checkpoint`	`Checkpoint`	The checkpoint to save.	required
`metadata`	`CheckpointMetadata`	Additional metadata to save with the checkpoint.	required
`new_versions`	`ChannelVersions`	New versions as of this write	required

Returns:

Name	Type	Description
`RunnableConfig`	`RunnableConfig`	The updated config containing the saved checkpoint's timestamp.

aput_writes `async` ¶

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

Asynchronous version of put_writes.

This method is an asynchronous wrapper around put_writes that runs the synchronous method in a separate thread using asyncio.

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig`	The config to associate with the writes.	required
`writes`	`Sequence[tuple[str, Any]]`	The writes to save, each as a (channel, value) pair.	required
`task_id`	`str`	Identifier for the task creating the writes.	required
`task_path`	`str`	Path of the task creating the writes.	`''`

Returns:

Type	Description
`None`	None

adelete_thread `async` ¶

adelete_thread(thread_id: str) -> None

Delete all checkpoints and writes associated with a thread ID.

Parameters:

Name	Type	Description	Default
`thread_id`	`str`	The thread ID to delete.	required

Returns:

Type	Description
`None`	None

get ¶

get(config: RunnableConfig) -> Checkpoint | None

Fetch a checkpoint using the given configuration.

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig`	Configuration specifying which checkpoint to retrieve.	required

Returns:

Type	Description
`Checkpoint \| None`	Optional[Checkpoint]: The requested checkpoint, or None if not found.

aget `async` ¶

aget(config: RunnableConfig) -> Checkpoint | None

Asynchronously fetch a checkpoint using the given configuration.

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig`	Configuration specifying which checkpoint to retrieve.	required

Returns:

Type	Description
`Checkpoint \| None`	Optional[Checkpoint]: The requested checkpoint, or None if not found.

PersistentDict ¶

Bases: defaultdict

Persistent dictionary with an API compatible with shelve and anydbm.

The dict is kept in memory, so the dictionary operations run as fast as a regular dictionary.

Write to disk is delayed until close or sync (similar to gdbm's fast mode).

Input file format is automatically discovered. Output file format is selectable between pickle, json, and csv. All three serialization formats are backed by fast C implementations.

Adapted from https://code.activestate.com/recipes/576642-persistent-dict-with-multiple-standard-file-format/

Methods:

Name	Description
`sync`	Write dict to disk

sync ¶

sync() -> None

Write dict to disk

Modules:

Name	Description
`aio`
`utils`

Classes:

Name	Description
`SqliteSaver`	A checkpoint saver that stores checkpoints in a SQLite database.

SqliteSaver ¶

Bases: BaseCheckpointSaver[str]

A checkpoint saver that stores checkpoints in a SQLite database.

Note

This class is meant for lightweight, synchronous use cases (demos and small projects) and does not scale to multiple threads. For a similar sqlite saver with async support, consider using AsyncSqliteSaver.

Parameters:

Name	Type	Description	Default
`conn`	`Connection`	The SQLite database connection.	required
`serde`	`Optional[SerializerProtocol]`	The serializer to use for serializing and deserializing checkpoints. Defaults to JsonPlusSerializerCompat.	`None`

Examples:

>>> import sqlite3
>>> from langgraph.checkpoint.sqlite import SqliteSaver
>>> from langgraph.graph import StateGraph
>>>
>>> builder = StateGraph(int)
>>> builder.add_node("add_one", lambda x: x + 1)
>>> builder.set_entry_point("add_one")
>>> builder.set_finish_point("add_one")
>>> # Create a new SqliteSaver instance
>>> # Note: check_same_thread=False is OK as the implementation uses a lock
>>> # to ensure thread safety.
>>> conn = sqlite3.connect("checkpoints.sqlite", check_same_thread=False)
>>> memory = SqliteSaver(conn)
>>> graph = builder.compile(checkpointer=memory)
>>> config = {"configurable": {"thread_id": "1"}}
>>> graph.get_state(config)
>>> result = graph.invoke(3, config)
>>> graph.get_state(config)
StateSnapshot(values=4, next=(), config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '0c62ca34-ac19-445d-bbb0-5b4984975b2a'}}, parent_config=None)

Methods:

Name	Description
`from_conn_string`	Create a new SqliteSaver instance from a connection string.
`setup`	Set up the checkpoint database.
`cursor`	Get a cursor for the SQLite database.
`get_tuple`	Get a checkpoint tuple from the database.
`list`	List checkpoints from the database.
`put`	Save a checkpoint to the database.
`put_writes`	Store intermediate writes linked to a checkpoint.
`delete_thread`	Delete all checkpoints and writes associated with a thread ID.
`aget_tuple`	Get a checkpoint tuple from the database asynchronously.
`alist`	List checkpoints from the database asynchronously.
`aput`	Save a checkpoint to the database asynchronously.
`get_next_version`	Generate the next version ID for a channel.
`get`	Fetch a checkpoint using the given configuration.
`aget`	Asynchronously fetch a checkpoint using the given configuration.
`aput_writes`	Asynchronously store intermediate writes linked to a checkpoint.
`adelete_thread`	Delete all checkpoints and writes associated with a specific thread ID.

Attributes:

Name	Type	Description
`config_specs`	`list`	Define the configuration options for the checkpoint saver.

config_specs `property` ¶

config_specs: list

Define the configuration options for the checkpoint saver.

Returns:

Name	Type	Description
`list`	`list`	List of configuration field specs.

from_conn_string `classmethod` ¶

from_conn_string(conn_string: str) -> Iterator[SqliteSaver]

Create a new SqliteSaver instance from a connection string.

Parameters:

Name	Type	Description	Default
`conn_string`	`str`	The SQLite connection string.	required

Yields:

Name	Type	Description
`SqliteSaver`	`SqliteSaver`	A new SqliteSaver instance.

Examples:

In memory:

    with SqliteSaver.from_conn_string(":memory:") as memory:
        ...

To disk:

    with SqliteSaver.from_conn_string("checkpoints.sqlite") as memory:
        ...

setup ¶

setup() -> None

Set up the checkpoint database.

This method creates the necessary tables in the SQLite database if they don't already exist. It is called automatically when needed and should not be called directly by the user.

cursor ¶

cursor(transaction: bool = True) -> Iterator[Cursor]

Get a cursor for the SQLite database.

This method returns a cursor for the SQLite database. It is used internally by the SqliteSaver and should not be called directly by the user.

Parameters:

Name	Type	Description	Default
`transaction`	`bool`	Whether to commit the transaction when the cursor is closed. Defaults to True.	`True`

Yields:

Type	Description
`Cursor`	sqlite3.Cursor: A cursor for the SQLite database.

get_tuple ¶

get_tuple(config: RunnableConfig) -> CheckpointTuple | None

Get a checkpoint tuple from the database.

This method retrieves a checkpoint tuple from the SQLite database based on the provided config. If the config contains a "checkpoint_id" key, the checkpoint with the matching thread ID and checkpoint ID is retrieved. Otherwise, the latest checkpoint for the given thread ID is retrieved.

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig`	The config to use for retrieving the checkpoint.	required

Returns:

Type	Description
`CheckpointTuple \| None`	Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.

Examples:

Basic:
>>> config = {"configurable": {"thread_id": "1"}}
>>> checkpoint_tuple = memory.get_tuple(config)
>>> print(checkpoint_tuple)
CheckpointTuple(...)

With checkpoint ID:

>>> config = {
...    "configurable": {
...        "thread_id": "1",
...        "checkpoint_ns": "",
...        "checkpoint_id": "1ef4f797-8335-6428-8001-8a1503f9b875",
...    }
... }
>>> checkpoint_tuple = memory.get_tuple(config)
>>> print(checkpoint_tuple)
CheckpointTuple(...)

list ¶

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

List checkpoints from the database.

This method retrieves a list of checkpoint tuples from the SQLite database based on the provided config. The checkpoints are ordered by checkpoint ID in descending order (newest first).

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig \| None`	The config to use for listing the checkpoints.	required
`filter`	`dict[str, Any] \| None`	Additional filtering criteria for metadata. Defaults to None.	`None`
`before`	`RunnableConfig \| None`	If provided, only checkpoints before the specified checkpoint ID are returned. Defaults to None.	`None`
`limit`	`int \| None`	The maximum number of checkpoints to return. Defaults to None.	`None`

Yields:

Type	Description
`CheckpointTuple`	Iterator[CheckpointTuple]: An iterator of checkpoint tuples.

Examples:

>>> from langgraph.checkpoint.sqlite import SqliteSaver
>>> with SqliteSaver.from_conn_string(":memory:") as memory:
... # Run a graph, then list the checkpoints
>>>     config = {"configurable": {"thread_id": "1"}}
>>>     checkpoints = list(memory.list(config, limit=2))
>>> print(checkpoints)
[CheckpointTuple(...), CheckpointTuple(...)]

>>> config = {"configurable": {"thread_id": "1"}}
>>> before = {"configurable": {"checkpoint_id": "1ef4f797-8335-6428-8001-8a1503f9b875"}}
>>> with SqliteSaver.from_conn_string(":memory:") as memory:
... # Run a graph, then list the checkpoints
>>>     checkpoints = list(memory.list(config, before=before))
>>> print(checkpoints)
[CheckpointTuple(...), ...]

put ¶

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

Save a checkpoint to the database.

This method saves a checkpoint to the SQLite database. The checkpoint is associated with the provided config and its parent config (if any).

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig`	The config to associate with the checkpoint.	required
`checkpoint`	`Checkpoint`	The checkpoint to save.	required
`metadata`	`CheckpointMetadata`	Additional metadata to save with the checkpoint.	required
`new_versions`	`ChannelVersions`	New channel versions as of this write.	required

Returns:

Name	Type	Description
`RunnableConfig`	`RunnableConfig`	Updated configuration after storing the checkpoint.

Examples:

>>> from langgraph.checkpoint.sqlite import SqliteSaver
>>> with SqliteSaver.from_conn_string(":memory:") as memory:
>>>     config = {"configurable": {"thread_id": "1", "checkpoint_ns": ""}}
>>>     checkpoint = {"ts": "2024-05-04T06:32:42.235444+00:00", "id": "1ef4f797-8335-6428-8001-8a1503f9b875", "channel_values": {"key": "value"}}
>>>     saved_config = memory.put(config, checkpoint, {"source": "input", "step": 1, "writes": {"key": "value"}}, {})
>>> print(saved_config)
{'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1ef4f797-8335-6428-8001-8a1503f9b875'}}

put_writes ¶

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

Store intermediate writes linked to a checkpoint.

This method saves intermediate writes associated with a checkpoint to the SQLite database.

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig`	Configuration of the related checkpoint.	required
`writes`	`Sequence[tuple[str, Any]]`	List of writes to store, each as (channel, value) pair.	required
`task_id`	`str`	Identifier for the task creating the writes.	required
`task_path`	`str`	Path of the task creating the writes.	`''`

delete_thread ¶

delete_thread(thread_id: str) -> None

Delete all checkpoints and writes associated with a thread ID.

Parameters:

Name	Type	Description	Default
`thread_id`	`str`	The thread ID to delete.	required

Returns:

Type	Description
`None`	None

aget_tuple `async` ¶

aget_tuple(
    config: RunnableConfig,
) -> CheckpointTuple | None

Get a checkpoint tuple from the database asynchronously.

Note

This async method is not supported by the SqliteSaver class. Use get_tuple() instead, or consider using AsyncSqliteSaver.

alist `async` ¶

alist(
    config: RunnableConfig | None,
    *,
    filter: dict[str, Any] | None = None,
    before: RunnableConfig | None = None,
    limit: int | None = None
) -> AsyncIterator[CheckpointTuple]

List checkpoints from the database asynchronously.

Note

This async method is not supported by the SqliteSaver class. Use list() instead, or consider using AsyncSqliteSaver.

aput `async` ¶

aput(
    config: RunnableConfig,
    checkpoint: Checkpoint,
    metadata: CheckpointMetadata,
    new_versions: ChannelVersions,
) -> RunnableConfig

Save a checkpoint to the database asynchronously.

Note

This async method is not supported by the SqliteSaver class. Use put() instead, or consider using AsyncSqliteSaver.

get_next_version ¶

get_next_version(current: str | None, channel: None) -> str

Generate the next version ID for a channel.

This method creates a new version identifier for a channel based on its current version.

Parameters:

Name	Type	Description	Default
`current`	`Optional[str]`	The current version identifier of the channel.	required

Returns:

Name	Type	Description
`str`	`str`	The next version identifier, which is guaranteed to be monotonically increasing.

get ¶

get(config: RunnableConfig) -> Checkpoint | None

Fetch a checkpoint using the given configuration.

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig`	Configuration specifying which checkpoint to retrieve.	required

Returns:

Type	Description
`Checkpoint \| None`	Optional[Checkpoint]: The requested checkpoint, or None if not found.

aget `async` ¶

aget(config: RunnableConfig) -> Checkpoint | None

Asynchronously fetch a checkpoint using the given configuration.

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig`	Configuration specifying which checkpoint to retrieve.	required

Returns:

Type	Description
`Checkpoint \| None`	Optional[Checkpoint]: The requested checkpoint, or None if not found.

aput_writes `async` ¶

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

Asynchronously store intermediate writes linked to a checkpoint.

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig`	Configuration of the related checkpoint.	required
`writes`	`Sequence[tuple[str, Any]]`	List of writes to store.	required
`task_id`	`str`	Identifier for the task creating the writes.	required
`task_path`	`str`	Path of the task creating the writes.	`''`

Raises:

Type	Description
`NotImplementedError`	Implement this method in your custom checkpoint saver.

adelete_thread `async` ¶

adelete_thread(thread_id: str) -> None

Delete all checkpoints and writes associated with a specific thread ID.

Parameters:

Name	Type	Description	Default
`thread_id`	`str`	The thread ID whose checkpoints should be deleted.	required

Classes:

Name	Description
`AsyncSqliteSaver`	An asynchronous checkpoint saver that stores checkpoints in a SQLite database.

AsyncSqliteSaver ¶

Bases: BaseCheckpointSaver[str]

An asynchronous checkpoint saver that stores checkpoints in a SQLite database.

This class provides an asynchronous interface for saving and retrieving checkpoints using a SQLite database. It's designed for use in asynchronous environments and offers better performance for I/O-bound operations compared to synchronous alternatives.

Attributes:

Name	Type	Description
`conn`	`Connection`	The asynchronous SQLite database connection.
`serde`	`SerializerProtocol`	The serializer used for encoding/decoding checkpoints.

Tip

Requires the aiosqlite package. Install it with pip install aiosqlite.

Warning

While this class supports asynchronous checkpointing, it is not recommended for production workloads due to limitations in SQLite's write performance. For production use, consider a more robust database like PostgreSQL.

Tip

Remember to close the database connection after executing your code, otherwise, you may see the graph "hang" after execution (since the program will not exit until the connection is closed).

The easiest way is to use the async with statement as shown in the examples.

async with AsyncSqliteSaver.from_conn_string("checkpoints.sqlite") as saver:
    # Your code here
    graph = builder.compile(checkpointer=saver)
    config = {"configurable": {"thread_id": "thread-1"}}
    async for event in graph.astream_events(..., config, version="v1"):
        print(event)

Examples:

Usage within StateGraph:

>>> import asyncio
>>>
>>> from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
>>> from langgraph.graph import StateGraph
>>>
>>> async def main():
>>>     builder = StateGraph(int)
>>>     builder.add_node("add_one", lambda x: x + 1)
>>>     builder.set_entry_point("add_one")
>>>     builder.set_finish_point("add_one")
>>>     async with AsyncSqliteSaver.from_conn_string("checkpoints.db") as memory:
>>>         graph = builder.compile(checkpointer=memory)
>>>         coro = graph.ainvoke(1, {"configurable": {"thread_id": "thread-1"}})
>>>         print(await asyncio.gather(coro))
>>>
>>> asyncio.run(main())
Output: [2]

Raw usage:

>>> import asyncio
>>> import aiosqlite
>>> from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
>>>
>>> async def main():
>>>     async with aiosqlite.connect("checkpoints.db") as conn:
...         saver = AsyncSqliteSaver(conn)
...         config = {"configurable": {"thread_id": "1", "checkpoint_ns": ""}}
...         checkpoint = {"ts": "2023-05-03T10:00:00Z", "data": {"key": "value"}, "id": "0c62ca34-ac19-445d-bbb0-5b4984975b2a"}
...         saved_config = await saver.aput(config, checkpoint, {}, {})
...         print(saved_config)
>>> asyncio.run(main())
{'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '0c62ca34-ac19-445d-bbb0-5b4984975b2a'}}

Methods:

Name	Description
`from_conn_string`	Create a new AsyncSqliteSaver instance from a connection string.
`get_tuple`	Get a checkpoint tuple from the database.
`list`	List checkpoints from the database asynchronously.
`put`	Save a checkpoint to the database.
`delete_thread`	Delete all checkpoints and writes associated with a thread ID.
`setup`	Set up the checkpoint database asynchronously.
`aget_tuple`	Get a checkpoint tuple from the database asynchronously.
`alist`	List checkpoints from the database asynchronously.
`aput`	Save a checkpoint to the database asynchronously.
`aput_writes`	Store intermediate writes linked to a checkpoint asynchronously.
`adelete_thread`	Delete all checkpoints and writes associated with a thread ID.
`get_next_version`	Generate the next version ID for a channel.
`get`	Fetch a checkpoint using the given configuration.
`aget`	Asynchronously fetch a checkpoint using the given configuration.

config_specs `property` ¶

config_specs: list

Define the configuration options for the checkpoint saver.

Returns:

Name	Type	Description
`list`	`list`	List of configuration field specs.

from_conn_string `async` `classmethod` ¶

from_conn_string(
    conn_string: str,
) -> AsyncIterator[AsyncSqliteSaver]

Create a new AsyncSqliteSaver instance from a connection string.

Parameters:

Name	Type	Description	Default
`conn_string`	`str`	The SQLite connection string.	required

Yields:

Name	Type	Description
`AsyncSqliteSaver`	`AsyncIterator[AsyncSqliteSaver]`	A new AsyncSqliteSaver instance.

get_tuple ¶

get_tuple(config: RunnableConfig) -> CheckpointTuple | None

Get a checkpoint tuple from the database.

This method retrieves a checkpoint tuple from the SQLite database based on the provided config. If the config contains a "checkpoint_id" key, the checkpoint with the matching thread ID and checkpoint ID is retrieved. Otherwise, the latest checkpoint for the given thread ID is retrieved.

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig`	The config to use for retrieving the checkpoint.	required

Returns:

Type	Description
`CheckpointTuple \| None`	Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.

list ¶

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

List checkpoints from the database asynchronously.

This method retrieves a list of checkpoint tuples from the SQLite database based on the provided config. The checkpoints are ordered by checkpoint ID in descending order (newest first).

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig \| None`	Base configuration for filtering checkpoints.	required
`filter`	`dict[str, Any] \| None`	Additional filtering criteria for metadata.	`None`
`before`	`RunnableConfig \| None`	If provided, only checkpoints before the specified checkpoint ID are returned. Defaults to None.	`None`
`limit`	`int \| None`	Maximum number of checkpoints to return.	`None`

Yields:

Type	Description
`CheckpointTuple`	Iterator[CheckpointTuple]: An iterator of matching checkpoint tuples.

put ¶

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

Save a checkpoint to the database.

This method saves a checkpoint to the SQLite database. The checkpoint is associated with the provided config and its parent config (if any).

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig`	The config to associate with the checkpoint.	required
`checkpoint`	`Checkpoint`	The checkpoint to save.	required
`metadata`	`CheckpointMetadata`	Additional metadata to save with the checkpoint.	required
`new_versions`	`ChannelVersions`	New channel versions as of this write.	required

Returns:

Name	Type	Description
`RunnableConfig`	`RunnableConfig`	Updated configuration after storing the checkpoint.

delete_thread ¶

delete_thread(thread_id: str) -> None

Delete all checkpoints and writes associated with a thread ID.

Parameters:

Name	Type	Description	Default
`thread_id`	`str`	The thread ID to delete.	required

Returns:

Type	Description
`None`	None

setup `async` ¶

setup() -> None

Set up the checkpoint database asynchronously.

This method creates the necessary tables in the SQLite database if they don't already exist. It is called automatically when needed and should not be called directly by the user.

aget_tuple `async` ¶

aget_tuple(
    config: RunnableConfig,
) -> CheckpointTuple | None

Get a checkpoint tuple from the database asynchronously.

This method retrieves a checkpoint tuple from the SQLite database based on the provided config. If the config contains a "checkpoint_id" key, the checkpoint with the matching thread ID and checkpoint ID is retrieved. Otherwise, the latest checkpoint for the given thread ID is retrieved.

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig`	The config to use for retrieving the checkpoint.	required

Returns:

Type	Description
`CheckpointTuple \| None`	Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.

alist `async` ¶

alist(
    config: RunnableConfig | None,
    *,
    filter: dict[str, Any] | None = None,
    before: RunnableConfig | None = None,
    limit: int | None = None
) -> AsyncIterator[CheckpointTuple]

List checkpoints from the database asynchronously.

This method retrieves a list of checkpoint tuples from the SQLite database based on the provided config. The checkpoints are ordered by checkpoint ID in descending order (newest first).

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig \| None`	Base configuration for filtering checkpoints.	required
`filter`	`dict[str, Any] \| None`	Additional filtering criteria for metadata.	`None`
`before`	`RunnableConfig \| None`	If provided, only checkpoints before the specified checkpoint ID are returned. Defaults to None.	`None`
`limit`	`int \| None`	Maximum number of checkpoints to return.	`None`

Yields:

Type	Description
`AsyncIterator[CheckpointTuple]`	AsyncIterator[CheckpointTuple]: An asynchronous iterator of matching checkpoint tuples.

aput `async` ¶

aput(
    config: RunnableConfig,
    checkpoint: Checkpoint,
    metadata: CheckpointMetadata,
    new_versions: ChannelVersions,
) -> RunnableConfig

Save a checkpoint to the database asynchronously.

This method saves a checkpoint to the SQLite database. The checkpoint is associated with the provided config and its parent config (if any).

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig`	The config to associate with the checkpoint.	required
`checkpoint`	`Checkpoint`	The checkpoint to save.	required
`metadata`	`CheckpointMetadata`	Additional metadata to save with the checkpoint.	required
`new_versions`	`ChannelVersions`	New channel versions as of this write.	required

Returns:

Name	Type	Description
`RunnableConfig`	`RunnableConfig`	Updated configuration after storing the checkpoint.

aput_writes `async` ¶

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

Store intermediate writes linked to a checkpoint asynchronously.

This method saves intermediate writes associated with a checkpoint to the database.

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig`	Configuration of the related checkpoint.	required
`writes`	`Sequence[tuple[str, Any]]`	List of writes to store, each as (channel, value) pair.	required
`task_id`	`str`	Identifier for the task creating the writes.	required
`task_path`	`str`	Path of the task creating the writes.	`''`

adelete_thread `async` ¶

adelete_thread(thread_id: str) -> None

Delete all checkpoints and writes associated with a thread ID.

Parameters:

Name	Type	Description	Default
`thread_id`	`str`	The thread ID to delete.	required

Returns:

Type	Description
`None`	None

get_next_version ¶

get_next_version(current: str | None, channel: None) -> str

Generate the next version ID for a channel.

This method creates a new version identifier for a channel based on its current version.

Parameters:

Name	Type	Description	Default
`current`	`Optional[str]`	The current version identifier of the channel.	required

Returns:

Name	Type	Description
`str`	`str`	The next version identifier, which is guaranteed to be monotonically increasing.

get ¶

get(config: RunnableConfig) -> Checkpoint | None

Fetch a checkpoint using the given configuration.

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig`	Configuration specifying which checkpoint to retrieve.	required

Returns:

Type	Description
`Checkpoint \| None`	Optional[Checkpoint]: The requested checkpoint, or None if not found.

aget `async` ¶

aget(config: RunnableConfig) -> Checkpoint | None

Asynchronously fetch a checkpoint using the given configuration.

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig`	Configuration specifying which checkpoint to retrieve.	required

Returns:

Type	Description
`Checkpoint \| None`	Optional[Checkpoint]: The requested checkpoint, or None if not found.

Classes:

Name	Description
`PostgresSaver`	Checkpointer that stores checkpoints in a Postgres database.

PostgresSaver ¶

Bases: BasePostgresSaver

Checkpointer that stores checkpoints in a Postgres database.

Methods:

Name	Description
`from_conn_string`	Create a new PostgresSaver instance from a connection string.
`setup`	Set up the checkpoint database asynchronously.
`list`	List checkpoints from the database.
`get_tuple`	Get a checkpoint tuple from the database.
`put`	Save a checkpoint to the database.
`put_writes`	Store intermediate writes linked to a checkpoint.
`delete_thread`	Delete all checkpoints and writes associated with a thread ID.
`get`	Fetch a checkpoint using the given configuration.
`aget`	Asynchronously fetch a checkpoint using the given configuration.
`aget_tuple`	Asynchronously fetch a checkpoint tuple using the given configuration.
`alist`	Asynchronously list checkpoints that match the given criteria.
`aput`	Asynchronously store a checkpoint with its configuration and metadata.
`aput_writes`	Asynchronously store intermediate writes linked to a checkpoint.
`adelete_thread`	Delete all checkpoints and writes associated with a specific thread ID.

Attributes:

Name	Type	Description
`config_specs`	`list`	Define the configuration options for the checkpoint saver.

config_specs `property` ¶

config_specs: list

Define the configuration options for the checkpoint saver.

Returns:

Name	Type	Description
`list`	`list`	List of configuration field specs.

from_conn_string `classmethod` ¶

from_conn_string(
    conn_string: str, *, pipeline: bool = False
) -> Iterator[PostgresSaver]

Create a new PostgresSaver instance from a connection string.

Parameters:

Name	Type	Description	Default
`conn_string`	`str`	The Postgres connection info string.	required
`pipeline`	`bool`	whether to use Pipeline	`False`

Returns:

Name	Type	Description
`PostgresSaver`	`Iterator[PostgresSaver]`	A new PostgresSaver instance.

setup ¶

setup() -> None

Set up the checkpoint database asynchronously.

This method creates the necessary tables in the Postgres database if they don't already exist and runs database migrations. It MUST be called directly by the user the first time checkpointer is used.

list ¶

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

List checkpoints from the database.

This method retrieves a list of checkpoint tuples from the Postgres database based on the provided config. The checkpoints are ordered by checkpoint ID in descending order (newest first).

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig \| None`	The config to use for listing the checkpoints.	required
`filter`	`dict[str, Any] \| None`	Additional filtering criteria for metadata. Defaults to None.	`None`
`before`	`RunnableConfig \| None`	If provided, only checkpoints before the specified checkpoint ID are returned. Defaults to None.	`None`
`limit`	`int \| None`	The maximum number of checkpoints to return. Defaults to None.	`None`

Yields:

Type	Description
`CheckpointTuple`	Iterator[CheckpointTuple]: An iterator of checkpoint tuples.

Examples:

>>> from langgraph.checkpoint.postgres import PostgresSaver
>>> DB_URI = "postgres://postgres:postgres@localhost:5432/postgres?sslmode=disable"
>>> with PostgresSaver.from_conn_string(DB_URI) as memory:
... # Run a graph, then list the checkpoints
>>>     config = {"configurable": {"thread_id": "1"}}
>>>     checkpoints = list(memory.list(config, limit=2))
>>> print(checkpoints)
[CheckpointTuple(...), CheckpointTuple(...)]

>>> config = {"configurable": {"thread_id": "1"}}
>>> before = {"configurable": {"checkpoint_id": "1ef4f797-8335-6428-8001-8a1503f9b875"}}
>>> with PostgresSaver.from_conn_string(DB_URI) as memory:
... # Run a graph, then list the checkpoints
>>>     checkpoints = list(memory.list(config, before=before))
>>> print(checkpoints)
[CheckpointTuple(...), ...]

get_tuple ¶

get_tuple(config: RunnableConfig) -> CheckpointTuple | None

Get a checkpoint tuple from the database.

This method retrieves a checkpoint tuple from the Postgres database based on the provided config. If the config contains a "checkpoint_id" key, the checkpoint with the matching thread ID and timestamp is retrieved. Otherwise, the latest checkpoint for the given thread ID is retrieved.

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig`	The config to use for retrieving the checkpoint.	required

Returns:

Type	Description
`CheckpointTuple \| None`	Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.

Examples:

Basic:
>>> config = {"configurable": {"thread_id": "1"}}
>>> checkpoint_tuple = memory.get_tuple(config)
>>> print(checkpoint_tuple)
CheckpointTuple(...)

With timestamp:

>>> config = {
...    "configurable": {
...        "thread_id": "1",
...        "checkpoint_ns": "",
...        "checkpoint_id": "1ef4f797-8335-6428-8001-8a1503f9b875",
...    }
... }
>>> checkpoint_tuple = memory.get_tuple(config)
>>> print(checkpoint_tuple)
CheckpointTuple(...)

put ¶

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

Save a checkpoint to the database.

This method saves a checkpoint to the Postgres database. The checkpoint is associated with the provided config and its parent config (if any).

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig`	The config to associate with the checkpoint.	required
`checkpoint`	`Checkpoint`	The checkpoint to save.	required
`metadata`	`CheckpointMetadata`	Additional metadata to save with the checkpoint.	required
`new_versions`	`ChannelVersions`	New channel versions as of this write.	required

Returns:

Name	Type	Description
`RunnableConfig`	`RunnableConfig`	Updated configuration after storing the checkpoint.

Examples:

>>> from langgraph.checkpoint.postgres import PostgresSaver
>>> DB_URI = "postgres://postgres:postgres@localhost:5432/postgres?sslmode=disable"
>>> with PostgresSaver.from_conn_string(DB_URI) as memory:
>>>     config = {"configurable": {"thread_id": "1", "checkpoint_ns": ""}}
>>>     checkpoint = {"ts": "2024-05-04T06:32:42.235444+00:00", "id": "1ef4f797-8335-6428-8001-8a1503f9b875", "channel_values": {"key": "value"}}
>>>     saved_config = memory.put(config, checkpoint, {"source": "input", "step": 1, "writes": {"key": "value"}}, {})
>>> print(saved_config)
{'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1ef4f797-8335-6428-8001-8a1503f9b875'}}

put_writes ¶

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

Store intermediate writes linked to a checkpoint.

This method saves intermediate writes associated with a checkpoint to the Postgres database.

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig`	Configuration of the related checkpoint.	required
`writes`	`Sequence[tuple[str, Any]]`	List of writes to store.	required
`task_id`	`str`	Identifier for the task creating the writes.	required

delete_thread ¶

delete_thread(thread_id: str) -> None

Delete all checkpoints and writes associated with a thread ID.

Parameters:

Name	Type	Description	Default
`thread_id`	`str`	The thread ID to delete.	required

Returns:

Type	Description
`None`	None

get ¶

get(config: RunnableConfig) -> Checkpoint | None

Fetch a checkpoint using the given configuration.

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig`	Configuration specifying which checkpoint to retrieve.	required

Returns:

Type	Description
`Checkpoint \| None`	Optional[Checkpoint]: The requested checkpoint, or None if not found.

aget `async` ¶

aget(config: RunnableConfig) -> Checkpoint | None

Asynchronously fetch a checkpoint using the given configuration.

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig`	Configuration specifying which checkpoint to retrieve.	required

Returns:

Type	Description
`Checkpoint \| None`	Optional[Checkpoint]: The requested checkpoint, or None if not found.

aget_tuple `async` ¶

aget_tuple(
    config: RunnableConfig,
) -> CheckpointTuple | None

Asynchronously fetch a checkpoint tuple using the given configuration.

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig`	Configuration specifying which checkpoint to retrieve.	required

Returns:

Type	Description
`CheckpointTuple \| None`	Optional[CheckpointTuple]: The requested checkpoint tuple, or None if not found.

Raises:

Type	Description
`NotImplementedError`	Implement this method in your custom checkpoint saver.

alist `async` ¶

alist(
    config: RunnableConfig | None,
    *,
    filter: dict[str, Any] | None = None,
    before: RunnableConfig | None = None,
    limit: int | None = None
) -> AsyncIterator[CheckpointTuple]

Asynchronously list checkpoints that match the given criteria.

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig \| None`	Base configuration for filtering checkpoints.	required
`filter`	`dict[str, Any] \| None`	Additional filtering criteria for metadata.	`None`
`before`	`RunnableConfig \| None`	List checkpoints created before this configuration.	`None`
`limit`	`int \| None`	Maximum number of checkpoints to return.	`None`

Returns:

Type	Description
`AsyncIterator[CheckpointTuple]`	AsyncIterator[CheckpointTuple]: Async iterator of matching checkpoint tuples.

Raises:

Type	Description
`NotImplementedError`	Implement this method in your custom checkpoint saver.

aput `async` ¶

aput(
    config: RunnableConfig,
    checkpoint: Checkpoint,
    metadata: CheckpointMetadata,
    new_versions: ChannelVersions,
) -> RunnableConfig

Asynchronously store a checkpoint with its configuration and metadata.

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig`	Configuration for the checkpoint.	required
`checkpoint`	`Checkpoint`	The checkpoint to store.	required
`metadata`	`CheckpointMetadata`	Additional metadata for the checkpoint.	required
`new_versions`	`ChannelVersions`	New channel versions as of this write.	required

Returns:

Name	Type	Description
`RunnableConfig`	`RunnableConfig`	Updated configuration after storing the checkpoint.

Raises:

Type	Description
`NotImplementedError`	Implement this method in your custom checkpoint saver.

aput_writes `async` ¶

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

Asynchronously store intermediate writes linked to a checkpoint.

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig`	Configuration of the related checkpoint.	required
`writes`	`Sequence[tuple[str, Any]]`	List of writes to store.	required
`task_id`	`str`	Identifier for the task creating the writes.	required
`task_path`	`str`	Path of the task creating the writes.	`''`

Raises:

Type	Description
`NotImplementedError`	Implement this method in your custom checkpoint saver.

adelete_thread `async` ¶

adelete_thread(thread_id: str) -> None

Delete all checkpoints and writes associated with a specific thread ID.

Parameters:

Name	Type	Description	Default
`thread_id`	`str`	The thread ID whose checkpoints should be deleted.	required

Classes:

Name	Description
`AsyncPostgresSaver`	Asynchronous checkpointer that stores checkpoints in a Postgres database.

AsyncPostgresSaver ¶

Bases: BasePostgresSaver

Asynchronous checkpointer that stores checkpoints in a Postgres database.

Methods:

Name	Description
`from_conn_string`	Create a new AsyncPostgresSaver instance from a connection string.
`setup`	Set up the checkpoint database asynchronously.
`alist`	List checkpoints from the database asynchronously.
`aget_tuple`	Get a checkpoint tuple from the database asynchronously.
`aput`	Save a checkpoint to the database asynchronously.
`aput_writes`	Store intermediate writes linked to a checkpoint asynchronously.
`adelete_thread`	Delete all checkpoints and writes associated with a thread ID.
`list`	List checkpoints from the database.
`get_tuple`	Get a checkpoint tuple from the database.
`put`	Save a checkpoint to the database.
`put_writes`	Store intermediate writes linked to a checkpoint.
`delete_thread`	Delete all checkpoints and writes associated with a thread ID.
`get`	Fetch a checkpoint using the given configuration.
`aget`	Asynchronously fetch a checkpoint using the given configuration.

Attributes:

Name	Type	Description
`config_specs`	`list`	Define the configuration options for the checkpoint saver.

config_specs `property` ¶

config_specs: list

Define the configuration options for the checkpoint saver.

Returns:

Name	Type	Description
`list`	`list`	List of configuration field specs.

from_conn_string `async` `classmethod` ¶

from_conn_string(
    conn_string: str,
    *,
    pipeline: bool = False,
    serde: SerializerProtocol | None = None
) -> AsyncIterator[AsyncPostgresSaver]

Create a new AsyncPostgresSaver instance from a connection string.

Parameters:

Name	Type	Description	Default
`conn_string`	`str`	The Postgres connection info string.	required
`pipeline`	`bool`	whether to use AsyncPipeline	`False`

Returns:

Name	Type	Description
`AsyncPostgresSaver`	`AsyncIterator[AsyncPostgresSaver]`	A new AsyncPostgresSaver instance.

setup `async` ¶

setup() -> None

Set up the checkpoint database asynchronously.

This method creates the necessary tables in the Postgres database if they don't already exist and runs database migrations. It MUST be called directly by the user the first time checkpointer is used.

alist `async` ¶

alist(
    config: RunnableConfig | None,
    *,
    filter: dict[str, Any] | None = None,
    before: RunnableConfig | None = None,
    limit: int | None = None
) -> AsyncIterator[CheckpointTuple]

List checkpoints from the database asynchronously.

This method retrieves a list of checkpoint tuples from the Postgres database based on the provided config. The checkpoints are ordered by checkpoint ID in descending order (newest first).

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig \| None`	Base configuration for filtering checkpoints.	required
`filter`	`dict[str, Any] \| None`	Additional filtering criteria for metadata.	`None`
`before`	`RunnableConfig \| None`	If provided, only checkpoints before the specified checkpoint ID are returned. Defaults to None.	`None`
`limit`	`int \| None`	Maximum number of checkpoints to return.	`None`

Yields:

Type	Description
`AsyncIterator[CheckpointTuple]`	AsyncIterator[CheckpointTuple]: An asynchronous iterator of matching checkpoint tuples.

aget_tuple `async` ¶

aget_tuple(
    config: RunnableConfig,
) -> CheckpointTuple | None

Get a checkpoint tuple from the database asynchronously.

This method retrieves a checkpoint tuple from the Postgres database based on the provided config. If the config contains a "checkpoint_id" key, the checkpoint with the matching thread ID and "checkpoint_id" is retrieved. Otherwise, the latest checkpoint for the given thread ID is retrieved.

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig`	The config to use for retrieving the checkpoint.	required

Returns:

Type	Description
`CheckpointTuple \| None`	Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.

aput `async` ¶

aput(
    config: RunnableConfig,
    checkpoint: Checkpoint,
    metadata: CheckpointMetadata,
    new_versions: ChannelVersions,
) -> RunnableConfig

Save a checkpoint to the database asynchronously.

This method saves a checkpoint to the Postgres database. The checkpoint is associated with the provided config and its parent config (if any).

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig`	The config to associate with the checkpoint.	required
`checkpoint`	`Checkpoint`	The checkpoint to save.	required
`metadata`	`CheckpointMetadata`	Additional metadata to save with the checkpoint.	required
`new_versions`	`ChannelVersions`	New channel versions as of this write.	required

Returns:

Name	Type	Description
`RunnableConfig`	`RunnableConfig`	Updated configuration after storing the checkpoint.

aput_writes `async` ¶

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

Store intermediate writes linked to a checkpoint asynchronously.

This method saves intermediate writes associated with a checkpoint to the database.

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig`	Configuration of the related checkpoint.	required
`writes`	`Sequence[tuple[str, Any]]`	List of writes to store, each as (channel, value) pair.	required
`task_id`	`str`	Identifier for the task creating the writes.	required

adelete_thread `async` ¶

adelete_thread(thread_id: str) -> None

Delete all checkpoints and writes associated with a thread ID.

Parameters:

Name	Type	Description	Default
`thread_id`	`str`	The thread ID to delete.	required

Returns:

Type	Description
`None`	None

list ¶

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

List checkpoints from the database.

This method retrieves a list of checkpoint tuples from the Postgres database based on the provided config. The checkpoints are ordered by checkpoint ID in descending order (newest first).

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig \| None`	Base configuration for filtering checkpoints.	required
`filter`	`dict[str, Any] \| None`	Additional filtering criteria for metadata.	`None`
`before`	`RunnableConfig \| None`	If provided, only checkpoints before the specified checkpoint ID are returned. Defaults to None.	`None`
`limit`	`int \| None`	Maximum number of checkpoints to return.	`None`

Yields:

Type	Description
`CheckpointTuple`	Iterator[CheckpointTuple]: An iterator of matching checkpoint tuples.

get_tuple ¶

get_tuple(config: RunnableConfig) -> CheckpointTuple | None

Get a checkpoint tuple from the database.

This method retrieves a checkpoint tuple from the Postgres database based on the provided config. If the config contains a "checkpoint_id" key, the checkpoint with the matching thread ID and "checkpoint_id" is retrieved. Otherwise, the latest checkpoint for the given thread ID is retrieved.

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig`	The config to use for retrieving the checkpoint.	required

Returns:

Type	Description
`CheckpointTuple \| None`	Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.

put ¶

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

Save a checkpoint to the database.

This method saves a checkpoint to the Postgres database. The checkpoint is associated with the provided config and its parent config (if any).

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig`	The config to associate with the checkpoint.	required
`checkpoint`	`Checkpoint`	The checkpoint to save.	required
`metadata`	`CheckpointMetadata`	Additional metadata to save with the checkpoint.	required
`new_versions`	`ChannelVersions`	New channel versions as of this write.	required

Returns:

Name	Type	Description
`RunnableConfig`	`RunnableConfig`	Updated configuration after storing the checkpoint.

put_writes ¶

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

Store intermediate writes linked to a checkpoint.

This method saves intermediate writes associated with a checkpoint to the database.

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig`	Configuration of the related checkpoint.	required
`writes`	`Sequence[tuple[str, Any]]`	List of writes to store, each as (channel, value) pair.	required
`task_id`	`str`	Identifier for the task creating the writes.	required
`task_path`	`str`	Path of the task creating the writes.	`''`

delete_thread ¶

delete_thread(thread_id: str) -> None

Delete all checkpoints and writes associated with a thread ID.

Parameters:

Name	Type	Description	Default
`thread_id`	`str`	The thread ID to delete.	required

Returns:

Type	Description
`None`	None

get ¶

get(config: RunnableConfig) -> Checkpoint | None

Fetch a checkpoint using the given configuration.

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig`	Configuration specifying which checkpoint to retrieve.	required

Returns:

Type	Description
`Checkpoint \| None`	Optional[Checkpoint]: The requested checkpoint, or None if not found.

aget `async` ¶

aget(config: RunnableConfig) -> Checkpoint | None

Asynchronously fetch a checkpoint using the given configuration.

Parameters:

Name	Type	Description	Default
`config`	`RunnableConfig`	Configuration specifying which checkpoint to retrieve.	required

Returns:

Type	Description
`Checkpoint \| None`	Optional[Checkpoint]: The requested checkpoint, or None if not found.

Checkpointers¶

CheckpointMetadata ¶

source instance-attribute ¶

step instance-attribute ¶

parents instance-attribute ¶

Checkpoint ¶

v instance-attribute ¶

id instance-attribute ¶

ts instance-attribute ¶

channel_values instance-attribute ¶

channel_versions instance-attribute ¶

versions_seen instance-attribute ¶

updated_channels instance-attribute ¶

BaseCheckpointSaver ¶

config_specs property ¶

get ¶

get_tuple ¶

list ¶

put ¶

put_writes ¶

delete_thread ¶

aget async ¶

aget_tuple async ¶

alist async ¶

aput async ¶

aput_writes async ¶

adelete_thread async ¶

get_next_version ¶

create_checkpoint ¶

SerializerProtocol ¶

CipherProtocol ¶

encrypt ¶

decrypt ¶

JsonPlusSerializer ¶

EncryptedSerializer ¶

dumps_typed ¶

from_pycryptodome_aes classmethod ¶

InMemorySaver ¶

config_specs property ¶

get_tuple ¶

list ¶

put ¶

put_writes ¶

delete_thread ¶

aget_tuple async ¶

alist async ¶

aput async ¶

aput_writes async ¶

adelete_thread async ¶

get ¶

aget async ¶

PersistentDict ¶

sync ¶

SqliteSaver ¶

config_specs property ¶

from_conn_string classmethod ¶

setup ¶

cursor ¶

get_tuple ¶

list ¶

put ¶

put_writes ¶

delete_thread ¶

aget_tuple async ¶

alist async ¶

aput async ¶

get_next_version ¶

get ¶

aget async ¶

aput_writes async ¶

adelete_thread async ¶

AsyncSqliteSaver ¶

config_specs property ¶

from_conn_string async classmethod ¶

get_tuple ¶

list ¶

put ¶

delete_thread ¶

setup async ¶

aget_tuple async ¶

source `instance-attribute` ¶

step `instance-attribute` ¶

parents `instance-attribute` ¶

v `instance-attribute` ¶

id `instance-attribute` ¶

ts `instance-attribute` ¶

channel_values `instance-attribute` ¶

channel_versions `instance-attribute` ¶

versions_seen `instance-attribute` ¶

updated_channels `instance-attribute` ¶

config_specs `property` ¶

aget `async` ¶

aget_tuple `async` ¶

alist `async` ¶

aput `async` ¶

aput_writes `async` ¶

adelete_thread `async` ¶

from_pycryptodome_aes `classmethod` ¶

config_specs `property` ¶

aget_tuple `async` ¶

alist `async` ¶

aput `async` ¶

aput_writes `async` ¶

adelete_thread `async` ¶

aget `async` ¶

config_specs `property` ¶

from_conn_string `classmethod` ¶

aget_tuple `async` ¶

alist `async` ¶

aput `async` ¶

aget `async` ¶

aput_writes `async` ¶

adelete_thread `async` ¶

config_specs `property` ¶

from_conn_string `async` `classmethod` ¶

setup `async` ¶

aget_tuple `async` ¶

alist `async` ¶

aput `async` ¶

aput_writes `async` ¶

adelete_thread `async` ¶

aget `async` ¶

config_specs `property` ¶

from_conn_string `classmethod` ¶

aget `async` ¶

aget_tuple `async` ¶

alist `async` ¶

aput `async` ¶

aput_writes `async` ¶

adelete_thread `async` ¶

config_specs `property` ¶

from_conn_string `async` `classmethod` ¶

setup `async` ¶

alist `async` ¶

aget_tuple `async` ¶

aput `async` ¶

aput_writes `async` ¶

adelete_thread `async` ¶

aget `async` ¶