利用Redis实现 cache 与 session

「利用 Redis 实现 Cache + Session」的完整实战版，适合接入：

• Python
• FastAPI
• AI Chat App
• Agent 系统
• Web 后端

你前面已经在做：

• FastAPI
• Chatbot API
• Memory
• RAG
• Agent

所以 Redis 基本会成为你的“系统状态中心”。

Redis 是什么

Redis

Redis 本质：

• 内存数据库
• key-value 存储
• 超快
• 支持过期时间（TTL）
• 常用于：
  • cache
  • session
  • rate limit
  • queue
  • memory
  • pub/sub

一、Redis 做 Cache（缓存）

为什么需要 Cache

正常流程：

用户请求 ↓ 数据库查询 ↓ 返回结果

问题：

• 数据库慢
• AI 调用贵
• API 请求耗时

所以：

用户请求 ↓ Redis Cache ↓ 有缓存 → 直接返回 没有缓存 → 查数据库/API → 写入Redis

Cache 核心思想

if redis中有数据: 直接返回 else: 查数据库 写入redis 返回结果

安装 Redis

Docker 启动 Redis

docker run -d \ --name redis \ -p 6379:6379 \ redis

安装 Python Redis SDK

pip install redis

Python 连接 Redis

import redis r = redis.Redis( host="localhost", port=6379, decode_responses=True ) r.set("name", "bb") print(r.get("name"))

Cache 示例

场景：

用户信息查询。

数据库很慢：

import time def get_user_from_db(user_id): time.sleep(3) return { "id": user_id, "name": "Tom" }

Redis Cache 版本

import redis import json import time r = redis.Redis( host="localhost", port=6379, decode_responses=True ) def get_user_from_db(user_id): time.sleep(3) return { "id": user_id, "name": "Tom" } def get_user(user_id): cache_key = f"user:{user_id}" # 1. 先查缓存 cached_user = r.get(cache_key) if cached_user: print("来自 Redis Cache") return json.loads(cached_user) # 2. 查数据库 print("来自 Database") user = get_user_from_db(user_id) # 3. 写入缓存 r.setex( cache_key, 60, json.dumps(user) ) return user print(get_user(1)) print(get_user(1))

setex 是什么

r.setex(key, ttl, value)

等价于：

SET + EXPIRE

例如：

r.setex("user:1", 60, "Tom")

意思：

60 秒后自动删除

Cache 常见场景

1. AI Response Cache

用户：

"什么是RAG"

缓存：

key = "llm:什么是RAG"

避免：

• 重复调用 LLM
• 重复花钱
• 重复等待

2. Embedding Cache

key = f"embedding:{text}"

避免重复 embedding。

3. SQL Query Cache

key = f"sql:{query}"

4. API Cache

比如天气 API。

二、Redis 做 Session

Session 是什么

HTTP 是无状态的：

请求1 请求2 请求3

服务器默认：

不知道是不是同一个用户

所以需要：

Session

保存用户状态。

例如：

用户已登录 用户购物车 用户聊天历史 用户权限

Session 工作流程

用户登录 ↓ 服务器生成 session_id ↓ session数据存入Redis ↓ session_id 返回给浏览器 ↓ 浏览器以后都带 session_id

Redis 为什么适合 Session

因为：

• 快
• 自动过期
• 多服务共享
• 分布式友好

Session 数据结构

例如：

session:abc123

值：

{ "user_id": 1, "username": "bb" }

Session 示例

登录

import uuid import json import redis r = redis.Redis( host="localhost", port=6379, decode_responses=True ) def login(user_id): session_id = str(uuid.uuid4()) session_data = { "user_id": user_id } r.setex( f"session:{session_id}", 3600, json.dumps(session_data) ) return session_id

获取 Session

def get_session(session_id): data = r.get(f"session:{session_id}") if not data: return None return json.loads(data)

使用

session_id = login(1) print(session_id) session = get_session(session_id) print(session)

三、FastAPI + Redis Session

安装

pip install fastapi uvicorn redis

完整示例

from fastapi import FastAPI, Request from fastapi.responses import JSONResponse import redis import uuid import json app = FastAPI() r = redis.Redis( host="localhost", port=6379, decode_responses=True ) @app.post("/login") def login(): session_id = str(uuid.uuid4()) session_data = { "user_id": 1, "name": "bb" } r.setex( f"session:{session_id}", 3600, json.dumps(session_data) ) response = JSONResponse({ "message": "login success" }) response.set_cookie( key="session_id", value=session_id ) return response @app.get("/me") def me(request: Request): session_id = request.cookies.get("session_id") if not session_id: return { "error": "not logged in" } session = r.get(f"session:{session_id}") if not session: return { "error": "session expired" } return json.loads(session)

Session 生命周期

用户登录

Redis: session:abc123

TTL：

3600 秒

用户长期不访问

Redis 自动删除：

session expired

四、AI Agent 里 Redis 的作用

你现在做的 Agent / Chatbot 系统里，Redis 会非常重要。

1. Conversation Memory

conversation:user123

保存：

[ {"role": "user", "content": "..."}, {"role": "assistant", "content": "..."} ]

2. Tool Cache

避免重复 tool calling。

3. RAG Cache

缓存 retrieval 结果。

4. Streaming 状态

保存：

正在生成 token 数 任务状态

5. Rate Limiting

user:1:request_count

防止 API 被刷。

五、Redis 常见数据结构

String

r.set("name", "bb")

List

聊天历史。

r.rpush("chat", "hello")

Hash

用户对象。

r.hset("user:1", mapping={ "name": "bb", "age": 18 })

Set

标签系统。

Sorted Set

排行榜。

六、生产环境最佳实践

1. 永远设置 TTL

否则：

Redis 内存爆炸

2. Cache Key 规范化

例如：

user:1 session:abc embedding:xxx

3. 不要缓存超大对象

Redis 是内存数据库。

4. Session 不要存敏感信息

不要直接存：

• 密码
• API Key

七、整体架构

┌──────────┐ │ Frontend │ └────┬─────┘ │ HTTP Request │ ┌──────▼──────┐ │ FastAPI │ └──────┬──────┘ │ ┌───────────┴───────────┐ │ │ ┌────▼────┐ ┌────▼────┐ │ Redis │ │Database │ │ Cache │ │Postgres │ │ Session │ │MongoDB │ └─────────┘ └─────────┘

八、你下一步最值得做的项目

你现在已经可以组合：

• FastAPI
• Redis
• Memory
• Tool Calling
• RAG

建议下一步：

做一个：

AI Chat App

包含：

• Redis Session
• Redis Conversation Memory
• Streaming
• Tool Calling
• RAG
• User Login
• Chat History
• Docker Compose

这已经是“真正 AI 产品”的架构了。