做RAG(检索增强生成)应用的人,早晚都会碰到同一个问题:同样的内容,用户换个说法反复问,你的系统就得重新检索、重新拼prompt、重新让大模型生成一遍。我最早在一个内部知识库问答项目上做统计,一天几千次请求,真正的问题种类不到两百个,剩下全是重复或近似重复,钱和时间都烧在重复劳动上。RAG缓存就是干这个的:把算过的东西存下来,下次遇到一样或者很像的问题,直接返回,不重复花时间、不重复烧token。
这篇文章适合正在用RAG做问答、客服、文档助理的开发者,也适合刚入门但想理解"缓存到底缓什么"的人。我会从链路成本拆解开始,讲清楚精确缓存、语义缓存、检索缓存三层各自的职责,给出能直接跑的代码和参数参考,最后把我踩过的坑和排查方法一起写出来。不整虚的,都是能落地的东西。
1. 先看一笔账:RAG请求的钱和延迟到底花在哪
1.1 一条完整RAG请求的三段成本
很多人觉得RAG贵在大模型生成,其实拆开来看,一条完整请求等于三段成本,每一段都有优化空间。
第一段是查询理解和向量化。用户问题要转成向量,不管用本地embedding模型还是云端接口,都有计算成本和接口调用成本。单次看起来便宜,但调用次数一旦上来,这部分就是账单上不容忽视的一项。
第二段是检索。向量数据库要做近似最近邻搜索,拿回top-k相关段落。这里花的是数据库查询时间、IO和网络开销。数据量大了以后,检索耗时可能从几十毫秒涨到几百毫秒,成为延迟的主要来源之一。
第三段是生成。把检索到的段落拼进prompt,交给大模型生成答案。这是三段里最贵的一环,按token计费,生成速度也最慢,通常在几百毫秒到几秒。
这三段里,最值得缓存的是最终生成结果,其次才是检索结果。原因很直接:生成成本最高、延迟最大,能省一次就赚一次;检索结果虽然比生成便宜,但你也不希望每天让向量库帮你重复跑同样的查询。
拿一个中等规模问答项目估算:假设每天5000次请求,其中真正不同的问题只有800个,大模型单次生成按输入2000 token、输出500 token算,一天下来就是几千次无效生成。如果缓存命中率做到40%,等于直接省掉上千次大模型调用,一个月累计下来非常可观。延迟上也明显,完整RAG链路通常1到2秒,缓存命中只要几十毫秒。
1.2 真实场景里的高频同义问题
我接触过的几个实际项目里,用户问题高度集中在少数主题上。比如某公司内部的制度问答,最常见的是这么一组问法:
"年假怎么休" "年假有几天" "我的年假天数怎么算" "年假制度具体是什么"
字面不一样,语义高度接近,答案基本一致。不做缓存的话,这四个问题会分别触发完整的"向量化+检索+生成"链路,钱和时间都花在重复劳动上。
还有一个容易忽略的特点:同一批用户会在不同时间反复问差不多的问题。早晨一拨人问,下午又有一拨人问,缓存的作用就是让后来的人直接复用前面算好的结果。高频场景里,热点问题往往只占问题总量的不到20%,却贡献了超过一半的请求量,这就是缓存收益最集中的地方。
1.3 哪些项目适合上RAG缓存,哪些不适合
先泼个冷水:如果知识库文档每个小时都在变,或者每个问题都强个性化,比如"我上个月工资为什么少了200",RAG缓存的效果会大打折扣,甚至可能让用户看到旧答案。
适合上缓存的特征有三条:知识库内容相对稳定;问题是典型的"高频同义反复";回答质量不随时间剧烈变化。常见的适合场景包括制度问答、产品使用手册问答、客服常见问题、教学答疑这类。
不适合的场景也有三条:文档频繁热更新且更新后答案必须立刻生效;答案强依赖用户身份或实时数据;问题涉及交易、权限、健康等高风险内容。在这些场景里,缓存带来的收益抵不上风险,建议谨慎使用或者干脆不做。
在这些前提下,缓存是性价比最高的优化手段之一。我在一个模拟项目X里做过统计,加了缓存之后,同样提问量下,大模型调用次数下降接近60%,平均响应时间从1.5秒降到200毫秒以内。这个数字足够说明问题。
2. 第一层缓存:先把一模一样的问法用哈希接住
2.1 精确缓存为什么是第一步
前面提到语义相似,但现实里最该优先处理的其实是完全相同的句子。同一个问题,用户可能问了一遍又一遍,或者同一入口反复提交。这种情况不需要任何相似度判断,一个哈希键就能解决。
精确缓存的好处是简单、绝对安全,命中率其实比很多人想象的高。真实环境里,用户并没有想象中那么多变,很多入口提交的是模板化文本,完全一样的query占比往往有20%到40%。
所以做RAG缓存的顺序,一定先是精确命中。它不解决"换说法"的问题,但能以最少代码拿到第一批收益,而且它是后续所有缓存层的基础判断逻辑。
2.2 查询归一化处理:决定命中率的第一细节
直接对原始字符串取哈希,是最容易踩的坑。用户的输入千奇百怪,同样一句话可能带着多余空格、全角半角混用、大小写不一致。不做归一化的话,"RAG 缓存"和"rag缓存"在系统眼里是两个不同问题。
我一般做这几步归一化:
- 统一转小写,英文部分按小写处理
- 全角转半角,中文标点统一成标准形式
- 去掉首尾空格,压缩连续空格
- 去掉无意义的语气词,比如"请问""呢""呀"这类,但要按业务谨慎处理
- 数字写法统一,比如"两个"和"2个"归一成同一形式
注意:归一化不是越狠越好。过度归一化会把"不知道"和"不知道怎么办"归成同一句话,影响命中后的答案准确性。我的习惯是先做保守归一化,观察对命中率和正确率的影响再逐步放开。
这步做好之后,精确缓存的命中率会立刻提高。我在项目里对比过,加上归一化之后,完全相同的query占比从不到20%涨到30%以上,效果非常明显。
2.3 精确缓存的生产级写法
精确缓存本身的逻辑不复杂,关键是每个缓存条目里要存什么。我的数据结构设计如下:
| 字段 | 作用 |
|---|---|
| 归一化后的query | 用来查重和人工排查 |
| 原始query示例 | 看真实用户是怎么问的 |
| 答案正文 | 缓存的核心价值 |
| 命中次数 | 统计热门问题、辅助淘汰 |
| 创建时间 | 配合TTL过期 |
| 知识库版本号 | 数据更新时整体失效 |
实现上我用的是目前主流的开源缓存存储Redis,value直接存JSON串,key用"cache:exact:v{version}:{sha256(normalized_query)}"。版本号放进key里,知识库一更新,新版本自然找不到旧缓存,旧缓存等TTL自然清理,不用手动清全部数据。
import hashlib import json import time import redis r = redis.Redis(host='127.0.0.1', port=6379, decode_responses=True) def normalize(text): # 简单归一化:小写、去空白、全角转半角 import unicodedata text = unicodedata.normalize('NFKC', text).strip().lower() return ' '.join(text.split()) def cache_key(version, norm_query): digest = hashlib.sha256(norm_query.encode()).hexdigest() return f"cache:exact:v{version}:{digest}" def exact_get(norm_query, version='1'): data = r.get(cache_key(version, norm_query)) if data: return json.loads(data)['answer'] return None def exact_set(norm_query, raw_query, answer, version='1', ttl=604800): payload = json.dumps({ 'query': norm_query, 'raw_query': raw_query, 'answer': answer, 'ts': time.time(), }, ensure_ascii=False) r.set(cache_key(version, norm_query), payload, ex=ttl)查询的顺序很重要:先查精确缓存,命中就直接返回;没命中再走语义缓存;再没命中才走完整RAG链路。一定要先精确后语义,因为精确判断是零成本、零误判的。
3. 第二层缓存:语义缓存让换说法的问法也能命中
3.1 思路:把问题向量化,用余弦相似度判断
精确缓存只能接住原封不动的句子,实际使用中,大部分重复问题是"意思一样,说法不同"。语义缓存就是为这个设计的:把历史问题向量化,新问题来了,先embedding成向量,和历史问题的向量算余弦相似度,超过阈值就把历史答案复用。
这里的关键是"缓存的是问题,返回的是答案"。我们不是简单比较两个字符串,而是比较它们在语义空间里的距离。"年假怎么休"和"年假有几天"字面差距大,但向量距离很近,所以能命中。这个思路就像一个"语义指纹",只要指纹够接近,就认为是同一个问题。
3.2 一个能直接跑的语义缓存代码
下面这段代码是入门级别的语义缓存实现,用Python写,依赖redis和numpy。embedding函数用占位,实际接本地模型或云端接口都行。
import hashlib import json import time import numpy as np import redis r = redis.Redis(host='127.0.0.1', port=6379, decode_responses=True) def get_embedding(text): # 实际使用中替换成你的embedding模型 # 返回 np.ndarray,维度固定即可 pass def cosine_score(a, b): a = a / max(np.linalg.norm(a), 1e-8) b = b / max(np.linalg.norm(b), 1e-8) return float(np.dot(a, b)) def semantic_get(query, threshold=0.90, version='1'): norm = normalize(query) q_emb = get_embedding(norm) best_key = None best_score = -1 cursor = '0' while True: cursor, keys = r.scan(cursor=cursor, match=f'cache:sem:v{version}:*', count=500) for key in keys: stored = r.hget(key, 'emb') if stored is None: continue emb = np.frombuffer(bytes.fromhex(stored), dtype=np.float32) score = cosine_score(q_emb, emb) if score > best_score: best_score = score best_key = key if cursor == '0': break if best_key is not None and best_score >= threshold: data = json.loads(r.hget(best_key, 'data')) r.hincrby(best_key, 'hits', 1) return data['answer'], best_score return None def semantic_set(query, answer, version='1', ttl=86400): norm = normalize(query) key = f'cache:sem:v{version}:' + hashlib.sha256(norm.encode()).hexdigest() emb = get_embedding(norm) record = { 'query': norm, 'raw_query': query, 'answer': answer, 'ts': time.time(), } r.hset(key, mapping={ 'data': json.dumps(record, ensure_ascii=False), 'emb': emb.astype(np.float32).tobytes().hex(), }) r.expire(key, ttl)这段代码我故意写成线性扫描所有缓存键的方式,因为它最容易看懂。等缓存量上来了,scan全量键会越来越慢,就要换成向量索引方案,后面专门讲。
3.3 相似度阈值怎么定:0.85、0.90、0.95的取舍
阈值是整个语义缓存里最敏感的旋钮。调太高,命中率低,缓存形同虚设;调太低,把本不该命中的问题当成相似问题,返回错误答案。
先给一个区间参考:
| 阈值区间 | 效果 | 适用场景 |
|---|---|---|
| 0.80以下 | 误命中风险高 | 基本不建议 |
| 0.85-0.90 | 命中率高,误命中开始显现 | 问题模式高度重复、对精确度要求不高的场景 |
| 0.90-0.95 | 命中与误命中相对平衡 | 大多数知识库问答的起点 |
| 0.95以上 | 几乎只有非常明确的同义句才命中 | 涉及交易、权限、高风险内容 |
注意这个数值不是绝对的,不同embedding模型产出的向量分布不同,同样一句话的内部相似度也不同。换新模型不能直接沿用旧阈值,必须重新做一轮评估。
比较靠谱的评估方法是:从历史日志里抽100到200组query对,人工标注"应该命中"和"不该命中",然后扫不同阈值下的精确率和召回率,画出曲线选拐点。别凭感觉拍板,这个数值得用数据定。
3.4 数据量变大后怎么升级:从全量扫描到向量索引
当缓存条目超过几百条以后,每次查询都scan全量Redis键,延迟会明显上升。这个阶段的常规方案是引入向量索引库FAISS。
做法是:内存里维护一个FAISS索引,专门存历史query的归一化向量;每次写入语义缓存时同步把向量和缓存key映射关系加进索引;查询时先走索引取top1,拿到相似度分数和对应的缓存key,再去Redis取答案。
import faiss import numpy as np dim = 768 # 按你的embedding维度改 index = faiss.IndexFlatIP(dim) id_to_key = {} # faiss内部id -> redis缓存key key_to_id = {} # redis缓存key -> faiss内部id def index_add(key, emb): # 归一化后加入索引,内积等价于余弦相似度 vec = emb.astype(np.float32).reshape(1, -1) faiss.normalize_L2(vec) idx = index.ntotal index.add(vec) id_to_key[idx] = key key_to_id[key] = idx def index_search(q_emb, topk=1): vec = q_emb.astype(np.float32).reshape(1, -1) faiss.normalize_L2(vec) scores, ids = index.search(vec, topk) return scores, ids索引本身要持久化,最省事的方式是定时把索引dump到本地文件,启动时重建。缓存条目过期删除时,也要同步从索引里移除对应的id,避免索引里留下死数据。这套结构和前面那段扫描代码的存储格式完全兼容,升级时不用改数据格式,只改查询路径。
4. 第三层缓存:把检索结果也复用好,省掉重复查询
4.1 检索缓存和响应缓存的区别在哪
响应缓存是直接把最终答案返回给用户,整个过程不走大模型,这很省钱。但有一种情况它管不了:query语义不太接近、不能走响应缓存,但检索出来的知识段落是重复的。
举个例子,用户问"年假申请流程是什么"和"年假申请需要几步",如果知识库里是同一篇制度文档,这两次检索命中的段落大概率高度重叠。虽然两个问题答案不同、都要走大模型生成,但检索结果可以复用。缓存这一层,能省掉向量数据库的重复查询和IO,延迟也能降一截。
4.2 检索缓存的实现要点
检索缓存的实现,核心是把"问题向量"映射到"检索引用的段落列表"。注意是段落列表,不是大模型答案。
和响应缓存不同的地方有两点:一是TTL要短得多,我一般用30分钟,因为文档源随时可能更新,检索结果的新鲜度要求比答案更高;二是value更大,存的是N个段落而不是一段答案,内存消耗要提前估算。
实现上复用语义缓存那套相似度判断就够了,只是value从答案换成chunks列表。
def retrieval_get(query, threshold=0.92, version='1'): norm = normalize(query) q_emb = get_embedding(norm) # 在已有的语义缓存结构里,找相近的历史检索query hit = semantic_find_similar(q_emb, threshold, version) if hit: return hit['chunks'], hit['score'] return None def retrieval_set(query, chunks, version='1', ttl=1800): norm = normalize(query) key = f'cache:retr:v{version}:' + hashlib.sha256(norm.encode()).hexdigest() emb = get_embedding(norm) record = { 'query': norm, 'chunks': chunks, 'ts': time.time(), } r.hset(key, mapping={ 'data': json.dumps(record, ensure_ascii=False), 'emb': emb.astype(np.float32).tobytes().hex(), }) r.expire(key, ttl)缓存不命中时走完整链路:检索、取段落、拼prompt、生成答案,然后把段落列表写进检索缓存。命中时直接拿段落拼prompt,大模型生成照常进行,省掉的只是检索这一段。
4.3 文档更新时怎么让缓存安全失效
检索缓存最怕旧数据。知识库文档一旦增删改,之前缓存的段落可能已经过时。我目前用的方案是"版本号+短TTL"双保险。
- 给知识库维护一个版本号,每次批量更新文档时版本号加一。所有缓存key都带当前版本号,新版本查询自然不会命中旧版本缓存。
- TTL兜底,避免版本号忘了更新时缓存长期不刷新。检索缓存TTL给30分钟,响应缓存TTL给24小时到7天,看业务容忍度。
提示:如果知识库支持单文档热更新,全局版本号会造成不必要的全体缓存失效,可以改成按文档维度做版本号,把文档ID放进缓存键里,只有被更新的文档对应缓存失效。这个方案稍复杂,但命中率会高很多。
5. 实战排坑记录:命中率低、误命中、缓存击穿
5.1 我踩过的三个典型坑
先说三个真实踩过的坑,每一个都花了不少时间。
第一个坑是换embedding模型后,旧的语义缓存全部"失忆"。换模型后向量分布完全不同,旧缓存不只是找不到,而是找出来的全是不该命中的,相似度分数整体漂移。那次我加了缓存发现命中率骤降,排查半天才发现是模型版本变了。正确做法:升级embedding模型时,直接把语义缓存和检索缓存全部清空重建,精确缓存可以保留,因为它不依赖向量。
第二个坑是意图相反的query误命中。有一阵子"删除文件A"和"恢复文件A"频繁互相命中,因为两个句子的embedding非常接近。后来我把这类操作型问题单独拎出来,不走语义缓存,或者在命中判断前先检查操作动词和对象,动作不同就不允许命中。这个问题没有通用解法,只能按业务规则卡。
第三个坑是缓存击穿。某个热门功能上线后,一批用户几乎同时问同一个问题,缓存里没有,大家全涌向大模型,瞬间把接口打满。后来我在缓存查询外面加了一个单飞锁,同一个归一化query同时只有一个请求去走RAG,其他请求等结果并共享。进程内用线程锁,跨进程用Redis锁。这个改动让高峰期的大模型调用次数明显下降。
5.2 常见问题速查表
| 现象 | 可能原因 | 处理方式 |
|---|---|---|
| 命中率很低 | 阈值太高 | 抽样标注后下调阈值 |
| 命中率很高但答非所问 | 阈值太低,误命中 | 上调阈值,对敏感query禁用语义缓存 |
| 缓存一直不命中 | 缓存key版本号对不上 | 检查知识库版本号和写入时版本号 |
| 缓存内存涨得快 | value太大或TTL太长 | 拆小value、缩短TTL、加淘汰机制 |
| 换模型后分数全变 | embedding模型不一致 | 清空语义缓存和检索缓存,重新预热 |
| 相同问题用户看到不同答案 | 精确缓存和语义缓存并存但答案不一致 | 统一写路径,保证同一问题只有一条权威缓存 |
5.3 缓存监控指标怎么埋
缓存没有监控等于没做。我至少会埋这几个指标:
- 精确缓存命中次数、语义缓存命中次数、检索缓存命中次数、完全miss次数
- 当前缓存条目数和占用内存
- 每个热门query的命中次数,用来发现哪些问题最值得预热
实现上不复杂,Redis的INCR或者计数器字段就能搞定,再拉个简单的趋势图。重点不是看单日数字,而是看命中率随知识库更新的变化曲线。如果每次更新文档后命中率掉一大截,然后慢慢回升,说明版本号失效策略生效了,属于正常现象。
6. 缓存做完之后,还有哪些值得尝试的方向
6.1 三个可选的进阶方向
缓存做到上面三层,已经能覆盖大部分收益。如果想更进一步,有三个方向值得考虑。
一是用LLM做query改写之后再缓存。先让一个小模型把用户问题改写成标准问法,再基于改写后的文本做精确缓存,很多语义缓存的需求被"降维"成精确缓存,既提高命中稳定性,又减少向量误判。代价是多一次模型调用,可以用轻量模型控制成本。
二是缓存部分生成结果。同一个上下文前缀配不同的用户问题,后半段生成不同,这种场景可以做前缀缓存,把prompt公共部分缓存住,只生成差异部分。这个对实现要求更高,但成本优化空间更大。
三是给缓存答案加置信度。语义命中时把相似度分数一起返回给前端,分数低于安全线就把答案标注成"仅供参考",高于安全线才直接展示。这个思路很适合客服和审核场景。
6.2 我的最终建议:什么时候该停下来
后面这几个方向没有统一套路,跟业务绑定很深。我的建议是先把前面三层做扎实,拿到真实数据后再看哪一步值得投入。缓存这东西,做到收益边际递减就该停手,不要为了技术完美把系统复杂度堆上去,后面维护成本都是实打实的。
最后分享一点个人体会。做RAG缓存,最容易犯的错是一上来就想做完美的语义缓存,结果项目卡在阈值调试和向量索引上迟迟上不了线。我回头复盘,最划算的路径其实是反过来的:先把精确缓存跑起来,看真实日志了解重复度,再上语义缓存,最后才考虑检索缓存和更高级的优化。技术方案越往后越复杂,但收益未必线性增长。
另外,缓存答案的"新鲜度"问题,永远比性能问题更难处理。我现在的原则是:宁可少缓存,也不要因为缓存给出过时答案。像合同条款、政策变动、价格信息这些内容,我宁可直接关闭缓存,也不愿冒让用户看到旧制度的风险。这个底线想清楚了,后面所有参数设计和失效策略都不会跑偏。