RAG 缓存与性能优化

RAG系统的性能瓶颈分析

一个典型的RAG请求链路是：Embedding查询→向量检索→文档重排序→上下文拼接→LLM生成。实际测试中，向量检索和LLM生成各占约40%的延迟，Embedding和重排序占20%。优化需要从每个环节入手，但最有价值的优化点是缓存——很多查询是重复或相似的，缓存可以将P99延迟从3秒降低到不到100毫秒。

多层次缓存架构

推荐三层缓存：L1-精确匹配缓存（完全相同查询→直接返回缓存结果，使用Redis String，TTL=1小时）、L2-语义相似缓存（相似度>0.95的查询→复用缓存结果，使用向量索引+语义去重）、L3-文档片段缓存（高频检索文档片段的Embedding预计算缓存，使用Redis Vector Set）。L1命中率约15-25%，L2命中率约20-30%，L3可将检索延迟降低50-70%。

Redis缓存实战

import redis, json, hashlib
import numpy as np
from openai import OpenAI

client = OpenAI(api_key="your-deepseek-api-key", base_url="https://api.deepseek.com")
redis_client = redis.Redis(host="localhost", port=6379, decode_responses=True)

class RAGCache:
    def __init__(self, similarity_threshold=0.95):
        self.threshold = similarity_threshold

    def _hash(self, text):
        return hashlib.md5(text.encode()).hexdigest()

    def get_exact(self, query):
        """L1: 精确匹配缓存"""
        return redis_client.get(f"rag:exact:{self._hash(query)}")

    def set_exact(self, query, result, ttl=3600):
        redis_client.setex(f"rag:exact:{self._hash(query)}", ttl,
                           json.dumps(result, ensure_ascii=False))

    def get_semantic(self, query_embedding):
        """L2: 语义相似缓存——查找相似的历史查询"""
        # Redis向量相似度搜索
        results = redis_client.ft("rag_semantic_idx").search(
            redis.commands.search.Query(
                f"*=>[KNN 1 @embedding $vec AS score]")
            .return_fields("query", "result", "score")
            .dialect(2),
            {"vec": np.array(query_embedding, dtype=np.float32).tobytes()}
        )
        if results.docs and float(results.docs[0].score) > self.threshold:
            return json.loads(results.docs[0].result)
        return None

    def embed_query(self, query):
        resp = client.embeddings.create(
            model="text-embedding-3-large", input=query)
        return resp.data[0].embedding

    def retrieve(self, query, retriever_func):
        """带缓存的检索"""
        # L1检查
        cached = self.get_exact(query)
        if cached:
            return {"source": "L1-cache", "result": json.loads(cached)}
        # L2检查
        emb = self.embed_query(query)
        cached = self.get_semantic(emb)
        if cached:
            return {"source": "L2-cache", "result": cached}
        # 实际检索
        result = retriever_func(query)
        self.set_exact(query, result)
        return {"source": "retriever", "result": result}

cache = RAGCache()
# 使用：cache.retrieve("什么是RAG?", my_retriever_func)

批量优化与异步处理

批量Embedding：将多次查询合并为一次API调用，Embedding API支持批量输入（最多2048条），可将Embedding阶段延迟降低90%。异步检索：向量检索和文档重排序是IO密集型操作，使用asyncio并发执行可将多路检索延迟从串行的N×t降低为近似的max(t)。连接池：Redis和向量数据库使用连接池避免频繁建连开销。预加载热点数据：根据统计将访问频率最高的20%文档预加载到内存。

成本优化

除了性能，缓存也大幅降低API调用成本。以OpenAI Embedding为例，text-embedding-3-large每百万token $0.13——看似便宜但每天10万次查询年成本超过$5000。精确缓存可减少15-25%的Embedding调用，语义缓存额外减少20-30%，两者结合年节省$2000-3000。对于高频LLM生成调用，如果缓存最终回复（而非中间检索结果），节省更为显著。

缓存失效策略与一致性保证

RAG系统引入缓存后面临一个经典问题：缓存一致性与时效性。当知识库文档更新后，相关的缓存条目需要失效。我们的解决方案是基于文档版本的缓存键——每个文档都有一个递增版本号（doc_v），缓存键从单纯的查询哈希变为query_hash+doc_version。当文档更新时，新的检索结果自动使用新版本号，旧版本缓存在TTL到期后自然淘汰。对于实时性要求极高的场景（如股市新闻的RAG），TTL设置为5分钟；对于相对稳定的场景（如法律条文），TTL可设置为24小时。我们还实现了主动缓存预热——分析历史查询日志识别高频查询模式，在知识库更新后主动为这些查询刷新缓存。这个策略将知识库更新后的"缓存冷启动"阶段的P95延迟从2.1秒（无预热）降低到0.3秒（预热后），用户体验几乎无感知。

RAG系统的冷启动优化

RAG系统新上线或缓存清空后，前几千次查询会经历"冷启动"——每次都需要完整的Embedding+向量检索+LLM生成流程，P99延迟可能是稳态的10倍。我们的冷启动优化策略：预热脚本——从历史查询日志中提取Top 1000高频查询（脱敏后），在服务启动时或缓存清空后，用预热脚本批量执行这些查询填充缓存；渐进式放量——新Pod启动后先进入"预热模式"（只接收10%流量），缓存命中率达到60%以上后再接收100%流量；缓存持久化——定期将L1精确匹配缓存dump到磁盘，新Pod启动时从磁盘恢复缓存。加上预热机制，冷启动P99延迟从2.8秒降低到0.5秒，用户体验的断裂感大幅减少。

语义缓存的实现细节与调优

语义缓存（L2缓存）的核心挑战是"相似但不完全相同"的查询识别。如果相似度阈值设得太高（>0.98），很少命中；设得太低（<0.8），可能返回不相关的缓存结果。我们的调优经验：阈值设定——对于事实型查询（如"Python版本"），相似度阈值设0.92即可；对于开放性查询（如"帮我写一首诗"），即使相似度0.98也不应使用缓存——诗歌应该是原创的。实现上，我们为每条查询附加一个"query_type"标签，根据类型决定是否允许使用语义缓存。缓存淘汰策略——基于TTL+LFU（最不频繁使用）的双重淘汰，热门查询自动延长TTL，冷门查询即使未到TTL也可能被淘汰。这个精细化策略将语义缓存的"坏命中率"（返回不相关缓存）从8%降低到1.5%。