一、问题：同样的答案，你付费了 1000 次

用户 A: "Python 怎么读取 CSV 文件？"    → LLM → $0.0003
用户 B: "How to read CSV in Python?"     → LLM → $0.0003
用户 C: "Python 读取csv文件的方法"        → LLM → $0.0003
用户 D: "python read csv file example"   → LLM → $0.0003

四个用户问了本质相同的问题，LLM 答了 4 次，你付了 4 次钱。如果有 100 万用户，30% 问题语义相似——每年多花 10 万美元。

LLM 缓存的特殊挑战：

• Redis 精确匹配只能命中完全相同的字符串
• 同义改写（“读CSV” vs “read csv”）精确匹配会 Miss
• 温度参数导致相同 prompt 可能不同输出

解决方案：精确缓存 + 语义缓存。

---

二、两级缓存架构

              ┌──────────────────────────┐
              │      Cache Manager       │
              │                          │
   Request ──►│  ┌────────────────────┐  │
              │  │ L1: Exact Cache    │  │  命中率 ~15%
              │  │  · MD5 hash        │  │  延迟 <1ms
              │  │  · 本地 LRU        │  │
              │  └───────┬────────────┘  │
              │          │ Miss          │
              │  ┌───────▼────────────┐  │
              │  │ L2: Semantic Cache │  │  命中率 ~25%
              │  │  · Embedding sim   │  │  延迟 ~10ms
              │  │  · Redis + FAISS   │  │
              │  └───────┬────────────┘  │
              │          │ Miss          │
              │  ┌───────▼────────────┐  │
              │  │ L3: LLM API        │  │  回源
              │  └────────────────────┘  │
              └──────────────────────────┘

---

三、完整实现

# semantic_cache.py - 两级 LLM 语义缓存
# pip install openai numpy sentence-transformers cachetools

import hashlib
import json
import time
import numpy as np
from typing import Optional, Dict, Any, List, Tuple
from dataclasses import dataclass, field
from cachetools import LRUCache
from openai import OpenAI


# ============================================================
# 1. 精确缓存 (L1)
# ============================================================

@dataclass
class CacheEntry:
    # 缓存条目
    query: str
    response: str
    usage: Dict[str, int]
    cost_saved: float
    created_at: float = field(default_factory=time.time)
    hit_count: int = 0


class ExactCache:
    # L1: 基于 MD5 的精确匹配缓存

    def __init__(self, max_size: int = 10000):
        self._cache = LRUCache(maxsize=max_size)

    def _key(self, messages: List[Dict], model: str,
             temperature: float) -> str:
        raw = json.dumps({
            "messages": messages,
            "model": model,
            "temperature": temperature,
        }, sort_keys=True, ensure_ascii=False)
        return hashlib.md5(raw.encode()).hexdigest()

    def get(self, messages: List[Dict], model: str,
            temperature: float = 0.0) -> Optional[CacheEntry]:
        key = self._key(messages, model, temperature)
        entry = self._cache.get(key)
        if entry:
            entry.hit_count += 1
        return entry

    def set(self, messages: List[Dict], model: str,
            temperature: float, response: str,
            usage: Dict, cost: float):
        key = self._key(messages, model, temperature)
        entry = CacheEntry(
            query=messages[-1]["content"] if messages else "",
            response=response, usage=usage, cost_saved=cost,
        )
        self._cache[key] = entry

    def stats(self) -> dict:
        total_hits = sum(e.hit_count for e in self._cache.values())
        return {
            "size": len(self._cache),
            "max_size": self._cache.maxsize,
            "total_hits": total_hits,
        }


# ============================================================
# 2. 语义缓存 (L2)
# ============================================================

class SemanticCache:
    # L2: 基于 Embedding 余弦相似度的语义缓存
    # 可选: 用本地 sentence-transformers 代替 OpenAI Embedding API

    def __init__(self, embedding_model: str = "text-embedding-3-small",
                 similarity_threshold: float = 0.92,
                 max_size: int = 50000,
                 use_local: bool = False):
        self.similarity_threshold = similarity_threshold
        self.embedding_model = embedding_model
        self.max_size = max_size
        self.use_local = use_local

        # 本地向量存储
        self._embeddings: List[np.ndarray] = []
        self._entries: List[CacheEntry] = []

        # 本地模型 (更快更便宜)
        self._local_model = None
        if use_local:
            try:
                from sentence_transformers import SentenceTransformer
                self._local_model = SentenceTransformer("all-MiniLM-L6-v2")
            except ImportError:
                pass

        self._openai = None

    def _get_client(self):
        if self._openai is None:
            self._openai = OpenAI()
        return self._openai

    def _get_embedding(self, text: str) -> np.ndarray:
        # 本地模型优先
        if self._local_model:
            emb = self._local_model.encode(
                text, normalize_embeddings=True
            )
            return np.array(emb)

        # 否则用 OpenAI API
        client = self._get_client()
        resp = client.embeddings.create(
            model=self.embedding_model, input=text,
        )
        emb = np.array(resp.data[0].embedding)
        return emb / np.linalg.norm(emb)

    def search(self, query: str) -> Optional[CacheEntry]:
        if not self._embeddings:
            return None

        query_emb = self._get_embedding(query)
        similarities = np.dot(np.array(self._embeddings), query_emb)

        best_idx = int(np.argmax(similarities))
        best_score = float(similarities[best_idx])

        if best_score >= self.similarity_threshold:
            entry = self._entries[best_idx]
            entry.hit_count += 1
            return entry
        return None

    def add(self, query: str, response: str,
            usage: Dict, cost: float):
        emb = self._get_embedding(query)

        if len(self._embeddings) >= self.max_size:
            # FIFO 淘汰
            self._embeddings.pop(0)
            self._entries.pop(0)

        self._embeddings.append(emb)
        self._entries.append(CacheEntry(
            query=query, response=response,
            usage=usage, cost_saved=cost,
        ))

    def stats(self) -> dict:
        total_hits = sum(e.hit_count for e in self._entries)
        return {
            "size": len(self._entries),
            "total_hits": total_hits,
        }


# ============================================================
# 3. 两级缓存管理器
# ============================================================

@dataclass
class CacheStats:
    l1_hits: int = 0
    l2_hits: int = 0
    misses: int = 0
    total_cost_saved: float = 0.0

    @property
    def total_requests(self) -> int:
        return self.l1_hits + self.l2_hits + self.misses

    @property
    def hit_rate(self) -> float:
        if self.total_requests == 0:
            return 0.0
        return (self.l1_hits + self.l2_hits) / self.total_requests

    def summary(self) -> str:
        return (
            f"Requests: {self.total_requests} | "
            f"Hit Rate: {self.hit_rate:.1%} | "
            f"L1: {self.l1_hits} L2: {self.l2_hits} Miss: {self.misses} | "
            f"Cost Saved: ${self.total_cost_saved:.4f}"
        )


class CachedLLM:
    # 带两级缓存的 LLM 客户端

    def __init__(self, openai_client: OpenAI,
                 exact_cache: ExactCache = None,
                 semantic_cache: SemanticCache = None,
                 auto_cache: bool = True):
        self.client = openai_client
        self.exact = exact_cache or ExactCache()
        self.semantic = semantic_cache or SemanticCache()
        self.auto_cache = auto_cache
        self.stats = CacheStats()

    def chat(self, messages: List[Dict[str, str]],
             model: str = "gpt-4o-mini",
             temperature: float = 0.0,
             max_tokens: int = 4096,
             enable_semantic: bool = True,
             **kwargs) -> Tuple[str, Dict]:
        # 返回 (response_text, usage_dict)

        # 提取最后一条 user message
        user_query = ""
        for m in reversed(messages):
            if m["role"] == "user":
                user_query = m["content"]
                break

        # L1: 精确缓存
        cached = self.exact.get(messages, model, temperature)
        if cached:
            self.stats.l1_hits += 1
            self.stats.total_cost_saved += cached.cost_saved
            return cached.response, cached.usage

        # L2: 语义缓存
        if enable_semantic and user_query:
            cached = self.semantic.search(user_query)
            if cached:
                self.stats.l2_hits += 1
                self.stats.total_cost_saved += cached.cost_saved
                return cached.response, cached.usage

        # L3: 调用 LLM
        self.stats.misses += 1

        resp = self.client.chat.completions.create(
            model=model, messages=messages,
            temperature=temperature, max_tokens=max_tokens, **kwargs,
        )

        response_text = resp.choices[0].message.content
        usage = {
            "prompt_tokens": resp.usage.prompt_tokens if resp.usage else 0,
            "completion_tokens": resp.usage.completion_tokens if resp.usage else 0,
            "total_tokens": resp.usage.total_tokens if resp.usage else 0,
        }

        # 估算成本
        rates = {"gpt-4o-mini": (0.15, 0.60)}  # per 1M tokens
        input_rate, output_rate = rates.get(model, (0.15, 0.60))
        cost = (
            usage["prompt_tokens"] / 1_000_000 * input_rate
            + usage["completion_tokens"] / 1_000_000 * output_rate
        )

        # 写入缓存
        if self.auto_cache:
            self.exact.set(messages, model, temperature,
                          response_text, usage, cost)
            if enable_semantic and user_query:
                self.semantic.add(user_query, response_text, usage, cost)

        return response_text, usage


# ============================================================
# 4. 基准测试
# ============================================================

if __name__ == "__main__":
    print("=" * 60)
    print("语义缓存基准测试 (模拟)")
    print("=" * 60)

    cached_llm = CachedLLM(
        openai_client=OpenAI(api_key="sk-fake"),
        exact_cache=ExactCache(max_size=1000),
        semantic_cache=SemanticCache(similarity_threshold=0.85),
        auto_cache=False,
    )

    queries = [
        "Python 如何读取 CSV 文件?",
        "How to read CSV file in Python?",
        "Python 读取csv文件的方法",
        "python read csv example",
        "read csv using pandas python",
        "Java 怎么读取 CSV?",
        "Python 如何写入 JSON 文件?",
        "python csv read tutorial",
        "how to parse csv in python",
        "What is the meaning of life?",
    ]

    fake_response = (
        "import csv\n"
        "with open('file.csv') as f:\n"
        "    reader = csv.reader(f)"
    )
    fake_usage = {"prompt_tokens": 50, "completion_tokens": 30, "total_tokens": 80}
    fake_cost = 50/1_000_000*0.15 + 30/1_000_000*0.60

    # 预热
    cached_llm.exact.set(
        [{"role": "user", "content": queries[0]}],
        "gpt-4o-mini", 0.0,
        fake_response, fake_usage, fake_cost,
    )
    print(f"\n已预热精确缓存: '{queries[0]}'")

    for i, q in enumerate(queries):
        print(f"\n--- Query {i+1}: '{q}' ---")
        l1 = cached_llm.exact.get(
            [{"role": "user", "content": q}], "gpt-4o-mini", 0.0
        )
        if l1:
            print("  L1 HIT! (exact)")
            cached_llm.stats.l1_hits += 1
            cached_llm.stats.total_cost_saved += l1.cost_saved
            continue

        cached_llm.stats.misses += 1
        if q in queries[:5] or q in queries[7:9]:
            print("  L2 WOULD HIT (similarity > 0.85)")
        else:
            print("  L2 miss, would call LLM")

    print(f"\n{'=' * 60}")
    print("统计汇总:")
    print(f"  10 条查询, 约 6 条可从缓存命中")
    print(f"  理论节省: ~60% LLM 调用")

---

四、缓存策略配置

# 生产环境推荐配置
class CacheConfig:
    # L1 精确缓存: temperature=0 的确定性任务
    L1_MAX_SIZE = 10000         # 本地 LRU, 内存 ~50MB

    # L2 语义缓存
    L2_SIMILARITY_THRESHOLD = 0.92  # 推荐区间 0.90-0.95
    L2_MAX_SIZE = 100000           # Redis + FAISS

    # 跳过缓存的情况
    SKIP_IF_TEMPERATURE_GT = 0.3   # temp > 0.3 不缓存
    SKIP_FOR_TOOL_CALLS = True      # tool call 不缓存
    SKIP_FOR_STREAMING = True       # 流式不缓存

相似度阈值选择指南：

阈值	命中率	准确率	适用场景
0.98	~5%	极高	金融/医疗零容忍
0.92	~20%	高	通用生产 (推荐)
0.85	~35%	中	客服/FAQ 容错场景
0.75	~50%	低	不推荐

---

五、成本收益分析

日均 100 万次请求, 每次 $0.0003:

缓存层	命中率	日节省	年节省
仅 L1 (精确)	~15%	$45	$16,425
L1 + L2 (语义)	~40%	$120	$43,800
L1 + L2 + 优化	~55%	$165	$60,225

额外成本:

• Embedding API: ~$0.02/1M tokens → 约 $10-20/月
• 本地模型 (all-MiniLM-L6-v2): 免费, CPU 即可
• Redis 内存: ~2GB (10 万条向量)

ROI: 额外成本 < $50/月, 年节省 $4-6 万。

---

六、生产化注意事项

1. 缓存一致性 — 模型升级后清空语义缓存 (embedding 可能变化)
2. 温度参数 — temperature > 0 不缓存 (输出随机)
3. Tool calls — 函数调用不缓存 (结果可能过时)
4. 多租户 — key = tenant_id:hash 隔离
5. 监控 — 命中率 / 节省成本 / 相似度分布面板

---

七、总结

两级语义缓存：

1. L1 精确缓存 — MD5 hash, <1ms, 命中率 ~15%
2. L2 语义缓存 — Embedding cosine sim, ~10ms, 命中率 ~25%
3. 组合命中率 ~40% — 年省 $4 万+ (百万日活)