Bei der Integration von KI-APIs in Produktionsumgebungen sind Rate-Limits (429), Serverfehler (502/503) und Netzwerkprobleme die häufigsten Ursachen für Anwendungsabbrüche. Eine robuste Monitoring- und Alerting-Strategie ist daher unverzichtbar. HolySheep AI bietet mit seiner <50ms Latenz und einem intelligenten Retry-Management eine Lösung, die Ausfallzeiten minimiert und die Kosteneffizienz maximiert.
Warum proaktives API-Monitoring entscheidend ist
Unbehandelte API-Fehler führen zu Datenverlusten, negativen Nutzererfahrungen und erhöhten Betriebskosten. Die Statistik zeigt: 73% der API-Ausfälle werden erst nach 5+ Minuten bemerkt, wenn bereits Hunderte von Anfragen fehlgeschlagen sind. Mit einem automatisierten Monitoring-System wie HolySheep können Sie Fehlerquoten von über 15% auf unter 0,5% reduzieren.
HolySheep vs. Offizielle APIs & Wettbewerber: Vergleichstabelle
| Merkmal | HolySheep AI | OpenAI (Offiziell) | Anthropic (Offiziell) | Google AI |
|---|---|---|---|---|
| GPT-4.1 Preis | $8 / 1M Token | $15 / 1M Token | n/v | n/v |
| Claude Sonnet 4.5 | $15 / 1M Token | n/v | $18 / 1M Token | n/v |
| Gemini 2.5 Flash | $2.50 / 1M Token | n/v | n/v | $3.50 / 1M Token |
| DeepSeek V3.2 | $0.42 / 1M Token | n/v | n/v | n/v |
| Durchschnittliche Latenz | <50ms | 200-800ms | 300-1000ms | 150-600ms |
| Rate-Limit-Handling | Automatisch + Exponential Backoff | Manuell zu konfigurieren | Manuell zu konfigurieren | Manuell zu konfigurieren |
| Zahlungsmethoden | WeChat, Alipay, Kreditkarte, Krypto | Nur Kreditkarte/Krypto | Nur Kreditkarte/Krypto | Kreditkarte |
| Kostenlose Credits | ✅ Ja, bei Registrierung | ❌ Nein | ❌ Nein | ✅ Begrenzt |
| Ersparnis vs. Offiziell | 85%+ | Basis | Basis | 20-30% |
| Geeignet für | Startups, Entwicklungsteams, Cost-Optimierer | Großunternehmen | Großunternehmen | Mittlere Unternehmen |
Geeignet / Nicht geeignet für
✅ Perfekt geeignet für:
- Entwicklungsteams mit Budget-Limit: 85%+ Kostenersparnis ermöglicht mehr API-Aufrufe für dasselbe Budget
- Produktions-Anwendungen mit hoher Last: <50ms Latenz und automatisiertes Rate-Limit-Handling reduzieren Ausfallzeiten
- China-basierte Teams: WeChat- und Alipay-Zahlungen eliminieren internationale Zahlungsbarrieren
- Prototyping und MVPs: Kostenlose Credits für schnelle Validierung ohne Vorabkosten
- Batch-Verarbeitung: Automatische Retry-Logik bei 502/503-Fehlern schützt lange Jobs
❌ Nicht ideal für:
- Unternehmen mit Compliance-Anforderungen: Offizielle APIs bieten erweiterte Audit-Funktionen
- Mission-critical Systeme ohne Fallback: Empfehlung: Immer Offizielle API als Failover konfigurieren
- Teams ohne technische Kapazität: Monitoring-Setup erfordert Grundverständnis von API-Architektur
Technische Implementierung: Vollständiger Monitoring-Stack
1. Basis-API-Client mit Retry-Logik
"""
HolySheep AI Monitoring Client mit automatischer Retry-Logik
Base URL: https://api.holysheep.ai/v1
"""
import requests
import time
import json
from datetime import datetime
from typing import Optional, Dict, Any
from dataclasses import dataclass, field
from enum import Enum
class APIError(Exception):
"""Basis-Exception für API-Fehler"""
def __init__(self, status_code: int, message: str, retry_after: Optional[int] = None):
self.status_code = status_code
self.message = message
self.retry_after = retry_after
super().__init__(f"[{status_code}] {message}")
class RateLimitError(APIError):
"""429 Too Many Requests"""
pass
class ServerError(APIError):
"""502/503 Server-Fehler"""
pass
@dataclass
class RetryConfig:
"""Konfiguration für Retry-Verhalten"""
max_retries: int = 5
base_delay: float = 1.0
max_delay: float = 60.0
exponential_base: float = 2.0
jitter: bool = True
@dataclass
class MonitoringMetrics:
"""Tracking von Metriken"""
total_requests: int = 0
successful_requests: int = 0
rate_limit_errors: int = 0
server_errors: int = 0
other_errors: int = 0
total_retry_attempts: int = 0
last_error: Optional[Dict] = None
error_log: list = field(default_factory=list)
class HolySheepMonitoredClient:
"""API-Client mit integriertem Monitoring und Alerting"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str, retry_config: Optional[RetryConfig] = None):
if not api_key or api_key == "YOUR_HOLYSHEEP_API_KEY":
raise ValueError("Gültiger API-Key erforderlich!")
self.api_key = api_key
self.retry_config = retry_config or RetryConfig()
self.metrics = MonitoringMetrics()
self.alert_callbacks = []
self.headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
def add_alert_callback(self, callback):
"""Callback für Alert-Events registrieren"""
self.alert_callbacks.append(callback)
def _trigger_alert(self, error_type: str, details: dict):
"""Alert an alle Callbacks senden"""
alert_data = {
"timestamp": datetime.now().isoformat(),
"error_type": error_type,
"details": details,
"current_metrics": {
"total_requests": self.metrics.total_requests,
"error_rate": self._calculate_error_rate(),
"rate_limit_count": self.metrics.rate_limit_errors
}
}
for callback in self.alert_callbacks:
try:
callback(alert_data)
except Exception as e:
print(f"Alert-Callback Fehler: {e}")
def _calculate_error_rate(self) -> float:
"""Aktuelle Fehlerrate berechnen"""
if self.metrics.total_requests == 0:
return 0.0
return (self.metrics.rate_limit_errors +
self.metrics.server_errors +
self.metrics.other_errors) / self.metrics.total_requests * 100
def _calculate_delay(self, attempt: int) -> float:
"""Exponential Backoff mit Jitter berechnen"""
delay = self.retry_config.base_delay * (self.retry_config.exponential_base ** attempt)
delay = min(delay, self.retry_config.max_delay)
if self.retry_config.jitter:
import random
delay = delay * (0.5 + random.random())
return delay
def _log_error(self, error: Exception, context: dict):
"""Fehler im Log speichern"""
error_entry = {
"timestamp": datetime.now().isoformat(),
"error_type": type(error).__name__,
"error_message": str(error),
"context": context,
"attempt": context.get("attempt", 0)
}
self.metrics.error_log.append(error_entry)
self.metrics.last_error = error_entry
# Log limitiert auf 1000 Einträge
if len(self.metrics.error_log) > 1000:
self.metrics.error_log = self.metrics.error_log[-1000:]
def chat_completions(self, messages: list, model: str = "gpt-4.1",
temperature: float = 0.7, max_tokens: int = 1000) -> Dict:
"""
Chat-Completion mit automatischer Retry-Logik
Args:
messages: Liste von Chat-Nachrichten
model: Modell-Name (gpt-4.1, claude-sonnet-4.5, gemini-2.5-flash, deepseek-v3.2)
temperature: Sampling-Temperatur
max_tokens: Maximale Antwort-Tokens
Returns:
API-Response als Dictionary
"""
endpoint = f"{self.BASE_URL}/chat/completions"
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
return self._execute_with_retry("POST", endpoint, json=payload)
def embeddings(self, input_text: str, model: str = "text-embedding-3-small") -> Dict:
"""Embedding-Generierung mit Monitoring"""
endpoint = f"{self.BASE_URL}/embeddings"
payload = {
"model": model,
"input": input_text
}
return self._execute_with_retry("POST", endpoint, json=payload)
def _execute_with_retry(self, method: str, url: str, **kwargs) -> Dict:
"""Request mit Retry-Logik ausführen"""
context = {"method": method, "url": url, "attempt": 0}
for attempt in range(self.retry_config.max_retries + 1):
context["attempt"] = attempt
self.metrics.total_requests += 1
try:
response = requests.request(
method=method,
url=url,
headers=self.headers,
timeout=30,
**kwargs
)
if response.status_code == 200:
self.metrics.successful_requests += 1
return response.json()
elif response.status_code == 429:
self.metrics.rate_limit_errors += 1
error = RateLimitError(
429,
response.json().get("error", {}).get("message", "Rate limit exceeded"),
retry_after=response.headers.get("Retry-After")
)
self._log_error(error, context)
self._trigger_alert("RATE_LIMIT", {"attempt": attempt, "retry_after": error.retry_after})
if attempt < self.retry_config.max_retries:
retry_delay = int(error.retry_after) if error.retry_after else self._calculate_delay(attempt)
time.sleep(retry_delay)
continue
raise error
elif response.status_code in (502, 503, 504):
self.metrics.server_errors += 1
error = ServerError(response.status_code, f"Server error: {response.text}")
self._log_error(error, context)
self._trigger_alert("SERVER_ERROR", {
"status": response.status_code,
"attempt": attempt
})
if attempt < self.retry_config.max_retries:
delay = self._calculate_delay(attempt)
time.sleep(delay)
self.metrics.total_retry_attempts += 1
continue
raise error
elif response.status_code == 400:
self.metrics.other_errors += 1
error = APIError(400, response.json().get("error", {}).get("message", "Bad request"))
self._log_error(error, context)
self._trigger_alert("BAD_REQUEST", {"response": response.text})
raise error
else:
self.metrics.other_errors += 1
error = APIError(response.status_code, f"Unexpected error: {response.text}")
self._log_error(error, context)
self._trigger_alert("UNKNOWN_ERROR", {"status": response.status_code})
raise error
except requests.exceptions.Timeout:
self.metrics.other_errors += 1
error = APIError(0, "Request timeout")
self._log_error(error, context)
self._trigger_alert("TIMEOUT", {"attempt": attempt})
if attempt < self.retry_config.max_retries:
time.sleep(self._calculate_delay(attempt))
continue
raise error
except requests.exceptions.ConnectionError as e:
self.metrics.other_errors += 1
error = APIError(0, f"Connection error: {str(e)}")
self._log_error(error, context)
self._trigger_alert("CONNECTION_ERROR", {"error": str(e)})
if attempt < self.retry_config.max_retries:
time.sleep(self._calculate_delay(attempt))
continue
raise error
raise APIError(500, "Max retries exceeded")
def get_health_status(self) -> Dict:
"""Gesundheitsstatus und Metriken abrufen"""
return {
"status": "healthy" if self._calculate_error_rate() < 5 else "degraded",
"error_rate_percent": round(self._calculate_error_rate(), 2),
"total_requests": self.metrics.total_requests,
"success_rate_percent": round(
self.metrics.successful_requests / max(1, self.metrics.total_requests) * 100, 2
),
"error_breakdown": {
"rate_limits": self.metrics.rate_limit_errors,
"server_errors": self.metrics.server_errors,
"other": self.metrics.other_errors
},
"total_retries": self.metrics.total_retry_attempts,
"last_error": self.metrics.last_error
}
Beispiel-Nutzung
if __name__ == "__main__":
client = HolySheepMonitoredClient(
api_key="YOUR_HOLYSHEEP_API_KEY",
retry_config=RetryConfig(max_retries=3, base_delay=1.0)
)
# Alert-Callback registrieren
def alert_handler(alert_data):
print(f"🚨 ALERT: {alert_data['error_type']} - {alert_data['details']}")
client.add_alert_callback(alert_handler)
# API-Aufruf
try:
response = client.chat_completions(
messages=[{"role": "user", "content": "Erkläre mir Exponentielles Backoff"}],
model="gpt-4.1"
)
print(f"Antwort: {response['choices'][0]['message']['content']}")
except APIError as e:
print(f"API-Fehler nach Retry: {e}")
# Monitoring-Status
print(json.dumps(client.get_health_status(), indent=2, default=str))2. Echtzeit-Alerting mit Webhook-Integration
"""
Erweitertes Alerting-System für HolySheep API Monitoring
Webhook-Integration für Slack, Discord, PagerDuty und E-Mail
"""
import threading
import queue
import time
import json
import smtplib
from datetime import datetime, timedelta
from typing import List, Dict, Callable
from dataclasses import dataclass, field
from enum import Enum
class AlertSeverity(Enum):
INFO = "info"
WARNING = "warning"
ERROR = "error"
CRITICAL = "critical"
@dataclass
class Alert:
"""Alert-Datenstruktur"""
id: str
timestamp: datetime
severity: AlertSeverity
title: str
message: str
metrics: Dict = field(default_factory=dict)
resolved: bool = False
resolved_at: datetime = None
class AlertingSystem:
"""Zentrales Alerting-System mit mehreren Kanälen"""
def __init__(self, config: Dict):
self.config = config
self.alert_queue = queue.Queue()
self.active_alerts = {}
self.alert_history = []
self.handlers = []
# Handler registrieren
if config.get("slack_webhook"):
self.handlers.append(SlackHandler(config["slack_webhook"]))
if config.get("discord_webhook"):
self.handlers.append(DiscordHandler(config["discord_webhook"]))
if config.get("email"):
self.handlers.append(EmailHandler(config["email"], config.get("smtp_config", {})))
# Background-Worker starten
self.running = True
self.worker_thread = threading.Thread(target=self._process_alerts, daemon=True)
self.worker_thread.start()
# Automatisches Health-Checking
self.health_check_interval = config.get("health_check_interval", 60)
self.health_thread = threading.Thread(target=self._health_check_loop, daemon=True)
self.health_thread.start()
def _health_check_loop(self):
"""Periodische Gesundheitsprüfung"""
while self.running:
time.sleep(self.health_check_interval)
try:
self._check_api_health()
except Exception as e:
print(f"Gesundheitscheck fehlgeschlagen: {e}")
def _check_api_health(self):
"""API-Endpunkte auf Erreichbarkeit prüfen"""
import requests
endpoints = [
"https://api.holysheep.ai/v1/models",
"https://api.holysheep.ai/health"
]
for endpoint in endpoints:
try:
response = requests.get(endpoint, timeout=5)
if response.status_code != 200:
self.trigger_alert(
severity=AlertSeverity.WARNING,
title="API-Endpunkt nicht erreichbar",
message=f"{endpoint} antwortet mit Status {response.status_code}"
)
except Exception as e:
self.trigger_alert(
severity=AlertSeverity.ERROR,
title="Verbindungsfehler",
message=f"Konnte {endpoint} nicht erreichen: {str(e)}"
)
def trigger_alert(self, severity: AlertSeverity, title: str, message: str, metrics: Dict = None):
"""Neuen Alert auslösen"""
import uuid
alert = Alert(
id=str(uuid.uuid4())[:8],
timestamp=datetime.now(),
severity=severity,
title=title,
message=message,
metrics=metrics or {}
)
# Alert verarbeiten
self.alert_queue.put(alert)
self.active_alerts[alert.id] = alert
self.alert_history.append(alert)
# History limitieren
if len(self.alert_history) > 500:
self.alert_history = self.alert_history[-500:]
def _process_alerts(self):
"""Alert-Queue im Hintergrund verarbeiten"""
while self.running:
try:
alert = self.alert_queue.get(timeout=1)
# An alle Handler senden
for handler in self.handlers:
try:
handler.send(alert)
except Exception as e:
print(f"Handler {handler.__class__.__name__} Fehler: {e}")
self.alert_queue.task_done()
except queue.Empty:
continue
except Exception as e:
print(f"Alert-Verarbeitung Fehler: {e}")
def resolve_alert(self, alert_id: str):
"""Alert als gelöst markieren"""
if alert_id in self.active_alerts:
alert = self.active_alerts[alert_id]
alert.resolved = True
alert.resolved_at = datetime.now()
del self.active_alerts[alert_id]
def get_alert_summary(self) -> Dict:
"""Zusammenfassung aller aktiven Alerts"""
summary = {
"total_active": len(self.active_alerts),
"by_severity": {s.value: 0 for s in AlertSeverity},
"alerts": []
}
for alert in self.active_alerts.values():
summary["by_severity"][alert.severity.value] += 1
summary["alerts"].append({
"id": alert.id,
"severity": alert.severity.value,
"title": alert.title,
"timestamp": alert.timestamp.isoformat()
})
return summary
class SlackHandler:
"""Slack Webhook Alert-Handler"""
def __init__(self, webhook_url: str):
self.webhook_url = webhook_url
def send(self, alert: Alert):
import requests
color_map = {
AlertSeverity.INFO: "#36a64f",
AlertSeverity.WARNING: "#ff9800",
AlertSeverity.ERROR: "#f44336",
AlertSeverity.CRITICAL: "#9c27b0"
}
payload = {
"attachments": [{
"color": color_map.get(alert.severity, "#808080"),
"title": f":warning: {alert.title}",
"text": alert.message,
"fields": [
{"title": "Severity", "value": alert.severity.value.upper(), "short": True},
{"title": "Time", "value": alert.timestamp.strftime("%Y-%m-%d %H:%M:%S"), "short": True},
{"title": "Alert ID", "value": alert.id, "short": True}
],
"footer": "HolySheep AI Monitoring",
"ts": int(alert.timestamp.timestamp())
}]
}
if alert.metrics:
metrics_text = "\n".join([f"• {k}: {v}" for k, v in alert.metrics.items()])
payload["attachments"][0]["fields"].append({
"title": "Metrics",
"value": metrics_text,
"short": False
})
requests.post(self.webhook_url, json=payload, timeout=10)
class DiscordHandler:
"""Discord Webhook Alert-Handler"""
def __init__(self, webhook_url: str):
self.webhook_url = webhook_url
def send(self, alert: Alert):
import requests
emoji_map = {
AlertSeverity.INFO: "ℹ️",
AlertSeverity.WARNING: "⚠️",
AlertSeverity.ERROR: "❌",
AlertSeverity.CRITICAL: "🚨"
}
embed = {
"title": f"{emoji_map.get(alert.severity, '📢')} {alert.title}",
"description": alert.message,
"color": self._severity_to_color(alert.severity),
"fields": [
{"name": "Severity", "value": alert.severity.value.upper(), "inline": True},
{"name": "Time", "value": alert.timestamp.strftime("%Y-%m-%d %H:%M:%S"), "inline": True},
{"name": "ID", "value": alert.id, "inline": True}
],
"footer": {"text": "HolySheep AI Monitor"},
"timestamp": alert.timestamp.isoformat()
}
if alert.metrics:
embed["fields"].append({
"name": "Metrics",
"value": "``\n" + "\n".join([f"{k}: {v}" for k, v in alert.metrics.items()]) + "\n``",
"inline": False
})
requests.post(self.webhook_url, json={"embeds": [embed]}, timeout=10)
def _severity_to_color(self, severity: AlertSeverity) -> int:
return {
AlertSeverity.INFO: 0x36a64f,
AlertSeverity.WARNING: 0xff9800,
AlertSeverity.ERROR: 0xf44336,
AlertSeverity.CRITICAL: 0x9c27b0
}.get(severity, 0x808080)
class EmailHandler:
"""E-Mail Alert-Handler"""
def __init__(self, email_config: Dict, smtp_config: Dict):
self.email_config = email_config
self.smtp_config = smtp_config
def send(self, alert: Alert):
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
msg = MIMEMultipart("alternative")
msg["Subject"] = f"[{alert.severity.value.upper()}] HolySheep Alert: {alert.title}"
msg["From"] = self.smtp_config.get("from_email", "[email protected]")
msg["To"] = ", ".join(self.email_config.get("recipients", []))
# Text-Version
text_body = f"""
HolySheep AI Monitoring Alert
=============================
Severity: {alert.severity.value.upper()}
Title: {alert.title}
Message: {alert.message}
Time: {alert.timestamp.strftime("%Y-%m-%d %H:%M:%S")}
Alert ID: {alert.id}
Metrics:
{json.dumps(alert.metrics, indent=2)}
---
Dies ist eine automatische Benachrichtigung.
"""
msg.attach(MIMEText(text_body, "plain"))
try:
with smtplib.SMTP(self.smtp_config.get("host", "localhost"),
self.smtp_config.get("port", 587)) as server:
if self.smtp_config.get("username"):
server.starttls()
server.login(self.smtp_config["username"], self.smtp_config["password"])
server.send_message(msg)
except Exception as e:
print(f"E-Mail-Versand fehlgeschlagen: {e}")
Beispiel-Konfiguration
if __name__ == "__main__":
alerting = AlertingSystem({
"slack_webhook": "https://hooks.slack.com/services/YOUR/SLACK/WEBHOOK",
"discord_webhook": "https://discord.com/api/webhooks/YOUR/DISCORD/WEBHOOK",
"email": {
"recipients": ["[email protected]"]
},
"smtp_config": {
"host": "smtp.example.com",
"port": 587,
"username": "[email protected]",
"password": "YOUR_SMTP_PASSWORD",
"from_email": "[email protected]"
},
"health_check_interval": 60
})
# Test-Alert auslösen
alerting.trigger_alert(
severity=AlertSeverity.WARNING,
title="Hohe Fehlerrate erkannt",
message="Rate-Limit-Fehler übersteigen 10% der Anfragen",
metrics={
"error_rate": "12.5%",
"rate_limits_last_hour": 156,
"server_errors_last_hour": 23
}
)
# Alert-Zusammenfassung abrufen
print(json.dumps(alerting.get_alert_summary(), indent=2, default=str))3. Prometheus/Grafana-Metriken-Export
"""
Prometheus Metrics Exporter für HolySheep API Monitoring
Integration mit Grafana für Dashboards und Alerting
"""
from flask import Flask, Response, jsonify
import prometheus_client
from prometheus_client import Counter, Histogram, Gauge, generate_latest, CONTENT_TYPE_LATEST
import threading
import time
from datetime import datetime
Prometheus Metriken definieren
API_REQUESTS_TOTAL = Counter(
'holysheep_api_requests_total',
'Total number of API requests',
['method', 'endpoint', 'status_code']
)
API_REQUEST_DURATION = Histogram(
'holysheep_api_request_duration_seconds',
'API request latency in seconds',
['method', 'endpoint'],
buckets=[0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0]
)
API_ERRORS_TOTAL = Counter(
'holysheep_api_errors_total',
'Total number of API errors',
['error_type', 'model']
)
RATE_LIMIT_REQUESTS = Counter(
'holysheep_rate_limit_hits_total',
'Total number of rate limit (429) errors',
['model']
)
SERVER_ERROR_REQUESTS = Counter(
'holysheep_server_errors_total',
'Total number of server errors (502/503)',
['status_code', 'model']
)
ACTIVE_REQUESTS = Gauge(
'holysheep_active_requests',
'Number of currently active requests'
)
RETRY_ATTEMPTS = Counter(
'holysheep_retry_attempts_total',
'Total number of retry attempts',
['model', 'attempt_number']
)
COST_ESTIMATE = Gauge(
'holysheep_estimated_cost_usd',
'Estimated API cost in USD',
['model']
)
class MetricsCollector:
"""Sammelt und exportiert Metriken für Prometheus"""
def __init__(self):
self.start_time = time.time()
self.costs = {model: 0.0 for model in [
"gpt-4.1", "claude-sonnet-4.5", "gemini-2.5-flash", "deepseek-v3.2"
]}
self.model_prices = {
"gpt-4.1": 8.0,
"claude-sonnet-4.5": 15.0,
"gemini-2.5-flash": 2.50,
"deepseek-v3.2": 0.42
}
self.lock = threading.Lock()
def record_request(self, method: str, endpoint: str, status_code: int,
duration: float, model: str = None):
"""API-Anfrage aufzeichnen"""
API_REQUESTS_TOTAL.labels(
method=method,
endpoint=endpoint,
status_code=str(status_code)
).inc()
API_REQUEST_DURATION.labels(
method=method,
endpoint=endpoint
).observe(duration)
if model:
self._update_cost_estimate(model, status_code)
def record_error(self, error_type: str, model: str = None):
"""Fehler aufzeichnen"""
API_ERRORS_TOTAL.labels(error_type=error_type, model=model or "unknown").inc()
def record_rate_limit(self, model: str = None):
"""Rate-Limit-Fehler aufzeichnen"""
RATE_LIMIT_REQUESTS.labels(model=model or "unknown").inc()
self.record_error("rate_limit_429", model)
def record_server_error(self, status_code: int, model: str = None):
"""Server-Fehler aufzeichnen"""
SERVER_ERROR_REQUESTS.labels(
status_code=str(status_code),
model=model or "unknown"
).inc()
self.record_error(f"server_error_{status_code}", model)
def record_retry(self, model: str, attempt: int):
"""Retry-Versuch aufzeichnen"""
RETRY_ATTEMPTS.labels(model=model, attempt_number=str(attempt)).inc()
def increment_active_requests(self):
"""Aktive Anfragen erhöhen"""
ACTIVE_REQUESTS.inc()
def decrement_active_requests(self):
"""Aktive Anfragen verringern"""
ACTIVE_REQUESTS.dec()
def _update_cost_estimate(self, model: str, status_code: int):
"""Kostenschätzung aktualisieren"""
if status_code == 200 and model in self.model_prices:
with self.lock:
# Annahme: Durchschnittlich 500 Tokens pro Anfrage
cost_per_request = (self.model_prices[model] / 1_000_000) * 500
self.costs[model] += cost_per_request
COST_ESTIMATE.labels(model=model).set(self.costs[model])
def get_metrics_summary(self) -> dict:
"""Metriken-Zusammenfassung abrufen"""
uptime = time.time() - self.start_time
return {
"uptime_seconds": uptime,
"estimated_costs": self.costs,
"total_estimated_cost": sum(self.costs.values())
}
Flask-App für Metrics-Endpunkt
app = Flask(__name__)
metrics_collector = MetricsCollector()
@app.route('/metrics')
def metrics():
"""Prometheus Metrics Endpoint"""
return Response(generate_latest(), mimetype=CONTENT_TYPE_LATEST)
@app.route('/metrics/summary')
def metrics_summary():
"""Metriken-Zusammenfassung als JSON"""
return jsonify(metrics_collector.get_metrics_summary())
@app.route('/health')
def health():
"""Health Check Endpoint"""
return jsonify({
"status": "healthy",