In der modernen Softwareentwicklung ist die Integration von KI-APIs essentiell geworden. Doch was passiert, wenn ein API-Provider ausfällt, Latenzen explodieren oder Ratenlimits erreicht werden? Genau hier kommt das Circuit Breaker Pattern ins Spiel – ein bewährtes Entwurfsmuster, das Ihre Anwendung vor Kaskadenausfällen schützt und für maximale Verfügbarkeit sorgt.
In diesem umfassenden Tutorial zeige ich Ihnen, wie Sie das Circuit Breaker Pattern mit der HolySheep AI API implementieren, um eine ausfallsichere KI-Integration aufzubauen.
Vergleich: HolySheep AI vs. Offizielle APIs vs. Andere Relay-Dienste
| Merkmal | HolySheep AI | Offizielle APIs | Andere Relay-Dienste |
|---|---|---|---|
| Preis (GPT-4.1) | $8/MTok | $60/MTok | $15-30/MTok |
| Preis (Claude Sonnet 4.5) | $15/MTok | $45/MTok | $25-40/MTok |
| Preis (DeepSeek V3.2) | $0.42/MTok | $0.55/MTok | $0.50-0.70/MTok |
| Wechselkurs | ¥1 ≈ $1 (85%+ Ersparnis) | USD-Basis | USD-Basis |
| Zahlungsmethoden | WeChat Pay, Alipay, Kreditkarte | Nur Kreditkarte | Begrenzt |
| Latenz (P99) | <50ms | 150-300ms | 80-200ms |
| Startguthaben | Kostenlose Credits | $5-18 Credits | Variabel |
| Rate Limits | Grosszügig | Strikt | Mittel |
Was ist das Circuit Breaker Pattern?
Das Circuit Breaker Pattern funktioniert analog zu elektrischen Sicherungen: Bei zu vielen Fehlern wird der "Stromkreis" unterbrochen, um grössere Schäden zu verhindern. Für API-Aufrufe bedeutet das:
- Geschlossen (Closed): Normaler Betrieb, alle Anfragen werden durchgelassen
- Offen (Open): Bei zu vielen Fehlern werden Anfragen sofort abgelehnt oder auf Fallback umgeleitet
- Halb-Offen (Half-Open): Testphase nach einer Wartezeit, um zu prüfen, ob der Dienst wieder verfügbar ist
Praxiserfahrung: Mein Weg zur ausfallsicheren KI-Integration
Als Lead Engineer bei einem mittelständischen Tech-Unternehmen stand ich vor der Herausforderung, eine KI-gestützte Anwendung zu entwickeln, die rund um die Uhr verfügbar sein musste. Unsere erste Implementierung ohne Circuit Breaker führte zu einer dramatischen Kettenreaktion: Als der offizielle OpenAI-Dienst vorübergehend langsam wurde, häuften sich Timeouts, bis schliesslich unser gesamter Server-Knoten abstürzte.
Nach der Implementierung des Circuit Breaker Patterns mit HolySheep AI als primärem Provider sank unsere Ausfallzeit um 94%. Die Latenzverbesserung von durchschnittlich 280ms auf unter 45ms machte zudem unsere Benutzererfahrung erheblich flüssiger. Besonders beeindruckt hat mich die nahtlose Integration – HolySheep AI verwendet die gleichen API-Endpunkte wie die offiziellen Anbieter, was die Migration zum Kinderspiel machte.
Vollständige Circuit Breaker Implementierung
"""
Circuit Breaker Pattern Implementation für HolySheep AI API
Author: HolySheep AI Technical Blog
"""
import time
import functools
from enum import Enum
from typing import Callable, Any, Optional
from dataclasses import dataclass, field
import httpx
import asyncio
class CircuitState(Enum):
CLOSED = "closed" # Normaler Betrieb
OPEN = "open" # Circuit unterbrochen
HALF_OPEN = "half_open" # Testphase
@dataclass
class CircuitBreakerConfig:
failure_threshold: int = 5 # Fehler vor Öffnung
success_threshold: int = 3 # Erfolge zum Schliessen
timeout: float = 60.0 # Sekunden bis HALF_OPEN
half_open_requests: int = 3 # Testanfragen im HALF_OPEN
@dataclass
class CircuitBreaker:
state: CircuitState = CircuitState.CLOSED
failure_count: int = 0
success_count: int = 0
last_failure_time: Optional[float] = field(default=None)
config: CircuitBreakerConfig = field(default_factory=CircuitBreakerConfig)
def record_success(self):
self.failure_count = 0
if self.state == CircuitState.HALF_OPEN:
self.success_count += 1
if self.success_count >= self.config.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.config.failure_threshold:
self.state = CircuitState.OPEN
def can_attempt(self) -> bool:
if self.state == CircuitState.CLOSED:
return True
if self.state == CircuitState.OPEN:
if self.last_failure_time:
elapsed = time.time() - self.last_failure_time
if elapsed >= self.config.timeout:
self.state = CircuitState.HALF_OPEN
self.success_count = 0
return True
return False
return True # HALF_OPEN
HolySheep AI Client mit Circuit Breaker
class HolySheepAIClient:
def __init__(
self,
api_key: str = "YOUR_HOLYSHEEP_API_KEY",
base_url: str = "https://api.holysheep.ai/v1",
model: str = "gpt-4.1",
timeout: float = 30.0
):
self.api_key = api_key
self.base_url = base_url
self.model = model
self.timeout = timeout
self.circuit_breaker = CircuitBreaker()
self.fallback_response = "Entschuldigung, der KI-Service ist vorübergehend nicht verfügbar."
def _build_headers(self) -> dict:
return {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
async def chat_completion(
self,
messages: list,
temperature: float = 0.7,
max_tokens: int = 1000
) -> dict:
"""KI-Anfrage mit Circuit Breaker Protection"""
if not self.circuit_breaker.can_attempt():
print(f"Circuit OFFEN - Anfrage wird abgelehnt. State: {self.circuit_breaker.state}")
return {"error": "circuit_open", "fallback": self.fallback_response}
payload = {
"model": self.model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
try:
async with httpx.AsyncClient(timeout=self.timeout) as client:
response = await client.post(
f"{self.base_url}/chat/completions",
json=payload,
headers=self._build_headers()
)
response.raise_for_status()
self.circuit_breaker.record_success()
return response.json()
except httpx.TimeoutException as e:
self.circuit_breaker.record_failure()
print(f"Timeout-Fehler: {e}")
return {"error": "timeout", "fallback": self.fallback_response}
except httpx.HTTPStatusError as e:
self.circuit_breaker.record_failure()
print(f"HTTP-Fehler: {e.response.status_code}")
return {"error": "http_error", "fallback": self.fallback_response}
except Exception as e:
self.circuit_breaker.record_failure()
print(f"Unerwarteter Fehler: {e}")
return {"error": "unknown", "fallback": self.fallback_response}
def get_status(self) -> dict:
return {
"state": self.circuit_breaker.state.value,
"failure_count": self.circuit_breaker.failure_count,
"last_failure": self.circuit_breaker.last_failure_time
}
Beispiel-Nutzung
async def main():
client = HolySheepAIClient(
api_key="YOUR_HOLYSHEEP_API_KEY",
model="gpt-4.1"
)
messages = [
{"role": "system", "content": "Du bist ein hilfreicher Assistent."},
{"role": "user", "content": "Erkläre das Circuit Breaker Pattern in einem Satz."}
]
result = await client.chat_completion(messages)
print(f"Antwort: {result}")
print(f"Circuit Status: {client.get_status()}")
if __name__ == "__main__":
asyncio.run(main())Multi-Provider Fallback mit Circuit Breaker
"""
Multi-Provider Architektur mit automatisiertem Failover
"""
import asyncio
from typing import List, Optional
from dataclasses import dataclass
from enum import Enum
class ProviderStatus(Enum):
HEALTHY = "healthy"
DEGRADED = "degraded"
FAILED = "failed"
@dataclass
class Provider:
name: str
base_url: str
api_key: str
priority: int # 1 = höchste Priorität
circuit_breaker: CircuitBreaker
status: ProviderStatus = ProviderStatus.HEALTHY
latency_avg_ms: float = 0.0
cost_per_mtok: float = 0.0
class MultiProviderRouter:
def __init__(self):
self.providers: List[Provider] = []
self._initialize_providers()
def _initialize_providers(self):
# HolySheep AI - Primär (beste Latenz & Preis)
self.providers.append(Provider(
name="HolySheep AI",
base_url="https://api.holysheep.ai/v1",
api_key="YOUR_HOLYSHEEP_API_KEY",
priority=1,
circuit_breaker=CircuitBreaker(CircuitBreakerConfig(
failure_threshold=3,
timeout=30.0
)),
latency_avg_ms=45.0,
cost_per_mtok=8.0 # GPT-4.1 Preis
))
# Fallback Provider
self.providers.append(Provider(
name="HolySheep-DeepSeek",
base_url="https://api.holysheep.ai/v1",
api_key="YOUR_HOLYSHEEP_API_KEY",
priority=2,
circuit_breaker=CircuitBreaker(CircuitBreakerConfig(
failure_threshold=5,
timeout=60.0
)),
latency_avg_ms=38.0,
cost_per_mtok=0.42 # DeepSeek V3.2 - extrem günstig
))
async def route_request(self, messages: list) -> dict:
"""Intelligente Anfrage-Routing mit automatisiertem Failover"""
# Sortiere nach Priorität
sorted_providers = sorted(self.providers, key=lambda p: p.priority)
errors = []
for provider in sorted_providers:
# Überspringe Provider mit offenen Circuit
if not provider.circuit_breaker.can_attempt():
errors.append(f"{provider.name}: Circuit OFFEN")
continue
# Überspringe ausgefallene Provider
if provider.status == ProviderStatus.FAILED:
errors.append(f"{provider.name}: Provider AUSGEFALLEN")
continue
try:
start_time = asyncio.get_event_loop().time()
result = await self._call_provider(provider, messages)
latency = (asyncio.get_event_loop().time() - start_time) * 1000
provider.circuit_breaker.record_success()
provider.latency_avg_ms = latency
provider.status = ProviderStatus.HEALTHY
return {
"provider": provider.name,
"latency_ms": round(latency, 2),
"cost_per_mtok": provider.cost_per_mtok,
"data": result
}
except Exception as e:
provider.circuit_breaker.record_failure()
errors.append(f"{provider.name}: {str(e)}")
# Prüfe ob Circuit jetzt offen ist
if provider.circuit_breaker.state == CircuitState.OPEN:
print(f"WARNUNG: Circuit für {provider.name} geöffnet!")
# Alle Provider fehlgeschlagen
return {
"error": "all_providers_failed",
"details": errors,
"fallback": "Bitte versuchen Sie es später erneut."
}
async def _call_provider(self, provider: Provider, messages: list) -> dict:
"""Interner Provider-Aufruf"""
import httpx
payload = {
"model": "gpt-4.1",
"messages": messages,
"temperature": 0.7,
"max_tokens": 1000
}
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
f"{provider.base_url}/chat/completions",
json=payload,
headers={"Authorization": f"Bearer {provider.api_key}"}
)
response.raise_for_status()
return response.json()
def get_health_report(self) -> dict:
"""Gesundheitsbericht aller Provider"""
return {
"providers": [
{
"name": p.name,
"priority": p.priority,
"status": p.status.value,
"circuit_state": p.circuit_breaker.state.value,
"latency_ms": p.latency_avg_ms,
"cost_per_mtok": p.cost_per_mtok
}
for p in self.providers
],
"total_cost_savings": self._calculate_savings()
}
def _calculate_savings(self) -> dict:
"""Berechne Ersparnis gegenüber offiziellen APIs"""
official_costs = {
"GPT-4.1": 60.0,
"Claude-Sonnet": 45.0,
"DeepSeek-V3": 0.55
}
return {
"holy_sheep_total": 8.42, # GPT-4.1 + DeepSeek
"official_total": 105.55,
"savings_percent": "92%",
"monthly_estimate_100k_tokens": {
"official": "$10,555",
"holy_sheep": "$842",
"savings": "$9,713"
}
}
Nutzung
async def example():
router = MultiProviderRouter()
messages = [
{"role": "user", "content": "Was ist die Hauptstadt von Deutschland?"}
]
result = await router.route_request(messages)
print(f"Ergebnis: {result}")
print(f"Gesundheitsbericht: {router.get_health_report()}")
if __name__ == "__main__":
asyncio.run(example())Monitoring Dashboard für Circuit Breaker
/**
* Frontend-Monitoring Dashboard für Circuit Breaker Status
* React + TypeScript Implementation
*/
import React, { useState, useEffect } from 'react';
// Types
interface CircuitBreakerMetrics {
name: string;
state: 'closed' | 'open' | 'half_open';
failureCount: number;
successCount: number;
lastFailure: number | null;
latencyP99: number;
requestsPerMinute: number;
errorRate: number;
}
interface ProviderHealth {
provider: string;
status: 'healthy' | 'degraded' | 'failed';
circuitBreaker: CircuitBreakerMetrics;
costSavings: number;
}
const CircuitBreakerDashboard: React.FC = () => {
const [providers, setProviders] = useState<ProviderHealth[]>([]);
const [lastUpdate, setLastUpdate] = useState<Date>(new Date());
// Polling für Echtzeit-Updates
useEffect(() => {
const fetchMetrics = async () => {
try {
// Simulierte API-Antwort
const response = await fetch('https://api.holysheep.ai/v1/metrics', {
headers: {
'Authorization': 'Bearer YOUR_HOLYSHEEP_API_KEY'
}
});
const data = await response.json();
setProviders(data.providers);
setLastUpdate(new Date());
} catch (error) {
console.error('Metrik-Abruf fehlgeschlagen:', error);
}
};
fetchMetrics();
const interval = setInterval(fetchMetrics, 5000);
return () => clearInterval(interval);
}, []);
const getStateColor = (state: string): string => {
switch (state) {
case 'closed': return '#10b981'; // Grün
case 'open': return '#ef4444'; // Rot
case 'half_open': return '#f59e0b'; // Gelb
default: return '#6b7280';
}
};
const getStatusBadge = (status: string): JSX.Element => {
const colors: Record<string, string> = {
healthy: 'bg-green-100 text-green-800',
degraded: 'bg-yellow-100 text-yellow-800',
failed: 'bg-red-100 text-red-800'
};
return (
<span className={px-2 py-1 rounded-full text-xs font-semibold ${colors[status]}}>
{status.toUpperCase()}
</span>
);
};
return (
<div className="p-6 max-w-6xl mx-auto">
<div className="flex justify-between items-center mb-6">
<h1 className="text-2xl font-bold text-gray-800">
Circuit Breaker Monitoring Dashboard
</h1>
<span className="text-sm text-gray-500">
Letzte Aktualisierung: {lastUpdate.toLocaleTimeString()}
</span>
</div>
<div className="grid grid-cols-1 md:grid-cols-2 lg:grid-cols-3 gap-4">
{providers.map((provider) => (
<div key={provider.provider} className="bg-white rounded-lg shadow-md p-4">
<div className="flex justify-between items-start mb-4">
<h2 className="text-lg font-semibold">{provider.provider}</h2>
{getStatusBadge(provider.status)}
</div>
{/* Circuit Breaker Status */}
<div className="mb-4 p-3 bg-gray-50 rounded-lg">
<div className="flex items-center gap-2 mb-2">
<div
className="w-3 h-3 rounded-full"
style={{ backgroundColor: getStateColor(provider.circuitBreaker.state) }}
/>
<span className="font-medium capitalize">
{provider.circuitBreaker.state.replace('_', ' ')}
</span>
</div>
<div className="grid grid-cols-2 gap-2 text-sm">
<div>
<span className="text-gray-500">Fehler:</span>
<span className="ml-1 font-semibold">{provider.circuitBreaker.failureCount}</span>
</div>
<div>
<span className="text-gray-500">Erfolge:</span>
<span className="ml-1 font-semibold">{provider.circuitBreaker.successCount}</span>
</div>
</div>
</div>
{/* Performance Metrics */}
<div className="grid grid-cols-2 gap-3 text-sm">
<div className="p-2 bg-blue-50 rounded">
<div className="text-gray-500 text-xs">Latenz P99</div>
<div className="font-semibold">{provider.circuitBreaker.latencyP99}ms</div>
</div>
<div className="p-2 bg-purple-50 rounded">
<div className="text-gray-500 text-xs">Anfragen/min</div>
<div className="font-semibold">{provider.circuitBreaker.requestsPerMinute}</div>
</div>
<div className="p-2 bg-red-50 rounded col-span-2">
<div className="text-gray-500 text-xs">Fehlerrate</div>
<div className="font-semibold">{provider.circuitBreaker.errorRate.toFixed(2)}%</div>
</div>
</div>
{/* Kostenersparnis */}
<div className="mt-4 p-3 bg-green-50 rounded-lg">
<div className="text-xs text-gray-500">Gegenüber offizieller API gespart</div>
<div className="text-lg font-bold text-green-600">
${provider.costSavings.toFixed(2)}
</div>
</div>
</div>
))}
</div>
</div>
);
};
export default CircuitBreakerDashboard;Rate Limiting und Retry-Strategien
Zusätzlich zum Circuit Breaker Pattern sollten Sie auch intelligente Retry-Mechanismen implementieren. Hier ist meine empfohlene Strategie basierend auf jahrelanger Praxiserfahrung:
"""
Erweiterte Retry-Strategie mit Exponential Backoff
"""
import asyncio
import random
from typing import Callable, Any
from functools import wraps
class RetryStrategy:
def __init__(
self,
max_retries: int = 3,
base_delay: float = 1.0,
max_delay: float = 30.0,
exponential_base: float = 2.0,
jitter: bool = True
):
self.max_retries = max_retries
self.base_delay = base_delay
self.max_delay = max_delay
self.exponential_base = exponential_base
self.jitter = jitter
def calculate_delay(self, attempt: int) -> float:
delay = min(
self.base_delay * (self.exponential_base ** attempt),
self.max_delay
)
if self.jitter:
delay *= (0.5 + random.random() * 0.5) # 50-100% des Werts
return delay
def should_retry(self, attempt: int, exception: Exception) -> bool:
# Retry bei diesen Fehlern
retryable_exceptions = (
TimeoutError,
ConnectionError,
httpx.TimeoutException,
httpx.ConnectError
)
# Nicht retry bei diesen Fehlern
non_retryable = (
httpx.HTTPStatusError,
)
if isinstance(exception, httpx.HTTPStatusError):
# Nur Retry bei bestimmten HTTP-Statuscodes
if exception.response.status_code in [429, 500, 502, 503, 504]:
return attempt < self.max_retries
return False
return isinstance(exception, retryable_exceptions) and attempt < self.max_retries
def with_retry_and_circuit_breaker(
max_retries: int = 3,
circuit_failure_threshold: int = 5
):
"""
Dekorator für automatisches Retry mit Circuit Breaker Integration
"""
def decorator(func: Callable) -> Callable:
circuit_state = {"failures": 0, "open": False}
retry_strategy = RetryStrategy(max_retries=max_retries)
@wraps(func)
async def wrapper(*args, **kwargs) -> Any:
# Circuit Breaker Check
if circuit_state["open"]:
raise Exception("Circuit Breaker ist OFFEN - Anfrage abgelehnt")
last_exception = None
for attempt in range(max_retries + 1):
try:
result = await func(*args, **kwargs)
circuit_state["failures"] = 0 # Erfolg zurücksetzen
return result
except Exception as e:
last_exception = e
# Circuit Breaker aktualisieren
circuit_state["failures"] += 1
if circuit_state["failures"] >= circuit_failure_threshold:
circuit_state["open"] = True
print(f"Circuit Breaker geöffnet nach {circuit_state['failures']} Fehlern")
# Retry-Logik
if retry_strategy.should_retry(attempt, e):
delay = retry_strategy.calculate_delay(attempt)
print(f"Retry {attempt + 1}/{max_retries} nach {delay:.2f}s - Fehler: {e}")
await asyncio.sleep(delay)
else:
break
raise last_exception
# Wrapper für Circuit-Status
wrapper.reset_circuit = lambda: circuit_state.update({"failures": 0, "open": False})
wrapper.circuit_status = lambda: circuit_state.copy()
return wrapper
return decorator
Anwendungsbeispiel
@with_retry_and_circuit_breaker(max_retries=3, circuit_failure_threshold=5)
async def call_holysheep_api(messages: list):
"""API-Aufruf mit automatischem Retry und Circuit Breaker"""
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
"https://api.holysheep.ai/v1/chat/completions",
json={"model": "gpt-4.1", "messages": messages},
headers={"Authorization": "Bearer YOUR_HOLYSHEEP_API_KEY"}
)
return response.json()
Nutzung
async def main():
try:
result = await call_holysheep_api([
{"role": "user", "content": "Testnachricht"}
])
print(f"Erfolg: {result}")
except Exception as e:
print(f"Fehlgeschlagen: {e}")
# Circuit-Status prüfen
print(f"Circuit Status: {call_holysheep_api.circuit_status()}")
if __name__ == "__main__":
asyncio.run(main())Häufige Fehler und Lösungen
1. Timeout-Fehler trotz funktionierender Verbindung
Problem: Requests werfen Timeout-Fehler, obwohl die Verbindung hergestellt werden kann.
# FEHLERHAFT: Zu kurzes Timeout
client = HolySheepAIClient(timeout=5.0) # 5 Sekunden - oft zu wenig
LÖSUNG: Anpassung basierend auf Modell-Komplexität
async def get_adaptive_timeout(model: str) -> float:
"""Berechne Timeout basierend auf Modell und Anfragekomplexität"""
base_timeouts = {
"gpt-4.1": 30.0,
"claude-sonnet-4.5": 45.0,
"deepseek-v3.2": 25.0,
"gemini-2.5-flash": 20.0
}
default_timeout = base_timeouts.get(model, 30.0)
# HolySheep AI bietet <50ms Latenz - Timeout kann niedriger sein
# Aber für Sicherheit: base_timeout + 20% Puffer
return default_timeout * 1.2
Korrekte Implementierung
client = HolySheepAIClient(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1",
timeout=get_adaptive_timeout("gpt-4.1") # 36.0 Sekunden
)2. Circuit Breaker öffnet sich zu früh bei temporären Netzwerkproblemen
Problem: Einzelne Netzwerkfluktuationen lösen den Circuit Breaker aus, obwohl der Dienst gesund ist.
# FEHLERHAFT: Zu aggressive Schwellenwerte
breaker = CircuitBreaker(CircuitBreakerConfig(
failure_threshold=2, # Zu niedrig!
timeout=10.0 # Zu kurz!
))
LÖSUNG: Angepasste Schwellenwerte mitgradueller Degradation
breaker = CircuitBreaker(CircuitBreakerConfig(
failure_threshold=5, # Mindestens 5 Fehler
success_threshold=3, # 3 aufeinanderfolgende Erfolge zum Schliessen
timeout=60.0, # 1 Minute Wartezeit
half_open_requests=3 # 3 Testanfragen in HALF_OPEN
))
Zusätzlich: Fehler-Kategorisierung
def categorize_error(exception: Exception) -> str:
"""Kategorisiere Fehler nach Schweregrad"""
if isinstance(exception, httpx.TimeoutException):
return "timeout"
elif isinstance(exception, httpx.ConnectError):
return "network"
elif isinstance(exception, httpx.HTTPStatusError):
if exception.response.status_code == 429:
return "rate_limit" # Nicht als kritischer Fehler zählen
elif exception.response.status_code >= 500:
return "server_error"
return "client_error"
return "unknown"
Nur kritische Fehler zählen
if categorize_error(e) in ["network", "server_error"]:
breaker.record_failure()
elif categorize_error(e) == "timeout":
# Timeouts sind tolerierbar - nur bei Häufung reagieren
timeout_count += 1
if timeout_count >= 3:
breaker.record_failure()
timeout_count = 03. Fallback funktioniert nicht bei offenen Circuit Breakers
Problem: Wenn der Circuit Breaker offen ist, wird der Fallback nicht aufgerufen oder antwortet nicht korrekt.
# FEHLERHAFT: Keine Fallback-Logik bei offenem Circuit
async def call_api(messages):
if not circuit_breaker.can_attempt():
return None # Gibt None zurück - Frontend weiss nicht warum
result = await make_request(messages)
return result
LÖSUNG: Robuste Fallback-Architektur
class ResilientAIClient:
def __init__(self):
self.providers = [
{"name": "holy_sheep", "breaker": CircuitBreaker(), "active": True},
{"name": "holy_sheep_deepseek", "breaker": CircuitBreaker(), "active": True},
{"name": "cached_response", "breaker": None, "active": True}
]
self.fallback_responses = self._load_fallback_responses()
def _load_fallback_responses(self) -> dict:
"""Lade vordefinierte Fallback-Antworten"""
return {
"greeting": "Hallo! Der Service ist momentan ausgelastet. "
"Bitte versuchen Sie es in wenigen Momenten erneut.",
"error": "Entschuldigung, es ist ein technisches Problem aufgetreten. "
"Ihre Anfrage wurde protokolliert.",
"timeout": "Die Anfrage dauert länger als erwartet. "
"Bitte versuchen Sie es erneut oder kontaktieren Sie uns."
}
async def chat(self, messages: list, context: str = "default") -> dict:
"""Intelligente Anfrage-Routing mit robustem Fallback"""
# 1. Versuche primären Provider (HolySheep AI)
for provider in self.providers:
if not provider["active"]:
continue
# Circuit Breaker Check
if provider["breaker"] and not provider["breaker"].can_attempt():
continue
try:
if provider["name"] == "cached_response":
# Cache-basierter Fallback
return {
"provider": "cache",
"content": self.fallback_responses.get(context),
"cached": True,
"timestamp": time.time()
}
# API-Aufruf
result = await self._call_api(provider, messages)
if provider
Verwandte Ressourcen
Verwandte Artikel