langgraph-checkpoint-gridgain

GridGain persistence for LangGraph: a checkpointer for short-term (thread) state, a store for long-term, cross-thread memory, and a node cache for compile(cache=...).

pip install langgraph-checkpoint-gridgain

from pygridgain import Client
from langgraph.cache.gridgain import GridGainNodeCache
from langgraph.checkpoint.gridgain import GridGainCheckpointSaver
from langgraph.store.gridgain import GridGainMemoryStore

client = Client()
client.connect("127.0.0.1", 10800)

graph = builder.compile(
    checkpointer=GridGainCheckpointSaver(client),
    store=GridGainMemoryStore(client),
)

Import paths

The classes are defined in langgraph.checkpoint.gridgain, langgraph.store.gridgain and langgraph.cache.gridgain — the namespace every LangGraph persistence backend publishes under, and the only one this package installs into:

from langgraph.checkpoint.gridgain import GridGainCheckpointSaver
from langgraph.store.gridgain import GridGainMemoryStore
from langgraph.cache.gridgain import GridGainNodeCache

There is no flat langgraph_checkpoint_gridgain module, for the same reason postgres, sqlite, redis and mongodb ship none: one object should have one import path.

Relationship to langchain-gridgain

This package depends on langchain-gridgain and borrows its client guard and TTL helper, so a checkpoint write cannot interleave with a vector search on the one socket a synchronous pygridgain client has. Installing this package installs that one too, which is how you get GridGain's LangChain components — a vector store, LLM caches, a byte store, a document loader and a chat message history — beside the LangGraph persistence.

Moving from langchain-gridgain 2.x? These two classes used to ship there. The 3.0.0 release moved them here:

-from langchain_gridgain import GridGainCheckpointSaver, GridGainMemoryStore
+from langgraph.checkpoint.gridgain import GridGainCheckpointSaver
+from langgraph.store.gridgain import GridGainMemoryStore

Behavior and stored data are unchanged.

Async

Both classes implement the async surface natively against pygridgain's AioClient, rather than running the sync path on a worker thread:

from pygridgain import AioClient

aio_client = AioClient()
await aio_client.connect("127.0.0.1", 10800)

checkpointer = GridGainCheckpointSaver(client, aio_client=aio_client)

Requirements

• Python 3.10+
• A GridGain 8 cluster reachable over the thin client protocol

GridGainCheckpointSaver

GridGainCheckpointSaver is a LangGraph checkpointer (BaseCheckpointSaver) that persists agent state in GridGain, enabling resume, human-in-the-loop, and time-travel. It stores each checkpoint, its payload and its intermediate writes in three SQL-backed caches — lg_checkpoints, lg_checkpoint_blobs and lg_checkpoint_writes — created automatically on first use. The payload is deliberately kept out of lg_checkpoints, which is the table every "latest checkpoint" query orders by; see Upgrading. Unlike the vector components, it uses only SQL, so it runs on any GridGain/Apache Ignite node (no vector license required).

Usage example:

from pygridgain import Client
from langgraph.checkpoint.gridgain import GridGainCheckpointSaver

client = Client()
client.connect("127.0.0.1", 10800)

checkpointer = GridGainCheckpointSaver(client)

graph = builder.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "conversation-1"}}
result = graph.invoke(inputs, config)

# Resume later on the same thread; inspect the latest state.
snapshot = graph.get_state(config)

For native async graphs (ainvoke / astream), pass a connected AioClient so the checkpointer uses non-blocking I/O:

from pygridgain import AioClient

aio_client = AioClient()
await aio_client.connect("127.0.0.1", 10800)

checkpointer = GridGainCheckpointSaver(aio_client=aio_client)

Pass both client and aio_client to serve sync and async graphs from one instance. A sync-only saver still works with ainvoke — it runs the sync path on a worker thread.

GridGainMemoryStore

GridGainMemoryStore is a LangGraph store (BaseStore) for memory that outlives a single thread — user preferences, extracted facts, anything an agent should recall across conversations. Items live in namespaces and support structured search, so this is the counterpart to the checkpointer's per-thread state.

It stores items in one cache (lg_store, created when the store is constructed) and, like the checkpointer, needs no vector license — it uses the key-value and SQL APIs only.

Usage example:

from pygridgain import Client
from langgraph.store.gridgain import GridGainMemoryStore

client = Client()
client.connect("127.0.0.1", 10800)

store = GridGainMemoryStore(client)

# Namespaced items
store.put(("users", "u1"), "prefs", {"theme": "dark", "score": 5})
store.get(("users", "u1"), "prefs").value  # {"theme": "dark", "score": 5}

# Search a namespace and everything nested under it
store.search(("users",), filter={"theme": "dark"})
store.search(("users",), filter={"score": {"$gte": 5}}, limit=10, offset=0)

# Explore the namespace hierarchy
store.list_namespaces(prefix=("users",))
store.list_namespaces(max_depth=1)

store.delete(("users", "u1"), "prefs")

graph = builder.compile(store=store)  # cross-thread memory for an agent

Filters support exact matches plus $eq, $ne, $gt, $gte, $lt, $lte, and are applied to the item's JSON value.

TTL is native. Unlike the Postgres/SQLite stores, expiry is enforced by the server rather than an expires_at column. Pass a per-item ttl (in minutes) and/or store-wide defaults, and by default reads refresh the expiry:

store = GridGainMemoryStore(
    client,
    ttl_config={"default_ttl": 60, "refresh_on_read": True},  # minutes
)

store.put(("users", "u1"), "session", {"step": 3}, ttl=10)  # expires in 10 minutes
store.get(("users", "u1"), "session")  # ...and that resets the clock
store.get(("users", "u1"), "session", refresh_ttl=False)  # read without refreshing

For native async graphs, pass a connected AioClient (optionally alongside the sync client, to serve both):

from pygridgain import AioClient

aio_client = AioClient()
await aio_client.connect("127.0.0.1", 10800)

store = GridGainMemoryStore(aio_client=aio_client)
await store.aput(("users", "u1"), "prefs", {"theme": "dark"})

Semantic search is not implemented yet. search(..., query="...") is accepted but the query is ignored (with a warning) and the structured-filter path runs; vector-backed search depends on the GridGain vector engine and lands separately.

GridGainNodeCache

GridGainNodeCache is a LangGraph node cache (langgraph.cache.base.BaseCache). When a node has a CachePolicy, the engine stores the node's output keyed by its input, and a later run with the same input skips the node. Attach it at compile(cache=...). Despite the shared interface name it is unrelated to the LangChain LLM cache above: GridGainCache (langchain_core.caches.BaseCache) caches model generations by prompt, GridGainNodeCache caches graph node runs.

It stores entries in one cache (lg_node_cache, created when the node cache is constructed with a sync client, or on first use with an aio_client only) and needs no vector license — it uses the key-value and SQL APIs only.

Usage example:

from pygridgain import Client
from langgraph.graph import END, START, StateGraph
from langgraph.types import CachePolicy
from langgraph.cache.gridgain import GridGainNodeCache

client = Client()
client.connect("127.0.0.1", 10800)

builder = StateGraph(State)
builder.add_node("expensive", expensive, cache_policy=CachePolicy(ttl=300))  # seconds
builder.add_edge(START, "expensive")
builder.add_edge("expensive", END)
graph = builder.compile(cache=GridGainNodeCache(client))

graph.invoke({"x": 3})  # runs `expensive`
graph.invoke({"x": 3})  # same input: the node's output comes from GridGain
graph.clear_cache(["expensive"])  # deletes those nodes' entries, one SQL DELETE

graph.clear_cache() with no arguments clears every node namespace of that graph (still one SQL DELETE). Only a direct node_cache.clear() empties the whole GridGain cache, other graphs' entries included.

What two graphs share. LangGraph's namespace for a node is the node's name plus its function's module.qualname; it carries no graph or application id. Two graphs share entries, hits and clears exactly when they have a node of the same name whose function has the same module.qualname: the same function, or two closures or functools.partials from one factory. That is what makes a hit survive a restart, and it is also how a factory-built node in one graph can serve another graph's cached output. The default lg_node_cache is shared across the cluster on purpose; give each application its own cache_name= (a plain identifier, [A-Za-z_][A-Za-z0-9_]*, not a SQL reserved word) when its node functions are built by a factory, or whenever two deployments must not see each other's entries.

Chat states need an explicit key_func. LangGraph's default cache key pickles the node's input. A message that came back from any cache, InMemoryCache included, carries a larger model_fields_set than a freshly built one and pickles differently, so in a model-plus-tools loop over an add_messages state only the first model step of a repeat run hits. Key on what the model sees instead:

def messages_key(state):
    return "|".join(f"{m.type}:{m.content}" for m in state["messages"])


builder.add_node(
    "call_model", call_model, cache_policy=CachePolicy(key_func=messages_key)
)

Key shapes. Namespace labels and cache keys must be built-in str. LangGraph produces those itself, except for the node name, which is whatever you passed to add_node: a StrEnum or (str, Enum) member is rejected with a TypeError, so spell it add_node(str(N.SUMMARIZE), ...).

TTL is native and in seconds. CachePolicy(ttl=...) is enforced by the server; an entry with no TTL never expires, and reads do not extend an entry's life. Entries live in GridGain, not in the process, so a second process — or a restart — sees the same hits.

For native async graphs, pass a connected AioClient (optionally alongside the sync client, to serve both):

graph = builder.compile(cache=GridGainNodeCache(aio_client=aio_client))
await graph.ainvoke({"x": 3})