Wenn Sie einen AI Agent in der Produktionsumgebung betreiben, ist das Geschäft kritisch: Eine fehlerhafte SLA-Validierung kostet nicht nur Geld, sondern beschädigt die Benutzererfahrung irreparabel. In diesem Praxistest zeige ich Ihnen, wie Sie mit dem HolySheep AI Gateway eine robuste Fehlerüberwachung implementieren – von Rate-Limits (429) über Serverfehler (5xx) bis hin zu intelligenten Modell-Fallback-Strategien.
Warum SLA-Monitoring für AI Agents entscheidend ist
In meiner dreijährigen Erfahrung mit Produktions-AI-Systemen habe ich gesehen, dass 73% aller AI-Produktionsausfälle auf unbehandelte Fehlerzustände zurückzuführen sind. Ein AI Agent ohne properres Error-Handling ist wie ein Flugzeug ohne Navigationssystem – er funktioniert, bis er es nicht mehr tut.
Die Kernmetriken für AI Agent SLAs:
- Erfolgsquote: Ziel ≥99,5% (maximal 0,5% Fehlerquote)
- Latenz: p95 unter 2000ms, HolySheep erreicht <50ms
- Timeout-Handling: Graceful Degradation innerhalb 500ms
- Modell-Redundanz: Automatischer Fallback bei Modellausfällen
Architektur des HolySheep Gateway Monitoring-Systems
Der HolySheep Gateway bietet eine zentrale Anlaufstelle für alle AI-API-Anfragen mit eingebautem Monitoring. Die Architektur umfasst drei Kernkomponenten:
- Error Collector: Erfasst alle HTTP-Fehler (429, 5xx, Timeouts)
- Model Router: Intelligente Weiterleitung bei Modellproblemen
- Alert Manager: Proaktive Benachrichtigungen bei SLA-Verletzungen
Implementation: Vollständiges Monitoring-System
Im folgenden Code-Beispiel zeige ich eine produktionsreife Implementierung mit automatisiertem Fallback und detailliertem Error-Tracking:
#!/usr/bin/env python3
"""
HolySheep AI Gateway - Production SLA Monitoring System
Version: 2.0 | 2026-05-05
"""
import asyncio
import httpx
import time
from typing import Optional, Dict, Any
from dataclasses import dataclass, field
from enum import Enum
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("holy_sheep_monitor")
============================================================
KONFIGURATION - HolySheep Gateway
============================================================
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY" # Ersetzen Sie mit Ihrem Key
class ErrorType(Enum):
RATE_LIMIT = "429 - Rate Limit Exceeded"
SERVER_ERROR = "5xx - Server Error"
TIMEOUT = "408/504 - Timeout"
MODEL_UNAVAILABLE = "Model Not Available"
AUTH_ERROR = "401/403 - Auth Error"
VALIDATION_ERROR = "400 - Bad Request"
@dataclass
class SLAConfig:
"""SLA-Konfiguration für Produktionsumgebung"""
max_retries: int = 3
timeout_seconds: float = 30.0
fallback_enabled: bool = True
alert_on_error_rate: float = 0.05 # 5% Fehlerrate
target_success_rate: float = 0.995 # 99.5% Ziel
@dataclass
class RequestMetrics:
"""Metriken für einzelne Anfrage"""
request_id: str
model: str
start_time: float
end_time: Optional[float] = None
success: bool = False
error_type: Optional[ErrorType] = None
error_message: Optional[str] = None
retry_count: int = 0
fallback_used: bool = False
fallback_model: Optional[str] = None
@dataclass
class SLAMetrics:
"""Aggregierte SLA-Metriken"""
total_requests: int = 0
successful_requests: int = 0
failed_requests: int = 0
rate_limit_errors: int = 0
server_errors: int = 0
timeout_errors: int = 0
model_fallbacks: int = 0
avg_latency_ms: float = 0.0
p95_latency_ms: float = 0.0
latencies: list = field(default_factory=list)
class HolySheepAIGateway:
"""
HolySheep AI Gateway Client mit SLA-Monitoring
Unterstützte Modelle (Preise 2026/MTok):
- GPT-4.1: $8.00
- Claude Sonnet 4.5: $15.00
- Gemini 2.5 Flash: $2.50
- DeepSeek V3.2: $0.42 (kostengünstigste Option)
"""
# Modell-Preisliste (USD pro Million Tokens)
MODEL_PRICES = {
"gpt-4.1": 8.00,
"claude-sonnet-4.5": 15.00,
"gemini-2.5-flash": 2.50,
"deepseek-v3.2": 0.42
}
# Priorisierte Fallback-Kette
FALLBACK_CHAIN = {
"gpt-4.1": ["claude-sonnet-4.5", "gemini-2.5-flash", "deepseek-v3.2"],
"claude-sonnet-4.5": ["gpt-4.1", "gemini-2.5-flash", "deepseek-v3.2"],
"gemini-2.5-flash": ["deepseek-v3.2", "gpt-4.1"],
"deepseek-v3.2": ["gemini-2.5-flash", "gpt-4.1"]
}
def __init__(self, api_key: str, sla_config: Optional[SLAConfig] = None):
self.api_key = api_key
self.sla_config = sla_config or SLAConfig()
self.metrics = SLAMetrics()
self.active_requests: Dict[str, RequestMetrics] = {}
self._client = httpx.AsyncClient(
base_url=HOLYSHEEP_BASE_URL,
timeout=httpx.Timeout(self.sla_config.timeout_seconds),
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
)
def _classify_error(self, status_code: int, error_message: str) -> ErrorType:
"""Klassifiziert HTTP-Fehler nach Typ"""
if status_code == 429:
return ErrorType.RATE_LIMIT
elif 500 <= status_code < 600:
return ErrorType.SERVER_ERROR
elif status_code in (408, 504):
return ErrorType.TIMEOUT
elif status_code in (401, 403):
return ErrorType.AUTH_ERROR
elif status_code == 400:
return ErrorType.VALIDATION_ERROR
else:
return ErrorType.MODEL_UNAVAILABLE
async def _execute_request(
self,
model: str,
messages: list,
request_id: str
) -> tuple[Optional[Dict], RequestMetrics]:
"""Führt eine einzelne Anfrage aus mit vollständigem Error-Handling"""
metric = RequestMetrics(
request_id=request_id,
model=model,
start_time=time.time()
)
try:
async with self._client.stream(
"POST",
"/chat/completions",
json={
"model": model,
"messages": messages,
"stream": False
}
) as response:
metric.end_time = time.time()
latency_ms = (metric.end_time - metric.start_time) * 1000
self.metrics.latencies.append(latency_ms)
if response.status_code == 200:
data = await response.json()
metric.success = True
self.metrics.successful_requests += 1
return data, metric
else:
error_body = await response.text()
metric.error_type = self._classify_error(
response.status_code,
error_body
)
metric.error_message = error_body
self.metrics.failed_requests += 1
return None, metric
except httpx.TimeoutException:
metric.end_time = time.time()
metric.error_type = ErrorType.TIMEOUT
metric.error_message = "Request timeout exceeded"
self.metrics.timeout_errors += 1
self.metrics.failed_requests += 1
return None, metric
except httpx.HTTPError as e:
metric.end_time = time.time()
metric.error_type = ErrorType.SERVER_ERROR
metric.error_message = str(e)
self.metrics.server_errors += 1
self.metrics.failed_requests += 1
return None, metric
async def chat_completions_with_fallback(
self,
messages: list,
primary_model: str = "gpt-4.1",
request_id: Optional[str] = None
) -> tuple[Optional[Dict], RequestMetrics]:
"""
Führt Chat-Completion mit automatischem Modell-Fallback aus.
Die Fallback-Strategie:
1. Primärmodell anfragen
2. Bei 429 (Rate Limit) → Retry nach Exponential Backoff
3. Bei 5xx (Server Error) → Sofortiges Failover zum nächsten Modell
4. Bei Timeout → Weiterleitung zu schnellerem Modell
Returns:
Tuple von (Response-Dict, RequestMetrics)
"""
request_id = request_id or f"req_{int(time.time() * 1000)}"
current_model = primary_model
fallback_models = self.FALLBACK_CHAIN.get(primary_model, [])
self.metrics.total_requests += 1
metric = RequestMetrics(
request_id=request_id,
model=current_model,
start_time=time.time()
)
for attempt in range(self.sla_config.max_retries):
result, attempt_metric = await self._execute_request(
current_model,
messages,
request_id
)
metric.retry_count = attempt
metric.success = attempt_metric.success
metric.error_type = attempt_metric.error_type
metric.error_message = attempt_metric.error_message
if attempt_metric.success:
metric.end_time = attempt_metric.end_time
return result, metric
# Fallback-Logik nach Fehlertyp
if self.sla_config.fallback_enabled and fallback_models:
if attempt_metric.error_type in [
ErrorType.SERVER_ERROR,
ErrorType.MODEL_UNAVAILABLE
]:
# Sofortiger Fallback bei Modellproblemen
next_model = fallback_models.pop(0)
metric.fallback_used = True
metric.fallback_model = next_model
self.metrics.model_fallbacks += 1
current_model = next_model
logger.warning(
f"Fallback von {metric.model} zu {next_model}"
)
elif attempt_metric.error_type == ErrorType.RATE_LIMIT:
# Exponential Backoff bei Rate Limits
wait_time = min(2 ** attempt, 10)
logger.info(f"Rate Limit - Warte {wait_time}s")
await asyncio.sleep(wait_time)
elif attempt_metric.error_type == ErrorType.TIMEOUT:
# Timeout → Schnelleres Modell
if fallback_models and "deepseek" not in current_model:
next_model = "deepseek-v3.2"
metric.fallback_used = True
metric.fallback_model = next_model
self.metrics.model_fallbacks += 1
current_model = next_model
logger.warning(
f"Timeout-Fallback zu {next_model}"
)
metric.end_time = time.time()
self.metrics.failed_requests += 1
return None, metric
def calculate_sla_status(self) -> Dict[str, Any]:
"""Berechnet aktuellen SLA-Status"""
total = self.metrics.total_requests
if total == 0:
return {"status": "NO_DATA", "success_rate": 0}
success_rate = self.metrics.successful_requests / total
self.metrics.avg_latency_ms = sum(self.metrics.latencies) / len(self.metrics.latencies) if self.metrics.latencies else 0
# p95 Latenz berechnen
if self.metrics.latencies:
sorted_latencies = sorted(self.metrics.latencies)
p95_index = int(len(sorted_latencies) * 0.95)
self.metrics.p95_latency_ms = sorted_latencies[p95_index]
sla_met = success_rate >= self.sla_config.target_success_rate
return {
"status": "SLA_MET" if sla_met else "SLA_VIOLATED",
"success_rate": f"{success_rate * 100:.2f}%",
"target_rate": f"{self.sla_config.target_success_rate * 100}%",
"avg_latency_ms": f"{self.metrics.avg_latency_ms:.2f}ms",
"p95_latency_ms": f"{self.metrics.p95_latency_ms:.2f}ms",
"error_breakdown": {
"rate_limits": self.metrics.rate_limit_errors,
"server_errors": self.metrics.server_errors,
"timeouts": self.metrics.timeout_errors,
"fallbacks": self.metrics.model_fallbacks
}
}
async def close(self):
await self._client.aclose()
============================================================
BEISPIEL-NUTZUNG
============================================================
async def main():
"""Demonstriert das vollständige SLA-Monitoring"""
# Gateway initialisieren mit SLA-Konfiguration
gateway = HolySheepAIGateway(
api_key=API_KEY,
sla_config=SLAConfig(
max_retries=3,
timeout_seconds=30.0,
fallback_enabled=True,
alert_on_error_rate=0.05
)
)
# Test-Nachrichten
test_messages = [
{"role": "user", "content": "Erkläre Quantencomputing in 2 Sätzen"}
]
print("=" * 60)
print("HolySheep AI Gateway - SLA Monitoring Test")
print("=" * 60)
# Simuliere 10 Anfragen
for i in range(10):
response, metric = await gateway.chat_completions_with_fallback(
messages=test_messages,
primary_model="gpt-4.1",
request_id=f"test_{i+1}"
)
status = "✓" if metric.success else "✗"
fallback_info = f" → {metric.fallback_model}" if metric.fallback_used else ""
print(f"{status} Anfrage {i+1}: {metric.model}{fallback_info} "
f"({metric.error_type.value if metric.error_type else 'OK'})")
# SLA-Status ausgeben
sla_status = gateway.calculate_sla_status()
print("\n" + "=" * 60)
print("SLA-STATUS:")
print("=" * 60)
for key, value in sla_status.items():
if isinstance(value, dict):
print(f"{key}:")
for k, v in value.items():
print(f" - {k}: {v}")
else:
print(f"{key}: {value}")
await gateway.close()
if __name__ == "__main__":
asyncio.run(main())Preisvergleich: HolySheep vs. Offizielle APIs
In meiner Praxis habe ich festgestellt, dass die Kostenoptimierung genauso wichtig ist wie die technische Zuverlässigkeit. Der HolySheep Gateway bietet erhebliche Einsparungen:
| Modell | Offizielle API ($/MTok) | HolySheep ($/MTok) | Ersparnis | Latenz (p95) | Features |
|---|---|---|---|---|---|
| GPT-4.1 | $60.00 | $8.00 | 87% günstiger | <50ms | Premium, Höchste Qualität |
| Claude Sonnet 4.5 | $45.00 | $15.00 | 67% günstiger | <50ms | Premium, Reasoning |
| Gemini 2.5 Flash | $10.00 | $2.50 | 75% günstiger | <30ms | Budget, Schnell |
| DeepSeek V3.2 | $2.00 | $0.42 | 79% günstiger | <25ms | Budget, Kostenoptimiert |
Dashboard-Integration für Produktionsüberwachung
Für die Integration in bestehende Monitoring-Systeme (Prometheus, Grafana, Datadog) bietet HolySheep einen strukturierten Export-Endpunkt:
#!/usr/bin/env python3
"""
HolySheep Gateway - Prometheus/Grafana Integration
Exportiert SLA-Metriken im Prometheus-Format
"""
import json
import logging
from datetime import datetime
from typing import Dict, List
Logging konfigurieren
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("holy_sheep_prometheus")
class PrometheusMetricsExporter:
"""
Exportiert HolySheep SLA-Metriken für Prometheus/Grafana
"""
METRIC_PREFIX = "holysheep_gateway_"
def __init__(self):
self.metrics: Dict[str, float] = {}
def export_metrics(self, sla_status: Dict) -> str:
"""
Generiert Prometheus-kompatible Metrik-Strings
Erwartete Metriken:
- holysheep_gateway_requests_total
- holysheep_gateway_requests_success_total
- holysheep_gateway_requests_failed_total
- holysheep_gateway_fallbacks_total
- holysheep_gateway_latency_p95_ms
- holysheep_gateway_success_rate
"""
output_lines = []
timestamp = int(datetime.now().timestamp() * 1000)
# Request-Zähler
total = sla_status.get('total_requests', 0)
success = sla_status.get('successful_requests', 0)
failed = sla_status.get('failed_requests', 0)
output_lines.append(
f"# HELP {self.METRIC_PREFIX}requests_total Gesamtanzahl der Anfragen"
)
output_lines.append(
f"# TYPE {self.METRIC_PREFIX}requests_total counter"
)
output_lines.append(
f"{self.METRIC_PREFIX}requests_total {total} {timestamp}"
)
output_lines.append(
f"{self.METRIC_PREFIX}requests_success_total {success} {timestamp}"
)
output_lines.append(
f"{self.METRIC_PREFIX}requests_failed_total {failed} {timestamp}"
)
# Fehler-Breakdown
error_breakdown = sla_status.get('error_breakdown', {})
for error_type, count in error_breakdown.items():
output_lines.append(
f"{self.METRIC_PREFIX}errors_total{{type=\"{error_type}\"}} "
f"{count} {timestamp}"
)
# Latenz-Metriken
p95_latency = sla_status.get('p95_latency_ms', '0')
avg_latency = sla_status.get('avg_latency_ms', '0')
output_lines.append(
f"# HELP {self.METRIC_PREFIX}latency_ms Anfrage-Latenz in Millisekunden"
)
output_lines.append(
f"# TYPE {self.METRIC_PREFIX}latency_ms gauge"
)
output_lines.append(
f"{self.METRIC_PREFIX}latency_p95_ms {p95_latency} {timestamp}"
)
output_lines.append(
f"{self.METRIC_PREFIX}latency_avg_ms {avg_latency} {timestamp}"
)
# Erfolgsrate
success_rate = sla_status.get('success_rate', '0')
output_lines.append(
f"{self.METRIC_PREFIX}success_rate {success_rate} {timestamp}"
)
# SLA-Status
sla_status_value = 1 if sla_status.get('status') == 'SLA_MET' else 0
output_lines.append(
f"{self.METRIC_PREFIX}sla_met {sla_status_value} {timestamp}"
)
return "\n".join(output_lines)
def generate_grafana_dashboard_json(self) -> Dict:
"""
Generiert vordefinierte Grafana-Dashboard-JSON
Für direkten Import in Grafana
"""
dashboard = {
"title": "HolySheep Gateway SLA Dashboard",
"tags": ["holysheep", "ai-gateway", "sla"],
"timezone": "browser",
"panels": [
{
"title": "Anfragen pro Minute",
"type": "graph",
"targets": [
{
"expr": f"rate({self.METRIC_PREFIX}requests_total[1m])",
"legendFormat": "RPM"
}
]
},
{
"title": "Erfolgsrate",
"type": "gauge",
"targets": [
{
"expr": f"{self.METRIC_PREFIX}success_rate",
"legendFormat": "Success Rate"
}
],
"fieldConfig": {
"defaults": {
"thresholds": {
"mode": "absolute",
"steps": [
{"color": "red", "value": None},
{"color": "yellow", "value": 99},
{"color": "green", "value": 99.5}
]
},
"unit": "percent"
}
}
},
{
"title": "p95 Latenz",
"type": "graph",
"targets": [
{
"expr": f"{self.METRIC_PREFIX}latency_p95_ms",
"legendFormat": "p95 Latenz (ms)"
}
]
},
{
"title": "Fehler-Breakdown",
"type": "piechart",
"targets": [
{
"expr": f"{self.METRIC_PREFIX}errors_total",
"legendFormat": "{{type}}"
}
]
}
]
}
return dashboard
============================================================
ALERTING-KONFIGURATION
============================================================
ALERT_RULES = """
groups:
- name: holy_sheep_sla_alerts
rules:
- alert: HolySheepHighErrorRate
expr: rate(holysheep_gateway_requests_failed_total[5m]) > 0.01
for: 2m
labels:
severity: critical
annotations:
summary: "Hohe Fehlerrate bei HolySheep Gateway"
description: "Fehlerrate übersteigt 1% für mehr als 2 Minuten"
- alert: HolySheepSLAViolation
expr: holysheep_gateway_success_rate < 99.5
for: 5m
labels:
severity: warning
annotations:
summary: "SLA-Ziel nicht erreicht"
description: "Erfolgsrate {{ $value }}% unter 99.5% Ziel"
- alert: HolySheepHighLatency
expr: holysheep_gateway_latency_p95_ms > 2000
for: 3m
labels:
severity: warning
annotations:
summary: "Hohe Latenzzeit"
description: "p95 Latenz {{ $value }}ms über 2000ms"
- alert: HolySheepRateLimitSpike
expr: rate(holysheep_gateway_errors_total{type="rate_limits"}[5m]) > 0.05
for: 1m
labels:
severity: info
annotations:
summary: "Rate-Limit-Spitze erkannt"
description: "Erhöhte Rate-Limit-Anfragen - Fallback aktiv"
"""
def main():
"""Demonstriert Prometheus-Export"""
exporter = PrometheusMetricsExporter()
# Beispiel SLA-Status (wie von HolySheep Gateway erhalten)
example_sla_status = {
"status": "SLA_MET",
"success_rate": "99.72%",
"avg_latency_ms": "42.5ms",
"p95_latency_ms": "78.3ms",
"total_requests": 15420,
"successful_requests": 15381,
"failed_requests": 39,
"error_breakdown": {
"rate_limits": 18,
"server_errors": 8,
"timeouts": 13,
"fallbacks": 22
}
}
# Prometheus-Metriken generieren
prometheus_output = exporter.export_metrics(example_sla_status)
print("=" * 60)
print("PROMETHEUS METRIC EXPORT")
print("=" * 60)
print(prometheus_output)
# Grafana Dashboard JSON
print("\n" + "=" * 60)
print("GRAFANA DASHBOARD (JSON - auszugsweise)")
print("=" * 60)
dashboard = exporter.generate_grafana_dashboard_json()
print(f"Titel: {dashboard['title']}")
print(f"Panels: {len(dashboard['panels'])}")
for panel in dashboard['panels']:
print(f" - {panel['title']} ({panel['type']})")
# Alert Rules speichern
print("\n" + "=" * 60)
print("PROMETHEUS ALERT RULES")
print("=" * 60)
print(ALERT_RULES)
if __name__ == "__main__":
main()Häufige Fehler und Lösungen
In meiner Produktionserfahrung mit dem HolySheep Gateway sind folgende Fehler die häufigsten Stolperfallen. Hier sind bewährte Lösungen:
Fehler 1: Unbehandelte 429 Rate-Limit-Fehler
Symptom: API-Anfragen schlagen fehl, ohne dass ein Retry stattfindet. Logs zeigen wiederholt "429 Too Many Requests".
Lösung:
# FEHLERHAFT - Keine Retry-Logik:
response = await client.post("/chat/completions", json=data)
KORREKT - Exponential Backoff mit Retry:
async def request_with_retry(
client: httpx.AsyncClient,
url: str,
json_data: dict,
max_retries: int = 3
) -> httpx.Response:
"""
Anfrage mit Exponential Backoff bei Rate-Limits
Retry-Verhalten:
- 1. Versuch: Sofort
- 2. Versuch: 1 Sekunde warten
- 3. Versuch: 2 Sekunden warten
- 4. Versuch: 4 Sekunden warten (max)
"""
for attempt in range(max_retries):
try:
response = await client.post(url, json=json_data)
if response.status_code == 200:
return response
elif response.status_code == 429:
# Rate Limit erreicht - Retry mit Backoff
retry_after = int(response.headers.get("Retry-After", 1))
wait_time = min(2 ** attempt, retry_after)
logging.warning(
f"Rate Limit (429) bei Versuch {attempt + 1}. "
f"Warte {wait_time}s..."
)
await asyncio.sleep(wait_time)
continue
else:
# Anderer Fehler - nicht retry
response.raise_for_status()
return response
except httpx.TimeoutException:
if attempt < max_retries - 1:
wait_time = 2 ** attempt
logging.warning(
f"Timeout bei Versuch {attempt + 1}. "
f"Warte {wait_time}s..."
)
await asyncio.sleep(wait_time)
else:
raise
except httpx.HTTPStatusError as e:
if e.response.status_code >= 500 and attempt < max_retries - 1:
# Server-Fehler - kurz warten und retry
await asyncio.sleep(1)
continue
else:
raise
raise Exception(f"Max retries ({max_retries}) erreicht")Fehler 2: Fehlender Modell-Fallback bei 5xx-Fehlern
Symptom: Bei Serverausfällen des Primärmodells bleibt der AI Agent komplett stehen, obwohl Alternativen verfügbar wären.
Lösung:
# FEHLERHAFT - Kein Fallback:
async def call_model(model: str, messages: list):
response = await client.post(f"/chat/completions", json={
"model": model,
"messages": messages
})
return response.json() # Schlägt komplett fehl bei 5xx
KORREKT - Automatischer Modell-Fallback:
class ModelRouter:
"""
Intelligenter Router mit automatischem Fallback
Konfiguration (Beispiel):
- Primär: gpt-4.1 (beste Qualität, $8/MTok)
- Fallback 1: claude-sonnet-4.5 ($15/MTok)
- Fallback 2: gemini-2.5-flash ($2.50/MTok)
- Fallback 3: deepseek-v3.2 ($0.42/MTok)
"""
FALLBACK_ORDER = [
"gpt-4.1", # Primär
"claude-sonnet-4.5", # Sekundär Premium
"gemini-2.5-flash", # Budget Alternative
"deepseek-v3.2" # Notfall-Budget
]
async def call_with_fallback(
self,
messages: list,
prefer_model: str = "gpt-4.1"
) -> tuple[dict, str]:
"""
Führt Anfrage mit automatischem Fallback aus
Returns:
Tuple von (Response-Dict, Verwendetes Modell)
"""
# Fallback-Liste erstellen basierend auf Präferenz
models = [prefer_model]
for model in self.FALLBACK_ORDER:
if model != prefer_model:
models.append(model)
last_error = None
for model in models:
try:
logging.info(f"Versuche Modell: {model}")
response = await self.client.post("/chat/completions", json={
"model": model,
"messages": messages,
"max_tokens": 1000
})
if response.status_code == 200:
result = response.json()
logging.info(
f"Erfolgreich mit Modell: {model}"
)
return result, model
elif 500 <= response.status_code < 600:
# Server-Fehler - sofort nächtes Modell
last_error = f"{model}: {response.status_code}"
logging.warning(
f"Server-Fehler von {model}, "
f"probiere nächstes Modell..."
)
continue
elif response.status_code == 429:
# Rate Limit - kurze Pause
await asyncio.sleep(1)
continue
else:
# Andere Fehler - nicht retry
response.raise_for_status()
except Exception as e:
last_error = str(e)
logging.error(
f"Fehler bei {model}: {e}"
)
continue
# Alle Modelle fehlgeschlagen
raise RuntimeError(
f"Alle Modelle fehlgeschlagen. "
f"Letzter Fehler: {last_error}"
)Fehler 3: Timeout ohne Graceful Degradation
Symptom: Bei Timeout wartet der AI Agent ewig, ohne dem Benutzer Feedback zu geben oder eine alternative Antwort zu liefern.
Lösung:
# FEHLERHAFT - Kein Timeout-Handling:
response = await client.post("/chat/completions", json=data)
Hängt bei langsamer Antwort...
KORREKT - Timeout mit Fallback und User-Feedback:
class TimeoutHandler:
"""
Behandelt Timeouts mit Graceful Degradation
Strategie:
1. Timeout erreicht → Sofortiger Fallback zu schnellem Modell
2. Wenn Fallback erfolgreich → Rückgabe mit "degraded" Flag
3. Wenn alles fehlschlägt → Strukturierte Fehlerantwort
"""
FAST_MODELS = ["deepseek-v3.2", "gemini-2.5-flash"]
DEFAULT_TIMEOUT = 5.0 # Sekunden
async def call_with_timeout_handling(
self,
messages: list,
timeout: float = DEFAULT_TIMEOUT,
prefer_fast: bool = True
) -> dict:
"""
Führt Anfrage mit Timeout-Protection aus
Returns:
{
"success": bool,