1. 为什么 AI Agent 绕不开 Redis 缓存这层
做 AI Agent 开发的人,十个里有八个会在某个阶段撞上同一堵墙:Agent 跑着跑着就慢了,Token 消耗像开了闸,多轮对话里模型反复处理同样的上下文,工具调用结果每次都重新算一遍。我最早搭 Agent 的时候也踩过这个坑,一个带工具调用的对话流程,单次响应要七八秒,账单一天下来能吓人一跳。后来把 Redis 缓存这层加进去,响应时间直接砍到两秒出头,Token 消耗降了将近四成。这不是什么玄学优化,就是把重复计算的东西存起来而已。
Redis 在 AI Agent 架构里的角色,说白了就是短期记忆 + 结果复用 + 状态协调三件事。Agent 和普通后端服务不一样的地方在于,它的"计算"特别贵——一次 LLM 调用可能几百毫秒到几秒,一次向量检索可能几十毫秒,一次工具调用可能涉及外部 API。这些结果如果每次都重算,成本和延迟都扛不住。Redis 作为内存级 KV 存储,读写延迟在亚毫秒级,天然适合放在 Agent 和昂贵计算之间当缓冲层。
这篇文章面向的是正在搭 Agent 或者准备给现有 Agent 加缓存层的开发者。不管你用的是 Python 还是 Rust 写 Agent,不管你跑的是单机还是分布式,Redis 这层的设计思路是相通的。我会从架构设计讲到具体落地,把缓存键怎么设计、失效策略怎么定、分布式锁怎么用、序列化怎么选这些实操细节全部拆开讲。看完你应该能直接照着搭一套能上生产的 Agent 缓存层。
需要先明确一点:Redis 缓存不是银弹。它解决的是"重复计算"和"状态共享"两个问题,解决不了模型本身慢的问题,也解决不了 Prompt 设计烂的问题。把它放对位置,收益很大;放错位置,就是给自己挖坑。下面我按实际项目里的落地顺序,一层层拆。
2. AI Agent 缓存层的整体架构设计
2.1 Agent 的哪些环节值得缓存
不是所有东西都值得往 Redis 里塞。我见过有人把整个对话历史全量缓存,结果内存爆了不说,命中率还低得可怜。判断一个环节值不值得缓存,看三个指标:计算成本高不高、结果可复用性强不强、数据时效性要求松不松。三个都满足,才值得缓存。
按这个标准过一遍 Agent 的典型流程,值得缓存的环节其实很清晰:
| 环节 | 计算成本 | 可复用性 | 时效要求 | 是否值得缓存 |
|---|---|---|---|---|
| LLM 完整响应 | 极高 | 中(取决于 Prompt 是否带随机性) | 松 | 值得,但要做归一化 |
| Embedding 向量 | 高 | 极高(同样文本永远同样向量) | 极松 | 强烈推荐 |
| 向量检索结果 | 中高 | 中(取决于知识库更新频率) | 中 | 值得,设短 TTL |
| 工具调用结果 | 差异大 | 差异大 | 差异大 | 按工具类型分别处理 |
| 对话历史 | 低 | 低 | 紧 | 缓存但设短 TTL |
| Prompt 模板渲染 | 低 | 高 | 松 | 可选,收益一般 |
Embedding 缓存是性价比最高的。同一段文本的向量是确定性的,缓存下来永远不会错,命中一次省一次模型调用。我实测过一个知识库问答 Agent,光 Embedding 缓存这一项,就把整体延迟降了 30% 以上,因为用户问的问题经常语义重复,归一化之后命中率能到 50% 往上。
LLM 响应缓存要谨慎。如果你的调用带了 temperature 大于 0 的随机性,或者 Prompt 里塞了时间戳、随机 ID 这类每次都变的东西,缓存命中率会趋近于零。做法是先对 Prompt 做归一化——去掉时间戳、把变量替换成占位符、统一空白字符——再算哈希当键。这样即使用户问法略有不同,只要归一化后一致就能命中。
2.2 缓存层放在 Agent 的哪个位置
架构位置上,Redis 缓存层应该夹在Agent 编排逻辑和昂贵计算之间。具体来说有两种放法,各有取舍。
第一种是旁路缓存(Cache-Aside),Agent 逻辑自己控制读写。查缓存 → 命中就返回 → 没命中就调模型 → 写回缓存。这种放法逻辑清晰,控制粒度细,出问题好排查。缺点是每个调用点都要写一遍缓存逻辑,代码有重复。我大部分项目用的都是这种,因为 Agent 的调用点其实不多,封装一个装饰器就解决了。
第二种是代理层缓存,在 Agent 和模型 API 之间加一个代理,代理自动做缓存。好处是 Agent 代码完全不用改,坏处是代理层很难理解业务语义,归一化做得粗糙,命中率上不去。除非你有大量现成的 Agent 不想改代码,否则我不太推荐这种。
import hashlib import json import redis r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=False) def cache_key(prefix: str, payload: dict) -> str: # 归一化:排序 key,去掉易变字段 normalized = json.dumps(payload, sort_keys=True, ensure_ascii=False) digest = hashlib.sha256(normalized.encode('utf-8')).hexdigest()[:32] return f"agent:{prefix}:{digest}" def cached_llm_call(prompt: str, model: str, ttl: int = 3600): key = cache_key("llm", {"prompt": prompt, "model": model}) hit = r.get(key) if hit is not None: return json.loads(hit) result = call_llm(prompt, model) # 真实调用 r.setex(key, ttl, json.dumps(result, ensure_ascii=False)) return result这段代码是旁路缓存的最小实现。注意sort_keys=True和ensure_ascii=False这两个参数,前者保证字典序列化顺序稳定,后者保证中文不被转义成\uXXXX,否则同样的内容会算出不同的哈希。这个坑我踩过,当时排查了半天才发现是序列化不一致导致缓存永远不命中。
2.3 单机 Redis 还是分布式 Redis
开发阶段单机 Redis 就够了,但上生产要考虑几个问题:Agent 实例可能多个,缓存要共享;Redis 挂了不能拖垮整个 Agent;缓存数据量可能超过单机内存。
多实例共享缓存是刚需。如果你的 Agent 部署了三个副本,每个副本各自缓存,命中率直接除以三,还浪费内存。这时候必须用同一个 Redis 实例或者集群。分布式缓存的意义就在这——所有 Agent 实例看到的是同一份缓存。
高可用方面,主从加哨兵是最常见的方案。主节点写,从节点读,哨兵负责故障转移。Docker 起主从其实很简单:
# 主节点 docker run -d --name redis-master -p 6379:6379 \ redis:7-alpine redis-server --appendonly yes # 从节点,指向主节点 docker run -d --name redis-replica -p 6380:6379 \ redis:7-alpine redis-server --replicaof redis-master 6379生产环境我更推荐用云托管的 Redis 服务,省去运维成本,故障转移也做得更成熟。自建的话,哨兵模式至少要三个哨兵节点才能可靠投票,这个部署复杂度不低。
注意:主从复制是异步的,主节点写入后从节点可能有毫秒级延迟。如果你的 Agent 对一致性要求极高(比如分布式锁场景),读操作也要走主节点,别图省事读从节点。
3. 缓存键设计与失效策略的实操细节
3.1 缓存键怎么设计才不踩坑
缓存键设计看着简单,实际是最容易出问题的地方。我总结了几条硬性规则,都是踩坑换来的。
键要有命名空间前缀。agent:llm:xxx、agent:embedding:xxx、agent:tool:xxx这样分门别类。好处是排查问题时可以用SCAN agent:llm:*快速定位,清理时也能按前缀批量删。千万别用裸哈希当键,到时候满屏十六进制字符串,根本不知道哪个是哪个。
键长度要控制。Redis 的键本身也占内存,一个 200 字节的键和 20 字节的键,存一千万条差距就是 1.8GB。做法是把长内容先哈希再当键,比如 Prompt 内容用 SHA256 取前 32 位十六进制,固定长度,碰撞概率也极低。
键里要包含影响结果的所有变量。模型名、温度参数、Prompt 版本号,这些变了结果就变了,必须进键。我见过有人缓存 LLM 响应时忘了带模型名,结果换了模型还在返回旧模型的缓存,排查了好久。
def build_llm_key(prompt: str, model: str, temperature: float, prompt_version: str) -> str: payload = { "p": prompt, "m": model, "t": round(temperature, 2), # 浮点数要 round,避免精度问题 "v": prompt_version, } return cache_key("llm", payload)round(temperature, 2)这步很关键。浮点数在不同语言、不同平台上的表示可能有微小差异,0.7和0.7000000001算出来的哈希完全不同。统一 round 到两位小数,能避免这种诡异的不命中。
3.2 TTL 设置:不同数据不同命
TTL 设太长,缓存脏数据;设太短,命中率上不去。我的经验是按数据特性分档:
- Embedding 向量:TTL 可以设很长,7 天甚至永久。文本不变向量就不变,除非你换了 Embedding 模型。换模型时手动清一次对应前缀就行。
- LLM 响应:1 到 24 小时。取决于你的 Prompt 是否包含时效性内容。如果是通用问答,24 小时没问题;如果涉及实时信息,1 小时甚至更短。
- 向量检索结果:10 分钟到 1 小时。知识库更新后旧结果就失效了,TTL 短一点更安全。
- 对话历史:30 分钟到 2 小时。用户会话结束后就没用了,设太长纯浪费内存。
- 工具调用结果:按工具定。查天气 10 分钟,查汇率 5 分钟,查静态配置可以几小时。
TTL 不要设成整齐的整数,比如全设 3600。这样会导致大量键在同一秒集中过期,Redis 的过期删除是惰性加定期结合的,集中过期会造成瞬时压力。加个随机抖动:
import random def jitter_ttl(base_ttl: int, ratio: float = 0.1) -> int: delta = int(base_ttl * ratio) return base_ttl + random.randint(-delta, delta)3.3 缓存失效的三种触发方式
TTL 到期是被动失效,但有些场景需要主动失效。我一般用三种方式组合。
主动删除:数据源更新时,直接DEL对应的键。比如知识库文档更新了,把相关的检索缓存键删掉。难点在于怎么知道哪些键受影响——如果键是哈希算出来的,你没法反推。解决办法是维护一个映射关系,或者用标签(tag)机制,把同一来源的键记在一个 Set 里,更新时遍历删除。
版本号失效:在键里带一个全局版本号,更新时版本号加一,旧键自然不再被访问,靠 TTL 慢慢清理。这种方式不用精确删除,实现简单,代价是有一段时间内存里同时存在新旧两版缓存。适合更新不频繁、内存不紧张的场景。
写时更新:更新数据源的同时更新缓存,保证缓存永远是最新的。这种方式一致性最好,但实现复杂,而且如果更新逻辑有 bug,缓存会一直错下去。我一般只在强一致要求的场景用。
实操心得:大部分 Agent 场景用 TTL 加主动删除就够了,别一上来就搞复杂的版本号机制。先跑起来,观察命中率和数据新鲜度,有问题再优化。
4. 分布式锁与并发控制:Agent 场景下的特殊考量
4.1 为什么 Agent 需要分布式锁
Agent 的并发场景比普通服务更复杂。多个用户同时问同一个问题,如果缓存没命中,会同时触发多次 LLM 调用,这就是典型的缓存击穿。更麻烦的是工具调用,比如两个 Agent 实例同时去调一个有限流的第三方 API,不加控制就会超限。
分布式锁解决的就是这类问题:同一时刻只允许一个实例去执行昂贵操作,其他实例等待结果。Redis 实现分布式锁的标准做法是SET key value NX PX ttl,NX 保证只有不存在时才设置成功,PX 设过期时间防止死锁。
import uuid import time def acquire_lock(conn, lock_key: str, ttl_ms: int = 10000): token = str(uuid.uuid4()) ok = conn.set(lock_key, token, nx=True, px=ttl_ms) return token if ok else None def release_lock(conn, lock_key: str, token: str): # Lua 脚本保证原子性:只有 token 匹配才删 lua = """ if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end """ conn.eval(lua, 1, lock_key, token)释放锁必须用 Lua 脚本,不能先GET再DEL。因为这两步之间锁可能已经过期被别的实例拿到了,你再删就把别人的锁删了。这个坑非常经典,面试也常考。
4.2 锁的粒度与等待策略
锁的粒度要尽量细。别用一个全局锁保护所有缓存操作,那样并发度直接归零。正确的做法是按缓存键加锁,不同的键互不影响。
等待策略有两种。一种是自旋等待,拿不到锁就循环重试,适合锁持有时间短的场景。另一种是直接返回旧值或降级,拿不到锁就用一个稍旧的结果,或者返回一个默认值。Agent 场景我更推荐后者,因为 LLM 调用可能好几秒,让用户干等体验很差。
def get_with_lock(conn, key: str, compute_fn, ttl: int, lock_ttl: int = 15000): val = conn.get(key) if val is not None: return json.loads(val) lock_key = f"{key}:lock" token = acquire_lock(conn, lock_key, lock_ttl) if token is None: # 没拿到锁,短暂等待后重试读缓存 time.sleep(0.1) val = conn.get(key) if val is not None: return json.loads(val) # 还没读到,降级:直接计算但不写缓存 return compute_fn() try: # 双重检查,防止拿到锁之前别人已经写好了 val = conn.get(key) if val is not None: return json.loads(val) result = compute_fn() conn.setex(key, ttl, json.dumps(result, ensure_ascii=False)) return result finally: release_lock(conn, lock_key, token)这段代码里的双重检查很重要。拿到锁之后要再读一次缓存,因为在你等锁的过程中,可能已经有别的实例算完写进去了。少了这一步,就会重复计算。
4.3 锁超时与续期
锁的 TTL 设多长是个难题。设短了,计算还没完锁就过期了,别的实例进来重复计算;设长了,万一持锁实例崩了,锁要等很久才释放。
我的做法是 TTL 设为预估计算时间的 2 到 3 倍,同时加一个看门狗机制,在计算过程中定期续期。续期逻辑要判断锁还是不是自己的,用 Lua 脚本做原子检查加续期。
def renew_lock(conn, lock_key: str, token: str, ttl_ms: int) -> bool: lua = """ if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('pexpire', KEYS[1], ARGV[2]) else return 0 end """ return conn.eval(lua, 1, lock_key, token, ttl_ms) == 1看门狗用一个后台线程,每隔 TTL 的三分之一时间续一次。计算结束后停掉线程并释放锁。这套机制在 Redisson 这类客户端库里已经封装好了,如果你用 Java 可以直接用,Python 的话自己实现也不复杂。
5. 序列化选型与数据类型实战
5.1 序列化格式怎么选
Redis 存的是字节,任何对象都要序列化。选错格式,要么占内存,要么慢,要么跨语言不通。常见几种对比:
| 格式 | 体积 | 速度 | 可读性 | 跨语言 | 适用场景 |
|---|---|---|---|---|---|
| JSON | 中 | 中 | 好 | 好 | 通用,调试友好 |
| MessagePack | 小 | 快 | 差 | 好 | 体积敏感 |
| Pickle | 中 | 快 | 差 | 差 | 仅 Python 内部 |
| Protobuf | 最小 | 最快 | 差 | 好 | 高性能、强 schema |
Agent 场景我默认用 JSON。原因很简单:调试方便,出问题能直接GET出来看内容,跨语言也没障碍。如果你的 Agent 是 Python 写的,但缓存要被 Rust 或 Go 的服务读,JSON 是唯一省心的选择。
Pickle 千万别用。它不仅能序列化数据,还能序列化可执行代码,反序列化不可信数据等于直接执行任意代码,安全风险极大。而且 Pickle 跨 Python 版本都可能不兼容,升级个版本缓存全废。
如果确实对体积和速度敏感,比如缓存大量 Embedding 向量,用 MessagePack 或者直接把向量转成二进制。一个 1536 维的 float32 向量,JSON 存要 20KB 左右,二进制只要 6KB,差了三倍多。
import numpy as np import msgpack def pack_vector(vec: np.ndarray) -> bytes: return msgpack.packb(vec.astype(np.float32).tobytes()) def unpack_vector(data: bytes, dim: int = 1536) -> np.ndarray: return np.frombuffer(msgpack.unpackb(data), dtype=np.float32).reshape(dim)5.2 Redis 数据类型在 Agent 里的用法
别只会用 String。Redis 的五种基础类型在 Agent 场景各有妙用。
String是最常用的,存 LLM 响应、Embedding、工具结果。注意单个 Value 别超过 10KB,太大了读写都慢,还容易阻塞。大对象考虑拆成多个键或者用 Hash。
Hash适合存结构化的对话状态。比如一个会话的元信息:用户 ID、当前轮次、最后活跃时间,用 Hash 存比多个 String 键省内存,还能用HGETALL一次取回。
List适合存对话历史。LPUSH加新消息,LRANGE取最近 N 条,LTRIM控制长度。天然就是消息队列的语义。
def append_message(conn, session_id: str, role: str, content: str, max_len: int = 50): key = f"agent:session:{session_id}:messages" msg = json.dumps({"role": role, "content": content}, ensure_ascii=False) pipe = conn.pipeline() pipe.lpush(key, msg) pipe.ltrim(key, 0, max_len - 1) pipe.expire(key, 7200) pipe.execute()用 pipeline 把三个命令打包发送,减少网络往返。LTRIM保证列表不会无限增长,EXPIRE保证会话结束后自动清理。
Set可以用来做去重。比如记录某个用户今天问过哪些问题,避免重复计费。Sorted Set适合做带权重的排序,比如按相关性给检索结果排序,或者做限流的时间窗口计数。
5.3 内存管理与淘汰策略
Redis 内存是有限的,必须配淘汰策略。maxmemory-policy有八种,Agent 场景常用的是这两个:
allkeys-lru:所有键里淘汰最近最少使用的。适合缓存场景,不区分数据重要性。volatile-lru:只在设了 TTL 的键里淘汰。适合混合场景,有些键你希望永久保留。
我一般用allkeys-lru,因为 Agent 缓存的数据基本都可以重建,淘汰了下次再算就行。但要注意,如果你的 Redis 同时存了分布式锁这类不能丢的数据,就不能用allkeys系列,得用volatile系列,并且给锁设 TTL。
maxmemory设多少?留出系统内存的 20% 给 Redis 自身开销和 fork 时的 Copy-On-Write。比如 16GB 的机器,Redis 最多设 12GB。别设成 0(不限制),那样内存涨满会被系统 OOM Killer 干掉。
注意:淘汰策略是全局的,没法针对单个键设置。所以重要数据要么单独用一个 Redis 实例,要么确保它不会被淘汰(比如不设 TTL 且用 volatile 策略)。
6. 缓存穿透、击穿、雪崩的排查与治理
6.1 三种缓存问题的区别与应对
这三个词经常被混着说,其实场景完全不同,治理手段也不一样。
缓存穿透:查一个根本不存在的数据,缓存永远不命中,每次都打到后端。恶意攻击或者逻辑 bug 都会造成。治理手段是缓存空值——查不到也往 Redis 里写一个特殊标记,设短 TTL,比如 60 秒。这样短时间内重复查同一个不存在的键,直接返回空,不打后端。
NULL_MARKER = "__NULL__" def get_or_compute(conn, key: str, compute_fn, ttl: int, null_ttl: int = 60): val = conn.get(key) if val is not None: return None if val == NULL_MARKER else json.loads(val) result = compute_fn() if result is None: conn.setex(key, null_ttl, NULL_MARKER) return None conn.setex(key, ttl, json.dumps(result, ensure_ascii=False)) return result缓存击穿:某个热点键过期瞬间,大量请求同时打到后端。治理手段就是前面讲的分布式锁,只让一个请求去计算,其他等待。
缓存雪崩:大量键在同一时间集中过期,后端瞬间压力暴增。治理手段是 TTL 加随机抖动,让过期时间分散开。前面jitter_ttl那个函数就是干这个的。
6.2 排查缓存问题的思路
线上缓存出问题,排查顺序我一般是这样的:
先看命中率。Redis 的INFO stats里有keyspace_hits和keyspace_misses,命中率等于 hits 除以两者之和。Agent 缓存命中率低于 30% 就要警惕了,要么键设计有问题,要么 TTL 太短。
再看慢查询。SLOWLOG GET 10看最近的慢命令。Agent 场景常见的是大 Value 的读写,或者KEYS *这种全量扫描。生产环境绝对禁止用KEYS,要用SCAN分批遍历。
然后看内存和淘汰。INFO memory看used_memory和maxmemory的比例,INFO stats看evicted_keys。如果淘汰量很大,说明内存不够,要么扩容要么优化存储。
最后看连接数。INFO clients看connected_clients。Agent 如果用了连接池没配好,可能连接泄漏,把 Redis 的连接数打满。
| 现象 | 可能原因 | 排查命令 | 解决方向 |
|---|---|---|---|
| 命中率骤降 | 键设计变更、TTL 太短 | INFO stats | 检查键生成逻辑 |
| 响应变慢 | 大 Value、慢命令 | SLOWLOG GET | 拆分大对象、禁用 KEYS |
| 内存告警 | 数据量增长、无淘汰 | INFO memory | 扩容或调淘汰策略 |
| 连接超时 | 连接池泄漏 | INFO clients | 检查连接释放逻辑 |
6.3 缓存一致性怎么保证
Agent 缓存和真实数据源之间的一致性是绕不开的问题。严格来说,缓存和数据库之间没有完美的一致性方案,只能根据业务容忍度选合适的。
先更新数据库再删缓存是最常用的方案。为什么不更新缓存而是删缓存?因为更新缓存可能失败,而且并发更新时顺序难保证。删缓存让下次读的时候自然重建,简单可靠。但有个小概率问题:读请求在删缓存之前读了旧值,删缓存之后才写回缓存,导致缓存里是旧值。这个概率很低,一般业务能接受。
延迟双删是对上面方案的加强:更新数据库后删一次缓存,等几百毫秒再删一次。第二次删是为了清掉可能被并发读请求写回的旧值。代价是有一段时间缓存不命中。
对于 Agent 场景,我的建议是别过度追求强一致。LLM 响应、Embedding 这些数据本身就有一定容错空间,缓存旧一点问题不大。真正需要强一致的是涉及金额、权限这类数据,那种场景干脆别用缓存,直接查库。
7. 不同技术栈下的落地要点
7.1 Python Agent 的缓存集成
Python 生态里redis-py是标准选择。连接池一定要配,别每次操作都新建连接。
from redis import ConnectionPool, Redis pool = ConnectionPool( host='localhost', port=6379, db=0, max_connections=50, socket_timeout=2, socket_connect_timeout=1, retry_on_timeout=True, decode_responses=False, ) r = Redis(connection_pool=pool)max_connections按你的并发量设,一般并发数的 1.5 倍。socket_timeout设 2 秒,别设太长,Redis 卡住时快速失败比一直等好。decode_responses=False保持字节模式,避免二进制数据被错误解码。
如果你用 LangChain 这类框架,它内置了RedisCache和RedisSemanticCache。语义缓存会用 Embedding 做相似度匹配,即使用户问法不同也能命中,命中率比精确匹配高不少。但语义缓存有误命中风险,相似度阈值要调好,一般 0.9 以上比较安全。
7.2 Rust Agent 的缓存集成
Rust 写 Agent 越来越常见,性能和内存控制是优势。Redis 客户端推荐redis-rs,异步场景配tokio。
use redis::AsyncCommands; use redis::aio::ConnectionManager; async fn get_cached(conn: &mut ConnectionManager, key: &str) -> redis::RedisResult<Option<String>> { conn.get(key).await } async fn set_cached(conn: &mut ConnectionManager, key: &str, val: &str, ttl: u64) -> redis::RedisResult<()> { conn.set_ex(key, val, ttl).await }ConnectionManager自带重连和连接池,比手动管理连接省心。Rust 的类型系统能帮你在编译期发现很多序列化问题,但要注意serde的序列化结果要和 Python 端保持一致,否则跨语言缓存会读不出来。JSON 字段名、数字精度这些都要对齐。
7.3 可视化工具与日常运维
命令行虽然强大,但日常排查有个可视化工具效率高很多。RedisInsight 是官方出的,免费,支持内存分析、慢查询查看、键浏览。Another Redis Desktop Manager 也不错,轻量,跨平台。
我平时排查问题习惯开两个窗口:一个 RedisInsight 看键的分布和内容,一个终端跑redis-cli --stat看实时 QPS 和内存变化。--stat每秒刷新一次,能直观看到流量波动,定位突发问题很有用。
日志方面,redis-cli MONITOR能实时打印所有命令,但生产环境慎用,高 QPS 下会拖慢 Redis。要看日志还是配slowlog,只记录超过阈值的命令,开销小。
8. 我踩过的坑和几条实在建议
先说几个具体的坑。第一个是键的编码问题。Python 的json.dumps默认ensure_ascii=True,中文会被转义成\uXXXX,导致同样的中文内容在不同代码路径下算出不同的键。统一加ensure_ascii=False能解决。
第二个是浮点数精度。前面提过 temperature 要 round,其实所有参与键计算的浮点数都要处理。Embedding 向量如果直接序列化当键,不同平台的浮点表示差异会导致永远不命中。向量的键应该用原始文本算,不是用向量算。
第三个是连接池耗尽。Agent 里如果有异步任务没正确释放连接,跑一段时间连接池就满了,所有请求卡住。排查时看INFO clients的connected_clients是不是持续增长。解决方法是确保每个连接用完归还,用上下文管理器或者try/finally。
第四个是序列化版本不兼容。有次升级 Python 版本,Pickle 格式变了,缓存全部读不出来,Agent 直接报错。从那以后我所有缓存都用 JSON,虽然体积大点,但稳定。
几条实在建议。缓存键加个版本前缀,比如v1:agent:llm:xxx。这样当你改了键的生成逻辑,直接换版本号,旧缓存自然失效,不用手动清理,也不会新旧混用。
监控命中率和延迟。这两个指标能反映缓存层的健康度。命中率低说明键设计或 TTL 有问题,延迟高说明 Redis 本身有压力。把这两个指标接到你的监控系统里,出问题能第一时间发现。
别把 Redis 当数据库用。它是缓存,数据可以丢,可以重建。如果你的 Agent 依赖 Redis 里的数据做持久化存储,那就要考虑持久化配置(RDB 加 AOF),但即便如此,也不如直接用正经数据库可靠。Redis 挂了,Agent 应该能降级到直接计算,而不是整个挂掉。
压测再上线。缓存层的性能瓶颈往往在并发高的时候才暴露。上线前用redis-benchmark压一下,看看你的配置能扛多少 QPS。Agent 的缓存 QPS 一般不会太高,但突发流量要有余量。
最后分享一个我常用的降级模式:Redis 不可用时,Agent 自动切换到无缓存模式,所有计算直接走。用一个熔断器包住 Redis 操作,连续失败几次就短路一段时间,定期探测恢复。这样 Redis 出问题不会拖垮整个 Agent,只是变慢而已。这个模式在线上救过我好几次,强烈建议加上。