生产环境 LLM 网关：失败切换、语义缓存和成本控制

# 危险：上游已经半死，重试只会放大 p99
import time
from openai import OpenAI, APIStatusError

client = OpenAI()  # 直连某家

def chat_bad(prompt: str) -> str:
    last_err = None
    for i in range(3):
        try:
            r = client.chat.completions.create(
                model="gpt-4o",
                messages=[{"role": "user", "content": prompt}],
                timeout=30,
            )
            return r.choices[0].message.content or ""
        except APIStatusError as e:
            last_err = e
            time.sleep(2 ** i)
    raise RuntimeError(f"still failing: {last_err}")

# litellm config 片段
model_list:
  - model_name: primary
    litellm_params:
      model: anthropic/claude-sonnet-4-0
      api_key: os.environ/ANTHROPIC_API_KEY
  - model_name: secondary
    litellm_params:
      model: openai/gpt-4o-mini
      api_key: os.environ/OPENAI_API_KEY
  - model_name: local
    litellm_params:
      model: ollama/qwen2.5:14b
      api_base: http://192.168.1.20:11434

router_settings:
  fallbacks:
    - primary: ["secondary", "local"]
  num_retries: 1          # 单上游最多轻试一次
  timeout: 45
  allowed_fails: 3        # 短窗内失败次数触发冷却
  cooldown_time: 30

curl -s http://127.0.0.1:4000/v1/chat/completions \
  -H "Authorization: Bearer sk-xxx" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "primary",
    "messages": [{"role":"user","content":"用一句话解释熔断"}]
  }'

# tiny_breaker.py
from __future__ import annotations
import time
from dataclasses import dataclass, field
from typing import Callable, TypeVar

T = TypeVar("T")

@dataclass
class CircuitBreaker:
    fail_threshold: int = 5
    cooldown_sec: float = 30.0
    fails: int = 0
    opened_at: float | None = None

    def allow(self) -> bool:
        if self.opened_at is None:
            return True
        if time.time() - self.opened_at >= self.cooldown_sec:
            return True  # HALF_OPEN：放行一次试探
        return False

    def record_success(self) -> None:
        self.fails = 0
        self.opened_at = None

    def record_failure(self) -> None:
        self.fails += 1
        if self.fails >= self.fail_threshold:
            self.opened_at = time.time()

@dataclass
class Upstream:
    name: str
    call: Callable[[str], str]
    breaker: CircuitBreaker = field(default_factory=CircuitBreaker)

def chat_with_fallback(prompt: str, chain: list[Upstream]) -> tuple[str, str]:
    errors: list[str] = []
    for u in chain:
        if not u.breaker.allow():
            errors.append(f"{u.name}: circuit open")
            continue
        try:
            text = u.call(prompt)
            u.breaker.record_success()
            return text, u.name
        except Exception as e:  # noqa: BLE001 — 示例聚合错误
            u.breaker.record_failure()
            errors.append(f"{u.name}: {e}")
    raise RuntimeError("; ".join(errors))

# cost_estimate.py
from dataclasses import dataclass

@dataclass(frozen=True)
class Price:
    input_per_mTok: float
    output_per_mTok: float

PRICE = {
    "gpt-4o-mini": Price(0.15, 0.60),
    "gpt-4o": Price(2.50, 10.00),
    "claude-sonnet": Price(3.00, 15.00),
}

def estimate_usd(model: str, prompt_tokens: int, completion_tokens: int) -> float:
    p = PRICE[model]
    return (
        prompt_tokens / 1_000_000 * p.input_per_mTok
        + completion_tokens / 1_000_000 * p.output_per_mTok
    )

# 用法：从响应 usage 取值
# cost = estimate_usd("gpt-4o-mini", u.prompt_tokens, u.completion_tokens)

# daily_budget.py
import threading
from datetime import date

class DailyBudget:
    def __init__(self, limit_usd: float):
        self.limit = limit_usd
        self._day = date.today()
        self._spent = 0.0
        self._lock = threading.Lock()

    def charge(self, usd: float) -> None:
        with self._lock:
            if date.today() != self._day:
                self._day = date.today()
                self._spent = 0.0
            if self._spent + usd > self.limit:
                raise RuntimeError(
                    f"budget exceeded: spent={self._spent:.4f} + {usd:.4f} > {self.limit}"
                )
            self._spent += usd

budget = DailyBudget(limit_usd=5.0)  # 调试环境一天 5 美元封顶

# semantic_cache.py —— 演示逻辑，生产可换 Redis + 向量库
from __future__ import annotations
import math
import hashlib
from dataclasses import dataclass

def fake_embed(text: str, dim: int = 32) -> list[float]:
    """演示用伪向量：稳定、无外部依赖。生产请换真实 embedding。"""
    h = hashlib.sha256(text.encode()).digest()
    vals = [((h[i % len(h)] / 255.0) * 2 - 1) for i in range(dim)]
    norm = math.sqrt(sum(v * v for v in vals)) or 1.0
    return [v / norm for v in vals]

def cosine(a: list[float], b: list[float]) -> float:
    return sum(x * y for x, y in zip(a, b))

@dataclass
class Entry:
    text: str
    vec: list[float]
    answer: str

class SemanticCache:
    def __init__(self, threshold: float = 0.92):
        self.threshold = threshold
        self.items: list[Entry] = []

    def get(self, prompt: str) -> str | None:
        q = fake_embed(prompt)
        best, score = None, -1.0
        for it in self.items:
            s = cosine(q, it.vec)
            if s > score:
                best, score = it, s
        if best and score >= self.threshold:
            return best.answer
        return None

    def put(self, prompt: str, answer: str) -> None:
        self.items.append(Entry(prompt, fake_embed(prompt), answer))