Redis Vector Search 与多级缓存：AI 服务的低延迟检索与缓存穿透防护
Redis Vector Search 与多级缓存：AI 服务的低延迟检索与缓存穿透防护
一、AI 服务的延迟瓶颈：检索链路上的每一毫秒都在算账
AI 应用的端到端延迟由多个环节叠加：用户请求解析（1-5ms）、向量嵌入计算（10-50ms）、向量检索（5-20ms）、LLM 推理（200-2000ms）、响应序列化（1-5ms）。其中向量检索和 LLM 推理是两个最大的延迟贡献者。LLM 推理的优化已有大量实践（量化、KV Cache、连续批处理），但向量检索的优化常被忽视。
生产环境中，向量检索的延迟不仅取决于索引算法，还取决于网络开销和缓存命中率。如果每次检索都要访问远程向量数据库（Milvus/Pinecone），网络往返可能增加 5-20ms 延迟。对于高频查询（如热门 FAQ），大量请求命中相同或相似的查询向量，重复计算嵌入和检索完全是浪费。
Redis Vector Search（Redis 8.0+ 的原生向量检索能力）和多级缓存设计，可以在毫秒级完成向量检索，同时将高频查询的端到端延迟降低 50% 以上。但 Redis 的向量检索能力有限（不支持 HNSW 以外的高级索引），缓存设计需要处理穿透、雪崩和一致性等经典问题。
二、多级缓存与 Redis Vector Search 的协作架构
多级缓存的核心思想是：离用户越近的缓存层延迟越低、容量越小；离用户越远的存储层延迟越高、容量越大。
flowchart TB A[用户请求] --> B{L1: 本地内存缓存} B -->|命中| C[直接返回] B -->|未命中| D{L2: Redis 向量缓存} D -->|命中| E[返回缓存结果] D -->|未命中| F{L3: Redis Vector Search} F -->|命中| G[返回检索结果 + 写入 L2] F -->|未命中| H{L4: Milvus/Pinecone 远程检索} H --> I[返回检索结果 + 写入 L2/L3] subgraph 缓存策略 J[精确匹配: 查询哈希 → 结果] K[语义匹配: 查询向量 → 相似查询的结果] L[部分匹配: 查询子集 → 部分结果] end B --> J D --> K F --> K subgraph Redis Vector Search 架构 M[FT.CREATE: 创建向量索引] N[FT.SEARCH: 向量 + 关键词混合检索] O[HNSW 索引: 内存中快速 ANN] P[元数据过滤: TAG/NUMERIC 字段] end F --> M F --> N F --> O F --> P subgraph 缓存一致性 Q[写入时更新: 同步刷新缓存] R[TTL 过期: 异步刷新] S[版本号: 检测数据变更] end G --> Q E --> R I --> SL1 本地内存缓存。使用进程内 LRU 缓存（如 Python 的cachetools.LRUCache），存储最近查询的精确匹配结果。命中时延迟 < 1ms，但容量受限于进程内存，且多实例间不共享。适合存储高频热门查询（如 Top 100 FAQ）。
L2 Redis 语义缓存。这是最关键的一层。不是缓存精确匹配的查询，而是缓存语义相似的查询——当新查询的向量与已缓存查询的向量余弦相似度超过阈值（如 0.95）时，直接返回缓存结果。Redis Vector Search 提供了在 Redis 内部做向量检索的能力，避免了外部向量数据库的网络往返。语义缓存的命中率远高于精确缓存，因为用户对同一问题的表述方式多种多样。
L3 Redis Vector Search。当语义缓存未命中时，在 Redis 的向量索引中做 ANN 检索。Redis 使用 HNSW 索引，在内存中完成检索，延迟通常在 1-5ms。相比远程访问 Milvus（5-20ms 网络延迟），Redis 的本地检索延迟更低。
L4 远程向量数据库。当 Redis 中的数据不足以回答查询时（如冷门查询、新入库的文档），回退到 Milvus/Pinecone 做全量检索。检索结果写入 Redis 缓存，后续相同或相似的查询可以直接从 Redis 返回。
三、多级缓存与 Redis Vector Search 的代码实现
import hashlib import json import time import numpy as np from dataclasses import dataclass from typing import Optional, Any from collections import OrderedDict import logging logger = logging.getLogger(__name__) # ============================================================ # L1: 本地内存缓存 # ============================================================ class LocalLRUCache: """进程内 LRU 缓存：存储热门查询的精确匹配结果""" def __init__(self, max_size: int = 1000): self._cache: OrderedDict[str, Any] = OrderedDict() self.max_size = max_size self.hits = 0 self.misses = 0 def get(self, key: str) -> Optional[Any]: if key in self._cache: self._cache.move_to_end(key) self.hits += 1 return self._cache[key] self.misses += 1 return None def put(self, key: str, value: Any) -> None: if key in self._cache: self._cache.move_to_end(key) self._cache[key] = value if len(self._cache) > self.max_size: self._cache.popitem(last=False) def stats(self) -> dict: total = self.hits + self.misses return { "size": len(self._cache), "hit_rate": round(self.hits / total, 4) if total > 0 else 0, } # ============================================================ # L2/L3: Redis 语义缓存与向量检索 # ============================================================ class RedisVectorCache: """Redis 语义缓存：基于向量相似度的查询结果缓存""" def __init__( self, redis_client=None, index_name: str = "query_cache", similarity_threshold: float = 0.95, ttl_seconds: int = 3600, ): self.redis = redis_client self.index_name = index_name self.similarity_threshold = similarity_threshold self.ttl_seconds = ttl_seconds self._index_created = False def _ensure_index(self, dim: int = 1024) -> None: """确保 Redis 向量索引存在""" if self._index_created: return try: # 检查索引是否已存在 self.redis.ft(self.index_name).info() self._index_created = True return except Exception: pass # 创建向量索引 from redis.commands.search.field import TextField, VectorField from redis.commands.search.indexDefinition import IndexDefinition schema = ( TextField("query_text"), TextField("result_json"), VectorField( "query_vector", "HNSW", { "TYPE": "FLOAT32", "DIM": dim, "DISTANCE_METRIC": "COSINE", "M": 16, "EF_CONSTRUCTION": 200, }, ), ) definition = IndexDefinition(prefix="cache:") self.redis.ft(self.index_name).create_index( fields=schema, definition=definition ) self._index_created = True logger.info("Redis 向量索引创建完成: %s", self.index_name) async def semantic_get( self, query_vector: list, query_text: str, ) -> Optional[dict]: """语义缓存查询：查找与当前查询向量最相似的缓存结果""" if self.redis is None: return None try: from redis.commands.search.query import Query # 将查询向量转为字节 query_vector_bytes = np.array( query_vector, dtype=np.float32 ).tobytes() # 在 Redis 中做向量检索，返回最相似的 1 个结果 q = ( Query(f"(*)=>[KNN 1 @query_vector $vec AS score]") .sort_by("score") .return_fields("query_text", "result_json", "score") .dialect(2) ) results = self.redis.ft(self.index_name).search( q, query_params={"vec": query_vector_bytes} ) if results.total == 0: return None doc = results.docs[0] similarity = 1 - float(doc.score) # COSINE 距离转相似度 if similarity >= self.similarity_threshold: logger.info( "语义缓存命中: 相似度 %.4f, 原始查询: %s", similarity, doc.query_text ) return json.loads(doc.result_json) return None except Exception as e: logger.warning("语义缓存查询失败: %s", str(e)) return None async def semantic_put( self, query_vector: list, query_text: str, result: dict, ) -> None: """将查询结果写入语义缓存""" if self.redis is None: return try: # 生成缓存键 cache_key = f"cache:{hashlib.md5(query_text.encode()).hexdigest()}" # 存储查询向量和结果 query_vector_bytes = np.array( query_vector, dtype=np.float32 ).tobytes() self.redis.hset( cache_key, mapping={ "query_text": query_text, "result_json": json.dumps(result, ensure_ascii=False), "query_vector": query_vector_bytes, }, ) self.redis.expire(cache_key, self.ttl_seconds) except Exception as e: logger.warning("语义缓存写入失败: %s", str(e)) # ============================================================ # 多级缓存编排器 # ============================================================ class MultiLevelCacheOrchestrator: """多级缓存编排器：协调 L1/L2/L3/L4 的查询流程""" def __init__( self, local_cache_size: int = 1000, redis_client=None, vector_db=None, embedding_model=None, similarity_threshold: float = 0.95, ): self.l1_cache = LocalLRUCache(max_size=local_cache_size) self.l2_cache = RedisVectorCache( redis_client=redis_client, similarity_threshold=similarity_threshold, ) self.vector_db = vector_db # Milvus/Pinecone self.embedding_model = embedding_model async def query( self, query_text: str, top_k: int = 5, category_filter: Optional[str] = None, ) -> dict: """多级缓存查询：依次尝试 L1 → L2 → L3 → L4""" start_time = time.perf_counter() # L1: 本地精确缓存 cache_key = self._make_cache_key(query_text, top_k, category_filter) result = self.l1_cache.get(cache_key) if result is not None: elapsed = (time.perf_counter() - start_time) * 1000 return {**result, "cache_level": "L1", "latency_ms": round(elapsed, 1)} # 计算查询向量（L1 未命中才计算，避免无谓开销） query_vector = await self._compute_embedding(query_text) # L2/L3: Redis 语义缓存 result = await self.l2_cache.semantic_get(query_vector, query_text) if result is not None: self.l1_cache.put(cache_key, result) elapsed = (time.perf_counter() - start_time) * 1000 return {**result, "cache_level": "L2", "latency_ms": round(elapsed, 1)} # L4: 远程向量数据库检索 result = await self._remote_search( query_vector, query_text, top_k, category_filter ) # 写入缓存 self.l1_cache.put(cache_key, result) await self.l2_cache.semantic_put(query_vector, query_text, result) elapsed = (time.perf_counter() - start_time) * 1000 return {**result, "cache_level": "L4", "latency_ms": round(elapsed, 1)} async def _compute_embedding(self, text: str) -> list: """计算查询向量""" if self.embedding_model is None: # 无嵌入模型时使用简单哈希作为伪向量（仅用于演示） return [0.0] * 1024 return self.embedding_model.encode(text).tolist() async def _remote_search( self, query_vector: list, query_text: str, top_k: int, category_filter: Optional[str], ) -> dict: """远程向量数据库检索""" if self.vector_db is None: return {"results": [], "total": 0} hits = self.vector_db.search( query_embedding=query_vector, top_k=top_k, category_filter=category_filter, ) return {"results": hits, "total": len(hits)} @staticmethod def _make_cache_key(query_text: str, top_k: int, category_filter: Optional[str]) -> str: """生成缓存键：包含查询文本和参数""" raw = f"{query_text}|{top_k}|{category_filter or ''}" return hashlib.md5(raw.encode()).hexdigest() def get_stats(self) -> dict: """获取缓存统计""" return { "l1": self.l1_cache.stats(), }上述代码的工程要点：L1 本地缓存使用OrderedDict实现 LRU 淘汰，命中时延迟 < 1ms；L2 语义缓存使用 Redis Vector Search 的 HNSW 索引做向量检索，相似度阈值 0.95 确保只返回语义高度一致的结果；缓存键包含查询参数（top_k、category_filter），防止参数不同的查询互相覆盖；远程检索结果同时写入 L1 和 L2，后续相同或相似的查询可以直接从缓存返回。
四、多级缓存的边界与一致性挑战
语义缓存的误命中。相似度阈值设置过高（如 0.99），命中率极低，缓存形同虚设；设置过低（如 0.85），可能返回语义相似但含义不同的结果。0.95 是一个合理的起点，但需要根据业务场景调整——事实性查询（如"Python 怎么安装"）可以用较低阈值，分析性查询（如"这个数据说明了什么"）需要更高阈值。
缓存一致性。当知识库更新（新增或修改文档）时，缓存中的旧结果可能与新数据不一致。解决方案有三种：写入时主动刷新（在文档更新接口中同步清除相关缓存）、TTL 过期自动刷新（设置合理的过期时间）、版本号检测（在缓存中存储数据版本号，查询时对比版本号）。
Redis Vector Search 的规模限制。Redis 的向量索引完全在内存中，向量数量受限于 Redis 实例的内存大小。对于百万级以上的向量，Redis 的内存成本可能高于 Milvus（Milvus 支持磁盘存储）。Redis Vector Search 适合作为缓存层存储热数据，而非全量数据存储。
适用边界：多级缓存适用于查询模式有明显热点分布的 AI 服务（如客服 FAQ、产品知识库）。对于查询完全随机的场景（如学术搜索），缓存命中率极低，多级缓存的收益不足以覆盖复杂度成本。Redis Vector Search 适合向量规模在百万级以内的场景，超过百万级应使用 Milvus。
五、总结
Redis Vector Search 与多级缓存设计的核心价值是：将高频查询的检索延迟从 10-20ms 降低到 1-5ms，同时减少远程向量数据库的查询压力。L1 本地缓存处理精确匹配的热门查询，L2 Redis 语义缓存处理语义相似的高频查询，L3 Redis Vector Search 处理本地向量检索，L4 远程向量数据库处理冷门查询的全量检索。落地时需要关注三个关键点：语义缓存的相似度阈值需要根据业务场景调整，0.95 是合理起点；知识库更新时必须同步刷新缓存，防止返回过期结果；Redis Vector Search 适合作为热数据缓存层，不适合存储全量向量数据。缓存不是越多越好，而是刚好覆盖热点查询——冷数据的缓存命中率低，维护成本高于收益。