此存储库包含 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())examples 目录包含展示使用 Redis 和 LangGraph 的 Jupyter 笔记本：
持久化_redis.ipynb：展示如何使用 Redis 检查点保存器与 LangGraph 结合
create-react-agent-memory.ipynb: 展示如何使用 Redis 创建具有持久化内存的代理
cross-thread-persistence.ipynb：展示跨线程持久化能力持久化功能.ipynb：展示使用 Redis 的持久化功能模式
要使用 Docker 运行示例笔记本：
-
导航到示例目录：cd examples -
启动 Docker 容器：docker compose up
-
在浏览器中打开控制台中显示的 URL（通常为 http://127.0.0.1:8888/tree），以访问 Jupyter。 -
完成后，停止容器：docker compose down
这个实现依赖于特定的 Redis 模块：
RedisJSON：用于将结构化 JSON 数据存储为原生 Redis 对象
RediSearch：用于在 JSON 数据上创建和查询索引
Redis 实现使用 RediSearch 创建这些主要索引：
检查点索引 ：存储检查点元数据和版本控制
频道值索引 ：存储与频道相关联的数据
写入索引 ：跟踪待处理的写入操作和中间状态
对于带有向量搜索的 Redis 存储：
存储索引 ：主键值存储
向量索引 ：可选的向量嵌入用于相似性搜索
Redis 检查点保存器和存储都利用了 Redis 的原生键过期功能：
原生 Redis TTL：使用 Redis 内置的EXPIRE命令
自动清理 ：Redis 会自动移除过期的键
可配置的默认过期时间(TTL)：以分钟为单位为所有键设置默认的 TTL（生存时间）
读取时刷新 TTL：可选的，在访问键时刷新 TTL
应用于所有相关键 ：TTL 适用于所有相关键（检查点、Blob、写入）
我们欢迎贡献！以下是你如何提供帮助：
-
克隆仓库：
git clone https://github.com/redis-developer/langgraph-redis cd langgraph-redis -
安装依赖项：
`poetry install --all-extras`
该项目包含几个用于开发的 make 命令：
-
测试 :make test # Run all tests make test-all # Run all tests including API tests
-
代码检测与格式化 :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:make redis-start # Start Redis Stack in Docker (includes RedisJSON and RediSearch modules) make redis-stop # Stop Redis container
为你的更改创建一个新的分支
为新功能编写测试
确保所有测试通过：make test
格式化你的代码：make format
运行 lint 检查：make lint
提交带有清晰变更描述的拉取请求
遵循 常规提交 消息
该项目根据 MIT 许可证授权。
