此存储库包含 LangGraph 的 Redis 实现，提供 Checkpoint 保存器和存储功能。
该项目由两个主要组件组成：
Redis Checkpoint 保存器 ：使用 Redis 存储和管理检查点的实现
Redis 存储 ：基于 Redis 的键值存储，具有可选的向量搜索功能
该项目需要以下主要的 Python 依赖项：
redis>=5.2.1redisvl>=0.5.1langgraph-checkpoint>=2.0.24
重要: 此库需要以下模块支持的 Redis：
RedisJSON - 用于存储和操作 JSON 数据
RediSearch - 用于搜索和索引功能
如果你使用的是 Redis 8.0 或更高版本，RedisJSON 和 RediSearch 模块已作为 Redis 核心发行版的一部分默认包含在内，无需额外安装。
如果你使用的 Redis 版本低于 8.0，你需要确保已安装这些模块：
使用 Redis Stack，它将 Redis 与这些模块捆绑在一起
或者分别在你的 Redis 实例中安装这些模块
如果缺少这些模块，将在索引创建和检查点操作期间引发错误。
使用 pip 安装库：
pip install langgraph-checkpoint-redis重要
首次使用 Redis 检查点时，请务必在其上调用 .setup() 方法以创建所需的索引。有关示例，请参见下文。
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))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())
浅层 Redis 检查点保存器仅将最新的检查点存储在 Redis 中。当不需要保留完整的检查点历史时，这些实现非常有用。
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 检查点保存器和存储器都支持时间戳功能以实现自动键过期：
# Configure TTL for checkpoint savers
ttl_config = {
"default_ttl": 60, # Default TTL in minutes
"refresh_on_read": True, # Refresh TTL when checkpoint is read
}
# Use with any checkpoint saver implementation
with RedisSaver.from_conn_string("redis://localhost:6379", ttl=ttl_config) as checkpointer:
checkpointer.setup()
# Use the checkpointer...
这使得管理存储变得容易，并确保临时数据自动被清理。
Redis 存储提供持久化的键值存储，并可选配向量搜索功能。
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 TTL configuration
ttl_config = {
"default_ttl": 60, # Default TTL in minutes
"refresh_on_read": True, # Refresh TTL when store entries are read
}
with RedisStore.from_conn_string(
"redis://localhost:6379",
index=index_config,
ttl=ttl_config
) as store:
store.setup()
# Use the store with vector search and TTL capabilities...from langgraph.store.redis.aio import AsyncRedisStore
async def main():
# TTL also works with async implementations
ttl_config = {
"default_ttl": 60, # Default TTL in minutes
"refresh_on_read": True, # Refresh TTL when store entries are read
}
async with AsyncRedisStore.from_conn_string(
"redis://localhost:6379",
ttl=ttl_config
) as store:
await store.setup()
# Use the store asynchronously...
asyncio.run(main())The examples directory contains Jupyter notebooks demonstrating the usage of Redis with LangGraph:
persistence_redis.ipynb: Demonstrates the usage of Redis checkpoint savers with LangGraphcreate-react-agent-memory.ipynb: Shows how to create an agent with persistent memory using Rediscross-thread-persistence.ipynb: Demonstrates cross-thread persistence capabilitiespersistence-functional.ipynb: Shows functional persistence patterns with Redis
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
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
The Redis implementation creates these main indices using RediSearch:
- 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
Both Redis checkpoint savers and stores leverage Redis's native key expiration:
- Native Redis TTL: Uses Redis's built-in
EXPIREcommand - 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)
We welcome contributions! Here's how you can help:
-
Clone the repository:
git clone https://github.com/redis-developer/langgraph-redis cd langgraph-redis -
Install dependencies:
`poetry install --all-extras`
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 Stack in Docker (includes RedisJSON and RediSearch modules) make redis-stop # Stop Redis container
- 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
This project is licensed under the MIT License.
