Đêm hôm đó, hệ thống AI của tôi sụp đổ hoàn toàn. Người dùng phản hồi chậm như rùa, dashboard báo lỗi liên tục, và điều tồi tệ nhất — tiền账单 tăng vọt vì hệ thống cứ gọi API thất bại đi gọi lại. Đó là lúc tôi phát hiện ra circuit breaker pattern — một kỹ thuật mà sau này trở thành xương sống của mọi ứng dụng AI production.
Điều Gì Xảy Ra Khi API "Trở Mặt"?
Trước khi hiểu circuit breaker, hãy tưởng tượng bạn gọi điện cho một người bạn. Lần đầu không nghe máy — bạn gọi lại. Lần hai — gọi lại. Lần ba — vẫn không ai nhấc. Nếu bạn cứ gọi liên tục 100 lần trong khi người đó đang bận họp hoặc điện thoại hỏng thì sao? Bạn sẽ phát điên, lãng phí thời gian, và có thể bỏ lỡ cuộc gọi quan trọng khác.
Đó chính xác là những gì xảy ra khi API AI gặp sự cố và code của bạn không có cơ chế dừng lại. Mỗi request thất bại tiêu tốn:
- Thời gian chờ: Trung bình 5-30 giây mỗi lần timeout
- Bandwidth: Request gửi đi, response nhận về (dù lỗi)
- Quota API: Mỗi lần gọi đều trừ vào giới hạn rate limit
- Truyền thống: Nếu dùng API Trung Quốc, tiền ¥ trả cho từng request thất bại
Tôi từng để một script chạy qua đêm với retry vô hạn — sáng hôm sau quota API đã "bốc hơi" hết 80%, và server logs có hơn 47,000 dòng lỗi timeout. Một bài học đắt giá về tầm quan trọng của circuit breaker.
Pattern Circuit Breaker Là Gì?
Circuit breaker hoạt động như AUTOMAT CẦU DAO điện trong nhà bạn. Khi dòng điện tăng đột ngột (quá tải), cb (circuit breaker) tự động ngắt để bảo vệ thiết bị. Khi điện ổn định trở lại, cb đóng lại và mọi thứ hoạt động bình thường.
Áp dụng vào lập trình, circuit breaker có 3 trạng thái:
- CLOSED (Đóng): Mọi thứ hoạt động tốt, request đi qua bình thường. Nếu có lỗi, đếm và ghi nhận.
- OPEN (Mở): Phát hiện quá nhiều lỗi → Ngắt ngay lập tức, không gọi API nữa. Trả về fallback response hoặc cache.
- HALF-OPEN (Nửa mở): Sau một khoảng thời gian chờ, thử gọi lại 1 request. Nếu thành công → Đóng cb và hoạt động bình thường. Nếu thất bại → Mở lại cb.
┌─────────────────────────────────────────────────────────────┐
│ CIRCUIT BREAKER FLOW │
├─────────────────────────────────────────────────────────────┤
│ │
│ Request ──▶ ┌──────┐ ──▶ [Success] ──▶ CLOSED ──▶ Response│
│ │ │ │
│ │ │ ──▶ [Failure] ──▶ Increment Counter │
│ │ │ │ │
│ └──┬───┘ ▼ │
│ │ ┌─────────────────┐ │
│ │ │ Failure ≥ Threshold? │
│ │ └────────┬────────┘ │
│ │ YES ─────┴───── NO │
│ │ │ │ │
│ │ ▼ │ │
│ │ ┌──────┐ │ │
│ │ │ OPEN │ │ │
│ │ └──┬───┘ │ │
│ │ │ │ │
│ │ ▼ Timeout? │ │
│ │ ┌──────────────┐ │ │
│ │ │ HALF-OPEN │────┘ │
│ │ └──────────────┘ │
│ │ │ │
│ │ ┌────────┴────────┐ │
│ │ ▼ ▼ │
│ │ Success Failure │
│ │ │ │ │
│ │ ▼ ▼ │
│ │ CLOSED OPEN │
│ │ │
└─────────────────────────────────────────────────────────────┘
Tại Sao Multi-Model Cần Circuit Breaker?
Khi bạn sử dụng nhiều model AI cùng lúc — ví dụ GPT-4.1 cho task phức tạp, Gemini 2.5 Flash cho task đơn giản, và DeepSeek V3.2 cho task tiết kiệm chi phí — mỗi model có thể gặp sự cố độc lập. Không ai muốn khi GPT-4.1 chết mà Gemini Flash cũng bị kéo chết theo.
Với HolyShehe AI, bạn có thể kết hợp nhiều model với chi phí cực kỳ thấp: DeepSeek V3.2 chỉ $0.42/MTok so với $8 của GPT-4.1 — tiết kiệm tới 85%. Nhưng dù dùng provider nào, bạn vẫn cần circuit breaker để đảm bảo hệ thống không sụp đổ khi một model gặp vấn đề.
Hướng Dẫn Triển Khai Chi Tiết
Bước 1: Cài Đặt Môi Trường
Tạo project mới
mkdir multi-model-circuit-breaker
cd multi-model-circuit-breaker
Tạo virtual environment
python -m venv venv
source venv/bin/activate # Linux/Mac
Hoặc: venv\Scripts\activate # Windows
Cài đặt dependencies
pip install httpx asyncio aiohttp
Bước 2: Triển Khai Circuit Breaker Class
Đây là phần quan trọng nhất. Tôi sẽ triển khai một class CircuitBreaker hoàn chỉnh với đầy đủ logic 3 trạng thái:
import time
import asyncio
from enum import Enum
from typing import Callable, Any, Optional
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 # Số lỗi để mở cb
success_threshold: int = 3 # Số thành công để đóng cb (half-open)
timeout_duration: float = 30.0 # Giây chờ trước khi thử lại (open→half-open)
half_open_max_calls: int = 1 # Số request thử trong trạng thái half-open
@dataclass
class CircuitBreakerMetrics:
total_calls: int = 0
successful_calls: int = 0
failed_calls: int = 0
rejected_calls: int = 0 # Request bị từ chối khi cb OPEN
state_changes: list = field(default_factory=list)
last_failure_time: Optional[float] = None
last_success_time: Optional[float] = None
class CircuitBreaker:
"""
Circuit Breaker Pattern Implementation
Bảo vệ API calls khỏi cascading failures
"""
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._half_open_calls = 0
self.metrics = CircuitBreakerMetrics()
@property
def state(self) -> CircuitState:
"""Kiểm tra và cập nhật trạng thái cb dựa trên timeout"""
if self._state == CircuitState.OPEN:
if self._should_attempt_reset():
self._transition_to_half_open()
return self._state
def _should_attempt_reset(self) -> bool:
"""Kiểm tra xem đã đến lúc thử reset chưa"""
if self._last_failure_time is None:
return True
elapsed = time.time() - self._last_failure_time
return elapsed >= self.config.timeout_duration
def _transition_to_half_open(self):
"""Chuyển sang trạng thái half-open"""
self._state = CircuitState.HALF_OPEN
self._half_open_calls = 0
self._record_state_change(CircuitState.HALF_OPEN)
print(f"[{self.name}] Circuit → HALF_OPEN (sẽ thử 1 request)")
def _transition_to_open(self):
"""Chuyển sang trạng thái open"""
self._state = CircuitState.OPEN
self._last_failure_time = time.time()
self._record_state_change(CircuitState.OPEN)
print(f"[{self.name}] Circuit → OPEN (ngừng gọi API)")
def _transition_to_closed(self):
"""Chuyển sang trạng thái closed"""
self._state = CircuitState.CLOSED
self._failure_count = 0
self._success_count = 0
self._record_state_change(CircuitState.CLOSED)
print(f"[{self.name}] Circuit → CLOSED (hoạt động bình thường)")
def _record_state_change(self, new_state: CircuitState):
self.metrics.state_changes.append({
"state": new_state.value,
"timestamp": time.time()
})
def call(self, func: Callable, *args, **kwargs) -> Any:
"""
Execute function với circuit breaker protection
Đồng bộ version
"""
self.metrics.total_calls += 1
# Kiểm tra trạng thái hiện tại
current_state = self.state
# Nếu đang OPEN hoặc half-open đã đạt limit
if current_state == CircuitState.OPEN:
self.metrics.rejected_calls += 1
raise CircuitBreakerOpenError(
f"Circuit {self.name} is OPEN. Request rejected. "
f"Try again in {self.config.timeout_duration}s"
)
if current_state == CircuitState.HALF_OPEN:
if self._half_open_calls >= self.config.half_open_max_calls:
self.metrics.rejected_calls += 1
raise CircuitBreakerOpenError(
f"Circuit {self.name} is HALF_OPEN. "
f"Max trial calls reached."
)
self._half_open_calls += 1
# Thực hiện call
try:
result = func(*args, **kwargs)
self._on_success()
return result
except Exception as e:
self._on_failure()
raise
async def call_async(self, func: Callable, *args, **kwargs) -> Any:
"""
Execute async function với circuit breaker protection
"""
self.metrics.total_calls += 1
current_state = self.state
if current_state == CircuitState.OPEN:
self.metrics.rejected_calls += 1
raise CircuitBreakerOpenError(
f"Circuit {self.name} is OPEN. Request rejected."
)
if current_state == CircuitState.HALF_OPEN:
if self._half_open_calls >= self.config.half_open_max_calls:
self.metrics.rejected_calls += 1
raise CircuitBreakerOpenError(
f"Circuit {self.name} is HALF_OPEN. Max trial calls reached."
)
self._half_open_calls += 1
try:
result = await func(*args, **kwargs)
self._on_success()
return result
except Exception as e:
self._on_failure()
raise
def _on_success(self):
self.metrics.successful_calls += 1
self.metrics.last_success_time = time.time()
if self._state == CircuitState.HALF_OPEN:
self._success_count += 1
if self._success_count >= self.config.success_threshold:
self._transition_to_closed()
elif self._state == CircuitState.CLOSED:
# Reset failure count on success
self._failure_count = 0
def _on_failure(self):
self.metrics.failed_calls += 1
self.metrics.last_failure_time = time.time()
self._failure_count += 1
if self._state == CircuitState.HALF_OPEN:
# Any failure in half-open → immediately open
self._transition_to_open()
elif self._state == CircuitState.CLOSED:
if self._failure_count >= self.config.failure_threshold:
self._transition_to_open()
def get_health_status(self) -> dict:
"""Lấy thông tin sức khỏe của circuit breaker"""
return {
"name": self.name,
"state": self.state.value,
"failure_count": self._failure_count,
"success_count": self._success_count,
"metrics": {
"total_calls": self.metrics.total_calls,
"success_rate": (
self.metrics.successful_calls / self.metrics.total_calls * 100
if self.metrics.total_calls > 0 else 0
),
"rejection_rate": (
self.metrics.rejected_calls / self.metrics.total_calls * 100
if self.metrics.total_calls > 0 else 0
)
}
}
class CircuitBreakerOpenError(Exception):
"""Exception raised when circuit breaker is open"""
pass
Bư�3: Triển Khai Multi-Model API Manager với HolySheep AI
Bây giờ chúng ta sẽ xây dựng một API Manager thông minh, có thể:
- Gọi nhiều model AI khác nhau
- Tự động chuyển sang model dự phòng khi model chính lỗi
- Sử dụng circuit breaker cho mỗi model riêng biệt
import asyncio
import httpx
from typing import Optional, Dict, List, Any
from dataclasses import dataclass
from enum import Enum
Import circuit breaker từ bước trước
from circuit_breaker import CircuitBreaker, CircuitBreakerConfig, CircuitBreakerOpenError
class ModelType(Enum):
GPT_4_1 = "gpt-4.1"
CLAUDE_SONNET_4_5 = "claude-sonnet-4.5"
GEMINI_FLASH_2_5 = "gemini-2.5-flash"
DEEPSEEK_V3_2 = "deepseek-v3.2"
@dataclass
class ModelConfig:
model_id: ModelType
api_url: str
fallback_models: List[ModelType]
circuit_breaker: CircuitBreaker
timeout: float = 30.0
class MultiModelAPIManager:
"""
Quản lý multi-model API calls với circuit breaker protection
Sử dụng HolySheep AI endpoint: https://api.holysheep.ai/v1
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
# Cấu hình circuit breaker cho mỗi model
# Threshold thấp hơn cho model đắt tiền hơn
self.model_configs: Dict[ModelType, ModelConfig] = {
ModelType.GPT_4_1: ModelConfig(
model_id=ModelType.GPT_4_1,
api_url=f"{self.base_url}/chat/completions",
fallback_models=[ModelType.GEMINI_FLASH_2_5, ModelType.DEEPSEEK_V3_2],
circuit_breaker=CircuitBreaker(
name="gpt-4.1",
config=CircuitBreakerConfig(
failure_threshold=3, # Mở cb sau 3 lỗi
success_threshold=2, # Đóng sau 2 thành công
timeout_duration=30.0 # Thử lại sau 30s
)
)
),
ModelType.CLAUDE_SONNET_4_5: ModelConfig(
model_id=ModelType.CLAUDE_SONNET_4_5,
api_url=f"{self.base_url}/chat/completions",
fallback_models=[ModelType.GPT_4_1, ModelType.DEEPSEEK_V3_2],
circuit_breaker=CircuitBreaker(
name="claude-sonnet-4.5",
config=CircuitBreakerConfig(
failure_threshold=3,
success_threshold=2,
timeout_duration=30.0
)
)
),
ModelType.GEMINI_FLASH_2_5: ModelConfig(
model_id=ModelType.GEMINI_FLASH_2_5,
api_url=f"{self.base_url}/chat/completions",
fallback_models=[ModelType.DEEPSEEK_V3_2],
circuit_breaker=CircuitBreaker(
name="gemini-2.5-flash",
config=CircuitBreakerConfig(
failure_threshold=5, # Model rẻ hơn → threshold cao hơn
success_threshold=3,
timeout_duration=15.0
)
)
),
ModelType.DEEPSEEK_V3_2: ModelConfig(
model_id=ModelType.DEEPSEEK_V3_2,
api_url=f"{self.base_url}/chat/completions",
fallback_models=[], # Model cuối cùng, không có fallback
circuit_breaker=CircuitBreaker(
name="deepseek-v3.2",
config=CircuitBreakerConfig(
failure_threshold=5,
success_threshold=3,
timeout_duration=10.0
)
)
)
}
# Cache cho fallback response
self._cache: Dict[str, Any] = {}
async def call_with_circuit_breaker(
self,
model: ModelType,
messages: List[Dict[str, str]],
temperature: float = 0.7,
max_tokens: int = 1000