Checkpointers

Checkpointers#

Checkpointers persist the state of a LangGraph workflow at each step, enabling time travel, fault recovery, and multi-turn conversations. The langgraph-checkpoint-redis library provides two primary checkpointer classes that store full checkpoint history.

RedisSaver#

RedisSaver is the synchronous checkpointer for standard Python applications.

Creating with a Connection String#

The from_conn_string class method is a context manager that handles connection lifecycle:

from langgraph.checkpoint.redis import RedisSaver

with RedisSaver.from_conn_string("redis://localhost:6379") as saver:
    saver.setup()
    graph = builder.compile(checkpointer=saver)
    result = graph.invoke(inputs, config)

The connection URL supports multiple schemes:

redis://host:port – standard unencrypted connection
rediss://host:port – SSL/TLS encrypted connection
redis+sentinel://host:26379/service_name/db – Redis Sentinel

Creating with a Redis Client#

Pass an existing Redis client for full control over connection settings:

from redis import Redis
from langgraph.checkpoint.redis import RedisSaver

client = Redis(host="localhost", port=6379, decode_responses=False)
saver = RedisSaver(redis_client=client)
saver.setup()

When you provide your own client, the saver does not close it automatically. You are responsible for managing the client lifecycle.

Calling setup()#

You must call setup() before using the checkpointer. This method creates the RediSearch indices that power checkpoint queries:

saver.setup()  # Creates indices if they do not already exist

The call is idempotent – it will not overwrite existing indices.

Core Operations#

put – Save a checkpoint:

next_config = saver.put(config, checkpoint, metadata, new_versions)

This is called automatically by LangGraph during graph execution. You rarely need to call it directly.

get_tuple – Retrieve the latest checkpoint for a thread:

config = {"configurable": {"thread_id": "thread-1"}}
checkpoint_tuple = saver.get_tuple(config)
# Returns: CheckpointTuple(config, checkpoint, metadata, parent_config, pending_writes)

To retrieve a specific checkpoint, include the checkpoint ID:

config = {
    "configurable": {
        "thread_id": "thread-1",
        "checkpoint_id": "01HZXYZ...",  # Specific ULID
    }
}
checkpoint_tuple = saver.get_tuple(config)

list – Iterate over checkpoint history:

config = {"configurable": {"thread_id": "thread-1"}}
for checkpoint_tuple in saver.list(config, limit=10):
    print(checkpoint_tuple.metadata)

The list method supports filtering:

# Filter by metadata fields
for ct in saver.list(config, filter={"source": "loop", "step": 3}):
    print(ct.checkpoint)

Thread Management#

delete_thread – Remove all checkpoints and writes for a thread:

saver.delete_thread("thread-1")

prune – Keep only recent checkpoints to manage storage:

# Keep only the latest checkpoint per namespace
saver.prune(["thread-1", "thread-2"])

# Keep the last 5 checkpoints per namespace
saver.prune(["thread-1"], keep_last=5)

# Delete all checkpoints for the given threads
saver.prune(["thread-1"], strategy="delete")

AsyncRedisSaver#

AsyncRedisSaver provides the same API for async applications, using async Redis client connections.

Creating an Async Checkpointer#

from langgraph.checkpoint.redis import AsyncRedisSaver

async with AsyncRedisSaver.from_conn_string("redis://localhost:6379") as saver:
    await saver.asetup()
    graph = builder.compile(checkpointer=saver)
    result = await graph.ainvoke(inputs, config)

Or with a direct async client:

from redis.asyncio import Redis as AsyncRedis
from langgraph.checkpoint.redis import AsyncRedisSaver

client = AsyncRedis(host="localhost", port=6379)
saver = AsyncRedisSaver(redis_client=client)
await saver.asetup()

For standalone Redis workloads that may schedule more concurrent operations than the connection pool can serve at once, use a bounded BlockingConnectionPool. When all connections are busy, additional operations wait for a connection instead of failing immediately with MaxConnectionsError:

from redis.asyncio import BlockingConnectionPool, Redis as AsyncRedis
from langgraph.checkpoint.redis import AsyncRedisSaver

pool = BlockingConnectionPool.from_url(
    "redis://localhost:6379",
    max_connections=20,
    timeout=5,
)
client = AsyncRedis(connection_pool=pool)

try:
    async with AsyncRedisSaver.from_conn_string(redis_client=client) as saver:
        graph = builder.compile(checkpointer=saver)
        result = await graph.ainvoke(inputs, config)
finally:
    await client.aclose()
    await pool.disconnect()

Choose max_connections based on the concurrency your Redis deployment can support. The timeout applies to each attempt to acquire a connection from the pool. If no connection becomes available within that interval, direct redis-py operations raise redis.exceptions.ConnectionError. Operations performed through RedisVL may instead raise RedisVLError, with the original connection error preserved as its cause.

This provides connection-level backpressure, but it does not limit how many asyncio tasks the application creates. Applications that can schedule an unbounded number of tasks should also use an application-level concurrency limit, such as an asyncio.Semaphore or a bounded worker queue.

BlockingConnectionPool is for standalone Redis clients. Redis Cluster and Redis Sentinel use topology-specific connection management; for those deployments, inject the appropriate async client and configure the limits it supports. Because the client in this example is supplied by the application, the application is also responsible for closing it and disconnecting its pool.

Async Operations#

The async methods mirror the sync API:

# Get a checkpoint
checkpoint_tuple = await saver.aget_tuple(config)

# List checkpoints
async for ct in saver.alist(config, limit=10):
    print(ct.metadata)

# Save writes
await saver.aput_writes(config, writes, task_id)

Using with LangGraph compile()#

The standard pattern for integrating a checkpointer with LangGraph:

from langgraph.graph import StateGraph

builder = StateGraph(MyState)
# ... add nodes and edges ...

# Compile with checkpointer
with RedisSaver.from_conn_string("redis://localhost:6379") as saver:
    saver.setup()
    graph = builder.compile(checkpointer=saver)

    # Each invocation with the same thread_id continues from the last checkpoint
    config = {"configurable": {"thread_id": "user-session-123"}}
    result = graph.invoke({"messages": [user_message]}, config)

Configuration Options#

The RedisSaver and AsyncRedisSaver constructors accept the following keyword arguments:

Parameter	Type	Default	Description
`redis_url`	`str`	`None`	Redis connection URL
`redis_client`	`Redis`	`None`	Existing Redis client
`connection_args`	`dict`	`None`	Extra connection keyword arguments
`ttl`	`dict`	`None`	TTL configuration (see TTL Configuration)
`checkpoint_prefix`	`str`	`"checkpoint"`	Key prefix for checkpoint documents
`checkpoint_write_prefix`	`str`	`"checkpoint_write"`	Key prefix for write documents

Cluster Support#

Both RedisSaver and AsyncRedisSaver support Redis Cluster. Cluster mode is detected automatically based on the client type:

from redis.cluster import RedisCluster

client = RedisCluster(host="cluster-node", port=7000)
saver = RedisSaver(redis_client=client)
saver.setup()

In cluster mode, the saver uses individual key operations instead of pipelines to avoid cross-slot errors.

Next Steps#

Shallow Checkpointers – use shallow savers for reduced storage
TTL Configuration – configure automatic expiration
Stores – add cross-thread key-value persistence

Version