Store (long-term memory) — API reference
Store (long-term memory) — API reference
Section titled “Store (long-term memory) — API reference”Verified against langgraph==1.2.2 (modules: langgraph.store.base, langgraph.store.memory).
Checkpointers give you short-term memory tied to a single thread_id. A Store gives you long-term memory that lives outside any thread — shared across conversations, users, and graph runs. Same backend pattern as checkpointers: one abstract base, multiple implementations.
Minimal runnable example
Section titled “Minimal runnable example”from langgraph.store.memory import InMemoryStore
store = InMemoryStore()
store.put(("users", "alice"), "prefs", {"theme": "dark", "lang": "en"})item = store.get(("users", "alice"), "prefs")print(item.value) # {'theme': 'dark', 'lang': 'en'}print(item.namespace) # ('users', 'alice')print(item.key) # 'prefs'
for hit in store.search(("users",), filter={"theme": "dark"}): print(hit.key, hit.value)Wire a store into a graph so nodes and tools can read/write to it:
from langgraph.graph import StateGraph, STARTfrom langgraph.store.memory import InMemoryStore
store = InMemoryStore()graph = ( StateGraph(State) .add_node("recall", recall_fn) .add_edge(START, "recall") .compile(store=store))Available backends
Section titled “Available backends”| Backend | Import | Persists? | Vector search | TTL |
|---|---|---|---|---|
InMemoryStore | langgraph.store.memory | No | Yes (numpy if installed) | Optional |
PostgresStore | langgraph.store.postgres1 | Yes | Yes (pgvector) | Yes |
AsyncPostgresStore | langgraph.store.postgres.aio1 | Yes | Yes (pgvector) | Yes |
AsyncBatchedBaseStore | langgraph.store.base.batch | Adapter | Same as wrapped | Same as wrapped |
1 Ships in the separate langgraph-checkpoint-postgres package — the same package as PostgresSaver.
Data model
Section titled “Data model”namespace: tuple[str, ...]— hierarchical path (e.g.,("users", "123", "memories")). The prefix is used for listing and scoped searches.key: str— unique within a namespace.value: dict[str, Any]— JSON-serializable payload. Keys are filterable.Item— returned byget/list_namespaces. Fields:value,key,namespace,created_at,updated_at.SearchItem(Item)— returned bysearch. Addsscore: float | None.
Any of these operations can raise InvalidNamespaceError (e.g., empty tuple, empty string label, "." in a label, or "langgraph" as the root segment).
BaseStore surface
Section titled “BaseStore surface”All methods have sync and a-prefixed async variants.
# Syncstore.get(namespace, key, *, refresh_ttl=None) -> Item | Nonestore.put(namespace, key, value, index=None, *, ttl=NOT_PROVIDED) -> Nonestore.delete(namespace, key) -> Nonestore.search( namespace_prefix, /, *, query=None, filter=None, limit=10, offset=0, refresh_ttl=None,) -> list[SearchItem]store.list_namespaces( *, prefix=None, suffix=None, max_depth=None, limit=100, offset=0,) -> list[tuple[str, ...]]store.batch(ops: Iterable[Op]) -> list[Result]
# Async equivalents — same signatures with awaitawait store.aget(namespace, key, *, refresh_ttl=None) -> Item | Noneawait store.aput(namespace, key, value, index=None, *, ttl=NOT_PROVIDED) -> Noneawait store.adelete(namespace, key) -> Noneawait store.asearch(namespace_prefix, /, *, query=None, filter=None, limit=10, offset=0, refresh_ttl=None) -> list[SearchItem]await store.alist_namespaces(*, prefix=None, suffix=None, max_depth=None, limit=100, offset=0) -> list[tuple[str, ...]]await store.abatch(ops: Iterable[Op]) -> list[Result]Under the hood, every single-item method funnels through batch/abatch. Submit mixed GetOp, PutOp, SearchOp, ListNamespacesOp for a single round-trip.
put() — details
Section titled “put() — details”store.put( namespace: tuple[str, ...], key: str, value: dict[str, Any], index: Literal[False] | list[str] | None = None, *, ttl: float | None | NotProvided = NOT_PROVIDED,) -> Noneindex=None— use the fields you configured on the store (or none if it is not indexed).index=False— skip embedding for this item even if the store is indexed.index=["metadata.title", "chapters[*].content"]— path selectors. Supports:- dot-separated nesting (
"a.b.c"), [*]for every array element (each embedded separately),[0]/[-1]for a specific index or the last element.
- dot-separated nesting (
ttl— minutes until expiry (not seconds). RaisesNotImplementedErrorif you pass a value and the backend hassupports_ttl = False.
# Store a regular item (uses store's default index config)store.put(("docs",), "d1", {"text": "Python tutorial", "lang": "python"})
# TTL: item expires after 30 minutes of inactivitystore.put(("cache",), "result-xyz", {"data": "..."}, ttl=30)
# Skip embedding for this item even if the store has vector searchstore.put(("docs",), "draft", {"text": "WIP..."}, index=False)
# Embed only specific fields, overriding the store's default fieldsstore.put(("docs",), "article", {"title": "Guide", "body": "...", "meta": "..."}, index=["title", "body"])
# Async variant — identical signatureawait store.aput(("docs",), "d1", {"text": "Python tutorial"})await store.aput(("cache",), "result-xyz", {"data": "..."}, ttl=30)await store.aput(("docs",), "draft", {"text": "WIP..."}, index=False)get() — details
Section titled “get() — details”store.get( namespace: tuple[str, ...], key: str, *, refresh_ttl: bool | None = None,) -> Item | None- Returns
Noneif the key does not exist. refresh_ttl=Trueresets the TTL countdown for the item on each access — useful for “last accessed” cache semantics.refresh_ttl=None(default) falls back to the store’sTTLConfig.refresh_on_readsetting (defaultTrue).
item = store.get(("users", "alice"), "prefs")if item: print(item.value) # {'theme': 'dark'} print(item.created_at) # datetime print(item.updated_at) # datetime
# Explicitly refresh TTL on this readitem = store.get(("cache",), "result-xyz", refresh_ttl=True)
# Async variantitem = await store.aget(("users", "alice"), "prefs")item = await store.aget(("cache",), "result-xyz", refresh_ttl=True)search() — filter + semantic
Section titled “search() — filter + semantic”Basic filtering
Section titled “Basic filtering”filter= accepts exact-match and comparison-operator expressions against top-level and nested value keys:
# Exact match (shorthand)results = store.search(("docs",), filter={"status": "active"})
# Exact match (explicit $eq — same as above)results = store.search(("docs",), filter={"status": {"$eq": "active"}})
# Comparison operatorsresults = store.search(("docs",), filter={"score": {"$gt": 4.99}})results = store.search(("docs",), filter={"score": {"$gte": 3.0}})results = store.search(("docs",), filter={"age": {"$lt": 30}})results = store.search(("docs",), filter={"age": {"$lte": 30}})results = store.search(("docs",), filter={"status": {"$ne": "deleted"}})
# Multiple conditions (AND — all must match)results = store.search( ("docs",), filter={"score": {"$gte": 3.0}, "color": "red"}, limit=20,)
# Nested dict filterresults = store.search( ("orders",), filter={"meta": {"priority": "high"}},)Filter operator reference
Section titled “Filter operator reference”| Operator | Meaning | Example |
|---|---|---|
| (plain value) | Equal (shorthand for $eq) | {"status": "active"} |
$eq | Equal | {"status": {"$eq": "active"}} |
$ne | Not equal | {"status": {"$ne": "deleted"}} |
$gt | Greater than | {"score": {"$gt": 4.0}} |
$gte | Greater than or equal | {"score": {"$gte": 4.0}} |
$lt | Less than | {"age": {"$lt": 30}} |
$lte | Less than or equal | {"age": {"$lte": 30}} |
Numeric comparisons use float() coercion internally, matching PostgreSQL JSONB behavior. Nested dict filters require the stored value to also be a dict with matching keys. Filtering is applied before semantic ranking, so filter= does not require an index.
Semantic (vector) search
Section titled “Semantic (vector) search”Requires the store to be created with index=IndexConfig(...):
from langchain.embeddings import init_embeddingsfrom langgraph.store.memory import InMemoryStore
store = InMemoryStore( index={ "dims": 1536, "embed": init_embeddings("openai:text-embedding-3-small"), # Optional: which fields within `value` to embed. Default: ["$"] (whole value). "fields": ["text", "summary"], })
store.put(("docs",), "d1", {"text": "Rust is a systems language", "type": "lang"})results = store.search( ("docs",), query="memory-safe low-level languages", filter={"type": "lang"}, limit=5,)for r in results: print(r.score, r.value["text"]) # r.score is float | NoneIf the store was not created with index=, the query= argument is silently ignored and search returns plain filtered results.
# Async variant — identical parametersresults = await store.asearch( ("docs",), query="memory-safe low-level languages", filter={"type": "lang"}, limit=5,)list_namespaces()
Section titled “list_namespaces()”Explore the namespace tree:
# All namespaces under "users", truncated to depth 2namespaces = store.list_namespaces(prefix=("users",), max_depth=2)# [('users', 'alice'), ('users', 'bob'), ...]
# Namespaces ending with "prefs" anywhere in the treenamespaces = store.list_namespaces(suffix=("prefs",))
# Wildcard: any namespace whose second segment is "config"namespaces = store.list_namespaces(prefix=("users", "*", "config"))
# Async variantnamespaces = await store.alist_namespaces(prefix=("users",), max_depth=2)prefix / suffix accept NamespacePath tuples; use "*" as a wildcard segment. max_depth caps the tuple length returned. Given existing namespaces ("a","b","c"), ("a","b","d","e"), ("a","b","f"):
store.list_namespaces(prefix=("a", "b"), max_depth=3)# [("a", "b", "c"), ("a", "b", "d"), ("a", "b", "f")]batch() — atomic multi-op
Section titled “batch() — atomic multi-op”Submit any mix of GetOp, PutOp, SearchOp, ListNamespacesOp in a single call. Results are returned in the same order as the operations. PutOp always returns None.
from langgraph.store.base import GetOp, PutOp, SearchOp, ListNamespacesOp
results = store.batch([ GetOp(namespace=("users", "123"), key="prefs"), PutOp(namespace=("users", "123"), key="cache", value={"data": "..."}), SearchOp(namespace_prefix=("users",), filter={"active": True}, limit=5), ListNamespacesOp(match_conditions=None, max_depth=2, limit=10, offset=0),])# results[0] -> Item | None# results[1] -> None (PutOp)# results[2] -> list[SearchItem]# results[3] -> list[tuple[str, ...]]
# Async variantresults = await store.abatch([ PutOp(("cache",), "key", {"data": "..."}), GetOp(("cache",), "key"),])IndexConfig
Section titled “IndexConfig”class IndexConfig(TypedDict, total=False): dims: int embed: Embeddings | EmbeddingsFunc | AEmbeddingsFunc | str fields: list[str] # default ["$"] — embed the entire value ann_index_config: ... # backend-specific (e.g., pgvector tuning) distance_type: Literal["l2", "inner_product", "cosine"]embed can be:
- a LangChain
Embeddingsinstance, - a sync
(list[str]) -> list[list[float]], - an async callable with the same shape,
- a provider string like
"openai:text-embedding-3-small"(LangChain resolves it).
Common model dimensions (from source docstring):
| Model | Dims |
|---|---|
openai:text-embedding-3-large | 3072 |
openai:text-embedding-3-small | 1536 |
openai:text-embedding-ada-002 | 1536 |
cohere:embed-english-v3.0 | 1024 |
cohere:embed-english-light-v3.0 | 384 |
cohere:embed-multilingual-v3.0 | 1024 |
cohere:embed-multilingual-light-v3.0 | 384 |
TTLConfig
Section titled “TTLConfig”class TTLConfig(TypedDict, total=False): refresh_on_read: bool # default True default_ttl: float | None # minutes for new items sweep_interval_minutes: int | Nonerefresh_on_read— ifTrue, everyget()orsearch()that returns an item resets its TTL. Can be overridden per-call withrefresh_ttl=onget/search.default_ttl— applied to allput()calls that do not specify their ownttl=.sweep_interval_minutes— how often the backend actively deletes expired items.InMemoryStoreevicts lazily (no background sweeper).
Only set ttl=... on put() if the backend supports TTL. InMemoryStore accepts the kwarg (supports_ttl = True) but does not run a background sweeper.
InMemoryStore
Section titled “InMemoryStore”from langgraph.store.memory import InMemoryStore
store = InMemoryStore(*, index: IndexConfig | None = None)- Pure-Python, process-local. Data is lost on exit.
- Vector search uses cosine similarity with numpy if installed, falls back to a pure-Python dot product otherwise.
pip install numpyfor any non-trivial corpus. - Exposes both sync and async methods;
abatchruns embedding calls viaasyncio.gatherand ThreadPoolExecutor for sync embedding models. supports_ttl = True— acceptsttl=onput(), but evicts lazily (no sweep thread).
With LangChain embeddings
Section titled “With LangChain embeddings”from langchain.embeddings import init_embeddingsfrom langgraph.store.memory import InMemoryStore
store = InMemoryStore( index={ "dims": 1536, "embed": init_embeddings("openai:text-embedding-3-small"), "fields": ["text"], # which fields to embed })
store.put(("docs",), "doc1", {"text": "Python tutorial"})store.put(("docs",), "doc2", {"text": "TypeScript guide"})
results = store.search(("docs",), query="python programming", limit=5)for hit in results: print(hit.key, hit.score) # SearchItem has .score: float | NoneWith a custom embed function
Section titled “With a custom embed function”from openai import OpenAIfrom langgraph.store.memory import InMemoryStore
client = OpenAI()
def embed_texts(texts: list[str]) -> list[list[float]]: response = client.embeddings.create( model="text-embedding-3-small", input=texts, ) return [e.embedding for e in response.data]
store = InMemoryStore(index={"dims": 1536, "embed": embed_texts})store.put(("docs",), "doc1", {"text": "Python tutorial"})results = store.search(("docs",), query="python programming", limit=5)With an async embed function
Section titled “With an async embed function”from openai import AsyncOpenAIfrom langgraph.store.memory import InMemoryStore
client = AsyncOpenAI()
async def aembed_texts(texts: list[str]) -> list[list[float]]: response = await client.embeddings.create( model="text-embedding-3-small", input=texts, ) return [e.embedding for e in response.data]
store = InMemoryStore(index={"dims": 1536, "embed": aembed_texts})
# Use async methods so the embed function is awaited properlyawait store.aput(("docs",), "doc1", {"text": "Python tutorial"})results = await store.asearch(("docs",), query="python programming")PostgresStore / AsyncPostgresStore
Section titled “PostgresStore / AsyncPostgresStore”from langgraph.store.postgres import PostgresStore
with PostgresStore.from_conn_string(DB_URI) as store: store.setup() # creates tables + pgvector extension if index is set graph = builder.compile(store=store) graph.invoke(..., cfg)from_conn_stringis a context manager (same pattern asPostgresSaver).setup()is required on first use.- Pass
index=IndexConfig(...)to enable pgvector semantic search. Requires thevectorextension in your database.
Async counterpart lives at langgraph.store.postgres.aio.AsyncPostgresStore with an async context manager and await store.setup().
Using a Store from a node
Section titled “Using a Store from a node”The Runtime.store attribute exposes whatever you passed to compile(store=...):
from langgraph.runtime import Runtime
def recall_node(state: State, runtime: Runtime) -> dict: if runtime.store is None: return {"memories": []} hits = runtime.store.search( ("memories", state["user_id"]), query=state["question"], limit=3, ) context = "\n".join(item.value["text"] for item in hits) return {"context": context}For async nodes use asearch / aget:
async def recall_node_async(state: State, runtime: Runtime) -> dict: hits = await runtime.store.asearch( ("memories", state["user_id"]), query=state["question"], limit=3, ) return {"context": "\n".join(h.value["text"] for h in hits)}Using a Store from a tool (InjectedStore)
Section titled “Using a Store from a tool (InjectedStore)”Tools get the store injected automatically when wrapped by ToolNode (from langgraph.prebuilt). The store argument is stripped from the schema the model sees, so the LLM cannot pass it.
import uuidfrom typing import Annotatedfrom langchain_core.tools import toolfrom langgraph.prebuilt import InjectedStore, ToolNodefrom langgraph.store.base import BaseStore
@tooldef remember_fact( fact: str, store: Annotated[BaseStore, InjectedStore()],) -> str: """Store a fact in long-term memory.""" store.put(("facts",), str(uuid.uuid4()), {"text": fact}) return f"Remembered: {fact}"
@tooldef save_fact( user_id: str, fact: str, store: Annotated[BaseStore, InjectedStore()],) -> str: """Save a fact scoped to a user.""" store.put(("facts", user_id), fact, {"text": fact}) return f"Saved for {user_id}"
tool_node = ToolNode([remember_fact, save_fact])InjectedState works the same way for whole-state injection; ToolRuntime bundles state + context + config + store + stream_writer + tool_call_id into one object.
Patterns
Section titled “Patterns”1. Per-user preferences
Section titled “1. Per-user preferences”ns = ("users", user_id, "prefs")store.put(ns, "theme", {"mode": "dark"})store.put(ns, "lang", {"code": "en"})for pref in store.list_namespaces(prefix=("users", user_id)): for item in store.search(pref): print(pref, item.key, item.value)2. Semantic memory with filtered recall
Section titled “2. Semantic memory with filtered recall”store = InMemoryStore(index={"dims": 1536, "embed": "openai:text-embedding-3-small"})store.put(("mem", "alice"), "m1", {"text": "Likes espresso", "kind": "food"})store.put(("mem", "alice"), "m2", {"text": "Works at Acme", "kind": "work"})
hits = store.search( ("mem", "alice"), query="favorite drink", filter={"kind": "food"}, limit=3,)3. Batched mixed operations in one round-trip
Section titled “3. Batched mixed operations in one round-trip”The batch() / abatch() methods execute any combination of GetOp, PutOp, SearchOp, and ListNamespacesOp in a single call. Use it when you need to atomically read and write, or simply avoid multiple network round-trips:
from langgraph.store.base import GetOp, PutOp, SearchOp, ListNamespacesOp
results = store.batch([ # Write three facts PutOp(("mem", user_id), "hobby", {"text": "cycling"}, index=None, ttl=None), PutOp(("mem", user_id), "city", {"text": "Berlin"}, index=None, ttl=None), # Read one fact in the same call GetOp(("mem", user_id), "name"), # Semantic search SearchOp( ("mem", user_id), query="favorite sport", filter=None, limit=3, offset=0, ), # List all namespaces under "mem" ListNamespacesOp(match_conditions=None, max_depth=2, limit=10, offset=0),])# results is a list aligned with the ops:# results[0] = None (PutOp returns None)# results[1] = None# results[2] = Item(...) or None# results[3] = list[SearchItem]# results[4] = list[tuple[str, ...]]put_hobby, put_city, get_name, search_results, namespaces = resultsUse abatch in async contexts:
results = await store.abatch([ PutOp(("cache",), "key", {"data": payload}), GetOp(("cache",), "key"),])4. Tools that read and write memory
Section titled “4. Tools that read and write memory”from typing import Annotatedfrom langchain_core.tools import toolfrom langgraph.prebuilt import InjectedStorefrom langgraph.store.base import BaseStore
@tooldef remember( user_id: str, text: str, store: Annotated[BaseStore, InjectedStore()],) -> str: store.put(("mem", user_id), f"note-{text[:16]}", {"text": text}) return "ok"
@tooldef recall( user_id: str, topic: str, store: Annotated[BaseStore, InjectedStore()],) -> list[str]: return [i.value["text"] for i in store.search(("mem", user_id), query=topic, limit=5)]5. TTL-bounded cache
Section titled “5. TTL-bounded cache”# Store result with a 30-minute TTLstore.put(("cache", "bing"), query, {"json": result}, ttl=30)
# Retrieve and reset the TTL countdown on each readhit = store.get(("cache", "bing"), query, refresh_ttl=True)
# Async equivalentsawait store.aput(("cache", "bing"), query, {"json": result}, ttl=30)hit = await store.aget(("cache", "bing"), query, refresh_ttl=True)6. Filter operators — comparison-based search
Section titled “6. Filter operators — comparison-based search”filter= on search() supports exact equality and six comparison operators, matching PostgreSQL JSONB behavior. No index required — filtering is applied before semantic ranking.
from langgraph.store.memory import InMemoryStore
store = InMemoryStore()
# Populate some itemsstore.put(("products",), "p1", {"name": "Widget A", "price": 9.99, "stock": 50})store.put(("products",), "p2", {"name": "Widget B", "price": 24.99, "stock": 0})store.put(("products",), "p3", {"name": "Gadget", "price": 4.99, "stock": 100})store.put(("products",), "p4", {"name": "Premium", "price": 99.99, "stock": 5})
# Items where stock > 0in_stock = store.search(("products",), filter={"stock": {"$gt": 0}})print([i.value["name"] for i in in_stock])# ['Widget A', 'Gadget', 'Premium']
# Price between 5 and 30 inclusive AND in stockaffordable = store.search( ("products",), filter={"price": {"$gte": 5.0}, "stock": {"$gt": 0}},)print([i.value["name"] for i in affordable])# ['Widget A']
# Not equalnot_widget = store.search(("products",), filter={"name": {"$ne": "Widget A"}})print([i.value["name"] for i in not_widget])# ['Widget B', 'Gadget', 'Premium']
# Nested field matchstore.put(("orders",), "o1", {"status": "pending", "meta": {"priority": "high"}})store.put(("orders",), "o2", {"status": "done", "meta": {"priority": "low"}})
high_priority = store.search(("orders",), filter={"meta": {"priority": "high"}})print([i.value["status"] for i in high_priority])# ['pending']All supported operators:
| Operator | Meaning | Example |
|---|---|---|
$eq | Equal | {"status": {"$eq": "active"}} — same as {"status": "active"} |
$ne | Not equal | {"status": {"$ne": "deleted"}} |
$gt | Greater than | {"score": {"$gt": 4.0}} |
$gte | Greater than or equal | {"score": {"$gte": 4.0}} |
$lt | Less than | {"age": {"$lt": 30}} |
$lte | Less than or equal | {"age": {"$lte": 30}} |
Operators apply to numeric comparisons via float() coercion. Nested dict filters require the value to also be a dict with matching keys.
7. Pagination with offset
Section titled “7. Pagination with offset”For large namespaces use offset to page through results:
from langgraph.store.memory import InMemoryStore
store = InMemoryStore()for i in range(100): store.put(("logs",), f"log-{i:03d}", {"msg": f"Event {i}", "level": "info"})
page_size = 20all_results = []offset = 0while True: batch = store.search(("logs",), limit=page_size, offset=offset) if not batch: break all_results.extend(batch) offset += page_size
print(f"Total retrieved: {len(all_results)}") # 1008. Async store operations in a FastAPI service
Section titled “8. Async store operations in a FastAPI service”import asynciofrom langgraph.store.memory import InMemoryStorefrom langgraph.store.base import GetOp, PutOp, SearchOp
store = InMemoryStore()
async def upsert_memory(user_id: str, key: str, text: str) -> None: await store.aput(("mem", user_id), key, {"text": text, "user": user_id})
async def recall_memories(user_id: str, query: str, top_k: int = 5) -> list[str]: hits = await store.asearch( ("mem", user_id), query=query, limit=top_k, ) return [h.value["text"] for h in hits]
async def bulk_write(user_id: str, facts: list[dict]) -> None: ops = [ PutOp(("mem", user_id), f"fact-{i}", {"text": f["text"]}) for i, f in enumerate(facts) ] await store.abatch(ops)
# Usage in an async context:async def main(): await upsert_memory("alice", "pref-lang", "Prefers Python") await bulk_write("alice", [ {"text": "Works at Acme Corp"}, {"text": "Enjoys hiking"}, ]) results = await recall_memories("alice", "programming language", top_k=3) print(results)
asyncio.run(main())9. Per-field vector indexing
Section titled “9. Per-field vector indexing”By default InMemoryStore embeds the entire value dict (fields=["$"]). Specify explicit paths to embed only relevant text and reduce embedding costs:
from langgraph.store.memory import InMemoryStore
store = InMemoryStore( index={ "dims": 1536, "embed": "openai:text-embedding-3-small", # Only embed these two fields; ignore "tags", "created_at", etc. "fields": ["title", "body"], })
store.put(("articles",), "a1", { "title": "Introduction to LangGraph", "body": "LangGraph is a framework for building stateful multi-actor apps...", "tags": ["langchain", "agents"], "created_at": "2025-01-01",})
store.put(("articles",), "a2", { "title": "Python async patterns", "body": "asyncio and structured concurrency in modern Python...", "tags": ["python", "async"], "created_at": "2025-02-01",})
# Semantic search only over title + bodyresults = store.search(("articles",), query="graph-based agent workflows", limit=3)for r in results: print(f"{r.score:.3f} — {r.value['title']}")
# Per-item override: skip indexing for a specific itemstore.put(("articles",), "a3", {"title": "Draft", "body": "WIP"}, index=False)
# Per-item override: embed only the body for this itemstore.put(("articles",), "a4", {"title": "Short", "body": "Detailed content here..."}, index=["body"])The [*] wildcard path selector embeds each array element separately:
store.put(("docs",), "d1", { "chapters": [ "Chapter 1: Introduction", "Chapter 2: State management", "Chapter 3: Multi-agent", ]}, index=["chapters[*]"])# Each chapter string is embedded separately and matched individually.Supported path selector syntax (from PutOp.index source):
| Syntax | Meaning |
|---|---|
"field" | Top-level field |
"parent.child.grandchild" | Nested field via dot notation |
"array[0]" | First element of an array |
"array[-1]" | Last element of an array |
"array[*]" | Each element separately (one vector per element) |
"a.b[*].c.d" | Complex nested path with array expansion |
"$" | Entire value object (default when fields is not set) |
Gotchas
Section titled “Gotchas”- Namespace rules. Each segment must be a non-empty string and must not contain
".".("", "x")raisesInvalidNamespaceError. The root segment"langgraph"is reserved and also raises. query=is ignored without an index. You will get filter-only results without any warning — always assertstorewas built withindex=IndexConfig(...)when you rely on semantic search.fields=["$"]means the entire value is stringified and embedded. Pick explicit fields for better recall and smaller embedding costs.InMemoryStoreis not Platform-safe. LangGraph Platform provides a managed store — don’t pass one when deploying there.- TTL is in minutes, not seconds.
ttl=30means 30 minutes, not 30 seconds. store.searchreturnslist[SearchItem], not an iterator. Always bounded bylimit(default 10). Paginate withoffset.deleteusesPutOp(...value=None)internally. If you subclassBaseStore,PutOp.value is Noneis the delete signal.supports_ttlcheck. Passingttl=toput()on a store withsupports_ttl = FalseraisesNotImplementedErrorat runtime, not at construction time.InMemoryStoreTTL is lazy-eviction only. There is no background sweep thread; items are removed when next accessed after expiry.
Breaking changes
Section titled “Breaking changes”| Version | Change |
|---|---|
| 1.2.1 | GetOp.refresh_ttl and SearchOp.refresh_ttl fields added; TTLConfig.sweep_interval_minutes added. |
| 1.1 | Semantic-search result SearchItem.score is consistently `float |
| 1.0 | Store moved out of experimental; InjectedStore is the stable way to pull the store into tools. |
| 0.6 | runtime.store replaces config["configurable"]["store"] for node injection. |