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.
writes	dict[str, Any]	The writes that were made between the previous checkpoint and this one.
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.

writes instance-attribute

writes: dict[str, Any]

The writes that were made between the previous checkpoint and this one.

Mapping from node name to writes emitted by that node.

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.
pending_sends	List[SendProtocol]	List of inputs pushed to nodes but not yet processed.

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.

pending_sends instance-attribute

pending_sends: List[SendProtocol]

List of inputs pushed to nodes but not yet processed. Cleared by the next 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[ConfigurableFieldSpec]

Define the configuration options for the checkpoint saver.

Returns:

Type	Description
list[ConfigurableFieldSpec]	list[ConfigurableFieldSpec]: List of configuration field specs.

get

get(config: RunnableConfig) -> Optional[Checkpoint]

Fetch a checkpoint using the given configuration.

Parameters:

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

Returns:

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

get_tuple

get_tuple(
    config: RunnableConfig,
) -> Optional[CheckpointTuple]

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
Optional[CheckpointTuple]	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: Optional[RunnableConfig],
    *,
    filter: Optional[Dict[str, Any]] = None,
    before: Optional[RunnableConfig] = None,
    limit: Optional[int] = None
) -> Iterator[CheckpointTuple]

List checkpoints that match the given criteria.

Parameters:

Name	Type	Description	Default
config	Optional[RunnableConfig]	Base configuration for filtering checkpoints.	required
filter	Optional[Dict[str, Any]]	Additional filtering criteria.	None
before	Optional[RunnableConfig]	List checkpoints created before this configuration.	None
limit	Optional[int]	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) -> Optional[Checkpoint]

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
Optional[Checkpoint]	Optional[Checkpoint]: The requested checkpoint, or None if not found.

aget_tuple async

aget_tuple(
    config: RunnableConfig,
) -> Optional[CheckpointTuple]

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
Optional[CheckpointTuple]	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: Optional[RunnableConfig],
    *,
    filter: Optional[Dict[str, Any]] = None,
    before: Optional[RunnableConfig] = None,
    limit: Optional[int] = None
) -> AsyncIterator[CheckpointTuple]

Asynchronously list checkpoints that match the given criteria.

Parameters:

Name	Type	Description	Default
config	Optional[RunnableConfig]	Base configuration for filtering checkpoints.	required
filter	Optional[Dict[str, Any]]	Additional filtering criteria for metadata.	None
before	Optional[RunnableConfig]	List checkpoints created before this configuration.	None
limit	Optional[int]	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: Optional[V], channel: ChannelProtocol
) -> 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	Optional[V]	The current version identifier (int, float, or str).	required
channel	ChannelProtocol	The channel being versioned.	required

Returns:

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

create_checkpoint

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

Create a checkpoint for the given channels.

Classes:

Name	Description
SerializerProtocol	Protocol for serialization and deserialization of objects.

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.

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
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	Optional[SerializerProtocol]	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[ConfigurableFieldSpec]	Define the configuration options for the checkpoint saver.

config_specs property

config_specs: list[ConfigurableFieldSpec]

Define the configuration options for the checkpoint saver.

Returns:

Type	Description
list[ConfigurableFieldSpec]	list[ConfigurableFieldSpec]: List of configuration field specs.

get_tuple

get_tuple(
    config: RunnableConfig,
) -> Optional[CheckpointTuple]

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
Optional[CheckpointTuple]	Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.

list

list(
    config: Optional[RunnableConfig],
    *,
    filter: Optional[dict[str, Any]] = None,
    before: Optional[RunnableConfig] = None,
    limit: Optional[int] = 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	Optional[RunnableConfig]	Base configuration for filtering checkpoints.	required
filter	Optional[dict[str, Any]]	Additional filtering criteria for metadata.	None
before	Optional[RunnableConfig]	List checkpoints created before this configuration.	None
limit	Optional[int]	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,
) -> Optional[CheckpointTuple]

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
Optional[CheckpointTuple]	Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.

alist async

alist(
    config: Optional[RunnableConfig],
    *,
    filter: Optional[dict[str, Any]] = None,
    before: Optional[RunnableConfig] = None,
    limit: Optional[int] = 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	Optional[RunnableConfig]	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) -> Optional[Checkpoint]

Fetch a checkpoint using the given configuration.

Parameters:

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

Returns:

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

aget async

aget(config: RunnableConfig) -> Optional[Checkpoint]

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
Optional[Checkpoint]	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[ConfigurableFieldSpec]	Define the configuration options for the checkpoint saver.

config_specs property

config_specs: list[ConfigurableFieldSpec]

Define the configuration options for the checkpoint saver.

Returns:

Type	Description
list[ConfigurableFieldSpec]	list[ConfigurableFieldSpec]: 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,
) -> Optional[CheckpointTuple]

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
Optional[CheckpointTuple]	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: Optional[RunnableConfig],
    *,
    filter: Optional[dict[str, Any]] = None,
    before: Optional[RunnableConfig] = None,
    limit: Optional[int] = 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	Optional[RunnableConfig]	The config to use for listing the checkpoints.	required
filter	Optional[dict[str, Any]]	Additional filtering criteria for metadata. Defaults to None.	None
before	Optional[RunnableConfig]	If provided, only checkpoints before the specified checkpoint ID are returned. Defaults to None.	None
limit	Optional[int]	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,
) -> Optional[CheckpointTuple]

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: Optional[RunnableConfig],
    *,
    filter: Optional[dict[str, Any]] = None,
    before: Optional[RunnableConfig] = None,
    limit: Optional[int] = 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: Optional[str], channel: ChannelProtocol
) -> 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
channel	BaseChannel	The channel being versioned.	required

Returns:

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

get

get(config: RunnableConfig) -> Optional[Checkpoint]

Fetch a checkpoint using the given configuration.

Parameters:

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

Returns:

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

aget async

aget(config: RunnableConfig) -> Optional[Checkpoint]

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
Optional[Checkpoint]	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[ConfigurableFieldSpec]

Define the configuration options for the checkpoint saver.

Returns:

Type	Description
list[ConfigurableFieldSpec]	list[ConfigurableFieldSpec]: 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,
) -> Optional[CheckpointTuple]

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
Optional[CheckpointTuple]	Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.

list

list(
    config: Optional[RunnableConfig],
    *,
    filter: Optional[dict[str, Any]] = None,
    before: Optional[RunnableConfig] = None,
    limit: Optional[int] = 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	Optional[RunnableConfig]	Base configuration for filtering checkpoints.	required
filter	Optional[dict[str, Any]]	Additional filtering criteria for metadata.	None
before	Optional[RunnableConfig]	If provided, only checkpoints before the specified checkpoint ID are returned. Defaults to None.	None
limit	Optional[int]	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,
) -> Optional[CheckpointTuple]

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
Optional[CheckpointTuple]	Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.

alist async

alist(
    config: Optional[RunnableConfig],
    *,
    filter: Optional[dict[str, Any]] = None,
    before: Optional[RunnableConfig] = None,
    limit: Optional[int] = 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	Optional[RunnableConfig]	Base configuration for filtering checkpoints.	required
filter	Optional[dict[str, Any]]	Additional filtering criteria for metadata.	None
before	Optional[RunnableConfig]	If provided, only checkpoints before the specified checkpoint ID are returned. Defaults to None.	None
limit	Optional[int]	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: Optional[str], channel: ChannelProtocol
) -> 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
channel	BaseChannel	The channel being versioned.	required

Returns:

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

get

get(config: RunnableConfig) -> Optional[Checkpoint]

Fetch a checkpoint using the given configuration.

Parameters:

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

Returns:

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

aget async

aget(config: RunnableConfig) -> Optional[Checkpoint]

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
Optional[Checkpoint]	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[ConfigurableFieldSpec]	Define the configuration options for the checkpoint saver.

config_specs property

config_specs: list[ConfigurableFieldSpec]

Define the configuration options for the checkpoint saver.

Returns:

Type	Description
list[ConfigurableFieldSpec]	list[ConfigurableFieldSpec]: 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: Optional[RunnableConfig],
    *,
    filter: Optional[dict[str, Any]] = None,
    before: Optional[RunnableConfig] = None,
    limit: Optional[int] = 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	Optional[RunnableConfig]	The config to use for listing the checkpoints.	required
filter	Optional[dict[str, Any]]	Additional filtering criteria for metadata. Defaults to None.	None
before	Optional[RunnableConfig]	If provided, only checkpoints before the specified checkpoint ID are returned. Defaults to None.	None
limit	Optional[int]	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,
) -> Optional[CheckpointTuple]

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
Optional[CheckpointTuple]	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) -> Optional[Checkpoint]

Fetch a checkpoint using the given configuration.

Parameters:

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

Returns:

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

aget async

aget(config: RunnableConfig) -> Optional[Checkpoint]

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
Optional[Checkpoint]	Optional[Checkpoint]: The requested checkpoint, or None if not found.

aget_tuple async

aget_tuple(
    config: RunnableConfig,
) -> Optional[CheckpointTuple]

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
Optional[CheckpointTuple]	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: Optional[RunnableConfig],
    *,
    filter: Optional[Dict[str, Any]] = None,
    before: Optional[RunnableConfig] = None,
    limit: Optional[int] = None
) -> AsyncIterator[CheckpointTuple]

Asynchronously list checkpoints that match the given criteria.

Parameters:

Name	Type	Description	Default
config	Optional[RunnableConfig]	Base configuration for filtering checkpoints.	required
filter	Optional[Dict[str, Any]]	Additional filtering criteria for metadata.	None
before	Optional[RunnableConfig]	List checkpoints created before this configuration.	None
limit	Optional[int]	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[ConfigurableFieldSpec]	Define the configuration options for the checkpoint saver.

config_specs property

config_specs: list[ConfigurableFieldSpec]

Define the configuration options for the checkpoint saver.

Returns:

Type	Description
list[ConfigurableFieldSpec]	list[ConfigurableFieldSpec]: List of configuration field specs.

from_conn_string async classmethod

from_conn_string(
    conn_string: str,
    *,
    pipeline: bool = False,
    serde: Optional[SerializerProtocol] = 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: Optional[RunnableConfig],
    *,
    filter: Optional[dict[str, Any]] = None,
    before: Optional[RunnableConfig] = None,
    limit: Optional[int] = 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	Optional[RunnableConfig]	Base configuration for filtering checkpoints.	required
filter	Optional[dict[str, Any]]	Additional filtering criteria for metadata.	None
before	Optional[RunnableConfig]	If provided, only checkpoints before the specified checkpoint ID are returned. Defaults to None.	None
limit	Optional[int]	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,
) -> Optional[CheckpointTuple]

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
Optional[CheckpointTuple]	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: Optional[RunnableConfig],
    *,
    filter: Optional[dict[str, Any]] = None,
    before: Optional[RunnableConfig] = None,
    limit: Optional[int] = 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	Optional[RunnableConfig]	Base configuration for filtering checkpoints.	required
filter	Optional[dict[str, Any]]	Additional filtering criteria for metadata.	None
before	Optional[RunnableConfig]	If provided, only checkpoints before the specified checkpoint ID are returned. Defaults to None.	None
limit	Optional[int]	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,
) -> Optional[CheckpointTuple]

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
Optional[CheckpointTuple]	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) -> Optional[Checkpoint]

Fetch a checkpoint using the given configuration.

Parameters:

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

Returns:

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

aget async

aget(config: RunnableConfig) -> Optional[Checkpoint]

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
Optional[Checkpoint]	Optional[Checkpoint]: The requested checkpoint, or None if not found.