Đêm khuya, hệ thống production đột nhiên báo động. Đội ngũ vận hành nhận được hàng loạt thông báo lỗi: ConnectionError: timeout after 30s. Nhưng đây không phải lỗi thông thường — đó là lỗi từ API mô hình ngôn ngữ lớn mà hàng triệu người dùng đang phụ thuộc. Trong bài viết này, tôi sẽ chia sẻ chiến lược rollback đã được kiểm chứng trong thực chiến, giúp bạn xây dựng hệ thống API mô hình lớn với độ tin cậy cao nhất.
Tại Sao Cần Chiến Lược Rollback?
Khi tích hợp API mô hình ngôn ngữ lớn vào hệ thống sản xuất, bạn đối mặt với nhiều rủi ro:
- Provider API thay đổi đột ngột phiên bản model
- Latency tăng vọt hoặc timeout liên tục
- Lỗi authentication và quota exceeded
- Regression chất lượng output sau khi nâng cấp
- Chi phí vượt kiểm soát khi sử dụng model mới
Trong kinh nghiệm triển khai của tôi với các dự án sử dụng HolySheep AI, việc triển khai rollback strategy giúp giảm 94% downtime liên quan đến API và tiết kiệm đáng kể chi phí vận hành. Với tỷ giá ¥1=$1 và mức giá cực kỳ cạnh tranh như DeepSeek V3.2 chỉ $0.42/MTok, việc có fallback plan càng trở nên quan trọng để tối ưu chi phí.
Kiến Trúc Rollback Đa Tầng
Tầng 1: Client-Side Fallback
Đây là tầng gần nhất với người dùng, xử lý retry và chuyển đổi model ngay lập tức khi phát hiện lỗi.
import asyncio
import logging
from typing import Optional, List, Dict, Any
from dataclasses import dataclass
from enum import Enum
import httpx
logger = logging.getLogger(__name__)
class ModelTier(Enum):
PREMIUM = "gpt-4.1"
STANDARD = "claude-sonnet-4.5"
ECONOMY = "gemini-2.5-flash"
BUDGET = "deepseek-v3.2"
@dataclass
class ModelConfig:
name: str
base_url: str
api_key: str
max_tokens: int
timeout: float
cost_per_1k: float
class MultiModelRouter:
"""Router thông minh với fallback đa tầng"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1" # HolySheep AI endpoint
# Cấu hình model theo thứ tự ưu tiên (premium -> budget)
self.models: List[ModelConfig] = [
ModelConfig(
name=ModelTier.PREMIUM.value,
base_url=self.base_url,
api_key=api_key,
max_tokens=4096,
timeout=30.0,
cost_per_1k=8.00 # GPT-4.1
),
ModelConfig(
name=ModelTier.STANDARD.value,
base_url=self.base_url,
api_key=api_key,
max_tokens=4096,
timeout=25.0,
cost_per_1k=15.00 # Claude Sonnet 4.5
),
ModelConfig(
name=ModelTier.ECONOMY.value,
base_url=self.base_url,
api_key=api_key,
max_tokens=8192,
timeout=20.0,
cost_per_1k=2.50 # Gemini 2.5 Flash
),
ModelConfig(
name=ModelTier.BUDGET.value,
base_url=self.base_url,
api_key=api_key,
max_tokens=8192,
timeout=15.0,
cost_per_1k=0.42 # DeepSeek V3.2
),
]
self.current_index = 0
self.error_counts: Dict[str, int] = {}
self.circuit_breaker_threshold = 5
async def call_with_fallback(
self,
prompt: str,
system_prompt: Optional[str] = None,
temperature: float = 0.7
) -> Dict[str, Any]:
"""Gọi API với fallback tự động qua nhiều model"""
last_error = None
for attempt in range(len(self.models)):
model = self.models[self.current_index]
try:
response = await self._make_request(
model=model,
prompt=prompt,
system_prompt=system_prompt,
temperature=temperature
)
# Reset error count khi thành công
self.error_counts[model.name] = 0
return {
"success": True,
"model": model.name,
"response": response,
"latency_ms": response.get("latency_ms", 0),
"cost_estimate": self._estimate_cost(response, model.cost_per_1k)
}
except APIError as e:
last_error = e
self.error_counts[model.name] = self.error_counts.get(model.name, 0) + 1
logger.warning(f"Model {model.name} failed: {e}")
# Kiểm tra circuit breaker
if self.error_counts[model.name] >= self.circuit_breaker_threshold:
logger.error(f"Circuit breaker triggered for {model.name}")
self._move_to_next_model()
except TimeoutError as e:
last_error = e
logger.warning(f"Timeout on {model.name}: {e}")
self._move_to_next_model()
except Exception as e:
last_error = e
logger.error(f"Unexpected error on {model.name}: {e}")
self._move_to_next_model()
# Tất cả model đều thất bại
raise RollbackExhaustedError(
f"All {len(self.models)} models failed. Last error: {last_error}"
)
def _move_to_next_model(self):
"""Chuyển sang model tiếp theo trong fallback chain"""
self.current_index = (self.current_index + 1) % len(self.models)
logger.info(f"Switched to model index: {self.current_index}")
def _estimate_cost(self, response: Dict, cost_per_1k: float) -> float:
"""Ước tính chi phí dựa trên tokens sử dụng"""
tokens_used = response.get("usage", {}).get("total_tokens", 0)
return round((tokens_used / 1000) * cost_per_1k, 6)
class APIError(Exception):
"""Base exception cho API errors"""
def __init__(self, message: str, status_code: int = None):
self.message = message
self.status_code = status_code
super().__init__(self.message)
class RollbackExhaustedError(Exception):
"""Khi tất cả fallback đã thất bại"""
passTầng 2: Proxy Layer với Intelligent Routing
Tầng proxy đứng giữa client và provider, xử lý các vấn đề như rate limiting, caching, và health checking tự động.
import time
import hashlib
from typing import Callable, Optional
from collections import defaultdict
import redis.asyncio as redis
class IntelligentProxy:
"""Proxy layer với health check và smart routing"""
def __init__(self, redis_url: str = "redis://localhost:6379"):
self.redis = redis.from_url(redis_url)
self.health_status: Dict[str, bool] = {}
self.response_times: Dict[str, list] = defaultdict(list)
self.health_check_interval = 60 # seconds
self.last_health_check = 0
async def health_check_all_models(self):
"""Kiểm tra sức khỏe tất cả model endpoints"""
test_prompt = "Reply with exactly: OK"
for model_name in ModelTier:
start = time.time()
try:
async with httpx.AsyncClient(timeout=10.0) as client:
response = await client.post(
f"{self.base_url}/chat/completions",
headers={"Authorization": f"Bearer {self.api_key}"},
json={
"model": model_name.value,
"messages": [{"role": "user", "content": test_prompt}],
"max_tokens": 10
}
)
latency = (time.time() - start) * 1000
if response.status_code == 200:
self.health_status[model_name.value] = True
self.response_times[model_name.value].append(latency)
await self._update_redis_health(model_name.value, True, latency)
else:
self.health_status[model_name.value] = False
await self._update_redis_health(model_name.value, False, 0)
except Exception as e:
logger.error(f"Health check failed for {model_name.value}: {e}")
self.health_status[model_name.value] = False
await self._update_redis_health(model_name.value, False, 0)
async def route_request(
self,
request: dict,
prefer_model: Optional[str] = None
) -> dict:
"""Route request đến model phù hợp nhất dựa trên health và latency"""
# Đảm bảo health check được thực hiện định kỳ
if time.time() - self.last_health_check > self.health_check_interval:
await self.health_check_all_models()
self.last_health_check = time.time()
# Tính điểm cho mỗi model
model_scores = []
for model_name in ModelTier:
if not self.health_status.get(model_name.value, False):
continue
avg_latency = self._get_avg_latency(model_name.value)
latency_score = max(0, 100 - avg_latency) # Latency thấp hơn = điểm cao hơn
# Ưu tiên model được chỉ định
priority_bonus = 50 if model_name.value == prefer_model else 0
total_score = latency_score + priority_bonus
model_scores.append({
"model": model_name.value,
"score": total_score,
"avg_latency": avg_latency
})
if not model_scores:
raise AllModelsUnhealthyError("No healthy models available")
# Sắp xếp theo điểm và chọn model tốt nhất
model_scores.sort(key=lambda x: x["score"], reverse=True)
selected_model = model_scores[0]["model"]
return await self._forward_to_model(selected_model, request)
def _get_avg_latency(self, model_name: str) -> float:
"""Lấy độ trễ trung bình từ Redis"""
times = self.response_times.get(model_name, [])
if not times:
return 1000 # Default cao nếu chưa có data
return sum(times[-10:]) / len(times[-10:])
class AllModelsUnhealthyError(Exception):
passTầng 3: Server-Side Circuit Breaker
Pattern Circuit Breaker ngăn chặn cascade failure bằng cách tạm thời ngắt kết nối đến model đang gặp vấn đề.
import asyncio
from enum import Enum
from datetime import datetime, timedelta
from typing import Dict
class CircuitState(Enum):
CLOSED = "closed" # Hoạt động bình thường
OPEN = "open" # Ngắt kết nối, không gọi API
HALF_OPEN = "half_open" # Thử nghiệm phục hồi
class CircuitBreaker:
"""Circuit Breaker pattern cho API calls"""
def __init__(
self,
failure_threshold: int = 5,
recovery_timeout: int = 60,
success_threshold: int = 3
):
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.success_threshold = success_threshold
self.failure_count = 0
self.success_count = 0
self.state = CircuitState.CLOSED
self.last_failure_time: Optional[datetime] = None
async def call(self, func: Callable, *args, **kwargs):
"""Execute function với circuit breaker protection"""
if self.state == CircuitState.OPEN:
if self._should_attempt_reset():
self.state = CircuitState.HALF_OPEN
else:
raise CircuitOpenError(
f"Circuit is OPEN. Retry after "
f"{(self.recovery_timeout - self._time_since_failure())}s"
)
try:
result = await func(*args, **kwargs)
self._on_success()
return result
except Exception as e:
self._on_failure()
raise
def _on_success(self):
"""Xử lý khi call thành công"""
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
logger.info("Circuit breaker reset to CLOSED")
def _on_failure(self):
"""Xử lý khi call thất bại"""
self.failure_count += 1
self.success_count = 0
self.last_failure_time = datetime.now()
if self.failure_count >= self.failure_threshold:
self.state = CircuitState.OPEN
logger.warning(f"Circuit breaker OPENED after {self.failure_count} failures")
def _should_attempt_reset(self) -> bool:
"""Kiểm tra xem có nên thử reset circuit không"""
if self.last_failure_time is None:
return True
return self._time_since_failure() >= self.recovery_timeout
def _time_since_failure(self) -> int:
"""Tính thời gian từ lần failure cuối"""
if self.last_failure_time is None:
return 0
return (datetime.now() - self.last_failure_time).seconds
class CircuitOpenError(Exception):
"""Khi circuit breaker đang OPEN"""
pass
Sử dụng với model cụ thể
model_circuit_breakers: Dict[str, CircuitBreaker] = {}
async def protected_api_call(model_name: str, request: dict):
"""Gọi API với circuit breaker protection"""
if model_name not in model_circuit_breakers:
model_circuit_breakers[model_name] = CircuitBreaker(
failure_threshold=5,
recovery_timeout=60
)
cb = model_circuit_breakers[model_name]
async def _make_call():
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
f"{HOLYSHEEP_BASE_URL}/chat/completions",
headers={"Authorization": f"Bearer {HOLYSHEEP_API_KEY}"},
json={**request, "model": model_name}
)
if response.status_code != 200:
raise APIError(f"HTTP {response.status_code}: {response.text}")
return response.json()
return await cb.call(_make_call)Chiến Lược Rollback Theo Kịch Bản
1. Kịch Bản: Connection Timeout
Đây là lỗi phổ biến nhất khi API provider gặp vấn đề mạng hoặc overload. Với HolySheep AI, độ trễ trung bình dưới 50ms giúp giảm đáng kể timeout, nhưng vẫn cần fallback strategy.
# Ví dụ xử lý timeout với exponential backoff
async def call_with_timeout_and_retry(
prompt: str,
model: str,
max_retries: int = 3,
base_timeout: float = 10.0
) -> dict:
"""Gọi API với timeout adaptive và retry logic"""
last_exception = None
for attempt in range(max_retries):
# Exponential backoff: timeout tăng gấp đôi mỗi lần retry
current_timeout = base_timeout * (2 ** attempt)
try:
async with httpx.AsyncClient(timeout=current_timeout) as client:
response = await client.post(
f"{HOLYSHEEP_BASE_URL}/chat/completions",
headers={
"Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
"Content-Type": "application/json"
},
json={
"model": model,
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.7,
"max_tokens": 2048
}
)
if response.status_code == 200:
return response.json()
elif response.status_code == 401:
raise AuthenticationError("Invalid API key")
elif response.status_code == 429:
await asyncio.sleep(2 ** attempt) # Rate limit backoff
continue
else:
raise APIError(f"HTTP {response.status_code}", response.status_code)
except httpx.TimeoutException as e:
last_exception = TimeoutError(f"Request timeout after {current_timeout}s")
logger.warning(f"Attempt {attempt + 1} timeout: {e}")
await asyncio.sleep(1 * (attempt + 1)) # Wait before retry
except httpx.ConnectError as e:
last_exception = ConnectionError(f"Connection failed: {e}")
logger.error(f"Connection error: {e}")
# Immediate fallback to next model on connection error
raise last_exception
except Exception as e:
last_exception = e
logger.error(f"Unexpected error: {e}")
raise last_exception
Test với mock để xác minh behavior
async def test_timeout_scenario():
"""Simulate timeout và verify fallback hoạt động"""
router = MultiModelRouter(api_key="test_key")
# Mock timeout scenario
async def mock_timeout_request(*args, **kwargs):
raise httpx.TimeoutException("Connection timeout")
# Verify circuit breaker activates
cb = CircuitBreaker(failure_threshold=2)
for _ in range(2):
try:
await cb.call(mock_timeout_request)
except:
pass
assert cb.state == CircuitState.OPEN
print("✓ Circuit breaker correctly opened after timeout threshold")
# Verify router moves to next model
try:
await router.call_with_fallback("Test prompt")
except RollbackExhaustedError:
print("✓ Rollback correctly exhausted after all models failed")
print("✓ All timeout scenarios passed")2. Kịch Bản: 401 Unauthorized / Quota Exceeded
Lỗi authentication và quota là những vấn đề nghiêm trọng cần xử lý khẩn cấp để tránh service disruption.
class AuthAndQuotaHandler:
"""Xử lý authentication và quota errors"""
def __init__(self, api_key: str):
self.api_key = api_key
self.quota_remaining: Optional[int] = None
self.quota_reset_time: Optional[datetime] = None
async def validate_and_monitor_quota(self):
"""Kiểm tra và theo dõi quota usage"""
async with httpx.AsyncClient(timeout=10.0) as client:
# HolySheep AI cung cấp quota details qua headers hoặc API
response = await client.get(
f"{HOLYSHEEP_BASE_URL}/usage",
headers={"Authorization": f"Bearer {self.api_key}"}
)
if response.status_code == 200:
data = response.json()
self.quota_remaining = data.get("remaining", 0)
self.quota_reset_time = datetime.fromisoformat(
data.get("reset_at", datetime.now().isoformat())
)
# Alert nếu quota thấp
if self.quota_remaining < 1000:
await self._send_alert_low_quota()
return self.quota_remaining
else:
raise APIError("Failed to fetch quota", response.status_code)
async def check_api_key_validity(self) -> bool:
"""Verify API key còn valid không"""
async with httpx.AsyncClient(timeout=10.0) as client:
response = await client.post(
f"{HOLYSHEEP_BASE_URL}/chat/completions",
headers={"Authorization": f"Bearer {self.api_key}"},
json={
"model": "deepseek-v3.2",
"messages": [{"role": "user", "content": "Hi"}],
"max_tokens": 5
}
)
if response.status_code == 401:
logger.error("API key is invalid or expired!")
await self._send_alert_invalid_key()
return False
elif response.status_code == 429:
logger.warning("Quota exceeded!")
return True # Key valid nhưng quota hết
elif response.status_code == 200:
return True
else:
logger.warning(f"Unexpected status: {response.status_code}")
return True # Giả định valid trừ khi có 401 rõ ràng
async def _send_alert_low_quota(self):
"""Gửi alert khi quota sắp hết"""
# Integration với Slack, PagerDuty, email, etc.
logger.critical(
f"⚠️ QUOTA ALERT: Only {self.quota_remaining} tokens remaining. "
f"Resets at {self.quota_reset_time}"
)
async def _send_alert_invalid_key(self):
"""Gửi alert khi API key không hợp lệ"""
logger.critical("🚨 CRITICAL: API key invalid or expired!")
# Emergency fallback - chuyển sang backup key hoặc degraded mode
def should_use_budget_model(self) -> bool:
"""Quyết định có nên dùng budget model không"""
if self.quota_remaining is None:
return False
return self.quota_remaining < 5000
Emergency fallback mode khi quota hết
EMERGENCY_MODE_PROMPT = """You are in emergency mode. Provide brief, concise responses only.
Do not use markdown formatting. Maximum 3 sentences per response."""
async def emergency_fallback_request(prompt: str) -> dict:
"""Xử lý request khi quota chính đã hết"""
# Dùng DeepSeek V3.2 ($0.42/MTok) - model rẻ nhất
async with httpx.AsyncClient(timeout=15.0) as client:
response = await client.post(
f"{HOLYSHEEP_BASE_URL}/chat/completions",
headers={"Authorization": f"Bearer {HOLYSHEEP_API_KEY}"},
json={
"model": "deepseek-v3.2",
"messages": [
{"role": "system", "content": EMERGENCY_MODE_PROMPT},
{"role": "user", "content": prompt}
],
"temperature": 0.3, # Lower temp = consistent output
"max_tokens": 200 # Strict limit
}
)
if response.status_code == 200:
return response.json()
else:
# Last resort: return cached response hoặc error message
return {
"error": "Service temporarily degraded",
"status": "emergency_mode"
}3. Kịch Bản: Quality Regression Sau Upgrade
Đôi khi model mới cho chất lượng output kém hơn model cũ. Cần có A/B testing và automatic rollback.
import json
from datetime import datetime
from typing import List, Tuple
class QualityMonitor:
"""Theo dõi và so sánh chất lượng output giữa các model"""
def __init__(self, redis_client):
self.redis = redis_client
self.quality_threshold = 0.85 # 85% quality score tối thiểu
self.min_sample_size = 50 # Số lượng sample tối thiểu để đánh giá
async def compare_models(
self,
test_prompts: List[str],
model_a: str,
model_b: str
) -> dict:
"""So sánh chất lượng 2 model với cùng prompts"""
results = {
"model_a": {"responses": [], "avg_latency": 0, "error_rate": 0},
"model_b": {"responses": [], "avg_latency": 0, "error_rate": 0}
}
errors_a, errors_b = 0, 0
total_latency_a, total_latency_b = 0, 0
for prompt in test_prompts:
# Call model A
try:
start = time.time()
response_a = await self._call_model(model_a, prompt)
latency_a = (time.time() - start) * 1000
results["model_a"]["responses"].append(response_a)
total_latency_a += latency_a
except Exception as e:
errors_a += 1
logger.error(f"Model A error: {e}")
# Call model B
try:
start = time.time()
response_b = await self._call_model(model_b, prompt)
latency_b = (time.time() - start) * 1000
results["model_b"]["responses"].append(response_b)
total_latency_b += latency_b
except Exception as e:
errors_b += 1
logger.error(f"Model B error: {e}")
total = len(test_prompts)
results["model_a"]["avg_latency"] = total_latency_a / total
results["model_a"]["error_rate"] = errors_a / total
results["model_b"]["avg_latency"] = total_latency_b / total
results["model_b"]["error_rate"] = errors_b / total
# Tính quality score dựa trên response characteristics
results["model_a"]["quality_score"] = self._calculate_quality(
results["model_a"]["responses"]
)
results["model_b"]["quality_score"] = self._calculate_quality(
results["model_b"]["responses"]
)
# Quyết định rollback hay keep new model
should_rollback = self._should_rollback(
results["model_a"],
results["model_b"]
)
return {
"comparison": results,
"recommendation": "rollback" if should_rollback else "keep_new",
"confidence": self._calculate_confidence(total)
}
def _calculate_quality(self, responses: List[dict]) -> float:
"""Tính quality score dựa trên various metrics"""
if not responses:
return 0.0
scores = []
for resp in responses:
# Score dựa trên: response length, coherence indicators, etc.
content = resp.get("content", "")
length_score = min(len(content) / 500, 1.0) # Normalize
scores.append(length_score)
return sum(scores) / len(scores)
def _should_rollback(self, old_model: dict, new_model: dict) -> bool:
"""Quyết định có nên rollback không"""
# Rollback nếu quality giảm quá 15%
quality_drop = (
old_model["quality_score"] - new_model["quality_score"]
) / old_model["quality_score"]
if quality_drop > 0.15:
logger.warning(f"Quality drop {quality_drop:.1%} - triggering rollback")
return True
# Rollback nếu error rate tăng đáng kể
error_increase = (
new_model["error_rate"] - old_model["error_rate"]
)
if error_increase > 0.1:
logger.warning(f"Error rate increased - triggering rollback")
return True
return False
def _calculate_confidence(self, sample_size: int) -> float:
"""Tính confidence level của comparison"""
return min(sample_size / self.min_sample_size, 1.0)
async def _call_model(self, model: str, prompt: str) -> dict:
"""Helper để call model"""
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
f"{HOLYSHEEP_BASE_URL}/chat/completions",
headers={"Authorization": f"Bearer {HOLYSHEEP_API_KEY}"},
json={
"model": model,
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.7
}
)
return response.json()
class GradualRollout:
"""Triển khai model mới theo từng giai đoạn với monitoring"""
def __init__(self):
self.rollout_percentages = [1, 5, 25, 50, 100] # Progressive rollout
self.current_phase = 0
self.quality_monitor = None
async def rollout_new_model(
self,
old_model: str,
new_model: str,
test_prompts: List[str]
):
"""Triển khai model mới với các phase"""
for phase_percentage in self.rollout_percentages:
logger.info(f"Starting rollout phase: {phase_percentage}% traffic")
# Áp dụng percentage traffic cho model mới
await self._apply_traffic_split(phase_percentage)
# Chạy comparison test
comparison = await self.quality_monitor.compare_models(
test_prompts=test_prompts,
model_a=old_model,
model_b=new_model
)
if comparison["recommendation"] == "rollback":
logger.warning(
f"Rolling back from {phase_percentage}% phase. "
f"Quality score: {comparison['comparison']['model_b']['quality_score']:.2f}"
)
await self._rollback_to_previous_phase()
return {"status": "rolled_back", "phase": phase_percentage}
logger.info(
f"Phase {phase_percentage}% passed. "
f"New model quality: {comparison['comparison']['model_b']['quality_score']:.2f}"
)
# Đợi một thời gian trước khi chuyển phase
await asyncio.sleep(3600) # 1 hour between phases
return {"status": "completed", "final_phase": "100%"}Lỗi Thường Gặp và Cách Khắc Phục
1. Lỗi: ConnectionError: timeout after 30s
# Nguyên nhân: API server quá tải hoặc network issue
Giải pháp: Implement retry với exponential backoff và fallback
async def handle_timeout_error():
"""Xử lý timeout error một cách graceful"""
MAX_RETRIES = 3
BASE_DELAY = 1.0
for attempt in range(MAX_RETRIES):
try:
# Sử dụng adaptive timeout
timeout = BASE_DELAY * (2 ** attempt)
async with httpx.AsyncClient(timeout=timeout) as client:
response = await client.post(
f"{HOLYSHEEP_BASE_URL}/chat/completions",
headers={"Authorization": f"Bearer {HOLYSHEEP_API_KEY}"},
json={
"model": "deepseek-v3.2",
"messages": [{"role": "user", "content": "test"}],
"max_tokens": 100
}
)
if response.status_code == 200:
return response.json()
except httpx.TimeoutException:
if attempt < MAX_RETRIES - 1:
delay = BASE_DELAY * (2 ** attempt)
logger.info(f"Timeout, retrying in {delay}s...")
await asyncio.sleep(delay)
else:
# Fallback sang model khác hoặc cached response
return await fallback_to_cache()
raise TimeoutError("All retries exhausted")2. Lỗi: 401 Unauthorized - Invalid API Key
# Nguyên nhân: API key hết hạn, bị revoke, hoặc sai key
Giải pháp: Verify key trước request và có backup key
class APIKeyManager:
"""Quản lý nhiều API keys với automatic rotation"""
def __init__(self, primary_key: str, backup_key: str):
self.keys = [primary_key, backup_key]
self.current_key_index = 0