Redis缓存策略实战：大模型推理服务的热点Token缓存与淘汰机制

大模型推理服务的缓存需求分析

大模型推理的瓶颈在显存带宽而非计算能力。KV Cache（键值缓存）占用推理过程中最大的显存资源。当多个用户请求包含相同的前缀Prompt时，重复计算KV Cache造成巨大浪费。Redis缓存层在模型外部拦截重复请求，对相同Prompt直接返回缓存结果，对共享前缀的请求缓存KV Cache的Token ID，减少重复计算量。实测中，FAQ类场景缓存命中后推理延迟从3-5秒降到5毫秒以内。

Redis缓存架构设计

大模型推理缓存分为两层：语义缓存（完整Prompt→完整响应）和前缀缓存（Prompt前缀→KV Cache指针）。Redis 7.x的多数据类型恰好满足这两种缓存模式：

# Redis缓存架构
# 1. 语义缓存：String类型存储完整响应
#    Key格式: llm:sem:{md5(prompt)}
#    Value: 完整JSON响应
#    TTL: 3600s (1小时)

# 2. 前缀缓存：Hash类型存储KV Cache元数据
#    Key格式: llm:prefix:{md5(prefix_prompt)}
#    Fields:
#      kv_cache_id: GPU显存中的KV Cache标识
#      token_count: 前缀Token数
#      model_id: 模型标识
#      created_at: 创建时间戳
#    TTL: 1800s (30分钟)

# 3. 热点统计：Sorted Set记录访问频率
#    Key格式: llm:hot:rank
#    Score: 访问次数
#    Member: prompt hash

语义缓存实现：精确匹配与模糊匹配

精确匹配以Prompt的MD5值作为缓存Key，模糊匹配通过向量相似度检索语义接近的Prompt：

import redis
import hashlib
import json

class SemanticCache:
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client
        self.ttl = 3600

    def _cache_key(self, prompt: str, model: str) -> str:
        raw = f"{model}:{prompt}"
        return f"llm:sem:{hashlib.md5(raw.encode()).hexdigest()}"

    def get(self, prompt: str, model: str) -> dict | None:
        key = self._cache_key(prompt, model)
        cached = self.redis.get(key)
        if cached:
            # 命中缓存，更新热点排名
            self.redis.zincrby("llm:hot:rank", 1, key)
            return json.loads(cached)
        return None

    def set(self, prompt: str, model: str, response: dict) -> None:
        key = self._cache_key(prompt, model)
        self.redis.setex(key, self.ttl, json.dumps(response, ensure_ascii=False))
        # 记录首次访问
        self.redis.zadd("llm:hot:rank", {key: 1})

前缀缓存实现：共享Prompt KV Cache复用

当多个请求的System Prompt相同（常见于同一应用的系统指令），可以复用已计算的KV Cache，避免重复推理前缀部分：

class PrefixCache:
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client
        self.ttl = 1800

    def _prefix_key(self, prefix_prompt: str, model: str) -> str:
        raw = f"prefix:{model}:{prefix_prompt}"
        return f"llm:prefix:{hashlib.md5(raw.encode()).hexdigest()}"

    def get_kv_cache_id(
        self, prefix_prompt: str, model: str
    ) -> tuple[str | None, int]:
        key = self._prefix_key(prefix_prompt, model)
        data = self.redis.hgetall(key)
        if data:
            return data[b'kv_cache_id'].decode(), int(data[b'token_count'])
        return None, 0

    def register_kv_cache(
        self, prefix_prompt: str, model: str,
        kv_cache_id: str, token_count: int
    ) -> None:
        key = self._prefix_key(prefix_prompt, model)
        self.redis.hset(key, mapping={
            'kv_cache_id': kv_cache_id,
            'token_count': token_count,
            'model_id': model,
            'created_at': str(int(time.time())),
        })
        self.redis.expire(key, self.ttl)

vLLM支持automatic prefix caching（APC），开启后在Redis中注册KV Cache指针，后续请求命中前缀缓存时直接跳过前缀Token的计算，推理速度提升显著。

缓存淘汰策略：LRU与LFU的选择

Redis默认的淘汰策略allkeys-lru适用于通用场景，但大模型缓存有其特殊性：热点的Prompt集中度高（FAQ类占比大），LFU策略更适合：

# Redis配置 (redis.conf)
maxmemory 16gb
maxmemory-policy allkeys-lfu

# LFU调优参数
# lfu-decay-time: 计数器衰减周期（分钟）
# lfu-log-factor: 计数器增长的对数因子
lfu-decay-time 30
lfu-log-factor 10

allkeys-lfu策略下，访问频率高的缓存Key存活时间长，低频Key被优先淘汰。配合前文的Sorted Set热点排名，可以主动驱逐排名最低的缓存，避免被动淘汰造成缓存抖动。

主动驱逐实现：

class CacheEvictionManager:
    def __init__(self, redis_client: redis.Redis, max_memory_pct=0.85):
        self.redis = redis_client
        self.max_memory_pct = max_memory_pct

    def check_and_evict(self) -> int:
        info = self.redis.info('memory')
        used = info['used_memory']
        max_mem = info['maxmemory']

        if max_mem == 0 or used / max_mem < self.max_memory_pct:
            return 0

        # 内存使用率超85%，主动驱逐低频缓存
        low_freq_keys = self.redis.zrange(
            "llm:hot:rank", 0, 99, withscores=True
        )

        evicted = 0
        for key, score in low_freq_keys:
            if score < 2:  # 访问不到2次的缓存
                self.redis.delete(key)
                self.redis.zrem("llm:hot:rank", key)
                evicted += 1

        return evicted

缓存一致性：模型更新与TTL协同

模型版本更新后，旧缓存结果与新模型输出不一致。需要在模型切换时批量清理缓存：

# 模型切换时清理语义缓存
# 使用SCAN避免阻塞Redis
import asyncio

async def invalidate_model_cache(
    redis_client: redis.Redis, model_id: str
) -> int:
    pattern = f"llm:sem:*"
    cursor = 0
    deleted = 0

    while True:
        cursor, keys = redis_client.scan(
            cursor, match=pattern, count=500
        )
        if keys:
            # 逐个检查是否属于目标模型
            pipe = redis_client.pipeline()
            for key in keys:
                pipe.get(key)
            results = pipe.execute()

            for key, value in zip(keys, results):
                if value:
                    data = json.loads(value)
                    if data.get('model') == model_id:
                        redis_client.delete(key)
                        deleted += 1

        if cursor == 0:
            break

    return deleted

更优雅的方案是在缓存Key中编码模型版本号：llm:sem:{model_version}:{md5(prompt)}，模型更新时旧版本Key自然过期，无需主动清理。

缓存监控与命中率优化

缓存命中率是衡量缓存策略效果的核心指标。在Prometheus中采集Redis缓存命中率：

# Prometheus告警规则
- alert: LLMBCacheHitRateLow
  expr: |
    rate(redis_keyspace_hits_total[5m])
    / (rate(redis_keyspace_hits_total[5m]) + rate(redis_keyspace_misses_total[5m]))
    < 0.3
  for: 10m
  labels:
    severity: warning
  annotations:
    summary: "大模型语义缓存命中率低于30%"
    description: "检查缓存Key设计是否合理，TTL是否过短"

命中率低于30%时需要排查：Prompt归一化是否足够（去除时间戳、随机ID等噪声字段）、TTL是否过短导致有效缓存提前过期、缓存容量是否不足触发大量淘汰。命中率优化到50%以上时，推理集群的GPU利用率可降低30-40%，成本节省显著。