LLM 语义缓存与模型路由：成本治理的两大杠杆

杠杆 1：语义缓存（Semantic Cache）
  用户查询 → 向量化 → 查缓存
  命中 → 直接返回缓存答案（0 推理成本、亚毫秒延迟）
  未命中 → 调模型 → 答案入库（供后续相似查询复用）

杠杆 2：模型路由（Model Routing）
  查询 → 复杂度分类 → 简单→小模型 / 复杂→大模型
  保证质量前提下，把多数流量路由到便宜模型

# semantic_cache.py — 语义缓存核心
import redis
import numpy as np

class SemanticCache:
    def __init__(self, redis_url, embed_fn, threshold: float = 0.92):
        """
        threshold 是命中阈值：向量余弦相似度 ≥ 此值视为同义。
        阈值越高越保守（少命中、防误判），越低越激进（多命中、有风险）。
        """
        self.r = redis.from_url(redis_url)
        self.embed_fn = embed_fn
        self.threshold = threshold
        self.PREFIX = "semcache"

    def get(self, query: str):
        """语义查询：命中返回缓存答案，未命中返回 None。"""
        q_vec = self.embed_fn(query)
        # 用 Redis 的向量检索（RedisVL / RediSearch）找最相似
        best = self.r.ft("idx:semcache").search(
            query_vector=q_vec, k=1, return_score=True)
        if not best.docs:
            return None
        score = best.docs[0].score
        if score >= self.threshold:
            return {
                "answer": best.docs[0].__getattribute__("answer"),
                "score": score,
                "cached": True,
            }
        return None

    def set(self, query: str, answer: str):
        """写入缓存：存向量 + 答案 + 元数据。"""
        q_vec = self.embed_fn(query)
        self.r.hset(f"{self.PREFIX}:{hash(query)}",
                    mapping={"vec": q_vec.tobytes(),
                             "answer": answer,
                             "ts": time.time()})

def calibrate_threshold(queries, embed_fn,
                        target_precision=0.99) -> float:
    """
    用一组已知"同义/非同义"查询对校准阈值。
    target_precision：误命中率控制在 1% 以内。
    """
    pairs = []   # [(q1, q2, is_same_intent)]
    sims = [cosine(embed_fn(q1), embed_fn(q2)) for q1, q2, _ in pairs]
    labels = [l for _, _, l in pairs]
    # 找到使"非同义对"误命中率 < 1% 的最小阈值
    for threshold in np.arange(0.85, 0.99, 0.005):
        false_pos = sum(s >= threshold and not l
                        for s, l in zip(sims, labels)) / sum(1 for _, _, l in pairs if not l)
        if false_pos <= 0.01:
            return threshold
    return 0.95   # 默认保守

def build_cache_key(query: str, user_id: str, feature: str,
                    context_hash: str, embed_fn) -> bytes:
    """复合语义键：向量 + 业务维度过滤。"""
    vec = embed_fn(query)
    # 向量本身做相似度检索，业务维度（用户/功能/上下文）做过滤
    return {
        "vec": vec.tobytes(),
        "user_id": user_id,
        "feature": feature,
        "context_hash": context_hash,
    }

CACHE_POLICY = {
    "faq_static":      {"ttl": 3600 * 6,  "cache": True},
    "product_price":   {"ttl": 300,       "cache": True, "versioned": True},
    "user_balance":    {"ttl": 0,         "cache": False},
    "order_status":    {"ttl": 0,         "cache": False},
}

def should_cache(query_meta: dict) -> tuple[bool, int]:
    """根据请求的功能类型决定缓存策略。"""
    policy = CACHE_POLICY.get(query_meta["feature"])
    if not policy:
        return False, 0
    return policy["cache"], policy["ttl"]

def invalidate_kb_scope(kb_version: str, cache_client, embed_fn):
    """知识库升级时，按 scope 批量失效缓存。"""
    # 每个缓存条目记录所属 scope（如 kb:v3）
    # 升级后旧版本 key 自然无法命中
    cache_client.delete_pattern("scope:kb:v2:*")
    print(f"已失效知识库 v2 缓存，当前激活 {kb_version}")

def get_or_compute(cache, query, compute_fn, lock_client):
    """缓存击穿防护：单飞（Single-flight）回填。"""
    cached = cache.get(query)
    if cached:
        return cached["answer"]

    lock_key = f"lock:{hash(query)}"
    if lock_client.setnx(lock_key, "1", ex=5):   # 抢锁成功者负责计算
        try:
            answer = compute_fn(query)
            cache.set(query, answer)
            return answer
        finally:
            lock_client.delete(lock_key)
    else:
        # 其他请求短暂等待后重试缓存（自旋）
        import time
        for _ in range(20):
            time.sleep(0.05)
            cached = cache.get(query)
            if cached:
                return cached["answer"]
        return compute_fn(query)   # 兜底：直接计算

路由优化前后对比（假设流量分布 70% 简单 / 25% 中等 / 5% 复杂）：
  全部用旗舰：成本 = 100% × 100x = 100
  按需路由：  70% × 1x + 25% × 10x + 5% × 100x = 70 + 250 + 500 = 8.2
  → 成本降至约 8%，质量不受损

# router.py — 三级路由
from dataclasses import dataclass

@dataclass
class RouteDecision:
    model: str
    reason: str

class ModelRouter:
    def __init__(self, classifiers, model_registry):
        self.classifiers = classifiers
        self.registry = model_registry   # {"light":..., "mid":..., "full":...}

    def route(self, query: str, feature: str) -> RouteDecision:
        # 1. 规则优先：强约束功能直接路由
        rule = ROUTE_RULES.get(feature)
        if rule:
            return RouteDecision(rule["model"], f"rule:{feature}")

        # 2. 分类器：判断复杂度
        complexity = self.classifiers.complexity(query)   # easy/medium/hard
        needs_tools = self.classifiers.needs_tools(query)
        if needs_tools:
            return RouteDecision("full", "needs_tools")
        return RouteDecision({
            "easy": "light", "medium": "mid", "hard": "full"
        }[complexity], f"complexity:{complexity}")

    def escalate_on_low_confidence(self, query, light_answer,
                                   verifier) -> RouteDecision:
        """反馈升级：小模型答案置信度低时升级到中端模型。"""
        conf = verifier.confidence(light_answer)
        if conf < 0.6:
            return RouteDecision("mid", "low_confidence_upgrade")
        return RouteDecision("light", "confident")

# complexity_classifier.py — 查询复杂度分类
COMPLEXITY_PROMPT = """判断查询的复杂度，只输出 easy/medium/hard 之一。
- easy：事实型、简单问答、无需推理
- medium：需要一定推理或多步组织
- hard：多步规划、代码、数学、长文档综合
查询：{query}"""

def classify_complexity(query, small_llm) -> str:
    label = small_llm(COMPLEXITY_PROMPT.format(query=query))
    return label.strip().lower() if label in {"easy", "medium", "hard"} else "medium"

def validate_router_quality(router, golden, judge_fn,
                            max_misroute_rate: float = 0.10) -> bool:
    """
    在评测集上验证：小模型处理的难题占比是否可控、质量是否达标。
    """
    misroutes = 0
    quality = []
    for case in golden.cases:
        decision = router.route(case["query"], case.get("feature", "default"))
        # 难题被路由到 light 模型 = 误路由
        if case["difficulty"] == "hard" and decision.model == "light":
            misroutes += 1
        answer = registry[decision.model](case["query"])
        quality.append(judge_fn(case["query"], answer, case["reference"])["total"])

    misroute_rate = misroutes / len(golden.cases)
    print(f"误路由率 {misroute_rate:.0%}，平均质量 {mean(quality):.1f}/25")
    assert misroute_rate <= max_misroute_rate, "难题过多被路由到小模型"
    return True

def shadow_compare(router, golden, models, judge_fn):
    """影子模式：新路由决策与"全用旗舰"的答案对比，量化损失。"""
    for case in golden.cases:
        decision = router.route(case["query"])
        routed_answer = models[decision.model](case["query"])
        full_answer = models["full"](case["query"])
        # 对比路由答案 vs 旗舰答案与参考的分数
        routed_score = judge_fn(case["query"], routed_answer, case["reference"])
        full_score = judge_fn(case["query"], full_answer, case["reference"])
        if routed_score["total"] - full_score["total"] < -3:
            log_regression(case["id"], decision, routed_score, full_score)

请求 → ① 语义缓存命中？ → 命中：直接返回（0 成本）
                           未命中 ↓
      ② 路由决策（复杂度分类）
      ③ 目标模型调用
      ④ 答案写入缓存（供相似查询复用）

def pipeline(query, cache, router, models, embed_fn):
    # 1. 先查缓存
    cached = cache.get(query)
    if cached:
        return cached["answer"]   # 0 推理成本

    # 2. 路由 + 调用
    decision = router.route(query)
    answer = models[decision.model](query)

    # 3. 回填缓存（仅静态内容）
    if should_cache({"feature": "default"})[0]:
        cache.set(query, answer)
    return answer

# monitor.py — 记录每请求的成本与路由信息
def trace_request(query, decision, answer, cost, latency_ms):
    push_metrics({
        "llm_cache_hit": decision.cached,
        "llm_model": decision.model,
        "llm_cost_usd": cost,
        "llm_latency_ms": latency_ms,
        "llm_feature": decision.feature,
    }, labels={"feature": decision.feature})

def weekly_cost_report(client) -> dict:
    """按功能/模型维度汇总周成本。"""
    return {
        "total": sum_query("llm_cost_usd"),
        "by_model": group_by("llm_model", "llm_cost_usd"),
        "cache_savings": estimate_cache_savings(),   # 命中省下的推理成本
    }

# litellm_router.py — 用 LiteLLM 做统一路由网关
import litellm
from litellm import Router

# 配置模型池：价格、限额、优先级
router = Router(
    model_list=[
        {"model_name": "main", "litellm_params": {"model": "gpt-4o-mini"},
         "model_info": {"cost_per_token": 0.0000001}},
        {"model_name": "main", "litellm_params": {"model": "gpt-4o"},
         "model_info": {"cost_per_token": 0.0000025}},
        {"model_name": "reasoning", "litellm_params": {"model": "o3-mini"}},
    ],
    routing_strategy="simple-shuffle",   # 或 usage-based / latency-based
)

resp = router.completion(model="main", messages=[...], max_tokens=500)
print(resp["model"], resp["cost"])    # 实际落点模型与成本

def estimate_roi(golden, router, cache, models, judge_fn,
                 traffic_model: dict) -> dict:
    """用评测集 + 流量分布估算优化后的成本与质量。"""
    # 1. 原方案成本
    baseline_cost = traffic_model["queries_per_day"] * \
        models["full"].cost_per_query(golden)

    # 2. 新方案：路由 + 缓存
    routed_cost = 0.0
    for case in golden.cases:
        decision = router.route(case["query"])
        routed_cost += models[decision.model].cost_per_query(case)
    # 加缓存命中折减（假设命中率 H）
    H = 0.4
    optimized_cost = routed_cost * (1 - H)

    # 3. 质量对比
    quality_loss = measure_quality_loss(golden, router, models, judge_fn)

    return {
        "baseline_cost_per_day": baseline_cost,
        "optimized_cost_per_day": optimized_cost,
        "savings_pct": 1 - optimized_cost / baseline_cost,
        "quality_loss": quality_loss,
    }

场景：电商客服 Agent，日均 10 万查询
流量构成：
  - 60% 高频 FAQ（重复/相似问题）→ 语义缓存主场
  - 30% 一般问答（订单、售后规则）→ 中端模型
  - 10% 复杂问题（多轮、协商）→ 旗舰模型
目标：成本降低 70%+，质量不降

# ecommerce_cost_governance.py
class EcommerceCostGovernance:
    def __init__(self, cache, router, models):
        self.cache = cache
        self.router = router
        self.models = models

    def answer(self, query: str, user_id: str, feature: str) -> dict:
        # 1. 功能级缓存策略
        cacheable, ttl = should_cache({"feature": feature})
        if cacheable:
            hit = self.cache.get(query)
            if hit:
                self._metric("cache_hit", feature)
                return {"answer": hit["answer"], "source": "cache"}

        # 2. 路由决策
        decision = self.router.route(query, feature)

        # 3. 调用（带失败升级）
        answer = self.models[decision.model](query)
        if self._low_confidence(answer) and decision.model != "full":
            answer = self.models["full"](query)
            decision = RouteDecision("full", "confidence_escalation")

        # 4. 回填缓存
        if cacheable:
            self.cache.set(query, answer)

        self._metric("model", feature, decision.model)
        return {"answer": answer, "source": "model",
                "model": decision.model}

    def _metric(self, *labels):
        push_metric("llm_governance", labels=labels)

发布策略：
阶段 1（影子）：新路由并行运行，只记录决策不生效，对比质量
阶段 2（金丝雀）：10% 流量走新路由，监控质量/延迟/成本
阶段 3（全量）：全流量切换，保留一键回滚开关
回滚条件：质量 Δ < -3 或 延迟 P99 超基线 ×1.3

def optimize_context_cost(system_prompt, messages, max_tokens):
    """三层降本：prompt 缓存 + 摘要压缩 + token 截断。"""
    # 1. system prompt 前缀缓存（多数厂商按前缀缓存打折）
    cached_cost = cacheable_prompt_cost(system_prompt)
    # 2. 超长历史压缩为摘要
    if estimate_tokens(messages) > max_tokens:
        messages = summarize_old_turns(messages, keep_last=5)
    # 3. 输出 token 上限约束
    return {"system": system_prompt, "messages": messages,
            "max_tokens": min(max_tokens, 2000)}

def batch_requests(queries: list[str], batch_size: int = 8,
                   model_fn) -> list[str]:
    """合并相似请求批量调用，降低单次请求固定开销。"""
    batched = [queries[i:i + batch_size] for i in range(0, len(queries), batch_size)]
    results = []
    for batch in batched:
        # 对支持 batch 的接口（如 OpenAI Chat Completions）一次调用
        results += model_fn.batch_completion(batch)
    return results

def adaptive_governance(monthly_budget, spent, remaining_days,
                        router):
    """预算接近上限时，自动把中端任务降级到轻量模型。"""
    daily_allow = (monthly_budget - spent) / max(remaining_days, 1)
    current_burn = daily_burn_rate()
    if current_burn > daily_allow * 1.2:    # 超支风险
        router.set_aggressive(True)         # 中端→轻量，仅 hard 用中端
        notify("成本压力：已降级路由档位")