Kịch bản lỗi thực tế: "ConnectionError: timeout" ở production
23:47 ngày 15/01/2026, hệ thống chatbot AI của tôi bị sập hoàn toàn. logs ghi nhận hàng nghìn dòng:
ERROR - ConnectionError: timeout after 30s
ERROR - HTTP 503 Service Unavailable
ERROR - RateLimitError: 429 Too Many Requests
ERROR - httpx.ReadTimeout: Gateway timeout
P99 latency tăng từ 200ms lên 45,000ms
Error rate từ 0.1% lên 89%
Queue overflow - 12,847 requests bị drop
Nguyên nhân gốc: Một feature mới tung ra đã tạo 50,000 requests/giây lên API AI, trong khi hệ thống chỉ handle được 1,000 requests/giây. Không có rate limiting, không có circuit breaker, không có fallback strategy. Toàn bộ hệ thống downstream bị chôn vùi.
Bài học đắt giá này dẫn tôi đến việc xây dựng một kiến trúc API Gateway hoàn chỉnh cho các ứng dụng AI. Trong bài viết này, tôi sẽ chia sẻ cách thiết kế hệ thống có thể xử lý 100,000+ concurrent requests mà vẫn duy trì latency dưới 100ms.
Tại sao cần API Gateway cho AI?
Khi sử dụng HolySheep AI - nền tảng API AI với chi phí tiết kiệm 85%+ (chỉ ¥1=$1), bạn cần một gateway để:
- Bảo vệ API key và quản lý authentication
- Implement rate limiting theo tier của người dùng
- Cân bằng tải giữa nhiều model providers
- Tự động fallback khi provider gặp sự cố
- Cache responses để giảm chi phí
Kiến trúc tổng quan
┌─────────────────────────────────────────────────────────────────┐
│ CLIENT REQUESTS │
│ (10,000 - 100,000 rps) │
└────────────────────────────┬────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ NGINX LOAD BALANCER │
│ (Layer 7 - HTTP/2, SSL Termination) │
│ Round Robin / Least Connections │
└────────────────────────────┬────────────────────────────────────┘
│
┌──────────────┴──────────────┐
▼ ▼
┌─────────────────────────┐ ┌─────────────────────────┐
│ API GATEWAY NODE 1 │ │ API GATEWAY NODE 2 │
│ ┌───────────────────┐ │ │ ┌───────────────────┐ │
│ │ Rate Limiter │ │ │ │ Rate Limiter │ │
│ │ (Token Bucket) │ │ │ │ (Token Bucket) │ │
│ └───────────────────┘ │ │ └───────────────────┘ │ │
│ ┌───────────────────┐ │ │ ┌───────────────────┐ │ │
│ │ Circuit Breaker │ │ │ │ Circuit Breaker │ │ │
│ │ (Half-Open) │ │ │ │ (Half-Open) │ │ │
│ └───────────────────┘ │ │ └───────────────────┘ │ │
│ ┌───────────────────┐ │ │ ┌───────────────────┐ │ │
│ │ Request Router │ │ │ │ Request Router │ │ │
│ └───────────────────┘ │ │ └───────────────────┘ │ │
└─────────────────────────┘ └─────────────────────────┘
│ │
└──────────────┬──────────────┘
│
┌──────────────┴──────────────┐
▼ ▼
┌─────────────────────────┐ ┌─────────────────────────┐
│ HolySheep AI API │ │ Fallback Provider │
│ api.holysheep.ai/v1 │ │ (Backup Model) │
│ GPT-4.1: $8/MTok │ │ DeepSeek: $0.42/MTok │
│ Claude: $15/MTok │ │ │
└─────────────────────────┘ └─────────────────────────┘
1. Cài đặt Rate Limiter với Token Bucket Algorithm
Rate limiting là lớp bảo vệ đầu tiên. Tôi sử dụng Token Bucket algorithm - cho phép burst traffic nhưng vẫn kiểm soát tổng consumption.
# rate_limiter.py
import time
import asyncio
from typing import Dict, Optional
from dataclasses import dataclass, field
from collections import defaultdict
import hashlib
@dataclass
class TokenBucket:
"""Token Bucket implementation cho rate limiting"""
capacity: int # Max tokens trong bucket
refill_rate: float # Tokens refill per second
tokens: float = field(init=False)
last_refill: float = field(init=False)
def __post_init__(self):
self.tokens = float(self.capacity)
self.last_refill = time.time()
def _refill(self):
"""Refill tokens based on elapsed time"""
now = time.time()
elapsed = now - self.last_refill
self.tokens = min(self.capacity, self.tokens + elapsed * self.refill_rate)
self.last_refill = now
def consume(self, tokens: int = 1) -> bool:
"""Attempt to consume tokens, return True if successful"""
self._refill()
if self.tokens >= tokens:
self.tokens -= tokens
return True
return False
class AIRateLimiter:
"""
Distributed rate limiter cho AI API Gateway
Hỗ trợ tiered rate limits và concurrent request tracking
"""
# Tier definitions (requests per minute)
TIERS = {
'free': {'rpm': 60, 'tpm': 10000, 'rpd': 1000},
'basic': {'rpm': 500, 'tpm': 100000, 'rpd': 50000},
'pro': {'rpm': 2000, 'tpm': 500000, 'rpd': 500000},
'enterprise': {'rpm': 10000, 'tpm': 2000000, 'rpd': -1} # Unlimited
}
def __init__(self, redis_client=None):
self.buckets: Dict[str, Dict[str, TokenBucket]] = defaultdict(dict)
self.daily_usage: Dict[str, int] = defaultdict(int)
self.concurrent_requests: Dict[str, int] = defaultdict(int)
self.max_concurrent = 100
self.redis = redis_client
def _get_user_tier(self, api_key: str) -> str:
"""Determine user tier from API key"""
if not api_key or api_key == 'YOUR_HOLYSHEEP_API_KEY':
return 'free'
# Hash key để identify user
key_hash = hashlib.sha256(api_key.encode()).hexdigest()[:8]
# Demo tier assignment (thực tế check database)
if key_hash > 'f' * 8:
return 'enterprise'
elif key_hash > 'c' * 8:
return 'pro'
elif key_hash > '5' * 8:
return 'basic'
return 'free'
async def check_rate_limit(
self,
api_key: str,
estimated_tokens: int = 1000
) -> tuple[bool, dict]:
"""
Check if request is within rate limits
Returns: (allowed: bool, info: dict)
"""
tier_name = self._get_user_tier(api_key)
tier = self.TIERS[tier_name]
user_id = hashlib.sha256(api_key.encode()).hexdigest()[:16]
now = time.time()
# Initialize buckets for user
if user_id not in self.buckets:
self.buckets[user_id] = {
'rpm': TokenBucket(capacity=tier['rpm'], refill_rate=tier['rpm']/60),
'tpm': TokenBucket(capacity=tier['tpm'], refill_rate=tier['tpm']/60),
}
buckets = self.buckets[user_id]
# Check concurrent requests
if self.concurrent_requests[user_id] >= self.max_concurrent:
return False, {
'error': 'concurrent_limit_exceeded',
'message': f'Max {self.max_concurrent} concurrent requests',
'retry_after': 5
}
# Check RPM (requests per minute)
if not buckets['rpm'].consume(1):
return False, {
'error': 'rate_limit_exceeded',
'message': f'RPM limit ({tier["rpm"]}) exceeded',
'retry_after': 60
}
# Check TPM (tokens per minute)
if not buckets['tpm'].consume(estimated_tokens):
return False, {
'error': 'token_limit_exceeded',
'message': f'TPM limit ({tier["tpm"]}) exceeded',
'retry_after': 60
}
# Check daily limit
if tier['rpd'] > 0 and self.daily_usage[user_id] >= tier['rpd']:
return False, {
'error': 'daily_limit_exceeded',
'message': f'Daily limit ({tier["rpd"]}) exceeded',
'retry_after': self._seconds_until_midnight()
}
# Track concurrent
self.concurrent_requests[user_id] += 1
return True, {
'tier': tier_name,
'remaining_rpm': int(buckets['rpm'].tokens),
'remaining_tpm': int(buckets['tpm'].tokens),
'daily_used': self.daily_usage[user_id]
}
def release(self, api_key: str, tokens_used: int = 0):
"""Release concurrent slot và update daily usage"""
user_id = hashlib.sha256(api_key.encode()).hexdigest()[:16]
if self.concurrent_requests[user_id] > 0:
self.concurrent_requests[user_id] -= 1
if tokens_used > 0:
self.daily_usage[user_id] += tokens_used
def _seconds_until_midnight(self) -> int:
"""Calculate seconds until midnight UTC"""
now = time.time()
midnight = int(now) + 86400 - (int(now) % 86400)
return midnight - int(now)
Usage example
async def main():
limiter = AIRateLimiter()
# Simulate request
allowed, info = await limiter.check_rate_limit(
api_key='YOUR_HOLYSHEEP_API_KEY',
estimated_tokens=500
)
if allowed:
print(f"✓ Request allowed: {info}")
else:
print(f"✗ Rate limited: {info['message']}")
print(f" Retry after: {info['retry_after']}s")
# Release after processing
limiter.release('YOUR_HOLYSHEEP_API_KEY', tokens_used=500)
if __name__ == '__main__':
asyncio.run(main())
2. Circuit Breaker Pattern - Ngăn chặn Cascade Failure
Circuit Breaker là pattern quan trọng nhất mà tôi đã thiếu trong kiến trúc cũ. Nó ngăn chặn một provider bị chôn vùi bởi quá nhiều requests khi đã có dấu hiệu failure.
# circuit_breaker.py
import asyncio
import time
from enum import Enum
from typing import Callable, Any, Optional
from dataclasses import dataclass
import logging
import httpx
logger = logging.getLogger(__name__)
class CircuitState(Enum):
CLOSED = "closed" # Normal operation
OPEN = "open" # Failing, reject requests
HALF_OPEN = "half_open" # Testing if recovered
@dataclass
class CircuitBreakerConfig:
failure_threshold: int = 5 # Failures before opening
success_threshold: int = 3 # Successes in half-open to close
timeout: float = 30.0 # Seconds before half-open
half_open_max_calls: int = 3 # Max calls in half-open state
latency_percentile: float = 0.95 # P95 latency threshold
latency_threshold_ms: float = 5000 # Consider slow if > 5s
class CircuitBreaker:
"""
Circuit Breaker implementation for AI API calls
States:
- CLOSED: Normal operation, all requests pass through
- OPEN: Circuit is tripped, all requests fail fast
- HALF_OPEN: Testing recovery with limited requests
"""
def __init__(self, name: str, config: CircuitBreakerConfig):
self.name = name
self.config = config
self.state = CircuitState.CLOSED
self.failure_count = 0
self.success_count = 0
self.last_failure_time: Optional[float] = None
self.half_open_calls = 0
# Metrics
self.total_calls = 0
self.successful_calls = 0
self.failed_calls = 0
self.rejected_calls = 0
self.latencies: list = []
def _should_allow_request(self) -> bool:
"""Check if request should be allowed based on current state"""
if self.state == CircuitState.CLOSED:
return True
if self.state == CircuitState.OPEN:
# Check if timeout has passed
if time.time() - self.last_failure_time >= self.config.timeout:
self._transition_to_half_open()
return True
return False
# HALF_OPEN state
if self.half_open_calls < self.config.half_open_max_calls:
self.half_open_calls += 1
return True
return False
def _transition_to_half_open(self):
"""Move to half-open state after timeout"""
self.state = CircuitState.HALF_OPEN
self.half_open_calls = 0
self.success_count = 0
logger.info(f"Circuit {self.name}: OPEN -> HALF_OPEN")
def _transition_to_closed(self):
"""Move to closed state after successful recovery"""
self.state = CircuitState.CLOSED
self.failure_count = 0
self.success_count = 0
logger.info(f"Circuit {self.name}: HALF_OPEN -> CLOSED (recovered)")
def _transition_to_open(self):
"""Move to open state after too many failures"""
self.state = CircuitState.OPEN
self.last_failure_time = time.time()
logger.warning(f"Circuit {self.name}: CLOSED -> OPEN (failure threshold reached)")
def record_success(self, latency_ms: float):
"""Record a successful call"""
self.total_calls += 1
self.successful_calls += 1
self.latencies.append(latency_ms)
# Keep only recent latencies
if len(self.latencies) > 1000:
self.latencies = self.latencies[-500:]
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
def record_failure(self, error_type: str):
"""Record a failed call"""
self.total_calls += 1
self.failed_calls += 1
self.last_failure_time = time.time()
logger.error(f"Circuit {self.name}: Call failed ({error_type})")
if self.state == CircuitState.HALF_OPEN:
self._transition_to_open()
elif self.state == CircuitState.CLOSED:
self.failure_count += 1
if self.failure_count >= self.config.failure_threshold:
self._transition_to_open()
def record_timeout(self, latency_ms: float):
"""Record a timeout (treated as failure)"""
self.record_failure(f"timeout ({latency_ms}ms)")
def get_stats(self) -> dict:
"""Get current circuit statistics"""
p95 = 0
if self.latencies:
sorted_latencies = sorted(self.latencies)
idx = int(len(sorted_latencies) * 0.95)
p95 = sorted_latencies[idx] if idx < len(sorted_latencies) else sorted_latencies[-1]
return {
'name': self.name,
'state': self.state.value,
'total_calls': self.total_calls,
'success_rate': self.successful_calls / self.total_calls if self.total_calls > 0 else 0,
'rejected_calls': self.rejected_calls,
'p95_latency_ms': round(p95, 2),
'failure_count': self.failure_count,
'time_until_half_open': max(0, self.config.timeout - (time.time() - self.last_failure_time))
if self.last_failure_time else 0
}
class AIGatewayWithCircuitBreaker:
"""AI Gateway with circuit breaker pattern"""
def __init__(self):
self.circuits: dict[str, CircuitBreaker] = {
'holysheep_gpt4': CircuitBreaker(
'holysheep_gpt4',
CircuitBreakerConfig(failure_threshold=3, timeout=30)
),
'holysheep_claude': CircuitBreaker(
'holysheep_claude',
CircuitBreakerConfig(failure_threshold=3, timeout=30)
),
'holysheep_deepseek': CircuitBreaker(
'holysheep_deepseek',
CircuitBreakerConfig(failure_threshold=5, timeout=15)
),
}
async def call_with_circuit_breaker(
self,
circuit_name: str,
api_key: str,
model: str,
messages: list,
temperature: float = 0.7,
max_tokens: int = 1000
) -> dict:
"""
Make AI API call with circuit breaker protection
"""
circuit = self.circuits.get(circuit_name)
if not circuit:
raise ValueError(f"Unknown circuit: {circuit_name}")
if not circuit._should_allow_request():
circuit.rejected_calls += 1
stats = circuit.get_stats()
raise Exception(
f"Circuit {circuit_name} is {stats['state']}. "
f"Retry after {int(stats['time_until_half_open'])}s"
)
start_time = time.time()
try:
async with httpx.AsyncClient(timeout=60.0) as client:
response = await client.post(
'https://api.holysheep.ai/v1/chat/completions',
headers={
'Authorization': f'Bearer {api_key}',
'Content-Type': 'application/json'
},
json={
'model': model,
'messages': messages,
'temperature': temperature,
'max_tokens': max_tokens
}
)
latency_ms = (time.time() - start_time) * 1000
if response.status_code == 200:
circuit.record_success(latency_ms)
return response.json()
else:
circuit.record_failure(f"HTTP {response.status_code}")
return {'error': response.text, 'status_code': response.status_code}
except httpx.TimeoutException:
latency_ms = (time.time() - start_time) * 1000
circuit.record_timeout(latency_ms)
raise Exception(f"Request timeout after {latency_ms}ms")
except httpx.ConnectError as e:
circuit.record_failure("connection_error")
raise Exception(f"Connection failed: {str(e)}")
Test circuit breaker
async def test_circuit_breaker():
gateway = AIGatewayWithCircuitBreaker()
# Check circuit stats
for name, circuit in gateway.circuits.items():
stats = circuit.get_stats()
print(f"{name}: {stats['state']} - P95: {stats['p95_latency_ms']}ms")
# Test actual call
try:
result = await gateway.call_with_circuit_breaker(
circuit_name='holysheep_gpt4',
api_key='YOUR_HOLYSHEEP_API_KEY',
model='gpt-4.1',
messages=[{'role': 'user', 'content': 'Hello!'}]
)
print(f"Success: {result.get('choices', [{}])[0].get('message', {}).get('content', '')[:100]}")
except Exception as e:
print(f"Error: {e}")
if __name__ == '__main__':
asyncio.run(test_circuit_breaker())
3. Graceful Degradation - Chiến lược Fallback thông minh
Khi primary provider gặp sự cố, hệ thống cần tự động fallback mà không ảnh hưởng user experience. Tôi implement một fallback strategy có chi phí tối ưu:
# graceful_degradation.py
import asyncio
import time
from typing import Optional, List, Dict, Any
from dataclasses import dataclass, field
from enum import Enum
import httpx
import logging
logger = logging.getLogger(__name__)
class ProviderStatus(Enum):
HEALTHY = "healthy"
DEGRADED = "degraded"
UNAVAILABLE = "unavailable"
@dataclass
class Provider:
name: str
base_url: str
api_key: str
models: List[str]
priority: int # 1 = highest
cost_per_1k_tokens: float # USD
avg_latency_ms: float
success_rate: float = 1.0
status: ProviderStatus = ProviderStatus.HEALTHY
last_health_check: float = field(default_factory=time.time)
@dataclass
class FallbackConfig:
max_retries: int = 2
retry_delay_ms: int = 500
health_check_interval: int = 30
min_success_rate: float = 0.95
latency_sla_ms: float = 5000
class IntelligentFallbackManager:
"""
Intelligent fallback manager với cost optimization
Fallback chain được thiết kế để:
1. Ưu tiên providers có độ trễ thấp nhất
2. Tự động bỏ qua providers có success rate thấp
3. Tối ưu chi phí khi primary unavailable
"""
def __init__(self, config: FallbackConfig):
self.config = config
self.providers: Dict[str, Provider] = {}
self.request_metrics: Dict[str, List[float]] = {}
def add_provider(self, provider: Provider):
"""Register a provider"""
self.providers[provider.name] = provider
self.request_metrics[provider.name] = []
logger.info(f"Registered provider: {provider.name} (priority: {provider.priority})")
async def _health_check(self, provider: Provider) -> bool:
"""Perform health check on provider"""
try:
async with httpx.AsyncClient(timeout=5.0) as client:
response = await client.post(
f"{provider.base_url}/chat/completions",
headers={'Authorization': f'Bearer {provider.api_key}'},
json={
'model': provider.models[0],
'messages': [{'role': 'user', 'content': 'ping'}],
'max_tokens': 1
}
)
if response.status_code in [200, 400, 401]: # 400/401 means reachable
provider.status = ProviderStatus.HEALTHY
provider.last_health_check = time.time()
return True
except Exception as e:
logger.warning(f"Health check failed for {provider.name}: {e}")
provider.status = ProviderStatus.UNAVAILABLE
return False
def _get_available_providers(self, model: str) -> List[Provider]:
"""Get providers sorted by priority that support the model"""
available = []
for provider in self.providers.values():
if model in provider.models and provider.status != ProviderStatus.UNAVAILABLE:
# Calculate composite score (lower = better)
# Score = latency_weight * latency + cost_weight * cost
latency_score = provider.avg_latency_ms / 1000 # Normalize
cost_score = provider.cost_per_1k_tokens
# Weight: 60% latency, 40% cost
composite_score = 0.6 * latency_score + 0.4 * cost_score
available.append((composite_score, provider))
# Sort by composite score (lower = better)
available.sort(key=lambda x: x[0])
return [p for _, p in available]
async def call_with_fallback(
self,
model: str,
messages: List[Dict],
temperature: float = 0.7,
max_tokens: int = 1000
) -> Dict[str, Any]:
"""
Make AI call with intelligent fallback
"""
providers = self._get_available_providers(model)
if not providers:
raise Exception(f"No available providers for model: {model}")
last_error = None
for attempt in range(self.config.max_retries + 1):
for provider in providers:
try:
start_time = time.time()
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
f"{provider.base_url}/chat/completions",
headers={'Authorization': f'Bearer {provider.api_key}'},
json={
'model': model,
'messages': messages,
'temperature': temperature,
'max_tokens': max_tokens
}
)
latency_ms = (time.time() - start_time) * 1000
self.request_metrics[provider.name].append(latency_ms)
# Update provider stats
provider.avg_latency_ms = (
0.9 * provider.avg_latency_ms + 0.1 * latency_ms
)
if response.status_code == 200:
result = response.json()
result['_provider'] = provider.name
result['_latency_ms'] = latency_ms
result['_cost_usd'] = self._estimate_cost(result, provider.cost_per_1k_tokens)
return result
elif response.status_code == 429:
# Rate limited - try next provider
logger.warning(f"Rate limited by {provider.name}, trying next...")
provider.status = ProviderStatus.DEGRADED
continue
else:
logger.error(f"Provider {provider.name} returned {response.status_code}")
provider.success_rate *= 0.95
continue
except httpx.TimeoutException:
logger.warning(f"Timeout from {provider.name}")
provider.avg_latency_ms *= 1.5 # Increase estimated latency
provider.success_rate *= 0.9
except httpx.ConnectError as e:
logger.error(f"Connection error to {provider.name}: {e}")
provider.status = ProviderStatus.UNAVAILABLE
except Exception as e:
logger.error(f"Unexpected error from {provider.name}: {e}")
last_error = e
# Small delay between providers
await asyncio.sleep(0.1)
# Wait before retry
if attempt < self.config.max_retries:
await asyncio.sleep(self.config.retry_delay_ms / 1000)
raise Exception(f"All providers failed. Last error: {last_error}")
def _estimate_cost(self, response: dict, cost_per_1k: float) -> float:
"""Estimate cost based on response tokens"""
try:
usage = response.get('usage', {})
total_tokens = usage.get('total_tokens', 0)
return (total_tokens / 1000) * cost_per_1k
except:
return 0.0
async def run_health_checks(self):
"""Background task to check provider health"""
while True:
for provider in self.providers.values():
await self._health_check(provider)
await asyncio.sleep(self.config.health_check_interval)
def get_cost_comparison(self, model: str) -> Dict[str, Dict]:
"""Compare costs across providers for a model"""
comparison = {}
for provider in self.providers.values():
if model in provider.models:
comparison[provider.name] = {
'cost_per_1m_tokens_usd': provider.cost_per_1k_tokens * 1000,
'avg_latency_ms': provider.avg_latency_ms,
'success_rate': provider.success_rate,
'status': provider.status.value
}
return comparison
Initialize with HolySheep AI providers
async def setup_fallback_manager():
manager = IntelligentFallbackManager(FallbackConfig())
# Primary: HolySheep AI (85%+ cheaper than OpenAI)
manager.add_provider(Provider(
name='holysheep_primary',
base_url='https://api.holysheep.ai/v1',
api_key='YOUR_HOLYSHEEP_API_KEY',
models=['gpt-4.1', 'gpt-4-turbo', 'gpt-3.5-turbo'],
priority=1,
cost_per_1k_tokens=0.008, # $8/MTok = $0.008/1K tokens
avg_latency_ms=45 # <50ms guarantee
))
# Fallback 1: DeepSeek (cheapest option)
manager.add_provider(Provider(
name='deepseek_fallback',
base_url='https://api.holysheep.ai/v1', # Same API, different model routing
api_key='YOUR_HOLYSHEEP_API_KEY',
models=['deepseek-v3.2', 'deepseek-chat'],
priority=2,
cost_per_1k_tokens=0.00042, # $0.42/MTok - ultra cheap
avg_latency_ms=80
))
# Fallback 2: Claude via HolySheep
manager.add_provider(Provider(
name='claude_fallback',
base_url='https://api.holysheep.ai/v1',
api_key='YOUR_HOLYSHEEP_API_KEY',
models=['claude-sonnet-4.5', 'claude-opus-3'],
priority=3,
cost_per_1k_tokens=0.015, # $15/MTok
avg_latency_ms=60
))
return manager
async def main():
manager = await setup_fallback_manager()
# Show cost comparison
print("=== Cost Comparison for GPT-4.1 ===")
comparison = manager.get_cost_comparison('gpt-4.1')
for provider, info in comparison.items():
print(f"{provider}: ${info['cost_per_1m_tokens_usd']:.2f}/1M tokens, "
f"{info['avg_latency_ms']}ms latency")
# Make request with fallback
print("\n=== Testing Fallback ===")
try:
result = await manager.call_with_fallback(
model='gpt-4.1',
messages=[{'role': 'user', 'content': 'Explain quantum computing in 2 sentences.'}]
)
print(f"Success via {result['_provider']}")
print(f"Latency: {result['_latency_ms']:.0f}ms")
print(f"Estimated cost: ${result['_cost_usd']:.6f}")
except Exception as e:
print(f"All providers failed: {e}")
if __name__ == '__main__':
asyncio.run(main())
4. Load Balancer với Weighted Round Robin
Để phân phối traffic hiệu quả giữa các model và provider, tôi sử dụng Weighted Round Robin với dynamic weights dựa trên health và latency:
# load_balancer.py
import random
import time
from typing import List, Dict, Optional, Callable
from dataclasses import dataclass, field
from collections import defaultdict
import asyncio
import logging
logger = logging.getLogger(__name__)
@dataclass
class Endpoint:
id: str
url: str
weight: int = 100 # Base weight (1-1000)
current_weight: int = 0
effective_weight: int = 100
# Health metrics
is_healthy: bool = True
consecutive_failures: int = 0
last_failure_time: float = 0
# Performance metrics
total_requests: int = 0
failed_requests: int = 0
total_latency_ms: float = 0
avg_latency_ms: float = 0
# Metadata
region: str = 'us-east'
max_concurrent: int = 100
current_concurrent: int = 0
class WeightedRoundRobinBalancer:
"""
Weighted Round Robin Load Balancer với:
- Dynamic weight adjustment based on health
- Least-connections awareness
- Geographic routing support
- Request coalescing for identical requests
"""
def __init__(self):
self.endpoints: Dict[str, Endpoint] = {}
self.current_index: int = -1
self.gcd_weight: int = 0
self.max_failures = 5
self.recovery_timeout = 30 # seconds
def add_endpoint(self, endpoint: Endpoint):
"""Register an endpoint"""
self.endpoints[endpoint.id] = endpoint
self._recalculate_gcd()
logger.info(f"Added endpoint: {endpoint.id} (weight: {endpoint.weight})")
def _recalculate_gcd(self):
"""Recalculate GCD of all weights"""
if not self.endpoints:
return
weights = [e.weight for e in self.endpoints.values() if e.weight > 0]
if weights:
self.gcd_weight = self._gcd(weights)
def _gcd(self, numbers: List[int]) -> int: