Implementation:Langchain ai Langgraph RedisCache

Knowledge Sources	Langchain_ai_Langgraph
Domains	Caching, Redis
Last Updated	2026-02-11 16:00 GMT

Overview

A Redis-based cache implementation with TTL support and configurable key prefixing for distributed LangGraph deployments.

Description

The RedisCache class extends `BaseCache` to provide a high-performance, distributed caching layer backed by Redis. It accepts any Redis client instance (sync or async) and stores serialized values using a configurable key prefix (defaulting to `"langgraph:cache:"`). Namespaces are encoded into the Redis key by joining namespace components with colons, producing keys in the format `prefix:ns1:ns2:key`.

Values are stored in a compact `"encoding:data"` binary format where the encoding type name is prepended to the serialized bytes. This allows the deserializer to determine the correct decoding strategy without requiring a separate metadata key. The class uses Redis `MGET` for efficient batch retrieval and `pipeline` for batch writes, minimizing round trips to the Redis server. When a TTL is provided, values are stored with `SETEX`; otherwise they persist indefinitely.

Error handling is designed for resilience: all Redis operations are wrapped in try/except blocks that silently return empty results or skip writes when Redis is unavailable. This ensures that cache failures do not crash the application; the system degrades gracefully to uncached behavior. The `clear` method supports both full prefix-based clearing and selective namespace-based eviction using Redis `KEYS` pattern matching.

Usage

Use `RedisCache` in production or multi-process deployments where cache data must be shared across multiple LangGraph instances. It is the recommended cache backend for distributed systems, containerized environments, and any scenario requiring high throughput and low latency caching with automatic TTL-based expiration managed natively by Redis.

Code Reference

Source Location

• Repository: Langchain_ai_Langgraph
• File: libs/checkpoint/langgraph/cache/redis/__init__.py

Signature

class RedisCache(BaseCache[ValueT]):
    def __init__(
        self,
        redis: Any,
        *,
        serde: SerializerProtocol | None = None,
        prefix: str = "langgraph:cache:",
    ) -> None: ...

    def get(self, keys: Sequence[FullKey]) -> dict[FullKey, ValueT]: ...
    async def aget(self, keys: Sequence[FullKey]) -> dict[FullKey, ValueT]: ...
    def set(self, mapping: Mapping[FullKey, tuple[ValueT, int | None]]) -> None: ...
    async def aset(self, mapping: Mapping[FullKey, tuple[ValueT, int | None]]) -> None: ...
    def clear(self, namespaces: Sequence[Namespace] | None = None) -> None: ...
    async def aclear(self, namespaces: Sequence[Namespace] | None = None) -> None: ...

Import

from langgraph.cache.redis import RedisCache

I/O Contract

Constructor Parameters

Parameter	Type	Required	Description
redis	`Any`	Yes	Redis client instance (e.g., `redis.Redis()` or `redis.asyncio.Redis()`)
serde	None`	No	Custom serializer; defaults to the base class default
prefix	`str`	No	Key prefix for all cache entries (default: `"langgraph:cache:"`)

get

Parameter	Type	Description
keys	`Sequence[FullKey]`	List of `(Namespace, key_str)` tuples to look up

Return Type	Description
`dict[FullKey, ValueT]`	Mapping of found keys to their deserialized values; empty dict on Redis failure

set

Parameter	Type	Description
mapping	None]]`	Keys mapped to `(value, ttl_seconds)` pairs; TTL of `None` means no expiry

clear

Parameter	Type	Description
namespaces	None`	Namespaces to clear; `None` clears all keys with the configured prefix

Internal Key Format

Component	Format	Example
Redis Key	`{prefix}{ns1}:{ns2}:{key}`	`langgraph:cache:embeddings:v1:doc_123`
Stored Value	`{encoding}:{serialized_bytes}`	`json:{"embedding": [0.1, 0.2]}`

Usage Examples

import redis
from langgraph.cache.redis import RedisCache

# Create a Redis-backed cache
client = redis.Redis(host="localhost", port=6379, db=0)
cache = RedisCache(redis=client, prefix="myapp:cache:")

# Store values with TTL
ns = ("embeddings", "v1")
cache.set({
    (ns, "doc_123"): ({"embedding": [0.1, 0.2, 0.3]}, 300),  # 5-minute TTL
    (ns, "doc_456"): ({"embedding": [0.4, 0.5, 0.6]}, None),  # No expiry
})

# Batch retrieve
results = cache.get([(ns, "doc_123"), (ns, "doc_456")])
for key, value in results.items():
    print(f"{key}: {value}")

# Clear a namespace
cache.clear(namespaces=[ns])

Related Pages

• Principle:Langchain_ai_Langgraph_Cache_Backend_Selection