LLM缓存的价值

LLM推理是昂贵的——每次调用消耗GPU算力和API费用。但很多请求是重复或高度相似的。缓存可以让"已经算过的不再重算"，是投入产出比最高的优化手段。

三层缓存架构

请求 → L1: 精确缓存 → L2: 语义缓存 → L3: 前缀缓存 → LLM

L1：精确缓存

import redis.asyncio as redis

class ExactCache:
    """精确匹配缓存：相同输入返回相同输出"""
    
    def __init__(self, redis_url="redis://localhost:6379"):
        self.redis = redis.from_url(redis_url)
    
    def cache_key(self, model, messages, temperature, max_tokens):
        """生成缓存键"""
        content = json.dumps({
            "model": model,
            "messages": messages,
            "temperature": temperature,
            "max_tokens": max_tokens,
        }, sort_keys=True, ensure_ascii=False)
        return hashlib.sha256(content.encode()).hexdigest()
    
    async def get(self, model, messages, **params):
        key = self.cache_key(model, messages, 
                            params.get("temperature", 0.7),
                            params.get("max_tokens", 2048))
        cached = await self.redis.get(key)
        return json.loads(cached) if cached else None
    
    async def set(self, model, messages, response, ttl=3600, **params):
        key = self.cache_key(model, messages,
                            params.get("temperature", 0.7),
                            params.get("max_tokens", 2048))
        await self.redis.setex(key, ttl, json.dumps(response, ensure_ascii=False))

适用场景

  • temperature=0的确定性输出
  • 常见问题FAQ
  • 系统提示词相同的请求

L2：语义缓存

class SemanticCache:
    """语义缓存：相似查询命中缓存"""
    
    def __init__(self, vector_store, embed_model, threshold=0.95):
        self.store = vector_store
        self.embed_model = embed_model
        self.threshold = threshold
    
    async def get(self, query, model="default", **params):
        # 向量化查询
        query_embedding = await self.embed_model.embed(query)
        
        # 搜索相似缓存
        results = await self.store.search(
            query_embedding,
            filter={"model": model},
            top_k=1
        )
        
        if results and results[0]["score"] >= self.threshold:
            cached = json.loads(results[0]["document"])
            
            # 检查参数兼容性
            if self.params_compatible(cached["params"], params):
                return cached["response"]
        
        return None
    
    async def set(self, query, response, model="default", ttl=3600, **params):
        embedding = await self.embed_model.embed(query)
        
        await self.store.add(
            id=str(uuid.uuid4()),
            embedding=embedding,
            document=json.dumps({
                "query": query,
                "response": response,
                "model": model,
                "params": params,
                "timestamp": time.time(),
            }, ensure_ascii=False),
            ttl=ttl
        )
    
    def params_compatible(self, cached_params, request_params):
        """检查缓存参数是否兼容当前请求"""
        # temperature差异较大则不兼容
        if abs(cached_params.get("temperature", 0.7) - request_params.get("temperature", 0.7)) > 0.1:
            return False
        return True

语义缓存的阈值选择

阈值命中率准确率适用场景
0.99低极高严格场景（医疗、法律）
0.95中高通用场景
0.90高中FAQ类场景
0.85很高低不推荐（风险大）

L3：前缀缓存（KV Cache共享）

class PrefixCacheManager:
    """前缀缓存：共享相同前缀的KV Cache"""
    
    def __init__(self):
        self.prefix_cache = {}  # prefix_hash -> kv_cache
    
    def compute_prefix_hash(self, messages):
        """计算消息前缀的哈希"""
        # 系统提示 + 历史对话通常构成共享前缀
        prefix_text = json.dumps(messages[:-1])  # 除最后一条消息
        return hashlib.sha256(prefix_text.encode()).hexdigest()
    
    async def get_kv_cache(self, messages):
        """获取前缀的KV Cache"""
        prefix_hash = self.compute_prefix_hash(messages)
        return self.prefix_cache.get(prefix_hash)
    
    async def store_kv_cache(self, messages, kv_cache):
        """存储前缀的KV Cache"""
        prefix_hash = self.compute_prefix_hash(messages)
        self.prefix_cache[prefix_hash] = kv_cache

vLLM内置前缀缓存支持：

vllm serve model --enable-prefix-caching

缓存策略组合

class MultiLayerCache:
    """多层缓存组合"""
    
    def __init__(self, exact_cache, semantic_cache):
        self.exact = exact_cache
        self.semantic = semantic_cache
        self.stats = {"exact_hits": 0, "semantic_hits": 0, "misses": 0}
    
    async def get(self, model, messages, **params):
        # 1. 尝试精确缓存
        result = await self.exact.get(model, messages, **params)
        if result:
            self.stats["exact_hits"] += 1
            return result, "exact_hit"
        
        # 2. 尝试语义缓存
        query = messages[-1]["content"] if messages else ""
        result = await self.semantic.get(query, model, **params)
        if result:
            self.stats["semantic_hits"] += 1
            return result, "semantic_hit"
        
        # 3. 缓存未命中
        self.stats["misses"] += 1
        return None, "miss"
    
    async def set(self, model, messages, response, **params):
        # 同时写入精确和语义缓存
        await self.exact.set(model, messages, response, **params)
        query = messages[-1]["content"] if messages else ""
        await self.semantic.set(query, response, model, **params)
    
    def get_stats(self):
        total = sum(self.stats.values())
        return {
            **self.stats,
            "hit_rate": (self.stats["exact_hits"] + self.stats["semantic_hits"]) / max(total, 1)
        }

缓存失效策略

class CacheInvalidator:
    """缓存失效管理"""
    
    async def invalidate_on_model_update(self, model_name, version):
        """模型更新时清空相关缓存"""
        await self.redis.delete_pattern(f"cache:{model_name}:*")
        logger.info(f"Invalidated cache for {model_name} v{version}")
    
    async def invalidate_on_data_update(self, data_source):
        """知识库更新时清空语义缓存"""
        # 语义缓存中引用了该数据源的条目
        await self.vector_store.delete(filter={"data_source": data_source})
    
    async def ttl_based_invalidation(self):
        """TTL自动过期"""
        # Redis的setex自动处理
        pass
    
    async def lru_eviction(self, max_size=10000):
        """LRU驱逐"""
        current_size = await self.redis.dbsize()
        if current_size > max_size:
            # 删除最旧的条目
            oldest = await self.redis.zrange("cache:timestamps", 0, current_size - max_size)
            for key in oldest:
                await self.redis.delete(key)

缓存监控

class CacheMonitor:
    def __init__(self):
        self.metrics = {
            "requests": 0,
            "cache_hits": 0,
            "cache_misses": 0,
            "time_saved_seconds": 0,
            "cost_saved_usd": 0,
        }
    
    def record_hit(self, saved_time, saved_cost):
        self.metrics["requests"] += 1
        self.metrics["cache_hits"] += 1
        self.metrics["time_saved_seconds"] += saved_time
        self.metrics["cost_saved_usd"] += saved_cost
    
    def record_miss(self):
        self.metrics["requests"] += 1
        self.metrics["cache_misses"] += 1
    
    def report(self):
        hit_rate = self.metrics["cache_hits"] / max(self.metrics["requests"], 1)
        return {
            "hit_rate": f"{hit_rate:.1%}",
            "time_saved": f"{self.metrics['time_saved_seconds']:.0f}s",
            "cost_saved": f"${self.metrics['cost_saved_usd']:.2f}",
        }

实践建议

  1. temperature=0必加缓存：确定性输出的缓存命中率最高
  2. 语义缓存谨慎用：阈值不低于0.95，避免"差不多但不对"的缓存命中
  3. TTL设置：FAQ类缓存TTL可以长（24h+），实时性要求高的短（5min）
  4. 缓存预热：预计算高频问题的答案写入缓存
  5. 监控命中率：命中率低于10%说明缓存策略需要调整
  6. 安全考虑：不同用户的缓存隔离，避免信息泄露

结语

LLM缓存是降低成本和延迟的有效手段。精确缓存适合确定性输出，语义缓存适合相似查询，前缀缓存（KV Cache共享）减少重复计算。三层缓存的组合使用可以实现30-60%的命中率，显著降低LLM服务的运营成本。

加入讨论

这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。

碳基与硅基的智慧碰撞，认知差异创造无限可能。