09-使用Redis保存大模型多轮对话上下文

使用 Redis 保存大模型多轮对话上下文

系列：Python + FastAPI 大模型应用基础（第 9 篇）
目标：使用 Python 异步 Redis 保存近期对话，理解 TTL、原子写入、用户隔离、缓存回源和并发边界。

1. Redis 在对话系统中解决什么问题

第 8 篇已经使用 MySQL 持久化正式对话。如果每次模型调用都从 MySQL 查询全部历史，会产生：

• 数据库读取压力；
• 历史序列化开销；
• 更高的接口时延；
• 多个 FastAPI 进程无法共享内存缓存。

Redis 是内存数据库，适合保存“近期、频繁读取、允许过期”的上下文：

MySQL：正式会话记录，长期保存，可审计
Redis：近期上下文缓存，快速读取，可以失效

推荐的 Cache-Aside（旁路缓存）流程：

读取对话上下文
↓
Redis 命中？
├── 是：直接使用
└── 否：从 MySQL 读取 → 写回 Redis → 使用

Redis 不应成为唯一事实来源。缓存被清空、过期或故障时，系统应该能够从 MySQL 恢复。

2. 为什么不能使用一个全局 Python 列表

# 错误示例：只属于当前进程，重启后丢失
conversation_history: dict[str, list[dict[str, str]]] = {}

在多进程部署中：

进程 A 的内存历史 ≠ 进程 B 的内存历史

同一个用户的两次请求可能被负载均衡到不同进程，导致上下文不一致。Redis 提供了多个进程共享的外部状态。

3. Redis 数据结构如何选择

常见选择：

本文使用 Redis List：

RPUSH：在右侧追加消息
LRANGE：读取消息
LTRIM：只保留最近 N 条
EXPIRE：设置过期时间

4. Key（键）设计

示例：

ai:context:tenant_001:10001:550e8400-e29b-41d4-a716-446655440000

组成部分：

• 固定前缀 ai:context；
• 租户标识；
• 已验证用户 ID；
• 会话 UUID。

不能只使用 conversation_id 就默认拥有权限。调用缓存前，业务层必须已经验证当前用户确实拥有该会话。

不要使用：

KEYS ai:context:*

KEYS 可能扫描整个键空间并阻塞 Redis。管理任务应使用 SCAN 分批迭代，普通请求不应遍历全部会话键。

5. 创建项目

redis_context_demo/
├── context_store.py
├── cache_service.py
├── main.py
├── tests/
│ └── test_context_store.py
└── requirements.txt

requirements.txt：

fastapi>=0.115,<1
uvicorn[standard]>=0.30,<1
redis>=5,<7
pydantic>=2.7,<3

# 测试依赖
pytest>=8,<9
pytest-asyncio>=0.23,<1
fakeredis>=2.23,<3

安装：

python –m venv .venv
.\\.venv\\Scripts\\python.exe –m pip install –r requirements.txt

6. 定义消息结构与 Redis Store

新建 context_store.py：

import json
import re
from dataclasses import asdict, dataclass
from datetime import datetime, timezone
from uuid import UUID

from redis.asyncio import Redis
from redis.exceptions import RedisError

class ContextStoreError(RuntimeError):
"""Redis 上下文存储统一异常。"""

class ContextDataCorruptedError(ContextStoreError):
"""Redis 中的消息结构损坏。"""

@dataclass(frozen=True)
class ContextMessage:
"""缓存中的一条对话消息。"""

role: str
content: str
created_at: str

@classmethod
def create(cls, role: str, content: str) –> "ContextMessage":
"""校验并创建消息。"""

if role not in {"user", "assistant"}:
# system 规则应由服务端维护，不允许从缓存恢复为可信指令
raise ValueError("role 只能是 user 或 assistant")

cleaned_content = content.strip()
if not cleaned_content:
raise ValueError("消息内容不能为空")

return cls(
role=role,
content=cleaned_content,
created_at=datetime.now(timezone.utc).isoformat(),
)

def to_json(self) –> str:
"""序列化为紧凑 JSON。"""

return json.dumps(
asdict(self),
ensure_ascii=False,
separators=(",", ":"),
)

@classmethod
def from_json(cls, raw_value: str) –> "ContextMessage":
"""从 Redis 文本恢复并严格校验。"""

try:
data = json.loads(raw_value)
except json.JSONDecodeError as exc:
raise ContextDataCorruptedError("消息不是合法 JSON") from exc

if not isinstance(data, dict):
raise ContextDataCorruptedError("消息 JSON 必须是对象")

role = data.get("role")
content = data.get("content")
created_at = data.get("created_at")
if role not in {"user", "assistant"}:
raise ContextDataCorruptedError("消息 role 不合法")
if not isinstance(content, str) or not content.strip():
raise ContextDataCorruptedError("消息 content 不合法")
if not isinstance(created_at, str) or not created_at:
raise ContextDataCorruptedError("消息 created_at 不合法")

return cls(role=role, content=content, created_at=created_at)

class RedisContextStore:
"""使用 Redis List 保存近期对话。"""

TENANT_PATTERN = re.compile(r"^[a-zA-Z0-9_-]{1,64}$")

def __init__(
self,
redis: Redis,
*,
ttl_seconds: int = 3600,
max_messages: int = 20,
) –> None:
if ttl_seconds <= 0:
raise ValueError("ttl_seconds 必须大于 0")
if max_messages < 2:
raise ValueError("max_messages 不能小于 2")

self.redis = redis
self.ttl_seconds = ttl_seconds
self.max_messages = max_messages

def build_key(
self,
tenant_id: str,
owner_id: int,
conversation_id: str,
) –> str:
"""构造经过校验的 Redis Key，拒绝任意用户输入拼接。"""

if not self.TENANT_PATTERN.fullmatch(tenant_id):
raise ValueError("tenant_id 格式不合法")
if owner_id <= 0:
raise ValueError("owner_id 必须大于 0")

try:
normalized_uuid = str(UUID(conversation_id))
except ValueError as exc:
raise ValueError("conversation_id 必须是合法 UUID") from exc

return (
f"ai:context:{tenant_id}:{owner_id}:{normalized_uuid}"
)

async def append(
self,
key: str,
message: ContextMessage,
) –> None:
"""原子执行追加、截断和设置 TTL。"""

try:
# transaction=True 会使用 MULTI/EXEC 作为一个事务批次执行
async with self.redis.pipeline(transaction=True) as pipeline:
pipeline.rpush(key, message.to_json())
pipeline.ltrim(key, –self.max_messages, –1)

# TTL 只在写入时刷新，单纯读取不会永久延长缓存寿命
pipeline.expire(key, self.ttl_seconds)
await pipeline.execute()
except RedisError as exc:
raise ContextStoreError("写入 Redis 上下文失败") from exc

async def replace(
self,
key: str,
messages: list[ContextMessage],
) –> None:
"""缓存未命中时，用数据库历史重建 Redis List。"""

trimmed_messages = messages[–self.max_messages :]
try:
async with self.redis.pipeline(transaction=True) as pipeline:
pipeline.delete(key)
if trimmed_messages:
pipeline.rpush(
key,
*(message.to_json() for message in trimmed_messages),
)
pipeline.expire(key, self.ttl_seconds)
await pipeline.execute()
except RedisError as exc:
raise ContextStoreError("重建 Redis 上下文失败") from exc

async def get(self, key: str) –> list[ContextMessage]:
"""按写入顺序读取缓存消息。"""

try:
raw_messages = await self.redis.lrange(key, 0, –1)
except RedisError as exc:
raise ContextStoreError("读取 Redis 上下文失败") from exc

# decode_responses=True 时返回 str；配置错误时可能返回 bytes
if any(not isinstance(item, str) for item in raw_messages):
raise ContextDataCorruptedError(
"Redis 客户端必须启用 decode_responses=True"
)

return [
ContextMessage.from_json(item)
for item in raw_messages
]

async def delete(self, key: str) –> None:
"""用户删除会话时同步删除缓存。"""

try:
await self.redis.delete(key)
except RedisError as exc:
raise ContextStoreError("删除 Redis 上下文失败") from exc

7. 为什么追加、截断和过期要放在一个事务中

如果分三次独立执行：

RPUSH 成功
LTRIM 失败
EXPIRE 未执行

这个 Key 可能永久存在并无限增长。

使用 MULTI/EXEC 可以保证命令不会被其他客户端命令插入。需要注意，Redis 事务和关系型数据库事务不同：运行期命令错误不会自动像 MySQL 一样回滚之前命令，因此仍要提前校验参数。

8. 实现 Cache-Aside 回源

新建 cache_service.py：

from collections.abc import Awaitable, Callable

from context_store import (
ContextMessage,
ContextStoreError,
RedisContextStore,
)

HistoryLoader = Callable[[], Awaitable[list[ContextMessage]]]

class ContextCacheService:
"""编排 Redis 缓存和 MySQL 回源。"""

def __init__(self, store: RedisContextStore) –> None:
self.store = store

async def load_context(
self,
key: str,
mysql_loader: HistoryLoader,
) –> list[ContextMessage]:
"""Redis 命中则返回；未命中或故障则从 MySQL 获取。"""

try:
cached_messages = await self.store.get(key)
if cached_messages:
return cached_messages
except ContextStoreError:
# 生产系统必须在这里记录脱敏日志和 Redis 故障指标
# Redis 故障不应直接让正式会话记录不可用
pass

database_messages = await mysql_loader()

try:
await self.store.replace(key, database_messages)
except ContextStoreError:
# 回写失败不覆盖数据库查询的正确结果
pass

return database_messages

上面为了突出控制流省略了日志实现。生产代码不能使用空 pass 静默处理，应记录错误代码、Redis 节点和请求 ID，但不能记录完整用户消息。

9. 在 FastAPI 生命周期中创建 Redis 客户端

新建 main.py：

import os
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

from fastapi import FastAPI, Request
from redis.asyncio import Redis

from context_store import ContextMessage, RedisContextStore

def required_env(name: str) –> str:
value = os.getenv(name, "").strip()
if not value:
raise RuntimeError(f"缺少环境变量：{name}")
return value

@asynccontextmanager
async def lifespan(app: FastAPI) –> AsyncIterator[None]:
"""创建一个供整个 FastAPI 进程复用的 Redis 连接池。"""

redis = Redis.from_url(
required_env("REDIS_URL"),
# SSE 和 JSON 内容按 UTF-8 字符串处理
decode_responses=True,
socket_connect_timeout=3,
socket_timeout=3,
health_check_interval=30,
)

# 启动时执行只读探测，配置错误则直接启动失败
await redis.ping()

app.state.context_store = RedisContextStore(
redis,
ttl_seconds=3600,
max_messages=20,
)

yield

await redis.aclose()

app = FastAPI(
title="Redis 多轮对话上下文",
lifespan=lifespan,
)

@app.post("/demo/context/{conversation_id}")
async def append_demo_message(
conversation_id: str,
request: Request,
) –> dict[str, int]:
"""仅演示 Store 调用；真实用户身份必须来自认证系统。"""

store: RedisContextStore = request.app.state.context_store

# 这里的用户和租户仅为本地演示固定值
key = store.build_key(
tenant_id="demo",
owner_id=10001,
conversation_id=conversation_id,
)
await store.append(
key,
ContextMessage.create(
role="user",
content="请解释 Redis List。",
),
)
messages = await store.get(key)
return {"message_count": len(messages)}

Redis URL 示例：

$env:REDIS_URL = "redis://:URL编码后的密码@127.0.0.1:6379/0"

不要把 Redis 暴露到公网，也不要使用无密码的生产实例。

10. 使用 fakeredis 运行自动化测试

新建 tests/test_context_store.py：

from uuid import uuid4

import fakeredis.aioredis
import pytest

from context_store import ContextMessage, RedisContextStore

@pytest.mark.asyncio
async def test_append_trim_and_ttl() –> None:
"""验证追加、最大消息数量和 TTL。"""

redis = fakeredis.aioredis.FakeRedis(decode_responses=True)
store = RedisContextStore(
redis,
ttl_seconds=60,
max_messages=4,
)
key = store.build_key("tenant_a", 10001, str(uuid4()))

for index in range(6):
await store.append(
key,
ContextMessage.create(
role="user" if index % 2 == 0 else "assistant",
content=f"消息 {index}",
),
)

messages = await store.get(key)

# 只保留最后 4 条
assert [item.content for item in messages] == [
"消息 2",
"消息 3",
"消息 4",
"消息 5",
]

ttl = await redis.ttl(key)
assert 0 < ttl <= 60
await redis.aclose()

@pytest.mark.asyncio
async def test_replace_and_delete() –> None:
"""验证缓存回源重建和删除。"""

redis = fakeredis.aioredis.FakeRedis(decode_responses=True)
store = RedisContextStore(redis, max_messages=10)
key = store.build_key("tenant_a", 10001, str(uuid4()))

source_messages = [
ContextMessage.create("user", "问题"),
ContextMessage.create("assistant", "回答"),
]
await store.replace(key, source_messages)
assert await store.get(key) == source_messages

await store.delete(key)
assert await store.get(key) == []
await redis.aclose()

运行：

.\\.venv\\Scripts\\python.exe –m pytest –q

fakeredis 可以验证基本命令和业务逻辑，但不能替代真实 Redis 的网络、持久化、主从切换、Cluster 和故障恢复测试。

11. Redis 故障时应该怎么办

如果 MySQL 是事实来源，推荐：

Redis 读取失败
↓
记录监控和脱敏日志
↓
从 MySQL 加载有限历史
↓
继续模型调用
↓
异步或尽力回写 Redis

如果 MySQL 也不可用，则不能假装有历史。应明确降级为“无法加载上下文”，或者拒绝需要多轮语义的请求。

12. TTL 应该怎么设置

TTL 没有通用最佳值，需要考虑：

• 用户多久会继续同一会话；
• Redis 内存预算；
• 历史从 MySQL 回源的成本；
• 对话数据敏感程度；
• 合规保留要求。

本文选择“写入时刷新 TTL，读取不刷新”。这样只读访问不会让敏感缓存永久存活。

13. 对抗性审查

13.1 Redis 不是权限系统

Key 中包含用户 ID 不代表已经鉴权。用户身份必须来自验证后的 Token 或 Session，并且 MySQL 查询仍要检查所有权。

13.2 List 截断可能破坏问答对

按消息条数保留最后 N 条，可能只剩 Assistant 回答而丢掉对应 User 问题。生产代码更适合按 Turn（轮次）保存，或每次按两条成对截断。

13.3 消息数量不等于 Token 预算

20 条短消息和 20 条长合同占用完全不同。下一篇将按模型 Token 预算构建上下文。

13.4 多请求可能交错

Redis 事务只能保证一次追加操作原子，不能保证“用户问题—调用模型—助手回答”整个长流程原子。同一会话并发控制仍应使用第 8 篇的 pending 状态或可靠分布式任务机制。

13.5 缓存内容仍然是敏感数据

Redis 需要网络隔离、认证、TLS（跨不可信网络时）、最小权限、备份策略和数据过期策略。不能因为数据会过期就忽略安全。

13.6 缓存击穿

一个热门会话过期时，大量请求可能同时回源 MySQL。可以使用短期互斥、单飞请求或后台刷新，但锁必须有超时、唯一令牌和安全释放逻辑。

13.7 Redis 持久化不等于业务持久化

RDB 或 AOF 能提高 Redis 恢复能力，但不能自动替代关系数据库中的业务约束、审计和所有权模型。

14. 本篇总结

本文完成了 Redis 上下文缓存基础：

• MySQL 保存正式记录，Redis 保存近期缓存；
• Redis List 负责顺序追加和有限截断；
• MULTI/EXEC 原子执行追加、截断和过期；
• Key 包含经过验证的租户、用户和会话标识；
• 缓存未命中或故障时回源 MySQL；
• TTL 在写入时刷新；
• 使用 fakeredis 验证核心行为；
• 明确消息条数、并发和 Token 预算仍需进一步处理。

下一篇将实现基于 Token 预算的滑动窗口、摘要压缩和重要信息保留。

15. 练习题