Implementation:Langchain ai Langgraph AsyncSqliteSaver

Knowledge Sources	Langchain_ai_Langgraph
Domains	Checkpointing, SQLite
Last Updated	2026-02-11 16:00 GMT

Overview

`AsyncSqliteSaver` is an asynchronous checkpoint saver that stores LangGraph checkpoints in a SQLite database using the `aiosqlite` driver.

Description

`AsyncSqliteSaver` extends `BaseCheckpointSaver[str]` to provide an asynchronous interface for persisting graph checkpoints and intermediate writes in SQLite. It uses `aiosqlite` for non-blocking database operations and manages two tables: `checkpoints` for storing the main checkpoint state, metadata, and parent lineage, and `writes` for persisting intermediate task writes linked to specific checkpoints.

The class handles automatic database setup with WAL (Write-Ahead Logging) journal mode for better concurrent read performance. Schema creation is performed lazily on first use via the `setup()` method, which is called automatically before any data access. Checkpoint data is serialized using either a custom `SerializerProtocol` or the built-in `JsonPlusSerializer`, and metadata is stored as JSON-encoded blobs.

`AsyncSqliteSaver` also provides synchronous wrapper methods (`get_tuple`, `list`, `put`, `put_writes`) that delegate to the async implementations via `asyncio.run_coroutine_threadsafe`, allowing use from background threads. Thread safety is ensured through an `asyncio.Lock`, and the class includes validation to prevent synchronous calls from the main event loop thread.

Usage

Use `AsyncSqliteSaver` for lightweight, file-based checkpoint persistence in async LangGraph applications. It is well suited for development, testing, prototyping, and small-scale deployments where the simplicity of SQLite is preferred over setting up a full database server. For production workloads with high write concurrency, consider `AsyncPostgresSaver` instead.

Code Reference

Source Location

• Repository: Langchain_ai_Langgraph
• File: libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/aio.py
• Lines: 1-631

Signature

class AsyncSqliteSaver(BaseCheckpointSaver[str]):
    def __init__(
        self,
        conn: aiosqlite.Connection,
        *,
        serde: SerializerProtocol | None = None,
    ):

Import

from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver

I/O Contract

Inputs

Name	Type	Required	Description
conn	aiosqlite.Connection	Yes	An aiosqlite async database connection.
serde	None	No	Custom serializer for encoding/decoding checkpoint data. Defaults to `JsonPlusSerializer`.

Outputs

Name	Type	Description
AsyncSqliteSaver	AsyncSqliteSaver	An instance of the async SQLite checkpoint saver, ready for use.

Key Methods

Method	Description
setup()	Creates checkpoint and writes tables with WAL journal mode. Called automatically on first use.
aget_tuple(config)	Retrieves a checkpoint tuple by thread ID and optional checkpoint ID, including pending writes.
alist(config, ...)	Lists checkpoint tuples matching filter criteria, ordered by checkpoint ID descending.
aput(config, checkpoint, metadata, new_versions)	Saves a checkpoint with serialized data and metadata.
aput_writes(config, writes, task_id, task_path)	Stores intermediate writes linked to a checkpoint.
from_conn_string(conn_string)	Async classmethod context manager that creates an instance from a SQLite file path or `":memory:"`.

Usage Examples

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")

    # Using the context manager (recommended)
    async with AsyncSqliteSaver.from_conn_string("checkpoints.db") as memory:
        graph = builder.compile(checkpointer=memory)
        result = await graph.ainvoke(1, {"configurable": {"thread_id": "thread-1"}})
        print(result)  # Output: 2

asyncio.run(main())

# Using a direct aiosqlite connection
import aiosqlite

async def raw_usage():
    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)

Related Pages

• Principle:Langchain_ai_Langgraph_Async_Checkpoint_Persistence