Produktionsarchitektur: Monitoring, Alarmierung, Skalierung
Auf dieser Seite
Ihre Anwendung in der Produktion: Die drei Säulen
Den Einsatz einer LLM-Anwendung in der Produktion zu realisieren, beschränkt sich nicht auf den Code. Drei Säulen garantieren Zuverlässigkeit: Monitoring zur Beobachtung, Alarmierung zur Reaktion und Skalierung zur Aufnahme von Last. Sie bilden eine Kette, und eine Kette ist nur so stark wie ihr schwächstes Glied: Metriken, die niemand betrachtet, sind nutzlos; Alarme ohne Reaktionsverfahren sind nur Lärm; und eine Skalierung, die zu spät einsetzt, ist lediglich ein Ausfall. Diese abschließende Lektion fasst alle Konzepte des Kurses zu einer vollständigen Produktionsarchitektur zusammen.
Beobachten: Was zu messen ist
Vier Metrikfamilien decken das Wesentliche ab: Anrufvolumen, Latenz, Fehler und Kosten, aufgeteilt nach Modell. Die Aufteilung nach Modell macht das Dashboard erst handlungsorientiert: Eine globale Fehlerrate von 3 % sagt Ihnen nichts, während derselbe Wert, konzentriert auf ein einzelnes Modell, sofort den Schuldigen identifiziert.
Der folgende Klassencode sammelt diese Metriken im Arbeitsspeicher, wobei ein gleitendes Fenster von 10.000 Latenzen verwendet wird, um zu verhindern, dass der Prozess unbegrenzt wächst. Beachten Sie die Verwendung von Perzentilen statt von Durchschnittswerten: Die durchschnittliche Latenz eines LLM-Dienstes ist fast immer irreführend, da eine Handvoll sehr langsamer Aufrufe im Durchschnitt verschwinden, genau die aber, über die sich Ihre Nutzer beschweren.
import time
from dataclasses import dataclass, field
from collections import defaultdict
@dataclass
class ProductionMetrics:
"""Sammelt wesentliche Produktionsmetriken."""
counters: dict = field(default_factory=lambda: defaultdict(int))
latencies: dict = field(default_factory=lambda: defaultdict(list))
errors: dict = field(default_factory=lambda: defaultdict(int))
costs: dict = field(default_factory=lambda: defaultdict(float))
def record_call(
self,
model: str,
latency: float,
input_tokens: int,
output_tokens: int,
cost: float,
success: bool,
):
self.counters[model] += 1
self.latencies[model].append(latency)
self.costs[model] += cost
if not success:
self.errors[model] += 1
# Behalten Sie nur die letzten 10.000 Latenzen
if len(self.latencies[model]) > 10_000:
self.latencies[model] = self.latencies[model][-10_000:]
def percentile(self, model: str, p: float) -> float:
"""Berechnet ein Latenz-Perzentil."""
values = sorted(self.latencies.get(model, []))
if not values:
return 0.0
idx = int(len(values) * p / 100)
return values[min(idx, len(values) - 1)]
def error_rate(self, model: str) -> float:
total = self.counters.get(model, 0)
if total == 0:
return 0.0
return self.errors.get(model, 0) / total
def dashboard(self) -> dict:
"""Generiert ein vollständiges Dashboard."""
dashboard = {}
for model in self.counters:
dashboard[model] = {
"calls": self.counters[model],
"latency_p50": f"{self.percentile(model, 50):.2f}s",
"latency_p99": f"{self.percentile(model, 99):.2f}s",
"error_rate": f"{self.error_rate(model):.2%}",
"total_cost": f"${self.costs[model]:.2f}",
}
return dashboardDie Erfassung muss natürlich systematisch erfolgen. Die manuelle Instrumentierung jedes Aufrufs garantiert, dass Sie einen vergessen werden – und wahrscheinlich genau den, der Probleme verursacht. Der folgende Dekorierer verlagert die Messung auf die Funktionsebene des Aufrufs: Der finally-Block zeichnet Latenz und Erfolg auf, egal was passiert, auch wenn eine Ausnahme propagiert wird.
import openai
import functools
metrics = ProductionMetrics()
def monitor(func):
"""Dekorierer, der jeden API-Aufruf überwacht."""
@functools.wraps(func)
def wrapper(*args, **kwargs):
model = kwargs.get("model", args[0] if args else "unknown")
start = time.perf_counter()
success = True
try:
result = func(*args, **kwargs)
return result
except Exception as e:
success = False
raise
finally:
latency = time.perf_counter() - start
metrics.record_call(
model=model,
latency=latency,
input_tokens=0,
output_tokens=0,
cost=0,
success=success,
)
return wrapperToken-Zähler und Kosten bleiben in dieser Version bei Null, da das Antwortobjekt zum Zeitpunkt der Erfassung noch nicht verfügbar ist. Ergänzen Sie diese durch das Lesen der Nutzungsfelder, die von der API zurückgegeben werden: Ohne diese Informationen wird Ihr Dashboard eine Spalte mit Nullkosten korrekt anzeigen.
Reagieren: Abstufte Alarme
Ein Alarm ist nur dann wertvoll, wenn er zu einer Handlung auffordert. Deshalb unterscheidet das System pro Metrik zwei Schwellenwerte: eine Warnschwelle, die auf eine Drift hinweist, die während des Tages geprüft werden soll, und eine kritische Schwelle, die das Wecken einer Person rechtfertigt. Die Verwechslung beider führt zum schlechtestmöglichen Ergebnis: Ein Team, das gelernt hat, Benachrichtigungen zu ignorieren.
from enum import Enum
class AlertLevel(Enum):
INFO = "info"
WARNING = "warning"
CRITICAL = "critical"
class AlertSystem:
"""Löst Alarme basierend auf Metriken aus."""
def __init__(self):
self.thresholds = {
"error_rate": {"warning": 0.05, "critical": 0.15},
"latency_p99": {"warning": 10.0, "critical": 30.0},
"hourly_cost": {"warning": 50.0, "critical": 200.0},
}
self.sent_alerts: list[dict] = []
def check(self, metrics: ProductionMetrics) -> list[dict]:
"""Überprüft Metriken und generiert Alarme."""
alerts = []
for model in metrics.counters:
# Fehlerrate
rate = metrics.error_rate(model)
if rate >= self.thresholds["error_rate"]["critical"]:
alerts.append({
"level": AlertLevel.CRITICAL,
"message": (
f"KRITISCHE Fehlerrate für {model}: "
f"{rate:.1%}"
),
"model": model,
})
elif rate >= self.thresholds["error_rate"]["warning"]:
alerts.append({
"level": AlertLevel.WARNING,
"message": (
f"Hohe Fehlerrate für {model}: "
f"{rate:.1%}"
),
"model": model,
})
# Latenz
p99 = metrics.percentile(model, 99)
if p99 >= self.thresholds["latency_p99"]["critical"]:
alerts.append({
"level": AlertLevel.CRITICAL,
"message": (
f"KRITISCHE p99-Latenz für {model}: "
f"{p99:.1f}s"
),
"model": model,
})
self.sent_alerts.extend(alerts)
return alerts
async def notify(self, alerts: list[dict]):
"""Sendet Benachrichtigungen (Webhook, E-Mail, Slack)."""
import httpx
for alert in alerts:
if alert["level"] == AlertLevel.CRITICAL:
# Sofortige Benachrichtigung
async with httpx.AsyncClient() as http:
await http.post(
"https://your-webhook.example.com/alerts",
json={
"level": alert["level"].value,
"message": alert["message"],
},
)Stündliche Kostenalarme sind die, die am häufigsten vergessen werden, und das ist schade: Ein unbegrenzter Wiederholungsloop oder ein Batch-Job, der zweimal gestartet wird, führen zu einer Rechnung, nicht zu einem Fehler. Es ist das einzige Sicherheitsnetz, das diesen Typ von Vorfällen während ihres Geschehens erkennt.
Absorbieren: Drei Resilienzmechanismen
Wenn die API einen Fehler zurückgibt, macht ein erneuter Versuch die Lage schlimmer: Sie belasten einen Dienst, der ohnehin in Schwierigkeiten ist, und lassen Ihre Nutzer warten, ohne dass etwas passiert. Der Circuit Breaker kehrt diese Logik um. Nach einer definierten Anzahl aufeinanderfolgender Fehler öffnet er den Schalter und lehnt nachfolgende Aufrufe sofort ab, bevor er nach einer Reset-Verzögerung erneut versucht.
class CircuitBreaker:
"""Unterbricht Aufrufe, wenn zu viele Fehler auftreten."""
def __init__(
self,
open_threshold: int = 5,
reset_delay: float = 60.0,
):
self.threshold = open_threshold
self.reset_delay = reset_delay
self.consecutive_failures = 0
self.opened_at: float | None = None
@property
def is_open(self) -> bool:
if self.opened_at is None:
return False
if time.time() - self.opened_at > self.reset_delay:
# Versuchen, den Schalter zu schließen
self.opened_at = None
self.consecutive_failures = 0
return False
return True
def record_success(self):
self.consecutive_failures = 0
self.opened_at = None
def record_failure(self):
self.consecutive_failures += 1
if self.consecutive_failures >= self.threshold:
self.opened_at = time.time()
def call(self, func, *args, **kwargs):
"""Führt die Funktion aus, wenn der Schalter geschlossen ist."""
if self.is_open:
raise RuntimeError(
"Schalter offen — API nicht verfügbar. "
f"Wiederholen Sie es in {self.reset_delay}s."
)
try:
result = func(*args, **kwargs)
self.record_success()
return result
except Exception as e:
self.record_failure()
raiseDer Circuit Breaker schützt Ihre Anwendung; er macht sie nicht funktionsfähig, obwohl das die Rolle des Modell-Fallbacks ist, der Alternativen in Reihenfolge versucht und diejenigen überspringt, deren Schalter bereits offen sind. Ein Designpunkt, an den Sie sich erinnern sollten: Die Reihenfolge der Liste ist eine Produktentscheidung, keine technische. Stellen Sie zuerst das Modell, das den erwarteten Dienst liefert, gefolgt von denen, die einen degradierten aber akzeptablen Dienst liefern, an – eine etwas weniger scharfe Antwort ist besser als eine nicht verfügbare Meldung.
class ModelFallback:
"""Wechselt auf ein Back-up-Modell im Falle eines Fehlers."""
def __init__(self):
self.models = ["gpt-5.6-terra", "gpt-5.6-sol", "gpt-5.6-luna"]
self.circuits = {
m: CircuitBreaker() for m in self.models
}
def call(self, prompt: str) -> str:
"""Versucht jedes Modell, bis eine Antwort erhalten wird."""
client = openai.OpenAI()
for model in self.models:
circuit = self.circuits[model]
if circuit.is_open:
continue
try:
response = circuit.call(
client.responses.create,
model=model,
input=prompt,
)
return response.output_text
except Exception:
continue
raise RuntimeError("Alle Modelle sind nicht verfügbar.")Der dritte Mechanismus betrifft die Dimensionierung. Die Steuerung der Anzahl der Worker basierend auf der Warteschlangenlänge statt auf der CPU-Last ist die richtige Reflex für einen Dienst, der die meiste Zeit darauf wartet, Netzwerkantworten zu erhalten: Die CPU sitzt untätig, während die Warteschlange wächst. Beachten Sie die Asymmetrie der angewandten Regel – wir verdoppeln bei der Skalierung nach oben und halbieren bei der Skalierung nach unten, mit weit getrennten Schwellenwerten (100 gegenüber 10). Dieser Abstand verhindert Oszillationen: Ohne ihn würde das System seine Zeit damit verbringen, Worker um den Wendepunkt herum zu erstellen und zu zerstören.
class AutoScaler:
"""Passt die Anzahl der Worker basierend auf der Last an."""
def __init__(
self,
min_workers: int = 2,
max_workers: int = 50,
scale_up_threshold: int = 100,
scale_down_threshold: int = 10,
):
self.min_workers = min_workers
self.max_workers = max_workers
self.up_threshold = scale_up_threshold
self.down_threshold = scale_down_threshold
self.active_workers = min_workers
def adjust(self, queue_size: int) -> int:
"""Gibt die empfohlene Anzahl der Worker zurück."""
if queue_size > self.up_threshold:
self.active_workers = min(
self.active_workers * 2,
self.max_workers,
)
elif queue_size < self.down_threshold:
self.active_workers = max(
self.active_workers // 2,
self.min_workers,
)
return self.active_workersProduktions-Einsatz-Checkliste
Dieser Kurs endet mit der Checkliste, die durchlaufen werden sollte, bevor Ihre Anwendung für die Öffentlichkeit geöffnet wird. Jeder Punkt bezieht sich auf eine Lektion, die Sie bereits behandelt haben; wenn einer unklar erscheint, lesen Sie diese Lektion erneut, bevor Sie den Einsatz durchführen.
- Latenz-Monitoring (p50, p95, p99) und Fehlerraten
- Alarme, die auf kritischen Schwellenwerten konfiguriert sind
- Circuit Breaker zum Schutz vor API-Fehlern
- Modell-Fallback konfiguriert
- Eingabe- und Ausgabe-Moderation aktiviert
- Client-seitige Rate-Limitierung implementiert
- Strukturierte Logs zur Fehlerbehebung
- Automatisierte Evals in CI/CD
- Kostenbudget mit Alarmen
- Versionierte Prompt-Dokumentation
Kernaussagen
- Monitoring umfasst Latenz, Fehler, Kosten und Antwortqualität
- Verfolgen Sie Latenz-Perzentilen, niemals nur den Durchschnitt
- Alarme müssen abgestuft sein: Warnung, dann kritisch
- Circuit Breaker schützt vor Fehlerkaskaden; Fallback stellt Verfügbarkeit sicher
- Eine Produktions-Checkliste verhindert kritische Übersehen
Testen Sie Ihr Wissen
Latenz, Kosten, Evals: Produktionsreflexe in fünf Fragen.
1. Was sind die wichtigsten Hebel zur Reduzierung der Latenz?
Antwort: Streaming (wahrgenommene Latenz), Prompt-Caching (stabile Präfixe), Predicted Outputs, wenn die Antwort weitgehend vorhersehbar ist (Chat Completions, bei Modellen, die am 14. Januar 2027 zurückgezogen wurden), und die Optimierung des Prompts selbst – gemessen von Ende bis Ende.
2. Wann sollten Sie die Batch-API verwenden?
Antwort: Für massive Verarbeitung ohne Echtzeit-Anforderungen: asynchrone Batches mit reduzierten Kosten – Ranking, Anreicherung oder Volumen-Generierung haben keinen Platz auf dem synchronen Pfad.
3. Was macht die Konversation-Kompaktierung?
Antwort: Sie komprimiert lange Historien (Zusammenfassungen, Beschneidung), um im Kontext zu passen und Kosten zu kontrollieren – unerlässlich für Agenten und lange Sitzungen.
4. Was ist der Zweck des Evals-Frameworks in CI/CD?
Antwort: Um Ihre Evaluierungen bei jeder Prompt- oder Modelländerung automatisch durchzuführen: Regression wird bei Commit erkannt, nicht in der Produktion.
5. Was beinhaltet eine vollständige Produktionsarchitektur?
Antwort: Monitoring (Latenz, Kosten, Qualität), Alarmierung, asynchrone Warteschlangen, Moderation im großen Maßstab und ein Skalierungsplan – KI in der Produktion ist ein System, kein API-Aufruf.
Messen, cachen, bündeln, kontinuierlich evaluieren: Vier Gewohnheiten, die Anwendungen schnell, schlank und zuverlässig machen.