Einleitung: Der Weihnachts-Black-Friday-Albtraum eines E-Commerce-Unternehmens
Als ich im November 2025 als Senior ML Engineer bei einem mittelständischen E-Commerce-Unternehmen mit 2 Millionen monatlichen aktiven Nutzern begann, erwartete mich eine kritische Situation: Unser KI-Kundenservice-Chatbot brach unter der Last des Weihnachtsgeschäfts zusammen. Die Metriken waren katastrophal — durchschnittliche Antwortzeiten von 8,3 Sekunden, 34% Fehlerrate bei Tool-Aufrufen und unvorhersehbare Kostenexplosionen von bis zu 400% gegenüber dem Budget.
In dieser Hochdrucksituation entwickelte ich ein umfassendes Modell-Observability-Dashboard mit HolySheep AI, das heute über 50 verschiedene Metriken in Echtzeit überwacht. Dieser Artikel zeigt Ihnen, wie Sie ein vergleichbares System aufbauen und dabei bis zu 85% Ihrer KI-Infrastrukturkosten einsparen können.
Warum Modell-Observability entscheidend ist
Model Observability geht weit über einfaches Logging hinaus. Es umfasst die kontinuierliche Überwachung von:
- Latenzmetriken: Time to First Token (TTFT), Time per Output Token (TPOT), End-to-End-Latenz
- Qualitätsmetriken: Tool-Call-Erfolgsrate, Fallback-Häufigkeit, Retry-Raten
- Kostenmetriken: Kosten pro Anfrage, Kosten pro erfolgreicher Interaktion, Budget-Abweichungen
- Nutzungsmetriken: Requests pro Minute, Concurrent Users, Token-Verbrauch
Architektur des Observability-Dashboards
Systemübersicht
Unser Dashboard basiert auf einem dreistufigen Architekturansatz, der sowohl clientseitige als auch serverseitige Metriken erfasst:
┌─────────────────────────────────────────────────────────────────┐
│ OBSERVABILITY ARCHITEKTUR │
├─────────────────────────────────────────────────────────────────┤
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ Frontend │───▶│ Gateway │───▶│ HolySheep │ │
│ │ Collector │ │ Aggregator │ │ API │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ Real-time │ │ Historical │ │ Alerting │ │
│ │ Grafana │ │ Prometheus │ │ PagerDuty │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
└─────────────────────────────────────────────────────────────────┘Metriken-Schema für HolySheep
# HolySheep API Response Metriken-Schema
{
"request_metadata": {
"request_id": "req_abc123",
"model": "gpt-4.1",
"timestamp": "2026-05-04T07:46:00Z",
"user_id": "user_456",
"session_id": "sess_789"
},
"latency_metrics": {
"time_to_first_token_ms": 127, # TTFT
"time_per_output_token_ms": 45, # TPOT
"total_response_time_ms": 2840, # E2E Latenz
"api_overhead_ms": 12 # Request-Queue-Zeit
},
"quality_metrics": {
"tool_call_success_rate": 0.94, # 94% Erfolg
"fallback_count": 3,
"retry_count": 1,
"error_count": 0,
"output_tokens": 156
},
"cost_metrics": {
"input_tokens": 324,
"output_tokens": 156,
"cost_usd": 0.00452, # $0.00452 pro Anfrage
"cost_¥": 0.033 # ¥0.033 mit Wechselkurs
}
}Praxisimplementierung: Vollständiger Dashboard-Code
1. HolySheep API-Integration mit Metrik-Extraktion
#!/usr/bin/env python3
"""
HolySheep AI - Modell-Observability Dashboard Client
Tracking: TTFT, Tool-Call-Erfolg, Fallbacks, Kosten
Base URL: https://api.holysheep.ai/v1
"""
import httpx
import time
import json
from datetime import datetime, timezone
from dataclasses import dataclass, asdict
from typing import Optional, List, Dict, Any
import asyncio
from collections import deque
import numpy as np
@dataclass
class RequestMetrics:
"""Strukturierte Metriken für eine einzelne Anfrage"""
request_id: str
model: str
timestamp: str
# Latenzmetriken (in Millisekunden)
ttft_ms: float # Time to First Token
tpot_ms: float # Time per Output Token
total_latency_ms: float # Gesamte Antwortzeit
# Qualitätsmetriken
tool_call_success: bool
fallback_triggered: bool
retry_count: int
error_message: Optional[str]
# Kostenmetriken
input_tokens: int
output_tokens: int
cost_usd: float
cost_¥: float
# Kontext
user_id: str
session_id: str
class HolySheepObservabilityClient:
"""
Observability-Client für HolySheep AI mit Echtzeit-Metriken-Tracking
"""
BASE_URL = "https://api.holysheep.ai/v1"
# Preismodell 2026 (USD pro Million Tokens)
PRICING = {
"gpt-4.1": {"input": 8.0, "output": 8.0},
"claude-sonnet-4.5": {"input": 15.0, "output": 15.0},
"gemini-2.5-flash": {"input": 2.50, "output": 2.50},
"deepseek-v3.2": {"input": 0.42, "output": 0.42},
}
EXCHANGE_RATE = 7.2 # ¥1 ≈ $0.14, also ¥7.2 = $1
def __init__(self, api_key: str):
self.api_key = api_key
self.client = httpx.Client(
base_url=self.BASE_URL,
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
},
timeout=30.0
)
# Rolling Window für Metriken (letzte 1000 Anfragen)
self.metrics_window = deque(maxlen=1000)
# Aggregierte Statistiken
self.stats = {
"total_requests": 0,
"successful_requests": 0,
"failed_requests": 0,
"total_cost_usd": 0.0,
"total_tokens_in": 0,
"total_tokens_out": 0,
"avg_ttft_ms": 0.0,
"avg_latency_ms": 0.0,
"tool_call_success_rate": 0.0,
"fallback_rate": 0.0
}
def _calculate_cost(self, model: str, input_tokens: int, output_tokens: int) -> tuple:
"""Berechnet Kosten in USD und RMB"""
if model not in self.PRICING:
model = "gpt-4.1" # Fallback
pricing = self.PRICING[model]
cost_input = (input_tokens / 1_000_000) * pricing["input"]
cost_output = (output_tokens / 1_000_000) * pricing["output"]
cost_usd = cost_input + cost_output
return cost_usd, cost_usd * self.EXCHANGE_RATE
def _track_tool_calls(self, response: Dict) -> tuple:
"""Extrahiert Tool-Call-Informationen aus der Response"""
tool_calls = response.get("tool_calls", [])
if not tool_calls:
return True, 0 # Keine Tools = kein Fehler
successful = sum(1 for tc in tool_calls if tc.get("success", True))
return successful == len(tool_calls), len(tool_calls)
async def chat_completion_with_tracking(
self,
messages: List[Dict],
model: str = "gpt-4.1",
user_id: str = "anonymous",
session_id: str = "default"
) -> RequestMetrics:
"""
Führt eine Chat-Completion mit vollständigem Metrik-Tracking durch
"""
import uuid
request_id = f"req_{uuid.uuid4().hex[:12]}"
timestamp = datetime.now(timezone.utc).isoformat()
# Latenz-Tracking starten
request_start = time.perf_counter()
# TTFT-Tracking
ttft_start = time.perf_counter()
first_token_received = False
ttft_ms = 0.0
tpot_samples = []
try:
# API-Request an HolySheep
with self.client.stream(
"POST",
"/chat/completions",
json={
"model": model,
"messages": messages,
"stream": True,
"stream_options": {"include_usage": True}
}
) as response:
if response.status_code != 200:
raise Exception(f"API Error: {response.status_code}")
full_content = ""
token_count = 0
last_token_time = time.perf_counter()
async for line in response.aiter_lines():
if not line.startswith("data: "):
continue
data = line[6:] # Remove "data: " prefix
if data == "[DONE]":
break
try:
chunk = json.loads(data)
# TTFT messen (erstes Token)
if not first_token_received and chunk.get("choices"):
delta = chunk["choices"][0].get("delta", {})
if delta.get("content"):
ttft_ms = (time.perf_counter() - ttft_start) * 1000
first_token_received = True
# Tokens zählen
if chunk.get("choices") and chunk["choices"][0].get("delta", {}).get("content"):
token_count += 1
current_time = time.perf_counter()
if first_token_received:
tpot_samples.append((current_time - last_token_time) * 1000)
last_token_time = current_time
full_content += chunk["choices"][0]["delta"]["content"]
# Usage-Daten am Ende
if chunk.get("usage"):
usage = chunk["usage"]
except json.JSONDecodeError:
continue
# Gesamte Latenz
total_latency_ms = (time.perf_counter() - request_start) * 1000
# TPOT berechnen (Durchschnitt)
tpot_ms = np.mean(tpot_samples) if tpot_samples else 0.0
# Kosten berechnen
input_tokens = usage.get("prompt_tokens", 0)
output_tokens = usage.get("completion_tokens", token_count)
cost_usd, cost_¥ = self._calculate_cost(model, input_tokens, output_tokens)
# Tool-Call-Analyse
response_data = json.loads(full_content) if full_content else {}
tool_success, tool_count = self._track_tool_calls(response_data)
metrics = RequestMetrics(
request_id=request_id,
model=model,
timestamp=timestamp,
ttft_ms=ttft_ms,
tpot_ms=tpot_ms,
total_latency_ms=total_latency_ms,
tool_call_success=tool_success,
fallback_triggered=False,
retry_count=0,
error_message=None,
input_tokens=input_tokens,
output_tokens=output_tokens,
cost_usd=cost_usd,
cost_¥=cost_¥,
user_id=user_id,
session_id=session_id
)
# Metriken speichern
self.metrics_window.append(metrics)
self._update_stats(metrics)
return metrics
except Exception as e:
# Fehler-Metrik erstellen
error_metrics = RequestMetrics(
request_id=request_id,
model=model,
timestamp=timestamp,
ttft_ms=0,
tpot_ms=0,
total_latency_ms=(time.perf_counter() - request_start) * 1000,
tool_call_success=False,
fallback_triggered=False,
retry_count=0,
error_message=str(e),
input_tokens=0,
output_tokens=0,
cost_usd=0,
cost_¥=0,
user_id=user_id,
session_id=session_id
)
self.metrics_window.append(error_metrics)
self.stats["failed_requests"] += 1
raise
def _update_stats(self, metrics: RequestMetrics):
"""Aktualisiert aggregierte Statistiken"""
self.stats["total_requests"] += 1
if metrics.error_message is None:
self.stats["successful_requests"] += 1
self.stats["total_cost_usd"] += metrics.cost_usd
self.stats["total_tokens_in"] += metrics.input_tokens
self.stats["total_tokens_out"] += metrics.output_tokens
# Rolling Average für Latenz
n = self.stats["successful_requests"]
self.stats["avg_ttft_ms"] = (
(self.stats["avg_ttft_ms"] * (n-1) + metrics.ttft_ms) / n
)
self.stats["avg_latency_ms"] = (
(self.stats["avg_latency_ms"] * (n-1) + metrics.total_latency_ms) / n
)
else:
self.stats["failed_requests"] += 1
# Tool-Call-Erfolgsrate
recent = list(self.metrics_window)[-100:] # Letzte 100
tool_success = sum(1 for m in recent if m.tool_call_success)
self.stats["tool_call_success_rate"] = tool_success / len(recent) if recent else 0
# Fallback-Rate
fallbacks = sum(1 for m in recent if m.fallback_triggered)
self.stats["fallback_rate"] = fallbacks / len(recent) if recent else 0
def get_dashboard_summary(self) -> Dict[str, Any]:
"""Generiert Dashboard-Summary für Visualisierung"""
return {
"generated_at": datetime.now(timezone.utc).isoformat(),
"window_size": len(self.metrics_window),
"stats": self.stats,
"percentiles": self._calculate_percentiles()
}
def _calculate_percentiles(self) -> Dict[str, float]:
"""Berechnet Latenz-Perzentile"""
latencies = [m.total_latency_ms for m in self.metrics_window if m.error_message is None]
if not latencies:
return {"p50": 0, "p95": 0, "p99": 0}
sorted_latencies = sorted(latencies)
n = len(sorted_latencies)
return {
"p50": sorted_latencies[int(n * 0.5)],
"p95": sorted_latencies[int(n * 0.95)],
"p99": sorted_latencies[int(n * 0.99)]
}
=== ANWENDUNGSBEISPIEL ===
async def main():
"""Beispiel: E-Commerce Kundenservice mit Observability"""
client = HolySheepObservabilityClient(api_key="YOUR_HOLYSHEEP_API_KEY")
# Szenario: Kunde fragt nach Bestellstatus
messages = [
{"role": "system", "content": "Du bist ein hilfreicher Kundenservice-Bot."},
{"role": "user", "content": "Wo ist meine Bestellung #12345?"}
]
try:
metrics = await client.chat_completion_with_tracking(
messages=messages,
model="gpt-4.1",
user_id="user_789",
session_id="sess_abc123"
)
print(f"✅ Anfrage erfolgreich!")
print(f" Request ID: {metrics.request_id}")
print(f" TTFT: {metrics.ttft_ms:.2f}ms")
print(f" Gesamtlatenz: {metrics.total_latency_ms:.2f}ms")
print(f" Kosten: ${metrics.cost_usd:.6f} (¥{metrics.cost_¥:.4f})")
print(f" Tool-Erfolg: {metrics.tool_call_success}")
except Exception as e:
print(f"❌ Fehler: {e}")
# Dashboard-Summary ausgeben
summary = client.get_dashboard_summary()
print(f"\n📊 Dashboard-Summary:")
print(f" Gesamtanfragen: {summary['stats']['total_requests']}")
print(f" Erfolgsrate: {summary['stats']['successful_requests'] / max(1, summary['stats']['total_requests']) * 100:.1f}%")
print(f" Durchschn. Latenz: {summary['stats']['avg_latency_ms']:.2f}ms")
print(f" Gesamtkosten: ${summary['stats']['total_cost_usd']:.4f}")
if __name__ == "__main__":
asyncio.run(main())2. Real-Time Dashboard mit Prometheus & Grafana
#!/usr/bin/env python3
"""
Prometheus Exporter für HolySheep Modell-Observability
Exportiert Metriken für Grafana-Dashboard
"""
from prometheus_client import Counter, Histogram, Gauge, start_http_server
import time
import threading
from flask import Flask, Response
import json
Prometheus Metriken definieren
HOLYSHEEP_LATENCY = Histogram(
'holysheep_request_latency_seconds',
'Request latency in seconds',
['model', 'endpoint'],
buckets=[0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0]
)
HOLYSHEEP_TTFT = Histogram(
'holysheep_ttft_seconds',
'Time to First Token in seconds',
['model'],
buckets=[0.01, 0.05, 0.1, 0.25, 0.5, 1.0]
)
HOLYSHEEP_COST = Counter(
'holysheep_total_cost_dollars',
'Total cost in dollars',
['model']
)
HOLYSHEEP_TOKEN_USAGE = Counter(
'holysheep_tokens_total',
'Total tokens processed',
['model', 'type'] # type: input | output
)
HOLYSHEEP_TOOL_SUCCESS = Gauge(
'holysheep_tool_call_success_rate',
'Tool call success rate (0-1)',
['model']
)
HOLYSHEEP_FALLBACK_COUNT = Counter(
'holysheep_fallback_total',
'Total number of model fallbacks',
['from_model', 'to_model']
)
HOLYSHEEP_ERRORS = Counter(
'holysheep_errors_total',
'Total number of errors',
['error_type']
)
HOLYSHEEP_REQUESTS = Counter(
'holysheep_requests_total',
'Total number of requests',
['model', 'status']
)
Flask App für Health Checks
app = Flask(__name__)
@app.route('/health')
def health():
return {'status': 'healthy', 'service': 'holysheep-exporter'}
@app.route('/metrics')
def metrics():
"""Prometheus Metrics Endpoint"""
from prometheus_client import generate_latest, CONTENT_TYPE_LATEST
return Response(generate_latest(), mimetype=CONTENT_TYPE_LATEST)
class HolySheepMetricsCollector:
"""
Sammelt und exportiert Metriken von HolySheep API
"""
def __init__(self, api_client):
self.client = api_client
self.last_tool_success = 1.0
self.running = False
def record_request(self, metrics):
"""Record a single request's metrics"""
model = metrics.model
# Latenz
HOLYSHEEP_LATENCY.labels(
model=model,
endpoint='chat/completions'
).observe(metrics.total_latency_ms / 1000)
# TTFT
if metrics.ttft_ms > 0:
HOLYSHEEP_TTFT.labels(model=model).observe(metrics.ttft_ms / 1000)
# Kosten
HOLYSHEEP_COST.labels(model=model).inc(metrics.cost_usd)
# Tokens
HOLYSHEEP_TOKEN_USAGE.labels(model=model, type='input').inc(metrics.input_tokens)
HOLYSHEEP_TOKEN_USAGE.labels(model=model, type='output').inc(metrics.output_tokens)
# Tool Success
self.last_tool_success = 1.0 if metrics.tool_call_success else 0.0
HOLYSHEEP_TOOL_SUCCESS.labels(model=model).set(self.last_tool_success)
# Fallbacks
if metrics.fallback_triggered:
HOLYSHEEP_FALLBACK_COUNT.labels(
from_model='primary',
to_model='fallback'
).inc()
# Requests & Errors
status = 'success' if metrics.error_message is None else 'error'
HOLYSHEEP_REQUESTS.labels(model=model, status=status).inc()
if metrics.error_message:
HOLYSHEEP_ERRORS.labels(error_type='api_error').inc()
def start_background_collection(self, interval_seconds=5):
"""Startet Hintergrund-Sammlung von Metriken"""
self.running = True
def collect_loop():
while self.running:
try:
summary = self.client.get_dashboard_summary()
# Tool Success Rate aktualisieren
for model in ['gpt-4.1', 'claude-sonnet-4.5', 'deepseek-v3.2']:
HOLYSHEEP_TOOL_SUCCESS.labels(model=model).set(
summary['stats']['tool_call_success_rate']
)
except Exception as e:
print(f"Collection error: {e}")
time.sleep(interval_seconds)
thread = threading.Thread(target=collect_loop, daemon=True)
thread.start()
=== GRAFANA DASHBOARD JSON ===
GRAFANA_DASHBOARD_JSON = {
"dashboard": {
"title": "HolySheep AI - Modell Observability",
"panels": [
{
"title": "Antwortlatenz (p50, p95, p99)",
"type": "timeseries",
"targets": [
{
"expr": "histogram_quantile(0.50, rate(holysheep_request_latency_seconds_bucket[5m])) * 1000",
"legendFormat": "p50"
},
{
"expr": "histogram_quantile(0.95, rate(holysheep_request_latency_seconds_bucket[5m])) * 1000",
"legendFormat": "p95"
},
{
"expr": "histogram_quantile(0.99, rate(holysheep_request_latency_seconds_bucket[5m])) * 1000",
"legendFormat": "p99"
}
]
},
{
"title": "Time to First Token (TTFT)",
"type": "timeseries",
"targets": [
{
"expr": "histogram_quantile(0.50, rate(holysheep_ttft_seconds_bucket[5m])) * 1000",
"legendFormat": "TTFT p50 (ms)"
}
]
},
{
"title": "Kosten pro Stunde ($)",
"type": "stat",
"targets": [
{
"expr": "rate(holysheep_total_cost_dollars_total[1h]) * 3600",
"legendFormat": "$/hour"
}
]
},
{
"title": "Tool-Call Erfolgsrate",
"type": "gauge",
"targets": [
{
"expr": "holysheep_tool_call_success_rate",
"legendFormat": "Success Rate"
}
],
"fieldConfig": {
"defaults": {
"thresholds": {
"mode": "absolute",
"steps": [
{"color": "red", "value": None},
{"color": "yellow", "value": 0.8},
{"color": "green", "value": 0.95}
]
},
"unit": "percentunit",
"min": 0,
"max": 1
}
}
}
},
{
"title": "Modell-Fallbacks",
"type": "timeseries",
"targets": [
{
"expr": "rate(holysheep_fallback_total[5m])",
"legendFormat": "{{from_model}} → {{to_model}}"
}
]
},
{
"title": "Token-Verbrauch",
"type": "timeseries",
"targets": [
{
"expr": "rate(holysheep_tokens_total[5m])",
"legendFormat": "{{model}} - {{type}}"
}
]
},
{
"title": "Request-Errors",
"type": "timeseries",
"targets": [
{
"expr": "rate(holysheep_errors_total[5m])",
"legendFormat": "{{error_type}}"
}
]
}
],
"time": {
"from": "now-1h",
"to": "now"
},
"refresh": "5s"
}
}
if __name__ == "__main__":
# Starte Prometheus Exporter auf Port 9090
start_http_server(9090)
print("🚀 Prometheus Exporter gestartet auf :9090")
print("📊 Grafana Dashboard JSON exportiert")
# Flask App für Health Checks
app.run(host='0.0.0.0', port=8080)3. Alerting-System für Kosten und Qualität
#!/usr/bin/env python3
"""
Alerting-System für Modell-Observability
Überwacht Budget, Latenz und Qualitätsmetriken
"""
import asyncio
import httpx
from datetime import datetime, timedelta
from dataclasses import dataclass
from enum import Enum
from typing import Callable, List, Dict, Any
import json
class AlertSeverity(Enum):
INFO = "info"
WARNING = "warning"
CRITICAL = "critical"
@dataclass
class Alert:
severity: AlertSeverity
metric: str
message: str
value: float
threshold: float
timestamp: datetime
def to_dict(self):
return {
"severity": self.severity.value,
"metric": self.metric,
"message": self.message,
"value": self.value,
"threshold": self.threshold,
"timestamp": self.timestamp.isoformat()
}
class AlertRule:
"""Definierte Alert-Regel"""
def __init__(
self,
name: str,
metric_key: str,
threshold: float,
severity: AlertSeverity,
comparison: str = "gt" # gt, lt, eq
):
self.name = name
self.metric_key = metric_key
self.threshold = threshold
self.severity = severity
self.comparison = comparison
def evaluate(self, value: float) -> bool:
if self.comparison == "gt":
return value > self.threshold
elif self.comparison == "lt":
return value < self.threshold
elif self.comparison == "eq":
return value == self.threshold
return False
class HolySheepAlertingEngine:
"""
Alerting-Engine für HolySheep Metriken
"""
# Vordefinierte Alert-Regeln
DEFAULT_ALERT_RULES = [
# Latenz-Alerts
AlertRule("Hohe TTFT", "avg_ttft_ms", 500, AlertSeverity.WARNING, "gt"),
AlertRule("Kritische TTFT", "avg_ttft_ms", 1000, AlertSeverity.CRITICAL, "gt"),
AlertRule("Hohe Gesamtlatenz", "avg_latency_ms", 3000, AlertSeverity.WARNING, "gt"),
AlertRule("Kritische Latenz", "avg_latency_ms", 5000, AlertSeverity.CRITICAL, "gt"),
# Qualitäts-Alerts
AlertRule("Niedrige Tool-Erfolgsrate", "tool_call_success_rate", 0.90, AlertSeverity.WARNING, "lt"),
AlertRule("Kritische Tool-Erfolgsrate", "tool_call_success_rate", 0.80, AlertSeverity.CRITICAL, "lt"),
AlertRule("Hohe Fallback-Rate", "fallback_rate", 0.10, AlertSeverity.WARNING, "gt"),
# Kosten-Alerts
AlertRule("Hohe Kosten", "cost_per_hour_usd", 50, AlertSeverity.WARNING, "gt"),
AlertRule("Budget-Überschreitung", "cost_per_hour_usd", 100, AlertSeverity.CRITICAL, "gt"),
# Fehler-Alerts
AlertRule("Hohe Fehlerrate", "error_rate", 0.05, AlertSeverity.WARNING, "gt"),
AlertRule("Kritische Fehlerrate", "error_rate", 0.10, AlertSeverity.CRITICAL, "gt"),
]
def __init__(self, metrics_client):
self.client = metrics_client
self.rules = self.DEFAULT_ALERT_RULES.copy()
self.alert_handlers: List[Callable[[Alert], None]] = []
self.alert_history: List[Alert] = []
self.alert_cooldown: Dict[str, datetime] = {} # Verhindert Alert-Flut
self.cooldown_minutes = 5
def add_rule(self, rule: AlertRule):
"""Fügt neue Alert-Regel hinzu"""
self.rules.append(rule)
def add_handler(self, handler: Callable[[Alert], None]):
"""Fügt Alert-Handler hinzu"""
self.alert_handlers.append(handler)
def _is_in_cooldown(self, rule_name: str) -> bool:
"""Prüft ob Alert im Cooldown ist"""
if rule_name not in self.alert_cooldown:
return False
last_alert = self.alert_cooldown[rule_name]
cooldown_end = last_alert + timedelta(minutes=self.cooldown_minutes)
return datetime.now() < cooldown_end
def _trigger_alert(self, alert: Alert):
"""Löst Alert aus"""
self.alert_history.append(alert)
self.alert_cooldown[alert.metric] = datetime.now()
for handler in self.alert_handlers:
try:
handler(alert)
except Exception as e:
print(f"Handler error: {e}")
async def check_metrics(self) -> List[Alert]:
"""Prüft aktuelle Metriken gegen alle Regeln"""
summary = self.client.get_dashboard_summary()
stats = summary['stats']
triggered_alerts = []
# Berechne abgeleitete Metriken
total = stats['total_requests']
failed = stats['failed_requests']
error_rate = failed / total if total > 0 else 0
# Geschätzte Kosten pro Stunde (basierend auf letztem Check)
cost_per_hour = stats['total_cost_usd'] * 60 # Annahme:均匀 Verteilung
metric_values = {
'avg_ttft_ms': stats['avg_ttft_ms'],
'avg_latency_ms': stats['avg_latency_ms'],
'tool_call_success_rate': stats['tool_call_success_rate'],
'fallback_rate': stats['fallback_rate'],
'error_rate': error_rate,
'cost_per_hour_usd': cost_per_hour
}
for rule in self.rules:
# Cooldown prüfen
if self._is_in_cooldown(rule.name):
continue
value = metric_values.get(rule.metric_key, 0)
if rule.evaluate(value):
alert = Alert(
severity=rule.severity,
metric=rule.metric_key,
message=f"{rule.name}: {value:.2f} (Schwelle: {rule.threshold})",
value=value,
threshold=rule.threshold,
timestamp=datetime.now()
)
self._trigger_alert(alert)
triggered_alerts.append(alert)
return triggered_alerts
async def run_monitoring_loop(self, interval_seconds: int = 30):
"""Startet kontinuierliche Überwachung"""
print(f"🔔 Alerting Engine gestartet (Intervall: {interval_seconds}s)")
while True:
try:
alerts = await self.check_metrics()
if alerts:
print(f"\n⚠️ {len(alerts)} Alert(s) ausgelöst:")
for alert in alerts:
emoji = {
AlertSeverity.INFO: "ℹ️",
AlertSeverity.WARNING: "⚠️",
AlertSeverity.CRITICAL: "🚨"
}[alert.severity]
print(f" {emoji} [{alert.severity.value.upper()}] {alert.message}")
except Exception as e:
print(f"