AI 缓存策略：Semantic Cache 与 Prompt Cache

原创灵阙教研团队

S 精选进阶最佳实践 | 约 8 分钟阅读更新于 2026-02-28

LLM OpenAI Anthropic 提示词工程缓存 Embedding 嵌入

AI 导读

AI 缓存策略：Semantic Cache 与 Prompt Cache 语义缓存（Embedding 相似度匹配）、精确缓存、Prompt Caching（Anthropic/OpenAI）与缓存失效策略引言 LLM 推理成本高、延迟大，而实际业务中大量请求是重复或高度相似的。据统计，典型的客服聊天机器人中 30-50%...

AI 缓存策略：Semantic Cache 与 Prompt Cache

语义缓存（Embedding 相似度匹配）、精确缓存、Prompt Caching（Anthropic/OpenAI）与缓存失效策略

引言

LLM 推理成本高、延迟大，而实际业务中大量请求是重复或高度相似的。据统计，典型的客服聊天机器人中 30-50% 的问题本质上是相同的。如果能将这些重复请求的结果缓存起来，既能显著降低成本，又能将延迟从秒级降到毫秒级。

但 LLM 缓存与传统 API 缓存有本质区别：用户很少用完全相同的文字提出相同的问题。"怎么退货"和"我要退商品"是不同的字符串，但语义完全一致。这就需要语义缓存（Semantic Cache）。

缓存策略全景

三层缓存体系

┌──────────────────────────────────────────────────┐
│                LLM 缓存金字塔                      │
│                                                  │
│              ┌───────────┐                       │
│              │ Prompt    │  Provider 端           │
│              │ Cache     │  prefix 复用            │
│              │ (API 级)   │  延迟: -50% 首 token    │
│              └─────┬─────┘                       │
│                    │                             │
│           ┌────────▼─────────┐                   │
│           │  Semantic Cache  │  应用端            │
│           │  (语义相似度)     │  embedding 匹配     │
│           │                  │  延迟: <50ms        │
│           └────────┬─────────┘                   │
│                    │                             │
│       ┌────────────▼──────────────┐              │
│       │     Exact Match Cache     │  应用端       │
│       │     (精确字符串匹配)        │  hash 查找    │
│       │                           │  延迟: <5ms   │
│       └───────────────────────────┘              │
└──────────────────────────────────────────────────┘

三种缓存对比

特性	精确匹配	语义缓存	Prompt Cache
匹配方式	字符串 hash	Embedding 相似度	前缀字符匹配
命中率	低 (5-15%)	中 (20-40%)	高 (60-80%)
额外成本	无	Embedding 计算	无（Provider 内置）
延迟减少	100%	100%	50-80%
成本减少	100%	100%	50-90%
实现复杂度	低	中	低（API 参数）
适用场景	固定模板查询	自然语言查询	长 System Prompt

精确匹配缓存

实现

# src/cache/exact_cache.py
import hashlib
import json
import time
from typing import Optional
import redis

class ExactMatchCache:
    """Cache LLM responses by exact input hash."""

    def __init__(
        self,
        redis_client: redis.Redis,
        default_ttl: int = 3600,
        prefix: str = "llm:exact:",
    ):
        self.redis = redis_client
        self.default_ttl = default_ttl
        self.prefix = prefix

    def _compute_key(self, messages: list[dict], model: str, temperature: float) -> str:
        """Deterministic hash of the full request."""
        payload = json.dumps({
            "messages": messages,
            "model": model,
            "temperature": temperature,
        }, sort_keys=True, ensure_ascii=False)

        return self.prefix + hashlib.sha256(payload.encode()).hexdigest()

    def get(self, messages: list[dict], model: str, temperature: float) -> Optional[dict]:
        key = self._compute_key(messages, model, temperature)
        cached = self.redis.get(key)
        if cached:
            data = json.loads(cached)
            # Record cache hit for metrics
            self.redis.incr(f"{self.prefix}hits")
            return data
        self.redis.incr(f"{self.prefix}misses")
        return None

    def set(
        self,
        messages: list[dict],
        model: str,
        temperature: float,
        response: dict,
        ttl: Optional[int] = None,
    ) -> None:
        key = self._compute_key(messages, model, temperature)
        self.redis.setex(
            key,
            ttl or self.default_ttl,
            json.dumps(response, ensure_ascii=False),
        )

    def get_hit_rate(self) -> float:
        hits = int(self.redis.get(f"{self.prefix}hits") or 0)
        misses = int(self.redis.get(f"{self.prefix}misses") or 0)
        total = hits + misses
        return hits / total if total > 0 else 0.0

语义缓存

核心原理

查询流程：

1. 用户输入 "怎么申请退货"
2. 计算 embedding: [0.12, -0.34, 0.56, ...]
3. 在向量库中搜索最相似的缓存条目
4. 找到 "如何退货退款" (相似度 0.95)
5. 相似度 > 阈值 (0.90) → 缓存命中
6. 直接返回缓存的 LLM 回答

未命中流程：
1. 用户输入 "你们支持什么支付方式"
2. 计算 embedding
3. 向量库中最相似的是 "怎么退货" (相似度 0.35)
4. 相似度 < 阈值 → 缓存未命中
5. 调用 LLM 获取回答
6. 将 (embedding, 回答) 存入缓存

完整实现

# src/cache/semantic_cache.py
import hashlib
import json
import time
from typing import Optional
from dataclasses import dataclass
import numpy as np

@dataclass
class CacheEntry:
    query: str
    response: dict
    embedding: list[float]
    model: str
    created_at: float
    hit_count: int = 0

class SemanticCache:
    """Semantic similarity-based LLM response cache."""

    def __init__(
        self,
        embedding_model: str = "text-embedding-3-small",
        similarity_threshold: float = 0.92,
        max_entries: int = 100_000,
        ttl_seconds: int = 86400,
    ):
        self.embedding_model = embedding_model
        self.similarity_threshold = similarity_threshold
        self.max_entries = max_entries
        self.ttl_seconds = ttl_seconds

        # Use Qdrant for vector storage
        from qdrant_client import QdrantClient, models
        self.qdrant = QdrantClient(url="http://localhost:6333")

        # Create collection if not exists
        try:
            self.qdrant.get_collection("semantic_cache")
        except Exception:
            self.qdrant.create_collection(
                collection_name="semantic_cache",
                vectors_config=models.VectorParams(
                    size=1536,  # text-embedding-3-small
                    distance=models.Distance.COSINE,
                ),
            )

    def _get_embedding(self, text: str) -> list[float]:
        import openai
        client = openai.OpenAI()
        response = client.embeddings.create(
            model=self.embedding_model,
            input=text,
        )
        return response.data[0].embedding

    def _extract_query(self, messages: list[dict]) -> str:
        """Extract the semantic query from messages."""
        # Use the last user message as the cache key
        user_messages = [m for m in messages if m["role"] == "user"]
        if not user_messages:
            return ""
        return user_messages[-1]["content"]

    def get(
        self,
        messages: list[dict],
        model: str,
    ) -> Optional[dict]:
        query = self._extract_query(messages)
        if not query:
            return None

        query_embedding = self._get_embedding(query)

        # Search for similar cached queries
        from qdrant_client import models
        results = self.qdrant.search(
            collection_name="semantic_cache",
            query_vector=query_embedding,
            limit=1,
            score_threshold=self.similarity_threshold,
            query_filter=models.Filter(
                must=[
                    models.FieldCondition(
                        key="model",
                        match=models.MatchValue(value=model),
                    ),
                ],
            ),
        )

        if results:
            hit = results[0]
            cached_response = json.loads(hit.payload["response"])

            # Update hit count
            self.qdrant.set_payload(
                collection_name="semantic_cache",
                payload={"hit_count": hit.payload.get("hit_count", 0) + 1},
                points=[hit.id],
            )

            return {
                **cached_response,
                "_cache": {
                    "hit": True,
                    "similarity": hit.score,
                    "original_query": hit.payload["query"],
                },
            }

        return None

    def set(
        self,
        messages: list[dict],
        model: str,
        response: dict,
    ) -> None:
        query = self._extract_query(messages)
        if not query:
            return

        query_embedding = self._get_embedding(query)

        import uuid
        from qdrant_client import models

        self.qdrant.upsert(
            collection_name="semantic_cache",
            points=[
                models.PointStruct(
                    id=str(uuid.uuid4()),
                    vector=query_embedding,
                    payload={
                        "query": query,
                        "response": json.dumps(response, ensure_ascii=False),
                        "model": model,
                        "created_at": time.time(),
                        "hit_count": 0,
                    },
                ),
            ],
        )

Provider Prompt Caching

Anthropic Prompt Caching

Anthropic 的 Prompt Caching 在 API 层面缓存长 System Prompt 的 KV-Cache，避免重复计算：

# Anthropic Prompt Caching
import anthropic

client = anthropic.Anthropic()

# Long system prompt (cached across requests)
system_prompt = """You are a customer service agent for AcmeCorp.
You have access to the following knowledge base:
[... 5000 words of product documentation ...]
[... 3000 words of return policy ...]
[... 2000 words of FAQ ...]
"""

response = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    system=[
        {
            "type": "text",
            "text": system_prompt,
            "cache_control": {"type": "ephemeral"},  # Enable caching
        },
    ],
    messages=[
        {"role": "user", "content": "How do I return a product?"},
    ],
)

# Check cache usage in response
print(f"Input tokens: {response.usage.input_tokens}")
print(f"Cache read tokens: {response.usage.cache_read_input_tokens}")
print(f"Cache creation tokens: {response.usage.cache_creation_input_tokens}")

# First request: cache_creation > 0, cache_read = 0
# Subsequent requests: cache_creation = 0, cache_read > 0 (90% cheaper)

OpenAI Prompt Caching

OpenAI 自动缓存长前缀，无需额外 API 参数：

import openai

client = openai.OpenAI()

# OpenAI automatically caches prefixes >= 1024 tokens
# that are identical across requests
long_system = "..." # 2000+ tokens of instructions

# Request 1: Full price
response1 = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": long_system},
        {"role": "user", "content": "Question 1"},
    ],
)

# Request 2: Same prefix, 50% discount on cached tokens
response2 = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": long_system},  # Same prefix
        {"role": "user", "content": "Question 2"},    # Different suffix
    ],
)

# Check cached tokens
print(response2.usage.prompt_tokens_details.cached_tokens)

Prompt Cache 优化技巧

最大化缓存命中的 Prompt 结构：

┌─────────────────────────────────────────┐
│ System Prompt (不变部分, 长且稳定)         │ ← 缓存命中
│   - 角色定义                              │
│   - 知识库内容                            │
│   - 输出格式规范                          │
│   - 工具定义                              │
├─────────────────────────────────────────┤
│ Few-shot Examples (不变部分)              │ ← 缓存命中
│   - 示例 1                               │
│   - 示例 2                               │
│   - 示例 3                               │
├─────────────────────────────────────────┤
│ Dynamic Context (会变部分)               │ ← 不缓存
│   - 检索到的文档                          │
│   - 当前会话历史                          │
│   - 用户输入                              │
└─────────────────────────────────────────┘

原则: 不变的放前面 (被缓存), 变化的放后面

缓存失效策略

多维度失效

# src/cache/invalidation.py
from enum import Enum
from typing import Optional
from datetime import datetime, timedelta

class InvalidationReason(Enum):
    TTL_EXPIRED = "ttl_expired"
    MODEL_UPDATED = "model_updated"
    KNOWLEDGE_UPDATED = "knowledge_updated"
    LOW_QUALITY = "low_quality"
    MANUAL = "manual"

class CacheInvalidator:
    def __init__(self, cache: SemanticCache):
        self.cache = cache

    def invalidate_by_ttl(self, max_age: timedelta) -> int:
        """Remove entries older than max_age."""
        cutoff = (datetime.now() - max_age).timestamp()
        # Delete old entries from Qdrant
        deleted = self.cache.qdrant.delete(
            collection_name="semantic_cache",
            points_selector=models.FilterSelector(
                filter=models.Filter(
                    must=[
                        models.FieldCondition(
                            key="created_at",
                            range=models.Range(lt=cutoff),
                        ),
                    ],
                ),
            ),
        )
        return deleted.operation_id

    def invalidate_by_model(self, model: str) -> int:
        """Clear cache when model is updated."""
        deleted = self.cache.qdrant.delete(
            collection_name="semantic_cache",
            points_selector=models.FilterSelector(
                filter=models.Filter(
                    must=[
                        models.FieldCondition(
                            key="model",
                            match=models.MatchValue(value=model),
                        ),
                    ],
                ),
            ),
        )
        return deleted.operation_id

    def invalidate_by_quality(self, min_hit_count: int = 0, max_age_days: int = 30) -> int:
        """Remove low-engagement entries (never hit, old)."""
        cutoff = (datetime.now() - timedelta(days=max_age_days)).timestamp()
        deleted = self.cache.qdrant.delete(
            collection_name="semantic_cache",
            points_selector=models.FilterSelector(
                filter=models.Filter(
                    must=[
                        models.FieldCondition(
                            key="hit_count",
                            range=models.Range(lte=min_hit_count),
                        ),
                        models.FieldCondition(
                            key="created_at",
                            range=models.Range(lt=cutoff),
                        ),
                    ],
                ),
            ),
        )
        return deleted.operation_id

监控指标

指标	计算方式	目标值	告警阈值
精确缓存命中率	hits / total	>10%	<5%
语义缓存命中率	hits / total	>25%	<15%
Prompt Cache 命中率	cached_tokens / total_input	>60%	<30%
缓存延迟 (P99)	从查询到返回	<50ms	>200ms
缓存节省成本	hit_cost_savings / total_cost	>20%	<10%
缓存错误率	stale_or_wrong / hits	<1%	>5%

总结

三层缓存互补：精确匹配处理确定性查询，语义缓存处理自然语言变体，Prompt Cache 处理公共前缀。
语义缓存的阈值需要调优：太低会返回不相关的结果，太高会降低命中率。0.90-0.95 是通常的安全区间。
Prompt Caching 是最低成本优化：只需要调整 Prompt 结构（不变的放前面），就能获得 50-90% 的输入 Token 折扣。
缓存失效比缓存命中更重要：错误的缓存比没有缓存更危险，多维度失效策略是必需的。
monitoring 驱动优化：持续监控命中率和节省成本，根据数据调整阈值和策略。

Maurice | [email protected]

深度加工（NotebookLM 生成）

基于本文内容生成的 PPT 大纲、博客摘要、短视频脚本与 Deep Dive 播客，用于多场景复用

PPT 大纲（5-8 张幻灯片）点击展开

AI 缓存策略：Semantic Cache 与 Prompt Cache — ppt

这是一份基于您上传的文章为您整理的 AI 缓存策略 PPT 大纲，共 7 张幻灯片。

幻灯片 1：引言与背景：为什么需要 AI 缓存？

核心痛点：LLM 推理面临着成本高、延迟大的问题，但在实际业务中，存在大量重复或高度相似的请求（如典型的客服聊天机器人中 30-50% 的问题本质是相同的）[1]。
传统 API 缓存的局限：传统缓存无法识别自然语言的“语义等价”，例如“怎么退货”和“我要退商品”字符串不同但语义完全一致 [1]。
缓存的价值：将重复请求的结果缓存起来，不仅能显著降低业务成本，还能将响应延迟从秒级大幅降到毫秒级 [1]。
解决方案：引入针对大模型的特殊缓存机制，包括精确缓存、语义缓存（Semantic Cache）和 Prompt Cache [1]。

幻灯片 2：AI 缓存金字塔全景

三层缓存体系：现代 AI 缓存架构是一套互补的三层金字塔结构 [1]。
第一层（应用端）：Exact Match Cache（精确字符串匹配），通过 Hash 查找，延迟 <5ms [1]。
第二层（应用端）：Semantic Cache（语义相似度缓存），基于 Embedding 匹配，延迟 <50ms [1]。
第三层（Provider 端 API 级）：Prompt Cache，基于前缀复用，可降低约 50% 的首 Token 延迟 [1]。

幻灯片 3：精确匹配缓存 (Exact Match Cache)

实现原理：通过将用户的完整请求输入（如 messages、模型版本、temperature）进行确定性的字符串 Hash 计算来比对 [1, 2]。
性能表现：命中率相对较低（5-15%），但完全没有额外的计算成本 [1]。
核心优势：一旦命中，可以实现 100% 的延迟减少和 100% 的成本减少 [1]。
适用场景：最适合处理固定模板查询以及系统中完全一致的高频重复请求 [1]。

幻灯片 4：语义缓存 (Semantic Cache)

核心机制：将用户输入转化为 Embedding 向量（如 [0.12, -0.34, 0.56...]），然后在向量库中搜索相似度最高的历史缓存条目 [2, 3]。
阈值控制：当检索结果的相似度大于设定的阈值（通常的安全区间是 0.90-0.95）时判定为命中，直接返回缓存结果 [2, 4]。
成本与命中率：命中率处于中等水平（20-40%），存在一定的 Embedding 计算额外成本 [1]。
适用场景：专门用于处理自然语言查询和用户表述中的多样性变体 [1]。

幻灯片 5：Prompt Cache (Provider 端缓存)

缓存原理：在服务商的 API 层面自动缓存长 System Prompt 的 KV-Cache，避免对相同的上下文前缀进行重复计算 [5]。
平台支持：Anthropic 通过 ephemeral 参数支持，而 OpenAI 则会针对大于 1024 tokens 的一致前缀自动进行缓存和打折 [5, 6]。
结构优化原则：最大化命中的关键是**“不变的放前面，变化的放后面”** [7]。
结构拆分示例：将系统角色设定、知识库和 Few-shot 示例等长且稳定的内容置于头部（可缓存），检索到的动态文档和用户当前输入置于尾部（不缓存） [7]。

幻灯片 6：多维度缓存失效策略 (Invalidation)

核心理念：错误的缓存比没有缓存更危险，因此需要建立完善的多维度失效策略 [4]。
时间失效 (TTL)：根据设定的生存周期（如默认 3600 秒或更长），自动清理过期的缓存条目 [2, 8]。
模型与知识更新失效：当底层大模型发生升级，或系统的知识库发生变更时，需定向清除对应缓存 [4, 8]。
质量失效：自动移除那些低参与度（从未被命中过）且老旧的低质量缓存数据，保障向量库的效率 [4]。

幻灯片 7：监控指标与核心总结

关键监控指标：需持续监控精确/语义缓存命中率、Prompt Cache 命中率（目标 >60%）、P99 缓存延迟（目标 <50ms）及缓存错误率等 [4]。
三层机制互补：精确匹配处理确定性查询，语义缓存处理自然语言变体，Prompt Cache 处理公共前缀 [4]。
阈值调优：语义缓存的相似度阈值需要基于监控数据持续调优，太低会导致不相关回复，太高会降低命中率 [4]。
最低成本优化：Prompt Caching 是最低成本的优化手段，只需调整 Prompt 结构，无需复杂开发即可获得 50-90% 的输入 Token 成本折扣 [4]。

博客摘要 + 核心看点点击展开

AI 缓存策略：Semantic Cache 与 Prompt Cache — summary

SEO 友好博客摘要

大语言模型（LLM）推理成本高且延迟大，如何有效优化？本文全面解析 AI 缓存策略的三层核心架构：精确匹配缓存、语义缓存（Semantic Cache）与 Prompt Cache [1]。通过在应用端使用 Embedding 相似度匹配解决自然语言变体问题，在模型端复用公共前缀，开发者能显著降低开销，并将延迟从秒级降至毫秒级 [1, 2]。文章深入对比了三种缓存的命中率与实现成本，并探讨了 Prompt 结构优化技巧及多维度缓存失效机制 [1, 3, 4]。掌握这些策略，助你构建低成本、高效率的 LLM 应用。

3 条核心看点