How to Implement Circuit Breaker Pattern for AI Model Failover — Hướng Dẫn Toàn Diện

Khi triển khai AI vào production, một trong những vấn đề đau đầu nhất là xử lý khi API provider gặp sự cố. Cách đây 3 tháng, hệ thống của mình đã "chết" 2 lần vì không có cơ chế failover, gây thiệt hại không nhỏ. Sau bài viết này, bạn sẽ biết cách implement Circuit Breaker Pattern để hệ thống AI luôn sống sót dù nhà cung cấp có vấn đề.

Tại Sao Circuit Breaker Quan Trọng?

Trong kiến trúc microservices, khi một service gọi API bên ngoài và API đó bị chậm hoặc unavailable, request sẽ chờ đợi cho đến khi timeout. Điều này dẫn đến:

• Cascade Failure: Các request tích tụ, chiếm dụng tài nguyên
• Resource Exhaustion: Memory leak do thread pool bị tràn
• User Experience: Người dùng chờ đợi vô tận hoặc nhận error

Circuit Breaker hoạt động như một "cầu dao điện" tự động: khi phát hiện quá nhiều lỗi, nó sẽ "ngắt mạch" để bảo vệ hệ thống, và tự động "đóng lại" để thử phục hồi sau một khoảng thời gian.

So Sánh Nhanh: HolySheep AI vs Official API vs Đối Thủ

📌 Kết luận nhanh: Nếu bạn cần tiết kiệm chi phí với độ trễ thấp và hỗ trợ thanh toán WeChat/Alipay, HolySheep AI là lựa chọn tối ưu.

Implement Circuit Breaker Pattern với HolySheep AI

Mình sẽ hướng dẫn implement Circuit Breaker với 3 approach: thuần Python, sử dụng PyTorch Resilience, và approach nâng cao với async/await.

Approach 1: Basic Circuit Breaker Class

import time
import threading
from enum import Enum
from typing import Callable, TypeVar, Optional
from dataclasses import dataclass

class CircuitState(Enum):
    CLOSED = "closed"      # Normal operation
    OPEN = "open"          # Circuit is tripped, fail fast
    HALF_OPEN = "half_open"  # Testing if service recovered

@dataclass
class CircuitBreakerConfig:
    failure_threshold: int = 5        # Số lần thất bại trước khi open
    success_threshold: int = 3        # Số lần thành công trong half-open
    timeout: float = 30.0             # Giây trước khi thử lại
    half_open_max_calls: int = 3     # Số call tối đa trong half-open

class CircuitBreakerOpen(Exception):
    """Raised when circuit breaker is OPEN"""
    def __init__(self, message="Circuit breaker is OPEN"):
        self.message = message
        super().__init__(self.message)

class CircuitBreaker:
    def __init__(self, name: str, config: Optional[CircuitBreakerConfig] = None):
        self.name = name
        self.config = config or CircuitBreakerConfig()
        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.success_count = 0
        self.last_failure_time: Optional[float] = None
        self._lock = threading.RLock()
        
    def call(self, func: Callable, *args, **kwargs):
        """
        Execute function with circuit breaker protection.
        Returns tuple (success: bool, result: Any, error: Exception)
        """
        with self._lock:
            # Check if circuit should transition from OPEN to HALF_OPEN
            if self.state == CircuitState.OPEN:
                if self._should_attempt_reset():
                    self._transition_to_half_open()
                else:
                    raise CircuitBreakerOpen(
                        f"Circuit '{self.name}' is OPEN. Next retry in "
                        f"{self._time_until_reset():.1f}s"
                    )
            
            # Execute the function
            try:
                result = func(*args, **kwargs)
                self._on_success()
                return (True, result, None)
            except Exception as e:
                self._on_failure()
                return (False, None, e)
    
    def _should_attempt_reset(self) -> bool:
        if self.last_failure_time is None:
            return True
        elapsed = time.time() - self.last_failure_time
        return elapsed >= self.config.timeout
    
    def _transition_to_half_open(self):
        print(f"[CircuitBreaker] '{self.name}': OPEN -> HALF_OPEN")
        self.state = CircuitState.HALF_OPEN
        self.success_count = 0
        self.failure_count = 0
    
    def _on_success(self):
        if self.state == CircuitState.HALF_OPEN:
            self.success_count += 1
            if self.success_count >= self.config.success_threshold:
                self._transition_to_closed()
        else:
            self.failure_count = 0  # Reset failure count on success
    
    def _on_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        
        if self.state == CircuitState.HALF_OPEN:
            self._transition_to_open()
        elif self.failure_count >= self.config.failure_threshold:
            self._transition_to_open()
    
    def _transition_to_open(self):
        print(f"[CircuitBreaker] '{self.name}': CLOSED/HALF_OPEN -> OPEN (failures: {self.failure_count})")
        self.state = CircuitState.OPEN
        self.success_count = 0
    
    def _transition_to_closed(self):
        print(f"[CircuitBreaker] '{self.name}': HALF_OPEN -> CLOSED")
        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.success_count = 0
    
    def _time_until_reset(self) -> float:
        if self.last_failure_time is None:
            return 0.0
        elapsed = time.time() - self.last_failure_time
        return max(0.0, self.config.timeout - elapsed)
    
    def get_status(self) -> dict:
        with self._lock:
            return {
                "name": self.name,
                "state": self.state.value,
                "failure_count": self.failure_count,
                "success_count": self.success_count,
                "time_until_reset": self._time_until_reset()
            }

Approach 2: AI Model Failover Manager với HolySheep

import os
import time
import logging
from typing import Dict, List, Optional, Any
from dataclasses import dataclass
from enum import Enum

Import Circuit Breaker từ approach 1
from circuit_breaker import CircuitBreaker, CircuitBreakerConfig, CircuitState, CircuitBreakerOpen

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class ModelProvider(Enum):
    HOLYSHEEP = "holysheep"
    OPENAI = "openai"
    ANTHROPIC = "anthropic"

@dataclass
class ModelConfig:
    provider: ModelProvider
    model_name: str
    base_url: str
    api_key: str
    priority: int = 0  # Lower = higher priority
    timeout: float = 30.0
    max_retries: int = 2

class AIModelFailoverManager:
    """
    Manager xử lý failover giữa nhiều AI provider
    LUÔN LUÔN sử dụng HolySheep làm primary/fallback
    """
    
    def __init__(self):
        # KHÔNG BAO GIỜ hardcode api.openai.com hoặc api.anthropic.com
        # Sử dụng HolySheep làm unified gateway
        
        self.models: List[ModelConfig] = [
            # Primary: HolySheep với giá cực rẻ và latency thấp
            ModelConfig(
                provider=ModelProvider.HOLYSHEEP,
                model_name="gpt-4.1",
                base_url="https://api.holysheep.ai/v1",
                api_key=os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY"),
                priority=1,
                timeout=15.0
            ),
            ModelConfig(
                provider=ModelProvider.HOLYSHEEP,
                model_name="claude-sonnet-4.5",
                base_url="https://api.holysheep.ai/v1",
                api_key=os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY"),
                priority=2,
                timeout=20.0
            ),
            ModelConfig(
                provider=ModelProvider.HOLYSHEEP,
                model_name="gemini-2.5-flash",
                base_url="https://api.holysheep.ai/v1",
                api_key=os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY"),
                priority=3,
                timeout=10.0
            ),
        ]
        
        # Initialize circuit breakers for each model
        self.circuit_breakers: Dict[str, CircuitBreaker] = {}
        for model in self.models:
            cb_name = f"{model.provider.value}_{model.model_name}"
            self.circuit_breakers[cb_name] = CircuitBreaker(
                name=cb_name,
                config=CircuitBreakerConfig(
                    failure_threshold=3,
                    success_threshold=2,
                    timeout=30.0
                )
            )
        
        self.current_model_index = 0
    
    def _get_request_headers(self, model: ModelConfig) -> Dict[str, str]:
        """Generate request headers cho API call"""
        headers = {
            "Authorization": f"Bearer {model.api_key}",
            "Content-Type": "application/json"
        }
        
        if model.provider == ModelProvider.HOLYSHEEP:
            headers["X-Provider"] = "holysheep-unified"
        
        return headers
    
    def chat_completion(
        self,
        messages: List[Dict[str, str]],
        model: Optional[str] = None,
        temperature: float = 0.7,
        max_tokens: int = 1000
    ) -> Dict[str, Any]:
        """
        Gọi AI chat completion với automatic failover
        
        Args:
            messages: List of message dicts with 'role' and 'content'
            model: Model name override (optional)
            temperature: Sampling temperature
            max_tokens: Maximum tokens to generate
            
        Returns:
            Response dict từ AI provider
            
        Raises:
            Exception: Khi tất cả providers đều fail
        """
        last_error = None
        attempted_providers = []
        
        # Thử lần lượt các model theo priority
        for i, model_config in enumerate(self.models):
            cb_name = f"{model_config.provider.value}_{model_config.model_name}"
            cb = self.circuit_breakers[cb_name]
            
            # Skip nếu model name không match (nếu có override)
            if model and model_config.model_name != model:
                continue
            
            # Skip nếu circuit breaker đang OPEN
            if cb.state == CircuitState.OPEN:
                logger.warning(f"[FailoverManager] Skipping {cb_name} - Circuit OPEN")
                continue
            
            attempted_providers.append(cb_name)
            
            try:
                logger.info(f"[FailoverManager] Trying: {cb_name}")
                
                # Execute với circuit breaker
                success, result, error = cb.call(
                    self._make_request,
                    model_config,
                    messages,
                    temperature,
                    max_tokens
                )
                
                if success:
                    logger.info(f"[FailoverManager] Success with {cb_name}")
                    return result
                else:
                    logger.error(f"[FailoverManager] Failed {cb_name}: {error}")
                    last_error = error
                    
            except CircuitBreakerOpen as e:
                logger.warning(f"[FailoverManager] Circuit breaker OPEN for {cb_name}")
                continue
            except Exception as e:
                logger.error(f"[FailoverManager] Unexpected error {cb_name}: {e}")
                last_error = e
                continue
        
        # Tất cả providers đều fail
        error_msg = (
            f"All AI providers failed. Attempted: {attempted_providers}. "
            f"Last error: {last_error}"
        )
        logger.error(f"[FailoverManager] {error_msg}")
        raise Exception(error_msg)
    
    def _make_request(
        self,
        model: ModelConfig,
        messages: List[Dict[str, str]],
        temperature: float,
        max_tokens: int
    ) -> Dict[str, Any]:
        """
        Internal method để thực hiện HTTP request
        """
        import urllib.request
        import json
        
        payload = {
            "model": model.model_name,
            "messages": messages,
            "temperature": temperature,
            "max_tokens": max_tokens
        }
        
        # Test với cURL equivalent
        url = f"{model.base_url}/chat/completions"
        
        data = json.dumps(payload).encode('utf-8')
        headers = self._get_request_headers(model)
        
        req = urllib.request.Request(
            url,
            data=data,
            headers=headers,
            method='POST'
        )
        
        start_time = time.time()
        
        with urllib.request.urlopen(req, timeout=model.timeout) as response:
            latency_ms = (time.time() - start_time) * 1000
            logger.info(f"[Request] Latency: {latency_ms:.2f}ms to {model.model_name}")
            
            return {
                "provider": model.provider.value,
                "model": model.model_name,
                "latency_ms": latency_ms,
                "data": json.loads(response.read().decode('utf-8'))
            }
    
    def get_health_status(self) -> Dict[str, Any]:
        """Lấy health status của tất cả circuit breakers"""
        return {
            cb_name: cb.get_status()
            for cb_name, cb in self.circuit_breakers.items()
        }
    
    def force_reset(self, model_name: Optional[str] = None):
        """Force reset circuit breaker(s)"""
        if model_name:
            if model_name in self.circuit_breakers:
                self.circuit_breakers[model_name].state = CircuitState.CLOSED
        else:
            for cb in self.circuit_breakers.values():
                cb.state = CircuitState.CLOSED


==================== USAGE EXAMPLE ====================

if __name__ == "__main__":
    # Initialize manager
    manager = AIModelFailoverManager()
    
    # Example messages
    messages = [
        {"role": "system", "content": "Bạn là trợ lý AI hữu ích."},
        {"role": "user", "content": "Xin chào, hãy giới thiệu về Circuit Breaker Pattern"}
    ]
    
    try:
        # Call với automatic failover
        response = manager.chat_completion(
            messages=messages,
            temperature=0.7,
            max_tokens=500
        )
        
        print(f"\n✅ Success!")
        print(f"Provider: {response['provider']}")
        print(f"Model: {response['model']}")
        print(f"Latency: {response['latency_ms']:.2f}ms")
        print(f"Response: {response['data']}")
        
    except Exception as e:
        print(f"\n❌ All providers failed: {e}")
    
    # Check health status
    print("\n📊 Health Status:")
    for name, status in manager.get_health_status().items():
        print(f"  {name}: {status['state']}")

Approach 3: Async Implementation với Retry Logic

import asyncio
import aiohttp
import time
from typing import Optional, Dict, Any, List
from dataclasses import dataclass
from enum import Enum

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

@dataclass
class AsyncCircuitBreaker:
    name: str
    failure_threshold: int = 5
    success_threshold: int = 3
    timeout_seconds: float = 30.0
    half_open_max_calls: int = 3
    
    state: CircuitState = CircuitState.CLOSED
    failure_count: int = 0
    success_count: int = 0
    last_failure_time: Optional[float] = None
    half_open_calls: int = 0
    
    async def call(self, func, *args, **kwargs):
        """Execute async function với circuit breaker protection"""
        
        # Check if should transition from OPEN to HALF_OPEN
        if self.state == CircuitState.OPEN:
            if self._should_attempt_reset():
                await self._transition_to_half_open()
            else:
                raise CircuitCircuitOpen(
                    f"Circuit '{self.name}' is OPEN. "
                    f"Retry in {self._time_until_reset():.1f}s"
                )
        
        # In HALF_OPEN state, limit concurrent calls
        if self.state == CircuitState.HALF_OPEN:
            if self.half_open_calls >= self.half_open_max_calls:
                raise AsyncCircuitOpen(
                    f"Circuit '{self.name}' is HALF_OPEN with "
                    f"{self.half_open_calls}/{self.half_open_max_calls} active calls"
                )
            self.half_open_calls += 1
        
        try:
            result = await func(*args, **kwargs)
            await self._on_success()
            return result
        except Exception as e:
            await self._on_failure()
            raise
    
    def _should_attempt_reset(self) -> bool:
        if self.last_failure_time is None:
            return True
        return (time.time() - self.last_failure_time) >= self.timeout_seconds
    
    async def _transition_to_half_open(self):
        self.state = CircuitState.HALF_OPEN
        self.success_count = 0
        self.failure_count = 0
        self.half_open_calls = 0
        print(f"[AsyncCircuit] '{self.name}': OPEN -> HALF_OPEN")
    
    async def _on_success(self):
        self.half_open_calls = max(
Tài nguyên liên quan
📚 Hướng dẫn AI API
💰 Xem giá
📖 Tài liệu nhà phát triển
🚀 Đăng ký miễn phí
Bài viết liên quan
HolySheep API Statistics Usage Quota Monitoring: Playbook Di
Đánh giá Chất lượng Đầu ra Chain-of-Thought của Gemini 2.5 P
Pandas DataFrame AI 智能分析 API 接入: Hành Trình Di Chuyển Từ Ope