☰
AI Agent 缓存实战:Redis 语义缓存与高并发稳定性设计
2026/10/6 11:06:06 网站建设 项目流程

1. 为什么 AI Agent 的缓存层不能照搬传统 Web 那套

很多人第一次给 AI Agent 加 Redis 缓存,脑子里浮现的还是那套经典画面:查数据库之前先查 Redis,命中就返回,没命中就回源写缓存。这套逻辑在传统 CRUD 业务里跑了十几年,稳得很。但把它原封不动搬到 AI Agent 上,你会发现缓存命中率低得可怜,甚至出现"缓存了反而更慢"的诡异现象。

根本原因在于,AI Agent 的请求特征和传统 Web 请求完全不是一个物种。传统 Web 请求是确定性的——同样的 URL 参数,永远得到同样的结果,所以缓存 key 可以直接用参数拼出来。而 AI Agent 的每一次调用,输入是自然语言、上下文、工具返回结果、历史对话的混合体,输出还带随机性(temperature 参数一调,同样的输入能给你三种不同措辞的回答)。你如果拿用户原始问题当 key,那基本等于没缓存,因为没人会一字不差地问两遍。

我在实际项目里踩过的第一个坑就是这个。当时做一个客服场景的 Agent,用户问"我的订单怎么还没到"和"订单为啥一直不发货",语义几乎一样,但字符串完全不同,缓存直接穿透,Redis 里躺着一堆只被访问过一次的 key,内存哗哗涨,命中率不到 5%。后来才想明白:AI Agent 的缓存,key 的设计逻辑必须从"字符串匹配"升级到"语义匹配",这是和传统缓存最本质的分水岭。

所以这篇文章不讲 Redis 的安装教程,也不重复SET/GET的基础命令——那些东西官方文档写得比我清楚。我要聊的是:当 Redis 遇上 AI Agent,缓存该在哪一层做、key 怎么设计、什么该缓存什么绝对不能缓存、并发上来之后怎么扛、以及那些只有真正上线跑过才会遇到的坑。适合已经会用 Redis、但正在把 Agent 往生产环境推的开发者。

2. AI Agent 里到底哪些东西值得进 Redis

给 Agent 做缓存,第一步不是写代码,而是做减法。你得先搞清楚 Agent 的一次完整调用链路上,哪些环节是"重复计算"的重灾区,哪些环节缓存了会出人命。我一般把 Agent 的缓存对象分成四类,优先级从高到低排。

2.1 第一优先级:Embedding 向量与语义检索结果

这是收益最高、最该缓存的东西。RAG 场景下,用户每问一个问题,系统都要把问题转成向量,再去向量库做相似度检索。而 Embedding 模型调用是要花钱、要耗时的——一次 text-embedding 调用动辄几十到几百毫秒。如果同一个问题被问两次,你完全没必要算两遍向量。

我的做法是把问题文本的哈希作为 key,向量数组作为 value 存进 Redis。这里有个细节:向量是浮点数组,直接 JSON 序列化又大又慢,我一般用numpy的tobytes()转成二进制再存,读取时frombuffer还原,体积能压到 JSON 的三分之一左右。实测一个 1536 维的向量,JSON 序列化后约 30KB,二进制只要 6KB 出头。

import hashlib import numpy as np import redis r = redis.Redis(host='localhost', port=6379, decode_responses=False) def get_embedding_cached(text, embed_fn): key = "emb:" + hashlib.sha256(text.encode()).hexdigest() cached = r.get(key) if cached: return np.frombuffer(cached, dtype=np.float32) vec = embed_fn(text).astype(np.float32) r.setex(key, 86400, vec.tobytes()) # 缓存一天 return vec

注意:向量缓存一定要设过期时间。模型版本升级后,旧向量和新向量不在一个语义空间里,混用会导致检索结果错乱。我一般把模型版本号也拼进 key,比如emb:v2:,升级时直接换前缀,旧的自然过期。

2.2 第二优先级:工具调用(Tool Call)的幂等结果

Agent 调用外部工具时,很多工具是幂等的——比如查天气、查汇率、查某个固定 ID 的商品详情。这类调用结果在短时间内不会变,缓存起来能省下大量外部 API 调用。但这里有个判断标准:结果随时间变化的频率,决定了 TTL 的长短。

天气数据我一般缓存 10 分钟,汇率缓存 1 分钟,商品详情缓存 5 分钟。这个 TTL 不是拍脑袋定的,而是根据业务对"数据新鲜度"的容忍度倒推的。你缓存太久,用户看到过期数据会投诉;缓存太短,等于没缓存。我通常会在配置里把这些 TTL 做成可调的常量,上线后根据实际命中率和数据时效投诉率再微调。

2.3 第三优先级:LLM 的完整响应(要非常谨慎)

这是争议最大的一块。缓存 LLM 的完整回答,收益巨大——一次 GPT-4 级别的调用可能几秒钟、几分钱,缓存命中直接省掉。但风险也巨大:Agent 的回答往往依赖上下文,同样的用户问题,在不同对话历史下答案完全不同。你如果只拿问题当 key,会把 A 用户的答案返回给 B 用户,这是灾难级的事故。

我的原则是:只有当输入完全确定、且不涉及任何用户私有上下文时,才缓存完整响应。比如一些固定的知识问答、FAQ 场景。而且 key 里必须包含模型名 + temperature + 完整 prompt 哈希,任何一个变了都不能命中。涉及用户数据的场景,我宁可不算这个缓存,也不冒串数据的风险。

2.4 绝对不能缓存的:会话状态与中间推理链

有些东西看着像缓存,其实是状态,必须用 Redis 的另一种数据结构存,而不是当缓存用。比如多轮对话的session上下文、Agent 的scratchpad(中间推理步骤)。这些数据的特点是必须强一致、必须可更新、不能过期丢失。用SETEX存这些,一旦过期,用户对话就断了。

这类数据我一般用 Redis 的Hash结构存,key 是session:{session_id},字段是各个上下文变量,配合EXPIRE做滑动过期(每次访问续期)。它和缓存的区别在于:缓存丢了可以回源重建,会话状态丢了用户就得重新开始。

缓存对象推荐结构典型 TTL能否容忍丢失
Embedding 向量String(二进制)24 小时能,回源重算
工具调用结果String(JSON)1-10 分钟能,重新调用
LLM 完整响应String(JSON)视场景能,但需防串数据
会话上下文Hash滑动续期不能,丢了对话断
中间推理链List / Stream会话周期不能,影响逻辑

3. 语义缓存:让"意思一样"的问题命中同一个 key

前面反复提到,AI Agent 缓存的命门在 key。传统缓存用精确字符串,Agent 必须用语义。这一节我把语义缓存的落地方式讲透,这也是整个方案里技术含量最高的部分。

3.1 精确哈希缓存的天花板在哪

先说清楚为什么精确哈希不够用。假设你用sha256(用户问题)当 key,那么"北京今天天气怎么样"和"今天北京天气如何"是两个完全不同的 key,缓存命中率为零。在真实业务里,用户表达同一意图的方式可能有几十种,精确匹配的命中率通常撑死到 10%-20%。这个数字在流量大的时候也能省点钱,但远远不够。

我做过一个统计:某客服 Agent 上线精确缓存一周,Redis 里存了 8 万个 key,其中被访问超过 2 次的不到 3000 个,命中率 3.7%。这个数据说明,不做语义归一化,缓存基本是白做。

3.2 用向量相似度做 key 匹配的完整链路

语义缓存的核心思路是:不比对字符串,而是比对向量。流程是这样的——用户问题先转成向量,然后去 Redis 里找"有没有哪个已缓存问题的向量,和当前向量足够接近"。如果接近度超过阈值,就认为语义相同,直接返回缓存结果。

这里的关键是"怎么在 Redis 里做向量相似度搜索"。早期大家用SCAN遍历所有 key 逐个算余弦相似度,数据量一上来就废了。现在 Redis 提供了向量检索能力(Redis Stack 的 RediSearch 模块),可以建向量索引,用KNN查询直接找出最相似的 top-k。

# 建立向量索引(Redis Stack) FT.CREATE idx:semantic ON HASH PREFIX 1 "sem:" SCHEMA \ vec VECTOR HNSW 6 TYPE FLOAT32 DIM 1536 DISTANCE_METRIC COSINE \ answer TEXT \ created_at NUMERIC

建好索引后,查询就变成了:

def semantic_lookup(query_vec, threshold=0.92): # KNN 查询最相似的 1 条 q = f"*=>[KNN 1 @vec $vec AS score]" res = r.ft("idx:semantic").search( q, query_params={"vec": query_vec.tobytes()} ) if res.docs: doc = res.docs[0] similarity = 1 - float(doc.score) # COSINE 距离转相似度 if similarity >= threshold: return doc.answer return None

阈值0.92是我反复调出来的经验值。太低(比如 0.85)会把"北京天气"和"上海天气"判成同一个,返回错误答案;太高(比如 0.98)又几乎匹配不上,等于没做。不同 Embedding 模型的相似度分布不一样,这个值必须在你自己的数据上标定,不能照抄。

3.3 阈值标定的实操方法

怎么标定阈值?我的做法是准备一批"语义相同"和"语义不同"的问题对,各几十组,然后跑一遍看相似度分布。语义相同的对,相似度应该集中在一个区间;语义不同的对,落在另一个区间。两个区间的分界点,就是你的阈值。

我实测下来,用主流的中文 Embedding 模型,"同义问题"的余弦相似度大多在 0.93-0.98 之间,"不同问题"大多在 0.7-0.88 之间。所以 0.92 是个相对安全的切分点。但如果你做的是法律、医疗这种对准确性要求极高的场景,我建议把阈值提到 0.95 以上,宁可少命中也不能答错。

提示:语义缓存一定要留"人工兜底"的口子。我一般会在返回缓存结果时打个标记,如果用户对缓存答案点了"不满意",就把这条缓存标记为可疑,后续降低它的权重或直接删除。缓存不是一劳永逸的,它需要被"喂养"和"清理"。

4. 并发场景下 Redis 缓存层的稳定性设计

Agent 服务一旦上量,缓存层承受的并发压力会比传统业务更猛——因为 Agent 单次请求耗时长,用户等待期间可能重试、可能并发发起多个子任务,瞬时 QPS 波动极大。这一节聊几个我在生产环境里真正用到的稳定性手段。

4.1 缓存击穿:热点 key 失效瞬间的雪崩

Agent 场景里有个特别典型的问题:某个热门问题(比如"帮我写个周报模板")被大量用户同时问,它的缓存 key 一旦过期,瞬间几百个请求同时回源去调 LLM,直接把下游打爆。这就是缓存击穿。

传统方案是加互斥锁,只让一个请求回源,其他等待。但在 Agent 场景里,回源可能要好几秒,让几百个请求干等几秒,用户体验很差。我的做法是逻辑过期 + 异步重建:缓存 value 里额外存一个"逻辑过期时间",物理上不设 TTL 或者设很长的 TTL。读取时如果发现逻辑过期了,先返回旧值(保证响应快),同时异步起一个任务去重建缓存。

import json, time, threading def get_with_logical_expire(key, rebuild_fn, logical_ttl=300): raw = r.get(key) if raw: data = json.loads(raw) if data["expire_at"] > time.time(): return data["value"] # 逻辑过期,异步重建,先返回旧值 threading.Thread( target=rebuild_fn, args=(key,), daemon=True ).start() return data["value"] # 完全没缓存,同步回源 value = rebuild_fn(key) return value

这套逻辑的好处是:用户永远拿到的是"有点旧但很快"的答案,而不是"很新但等半天"的答案。对于 Agent 这种对延迟敏感的场景,这个取舍非常值。

4.2 缓存穿透:恶意或异常 key 的防护

穿透指的是查询一个根本不存在的 key,每次都回源。Agent 场景里,用户乱输、爬虫扫接口都会造成大量无效查询。防护手段有两个:一是空值缓存,查不到也往 Redis 里写个短 TTL 的空标记,避免重复回源;二是布隆过滤器,在缓存前面挡一层,快速判断"这个 key 一定不存在"。

我一般用空值缓存就够了,简单有效。但要注意空值的 TTL 要短(比如 60 秒),否则数据后来真的有了,你还一直返回空。

4.3 连接管理:别让连接池成为瓶颈

Agent 服务通常是异步的(FastAPI、asyncio),如果用同步的 Redis 客户端,会把事件循环堵死。我踩过这个坑:一开始用redis-py的同步客户端,QPS 一上来,整个服务卡成 PPT。后来换成redis.asyncio,问题立刻消失。

import redis.asyncio as aioredis pool = aioredis.ConnectionPool.from_url( "redis://localhost:6379", max_connections=50, decode_responses=False ) r = aioredis.Redis(connection_pool=pool)

max_connections这个值要结合你的并发量调。太小了请求排队,太大了 Redis 服务端扛不住。我的经验值是:并发峰值 × 1.5左右,然后压测验证。

4.4 超时与降级:Redis 挂了 Agent 不能挂

这是最容易被忽略的一点。很多人把 Redis 当成"必须有"的组件,一旦 Redis 抖动,整个 Agent 服务跟着报错。正确的姿势是:缓存层永远是可降级的。Redis 连不上,就跳过缓存直接回源,业务照常跑,只是慢一点、贵一点。

async def safe_cache_get(key): try: return await r.get(key) except Exception as e: logger.warning(f"Redis get failed: {e}, fallback to source") return None

配合合理的超时设置(我一般设socket_timeout=0.5秒),Redis 慢的时候快速失败,不拖累主流程。这个设计在线上救过我好几次——Redis 集群做迁移的那次,Agent 服务全程无感知。

5. 那些上线后才暴露的缓存治理问题

代码写完、压测通过,不代表缓存就稳了。真正的问题往往在上线后一两周才冒出来。这一节聊几个我亲身经历的、文档里不会写的坑。

5.1 内存悄悄涨满:key 没有统一的生命周期管理

Agent 缓存的 key 种类多、来源杂,很容易出现"这个模块设了 TTL,那个模块忘了设"的情况。结果就是 Redis 内存缓慢上涨,某天突然触发maxmemory淘汰策略,把正在用的会话数据也淘汰了。

我的治理办法是给所有 key 加统一前缀 + 强制 TTL 检查。在代码层面封装一个cache_set函数,如果调用方没传 TTL,直接抛异常,从源头杜绝"永久 key"。同时用INFO memory和--bigkeys定期巡检,找出异常大的 key。

def cache_set(key, value, ttl=None): if ttl is None: raise ValueError("TTL is required for cache keys") r.setex(key, ttl, value)

5.2 序列化格式不统一导致的"读不出来"

这个坑特别隐蔽。项目里不同的人用了不同的序列化方式——有人用pickle,有人用json,有人直接存字符串。结果 A 模块写的缓存,B 模块读出来是乱码,还以为是缓存没命中。我后来强制规定:所有缓存 value 统一用 JSON(二进制数据除外),并在 value 里带一个_v字段标识版本。读取时先检查版本,不匹配就当没命中。

5.3 缓存与真实数据不一致的排查链路

有一次线上出现用户投诉:Agent 返回的商品价格是旧的。排查过程我记录一下,因为这类问题很典型。

第一步,确认是不是缓存问题——直接查数据库,发现数据库价格已经更新了,但 Agent 返回旧的,基本锁定缓存。第二步,查这个 key 的 TTL,发现设了 1 小时,而商品价格是运营手动改的,改完没清缓存。第三步,定位到根因:写路径没有联动清缓存。数据库更新了,但没人去DEL对应的 Redis key。

修复方案是在数据更新的地方加缓存失效逻辑。但这里又有个坑:直接DEL会导致下一次请求回源,如果更新频繁,缓存反复失效等于没有。我最后用的是延迟双删——更新后立即删一次,隔 500 毫秒再删一次,覆盖掉"更新期间被旧数据回填"的窗口。

5.4 监控指标:没有度量就没有优化

最后说监控。缓存做得好不好,不能靠感觉,要看数据。我必看的几个指标:命中率(低于 60% 说明 key 设计有问题)、平均响应时间(缓存命中应该比回源快一个数量级)、内存使用率(超过 70% 要警惕)、大 key 数量(单个 key 超过 10KB 就要关注)。

这些指标我用 Redis 自带的INFO stats配合 Prometheus 采集,做成看板。有一次命中率突然从 75% 掉到 40%,看板报警,一查发现是某个上游改了 prompt 模板,导致所有 key 都变了。如果没有监控,这个问题可能要等用户投诉才发现。

6. 一套可复用的 Agent 缓存分层落地清单

聊了这么多原理和坑,最后我把整套方案收敛成一份可以直接照着搭的分层清单。这套结构我在两个项目里用过,基本能覆盖大部分 Agent 场景。

第一层:精确缓存。用sha256(完整输入)当 key,存那些输入完全确定的场景,比如固定 FAQ、系统提示词相关的固定问答。TTL 可以长一点,几小时到一天。这一层命中率不高,但实现最简单,作为兜底。

第二层:语义缓存。用向量相似度匹配,这是主力层。key 是问题的 Embedding,value 是答案,配合 RediSearch 的向量索引做 KNN 查询。阈值根据业务标定,一般 0.92-0.95。TTL 视数据时效性定,知识类可以长,实时类要短。

第三层:工具结果缓存。针对幂等的工具调用,key 是工具名 + 参数哈希,TTL 按数据变化频率定。这一层能省下大量外部 API 成本。

第四层:会话状态。严格来说不算缓存,用 Hash 结构存,滑动过期,绝不主动删。和前三层物理隔离,最好用不同的 Redis 库(SELECT不同的 db)或者不同的 key 前缀,避免误操作。

配套的治理动作:所有 key 强制 TTL、统一 JSON 序列化带版本号、写路径联动失效、监控命中率和内存、定期巡检大 key。这几条做到位,Agent 的缓存层基本就能稳定跑了。

我个人最大的体会是:Agent 缓存的核心难点从来不是 Redis 本身,而是"什么该缓存、key 怎么设计、什么时候该失效"这三个业务判断。Redis 只是个工具,真正决定成败的是你对 Agent 调用链路的理解深度。把调用链路拆清楚,知道每一步的输入输出特征,缓存方案自然就出来了。反过来,如果连 Agent 一次请求到底经过了哪些环节都没搞明白,上来就堆 Redis,那缓存只会变成新的故障源。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询