LangGraph Redis

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

Overview

The project consists of three main components:

1. Redis Checkpoint Savers: Implementations for storing and managing checkpoints using Redis
2. Redis Stores: Redis-backed key-value stores with optional vector search capabilities
3. Redis Middleware: LangChain agent middleware for semantic caching, tool caching, and conversation memory

Dependencies

Python Dependencies

The project requires the following main Python dependencies:

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

Redis Modules Requirements

IMPORTANT: This library requires Redis with the following modules:

• RedisJSON - For storing and manipulating JSON data
• RediSearch - For search and indexing capabilities

Redis 8.0+

If you're using Redis 8.0 or higher, both RedisJSON and RediSearch modules are included by default as part of the core Redis distribution. No additional installation is required.

Redis < 8.0

If you're using a Redis version lower than 8.0, you'll need to ensure these modules are installed:

• Use Redis Stack, which bundles Redis with these modules
• Or install the modules separately in your Redis instance

Failure to have these modules available will result in errors during index creation and checkpoint operations.

Azure Managed Redis / Azure Cache for Redis / Redis Enterprise Configuration

If you're using Azure Managed Redis, Azure Cache for Redis (especially Enterprise tier) or Redis Enterprise, there are important configuration considerations:

Client Configuration

Azure Managed Redis, Azure Cache for Redis and Redis Enterprise use a proxy layer that makes the cluster appear as a single endpoint. This requires using a standard Redis client, not a cluster-aware client:

fromredisimportRedisfromlanggraph.checkpoint.redisimportRedisSaver# ✅ CORRECT: Use standard Redis client for Azure/Enterpriseclient=Redis(
host="your-cache.redis.cache.windows.net", # or your Redis Enterprise endpointport=6379, # or 10000 for Azure Managed Redis / Azure Enterprise with TLSpassword="your-access-key",
ssl=True, # Azure/Enterprise typically requires SSLssl_cert_reqs="required", # or "none" for self-signed certsdecode_responses=False# RedisSaver expects bytes
)
# Pass the configured client to RedisSaversaver=RedisSaver(redis_client=client)
saver.setup()
# ❌ WRONG: Don't use RedisCluster client with Azure/Enterprise# from redis.cluster import RedisCluster# cluster_client = RedisCluster(...) # This will fail with proxy-based deployments

Why This Matters

• Proxy Architecture: Azure Managed Redis, Azure Cache for Redis and Redis Enterprise use a proxy layer that handles cluster operations internally
• Automatic Detection: RedisSaver will correctly detect this as non-cluster mode when using the standard client
• No Cross-Slot Errors: The proxy handles key distribution, avoiding cross-slot errors

Azure Specific Settings

For Azure Managed Redis & Azure Cache for Redis Enterprise tier:

• Port: Use port 10000 with TLS, or 6379 for standard
• Modules: RediSearch and RedisJSON need to be selected at creation
• SSL/TLS: Always enabled, minimum TLS 1.2

Example for Azure Managed Redis, Azure Cache for Redis Enterprise:

client=Redis(
host="your-host-endpoint",
port=10000, # Enterprise TLS portpassword="your-access-key",
ssl=True,
ssl_cert_reqs="required",
decode_responses=False
)

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

fromlanggraph.checkpoint.redisimportRedisSaverwrite_config= {"configurable": {"thread_id": "1", "checkpoint_ns": ""}}
read_config= {"configurable": {"thread_id": "1"}}
withRedisSaver.from_conn_string("redis://localhost:6379") ascheckpointer:
# Call setup to initialize indicescheckpointer.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 checkpointcheckpointer.put(write_config, checkpoint, {}, {})
# Retrieve checkpointloaded_checkpoint=checkpointer.get(read_config)
# List all checkpointscheckpoints=list(checkpointer.list(read_config))

Async Implementation

fromlanggraph.checkpoint.redis.aioimportAsyncRedisSaverasyncdefmain():
write_config= {"configurable": {"thread_id": "1", "checkpoint_ns": ""}}
read_config= {"configurable": {"thread_id": "1"}}
asyncwithAsyncRedisSaver.from_conn_string("redis://localhost:6379") ascheckpointer:
# Call setup to initialize indicesawaitcheckpointer.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 checkpointawaitcheckpointer.aput(write_config, checkpoint, {}, {})
# Retrieve checkpointloaded_checkpoint=awaitcheckpointer.aget(read_config)
# List all checkpointscheckpoints= [casyncforcincheckpointer.alist(read_config)]
# Run the async main functionimportasyncioasyncio.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.

fromlanggraph.checkpoint.redis.shallowimportShallowRedisSaver# For async version: from langgraph.checkpoint.redis.ashallow import AsyncShallowRedisSaverwrite_config= {"configurable": {"thread_id": "1", "checkpoint_ns": ""}}
read_config= {"configurable": {"thread_id": "1"}}
withShallowRedisSaver.from_conn_string("redis://localhost:6379") ascheckpointer:
checkpointer.setup()
# ... rest of the implementation follows similar pattern

Redis Checkpoint TTL Support

Both Redis checkpoint savers and stores support automatic expiration using Redis TTL:

# Configure automatic expirationttl_config= {
"default_ttl": 60, # Expire checkpoints after 60 minutes"refresh_on_read": True, # Reset expiration time when reading checkpoints
}
withRedisSaver.from_conn_string("redis://localhost:6379", ttl=ttl_config) assaver:
saver.setup()
# Checkpoints will expire after 60 minutes of inactivity

When no TTL is configured, checkpoints are persistent (never expire automatically).

Removing TTL (Pinning Threads)

You can make specific checkpoints persistent by removing their TTL. This is useful for "pinning" important threads that should never expire:

fromlanggraph.checkpoint.redisimportRedisSaver# Create saver with default TTLsaver=RedisSaver.from_conn_string("redis://localhost:6379", ttl={"default_ttl": 60})
saver.setup()
# Save a checkpointconfig= {"configurable": {"thread_id": "important-thread", "checkpoint_ns": ""}}
saved_config=saver.put(config, checkpoint, metadata, {})
# Remove TTL from the checkpoint to make it persistentcheckpoint_id=saved_config["configurable"]["checkpoint_id"]
checkpoint_key=f"checkpoint:important-thread:__empty__:{checkpoint_id}"saver._apply_ttl_to_keys(checkpoint_key, ttl_minutes=-1)
# The checkpoint is now persistent and won't expire

When no TTL configuration is provided, checkpoints are persistent by default (no expiration).

This makes it easy to manage storage and ensure ephemeral data is automatically cleaned up while keeping important data persistent.

Redis Stores

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

Synchronous Implementation

fromlanggraph.store.redisimportRedisStore# Basic usagewithRedisStore.from_conn_string("redis://localhost:6379") asstore:
store.setup()
# Use the store...# With vector search configurationindex_config= {
"dims": 1536, # Vector dimensions"distance_type": "cosine", # Distance metric"fields": ["text"], # Fields to index
}
# With TTL configurationttl_config= {
"default_ttl": 60, # Default TTL in minutes"refresh_on_read": True, # Refresh TTL when store entries are read
}
withRedisStore.from_conn_string(
"redis://localhost:6379",
index=index_config,
ttl=ttl_config
) asstore:
store.setup()
# Use the store with vector search and TTL capabilities...

Async Implementation

fromlanggraph.store.redis.aioimportAsyncRedisStoreasyncdefmain():
# TTL also works with async implementationsttl_config= {
"default_ttl": 60, # Default TTL in minutes"refresh_on_read": True, # Refresh TTL when store entries are read
}
asyncwithAsyncRedisStore.from_conn_string(
"redis://localhost:6379",
ttl=ttl_config
) asstore:
awaitstore.setup()
# Use the store asynchronously...asyncio.run(main())

Redis Middleware for LangChain Agents

Redis middleware provides semantic caching, tool result caching, conversation memory, and semantic routing for LangChain agents. These middleware components integrate directly with langchain.agents.create_agent().

Key Features

• SemanticCacheMiddleware: Cache LLM responses by semantic similarity, reducing costs and latency
• ToolResultCacheMiddleware: Cache expensive tool executions (API calls, computations)
• ConversationMemoryMiddleware: Inject semantically relevant past messages into context
• SemanticRouterMiddleware: Route requests based on semantic matching

Quick Start

importastimportoperatorasopfromlangchain.agentsimportcreate_agentfromlangchain_core.messagesimportHumanMessagefromlangchain_core.toolsimporttoolfromlanggraph.middleware.redisimport (
SemanticCacheMiddleware,
SemanticCacheConfig,
ToolResultCacheMiddleware,
ToolCacheConfig,
)
# Safe math expression evaluator (no arbitrary code execution)SAFE_OPERATORS= {
ast.Add: op.add, ast.Sub: op.sub, ast.Mult: op.mul,
ast.Div: op.truediv, ast.Pow: op.pow, ast.USub: op.neg,
}
def_eval_expr(node):
ifisinstance(node, ast.Constant):
returnnode.valueelifisinstance(node, ast.BinOp) andtype(node.op) inSAFE_OPERATORS:
returnSAFE_OPERATORS[type(node.op)](_eval_expr(node.left), _eval_expr(node.right))
elifisinstance(node, ast.UnaryOp) andtype(node.op) inSAFE_OPERATORS:
returnSAFE_OPERATORS[type(node.op)](_eval_expr(node.operand))
raiseValueError(f"Unsupported expression")
defsafe_eval(expr: str) ->float:
return_eval_expr(ast.parse(expr, mode='eval').body)
# Define tools with cacheability metadata@tooldefcalculate(expression: str) ->str:
"""Evaluate a math expression."""returnstr(safe_eval(expression))
calculate.metadata= {"cacheable": True} # Deterministic - safe to cache@tooldefget_stock_price(symbol: str) ->str:
"""Get current stock price."""returnfetch_price(symbol)
get_stock_price.metadata= {"cacheable": False} # Temporal - don't cache# Create middlewaresemantic_cache=SemanticCacheMiddleware(
SemanticCacheConfig(
redis_url="redis://localhost:6379",
name="llm_cache",
distance_threshold=0.15,
ttl_seconds=3600,
deterministic_tools=["calculate"], # Safe to cache after these tools
)
)
tool_cache=ToolResultCacheMiddleware(
ToolCacheConfig(
redis_url="redis://localhost:6379",
name="tool_cache",
ttl_seconds=1800,
)
)
# Create agent with middlewareagent=create_agent(
model="gpt-4o-mini",
tools=[calculate, get_stock_price],
middleware=[semantic_cache, tool_cache],
)
# Use async invocation (middleware is async-first)result=awaitagent.ainvoke({"messages": [HumanMessage(content="Calculate 25 * 4")]})

Tool Cacheability

The tool cache uses a priority chain (inspired by SQL function volatility and MCP ToolAnnotations) to decide whether a tool call should be cached. The checks are evaluated in order — the first match wins:

Priority	Check	Result
1	metadata["cacheable"] is set	Use its boolean value (highest priority)
2	metadata["destructive"] == True	Never cache
3	metadata["volatile"] == True	Never cache
4	metadata["read_only"] and metadata["idempotent"]	Cache
5	Tool name matches a side-effect prefix	Never cache
6	Call args contain a volatile arg name	Never cache
7	Config whitelist / blacklist	Existing fallback

Tool Metadata

Control cacheability per-tool using LangChain's native metadata:

# Explicit cacheable flag (priority 1 — overrides everything)@tooldefsearch(query: str) ->str:
"""Search the web."""returnweb_search(query)
search.metadata= {"cacheable": True}
# MCP-style annotations (priorities 2–4)@tooldefsend_email(to: str, body: str) ->str:
"""Send an email."""returnsmtp_send(to, body)
send_email.metadata= {"destructive": True} # Never cached@tooldefget_exchange_rate(pair: str) ->str:
"""Get live exchange rate."""returnfetch_rate(pair)
get_exchange_rate.metadata= {"volatile": True} # Never cached@tooldeflookup_zip(code: str) ->str:
"""Look up a ZIP code."""returnzip_db.get(code)
lookup_zip.metadata= {"read_only": True, "idempotent": True} # Cached# StructuredTool with metadatafromlangchain_core.toolsimportStructuredToolget_weather=StructuredTool.from_function(
func=fetch_weather,
name="get_weather",
description="Get current weather",
metadata={"cacheable": False}, # Real-time data
)

Advanced Cache Key Configuration

ToolCacheConfig supports three opt-in fields for fine-grained cache key control. All default to None (disabled), so existing behavior is unchanged.

fromlanggraph.middleware.redisimport (
ToolResultCacheMiddleware,
ToolCacheConfig,
DEFAULT_VOLATILE_ARG_NAMES,
DEFAULT_SIDE_EFFECT_PREFIXES,
)
tool_cache=ToolResultCacheMiddleware(
ToolCacheConfig(
redis_url="redis://localhost:6379",
name="tool_cache",
ttl_seconds=1800,
# --- Volatile arg names (priority 6) ---# Tool calls containing these arg names at any nesting depth# are never cached. Useful for temporal arguments.# Use the built-in defaults or provide your own set.volatile_arg_names=DEFAULT_VOLATILE_ARG_NAMES,
# DEFAULT_VOLATILE_ARG_NAMES includes:# "timestamp", "current_time", "now", "date",# "today", "current_date", "current_timestamp"# --- Ignored arg names ---# These arg names are stripped from the cache key before# serialization. Two calls differing only in ignored args# will share the same cache entry.ignored_arg_names={"request_id", "trace_id", "correlation_id"},
# --- Side-effect prefixes (priority 5) ---# Tool names starting with these prefixes are never cached.# Use the built-in defaults or provide your own tuple.side_effect_prefixes=DEFAULT_SIDE_EFFECT_PREFIXES,
# DEFAULT_SIDE_EFFECT_PREFIXES includes:# "send_", "delete_", "create_", "update_", "remove_",# "write_", "post_", "put_", "patch_"
)
)

Volatile arg names — When a tool call's arguments contain a key matching one of these names (checked recursively at any nesting depth), the call is never cached. This prevents stale results for time-dependent queries like {"query": "weather", "timestamp": 1709827200}.

Ignored arg names — Per-request noise like request_id or trace_id inflates cache misses without affecting the tool's output. Stripping them from the cache key means two otherwise-identical calls will share a cache entry regardless of their tracking IDs.

Side-effect prefixes — Tools whose names start with mutating prefixes (e.g., send_email, delete_record, create_user) are never cached, since their results represent actions that should always execute. This can be overridden per-tool with metadata["cacheable"] = True.

Middleware Composition

Combine multiple middleware using MiddlewareStack or factory functions:

fromlanggraph.middleware.redisimportMiddlewareStack, from_configs# Option 1: Create stack directlystack=MiddlewareStack([
SemanticCacheMiddleware(SemanticCacheConfig(redis_url="redis://localhost:6379", name="llm_cache")),
ToolResultCacheMiddleware(ToolCacheConfig(redis_url="redis://localhost:6379", name="tool_cache")),
])
# Option 2: Use from_configs factory (shares Redis connection)stack=from_configs(
configs=[
SemanticCacheConfig(name="llm_cache", ttl_seconds=3600),
ToolCacheConfig(name="tool_cache", ttl_seconds=1800),
],
redis_url="redis://localhost:6379",
)
agent=create_agent(model="gpt-4o-mini", tools=tools, middleware=[stack])

Connection Sharing with Checkpointer

Share Redis connections between middleware and checkpointer:

fromlanggraph.checkpoint.redis.aioimportAsyncRedisSaverfromlanggraph.middleware.redisimportIntegratedRedisMiddleware# Create checkpointercheckpointer=AsyncRedisSaver(redis_url="redis://localhost:6379")
awaitcheckpointer.asetup()
# Create middleware that shares the connectionmiddleware=IntegratedRedisMiddleware.from_saver(
checkpointer,
configs=[
SemanticCacheConfig(name="llm_cache"),
ToolCacheConfig(name="tool_cache"),
],
)
agent=create_agent(
model="gpt-4o-mini",
tools=tools,
checkpointer=checkpointer,
middleware=[middleware],
)

Example Notebooks

See the examples/middleware/ directory for detailed notebooks:

• middleware_semantic_cache.ipynb: LLM response caching with semantic matching
• middleware_tool_caching.ipynb: Tool result caching with metadata-based control
• middleware_conversation_memory.ipynb: Semantic conversation history retrieval
• middleware_composition.ipynb: Combining middleware with checkpointers

Examples

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

Checkpoint and Store Examples

• 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

Middleware Examples (examples/middleware/)

• middleware_semantic_cache.ipynb: LLM response caching with semantic similarity matching
• middleware_tool_caching.ipynb: Tool result caching with metadata-based cacheability control
• middleware_conversation_memory.ipynb: Semantic conversation history and context injection
• middleware_composition.ipynb: Combining multiple middleware with shared Redis connections

Running Example Notebooks

To run the example notebooks with Docker:

1. Navigate to the examples directory:

   cd examples

2. Start the Docker containers:

   docker compose up

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

4. When finished, stop the containers:

   docker compose down

Implementation Details

Redis Module Usage

This implementation relies on specific Redis modules:

• RedisJSON: Used for storing structured JSON data as native Redis objects
• RediSearch: Used for creating and querying indices on JSON data

Indexing

The Redis implementation creates these main indices using RediSearch:

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

For Redis Stores with vector search:

1. Store Index: Main key-value store
2. Vector Index: Optional vector embeddings for similarity search

TTL Implementation

Both Redis checkpoint savers and stores leverage Redis's native key expiration:

• Native Redis TTL: Uses Redis's built-in EXPIRE command for setting TTL
• TTL Removal: Uses Redis's PERSIST command to remove TTL (with ttl_minutes=-1)
• Automatic Cleanup: Redis automatically removes expired keys
• Configurable Default TTL: Set a default TTL for all keys in minutes
• TTL Refresh on Read: Optionally refresh TTL when keys are accessed
• Applied to All Related Keys: TTL is applied to all related keys (checkpoint, blobs, writes)
• Persistent by Default: When no TTL is configured, keys are persistent (no expiration)

Contributing

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

Development Setup

1. Clone the repository:

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

2. 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

• Code Quality:

  make test-coverage # Run tests with coverage reporting
  make coverage-report # Generate coverage report without running tests
  make coverage-html # Generate HTML coverage report (opens in htmlcov/)
  make find-dead-code # Find unused code with vulture

• Redis for Development/Testing:

  make redis-start # Start Redis Stack in Docker (includes RedisJSON and RediSearch modules)
  make redis-stop # Stop Redis container

Contribution Guidelines

1. Create a new branch for your changes
2. Write tests for new functionality
3. Ensure all tests pass: make test
4. Format your code: make format
5. Run linting checks: make lint
6. Submit a pull request with a clear description of your changes
7. Follow Conventional Commits for commit messages

License

This project is licensed under the MIT License.