RAG 语义缓存设计：用 Redis 与向量相似度避免重复 LLM 调用-平芜编程栈

RAG 语义缓存设计：用 Redis 与向量相似度避免重复 LLM 调用

在做基于 RAG（检索增强生成）的大模型应用时，调用 LLM 的 API 费用和首字延迟（TTFT）是两个绕不开的麻烦。实际在企业知识库或者客服问答里看日志会发现，很多用户的提问在意思上是高度重复的（重复查询占比常在 30%~50% 左右）。

如果用户每次问个类似的问题，都去跑一遍“文本向量化 ➔ 检索向量库 ➔ 拼接 Prompt ➔ 调 LLM 推理”的完整流程，不仅浪费 Token 费用，用户也要白白等上好几秒。

用语义缓存（Semantic Cache）能解决这个问题。跟传统按字符串精确匹配的 Redis 缓存不同，语义缓存是用 Vector Embedding 和余弦相似度算法，找到那些意思一样但说话方式不一样的提问（比如“怎么申请报销”和“报销流程是啥”），直接把之前生成好的答案返回给用户。

但是在写语义缓存的时候，不能简单地“算个向量距离就直接给旧答案”。如果没有做好租户隔离、权限校验和知识库版本更新，语义缓存很容易把敏感数据吐给没有权限的用户。

语义缓存核心架构：“向量相似不等于业务可复用”

传统 KV 缓存看的是Key == Query，逻辑是确定的。而语义缓存算的是向量空间距离，属于概率型近似匹配。

这种近似匹配虽然提高了命中率，但也带了一个风险：在业务层面，向量相似不代表答案就能直接拿来用。
比如：

租户 A 的员工问“公司报销额度多少？”，租户 B 的员工也问“公司报销额度多少？”，两个人的 Query 向量一模一样，但答案涉及各自公司的规定，绝对不能跨租户混用。
用户问“现在还有多少库存？”，虽然和 10 分钟前的提问语义相同，但库存是实时变动的，给缓存的答案就会出业务事故。

所以，RAG 的语义缓存必须做成“硬性条件过滤 ➔ 向量相似度匹配 ➔ 业务校验”的三层流水线：

flowchart TD UserQuery[用户输入 Query] --> Embed[生成 Query 向量 Vector_Q] Embed --> HardFilter{第一层: 硬性条件过滤<br/>租户ID / 知识库版本 / 提示词版本} HardFilter -->|过滤后的向量空间| VectorSearch[第二层: Redis 向量 KNN 检索] VectorSearch --> SimCheck{相似度 Sim >= Threshold?} SimCheck -->|否: Cache Miss| MissPath[走完整 RAG 流程: 查库 + 调 LLM] MissPath --> PutCache[写入 Redis 语义缓存] PutCache --> ReturnNew[返回新生成的答案] SimCheck -->|是: 命中候选| BizVerify{第三层: 业务数据与权限校验} BizVerify -->|校验通过: Cache Hit| ReturnCached[返回缓存答案] BizVerify -->|校验失败: 判定过期| Invalidate[删除失效 Key ➔ 转 Miss 流程] Invalidate --> MissPath

租户隔离、权限过滤与知识库版本控制设计

为了保证语义缓存不会导致越权泄露，需要在构造 Redis 缓存 Key 和向量查询范围时带上这几个维度：

1. 租户与权限隔离

不能把所有用户的向量都塞在同一个没有隔离的索引里。用 RediSearch 建向量索引的时候，要把tenant_id和role_scope设成TAG 属性。
在做 KNN 向量检索时，用带预过滤的查询：
(@tenant_id:{tenant_A} @role_scope:{admin})=>[KNN 1 @vector $BLOB AS score]
让向量距离计算只在当前租户授权的子空间里进行。

2. 知识库版本控制与失效

当管理员在后台更新或删除了某份 PDF 文档，之前基于这份文档生成的缓存就要废弃。

简易方案（版本号递增）：在 Redis 里给每个知识库记一个递增的版本号kb_version，缓存项里也保存生成时的版本号。命中缓存时比对cached.kb_version == current_kb_version；如果版本不一致，当成 Cache Miss，顺手把旧缓存删掉。
精细方案（反向索引）：写入缓存时，记录answer_id ➔ [doc_id_1, doc_id_2]的映射。当doc_id_1被更新时，查反向索引精准删除关联的缓存 Key。

生产级 Python / Redis 向量语义缓存实现

下面是一段用redis-py写的语义缓存组件，包含了向量检索、租户硬过滤以及相似度阈值控制：

import time import json import logging import numpy as np import redis from redis.commands.search.query import Query from typing import Optional, Dict, Any, List logger = logging.getLogger("semantic_cache") class SemanticCache: def __init__(self, redis_url: str, vector_dim: int = 1536, threshold: float = 0.92): self.redis_client = redis.from_url(redis_url) self.vector_dim = vector_dim self.threshold = threshold self.index_name = "idx:semantic_cache" self._ensure_index() def _ensure_index(self): """初始化 RediSearch 向量索引""" try: self.redis_client.ft(self.index_name).info() except redis.exceptions.ResponseError: from redis.commands.search.field import TagField, VectorField schema = ( TagField("tenant_id"), TagField("kb_version"), VectorField( "vector", "HNSW", { "TYPE": "FLOAT32", "DIM": self.vector_dim, "DISTANCE_METRIC": "COSINE", } ) ) self.redis_client.ft(self.index_name).create_index(schema) logger.info("Created RediSearch vector index: %s", self.index_name) def get(self, tenant_id: str, kb_version: str, query_vector: List[float]) -> Optional[Dict[str, Any]]: """带租户和版本过滤的语义缓存查询""" query_bytes = np.array(query_vector, dtype=np.float32).tobytes() filter_str = f"(@tenant_id:{{{tenant_id}}} @kb_version:{{{kb_version}}})" q_str = f"{filter_str}=>[KNN 1 @vector $vec AS score]" q = ( Query(q_str) .sort_by("score") .paging(0, 1) .return_fields("score", "payload") .dialect(2) ) try: res = self.redis_client.ft(self.index_name).search(q, query_params={"vec": query_bytes}) if not res.docs: return None top_doc = res.docs[0] # 余弦距离 (Cosine Distance): score 越小越相似 cosine_dist = float(top_doc.score) similarity = 1.0 - cosine_dist logger.info("Semantic search top hit similarity: %.4f (threshold: %.4f)", similarity, self.threshold) if similarity >= self.threshold: payload = json.loads(top_doc.payload) payload["hit_similarity"] = similarity return payload except Exception as e: logger.error("Failed to query semantic cache: %s", e) return None def put(self, tenant_id: str, kb_version: str, query_text: str, query_vector: List[float], answer: str, docs_cited: List[str], ttl_seconds: int = 86400): """存入语义缓存，绑定元数据与 TTL""" doc_id = f"cache:{tenant_id}:{hash(query_text)}" query_bytes = np.array(query_vector, dtype=np.float32).tobytes() payload = { "query": query_text, "answer": answer, "docs_cited": docs_cited, "created_at": time.time() } mapping = { "tenant_id": tenant_id, "kb_version": kb_version, "vector": query_bytes, "payload": json.dumps(payload, ensure_ascii=False) } pipeline = self.redis_client.pipeline() pipeline.hset(doc_id, mapping=mapping) pipeline.expire(doc_id, ttl_seconds) pipeline.execute() logger.info("Successfully cached semantic answer for doc_id: %s", doc_id)

缓存调优与坑点

实际用语义缓存时，需要在命中率和准确性之间取舍：

1. 相似度阈值（Threshold）调优

阈值设得太高（比如 >= 0.98）：只有两句话几乎一模一样才能命中，语义缓存就退化成了普通字符串缓存，省不了多少费用。
阈值设得太低（比如 <= 0.85）：很容易误命中，把关于“退货政策”的答案错给询问“换货政策”的用户。
建议：刚上线时先把阈值设在0.92~0.95这种比较稳妥的位置，把落在[0.85, 0.95]区间的命中结果打日志存下来，人工抽检看看，再根据实际业务微调。

2. 区分静态知识和动态数据

不是所有 RAG 场景都适合加语义缓存：

graph LR Sub1[适合语义缓存的场景] --> A[产品 FAQ 与操作手册] Sub1 --> B[公司 HR 制度与政策问答] Sub1 --> C[固定技术文档与 API 解答] Sub2[不适合语义缓存的场景] --> D[实时价格、库存与优惠券计算] Sub2 --> E[个性化推荐与个人隐私数据] Sub2 --> F[强时效性新闻与大盘行情]

3. 监控指标

语义缓存接入后，监控要看这三个指标：

semantic_cache_hit_ratio：缓存命中率（按租户统计）；
semantic_cache_saved_tokens：累计省下来的输入和输出 Token 数量；
semantic_cache_latency_p99：缓存查询的耗时（一般在 5~10ms，远快于 LLM 的上秒等待）。

总结

做 RAG 系统，语义缓存是降低延迟和 Token 费用很管用的手段。

不过，语义缓存不能只看向量匹配度。先做租户隔离、权限过滤和版本校验，在这个前提下再去算向量距离，才能在省钱省时间的同时，不把数据搞错。

参考资料

Redis Vector Search
OWASP Top 10 for LLM Applications
OpenAI Prompt Caching Guide

RAG 语义缓存设计：用 Redis 与向量相似度避免重复 LLM 调用