Skip to main content

Documentation Index

Fetch the complete documentation index at: https://mintlify.com/langchain-ai/langgraph/llms.txt

Use this file to discover all available pages before exploring further.

SqliteSaver is a checkpoint saver that stores checkpoints in a SQLite database. It provides a lightweight, file-based persistence solution for LangGraph agents.

​
Overview

SqliteSaver is designed for:
  • Lightweight, synchronous use cases
  • Demos and small projects
  • Local development and testing
  • Single-threaded applications
SqliteSaver does not scale to multiple threads. For production workloads or async applications, consider using AsyncSqliteSaver or PostgresSaver.

​
Class Definition

from langgraph.checkpoint.sqlite import SqliteSaver

class SqliteSaver(BaseCheckpointSaver[str]):
    """A checkpoint saver that stores checkpoints in a SQLite database."""
Source: langgraph.checkpoint.sqlite.__init__:38

​
Installation

SqliteSaver is included in the base langgraph-checkpoint-sqlite package: 
pip install langgraph-checkpoint-sqlite

​
Constructor

def __init__(
    self,
    conn: sqlite3.Connection,
    *,
    serde: SerializerProtocol | None = None,
) -> None

​
Parameters

  • conn (sqlite3.Connection): The SQLite database connection
  • serde (SerializerProtocol, optional): The serializer for encoding/decoding checkpoints. Defaults to JsonPlusSerializer
Source: langgraph.checkpoint.sqlite.__init__:78

​
Usage

​
Basic Setup

import sqlite3
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.graph import StateGraph

# Create a graph
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 database connection and checkpointer
# Note: check_same_thread=False is OK as the implementation uses a lock
conn = sqlite3.connect("checkpoints.sqlite", check_same_thread=False)
memory = SqliteSaver(conn)

# Compile graph with checkpointer
graph = builder.compile(checkpointer=memory)

# Use the graph
config = {"configurable": {"thread_id": "1"}}
result = graph.invoke(3, config)
print(result)  # Output: 4

# Get state
state = graph.get_state(config)
print(state)
# StateSnapshot(values=4, next=(), config={...}, parent_config=None)

​
Using from_conn_string

from langgraph.checkpoint.sqlite import SqliteSaver

# In-memory database
with SqliteSaver.from_conn_string(":memory:") as memory:
    graph = builder.compile(checkpointer=memory)
    config = {"configurable": {"thread_id": "1"}}
    result = graph.invoke(3, config)

# Persistent database file
with SqliteSaver.from_conn_string("checkpoints.sqlite") as memory:
    graph = builder.compile(checkpointer=memory)
    config = {"configurable": {"thread_id": "1"}}
    result = graph.invoke(3, config)

​
Class Methods

​
from_conn_string

@classmethod
@contextmanager
def from_conn_string(cls, conn_string: str) -> Iterator[SqliteSaver]
Create a new SqliteSaver instance from a connection string. Parameters:
  • conn_string (str): The SQLite connection string. Use :memory: for in-memory database or a file path for persistent storage
Returns:
  • Iterator[SqliteSaver]: A context manager yielding a SqliteSaver instance
Example: 
# In-memory
with SqliteSaver.from_conn_string(":memory:") as saver:
    # Use saver
    pass

# Persistent file
with SqliteSaver.from_conn_string("checkpoints.sqlite") as saver:
    # Use saver
    pass
Source: langgraph.checkpoint.sqlite.__init__:90

​
Instance Methods

​
setup

def setup(self) -> None
Set up the checkpoint database. Creates the necessary tables if they don’t exist. Note: This method is called automatically when needed and should not be called directly by users. Source: langgraph.checkpoint.sqlite.__init__:122

​
get_tuple

def get_tuple(self, config: RunnableConfig) -> CheckpointTuple | None
Get a checkpoint tuple from the database. Parameters:
  • config (RunnableConfig): Configuration containing thread_id and optionally checkpoint_id
Returns:
  • CheckpointTuple | None: The checkpoint tuple, or None if not found
Example: 
# Get latest checkpoint
config = {"configurable": {"thread_id": "1"}}
checkpoint_tuple = memory.get_tuple(config)

# Get specific checkpoint
config = {
    "configurable": {
        "thread_id": "1",
        "checkpoint_ns": "",
        "checkpoint_id": "1ef4f797-8335-6428-8001-8a1503f9b875",
    }
}
checkpoint_tuple = memory.get_tuple(config)
Source: langgraph.checkpoint.sqlite.__init__:184

​
list

def list(
    self,
    config: RunnableConfig | None,
    *,
    filter: dict[str, Any] | None = None,
    before: RunnableConfig | None = None,
    limit: int | None = None,
) -> Iterator[CheckpointTuple]
List checkpoints from the database. Parameters:
  • config (RunnableConfig | None): Base configuration for filtering
  • filter (dict[str, Any] | None): Additional metadata filtering criteria
  • before (RunnableConfig | None): Only return checkpoints before this checkpoint ID
  • limit (int | None): Maximum number of checkpoints to return
Returns:
  • Iterator[CheckpointTuple]: Iterator of checkpoint tuples, ordered by checkpoint ID (newest first)
Example: 
# List all checkpoints for a thread
config = {"configurable": {"thread_id": "1"}}
checkpoints = list(memory.list(config))

# List with limit
checkpoints = list(memory.list(config, limit=5))

# List checkpoints before a specific checkpoint
before_config = {
    "configurable": {
        "checkpoint_id": "1ef4f797-8335-6428-8001-8a1503f9b875"
    }
}
checkpoints = list(memory.list(config, before=before_config))
Source: langgraph.checkpoint.sqlite.__init__:288

​
put

def put(
    self,
    config: RunnableConfig,
    checkpoint: Checkpoint,
    metadata: CheckpointMetadata,
    new_versions: ChannelVersions,
) -> RunnableConfig
Save a checkpoint to the database. Parameters:
  • config (RunnableConfig): Configuration for the checkpoint
  • checkpoint (Checkpoint): The checkpoint to save
  • metadata (CheckpointMetadata): Additional metadata
  • new_versions (ChannelVersions): New channel versions
Returns:
  • RunnableConfig: Updated configuration with the new checkpoint ID
Example: 
config = {"configurable": {"thread_id": "1", "checkpoint_ns": ""}}
checkpoint = {
    "v": 1,
    "ts": "2024-05-04T06:32:42.235444+00:00",
    "id": "1ef4f797-8335-6428-8001-8a1503f9b875",
    "channel_values": {"key": "value"},
    "channel_versions": {},
    "versions_seen": {},
}
metadata = {"source": "input", "step": 1}
saved_config = memory.put(config, checkpoint, metadata, {})
Source: langgraph.checkpoint.sqlite.__init__:380

​
put_writes

def put_writes(
    self,
    config: RunnableConfig,
    writes: Sequence[tuple[str, Any]],
    task_id: str,
    task_path: str = "",
) -> None
Store intermediate writes linked to a checkpoint. Parameters:
  • config (RunnableConfig): Configuration of the related checkpoint
  • writes (Sequence[tuple[str, Any]]): List of (channel, value) pairs to store
  • task_id (str): Identifier for the task creating the writes
  • task_path (str): Path of the task (default: "")
Source: langgraph.checkpoint.sqlite.__init__:438

​
delete_thread

def delete_thread(self, thread_id: str) -> None
Delete all checkpoints and writes associated with a thread ID. Parameters:
  • thread_id (str): The thread ID to delete
Example: 
memory.delete_thread("thread-1")
Source: langgraph.checkpoint.sqlite.__init__:477

​
get_next_version

def get_next_version(self, current: str | None, channel: None) -> str
Generate the next version ID for a channel. Parameters:
  • current (str | None): The current version identifier
  • channel (None): Deprecated parameter
Returns:
  • str: The next version identifier (format: "{version:032}.{random:016}")
Source: langgraph.checkpoint.sqlite.__init__:537

​
Database Schema

SqliteSaver creates two tables:

​
checkpoints table

CREATE TABLE checkpoints (
    thread_id TEXT NOT NULL,
    checkpoint_ns TEXT NOT NULL DEFAULT '',
    checkpoint_id TEXT NOT NULL,
    parent_checkpoint_id TEXT,
    type TEXT,
    checkpoint BLOB,
    metadata BLOB,
    PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id)
);

​
writes table

CREATE TABLE writes (
    thread_id TEXT NOT NULL,
    checkpoint_ns TEXT NOT NULL DEFAULT '',
    checkpoint_id TEXT NOT NULL,
    task_id TEXT NOT NULL,
    idx INTEGER NOT NULL,
    channel TEXT NOT NULL,
    type TEXT,
    value BLOB,
    PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id, task_id, idx)
);

​
AsyncSqliteSaver

For async applications, use AsyncSqliteSaver: 
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)
        result = await graph.ainvoke(1, {"configurable": {"thread_id": "1"}})
        print(result)  # Output: 2

asyncio.run(main())
AsyncSqliteSaver requires the aiosqlite package:
pip install aiosqlite
Source: langgraph.checkpoint.sqlite.aio:31

​
Limitations

  • Not suitable for production workloads with high concurrency
  • Does not scale to multiple threads (use AsyncSqliteSaver or PostgresSaver instead)
  • SQLite’s write performance is limited compared to dedicated databases
  • File locking can cause issues in distributed environments

​
See Also

Build docs developers (and LLMs) love

Get started for freeTalk to us