redis存储状态

发表于 更新于 分类于 阅读次数：

为什么用redis

Redis通过 RedisSessionManager 类来管理用户会话，存储结构如下：

1
2
3
4
5
6
7
session:{user_id} -> {
  "session_id": "会话ID",
  "status": "idle|running|interrupted|completed|error",
  "last_response": "上次智能体响应",
  "last_query": "用户上次查询",
  "last_updated": "最后更新时间戳"
}
image-20250809232853613

主要功能

  • 会话创建与维护 ：为每个用户创建唯一会话，支持会话超时自动清理
  • 状态跟踪 ：实时跟踪智能体执行状态（空闲、运行中、中断、完成、错误）
  • 中断恢复支持 ：当智能体需要人工干预时，Redis保存中断状态，支持后续恢复执行
  • 用户管理 ：统计活跃用户数量，管理多用户并发访问

与PostgreSQL的分工

  • Redis ：负责临时会话状态和实时数据（快速读写）
  • PostgreSQL ：负责智能体的长期记忆存储（通过LangGraph的checkpointer）

为什么不使用pgsql完成对状态的存储

频繁读写 ：会话状态需要频繁更新（每次请求都要更新状态），PostgreSQL的磁盘I/O比Redis内存操作慢很多4

短期记忆（PostgreSQL + LangGraph Checkpointer）

临时状态记忆（Redis）

redis实现状态存储业务逻辑总览图

image-20250806164348663

使用redis的根本逻辑：存储对话的状态，当出现由工具调用或者客户端崩溃导致的中断时，可以存储状态在redis，在开始对话时，通过session_id获取redis的状态，并根据状态判断是要恢复中断还是正常对话

存储的redis（调用invoke_agent接口）：开始（创建）对话时要根据会话user_id获取或创建redis；再调用agent后，根据响应是否存在status字段是否是”interrupt”，判断是否有终端，最后更新redis状态

恢复的redis（调用resume_agent接口）：获取redis状态，并根据请求的恢复内容，使用Command命令恢复agent，最后更新redis

image-20250809233037563

redis类

1
2
3
4
5
6
7
8
9
10
11
12
13
14
# 初始化异步 Redis 连接和会话配置
def __init__(self, redis_host, redis_port, redis_db, session_timeout):
    self.redis_client = redis.Redis(
        host=redis_host,
        port=redis_port,
        db=redis_db,
        decode_responses=True
    )
    self.session_timeout = session_timeout  # 会话过期时间（秒）

# 关闭 Redis 连接
async def close(self):
    await self.redis_client.close()

方法名 作用 输入参数 返回值 备注
__init__ 建立与 Redis 的异步连接并设置会话超时 redis_host, redis_port, redis_db, session_timeout - decode_responses=True 使 Redis 返回字符串而非字节
close 优雅关闭 Redis 连接 - - 异步方法，需 await
create_session 为指定用户新建（或覆盖）会话记录 user_id, 可选 session_id, status, last_query, last_response, last_updated str：生成的 session_id 会话键格式：session:{user_id}；过期时间为 session_timeout
get_session 读取指定用户的完整会话字典 user_id dict 或 None 自动将 JSON 里的 last_response 反序列化为 AgentResponse 对象
update_session 增量更新已有会话的字段 user_id, 可选 status, last_query, last_response, last_updated bool：True 更新成功，False 用户不存在 更新后刷新过期时间
delete_session 删除单个用户的会话 user_id bool：True 删除成功 直接删除 session:{user_id}
get_session_count 计算当前活跃会话总数 - int 使用异步扫描 session:* 键空间
get_all_user_ids 取出所有已创建会话的 user_id - List[str] 同样基于 session:* 扫描
user_id_exists 快速判断某用户是否已有会话 user_id bool 利用 EXISTS 命令

安装redis

linux系统

1
2
3
4
5
6
sudo apt update
sudo apt install -y redis-server
# 启动 Redis 服务
sudo service redis-server start
# 检查 Redis 服务状态
sudo service redis-server status

docker

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
# Docker Compose 配置文件，用于启动 Redis 服务
# 该配置为 FastAPI 应用提供 Redis 后端，支持分布式会话管理
version: '3.8'

services:
  redis:
    # 使用官方 Redis 镜像
    image: redis:latest
    # 服务名称
    container_name: redis
    # 映射 Redis 默认端口到主机
    ports:
      - "6379:6379"
    # 持久化存储配置（可选）
    volumes:
      - redis-data:/data
    # 确保容器在重启时自动启动
    restart: unless-stopped
    # 健康检查：验证 Redis 服务是否正常运行
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 30s
      timeout: 10s
      retries: 3
      start_period: 10s
    # 网络配置
    networks:
      - app-network

# 定义持久化存储卷
volumes:
  redis-data:
    name: redis-data

# 定义网络
networks:
  app-network:
    driver: bridge
1
docker run -d  --name redis -p 6379:6379  -v redis-data:/data redis:latest