1. 为什么 AI Agent 必须认真对待缓存这件事
做过 AI Agent 项目的人都有一个共同体会:真正让系统变慢的,往往不是大模型推理本身,而是那些看起来不起眼的重复查询、重复工具调用和重复上下文拼装。一个稍微复杂点的 Agent,一次用户请求背后可能触发十几轮工具调用、几十次向量检索、上百次状态读写。如果每一轮都老老实实打到数据库或者外部 API,延迟会像滚雪球一样涨上去,成本也会失控。
我最早做 Agent 的时候没太在意缓存,觉得“先把功能跑通再说”。结果上线第一周就出问题:同一个用户连续问三个相似问题,Agent 每次都重新检索知识库、重新调用天气接口、重新拼 prompt,响应时间从 2 秒飙到 8 秒,API 账单也翻了好几倍。后来把 Redis 引进来做缓存层,情况立刻好转——重复的工具调用结果直接命中缓存,会话状态用 Redis 存,向量检索结果做短时缓存,整体 P95 延迟降了 60% 以上。
所以这篇内容想聊的就是AI Agent 和 Redis 缓存怎么配合。核心关键词就三个:AI Agent、Redis、缓存。我会从架构设计、数据类型选型、实操搭建、并发处理、缓存治理几个角度,把我在实际项目里踩过的坑和总结出来的方案完整讲一遍。适合正在搭 Agent 的开发者、后端工程师,也适合对缓存治理感兴趣但还没系统梳理过的朋友。哪怕你之前只用过 Redis 做简单的 KV 存储,看完也能把思路迁移到 Agent 场景里。
需要先说明一点:Agent 的缓存和传统 Web 缓存不是一回事。传统缓存大多是“请求-响应”级别的,键值明确、生命周期清晰。Agent 的缓存对象更杂——有工具调用结果、有中间推理状态、有向量检索片段、有会话记忆、还有模型输出的部分结果。这些东西的失效策略、一致性要求、数据大小差异极大,不能一把梭全塞进一个 Redis 实例里用同一套 TTL。下面我会逐个拆开讲。
2. AI Agent 缓存体系整体设计与选型思路
2.1 Agent 场景下缓存到底缓存什么
先把缓存对象理清楚,这是设计的地基。我在项目里通常把 Agent 的缓存分成五类,每一类的特征和策略都不一样。
第一类是工具调用结果缓存。Agent 调用外部 API(天气、搜索、数据库查询)得到的结果,如果输入参数相同,短时间内完全可以复用。这类缓存键通常是“工具名 + 参数哈希”,TTL 设得比较短,几分钟到几十分钟,因为外部数据可能变化。
第二类是向量检索结果缓存。用户问了一个问题,Agent 去向量库检索 top-k 片段。如果短时间内有相似 query,检索结果可以复用。这类缓存要注意 query 的归一化处理,否则“北京天气”和“北京的天气”会被当成两个键。
第三类是会话状态与记忆缓存。多轮对话里,Agent 需要记住上下文。把会话历史放 Redis,比每次从数据库读快得多,而且天然支持过期清理。这类数据一致性要求高,不能随便丢。
第四类是模型输出片段缓存。有些 Agent 会做流式输出,或者把大任务拆成子任务。子任务的输出如果可复用,缓存下来能省不少 token。
第五类是限流与并发控制状态。这个严格说不算“缓存”,但实践中经常和缓存放一起,用 Redis 的原子操作做计数器和分布式锁。
把这五类分清楚之后,你会发现它们的 TTL、数据结构、一致性要求完全不同。我的做法是按类别分 key 前缀,甚至分不同的 Redis 逻辑库或实例,避免互相干扰。
2.2 为什么选 Redis 而不是本地缓存
有人会问:Agent 服务本身可以用进程内缓存(比如 Python 的functools.lru_cache或者 Caffeine),为什么还要引入 Redis?
本地缓存确实快,纳秒级访问,但它有几个硬伤。第一,多实例部署时不一致。Agent 服务通常要水平扩展,三个实例各自缓存一份,命中率被稀释,而且状态类数据没法共享。第二,重启即失效。进程一重启,缓存全没了,冷启动期间压力全打到后端。第三,容量受限。本地内存要留给模型加载、推理中间结果,能分给缓存的空间有限。
Redis 作为独立缓存层,解决了共享、持久化、容量这几个问题。而且 Redis 的数据结构丰富,字符串、哈希、列表、有序集合、Stream 都能用上,特别适合 Agent 这种数据形态多样的场景。比如会话历史用 List 或 Stream,工具调用计数用 String 的 INCR,向量检索结果用 Hash 存多字段,限流用有序集合做滑动窗口。
当然 Redis 也不是没代价,网络往返是主要开销。我的经验是:热数据放本地 + 冷数据放 Redis 做两级缓存,本地缓存 TTL 设得很短(几秒到几十秒),Redis 做兜底。这样既拿到本地缓存的低延迟,又保证多实例间基本一致。
2.3 缓存键设计:Agent 场景最容易翻车的地方
键设计看着简单,实际是 Agent 缓存里最容易出问题的地方。我见过太多项目因为键设计粗糙,导致缓存命中率极低或者命中错误数据。
核心原则是:键必须能唯一标识一次可复用的计算。对于工具调用,键应该是tool:{tool_name}:{hash(params)}。这里的 hash 要稳定,参数顺序不同但语义相同的调用应该映射到同一个键。我一般用参数排序后做 JSON 序列化再哈希,避免字典顺序问题。
对于向量检索,键是retrieval:{hash(normalized_query)}:{top_k}。query 归一化包括去空格、转小写、去掉标点。top_k 也要进键,因为不同 top_k 结果不同。
对于会话状态,键是session:{session_id},这个最直接。
有个坑要特别注意:不要把用户 ID 或者敏感信息直接拼进键里。一方面键会变长,另一方面如果 Redis 被未授权访问,键名本身就泄露信息。我通常对用户标识做一次哈希再拼进去。
还有一个细节是键的版本管理。当你的缓存逻辑升级(比如换了序列化方式、改了数据结构),旧键和新键会冲突。我的做法是在键前缀里加版本号,比如v2:tool:...,升级时直接换版本号,旧数据自然过期,不用手动清理。
3. Redis 数据类型选型与 Agent 缓存实操
3.1 五种核心数据类型在 Agent 里的具体用法
Redis 的数据类型不是拿来炫技的,每种都有它最适合的场景。我把 Agent 项目里最常用的五种列出来,配上具体用法。
String是最基础的,适合存单个值。工具调用结果、模型输出片段、简单的计数器都用它。比如SET tool:weather:hash123 "{...}" EX 600。注意大 value 要控制大小,超过 10KB 的结果我一般会压缩后再存,或者干脆不缓存。
Hash适合存结构化对象。会话状态里有很多字段(用户 ID、历史消息、当前意图、临时变量),用 Hash 存比存一个大 JSON 字符串更灵活,可以单独更新某个字段。HSET session:abc history "..." intent "query_weather",读取时HGETALL一次拿全。
List适合存有序的消息历史。Agent 的多轮对话历史天然是有序的,用LPUSH加新消息,LRANGE取最近 N 条,LTRIM控制长度。这样会话历史不会无限增长。
Sorted Set适合做限流和优先级队列。滑动窗口限流就是典型用法:把每次请求的时间戳作为 score 存进去,查询时用ZCOUNT统计窗口内的请求数,超了就拒绝。Agent 调用外部 API 时做限流特别有用。
Stream适合做事件日志和异步任务队列。Agent 的任务编排如果涉及异步步骤,可以用 Stream 做消息传递,支持消费者组,比 List 更专业。
下面这张表是我总结的选型对照,实际项目里直接照着选基本不会错。
| 数据类型 | Agent 场景 | 典型命令 | 注意事项 |
|---|---|---|---|
| String | 工具结果、计数器、锁 | SET/GET/INCR/SETNX | 大 value 要压缩,锁要设过期 |
| Hash | 会话状态、结构化对象 | HSET/HGETALL/HDEL | 字段多时注意内存,可拆分 |
| List | 消息历史、任务队列 | LPUSH/LRANGE/LTRIM | 必须 LTRIM 控制长度 |
| Sorted Set | 限流、优先级队列 | ZADD/ZCOUNT/ZREMRANGEBYSCORE | 定期清理过期成员 |
| Stream | 事件日志、异步任务 | XADD/XREADGROUP | 注意消费者组和 pending 处理 |
3.2 序列化方式的选择与性能影响
存对象就要序列化,这一步的选择直接影响性能和兼容性。常见的有 JSON、MessagePack、Pickle、Protobuf 几种。
JSON 最通用,可读性好,跨语言无障碍,缺点是体积大、序列化慢。MessagePack 体积小、速度快,但可读性差,调试时得专门工具解。Pickle 是 Python 专用,快但跨语言不行,而且有安全风险,绝对不要反序列化不可信数据。Protobuf 体积最小、最快,但需要预定义 schema,改结构麻烦。
我的实际选择是:内部服务间通信用 MessagePack,需要人工排查的用 JSON,跨语言且性能敏感的用 Protobuf。Agent 项目里工具调用结果我一般用 JSON,因为要经常看日志排查;会话状态用 MessagePack,因为读写频繁且不需要人看。
这里有个容易忽略的点:序列化后的数据要加类型标记。否则你改了数据结构,旧缓存反序列化会报错。我的做法是在 value 前面加一个短前缀标识版本,比如j:表示 JSON v1,m:表示 MessagePack v1。读取时先看前缀决定怎么解。
3.3 从零搭建:Redis 安装与 Agent 缓存层接入
先把环境搭起来。Redis 的安装方式很多,我按不同系统分别说。
Linux 上最省事的是包管理器,apt install redis-server或者yum install redis。装完systemctl start redis启动,redis-cli ping返回 PONG 就通了。macOS 上用 Homebrew,brew install redis,然后brew services start redis。Windows 官方不推荐原生跑,一般用 Docker 或者 WSL。
Docker 方式我最推荐,环境干净、版本可控。docker run -d --name redis -p 6379:6379 redis:7-alpine,生产环境记得挂载配置文件和持久化目录,加上密码。
docker run -d --name redis \ -p 6379:6379 \ -v /data/redis/conf:/usr/local/etc/redis \ -v /data/redis/data:/data \ redis:7-alpine \ redis-server /usr/local/etc/redis/redis.conf --requirepass yourpassword配置文件里几个关键参数要调。maxmemory设成物理内存的 70% 左右,留出余量。maxmemory-policy用allkeys-lru或volatile-lru,前者淘汰所有键,后者只淘汰设了过期时间的键。Agent 场景我倾向volatile-lru,因为会话状态这类不能丢的数据我不设过期,就不会被淘汰。
Python 接入用redis-py,连接池一定要配。redis.Redis(host='localhost', port=6379, password='xxx', decode_responses=True, max_connections=50)。decode_responses=True让返回的是字符串而不是 bytes,省得手动解码。连接池大小根据并发量调,一般 QPS 的 1.5 倍左右。
封装一个缓存工具类,把 get/set/delete 和序列化逻辑包起来,业务代码只调方法,不直接碰 Redis 命令。这样以后换缓存实现或者加监控都方便。
3.4 工具调用结果缓存的完整实现
这是 Agent 缓存里收益最直接的一块。我写一个完整的实现思路。
首先定义缓存键生成函数。工具名加上参数哈希,参数要先规范化。
import hashlib import json def make_tool_cache_key(tool_name, params, version="v1"): normalized = json.dumps(params, sort_keys=True, ensure_ascii=False) param_hash = hashlib.sha256(normalized.encode()).hexdigest()[:16] return f"{version}:tool:{tool_name}:{param_hash}"然后封装带缓存的工具调用。先查缓存,命中直接返回,未命中执行真实调用再写缓存。
def call_tool_with_cache(redis_client, tool_name, params, ttl=600): key = make_tool_cache_key(tool_name, params) cached = redis_client.get(key) if cached: return json.loads(cached) result = execute_tool(tool_name, params) redis_client.setex(key, ttl, json.dumps(result, ensure_ascii=False)) return resultTTL 的选择要看数据特性。天气这类变化快的设 5 到 10 分钟,知识库检索设 30 分钟到 1 小时,静态配置类可以设几小时甚至一天。我一般会按工具类型配一个 TTL 映射表,而不是写死。
有个细节:空结果也要缓存。如果某个查询确实没结果,不缓存的话每次都会重新查,白白浪费。但空结果的 TTL 要短一些,避免数据更新后还返回空。
3.5 会话状态用 Redis 存的具体方案
会话状态是 Agent 的核心数据,丢了用户体验直接崩。我用 Hash 存,结构清晰。
def save_session(redis_client, session_id, state, ttl=3600): key = f"session:{session_id}" redis_client.hset(key, mapping=state) redis_client.expire(key, ttl) def load_session(redis_client, session_id): key = f"session:{session_id}" return redis_client.hgetall(key)消息历史单独用 List 存,避免 Hash 里塞大数组。
def append_message(redis_client, session_id, message, max_len=50): key = f"session:{session_id}:history" redis_client.lpush(key, json.dumps(message, ensure_ascii=False)) redis_client.ltrim(key, 0, max_len - 1) redis_client.expire(key, 3600)LTRIM这步很关键,不控制长度的话历史会无限增长,内存迟早爆。我一般保留最近 50 条,够 Agent 理解上下文了。
会话状态的 TTL 要设得比单次对话长,但也不能太长。我设 1 小时,用户一小时内没新消息就自动清理。如果业务要求长期记忆,那应该落库,Redis 只做热数据缓存。
4. AI Agent 高并发场景下的缓存策略
4.1 Agent 怎么扛并发:缓存是第一道防线
“AI Agent 怎么扛并发”是个高频问题。我的答案很直接:先把缓存做扎实,再谈其他。
Agent 的并发压力主要来自三处:大模型调用、外部工具调用、向量检索。这三处都是慢操作,动辄几百毫秒到几秒。如果每个请求都实打实走一遍,并发能力会被死死卡住。
缓存能挡掉多少?根据我的实测,在问答类 Agent 里,工具调用结果的缓存命中率能到 40% 到 60%,向量检索命中率 30% 到 50%,会话状态读取几乎 100% 走 Redis。综合下来,后端实际压力能降一半以上。
除了缓存,还有几个配合手段。请求合并:短时间内相同的请求只执行一次,其他等结果。异步化:能并行的工具调用并行执行。降级:缓存没命中且后端压力大时,返回兜底结果而不是硬扛。这些后面细说。
4.2 缓存穿透、击穿、雪崩在 Agent 里的表现与对策
这三个经典问题在 Agent 场景里有独特表现,得针对性处理。
缓存穿透指查询一个不存在的数据,缓存和数据库都没有,每次都打到后端。Agent 里常见于用户问了一个知识库里没有的问题,每次都要走一遍向量检索。对策是缓存空结果,用一个特殊标记存起来,TTL 设短一点。或者用布隆过滤器预判,但 Agent 的 query 空间太大,布隆过滤器不太实用,我一般就用空结果缓存。
缓存击穿指某个热点键过期瞬间,大量请求同时打到后端。Agent 里常见于热门问题或者热门工具。对策是互斥锁重建:缓存失效时只让一个请求去重建,其他请求等待或返回旧值。
def get_with_lock(redis_client, key, rebuild_func, ttl=600): value = redis_client.get(key) if value: return json.loads(value) lock_key = f"lock:{key}" if redis_client.set(lock_key, "1", nx=True, ex=10): try: value = rebuild_func() redis_client.setex(key, ttl, json.dumps(value)) return value finally: redis_client.delete(lock_key) else: # 等待重建完成或返回兜底 time.sleep(0.1) value = redis_client.get(key) return json.loads(value) if value else None缓存雪崩指大量键同时过期,请求全打到后端。对策是TTL 加随机抖动。比如本来设 600 秒,实际设 600 加上 0 到 120 的随机值,让过期时间分散开。
import random ttl = 600 + random.randint(0, 120) redis_client.setex(key, ttl, value)这个技巧看着简单,效果非常明显。我有个项目就是因为没加抖动,每天整点缓存集体过期,后端瞬间被打满,加了抖动之后再没出现过。
4.3 分布式锁:Agent 并发控制的关键工具
Agent 里很多操作需要互斥,比如同一个会话的并发请求要串行处理,避免状态错乱。Redis 的SET NX EX是实现分布式锁的标准方式。
def acquire_lock(redis_client, lock_name, ttl=10): lock_key = f"lock:{lock_name}" token = str(uuid.uuid4()) if redis_client.set(lock_key, token, nx=True, ex=ttl): return token return None def release_lock(redis_client, lock_name, token): lock_key = f"lock:{lock_name}" # 用 Lua 保证原子性 lua = """ if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end """ redis_client.eval(lua, 1, lock_key, token)释放锁必须用 Lua 脚本保证“判断 token 再删除”的原子性,否则可能误删别人的锁。这个坑我踩过,当时用 GET 再 DEL 两步操作,高并发下删错了锁,导致两个请求同时进入临界区,会话状态直接乱了。
锁的 TTL 要设得比操作耗时长一点,但也不能太长,否则锁泄漏后要等很久。我一般设操作预估耗时的 2 到 3 倍。如果操作可能超时,还要加续期机制,用后台线程定期延长 TTL。
4.4 缓存与数据库的一致性怎么保证
Agent 的会话状态如果同时存 Redis 和数据库,就有一致性问题。我的策略是Redis 做主存储,数据库做持久化备份,而不是两边都当主。
具体做法是:写操作先写 Redis,然后异步落库。读操作只读 Redis,Redis 没有才回源数据库并回填。这样 Redis 是唯一的事实来源,不存在两边不一致的问题。异步落库用消息队列或者后台任务,失败重试。
如果业务要求强一致,那就得用先更新数据库再删除缓存的经典模式,配合延迟双删。但 Agent 场景大多不需要强一致,会话状态丢一两条消息用户基本无感,所以我的方案够用了。
有个细节:删除缓存比更新缓存更安全。更新缓存可能因为并发导致旧值覆盖新值,删除缓存让下次读时自然回源,更稳妥。
5. 缓存治理与线上问题排查实录
5.1 缓存治理:监控、清理与容量规划
缓存上线只是开始,治理才是长期工作。我关注几个核心指标。
命中率是最重要的。工具调用缓存命中率低于 30% 就要查原因,可能是键设计有问题,或者 TTL 太短。内存使用率要盯着,接近 maxmemory 就要扩容或清理。慢查询用SLOWLOG看,超过 10ms 的命令要优化。连接数接近上限要调连接池。
清理策略上,我一般不用KEYS命令,它会阻塞 Redis。用SCAN游标遍历,分批删除。或者干脆靠 TTL 自然过期,省事又安全。
容量规划有个粗略公式:所需内存 = 平均 value 大小 × 键数量 × 1.5。1.5 是 Redis 自身的开销系数。比如平均 value 2KB,100 万个键,大概需要 3GB。实际还要留余量,我一般按算出来的 1.5 倍配。
5.2 常见问题速查表
下面这张表是我这几年攒下来的,基本覆盖了 Agent 缓存 90% 的线上问题。
| 现象 | 可能原因 | 排查方法 | 解决方案 |
|---|---|---|---|
| 命中率突然下降 | 键设计变更、TTL 调整 | 对比变更前后的键样本 | 回滚变更,检查键生成逻辑 |
| 内存持续增长 | 键没设 TTL、历史没 LTRIM | INFO memory看 used_memory | 补 TTL,加 LTRIM,清理无用键 |
| 响应变慢 | 慢查询、大 key | SLOWLOG GET、--bigkeys | 拆分大 key,优化命令 |
| 连接超时 | 连接池太小、网络抖动 | 看客户端连接数指标 | 调大连接池,加重试 |
| 数据不一致 | 并发写、锁失效 | 查日志时间线 | 用 Lua 保证原子性,加锁 |
| 缓存雪崩 | TTL 集中过期 | 看过期时间分布 | TTL 加随机抖动 |
| 序列化报错 | 数据结构变更 | 看报错堆栈 | 加版本前缀,旧数据自然过期 |
5.3 几个我踩过的坑和独家经验
坑一:大 key 拖垮整个实例。有次我把整个会话历史塞进一个 Hash 字段,单个 key 几十 KB,读取时网络传输慢,还阻塞了其他命令。后来拆成 List 存消息,Hash 只存元数据,问题解决。经验是单个 value 控制在 10KB 以内,超了就拆。
坑二:TTL 设成 0 导致数据永不过期。有次代码里 TTL 变量没初始化,默认 0,结果缓存永远不过期,内存慢慢涨满。后来加了参数校验,TTL 必须大于 0,否则抛异常。
坑三:序列化方式混用。项目里有人用 JSON 有人用 Pickle,读的时候解错格式直接崩。后来统一封装序列化层,加类型前缀,才彻底解决。
坑四:忘记处理 Redis 不可用。有次 Redis 挂了,Agent 服务直接全线报错。后来加了降级逻辑,Redis 不可用时直接走后端,虽然慢但不至于挂。缓存是加速手段,不能成为单点依赖。
坑五:锁没设过期时间。早期用SETNX没加EX,进程崩溃后锁永远不释放,整个会话卡死。现在一律用SET key value NX EX ttl,原子且安全。
这些坑的共同点是:都是细节没处理好导致的。缓存这东西,原理不难,难的是把每个细节都想到。我的建议是上线前做一轮压力测试,专门测缓存失效、Redis 重启、高并发这些边界场景,能提前发现大部分问题。
5.4 缓存失效策略的进阶玩法
基础的 TTL 过期够用,但有些场景需要更精细的控制。
主动失效:数据源更新时主动删缓存。比如知识库更新了,把相关的检索缓存键删掉。实现上可以用键前缀扫描,或者维护一个键索引。
分级 TTL:热数据 TTL 长,冷数据 TTL 短。判断冷热可以用访问计数,Redis 的OBJECT FREQ能拿到访问频率(需要开启 LFU 策略)。
预热:服务启动或者缓存大面积失效后,提前把热点数据加载进缓存,避免冷启动压力。我一般写个预热脚本,把 top 1000 的热点 query 提前跑一遍。
多级缓存:本地缓存加 Redis 加数据库三层。本地缓存挡最热的请求,Redis 挡中等热度,数据库兜底。本地缓存 TTL 设几秒,保证多实例间基本一致。
这套组合拳打下来,Agent 的并发能力和响应速度会有质的提升。我在一个日活几万的 Agent 项目里用这套方案,单实例 QPS 从 50 提到了 300 以上,P99 延迟从 5 秒降到 1.5 秒。
最后分享一个小技巧:给缓存加个“命中来源”标记。在返回结果里带上是从本地缓存、Redis 还是后端来的,排查问题时一眼就能看出缓存有没有生效。这个标记只在调试模式返回,生产环境关掉,不影响性能。
这套东西不是一次设计出来的,是我在几个项目里反复迭代、踩坑、修正攒下来的。你要是刚开始做 Agent 缓存,建议先从工具调用结果缓存和会话状态缓存这两块入手,收益最直接,实现也简单。等这两块跑稳了,再逐步加上限流、分布式锁、多级缓存这些进阶能力。缓存治理是个持续活儿,别指望一次做到完美,边跑边调才是常态。