langgraph-checkpoint-redis

LangGraph Redis

This repository contains Redis implementations for LangGraph, providing both Checkpoint Savers and Stores functionality.

Overview

The project consists of two main components:

• Redis Checkpoint Savers: Implementations for storing and managing checkpoints using Redis
• Redis Stores: Redis-backed key-value stores with optional vector search capabilities

Dependencies

The project requires the following main dependencies:

• redis>=5.2.1
• redisvl>=0.5.1
• langgraph-checkpoint>=2.0.24
• langgraph>=0.3.0

Installation

Install the library using pip:

pip install langgraph-checkpoint-redis

Redis Checkpoint Savers

Important Notes

[!IMPORTANT] When using Redis checkpointers for the first time, make sure to call .setup() method on them to create required indices. See examples below.

Standard Implementation

from langgraph.checkpoint.redis import RedisSaver

write_config = {"configurable": {"thread_id": "1", "checkpoint_ns": ""}}
read_config = {"configurable": {"thread_id": "1"}}

with RedisSaver.from_conn_string("redis://localhost:6379") as checkpointer:
    # Call setup to initialize indices
    checkpointer.setup()
    checkpoint = {
        "v": 1,
        "ts": "2024-07-31T20:14:19.804150+00:00",
        "id": "1ef4f797-8335-6428-8001-8a1503f9b875",
        "channel_values": {
            "my_key": "meow",
            "node": "node"
        },
        "channel_versions": {
            "__start__": 2,
            "my_key": 3,
            "start:node": 3,
            "node": 3
        },
        "versions_seen": {
            "__input__": {},
            "__start__": {
                "__start__": 1
            },
            "node": {
                "start:node": 2
            }
        },
        "pending_sends": [],
    }

    # Store checkpoint
    checkpointer.put(write_config, checkpoint, {}, {})

    # Retrieve checkpoint
    loaded_checkpoint = checkpointer.get(read_config)

    # List all checkpoints
    checkpoints = list(checkpointer.list(read_config))

Async Implementation

from langgraph.checkpoint.redis.aio import AsyncRedisSaver

async def main():
    write_config = {"configurable": {"thread_id": "1", "checkpoint_ns": ""}}
    read_config = {"configurable": {"thread_id": "1"}}

    async with AsyncRedisSaver.from_conn_string("redis://localhost:6379") as checkpointer:
        # Call setup to initialize indices
        await checkpointer.asetup()
        checkpoint = {
            "v": 1,
            "ts": "2024-07-31T20:14:19.804150+00:00",
            "id": "1ef4f797-8335-6428-8001-8a1503f9b875",
            "channel_values": {
                "my_key": "meow",
                "node": "node"
            },
            "channel_versions": {
                "__start__": 2,
                "my_key": 3,
                "start:node": 3,
                "node": 3
            },
            "versions_seen": {
                "__input__": {},
                "__start__": {
                    "__start__": 1
                },
                "node": {
                    "start:node": 2
                }
            },
            "pending_sends": [],
        }

        # Store checkpoint
        await checkpointer.aput(write_config, checkpoint, {}, {})

        # Retrieve checkpoint
        loaded_checkpoint = await checkpointer.aget(read_config)

        # List all checkpoints
        checkpoints = [c async for c in checkpointer.alist(read_config)]

# Run the async main function
import asyncio
asyncio.run(main())

Shallow Implementations

Shallow Redis checkpoint savers store only the latest checkpoint in Redis. These implementations are useful when retaining a complete checkpoint history is unnecessary.

from langgraph.checkpoint.redis.shallow import ShallowRedisSaver
# For async version: from langgraph.checkpoint.redis.ashallow import AsyncShallowRedisSaver

write_config = {"configurable": {"thread_id": "1", "checkpoint_ns": ""}}
read_config = {"configurable": {"thread_id": "1"}}

with ShallowRedisSaver.from_conn_string("redis://localhost:6379") as checkpointer:
    checkpointer.setup()
    # ... rest of the implementation follows similar pattern

Redis Stores

Redis Stores provide a persistent key-value store with optional vector search capabilities.

Synchronous Implementation

from langgraph.store.redis import RedisStore

# Basic usage
with RedisStore.from_conn_string("redis://localhost:6379") as store:
    store.setup()
    # Use the store...

# With vector search configuration
index_config = {
    "dims": 1536,  # Vector dimensions
    "distance_type": "cosine",  # Distance metric
    "fields": ["text"],  # Fields to index
}

with RedisStore.from_conn_string("redis://localhost:6379", index=index_config) as store:
    store.setup()
    # Use the store with vector search capabilities...

Async Implementation

from langgraph.store.redis.aio import AsyncRedisStore

async def main():
    async with AsyncRedisStore.from_conn_string("redis://localhost:6379") as store:
        await store.setup()
        # Use the store asynchronously...

asyncio.run(main())

Examples

The examples directory contains Jupyter notebooks demonstrating the usage of Redis with LangGraph:

• persistence_redis.ipynb: Demonstrates the usage of Redis checkpoint savers with LangGraph
• create-react-agent-memory.ipynb: Shows how to create an agent with persistent memory using Redis
• cross-thread-persistence.ipynb: Demonstrates cross-thread persistence capabilities
• persistence-functional.ipynb: Shows functional persistence patterns with Redis

Running Example Notebooks

To run the example notebooks with Docker:

• Navigate to the examples directory:

  cd examples
  

• Start the Docker containers:

  docker compose up
  

• Open the URL shown in the console (typically http://127.0.0.1:8888/tree) in your browser to access Jupyter.

• When finished, stop the containers:

  docker compose down
  

Implementation Details

Indexing

The Redis implementation creates these main indices:

• Checkpoints Index: Stores checkpoint metadata and versioning
• Channel Values Index: Stores channel-specific data
• Writes Index: Tracks pending writes and intermediate states

For Redis Stores with vector search:

• Store Index: Main key-value store
• Vector Index: Optional vector embeddings for similarity search

Contributing

We welcome contributions! Here's how you can help:

Development Setup

• Clone the repository:

  git clone https://github.com/redis-developer/langgraph-redis
  cd langgraph-redis
  

• Install dependencies:

  poetry install --all-extras
  

Available Commands

The project includes several make commands for development:

• Testing:

  make test           # Run all tests
  make test-all       # Run all tests including API tests
  

• Linting and Formatting:

  make format        # Format all files with Black and isort
  make lint          # Run formatting, type checking, and other linters
  make check-types   # Run mypy type checking
  

• Redis for Development/Testing:

  make redis-start   # Start Redis in Docker
  make redis-stop    # Stop Redis container
  

Contribution Guidelines

• Create a new branch for your changes
• Write tests for new functionality
• Ensure all tests pass: make test
• Format your code: make format
• Run linting checks: make lint
• Submit a pull request with a clear description of your changes
• Follow Conventional Commits for commit messages

License

This project is licensed under the MIT License.