Tưởng tượng bạn đang vận hành một hệ thống chatbot phục vụ hàng nghìn người dùng. Vào lúc 3 giờ sáng, hệ thống báo lỗi ConnectionError: timeout after 30s. Bạn vội vàng kiểm tra dashboard, nhận ra provider chính đã downtime. Nhưng lúc đó, ứng dụng của bạn đã "chết" hoàn toàn — không có đơn hàng nào được xử lý, khách hàng chờ đợi và feedback tiêu cực tràn ngập.
Bài viết này sẽ hướng dẫn bạn xây dựng một hệ thống Fallback thông minh — tự động chuyển đổi sang model dự phòng khi model chính gặp sự cố. Tôi đã triển khai kiến trúc này cho nhiều dự án sản xuất và sẽ chia sẻ những bài học thực chiến quý giá.
Tại sao cần Fallback Mechanism?
Trong thực tế vận hành, tôi đã gặp những tình huống không thể lường trước:
- Provider API downtime — API key hết hạn, rate limit exceeded, server quá tải
- Network partition — Kết nối bị ngắt đột ngột, DNS resolution thất bại
- Model-specific errors — Một số model không hỗ trợ function calling, context length vượt quá giới hạn
- Cost optimization — Tự động chuyển sang model rẻ hơn khi không cần thiết phải dùng model đắt tiền
Kiến trúc Fallback với HolySheep AI
Đăng ký tại đây để trải nghiệm API với độ trễ dưới 50ms và chi phí tiết kiệm đến 85%. HolySheep hỗ trợ đa dạng model với giá cạnh tranh: DeepSeek V3.2 chỉ $0.42/MTok trong khi GPT-4.1 là $8/MTok.
Triển khai Fallback Client — Code mẫu hoàn chỉnh
1. Python Implementation cơ bản
import requests
import time
import logging
from typing import List, Dict, Optional, Any
from dataclasses import dataclass
from enum import Enum
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class ModelTier(Enum):
PREMIUM = "premium" # GPT-4.1, Claude Sonnet 4.5
STANDARD = "standard" # Gemini 2.5 Flash
ECONOMY = "economy" # DeepSeek V3.2
@dataclass
class ModelConfig:
name: str
tier: ModelTier
base_url: str = "https://api.holysheep.ai/v1"
max_tokens: int = 4096
timeout: int = 30
max_retries: int = 3
Cấu hình model theo thứ tự ưu tiên
MODEL_CHAIN = [
ModelConfig(
name="gpt-4.1",
tier=ModelTier.PREMIUM,
max_tokens=8192,
timeout=30
),
ModelConfig(
name="claude-sonnet-4.5",
tier=ModelTier.PREMIUM,
max_tokens=8192,
timeout=35
),
ModelConfig(
name="gemini-2.5-flash",
tier=ModelTier.STANDARD,
max_tokens=4096,
timeout=20
),
ModelConfig(
name="deepseek-v3.2",
tier=ModelTier.ECONOMY,
max_tokens=4096,
timeout=25
),
]
class FallbackLLMClient:
"""Client với cơ chế Fallback tự động"""
def __init__(self, api_key: str, enable_fallback: bool = True):
self.api_key = api_key
self.enable_fallback = enable_fallback
self.current_model_index = 0
self.stats = {
"total_requests": 0,
"fallback_count": 0,
"model_usage": {}
}
def chat(
self,
messages: List[Dict[str, str]],
temperature: float = 0.7,
**kwargs
) -> Dict[str, Any]:
"""
Gửi request với automatic fallback
"""
self.stats["total_requests"] += 1
errors = []
for i, model_config in enumerate(MODEL_CHAIN):
try:
result = self._call_model(model_config, messages, temperature, **kwargs)
# Log thành công
self.stats["model_usage"][model_config.name] = \
self.stats["model_usage"].get(model_config.name, 0) + 1
if i > 0:
self.stats["fallback_count"] += 1
logger.warning(
f"✅ Fallback thành công: {MODEL_CHAIN[0].name} → {model_config.name}"
)
return {
"success": True,
"model": model_config.name,
"tier": model_config.tier.value,
"response": result,
"fallback_used": i > 0
}
except RateLimitError as e:
logger.warning(f"⚠️ Rate limit cho {model_config.name}: {e}")
errors.append(f"{model_config.name}: RateLimitError")
continue
except AuthenticationError as e:
# API key lỗi - không fallback, raise ngay
logger.error(f"❌ Authentication error: {e}")
raise
except ModelNotSupportedError as e:
logger.warning(f"⚠️ Model không hỗ trợ: {model_config.name}")
errors.append(f"{model_config.name}: NotSupported")
continue
except Exception as e:
logger.warning(f"⚠️ Lỗi {model_config.name}: {type(e).__name__}: {e}")
errors.append(f"{model_config.name}: {type(e).__name__}")
continue
# Tất cả model đều thất bại
raise AllModelsFailedError(
f"Tất cả model đều thất bại sau {len(MODEL_CHAIN)} lần thử. "
f"Chi tiết: {errors}"
)
def _call_model(
self,
config: ModelConfig,
messages: List[Dict[str, str]],
temperature: float,
**kwargs
) -> Dict[str, Any]:
"""Gọi API của model cụ thể"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": config.name,
"messages": messages,
"temperature": temperature,
"max_tokens": config.max_tokens,
**kwargs
}
start_time = time.time()
response = requests.post(
f"{config.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=config.timeout
)
latency = (time.time() - start_time) * 1000 # ms
if response.status_code == 401:
raise AuthenticationError("API key không hợp lệ")
if response.status_code == 429:
raise RateLimitError("Rate limit exceeded")
if response.status_code == 400:
error_detail = response.json()
if "not supported" in str(error_detail).lower():
raise ModelNotSupportedError(f"Model không hỗ trợ feature này")
raise ValueError(f"Bad request: {error_detail}")
if response.status_code != 200:
raise APIError(f"HTTP {response.status_code}: {response.text}")
return {
"content": response.json()["choices"][0]["message"]["content"],
"latency_ms": round(latency, 2),
"usage": response.json().get("usage", {})
}
def get_stats(self) -> Dict[str, Any]:
"""Lấy thống kê sử dụng"""
fallback_rate = (
self.stats["fallback_count"] / self.stats["total_requests"] * 100
if self.stats["total_requests"] > 0 else 0
)
return {
**self.stats,
"fallback_rate": f"{fallback_rate:.2f}%"
}
Custom Exceptions
class RateLimitError(Exception):
pass
class AuthenticationError(Exception):
pass
class ModelNotSupportedError(Exception):
pass
class APIError(Exception):
pass
class AllModelsFailedError(Exception):
pass2. Async Implementation cho High-Throughput
import asyncio
import aiohttp
from typing import List, Dict, Optional, Any
from dataclasses import dataclass
import time
@dataclass
class AsyncModelConfig:
name: str
tier: str
base_url: str = "https://api.holysheep.ai/v1"
max_tokens: int = 4096
timeout: float = 30.0
priority: int = 1 # Thứ tự ưu tiên (số càng nhỏ càng ưu tiên)
Cấu hình với async models
ASYNC_MODEL_CHAIN = [
AsyncModelConfig(
name="gpt-4.1",
tier="premium",
max_tokens=8192,
timeout=30.0,
priority=1
),
AsyncModelConfig(
name="gemini-2.5-flash",
tier="standard",
max_tokens=4096,
timeout=20.0,
priority=2
),
AsyncModelConfig(
name="deepseek-v3.2",
tier="economy",
max_tokens=4096,
timeout=25.0,
priority=3
),
]
class AsyncFallbackClient:
"""
Async client với parallel fallback và circuit breaker pattern
"""
def __init__(self, api_key: str):
self.api_key = api_key
self._circuit_breakers: Dict[str, "CircuitBreaker"] = {}
# Khởi tạo circuit breaker cho mỗi model
for config in ASYNC_MODEL_CHAIN:
self._circuit_breakers[config.name] = CircuitBreaker(
failure_threshold=5,
recovery_timeout=60.0,
expected_exception=Exception
)
async def chat_completion(
self,
messages: List[Dict[str, str]],
model_preference: Optional[str] = None,
**kwargs
) -> Dict[str, Any]:
"""
Async chat completion với fallback thông minh
"""
# Nếu chỉ định model cụ thể
if model_preference:
models_to_try = [
m for m in ASYNC_MODEL_CHAIN if m.name == model_preference
] or ASYNC_MODEL_CHAIN
else:
models_to_try = ASYNC_MODEL_CHAIN
errors = []
for model_config in models_to_try:
circuit_breaker = self._circuit_breakers[model_config.name]
# Kiểm tra circuit breaker
if not circuit_breaker.can_execute():
logger.info(f"Circuit breaker OPEN cho {model_config.name}")
errors.append(f"{model_config.name}: CircuitOpen")
continue
try:
start = time.time()
result = await self._async_call(model_config, messages, **kwargs)
latency = (time.time() - start) * 1000
# Reset circuit breaker khi thành công
circuit_breaker.record_success()
return {
"success": True,
"model": model_config.name,
"tier": model_config.tier,
"latency_ms": round(latency, 2),
"content": result["content"],
"usage": result.get("usage", {})
}
except Exception as e:
circuit_breaker.record_failure()
errors.append(f"{model_config.name}: {type(e).__name__}")
continue
raise AllModelsFailedError(f"Failed after trying {len(errors)} models: {errors}")
async def _async_call(
self,
config: AsyncModelConfig,
messages: List[Dict[str, str]],
temperature: float = 0.7,
**kwargs
) -> Dict[str, Any]:
"""Async HTTP call đến API"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": config.name,
"messages": messages,
"temperature": temperature,
"max_tokens": config.max_tokens,
**kwargs
}
timeout = aiohttp.ClientTimeout(total=config.timeout)
async with aiohttp.ClientSession(timeout=timeout) as session:
async with session.post(
f"{config.base_url}/chat/completions",
headers=headers,
json=payload
) as response:
if response.status == 401:
raise AuthenticationError("Invalid API key")
if response.status == 429:
raise RateLimitError("Rate limit exceeded")
if response.status >= 500:
raise ServerError(f"Server error: {response.status}")
data = await response.json()
return {
"content": data["choices"][0]["message"]["content"],
"usage": data.get("usage", {}),
"model": data.get("model", config.name)
}
class CircuitBreaker:
"""
Circuit Breaker Pattern - Ngăn chặn cascade failure
"""
def __init__(
self,
failure_threshold: int = 5,
recovery_timeout: float = 60.0,
expected_exception: type = Exception
):
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.expected_exception = expected_exception
self.failure_count = 0
self.last_failure_time: Optional[float] = None
self.state = "CLOSED" # CLOSED, OPEN, HALF_OPEN
def can_execute(self) -> bool:
if self.state == "CLOSED":
return True
if self.state == "OPEN":
if self.last_failure_time:
elapsed = time.time() - self.last_failure_time
if elapsed >= self.recovery_timeout:
self.state = "HALF_OPEN"
return True
return False
# HALF_OPEN - cho phép một request thử
return True
def record_success(self):
self.failure_count = 0
self.state = "CLOSED"
def record_failure(self):
self.failure_count += 1
self.last_failure_time = time.time()
if self.failure_count >= self.failure_threshold:
self.state = "OPEN"
print(f"Circuit breaker OPENED after {self.failure_count} failures")
Sử dụng
async def main():
client = AsyncFallbackClient(api_key="YOUR_HOLYSHEEP_API_KEY")
messages = [
{"role": "system", "content": "Bạn là trợ lý AI thông minh."},
{"role": "user", "content": "Giải thích khái niệm Fallback Mechanism"}
]
try:
result = await client.chat_completion(
messages,
temperature=0.7
)
print(f"✅ Response từ {result['model']}:")
print(f" Latency: {result['latency_ms']}ms")
print(f" Content: {result['content'][:100]}...")
except AllModelsFailedError as e:
print(f"❌ Tất cả models đều thất bại: {e}")
if __name__ == "__main__":
asyncio.run(main())Tính năng nâng cao: Smart Routing
from enum import Enum
from typing import Callable, Any
import hashlib
class RequestType(Enum):
COMPLEX_REASONING = "complex_reasoning" # Cần model mạnh
GENERAL_CHAT = "general_chat" # Model tiêu chuẩn
BATCH_PROCESSING = "batch_processing" # Model rẻ, xử lý nhiều
REAL_TIME = "real_time" # Latency thấp là ưu tiên
class SmartRouter:
"""
Router thông minh - Chọn model phù hợp dựa trên yêu cầu
"""
def __init__(self, client: FallbackLLMClient):
self.client = client
# Mapping request type -> model preferences
self.route_map = {
RequestType.COMPLEX_REASONING: [
"gpt-4.1", "claude-sonnet-4.5", "gemini-2.5-flash"
],
RequestType.GENERAL_CHAT: [
"gemini-2.5-flash", "deepseek-v3.2"
],
RequestType.BATCH_PROCESSING: [
"deepseek-v3.2", "gemini-2.5-flash"
],
RequestType.REAL_TIME: [
"gemini-2.5-flash", "deepseek-v3.2"
],
}
# Cost tracking
self.cost_per_1k_tokens = {
"gpt-4.1": 8.0,
"claude-sonnet-4.5": 15.0,
"gemini-2.5-flash": 2.50,
"deepseek-v3.2": 0.42,
}
self.total_cost = 0.0
self.total_tokens = 0
def detect_request_type(
self,
messages: list,
query: str = ""
) -> RequestType:
"""
Tự động phát hiện loại request dựa trên nội dung
"""
# Mã hóa query để phân tích
query_lower = query.lower()
combined = " ".join([
m.get("content", "").lower()
for m in messages
])
# Complex reasoning keywords
complex_keywords = [
"phân tích", "đánh giá", "so sánh", "lập luận",
"chứng minh", "suy luận", "reasoning", "analyze"
]
# Real-time keywords
realtime_keywords = [
"ngay lập tức", "nhanh", "tức thì", "gấp",
"urgent", "immediately", "asap"
]
# Batch keywords
batch_keywords = [
"xử lý hàng loạt", "batch", "danh sách", "nhiều",
"tất cả", "loop", "iterate"
]
if any(kw in combined for kw in complex_keywords):
return RequestType.COMPLEX_REASONING
if any(kw in query_lower for kw in realtime_keywords):
return RequestType.REAL_TIME
if any(kw in combined for kw in batch_keywords):
return RequestType.BATCH_PROCESSING
return RequestType.GENERAL_CHAT
def chat_with_routing(
self,
messages: list,
prefer_type: RequestType = None,
max_cost_per_request: float = 0.10,
**kwargs
) -> dict:
"""
Chat với smart routing - Tự động chọn model tối ưu
"""
# Bước 1: Xác định request type
request_type = prefer_type or self.detect_request_type(messages)
# Bước 2: Lấy danh sách model phù hợp
available_models = self.route_map[request_type]
# Bước 3: Thử các model theo thứ tự ưu tiên
for model_name in available_models:
model_config = next(
(m for m in MODEL_CHAIN if m.name == model_name),
None
)
if not model_config:
continue
try:
result = self.client.chat(messages, **kwargs)
# Tính cost
usage = result.get("response", {}).get("usage", {})
prompt_tokens = usage.get("prompt_tokens", 0)
completion_tokens = usage.get("completion_tokens", 0)
total_tokens = prompt_tokens + completion_tokens
cost = (total_tokens / 1000) * self.cost_per_1k_tokens[model_name]
# Kiểm tra budget
if cost > max_cost_per_request:
continue
# Update stats
self.total_cost += cost
self.total_tokens += total_tokens
return {
**result,
"request_type": request_type.value,
"cost_usd": round(cost, 4),
"total_cost_usd": round(self.total_cost, 4)
}
except Exception as e:
continue
raise Exception(f"Không có model nào phù hợp cho request type: {request_type}")
Ví dụ sử dụng
def demo():
client = FallbackLLMClient(api_key="YOUR_HOLYSHEEP_API_KEY")
router = SmartRouter(client)
# Test case 1: Complex reasoning
messages = [
{"role": "user", "content": "Phân tích ưu nhược điểm của microservices vs monolith architecture"}
]
result = router.chat_with_routing(messages)
print(f"Request Type: {result['request_type']}")
print(f"Model used: {result['model']}")
print(f"Cost: ${result['cost_usd']}")
print(f"Total cost: ${result['total_cost_usd']}")
# Test case 2: Real-time
messages = [
{"role": "user", "content": "Trả lời ngay: 1 + 1 = ?"}
]
result = router.chat_with_routing(messages, prefer_type=RequestType.REAL_TIME)
print(f"Request Type: {result['request_type']}")
print(f"Model used: {result['model']}")
if __name__ == "__main__":
demo()Monitoring và Alerting
import json
import smtplib
from datetime import datetime, timedelta
from dataclasses import dataclass, field
from typing import List, Dict
from collections import defaultdict
@dataclass
class FallbackMetrics:
"""Theo dõi metrics cho fallback system"""
timestamp: datetime = field(default_factory=datetime.now)
model_name: str = ""
request_id: str = ""
latency_ms: float = 0
success: bool = False
fallback_triggered: bool = False
error_type: str = ""
cost_usd: float = 0
class FallbackMonitor:
"""
Monitoring system cho fallback mechanism
"""
def __init__(self):
self.metrics: List[FallbackMetrics] = []
self.alerts: List[Dict] = []
# Thresholds
self.fallback_rate_threshold = 0.15 # 15%
self.latency_p95_threshold = 5000 # 5000ms
self.error_rate_threshold = 0.05 # 5%
def record(self, metrics: FallbackMetrics):
"""Ghi nhận một request"""
self.metrics.append(metrics)
# Kiểm tra alert conditions
self._check_alerts(metrics)
def _check_alerts(self, metrics: FallbackMetrics):
"""Kiểm tra điều kiện alert"""
now = datetime.now()
window_start = now - timedelta(hours=1)
recent_metrics = [
m for m in self.metrics
if m.timestamp >= window_start
]
if not recent_metrics:
return
# Tính fallback rate
fallback_count = sum(1 for m in recent_metrics if m.fallback_triggered)
fallback_rate = fallback_count / len(recent_metrics)
if fallback_rate > self.fallback_rate_threshold:
self.alerts.append({
"type": "HIGH_FALLBACK_RATE",
"severity": "WARNING",
"message": f"Fallback rate {fallback_rate:.1%} vượt ngưỡng {self.fallback_rate_threshold:.1%}",
"time": now.isoformat(),
"models": self._get_model_failure_stats(recent_metrics)
})
# Tính error rate
error_count = sum(1 for m in recent_metrics if not m.success)
error_rate = error_count / len(recent_metrics)
if error_rate > self.error_rate_threshold:
self.alerts.append({
"type": "HIGH_ERROR_RATE",
"severity": "CRITICAL",
"message": f"Error rate {error_rate:.1%} vượt ngưỡng {self.error_rate_threshold:.1%}",
"time": now.isoformat()
})
def _get_model_failure_stats(self, metrics_list: List[FallbackMetrics]) -> Dict:
"""Thống kê lỗi theo model"""
stats = defaultdict(lambda: {"total": 0, "errors": 0})
for m in metrics_list:
stats[m.model_name]["total"] += 1
if not m.success:
stats[m.model_name]["errors"] += 1
return {
model: {
"total_requests": data["total"],
"error_count": data["errors"],
"error_rate": data["errors"] / data["total"] if data["total"] > 0 else 0
}
for model, data in stats.items()
}
def get_dashboard_data(self) -> Dict:
"""Generate data cho dashboard"""
now = datetime.now()
last_24h = now - timedelta(hours=24)
last_1h = now - timedelta(hours=1)
metrics_24h = [m for m in self.metrics if m.timestamp >= last_24h]
metrics_1h = [m for m in self.metrics if m.timestamp >= last_1h]
# Model usage
model_usage = defaultdict(int)
for m in metrics_24h:
model_usage[m.model_name] += 1
# Calculate latency P95
latencies = sorted([m.latency_ms for m in metrics_1h])
p95_index = int(len(latencies) * 0.95)
latency_p95 = latencies[p95_index] if latencies else 0
# Calculate total cost
total_cost = sum(m.cost_usd for m in metrics_24h)
total_tokens = sum(
m.latency_ms for m in metrics_24h # placeholder
)
return {
"summary": {
"total_requests_24h": len(metrics_24h),
"total_requests_1h": len(metrics_1h),
"fallback_rate_1h": self._calculate_fallback_rate(metrics_1h),
"error_rate_1h": self._calculate_error_rate(metrics_1h),
"latency_p95_1h_ms": round(latency_p95, 2),
"total_cost_24h_usd": round(total_cost, 4)
},
"model_usage": dict(model_usage),
"recent_alerts": self.alerts[-10:],
"timestamp": now.isoformat()
}
def _calculate_fallback_rate(self, metrics_list: List[FallbackMetrics]) -> float:
if not metrics_list:
return 0
return sum(1 for m in metrics_list if m.fallback_triggered) / len(metrics_list)
def _calculate_error_rate(self, metrics_list: List[FallbackMetrics]) -> float:
if not metrics_list:
return 0
return sum(1 for m in metrics_list if not m.success) / len(metrics_list)
def send_alert_email(self, smtp_config: Dict, recipients: List[str]):
"""Gửi email alert"""
if not self.alerts:
return
subject = f"[HolySheep AI] Fallback Alert - {len(self.alerts)} issues"
body = "Fallback System Alerts:\n\n"
for alert in self.alerts[-5:]:
body += f"[{alert['severity']}] {alert['type']}\n"
body += f"Message: {alert['message']}\n"
body += f"Time: {alert['time']}\n\n"
# Gửi email logic here
print(f"Alert Email:\nSubject: {subject}\nBody: {body}")
Sử dụng monitor
monitor = FallbackMonitor()
Record metrics
monitor.record(FallbackMetrics(
model_name="gpt-4.1",
request_id="req_001",
latency_ms=1250.5,
success=True,
fallback_triggered=False,
cost_usd=0.0025
))
monitor.record(FallbackMetrics(
model_name="gemini-2.5-flash",
request_id="req_002",
latency_ms=890.2,
success=True,
fallback_triggered=True,
cost_usd=0.0012
))
Get dashboard data
dashboard = monitor.get_dashboard_data()
print(json.dumps(dashboard, indent=2))Bảng giá tham khảo 2026
| Model | Giá/MTok | Đề xuất sử dụng |
|---|---|---|
| GPT-4.1 | $8.00 | Complex reasoning, coding |
| Claude Sonnet 4.5 | $15.00 | Long-form writing, analysis |
| Gemini 2.5 Flash | $2.50 | General chat, real-time |
| DeepSeek V3.2 | $0.42 | Batch processing, cost-saving |
Tỷ giá: ¥1 = $1. Thanh toán qua WeChat/Alipay. Độ trễ trung bình <50ms.
Lỗi thường gặp và cách khắc phục
1. Lỗi 401 Unauthorized — API Key không hợp lệ
# ❌ Sai cách - Hardcode API key trong code
client = FallbackLLMClient(api_key="sk-xxxxx...")
✅ Đúng cách - Sử dụng environment variable
import os
from dotenv import load_dotenv
load_dotenv()
api_key = os.getenv("HOLYSHEEP_API_KEY")
if not api_key:
raise ValueError("HOLYSHEEP_API_KEY không được set trong environment")
client = FallbackLLMClient(api_key=api_key)
Hoặc sử dụng config file (.env)
HOLYSHEEP_API_KEY=your_key_here
HOLYSHEEP_BASE_URL=https://api.holysheep.ai/v1
Nguyên nhân: API key bị hết hạn, sai format, hoặc chưa được kích hoạt. Khắc phục: Kiểm tra lại API key tại dashboard HolySheep, đảm bảo còn credits.
2. Lỗi 429 Rate Limit — Vượt quá giới hạn request
# ❌ Sai cách - Retry ngay lập tức khi rate limit
for _ in range(10):
try:
response = client.chat(messages)
break
except RateLimitError:
continue # Vòng lặp liên tục, không backoff
✅ Đúng cách - Exponential backoff
import time
import random
def chat_with_backoff(client, messages, max_retries=5):
for attempt in range(max_retries):
try:
return client.chat(messages)
except RateLimitError as e:
if attempt == max_retries - 1:
raise
# Exponential backoff: 1s, 2s, 4s, 8s, 16s + jitter
wait_time = (2 ** attempt) + random.uniform(0, 1)
print(f"Rate limit hit. Retry sau {wait_time:.2f}s...")
time.sleep(wait_time)
except Exception as e:
raise
Hoặc sử dụng tenacity library
from tenacity import (
retry, stop_after_attempt, wait_exponential,
retry_if_exception_type
)
@retry(
stop=stop_after_attempt(5),
wait=wait_exponential(multiplier=1, min=1, max=60),
retry=retry_if_exception_type(RateLimitError)
)
def chat_with_tenacity(client, messages):
return client.chat(messages)Nguyên nhân: Gửi quá nhiều request trong thời gian ngắn. Khắc phục: Triển khai exponential backoff, giới hạn rate request hoặc nâng cấp gói subscription.
3. Lỗi Connection Timeout — Server không phản hồi
# ❌ Sai cách - Timeout quá ngắn hoặc không set
response = requests.post(