Alarme bei Latenz, Kosten und Qualitätsverlust
Definieren Sie Alarmschwellen für die p99-Latenz, die Kosten pro Anfrage und automatisierte Qualitätswerte und leiten Sie Alarme an Slack oder PagerDuty weiter, wenn sich Ihre LLM-Pipeline verschlechtert.
Alarme bei Latenz, Kosten und Qualitätsverlust ist eine kostenlose AI Engineering Academy-Lektion auf CoddyKit. Dies ist Lektion 4 von 4. Du kannst die komplette Lektion unten kostenlos lesen – dann übst du sie direkt im Browser mit einem integrierten Code-Editor und einem KI-Tutor rund um die Uhr. Sie ist Teil des AI Engineering Academy-Lernpfads, und dein Fortschritt wird über Web und CoddyKit-App synchronisiert. Der AI Engineering Academy-Kurs umfasst insgesamt 4 Lektionen.
Warum Benachrichtigungen für LLM-Systeme wichtig sind
LLM-Anwendungen fallen auf subtile und schleichende Weise aus. Eine Prompt-Änderung kann die durchschnittliche Latenz um 30 % erhöhen, eine Verbesserung des Abrufs die Antwortqualität leicht verringern oder ein Nutzungsanstieg die täglichen Kosten auf das Fünffache des Budgets treiben. Ohne proaktive Benachrichtigungen entdecken Sie diese Probleme erst, wenn sich Benutzer beschweren oder die monatliche Rechnung eintrifft. Benachrichtigungen machen aus reaktiver Brandbekämpfung einen proaktiven Betrieb.
Die drei Kategorien von Benachrichtigungen
Benachrichtigungen für LLM-Anwendungen lassen sich in drei Kategorien einteilen. Latenzbenachrichtigungen werden ausgelöst, wenn die Antwortzeit einen Schwellenwert für die Benutzererfahrung überschreitet (z. B. p99 > 10 Sekunden). Kostenbenachrichtigungen werden ausgelöst, wenn die Kosten pro Anfrage oder die täglichen Ausgaben Budgetgrenzen überschreiten, und verhindern so böse Überraschungen bei der Abrechnung. Qualitätsbenachrichtigungen werden ausgelöst, wenn automatisierte Qualitätsmetriken (LLM-as-Judge-Bewertungen, Benutzerzufriedenheitsraten, Faktentreuewerte) unter ein akzeptables Mindestniveau fallen. Jede Kategorie erfordert eine andere Instrumentierung und andere Benachrichtigungskanäle.
from dataclasses import dataclass
@dataclass
class AlertThresholds:
# Latency (milliseconds)
p50_latency_ms: int = 2000 # median should be under 2s
p99_latency_ms: int = 10000 # 99th percentile under 10s
# Cost (USD)
max_cost_per_request: float = 0.05 # alert if one request costs > 5 cents
max_daily_spend: float = 50.00 # alert if daily spend exceeds $50
# Quality (0.0 to 1.0 scale)
min_quality_score: float = 0.75 # alert if rolling avg drops below 75%
min_faithfulness: float = 0.80 # alert if RAGAS faithfulness drops below 80%
DEFAULT_THRESHOLDS = AlertThresholds()Latenzmetriken erfassen
Um Benachrichtigungen für die Latenz einzurichten, müssen Sie diese zunächst konsistent erfassen. Messen Sie drei Latenzkomponenten getrennt: die Zeit bis zum ersten Token (TTFT, entscheidend für die Streaming-Benutzererfahrung), die gesamte Antwortzeit und die Latenzen der einzelnen Schritte für Abruf und Generierung. Speichern Sie diese als Zeitreihendaten (einen Datenpunkt pro Anfrage), damit Sie Perzentile und gleitende Mittelwerte für die vergangenen N Minuten oder Stunden berechnen können.
import time
from collections import deque
from statistics import quantiles
class LatencyTracker:
def __init__(self, window_size=100):
self.window = deque(maxlen=window_size) # rolling window of latencies
def record(self, latency_ms: float):
self.window.append(latency_ms)
def p50(self) -> float:
if not self.window:
return 0
return quantiles(self.window, n=100)[49]
def p99(self) -> float:
if not self.window:
return 0
return quantiles(self.window, n=100)[98]
def check_alert(self, thresholds: AlertThresholds) -> list[str]:
alerts = []
if self.p99() > thresholds.p99_latency_ms:
alerts.append(f'p99 latency {self.p99():.0f}ms exceeds {thresholds.p99_latency_ms}ms')
if self.p50() > thresholds.p50_latency_ms:
alerts.append(f'p50 latency {self.p50():.0f}ms exceeds {thresholds.p50_latency_ms}ms')
return alerts
latency_tracker = LatencyTracker()Kosten verfolgen und Benachrichtigungen einrichten
Für die Kostenüberwachung müssen Sie Ausgaben auf mehreren Granularitätsebenen verfolgen: Kosten pro Anfrage (um teure Ausreißerabfragen zu erkennen), stündliche und tägliche Summen (um Nutzungsspitzen zu erkennen) sowie Kosten pro Funktion (um den teuersten Teil Ihrer Anwendung zu identifizieren). Benachrichtigen Sie sofort bei Anomalien bei den Kosten einzelner Anfragen und verwenden Sie geplante Prüfungen für tägliche Budgetgrenzen.
from datetime import datetime, date
import threading
class CostTracker:
def __init__(self):
self._lock = threading.Lock()
self._daily_spend = {} # date -> total USD
self._per_request = [] # list of (timestamp, cost)
def record(self, cost_usd: float) -> list[str]:
today = date.today().isoformat()
alerts = []
with self._lock:
# Track daily spend
self._daily_spend[today] = self._daily_spend.get(today, 0) + cost_usd
self._per_request.append((datetime.now(), cost_usd))
# Alert on expensive single requests
if cost_usd > DEFAULT_THRESHOLDS.max_cost_per_request:
alerts.append(f'Expensive request: ${cost_usd:.4f} (threshold: ${DEFAULT_THRESHOLDS.max_cost_per_request})')
# Alert on daily budget exceeded
daily_total = self._daily_spend[today]
if daily_total > DEFAULT_THRESHOLDS.max_daily_spend:
alerts.append(f'Daily budget exceeded: ${daily_total:.2f} > ${DEFAULT_THRESHOLDS.max_daily_spend}')
return alerts
cost_tracker = CostTracker()Benachrichtigungen für Qualitätsbewertungen
Qualitätsbenachrichtigungen sind die komplexeste Kategorie, da sich Qualität nicht allein anhand von Infrastrukturmetriken messen lässt – dafür ist eine semantische Evaluierung erforderlich. Der am besten skalierbare Ansatz ist die automatisierte Evaluierung anhand von Stichproben: Führen Sie für eine zufällige Stichprobe von Produktionsanfragen (5–10 %) asynchron einen LLM-as-Judge-Evaluator aus, speichern Sie die Bewertungen und berechnen Sie einen gleitenden Mittelwert. Lösen Sie eine Benachrichtigung aus, wenn der gleitende Mittelwert unter Ihr Qualitätsmindestniveau fällt.
import random
from openai import OpenAI
client = OpenAI()
def evaluate_response_quality(question: str, answer: str) -> float:
judge_prompt = f'''Rate the quality of this AI assistant response on a scale from 0 to 1.
Question: {question}
Answer: {answer}
Return only a JSON object: {{"score": 0.85, "reason": "brief reason"}}
Scoring guide:
1.0 = Perfect, accurate, helpful
0.75 = Good, minor issues
0.5 = Acceptable but incomplete
0.25 = Poor, major gaps
0.0 = Completely wrong or harmful'''
response = client.chat.completions.create(
model='gpt-4o-mini', # cheaper model for evaluation
messages=[{'role': 'user', 'content': judge_prompt}],
response_format={'type': 'json_object'}
)
import json
result = json.loads(response.choices[0].message.content)
return float(result['score'])
def maybe_evaluate(question: str, answer: str, sample_rate=0.1):
if random.random() < sample_rate:
score = evaluate_response_quality(question, answer)
quality_tracker.record(score)
return score
return NoneGleitende Mittelwerte zur Trenderkennung
Eine einzelne schlechte Antwort weist nicht auf ein systemisches Problem hin. Verwenden Sie gleitende Mittelwerte über ein Zeitfenster (z. B. die letzte Stunde oder die letzten 100 Anfragen), um Trends zu erkennen. Ein gleitender Mittelwert, der einen Schwellenwert überschreitet und dort mindestens 10 Minuten bleibt, weist auf ein echtes Problem hin, während eine kurze Spitze ein Messrauschen sein kann. Exponentiell gewichtete gleitende Mittelwerte (EWMA) reagieren schneller auf aktuelle Änderungen als einfache gleitende Mittelwerte.
class RollingQualityMonitor:
def __init__(self, window=50, alert_threshold=0.75, ewma_alpha=0.1):
self.scores = []
self.window = window
self.alert_threshold = alert_threshold
self.alpha = ewma_alpha # EWMA decay factor
self.ewma_score = None
def record(self, score: float):
self.scores.append(score)
if len(self.scores) > self.window:
self.scores.pop(0)
# Update exponentially weighted moving average
if self.ewma_score is None:
self.ewma_score = score
else:
self.ewma_score = self.alpha * score + (1 - self.alpha) * self.ewma_score
def should_alert(self) -> bool:
if len(self.scores) < 10: # not enough data yet
return False
return self.ewma_score < self.alert_threshold
def summary(self) -> dict:
if not self.scores:
return {}
return {
'rolling_avg': sum(self.scores) / len(self.scores),
'ewma': self.ewma_score,
'sample_count': len(self.scores),
'alert': self.should_alert()
}
quality_tracker = RollingQualityMonitor()Weiterleitung von Benachrichtigungen: Slack und PagerDuty
Benachrichtigungen sind nur dann nützlich, wenn sie die richtige Person über den richtigen Kanal erreichen. Leiten Sie Benachrichtigungen nach Schweregrad weiter: Benachrichtigungen über Qualitätsverschlechterungen und Kostenüberschreitungen gehen an einen Slack-Kanal, den Ihr Team während der Geschäftszeiten überwacht. Latenzbenachrichtigungen oberhalb kritischer Schwellenwerte (z. B. p99 > 30 Sekunden) und plötzliche Kostenspitzen (z. B. innerhalb von 5 Minuten das Zehnfache des Normalwerts) gehen für eine sofortige Eskalation an den Bereitschaftsdienst an PagerDuty.
import requests
def send_slack_alert(message: str, severity: str, webhook_url: str):
color = {'critical': '#ff0000', 'warning': '#ff9900', 'info': '#36a64f'}[severity]
payload = {
'attachments': [{
'color': color,
'title': f'LLM Alert [{severity.upper()}]',
'text': message,
'footer': 'AI Pipeline Monitor'
}]
}
requests.post(webhook_url, json=payload)
def send_pagerduty_alert(title: str, details: str, routing_key: str):
payload = {
'routing_key': routing_key,
'event_action': 'trigger',
'payload': {
'summary': title,
'severity': 'critical',
'source': 'ai-pipeline-monitor',
'custom_details': {'details': details}
}
}
requests.post('https://events.pagerduty.com/v2/enqueue', json=payload)
def route_alert(alert_text: str, severity: str):
send_slack_alert(alert_text, severity, SLACK_WEBHOOK_URL)
if severity == 'critical':
send_pagerduty_alert(alert_text, alert_text, PAGERDUTY_ROUTING_KEY)Die Auswertungsschleife für Benachrichtigungen
Die Logik für Benachrichtigungen sollte in einer geplanten Schleife ausgeführt werden, nicht inline bei der Verarbeitung von Anfragen. Durch die asynchrone Ausführung der Prüfungen wird verhindert, dass Produktionsanfragen zusätzlich verlangsamt werden. Für die meisten Anforderungen an Benachrichtigungen genügt ein Hintergrundthread oder ein geplanter Job, der alle 60 Sekunden ausgeführt wird. Erfassen Sie Metriken im Anfragepfad und werten Sie die Schwellenwerte in der Hintergrundschleife aus.
import threading
import time
def alert_evaluation_loop(interval_seconds=60):
while True:
try:
# Check latency
latency_alerts = latency_tracker.check_alert(DEFAULT_THRESHOLDS)
for alert in latency_alerts:
route_alert(f'LATENCY: {alert}', 'warning')
# Check quality
quality_summary = quality_tracker.summary()
if quality_summary.get('alert'):
msg = f'QUALITY DEGRADATION: Rolling avg {quality_summary["ewma"]:.2f} below threshold {DEFAULT_THRESHOLDS.min_quality_score}'
route_alert(msg, 'warning')
print(f'Alert check complete. Quality: {quality_summary.get("ewma", "N/A")}')
except Exception as e:
print(f'Alert evaluation error: {e}')
time.sleep(interval_seconds)
# Start in background thread
alert_thread = threading.Thread(target=alert_evaluation_loop, daemon=True)
alert_thread.start()Benachrichtigungsmüdigkeit und Abstimmung von Schwellenwerten
Alarmmüdigkeit tritt auf, wenn Alarme so häufig ausgelöst werden, dass das Team sie zu ignorieren beginnt. Verhindern Sie dies, indem Sie mit konservativen (hohen) Schwellenwerten beginnen und diese anpassen, sobald Sie Ihre Ausgangswerte kennen, Alarme erst auslösen, wenn die Bedingung über mehrere Auswertungszyklen hinweg anhält, zusammengehörige Alarme in einzelnen Benachrichtigungen bündeln und regelmäßig Alarme überprüfen und deaktivieren, die nie zu sinnvollen Maßnahmen führen.
class AlertDebouncer:
def __init__(self, required_consecutive_fires=3):
self.required = required_consecutive_fires
self.fire_counts = {} # alert_name -> consecutive fires
self.already_firing = set() # alert_names currently active
def should_fire(self, alert_name: str, condition: bool) -> bool:
if condition:
self.fire_counts[alert_name] = self.fire_counts.get(alert_name, 0) + 1
if self.fire_counts[alert_name] >= self.required and alert_name not in self.already_firing:
self.already_firing.add(alert_name)
return True # fire the alert (first time condition sustained)
else:
if self.fire_counts.get(alert_name, 0) > 0:
self.fire_counts[alert_name] = 0
self.already_firing.discard(alert_name) # condition cleared
return False # either not sustained or already firing (no duplicate alert)
debouncer = AlertDebouncer(required_consecutive_fires=3)Kostenanomalien mit statistischen Methoden überwachen
Einfache Schwellenwertalarme erkennen subtile Kostenanomalien nicht, etwa einen allmählichen Kostenanstieg um 50 % innerhalb von zwei Tagen. Verwenden Sie die statistische Anomalieerkennung: Berechnen Sie den gleitenden Mittelwert und die Standardabweichung Ihrer stündlichen Kosten und lösen Sie einen Alarm aus, wenn die Kosten der aktuellen Stunde um mehr als N Standardabweichungen über dem Mittelwert liegen (Z-Score-Überwachung). Dadurch passt sich die Überwachung automatisch an natürliche Nutzungsmuster wie den Datenverkehr an Wochentagen und Wochenenden an.
import statistics
def compute_z_score(recent_value: float, historical_values: list[float]) -> float:
if len(historical_values) < 5:
return 0 # not enough data
mean = statistics.mean(historical_values)
stdev = statistics.stdev(historical_values)
if stdev == 0:
return 0
return (recent_value - mean) / stdev
def check_cost_anomaly(current_hour_cost: float, historical_hourly_costs: list[float]) -> str | None:
z = compute_z_score(current_hour_cost, historical_hourly_costs)
if z > 3.0: # more than 3 standard deviations above mean
mean = statistics.mean(historical_hourly_costs)
return f'Cost anomaly: ${current_hour_cost:.2f} this hour (normal: ${mean:.2f}, z={z:.1f})'
return NoneEin Ops-Dashboard erstellen
Alarme bilden die reaktive Ebene; ein Live-Betriebsdashboard ist die proaktive Ebene. Erstellen Sie ein einfaches Dashboard mit Echtzeitmetriken: aktuelle p50-/p99-Latenz, Anfragen pro Minute, aktueller Qualitätswert (gleitende EWMA), bisherige Tageskosten im Vergleich zum Budget und die fünf teuersten Anfragen der letzten Stunde. So erhält Ihr Team auf einen Blick Einblick in den Systemzustand, ohne auf das Auslösen eines Alarms warten zu müssen.
Schnelltest
Testen Sie Ihr Verständnis der Alarmierung für Metriken von LLM-Anwendungen aus dieser Lektion.
Zusammenfassung der Lektion
In dieser Lektion haben Sie gelernt: Drei Alarmkategorien decken LLM-Systeme ab — Latenz, Kosten und Qualität — und erfordern jeweils eine unterschiedliche Instrumentierung. Gleitende Mittelwerte und EWMA erkennen eine allmähliche Qualitätsverschlechterung besser als Schwellenwertprüfungen einzelner Anfragen. Alarm-Entprellung verhindert Alarmmüdigkeit, indem Bedingungen über mehrere Auswertungszyklen hinweg bestehen müssen, bevor ein Alarm ausgelöst wird. Als Nächstes befassen wir uns mit Prompt-Injection-Angriffen und KI-Sicherheit.
Häufig gestellte Fragen
Ist die Lektion „Alarme bei Latenz, Kosten und Qualitätsverlust“ kostenlos?
Ja — der vollständige Text von „Alarme bei Latenz, Kosten und Qualitätsverlust“ ist hier im Web kostenlos zu lesen. Um sie interaktiv zu üben (integrierter Code-Editor und 24/7 KI-Tutor) und den Rest des AI Engineering Academy-Kurses freizuschalten, upgrade auf CoddyKit PRO. Der AI Engineering Academy-Kurs umfasst insgesamt 4 Lektionen.
Was lerne ich in „Alarme bei Latenz, Kosten und Qualitätsverlust“?
Definieren Sie Alarmschwellen für die p99-Latenz, die Kosten pro Anfrage und automatisierte Qualitätswerte und leiten Sie Alarme an Slack oder PagerDuty weiter, wenn sich Ihre LLM-Pipeline verschlech… Du übst AI Engineering Academy mit praktischem Code, den du direkt im Browser ausführst, und ein 24/7 KI-Tutor beantwortet deine Fragen während du die Lektion bearbeitest.
Brauche ich Erfahrung, um AI Engineering Academy zu starten?
Keine Vorkenntnisse erforderlich. AI Engineering Academy auf CoddyKit ist für Anfänger bis fortgeschrittene Lernende strukturiert, sodass du hier starten oder von Anfang an beginnen und in deinem eigenen Tempo voranschreiten kannst. Dies ist Lektion 4 von 4.
Wie lange dauert die Lektion „Alarme bei Latenz, Kosten und Qualitätsverlust“?
Die meisten CoddyKit-Lektionen dauern etwa 5–10 Minuten. Jede ist kompakt und interaktiv, sodass du stetig Fortschritte machst und genau dort weitermachst, wo du aufgehört hast – im Web und in der App.
Kann ich in dieser AI Engineering Academy-Lektion Code schreiben und ausführen?
Ja. Jede AI Engineering Academy-Lektion enthält einen integrierten Code-Editor, sodass du echten Code direkt in deinem Browser schreibst und ausführst und sofort KI-Feedback erhältst — ohne lokale Einrichtung erforderlich.
Alle Lektionen in diesem Kurs
- Warum sich LLM-Anwendungen schwer debuggen lassen
- Tracing mit LangSmith
- Langfuse für modellunabhängige Observability
- Alarme bei Latenz, Kosten und Qualitätsverlust