1. 从一次线上抖动说起:AI Agent 为什么绕不开 Redis
去年冬天我接手了一个内部 AI Agent 项目,功能不复杂——用户丢一段自然语言进来,Agent 负责拆解意图、调用工具、拼装结果返回。上线第一周风平浪静,第二周开始陆续有用户反馈"有时候要等十几秒才出结果"。我第一反应是模型推理慢,查了链路耗时才发现,真正吃掉时间的是 Agent 反复去查同一份知识库文档、反复调用同一个外部接口拿配置。一个会话里,同一个工具被调了七八次,每次都是全量请求。
这就是 AI Agent 和传统 CRUD 应用最不一样的地方:Agent 的决策是动态的、多轮的、带工具调用的,它不像一个固定接口那样"一次请求一次响应",而是在一次用户交互里可能触发几十次内部调用。如果这些调用每次都打到数据库、外部 API 或者向量库,延迟会成倍叠加,成本也会失控。Redis 在这里扮演的角色,不是"锦上添花的缓存",而是Agent 能不能扛住并发、能不能把响应压到可接受范围的关键基础设施。
我后来把整个 Agent 的缓存层重新梳理了一遍,用 Redis 做了会话状态、工具结果、向量检索结果、限流计数四类缓存,P95 延迟从 12 秒降到 2.3 秒,外部 API 调用量降了大概七成。这篇文章就把这套东西拆开讲清楚:AI Agent 场景下 Redis 到底缓存什么、key 怎么设计、过期策略怎么定、并发和一致性怎么处理,以及我踩过的那些坑。适合正在搭 Agent、或者 Agent 已经上线但开始遇到性能瓶颈的开发者看,对 Redis 本身有一定了解会更好,但基础部分我也会补。
需要先说明一点:下面所有方案都是基于我实际项目(Python + FastAPI + LangChain 风格的工具调用链)总结的,不同技术栈细节会有差异,但缓存的分层思路是通用的。
2. AI Agent 的缓存需求和普通 Web 缓存根本不是一回事
2.1 Agent 一次交互里到底发生了什么
要理解缓存怎么设计,先得看清楚 Agent 的执行链路。一个典型的 Agent 处理一次用户输入,大致会经历这些步骤:
- 接收用户消息,加载会话历史(多轮对话上下文)
- 把历史 + 当前输入 + 工具描述一起送给大模型,让模型决定下一步
- 模型返回"我要调用某个工具,参数是 XXX"
- 执行工具(可能是查数据库、调外部 API、检索向量库、跑一段代码)
- 把工具结果塞回上下文,再次请求模型
- 重复 3-5,直到模型认为可以给出最终答案
- 返回结果,写回会话历史
注意第 3 到第 6 步是一个循环,循环次数取决于任务复杂度,简单问答可能一轮就结束,复杂任务(比如"帮我分析这份报表并生成摘要")可能循环十几轮。每一轮里,工具调用都是潜在的缓存点,会话历史是另一个缓存点,模型本身的输出在特定条件下也能缓存。
普通 Web 应用的缓存模型很简单:一个 URL 对应一份数据,读多写少,缓存 key 基本就是 URL 或者资源 ID。但 Agent 的缓存对象是动态生成的中间产物,它的 key 需要从"工具名 + 参数"里推导出来,而且这些中间产物可能只在一次会话内有效,跨会话复用价值有限。这个差异直接决定了后面所有的设计选择。
2.2 四类缓存对象,价值密度完全不同
我在项目里把 Agent 的缓存对象分成四类,它们的特性差异很大,不能一刀切:
| 缓存类型 | 典型内容 | 生命周期 | 复用范围 | 命中收益 |
|---|---|---|---|---|
| 会话状态 | 对话历史、当前任务进度 | 分钟到小时级 | 单用户单会话 | 避免重复加载上下文 |
| 工具结果 | API 返回、DB 查询、计算结果 | 秒到分钟级 | 跨用户可复用 | 直接省掉一次外部调用 |
| 检索结果 | 向量检索 top-k、关键词召回 | 分钟到小时级 | 跨用户可复用 | 省掉一次向量库查询 |
| 计数限流 | 调用次数、token 消耗 | 秒到分钟级 | 单用户/单租户 | 保护下游不被打爆 |
会话状态是私有的,一个用户的对话历史绝不能被另一个用户读到,所以 key 里必须带用户或会话标识,而且要考虑隐私和过期清理。工具结果和检索结果则往往是可共享的——比如"查询北京今天的天气"这个工具调用,所有问这个问题的用户结果都一样,完全可以共享一份缓存。计数限流是控制类数据,它的价值不在省时间,而在保护系统。
把这四类混在一起用同一套过期策略,是我见过最常见的错误。会话状态设 5 分钟过期,用户聊到一半上下文没了;工具结果设 1 小时过期,结果拿到的是过期数据。分开对待,是设计的第一步。
2.3 为什么是 Redis,而不是本地内存或数据库
有人会问:Agent 通常部署在单机或者少量实例上,用进程内内存(比如 Python 的 dict、functools.lru_cache)做缓存不就行了,为什么要引入 Redis?
我一开始也是这么想的,直到遇到三个问题。第一,多实例部署。Agent 服务为了扛并发通常会起多个 worker 或多个 Pod,进程内缓存各存各的,命中率直接除以实例数,而且会话状态如果存在本地,用户请求被负载均衡打到另一个实例就"失忆"了。第二,重启即失效。进程内缓存一重启就没了,Agent 冷启动时所有请求都穿透到下游,容易把外部 API 打挂。第三,缺少过期和淘汰策略。自己用 dict 实现 LRU + TTL 不是不行,但并发安全、内存上限、淘汰算法都要自己写,容易出 bug。
Redis 恰好把这三点都解决了:它是独立进程,多实例共享;支持持久化,重启不丢(视配置);原生支持 TTL、多种淘汰策略、原子操作。而且它的数据结构丰富,字符串、哈希、列表、有序集合、HyperLogLog 都能在 Agent 场景里找到用武之地。代价是引入一次网络往返(本地几毫秒,跨机房可能十几毫秒),但对于动辄几百毫秒到几秒的模型调用来说,这点开销完全可以接受。
提示:如果你的 Agent 是纯单机、低并发、可接受冷启动穿透,进程内缓存确实够用,不必为了"架构好看"硬上 Redis。但只要涉及多实例或会话状态,Redis 基本是必选项。
3. Key 设计:Agent 缓存最容易翻车的地方
3.1 工具结果缓存的 key 怎么拼才不冲突
工具结果缓存是收益最大的一类,也是最容易出问题的。核心问题是:怎么把一个工具调用唯一地映射成一个 key。
最朴素的做法是tool:{工具名}:{参数},但参数往往是 JSON,直接拼进去又长又乱,还可能有顺序问题({"a":1,"b":2}和{"b":2,"a":1}语义相同但字符串不同)。我的做法是:
- 对参数字典做规范化排序(按 key 字典序)
- 序列化成紧凑 JSON(无空格)
- 对序列化结果做哈希(我用 SHA-256,取前 16 字节的十六进制)
- key 拼成
agent:tool:{工具名}:{参数哈希}
这样无论参数多复杂,key 长度都可控,且语义相同的参数一定得到相同的 key。哈希碰撞概率在 16 字节(128 位)下可以忽略。
但这里有个坑:不是所有工具都适合缓存。带副作用的工具(发消息、下单、写数据库)绝对不能缓存结果,否则用户点两次只执行一次。只有幂等的、只读的工具才适合缓存。我在工具注册时就给每个工具打了一个cacheable标记,只有标记为 true 的才走缓存逻辑。这个标记必须由开发者显式声明,不能靠自动推断,因为"这个工具是否幂等"是业务语义,机器猜不准。
还有一个细节:参数里如果包含时间戳、随机数、用户 ID这类每次都变的值,缓存永远不会命中。我遇到过一个大坑——某个工具的参数里带了一个request_id,导致缓存命中率长期为 0,排查了半天才发现。后来我在工具层加了一个"缓存参数白名单",只把真正影响结果的参数纳入 key 计算,无关参数直接剔除。
3.2 会话状态的 key 与隐私边界
会话状态的 key 相对简单,但要考虑隐私。我用的是agent:session:{租户ID}:{会话ID},租户 ID 放在前面是为了后续做批量清理和隔离。会话 ID 是服务端生成的 UUID,不直接用用户 ID,避免用户 ID 泄露在 key 里(Redis 的 key 在某些运维场景下是可见的)。
会话状态我存成 Redis Hash,字段包括history(对话历史,序列化后存)、last_active(最后活跃时间)、task_state(当前任务进度,如果 Agent 支持中断续跑)。用 Hash 而不是 String 的好处是可以单独更新某个字段,比如只刷新last_active而不用把整个历史读出来再写回去,省带宽也省 CPU。
注意:会话历史可能很长(多轮对话累积),单个 value 可能到几十 KB 甚至更大。Redis 单 value 建议不要超过 100KB,否则网络传输和序列化都会变慢。我的做法是只保留最近 N 轮(比如 20 轮)在 Redis 里,更早的历史归档到数据库,需要时再按需加载。
3.3 检索结果缓存:向量检索也能缓存
向量检索是 Agent 里另一个耗时大户,一次 top-k 检索在百万级向量库里可能要几十到几百毫秒。但检索结果能不能缓存,取决于查询是否重复。用户问"公司年假政策是什么"和"年假怎么休",语义相同但字面不同,如果按字面做 key,缓存命中率会很低。
我的方案是两级 key:第一级用查询文本的规范化哈希(去空格、转小写、去标点)做 key,命中就直接返回;第二级用查询的 embedding 向量做近似匹配(这个成本较高,只在第一级未命中时尝试)。实际跑下来,第一级命中率大概 30%-40%,第二级能再补 10% 左右。第二级实现复杂,如果团队资源有限,只做第一级也够用。
检索结果的 TTL 我设得比较短(5-15 分钟),因为知识库可能更新,缓存太久会返回过期内容。如果知识库更新频率很低(比如一周一次),可以适当延长。
4. 过期、淘汰与一致性:让缓存"该走的时候走"
4.1 TTL 不是拍脑袋定的,要按数据特性分层
TTL 设置是缓存设计里最需要动脑子的地方。设太短,命中率低,缓存形同虚设;设太长,数据过期,用户拿到错误结果。我的经验是按数据特性分四档:
- 秒级(5-30 秒):高频变化的数据,比如实时股价、库存数量。这类数据缓存价值有限,但能挡住瞬时并发。
- 分钟级(1-15 分钟):大部分工具结果和检索结果。这是主力档位,兼顾命中率和新鲜度。
- 小时级(1-24 小时):变化很慢的配置、字典、静态知识。比如"公司部门列表"这种,一天变一次都算频繁。
- 会话级(跟随会话生命周期):会话状态,TTL 设为"用户可能的最长思考时间 + 缓冲",我设的是 30 分钟,每次访问刷新。
这里有个反直觉的点:TTL 不要设成整齐的整数。如果所有 key 都在同一时刻过期(比如都设 600 秒且同时写入),会造成"缓存雪崩"——大量 key 同时失效,请求瞬间全打到下游。我的做法是在基础 TTL 上加一个随机抖动,比如 600 秒 ± 60 秒,让过期时间分散开。
4.2 内存满了怎么办:淘汰策略的选择
Redis 内存是有限的,当内存达到maxmemory时,需要按策略淘汰 key。Agent 场景我推荐allkeys-lru或allkeys-lfu:
allkeys-lru:淘汰最久未使用的。适合访问模式比较均匀的场景。allkeys-lfu:淘汰访问频率最低的。适合有明显热点的场景,比如少数几个工具被高频调用。
我实测下来,Agent 场景用allkeys-lfu效果更好,因为工具调用有明显的长尾——少数几个工具占了大部分调用量,LFU 能更好地保住这些热点。但要注意 LFU 需要 Redis 4.0+,且计数器有衰减机制,配置不当可能导致新 key 难以积累频率。
提示:千万不要用
noeviction。内存满了之后写操作直接报错,Agent 会因为写不进缓存而整个链路失败。这个坑我在测试环境踩过一次,生产环境务必避开。
4.3 缓存和真实数据不一致了怎么办
这是缓存绕不开的经典问题。Agent 场景下,不一致主要来自两个地方:工具结果缓存和知识库检索缓存。
对于工具结果,我的策略是短 TTL + 主动失效。短 TTL 保证即使不一致,窗口也很小(最多几分钟)。主动失效用于那些"数据变更后必须立即生效"的场景,比如用户更新了自己的资料,那么所有涉及该用户资料的缓存 key 都要删掉。主动失效的难点是"怎么知道哪些 key 受影响",我的做法是在 key 里嵌入可识别的维度(比如用户 ID),变更时用SCAN+ 模式匹配删除。注意不要用KEYS命令,它会阻塞 Redis,生产环境用SCAN分批扫。
对于知识库检索,我用的是版本号机制。知识库每次更新,版本号 +1,缓存 key 里带上版本号。更新后旧版本的 key 自然不再被访问,等 TTL 到期自动清理。这样避免了主动删除的复杂性,代价是更新后有一段时间新旧版本共存(旧版本还在被 TTL 内的请求命中),但因为我们更新不频繁,这个窗口可以接受。
5. 并发场景:Agent 扛并发的缓存实战
5.1 缓存击穿:热点 key 失效瞬间的雪崩
"缓存击穿"指的是某个热点 key 突然失效,大量并发请求同时发现缓存没有,一起去查下游,把下游打挂。Agent 场景里,热点 key 可能是某个高频工具的结果,或者某个热门问题的检索结果。
解决方案是互斥锁 + 双重检查。逻辑是:缓存未命中时,先尝试获取一个分布式锁(用 Redis 的SET key value NX PX),拿到锁的请求去查下游并写缓存,没拿到锁的请求短暂等待后重试读缓存。这样只有一个请求真正打到下游。
import redis import time import json r = redis.Redis(host='localhost', port=6379, decode_responses=True) def get_with_lock(cache_key, lock_key, fetch_func, ttl=600, lock_ttl=10): # 第一次检查 cached = r.get(cache_key) if cached is not None: return json.loads(cached) # 尝试获取锁 got_lock = r.set(lock_key, "1", nx=True, px=lock_ttl * 1000) if got_lock: try: # 双重检查,防止在等锁期间别人已经写好 cached = r.get(cache_key) if cached is not None: return json.loads(cached) data = fetch_func() r.set(cache_key, json.dumps(data), ex=ttl) return data finally: r.delete(lock_key) else: # 没拿到锁,短暂等待后重试 time.sleep(0.05) cached = r.get(cache_key) if cached is not None: return json.loads(cached) # 兜底:直接查下游,避免无限等待 return fetch_func()这段代码有几个细节值得说。lock_ttl必须设置,否则拿锁的进程崩溃后锁永远不释放,其他请求全部卡死。等待重试的次数要有上限,不能无限循环,我一般最多重试 3 次,超过就直接查下游兜底,保证可用性优先。锁的 value 最好用唯一标识(比如 UUID),释放时校验,避免误删别人的锁——上面为了简洁省略了,生产代码建议加上。
5.2 缓存穿透:查不存在的数据
"缓存穿透"指的是查询一个根本不存在的数据,缓存永远不命中,每次都打到下游。Agent 场景里,用户可能问一个知识库里没有的问题,或者调用一个参数非法的工具,如果每次都穿透,下游压力会很大。
解决方案有两个。一是缓存空结果,查不到时也写一个特殊值(比如__NULL__)进缓存,TTL 设短一点(比如 60 秒)。二是布隆过滤器,在缓存前面加一层,快速判断"这个 key 是否可能存在",不存在直接返回。布隆过滤器实现复杂,我一般用空结果缓存就够了,简单有效。
注意:空结果缓存的 TTL 不能太长,否则数据后来真的存在了,用户还是拿到"不存在"。60 秒是个比较稳妥的值。
5.3 分布式锁在 Agent 里的正确用法
Agent 里需要分布式锁的场景其实不多,主要是两类:一是上面说的缓存击穿保护,二是防止同一个任务被重复执行。比如用户重复点击"生成报告",如果不加锁,可能触发两次相同的 Agent 任务,浪费资源还可能产生重复结果。
用 Redis 做分布式锁,核心是SET key value NX PX ttl这个原子命令。value 用唯一 ID,释放时用 Lua 脚本校验后删除,保证原子性:
-- 释放锁的 Lua 脚本 if redis.call("get", KEYS[1]) == ARGV[1] then return redis.call("del", KEYS[1]) else return 0 end这个脚本保证"校验 value 相等"和"删除"是一个原子操作,避免删掉别人的锁。锁的 TTL 要大于任务最长执行时间,否则任务还没跑完锁就过期了,另一个请求又能拿到锁。如果任务时间不确定,可以用"看门狗"机制定期续期,但这会增加复杂度,我一般把 TTL 设得足够大(比如任务预估时间的 3 倍)来规避。
6. 我踩过的坑和几条硬经验
6.1 序列化选错,性能差一个数量级
缓存 value 的序列化方式对性能影响很大。我一开始用 Python 的pickle,方便是方便,但有两个问题:一是跨语言不兼容(后来有 Go 服务要读同一份缓存,读不了),二是 pickle 反序列化有安全风险(如果缓存被污染,可能执行任意代码)。后来换成 JSON,可读性和兼容性都好,但体积比 pickle 大,序列化速度也慢一些。
再后来我试了msgpack,体积比 JSON 小 30% 左右,速度也快,跨语言支持好。最终我选了 msgpack 作为主力,JSON 作为需要人工排查时的备选。如果你的 Agent 是纯 Python 且不跨语言,pickle 也能用,但一定要确保缓存来源可信。
6.2 大 key 是 Redis 的隐形杀手
Agent 的会话历史、工具返回的大 JSON,很容易变成"大 key"(单个 value 超过 10KB 甚至 100KB)。大 key 的危害是:读写慢、网络传输慢、删除时阻塞 Redis(删除大 key 是 O(n) 操作)、集群模式下可能导致数据倾斜。
我的应对是:拆分 + 压缩 + 限制。会话历史按轮次拆成多个 key,或者只存最近 N 轮;大 JSON 先压缩(gzip 或 zstd)再存;在写入前检查 value 大小,超过阈值就告警并拒绝缓存。Redis 4.0+ 提供了UNLINK命令,异步删除大 key,比DEL安全,生产环境删大 key 优先用UNLINK。
6.3 监控没做好,出了问题两眼一抹黑
缓存上线后,必须监控几个核心指标:命中率(hit rate)、内存使用率、慢查询、连接数。命中率低于预期,说明 key 设计或 TTL 有问题;内存使用率持续高位,说明淘汰策略或容量需要调整;慢查询说明有大 key 或复杂命令;连接数暴涨可能是连接泄漏。
我用的是 Redis 自带的INFO命令 + Prometheus 采集,重点看keyspace_hits、keyspace_misses、used_memory、evicted_keys。evicted_keys持续增长说明内存不够,在淘汰 key,这时候要么扩容要么优化 TTL。这些指标我配了告警,命中率跌破阈值或者内存超过 80% 就通知。
6.4 几个容易被忽略的配置项
最后列几个我实际调过的配置,都是踩坑之后才重视的:
maxmemory-policy:设成allkeys-lfu,别用默认的noeviction。timeout:客户端空闲连接超时,设成 300 秒,避免连接泄漏。tcp-keepalive:设成 60 秒,保持长连接活跃。save:持久化策略,Agent 缓存对持久化要求不高,可以关掉 RDB 或者降低频率,减少 fork 开销。appendonly:如果缓存丢了能接受,AOF 也可以关掉,提升写入性能。
这些配置没有绝对的对错,取决于你的可用性要求和性能要求。我的原则是:缓存是加速层,不是数据源,丢了能重建的,就大胆牺牲持久化换性能。
7. 写在最后的一点个人体会
这套 Redis 缓存方案在我项目里跑了半年多,中间经历过一次大促级别的流量冲击,Agent 服务没有出现雪崩,P95 延迟稳定在 2 秒出头。回头看,最关键的不是用了多高级的技术,而是把缓存对象分清楚了、把 key 设计规范了、把过期和并发处理想透了。很多团队上缓存出问题,不是 Redis 不好用,而是把缓存当成了一个"随手塞东西的桶",没有认真对待它的生命周期和一致性。
如果你正准备给 Agent 加缓存,我的建议是先别急着写代码,拿张纸把"哪些数据可以缓存、缓存多久、key 长什么样、失效了怎么办"这四个问题写清楚,再动手。这四个问题想明白了,代码其实没多少。至于具体的 Redis 版本、部署方式、客户端选型,反而是次要的——用你团队最熟悉的那套就行,别为了追新引入不必要的复杂度。