### Run Docker Compose for Examples

Source: https://github.com/redis-developer/langgraph-redis/blob/main/CLAUDE.md

Use this command to start Redis and other necessary services for running the example notebooks.

```bash
cd examples
docker compose up
```

--------------------------------

### Install Dependencies and Run Jupyter Notebook

Source: https://github.com/redis-developer/langgraph-redis/blob/main/CLAUDE.md

Install required Python packages and start a Jupyter Notebook server to access and run the examples locally.

```bash
pip install langgraph-checkpoint-redis jupyter
jupyter notebook
```

--------------------------------

### AsyncRedisStore Setup

Source: https://github.com/redis-developer/langgraph-redis/blob/main/docs/user_guide/stores.md

Illustrates the asynchronous initialization and setup of AsyncRedisStore.

```APIDOC
## AsyncRedisStore Setup

### Description
Initialize and set up the `AsyncRedisStore` for asynchronous key-value persistence.

### Method
`AsyncRedisStore.from_conn_string()`

### Endpoint
N/A (Client-side library)

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
```python
from langgraph.store.redis import AsyncRedisStore

async with AsyncRedisStore.from_conn_string("redis://localhost:6379") as store:
    await store.asetup()
    # Use the store...
```

### Response
#### Success Response (200)
N/A (Initialization)

#### Response Example
N/A
```

--------------------------------

### Navigate to Examples Directory

Source: https://github.com/redis-developer/langgraph-redis/blob/main/README.md

Change the current directory to the examples folder to run the notebooks.

```bash
cd examples
```

--------------------------------

### Initialize Agent with Tools

Source: https://github.com/redis-developer/langgraph-redis/blob/main/examples/middleware/middleware_semantic_cache.ipynb

Creates an agent using `create_agent` with a specified model and a list of tools. This example shows the setup for both default and Responses API modes.

```python
from langchain.agents import create_agent
from langchain_core.messages import HumanMessage

# Create temporary agents (no middleware) to show raw response formats
agent_default_raw = create_agent(model=model_default, tools=tools)
agent_responses_raw = create_agent(model=model_responses_api, tools=tools)

# Same question, two different API modes
question = "What is 2 + 2?"

result_default = await agent_default_raw.ainvoke(
    {"messages": [HumanMessage(content=question)]}
)
inspect_response(result_default, label="Default Mode (Chat Completions)")

result_responses = await agent_responses_raw.ainvoke(
    {"messages": [HumanMessage(content=question)]}
)
inspect_response(result_responses, label="Responses API Mode")

print("\nNotice: Responses API content is a list of blocks, each with an embedded 'id'.")
print("The middleware strips these IDs from cached content to prevent duplicate ID errors.")
```

--------------------------------

### RedisSaver - setup()

Source: https://github.com/redis-developer/langgraph-redis/blob/main/docs/user_guide/checkpointers.md

The `setup()` method must be called before using the checkpointer to create necessary RediSearch indices.

```APIDOC
## RedisSaver - setup()

### Description
The `setup()` method must be called before using the checkpointer. It creates the RediSearch indices required for checkpoint queries. This operation is idempotent.

### Method
`setup()`

### Endpoint
N/A (Method on RedisSaver instance)

### Parameters
None

### Request Example
```python
saver.setup()  # Creates indices if they do not already exist
```

### Response
None

#### Success Response (200)
N/A

#### Response Example
N/A
```

--------------------------------

### RedisStore Setup

Source: https://github.com/redis-developer/langgraph-redis/blob/main/docs/user_guide/stores.md

Demonstrates how to initialize and set up RedisStore using a connection string or an existing Redis client.

```APIDOC
## RedisStore Setup

### Description
Initialize and set up the `RedisStore` for key-value persistence.

### Method
`RedisStore.from_conn_string()` or `RedisStore(client)`

### Endpoint
N/A (Client-side library)

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
```python
from langgraph.store.redis import RedisStore

# Using connection string
with RedisStore.from_conn_string("redis://localhost:6379") as store:
    store.setup()
    # Use the store...

# Using an existing Redis client
from redis import Redis
client = Redis.from_url("redis://localhost:6379")
store = RedisStore(client)
store.setup()
```

### Response
#### Success Response (200)
N/A (Initialization)

#### Response Example
N/A
```

--------------------------------

### Install All Dependencies with Poetry

Source: https://github.com/redis-developer/langgraph-redis/blob/main/CLAUDE.md

Installs all project dependencies, including extras, using Poetry. This command is essential for setting up the development environment.

```bash
poetry install --all-extras
```

--------------------------------

### Run Docker Compose

Source: https://github.com/redis-developer/langgraph-redis/blob/main/examples/README.md

Use this command to start the Docker containers for running the Jupyter notebooks. Ensure Docker and Docker Compose are installed.

```bash
docker compose up
```

--------------------------------

### Setup RedisSaver Indices

Source: https://github.com/redis-developer/langgraph-redis/blob/main/docs/user_guide/checkpointers.md

Call `setup()` to create necessary RediSearch indices for checkpoint queries. This operation is idempotent.

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

--------------------------------

### Create RedisStore with Existing Client

Source: https://github.com/redis-developer/langgraph-redis/blob/main/docs/user_guide/stores.md

Instantiate RedisStore by passing an existing Redis client. Remember to call `setup()` on the store.

```python
from redis import Redis
from langgraph.store.redis import RedisStore

client = Redis.from_url("redis://localhost:6379")
store = RedisStore(client)
store.setup()
```

--------------------------------

### Complete Azure Redis Example for LangGraph

Source: https://github.com/redis-developer/langgraph-redis/blob/main/docs/user_guide/azure_enterprise.md

This example demonstrates setting up RedisSaver and RedisStore with TTL configuration for a LangGraph application deployed on Azure. It shows how to initialize the Redis client, configure TTL for 2-hour expiration with refresh on read, and compile the graph with these components. The example also includes running the graph with a specific thread configuration.

```python
from redis import Redis
from langgraph.checkpoint.redis import RedisSaver
from langgraph.store.redis import RedisStore

# Single client for Azure Managed Redis
client = Redis(
    host="my-redis.eastus.redis.azure.net",
    port=10000,
    password="your-access-key",
    ssl=True,
)

# TTL configured for 2-hour expiration
ttl_config = {"default_ttl": 120, "refresh_on_read": True}

saver = RedisSaver(redis_client=client, ttl=ttl_config)
saver.setup()

store = RedisStore(client, ttl=ttl_config)
store.setup()

graph = builder.compile(checkpointer=saver, store=store)

# Run with thread management
config = {"configurable": {"thread_id": "azure-session-1"}}
result = graph.invoke({"messages": ["Hello from Azure"]}, config)
```

--------------------------------

### Install langgraph-checkpoint-redis with Poetry

Source: https://github.com/redis-developer/langgraph-redis/blob/main/docs/user_guide/installation.md

Use this command to install the package using Poetry.

```bash
poetry add langgraph-checkpoint-redis
```

--------------------------------

### Create RedisStore with Managed Connection

Source: https://github.com/redis-developer/langgraph-redis/blob/main/docs/user_guide/stores.md

Use `from_conn_string` for a RedisStore with managed connection lifecycle. Ensure to call `setup()` after creation.

```python
from langgraph.store.redis import RedisStore

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

--------------------------------

### Install LangGraph and Langchain-OpenAI

Source: https://github.com/redis-developer/langgraph-redis/blob/main/examples/react-agent/create-react-agent-memory.ipynb

Installs the necessary libraries for LangGraph and OpenAI integration. Use this command in your environment to set up the project.

```python
%%capture --no-stderr
%pip install -U langgraph langchain-openai
```

--------------------------------

### Install LangGraph

Source: https://github.com/redis-developer/langgraph-redis/blob/main/examples/human_in_the_loop/dynamic_breakpoints.ipynb

Install the LangGraph library. This command ensures you have the latest version for development.

```python
%%capture --no-stderr
%pip install -U langgraph
```

--------------------------------

### Install LangGraph Redis and Langchain Packages

Source: https://github.com/redis-developer/langgraph-redis/blob/main/examples/middleware/middleware_tool_caching.ipynb

Installs necessary packages for LangGraph Redis and Langchain. It checks if langgraph-checkpoint-redis is already installed and installs it from PyPI if not. It also installs langchain and langchain-openai.

```python
%%capture --no-stderr
# When running via docker-compose, the local library is already installed via editable mount.
# Only install from PyPI if not already available.
try:
    import langgraph.middleware.redis
    print("langgraph-checkpoint-redis already installed")
except ImportError:
    %pip install -U langgraph-checkpoint-redis

%pip install -U langchain langchain-openai sentence-transformers
```

--------------------------------

### Launch Jupyter Notebook

Source: https://github.com/redis-developer/langgraph-redis/blob/main/examples/README.md

Starts the Jupyter Notebook server locally. Navigate to the desired notebook in your browser to run it.

```bash
jupyter notebook
```

--------------------------------

### Initialize and Use Redis Store

Source: https://github.com/redis-developer/langgraph-redis/blob/main/docs/concepts/stores.md

Demonstrates initializing a RedisStore and performing basic put and get operations. Requires a Redis client connection.

```python
from langgraph.store.redis import RedisStore

store = RedisStore(conn=redis_client)
store.setup()

# Store a user preference
store.put(("users", "user-123", "preferences"), "theme", {"value": "dark"})

# Retrieve it
item = store.get(("users", "user-123", "preferences"), "theme")

# List all items under a namespace prefix
items = store.search(("users", "user-123"))
```

--------------------------------

### Install langgraph-checkpoint-redis

Source: https://github.com/redis-developer/langgraph-redis/blob/main/docs/index.md

Install the package using pip. This is the first step before using any of the Redis-backed features for LangGraph.

```bash
pip install langgraph-checkpoint-redis
```

--------------------------------

### Install LangGraph and OpenAI Packages

Source: https://github.com/redis-developer/langgraph-redis/blob/main/examples/cross-thread/cross-thread-persistence.ipynb

Installs the necessary packages for LangGraph and OpenAI integration. Use this command in your environment to set up the project.

```python
%%capture --no-stderr
%pip install -U langchain_openai langgraph
```

--------------------------------

### Start Redis Stack with Docker

Source: https://github.com/redis-developer/langgraph-redis/blob/main/docs/user_guide/installation.md

For Redis versions prior to 8.0, use this command to start Redis Stack, which bundles the required RedisJSON and RediSearch modules. Verify module presence with `redis-cli INFO modules`.

```bash
docker run -d --name redis -p 6379:6379 redis/redis-stack-server:latest
```

--------------------------------

### Install LangGraph and Dependencies

Source: https://github.com/redis-developer/langgraph-redis/blob/main/examples/human_in_the_loop/review-tool-calls-openai.ipynb

Installs the necessary Python packages for LangGraph, Langchain-OpenAI, and httpx. Use this command in your environment to set up the project.

```python
%%capture --no-stderr
%pip install --quiet -U langgraph langchain-openai "httpx>=0.24.0,<1.0.0"
```

--------------------------------

### Install LangGraph and Dependencies

Source: https://github.com/redis-developer/langgraph-redis/blob/main/examples/human_in_the_loop/review-tool-calls.ipynb

Installs the required packages for LangGraph, Langchain Anthropic, and HTTP client. Use this command in your environment to set up the project.

```python
%%capture --no-stderr
%pip install --quiet -U langgraph langchain_anthropic "httpx>=0.24.0,<1.0.0"
```

--------------------------------

### Install LangGraph, Langchain, and OpenAI

Source: https://github.com/redis-developer/langgraph-redis/blob/main/examples/memory/semantic-search.ipynb

Installs the necessary libraries for building LangGraph applications with OpenAI embeddings and chat models.

```python
%%capture --no-stderr
%pip install -U langgraph langchain-openai langchain
```

--------------------------------

### Setup Redis Checkpointer and Workflow

Source: https://github.com/redis-developer/langgraph-redis/blob/main/examples/functional/persistence-functional.ipynb

Configures a Redis checkpointer for state persistence and defines the main workflow. The checkpointer is passed to the entrypoint decorator.

```python
from langchain_core.messages import BaseMessage
from langgraph.graph import add_messages
from langgraph.func import entrypoint, task
from langgraph.checkpoint.redis import RedisSaver


@task
def call_model(messages: list[BaseMessage]):
    response = model.invoke(messages)
    return response


# Set up Redis connection for checkpointer
REDIS_URI = "redis://redis:6379"
checkpointer = None
with RedisSaver.from_conn_string(REDIS_URI) as cp:
    cp.setup()
    checkpointer = cp


@entrypoint(checkpointer=checkpointer)
def workflow(inputs: list[BaseMessage], *, previous: list[BaseMessage]):
    if previous:
        inputs = add_messages(previous, inputs)

    response = call_model(inputs).result()
    return entrypoint.final(value=response, save=add_messages(inputs, response))
```

--------------------------------

### Install LangGraph and Langchain Anthropic

Source: https://github.com/redis-developer/langgraph-redis/blob/main/examples/memory/delete-messages.ipynb

Installs necessary libraries for LangGraph and Anthropic integration. Use this command to set up your environment.

```bash
%pip install --quiet -U langgraph langchain_anthropic

```

--------------------------------

### Async Redis Checkpoint Saver Setup and Usage

Source: https://github.com/redis-developer/langgraph-redis/blob/main/README.md

Demonstrates the asynchronous implementation of the Redis saver. Ensure to use `await` for all operations and `async with` for context management.

```python
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())
```

--------------------------------

### LangGraph RedisSaver Usage Example

Source: https://github.com/redis-developer/langgraph-redis/blob/main/docs/concepts/checkpointing.md

Example of initializing RedisSaver, setting up the graph, and invoking it with a specific thread configuration. This demonstrates how to save and resume graph execution states.

```python
from langgraph.checkpoint.redis import RedisSaver

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

    config = {"configurable": {"thread_id": "thread-1"}}
    result = graph.invoke({"messages": [("human", "Hi")]}, config)

    # Resume the same thread later
    result = graph.invoke({"messages": [("human", "What did I say?")]}, config)
```

--------------------------------

### Install LangGraph and OpenAI Packages

Source: https://github.com/redis-developer/langgraph-redis/blob/main/examples/react-agent/create-react-agent-manage-message-history.ipynb

Installs or upgrades the required packages for LangGraph and OpenAI integration, including a specific version of httpx.

```python
%%capture --no-stderr
%pip install -U langgraph langchain-openai "httpx>=0.24.0,<1.0.0"
```

--------------------------------

### Initialize and Use ShallowRedisSaver

Source: https://github.com/redis-developer/langgraph-redis/blob/main/docs/user_guide/shallow_checkpointers.md

Instantiate and use the synchronous shallow checkpointer with a Redis connection string. Ensure to call `setup()` before compiling the graph.

```python
from langgraph.checkpoint.redis.shallow import ShallowRedisSaver

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

    config = {"configurable": {"thread_id": "shallow-thread"}}
    result = graph.invoke(inputs, config)
```

--------------------------------

### Perform Fresh Start Data Migration

Source: https://github.com/redis-developer/langgraph-redis/blob/main/docs/user_guide/migration.md

For development environments, clear existing checkpoint data from Redis and re-initialize indices using RedisSaver. This is the recommended approach for starting fresh.

```python
import redis

client = redis.from_url("redis://localhost:6379")
# Clear old checkpoint data
for key in client.scan_iter("checkpoint:*"):
    client.delete(key)
for key in client.scan_iter("checkpoint_write:*"):
    client.delete(key)

# Recreate indices with the new version
from langgraph.checkpoint.redis import RedisSaver
with RedisSaver.from_conn_string("redis://localhost:6379") as saver:
    saver.setup()
```

--------------------------------

### Start Redis Stack in Docker

Source: https://github.com/redis-developer/langgraph-redis/blob/main/README.md

Launch Redis Stack, including RedisJSON and RediSearch modules, using Docker.

```bash
make redis-start
```

--------------------------------

### Configure Checkpointer TTL with Direct Constructor

Source: https://github.com/redis-developer/langgraph-redis/blob/main/docs/user_guide/ttl.md

Instantiate RedisSaver directly, providing the Redis URL and TTL configuration. Ensure setup is called.

```python
from langgraph.checkpoint.redis import RedisSaver

saver = RedisSaver(
    redis_url="redis://localhost:6379",
    ttl={"default_ttl": 60, "refresh_on_read": True},
)
saver.setup()
```

--------------------------------

### Start Redis 8 with Docker

Source: https://github.com/redis-developer/langgraph-redis/blob/main/docs/user_guide/installation.md

This command starts a Redis 8 server using Docker, which includes RedisJSON and RediSearch by default. Ensure RedisJSON and RediSearch are present by running `redis-cli INFO modules`.

```bash
docker run -d --name redis -p 6379:6379 redis:8
```

--------------------------------

### Vector Search Configuration

Source: https://github.com/redis-developer/langgraph-redis/blob/main/docs/user_guide/stores.md

Guides on configuring RedisStore for vector similarity search, including index setup and embedding models.

```APIDOC
## Vector Search Configuration

### Description
Configure `RedisStore` to support vector similarity search by defining an index configuration.

### Method
`RedisStore.from_conn_string` with `index` parameter

### Endpoint
N/A (Client-side library)

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
```python
from langgraph.store.redis import RedisStore
from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

index_config = {
    "dims": 1536,
    "distance_type": "cosine",
    "fields": ["text"],
    "embed": embeddings,
}

with RedisStore.from_conn_string(
    "redis://localhost:6379",
    index=index_config,
) as store:
    store.setup()
    # Store items with text fields for embedding
    store.put(
        ("docs",), "intro",
        {"text": "LangGraph enables stateful AI workflows."},
    )
```

### Response
#### Success Response (200)
N/A (Configuration)

#### Response Example
N/A
```

--------------------------------

### Apply TTL Propagation Example

Source: https://github.com/redis-developer/langgraph-redis/blob/main/docs/concepts/ttl.md

Illustrates how TTL is applied atomically to a checkpoint and its related keys (blob, pending write) using Redis commands.

```text
checkpoint:thread-1::01JEXAMPLE          -> EXPIRE 3600
checkpoint_blob:thread-1::messages:v3    -> EXPIRE 3600
checkpoint_write:thread-1::01JEXAMPLE:t1 -> EXPIRE 3600
```

--------------------------------

### Standard Redis Checkpoint Saver Setup and Usage

Source: https://github.com/redis-developer/langgraph-redis/blob/main/README.md

Demonstrates setting up and using the standard Redis saver for checkpointing. Remember to call `.setup()` to initialize required indices before use.

```python
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))
```

--------------------------------

### Add Redis Persistence to LangGraph

Source: https://github.com/redis-developer/langgraph-redis/blob/main/docs/user_guide/getting_started.md

Compiles a LangGraph with Redis persistence using RedisSaver. Requires a running Redis 8.0+ instance and `langgraph-checkpoint-redis` installed. The `setup()` method creates necessary search indices.

```python
from langgraph.checkpoint.redis import RedisSaver

REDIS_URL = "redis://localhost:6379"

with RedisSaver.from_conn_string(REDIS_URL) as checkpointer:
    checkpointer.setup()  # Create search indices in Redis

    # Compile the graph with the Redis checkpointer
    graph = builder.compile(checkpointer=checkpointer)

    # Run the graph with a thread ID
    config = {"configurable": {"thread_id": "my-first-thread"}}
    result = graph.invoke({"messages": ["Hi there"]}}, config)
    print(result["messages"])
    # Output: ['Hi there', 'Hello! How can I help you?', 'Goodbye!']
```

--------------------------------

### Async Checkpoint Persistence with AsyncRedisSaver

Source: https://context7.com/redis-developer/langgraph-redis/llms.txt

The async variant of the checkpointer for use with asyncio applications. It supports async setup, invocation, getting tuples, and listing checkpoints. It can be initialized with a connection string or a direct async Redis client.

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


async def main():
    # Using connection string
    async with AsyncRedisSaver.from_conn_string("redis://localhost:6379") as saver:
        await saver.asetup()

        graph = builder.compile(checkpointer=saver)

        config = {"configurable": {"thread_id": "async-thread"}}
        result = await graph.ainvoke({"messages": ["Hi"]}}, config)
        print(result["messages"])

        # Async get tuple
        checkpoint_tuple = await saver.aget_tuple(config)

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

    # Or with a direct async client
    client = AsyncRedis(host="localhost", port=6379)
    saver = AsyncRedisSaver(redis_client=client)
    await saver.asetup()
    # ... use saver ...
    await client.aclose()


asyncio.run(main())
```

--------------------------------

### Install LangGraph and Dependencies

Source: https://github.com/redis-developer/langgraph-redis/blob/main/examples/cross-thread/cross-thread-persistence-functional.ipynb

Install the necessary packages for LangGraph, Anthropic, and OpenAI. The `--capture --no-stderr` flags are used to suppress output during installation.

```python
%%capture --no-stderr
%pip install -U langchain_anthropic langchain_openai langgraph
```

--------------------------------

### Cache Key Format Example

Source: https://github.com/redis-developer/langgraph-redis/blob/main/docs/design/tool-caching.md

Demonstrates the structure of a tool cache key, combining configuration name, tool name, and sorted JSON arguments. This ensures consistent cache lookups.

```python
toolcache:search:{"page": 1, "query": "redis"}
```

--------------------------------

### Build a ReAct-style Agent with Redis Checkpointing

Source: https://github.com/redis-developer/langgraph-redis/blob/main/examples/human_in_the_loop/time-travel.ipynb

Sets up a LangGraph agent using OpenAI models, Redis for state persistence, and tools for actions. This example defines tools, a model, and the graph structure for a ReAct agent.

```python
# Set up the tool
import uuid
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph.graph import MessagesState, START
from langgraph.prebuilt import ToolNode
from langgraph.graph import END, StateGraph
from langgraph.checkpoint.redis import RedisSaver

# Set up Redis connection
REDIS_URI = "redis://redis:6379"
memory = None
with RedisSaver.from_conn_string(REDIS_URI) as cp:
    cp.setup()
    memory = cp


@tool
def play_song_on_spotify(song: str):
    """Play a song on Spotify"""
    # Call the spotify API ...
    return f"Successfully played {song} on Spotify!"


@tool
def play_song_on_apple(song: str):
    """Play a song on Apple Music"""
    # Call the apple music API ...
    return f"Successfully played {song} on Apple Music!"


tools = [play_song_on_apple, play_song_on_spotify]
tool_node = ToolNode(tools)

# Set up the model

model = ChatOpenAI(model="gpt-4o-mini")
model = model.bind_tools(tools, parallel_tool_calls=False)


# Define nodes and conditional edges


# Define the function that determines whether to continue or not
def should_continue(state):
    messages = state["messages"]
    last_message = messages[-1]
    # If there is no function call, then we finish
    if not last_message.tool_calls:
        return "end"
    # Otherwise if there is, we continue
    else:
        return "continue"


# Define the function that calls the model
def call_model(state):
    messages = state["messages"]
    response = model.invoke(messages)
    # We return a list, because this will get added to the existing list
    return {"messages": [response]}


# Define a new graph
workflow = StateGraph(MessagesState)

# Define the two nodes we will cycle between
workflow.add_node("agent", call_model)
workflow.add_node("action", tool_node)

# Set the entrypoint as `agent`
# This means that this node is the first one called
workflow.add_edge(START, "agent")

```

--------------------------------

### Install LangMem Package

Source: https://github.com/redis-developer/langgraph-redis/blob/main/examples/react-agent/create-react-agent-manage-message-history.ipynb

Install the langmem package to utilize its features for memory management in LangGraph applications.

```python
%pip install -U langmem
```

--------------------------------

### Set up Redis connection for checkpointer

Source: https://github.com/redis-developer/langgraph-redis/blob/main/examples/memory/delete-messages.ipynb

Initializes a RedisSaver checkpointer using a connection string and sets up the workflow with this checkpointer.

```python
REDIS_URI = "redis://redis:6379"
memory = None
with RedisSaver.from_conn_string(REDIS_URI) as cp:
    cp.setup()
    memory = cp

app = workflow.compile(checkpointer=memory)
```

--------------------------------

### Define and Initialize Tools and Model

Source: https://github.com/redis-developer/langgraph-redis/blob/main/examples/human_in_the_loop/wait-user-input.ipynb

Sets up a search tool, a mock 'ask_human' tool using Pydantic, and an Anthropic chat model bound to these tools. Requires langchain-core >= 0.3 for Pydantic v2 compatibility.

```python
# Set up the state
from langgraph.graph import MessagesState, START

# Set up the tool
# We will have one real tool - a search tool
# We'll also have one "fake" tool - a "ask_human" tool
# Here we define any ACTUAL tools
from langchain_core.tools import tool
from langgraph.prebuilt import ToolNode


@tool
def search(query: str):
    """Call to surf the web."""
    # This is a placeholder for the actual implementation
    # Don't let the LLM know this though 😊
    return f"I looked up: {query}. Result: It's sunny in San Francisco, but you better look out if you're a Gemini 😈."


tools = [search]
tool_node = ToolNode(tools)

# Set up the model
from langchain_anthropic import ChatAnthropic

model = ChatAnthropic(model="claude-3-5-sonnet-latest")

from pydantic import BaseModel


# We are going "bind" all tools to the model
# We have the ACTUAL tools from above, but we also need a mock tool to ask a human
# Since `bind_tools` takes in tools but also just tool definitions,
# We can define a tool definition for `ask_human`
class AskHuman(BaseModel):
    """Ask the human a question"""

    question: str


model = model.bind_tools(tools + [AskHuman])
```

--------------------------------

### Configure RedisSaver with TTL

Source: https://github.com/redis-developer/langgraph-redis/blob/main/docs/concepts/ttl.md

Initialize RedisSaver with a default TTL of 60 minutes and enable TTL refresh on read.

```python
from langgraph.checkpoint.redis import RedisSaver

saver = RedisSaver.from_conn_string(
    "redis://localhost:6379",
    ttl={"default_ttl": 60, "refresh_on_read": True},
)
```

--------------------------------

### Set up the Language Model

Source: https://github.com/redis-developer/langgraph-redis/blob/main/examples/human_in_the_loop/breakpoints.ipynb

Initializes the ChatAnthropic model and binds it with the defined tools. This model will be used by the agent to decide actions.

```python
model = ChatAnthropic(model="claude-3-5-sonnet-20240620")
model = model.bind_tools(tools)
```

--------------------------------

### Run LangGraph to Approve Tool Call

Source: https://github.com/redis-developer/langgraph-redis/blob/main/examples/human_in_the_loop/review-tool-calls-openai.ipynb

Demonstrates running the LangGraph with an input that prompts a tool call. This setup is for approving a tool call, implying a subsequent human review step would occur.

```python
# Input
initial_input = {"messages": [{"role": "user", "content": "what's the weather in sf?"}]}

# Thread
thread = {"configurable": {"thread_id": "openai-2"}}


```

--------------------------------

### Install or Upgrade LangGraph

Source: https://github.com/redis-developer/langgraph-redis/blob/main/docs/user_guide/migration.md

If you encounter an ImportError for Interrupt, ensure that langgraph version 1.0.0 or later is installed. Upgrade using pip.

```bash
pip install --upgrade langgraph>=1.0.0
```

--------------------------------

### Build a ReAct Agent with Redis Checkpointer

Source: https://github.com/redis-developer/langgraph-redis/blob/main/examples/memory/delete-messages.ipynb

Sets up a ReAct agent using LangChain and LangGraph, integrating with Redis for state persistence. This example demonstrates tool usage and conditional graph execution.

```python
from typing import Literal

from langchain_anthropic import ChatAnthropic
from langchain_core.tools import tool

from langgraph.checkpoint.redis import RedisSaver
from langgraph.graph import MessagesState, StateGraph, START, END
from langgraph.prebuilt import ToolNode

# Set up Redis connection for checkpointer
REDIS_URI = "redis://redis:6379"
memory = None
with RedisSaver.from_conn_string(REDIS_URI) as cp:
    cp.setup()
    memory = cp


@tool
def search(query: str):
    """Call to surf the web."""
    # This is a placeholder for the actual implementation
    # Don't let the LLM know this though 😊
    return "It's sunny in San Francisco, but you better look out if you're a Gemini 😈."

tools = [search]
tool_node = ToolNode(tools)
model = ChatAnthropic(model_name="claude-3-haiku-20240307")
bound_model = model.bind_tools(tools)


def should_continue(state: MessagesState):
    """Return the next node to execute."""
    last_message = state["messages"][-1]
    # If there is no function call, then we finish
    if not last_message.tool_calls:
        return END
    # Otherwise if there is, we continue
    return "action"


# Define the function that calls the model
def call_model(state: MessagesState):
    response = model.invoke(state["messages"])
    # We return a list, because this will get added to the existing list
    return {"messages": response}


# Define a new graph
workflow = StateGraph(MessagesState)

# Define the two nodes we will cycle between
workflow.add_node("agent", call_model)
workflow.add_node("action", tool_node)

# Set the entrypoint as `agent`
# This means that this node is the first one called
workflow.add_edge(START, "agent")

# We now add a conditional edge
workflow.add_conditional_edges(
    # First, we define the start node. We use `agent`.
    # This means these are the edges taken after the `agent` node is called.
    "agent",
    # Next, we pass in the function that will determine which node is called next.
    should_continue,
    # Next, we pass in the path map - all the possible nodes this edge could go to
    ["action", END],
)

# We now add a normal edge from `tools` to `agent`.
# This means that after `tools` is called, `agent` node is called next.
workflow.add_edge("action", "agent")

# Finally, we compile it!
# This compiles it into a LangChain Runnable,

```

--------------------------------

### Initialize Redis Checkpointer and Graph

Source: https://github.com/redis-developer/langgraph-redis/blob/main/examples/subgraph/subgraphs-manage-state.ipynb

Set up a Redis connection for the checkpointer and define a graph with double-nested subgraphs. This example includes state definitions and node configurations for routing.

```python
from typing import Literal
from typing_extensions import TypedDict
from langgraph.checkpoint.redis import RedisSaver

# Set up Redis connection for checkpointer
REDIS_URI = "redis://redis:6379"
memory = None
with RedisSaver.from_conn_string(REDIS_URI) as cp:
    cp.setup()
    memory = cp


class RouterState(MessagesState):
    route: Literal["weather", "other"]


class Router(TypedDict):
    route: Literal["weather", "other"]


router_model = raw_model.with_structured_output(Router)


def router_node(state: RouterState):
    system_message = "Classify the incoming query as either about weather or not."
    messages = [{"role": "system", "content": system_message}] + state["messages"]
    route = router_model.invoke(messages)
    return {"route": route["route"]}


def normal_llm_node(state: RouterState):
    response = raw_model.invoke(state["messages"])
    return {"messages": [response]}


def route_after_prediction(
        state: RouterState,
) -> Literal["weather_graph", "normal_llm_node"]:
    if state["route"] == "weather":
        return "weather_graph"
    else:
        return "normal_llm_node"


graph = StateGraph(RouterState)
graph.add_node(router_node)
graph.add_node(normal_llm_node)
graph.add_node("weather_graph", subgraph)
graph.add_edge(START, "router_node")
graph.add_conditional_edges("router_node", route_after_prediction)
graph.add_edge("normal_llm_node", END)
graph.add_edge("weather_graph", END)
graph = graph.compile(checkpointer=memory)
```

--------------------------------

### Create RedisStore Instance

Source: https://github.com/redis-developer/langgraph-redis/blob/main/examples/cross-thread/cross-thread-persistence-functional.ipynb

Instantiate a RedisStore for persistence. Ensure your Redis connection string is correctly formatted.

```python
from langgraph.store.redis import RedisStore, BaseStore

store = RedisStore.from_conn_string("redis://redis:6379")
```

--------------------------------

### Install LangGraph and Anthropic Packages

Source: https://github.com/redis-developer/langgraph-redis/blob/main/examples/functional/persistence-functional.ipynb

Installs the required Python packages for LangGraph and Anthropic integration. Use this command in your environment to set up the project.

```python
%%capture --no-stderr
%pip install --quiet -U langgraph langchain_anthropic
```

--------------------------------

### Configure and Run Agent Stream

Source: https://github.com/redis-developer/langgraph-redis/blob/main/examples/react-agent/create-react-agent-hitl.ipynb

Sets up configuration with a unique thread ID and initial user input, then streams the agent's execution using the print_stream utility.

```python
import uuid

config = {"configurable": {"thread_id": str(uuid.uuid4())}}
inputs = {"messages": [("user", "what is the weather in SF, CA?")]}

print_stream(graph.stream(inputs, config, stream_mode="values"))
```

--------------------------------

### Start a New Conversation (Thread 2)

Source: https://github.com/redis-developer/langgraph-redis/blob/main/examples/functional/persistence-functional.ipynb

Starts a new conversation with a different thread ID, demonstrating that memory is reset for new threads.

```python
input_message = {"role": "user", "content": "what's my name?"}
for chunk in workflow.stream(
        [input_message],
        {"configurable": {"thread_id": "2"}},
        stream_mode="values",
):
    chunk.pretty_print()
```

--------------------------------

### Set up Redis Connection for State Saving

Source: https://github.com/redis-developer/langgraph-redis/blob/main/examples/human_in_the_loop/review-tool-calls.ipynb

Establishes a connection to Redis using a connection string and sets up the saver for state persistence. Ensure Redis is running at the specified URI.

```python
REDIS_URI = "redis://redis:6379"
memory = None
with RedisSaver.from_conn_string(REDIS_URI) as cp:
    cp.setup()
    memory = cp
```

--------------------------------

### Install LangGraph and OpenAI Packages

Source: https://github.com/redis-developer/langgraph-redis/blob/main/examples/human_in_the_loop/time-travel.ipynb

Installs the necessary Python packages for LangGraph and OpenAI integration. Use this command in your environment to set up the required libraries.

```python
%%capture --no-stderr
%pip install --quiet -U langgraph langchain_openai
```

--------------------------------

### Check Installed Versions

Source: https://github.com/redis-developer/langgraph-redis/blob/main/MIGRATION_0.2.0.md

Command to check the installed versions of relevant LangGraph and related packages. This is crucial for ensuring compatibility and troubleshooting migration issues.

```bash
# Check installed versions
pip show langgraph-checkpoint-redis langgraph langgraph-checkpoint langchain-core

# Expected versions for 0.2.0:
# langgraph-checkpoint-redis: 0.2.0
# langgraph: >=1.0.0
# langgraph-checkpoint: >=3.0.0
# langchain-core: >=1.0.0
```

--------------------------------

### Configure parallel Redis deployments for migration

Source: https://github.com/redis-developer/langgraph-redis/blob/main/MIGRATION_0.2.0.md

This example shows how to set up two Redis Saver instances using different database numbers or key prefixes to run old and new versions in parallel during a production migration. Traffic is gradually shifted to the new version.

```python
# Deploy 0.2.0 with a different key prefix or database
REDIS_URL_NEW = "redis://localhost:6379/1"  # Different database
REDIS_URL_OLD = "redis://localhost:6379/0"  # Old database

# New code uses 0.2.0
from langgraph.checkpoint.redis import RedisSaver
saver_new = RedisSaver.from_conn_string(REDIS_URL_NEW)

# Gradually migrate traffic to new version
# Old traffic continues on 0.1.x until fully migrated
```

--------------------------------

### Check Installed Package Versions

Source: https://github.com/redis-developer/langgraph-redis/blob/main/docs/user_guide/migration.md

Verify the installed versions of key packages, including langgraph-checkpoint-redis, langgraph, and langchain-core, using the pip show command.

```bash
pip show langgraph-checkpoint-redis langgraph langgraph-checkpoint langchain-core
```

--------------------------------

### Shared Redis Client Connection Example

Source: https://github.com/redis-developer/langgraph-redis/blob/main/docs/concepts/architecture.md

Demonstrates how to create a single Redis client instance and inject it into multiple components (RedisSaver and RedisStore) to share the connection across checkpointers, stores, and middleware.

```python
from redis import Redis

redis_client = Redis.from_url("redis://localhost:6379")

saver = RedisSaver(redis_client=redis_client)
store = RedisStore(conn=redis_client)
```

--------------------------------

### Install LangGraph Redis Dependencies

Source: https://github.com/redis-developer/langgraph-redis/blob/main/examples/README.md

Installs the necessary Python packages for running LangGraph notebooks with Redis persistence locally. Ensure Redis is running.

```bash
pip install langgraph-checkpoint-redis
pip install langgraph>=0.3.0
pip install jupyter redis>=5.2.1 redisvl>=0.11.0
pip install langchain-openai langchain-anthropic
pip install python-ulid "httpx>=0.24.0,<1.0.0"
```

--------------------------------

### Configure RedisStore with Default TTL

Source: https://github.com/redis-developer/langgraph-redis/blob/main/docs/concepts/ttl.md

Initialize a RedisStore with a default TTL of 24 hours (1440 minutes) for all items.

```python
from langgraph.store.redis import RedisStore

store = RedisStore(
    conn=redis_client,
    ttl={"default_ttl": 1440},
)
```

--------------------------------

### Create Async Redis Checkpointer and Middleware

Source: https://github.com/redis-developer/langgraph-redis/blob/main/examples/middleware/middleware_composition.ipynb

This Python snippet demonstrates setting up an asynchronous Redis checkpointer and an integrated Redis middleware stack. The middleware shares the Redis connection with the checkpointer, optimizing resource usage. Ensure REDIS_URL is correctly configured.

```python
from langgraph.checkpoint.redis.aio import AsyncRedisSaver
from langgraph.middleware.redis import IntegratedRedisMiddleware

# Create async checkpointer
async_checkpointer = AsyncRedisSaver(redis_url=REDIS_URL)
await async_checkpointer.asetup()

# Create middleware stack that shares connection with checkpointer
integrated_cache_name = f"integrated_cache_{uuid.uuid4().hex[:8]}"

integrated_stack = IntegratedRedisMiddleware.from_saver(
    async_checkpointer,
    configs=[
        SemanticCacheConfig(name=integrated_cache_name, ttl_seconds=3600),
    ],
)

print("Created IntegratedRedisMiddleware from AsyncRedisSaver!")
print(f"Number of middleware: {len(integrated_stack._middlewares)}")
```

--------------------------------

### Set up Redis connection for checkpointer

Source: https://github.com/redis-developer/langgraph-redis/blob/main/examples/memory/semantic-search.ipynb

Initializes a RedisSaver instance for checkpointing. Ensure the Redis server is accessible at the specified URI.

```python
REDIS_URI = "redis://redis:6379"
checkpointer = None
with RedisSaver.from_conn_string(REDIS_URI) as cp:
    cp.setup()
    checkpointer = cp
```

--------------------------------

### Basic Redis Store Usage

Source: https://github.com/redis-developer/langgraph-redis/blob/main/README.md

Initialize and use a RedisStore for basic key-value storage. Ensure the connection string is correct.

```python
from langgraph.store.redis import RedisStore

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

--------------------------------

### Get Current Messages

Source: https://github.com/redis-developer/langgraph-redis/blob/main/examples/memory/delete-messages.ipynb

Retrieve the current list of messages in the thread to identify messages for deletion.

```python
messages = app.get_state(config).values["messages"]
messages
```

--------------------------------

### Initialize LLM with Tool Bindings

Source: https://github.com/redis-developer/langgraph-redis/blob/main/examples/human_in_the_loop/review-tool-calls.ipynb

Initializes the ChatAnthropic model and binds the defined weather_search tool to it, enabling the model to use this tool.

```python
model = ChatAnthropic(model_name="claude-3-5-sonnet-latest").bind_tools([weather_search])
```

--------------------------------

### Check Python Version

Source: https://github.com/redis-developer/langgraph-redis/blob/main/docs/user_guide/installation.md

Verify that your Python installation meets the minimum requirement of Python 3.10 or higher.

```bash
python --version
# Should show 3.10.x or higher
```

--------------------------------

### Configure ShallowRedisSaver

Source: https://github.com/redis-developer/langgraph-redis/blob/main/MIGRATION_0.2.0.md

Example of configuring ShallowRedisSaver with key and channel cache sizes. This is used for shallow checkpointing.

```python
from langgraph.checkpoint.redis.shallow import ShallowRedisSaver

saver = ShallowRedisSaver(
    redis_url="redis://localhost:6379",
    key_cache_max_size=2000,
    channel_cache_max_size=200
)
```

--------------------------------

### Set up Redis Connection for Checkpointer

Source: https://github.com/redis-developer/langgraph-redis/blob/main/examples/react-agent/create-react-agent-manage-message-history.ipynb

Initializes a RedisSaver checkpointer using a connection string. This checkpointer is then passed to the create_react_agent function.

```python
REDIS_URI = "redis://redis:6379"
checkpointer = None
with RedisSaver.from_conn_string(REDIS_URI) as cp:
    cp.setup()
    checkpointer = cp

graph = create_react_agent(
    model,
    tools,
    # highlight-next-line
    pre_model_hook=pre_model_hook,
    checkpointer=checkpointer,
)
```

--------------------------------

### Upgrade redisvl for Security

Source: https://github.com/redis-developer/langgraph-redis/blob/main/MIGRATION_0.2.0.md

Mitigate known vulnerabilities (CVE-2025-64439) by ensuring you have redisvl version 0.11.0 or later installed.

```bash
pip install --upgrade redisvl>=0.11.0
```

--------------------------------

### Async Redis Checkpointer Setup

Source: https://context7.com/redis-developer/langgraph-redis/llms.txt

Use AsyncShallowRedisSaver for asynchronous checkpointing with Redis. Ensure Redis is running on localhost:6379.

```python
from langgraph.checkpoint.redis import AsyncShallowRedisSaver

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

--------------------------------

### Define State and Graph Nodes for Breakpoint Example

Source: https://github.com/redis-developer/langgraph-redis/blob/main/examples/human_in_the_loop/breakpoints.ipynb

Sets up the state definition using TypedDict and defines simple functions for graph nodes. It also initializes the StateGraph builder.

```python
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.redis import RedisSaver
from IPython.display import Image, display


class State(TypedDict):
    input: str


def step_1(state):
    print("---")
    pass


def step_2(state):
    print("---")
    pass


def step_3(state):
    print("---")
    pass


builder = StateGraph(State)
builder.add_node("step_1", step_1)
builder.add_node("step_2", step_2)
builder.add_node("step_3", step_3)
builder.add_edge(START, "step_1")
builder.add_edge("step_1", "step_2")
builder.add_edge("step_2", "step_3")
builder.add_edge("step_3", END)
```

--------------------------------

### Create Interrupt (Before 0.1.x)

Source: https://github.com/redis-developer/langgraph-redis/blob/main/MIGRATION_0.2.0.md

Example of creating an Interrupt object with the older API, including `resumable`, `ns`, and `when` fields.

```python
from langgraph.types import Interrupt

# Creating an interrupt
interrupt = Interrupt(
    value={"user_input": "data"},
    resumable=True,
    ns="my_namespace",
    when="before"
)

# Accessing fields
if interrupt.resumable:
    process(interrupt.value)
```