GPT-5.5 vs. V4-Flash: Kostenanalyse und Optimierungsstrategien für Produktionsumgebungen

Concurrency-Control und Rate-Limiting

Bei High-Throughput-Anwendungen ist intelligentes Rate-Limiting essentiell. Hier ist meine Production-Implementierung mit Token-Bucket-Algorithmus und automatischer Skalierung:

"""
Production-Grade Concurrency Controller mit Auto-Scaling
Features: Token-Bucket, Circuit Breaker, Request-Queuing
"""
import asyncio
import time
import logging
from collections import deque
from dataclasses import dataclass, field
from typing import Optional, Callable, Any
from contextlib import asynccontextmanager
import threading

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@dataclass
class RateLimitConfig:
    """Konfiguration für verschiedene API-Provider."""
    requests_per_minute: int = 60
    tokens_per_minute: int = 100_000
    burst_size: int = 10
    cooldown_seconds: float = 5.0

@dataclass
class CircuitBreakerState:
    """State für Circuit Breaker Pattern."""
    failure_count: int = 0
    last_failure_time: float = 0.0
    state: str = "CLOSED"  # CLOSED, OPEN, HALF_OPEN
    success_count: int = 0

class ConcurrencyController:
    """
    Production-Grade Controller mit:
    - Token-Bucket Rate Limiting
    - Circuit Breaker für Fault Tolerance
    - Request Coalescing
    - Automatic Backpressure
    """
    
    def __init__(
        self,
        config: RateLimitConfig,
        model_router: IntelligentModelRouter,
        holy_api_key: str,
        base_url: str = "https://api.holysheep.ai/v1"
    ):
        self.config = config
        self.router = model_router
        self.api_key = holy_api_key
        self.base_url = base_url
        
        # Token Bucket State
        self.tokens = config.burst_size
        self.last_refill = time.time()
        
        # Request Tracking
        self.active_requests = 0
        self.request_queue = deque()
        self.processed_count = 0
        self.total_cost = 0.0
        
        # Circuit Breaker
        self.circuit_breaker = CircuitBreakerState()
        self.circuit_failure_threshold = 5
        self.circuit_recovery_timeout = 30.0
        
        # Lock für Thread-Safety
        self._lock = threading.Lock()
    
    def _refill_tokens(self):
        """Refill Token-Bucket basierend auf Zeit."""
        now = time.time()
        elapsed = now - self.last_refill
        
        # Tokens pro Sekunde berechnen
        refill_rate = self.config.requests_per_minute / 60.0
        new_tokens = elapsed * refill_rate * self.config.burst_size
        
        self.tokens = min(self.config.burst_size, self.tokens + new_tokens)
        self.last_refill = now
    
    async def acquire(self, priority: int = 0) -> bool:
        """
        Acquired Permission für Request.
        Priority: 0 (normal), 1 (high), 2 (critical)
        """
        while True:
            with self._lock:
                self._refill_tokens()
                
                # Circuit Breaker Check
                if self._is_circuit_open():
                    await asyncio.sleep(1.0)
                    continue
                
                # Token verfügbar?
                if self.tokens >= 1:
                    self.tokens -= 1
                    self.active_requests += 1
                    return True
            
            # Backpressure: Warte auf Token
            await asyncio.sleep(0.05)
    
    def release(self, success: bool = True):
        """Releases Request-Slot und aktualisiert Circuit Breaker."""
        with self._lock:
            self.active_requests -= 1
            
            if success:
                self.circuit_breaker.success_count += 1
                self.circuit_breaker.failure_count = 0
                
                # Circuit Breaker zurücksetzen
                if self.circuit_breaker.state == "HALF_OPEN":
                    self.circuit_breaker.state = "CLOSED"
            else:
                self._handle_failure()
    
    def _is_circuit_open(self) -> bool:
        """Prüft Circuit Breaker Status."""
        cb = self.circuit_breaker
        
        if cb.state == "CLOSED":
            return False
        
        if cb.state == "OPEN":
            if time.time() - cb.last_failure_time > self.circuit_recovery_timeout:
                cb.state = "HALF_OPEN"
                logger.info("Circuit Breaker: OPEN -> HALF_OPEN")
                return False
            return True
        
        return False
    
    def _handle_failure(self):
        """Behandelt Failure und aktualisiert Circuit Breaker."""
        cb = self.circuit_breaker
        cb.failure_count += 1
        cb.last_failure_time = time.time()
        
        if cb.failure_count >= self.circuit_failure_threshold:
            cb.state = "OPEN"
            logger.warning(
                f"Circuit Breaker geöffnet nach {cb.failure_count} Failures"
            )
    
    @asynccontextmanager
    async def managed_request(self, priority: int = 0):
        """Context Manager für automatisches Acquire/Release."""
        await self.acquire(priority)
        try:
            yield
        finally:
            self.release(success=True)
    
    async def execute_with_fallback(
        self,
        prompt: str,
        system_prompt: str = "",
        max_retries: int = 3
    ) -> dict:
        """
        Führt Request mit automatischem Fallback aus.
        Falls Primary Model fehlschlägt, probiere günstigere Alternativen.
        """
        classification = self.router.classify_request(prompt, system_prompt)
        primary_model, _ = self.router.select_model(classification)
        
        models_to_try = [primary_model]
        
        # Fallback-Liste erstellen
        if "gpt" in primary_model:
            models_to_try.extend(["holy-gemini-flash", "holy-deepseek"])
        elif "gemini" in primary_model:
            models_to_try.append("holy-deepseek")
        
        last_error = None
        
        for model in models_to_try:
            for attempt in range(max_retries):
                try:
                    async with self.managed_request():
                        result = await self._call_api(
                            model=model,
                            prompt=prompt,
                            system_prompt=system_prompt
                        )
                        
                        # Cost Tracking
                        cost_info = self.router.estimate_cost(
                            model,
                            result.get("usage", {}).get("total_tokens", 0)
                        )
                        self.total_cost += cost_info["cost_usd"]
                        self.processed_count += 1
                        
                        return {
                            "model": model,
                            "response": result["choices"][0]["message"]["content"],
                            "usage": result.get("usage", {}),
                            "cost": cost_info
                        }
                
                except Exception as e:
                    last_error = e
                    logger.warning(f"Model {model} attempt {attempt + 1} failed: {e}")
                    self.release(success=False)
                    await asyncio.sleep(2 ** attempt)  # Exponential Backoff
        
        raise RuntimeError(f"Alle Modelle fehlgeschlagen: {last_error}")
    
    async def _call_api(
        self,
        model: str,
        prompt: str,
        system_prompt: str
    ) -> dict:
        """Interner API-Call zu HolySheep AI."""
        import aiohttp
        
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        payload = {
            "model": model,
            "messages": [
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": prompt}
            ],
            "temperature": 0.7,
            "max_tokens": 2048
        }
        
        async with aiohttp.ClientSession() as session:
            async with session.post(
                f"{self.base_url}/chat/completions",
                headers=headers,
                json=payload,
                timeout=aiohttp.ClientTimeout(total=30)
            ) as response:
                if response.status != 200:
                    error_text = await response.text()
                    raise RuntimeError(f"API Error {response.status}: {error_text}")
                
                return await response.json()
    
    def get_stats(self) -> dict:
        """Gibt aktuelle Statistiken zurück."""
        return {
            "active_requests": self.active_requests,
            "processed_count": self.processed_count,
            "total_cost_usd": round(self.total_cost, 2),
            "avg_cost_per_request": round(
                self.total_cost / max(self.processed_count, 1), 4
            ),
            "circuit_breaker_state": self.circuit_breaker.state,
            "available_tokens": round(self.tokens, 2)
        }

Benchmark: 1000 Requests simulieren
async def benchmark():
    """Simuliert 1000 Requests mit Cost-Tracking."""
    router = IntelligentModelRouter()
    controller = ConcurrencyController(
        config=RateLimitConfig(requests_per_minute=500, tokens_per_minute=1_000_000),
        model_router=router,
        holy_api_key="YOUR_HOLYSHEEP_API_KEY"
    )
    
    test_prompts = [
        ("Was ist die Hauptstadt von Frankreich?", 0),
        ("Erkläre Quantencomputing mit Beispielen aus der Praxis.", 1),
        ("Schreibe Python-Code für einen Webserver.", 0),
        ("Analysiere die Auswirkungen von KI auf den Arbeitsmarkt.", 1),
    ] * 250  # 1000 Requests
    
    start_time = time.time()
    
    # Requests parallel ausführen (max 50 gleichzeitig)
    semaphore = asyncio.Semaphore(50)
    
    async def bounded_request(prompt, priority):
        async with semaphore:
            try:
                result = await controller.execute_with_fallback(prompt)
                return result["cost"]["cost_cents"]
            except Exception as e:
                return 0.0
    
    tasks = [
        bounded_request(prompt, priority) 
        for prompt, priority in test_prompts
    ]
    
    costs = await asyncio.gather(*tasks)
    elapsed = time.time() - start_time
    
    stats = controller.get_stats()
    
    print("=" * 60)
    print("BENCHMARK ERGEBNISSE")
    print("=" * 60)
    print(f"Requests:      {stats['processed_count']}")
    print(f"Dauer:         {elapsed:.2f} Sekunden")
    print(f"Throughput:    {stats['processed_count'] / elapsed:.1f} req/s")
    print(f"Gesamtkosten:  ${stats['total_cost_usd']:.2f}")
    print(f"Ø Kosten/Req: ${stats['avg_cost_per_request']:.4f}")
    print(f"Vs. GPT-5.5:   ${stats['processed_count'] * 0.03:.2f} (100% teuer)")
    print(f"Ersparnis:     {100 - (stats['total_cost_usd'] / (stats['processed_count'] * 0.03) * 100):.1f}%")
    print("=" * 60)

Benchmark ausführen
asyncio.run(benchmark())

"""
Semantic Cache für AI API Responses
Implementiert: Exact-Match, Fuzzy-Match, TTL, Cost-Savings Tracking
"""
import hashlib
import json
import time
from collections import OrderedDict
from dataclasses import dataclass
from typing import Optional, Any, Callable
import numpy as np

@dataclass
class CacheEntry:
    """Single Cache Entry mit Metadaten."""
    key: str
    response: Any
    model: str
    created_at: float
    last_accessed: float
    access_count: int
    tokens_used: int
    cost_saved: float

class SemanticCache:
    """
    Two-Level Cache:
    1. L1: Exact-Match mit MD5-Hash (O(1) Lookup)
    2. L2: Semantic Similarity mit Embeddings (für leicht abweichende Prompts)
    """
    
    def __init__(
        self,
        max_size: int = 10000,
        ttl_seconds: float = 3600,
        similarity_threshold: float = 0.95
    ):
        # L1 Cache: Exact Match
        self.exact_cache: OrderedDict[str, CacheEntry] = OrderedDict()
        self.max_size = max_size
        self.ttl = ttl_seconds
        self.similarity_threshold = similarity_threshold
        
        # L2 Cache: Semantic (vereinfacht mit Hash-Splitting)
        self.semantic_cache: dict[str, list[CacheEntry]] = {}
        
        # Statistics
        self.stats = {
            "hits": 0,
            "misses": 0,
            "exact_hits": 0,
            "semantic_hits": 0,
            "total_savings_cents": 0.0,
            "evictions": 0
        }
    
    def _normalize_prompt(self, prompt: str) -> str:
        """Normalisiert Prompt für besseren Cache-Hit."""
        return (
            prompt
            .strip()
            .lower()
            .replace("\\n", " ")
            .replace("\\t", " ")
        )
    
    def _compute_key(self, prompt: str, system_prompt: str = "") -> str:
        """Compute Cache-Key mit Hash."""
        combined = f"{system_prompt}||{prompt}"
        return hashlib.md5(combined.encode()).hexdigest()
    
    def _compute_semantic_key(self, prompt: str) -> str:
        """
        Compute Semantic-Key für Fuzzy-Matching.
        Verwendet vereinfachte Keyword-Extraktion.
        """
        words = self._normalize_prompt(prompt).split()
        # Wichtige Wörter behalten (>3 Zeichen, nicht Stopwords)
        stopwords = {"und", "oder", "der", "die", "das", "ist", "von", "mit"}
        keywords = sorted([w for w in words if len(w) > 3 and w not in stopwords])
        return " ".join(keywords[:10])  # Max 10 Keywords
    
    def get(self, prompt: str, system_prompt: str = "") -> Optional[CacheEntry]:
        """Retrieves Cached Response oder None."""
        normalized = self._normalize_prompt(prompt)
        
        # L1: Exact Match
        key = self._compute_key(normalized, system_prompt)
        entry = self.exact_cache.get(key)
        
        if entry and self._is_valid(entry):
            entry.last_accessed = time.time()
            entry.access_count += 1
            self.stats["hits"] += 1
            self.stats["exact_hits"] += 1
            return entry
        
        # L2: Semantic Match
        semantic_key = self._compute_semantic_key(normalized)
        semantic_entries = self.semantic_cache.get(semantic_key, [])
        
        for entry in semantic_entries:
            if self._is_valid(entry):
                entry.last_accessed = time.time()
                entry.access_count += 1
                self.stats["hits"] += 1
                self.stats["semantic_hits"] += 1
                return entry
        
        self.stats["misses"] += 1
        return None
    
    def put(
        self,
        prompt: str,
        system_prompt: str,
        response: Any,
        model: str,
        tokens_used: int,
        cost_per_million: float
    ) -> None:
        """Speichert Response im Cache."""
        normalized = self._normalize_prompt(prompt)
        key = self._compute_key(normalized, system_prompt)
        
        cost_saved = (tokens_used / 1_000_000) * cost_per_million
        
        entry = CacheEntry(
            key=key,
            response=response,
            model=model,
            created_at=time.time(),
            last_accessed=time.time(),
            access_count=1,
            tokens_used=tokens_used,
            cost_saved=cost_saved
        )
        
        # L1 Cache aktualisieren
        if key in self.exact_cache:
            del self.exact_cache[key]
        self.exact_cache[key] = entry
        
        # L2 Semantic Cache
        semantic_key = self._compute_semantic_key(normalized)
        if semantic_key not in self.semantic_cache:
            self.semantic_cache[semantic_key] = []
        self.semantic_cache[semantic_key].append(entry)
        
        # Eviction wenn über Max-Size
        self._evict_if_needed()
    
    def _is_valid(self, entry: CacheEntry) -> bool:
        """Prüft ob Entry noch valide (nicht TTL-abgelaufen)."""
        return (time.time() - entry.created_at) < self.ttl
    
    def _evict_if_needed(self):
        """Evicted älteste Entries wenn Cache voll."""
        while len(self.exact_cache) > self.max_size:
            oldest_key = next(iter(self.exact_cache))
            self.exact_cache.pop(oldest_key)
            self.stats["evictions"] += 1
    
    def get_stats(self) -> dict:
        """Gibt Cache-Statistiken zurück."""
        total_requests = self.stats["hits"] + self.stats["misses"]
        hit_rate = self.stats["hits"] / max(total_requests, 1)
        
        return {
            "hit_rate": f"{hit_rate * 100:.1f}%",
            "exact_hits": self.stats["exact_hits"],
            "semantic_hits": self.stats["semantic_hits"],
            "misses": self.stats["misses"],
            "total_savings": f"${self.stats['total_savings_cents'] / 100:.2f}",
            "cache_size": len(self.exact_cache),
            "evictions": self.stats["evictions"]
        }

Production Integration mit Router
class CachedModelClient:
    """
    Production-Client mit Cache-Integration.
    Nahtloses Caching mit automatischer Kostenverfolgung.
    """
    
    def __init__(
        self,
        api_key: str,
        router: IntelligentModelRouter,
        cache: SemanticCache
    ):
        self.api_key = api_key
        self.router = router
        self.cache = cache
        self.base_url = "https://api.holysheep.ai/v1"
    
    async def complete(
        self,
        prompt: str,
        system_prompt: str = "",
        use_cache: bool = True
    ) -> dict:
        """
        Führt AI-Completion mit Caching aus.
        """
        # Cache prüfen
        if use_cache:
            cached = self.cache.get(prompt, system_prompt)
            if cached:
                return {
                    "response": cached.response,
                    "cached": True,
                    "model": cached.model,
                    "cost_saved": cached.cost_saved
                }
        
        # Request an API
        classification = self.router.classify_request(prompt, system_prompt)
        model, cost_per_1k = self.router.select_model(classification)
        
        # API Call (hier vereinfacht)
        response = await self._call_api(model, prompt, system_prompt)
        tokens = response.get("usage", {}).get("total_tokens", 0)
        
        # In Cache speichern
        if use_cache:
            self.cache.put(
                prompt=prompt,
                system_prompt=system_prompt,
                response=response["choices"][0]["message"]["content"],
                model=model,
                tokens_used=tokens,
                cost_per_million=cost_per_1k * 1000
            )
        
        return {
            "response": response["choices"][0]["message"]["content"],
            "cached": False,
            "model": model,
            "tokens": tokens
        }
    
    async def _call_api(self, model: str, prompt: str, system: str) -> dict:
        """API Call zu HolySheep AI."""
        import aiohttp
        
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        payload = {
            "model": model,
            "messages": [
                {"role": "system", "content": system},
                {"role": "user", "content": prompt}
            ]
        }
        
        async with aiohttp.ClientSession() as session:
            async with session.post(
                f"{self.base_url}/chat/completions",
                headers=headers,
                json=payload,
                timeout=aiohttp.ClientTimeout(total=30)
            ) as resp:
                return await resp.json()

Demonstration
cache = SemanticCache(max_size=5000, ttl_seconds=1800)
router = IntelligentModelRouter()

Test-Szenarien
test_cases = [
    ("Was ist Python?", "Du bist ein hilfreicher Assistent."),
    ("Was ist Python?", "Du bist ein hilfreicher Assistent."),  # Cache Hit!
    ("Erkläre Python", "Du bist ein hilfreicher Assistent."),   # Semantic Hit
    ("Was ist die Antwort auf alles?", ""),                     # Cache Miss
]

print("CACHE DEMO")
print("-" * 40)
for prompt, system in test_cases:
    result = cache.get(prompt, system)
    if result:
        print(f"CACHE HIT: {prompt[:30]}... (Model: {result.model})")
    else:
        print(f"CACHE MISS: {prompt[:30]}...")
        # Simulate cache fill
        cache.put(prompt, system, "Cached Response", "gpt-4.1", 100, 8.0)

print("\\n" + "-" * 40)
print("FINAL STATS:", cache.get_stats())