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, dtypenp.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 fcache:{tenant_id}:{hash(query_text)} query_bytes np.array(query_vector, dtypenp.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_asciiFalse) } pipeline self.redis_client.pipeline() pipeline.hset(doc_id, mappingmapping) 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 SearchOWASP Top 10 for LLM ApplicationsOpenAI Prompt Caching Guide