LLM 部署与缓存策略：从推理加速到成本优化的工程实践

一、推理成本与延迟的双重压力：LLM 部署的工程瓶颈

大语言模型（LLM）的部署面临两个核心挑战：推理延迟和成本。一次 GPT-4 级别模型的推理请求，首 Token 延迟（TTFT）可达 2-5 秒，生成延迟约 50-100ms/Token。在实时对话场景中，用户对首响应的容忍度通常在 2 秒以内，超过 3 秒即可感知到明显卡顿。成本方面，GPU 推理的单次请求成本在 0.01-0.1 元之间，日请求量百万级时月成本可达百万元。更关键的是，大量请求存在高度重复性——相同或相似的 Prompt 反复发送给模型，每次都重新计算一遍。通过缓存策略消除重复计算，是降低延迟和成本最直接有效的手段。

二、LLM 缓存的核心机制：KV Cache、语义缓存与前缀缓存

LLM 缓存策略分为三个层次，从底层到高层逐步消除重复计算。

flowchart TB
    subgraph 请求层["请求接入"]
        A[用户请求] --> B[语义缓存查询]
    end

    subgraph 缓存层["三级缓存体系"]
        B -->|命中| C[语义缓存<br/>相似问题直接返回]
        B -->|未命中| D[前缀缓存查询]
        D -->|命中| E[前缀缓存<br/>复用 KV Cache 前缀]
        D -->|未命中| F[完整推理]
        F --> G[生成 KV Cache]
        G --> H[写入前缀缓存]
        G --> I[写入语义缓存]
    end

    subgraph 推理层["推理引擎"]
        E --> J[从缓存点继续推理]
        F --> J
        J --> K[流式输出]
    end

    subgraph 存储层["缓存存储"]
        C -.->|Redis| L1[语义向量索引]
        E -.->|GPU HBM| L2[KV Cache 池]
        H -.->|GPU HBM| L2
    end

KV Cache（键值缓存）：这是最底层的缓存，由推理引擎自动管理。在 Transformer 的自回归生成过程中，每生成一个新 Token 都需要对前面所有 Token 做注意力计算。KV Cache 将已计算的 Key 和 Value 向量缓存下来，避免重复计算。KV Cache 的大小与序列长度和模型维度成正比，一个 70B 模型在 4096 Token 序列上的 KV Cache 约占 2GB 显存。KV Cache 是推理引擎的内部优化，应用层无需干预。

前缀缓存（Prefix Cache）：当多个请求共享相同的 Prompt 前缀时（如系统提示词 + 上下文），前缀缓存将共享前缀的 KV Cache 保留在 GPU 显存中，新请求只需计算差异部分的 KV。前缀缓存的命中率取决于请求间的 Prompt 相似度。在客服对话场景中，系统提示词通常固定，前缀缓存命中率可达 60%-80%。前缀缓存存储在 GPU HBM 中，受显存容量限制，需要 LRU 策略淘汰。

语义缓存（Semantic Cache）：这是最高层的缓存，基于请求的语义相似度匹配。当新请求与已缓存的请求在语义上足够相似时，直接返回缓存的响应，无需调用模型推理。语义相似度通过向量嵌入计算，阈值通常设为 0.92-0.95。语义缓存存储在 Redis 等外部存储中，容量远大于 GPU 显存，但查询延迟（1-5ms）高于前缀缓存。

三、生产级 LLM 缓存系统：语义缓存与前缀缓存的实现

以下代码展示了语义缓存系统的核心实现，包括向量索引和相似度匹配：

/**
 * LLM 语义缓存系统
 * 核心原理：将请求 Prompt 转为向量嵌入，在向量索引中查找相似请求
 * 相似度超过阈值时直接返回缓存的响应
 */
public class SemanticCache {

    private final EmbeddingService embeddingService;  // 向量嵌入服务
    private final VectorIndex vectorIndex;            // 向量索引（HNSW）
    private final CacheStorage cacheStorage;          // 缓存存储（Redis）

    // 语义相似度阈值：低于此值不命中缓存
    // 阈值越高，缓存越精确但命中率越低
    private static final double SIMILARITY_THRESHOLD = 0.93;
    // 缓存过期时间：避免返回过时的响应
    private static final Duration CACHE_TTL = Duration.ofHours(24);

    /**
     * 查询语义缓存
     * 返回 Optional.empty() 表示未命中，需要调用模型推理
     */
    public Optional<CachedResponse> query(String prompt,
            Map<String, String> context) {
        // 1. 将 Prompt 转为向量嵌入
        float[] embedding;
        try {
            embedding = embeddingService.embed(prompt);
        } catch (EmbeddingException e) {
            // 嵌入服务不可用时，跳过缓存直接推理
            log.warn("嵌入服务异常，跳过语义缓存: {}", e.getMessage());
            return Optional.empty();
        }

        // 2. 在向量索引中查找最相似的缓存条目
        List<SearchResult> results = vectorIndex.search(
            embedding, 5); // 返回 Top 5 候选

        if (results.isEmpty()) {
            return Optional.empty();
        }

        // 3. 验证相似度是否超过阈值
        SearchResult bestMatch = results.get(0);
        if (bestMatch.getSimilarity() < SIMILARITY_THRESHOLD) {
            // 最高相似度仍低于阈值，缓存未命中
            return Optional.empty();
        }

        // 4. 从缓存存储中获取响应
        String cacheKey = bestMatch.getCacheKey();
        CachedResponse cached = cacheStorage.get(cacheKey);

        if (cached == null || cached.isExpired()) {
            // 缓存已过期或被清除
            return Optional.empty();
        }

        // 5. 上下文验证：确保缓存响应的上下文与当前请求兼容
        if (!isContextCompatible(cached.getContext(), context)) {
            return Optional.empty();
        }

        // 记录缓存命中指标
        MetricsCollector.recordCacheHit("semantic",
            bestMatch.getSimilarity());

        return Optional.of(cached);
    }

    /**
     * 将推理结果写入语义缓存
     */
    public void put(String prompt, Map<String, String> context,
            String response) {
        try {
            float[] embedding = embeddingService.embed(prompt);
            String cacheKey = generateCacheKey(embedding);

            CachedResponse cached = new CachedResponse(
                response, context, Instant.now(), CACHE_TTL);

            // 同时写入向量索引和缓存存储
            vectorIndex.upsert(embedding, cacheKey);
            cacheStorage.set(cacheKey, cached, CACHE_TTL);

            MetricsCollector.recordCachePut("semantic");
        } catch (Exception e) {
            // 缓存写入失败不影响正常推理
            log.warn("语义缓存写入失败: {}", e.getMessage());
        }
    }

    /**
     * 上下文兼容性检查
     * 即使 Prompt 相似，不同的上下文（如不同的系统提示词）
     * 也可能导致模型给出不同的响应
     */
    private boolean isContextCompatible(
            Map<String, String> cachedContext,
            Map<String, String> currentContext) {
        // 系统提示词必须完全一致
        String cachedSystem = cachedContext.get("system_prompt");
        String currentSystem = currentContext.get("system_prompt");
        if (cachedSystem != null && currentSystem != null
                && !cachedSystem.equals(currentSystem)) {
            return false;
        }
        // 模型版本必须一致
        String cachedModel = cachedContext.get("model_version");
        String currentModel = currentContext.get("model_version");
        if (cachedModel != null && currentModel != null
                && !cachedModel.equals(currentModel)) {
            return false;
        }
        return true;
    }
}

前缀缓存的管理需要考虑 GPU 显存的有限性：

"""
LLM 前缀缓存管理器
核心职责：管理 GPU 显存中的 KV Cache 前缀，LRU 淘汰策略
适配 vLLM 等推理引擎的 Prefix Caching 接口
"""
from dataclasses import dataclass
from typing import Optional
import hashlib
import threading
import time


@dataclass
class PrefixCacheEntry:
    """前缀缓存条目"""
    prefix_hash: str           # 前缀内容的哈希
    token_count: int           # 前缀 Token 数
    kv_cache_size_mb: float    # KV Cache 占用显存（MB）
    hit_count: int             # 命中次数
    last_access_time: float    # 最后访问时间
    created_time: float        # 创建时间


class PrefixCacheManager:
    """
    前缀缓存管理器
    采用 LRU 策略管理 GPU 显存中的 KV Cache 前缀
    当显存不足时，优先淘汰最久未使用的前缀
    """

    def __init__(self, max_memory_mb: float,
                 eviction_threshold: float = 0.85):
        self.max_memory_mb = max_memory_mb
        self.eviction_threshold = eviction_threshold
        self.entries: dict[str, PrefixCacheEntry] = {}
        self.used_memory_mb = 0.0
        self.lock = threading.Lock()

    def lookup(self, prefix_tokens: list[int]) -> Optional[str]:
        """
        查找前缀缓存
        返回前缀哈希（命中）或 None（未命中）
        """
        prefix_hash = self._compute_hash(prefix_tokens)

        with self.lock:
            entry = self.entries.get(prefix_hash)
            if entry is not None:
                entry.hit_count += 1
                entry.last_access_time = time.time()
                return prefix_hash
            return None

    def insert(self, prefix_tokens: list[int],
               kv_cache_size_mb: float) -> bool:
        """
        插入前缀缓存
        当显存不足时触发 LRU 淘汰
        返回是否插入成功
        """
        prefix_hash = self._compute_hash(prefix_tokens)

        with self.lock:
            # 已存在则更新
            if prefix_hash in self.entries:
                self.entries[prefix_hash].last_access_time = \
                    time.time()
                return True

            # 检查显存是否足够
            if self.used_memory_mb + kv_cache_size_mb > \
                    self.max_memory_mb * self.eviction_threshold:
                # 触发 LRU 淘汰，释放至少 20% 的空间
                self._evict(
                    kv_cache_size_mb +
                    self.max_memory_mb * 0.2
                )

            # 再次检查
            if self.used_memory_mb + kv_cache_size_mb > \
                    self.max_memory_mb:
                return False  # 淘汰后仍不足

            entry = PrefixCacheEntry(
                prefix_hash=prefix_hash,
                token_count=len(prefix_tokens),
                kv_cache_size_mb=kv_cache_size_mb,
                hit_count=0,
                last_access_time=time.time(),
                created_time=time.time(),
            )
            self.entries[prefix_hash] = entry
            self.used_memory_mb += kv_cache_size_mb
            return True

    def _evict(self, target_free_mb: float):
        """
        LRU 淘汰：按最后访问时间排序，逐个淘汰
        直到释放的显存达到目标值
        """
        sorted_entries = sorted(
            self.entries.values(),
            key=lambda e: e.last_access_time
        )

        freed = 0.0
        for entry in sorted_entries:
            if freed >= target_free_mb:
                break
            # 跳过高频命中的前缀（保护热数据）
            if entry.hit_count > 100 and \
                    (time.time() - entry.last_access_time) < 300:
                continue
            del self.entries[entry.prefix_hash]
            self.used_memory_mb -= entry.kv_cache_size_mb
            freed += entry.kv_cache_size_mb

    @staticmethod
    def _compute_hash(tokens: list[int]) -> str:
        """计算前缀 Token 序列的哈希"""
        data = ",".join(str(t) for t in tokens)
        return hashlib.sha256(data.encode()).hexdigest()[:16]

四、LLM 缓存的代价：一致性风险、显存占用与语义漂移

LLM 缓存策略在降低延迟和成本的同时，也引入了需要审慎评估的代价。

一致性风险：语义缓存返回的是历史响应，而非实时推理结果。当模型版本更新、系统提示词变更或外部知识库更新后，缓存的响应可能已过时。在金融、医疗等对准确性要求极高的场景中，返回过时响应可能导致严重后果。缓解方案是为缓存设置较短的 TTL（如 1-4 小时），并在模型版本或提示词变更时主动清除相关缓存。但短 TTL 会降低缓存命中率，需要在准确性和成本之间取舍。

显存占用：前缀缓存的 KV Cache 存储在 GPU HBM 中，直接挤占可用于推理的显存。一张 A100 的 80GB 显存中，如果分配 20GB 给前缀缓存，推理的可用显存减少 25%，最大并发请求数相应降低。在高并发场景下，前缀缓存带来的并发损失可能抵消其延迟收益。建议根据前缀缓存的命中率动态调整分配比例——命中率低于 30% 时减少缓存分配，命中率高于 60% 时增加分配。

语义漂移：语义缓存基于向量相似度匹配，但"语义相似"不等于"答案相同"。例如，"Python 如何读取文件"和"Python 如何写入文件"的语义相似度可能超过 0.85，但答案完全不同。阈值设置过高（>0.95）会大幅降低命中率，设置过低（<0.90）则会增加错误缓存的比例。建议对缓存命中结果增加"相关性验证"——用一个轻量级分类模型判断缓存响应是否适用于当前请求，但这也增加了额外的推理开销。

适用边界：LLM 缓存策略适合请求重复度高（相似 Prompt 占比 > 30%）、延迟敏感且成本敏感的场景。对于每次请求都独特的创意写作场景，缓存命中率极低，引入缓存反而增加了不必要的开销。

五、总结

LLM 部署中的缓存策略是降低推理延迟和成本的关键手段。本文从 KV Cache、前缀缓存和语义缓存三个层次设计了缓存体系，并给出了语义缓存和前缀缓存管理器的生产级实现。

落地建议：第一步，确保推理引擎启用了 KV Cache，这是最基本的优化；第二步，分析请求的 Prompt 分布，识别共享前缀的模式，启用前缀缓存；第三步，对于重复性高的业务场景（如客服、FAQ），部署语义缓存，将相似请求的响应直接返回；第四步，建立缓存命中率监控，根据命中率动态调整缓存参数和显存分配；第五步，在模型版本更新或提示词变更时，建立缓存清除机制，确保响应的一致性。缓存是空间换时间的经典策略，在 LLM 部署中，合理利用缓存可以将推理成本降低 30%-50%，同时将平均响应延迟缩短 40% 以上。