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: """生成缓存 key：query + 参数的稳定哈希""" normalized = json.dumps({"query": query.strip().lower(), **params}, sort_keys=True) digest = hashlib.md5(normalized.encode()).hexdigest()[:16] return f"rag: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_exceptions=True) 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_responses=True) config = CacheConfig( strategy=InvalidationStrategy.VERSIONED, default_ttl=600, max_ttl=7200, ) 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 的"脑容量"瘦身。
