LangGraph Redis Checkpoint

Warnings

• breaking Version 0.1.0 introduced breaking changes to the internal storage format. Checkpoints created with pre-0.1.0 versions are not readable by 0.1.0+ without manual migration.
  Fix: For new deployments, start with version 0.1.0 or newer. For existing data, migration is not automatically supported; consider clearing old checkpoints or using new thread IDs.
• gotcha Redis requires the 'RedisJSON' and 'RediSearch' modules to be enabled for `langgraph-checkpoint-redis` functionality, especially for index creation and efficient data access. Redis 8.0+ includes these by default.
  Fix: Ensure your Redis instance (local, Redis Stack, or managed service) has RedisJSON and RediSearch modules enabled. For Redis < 8.0, use Redis Stack or install modules separately. Failure to do so will result in errors during `.setup()` or checkpoint operations.
• gotcha The `.setup()` method must be called on RedisSaver/AsyncRedisSaver instances upon initial setup or application start to create necessary RediSearch indices. Failure to do so will lead to runtime errors.
  Fix: Always call `checkpointer.setup()` (for synchronous) or `await checkpointer.asetup()` (for asynchronous) after creating your `RedisSaver` or `AsyncRedisSaver` instance.
• gotcha Potential data loss due to Redis persistence settings. Default Redis configurations might not guarantee durability if Redis crashes unexpectedly before RDB snapshot or AOF write operations complete.
  Fix: Tune Redis persistence settings: use `save` directives (e.g., `save 60 100`), enable AOF (`appendonly yes`) with `appendfsync everysec`, and consider using both RDB and AOF with `aof-use-rdb-preamble yes`. For high availability, consider Redis Sentinel or Redis Cluster.
• gotcha Incorrect message serialization can lead to `MESSAGE_COERCION_FAILURE` errors when using LangGraph's checkpointers. This often happens when `message.to_dict()` is stored instead of `BaseMessage` objects or simple `{role, content}` dicts.
  Fix: Ensure that `BaseMessage` objects or `{role, content}` dictionaries are passed directly to the state or event streams, and avoid explicit calls to `message.to_dict()` for persistence.
• gotcha High memory usage in Redis deployments can lead to latency, failed runs, or Out of Memory (OOM) errors, especially in high-load or limited-resource environments.
  Fix: Monitor Redis memory usage. Implement TTL (Time To Live) settings for checkpoints (e.g., `defaultTTL` when creating saver) to manage data retention. Use `ShallowRedisSaver` for memory-optimized scenarios. Consider increasing Redis memory limits or using production-grade Redis deployments.
• gotcha Compatibility issues have been reported with Valkey (a Redis fork), which may cause hanging requests and connection problems.
  Fix: It is strongly recommended to use a genuine Redis instance instead of Valkey for reliable operation with `langgraph-checkpoint-redis`.

Install

• pip install langgraph-checkpoint-redis langgraph redis Install core packages

Imports

• RedisSaver

  from langgraph.checkpoint.redis import RedisSaver

• AsyncRedisSaver

  from langgraph.checkpoint.redis.aio import AsyncRedisSaver

• ShallowRedisSaver

  from langgraph.checkpoint.redis.shallow import ShallowRedisSaver

• AsyncShallowRedisSaver

  from langgraph.checkpoint.redis.ashallow import AsyncShallowRedisSaver

• RedisStore

  from langgraph.store.redis import RedisStore

Quickstart

This quickstart demonstrates how to integrate `RedisSaver` with a basic LangGraph `StateGraph`. It sets up a Redis connection, initializes the `RedisSaver` (including calling `.setup()` for index creation), and then compiles a simple graph to persist its state across invocations using a `thread_id`.

import os
from typing import Annotated
from langgraph.graph import StateGraph, START
from langgraph.checkpoint.redis import RedisSaver
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage

# Set up Redis connection string, e.g., 'redis://localhost:6379/0'
REDIS_URL = os.environ.get('REDIS_URL', 'redis://localhost:6379/0')

# Define your graph state
class AgentState:
    messages: Annotated[list[BaseMessage], lambda x, y: x + y]

# Define a simple node
def simple_agent_node(state: AgentState) -> dict:
    new_message = AIMessage(content=f"Echo: {state.messages[-1].content}")
    return {"messages": [new_message]}

# Create the Redis Checkpoint Saver
# Make sure Redis is running and accessible at REDIS_URL
# For production, ensure RedisJSON and RediSearch modules are enabled (Redis 8.0+ includes them)
with RedisSaver.from_conn_string(REDIS_URL) as checkpointer:
    # Important: Call setup() to initialize required RediSearch indices
    checkpointer.setup()

    # Build the graph
    workflow = StateGraph(AgentState)
    workflow.add_node("agent", simple_agent_node)
    workflow.set_entry_point(START)
    workflow.set_finish_point("agent")

    # Compile the graph with the checkpointer
    app = workflow.compile(checkpointer=checkpointer)

    # Example usage with a specific thread_id for persistence
    thread_id = "test_conversation_123"
    config = {"configurable": {"thread_id": thread_id}}

    print(f"\n--- Invoking agent for thread: {thread_id} ---")
    inputs = {"messages": [HumanMessage(content="Hello LangGraph with Redis!")]}
    result = app.invoke(inputs, config)
    print("Result 1:", result)

    print(f"\n--- Invoking agent again for the same thread: {thread_id} ---")
    inputs = {"messages": [HumanMessage(content="How are you doing?")]}
    result = app.invoke(inputs, config)
    print("Result 2:", result)

    # List checkpoints (optional)
    print(f"\n--- Checkpoints for thread {thread_id} ---")
    for checkpoint in checkpointer.list(config):
        print(checkpoint)