A checkpoint saver that stores checkpoints in a SQLite database.
SqliteSaver(
self,
conn: sqlite3.Connection,
*,
serde: SerializerProtocol | None = None
)| Name | Type | Description |
|---|---|---|
conn* | sqlite3.Connection | |
serde | Optional[SerializerProtocol] | Default: None |
| Name | Type |
|---|---|
| conn | sqlite3.Connection |
| serde | SerializerProtocol | None |
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.
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)
The SQLite database connection.
The serializer to use for serializing and deserializing checkpoints. Defaults to JsonPlusSerializerCompat.
Create a new SqliteSaver instance from a connection string.
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.
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.
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.
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).
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).
Store intermediate writes linked to a checkpoint.
This method saves intermediate writes associated with a checkpoint to the SQLite database.
Delete all checkpoints and writes associated with a thread ID.
Fast-path override of BaseCheckpointSaver.get_delta_channel_history.
Two-stage query:
Stage 1 (paged): newest-first slice of checkpoints returning
(checkpoint_id, parent_checkpoint_id, type, checkpoint) per
ancestor. Sqlite has no JSONB, so we ship the full serialized
checkpoint blob and inspect channel_values in Python. Pages
newest-first by checkpoint_id with a < cursor predicate;
page size is DELTA_PAGE_SIZE. Stops paging when every channel
has found its seed or the chain is exhausted.
Stage 2 (per-channel UNION ALL): one branch per channel reading
writes filtered to that channel's specific chain_cids. No
separate seed-blob fetch — sqlite stores channel_values inline
in the checkpoint blob, so seeds come back from stage 1.
Get a checkpoint tuple from the database asynchronously.
List checkpoints from the database asynchronously.
Save a checkpoint to the database asynchronously.
Generate the next version ID for a channel.
This method creates a new version identifier for a channel based on its current version.