LLM 缓存策略：语义缓存与多级缓存架构

from dataclasses import dataclass
import numpy as np

@dataclass
class CacheEntry:
    query: str
    query_embedding: np.ndarray
    response: str
    model: str
    tokens_used: int
    timestamp: float
    hit_count: int = 0
    ttl_seconds: int = 3600

class SemanticCache:
    """
    语义缓存：基于向量相似度的缓存命中
    核心思想：如果新查询与已缓存查询的语义距离 < 阈值，则复用缓存结果
    """
    def __init__(self, embedding_model, similarity_threshold: float = 0.95):
        self.embedding_model = embedding_model
        self.threshold = similarity_threshold
        self.cache: list[CacheEntry] = []
        self.index = None  # 可选：FAISS/Annoy 加速

    async def get(self, query: str) -> tuple[bool, str | None]:
        query_embedding = await self.embedding_model.embed(query)
        # 线性搜索（生产环境用向量索引）
        for entry in self.cache:
            similarity = self._cosine_similarity(query_embedding, entry.query_embedding)
            if similarity >= self.threshold:
                entry.hit_count += 1
                return True, entry.response
        return False, None

    async def set(self, query: str, response: str, model: str, tokens: int):
        embedding = await self.embedding_model.embed(query)
        entry = CacheEntry(
            query=query,
            query_embedding=embedding,
            response=response,
            model=model,
            tokens_used=tokens,
            timestamp=time.time()
        )
        self.cache.append(entry)

    def _cosine_similarity(self, v1: np.ndarray, v2: np.ndarray) -> float:
        return float(np.dot(v1, v2) / (np.linalg.norm(v1) * np.linalg.norm(v2)))

import matplotlib.pyplot as plt

class ThresholdOptimizer:
    """语义缓存阈值优化器"""
    def __init__(self, cache: SemanticCache, test_data: list[tuple[str, str, bool]]):
        """
        test_data: [(query1, cached_query, should_hit), ...]
        """
        self.cache = cache
        self.test_data = test_data

    def evaluate(self, threshold: float) -> dict:
        self.cache.threshold = threshold
        true_positives = 0
        false_positives = 0
        false_negatives = 0
        true_negatives = 0

        for query, cached, should_hit in self.test_data:
            # 手动计算相似度
            e1 = await self.cache.embedding_model.embed(query)
            e2 = await self.cache.embedding_model.embed(cached)
            sim = self.cache._cosine_similarity(e1, e2)
            is_hit = sim >= threshold

            if should_hit and is_hit:
                true_positives += 1
            elif not should_hit and is_hit:
                false_positives += 1
            elif should_hit and not is_hit:
                false_negatives += 1
            else:
                true_negatives += 1

        precision = true_positives / (true_positives + false_positives) if (true_positives + false_positives) > 0 else 0
        recall = true_positives / (true_positives + false_negatives) if (true_positives + false_negatives) > 0 else 0
        f1 = 2 * precision * recall / (precision + recall) if (precision + recall) > 0 else 0

        return {"threshold": threshold, "precision": precision, "recall": recall, "f1": f1}

    def find_optimal_threshold(self) -> float:
        thresholds = np.arange(0.80, 1.0, 0.01)
        results = [self.evaluate(t) for t in thresholds]
        best = max(results, key=lambda x: x["f1"])
        return best["threshold"]

┌─────────────────────────────────────────────────┐
│                  LLM Request                     │
└─────────────────────┬───────────────────────────┘
                      │
                      ▼
            ┌─────────────────┐
            │   L1: 精确缓存   │  (Redis: exact query hash)
            │   命中率: 40%    │
            │   延迟: 1ms     │
            └────────┬────────┘
                     │ Miss
                     ▼
            ┌─────────────────┐
            │   L2: 语义缓存   │  (Redis + Vector Index)
            │   命中率: 30%    │
            │   延迟: 10-50ms  │
            └────────┬────────┘
                     │ Miss
                     ▼
            ┌─────────────────┐
            │   L3: 热模型缓存 │  (本地推理: small model)
            │   命中率: 15%    │
            │   延迟: 100ms    │
            └────────┬────────┘
                     │ Miss
                     ▼
            ┌─────────────────┐
            │   L4: 远程LLM    │  (OpenAI/Anthropic)
            │   命中率: 15%    │
            │   延迟: 500-3000ms│
            └─────────────────┘

import hashlib
import redis.asyncio as redis

class ExactCache:
    """精确匹配缓存：查询文本的确定性哈希"""
    def __init__(self, redis_client: redis.Redis, prefix: str = "llm:exact"):
        self.redis = redis_client
        self.prefix = prefix
        self.default_ttl = 3600

    def _hash(self, query: str, model: str, params: dict) -> str:
        """确定性哈希：相同输入相同哈希"""
        canonical = f"{model}|{query}|{json.dumps(params, sort_keys=True)}"
        return hashlib.sha256(canonical.encode()).hexdigest()[:32]

    async def get(self, query: str, model: str, params: dict) -> str | None:
        key = f"{self.prefix}:{self._hash(query, model, params)}"
        return await self.redis.get(key)

    async def set(self, query: str, model: str, params: dict, response: str,
                  ttl: int = None, tokens: int = 0):
        key = f"{self.prefix}:{self._hash(query, model, params)}"
        value = json.dumps({"response": response, "tokens": tokens, "timestamp": time.time()})
        await self.redis.setex(key, ttl or self.default_ttl, value)

class RedisSemanticCache:
    """Redis 存储向量，FAISS 加速检索"""
    def __init__(self, redis_client: redis.Redis, embedding_model):
        self.redis = redis_client
        self.embedding_model = embedding_model
        self.prefix = "llm:semantic"
        self.threshold = 0.92
        # FAISS 索引（内存中）
        self.index = None
        self.id_to_key: list[str] = []

    async def build_index(self):
        """从 Redis 重建向量索引"""
        keys = await self.redis.keys(f"{self.prefix}:*")
        embeddings = []
        self.id_to_key = []

        for key in keys:
            data = await self.redis.get(key)
            if data:
                entry = json.loads(data)
                embeddings.append(entry["embedding"])
                self.id_to_key.append(key)

        if embeddings:
            embeddings = np.array(embeddings, dtype=np.float32)
            self.index = faiss.IndexFlatIP(embeddings.shape[1])
            faiss.normalize_L2(embeddings)
            self.index.add(embeddings)

    async def get(self, query: str) -> tuple[bool, str | None]:
        query_embedding = await self.embedding_model.embed(query)
        query_embedding = np.array([query_embedding], dtype=np.float32)
        faiss.normalize_L2(query_embedding)

        if self.index is None or self.index.ntotal == 0:
            return False, None

        D, I = self.index.search(query_embedding, k=5)
        for dist, idx in zip(D[0], I[0]):
            if dist >= self.threshold and idx < len(self.id_to_key):
                key = self.id_to_key[idx]
                data = await self.redis.get(key)
                if data:
                    entry = json.loads(data)
                    entry["hit_count"] = entry.get("hit_count", 0) + 1
                    await self.redis.set(key, json.dumps(entry))
                    return True, entry["response"]

        return False, None

    async def set(self, query: str, response: str, model: str):
        embedding = await self.embedding_model.embed(query)
        key = f"{self.prefix}:{hashlib.md5(query.encode()).hexdigest()}"
        value = {
            "query": query,
            "response": response,
            "model": model,
            "embedding": embedding.tolist(),
            "timestamp": time.time(),
            "hit_count": 0
        }
        await self.redis.setex(key, 86400, json.dumps(value))
        # 更新索引
        if self.index is not None:
            emb = np.array([embedding], dtype=np.float32)
            faiss.normalize_L2(emb)
            self.index.add(emb)
            self.id_to_key.append(key)

class LocalModelCache:
    """
    本地小模型缓存：对于简单查询，用本地模型快速响应
    适用场景：FAQ、简单问答、格式化任务
    """
    def __init__(self, local_model_path: str, complexity_classifier):
        self.model = self._load_local_model(local_model_path)
        self.classifier = complexity_classifier

    def _load_local_model(self, path):
        from transformers import AutoModelForCausalLM, AutoTokenizer
        tokenizer = AutoTokenizer.from_pretrained(path)
        model = AutoModelForCausalLM.from_pretrained(path, device_map="auto")
        return {"tokenizer": tokenizer, "model": model}

    async def can_handle(self, query: str) -> bool:
        """判断查询复杂度是否适合本地模型"""
        complexity = await self.classifier.predict(query)
        return complexity in ["simple", "faq", "formatting"]

    async def generate(self, query: str, max_tokens: int = 512) -> str:
        inputs = self.model["tokenizer"](query, return_tensors="pt").to("cuda")
        outputs = self.model["model"].generate(**inputs, max_new_tokens=max_tokens)
        return self.model["tokenizer"].decode(outputs[0], skip_special_tokens=True)

class MultiLevelCacheCoordinator:
    """多级缓存协调器：按顺序查询各级缓存"""
    def __init__(self, l1: ExactCache, l2: RedisSemanticCache,
                 l3: LocalModelCache, remote_llm):
        self.l1 = l1
        self.l2 = l2
        self.l3 = l3
        self.llm = remote_llm
        self.stats = {"l1_hits": 0, "l2_hits": 0, "l3_hits": 0, "llm_calls": 0}

    async def query(self, request: LLMRequest) -> LLMResponse:
        # L1: 精确缓存
        cached = await self.l1.get(request.query, request.model, request.params)
        if cached:
            self.stats["l1_hits"] += 1
            return LLMResponse(content=cached, source="cache_l1")

        # L2: 语义缓存
        hit, response = await self.l2.get(request.query)
        if hit:
            self.stats["l2_hits"] += 1
            # 回填 L1
            await self.l1.set(request.query, request.model, request.params, response)
            return LLMResponse(content=response, source="cache_l2")

        # L3: 本地模型
        if await self.l3.can_handle(request.query):
            self.stats["l3_hits"] += 1
            response = await self.l3.generate(request.query)
            # 回填 L1, L2
            await self._backfill(request, response)
            return LLMResponse(content=response, source="cache_l3")

        # L4: 远程 LLM
        self.stats["llm_calls"] += 1
        response = await self.llm.chat(request)
        await self._backfill(request, response.content)
        return LLMResponse(content=response.content, source="remote_llm")

    async def _backfill(self, request: LLMRequest, response: str):
        await asyncio.gather(
            self.l1.set(request.query, request.model, request.params, response),
            self.l2.set(request.query, response, request.model)
        )

    def get_stats(self) -> dict:
        total = sum(self.stats.values())
        return {
            **self.stats,
            "l1_hit_rate": self.stats["l1_hits"] / total if total > 0 else 0,
            "l2_hit_rate": self.stats["l2_hits"] / total if total > 0 else 0,
            "l3_hit_rate": self.stats["l3_hits"] / total if total > 0 else 0,
            "total_requests": total,
        }

class SmartTTL:
    """智能 TTL：根据内容类型和更新频率动态调整"""
    TTL_RULES = {
        "fact": 86400 * 30,      # 事实类：30天
        "news": 3600,            # 新闻类：1小时
        "code": 86400 * 7,      # 代码类：7天
        "chat": 300,            # 对话类：5分钟
        "weather": 600,          # 天气：10分钟
        "stock": 10,            # 股票：10秒
    }

    async def get_ttl(self, query: str, classifier) -> int:
        query_type = await classifier.classify(query)
        return self.TTL_RULES.get(query_type, 3600)

class SlidingWindowTTL:
    """滑动窗口 TTL：被频繁访问的缓存条目自动延长 TTL"""
    def __init__(self, base_ttl: int = 3600, extend_factor: float = 1.5,
                 max_ttl: int = 86400):
        self.base_ttl = base_ttl
        self.extend_factor = extend_factor
        self.max_ttl = max_ttl

    async def on_hit(self, key: str, redis: redis.Redis):
        current_ttl = await redis.ttl(key)
        if current_ttl > 0:
            new_ttl = min(int(current_ttl * self.extend_factor), self.max_ttl)
            await redis.expire(key, new_ttl)

class CacheInvalidator:
    """主动失效机制"""
    def __init__(self, redis: redis.Redis):
        self.redis = redis
        self.pubsub = redis.pubsub()

    async def subscribe_invalidations(self):
        await self.pubsub.subscribe("cache:invalidate")
        async for message in self.pubsub.listen():
            if message["type"] == "message":
                data = json.loads(message["data"])
                await self._handle_invalidation(data)

    async def _handle_invalidation(self, data: dict):
        pattern = data.get("pattern")
        if pattern:
            keys = await self.redis.keys(pattern)
            if keys:
                await self.redis.delete(*keys)

    async def invalidate_by_model(self, model: str):
        """模型更新时失效所有相关缓存"""
        await self.redis.publish("cache:invalidate", json.dumps({
            "pattern": f"llm:*:{model}:*"
        }))

    async def invalidate_by_source(self, source_id: str):
        """数据源更新时失效相关缓存"""
        await self.redis.publish("cache:invalidate", json.dumps({
            "pattern": f"llm:*:source:{source_id}:*"
        }))

class CacheWarmup:
    """缓存预热：系统启动时加载热点数据"""
    def __init__(self, coordinator: MultiLevelCacheCoordinator):
        self.coordinator = coordinator

    async def warmup(self, hot_queries_file: str = "hot_queries.json"):
        """从历史日志中提取热点查询，提前缓存"""
        with open(hot_queries_file) as f:
            hot_queries = json.load(f)

        tasks = []
        for item in hot_queries[:100]:  # 预热 top 100
            request = LLMRequest(
                query=item["query"],
                model=item["model"],
                params=item.get("params", {})
            )
            # 查询 LLM 并缓存结果
            tasks.append(self.coordinator.query(request))

        await asyncio.gather(*tasks, return_exceptions=True)
        print(f"Warmed up {len(tasks)} queries")

    async def periodic_refresh(self, interval: int = 3600):
        """定期刷新热点缓存"""
        while True:
            await asyncio.sleep(interval)
            hot_keys = await self.coordinator.l1.redis.zrevrange(
                "cache:hot_queries", 0, 50
            )
            for key in hot_keys:
                # 重新请求 LLM 更新缓存
                ...

class CacheMetrics:
    """缓存性能指标"""
    def __init__(self):
        self.hit_latencies: list[float] = []
        self.miss_latencies: list[float] = []
        self.cost_saved: float = 0.0

    def record(self, source: str, latency: float, tokens: int):
        if source.startswith("cache"):
            self.hit_latencies.append(latency)
            # 计算节省的成本
            self.cost_saved += tokens * 0.002 / 1000  # 假设 GPT-4o-mini 价格
        else:
            self.miss_latencies.append(latency)

    def get_summary(self) -> dict:
        return {
            "avg_hit_latency_ms": np.mean(self.hit_latencies) * 1000 if self.hit_latencies else 0,
            "avg_miss_latency_ms": np.mean(self.miss_latencies) * 1000 if self.miss_latencies else 0,
            "total_cost_saved_usd": self.cost_saved,
            "latency_improvement_factor": (
                np.mean(self.miss_latencies) / np.mean(self.hit_latencies)
                if self.hit_latencies else 0
            ),
        }