Redis 语义缓存初探用向量距离命中历史相似 Query在大模型问答和企业知识库系统的生产运营账本里Token 费用与首字延迟TTFT永远是两项最沉重的负担。在真实的客服工单和业务咨询场景中用户的提问具有极强的语义聚集性用户 A 问“如何开具增值税电子发票”用户 B 问“请问电子增值税发票怎么开”用户 C 问“开电子专票的具体流程是什么”这三个问题在字面字符上完全不同传统的精确 Key-Value 缓存如MD5(query)完全无法命中。结果是系统不得不对每一个提问都重新执行一次全套的“向量检索 重排 大模型推理”白白消耗了数万次昂贵的 API 调用和数秒的推理时间。利用Redis 8 / RediSearch 向量检索实现的语义缓存Semantic Cache通过计算 Query 之间的高维余弦相似度能够直接在毫秒级命中历史相似提问的高质量答案实现成本与延迟的断崖式下降。语义缓存的两段式工作流语义缓存的核心架构可以分为前置检索判定与异步结果回填两个阶段[ 终端用户 Query ] | v 生成 Query Embedding (5ms) ---------------- Redis 语义缓存层 ---------------- | 1. FT.SEARCH 近邻探查历史缓存 Query 向量 | | 2. 计算余弦相似度 Score: | | - 若 Score 0.92 (极高置信度命中): | | -- 直接返回历史 Cache Answer (响应耗时 8ms) | | - 若 Score 0.92 (未命中): | | -- 穿透至后端 RAG 链路与大模型推理 | --------------------------------------------------- | (未命中回源) v [ 执行 RAG 检索 LLM 生成完整回答 (耗时 1500ms) ] | v 异步写入 Redis [ 存入新的 Query 向量与 Answer 到语义缓存库 (TTL 7天) ]生产级 Python 语义缓存实现以下是一套经过高并发压测检验的轻量级 Redis 语义缓存核心实现import redis from redis.commands.search.field import VectorField, TextField, NumericField from redis.commands.search.indexDefinition import IndexDefinition, IndexType from redis.commands.search.query import Query import numpy as np import time from typing import Optional, Dict, Any class RedisSemanticCache: def __init__( self, redis_client: redis.Redis, embedding_dim: int 768, similarity_threshold: float 0.90, ttl_seconds: int 86400 * 7 # 缓存 7 天 ): self.r redis_client self.dim embedding_dim self.threshold similarity_threshold self.ttl ttl_seconds self.index_name idx:semantic_cache self.prefix cache:query: self._ensure_index() def _ensure_index(self): 确保 Redis 向量索引已创建 try: self.r.ft(self.index_name).info() except Exception: schema ( TextField(query_text), TextField(answer_text), NumericField(created_at), VectorField( query_vector, HNSW, { TYPE: FLOAT32, DIM: self.dim, DISTANCE_METRIC: COSINE, M: 16, EF_CONSTRUCTION: 200 } ) ) self.r.ft(self.index_name).create_index( schema, definitionIndexDefinition(prefix[self.prefix], index_typeIndexType.HASH) ) def get_similar_answer(self, query_vector: np.ndarray) - Optional[Dict[str, Any]]: 探查是否存在相似度高于阈值的历史答案 raw_bytes query_vector.astype(np.float32).tobytes() # 查找 Top-1 最近邻 q ( Query(*[KNN 1 query_vector $vec AS score]) .sort_by(score) .return_fields(query_text, answer_text, score, created_at) .dialect(2) ) results self.r.ft(self.index_name).search(q, query_params{vec: raw_bytes}) if results.docs: top_doc results.docs[0] # RediSearch COSINE 返回的是余弦距离 (Distance 1 - Cosine_Similarity) cosine_distance float(top_doc.score) similarity 1.0 - cosine_distance if similarity self.threshold: return { matched_query: top_doc.query_text, cached_answer: top_doc.answer_text, similarity: similarity, hit: True } return None def set_cache(self, query_text: str, query_vector: np.ndarray, answer_text: str): 异步写入新的问答缓存 doc_id f{self.prefix}{int(time.time() * 1000)} raw_bytes query_vector.astype(np.float32).tobytes() mapping { query_text: query_text, answer_text: answer_text, created_at: int(time.time()), query_vector: raw_bytes } # 写入 Hash 并设置过期时间 self.r.hset(doc_id, mappingmapping) self.r.expire(doc_id, self.ttl)生产环境的三大核心避坑点阈值设定的“语义漂移”陷阱阈值设得太低如 0.80会导致“如何注销账号”误命中“如何修改账号密码”发生严重的答非所问阈值设得太高如 0.98会导致稍微换个同义词就无法命中缓存利用率形同虚设黄金区间对于 768 维高质量 Embedding 模型如 BGE/Text-Embedding-30.90 ~ 0.92是误判率低于 0.5% 且命中率最优的黄金甜点位。时效性与数据一致性对于产品价格、库存、动态政策等敏感问答不能无脑长期缓存。必须结合业务标签TagField存储版本号一旦后端知识库发生更新通过发布订阅Pub/Sub或扫描 Tag 批量清除相关旧缓存。缓存冷启动与击穿在大促或新功能上线前可以从历史客服日志中提取 Top-500 高频问答离线运行大模型生成标准答案并预热加载进 Redis 语义缓存库直接为线上拦截 60% 以上的突发流量。总结引入 Redis 语义缓存后企业大模型服务的端到端 P99 响应延迟可从 1500ms 骤降至 10ms 以内API 调用账单直降 40%~60%。这是所有落地大模型应用的团队必须优先构建的高 ROI 基础设施。 Redis 语义缓存初探：用向量距离命中历史相似 Query
Redis 语义缓存初探用向量距离命中历史相似 Query在大模型问答和企业知识库系统的生产运营账本里Token 费用与首字延迟TTFT永远是两项最沉重的负担。在真实的客服工单和业务咨询场景中用户的提问具有极强的语义聚集性用户 A 问“如何开具增值税电子发票”用户 B 问“请问电子增值税发票怎么开”用户 C 问“开电子专票的具体流程是什么”这三个问题在字面字符上完全不同传统的精确 Key-Value 缓存如MD5(query)完全无法命中。结果是系统不得不对每一个提问都重新执行一次全套的“向量检索 重排 大模型推理”白白消耗了数万次昂贵的 API 调用和数秒的推理时间。利用Redis 8 / RediSearch 向量检索实现的语义缓存Semantic Cache通过计算 Query 之间的高维余弦相似度能够直接在毫秒级命中历史相似提问的高质量答案实现成本与延迟的断崖式下降。语义缓存的两段式工作流语义缓存的核心架构可以分为前置检索判定与异步结果回填两个阶段[ 终端用户 Query ] | v 生成 Query Embedding (5ms) ---------------- Redis 语义缓存层 ---------------- | 1. FT.SEARCH 近邻探查历史缓存 Query 向量 | | 2. 计算余弦相似度 Score: | | - 若 Score 0.92 (极高置信度命中): | | -- 直接返回历史 Cache Answer (响应耗时 8ms) | | - 若 Score 0.92 (未命中): | | -- 穿透至后端 RAG 链路与大模型推理 | --------------------------------------------------- | (未命中回源) v [ 执行 RAG 检索 LLM 生成完整回答 (耗时 1500ms) ] | v 异步写入 Redis [ 存入新的 Query 向量与 Answer 到语义缓存库 (TTL 7天) ]生产级 Python 语义缓存实现以下是一套经过高并发压测检验的轻量级 Redis 语义缓存核心实现import redis from redis.commands.search.field import VectorField, TextField, NumericField from redis.commands.search.indexDefinition import IndexDefinition, IndexType from redis.commands.search.query import Query import numpy as np import time from typing import Optional, Dict, Any class RedisSemanticCache: def __init__( self, redis_client: redis.Redis, embedding_dim: int 768, similarity_threshold: float 0.90, ttl_seconds: int 86400 * 7 # 缓存 7 天 ): self.r redis_client self.dim embedding_dim self.threshold similarity_threshold self.ttl ttl_seconds self.index_name idx:semantic_cache self.prefix cache:query: self._ensure_index() def _ensure_index(self): 确保 Redis 向量索引已创建 try: self.r.ft(self.index_name).info() except Exception: schema ( TextField(query_text), TextField(answer_text), NumericField(created_at), VectorField( query_vector, HNSW, { TYPE: FLOAT32, DIM: self.dim, DISTANCE_METRIC: COSINE, M: 16, EF_CONSTRUCTION: 200 } ) ) self.r.ft(self.index_name).create_index( schema, definitionIndexDefinition(prefix[self.prefix], index_typeIndexType.HASH) ) def get_similar_answer(self, query_vector: np.ndarray) - Optional[Dict[str, Any]]: 探查是否存在相似度高于阈值的历史答案 raw_bytes query_vector.astype(np.float32).tobytes() # 查找 Top-1 最近邻 q ( Query(*[KNN 1 query_vector $vec AS score]) .sort_by(score) .return_fields(query_text, answer_text, score, created_at) .dialect(2) ) results self.r.ft(self.index_name).search(q, query_params{vec: raw_bytes}) if results.docs: top_doc results.docs[0] # RediSearch COSINE 返回的是余弦距离 (Distance 1 - Cosine_Similarity) cosine_distance float(top_doc.score) similarity 1.0 - cosine_distance if similarity self.threshold: return { matched_query: top_doc.query_text, cached_answer: top_doc.answer_text, similarity: similarity, hit: True } return None def set_cache(self, query_text: str, query_vector: np.ndarray, answer_text: str): 异步写入新的问答缓存 doc_id f{self.prefix}{int(time.time() * 1000)} raw_bytes query_vector.astype(np.float32).tobytes() mapping { query_text: query_text, answer_text: answer_text, created_at: int(time.time()), query_vector: raw_bytes } # 写入 Hash 并设置过期时间 self.r.hset(doc_id, mappingmapping) self.r.expire(doc_id, self.ttl)生产环境的三大核心避坑点阈值设定的“语义漂移”陷阱阈值设得太低如 0.80会导致“如何注销账号”误命中“如何修改账号密码”发生严重的答非所问阈值设得太高如 0.98会导致稍微换个同义词就无法命中缓存利用率形同虚设黄金区间对于 768 维高质量 Embedding 模型如 BGE/Text-Embedding-30.90 ~ 0.92是误判率低于 0.5% 且命中率最优的黄金甜点位。时效性与数据一致性对于产品价格、库存、动态政策等敏感问答不能无脑长期缓存。必须结合业务标签TagField存储版本号一旦后端知识库发生更新通过发布订阅Pub/Sub或扫描 Tag 批量清除相关旧缓存。缓存冷启动与击穿在大促或新功能上线前可以从历史客服日志中提取 Top-500 高频问答离线运行大模型生成标准答案并预热加载进 Redis 语义缓存库直接为线上拦截 60% 以上的突发流量。总结引入 Redis 语义缓存后企业大模型服务的端到端 P99 响应延迟可从 1500ms 骤降至 10ms 以内API 调用账单直降 40%~60%。这是所有落地大模型应用的团队必须优先构建的高 ROI 基础设施。