上下文压缩后 RAG 检索降级的 4 种应对策略

上下文压缩后 RAG 检索降级

问题:压缩后 Embedding 变了

给 RAG 对话系统加上上下文压缩(对话历史摘要化)后,向量检索缓存命中率从 60% 跌到接近 0。这不是缓存本身坏了,而是压缩改变了对话历史的向量空间:同一个问题在压缩前 embedding 相似度 0.91,压缩后仅 0.73,跌破 0.85 命中阈值。

实测中,一次摘要化压缩可使语义指纹偏差达 0.2。错误解法有两个:一是忽略这个问题继续用旧缓存,相关性大幅下降;二是每次查询都重新 Embedding,白耗算力。

策略一:版本化缓存

每个压缩版本分配独立缓存命名空间。查询进来时,先查当前版本是否有命中的缓存;版本切换时,保留旧版本缓存供回退使用,新请求写入新空间。

# 版本化缓存 key 结构
def cache_key(version: int, query: str) -> str:
    return f"rag:v{version}:{hashlib.md5(query.encode()).hexdigest()}"

# 压缩触发时版本号递增
VERSION = "v1"  # 初始版本
def compress_and_upgrade():
    global VERSION
    # 摘要化压缩逻辑...
    VERSION = f"v{int(VERSION[1:]) + 1}"  # v1 → v2

适用场景:对话轮次明确、版本切换频率可预期(每 N 轮压缩一次)的客服或助手场景。缺点是版本积累后缓存膨胀,需要定期清理旧版本。

策略二:旁路 LRU 缓存

压缩后直接清空语义缓存,改用旁路 LRU。旁路缓存基于关键词或实体匹配,不依赖完整向量相似度,压缩前后行为一致。

from collections import OrderedDict

class BypassLRUCache:
    def __init__(self, maxsize=128):
        self.cache = OrderedDict()
        self.maxsize = maxsize

    def get(self, key: str):
        """key = query text, 不涉及 embedding"""
        return self.cache.get(key)

    def set(self, key: str, value):
        if key in self.cache:
            self.cache.move_to_end(key)
        else:
            if len(self.cache) >= self.maxsize:
                self.cache.popitem(last=False)
        self.cache[key] = value

    def invalidate_on_compress(self):
        """压缩触发时清空,强制重建"""
        self.cache.clear()

旁路 LRU 的代价是牺牲语义相似度匹配能力,只用字符串精确匹配。适合压缩频繁(每 3-5 轮就压缩)的场景。

策略三:语义指纹 + 关键词双重匹配

语义缓存 key 包含两层:Embedding 向量的量化指纹(8bytes)+ 关键词集合的哈希。压缩前后只要关键词集合没变,即使向量变了,也能命中缓存。

import hashlib, re

def semantic_cache_key(query: str, top_k=3) -> str:
    # 第一层:提取关键词集合哈希
    words = re.findall(r'\w+', query.lower())
    kw_hash = hashlib.md5(','.join(sorted(set(words[:top_k]))).encode()).hexdigest()[:8]
    # 第二层:query 本身 MD5(精确匹配兜底)
    query_hash = hashlib.md5(query.encode()).hexdigest()[:12]
    return f"sem:{kw_hash}:{query_hash}"

# 命中条件:关键词集合相似 AND query 精确匹配
def cache_lookup(query: str, cache: dict) -> Optional[str]:
    key = semantic_cache_key(query)
    return cache.get(key)

这种双层设计让压缩前后关键词重叠度高时仍能命中,同时精确匹配保证了结果相关性。缺点是精确匹配覆盖率不高(约 40-60%)。

策略四:分层压缩策略

不等到对话历史膨胀再压缩,而是分层处理:最近 N 轮保持完整语义,中间 M 轮轻度压缩(保留关键实体和意图),更早的轮次才做深度摘要。这样每个压缩版本只影响局部向量空间,全局缓存不会突然失效。

def layered_compress(history: list, config: dict) -> dict:
    """
    history: [{role, content}, ...]
    config: {keep_rounds=5, light_compress_rounds=10, deep_compress_before=10}
    """
    total = len(history)
    keep = history[-config["keep_rounds"]:]
    light = history[-config["keep_rounds"]-config["light_compress_rounds"]:-config["keep_rounds"]]
    deep = history[:-config["keep_rounds"]+config["light_compress_rounds"]]

    return {
        "keep": keep,                        # 完整语义,不压缩
        "light": [compress_light(t) for t in light],  # 保留实体+意图
        "deep": [compress_deep(t) for t in deep]        # 深度摘要
    }

分层压缩让向量空间变化是渐进的,不是断崖式的。实测中,这种策略比全量压缩的缓存命中率高出约 35%。

现在你可以做什么

第一步:在 RAG 缓存层加版本号字段,压缩触发时 version++,缓存 key 从此带上版本前缀。不想自己写,用 Redis 的 SCAN + DEL 批量清理也能快速落地。

第二步:如果版本化改动太大,先加旁路 LRU(20 行代码),关键词精确匹配兜底,压缩后自动 invalidate。

第三步:跑一周真实流量,统计压缩前后缓存命中率和平均 Token 消耗,用真实数据决定用哪种策略,而不是抄网上的"80%就该压缩"。

© 版权声明
THE END
喜欢就支持一下吧
点赞13 分享