做RAG应用初期,有件事特别容易被忽略——缓存。我印象很深的是帮某团队看一个知识库问答系统,功能早就跑通了,文档切好了,向量也建好了,但日活一上来,大家发现系统又慢又费钱:同一个问题,换一种问法,整个检索和生成链路就全部重来一遍。这就是我写下“小白的RAG缓存”这篇文章的由来。RAG缓存不是那种看起来很酷的技术,但它可能是你从demo走向生产时,投入产出比最高的一步。
这篇文章适合谁?如果你正在搭自己的RAG项目,已经能完成“提问—召回—生成”这条主链路,但对性能、成本还没有系统概念,那这篇文章就是给你写的。我会从缓存的底层动机讲起,把精确缓存、语义缓存这些容易混淆的概念拆开,再给一份能直接接到现有RAG流程里的实现示例,最后聊一聊我实际踩过的坑。看完之后,你应该有能力判断:自己的RAG项目要不要上缓存、上哪种缓存、以及参数该怎么调。
1. RAG缓存到底在解决什么问题
1.1 先复习一下RAG这条主链路
RAG的常规流程,大多数人都有印象:先加载文档、把文档切块,然后对每块文本做向量化,存进向量库;用户提问时,同样把问题向量化,去向量库中找到最相似的文档片段;把召回到的片段放进提示词中,交给大模型生成最终回答。整个链路看起来不复杂,但从工程角度拆开看,每一步都有成本。向量化要调用嵌入模型服务,检索要扫描向量库,生成更是要调用大模型,而大模型生成这一步往往是最慢、最贵的。
小白阶段写demo,输入一个问题,等两三秒出答案,大家觉得无所谓。一旦系统面向真实用户,同样的问答一天被问几百次,问题就暴露了:没有缓存的情况下,每一次提问都是从头开始。哪怕用户一个字都没改,系统也不会记得上一次算过什么。RAG系统的无状态特性,让重复计算变得非常普遍。
1.2 三个真实的痛点:慢、贵、重复
第一个是延迟。假设你的文档有几千份甚至更多,向量库的检索可能就要几十毫秒到几百毫秒,紧随其后的大模型生成又要好几秒。现实场景里,用户等不了那么久。尤其很多问题本来就是热门问题,答案早就被算出来过,系统却不记得。
第二个是成本。现在大家做大模型应用,token都是按量计费的。同一句话问一百次,意味着这一百次都在为大模型的回复埋单。缓存的意义在于:第一次生成的答案已经花过钱了,第二次、第三次命中缓存时,你只需要返回上次的结果,token费用直接归零。
第三个是体验上的重复感。用户遇到同一个问题,每次打开对话框都要重新等待,这种体验很消耗耐心。这里有个生活化的类比:缓存相当于平时做好的半成品菜,客人点餐时热一下就能上桌,而不是每次从洗菜、切菜开始。RAG的检索相当于洗菜切菜,大模型生成相当于炒菜,缓存就是那个保温柜。
1.3 加缓存到底能带来什么收益
用一张表整理一下常见的收益指标,做方案汇报时可以拿来直接用:
| 收益维度 | 说明 |
|---|---|
| 响应延迟 | 命中缓存后,省去向量检索、重排序、大模型生成等多个环节,延迟可能从数秒降到毫秒级 |
| token成本 | 命中后不再调用大模型生成接口,相关费用直接为0 |
| 系统吞吐量 | 同样的算力可以支撑更多的并发请求,因为大量请求不需要进入完整链路 |
| 服务稳定性 | 核心模型服务的压力下降,高峰期被打爆的概率明显降低 |
注意,RAG缓存并不是没有代价。缓存需要存储空间,语义缓存还需要额外的向量化计算和相似度比较。但总体来看,对于一个以问答为核心的RAG应用,缓存带来的收益远大于成本。尤其是当你想进一步压测系统,或者面对成本预算压力时,缓存几乎是第一优先级优化项。
2. RAG缓存的几种形态,别傻傻分不清
2.1 精确缓存:最简单但远不够用
精确缓存很好理解:把用户问题标准化后作为键,把答案作为值,存进一张哈希表。完全一样的问题来了,直接哈希命中,返回答案。“完全一样”这个条件很苛刻,真实用户几乎不会打一字不差的问题。所以精确缓存一般不是主角,而是作为多级缓存的第一层,先把那些重复提交、同字同序的问题挡下来。
标准化这一步值得多花点心思。至少要做大小写统一、多余空格清理、全角半角统一。我见过不少团队连标点符号都没处理,同一个问题因为末尾多了一个句号就缓存不中,非常可惜。标准化的目标不是消除所有差异,而是为完全一致的文本建立统一的键。
2.2 语义缓存:RAG场景最常用的一层
语义缓存解决的核心问题,是“问题字面不同但意思相近”。做法是把用户问题变成向量,与缓存中已有的问题向量逐一计算相似度,超过一定阈值就认为是同一个问题,可以直接复用缓存答案。
这里的关键参数就是相似度阈值。阈值设高了,很多明明很像的问题也命中不了,缓存命中率低;阈值设低了,明显不同的问题却被当成同一问题,返回牛头不对马嘴的答案。阈值怎么调,我后面会专门展开讲,这里先记住一个原则:不要照抄别人的经验值,必须结合你自己的嵌入模型和业务数据实际校准。
另外,语义缓存每次都要做一次向量化,这个成本通常比一次完整RAG链路低得多,属于可以接受的代价。
2.3 检索结果缓存:把召回阶段的结果也缓存起来
还有一种经常被误会的缓存,是只缓存检索阶段的产出,不缓存大模型生成的最终答案。比如知识库预览、文档引用列表、二次处理后需要用户确认的中间结果,这些场景并不需要大模型生成,只要拿到检索结果就够了。
检索结果缓存对流式输出也有帮助。用户先看到文档引用的来源,再等着答案流式返回,体验会更稳。不过对大多数小白项目来说,这层缓存属于锦上添花,前期可以不做。
2.4 提示词与中间产物缓存
除了问答层,RAG系统内部其实还有一层容易被忽略的缓存:相同的系统提示词、固定指令、长文档片段如果反复出现在提示词中,成熟的模型推理框架会在底层复用键值状态。这一层对小白来说不需要自己实现,了解它的存在就好,避免后面遇到类似话题时把不同层次的缓存搞混。
2.5 用一张表搞定选型
| 缓存类型 | 命中条件 | 主要收益 | 实现复杂度 |
|---|---|---|---|
| 精确缓存 | 问题标准化后完全一致 | 省去整条链路 | 很低 |
| 语义缓存 | 问题向量相似度超阈值 | 省去整条链路 | 中等 |
| 检索结果缓存 | 检索条件一致 | 省去检索和重排序 | 中等 |
| 提示词缓存 | 提示词片段重复 | 省去大模型重复处理前缀 | 依赖推理框架 |
给小白一个实用建议:起步阶段,先做精确缓存加语义缓存两层,效果已经很明显。等系统复杂度上来,再根据监控数据决定要不要把检索结果缓存加进来。
3. 手把手实现一个可用的RAG语义缓存
3.1 先想清楚缓存什么、怎么判断命中
实现之前先把粒度定下来。我推荐以“问题—答案”为单位做语义缓存,这是最容易理解也最好接入的方式。命中时,直接把上一次生成的回答原文返回,附带当时引用的文档来源,保持可追溯。
判断命中则分两步:先做精确匹配,再做语义匹配。精确匹配可以用标准化后的文本作为键,语义匹配用向量相似度判断。语义命中后,返回的答案还需要让用户知道“这是基于哪几篇文档生成的”,这点在缓存方案里经常被漏掉,但影响信任感。
3.2 缓存条目应该存哪些字段
一个缓存条目不能只存问题和答案,否则后面很难做清理和审计。我建议至少包含这些字段:问题原文、标准化键、问题向量、答案文本、来源文档引用、创建时间、过期时间、命中次数。
命中次数很有用,淘汰缓存时优先淘汰那些长期无人问津的条目,而不是随机淘汰。来源文档引用也很关键,RAG本来就是靠可追溯性建立可信度的,缓存不能把信任的基础破坏掉。
TTL的选择要匹配业务特性。知识库如果每周更新,TTL设为24小时到7天都合理;如果文档会频繁变化,TTL就要相应缩短。这个时间一开始可以拍一个值,后面根据命中率数据再调。
3.3 相似度阈值到底怎么定
这可能是新手最容易蹦出来的问题。我也经历了随手设一个0.85,看起来还挺像那么回事。后来在某一次测试中发现,一个项目的问题问法相似度极高的都在0.90以上,不同问题大多在0.70左右,我设的0.85明显偏低,缓存会错误命中。
正确的做法是做一组校准样本。找二十组“应该视为同一问题”的相似问法,再找二十组“必须区分”的不同问题,分别算相似度,画一张分数分布图,取两类分布之间的分界值,再留10%左右的安全余量。
不同嵌入模型产出的向量,相似度分布会有明显差异,别人博客里的0.9你直接搬过去大概率不准。这是需要亲自踩一遍的坑。
3.4 一个可运行的简化实现
下面的参考实现里,我把具体嵌入服务的调用隔离成了embed_text_with_your_service,你需要替换成自己实际使用的嵌入模型。缓存存储先用内存列表遍历,条目超过几百上千后,建议换成向量索引,否则语义查询就是O(n)。
import math import time class RAGSemanticCache: def __init__(self, max_items=1000, ttl_seconds=86400): self.store = {} self.max_items = max_items self.ttl_seconds = ttl_seconds @staticmethod def _normalize(text: str) -> str: # 统一小写、清理连续空格,为精确缓存生成稳定键 return " ".join(text.strip().lower().split()) def _embed(self, text: str) -> list[float]: # 将文本转换为向量,替换成你实际使用的嵌入服务。 # 注意:同一个系统必须使用同一个嵌入模型,否则向量空间不一致,相似度没有意义。 return embed_text_with_your_service(text) @staticmethod def _cosine_similarity(vec_a: list[float], vec_b: list[float]) -> float: dot = 0.0 norm_a = 0.0 norm_b = 0.0 for x, y in zip(vec_a, vec_b): dot += x * y norm_a += x * x norm_b += y * y if norm_a == 0.0 or norm_b == 0.0: return 0.0 return dot / (math.sqrt(norm_a) * math.sqrt(norm_b)) def get(self, question: str, threshold: float = 0.90): normalized = self._normalize(question) # 第一层:精确命中 if normalized in self.store: item = self.store[normalized] if self._is_expired(item): del self.store[normalized] else: item["hit_count"] += 1 return item["answer"], 1.0, item["source_refs"] # 第二层:语义命中 question_vec = self._embed(question) best_score = 0.0 best_item = None for item in self.store.values(): if self._is_expired(item): continue score = self._cosine_similarity(question_vec, item["question_vec"]) if score >= threshold and score > best_score: best_score = score best_item = item if best_item is not None: best_item["hit_count"] += 1 return best_item["answer"], best_score, best_item["source_refs"] return None, 0.0, None def set(self, question: str, answer: str, source_refs: list[str]): if len(self.store) >= self.max_items: self._evict() normalized = self._normalize(question) now = time.time() self.store[normalized] = { "question": question, "question_vec": self._embed(question), "answer": answer, "source_refs": source_refs, "created_at": now, "expires_at": now + self.ttl_seconds, "hit_count": 1, } def _is_expired(self, item) -> bool: return time.time() > item["expires_at"] def _evict(self): # 先清掉过期的条目,再按命中次数淘汰长期无人问津的条目 now = time.time() expired_keys = [k for k, v in self.store.items() if v["expires_at"] < now] for k in expired_keys: del self.store[k] while len(self.store) >= self.max_items: least_key = min(self.store.keys(), key=lambda k: self.store[k]["hit_count"]) del self.store[least_key]代码里有两个地方需要你自己替换:文本嵌入服务,以及向量存储。如果你现在连向量库都还没上,主流程里用的是普通的内存检索,那语义缓存里的全表遍历还能用;条目一多,就必须引入向量索引,不要硬扛。
3.5 把缓存接进你的RAG主流程
假设你的主流程长这样:retrieve_documents去检索文档,generate_answer去调用大模型。接上缓存之后,逻辑就变成下面这样:
def get_answer(question: str, threshold: float = 0.90): cached_answer, score, source_refs = cache.get(question, threshold) if cached_answer is not None: log_hit(question, score) return cached_answer, source_refs source_refs = retrieve_documents(question) answer = generate_answer(question, source_refs) cache.set(question, answer, source_refs) log_miss(question) return answer, source_refs有两点容易被忽略。一是日志,没有命中率日志,你后面根本不知道阈值调得对不对。二是来源引用,缓存命中时也必须把来源文档返回给用户,否则用户在缓存帮助下虽然秒答,却无法确认答案来源,时间长了就成了不可信的答案。
4. 避坑手册:我在RAG缓存上踩过的几个坑
4.1 阈值拍脑袋,命中率忽高忽低
我已经说过别照搬阈值,这里再讲一个我实际遇到过的案例。某模拟项目一开始用0.85,结果语义缓存大量误命中,用户问“报销流程是什么”,系统把“请假流程是什么”的缓存答案返回来了,因为两者相似度恰好超过0.85。后来把阈值提高到0.92,错误命中立刻消失,但命中率从70%掉到40%。
再后来把嵌入模型换成了更适合业务语料的模型,相似度分布整体拉开,0.90成了更合适的点。这说明阈值必须结合模型和业务重新标定,没有一劳永逸的数字。
4.2 缓存了错误答案,后果被放大
RAG系统生成答案是有可能跑偏的。如果你不判断生成质量,不管三七二十一都写进缓存,一个错误的回答就会被后面几十个相似问题共同复用,错误被放大了。
我现在的做法是:加一个反馈位。推荐类问答可以默认缓存,但事实核查类问题要等用户点踩或点赞后再决定是否缓存。更简单的方式是,先将置信度不高的答案标记为“低质量,顺手缓存但不长期生效”,或者直接将低质量答案排除在缓存之外,宁可多花一次token,也不能把坏答案扩散出去。
4.3 多用户场景下注意缓存数据隔离
如果你的系统有登录体系,比如不同团队各自上传文档,那缓存键里一定要带上空间ID,否则A用户问出的答案,可能包含他私有文档的引用,却被B用户的相似问题命中。信息泄露这种事在真实项目里不是技术问题,是事故。
为避免事故,缓存写入时把空间ID拼进标准化键,读取时同一个空间ID才允许命中。如果涉及更细的权限边界,还要在查询前校验当前用户对该缓存条目来源文档的访问权限。
4.4 缓存穿透、击穿、雪崩的RAG版
这三个词在Web开发里很常见,在RAG场景下同样存在:
| 问题 | 现象 | 常见解法 |
|---|---|---|
| 缓存穿透 | 大量新问题都没有缓存,每次都走完整链路 | 语义缓存能聚合相似问题;同步限制涌入口 |
| 缓存击穿 | 某个热门问题缓存过期,瞬间大量请求同时miss | 加锁,让同一个缓存键同一时间只有一个线程去重建 |
| 缓存雪崩 | 大量缓存条目在同一时间过期 | TTL加入随机偏移,避免齐刷刷失效 |
RAG场景下的击穿更危险,因为热门问题的重建不仅要做检索,还要调用大模型生成,成本高、耗时长。加锁是必要的,否则流量一起来,模型服务直接被冲垮。
4.5 有些场景就是不该做缓存
不是所有RAG都适合缓存。如果答案高度依赖实时状态,比如股票报价、实时环境监控、日程查询,缓存就是帮倒忙。如果用户的输入包含私有数据和动态token,缓存键也难以稳定。
还有合规要求严格的场景,需要对每次回答完整记录生成链路,缓存命中可能让审计记录缺失。这些情况不是绝对不能用缓存,而是要额外设计审计字段,把命中时间和命中原因也记下来。
4.6 常见问题速查
| 症状 | 可能原因 | 解决思路 |
|---|---|---|
| 缓存命中率极低 | 阈值太高,语义匹配几乎不触发 | 用校准样本重新标定阈值 |
| 答案牛头不对马嘴 | 阈值太低,不同问题被误判为同一问题 | 调高阈值,观察误命中日志 |
| 缓存占用越来越大 | 没有淘汰策略或TTL太长 | 增加容量上限、缩短TTL |
| 同一个问题重复生成 | 标准化没做好,精确缓存没启用 | 先处理文本归一化,再启用语义层 |
| 缓存答案来源不清 | 没存source_refs字段 | 重新设计缓存条目结构,补上文档引用 |
5. 别只顾着写码:这几招帮你把缓存效果量化出来
5.1 三个核心指标必须盯住
接入缓存后,至少要看三个指标:命中率、平均响应时间、token费用。命中率代表缓存的有效性,平均响应时间是用户体感的核心,token费用则直接关系到钱。
日志字段建议包含:问题摘要、是否命中、命中类型(精确还是语义)、相似度得分、响应耗时、来源文档数量、当时缓存条目总数。有了这些,你才能回答“缓存到底有没有用”这个灵魂问题。
5.2 一次有代表性的基准对比
我整理一个虚拟但很典型的场景:某知识库问答系统,文档约5000份,日均查询约3万次,之前没有缓存,平均响应在3到5秒之间。加装语义缓存后,缓存命中率稳定在60%以上,命中请求的耗时降到1秒以内,token费用大约下降了五成到七成。
这个数字不是绝对值,不同项目差异会很大。但有一点是共通的:先从日志里看看你的问题分布,如果存在一批高频相似问题,缓存收益一定明显;如果每个问题都是独立的、一次性的,那缓存收益会非常有限。
5.3 命中率低不代表缓存没用
很多人一看到命中率只有20%,就觉得缓存方案失败了。实际上,缓存省下的不只是命中那部分,还帮你摸清了用户的提问规律。把缓存日志里的“近似命中”问题收集下来,定期做人工归并,你会发现很多用户其实在反复问同一件事。这些数据反过来又能用来优化知识库本身的文档结构,属于意外收获。
5.4 调优顺序建议
先精确缓存,再语义缓存;先单机内存,再换分布式存储层;先默认阈值,再用校准样本修正;先观察日志,再逐步调整TTL和淘汰策略。每改一个参数,都留出观察窗口,不要同时动两三个变量,否则出了问题你根本不知道罪魁祸首是谁。
6. 最后再分享一点个人体会
做RAG缓存这件事,技术难度并不高,难的是想清楚“要不要、为什么、做到什么程度”。我在好几个项目上见过类似的情况:团队花大力气优化了检索算法、换了更大的模型,却忽略了缓存这种朴素的手段。实际上,一次配置合理的语义缓存,往往比模型层面的小升级带来更直接的体感提升。
如果你现在正准备给自己的RAG项目加缓存,我的建议是从很小的规模开始:先在一个接口后面包一层,跑一周日志,看看问题分布和命中率,再决定要不要全面铺开。这样既能快速见效,又不会把系统改得太复杂。
最后一个小技巧:把缓存日志里那些“差一点就命中”的问题定期导出,人工看看用户到底在怎么问。很多时候,你会从中发现知识库缺失的文档、容易混淆的术语,甚至新的产品需求。缓存不只是省钱的工具,它还是让你更懂用户的一扇窗。