Trong hành trình triển khai hệ thống AI production, tôi đã gặp không ít lần "cháy" API key vì không kiểm soát được chi phí, hoặc bị rate limit vì concurrency không được quản lý tốt. Bài viết này tổng hợp kinh nghiệm thực chiến với HolySheep AI — nền tảng API hỗ trợ Claude, GPT, Gemini với chi phí thấp hơn 85% so với các provider khác.
Tại Sao Quản Lý API Key Quan Trọng?
Khi làm việc với các API AI, có 3 rủi ro lớn thường gặp:
- Cháy budget: Không có giới hạn → chi phí phình to không kiểm soát
- Rate limit: Gửi quá nhiều request đồng thời → bị chặn
- Bảo mật: Key bị lộ → bị sử dụng trái phép
Với HolySheep AI, chi phí được tối ưu rõ rệt: Claude Sonnet 4.5 chỉ $15/1M tokens so với $18 của Anthropic, trong khi DeepSeek V3.2 chỉ $0.42/1M tokens — phù hợp cho các task không cần model đắt đỏ.
Kiến Trúc Quản Lý API Key An Toàn
1. Lưu Trữ Key Với Environment Variables
Nguyên tắc vàng: Không bao giờ hardcode API key trong source code. Sử dụng environment variables với validation nghiêm ngặt.
# config.py - Quản lý configuration an toàn
import os
from dataclasses import dataclass
from typing import Optional
@dataclass
class APIConfig:
api_key: str
base_url: str = "https://api.holysheep.ai/v1"
max_retries: int = 3
timeout: float = 30.0
@classmethod
def from_env(cls) -> "APIConfig":
api_key = os.getenv("HOLYSHEEP_API_KEY")
if not api_key:
raise ValueError("HOLYSHEEP_API_KEY not found in environment")
# Validate key format (HolySheep keys thường có prefix)
if not api_key.startswith("sk-") and not api_key.startswith("hs-"):
raise ValueError("Invalid API key format")
return cls(
api_key=api_key,
base_url=os.getenv("HOLYSHEEP_BASE_URL", "https://api.holysheep.ai/v1"),
max_retries=int(os.getenv("API_MAX_RETRIES", "3")),
timeout=float(os.getenv("API_TIMEOUT", "30.0"))
)
Sử dụng
config = APIConfig.from_env()
print(f"✅ Config loaded: base_url={config.base_url}")2. Implement Retry Logic Với Exponential Backoff
Rate limit là vấn đề thường gặp khi xử lý batch. Retry thông minh giúp giảm failed requests.
# retry_handler.py - Exponential backoff với jitter
import time
import asyncio
import aiohttp
from typing import Callable, Any
from functools import wraps
class RateLimitHandler:
def __init__(self, max_retries: int = 5, base_delay: float = 1.0):
self.max_retries = max_retries
self.base_delay = base_delay
self.request_count = 0
self.last_reset = time.time()
def should_retry(self, status_code: int) -> bool:
# HolySheep trả về 429 khi rate limit
return status_code in [429, 500, 502, 503, 504]
def get_delay(self, attempt: int) -> float:
# Exponential backoff: 1s, 2s, 4s, 8s, 16s
delay = self.base_delay * (2 ** attempt)
# Thêm jitter ±25% để tránh thundering herd
import random
jitter = delay * 0.25 * (2 * random.random() - 1)
return delay + jitter
async def execute_with_retry(
self,
func: Callable,
*args,
**kwargs
) -> Any:
last_exception = None
for attempt in range(self.max_retries):
try:
result = await func(*args, **kwargs)
self.request_count += 1
return result
except aiohttp.ClientResponseError as e:
last_exception = e
if e.status == 429:
# Rate limit - đợi đến khi được phép
retry_after = e.headers.get('Retry-After', '60')
wait_time = int(retry_after) if retry_after.isdigit() else 60
print(f"⏳ Rate limited. Waiting {wait_time}s...")
await asyncio.sleep(wait_time)
continue
elif not self.should_retry(e.status):
raise
delay = self.get_delay(attempt)
print(f"🔄 Retry {attempt + 1}/{self.max_retries} after {delay:.2f}s")
await asyncio.sleep(delay)
except Exception as e:
last_exception = e
if attempt == self.max_retries - 1:
raise
delay = self.get_delay(attempt)
await asyncio.sleep(delay)
raise last_exception
Sử dụng
handler = RateLimitHandler(max_retries=5, base_delay=1.0)Kiểm Soát Chi Phí & Tối Ưu Token Usage
3. Budget Guard - Ngăn Chặn Chi Phí Phình To
Đây là tính năng cực kỳ quan trọng mà nhiều developer bỏ qua. Implement budget guard giúp bạn ngủ ngon hơn.
# budget_guard.py - Giới hạn chi phí tự động
import time
from dataclasses import dataclass, field
from typing import Dict, Optional
from threading import Lock
@dataclass
class BudgetConfig:
daily_limit: float = 100.0 # USD
monthly_limit: float = 2000.0 # USD
per_request_limit: float = 5.0 # USD (ngăn request quá lớn)
@dataclass
class TokenUsage:
tokens: int
cost: float
model: str
timestamp: float = field(default_factory=time.time)
class BudgetGuard:
def __init__(self, config: BudgetConfig):
self.config = config
self.daily_usage: float = 0.0
self.monthly_usage: float = 0.0
self.last_daily_reset = time.time()
self.last_monthly_reset = time.time()
self.usage_history: Dict[str, TokenUsage] = {}
self.lock = Lock()
def check_request_allowed(self, estimated_cost: float) -> bool:
"""Kiểm tra xem request có được phép thực hiện không"""
self._reset_if_needed()
with self.lock:
if estimated_cost > self.config.per_request_limit:
print(f"❌ Request rejected: cost ${estimated_cost:.2f} exceeds limit ${self.config.per_request_limit:.2f}")
return False
if self.daily_usage + estimated_cost > self.config.daily_limit:
print(f"❌ Daily budget exceeded: ${self.daily_usage:.2f} + ${estimated_cost:.2f} > ${self.config.daily_limit:.2f}")
return False
if self.monthly_usage + estimated_cost > self.config.monthly_limit:
print(f"❌ Monthly budget exceeded: ${self.monthly_usage:.2f} + ${estimated_cost:.2f} > ${self.config.monthly_limit:.2f}")
return False
return True
def record_usage(self, request_id: str, usage: TokenUsage):
"""Ghi nhận usage sau khi request hoàn thành"""
self._reset_if_needed()
with self.lock:
self.daily_usage += usage.cost
self.monthly_usage += usage.cost
self.usage_history[request_id] = usage
# Alert khi approaching limits
daily_pct = (self.daily_usage / self.config.daily_limit) * 100
monthly_pct = (self.monthly_usage / self.config.monthly_limit) * 100
if daily_pct >= 80:
print(f"⚠️ Daily budget at {daily_pct:.1f}%")
if monthly_pct >= 80:
print(f"⚠️ Monthly budget at {monthly_pct:.1f}%")
def _reset_if_needed(self):
"""Reset counters định kỳ"""
now = time.time()
# Reset daily (mỗi 24h)
if now - self.last_daily_reset > 86400:
self.daily_usage = 0.0
self.last_daily_reset = now
# Reset monthly (mỗi 30 days)
if now - self.last_monthly_reset > 2592000:
self.monthly_usage = 0.0
self.last_monthly_reset = now
def get_stats(self) -> Dict:
return {
"daily_usage": self.daily_usage,
"daily_limit": self.config.daily_limit,
"daily_pct": (self.daily_usage / self.config.daily_limit) * 100,
"monthly_usage": self.monthly_usage,
"monthly_limit": self.config.monthly_limit,
"monthly_pct": (self.monthly_usage / self.config.monthly_limit) * 100,
"total_requests": len(self.usage_history)
}
Pricing lookup cho HolySheep (USD per 1M tokens)
HOLYSHEEP_PRICING = {
"gpt-4.1": {"input": 8.0, "output": 8.0},
"claude-sonnet-4.5": {"input": 15.0, "output": 15.0},
"gemini-2.5-flash": {"input": 2.50, "output": 2.50},
"deepseek-v3.2": {"input": 0.42, "output": 0.42}
}
def estimate_cost(model: str, input_tokens: int, output_tokens: int) -> float:
"""Ước tính chi phí cho một request"""
if model not in HOLYSHEEP_PRICING:
return 0.0
pricing = HOLYSHEEP_PRICING[model]
input_cost = (input_tokens / 1_000_000) * pricing["input"]
output_cost = (output_tokens / 1_000_000) * pricing["output"]
return input_cost + output_cost
Sử dụng
budget = BudgetGuard(BudgetConfig(daily_limit=50.0, monthly_limit=500.0))
Test
estimated = estimate_cost("deepseek-v3.2", 10000, 5000)
print(f"💰 Estimated cost for request: ${estimated:.4f}")
print(f"✅ Request allowed: {budget.check_request_allowed(estimated)}")4. Model Selection Thông Minh - Chọn Đúng Model Cho Đúng Task
Việc chọn đúng model có thể tiết kiệm 90%+ chi phí. DeepSeek V3.2 chỉ $0.42/1M tokens trong khi Claude Sonnet 4.5 là $15/1M tokens — gấp 35 lần.
# model_selector.py - Chọn model tối ưu chi phí
from enum import Enum
from dataclasses import dataclass
from typing import List, Optional, Callable
import re
class TaskComplexity(Enum):
TRIVIAL = "trivial" # Classification, sentiment
SIMPLE = "simple" # Summarization, extraction
MODERATE = "moderate" # Translation, rewrite
COMPLEX = "complex" # Analysis, reasoning
ADVANCED = "advanced" # Code generation, creative
@dataclass
class ModelRecommendation:
model: str
provider: str
estimated_cost_per_1k: float # USD per 1000 tokens
complexity: TaskComplexity
latency_ms: int
reason: str
class SmartModelSelector:
def __init__(self):
# HolySheep pricing (USD per 1M tokens)
self.models = {
# DeepSeek - rẻ nhất, phù hợp task đơn giản
"deepseek-v3.2": {
"cost_input": 0.42,
"cost_output": 0.42,
"complexity": [TaskComplexity.TRIVIAL, TaskComplexity.SIMPLE],
"latency_ms": 45,
"provider": "HolySheep"
},
# Gemini Flash - cân bằng cost/performance
"gemini-2.5-flash": {
"cost_input": 2.50,
"cost_output": 2.50,
"complexity": [TaskComplexity.SIMPLE, TaskComplexity.MODERATE],
"latency_ms": 35,
"provider": "HolySheep"
},
# GPT-4.1 - mạnh, giá vừa phải
"gpt-4.1": {
"cost_input": 8.0,
"cost_output": 8.0,
"complexity": [TaskComplexity.MODERATE, TaskComplexity.COMPLEX],
"latency_ms": 55,
"provider": "HolySheep"
},
# Claude Sonnet - đắt nhất nhưng mạnh nhất
"claude-sonnet-4.5": {
"cost_input": 15.0,
"cost_output": 15.0,
"complexity": [TaskComplexity.COMPLEX, TaskComplexity.ADVANCED],
"latency_ms": 65,
"provider": "HolySheep"
}
}
def analyze_task(self, prompt: str, context_length: int = 0) -> TaskComplexity:
"""Phân tích độ phức tạp của task dựa trên prompt"""
prompt_lower = prompt.lower()
# Keywords gợi ý complexity cao
advanced_keywords = ["analyze", "evaluate", "design", "architect",
"generate code", "debug", "optimize", "creative"]
complex_keywords = ["compare", "summarize", "explain", "translate",
"rewrite", "classify", "extract"]
# Check for code/math
has_code = bool(re.search(r'```|\ndef |\nclass |function ', prompt))
has_math = bool(re.search(r'\d+\s*[\+\-\*/\%]\s*\d+|calculate|solve', prompt))
if any(kw in prompt_lower for kw in advanced_keywords) or (has_code and has_math):
return TaskComplexity.ADVANCED
elif any(kw in prompt_lower for kw in complex_keywords):
return TaskComplexity.MODERATE
elif context_length > 10000:
return TaskComplexity.MODERATE
else:
return TaskComplexity.TRIVIAL
def recommend(
self,
prompt: str,
required_complexity: Optional[TaskComplexity] = None,
min_quality: str = "good" # "fast", "good", "best"
) -> ModelRecommendation:
"""Đưa ra model recommendation tối ưu"""
complexity = required_complexity or self.analyze_task(prompt, len(prompt))
# Tìm model phù hợp với complexity
candidates = []
for model_name, model_info in self.models.items():
if complexity in model_info["complexity"]:
candidates.append((model_name, model_info))
if not candidates:
# Fallback to most capable
candidates = [("claude-sonnet-4.5", self.models["claude-sonnet-4.5"])]
# Chọn model dựa trên yêu cầu quality
if min_quality == "fast":
# Chọn model có latency thấp nhất
best = min(candidates, key=lambda x: x[1]["latency_ms"])
elif min_quality == "best":
# Chọn model mạnh nhất
best = max(candidates, key=lambda x: x[1]["cost_input"])
else:
# Cân bằng cost/performance - chọn model rẻ nhất phù hợp
best = min(candidates, key=lambda x: x[1]["cost_input"])
model_name, model_info = best
# Generate reason
reasons = {
TaskComplexity.TRIVIAL: "Task đơn giản, dùng model giá rẻ",
TaskComplexity.SIMPLE: "Task vừa phải, tối ưu chi phí",
TaskComplexity.MODERATE: "Cần reasoning tốt hơn",
TaskComplexity.COMPLEX: "Cần model mạnh với context dài",
TaskComplexity.ADVANCED: "Task phức tạp cao, cần model tốt nhất"
}
return ModelRecommendation(
model=model_name,
provider=model_info["provider"],
estimated_cost_per_1k=model_info["cost_input"] / 1000,
complexity=complexity,
latency_ms=model_info["latency_ms"],
reason=reasons.get(complexity, "")
)
Sử dụng
selector = SmartModelSelector()
test_prompts = [
"Classify this email as spam or not spam",
"Summarize the following article",
"Debug this Python code and suggest fixes",
"Design a microservices architecture for e-commerce"
]
for prompt in test_prompts:
rec = selector.recommend(prompt, min_quality="good")
print(f"\n📝 Task: {prompt[:50]}...")
print(f" 🎯 Recommended: {rec.model}")
print(f" 💰 Cost: ${rec.estimated_cost_per_1k:.4f}/1K tokens")
print(f" ⚡ Latency: ~{rec.latency_ms}ms")Concurrency Control - Kiểm Soát Request Đồng Thời
Đây là phần nhiều developer gặp vấn đề nhất. Gửi quá nhiều request đồng thời sẽ bị rate limit. Với HolySheep AI có latency trung bình <50ms, bạn có thể tận dụng điều này để tối ưu throughput.
# concurrency_control.py - Semaphore-based rate limiting
import asyncio
import time
from typing import List, Dict, Any, Optional
from dataclasses import dataclass
from collections import deque
import aiohttp
@dataclass
class RateLimitConfig:
requests_per_second: float = 10.0
requests_per_minute: float = 500.0
requests_per_day: float = 100000.0
burst_size: int = 20
class TokenBucket:
"""Token bucket algorithm cho smooth rate limiting"""
def __init__(self, rate: float, capacity: int):
self.rate = rate # tokens per second
self.capacity = capacity
self.tokens = capacity
self.last_update = time.time()
self.lock = asyncio.Lock()
async def acquire(self, tokens: int = 1) -> float:
"""Acquire tokens, return wait time if needed"""
async with self.lock:
now = time.time()
elapsed = now - self.last_update
# Refill tokens based on elapsed time
self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
self.last_update = now
if self.tokens >= tokens:
self.tokens -= tokens
return 0.0
else:
# Calculate wait time
wait_time = (tokens - self.tokens) / self.rate
return wait_time
class ConcurrencyController:
"""Controller quản lý concurrency với multiple rate limits"""
def __init__(self, config: RateLimitConfig):
self.rps_bucket = TokenBucket(config.requests_per_second, config.burst_size)
self.rpm_bucket = TokenBucket(config.requests_per_minute / 60, config.requests_per_minute // 60 + 10)
self.rpd_bucket = TokenBucket(config.requests_per_day / 86400, int(config.requests_per_day / 86400) + 100)
self.semaphore = asyncio.Semaphore(config.burst_size)
self.stats = {
"total_requests": 0,
"successful": 0,
"rate_limited": 0,
"total_wait_time": 0.0
}
async def execute(
self,
coro: Any,
session: aiohttp.ClientSession
) -> Any:
"""Execute coroutine với rate limiting"""
# Acquire all rate limit tokens
start_wait = time.time()
wait_rps = await self.rps_bucket.acquire(1)
wait_rpm = await self.rpm_bucket.acquire(1)
wait_rpd = await self.rpd_bucket.acquire(1)
total_wait = max(wait_rps, wait_rpm, wait_rpd)
self.stats["total_wait_time"] += total_wait
if total_wait > 0:
await asyncio.sleep(total_wait)
# Acquire semaphore slot
async with self.semaphore:
self.stats["total_requests"] += 1
try:
result = await coro
self.stats["successful"] += 1
return result
except aiohttp.ClientResponseError as e:
if e.status == 429:
self.stats["rate_limited"] += 1
raise RateLimitExceeded()
raise
def get_stats(self) -> Dict:
return {
**self.stats,
"avg_wait_time": self.stats["total_wait_time"] / max(1, self.stats["total_requests"])
}
class RateLimitExceeded(Exception):
"""Custom exception for rate limit"""
pass
Ví dụ sử dụng với HolySheep API
async def call_holysheep(
controller: ConcurrencyController,
session: aiohttp.ClientSession,
prompt: str,
model: str = "deepseek-v3.2"
):
"""Gọi HolySheep API với concurrency control"""
url = "https://api.holysheep.ai/v1/chat/completions"
headers = {
"Authorization": f"Bearer {os.getenv('HOLYSHEEP_API_KEY')}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": 1000
}
async def request():
async with session.post(url, json=payload, headers=headers) as resp:
return await resp.json()
return await controller.execute(request(), session)
Test
async def main():
controller = ConcurrencyController(
RateLimitConfig(
requests_per_second=50.0, # 50 req/s
requests_per_minute=2000.0, # 2000 req/min
burst_size=100
)
)
async with aiohttp.ClientSession() as session:
tasks = [
call_holysheep(controller, session, f"Analyze document #{i}")
for i in range(100)
]
start = time.time()
results = await asyncio.gather(*tasks, return_exceptions=True)
elapsed = time.time() - start
stats = controller.get_stats()
print(f"✅ Completed {stats['successful']} requests in {elapsed:.2f}s")
print(f"📊 Throughput: {stats['successful'] / elapsed:.2f} req/s")
print(f"⏱️ Avg wait time: {stats['avg_wait_time'] * 1000:.2f}ms")Monitoring & Alerting - Theo Dõi Trong Thời Gian Thực
# monitoring.py - Dashboard metrics cho production
import time
import json
from dataclasses import dataclass, asdict
from typing import Dict, List
from datetime import datetime, timedelta
from collections import defaultdict
@dataclass
class RequestMetric:
request_id: str
model: str
input_tokens: int
output_tokens: int
latency_ms: float
cost: float
status: str
timestamp: float
class APIMonitor:
"""Monitor và alert cho API usage"""
def __init__(self, alert_thresholds: Dict = None):
self.thresholds = alert_thresholds or {
"p99_latency_ms": 500,
"error_rate_pct": 5,
"daily_cost_usd": 100
}
self.metrics: List[RequestMetric] = []
self.alerts: List[Dict] = []
def record(self, metric: RequestMetric):
self.metrics.append(metric)
self._check_alerts(metric)
def _check_alerts(self, metric: RequestMetric):
"""Check various alert conditions"""
alerts = []
# High latency alert
if metric.latency_ms > self.thresholds["p99_latency_ms"]:
alerts.append({
"type": "high_latency",
"message": f"High latency: {metric.latency_ms:.0f}ms",
"severity": "warning"
})
# Error alert
if metric.status == "error":
alerts.append({
"type": "error",
"message": f"Request failed: {metric.request_id}",
"severity": "error"
})
# High cost alert
if metric.cost > 1.0: # > $1 per request
alerts.append({
"type": "high_cost",
"message": f"High cost request: ${metric.cost:.2f}",
"severity": "warning"
})
self.alerts.extend(alerts)
def get_dashboard(self) -> Dict:
"""Generate dashboard data"""
now = time.time()
last_hour = now - 3600
last_24h = now - 86400
recent = [m for m in self.metrics if m.timestamp > last_hour]
daily = [m for m in self.metrics if m.timestamp > last_24h]
# Calculate statistics
total_cost = sum(m.cost for m in daily)
total_tokens = sum(m.input_tokens + m.output_tokens for m in daily)
error_count = sum(1 for m in recent if m.status == "error")
# Latency percentiles
latencies = sorted([m.latency_ms for m in recent])
p50 = latencies[len(latencies) // 2] if latencies else 0
p99 = latencies[int(len(latencies) * 0.99)] if latencies else 0
return {
"timestamp": datetime.now().isoformat(),
"last_hour": {
"requests": len(recent),
"error_rate": error_count / max(1, len(recent)) * 100,
"p50_latency_ms": p50,
"p99_latency_ms": p99
},
"last_24h": {
"requests": len(daily),
"total_cost_usd": round(total_cost, 2),
"total_tokens": total_tokens,
"avg_cost_per_request": round(total_cost / max(1, len(daily)), 4)
},
"model_breakdown": self._model_breakdown(daily),
"alerts": self.alerts[-10:] # Last 10 alerts
}
def _model_breakdown(self, metrics: List[RequestMetric]) -> Dict:
breakdown = defaultdict(lambda: {"requests": 0, "cost": 0, "tokens": 0})
for m in metrics:
breakdown[m.model]["requests"] += 1
breakdown[m.model]["cost"] += m.cost
breakdown[m.model]["tokens"] += m.input_tokens + m.output_tokens
return dict(breakdown)
def export_json(self, filepath: str):
"""Export metrics to JSON for external monitoring"""
with open(filepath, 'w') as f:
json.dump(self.get_dashboard(), f, indent=2)
Sử dụng
monitor = APIMonitor()
Record sample metrics
for i in range(100):
monitor.record(RequestMetric(
request_id=f"req_{i}",
model="deepseek-v3.2",
input_tokens=500,
output_tokens=200,
latency_ms=45 + (i % 50),
cost=0.00035,
status="success",
timestamp=time.time()
))
dashboard = monitor.get_dashboard()
print(json.dumps(dashboard, indent=2))Lỗi Thường Gặp Và Cách Khắc Phục
Lỗi 1: "Invalid API Key" Hoặc Authentication Failed
Nguyên nhân: Key không đúng format, đã bị revoke, hoặc chưa set đúng environment variable.
# Debug authentication issues
import os
def validate_api_key(key: str) -> tuple[bool, str]:
"""Validate HolySheep API key format"""
# Check basic format
if not key:
return False, "API key is empty"
# HolySheep keys thường có prefix sk- hoặc hs-
if not (key.startswith("sk-") or key.startswith("hs-")):
return False, f"Invalid key prefix. Key should start with 'sk-' or 'hs-', got: {key[:5]}***"
# Check length (typically 32+ characters)
if len(key) < 32:
return False, f"API key too short: {len(key)} characters"
# Check for whitespace
if ' ' in key:
return False, "API key contains whitespace characters"
return True, "API key format is valid"
Test
test_keys = [
"sk-holysheep-test-key-1234567890abcdef",
"hs-abc123",
"",
"invalid_key_format",
"sk-key with spaces"
]
for key in test_keys:
valid, msg = validate_api_key(key)
status = "✅" if valid else "❌"
print(f"{status} {msg}")Lỗi 2: Rate Limit (429 Too Many Requests)
Nguyên nhân: Gửi quá nhiều request trong thời gian ngắn. HolySheep có rate limit tùy theo plan.
# Handle rate limit với smart backoff
import time
import asyncio
async def smart_retry_with_backoff(
func,
max_retries: int = 5,
base_delay: float = 1.0,
max_delay: float = 60.0
):
"""
Retry với exponential backoff thông minh
- Đọc Retry-After header nếu có
- Exponential backoff khi không có header
- Jitter để tránh thundering herd
"""
last_exception = None
for attempt in range(max_retries):
try:
return await func()
except Exception as e:
last_exception = e
# Parse error
if hasattr(e, 'status') and e.status == 429:
# Get retry-after from headers if available
retry_after = getattr(e, 'retry_after', None)
if retry_after:
wait_time = retry_after
print(f"⏳ Server says wait {wait_time}s")
else:
# Exponential backoff
wait_time = min(base_delay * (2 ** attempt), max_delay)
# Add random jitter ±25%
import random
wait_time *= (0.75 + random.random() * 0.5)
print(f"🔄 Rate limited. Attempt {attempt + 1}/{max_retries}. Waiting {wait_time:.1f}s...")
await asyncio.sleep(wait_time)
else:
# Non-rate-limit error, re-raise immediately
raise
raise last_exception
Alternative: Pre-emptive rate limiting
class PreemptiveRateLimiter:
"""Limit requests before gửi đi"""
def __init__(self, calls_per_second: float = 10.0):
self.interval = 1.0 / calls_per_second
self.last_call = 0.0
self.lock = asyncio.Lock()
async def wait_if_needed(self):
async with self.lock:
now = time.time()
elapsed = now - self.last_call
if elapsed < self.interval:
wait = self.interval - elapsed
await asyncio.sleep(wait)
self.last_call = time.time()
Sử dụng