Implementation:Langchain ai Langgraph BaseStore And InMemoryStore

Attribute	Value
Page Type	Implementation (API Doc)
Library	langgraph (checkpoint, checkpoint-postgres)
Workflow	Persistence_and_Memory_Setup
Principle	Langchain_ai_Langgraph_Store_Configuration
Implementation	BaseStore_And_InMemoryStore
Source	`libs/checkpoint/langgraph/store/base/__init__.py:L700-1253`, `libs/checkpoint/langgraph/store/memory/__init__.py:L136-477`, `libs/checkpoint-postgres/langgraph/store/postgres/base.py:L640-805`

Overview

BaseStore is the abstract base class defining the persistent key-value store interface. InMemoryStore is the dictionary-backed implementation for development and testing. PostgresStore is the PostgreSQL-backed implementation for production, with optional vector search via pgvector and TTL support.

Description

All store implementations share the same high-level API defined by BaseStore: get, put, search, delete, and list_namespaces (plus async variants). The underlying execution model is batch-based: all convenience methods delegate to batch() or abatch(), which processes a list of operation objects in a single call.

Usage

Import

from langgraph.store.memory import InMemoryStore
from langgraph.store.postgres import PostgresStore

Code Reference

Source Location

Class	File	Lines
`BaseStore`	`libs/checkpoint/langgraph/store/base/__init__.py`	L700-1253
`InMemoryStore`	`libs/checkpoint/langgraph/store/memory/__init__.py`	L136-477
`PostgresStore`	`libs/checkpoint-postgres/langgraph/store/postgres/base.py`	L640-805

BaseStore Signature

class BaseStore(ABC):
    supports_ttl: bool = False
    ttl_config: TTLConfig | None = None

    @abstractmethod
    def batch(self, ops: Iterable[Op]) -> list[Result]: ...

    @abstractmethod
    async def abatch(self, ops: Iterable[Op]) -> list[Result]: ...

    def get(
        self,
        namespace: tuple[str, ...],
        key: str,
        *,
        refresh_ttl: bool | None = None,
    ) -> Item | None: ...

    def search(
        self,
        namespace_prefix: tuple[str, ...],
        /,
        *,
        query: str | None = None,
        filter: dict[str, Any] | None = None,
        limit: int = 10,
        offset: int = 0,
        refresh_ttl: bool | None = None,
    ) -> list[SearchItem]: ...

    def put(
        self,
        namespace: tuple[str, ...],
        key: str,
        value: dict[str, Any],
        index: Literal[False] | list[str] | None = None,
        *,
        ttl: float | None | NotProvided = NOT_PROVIDED,
    ) -> None: ...

    def delete(self, namespace: tuple[str, ...], key: str) -> None: ...

    def list_namespaces(
        self,
        *,
        prefix: NamespacePath | None = None,
        suffix: NamespacePath | None = None,
        max_depth: int | None = None,
        limit: int = 100,
        offset: int = 0,
    ) -> list[tuple[str, ...]]: ...

Async variants: aget, asearch, aput, adelete, alist_namespaces.

InMemoryStore Signature

class InMemoryStore(BaseStore):
    def __init__(self, *, index: IndexConfig | None = None) -> None: ...

PostgresStore Signature

class PostgresStore(BaseStore, BasePostgresStore[_pg_internal.Conn]):
    supports_ttl: bool = True

    def __init__(
        self,
        conn: _pg_internal.Conn,
        *,
        pipe: Pipeline | None = None,
        deserializer: Callable[[bytes | orjson.Fragment], dict[str, Any]] | None = None,
        index: PostgresIndexConfig | None = None,
        ttl: TTLConfig | None = None,
    ) -> None: ...

    @classmethod
    @contextmanager
    def from_conn_string(
        cls,
        conn_string: str,
        *,
        pipeline: bool = False,
        pool_config: PoolConfig | None = None,
        index: PostgresIndexConfig | None = None,
        ttl: TTLConfig | None = None,
    ) -> Iterator[PostgresStore]: ...

Supporting Types

class Item:
    value: dict[str, Any]
    key: str
    namespace: tuple[str, ...]
    created_at: datetime
    updated_at: datetime

class SearchItem(Item):
    score: float | None

class IndexConfig(TypedDict, total=False):
    dims: int
    embed: Embeddings | EmbeddingsFunc | AEmbeddingsFunc | str
    fields: list[str] | None

class TTLConfig(TypedDict, total=False):
    refresh_on_read: bool
    default_ttl: float | None
    sweep_interval_minutes: int | None

I/O Contract

Method	Input	Output	Description
`get`	`namespace: tuple[str, ...], key: str`	None	Retrieve a single item by namespace and key
`put`	`namespace, key, value, index, ttl`	`None`	Store or update an item
`search`	`namespace_prefix, query, filter, limit, offset`	`list[SearchItem]`	Search items within a namespace prefix
`delete`	`namespace, key`	`None`	Delete an item
`list_namespaces`	`prefix, suffix, max_depth, limit, offset`	`list[tuple[str, ...]]`	List namespaces matching criteria

Usage Examples

Basic InMemoryStore

from langgraph.store.memory import InMemoryStore

store = InMemoryStore()

# Store data
store.put(("users", "123"), "prefs", {"theme": "dark"})

# Retrieve data
item = store.get(("users", "123"), "prefs")
print(item.value)  # {"theme": "dark"}

# Search within a namespace
results = store.search(("users",))

InMemoryStore with Vector Search

from langchain.embeddings import init_embeddings
from langgraph.store.memory import InMemoryStore

store = InMemoryStore(
    index={
        "dims": 1536,
        "embed": init_embeddings("openai:text-embedding-3-small"),
        "fields": ["text"],
    }
)

# Store documents
store.put(("docs",), "doc1", {"text": "Python tutorial"})
store.put(("docs",), "doc2", {"text": "TypeScript guide"})

# Semantic search
results = store.search(("docs",), query="programming languages")
for item in results:
    print(item.key, item.score)

PostgresStore with from_conn_string

from langgraph.store.postgres import PostgresStore

conn_string = "postgresql://user:pass@localhost:5432/dbname"

with PostgresStore.from_conn_string(conn_string) as store:
    store.setup()  # Run migrations (required on first use)

    store.put(("users", "123"), "prefs", {"theme": "dark"})
    item = store.get(("users", "123"), "prefs")

PostgresStore with Vector Search and TTL

from langchain.embeddings import init_embeddings
from langgraph.store.postgres import PostgresStore

with PostgresStore.from_conn_string(
    conn_string,
    index={
        "dims": 1536,
        "embed": init_embeddings("openai:text-embedding-3-small"),
        "fields": ["text"],
    },
    ttl={"default_ttl": 60, "refresh_on_read": True},
) as store:
    store.setup()

    # Items expire after 60 minutes of inactivity
    store.put(("cache",), "result1", {"text": "cached value"})

    # Override TTL per item
    store.put(("cache",), "result2", {"text": "short-lived"}, ttl=5)

Compiling a Graph with Store

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.store.memory import InMemoryStore
from langgraph.graph import StateGraph

checkpointer = InMemorySaver()
store = InMemoryStore()

graph = builder.compile(checkpointer=checkpointer, store=store)

Related Pages

Page Connections

Double-click a node to navigate. Hold to expand connections.

Principle

Implementation

Heuristic

Environment