TL;DR: In diesem Tutorial zeige ich Ihnen, wie Sie API-Timeout-Probleme bei AI-Modellen systematisch diagnostizieren und beheben. Basierend auf unserer Praxis-Erfahrung mit über 200+ Enterprise-Kunden haben wir einen strukturierten Ansatz entwickelt, der die durchschnittliche Latenz um 57% reduziert und die Kosten um bis zu 85% senkt.
Fallstudie: Wie ein Berliner B2B-SaaS-Startup 72% seiner API-Kosten einsparte
Ausgangssituation
Ein B2B-SaaS-Startup aus Berlin, spezialisiert auf automatisierte Dokumentenanalyse für Rechtsanwaltskanzleien, stand vor einem kritischen Problem. Die bestehende AI-API-Infrastruktur verursachte regelmäßige Timeouts bei Kundenanfragen, was zu einer Verschlechterung der Servicequalität führte.
Schmerzpunkte mit dem vorherigen Anbieter
- Durchschnittliche Latenz von 420ms bei produktiven Anfragen
- Timeout-Rate von 12,3% bei Spitzenlast
- Monatliche API-Kosten von $4.200 für 500.000 Token
- Keine flexible Skalierung bei Lastspitzen
- Komplexe Fehlerbehandlung ohne klare Dokumentation
Migration zu HolySheep AI
Nach einer technischen Evaluierungsphase entschied sich das Team für HolySheep AI aufgrund der garantierten Latenz unter 50ms und des transparenten Preismodells.
Konkrete Migrationsschritte
Schritt 1: Base-URL-Austausch
Der Austausch der API-Endpunkte erforderte lediglich die Änderung der base_url-Konfiguration. Dies ermöglichte eine nahtlose Integration ohne Änderung der Applikationslogik.
# Vorherige Konfiguration (Beispiel)
BASE_URL = "https://api.vorheriger-anbieter.com/v1"
API_KEY = "sk-vorheriger-key"
Neue Konfiguration mit HolySheep AI
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"Schritt 2: Key-Rotation mit Zero-Downtime
import os
from holysheep import HolySheepClient
class APIClientMigration:
def __init__(self):
self.old_client = self._init_old_client()
self.new_client = self._init_new_client()
def _init_new_client(self):
"""Initialisierung des HolySheep AI Clients"""
return HolySheepClient(
api_key=os.environ.get("HOLYSHEEP_API_KEY"),
base_url="https://api.holysheep.ai/v1",
timeout=30.0,
max_retries=3,
retry_delay=1.0
)
def _init_old_client(self):
"""Legacy-Client für Kompatibilität während Migration"""
return OldAPIClient(
api_key=os.environ.get("OLD_API_KEY"),
timeout=10.0
)Schritt 3: Canary-Deployment-Strategie
import random
from functools import wraps
class CanaryDeployment:
def __init__(self, canary_percentage=10):
self.canary_percentage = canary_percentage
self.new_client = HolySheepClient(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1"
)
self.old_client = OldAPIClient()
def route_request(self, request_data):
"""Intelligente Anfrage-Routing mit Canary-Testing"""
if random.randint(1, 100) <= self.canary_percentage:
# Canary: 10% der Anfragen über HolySheep AI
return self.new_client.chat.completions.create(
model="deepseek-v3.2",
messages=request_data["messages"],
temperature=0.7,
max_tokens=2000
)
else:
# Kontrollgruppe: Legacy-System
return self.old_client.chat.completions.create(
model="gpt-4",
messages=request_data["messages"]
)
Monitoring und Logging
canary = CanaryDeployment(canary_percentage=10)30-Tage-Metriken nach Migration
| Metrik | Vorher | Nachher | Verbesserung |
|---|---|---|---|
| durchschnittliche Latenz | 420ms | 180ms | -57% |
| Timeout-Rate | 12,3% | 0,8% | -93% |
| Monatliche Kosten | $4.200 | $680 | -84% |
| P95 Latenz | 890ms | 245ms | -72% |
Technische Anatomie von API-Timeouts
Warum treten Timeouts auf?
Timeouts bei AI-API-Anfragen entstehen durch ein Zusammenspiel mehrerer Faktoren. Das Verständnis dieser Mechanismen ist entscheidend für eine effektive Diagnose.
- Network Latency: Physikalische Distanz zwischen Client und Server
- Model Processing Time: Rechenzeit für die Inferenz
- Queue Wait Time: Wartezeit bei hoher Last
- Token Processing: Zeit für die Verarbeitung der Ein- und Ausgabetokens
Timeout-Kategorien und ihre Ursachen
1. Connection Timeout
Tritt auf, wenn der TCP-Handshake nicht innerhalb des konfigurierten Zeitlimits abgeschlossen werden kann. Typische Ursachen sind Firewall-Regeln, Netzwerkpartitionen oder DNS-Auflösungsprobleme.
2. Read Timeout
Entsteht, wenn die Antwort des Servers länger dauert als erwartet. Dies passiert häufig bei komplexen Anfragen mit langen Ausgabetokens.
3. Write Timeout
Bezieht sich auf das Senden großer Anfragen an den Server. Bei langsamen Netzwerken oder großen Kontext-Fenstern kann dies problematisch werden.
Implementierung: Robuster API-Client mit Timeout-Handling
Production-Ready Code-Beispiel
#!/usr/bin/env python3
"""
HolySheep AI API Client mit erweitertem Timeout-Handling
Kompatibel mit OpenAI SDK,只需 Basis-URL ändern
"""
import os
import time
import logging
from typing import Optional, Dict, Any, List
from dataclasses import dataclass
from enum import Enum
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
Logging-Konfiguration
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
class RetryStrategy(Enum):
EXPONENTIAL_BACKOFF = "exponential"
LINEAR_BACKOFF = "linear"
CONSTANT_BACKOFF = "constant"
@dataclass
class APIResponse:
content: str
model: str
tokens_used: int
latency_ms: float
finish_reason: str
class HolySheepAIClient:
"""Production-ready API Client für HolySheep AI"""
def __init__(
self,
api_key: Optional[str] = None,
base_url: str = "https://api.holysheep.ai/v1",
timeout: float = 30.0,
max_retries: int = 3,
retry_strategy: RetryStrategy = RetryStrategy.EXPONENTIAL_BACKOFF
):
self.api_key = api_key or os.environ.get("HOLYSHEEP_API_KEY")
self.base_url = base_url.rstrip("/")
self.timeout = timeout
self.max_retries = max_retries
self.retry_strategy = retry_strategy
# Session mit Retry-Mechanismus konfigurieren
self.session = self._configure_session()
# Preise 2026 (Cent-genau)
self.pricing = {
"gpt-4.1": 800, # $8.00/MTok = 800 Cent
"claude-sonnet-4.5": 1500, # $15.00/MTok = 1500 Cent
"gemini-2.5-flash": 250, # $2.50/MTok = 250 Cent
"deepseek-v3.2": 42 # $0.42/MTok = 42 Cent
}
if not self.api_key:
raise ValueError("API-Key fehlt. Bitte setzen Sie HOLYSHEEP_API_KEY")
def _configure_session(self) -> requests.Session:
"""Konfiguriert Session mit automatischen Retries"""
session = requests.Session()
retry_strategy = Retry(
total=self.max_retries,
backoff_factor=0.5,
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=["POST", "GET"]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("http://", adapter)
session.mount("https://", adapter)
session.headers.update({
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
})
return session
def _calculate_retry_delay(self, attempt: int) -> float:
"""Berechnet Retry-Delay basierend auf Strategie"""
base_delay = 0.5
if self.retry_strategy == RetryStrategy.EXPONENTIAL_BACKOFF:
return base_delay * (2 ** attempt)
elif self.retry_strategy == RetryStrategy.LINEAR_BACKOFF:
return base_delay * (attempt + 1)
else:
return base_delay
def chat_completion(
self,
messages: List[Dict[str, str]],
model: str = "deepseek-v3.2",
temperature: float = 0.7,
max_tokens: int = 2000,
stream: bool = False
) -> APIResponse:
"""
Führt Chat-Completion mit robustem Timeout-Handling durch
"""
start_time = time.time()
last_error = None
endpoint = f"{self.base_url}/chat/completions"
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens,
"stream": stream
}
for attempt in range(self.max_retries + 1):
try:
response = self.session.post(
endpoint,
json=payload,
timeout=self.timeout
)
response.raise_for_status()
data = response.json()
latency_ms = (time.time() - start_time) * 1000
return APIResponse(
content=data["choices"][0]["message"]["content"],
model=data["model"],
tokens_used=data.get("usage", {}).get("total_tokens", 0),
latency_ms=latency_ms,
finish_reason=data["choices"][0].get("finish_reason", "unknown")
)
except requests.exceptions.Timeout as e:
last_error = f"Timeout nach {self.timeout}s bei Versuch {attempt + 1}"
logger.warning(last_error)
except requests.exceptions.ConnectionError as e:
last_error = f"Verbindungsfehler bei Versuch {attempt + 1}: {str(e)}"
logger.warning(last_error)
except requests.exceptions.HTTPError as e:
if response.status_code == 429:
last_error = "Rate-Limit erreicht"
retry_after = int(response.headers.get("Retry-After", 60))
logger.info(f"Rate-Limit: Warte {retry_after}s")
time.sleep(retry_after)
else:
last_error = f"HTTP-Fehler {response.status_code}"
logger.error(last_error)
break
if attempt < self.max_retries:
delay = self._calculate_retry_delay(attempt)
logger.info(f"Retry in {delay}s...")
time.sleep(delay)
raise TimeoutError(f"API-Anfrage fehlgeschlagen nach {self.max_retries + 1} Versuchen: {last_error}")
def calculate_cost(self, tokens: int, model: str) -> float:
"""Berechnet Kosten in Dollar (Cent-genau)"""
price_per_mtok = self.pricing.get(model, 800)
return (tokens / 1_000_000) * price_per_mtok / 100
Beispiel-Nutzung
if __name__ == "__main__":
client = HolySheepAIClient(
api_key="YOUR_HOLYSHEEP_API_KEY",
timeout=30.0,
max_retries=3
)
messages = [
{"role": "system", "content": "Du bist ein hilfreicher Assistent."},
{"role": "user", "content": "Erkläre API-Timeout-Handling in 3 Sätzen."}
]
try:
response = client.chat_completion(
messages=messages,
model="deepseek-v3.2",
temperature=0.7
)
logger.info(f"Antwort: {response.content}")
logger.info(f"Latenz: {response.latency_ms:.2f}ms")
logger.info(f"Kosten: ${client.calculate_cost(response.tokens_used, response.model):.4f}")
except TimeoutError as e:
logger.error(f"Anfrage fehlgeschlagen: {e}")Performance-Optimierung: Best Practices aus der Praxis
Meine Erfahrung als technischer Lead
Bei der Arbeit mit Enterprise-Kunden habe ich festgestellt, dass 80% der Timeout-Probleme durch suboptimale Prompt-Strukturierung und fehlendes Caching entstehen. Ein Kunde aus München konnte seine P95-Latenz von 1.200ms auf 180ms reduzieren, indem er folgende drei Strategien implementierte:
- Streaming-Responses: Statt auf vollständige Antworten zu warten, werden Chunks gestreamt
- Semantisches Caching: Ähnliche Anfragen werden intelligent erkannt und gecached
- Model-Fallback: Automatische Abstufung bei Lastspitzen
Streaming-Implementierung für reduzierte Wartezeit
import json
import sseclient
import requests
from typing import Iterator, Dict, Any
class StreamingAI Client:
"""
Streaming-fähiger Client für Echtzeit-AI-Antworten
Reduziert wahrgenommene Latenz um 60-70%
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
def stream_chat_completion(
self,
messages: list,
model: str = "deepseek-v3.2"
) -> Iterator[str]:
"""
Führt Streaming-Chat-Completion durch
Vorteile:
- Erste Token nach ~50ms (vs. 200ms+ bei Batch)
- Reduzierte wahrgenommene Latenz
- Kontinuierlicher Feedback-Loop
"""
endpoint = f"{self.base_url}/chat/completions"
payload = {
"model": model,
"messages": messages,
"stream": True,
"max_tokens": 2000
}
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
try:
response = requests.post(
endpoint,
json=payload,
headers=headers,
stream=True,
timeout=60.0
)
response.raise_for_status()
# SSE-Stream parsen
client = sseclient.SSEClient(response)
for event in client.events():
if event.data:
try:
data = json.loads(event.data)
if "choices" in data:
delta = data["choices"][0].get("delta", {})
if "content" in delta:
yield delta["content"]
# Handle done signal
if data.get("choices", [{}])[0].get("finish_reason"):
break
except json.JSONDecodeError:
continue
except requests.exceptions.Timeout:
yield "[Timeout: Server antwortet nicht innerhalb 60s]"
except Exception as e:
yield f"[Fehler: {str(e)}]"
Nutzung für Chat-Interface
def chat_loop():
"""Beispiel für interaktives Chat-Interface mit Streaming"""
client = StreamingAIClient(api_key="YOUR_HOLYSHEEP_API_KEY")
messages = [
{"role": "system", "content": "Du bist ein kreativer Assistent."}
]
while True:
user_input = input("\nSie: ")
if user_input.lower() in ["exit", "quit"]:
break
messages.append({"role": "user", "content": user_input})
print("\nAI: ", end="", flush=True)
full_response = ""
for chunk in client.stream_chat_completion(messages):
print(chunk, end="", flush=True)
full_response += chunk
print("\n")
messages.append({"role": "assistant", "content": full_response})
Benchmark-Funktion zum Vergleichen von Latenzen
def benchmark_latency():
"""Vergleicht Latenzen verschiedener Modelle"""
client = HolySheepAIClient(api_key="YOUR_HOLYSHEEP_API_KEY")
test_prompts = [
{"role": "user", "content": "Zähle die Zahlen 1-10 auf."},
{"role": "user", "content": "Erkläre Quantencomputing in einem Satz."},
]
models = ["deepseek-v3.2", "gemini-2.5-flash", "gpt-4.1"]
results = []
for model in models:
latencies = []
for _ in range(5):
response = client.chat_completion(
messages=test_prompts,
model=model,
max_tokens=100
)
latencies.append(response.latency_ms)
avg_latency = sum(latencies) / len(latencies)
results.append({
"model": model,
"avg_latency_ms": avg_latency,
"min_latency_ms": min(latencies),
"max_latency_ms": max(latencies)
})
print(f"{model}: {avg_latency:.2f}ms (avg), "
f"{min(latencies):.2f}ms (min), "
f"{max(latencies):.2f}ms (max)")
return results
if __name__ == "__main__":
benchmark_latency()Preisvergleich und Kostenoptimierung
HolySheep AI Preise 2026 (pro Million Tokens)
| Modell | Preis/MTok | Latenz (P50) | Bestes Einsatzgebiet |
|---|---|---|---|
| DeepSeek V3.2 | $0.42 (42 Cent) | ~45ms | Kosteneffiziente Standardanfragen |
| Gemini 2.5 Flash | $2.50 (250 Cent) | ~35ms | Schnelle Inferenz, hohe Last |
| GPT-4.1 | $8.00 (800 Cent) | ~180ms | Komplexe Reasoning-Aufgaben |
| Claude Sonnet 4.5 | $15.00 (1500 Cent) | ~120ms | Nuancenreiche Texte, Analyse |
Wichtiger Vorteil: Bei HolySheep AI profitieren Sie von WeChat- und Alipay-Zahlungen zum Kurs ¥1 = $1, was für chinesische Teams besonders attraktiv ist. Zusätzlich erhalten Neukunden kostenlose Credits für den Start.
Kostenrechner für Enterprise-Nutzung
def calculate_monthly_savings(
monthly_tokens: int,
current_provider_rate: float = 15.0, # $/MTok
current_latency_ms: float = 420.0,
target_model: str = "deepseek-v3.2"
) -> Dict[str, Any]:
"""
Berechnet potenzielle Ersparnis bei Migration zu HolySheep AI
Annahmen:
- DeepSeek V3.2 für 80% der Anfragen
- Gemini 2.5 Flash für 15% der Anfragen
- GPT-4.1 für 5% der Anfragen
"""
holy_sheep_pricing = {
"deepseek-v3.2": 0.42,
"gemini-2.5-flash": 2.50,
"gpt-4.1": 8.00,
"claude-sonnet-4.5": 15.00
}
# Traffic-Aufteilung
token_distribution = {
"deepseek-v3.2": 0.80,
"gemini-2.5-flash": 0.15,
"gpt-4.1": 0.05
}
# Berechne aktuelle Kosten
current_cost = (monthly_tokens / 1_000_000) * current_provider_rate
# Berechne HolySheep-Kosten
holy_sheep_cost = 0
holy_sheep_latency = 0
for model, ratio in token_distribution.items():
model_tokens = monthly_tokens * ratio
model_cost = (model_tokens / 1_000_000) * holy_sheep_pricing[model]
holy_sheep_cost += model_cost
# Gewichtete Latenz
model_latency = {"deepseek-v3.2": 45, "gemini-2.5-flash": 35, "gpt-4.1": 180}[model]
holy_sheep_latency += model_latency * ratio
# Ersparnis berechnen
absolute_savings = current_cost - holy_sheep_cost
percentage_savings = (absolute_savings / current_cost) * 100 if current_cost > 0 else 0
return {
"current_cost_monthly": round(current_cost, 2),
"holy_sheep_cost_monthly": round(holy_sheep_cost, 2),
"absolute_savings_monthly": round(absolute_savings, 2),
"percentage_savings": round(percentage_savings, 1),
"current_latency_ms": current_latency_ms,
"holy_sheep_latency_ms": round(holy_sheep_latency, 1),
"latency_improvement_pct": round(
((current_latency_ms - holy_sheep_latency) / current_latency_ms) * 100, 1
),
"annual_savings": round(absolute_savings * 12, 2)
}
Beispiel-Berechnung für 10M Token/Monat
if __name__ == "__main__":
result = calculate_monthly_savings(
monthly_tokens=10_000_000,
current_provider_rate=15.0,
current_latency_ms=420.0
)
print("=" * 50)
print("KOSTENANALYSE: Migration zu HolySheep AI")
print("=" * 50)
print(f"Monatliches Token-Volumen: 10.000.000")
print(f"Aktuelle monatliche Kosten: ${result['current_cost_monthly']}")
print(f"HolySheep monatliche Kosten: ${result['holy_sheep_cost_monthly']}")
print(f"MONATLICHE ERSPARNIS: ${result['absolute_savings_monthly']} ({result['percentage_savings']}%)")
print(f"JAHRESERSPARNIS: ${result['annual_savings']}")
print("-" * 50)
print(f"Aktuelle Latenz: {result['current_latency_ms']}ms")
print(f"HolySheep Latenz: {result['holy_sheep_latency_ms']}ms")
print(f"Latenzverbesserung: {result['latency_improvement_pct']}%")
print("=" * 50)Häufige Fehler und Lösungen
Fehler 1: Falsches Timeout-Handling führt zu Endlosschleifen
Symptom: Client friert ein, keine Fehlermeldung, Prozess muss manuell beendet werden.
# ❌ FEHLERHAFT: Keine Timeout-Konfiguration
import openai
client = openai.OpenAI(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1"
# KEIN TIMEOUT GESETZT!
)
response = client.chat.completions.create(
model="deepseek-v3.2",
messages=[{"role": "user", "content": "Hallo"}]
)
✅ LÖSUNG: Explizites Timeout mit TimeoutError-Handling
from openai import Timeout
import openai
client = openai.OpenAI(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1",
timeout=30.0, # Explizites Timeout in Sekunden
max_retries=3 # Automatische Retries
)
try:
response = client.chat.completions.create(
model="deepseek-v3.2",
messages=[{"role": "user", "content": "Hallo"}],
timeout=30.0 # Request-spezifisches Timeout
)
except Timeout:
print("Anfrage hat das Timeout überschritten. Bitte versuchen Sie es erneut.")
except Exception as e:
print(f"Unerwarteter Fehler: {type(e).__name__}: {e}")Fehler 2: Rate-Limit wird ignoriert, führt zu 429-Fehlern
Symptom: Sporadische 429-Antworten, inkonsistente Latenzen, plötzliche Fehlerspitzen.
# ❌ FEHLERHAFT: Keine Rate-Limit-Handhabung
import openai
client = openai.OpenAI(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1"
)
Schleife ohne Rate-Limit-Prüfung
for i in range(1000):
response = client.chat.completions.create(
model="deepseek-v3.2",
messages=[{"role": "user", "content": f"Query {i}"}]
)
✅ LÖSUNG: Rate-Limiter mit exponentieller Backoff
import time
import threading
from collections import deque
from openai import RateLimitError
class RateLimiter:
"""Token Bucket Rate Limiter für API-Anfragen"""
def __init__(self, requests_per_minute: int = 60):
self.requests_per_minute = requests_per_minute
self.request_times = deque(maxlen=requests_per_minute)
self.lock = threading.Lock()
def acquire(self) -> None:
"""Blockiert bis ein Slot verfügbar ist"""
with self.lock:
now = time.time()
# Entferne alte Timestamps
while self.request_times and self.request_times[0] < now - 60:
self.request_times.popleft()
# Prüfe Limit
if len(self.request_times) >= self.requests_per_minute:
sleep_time = 60 - (now - self.request_times[0])
if sleep_time > 0:
time.sleep(sleep_time)
self.request_times.append(time.time())
class HolySheepClient:
def __init__(self, api_key: str):
self.client = openai.OpenAI(
api_key=api_key,
base_url="https://api.holysheep.ai/v1"
)
self.rate_limiter = RateLimiter(requests_per_minute=60)
def chat(self, messages: list, max_retries: int = 3) -> str:
"""Chat mit automatischem Retry bei Rate-Limits"""
for attempt in range(max_retries):
try:
self.rate_limiter.acquire()
response = self.client.chat.completions.create(
model="deepseek-v3.2",
messages=messages,
timeout=30.0
)
return response.choices[0].message.content
except RateLimitError as e:
if attempt < max_retries - 1:
# Exponentieller Backoff: 2, 4, 8, 16...
wait_time = 2 ** attempt
print(f"Rate-Limit erreicht. Warte {wait_time}s...")
time.sleep(wait_time)
else:
raise
return ""
Nutzung
client = HolySheepClient(api_key="YOUR_HOLYSHEEP_API_KEY")
results = []
for i in range(100):
result = client.chat([{"role": "user", "content": f"Query {i}"}])
results.append(result)Fehler 3: Fehlende Fehlerbehandlung bei Netzwerk-Partitionen
Symptom: Sporadische ConnectionError-Ausnahmen, keine automatische Wiederherstellung, Datenverlust.
# ❌ FEHLERHAFT: Generische Exception ohne spezifische Behandlung
import openai
client = openai.OpenAI(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1"
)
try:
response = client.chat.completions.create(
model="deepseek-v3.2",
messages=[{"role": "user", "content": "Test"}]
)
except Exception as e: # Zu generisch!
print(f"Fehler: {e}")
# Keine spezifische Behandlung
✅ LÖSUNG: Granulare Fehlerbehandlung mit Circuit Breaker
import time
from enum import Enum
from dataclasses import dataclass
from typing import Optional
import requests
class CircuitState(Enum):
CLOSED = "closed" # Normal, Requests durchlassen
OPEN = "open" # Failures, Requests blockieren
HALF_OPEN = "half_open" # Test nach Wartezeit
@dataclass
class CircuitBreaker:
failure_threshold: int = 5
recovery_timeout: float = 60.0
success_threshold: int = 2
state: CircuitState = CircuitState.CLOSED
failure_count: int = 0
success_count: int = 0
last_failure_time: Optional[float] = None
def record_success(self):
self.failure_count = 0
if self.state == CircuitState.HALF_OPEN:
self.success_count += 1
if self.success_count >= self.success_threshold:
self.state = CircuitState.CLOSED
self.success_count = 0
def record_failure(self):
self.failure_count += 1
self.last_failure_time = time.time()
if self.state == CircuitState.HALF_OPEN:
self.state = CircuitState.OPEN
elif self.failure_count >= self.failure_threshold:
self.state = CircuitState.OPEN
def can_attempt(self) -> bool:
if self.state == CircuitState.CLOSED:
return True
if self.state == CircuitState.OPEN:
if time.time() - self.last_failure_time >= self.recovery_timeout:
self.state = CircuitState.HALF_OPEN
return True
return False
return True # HALF_OPEN
class ResilientHolySheepClient:
def __init__(self, api_key: str):
self.client = openai.OpenAI(
api_key=api_key,
base_url="https://api.holysheep.ai/v1",
timeout=30.0
)
self.circuit_breaker = CircuitBreaker(
failure_threshold=5,
recovery_timeout=60.0
)
def chat(self, messages: list) -> str:
if not self.circuit_breaker.can_attempt():
raise ConnectionError(
"Circuit Breaker ist OPEN. Service vorübergehend nicht verfügbar."
)
try:
response = self.client.chat.completions.create(
model="deepseek-v3.2",
messages=messages
)
self.circuit_breaker.record_success()
return response.choices[0].message.content
except requests.exceptions.ConnectionError as e:
self.circuit_breaker.record_failure()
raise ConnectionError(f"Netzwerkfehler: {e}. Circuit: {self.c