KI-Kostenexplosion: Warum sich Ihre Azure-OpenAI-Rechnung verdreifacht hat und wie Sie das beheben

Tägliche Token-Aufschlüsselung (echte Kundendaten, anonymisiert):
────────────────────────────────────────────────────
System-Prompt:         800 Token  x 50.000 = 40M Input-Token
Abgerufener Kontext: 2.500 Token  x 50.000 = 125M Input-Token
Nutzeranfrage:         150 Token  x 50.000 = 7,5M Input-Token
────────────────────────────────────────────────────
Gesamt Input:                                 172,5M Token/Tag
Modellantwort:         400 Token  x 50.000 = 20M Output-Token

Tägliche Kosten (GPT-4o):
  Input:  172,5M / 1M * $2,50  = $431,25
  Output: 20M / 1M * $10,00    = $200,00
  Gesamt:                        $631,25/Tag = ~$19.000/Monat

from enum import Enum
from dataclasses import dataclass

class ModelTier(Enum):
    NANO = "gpt-4.1-nano"      # Klassifizierung, einfache Extraktion
    MINI = "gpt-4o-mini"       # Standard Q&A, Zusammenfassung
    STANDARD = "gpt-4o"        # Komplexes Reasoning, nuancierte Antworten
    REASONING = "o3-mini"      # Mehrstufige Logik, Mathematik, Analyse

@dataclass
class RoutingDecision:
    model: ModelTier
    reason: str
    estimated_cost_ratio: float  # Relativ zu GPT-4o

class ModelRouter:
    """Leitet Anfragen an das günstigste Modell, das Qualitätsanforderungen erfüllt."""

    async def route(self, query: str, context: dict) -> RoutingDecision:
        # Regelbasierter Schnellpfad für bekannte Muster
        if context.get("task_type") == "classification":
            return RoutingDecision(ModelTier.NANO, "classification_task", 0.04)
        if context.get("task_type") == "extraction":
            return RoutingDecision(ModelTier.NANO, "extraction_task", 0.04)

        # Nano-Modell zur Klassifizierung nutzen (kostet fast nichts)
        complexity = await self._classify_complexity(query)

        routing_map = {
            "SIMPLE": RoutingDecision(ModelTier.NANO, "simple_query", 0.04),
            "MODERATE": RoutingDecision(ModelTier.MINI, "moderate_query", 0.06),
            "COMPLEX": RoutingDecision(ModelTier.STANDARD, "complex_query", 1.0),
            "REASONING": RoutingDecision(ModelTier.REASONING, "reasoning_query", 0.44),
        }
        return routing_map.get(complexity,
            RoutingDecision(ModelTier.MINI, "default_moderate", 0.06))

import numpy as np
from redis import Redis
from openai import AzureOpenAI

class SemanticCache:
    def __init__(self, redis_client: Redis, openai_client: AzureOpenAI,
                 similarity_threshold: float = 0.95,
                 ttl_seconds: int = 3600):
        self.redis = redis_client
        self.openai = openai_client
        self.threshold = similarity_threshold
        self.ttl = ttl_seconds

    async def get_or_generate(self, query: str, system_prompt: str,
                               context: str, generate_fn) -> dict:
        # Embedding für die Anfrage generieren
        query_embedding = await self._get_embedding(query)

        # Nach ähnlichen gecachten Anfragen suchen
        cached = await self._search_cache(query_embedding)

        if cached and cached["similarity"] >= self.threshold:
            return {
                "response": cached["response"],
                "cached": True,
                "tokens_saved": cached["total_tokens"],
                "cost_saved": cached["estimated_cost"],
            }

        # Cache Miss — frische Antwort generieren
        result = await generate_fn(query, system_prompt, context)

        # Im Cache speichern
        await self._store_in_cache(
            query=query,
            embedding=query_embedding,
            response=result["response"],
            total_tokens=result["usage"]["total_tokens"],
        )

        return {**result, "cached": False}

class ContextCompressor:
    """Reduziert RAG-Kontext auf nur die relevantesten Chunks."""

    def compress_context(self, chunks: list, query: str,
                          max_tokens: int = 1500) -> list:
        ranked = sorted(chunks, key=lambda c: c["score"], reverse=True)
        relevant = [c for c in ranked if c["score"] >= 0.82]

        selected = []
        current_tokens = 0
        for chunk in relevant:
            chunk_tokens = len(chunk["text"].split()) * 1.3
            if current_tokens + chunk_tokens > max_tokens:
                break
            selected.append(chunk)
            current_tokens += chunk_tokens

        return selected

# VORHER: 800 Token
BLOATED_SYSTEM_PROMPT = """
Du bist ein hilfreicher, sachkundiger und freundlicher Kundensupport-
Assistent für die Acme Corporation. Deine Aufgabe ist es, Kunden
bei ihren Fragen zu unseren Produkten, Dienstleistungen, Abrechnung
und technischen Problemen zu helfen. Du solltest immer höflich,
professionell und gründlich in deinen Antworten sein...
"""

# NACHHER: 180 Token (78% Reduktion)
OPTIMIZED_SYSTEM_PROMPT = """Acme-Corp-Support-Assistent.
Regeln: Nur aus Kontext antworten. Sage "Ich weiß es nicht" bei Unsicherheit.
Keine persönlichen Daten. Max 150 Wörter. Quellen angeben."""

response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=messages,
    max_tokens=300,        # Hartes Limit für Output-Länge
    temperature=0.3,       # Niedrigere Temperatur = prägnanter
)

import json
from openai import AzureOpenAI

client = AzureOpenAI(
    azure_endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
    api_version="2025-04-01-preview",
)

# Batch-Datei vorbereiten (JSONL-Format)
batch_requests = []
for doc in documents_to_process:
    batch_requests.append({
        "custom_id": doc["id"],
        "method": "POST",
        "url": "/chat/completions",
        "body": {
            "model": "gpt-4o-mini",
            "messages": [
                {"role": "system", "content": "Extrahiere Schlüssel-Entitäten aus diesem Dokument."},
                {"role": "user", "content": doc["text"][:4000]},
            ],
            "temperature": 0.1,
        }
    })

# JSONL schreiben und Batch erstellen
with open("batch_input.jsonl", "w") as f:
    for req in batch_requests:
        f.write(json.dumps(req) + "\n")

batch_file = client.files.create(file=open("batch_input.jsonl", "rb"), purpose="batch")
batch_job = client.batches.create(
    input_file_id=batch_file.id,
    endpoint="/chat/completions",
    completion_window="24h",
)
# Ergebnisse innerhalb von 24h verfügbar, zu 50% der Kosten

// KQL: Tägliche Azure-OpenAI-Kostenaufschlüsselung nach Modell
AzureDiagnostics
| where ResourceProvider == "MICROSOFT.COGNITIVESERVICES"
| where Category == "RequestResponse"
| extend model = tostring(parse_json(properties_s).model)
| extend promptTokens = toint(parse_json(properties_s).promptTokens)
| extend completionTokens = toint(parse_json(properties_s).completionTokens)
| extend inputCost = case(
    model startswith "gpt-4o-mini", promptTokens / 1000000.0 * 0.15,
    model startswith "gpt-4o", promptTokens / 1000000.0 * 2.50,
    model startswith "gpt-4.1-nano", promptTokens / 1000000.0 * 0.10,
    0.0)
| extend outputCost = case(
    model startswith "gpt-4o-mini", completionTokens / 1000000.0 * 0.60,
    model startswith "gpt-4o", completionTokens / 1000000.0 * 10.00,
    model startswith "gpt-4.1-nano", completionTokens / 1000000.0 * 0.40,
    0.0)
| summarize
    TotalRequests = count(),
    EstimatedTotalCost = sum(inputCost) + sum(outputCost)
    by bin(TimeGenerated, 1d), model
| order by TimeGenerated desc