LLM API-Failover in der Praxis: Routing-Strategie mit Latenz- und Kostengewichtung

Block 2 — Vollständiger Kosten-Latenz-Router

from dataclasses import dataclass, field
from typing import Dict, List
import time, requests

@dataclass
class ModelSpec:
    name: str
    cost_per_mtok_usd: float   # offizieller Listenpreis USD / 1M Output-Tokens
    holy_cost_per_mtok_usd: float  # HolySheep-Tarif (Kurs ¥1=$1, -85%)
    avg_latency_ms: float = 60.0
    health: float = 1.0
    window: List[float] = field(default_factory=list)

class WeightedRouter:
    BASE_URL = "https://api.holysheep.ai/v1"
    MODELS: Dict[str, ModelSpec] = {
        "gpt-4.1":           ModelSpec("gpt-4.1",           8.00, 1.20, 50.0),
        "claude-sonnet-4.5": ModelSpec("claude-sonnet-4.5", 15.00, 2.25, 65.0),
        "gemini-2.5-flash":  ModelSpec("gemini-2.5-flash",   2.50, 0.375, 38.0),
        "deepseek-v3.2":     ModelSpec("deepseek-v3.2",      0.42, 0.063, 42.0),
    }

    def __init__(self, api_key: str, w_cost: float = 0.45, w_lat: float = 0.55):
        self.api_key = api_key
        self.w_cost, self.w_lat = w_cost, w_lat
        self.max_cost = max(m.cost_per_mtok_usd for m in self.MODELS.values())
        self.max_lat  = max(m.avg_latency_ms for m in self.MODELS.values())

    def score(self, m: ModelSpec) -> float:
        cost_part = 1 - (m.cost_per_mtok_usd / self.max_cost)
        lat_part  = 1 - (m.avg_latency_ms / self.max_lat)
        return (self.w_cost * cost_part + self.w_lat * lat_part) * m.health

    def ranking(self) -> List[str]:
        return [m.name for m in sorted(self.MODELS.values(),
                                       key=self.score, reverse=True)]

    def call(self, messages: List[dict], max_tokens: int = 512):
        last_err = None
        for name in self.ranking():
            spec = self.MODELS[name]
            t0 = time.perf_counter()
            try:
                r = requests.post(
                    f"{self.BASE_URL}/chat/completions",
                    headers={"Authorization": f"Bearer {self.api_key}"},
                    json={"model": name, "messages": messages,
                          "max_tokens": max_tokens},
                    timeout=8,
                )
                r.raise_for_status()
                ms = (time.perf_counter() - t0) * 1000
                self._record(name, ms, ok=True)
                return {"model": name,
                        "latency_ms": round(ms, 1),
                        "cost_usd_per_mtok": spec.holy_cost_per_mtok_usd,
                        "data": r.json()}
            except Exception as e:
                last_err = e
                self._record(name, 0.0, ok=False)
        raise RuntimeError(f"Alle Modelle fehlgeschlagen: {last_err}")

    def _record(self, name: str, ms: float, ok: bool):
        spec = self.MODELS[name]
        if ok:
            spec.window.append(ms)
            if len(spec.window) > 20: spec.window.pop(0)
            spec.avg_latency_ms = sum(spec.window) / len(spec.window)
            spec.health = min(1.0, spec.health + 0.05)
        else:
            spec.health = max(0.0, spec.health * 0.5)

--- Nutzung ---
router = WeightedRouter("YOUR_HOLYSHEEP_API_KEY")
result = router.call([{"role": "user",
                       "content": "Erkläre Latenz-Routing in 3 Sätzen."}])
print(result["model"], result["latency_ms"], "ms",
      "→", result["cost_usd_per_mtok"], "USD/Mtok")

import time
from typing import Dict

class CircuitBreaker:
    """Sperrt ein Modell nach n Fehlversuchen für reset_timeout Sekunden."""
    def __init__(self, fail_threshold: int = 5, reset_timeout: int = 30):
        self.fail_threshold = fail_threshold
        self.reset_timeout = reset_timeout
        self.failures: Dict[str, int] = {}
        self.opened_at: Dict[str, float] = {}

    def allow(self, key: str) -> bool:
        if self.failures.get(key, 0) < self.fail_threshold:
            return True
        if time.time() - self.opened_at.get(key, 0) > self.reset_timeout:
            self.failures[key] = 0  # Half-Open-Test
            return True
        return False

    def on_failure(self, key: str):
        self.failures[key] = self.failures.get(key, 0) + 1
        if self.failures[key] == self.fail_threshold:
            self.opened_at[key] = time.time()

    def on_success(self, key: str):
        self.failures[key] = 0

In WeightedRouter.call() einbinden:
  if not breaker.allow(name): continue
  try: ... ; breaker.on_success(name)
  except: breaker.on_failure(name)

FALSCH
spec.health += 0.05

RICHTIG — exponentielles Decay mit Decke
spec.health = min(1.0, spec.health + 0.05)
spec.health = max(0.0, spec.health * 0.5)   # bei Fehler

FALSCH — Mittelwert reagiert zu empfindlich
spec.avg_latency_ms = sum(spec.window[-5:]) / 5

RICHTIG — EMA (Exponential Moving Average)
alpha = 0.2
spec.avg_latency_ms = alpha * new_ms + (1 - alpha) * spec.avg_latency_ms

FALSCH — globaler Timeout
requests.post(url, json=payload, timeout=30)

RICHTIG — modell-spezifischer Timeout
TIMEOUTS = {"gpt-4.1": 6, "claude-sonnet-4.5": 8,
            "gemini-2.5-flash": 4, "deepseek-v3.2": 5}
requests.post(url, json=payload, timeout=TIMEOUTS[model])

RICHTIG — bei jedem Ranking-Aufruf neu berechnen
self.max_cost = max(m.cost_per_mtok_usd for m in self.MODELS.values())
self.max_lat  = max(m.avg_latency_ms for m in self.MODELS.values())