Redis 缓存一致性：向量搜索场景下缓存失效的三种策略及其代价

Redis 缓存一致性向量搜索场景下缓存失效的三种策略及其代价一、深度引言与场景痛点Redis 在 RAG 系统里的角色就像你办公桌上的便利贴——把常用的东西贴在触手可及的地方免得每次都翻箱倒柜。在向量检索场景中最常见的缓存对象是搜索结果对于高频查询比如如何重置密码与其每次走完 embedding → 向量检索 → rerank 的完整链路不如直接把结果缓存起来下次命中直接返回。但缓存的本质矛盾在于你存了一份数据的副本而源头数据可能随时变化。向量检索的缓存场景比普通的 KV 缓存更复杂因为这里的变化来源有两个维度第一是文档维度——知识库里的文档被新增、修改或删除之前缓存的搜索结果可能变得不准确。第二是模型维度——你升级了 embedding 模型同样的 query 变成了不同的向量但缓存的还是基于旧模型算出来的结果。最典型的翻车现场是这样的你上线了一个新的 FAQ 文档里面有退货流程的详细说明。但 Redis 里缓存了三天的退货搜索结果还是旧的——因为没有人触发缓存失效。用户在问怎么退货时AI 给出的答案里完全没有提到新文档直到有人投诉才发现问题。本文不泛泛地谈缓存一致性而是聚焦于向量搜索场景下三种缓存失效策略的具体实现、它们各自的性能代价和一致性保证。二、底层机制与原理深度剖析缓存一致性问题本质上是何时让缓存里的数据和源头保持一致的决策。三种主流策略对应了三种不同的回答flowchart TD A[文档变更事件] -- B{缓存失效策略} subgraph S1 [策略1: TTL 惰性失效] B -- C1[设置过期时间] C1 -- D1[缓存自然过期] D1 -- E1[下次查询回源] end subgraph S2 [策略2: 主动失效] B -- C2[监听文档变更] C2 -- D2{变更类型} D2 --|新增文档| E2[失效相关 query 缓存] D2 --|修改文档| F2[失效该文档相关所有缓存] D2 --|删除文档| G2[清理 失效关联] end subgraph S3 [策略3: 版本号驱逐] B -- C3[全局版本号 1] C3 -- D3[写入新版本号] D3 -- E3[读取时对比版本] E3 -- F3{版本一致?} F3 --|一致| G3[返回缓存] F3 --|不一致| H3[回源 更新缓存版本] end style S1 fill:#e1f5fe style S2 fill:#fff3e0 style S3 fill:#e8f5e9策略一TTL 惰性失效——我不管数据变没变反正过了 N 分钟缓存就作废。最简单零维护成本但一致性最差。在文档变更到缓存过期之间的这段时间窗口内用户拿到的是过期结果。窗口大小由 TTL 决定TTL 越小一致性越好但缓存命中率越低。策略二主动失效——文档一变我就去把相关的缓存全清了。一致性最好但实现复杂度最高。难点在于相关性的计算——文档 D 被改了哪些查询的结果里包含 D如果做一个反向索引query → docs维护成本相当高。策略三版本号驱逐——全系统维护一个版本号数据一变就加一缓存读取时比较版本号。一致性和性能的折中。相比策略二它不需要精确知道哪些缓存需要失效但代价是只要有文档变动所有查询的缓存都会在版本号递增后的下一次命中时失效容易引发缓存雪崩。三、生产级代码实现import asyncio import hashlib import json import time import logging from typing import Any, Dict, List, Optional, Callable from dataclasses import dataclass, field from enum import Enum from functools import wraps import redis.asyncio as redis logger logging.getLogger(__name__) class InvalidationStrategy(Enum): TTL ttl ACTIVE active VERSIONED versioned dataclass class CacheConfig: 缓存配置 strategy: InvalidationStrategy InvalidationStrategy.VERSIONED default_ttl: int 300 # 秒 max_ttl: int 3600 version_key: str rag:cache:version doc_index_key: str rag:cache:doc_index # doc_id → query cache keys class SearchCache: 向量搜索缓存层 同时支持三种失效策略通过 config.strategy 切换 def __init__(self, redis_client: redis.Redis, config: Optional[CacheConfig] None): self.redis redis_client self.config config or CacheConfig() self._local_version: int 0 def _cache_key(self, query: str, **params) - str: 生成缓存 keyquery 参数的稳定哈希 normalized json.dumps({query: query.strip().lower(), **params}, sort_keysTrue) digest hashlib.md5(normalized.encode()).hexdigest()[:16] return frag:search:{digest} async def get(self, query: str, **params) - Optional[List[Dict]]: 获取缓存 key self._cache_key(query, **params) if self.config.strategy InvalidationStrategy.VERSIONED: return await self._get_versioned(key) else: # TTL 和 Active 策略在读取上逻辑相同 cached await self.redis.get(key) if cached: return json.loads(cached) return None async def _get_versioned(self, key: str) - Optional[List[Dict]]: 版本号策略的读取 pipe self.redis.pipeline() pipe.get(key) pipe.get(self.config.version_key) results await pipe.execute() cached_data, version_bytes results if not cached_data or not version_bytes: return None cached json.loads(cached_data) current_version int(version_bytes) # 比较版本号 if cached.get(_version, -1) ! current_version: logger.debug(f缓存版本不匹配key{key}) await self.redis.delete(key) return None return cached.get(data) async def set(self, query: str, results: List[Dict], **params): 写入缓存 key self._cache_key(query, **params) if self.config.strategy InvalidationStrategy.VERSIONED: await self._set_versioned(key, results) elif self.config.strategy InvalidationStrategy.TTL: await self.redis.setex(key, self.config.default_ttl, json.dumps(results)) elif self.config.strategy InvalidationStrategy.ACTIVE: await self.redis.setex(key, self.config.max_ttl, json.dumps(results)) # 记录 doc_id → cache_key 的反向索引 doc_ids [r.get(id) for r in results if r.get(id)] if doc_ids: pipe self.redis.pipeline() for doc_id in doc_ids: pipe.sadd(f{self.config.doc_index_key}:{doc_id}, key) await pipe.execute() async def _set_versioned(self, key: str, results: List[Dict]): 版本号策略的写入 current_version await self.redis.get(self.config.version_key) version int(current_version or 0) payload json.dumps({ _version: version, data: results, _cached_at: time.time(), }) await self.redis.setex(key, self.config.max_ttl, payload) async def invalidate_by_doc(self, doc_id: str): 主动失效根据文档 ID 清理所有相关缓存 if self.config.strategy ! InvalidationStrategy.ACTIVE: return index_key f{self.config.doc_index_key}:{doc_id} cache_keys await self.redis.smembers(index_key) if cache_keys: pipe self.redis.pipeline() for key in cache_keys: pipe.delete(key) pipe.delete(index_key) await pipe.execute() logger.info(f主动失效 {len(cache_keys)} 个缓存, doc{doc_id}) async def invalidate_all(self): 全局版本号递增触发全量缓存重建 new_version await self.redis.incr(self.config.version_key) self._local_version new_version logger.info(f全局版本递增至 {new_version}所有缓存将在下次读取时失效) async def warmup(self, queries: List[str], search_fn: Callable): 缓存预热对高频查询提前计算并缓存 tasks [] for query in queries: # 检查缓存是否已存在 tasks.append(self._warmup_single(query, search_fn)) results await asyncio.gather(*tasks, return_exceptionsTrue) hit_count sum(1 for r in results if isinstance(r, bool) and not r) logger.info(f缓存预热完成: {len(queries)} 条, 命中 {hit_count} 条) async def _warmup_single(self, query: str, search_fn: Callable) - bool: 预热单条查询返回 True 表示跳过已缓存 cached await self.get(query) if cached: return True results await search_fn(query) await self.set(query, results) return False # 文档变更 Hook class DocumentChangeHook: 文档变更回调——对接主动失效策略 def __init__(self, cache: SearchCache): self.cache cache async def on_document_created(self, doc_id: str, content: str): 文档新增版本递增 可选主动失效 if self.cache.config.strategy in (InvalidationStrategy.VERSIONED,): await self.cache.invalidate_all() async def on_document_updated(self, doc_id: str, content: str): 文档修改主动失效该文档 版本递增 if self.cache.config.strategy InvalidationStrategy.ACTIVE: await self.cache.invalidate_by_doc(doc_id) if self.cache.config.strategy InvalidationStrategy.VERSIONED: await self.cache.invalidate_all() async def on_document_deleted(self, doc_id: str): 文档删除清理相关缓存 版本递增 if self.cache.config.strategy InvalidationStrategy.ACTIVE: await self.cache.invalidate_by_doc(doc_id) if self.cache.config.strategy InvalidationStrategy.VERSIONED: await self.cache.invalidate_all() # 使用示例 async def main(): r redis.Redis(decode_responsesTrue) config CacheConfig( strategyInvalidationStrategy.VERSIONED, default_ttl600, max_ttl7200, ) cache SearchCache(r, config) # 模拟搜索函数 async def expensive_search(query: str) - List[Dict]: await asyncio.sleep(0.3) # 模拟实际搜索延迟 return [{id: doc_1, score: 0.95, title: query}] # 第一次调用缓存未命中走搜索 results await cache.get(如何配置 Redis) if not results: results await expensive_search(如何配置 Redis) await cache.set(如何配置 Redis, results) # 第二次调用缓存命中 results2 await cache.get(如何配置 Redis) print(命中 if results2 else 未命中) # 文档变更后 await cache.invalidate_all() results3 await cache.get(如何配置 Redis) print(失效后命中 if results3 else 失效后未命中需回源) if __name__ __main__: asyncio.run(main())四、边界分析与架构权衡1. 三种策略的选择指南场景推荐策略理由知识库几乎不变周更TTL简单有效护成本为零知识库频繁变更日更Active一致性最好知识库中等变更 查询量极大Versioned一致性可接受、无反向索引负担混合场景Versioned 对核心文档 Active覆盖面最广2. 版本号策略的雪崩风险当全局版本号递增后所有缓存在下一次读取时都会发现自己过期产生大量回源请求。如果 QPS 很高500这可能导致后端服务瞬间压力过大。缓解措施是过期重建时加随机抖动在各读取请求的回源时机上增加一个随机延迟0-3秒让回源请求分散开而不是集中爆发。3. 主动失效的漏网之鱼invalidate_by_doc的反向索引只记录了当时搜索结果里有哪些 doc但无法覆盖未来可能出现的搜索结果。如果你新增了一个文档而某个查询的缓存里没有这个文档因为创建缓存时文档还不存在主动失效就漏掉了这个缓存。解决方案是文档新增也应该触发版本号递增作为主动失效的补丁。4. 缓存的正确性窗口没有任何缓存策略能保证绝对的实时一致性而在向量搜索场景下追求实时一致性也往往没有必要——用户对搜索结果延迟的容忍度远低于对少了一篇新文档的容忍度。与其追求完美一致性不如明确定义你的最大一致延迟如 5 分钟然后选择能满足这个 SLA 的最简方案。五、总结在向量搜索场景下做缓存核心是三件事缓存什么搜索结果而非文档本身、何时失效TTL / 主动 / 版本号、如何预热高频查询提前加载。如果你是一个人的团队、文档变更不频繁——TTL 策略完全够用别把事情搞复杂。如果你的系统有专门的内容管理后台、文档变更频繁——上主动失效配合反向索引。如果你是高 QPS 的大规模服务——版本号策略是性价比最高的选择但要记住加抖动逻辑防止雪崩。下一篇聊聊 Agent 上下文窗口管理——当你的 System Prompt 长得像个小型论文时怎么给 LLM 的脑容量瘦身。