Redis 在 AI Agent 里到底扮演什么角色,这个问题我在过去大半年里被问过不下二十次。每次有人听说我在 Agent 项目里用 Redis 做缓存,第一反应往往是"不就是个 key-value 存储吗,能有多复杂"。但真正把 Agent 跑在生产环境里、面对几十上百个并发会话、还要控制 token 成本和响应延迟的时候,你会发现 Redis 的用法和传统的 Web 缓存完全不是一回事。Agent 的缓存对象是对话上下文、工具调用结果、向量检索片段、模型推理中间态,这些东西的失效策略、序列化方式、并发访问模式都和普通业务缓存有本质区别。这篇内容就是把我踩过的坑、调过的参数、以及最终沉淀下来的一套可复现方案完整讲清楚,适合正在搭建 AI Agent 或者准备给现有 Agent 加缓存层的朋友参考,不管你是用 Python 还是 Rust,思路是通用的。
1. 为什么 AI Agent 的缓存不能照搬 Web 缓存那套
1.1 Agent 请求链路里真正昂贵的是什么
普通 Web 应用的缓存逻辑很直白:数据库查询慢,就在前面加一层 Redis,把查询结果按主键缓存起来,设个 TTL,命中就返回,不命中就回源。这套逻辑之所以成立,是因为 Web 请求的输入是高度重复的——同一个用户 ID 查同一份资料,结果就是一样的。
但 AI Agent 的请求链路完全不是这个结构。一次 Agent 调用通常包含这几个阶段:接收用户输入、拼接系统提示词和历史对话、调用大模型推理、解析模型返回的工具调用意图、执行工具、把工具结果再喂回模型、生成最终回复。这里面真正耗时的是大模型推理,一次调用动辄几秒到几十秒,而工具执行和数据库查询反而可能是毫秒级的。
这就带来一个反直觉的结论:Agent 缓存的第一优先级不是缓存数据库查询,而是缓存模型推理结果和工具调用结果。我见过不少团队上来就给数据库加缓存,结果发现端到端延迟几乎没降,因为瓶颈根本不在那儿。你得先搞清楚自己的 Agent 时间花在哪里,再决定缓存什么。
1.2 对话上下文的缓存粒度问题
Agent 的对话是有状态的,这是和普通 Web 请求最大的区别。用户问"帮我查一下北京明天的天气",Agent 调用天气工具返回结果,用户接着问"那后天呢",这时候 Agent 需要知道"那"指的是天气查询这个上下文。
如果你把整个对话历史当成一个整体去缓存,粒度就太粗了,任何一轮新对话都会导致缓存失效。正确的做法是按轮次或者按消息单元缓存,每一轮的用户输入加助手回复作为一个缓存单元,同时维护一个会话级别的消息列表引用。这样新的一轮对话只需要追加,不需要重建整个上下文。
我在实际项目里用的是这样的结构:会话 ID 作为主键,存储一个有序列表,列表里每个元素是一轮对话的序列化结果。读取的时候按需取最近 N 轮,而不是一次性把全部历史拉出来。这个设计在长对话场景下能省掉大量网络传输和反序列化开销。
1.3 缓存命中率为什么在 Agent 场景下天然偏低
Web 缓存的命中率做到 90% 以上很常见,但 Agent 缓存能做到 40% 就已经不错了。原因在于 Agent 的输入是自然语言,同样的意图可以有无数种表达方式。"北京天气怎么样"和"查一下北京今天天气"在语义上几乎一样,但字符串完全不同,如果你用输入文本的哈希做缓存键,这两个请求就是两次独立的模型调用。
所以 Agent 缓存的键设计必须做语义归一化。常见的做法是先用一个轻量级的意图识别或者 embedding 相似度匹配,把语义相同的请求映射到同一个缓存键上。这一步本身也有成本,所以要在"归一化开销"和"缓存收益"之间找平衡点。我的经验是,对于高频的、模板化的查询(比如天气、汇率、订单状态),做语义归一化收益明显;对于开放式的创作类请求,归一化成本高且收益低,不如不做。
2. Redis 数据结构的选型:别只会用 String
2.1 会话上下文用 Hash 还是 List
这是我在项目初期纠结最久的一个问题。会话上下文本质上是一个有序的消息序列,直觉上应该用 List,因为 List 天然支持顺序和范围读取。但实际用下来,Hash 在很多场景下反而更合适。
List 的问题是,如果你想更新中间某一轮对话(比如用户编辑了历史消息),操作会非常别扭,需要重建整个列表。而 Hash 可以用轮次编号作为 field,每一轮独立存储,更新某一轮就是一次 HSET,读取最近 N 轮可以用 HGETALL 配合应用层排序,或者用 Sorted Set 维护轮次索引。
我最终的方案是Hash 存内容加 Sorted Set 存索引的组合。Hash 的 field 是轮次 ID,value 是序列化后的对话内容;Sorted Set 的 score 是时间戳,member 是轮次 ID。这样既能快速定位最近几轮,又能独立更新任意一轮,还能按时间范围清理过期数据。多维护一个 Sorted Set 的成本,换来的是操作灵活性的大幅提升,这笔账很划算。
2.2 工具调用结果缓存为什么适合用 String 加 TTL
工具调用结果的特点是:输入参数确定,输出就确定,而且大部分工具结果有时效性。比如查天气,五分钟前的数据和现在的数据可能就不一样了;查股票价格,秒级变化。这类数据用 String 存储最合适,键是工具名加参数哈希,值是序列化的结果,TTL 根据数据时效性设置。
这里有个细节值得说:TTL 不要设成固定值,而是根据数据特性分层设置。天气数据设 5 到 10 分钟,汇率设 1 分钟,静态知识库查询可以设几小时甚至一天。我见过有人图省事全部设 60 秒,结果静态数据频繁回源,白白浪费了缓存空间和网络开销。
另外,工具调用结果的序列化格式建议用 MessagePack 或者 JSON 的紧凑模式,不要用 Java 原生序列化或者 Pickle。前者跨语言兼容性好,后者有安全风险且体积大。在 Python 项目里我一般用 orjson,序列化速度比标准库快好几倍,体积也小。
2.3 向量检索结果用哪种结构存
Agent 经常需要做 RAG 检索,把用户问题转成向量,去向量库查相似片段。这个检索过程本身可能就要几十到几百毫秒,如果同一个问题反复问,缓存检索结果收益很大。
但向量检索结果的缓存有个特殊性:它不是精确匹配,而是相似度匹配。你没法用问题文本做精确的键,因为换个说法向量就变了。我的做法是对查询向量做量化后作为键的一部分,比如把 1536 维的向量降维到 64 维再量化成字符串,作为 Hash 的 field。这样语义相近的查询有较大概率落到同一个桶里,虽然会有一定的误命中,但在 RAG 场景下,检索到相似但不完全相同的片段通常也是可接受的。
这个方案不是银弹,量化会损失精度,误命中率需要根据你的业务容忍度来调。如果对准确性要求极高,那就老老实实每次检索,或者用更精细的局部敏感哈希方案。
2.4 分布式锁在 Agent 并发控制中的实际用法
Agent 处理并发请求时,有些操作必须串行化。比如同一个会话的连续消息,如果两个请求同时到达,可能会出现上下文错乱。这时候就需要分布式锁。
Redis 做分布式锁的标准做法是 SET key value NX PX timeout,value 用唯一标识(比如 UUID),释放锁的时候用 Lua 脚本校验 value 再删除,防止误删别人的锁。这个模式网上的资料很多,但 Agent 场景下有个坑:锁的粒度要按会话 ID 来,而不是全局锁。全局锁会让所有并发请求排队,吞吐量直接归零。按会话加锁,不同会话之间互不影响,只有同一会话的并发请求才会串行。
锁的超时时间也要仔细设。设太短,业务没执行完锁就释放了,会出现并发问题;设太长,万一持有锁的进程崩溃,其他请求要等很久。我的经验值是设为业务平均耗时的 3 到 5 倍,同时配合看门狗机制定期续期。
3. 缓存失效策略:Agent 场景下的特殊考量
3.1 主动失效和被动过期的取舍
Web 缓存里被动过期(TTL 到期自动删除)是主流,因为数据变化不频繁,容忍一定的陈旧度。但 Agent 场景下,有些数据必须主动失效。
最典型的是用户主动修改了偏好设置或者知识库内容,这时候相关的缓存必须立即清除,否则 Agent 会基于旧数据做出错误决策。我的做法是维护一套失效事件订阅机制:当底层数据发生变化时,发布一个失效事件,缓存层订阅这个事件并清除对应的键。Redis 的 Pub/Sub 或者 Stream 都能实现这个模式。
但要注意,Pub/Sub 是不可靠的,消息可能丢失。如果对一致性要求高,用 Stream 加消费者组,保证消息至少被消费一次。代价是复杂度上升,需要处理重复消费和消费位点管理。
3.2 缓存雪崩在 Agent 场景下的表现和预防
缓存雪崩指的是大量缓存同时失效,导致请求全部打到后端。Agent 场景下这个问题更严重,因为后端是大模型 API,有速率限制,一旦被打爆,整个服务就不可用了。
预防雪崩的核心是给 TTL 加随机抖动。比如你本来想设 300 秒,那就设成 300 加上 0 到 60 之间的随机数。这样即使一批缓存是同时写入的,过期时间也会分散开,不会集中失效。
另一个手段是热点数据永不过期加后台刷新。对于访问频率极高的数据,不设 TTL,而是起一个后台任务定期更新。这样缓存永远有值,不会出现失效瞬间的穿透。代价是需要额外的后台任务管理,以及要处理更新失败时的降级逻辑。
3.3 缓存穿透:不存在的键反复查询
缓存穿透是指查询一个根本不存在的键,缓存里没有,每次都打到后端。Agent 场景下,用户可能问一些 Agent 根本无法处理的问题,如果每次都去查一遍,既浪费资源又拉高延迟。
解决方案是缓存空结果。查询不到的时候,也往 Redis 里写一个特殊标记(比如空字符串或者特定的占位符),设一个较短的 TTL。下次同样的查询进来,直接返回空结果,不再回源。TTL 要短一些,因为数据可能后来才存在,设太长会导致新数据查不到。
对于恶意构造的不存在键,还可以用布隆过滤器做前置拦截。把所有可能存在的键放进布隆过滤器,查询前先过一遍,不存在就直接返回。布隆过滤器有误判率,但不会漏判,适合做这种前置过滤。
3.4 缓存一致性:Agent 读到的数据是旧的吗
这是最容易被忽视的问题。Agent 基于缓存数据做决策,如果缓存是旧的,决策就可能出错。比如用户刚把收货地址改了,Agent 还用旧地址下单,这就是事故。
强一致性在分布式缓存里很难做到,通常只能做到最终一致。我的策略是根据数据敏感度分级:对于地址、支付信息这类强敏感数据,不走缓存或者缓存 TTL 设得极短(几秒),并且写操作后主动失效;对于商品描述、知识库这类弱敏感数据,容忍几分钟的陈旧度。
还有一个技巧是版本号机制。每次数据更新,版本号加一,缓存里存版本号。读取的时候对比版本号,如果缓存版本落后于数据库版本,就回源。这需要在数据库侧维护版本号,增加了一点复杂度,但能有效避免读到旧数据。
4. 序列化与性能:那些拖慢 Agent 的隐形杀手
4.1 序列化格式的选择直接影响延迟
Agent 的缓存读写非常频繁,序列化的开销会被放大。我做过一组对比测试,同样的数据结构,用不同序列化方式,单次读写耗时能差出好几倍。
| 序列化方式 | 相对体积 | 序列化速度 | 反序列化速度 | 跨语言 |
|---|---|---|---|---|
| JSON (标准库) | 1.0 | 1.0 | 1.0 | 好 |
| orjson | 0.7 | 3.5 | 4.0 | 好 |
| MessagePack | 0.6 | 2.8 | 3.2 | 好 |
| Pickle | 0.9 | 2.0 | 2.5 | 差 |
| Protobuf | 0.4 | 4.5 | 5.0 | 好 |
从数据看,orjson 和 MessagePack 是性价比最高的选择。Protobuf 体积最小速度最快,但需要预定义 schema,改起来麻烦,适合结构稳定的场景。Pickle 虽然速度还行,但跨语言差且有安全风险,不建议在生产环境用。
4.2 大 key 问题:一个会话存了几百 KB
Redis 是单线程处理命令的,一个大 key 的读写会阻塞其他请求。Agent 的会话上下文如果一直追加,很容易变成大 key。我见过一个会话存了 500KB 的对话历史,每次读取都要几毫秒,并发一上来整个 Redis 就卡住了。
解决办法是分片存储。把长对话按轮次切分,每 N 轮存一个 key,读取的时候按需加载。或者用 Hash 结构,每个 field 存一轮,避免单个 value 过大。Redis 官方建议单个 value 不要超过 10KB,超过就要考虑拆分。
另外要定期清理不活跃的会话。可以给每个会话的 key 设 TTL,每次访问时续期,长时间不访问就自动过期。这样既能控制内存占用,又能避免大 key 堆积。
4.3 连接池配置:别让连接成为瓶颈
Agent 服务通常是多线程或者异步的,每个请求都要访问 Redis。如果每次请求都新建连接,开销会非常大。必须用连接池。
连接池的大小要根据并发量来定。太小了请求排队,太大了浪费资源且可能触发 Redis 的最大连接数限制。我的经验公式是:连接池大小等于平均并发数乘以 1.5。比如你的 Agent 服务平均同时处理 50 个请求,连接池设 75 左右比较合适。
还有一个容易忽略的点是连接的超时设置。Redis 连接如果长时间空闲,可能被服务端或者中间网络设备断开。客户端需要配置合理的超时和重连策略。我一般设连接超时 2 秒,读写超时 1 秒,同时开启自动重连。
4.4 批量操作减少网络往返
Agent 一次请求可能需要读写多个缓存键,如果每个键都单独发一次命令,网络往返的开销会累积。Redis 支持 Pipeline 和批量命令,能把多次往返合并成一次。
比如你要读取会话上下文、工具缓存、用户偏好三个键,用 MGET 一次就能拿到,比三次 GET 快得多。写入的时候用 Pipeline 把多个命令打包发送,也能显著降低延迟。
但要注意,Pipeline 里的命令不是原子的,中间可能插入其他客户端的命令。如果需要原子性,用 MULTI/EXEC 事务或者 Lua 脚本。Lua 脚本在 Agent 场景下特别有用,比如"检查缓存是否存在,不存在则写入并返回"这种逻辑,用 Lua 能保证原子性且减少往返。
5. 从零搭一套 Agent 缓存层的实操路径
5.1 环境准备和 Redis 部署方式选择
先说部署。本地开发用 Docker 起一个单机 Redis 就够了,命令很简单:
docker run -d --name agent-redis -p 6379:6379 redis:7-alpine --maxmemory 512mb --maxmemory-policy allkeys-lru这里有两个参数值得解释。--maxmemory限制内存使用,防止 Redis 吃光服务器内存。--maxmemory-policy allkeys-lru是淘汰策略,内存满了之后淘汰最近最少使用的键。Agent 缓存场景下,LRU 通常比 LFU 更合适,因为访问模式变化快,新数据很快会变成热点。
生产环境建议用主从加哨兵,或者直接用 Redis Cluster。主从保证高可用,哨兵负责故障转移。Cluster 适合数据量大、需要分片的场景,但配置复杂,小规模 Agent 服务用主从就够了。
macOS 上如果不想用 Docker,也可以用 Homebrew 安装:brew install redis,然后brew services start redis启动。Windows 用户建议用 WSL2 里的 Docker,原生 Windows 版 Redis 版本落后且维护不活跃。
5.2 缓存键的命名规范设计
键的命名看起来是小事,但项目一大就会乱。我建议用冒号分隔的层级结构,格式是业务:实体:标识:属性。比如:
agent:session:abc123:messages会话消息agent:tool:weather:hash123工具调用结果agent:user:user456:prefs用户偏好
这样的命名一眼就能看出是什么数据,方便排查问题,也方便用SCAN命令按前缀批量操作。注意不要用KEYS命令做前缀匹配,它会阻塞 Redis,生产环境用SCAN游标遍历。
键名不要太长,会浪费内存。但也不要为了省内存用无意义的缩写,可维护性比省那点内存重要得多。
5.3 缓存读写逻辑的代码骨架
下面是一个 Python 的缓存层骨架,用 redis-py 客户端,展示了核心的读写和失效逻辑:
import orjson import redis from typing import Any, Optional class AgentCache: def __init__(self, redis_client: redis.Redis): self.r = redis_client def get_session_messages(self, session_id: str, limit: int = 10) -> list: key = f"agent:session:{session_id}:messages" # 用 Sorted Set 拿最近 limit 轮的 ID round_ids = self.r.zrevrange(key + ":index", 0, limit - 1) if not round_ids: return [] # 批量拿内容 pipe = self.r.pipeline() for rid in round_ids: pipe.hget(key, rid) results = pipe.execute() return [orjson.loads(r) for r in results if r] def append_message(self, session_id: str, round_id: str, message: dict, ttl: int = 3600): key = f"agent:session:{session_id}:messages" pipe = self.r.pipeline() pipe.hset(key, round_id, orjson.dumps(message)) pipe.zadd(key + ":index", {round_id: float(round_id)}) pipe.expire(key, ttl) pipe.expire(key + ":index", ttl) pipe.execute() def get_tool_result(self, tool_name: str, params_hash: str) -> Optional[Any]: key = f"agent:tool:{tool_name}:{params_hash}" data = self.r.get(key) return orjson.loads(data) if data else None def set_tool_result(self, tool_name: str, params_hash: str, result: Any, ttl: int = 300): key = f"agent:tool:{tool_name}:{params_hash}" self.r.set(key, orjson.dumps(result), ex=ttl)这段代码里有几个设计点值得说明。会话消息用 Hash 加 Sorted Set 的组合,Hash 存内容,Sorted Set 存索引,读取时先拿索引再批量取内容。工具结果用简单的 String 加 TTL。所有序列化都用 orjson,速度快体积小。
5.4 并发场景下的锁实现
前面提到会话级别的分布式锁,这里给出具体实现:
import uuid import time class SessionLock: def __init__(self, redis_client: redis.Redis): self.r = redis_client def acquire(self, session_id: str, timeout: int = 10) -> Optional[str]: key = f"agent:lock:session:{session_id}" token = str(uuid.uuid4()) if self.r.set(key, token, nx=True, ex=timeout): return token return None def release(self, session_id: str, token: str) -> bool: key = f"agent:lock:session:{session_id}" lua = """ if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end """ return bool(self.r.eval(lua, 1, key, token))释放锁用 Lua 脚本保证"校验加删除"的原子性,这是防止误删别人锁的关键。如果不用 Lua,先 GET 再 DEL 中间可能插入其他操作,导致删错。
实际使用的时候,获取锁失败不要立即返回错误,而是短暂重试几次,因为锁的持有时间通常很短。重试间隔用指数退避,避免大量请求同时重试造成惊群。
5.5 监控指标:怎么知道缓存层是否健康
缓存上线不是终点,得持续监控。我关注这几个核心指标:
| 指标 | 含义 | 健康范围 |
|---|---|---|
| 命中率 | 缓存命中次数除以总查询次数 | 30% 以上 |
| 平均延迟 | 单次缓存操作耗时 | 5ms 以内 |
| 内存使用率 | 已用内存除以最大内存 | 80% 以下 |
| 淘汰键数量 | 单位时间被淘汰的键数 | 越低越好 |
| 连接数 | 当前活跃连接数 | 低于最大连接数的 70% |
命中率低于 30% 说明缓存设计有问题,要么键设计不合理,要么 TTL 太短。内存使用率长期高于 80% 要考虑扩容或者优化数据结构。淘汰键数量突然升高说明内存不够用了,需要及时处理。
这些指标可以通过 Redis 的 INFO 命令获取,也可以用 Prometheus 加 redis_exporter 做可视化监控。我一般会在 Grafana 上配一个看板,把这些指标和 Agent 的端到端延迟放在一起看,方便定位问题。
6. 几个真实踩过的坑和对应的解法
6.1 序列化不一致导致的脏读
项目早期我用 JSON 序列化,但不同模块用了不同的库,有的用标准库 json,有的用 orjson。结果就是 A 模块写进去的数据,B 模块读出来报错,因为 orjson 默认把 datetime 序列化成 ISO 格式字符串,而标准库 json 遇到 datetime 直接抛异常。
这个坑排查了很久,因为错误信息不直观。后来统一了序列化库,并且在缓存层做了版本标记,键名里带上序列化格式的版本号,升级格式的时候新旧数据可以共存,平滑过渡。
教训就是:缓存层的数据格式必须统一管理,不能各写各的。最好封装一个统一的序列化模块,所有读写都走这个模块。
6.2 TTL 设置不当引发的连锁反应
有一次线上出现大面积超时,排查发现是某个工具调用的缓存 TTL 设成了 1 秒,导致几乎每次请求都回源,后端 API 被打爆,触发限流,然后所有依赖这个工具的 Agent 请求都超时。
这个问题的根源是 TTL 设得太短,缓存形同虚设。后来我定了个规矩:任何缓存 TTL 不得低于 30 秒,除非有明确的业务理由。同时加了监控,当某个键的回源率超过阈值时告警。
6.3 大 key 导致的 Redis 阻塞
前面提到过大 key 问题,我实际遇到过一次。某个用户的会话历史特别长,存了将近 1MB,每次读取都要几毫秒。平时没事,但赶上并发高峰,多个大 key 同时读写,Redis 单线程被阻塞,其他请求全部排队,延迟飙升。
解决办法是给会话历史做分片,每 20 轮存一个 key,读取的时候按需加载。同时加了会话长度限制,超过一定轮数就做摘要压缩,把早期对话总结成一段文字,减少存储量。
6.4 主从切换时的缓存丢失
用主从架构的时候,主节点故障切换到从节点,如果从节点还没同步完数据,就会丢失一部分缓存。对于 Agent 来说,丢失工具缓存问题不大,重新查一次就行;但丢失会话上下文就麻烦了,用户会发现 Agent 突然"失忆"。
我的应对策略是会话上下文做持久化,不能只存在 Redis 里。每次写入 Redis 的同时,异步写一份到数据库。Redis 挂了从数据库恢复,虽然慢一点但不会丢数据。工具缓存这类可重建的数据就不需要持久化,丢了就丢了。
6.5 缓存预热:冷启动时的性能抖动
服务重启或者 Redis 重启后,缓存是空的,所有请求都回源,这时候延迟会明显升高。如果赶上流量高峰,可能直接把后端打挂。
解决办法是缓存预热。服务启动时,把高频访问的数据提前加载到缓存里。哪些是高频数据?可以从历史访问日志里统计,或者维护一个热点键列表。预热的时机要选在流量低峰期,避免预热本身造成压力。
我现在项目里的做法是,服务启动后先加载最近一小时访问频率最高的 1000 个键,然后再开始接收流量。这样冷启动的抖动基本消除了。
7. 关于 Agent 缓存治理的一些个人体会
缓存治理这个词听起来很正式,但落到实处就是几件具体的事:定期清理无用键、监控命中率和内存、根据业务变化调整 TTL、以及处理各种异常情况。我在实际项目里最大的体会是,缓存不是加得越多越好,而是要加得准。
有些团队一上来就给所有东西加缓存,结果缓存层变得极其复杂,维护成本高,还容易出 bug。我的建议是先不加缓存,把 Agent 跑通,用监控找出真正的性能瓶颈,然后针对性地加缓存。加的时候也要小步验证,先在一个场景试点,确认收益和稳定性后再推广。
另一个体会是,缓存的失效逻辑比写入逻辑更重要。写入逻辑错了,最多是缓存没生效;失效逻辑错了,会导致读到脏数据,可能引发业务事故。所以每次设计缓存的时候,我都会先想清楚:这个数据什么时候会变?变了之后怎么让缓存失效?失效不及时会有什么后果?把这些问题想明白了再动手写代码。
Redis 本身是个很成熟的工具,坑大多不在 Redis 本身,而在于你怎么用它。Agent 场景的特殊性在于数据形态复杂、访问模式多变、对延迟敏感,这就要求我们在用 Redis 的时候多想一想,而不是套用现成的模板。希望这些经验能帮你少走一些弯路。