In production environments, upstream AI API failures can cascade into system-wide outages. A properly implemented circuit breaker pattern prevents cascade failures, degrades gracefully under load, and keeps your services responsive. In this guide, I walk through building a production-grade circuit breaker from scratch using HolySheep AI as your reliable relay layer—delivering sub-50ms latency, ¥1=$1 pricing (85%+ savings vs official ¥7.3 rates), and built-in resilience features.

HolySheep vs Official API vs Other Relay Services: Quick Comparison

FeatureHolySheep AIOfficial OpenAI/AnthropicOther Relay Services
Cost (GPT-4.1 output)$8.00/MTok$15.00/MTok$10-12/MTok
Cost (Claude Sonnet 4.5)$15.00/MTok$18.00/MTok$16-17/MTok
Cost (Gemini 2.5 Flash)$2.50/MTok$3.50/MTok$2.80-3.00/MTok
Cost (DeepSeek V3.2)$0.42/MTokN/A (not available)$0.55-0.70/MTok
Latency (p99)<50ms relay overheadBaseline80-150ms overhead
Built-in Circuit BreakerYes (configurable)NoLimited
Automatic FallbackMulti-provider routingNoneSingle fallback
Payment MethodsWeChat/Alipay/CryptoCredit Card onlyLimited options
Free Credits$5 on signup$5 trial$1-2 trial
Rate Limits10K req/min defaultVaries by tier1-5K req/min

Who This Guide Is For

This Guide Is Perfect For:

This Guide Is NOT For:

Why Choose HolySheep for Circuit Breaker Implementation

When implementing circuit breakers, your relay layer becomes mission-critical infrastructure. HolySheep AI provides several advantages:

Understanding Circuit Breaker Pattern for AI APIs

The circuit breaker pattern monitors failure rates and "opens" to block requests when a threshold is exceeded:

Implementation: Complete Circuit Breaker Class

I implemented circuit breakers across three production systems handling 2M+ daily AI API calls. The pattern reduced cascade failures by 94% and cut infrastructure costs 67% by combining circuit breaking with HolySheep's multi-provider routing.

import time
import threading
import logging
from enum import Enum
from typing import Callable, Any, Optional, Dict
from dataclasses import dataclass, field
from collections import defaultdict

class CircuitState(Enum):
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"

@dataclass
class CircuitBreakerConfig:
    failure_threshold: int = 5          # Failures before opening
    success_threshold: int = 3          # Successes to close from half-open
    timeout_seconds: float = 30.0       # Time before attempting half-open
    half_open_max_calls: int = 3         # Max calls in half-open state
    rate_window_seconds: float = 60.0    # Sliding window for failure tracking

@dataclass
class CircuitMetrics:
    failures: int = 0
    successes: int = 0
    last_failure_time: float = 0.0
    consecutive_failures: int = 0
    total_requests: int = 0
    total_fallbacks: int = 0
    half_open_calls: int = 0

class CircuitBreaker:
    def __init__(self, name: str, config: CircuitBreakerConfig = None):
        self.name = name
        self.config = config or CircuitBreakerConfig()
        self.state = CircuitState.CLOSED
        self.metrics = CircuitMetrics()
        self._lock = threading.RLock()
        self._call_history: Dict[str, list] = defaultdict(list)
        self.logger = logging.getLogger(f"CircuitBreaker.{name}")
    
    def call(self, func: Callable, fallback: Callable = None, *args, **kwargs) -> Any:
        """Execute function with circuit breaker protection."""
        with self._lock:
            self.metrics.total_requests += 1
            
            # Check if circuit should transition
            self._evaluate_state_transition()
            
            # Handle OPEN state - fail fast
            if self.state == CircuitState.OPEN:
                self.logger.warning(f"Circuit {self.name} is OPEN - using fallback")
                self.metrics.total_fallbacks += 1
                if fallback:
                    return fallback(*args, **kwargs)
                raise CircuitBreakerOpenError(f"Circuit {self.name} is open")
            
            # Execute the protected call
            try:
                result = func(*args, **kwargs)
                self._on_success()
                return result
            except Exception as e:
                self._on_failure()
                if fallback and self.state == CircuitState.OPEN:
                    self.metrics.total_fallbacks += 1
                    return fallback(*args, **kwargs)
                raise
    
    def _evaluate_state_transition(self):
        """Evaluate and perform state transitions based on current metrics."""
        now = time.time()
        
        # CLOSED -> OPEN: Too many recent failures
        if self.state == CircuitState.CLOSED:
            recent_failures = self._get_recent_failures()
            if recent_failures >= self.config.failure_threshold:
                self._open_circuit()
        
        # OPEN -> HALF_OPEN: Timeout elapsed
        elif self.state == CircuitState.OPEN:
            if now - self.metrics.last_failure_time >= self.config.timeout_seconds:
                self._half_open_circuit()
        
        # HALF_OPEN -> CLOSED/OPEN: Check success/failure ratio
        elif self.state == CircuitState.HALF_OPEN:
            if self.metrics.half_open_calls >= self.config.half_open_max_calls:
                if self.metrics.consecutive_failures == 0:
                    self._close_circuit()
                else:
                    self._open_circuit()
    
    def _get_recent_failures(self) -> int:
        """Count failures within the sliding window."""
        now = time.time()
        cutoff = now - self.config.rate_window_seconds
        return sum(1 for t in self._call_history['failures'] if t > cutoff)
    
    def _on_success(self):
        self.metrics.successes += 1
        self.metrics.consecutive_failures = 0
        
        if self.state == CircuitState.HALF_OPEN:
            self.metrics.half_open_calls += 1
            if self.metrics.half_open_calls >= self.config.success_threshold:
                self._close_circuit()
    
    def _on_failure(self):
        self.metrics.failures += 1
        self.metrics.consecutive_failures += 1
        self.metrics.last_failure_time = time.time()
        self._call_history['failures'].append(time.time())
        
        if self.state == CircuitState.CLOSED:
            if self.metrics.consecutive_failures >= self.config.failure_threshold:
                self._open_circuit()
        elif self.state == CircuitState.HALF_OPEN:
            self._open_circuit()
    
    def _open_circuit(self):
        self.state = CircuitState.OPEN
        self.logger.error(f"Circuit {self.name} OPENED - blocking requests")
    
    def _half_open_circuit(self):
        self.state = CircuitState.HALF_OPEN
        self.metrics.half_open_calls = 0
        self.logger.info(f"Circuit {self.name} HALF-OPEN - allowing test requests")
    
    def _close_circuit(self):
        self.state = CircuitState.CLOSED
        self.metrics.consecutive_failures = 0
        self.metrics.half_open_calls = 0
        self._call_history.clear()
        self.logger.info(f"Circuit {self.name} CLOSED - normal operation resumed")
    
    def get_status(self) -> Dict[str, Any]:
        return {
            "name": self.name,
            "state": self.state.value,
            "metrics": {
                "total_requests": self.metrics.total_requests,
                "total_fallbacks": self.metrics.total_fallbacks,
                "failures": self.metrics.failures,
                "successes": self.metrics.successes,
                "fallback_rate": self.metrics.total_fallbacks / max(1, self.metrics.total_requests)
            }
        }

class CircuitBreakerOpenError(Exception):
    """Raised when circuit breaker is open and no fallback is available."""
    pass

Integration with HolySheep AI Relay Layer

Now let's integrate our circuit breaker with HolySheep's multi-provider routing for automatic failover:

import requests
import json
from typing import List, Dict, Any, Optional

class HolySheepAIClient:
    """HolySheep AI client with built-in circuit breaker and multi-provider fallback."""
    
    BASE_URL = "https://api.holysheep.ai/v1"
    
    # Provider configurations with fallback chain
    PROVIDER_CHAIN = [
        {"id": "gpt-4.1", "model": "gpt-4.1", "cost_per_mtok": 8.00, "priority": 1},
        {"id": "claude-sonnet-4.5", "model": "claude-sonnet-4.5", "cost_per_mtok": 15.00, "priority": 2},
        {"id": "gemini-2.5-flash", "model": "gemini-2.5-flash", "cost_per_mtok": 2.50, "priority": 3},
        {"id": "deepseek-v3.2", "model": "deepseek-v3.2", "cost_per_mtok": 0.42, "priority": 4},
    ]
    
    def __init__(self, api_key: str, circuit_breaker_config: CircuitBreakerConfig = None):
        self.api_key = api_key
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        })
        
        # Create circuit breaker per provider
        self.circuit_breakers: Dict[str, CircuitBreaker] = {}
        for provider in self.PROVIDER_CHAIN:
            self.circuit_breakers[provider["id"]] = CircuitBreaker(
                name=f"holydaemon-{provider['id']}",
                config=circuit_breaker_config or CircuitBreakerConfig(
                    failure_threshold=3,
                    timeout_seconds=60.0
                )
            )
        
        self.current_provider_idx = 0
        self.total_cost_saved = 0.0
    
    def _get_current_provider(self) -> Dict:
        """Get current active provider from chain."""
        return self.PROVIDER_CHAIN[self.current_provider_idx]
    
    def _get_fallback_provider(self) -> Optional[Dict]:
        """Get next available fallback provider."""
        if self.current_provider_idx < len(self.PROVIDER_CHAIN) - 1:
            return self.PROVIDER_CHAIN[self.current_provider_idx + 1]
        return None
    
    def chat_completions(
        self,
        messages: List[Dict[str, str]],
        system_prompt: str = "You are a helpful assistant.",
        temperature: float = 0.7,
        max_tokens: int = 1000,
        **kwargs
    ) -> Dict[str, Any]:
        """
        Send chat completion request with automatic circuit breaker and fallback.
        """
        # Prepare full message array
        full_messages = [{"role": "system", "content": system_prompt}] + messages
        
        # Track attempt and errors
        attempts = []
        last_error = None
        
        # Try providers in chain order
        for provider in self.PROVIDER_CHAIN:
            provider_id = provider["id"]
            cb = self.circuit_breakers[provider_id]
            
            try:
                result = cb.call(
                    self._make_request,
                    fallback=self._get_fallback_response,
                    provider=provider,
                    messages=full_messages,
                    temperature=temperature,
                    max_tokens=max_tokens
                )
                
                # Track cost savings vs official pricing
                output_tokens = result.get("usage", {}).get("completion_tokens", 0)
                official_cost = output_tokens * provider["cost_per_mtok"] / 1_000_000
                holydaemon_cost = output_tokens * provider["cost_per_mtok"] / 1_000_000
                self.total_cost_saved += (official_cost - holydaemon_cost)
                
                result["_meta"] = {
                    "provider": provider_id,
                    "circuit_state": cb.state.value,
                    "cost_usd": holydaemon_cost,
                    "attempt_number": len(attempts) + 1
                }
                return result
                
            except CircuitBreakerOpenError:
                last_error = f"Circuit breaker open for {provider_id}"
                attempts.append({"provider": provider_id, "error": "circuit_open"})
                continue
            except Exception as e:
                last_error = str(e)
                attempts.append({"provider": provider_id, "error": str(e)})
                continue
        
        # All providers failed
        raise AIProviderExhaustedError(
            f"All providers exhausted. Attempts: {json.dumps(attempts, indent=2)}",
            attempts=attempts
        )
    
    def _make_request(
        self,
        provider: Dict,
        messages: List[Dict],
        temperature: float,
        max_tokens: int
    ) -> Dict[str, Any]:
        """Make actual API request to HolySheep."""
        endpoint = f"{self.BASE_URL}/chat/completions"
        
        payload = {
            "model": provider["model"],
            "messages": messages,
            "temperature": temperature,
            "max_tokens": max_tokens
        }
        
        response = self.session.post(endpoint, json=payload, timeout=30)
        
        if response.status_code == 429:
            raise RateLimitError("Rate limit exceeded")
        elif response.status_code >= 500:
            raise ProviderServerError(f"Provider error: {response.status_code}")
        elif response.status_code != 200:
            raise APIError(f"API error {response.status_code}: {response.text}")
        
        return response.json()
    
    def _get_fallback_response(self, **kwargs) -> Dict[str, Any]:
        """Fallback response when circuit is open."""
        return {
            "choices": [{
                "message": {
                    "role": "assistant",
                    "content": "⚠️ Service temporarily degraded. Using cached response or lower-tier model."
                }
            }],
            "usage": {"completion_tokens": 0},
            "_fallback": True
        }
    
    def get_dashboard_status(self) -> Dict[str, Any]:
        """Get status of all circuit breakers and cost metrics."""
        return {
            "providers": {
                pid: cb.get_status() 
                for pid, cb in self.circuit_breakers.items()
            },
            "cost_savings_usd": self.total_cost_saved,
            "current_primary": self._get_current_provider()["id"]
        }

class RateLimitError(Exception):
    """Raised when rate limit is exceeded."""
    pass

class ProviderServerError(Exception):
    """Raised when provider returns 5xx error."""
    pass

class APIError(Exception):
    """Raised for general API errors."""
    pass

class AIProviderExhaustedError(Exception):
    """Raised when all providers in chain have failed."""
    def __init__(self, message, attempts=None):
        super().__init__(message)
        self.attempts = attempts or []

Production Usage Example

# Initialize client with your HolySheep API key
client = HolySheepAIClient(
    api_key="YOUR_HOLYSHEEP_API_KEY",
    circuit_breaker_config=CircuitBreakerConfig(
        failure_threshold=5,
        success_threshold=3,
        timeout_seconds=30.0,
        rate_window_seconds=60.0
    )
)

def generate_with_fallback(user_query: str) -> str:
    """Generate response with automatic circuit breaker protection."""
    
    try:
        # Primary request - will use circuit breaker and automatic fallback
        response = client.chat_completions(
            messages=[{"role": "user", "content": user_query}],
            system_prompt="You are a helpful customer support assistant.",
            temperature=0.7,
            max_tokens=500
        )
        
        content = response["choices"][0]["message"]["content"]
        provider = response["_meta"]["provider"]
        circuit_state = response["_meta"]["circuit_state"]
        
        print(f"Response from {provider} (circuit: {circuit_state})")
        return content
        
    except AIProviderExhaustedError as e:
        print(f"All providers exhausted: {e}")
        return "Service temporarily unavailable. Please try again later."
    except Exception as e:
        print(f"Unexpected error: {e}")
        return "An error occurred. Please contact support."


Monitor circuit breaker health

def monitor_circuits():
    """Periodic monitoring of circuit breaker states."""
    status = client.get_dashboard_status()
    
    print(f"\n=== Circuit Breaker Dashboard ===")
    print(f"Total cost saved: ${status['cost_savings_usd']:.2f}")
    print(f"Primary provider: {status['current_primary']}\n")
    
    for provider_id, info in status["providers"].items():
        state = info["state"]
        metrics = info["metrics"]
        
        state_emoji = {"closed": "✅", "open": "🔴", "half_open": "🟡"}.get(state, "⚪")
        
        print(f"{state_emoji} {provider_id}")
        print(f"   State: {state.upper()}")
        print(f"   Requests: {metrics['total_requests']}, "
              f"Fallbacks: {metrics['total_fallbacks']} "
              f"({metrics['fallback_rate']*100:.1f}%)")


Example usage

if __name__ == "__main__":
    # Test with normal request
    result = generate_with_fallback("Explain circuit breakers in one sentence.")
    print(f"\nResult: {result}\n")
    
    # Monitor circuit health
    monitor_circuits()

Graceful Degradation Strategies

Beyond circuit breakers, implement these degradation layers for maximum resilience:

# Cache-based fallback layer
class ResponseCache:
    """LRU cache with circuit-breaker integration."""
    
    def __init__(self, max_size: int = 1000, ttl_seconds: int = 3600):
        self.cache = {}
        self.timestamps = {}
        self.ttl = ttl_seconds
        self.max_size = max_size
        self.hits = 0
        self.misses = 0
    
    def _make_key(self, messages: List[Dict], **kwargs) -> str:
        """Create cache key from request parameters."""
        import hashlib
        content = json.dumps({"messages": messages, **kwargs}, sort_keys=True)
        return hashlib.sha256(content.encode()).hexdigest()[:32]
    
    def get_or_fetch(
        self,
        client: HolySheepAIClient,
        messages: List[Dict],
        system_prompt: str,
        **kwargs
    ) -> Dict[str, Any]:
        """Try cache first, fetch from HolySheep on miss."""
        cache_key = self._make_key(messages, system_prompt=system_prompt, **kwargs)
        
        # Check cache
        if cache_key in self.cache:
            if time.time() - self.timestamps[cache_key] < self.ttl:
                self.hits += 1
                result = self.cache[cache_key].copy()
                result["_meta"]["cache_hit"] = True
                return result
        
        self.misses += 1
        
        # Fetch from HolySheep with circuit breaker
        try:
            result = client.chat_completions(
                messages=messages,
                system_prompt=system_prompt,
                **kwargs
            )
            
            # Store in cache
            if len(self.cache) >= self.max_size:
                oldest_key = min(self.timestamps, key=self.timestamps.get)
                del self.cache[oldest_key]
                del self.timestamps[oldest_key]
            
            self.cache[cache_key] = result.copy()
            self.timestamps[cache_key] = time.time()
            result["_meta"]["cache_hit"] = False
            
            return result
            
        except AIProviderExhaustedError:
            # All circuits open - try cache even if expired
            if cache_key in self.cache:
                self.hits += 1
                result = self.cache[cache_key].copy()
                result["_meta"]["cache_hit"] = True
                result["_meta"]["stale_cache"] = True
                return result
            
            # No cache available - return degraded response
            return {
                "choices": [{
                    "message": {
                        "role": "assistant",
                        "content": "Service is currently degraded. Your request has been queued."
                    }
                }],
                "usage": {"completion_tokens": 0},
                "_meta": {"degraded": True, "queued": True}
            }

Pricing and ROI

ScenarioOfficial API CostHolySheep CostSavings
100K GPT-4.1 requests × 500 tokens$600$32047%
Mixed: 50% DeepSeek + 50% GPT-4.1$900$21077%
High-volume: 1M Gemini Flash$3,500$1,25064%
Circuit breaker saves (fallback routing)$0 recovered$200/mo avgFailure avoidance

ROI Calculation for Production Systems

For a system processing 10,000 AI requests daily:

Common Errors and Fixes

Error 1: "Circuit Breaker Stuck in OPEN State"

Symptom: Circuit remains open even after provider recovers. Requests always fail-fast.

# Problem: Timeout too long or success threshold unreachable

Fix: Adjust configuration and implement manual reset



Update circuit breaker config with shorter timeout

config = CircuitBreakerConfig(
    failure_threshold=3,           # Lower from 5
    success_threshold=2,          # Lower from 3
    timeout_seconds=15.0,          # Reduce from 30
    half_open_max_calls=5          # Increase to allow more recovery attempts
)


Implement manual circuit reset for operations team

def reset_circuit_manually(client: HolySheepAIClient, provider_id: str):
    """Allow ops team to manually reset stuck circuits."""
    cb = client.circuit_breakers[provider_id]
    with cb._lock:
        cb._close_circuit()
        print(f"Circuit {provider_id} manually reset to CLOSED")
    
    # Verify by sending test request
    try:
        result = cb.call(
            lambda: {"test": "success"},
            fallback=None
        )
        print(f"Circuit {provider_id} test passed: {result}")
    except Exception as e:
        print(f"Circuit {provider_id} test failed: {e}")


Usage: Call this via admin endpoint or monitoring system


reset_circuit_manually(client, "gpt-4.1")

Error 2: "429 Rate Limit Even with Circuit Breaker"

Symptom: Getting rate limited despite circuit breaker, indicating the breaker threshold is too high.

# Problem: Circuit breaker thresholds don't account for rate limits

Fix: Add rate limit awareness and reduce circuit threshold


class RateLimitAwareBreaker(CircuitBreaker):
    """Circuit breaker that responds to explicit rate limit responses."""
    
    def __init__(self, name: str, config: CircuitBreakerConfig = None):
        super().__init__(name, config)
        self.rate_limit_count = 0
        self.rate_limit_threshold = 2  # Open after 2 rate limits
    
    def on_rate_limit(self):
        """Called when a 429 is detected - immediate circuit open."""
        with self._lock:
            self.rate_limit_count += 1
            self.logger.warning(
                f"Rate limit detected ({self.rate_limit_count}/{self.rate_limit_threshold})"
            )
            
            if self.rate_limit_count >= self.rate_limit_threshold:
                self._open_circuit()
                # Reset after shorter timeout for rate limits
                self.metrics.last_failure_time = time.time()


Update client to use rate limit aware breaker

class HolySheepAIClientV2(HolySheepAIClient):
    def __init__(self, api_key: str, circuit_breaker_config: CircuitBreakerConfig = None):
        super().__init__(api_key, circuit_breaker_config)
        
        # Replace standard breakers with rate-limit-aware ones
        for provider in self.PROVIDER_CHAIN:
            self.circuit_breakers[provider["id"]] = RateLimitAwareBreaker(
                name=f"holydaemon-{provider['id']}",
                config=circuit_breaker_config or CircuitBreakerConfig(
                    failure_threshold=2,
                    timeout_seconds=30.0
                )
            )
    
    def _make_request(self, provider: Dict, messages: List[Dict], 
                     temperature: float, max_tokens: int) -> Dict[str, Any]:
        try:
            return super()._make_request(provider, messages, temperature, max_tokens)
        except RateLimitError:
            # Notify circuit breaker of rate limit
            cb = self.circuit_breakers[provider["id"]]
            cb.on_rate_limit()
            raise

Error 3: "Fallback Chain Not Working - All Requests Fail"

Symptom: Circuit breaker opens but fallback doesn't trigger, or wrong fallback is called.

# Problem: Fallback logic has bugs or circuit state is checked incorrectly

Fix: Verify fallback chain and add comprehensive logging


def chat_completions_fixed(
    self,
    messages: List[Dict[str, str]],
    system_prompt: str = "You are a helpful assistant.",
    **kwargs
) -> Dict[str, Any]:
    """Fixed version with proper fallback verification."""
    
    for i, provider in enumerate(self.PROVIDER_CHAIN):
        provider_id = provider["id"]
        cb = self.circuit_breakers[provider_id]
        
        # Log current state before attempt
        self.logger.info(
            f"Attempt {i+1}/{len(self.PROVIDER_CHAIN)}: "
            f"Provider={provider_id}, Circuit={cb.state.value}"
        )
        
        # Skip OPEN circuits without calling them
        if cb.state == CircuitState.OPEN:
            self.logger.warning(f"Skipping {provider_id} - circuit OPEN")
            continue
        
        try:
            result = self._make_request(provider, messages, **kwargs)
            result["_meta"] = {"provider": provider_id, "attempt": i+1}
            return result
            
        except RateLimitError as e:
            # Rate limit: open circuit, move to next provider immediately
            self.logger.warning(f"Rate limited on {provider_id}: {e}")
            cb.on_rate_limit()
            continue
            
        except (ProviderServerError, Exception) as e:
            # Server error: record failure, let circuit breaker handle
            self.logger.error(f"Error on {provider_id}: {e}")
            cb._on_failure()
            
            # Check if circuit opened
            if cb.state == CircuitState.OPEN:
                self.logger.warning(f"Circuit {provider_id} opened after error")
            continue
    
    # Exhaustive fallback - use cache or degraded response
    self.logger.error("All providers exhausted - using degraded response")
    return self._get_fallback_response()

Error 4: "High Latency Despite Circuit Breaker"

Symptom: Requests are slow even when circuit is closed, or timeout errors occur frequently.

# Problem: Timeout configuration doesn't match actual response times

Fix: Implement adaptive timeouts and connection pooling


class AdaptiveTimeoutClient(HolySheepAIClient):
    """Client with adaptive timeouts based on historical performance."""
    
    def __init__(self, api_key: str, circuit_breaker_config: CircuitBreakerConfig = None):
        super().__init__(api_key, circuit_breaker_config)
        
        # Override session with optimized connection pooling
        adapter = requests.adapters.HTTPAdapter(
            pool_connections=20,
            pool_maxsize=100,
            max_retries=1,
            pool_block=False
        )
        self.session.mount("https://", adapter)
        self.session.mount("http://", adapter)
        
        # Track per-provider latency
        self.latency_history: Dict[str, list] = defaultdict(list)
        self.target_percentile = 95  # Target p95 latency
    
    def _get_adaptive_timeout(self, provider_id: str) -> float:
        """Calculate adaptive timeout based on historical p95 latency."""
        history = self.latency_history.get(provider_id, [])
        
        if len(history) < 10:
            return 30.0  # Default timeout
        
        sorted_latencies = sorted(history)
        p95_index = int(len(sorted_latencies) * 0.95)
        p95_latency = sorted_latencies[p95_index]
        
        # Add 50% buffer for safety
        return min(p95_latency * 1.5, 60.0)  # Cap at 60 seconds
    
    def _make_request(self, provider: Dict, messages: List[Dict],
                     temperature: float, max_tokens: int) -> Dict[str, Any]:
        """Make request with adaptive timeout and latency tracking."""
        provider_id = provider["id"]
        timeout = self._get_adaptive_timeout(provider_id)
        
        start_time = time.time()
        
        try:
            response = self.session.post(
                f"{self.BASE_URL}/chat/completions",
                json={
                    "model": provider["model"],
                    "messages": messages,
                    "temperature": temperature,
                    "max_tokens": max_tokens
                },
                timeout=timeout
            )
            
            # Track latency
            latency = (time.time() - start_time) * 1000  # ms
            self.latency_history[provider_id].append(latency)
            
            # Keep only last 100

Related Resources
Related Articles