Kostenoptimierung im großen Maßstab

Kosten als ingenieurtechnische Randbedingung

Ein schlecht optimierter Prompt, der 0,002 USD pro Aufruf kostet, ist während der Entwicklung vernachlässigbar. Bei einer Million Aufrufen beläuft sich die Kosten auf 2.000 USD – und diese Kosten werden jeden Monat weiter anfallen, bis jemand den Code neu schreibt. Dieser Wandel im Maßstab verwandelt Kosten in eine ingenieurtechnische Randbedingung: Sie lösen das Problem nicht durch Vertragsverhandlungen, sondern indem Sie den Code ändern.

Sichtbarkeit ist die Voraussetzung für jede Optimierung. Eine monatliche Rechnung zeigt Ihnen nur, wie viel Sie ausgegeben haben, nicht wo. Deshalb beginnt diese Lektion mit der Instrumentierung, bevor sie die Hebel zur Reduzierung behandelt.

Wo das Geld hingeht

Vier Elemente bestimmen die Kosten eines Aufrufs, und sie verhalten sich nicht gleichartig. Eingabetokens, die den Systemprompt, die Gesprächshistorie und die Benutzeranfrage aggregieren, wachsen mit der Sitzungsdauer. Ausgabetokens – die generierte Antwort – werden deutlich höher als Eingabetokens berechnet. Eingabetokens, die im Cache liegen, erhalten einen Rabatt von etwa 50 % gegenüber dem Eingabesatz, was ihre Erfassung für das Verständnis jeder Rechnung unerlässlich macht. Schließlich das Modell selbst: GPT-5.6 Luna ist deutlich günstiger als GPT-5.6 Sol, und dieser Faktor dominiert alle anderen.

Tracking bedeutet, jedes usage-Objekt, das die API zurückgibt, in einen Geldbetrag umzuwandeln, den korrekten Satz anzuwenden und den gecachten Anteil zu isolieren.

import openai
from dataclasses import dataclass, field

client = openai.OpenAI()

# Sätze pro Million Tokens (Juli 2026) — cached_input ist repräsentativ (~50 % von input),
# anpassen Sie dies gemäß dem offiziellen Preisgitter
RATES = {
    "gpt-5.6-luna": {"input": 0.20, "output": 1.20, "cached_input": 0.02},
    "gpt-5.6-terra": {"input": 2.00, "output": 12.00, "cached_input": 0.20},
    "gpt-5.6-sol": {"input": 5.00, "output": 30.00, "cached_input": 0.50},
}

@dataclass
class CostTracker:
    """Verfolgt Kosten in Echtzeit."""
    calls: list[dict] = field(default_factory=list)

    def record(self, model: str, usage: dict) -> float:
        input_tokens = usage.input_tokens
        cached_tokens = getattr(
            usage.input_tokens_details, "cached_tokens", 0
        )
        output_tokens = usage.output_tokens
        uncached_input_tokens = input_tokens - cached_tokens

        rate = RATES.get(model, RATES["gpt-5.6-terra"])
        cost = (
            (uncached_input_tokens / 1_000_000) * rate["input"]
            + (cached_tokens / 1_000_000) * rate["cached_input"]
            + (output_tokens / 1_000_000) * rate["output"]
        )

        self.calls.append({
            "model": model,
            "input_tokens": input_tokens,
            "cached_tokens": cached_tokens,
            "output_tokens": output_tokens,
            "cost": cost,
        })
        return cost

    @property
    def total_cost(self) -> float:
        return sum(a["cost"] for a in self.calls)

    def report(self) -> str:
        if not self.calls:
            return "Keine Aufrufe aufgezeichnet."
        return (
            f"Aufrufe: {len(self.calls)} | "
            f"Gesamtkosten: ${self.total_cost:.4f} | "
            f"Durchschnittskosten: ${self.total_cost / len(self.calls):.6f}"
        )

tracker = CostTracker()

Die Tabelle RATES erfordert ständige Aufmerksamkeit: Sie wird veraltet, sobald sich das offizielle Preisgitter ändert. Behandeln Sie sie als Konfigurationsdaten zur Überprüfung und nicht als Programmkonstante, und machen Sie die durchschnittlichen Kosten pro Aufruf zu einer Kennzahl, die Sie wöchentlich überprüfen.

Vier Hebel zur Reduzierung

Der erste Hebel und bei weitem der profitabelste ist die Vermeidung der Bezahlung für ein High-End-Modell, wenn die Aufgabe dies nicht erfordert. Eine explizite Routierung basierend auf der Komplexität verhindert den Standard-Drift, bei dem der gesamte Verkehr auf das teuerste Modell landet, nur weil es im ersten kopierten Beispiel vorkam.

async def economic_router(prompt: str, complexity: str) -> str:
    """Routiert zum günstigsten fähigen Modell."""
    if complexity == "simple":
        model = "gpt-5.6-luna"
    elif complexity == "medium":
        model = "gpt-5.6-terra"
    else:
        model = "gpt-5.6-sol"

    response = client.responses.create(model=model, input=prompt)
    tracker.record(model, response.usage)
    return response.output_text

Der zweite Hebel geht weiter, wenn die Komplexität nicht im Voraus bekannt ist. Anstatt die Anfrage zu klassifizieren, versucht die Kaskade zunächst mit dem günstigsten Modell und eskaliert nur, wenn die Antwort eine Qualitätsprüfung nicht besteht. Die hier verwendete Prüfung ist absichtlich rudimentär – eine Mindestlänge und das Fehlen einer Eingeständnis der Unwissenheit – und dies ist der Punkt, den Sie verfeinern müssen: Eine Kaskade, die zu oft eskaliert, wird teurer als ein direkter Aufruf am richtigen Modell.

async def model_cascade(prompt: str) -> str:
    """Versucht zuerst mit dem günstigsten Modell."""
    models = ["gpt-5.6-luna", "gpt-5.6-terra", "gpt-5.6-sol"]

    for model in models:
        response = client.responses.create(
            model=model,
            input=prompt,
            max_output_tokens=500,
        )
        tracker.record(model, response.usage)

        # Qualitätsprüfung (einfache Heuristik)
        text = response.output_text
        if len(text) > 50 and "i don't know" not in text.lower():
            return text

    return text  # Letztes Ergebnis, auch wenn es unvollkommen ist

Der dritte Hebel zielt auf Eingabetokens ab, die in jeder konversationellen Anwendung stumm wachsen: Bis zum zwanzigsten Turn zahlen Sie für die vorherigen neunzehn mit jeder neuen Nachricht. Das Komprimieren der Historie, indem man die erste Nachricht (die den Kontext enthält) und die letzte (die die Frage enthält) behält und den verbleibenden Budgetteil mit den neuesten Austausch füllt, bewahrt die wesentliche Kohärenz zu einem Bruchteil der Kosten.

def compress_history(
    messages: list[dict],
    max_tokens: int = 4000,
) -> list[dict]:
    """Behält die neuesten Nachrichten im Rahmen des Budgets."""
    # Behalten Sie immer die erste (Kontext) und die letzte (Frage)
    if len(messages) <= 2:
        return messages

    first = messages[0]
    last = messages[-1]
    middle = messages[1:-1]

    # Schätzung der Tokens (Annäherung: 1 Token ≈ 4 Zeichen)
    fixed_tokens = (len(first["content"]) + len(last["content"])) // 4
    middle_budget = max_tokens - fixed_tokens

    # Behalten Sie die neuesten Nachrichten
    result = []
    tokens_used = 0
    for msg in reversed(middle):
        msg_tokens = len(msg["content"]) // 4
        if tokens_used + msg_tokens > middle_budget:
            break
        result.insert(0, msg)
        tokens_used += msg_tokens

    return [first] + result + [last]

Der vierte Hebel erfordert keine algorithmische Feinheit, sondern nur Disziplin: Die Batch-API bietet eine 50 %ige Reduzierung, und Sie müssen lediglich identifizieren, was warten kann. Lassen Sie die Aufgabe selbst die Dringlichkeit von ihrer Erstellung her tragen, anstatt sie danach abzuleiten – dies ist der einzige Weg, um zu verhindern, dass alles standardmäßig in die Echtzeit-Warteschlange landet.

def classify_urgency(tasks: list[dict]) -> dict:
    """Trennt Aufgaben in Echtzeit und Batch."""
    real_time = []
    batch = []

    for task in tasks:
        if task.get("urgency") == "immediate":
            real_time.append(task)
        else:
            batch.append(task)

    return {
        "real_time": real_time,
        "batch": batch,
        "estimated_savings": len(batch) * 0.001,  # Beispiel
    }

Warnung erhalten, bevor die Rechnung kommt

Alle diese Optimierungen können durch eine einzelne unglückliche Bereitstellung zunichte gemacht werden: ein ungebundener Retry-Loop, ein verdreifachter Systemprompt, ein versehentliches Modellwechsel. Eine Kostenwarnung ist das Sicherheitsnetz, das Sie innerhalb weniger Stunden warnt, nicht erst Monatsende. Beachten Sie die Zwischen-Schwelle bei 80 %: Sie verwandelt eine brutale Alarmierung in ein frühes Signal und gibt Ihnen Zeit, bevor Sie die Grenze überschreiten.

class CostAlert:
    """Löst Warnungen aus, wenn Kosten eine Schwelle überschreiten."""

    def __init__(self, daily_threshold: float = 50.0):
        self.threshold = daily_threshold
        self.tracker = CostTracker()

    def check(self) -> str | None:
        if self.tracker.total_cost > self.threshold:
            return (
                f"ALERT: tägliche Kosten ${self.tracker.total_cost:.2f} "
                f"überschreiten die Schwelle von ${self.threshold:.2f}"
            )
        if self.tracker.total_cost > self.threshold * 0.8:
            return (
                f"WARNUNG: Kosten bei {self.tracker.total_cost / self.threshold:.0%} "
                f"der täglichen Schwelle"
            )
        return None

Kernaussagen

• Verfolgen Sie Ihre Kosten in Echtzeit mit response.usage
• Routieren Sie zum günstigsten Modell, das die Aufgabe bewältigen kann
• Nutzen Sie die Batch-API (-50 %) für alles, was keine Echtzeitverarbeitung erfordert
• Reduzieren Sie Eingabetokens durch Komprimierung der Historie
• Richten Sie Kostenwarnungen ein, um Überraschungen zu vermeiden