Năm ngoái, vào lúc 3 giờ sáng, hệ thống sản xuất của tôi bị sập hoàn toàn. Logs tràn ngập lỗi: ConnectionError: timeout after 30s và 401 Unauthorized. Nguyên nhân? API key hết hạn và không có fallback. Bài học đắt giá đó đã thay đổi hoàn toàn cách tôi thiết kế kiến trúc phụ thuộc AI. Trong bài viết này, tôi sẽ chia sẻ cách phân tích và tối ưu hóa kiến trúc AI service để đạt được 99.9% uptime với chi phí tối ưu nhất.
Tại sao phân tích phụ thuộc AI quan trọng?
Khi tích hợp AI vào hệ thống, nhiều developer chỉ quan tâm đến việc "gọi được API" mà bỏ qua các yếu tố quan trọng:
- Latency: Độ trễ ảnh hưởng trực tiếp đến trải nghiệm người dùng
- Cost: Chi phí API có thể tăng đột biến nếu không kiểm soát
- Reliability: Single point of failure có thể gây sập toàn hệ thống
- Scalability: Kiến trúc không tối ưu sẽ không xử lý được traffic cao đột xuất
Kiến trúc đa tầng với HolySheep AI
HolySheep AI cung cấp endpoint unified hoạt động với nhiều model AI hàng đầu: GPT-4.1 ($8/MTok), Claude Sonnet 4.5 ($15/MTok), Gemini 2.5 Flash ($2.50/MTok), và DeepSeek V3.2 chỉ $0.42/MTok — tiết kiệm đến 85%+ so với các provider khác. Thời gian phản hồi trung bình dưới 50ms, hỗ trợ WeChat/Alipay thanh toán, và tín dụng miễn phí khi đăng ký tại đây.
Cấu hình Client với Retry Logic và Fallback
import httpx
import asyncio
import logging
from typing import Optional, Dict, Any
from datetime import datetime, timedelta
logger = logging.getLogger(__name__)
class HolySheepAIClient:
"""
Client AI service với kiến trúc đa tầng:
- Retry với exponential backoff
- Circuit breaker pattern
- Automatic fallback giữa các model
"""
BASE_URL = "https://api.holysheep.ai/v1"
# Thứ tự ưu tiên model (fallback chain)
MODEL_PRIORITY = [
"gpt-4.1",
"claude-sonnet-4.5",
"gemini-2.5-flash",
"deepseek-v3.2"
]
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = self.BASE_URL
self.current_model_index = 0
# Cấu hình HTTP client với timeout
self.client = httpx.AsyncClient(
timeout=httpx.Timeout(
connect=10.0,
read=60.0,
write=10.0,
pool=30.0
),
limits=httpx.Limits(max_keepalive_connections=20, max_connections=100)
)
# Circuit breaker state
self.failure_count = 0
self.last_failure_time: Optional[datetime] = None
self.circuit_open = False
self.circuit_open_time: Optional[datetime] = None
self.CIRCUIT_BREAKER_THRESHOLD = 5
self.CIRCUIT_BREAKER_TIMEOUT = 60 # seconds
def _get_current_model(self) -> str:
"""Lấy model hiện tại trong fallback chain"""
return self.MODEL_PRIORITY[self.current_model_index]
def _should_try_fallback(self) -> bool:
"""Kiểm tra xem có nên thử model tiếp theo không"""
if self.current_model_index < len(self.MODEL_PRIORITY) - 1:
return True
return False
def _open_circuit(self):
"""Mở circuit breaker"""
self.circuit_open = True
self.circuit_open_time = datetime.now()
logger.warning(f"Circuit breaker OPENED at {self.circuit_open_time}")
def _close_circuit(self):
"""Đóng circuit breaker"""
self.circuit_open = False
self.circuit_open_time = None
self.failure_count = 0
logger.info("Circuit breaker CLOSED")
def _check_circuit_state(self):
"""Kiểm tra và cập nhật trạng thái circuit breaker"""
if self.circuit_open and self.circuit_open_time:
elapsed = (datetime.now() - self.circuit_open_time).total_seconds()
if elapsed >= self.CIRCUIT_BREAKER_TIMEOUT:
self._close_circuit()
logger.info("Circuit breaker HALF-OPEN, attempting recovery")
async def chat_completion(
self,
messages: list,
system_prompt: Optional[str] = None,
temperature: float = 0.7,
max_tokens: int = 2048,
retries: int = 3
) -> Dict[str, Any]:
"""
Gọi API với retry logic và circuit breaker
"""
self._check_circuit_state()
if self.circuit_open:
# Nếu circuit đang open, thử fallback sang model khác
if self._should_try_fallback():
self.current_model_index += 1
logger.info(f"Falling back to {self._get_current_model()}")
else:
raise Exception("All AI services unavailable (circuit breaker open)")
all_messages = messages.copy()
if system_prompt:
all_messages.insert(0, {"role": "system", "content": system_prompt})
model = self._get_current_model()
for attempt in range(retries):
try:
response = await self._make_request(
model=model,
messages=all_messages,
temperature=temperature,
max_tokens=max_tokens
)
# Thành công - reset state
self.failure_count = 0
self.current_model_index = 0 # Reset về model tốt nhất
return response
except httpx.TimeoutException as e:
logger.warning(f"Timeout on attempt {attempt + 1}: {e}")
self.failure_count += 1
if self.failure_count >= self.CIRCUIT_BREAKER_THRESHOLD:
self._open_circuit()
raise Exception("Circuit breaker triggered: too many timeouts")
if attempt < retries - 1:
# Exponential backoff
await asyncio.sleep(2 ** attempt)
except httpx.HTTPStatusError as e:
if e.response.status_code == 401:
logger.error("Authentication failed - check API key")
raise Exception(f"401 Unauthorized - Invalid API key")
elif e.response.status_code == 429:
# Rate limit - chờ và thử lại
wait_time = int(e.response.headers.get("Retry-After", 60))
logger.warning(f"Rate limited, waiting {wait_time}s")
await asyncio.sleep(wait_time)
else:
logger.error(f"HTTP error {e.response.status_code}: {e}")
raise
except httpx.ConnectError as e:
logger.error(f"Connection error: {e}")
self.failure_count += 1
if attempt < retries - 1:
await asyncio.sleep(2 ** attempt)
# Thử fallback nếu có model khả dụng
if self._should_try_fallback():
self.current_model_index += 1
logger.info(f"Retrying with fallback model: {self._get_current_model()}")
return await self.chat_completion(
messages=messages,
system_prompt=system_prompt,
temperature=temperature,
max_tokens=max_tokens,
retries=retries
)
raise Exception(f"Failed after {retries} retries with all available models")
async def _make_request(
self,
model: str,
messages: list,
temperature: float,
max_tokens: int
) -> Dict[str, Any]:
"""Thực hiện HTTP request đến HolySheep API"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
start_time = datetime.now()
response = await self.client.post(
f"{self.base_url}/chat/completions",
json=payload,
headers=headers
)
response.raise_for_status()
elapsed = (datetime.now() - start_time).total_seconds() * 1000
logger.info(f"Request completed in {elapsed:.2f}ms with model {model}")
return response.json()
Sử dụng
async def main():
client = HolySheepAIClient(api_key="YOUR_HOLYSHEEP_API_KEY")
try:
result = await client.chat_completion(
messages=[
{"role": "user", "content": "Giải thích kiến trúc microservices?"}
],
system_prompt="Bạn là một chuyên gia về kiến trúc hệ thống",
temperature=0.7,
max_tokens=1000
)
print(result)
except Exception as e:
logger.error(f"Request failed: {e}")
if __name__ == "__main__":
asyncio.run(main())
Monitoring và Health Check System
import asyncio
import aiohttp
from dataclasses import dataclass
from typing import Dict, List, Optional
from datetime import datetime, timedelta
import statistics
@dataclass
class ServiceHealth:
name: str
is_healthy: bool
latency_avg: float
latency_p95: float
success_rate: float
last_check: datetime
error_message: Optional[str] = None
class AIMonitoringSystem:
"""
Hệ thống monitoring theo dõi sức khỏe của các AI service
"""
def __init__(self):
self.services: Dict[str, dict] = {}
self.health_history: Dict[str, List[ServiceHealth]] = {}
self.alert_callbacks: List[callable] = []
async def health_check(self, service_name: str, api_key: str) -> ServiceHealth:
"""Kiểm tra sức khỏe của một AI service"""
latencies = []
successes = 0
errors = []
# Thực hiện 10 health check requests
async with aiohttp.ClientSession() as session:
for _ in range(10):
start = datetime.now()
try:
async with session.post(
"https://api.holysheep.ai/v1/chat/completions",
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
},
json={
"model": "gpt-4.1",
"messages": [{"role": "user", "content": "Hi"}],
"max_tokens": 5
},
timeout=aiohttp.ClientTimeout(total=10)
) as resp:
if resp.status == 200:
successes += 1
else:
errors.append(f"HTTP {resp.status}")
except asyncio.TimeoutError:
errors.append("Timeout")
except Exception as e:
errors.append(str(e))
finally:
latency = (datetime.now() - start).total_seconds() * 1000
latencies.append(latency)
await asyncio.sleep(0.5)
latencies.sort()
p95_index = int(len(latencies) * 0.95)
health = ServiceHealth(
name=service_name,
is_healthy=successes >= 8, # 80% success rate threshold
latency_avg=statistics.mean(latencies),
latency_p95=latencies[p95_index] if latencies else 0,
success_rate=successes / 10,
last_check=datetime.now(),
error_message="; ".join(errors[:3]) if errors else None
)
# Lưu vào history
if service_name not in self.health_history:
self.health_history[service_name] = []
self.health_history[service_name].append(health)
# Giữ chỉ 100 records gần nhất
if len(self.health_history[service_name]) > 100:
self.health_history[service_name] = self.health_history[service_name][-100:]
# Check và alert nếu cần
await self._check_alerts(health)
return health
async def _check_alerts(self, health: ServiceHealth):
"""Kiểm tra điều kiện alert"""
alert_conditions = []
if not health.is_healthy:
alert_conditions.append(f"Service unhealthy (success rate: {health.success_rate:.1%})")
if health.latency_avg > 1000: # > 1s
alert_conditions.append(f"High latency: {health.latency_avg:.0f}ms")
if health.latency_p95 > 3000: # > 3s
alert_conditions.append(f"P95 latency critical: {health.latency_p95:.0f}ms")
if alert_conditions:
message = f"[ALERT] {health.name}: " + "; ".join(alert_conditions)
for callback in self.alert_callbacks:
await callback(message)
def get_service_stats(self, service_name: str, hours: int = 24) -> dict:
"""Lấy thống kê service trong N giờ"""
if service_name not in self.health_history:
return {}
cutoff = datetime.now() - timedelta(hours=hours)
recent = [
h for h in self.health_history[service_name]
if h.last_check >= cutoff
]
if not recent:
return {}
return {
"total_checks": len(recent),
"avg_success_rate": statistics.mean(h.success_rate for h in recent),
"avg_latency": statistics.mean(h.latency_avg for h in recent),
"p95_latency": statistics.mean(h.latency_p95 for h in recent),
"uptime_percent": sum(1 for h in recent if h.is_healthy) / len(recent) * 100
}
Prometheus metrics exporter
class PrometheusMetrics:
"""Export metrics cho Prometheus"""
def __init__(self):
self.metrics = {
"ai_request_total": 0,
"ai_request_success": 0,
"ai_request_failed": 0,
"ai_latency_ms_sum": 0.0,
"ai_latency_ms_count": 0,
}
def record_request(self, success: bool, latency_ms: float, model: str):
self.metrics["ai_request_total"] += 1
if success:
self.metrics["ai_request_success"] += 1
else:
self.metrics["ai_request_failed"] += 1
self.metrics["ai_latency_ms_sum"] += latency_ms
self.metrics["ai_latency_ms_count"] += 1
def get_prometheus_format(self) -> str:
success_rate = (
self.metrics["ai_request_success"] / self.metrics["ai_request_total"]
if self.metrics["ai_request_total"] > 0 else 0
)
avg_latency = (
self.metrics["ai_latency_ms_sum"] / self.metrics["ai_latency_ms_count"]
if self.metrics["ai_latency_ms_count"] > 0 else 0
)
return f"""# HELP ai_request_total Total AI API requests
TYPE ai_request_total counter
ai_request_total {self.metrics['ai_request_total']}
HELP ai_request_success_total Successful requests
TYPE ai_request_success_total counter
ai_request_success_total {self.metrics['ai_request_success']}
HELP ai_request_success_rate Current success rate
TYPE ai_request_success_rate gauge
ai_request_success_rate {success_rate:.4f}
HELP ai_request_latency_ms_avg Average latency in milliseconds
TYPE ai_request_latency_ms_avg gauge
ai_request_latency_ms_avg {avg_latency:.2f}
"""
Rate Limiter và Cost Control
import time
import asyncio
from collections import defaultdict
from dataclasses import dataclass, field
from typing import Optional
import hashlib
@dataclass
class RateLimitConfig:
"""Cấu hình rate limit cho từng tier"""
requests_per_minute: int
requests_per_hour: int
tokens_per_minute: int
tokens_per_hour: int
max_cost_per_day: float
class TokenBucketRateLimiter:
"""
Token Bucket algorithm cho rate limiting
- Đảm bảo không vượt quá rate limit
- Cho phép burst nhưng duy trì average rate
"""
def __init__(self, config: RateLimitConfig):
self.config = config
self.tokens = config.requests_per_minute
self.last_refill = time.time()
self.lock = asyncio.Lock()
# Track usage
self.hourly_requests = 0
self.hourly_tokens = 0
self.daily_cost = 0.0
self.last_hourly_reset = time.time()
self.last_daily_reset = time.time()
# Cost tracking per model
self.model_costs = defaultdict(float)
async def acquire(self, estimated_tokens: int, model: str) -> bool:
"""Acquire permission to make a request"""
async with self.lock:
self._refill_tokens()
self._check_periodic_resets()
# Check all limits
if self.tokens < 1:
wait_time = (1 - self.tokens) * (60 / self.config.requests_per_minute)
raise RateLimitExceeded(f"Rate limit reached, wait {wait_time:.1f}s")
if self.hourly_requests >= self.config.requests_per_hour:
raise RateLimitExceeded("Hourly request limit exceeded")
if self.hourly_tokens + estimated_tokens > self.config.tokens_per_hour:
raise RateLimitExceeded("Hourly token limit exceeded")
if self.daily_cost >= self.config.max_cost_per_day:
raise RateLimitExceeded("Daily cost limit exceeded")
# Consume token
self.tokens -= 1
self.hourly_requests += 1
self.hourly_tokens += estimated_tokens
return True
def record_cost(self, model: str, tokens_used: int, cost: float):
"""Ghi nhận chi phí thực tế"""
self.daily_cost += cost
self.model_costs[model] += cost
def _refill_tokens(self):
"""Refill tokens dựa trên thời gian trôi qua"""
now = time.time()
elapsed = now - self.last_refill
refill_rate = self.config.requests_per_minute / 60.0
self.tokens = min(
self.config.requests_per_minute,
self.tokens + elapsed * refill_rate
)
self.last_refill = now
def _check_periodic_resets(self):
"""Reset counters theo chu kỳ"""
now = time.time()
# Reset hourly
if now - self.last_hourly_reset >= 3600:
self.hourly_requests = 0
self.hourly_tokens = 0
self.last_hourly_reset = now
# Reset daily
if now - self.last_daily_reset >= 86400:
self.daily_cost = 0.0
self.model_costs.clear()
self.last_daily_reset = now
def get_remaining(self) -> dict:
"""Lấy thông tin remaining quota"""
return {
"tokens_remaining": int(self.tokens),
"requests_this_hour": self.hourly_requests,
"tokens_this_hour": self.hourly_tokens,
"cost_today": round(self.daily_cost, 4),
"cost_by_model": dict(self.model_costs)
}
class CostAwareRouter:
"""
Router thông minh chọn model dựa trên:
- Yêu cầu về latency/quality
- Budget còn lại
- Load hiện tại
"""
# Pricing từ HolySheep AI (USD per 1M tokens)
MODEL_COSTS = {
"gpt-4.1": {"input": 2.0, "output": 8.0, "latency": "medium", "quality": "highest"},
"claude-sonnet-4.5": {"input": 3.0, "output": 15.0, "latency": "medium", "quality": "highest"},
"gemini-2.5-flash": {"input": 0.35, "output": 2.50, "latency": "low", "quality": "high"},
"deepseek-v3.2": {"input": 0.07, "output": 0.42, "latency": "medium", "quality": "high"}
}
def select_model(
self,
task_type: str,
budget_remaining: float,
priority: str = "balanced" # "cost", "quality", "latency", "balanced"
) -> str:
"""Chọn model tối ưu dựa trên context"""
candidates = []
for model, info in self.MODEL_COSTS.items():
score = 0
if priority == "cost":
# Ưu tiên chi phí thấp
score += (20 - info["output"]) * 10
elif priority == "quality":
# Ưu tiên chất lượng
score += {"highest": 100, "high": 70}.get(info["quality"], 0)
elif priority == "latency":
# Ưu tiên tốc độ
score += {"low": 100, "medium": 60}.get(info["latency"], 0)
else: # balanced
score = 100 - info["output"] * 5 # Cost factor
score += {"highest": 30, "high": 20}.get(info["quality"], 0)
# Penalize nếu vượt budget
avg_cost = (info["input"] + info["output"]) / 2
if avg_cost > budget_remaining * 10: # Assume 10 requests/day
score *= 0.1
candidates.append((model, score))
# Sort và return model tốt nhất
candidates.sort(key=lambda x: x[1], reverse=True)
return candidates[0][0]
def estimate_cost(
self,
model: str,
input_tokens: int,
output_tokens: int
) -> float:
"""Ước tính chi phí cho một request"""
info = self.MODEL_COSTS.get(model, {})
input_cost = (input_tokens / 1_000_000) * info.get("input", 0)
output_cost = (output_tokens / 1_000_000) * info.get("output", 0)
return round(input_cost + output_cost, 6)
class RateLimitExceeded(Exception):
"""Exception khi vượt rate limit"""
pass
Ví dụ sử dụng
async def example_usage():
config = RateLimitConfig(
requests_per_minute=60,
requests_per_hour=1000,
tokens_per_minute=100000,
tokens_per_hour=1000000,
max_cost_per_day=50.0 # $50/day limit
)
limiter = TokenBucketRateLimiter(config)
router = CostAwareRouter()
# Chọn model tiết kiệm cho task đơn giản
model = router.select_model(
task_type="simple_classification",
budget_remaining=50.0,
priority="cost"
)
print(f"Selected model: {model}")
# Estimate cost
cost = router.estimate_cost(model, input_tokens=500, output_tokens=200)
print(f"Estimated cost: ${cost}")
# Try to make request
try:
await limiter.acquire(estimated_tokens=700, model=model)
print("Request allowed")
# Sau khi request thành công
limiter.record_cost(model, 700, cost)
print(f"Remaining quota: {limiter.get_remaining()}")
except RateLimitExceeded as e:
print(f"Rate limit exceeded: {e}")
if __name__ == "__main__":
asyncio.run(example_usage())
Lỗi thường gặp và cách khắc phục
1. Lỗi 401 Unauthorized - Authentication Failed
Mô tả lỗi: Khi gọi API nhận được response 401 Unauthorized hoặc AuthenticationError.
Nguyên nhân thường gặp:
- API key sai hoặc có khoảng trắng thừa
- API key đã bị revoke hoặc hết hạn
- Sai định dạng header Authorization
Mã khắc phục:
# Cách kiểm tra và khắc phục 401
1. Kiểm tra format API key (không có khoảng trắng)
api_key = "YOUR_HOLYSHEEP_API_KEY".strip()
2. Verify key format trước khi gọi
def validate_api_key(key: str) -> bool:
if not key:
return False
if len(key) < 20:
return False
# HolySheep AI key format: hs_live_xxxx hoặc hs_test_xxxx
return key.startswith(("hs_live_", "hs_test_"))
3. Test authentication
async def test_authentication(api_key: str) -> dict:
async with httpx.AsyncClient() as client:
try:
response = await client.get(
"https://api.holysheep.ai/v1/models",
headers={"Authorization": f"Bearer {api_key}"}
)
if response.status_code == 200:
return {"status": "valid", "models": response.json()}
elif response.status_code == 401:
return {"status": "invalid", "error": "Invalid API key"}
else:
return {"status": "error", "code": response.status_code}
except Exception as e:
return {"status": "error", "exception": str(e)}
4. Retry với fresh key
async def call_with_key_refresh(api_key: str):
try:
# Thử với key hiện tại
result = await make_api_call(api_key)
return result
except AuthenticationError:
# Key có thể đã hết hạn - lấy key mới
new_key = await refresh_api_key()
if new_key:
return await make_api_call(new_key)
raise
2. Lỗi ConnectionError: timeout after 30s
Mô tả lỗi: Request bị timeout liên tục, không nhận được response.
Nguyên nhân thường gặp:
- Mạng không ổn định hoặc bị chặn firewall
- Server quá tải, response time > 30s
- SSL/TLS handshake thất bại
- DNS resolution không hoạt động
Mã khắc phục:
import socket
import ssl
1. Cấu hình timeout linh hoạt
TIMEOUT_CONFIG = {
"connect": 10.0, # 10s để thiết lập connection
"read": 60.0, # 60s để nhận data
"pool": 30.0 # 30s cho connection pool
}
2. Retry với exponential backoff
async def call_with_smart_timeout(api_key: str, base_timeout: float = 30.0):
max_retries = 3
for attempt in range(max_retries):
try:
# Tăng timeout cho attempt sau
current_timeout = base_timeout * (1.5 ** attempt)
async with httpx.AsyncClient(
timeout=httpx.Timeout(current_timeout)
) as client:
response = await client.post(
"https://api.holysheep.ai/v1/chat/completions",
headers={"Authorization": f"Bearer {api_key}"},
json={"model": "gpt-4.1", "messages": [...], "max_tokens": 1000}
)
return response.json()
except httpx.TimeoutException:
if attempt == max_retries - 1:
# Thử fallback sang model có latency thấp hơn
return await call_with_fallback_model(api_key, "gemini-2.5-flash")
await asyncio.sleep(2 ** attempt)
except httpx.ConnectError:
# Kiểm tra kết nối internet
if not await check_internet_connection():
raise ConnectionError("No internet connection")
await asyncio.sleep(1)
3. Health check trước khi call
async def health_check_and_retry(api_key: str):
# Ping để check connectivity
try:
async with httpx.AsyncClient() as client:
start = time.time()
await client.get("https://api.holysheep.ai/v1/models",
headers={"Authorization": f"Bearer {api_key}"},
timeout=5.0)
latency = (time.time() - start) * 1000
if latency > 2000:
logger.warning(f"High latency detected: {latency}ms")
# Switch sang model có latency thấp hơn
return "gemini-2.5-flash"
return "gpt-4.1"
except:
return "deepseek-v3.2" # Fallback cuối cùng
async def check_internet_connection() -> bool:
try:
async with httpx.AsyncClient() as client:
await client.get("https://www.google.com", timeout=5.0)
return True
except:
return False
3. Lỗi 429 Rate Limit Exceeded
Mô tả lỗi: Nhận được HTTP 429 hoặc message "Rate limit exceeded".
Nguyên nhân thường gặp:
- Gửi quá nhiều request trong thời gian ngắn
- Vượt quota tokens cho phép
- Không có delay giữa các request
Mã khắc phục:
# Xử lý rate limit với adaptive throttling
class AdaptiveRateLimiter:
def __init__(self, initial_rpm: int = 60):
self.current_rpm = initial_rpm
self.request_times = []
self.penalty_count = 0
async def acquire(self):
"""Chờ cho đến khi được phép gửi request"""
now = time.time()
window = 60.0 # 1 phút
# Remove requests cũ hơn 1 phút
self.request_times = [t for t in self.request_times if now - t < window]
if len(self.request_times) >= self.current_rpm:
# Tính thời gian chờ
oldest = self.request_times[0]
wait_time = window - (now - oldest)
if wait_time > 0:
logger.info(f"Rate limit: waiting {wait_time:.1f}s")
await asyncio.sleep(wait_time)
self.request_times.append(time.time())
def handle_429(self, retry_after: Optional[int] = None):
"""Xử lý khi nhận 429 response"""
self.penalty_count += 1
# Giảm RPM tạm thời
self.current_rpm = max(10, self.current_rpm * 0.8)
# Tăng lại sau 5 phút không có lỗi
if self.penalty_count == 0:
self.current_rpm = min(100, self.current_rpm * 1.1)
def calculate_delay(self) -> float:
"""Tính delay cần thiết giữa các request"""
if self.request_times:
avg_interval = 60.0 / self.current_rpm
return max(0.1, avg_interval)
return 0.1
Sử dụng trong request loop
async def batch_process(items: list, api_key: str):
limiter = AdaptiveRateLimiter(initial_rpm=60)
for item in items:
await limiter.acquire()
try:
result = await make_api_call(item, api_key)
limiter.handle_success()
except httpx.HTTPStatusError as e:
if e.response.status_code == 429:
retry_after = int(e.response.headers.get("Retry-After", 60))
await asyncio.sleep(retry_after)
limiter.handle_429(retry_after)
# Retry sau khi chờ
result = await make_api_call