Implementation:Langchain ai Langgraph AsyncSqliteStore

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

Overview

`AsyncSqliteStore` is an asynchronous SQLite-backed key-value store with optional vector search via `sqlite-vec` and TTL-based item expiration.

Description

`AsyncSqliteStore` extends both `AsyncBatchedBaseStore` and `BaseSqliteStore` to provide a fully asynchronous interface for storing, retrieving, and searching key-value data in SQLite. Items are organized by namespace tuples and identified by string keys. The store uses `aiosqlite` for non-blocking database access and supports transactional operations through its cursor management.

The store provides optional vector search capabilities through the `sqlite-vec` extension. When an `index` configuration is provided with embedding dimensions and an embedding model, the store creates a `store_vectors` table and automatically generates vector embeddings for specified document fields. This enables semantic similarity search with configurable distance metrics (cosine, L2, inner product) through the standard `asearch` interface.

`AsyncSqliteStore` also supports TTL (time-to-live) for automatic item expiration. Items can be assigned expiration timestamps and a `ttl_minutes` value that controls their lifetime. A background sweeper task, started via `start_ttl_sweeper()`, periodically removes expired items. The TTL feature also supports refresh-on-read behavior, extending an item's expiration each time it is accessed. All database operations are protected by an `asyncio.Lock` to ensure thread safety.

Usage

Use `AsyncSqliteStore` when you need a lightweight, file-based key-value store for async LangGraph applications. It is ideal for development, testing, and small-scale deployments where setting up PostgreSQL is not warranted. Use the vector search feature for semantic retrieval of stored items in memory-constrained environments.

Code Reference

Source Location

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

Signature

class AsyncSqliteStore(AsyncBatchedBaseStore, BaseSqliteStore):
    def __init__(
        self,
        conn: aiosqlite.Connection,
        *,
        deserializer: Callable[[bytes | str | orjson.Fragment], dict[str, Any]] | None = None,
        index: SqliteIndexConfig | None = None,
        ttl: TTLConfig | None = None,
    ):

Import

from langgraph.store.sqlite import AsyncSqliteStore

I/O Contract

Inputs

Name	Type	Required	Description
conn	aiosqlite.Connection	Yes	An aiosqlite async database connection.
deserializer	None	No	Optional custom deserializer for stored values.
index	None	No	Optional vector search configuration with dims, embed model, and fields.
ttl	None	No	Optional time-to-live configuration for automatic item expiration.

Outputs

Name	Type	Description
AsyncSqliteStore	AsyncSqliteStore	An instance of the async SQLite store, ready for use after calling `setup()`.

Key Methods

Method	Description
setup()	Creates store tables, applies migrations, loads sqlite-vec extension, and sets up vector tables if configured.
abatch(ops)	Executes a batch of store operations (Get, Put, Search, ListNamespaces) asynchronously within a transaction.
from_conn_string(conn_string, ...)	Async classmethod context manager that creates an instance from a SQLite file path or `":memory:"`.
sweep_ttl()	Deletes expired store items based on TTL. Returns the number of deleted items.
start_ttl_sweeper()	Starts a background async task that periodically removes expired items.
stop_ttl_sweeper(timeout)	Gracefully stops the TTL sweeper task.

Usage Examples

from langgraph.store.sqlite import AsyncSqliteStore

# Basic key-value storage
async with AsyncSqliteStore.from_conn_string(":memory:") as store:
    await store.setup()

    await store.aput(("users", "123"), "prefs", {"theme": "dark"})
    item = await store.aget(("users", "123"), "prefs")

# With vector search
from langchain_openai import OpenAIEmbeddings

async with AsyncSqliteStore.from_conn_string(
    ":memory:",
    index={
        "dims": 1536,
        "embed": OpenAIEmbeddings(),
        "fields": ["text"],
    },
) as store:
    await store.setup()

    await store.aput(("docs",), "doc1", {"text": "Python tutorial"})
    await store.aput(("docs",), "doc2", {"text": "TypeScript guide"})
    await store.aput(("docs",), "doc3", {"text": "Other guide"}, index=False)

    results = await store.asearch(("docs",), query="programming guides", limit=2)

# With TTL configuration
async with AsyncSqliteStore.from_conn_string(
    "store.db",
    ttl={"default_ttl": 60, "refresh_on_read": True, "sweep_interval_minutes": 5},
) as store:
    await store.setup()
    await store.start_ttl_sweeper()
    # Items will expire after 60 minutes, refreshed on read

Related Pages

• Principle:Langchain_ai_Langgraph_Async_Store_Operations