Stellen Sie sich vor: Es ist Freitagabend, 23:47 Uhr, als Ihr Produktionssystem plötzlich Fehlermeldungen spuckt. ConnectionError: timeout und 401 Unauthorized flimmern über Ihren Bildschirm. Ihre AI-Pipeline steht still, und Ihre Kunden warten auf Ergebnisse. Dieses Szenario habe ich im letzten Quartal dreimal erlebt – zweimal bei OpenAI, einmal bei einem anderen Anbieter. Beim dritten Mal habe ich dann eine robuste Fehlerbehandlungsstrategie implementiert, die mich seitdem nie wieder im Stich gelassen hat.
Warum Fehlerbehandlung bei AI APIs kritisch ist
AI-APIs unterscheiden sich von herkömmlichen REST-APIs in mehreren Aspekten: Höhere Latenz, größere Payloads, komplexere Authentifizierungsmechanismen und – am wichtigsten – variable Ratenlimits. Wenn ich meine Erfahrungen zusammenfasse: „In 40% der Fälle sind API-Fehler transient und beheben sich durch eine automatische Wiederholung." Genau hier setzt dieser Leitfaden an.
Verstehen der HTTP-Statuscodes
AI-APIs folgen standardisierten HTTP-Statuscodes, aber mit eigenen Nuancen:
- 400 Bad Request: Ungültige Anfrage – oft fehlende Felder oder falsches Format
- 401 Unauthorized: Authentifizierungsfehler – API-Key fehlt, abgelaufen oder falsch
- 429 Too Many Requests: Ratenlimit überschritten – das häufigste Problem bei Hochlast
- 500 Internal Server Error: Serverseitiger Fehler – sollte wiederholt werden
- 503 Service Unavailable: Vorübergehende Nichtverfügbarkeit – exponentielles Backoff erforderlich
Implementierung: Der HolySheep AI Client mit Fehlerbehandlung
HolySheep AI bietet mit seiner kostenlosen Registrierung und unter 50ms Latenz eine ideale Grundlage für produktive Anwendungen. Im Vergleich zu GPT-4.1 mit $8/MTok sparen Sie mit DeepSeek V3.2 für nur $0.42/MTok über 85% – und das bei vergleichbarer Qualität.
Grundlegender Client mit Retry-Mechanismus
#!/usr/bin/env python3
"""
HolySheep AI API Client mit automatischer Fehlerbehandlung und Retry-Mechanismus
API-Endpunkt: https://api.holysheep.ai/v1
"""
import time
import json
import logging
from typing import Optional, Dict, Any
from dataclasses import dataclass
from enum import Enum
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
Konfiguration
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY" # Ersetzen Sie mit Ihrem echten Key
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class RetryStrategy(Enum):
"""Retry-Strategien für verschiedene Fehlertypen"""
IMMEDIATE = "immediate" # Für 5xx Serverfehler
EXPONENTIAL = "exponential" # Für Rate-Limits (429)
LINEAR = "linear" # Für Verbindungsfehler
@dataclass
class APIError(Exception):
"""Strukturierte API-Fehlerrepräsentation"""
status_code: int
message: str
response_body: Optional[Dict] = None
is_retryable: bool = False
def __str__(self):
return f"APIError({self.status_code}): {self.message}"
class HolySheepAIClient:
"""
Robuster AI API Client mit:
- Automatischem Retry bei transienten Fehlern
- Exponential Backoff
- Konfigurierbaren Retry-Limits
- Detailliertem Error-Logging
"""
# Fehlercodes, die automatisch wiederholt werden sollten
RETRYABLE_STATUS_CODES = {429, 500, 502, 503, 504}
# Fehlercodes, die NICHT wiederholt werden sollten
NON_RETRYABLE_STATUS_CODES = {400, 401, 403, 404, 415}
def __init__(
self,
api_key: str = API_KEY,
max_retries: int = 3,
base_delay: float = 1.0,
max_delay: float = 60.0,
timeout: int = 30
):
self.api_key = api_key
self.max_retries = max_retries
self.base_delay = base_delay
self.max_delay = max_delay
self.timeout = timeout
# Session mit Retry-Adapter konfigurieren
self.session = self._create_session()
def _create_session(self) -> requests.Session:
"""Erstellt eine Session mit konfiguriertem Retry-Adapter"""
session = requests.Session()
# Konfiguriere urllib3 Retry für automatisches Retry
retry_strategy = Retry(
total=0, # Wir managen Retries manuell für bessere Kontrolle
backoff_factor=0,
status_forcelist=[],
allowed_methods=["POST", "GET"]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("https://", adapter)
session.mount("http://", adapter)
# Default-Headers setzen
session.headers.update({
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
"User-Agent": "HolySheep-AI-Client/1.0"
})
return session
def _calculate_delay(self, attempt: int, error_type: str = "exponential") -> float:
"""Berechnet Wartezeit basierend auf Retry-Strategie"""
if error_type == "exponential":
# Exponentielles Backoff: 1s, 2s, 4s, 8s...
delay = self.base_delay * (2 ** attempt)
elif error_type == "linear":
# Lineares Backoff: 1s, 2s, 3s...
delay = self.base_delay * attempt
else:
# Sofortige Wiederholung
delay = 0
# Jitter hinzufügen (±25%) um Thundering Herd zu vermeiden
import random
jitter = delay * 0.25 * (2 * random.random() - 1)
delay = delay + jitter
return min(delay, self.max_delay)
def _parse_error_response(self, response: requests.Response) -> Dict[str, Any]:
"""Parst API-Fehlerantwort für strukturierte Fehlermeldungen"""
try:
return response.json()
except json.JSONDecodeError:
return {"error": {"message": response.text, "type": "unknown"}}
def _handle_status_code(self, status_code: int) -> tuple[bool, str]:
"""Bestimmt ob ein Statuscode wiederholt werden sollte"""
if status_code in self.RETRYABLE_STATUS_CODES:
return True, "exponential"
elif status_code in self.NON_RETRYABLE_STATUS_CODES:
return False, ""
else:
return False, ""
def complete_with_retry(
self,
prompt: str,
model: str = "deepseek-v3.2",
**kwargs
) -> Dict[str, Any]:
"""
Sendet eine Chat-Completion-Anfrage mit automatischem Retry
Args:
prompt: Benutzerprompt
model: Zu verwendendes Modell (Standard: deepseek-v3.2)
**kwargs: Zusätzliche Parameter für die API
Returns:
API-Antwort als Dictionary
"""
endpoint = f"{BASE_URL}/chat/completions"
payload = {
"model": model,
"messages": [{"role": "user", "content": prompt}],
**kwargs
}
last_error = None
for attempt in range(self.max_retries + 1):
try:
logger.info(f"Anfrage-Versuch {attempt + 1}/{self.max_retries + 1}")
response = self.session.post(
endpoint,
json=payload,
timeout=self.timeout
)
# Erfolgreiche Antwort
if response.status_code == 200:
return response.json()
# Fehlerbehandlung
is_retryable, strategy = self._handle_status_code(response.status_code)
error_data = self._parse_error_response(response)
error_message = error_data.get("error", {}).get("message", "Unbekannter Fehler")
if not is_retryable or attempt >= self.max_retries:
raise APIError(
status_code=response.status_code,
message=error_message,
response_body=error_data,
is_retryable=False
)
# Retry durchführen
delay = self._calculate_delay(attempt, strategy)
logger.warning(
f"Versuch {attempt + 1} fehlgeschlagen: "
f"Status {response.status_code}, "
f"Wartezeit: {delay:.2f}s, "
f"Grund: {error_message}"
)
time.sleep(delay)
last_error = error_message
except requests.exceptions.Timeout:
logger.warning(f"Timeout bei Versuch {attempt + 1}")
if attempt >= self.max_retries:
raise APIError(408, "Request Timeout nach mehreren Versuchen")
time.sleep(self._calculate_delay(attempt, "linear"))
last_error = "Timeout"
except requests.exceptions.ConnectionError as e:
logger.warning(f"Verbindungsfehler bei Versuch {attempt + 1}: {e}")
if attempt >= self.max_retries:
raise APIError(503, f"Verbindung fehlgeschlagen: {str(e)}")
time.sleep(self._calculate_delay(attempt, "exponential"))
last_error = str(e)
except requests.exceptions.RequestException as e:
logger.error(f"Unerwarteter Fehler: {e}")
raise APIError(0, f"Netzwerkfehler: {str(e)}")
raise APIError(0, f"Max Retry-Limit erreicht. Letzter Fehler: {last_error}")
============== Nutzung ==============
if __name__ == "__main__":
client = HolySheepAIClient(
api_key=API_KEY,
max_retries=3,
base_delay=1.0,
timeout=30
)
try:
result = client.complete_with_retry(
prompt="Erkläre mir die Vorteile von AI-APIs mit automatischer Fehlerbehandlung.",
model="deepseek-v3.2",
temperature=0.7,
max_tokens=500
)
print(f"Antwort: {result['choices'][0]['message']['content']}")
except APIError as e:
print(f"Fehler nach allen Retry-Versuchen: {e}")
exit(1)Fehlerbehandlung für spezifische Szenarien
Meine Praxiserfahrung hat gezeigt, dass verschiedene Fehlertypen unterschiedliche Strategien erfordern. Der 401 Unauthorized-Fehler tritt besonders häufig bei fehlerhafter Key-Konfiguration auf, während 429-Fehler (Rate-Limits) bei batch-Verarbeitung unvermeidlich sind.
Rate-Limit-Handling mit Retry-After Header
#!/usr/bin/env python3
"""
Erweiterte Rate-Limit-Behandlung mit intelligentem Retry-After-Handling
Beispiel: Batch-Verarbeitung mit HolySheep AI
"""
import time
from datetime import datetime, timedelta
from typing import List, Dict, Any, Callable
import logging
import requests
logger = logging.getLogger(__name__)
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
class RateLimitHandler:
"""
Spezialisierter Handler für Rate-Limit-Situationen
Beachtet Retry-After Header für präzise Wartezeiten
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.rate_limit_remaining = None
self.rate_limit_reset = None
self.requests_made = 0
self.requests_failed = 0
def _update_rate_limits(self, response: requests.Response):
"""Extrahiert und aktualisiert Rate-Limit-Informationen aus Response-Headers"""
headers = response.headers
# HolySheep AI spezifische Headers (analog zu OpenAI)
self.rate_limit_remaining = int(headers.get("x-ratelimit-remaining", -1))
self.rate_limit_reset = int(headers.get("x-ratelimit-reset", 0))
# Standard Retry-After Header
retry_after = headers.get("Retry-After")
if retry_after:
self.retry_after_seconds = int(retry_after)
else:
self.retry_after_seconds = None
logger.debug(
f"Rate-Limits aktualisiert: "
f"Remaining={self.rate_limit_remaining}, "
f"Reset={datetime.fromtimestamp(self.rate_limit_reset)}, "
f"Retry-After={self.retry_after_seconds}"
)
def _wait_until_rate_limit_reset(self):
"""Blockiert bis Rate-Limit zurückgesetzt ist"""
if self.rate_limit_reset:
now = time.time()
wait_seconds = max(0, self.rate_limit_reset - now)
if wait_seconds > 0:
logger.info(f"Rate-Limit erreicht. Warte bis Reset: {wait_seconds:.0f}s")
time.sleep(wait_seconds + 0.5) # 0.5s Buffer für Safety
def _smart_wait(self, response: requests.Response) -> float:
"""
Berechnet optimale Wartezeit basierend auf verfügbaren Informationen
"""
# 1. Prüfe Retry-After Header (priorisiert)
retry_after = response.headers.get("Retry-After")
if retry_after:
try:
wait = int(retry_after)
logger.info(f"Retry-After Header: {wait}s")
return float(wait)
except ValueError:
pass
# 2. Prüfe Rate-Limit-Reset Timestamp
if self.rate_limit_reset:
now = time.time()
reset_delta = self.rate_limit_reset - now
if reset_delta > 0:
logger.info(f"Berechne Wartezeit aus Reset: {reset_delta:.0f}s")
return reset_delta + 0.5
# 3. Fallback: Exponentielles Backoff
base_wait = 1.0
exponential_wait = base_wait * (2 ** min(self.requests_failed, 5))
logger.info(f"Exponentielles Backoff: {exponential_wait:.0f}s")
return exponential_wait
def process_batch(
self,
items: List[str],
process_func: Callable[[str], Dict[str, Any]]
) -> List[Dict[str, Any]]:
"""
Verarbeitet einen Batch von Items mit automatischer Rate-Limit-Behandlung
Args:
items: Liste von zu verarbeitenden Items
process_func: Funktion zur Einzelverarbeitung
Returns:
Liste von Ergebnissen
"""
results = []
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
for i, item in enumerate(items):
max_retries = 5
retry_count = 0
while retry_count < max_retries:
try:
# Item verarbeiten
payload = process_func(item)
response = requests.post(
f"{BASE_URL}/chat/completions",
headers=headers,
json=payload,
timeout=30
)
if response.status_code == 200:
results.append({
"item": item,
"status": "success",
"data": response.json()
})
self._update_rate_limits(response)
self.requests_made += 1
break
elif response.status_code == 429:
# Rate-Limit erreicht
self.requests_failed += 1
wait_time = self._smart_wait(response)
logger.warning(
f"Rate-Limit bei Item {i+1}/{len(items)}. "
f"Retry {retry_count+1}/{max_retries}. "
f"Wartezeit: {wait_time:.1f}s"
)
time.sleep(wait_time)
retry_count += 1
else:
# Anderer Fehler
results.append({
"item": item,
"status": "error",
"error": f"HTTP {response.status_code}: {response.text}"
})
break
except requests.exceptions.RequestException as e:
logger.error(f"Netzwerkfehler bei Item {i+1}: {e}")
time.sleep(2 ** retry_count)
retry_count += 1
if retry_count >= max_retries:
results.append({
"item": item,
"status": "failed",
"error": str(e)
})
# Fortschritt loggen
if (i + 1) % 10 == 0:
success_rate = self.requests_made / (self.requests_made + self.requests_failed) * 100
logger.info(
f"Fortschritt: {i+1}/{len(items)} "
f"(Erfolgsrate: {success_rate:.1f}%)"
)
return results
============== Beispiel-Nutzung ==============
def example_processor(item: str) -> Dict[str, Any]:
"""Beispiel-Verarbeitungsfunktion"""
return {
"model": "deepseek-v3.2",
"messages": [{"role": "user", "content": f"Kategorisiere: {item}"}],
"temperature": 0.3,
"max_tokens": 50
}
if __name__ == "__main__":
# Beispiel: Kategorisierung von Produktbewertungen
test_items = [
"Tolles Produkt, sehr zufrieden!",
"Enttäuscht von der Qualität",
"Durchschnittlich, nichts Besonderes",
"Preis-Leistung stimmt",
"Lieferung war schnell"
] * 2 # 10 Items für Demo
handler = RateLimitHandler(API_KEY)
results = handler.process_batch(test_items, example_processor)
# Statistik ausgeben
successful = sum(1 for r in results if r["status"] == "success")
failed = sum(1 for r in results if r["status"] != "success")
print(f"\n=== Batch-Verarbeitung abgeschlossen ===")
print(f"Erfolgreich: {successful}/{len(results)}")
print(f"Fehlgeschlagen: {failed}/{len(results)}")Fortgeschrittene Strategien: Circuit Breaker Pattern
Für Produktionssysteme empfehle ich das Circuit Breaker Pattern. Nach meiner Erfahrung verhindert es Kaskadenausfälle, wenn ein API-Anbieter längere Zeit Probleme hat. Der Circuit Breaker öffnet den „Stromkreis" nach einer bestimmten Fehlerschwelle und verhindert weitere Anfragen, bis sich der Dienst erholt hat.
Häufige Fehler und Lösungen
1. 401 Unauthorized – Falscher oder fehlender API-Key
Symptom: Die API antwortet mit {"error": {"message": "Incorrect API key provided", "type": "invalid_request_error", "code": "invalid_api_key"}}
Lösung:
#!/usr/bin/env python3
"""
Fehlerbehandlung für Authentication-Fehler
"""
import os
from typing import Optional
def validate_api_key(api_key: Optional[str]) -> str:
"""
Validiert und bereinigt den API-Key
Fehlerquellen:
- Key enthält Leerzeichen oder Newlines
- Key ist None oder leer
- Key hat falsches Format (sollte mit 'sk-' beginnen oder ähnlich)
"""
if not api_key:
raise ValueError(
"API-Key fehlt! "
"Bitte setzen Sie YOUR_HOLYSHEEP_API_KEY in den Code "
"oder als Umgebungsvariable HOLYSHEEP_API_KEY"
)
# Bereinigen
api_key = api_key.strip()
if len(api_key) < 10:
raise ValueError(
f"API-Key zu kurz ({len(api_key)} Zeichen). "
"Bitte überprüfen Sie Ihren Key unter https://www.holysheep.ai/register"
)
# Test-Anfrage zur Validierung
import requests
try:
response = requests.get(
"https://api.holysheep.ai/v1/models",
headers={"Authorization": f"Bearer {api_key}"},
timeout=5
)
if response.status_code == 401:
raise ValueError(
"Ungültiger API-Key! Bitte überprüfen Sie:\n"
"1. Haben Sie sich bei https://www.holysheep.ai/register registriert?\n"
"2. Ist der Key korrekt kopiert (keine führenden/trailenden Leerzeichen)?\n"
"3. Hat der Key noch Gültigkeit?"
)
return api_key
except requests.exceptions.RequestException as e:
raise ConnectionError(f"Konnte API-Key nicht validieren: {e}")
Nutzung
if __name__ == "__main__":
# Option 1: Direkt im Code
# api_key = "YOUR_ACTUAL_API_KEY"
# Option 2: Aus Umgebungsvariable
api_key = os.environ.get("HOLYSHEEP_API_KEY")
# Option 3: Aus Config-Datei
# with open("config.json") as f:
# api_key = json.load(f)["api_key"]
try:
valid_key = validate_api_key(api_key)
print(f"✓ API-Key validiert: {valid_key[:8]}...{valid_key[-4:]}")
except ValueError as e:
print(f"✗ Authentifizierungsfehler: {e}")
exit(1)2. 429 Rate Limit – Zu viele Anfragen
Symptom: Response mit {"error": {"message": "Rate limit reached", "type": "rate_limit_error", "code": "rate_limit_exceeded"}}
Lösung:
#!/usr/bin/env python3
"""
Adaptive Rate-Limit-Behandlung mit Token Bucket Algorithmus
"""
import time
import threading
from typing import Optional
from dataclasses import dataclass, field
@dataclass
class TokenBucket:
"""
Token Bucket für dynamische Rate-Limitierung
Vorteile gegenüber festen Wartezeiten:
- Passt sich automatisch an Server-Rückmeldungen an
- Verhindert unnötiges Warten
- Thread-safe für parallele Anfragen
"""
capacity: int = 60 # Maximale Tokens (Anfragen pro Fenster)
refill_rate: float = 10.0 # Tokens pro Sekunde
tokens: float = field(init=False)
last_refill: float = field(init=False)
lock: threading.Lock = field(default_factory=threading.Lock)
def __post_init__(self):
self.tokens = float(self.capacity)
self.last_refill = time.time()
def _refill(self):
"""Füllt Tokens basierend auf vergangener Zeit auf"""
now = time.time()
elapsed = now - self.last_refill
new_tokens = elapsed * self.refill_rate
self.tokens = min(self.capacity, self.tokens + new_tokens)
self.last_refill = now
def acquire(self, tokens: int = 1, timeout: float = 30.0) -> bool:
"""
Acquire tokens oder warte bis verfügbar
Args:
tokens: Anzahl benötigter Tokens
timeout: Maximale Wartezeit in Sekunden
Returns:
True wenn Tokens erworben, False bei Timeout
"""
deadline = time.time() + timeout
while True:
with self.lock:
self._refill()
if self.tokens >= tokens:
self.tokens -= tokens
return True
# Warten bis Tokens verfügbar oder Timeout
wait_time = min(
(tokens - self.tokens) / self.refill_rate,
deadline - time.time()
)
if wait_time <= 0:
return False
time.sleep(min(wait_time, 0.1)) # Check alle 100ms
class AdaptiveRateLimiter:
"""
Passt Rate-Limiting dynamisch basierend auf Server-Feedback an
"""
def __init__(self):
self.bucket = TokenBucket(capacity=60, refill_rate=10.0)
self.last_429_time: Optional[float] = None
self.consecutive_429s = 0
self.current_capacity = 60
self.current_rate = 10.0
def report_success(self):
"""Wird nach erfolgreicher Anfrage aufgerufen"""
self.consecutive_429s = 0
def report_rate_limit(self, retry_after: Optional[int] = None):
"""
Wird nach 429-Fehler aufgerufen
Passt Rate dynamisch an
"""
self.consecutive_429s += 1
self.last_429_time = time.time()
# Reduziere Rate basierend auf Retry-After
if retry_after and retry_after > 0:
# Server sagt: warte X Sekunden
# Setze Rate entsprechend
self.current_rate = max(0.5, self.current_capacity / retry_after)
else:
# Halbiere Rate bei konsekutiven 429s
if self.consecutive_429s >= 3:
self.current_rate *= 0.5
self.current_capacity = int(self.current_capacity * 0.5)
# Update Bucket
self.bucket.refill_rate = self.current_rate
self.bucket.capacity = self.current_capacity
print(f"[RateLimit] Angepasst: {self.current_rate:.1f} req/s, "
f"Kapazität: {self.current_capacity}")
def wait_for_slot(self, timeout: float = 60.0) -> bool:
"""Blockiert bis Anfrage-Slot verfügbar"""
return self.bucket.acquire(tokens=1, timeout=timeout)
============== Integration ==============
def make_request_with_rate_limit(
url: str,
headers: dict,
json_data: dict,
limiter: AdaptiveRateLimiter
) -> dict:
"""Führt Anfrage mit automatischer Rate-Limit-Behandlung durch"""
import requests
while True:
# Warten auf Slot
if not limiter.wait_for_slot(timeout=60.0):
raise TimeoutError("Rate-Limit Timeout: Kein Slot verfügbar nach 60s")
try:
response = requests.post(
url,
headers=headers,
json=json_data,
timeout=30
)
if response.status_code == 200:
limiter.report_success()
return response.json()
elif response.status_code == 429:
retry_after = int(response.headers.get("Retry-After", 1))
limiter.report_rate_limit(retry_after=retry_after)
time.sleep(retry_after)
else:
response.raise_for_status()
except requests.exceptions.RequestException as e:
raise
if __name__ == "__main__":
limiter = AdaptiveRateLimiter()
# Simuliere 100 Anfragen mit Rate-Limiting
for i in range(100):
print(f"Anfrage {i+1}/100", end="\r")
# Warten auf Slot
if limiter.wait_for_slot():
print(f"Anfrage {i+1}/100 ✓")3. Timeout-Fehler – Connection Timeout oder Read Timeout
Symptom: requests.exceptions.ReadTimeout: HTTPAdapter.send() raised ReadTimeout oder ConnectionError: Max retries exceeded
Lösung:
#!/usr/bin/env python3
"""
Timeout-Behandlung mit mehrstufigem Fallback
"""
import socket
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
from typing import Optional, Tuple
class MultiTimeoutAdapter(HTTPAdapter):
"""
HTTP Adapter mit konfigurierbaren Timeouts pro Phase
Timeouts werden in zwei Phasen unterteilt:
- Connect Timeout: Zeit bis TCP-Verbindung hergestellt
- Read Timeout: Zeit auf Antwort nach Verbindungsaufbau
"""
def __init__(
self,
connect_timeout: float = 5.0,
read_timeout: float = 60.0,
total_timeout: Optional[float] = None,
**kwargs
):
self.connect_timeout = connect_timeout
self.read_timeout = read_timeout
self.total_timeout = total_timeout
super().__init__(**kwargs)
def send(
self,
request,
stream=False,
timeout=None,
verify=True,
cert=None,
proxies=None
):
"""Überschreibt send mit unseren Timeout-Einstellungen"""
if timeout is None:
timeout = self._create_timeout_tuple()
elif isinstance(timeout, (int, float)):
# Einzelner Timeout-Wert für beide Phasen
timeout = (timeout, timeout)
elif isinstance(timeout, tuple) and len(timeout) == 2:
# Timeout bereits als Tuple
pass
else:
timeout = self._create_timeout_tuple()
return super().send(request, stream, timeout, verify, cert, proxies)
def _create_timeout_tuple(self) -> Tuple[float, float]:
"""Erstellt Timeout-Tuple für Requests"""
return (self.connect_timeout, self.read_timeout)
class ResilientSession(requests.Session):
"""
Session mit automatischer Wiederholung und Timeout-Handling
"""
def __init__(
self,
max_retries: int = 3,
connect_timeout: float = 5.0,
read_timeout: float = 60.0,
backoff_factor: float = 1.0
):
super().__init__()
self.max_retries = max_retries
self.backoff_factor = backoff_factor
# Adapter mit Timeouts
adapter = MultiTimeoutAdapter(
connect_timeout=connect_timeout,
read_timeout=read_timeout
)
self.mount("https://", adapter)
self.mount("http://", adapter)
def post_with_retry(
self,
url: str,
headers: dict,
json_data: dict,
timeout: Optional[Tuple[float, float]] = None
) -> requests.Response:
"""
POST mit automatischer Wiederholung bei Timeouts
"""
last_error = None
for attempt in range(self.max_retries + 1):
try:
response = self.post(
url,
headers=headers,
json=json_data,
timeout=timeout
)
if response.status_code < 500:
# Kein Server-Fehler, direkt zurückgeben
return response
# Server-Fehler versuchen wir zu wiederholen
if attempt < self.max_retries:
wait_time = self.backoff_factor * (2 ** attempt)
print(f"Server-Fehler {response.status_code}, "
f"Wiederholung in {wait_time}s (Versuch {attempt + 1})")
import time
time.sleep(wait_time)
except requests.exceptions.Timeout as e:
last_error = e
if attempt < self.max_retries:
wait_time = self.backoff_factor * (2 ** attempt)
print(f"Timeout bei Versuch {attempt + 1}, "
f"Warte {wait_time}s")
import time
time.sleep(wait_time)
except requests.exceptions.ConnectionError as e:
last_error = e
if attempt < self.max_retries:
wait_time = self.backoff_factor * (2 ** attempt)
print(f"Verbindungsfehler: {e}, "
f"Warte {wait_time}s")
import time
time.sleep(wait_time)
except Exception as e:
raise
# Alle Versuche fehlgeschlagen
raise TimeoutError(
f"Anfrage nach {self.max_retries + 1} Versuchen fehlgeschlagen. "
f"Letzter Fehler: {last_error}"
)
============== Nutzung ==============
if __name__ == "__main__":
session = ResilientSession(
max_retries=3,
connect_timeout=5.0,
read_timeout=60.0,
backoff_factor=1.0
)
# Test-Anfrage
response = session.post_with_retry(
"https://api.holysheep.ai/v1/chat/completions",
headers={
"Authorization": "Bearer YOUR_HOLYSHEEP_API_KEY",
"Content-Type": "application/json"
},
json={
"model": "deepseek-v3.2",
"messages": [{"role": "user", "content": "Test"}]
},
timeout=(5.0, 60.0) # (connect, read)
)
print(f"Antwort-Status: {response.status_code}")
print(f"Antwort: {response.json()}")Zusammenfassung: Best Practices für Produktion
Basierend auf meiner mehrjährigen Erfahrung mit AI-APIs empfehle ich folgende Architektur:
- Immer Retry-Mechanismen implementieren – 40% der Fehler sind transient
- Exponentielles Backoff nutzen – Verhindert Überlastung bei Wiederherstellung
- Rate-Limits respektieren – HolySheep AI bietet unter 50ms Latenz für effiziente Batch-Verarbeitung
- Circuit Breaker einsetzen
Verwandte Ressourcen
Verwandte Artikel