#Python 操作 Redis 与缓存实战

阅读提示 本笔记面向已有 Python 基础、正在做 AI/后端应用的学习者，不假设你用过 Redis。主线是：从裸连接到生产级缓存封装，以及三类缓存事故到底是怎么发生的。

素材来自 redis_example/ 的两个文件（redis_demo.py / redis_client.py）。原代码里有多处实质性错误或反模式，笔记中已逐一纠正并给出正确写法 —— 不要直接复制原文件。

符号约定：⚠️ = 常见陷阱　🆚 = 对比辨析　💡 = 选择建议

#目录

• 一、为什么需要 Redis：从一次 LLM 调用讲起
• 二、连接 Redis：客户端与连接池
• 三、序列化：把 Python 对象存进 String
• 四、过期时间：缓存之所以是缓存
• 五、Cache-Aside 模式：get_or_set 到底做了什么
• 六、三类缓存事故：穿透、击穿、雪崩
• 七、实战：给 LLM 调用加缓存
• 八、延伸：Redis 作为 Agent 的短期记忆
• 速查表

核心概念 Redis 是一个内存键值数据库：key 是二进制安全的字符串，value 则有 String、Hash、List、Set、Sorted Set、Stream 等多种原生类型。本笔记的缓存场景主要用 String（第三~七节）和 List（第八节）。用它做缓存时，你实际只在做四件事：连得对（连接池 + 不硬编码密码）、存得进（序列化成 JSON 字符串）、会过期（TTL）、读写顺序对（Cache-Aside）。

而所谓的「穿透 / 击穿 / 雪崩」，全部是第四步在特定流量形态下的失效方式 —— 不是三种玄学，是同一个流程图上的三个漏点。

#一、为什么需要 Redis：从一次 LLM 调用讲起

一句话：Redis 用内存换时间，把「慢且贵」的操作结果暂存起来，下次直接取。

看一个真实场景 —— 调用大模型 API 回答问题：

start = time.time()
res = get_llm_response('深度学习是什么')
end = time.time()
print(f"耗时: {end - start:.2f}秒")   # 首次：3~10 秒，并且计费

同一个问题，一天之内可能被上百个用户问到。每次都调 API，意味着：

数据库查询同理。只要「获取数据的代价」远大于「读一次内存的代价」，就值得加缓存。

Redis 为什么快 数据全部放在内存里，不走磁盘 I/O；命令由单线程事件循环串行执行，单条命令内部不需要加锁。注意「Redis 是单线程的」只对命令执行这一层成立 —— 现代 Redis 仍有后台线程负责 AOF 落盘、大对象异步释放（lazy free）和网络 I/O。

代价是内存容量有限且成本高。至于它能不能当权威数据源，取决于你对持久化（RDB/AOF）、复制和一致性的要求 —— Redis 本身是支持这些的，但本笔记讨论的是它最常见的角色：缓存，数据丢了可以从数据源重建。

#二、连接 Redis：客户端与连接池

#2.1 最简连接

import redis
 
r = redis.Redis(host='localhost', port=6379, db=0,
                password='...', decode_responses=True)
print(r.ping())   # True → 连接成功

几个参数的实际含义：

逻辑库的三条边界

• 16 不是固定规则，它来自配置项 databases，可以改。
• Redis Cluster 只支持 DB 0，一旦上了集群就没有多库可用 —— 所以别把「用 db 分环境」写进架构假设。
• 逻辑库只是命名空间，不是隔离。同一实例上所有库共享内存、CPU 和阻塞，FLUSHALL 一把清空全部。官方也不建议用它承载多个不相关的应用，那种情况应该用独立实例或 key 前缀。

⚠️ decode_responses 忘了设的典型症状 错误操作： r = redis.Redis(host='localhost')，然后 r.get('name')

实际结果： 返回 b'\xe5\xbc\xa0\xe6\x98\x8a' 而不是 '张昊'，拿去做 json.loads() 或字符串拼接时行为诡异。

原因： Redis 协议层传输的是字节流，客户端默认不替你猜编码。

正确做法： 建连接时统一加 decode_responses=True，全项目保持一致。

#2.2 连接池：为什么不能每次请求新建客户端

一句话：建连接（TCP 三次握手 + AUTH）不是免费的，连接池的价值在于让同一批连接被反复复用。

先澄清一个容易记岔的点：redis.Redis(...) 本身默认就自带一个连接池，而且连接是执行命令时才按需建立的 —— 构造客户端对象这一步并不会立刻握手。真正的问题在于：如果你在每个请求里都 redis.Redis(...) 一次，就等于造了一堆互不共享的池子，连接从头到尾没被复用，服务端连接数还会被迅速堆高。

所以正确姿势是显式建一个池、全进程共用：

import redis
 
pool = redis.ConnectionPool(
    host=host, port=port, db=db, password=password,
    max_connections=20,          # 池子上限；无空闲连接且已达上限时抛 MaxConnectionsError
    decode_responses=True,
    socket_timeout=5,            # 读写超时（秒）：防止业务被卡死的 Redis 挂住
    socket_connect_timeout=5,    # 建连超时（秒）：Redis 不可达时快速失败
    health_check_interval=30     # 空闲超过 30 秒的连接，取用前先 PING 一次
)
client = redis.Redis(connection_pool=pool)

⚠️ max_connections 达到上限时是报错**，不是排队** 错误认知： 「池子满了，后面的请求会排队等一个连接空出来。」

实际结果： 普通 ConnectionPool 在没有空闲连接、且已创建到 max_connections 时，直接抛 redis.exceptions.MaxConnectionsError（ConnectionError 的子类）。高并发下这表现为一片突发的 5xx，而不是「变慢」。

原因： 连接池是按需创建 + 归还复用，并不是预先建好一批再分发；它也没有内置的等待队列。

正确做法： 需要「等」的语义就换 redis.BlockingConnectionPool(max_connections=20, timeout=5)，它会阻塞至多 timeout 秒再抛错。但更根本的是按并发量把 max_connections 估够 —— 靠阻塞兜底不如把池开对。

三个超时/探活参数是生产环境和玩具代码的分界线：

• 没有 socket_timeout：Redis 卡住时，你的 Web 请求会一直挂着，线程池很快被耗尽 —— 缓存故障演变成全站故障。
• 没有 socket_connect_timeout：Redis 宕机或遇到网络黑洞时，光是建连阶段就能挂住几十秒（取决于系统 TCP 重传策略），一点也「快速失败」不起来。
• 没有 health_check_interval：中间的防火墙或云负载均衡会静默掐断空闲连接，池子里留着一堆「看起来还在、实际已断」的连接，下次使用直接报错。设了它之后，空闲超过该间隔的连接在被取用时会先 PING 一次，失败则重连重试 —— 注意是取用时检查，不是有个后台线程在定时清扫。

💡 连接池应当是模块级单例 ConnectionPool 要在模块导入时创建一次，不要写在函数里。写在函数里等于每次调用建一个新池子，连接池就完全失去了意义，还会迅速耗尽服务端的最大连接数。

#2.3 配置绝不能硬编码

🔴 源码中的严重问题：硬编码 IP 与密码 错误操作：（redis_demo.py:12、redis_client.py:84）

r = redis.Redis(host='192.168.142.128', port=6379, password='123456')
redis_cli = RedisClient(host='192.168.142.128', password='123456')

实际结果： ① 密码随代码进 Git 历史，仓库一旦公开或转手就等于泄露，且改密码要改代码重新发版；② 内网 IP 写死后，换机器、上测试环境、同学各自本机跑，全都要改源码；③ 多人共用同一实例还会互相覆盖 key。

原因： 把「环境相关配置」和「业务代码」耦合在了一起。二者的变更频率和可见范围完全不同。

正确做法： 一律走环境变量，配 .env 并把它加进 .gitignore：

import os
from dotenv import load_dotenv
 
load_dotenv()
 
REDIS_HOST = os.getenv("REDIS_HOST", "localhost")
REDIS_PORT = int(os.getenv("REDIS_PORT", "6379"))
REDIS_PASSWORD = os.getenv("REDIS_PASSWORD")     # 无默认值，缺失即为 None
REDIS_DB = int(os.getenv("REDIS_DB", "0"))

已经提交过的密码要视为已泄露，必须在服务端轮换，而不是删掉代码了事 —— Git 历史里还留着。

#三、序列化：把 Python 对象存进 String

一句话：String 类型的 value 是一段字节序列，所以 SET 一个 dict、list 之前必须先序列化。

🆚 「Redis 只认字符串」是个流传很广的误解 Redis 的 key 确实是二进制安全的字符串，但 value 有多种原生类型：String、Hash、List、Set、Sorted Set、Stream、Bitmap、HyperLogLog……本节讲的限制只属于 String 类型的 SET/GET。

所以「把一个 dict 存进 Redis」至少有两条路：序列化成 JSON 塞进 String（整存整取、最简单，本节的做法），或者用 Hash（HSET/HGET，能只改一个字段而不必读回整个对象）。第八节的会话历史用的则是 List。选哪个取决于你的访问模式，不是「Redis 只能这样」。

import json
 
# ❌ 直接存字典
r.set("user:1001", {"name": "Alice"})
# DataError: Invalid input of type dict
 
# ✅ 先转 JSON 字符串
r.set("user:1001", json.dumps({"name": "Alice"}, ensure_ascii=False))
data = json.loads(r.get("user:1001"))   # 读出来再反序列化

封装成一对方法就不用每次手写：

def set_json(self, key: str, value: Any, ex: Optional[int] = None) -> bool:
    """存储 JSON 对象"""
    try:
        self.client.set(key, json.dumps(value, ensure_ascii=False), ex=ex)
        return True
    except RedisError as e:
        logger.error("set_json 失败 key=%s: %s", key, e)
        return False
 
def get_json(self, key: str) -> Optional[Any]:
    """获取 JSON 对象，不存在或异常返回 None"""
    try:
        data = self.client.get(key)
        return json.loads(data) if data is not None else None
    except (RedisError, json.JSONDecodeError) as e:
        logger.error("get_json 失败 key=%s: %s", key, e)
        return None

三个细节值得注意：

• ensure_ascii=False：不加的话「张昊」会被转成 "\u5f20\u660a" 这种转义序列，虽然读回来能正常还原，但你在 redis-cli 里排查问题时完全看不懂，而且每个汉字从 3 字节膨胀到 6 字节。
• Optional[int] 而非 int = None：原代码写的是 ex: int = None，类型注解和默认值矛盾，mypy 会直接报错。
• if data is not None 而非 if data：原代码用的是后者。如果某个 key 存的值恰好是空字符串 ""，if data 判定为假会返回 None，让调用方误以为「key 不存在」。

序列化不是 Redis 特有的概念 只要数据要跨越进程边界，就必须序列化 —— 网络传输、写磁盘、进消息队列，全都一样。JSON 只是最通用的一种格式；追求性能可以换 pickle（Python 专用、有安全风险）或 msgpack（跨语言、更紧凑）。

#四、过期时间：缓存之所以是缓存

一句话：一个持续写入新 key、却既不设 TTL 也不设容量上限的 Redis，内存只会一直涨。

（严格说，maxmemory + maxmemory-policy 淘汰策略同样能给内存兜底，长期存在的持久 key 也是合法用法。但对缓存场景，TTL 是最直接、最该先用上的那道闸。）

r.set('name', '张昊', ex=5)      # 5 秒后自动删除
print(r.get('name'))            # '张昊'
# ...等待 6 秒...
print(r.get('name'))            # None
 
print(r.ttl('name'))            # 剩余秒数；-1=永不过期，-2=key 不存在

TTL 同时解决了两个问题：

1. 内存回收 —— 内存是有限且昂贵的，冷数据必须自动淘汰。
2. 数据新鲜度 —— 缓存里是数据源的旧副本。源数据改了而缓存没改，就出现了不一致。在没有并发竞态、缓存也能正常刷新的前提下，TTL 给这种不一致划了一个近似上界，是一种最省事的最终一致性。 ⚠️ 只是「近似」：并发回填时，一个慢请求可能把更旧的值写在更新的值之后，这条陈旧数据会重新拿到一个完整 TTL；回源持续失败时陈旧数据也可能被反复续上。

💡 TTL 怎么定：

⚠️ 「先更新数据库，再删缓存」而不是「更新缓存」 错误操作： 数据变更时同时写数据库和写缓存（双写）。

实际结果： 高并发下两个请求的写入顺序可能交错，缓存里留下的是先算出但后写入的旧值，且这个错误值会一直存在到 TTL 到期。

原因： 数据库和 Redis 是两个独立系统，无法保证跨系统的写入顺序。

常用做法： 更新数据源后删除缓存（Cache-Aside 的写路径），让下一次读请求自然回填。删除是幂等的，重复执行无副作用。

但要清楚它是默认首选，不是唯一解，也没有消除竞态 —— 典型残留场景：一个读请求在你删缓存之前读到了旧值，却在删除之后才把这个旧值写回去。要求更强时还有：延迟双删、监听 binlog/CDC 触发失效、给缓存值带版本号、或者 Write-Through 由缓存层统一承接写入。

#五、Cache-Aside 模式：get_or_set 到底做了什么

一句话：读的时候先查缓存，没有再查数据源并回填 —— 这就是 Cache-Aside（旁路缓存），最常用的缓存模式。

读图顺序：从左上角的请求出发，走到中间的菱形判断处分叉 —— 命中就向右直接返回，未命中则向下查数据源、回写缓存、再返回。虚线表示回写之后，下一次同样的请求就会走上面那条快路径。

代码形态：

def get_or_set(self, key: str, func: Callable, ex: int = 60) -> Optional[Any]:
    """
    :param key: 缓存键
    :param func: 缓存未命中时用来获取数据的回调
    :param ex: 过期时间（秒）
    """
    cached = self.get_json(key)      # ① 查缓存
    if cached is not None:
        return cached                # ② 命中，直接返回
 
    data = func()                    # ③ 未命中，查数据源
    if data is not None:
        self.set_json(key, data, ex) # ④ 回写缓存
    return data

使用时把「怎么取数据」作为函数传进来，调用方完全感知不到缓存的存在：

def fetch_from_db():
    print(">>> 缓存未命中，正在从数据库查询...")
    return {"id": 1002, "name": "Bob"}
 
redis_cli.get_or_set("user:1002", fetch_from_db, ex=60)   # 打印查询日志
redis_cli.get_or_set("user:1002", fetch_from_db, ex=60)   # 静默返回缓存

🔴 源码中的严重问题：这段代码防不住缓存穿透 错误操作： redis_client.py 里 get_or_set 的 docstring 写的是「缓存穿透保护」。

实际结果： 它恰恰不能防穿透。因为 if data is not None 这个判断意味着：当数据源返回空（查不到这条记录）时，什么都不会写进缓存。于是同一个不存在的 key，第 1 次、第 100 万次请求都会穿过 Redis 打到数据库。

原因： 混淆了两件事 —— 这段代码实现的是 Cache-Aside 模式（一种读取模式），而缓存穿透防护（缓存空值 / 布隆过滤器）是需要额外加的一层保护。前者不蕴含后者。

正确做法： 用一个哨兵值把「查过了，确实没有」这个事实也缓存起来，并给它一个明显更短的 TTL：

_EMPTY = "__EMPTY__"          # 哨兵值：代表「数据源里确实没有」
_EMPTY_TTL = 60               # 空值 TTL 要远短于正常 TTL
 
def get_or_set(self, key: str, func: Callable, ex: int = 300) -> Optional[Any]:
    cached = self.get_json(key)
    if cached == _EMPTY:      # 命中空值缓存，直接返回，不打数据源
        return None
    if cached is not None:
        return cached
 
    data = func()
    if data is not None:
        self.set_json(key, data, ex)
    else:
        self.set_json(key, _EMPTY, _EMPTY_TTL)   # ← 关键：空结果也要缓存
    return data

空值 TTL 必须短，否则数据真的被创建出来之后，用户还要等一个完整 TTL 才能看到。

顺带一提，同一个类里的 delete() 也有个小毛病：

def delete(self, key: str) -> bool:
    try:
        self.client.delete(key)
        return True          # ← 不管 key 存不存在都返回 True
    except RedisError:
        return False
 
# 更诚实的写法：DEL 命令本身会返回实际删除的键数量
def delete(self, key: str) -> bool:
    try:
        return self.client.delete(key) > 0
    except RedisError as e:
        logger.error("delete 失败 key=%s: %s", key, e)
        return False

#六、三类缓存事故：穿透、击穿、雪崩

这三个词长得像，中文名还容易混，但它们描述的是完全不同的三个失效点。

读图顺序：三栏结构完全一样（请求 → Redis → 数据库 → 解法），差别只在中间 Redis 层的状态。左栏是「key 本来就不存在」，中栏是「一个热点 key 刚好过期」，右栏是「一大批 key 同时失效」。红色箭头的粗细代表打到数据库的流量大小。

⚠️ 源码注释里的定义写反了 错误操作： redis_client.py 结尾的注释写道：

「当一个 key 在 Redis 中不存在，而数据库中存在时，就会发生缓存穿透。」

实际结果： 按这个定义，每一次正常的缓存未命中都成了「穿透」—— 那缓存就没法用了。同一段注释后面又写「使用空值来缓存不存在的 key，避免缓存击穿」，也把解法安错了对象（缓存空值防的是穿透）。

原因： 把「普通未命中」当成了穿透。未命中是缓存的正常工作状态，回填一次就好了；穿透的要害在于数据源里也没有，所以永远回填不上，这个 key 会无限次打到数据库。

正确做法： 按下表记忆，三者的判定依据分别是「数据源里有没有」「失效的是一个还是一批」。

三种解法的关键实现要点：

① 穿透 —— 缓存空值 + 布隆过滤器

缓存空值的代码见第五节。布隆过滤器则是在查缓存之前加一道拦截：把所有合法 ID 预先放进一个位数组，查询时先问它「这个 ID 可能存在吗」。它的特性是**「说不存在就一定不存在，说存在则可能误判」**，正好适合做前置过滤 —— 误判只会导致多查一次库，漏判则不会发生。

② 击穿 —— 把回源收敛成一次

热点 key 过期的瞬间，成千上万个请求同时发现「缓存没有」，于是同时去查数据库。思路都是「只让一次回源真正发生」，常见有三条路：

分布式锁版本：

import uuid
 
# 释放锁必须是「比对 + 删除」的原子操作，中间不能被别的命令插进来
_UNLOCK_LUA = """
if redis.call('GET', KEYS[1]) == ARGV[1] then
    return redis.call('DEL', KEYS[1])
else
    return 0
end
"""
 
def get_with_lock(self, key: str, func: Callable, ex: int = 300):
    cached = self.get_json(key)
    if cached is not None:
        return cached
 
    lock_key = f"lock:{key}"
    token = uuid.uuid4().hex               # ← 每个持有者一个唯一 token
    # SET NX EX：只有 key 不存在时才设置成功，且锁自带过期时间防死锁
    if self.client.set(lock_key, token, nx=True, ex=10):
        try:
            data = func()                  # 只有抢到锁的这个请求去回源
            if data is not None:
                self.set_json(key, data, ex)
            return data
        finally:
            # 只有值仍等于自己的 token 才删；用 Lua 保证「比对」和「删除」之间不被打断
            self.client.eval(_UNLOCK_LUA, 1, lock_key, token)
    else:
        # TODO(human): 没抢到锁时的等待策略
        return None

🔴 固定值加锁 + 无条件 DEL = 会误删别人的锁 错误操作： self.client.set(lock_key, "1", nx=True, ex=10)，然后在 finally 里直接 self.client.delete(lock_key)。

实际结果： 如果 func() 跑了超过 10 秒，锁会先自己过期，请求 B 随即合法地抢到同一把锁开始回源；此时请求 A 执行到 finally，把 B 的锁删掉了。于是 C 又能抢到锁 —— 互斥当场失效，多个请求同时回源，击穿防护形同虚设。

原因： 锁的值是常量 "1"，谁也分辨不出这把锁是不是自己那把；而 DEL 是无条件执行的。

正确做法： ① 加锁时写入唯一 token（uuid4().hex）；② 释放时用 Lua 做「值等于我的 token 才删」的原子比较删除。

注意 token 只能防误删，防不住「A 的锁已过期、A 却还在跑」这种双重回源。业务时长不可控时还需要看门狗续期、fencing token，或者直接用成熟实现 —— redis-py 自带的 client.lock(...) 就已经做好了 token + Lua 释放。

另外 nx=True, ex=10 必须一起设：只有 nx 没有 ex，一旦持锁进程崩溃，这把锁永不释放，后续请求全部卡死。

⚠️ 没抢到锁的分支，不能只睡一次就返回 错误操作： time.sleep(0.05) 之后 return self.get_json(key)。

实际结果： 如果回源要 3 秒（LLM 调用非常常见），50 毫秒后缓存里还什么都没有，这个请求直接返回 None —— 用户拿到的是空结果，而不是数据。原本只是「慢一点」的场景，被写成了「错一个」。

原因： 把「锁被别人持有」误当成「稍等就一定有」，但等待时长和回源耗时毫无关系。

正确做法： 这个分支的等待策略要按「回源大概多久」和「调用方的超时预算」来定，见上方代码里留空的那个分支。

③ 雪崩 —— TTL 加随机抖动

如果启动时批量预热了 10000 个 key、TTL 全设成 3600 秒，那一小时后它们会在同一秒集体失效。加个随机量就能把失效时间打散：

import random
 
def set_with_jitter(self, key: str, value, base_ex: int = 3600):
    """在基础 TTL 上叠加 ±10% 的随机抖动，避免集中失效"""
    jitter = random.randint(-base_ex // 10, base_ex // 10)
    return self.set_json(key, value, ex=base_ex + jitter)

另一半的雪崩是 Redis 本身挂了，那属于架构问题：主从复制 + 哨兵/集群保证可用性，RDB/AOF 持久化能减少重启后需要重建的数据量（究竟能恢复多少取决于持久化配置和最后一次落盘时间，不等于「一定不用预热」），同时业务侧要有降级能力 —— Redis 不可用时直接查数据库并限流，而不是整个接口报 500。

#七、实战：给 LLM 调用加缓存

把前面的东西合起来，就是一个能用的 LLM 响应缓存。先看原代码的问题：

⚠️ 不要把原始 prompt 直接拼进 key 错误操作：（redis_demo.py:30）

cache_key = f"llm_cache:{prompt}"

实际结果： ① prompt 动辄几千字符，Redis 的 key 会变得极其臃肿（key 本身也占内存，且 KEYS/SCAN 排查时完全没法看）；② prompt 里的换行、冒号、空格会破坏 key 的命名层次；③ 如果 prompt 含用户隐私，这些内容就明文躺在 Redis 里，任何能连上实例的人 SCAN 一遍就全看到了。

原因： key 的职责是唯一标识，不是存储内容。

正确做法： 用哈希摘要做 key，长度固定、字符安全、key 里也不再有 prompt 明文：

import hashlib
 
def _cache_key(prompt: str, model: str) -> str:
    digest = hashlib.sha256(prompt.encode("utf-8")).hexdigest()[:16]
    return f"llm_cache:{model}:{digest}"

注意把 model 名字也放进 key：同一个 prompt 用不同模型回答结果不同，只按 prompt 做 key 会导致换模型后读到旧模型的回答。

两点补充，别把这个函数当终点：

• 摘要不是加密。 它只是让 key 里不再有明文。prompt 空间小的时候（比如枚举城市名、商品名），攻击者照样能把候选值挨个哈希去撞 key。真正敏感的内容要靠网络隔离、访问控制和 Redis 自身的鉴权/加密来保护，而不是靠哈希。
• 参与哈希的输入要够全。 只有 prompt + model 并不足以唯一标识一次请求 —— 凡是会改变输出的输入都该进 key：system prompt、temperature / top_p、工具定义、RAG 检索用的知识库版本、模型的具体版本号，以及租户或权限范围（否则会跨用户串答案）。截断到 16 个十六进制字符只有 64 bit，做缓存够用，但别宣称它「保证不碰撞」。

完整的修正版：

import os
import json
import hashlib
import redis
from redis.exceptions import RedisError
from dotenv import load_dotenv
from utils.model_utils import get_model
 
load_dotenv()
 
# 连接池：模块级单例，全进程复用
_pool = redis.ConnectionPool(
    host=os.getenv("REDIS_HOST", "localhost"),
    port=int(os.getenv("REDIS_PORT", "6379")),
    password=os.getenv("REDIS_PASSWORD"),
    db=int(os.getenv("REDIS_DB", "0")),
    decode_responses=True,
    max_connections=20,
    socket_timeout=5,
    socket_connect_timeout=5,
    health_check_interval=30,
)
r = redis.Redis(connection_pool=_pool)
 
CACHE_TTL = 86400   # 24 小时
 
 
def _cache_key(prompt: str, model: str, **params) -> str:
    """用摘要做 key：长度固定、不含特殊字符、key 里不留 prompt 明文
 
    凡是会影响输出的输入都要参与哈希（system prompt、temperature、工具定义、
    知识库版本、租户……），否则会读到「本不该属于这次请求」的缓存。
    """
    payload = json.dumps(
        {"prompt": prompt, "model": model, **params},
        sort_keys=True,          # 排序：保证 dict 的书写顺序不影响 key
        ensure_ascii=False,
    )
    digest = hashlib.sha256(payload.encode("utf-8")).hexdigest()[:16]
    return f"llm_cache:{model}:{digest}"
 
 
def get_llm_response(prompt: str, model: str = "qwen") -> str:
    """获取 LLM 回答：有缓存就返回缓存，没有就调 API 并回写"""
    key = _cache_key(prompt, model)
 
    # ① 查缓存。Redis 故障不应该让业务挂掉，降级为直接调 API
    try:
        cached = r.get(key)
        if cached is not None:
            print(f"✅ 命中缓存：{prompt[:20]}...")
            return cached
    except RedisError as e:
        print(f"⚠️ Redis 读取失败，降级直连 API：{e}")
 
    # ② 未命中，调用 LLM
    print(f"🔄 调用 API：{prompt[:20]}...")
    client = get_model(model)
    content = client.invoke(prompt).content
 
    # ③ 回写缓存。写失败同样不影响本次返回
    try:
        r.set(key, content, ex=CACHE_TTL)
    except RedisError as e:
        print(f"⚠️ Redis 写入失败，本次不缓存：{e}")
 
    return content

相比原版的四处改动：

💡 什么样的 LLM 调用适合缓存 适合：固定知识问答、文档摘要、翻译、分类打标 —— 相同输入期望相同输出。 不适合：多轮对话（完整上下文参与 key 时命中率通常很低 —— 但用户重试、会话分支复用、共享前缀等场景仍可能命中，值得先量一量再下结论）、需要创意多样性的生成（缓存会让每次结果一模一样）、带时效性的查询。

#八、延伸：Redis 作为 Agent 的短期记忆

除了当缓存，Redis 在 AI 应用里还有一个常见角色：存放 Agent 的短期记忆。

分工大致是这样的：

用 Redis 的 List 结构存会话历史很自然：

def append_message(session_id: str, role: str, content: str, ttl: int = 1800) -> bool:
    """追加一条对话消息，并刷新整个会话的存活时间"""
    key = f"chat:{session_id}"
    msg = json.dumps({"role": role, "content": content}, ensure_ascii=False)
    try:
        # pipeline(transaction=True)：三条命令打包成一次 MULTI/EXEC 发出，
        # 避免留下「追加了但没截断」或「有历史但没 TTL」的半完成状态
        with r.pipeline(transaction=True) as pipe:
            pipe.rpush(key, msg)
            pipe.ltrim(key, -20, -1)   # 只保留最近 20 条，防止无限膨胀撑爆内存
            pipe.expire(key, ttl)      # 每次交互都续期：30 分钟无操作则整个会话自动清理
            pipe.execute()
        return True
    except RedisError as e:
        logger.error("append_message 失败 session=%s: %s", session_id, e)
        return False
 
 
def load_history(session_id: str) -> list:
    """读取会话历史供下一轮拼进 prompt；Redis 故障时降级为空历史"""
    try:
        return [json.loads(m) for m in r.lrange(f"chat:{session_id}", 0, -1)]
    except (RedisError, json.JSONDecodeError) as e:
        logger.error("load_history 失败 session=%s: %s", session_id, e)
        return []

三个设计要点：

• ltrim 是必须的。不做截断的话，长会话的历史会一直增长，既撑内存又超出模型的上下文窗口。
• expire 要每次刷新，实现「滑动过期」—— 用户还在聊就一直续命，停下 30 分钟才回收。
• 三条命令要打包。分三次单发的话，中途任意一步失败都会留下坏状态：只 RPUSH 成功 → 历史无限增长；只有前两条成功 → 这个会话永不过期，成了内存里的僵尸数据。pipeline 顺带还把三次往返压成一次，是白得的性能。

概念背景可参考 Agent概念，整体学习路径见 学习路线。

#速查表

常用命令 / 方法

三类事故决策表

key 命名约定

用冒号分层，形如 业务:实体:标识，例如 user:1001、llm_cache:qwen:a3f9c2...、chat:sess_88、lock:user:1001。好处是在 redis-cli 里可以用 SCAN 0 MATCH "llm_cache:*" 按前缀排查。生产环境永远不要用 KEYS * —— 它会阻塞整个 Redis 实例。

复习重点

1. value 有多种原生类型（String / Hash / List / Set / ZSet / Stream），需要序列化的只是 String 的 SET/GET：存对象要 json.dumps，中文记得 ensure_ascii=False，客户端记得 decode_responses=True。
2. 连接池是模块级单例，三个超时/探活参数是生产必备 —— 否则 Redis 卡住会拖垮整个应用。记牢：普通池达到 max_connections 时抛 MaxConnectionsError，不是排队，要「等」得用 BlockingConnectionPool。
3. Cache-Aside 四步：查缓存 → 命中返回 → 未命中查源 → 回写（带 TTL）。写路径是「更新数据库后删除缓存」，不是更新缓存。
4. 穿透 / 击穿 / 雪崩靠两个问题区分：数据源里有没有这条数据？失效的是一个 key 还是一批？对应解法是 缓存空值+布隆过滤器 / 收敛回源 / TTL 抖动+集群。
5. Cache-Aside 不等于穿透防护。源码里 get_or_set 因为「空结果不写缓存」，恰恰防不住穿透 —— 必须显式缓存哨兵空值。
6. 分布式锁的两条命：加锁写唯一 token、释放用 Lua 比对后再删。固定值 "1" + 无条件 DEL 会在业务超时后误删别人的锁，互斥直接失效。
7. 缓存是优化不是依赖：对外暴露的缓存读写都要捕获 RedisError 并降级到数据源（本笔记的 get_json / set_json / get_llm_response / append_message 都这么做；get_with_lock 为突出锁逻辑省略了，实际使用同样要包一层）。缓存故障绝不能升级成业务故障。
8. 配置走环境变量，key 用哈希摘要而非原始内容（且所有影响输出的参数都要参与哈希），会话类数据必须 ltrim + expire 双保险、并打包进 pipeline。