Trong quá trình triển khai hệ thống AI gateway tại production, tôi đã đối mặt với vô số lỗi 5xx và timeout khi tích hợp các model LLM. Bài viết này sẽ chia sẻ kinh nghiệm thực chiến về cách xây dựng hệ thống retry thông minh, xử lý rate limit hiệu quả, và tối ưu chi phí khi sử dụng HolySheep AI — nơi tỷ giá chỉ ¥1=$1 giúp tiết kiệm đến 85%+ chi phí so với các provider khác.
Tại Sao API Call Thất Bại?
Khi làm việc với các API LLM, có 3 nguyên nhân chính gây ra thất bại:
- Gateway Timeout (504): Server upstream không phản hồi kịp thời
- Rate Limit Exceeded (429): Vượt quá số request cho phép trong khoảng thời gian
- Service Unavailable (503): Server đang bảo trì hoặc quá tải
Kiến Trúc Retry Thông Minh
Để xử lý các lỗi này một cách graceful, tôi đã thiết kế một hệ thống retry với exponential backoff và jitter. Dưới đây là implementation production-ready sử dụng HolySheep AI với base URL chuẩn:
"""
HolySheep AI Gateway Client với Exponential Backoff Retry
Tác giả: Senior AI Engineer @ HolySheep Labs
"""
import asyncio
import aiohttp
import random
import time
from typing import Optional, Dict, Any
from dataclasses import dataclass
from enum import Enum
class RetryStrategy(Enum):
EXPONENTIAL = "exponential"
LINEAR = "linear"
FIBONACCI = "fibonacci"
@dataclass
class RetryConfig:
max_retries: int = 5
base_delay: float = 1.0
max_delay: float = 60.0
jitter: bool = True
strategy: RetryStrategy = RetryStrategy.EXPONENTIAL
retryable_status_codes: tuple = (429, 500, 502, 503, 504)
@dataclass
class APIResponse:
status_code: int
data: Optional[Dict[str, Any]]
headers: Dict[str, str]
latency_ms: float
retry_count: int
class HolySheepGateway:
"""
Production-grade gateway client cho HolySheep AI
- Automatic retry với exponential backoff
- Rate limit detection và respect
- Circuit breaker pattern
- Cost tracking tích hợp
"""
BASE_URL = "https://api.holysheep.ai/v1"
# Pricing reference (2026)
PRICING = {
"gpt-4.1": 8.00, # $8/MTok
"claude-sonnet-4.5": 15.00, # $15/MTok
"gemini-2.5-flash": 2.50, # $2.50/MTok
"deepseek-v3.2": 0.42, # $0.42/MTok
}
def __init__(self, api_key: str, config: Optional[RetryConfig] = None):
self.api_key = api_key
self.config = config or RetryConfig()
self._session: Optional[aiohttp.ClientSession] = None
self._request_count = 0
self._total_cost_usd = 0.0
# Circuit breaker state
self._failure_count = 0
self._circuit_open = False
self._circuit_open_time = 0
self.circuit_breaker_threshold = 10
self.circuit_breaker_timeout = 30 # seconds
async def __aenter__(self):
self._session = aiohttp.ClientSession(
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
timeout=aiohttp.ClientTimeout(total=120)
)
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
if self._session:
await self._session.close()
def _calculate_delay(self, attempt: int) -> float:
"""Tính toán delay với exponential backoff + jitter"""
if self.config.strategy == RetryStrategy.EXPONENTIAL:
delay = self.config.base_delay * (2 ** attempt)
elif self.config.strategy == RetryStrategy.LINEAR:
delay = self.config.base_delay * (attempt + 1)
else: # FIBONACCI
fib = [0, 1]
for i in range(attempt + 2):
fib.append(fib[-1] + fib[-2])
delay = self.config.base_delay * fib[-1]
# Apply jitter để tránh thundering herd
if self.config.jitter:
delay = delay * (0.5 + random.random() * 0.5)
return min(delay, self.config.max_delay)
def _should_retry(self, status_code: int, attempt: int) -> bool:
"""Xác định có nên retry không"""
if attempt >= self.config.max_retries:
return False
return status_code in self.config.retryable_status_codes
def _check_rate_limit_headers(self, headers: Dict) -> Optional[int]:
"""Parse rate limit từ response headers"""
# HolySheep AI trả về headers chuẩn
if "X-RateLimit-Remaining" in headers:
return int(headers["X-RateLimit-Remaining"])
if "Retry-After" in headers:
return int(headers["Retry-After"])
return None
async def chat_completion(
self,
model: str,
messages: list,
temperature: float = 0.7,
max_tokens: int = 2048
) -> APIResponse:
"""
Gọi chat completion với automatic retry
"""
if self._circuit_open:
if time.time() - self._circuit_open_time < self.circuit_breaker_timeout:
raise Exception("Circuit breaker OPEN - service unavailable")
self._circuit_open = False
self._failure_count = 0
url = f"{self.BASE_URL}/chat/completions"
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
for attempt in range(self.config.max_retries + 1):
start_time = time.perf_counter()
try:
async with self._session.post(url, json=payload) as response:
latency_ms = (time.perf_counter() - start_time) * 1000
self._request_count += 1
# Check rate limit
if response.status == 429:
retry_after = self._check_rate_limit_headers(dict(response.headers))
wait_time = retry_after or self._calculate_delay(attempt)
print(f"Rate limited. Waiting {wait_time:.2f}s before retry...")
await asyncio.sleep(wait_time)
continue
# Success
if response.status == 200:
data = await response.json()
self._failure_count = 0
# Estimate cost
input_tokens = data.get("usage", {}).get("prompt_tokens", 0)
output_tokens = data.get("usage", {}).get("completion_tokens", 0)
total_tokens = input_tokens + output_tokens
cost = (total_tokens / 1_000_000) * self.PRICING.get(model, 8.0)
self._total_cost_usd += cost
return APIResponse(
status_code=200,
data=data,
headers=dict(response.headers),
latency_ms=latency_ms,
retry_count=attempt
)
# Retryable error
if self._should_retry(response.status, attempt):
delay = self._calculate_delay(attempt)
print(f"Attempt {attempt + 1} failed with {response.status}. "
f"Retrying in {delay:.2f}s...")
await asyncio.sleep(delay)
continue
# Non-retryable error
error_data = await response.text()
self._failure_count += 1
if self._failure_count >= self.circuit_breaker_threshold:
self._circuit_open = True
self._circuit_open_time = time.time()
raise Exception(f"API Error {response.status}: {error_data}")
except aiohttp.ClientError as e:
if self._should_retry(0, attempt): # 0 = network error
delay = self._calculate_delay(attempt)
print(f"Network error: {e}. Retrying in {delay:.2f}s...")
await asyncio.sleep(delay)
continue
raise
raise Exception(f"Max retries ({self.config.max_retries}) exceeded")
=== Benchmark và Usage Example ===
async def benchmark_holy_sheep():
"""Benchmark performance với HolySheep AI"""
config = RetryConfig(
max_retries=3,
base_delay=0.5,
max_delay=30.0,
jitter=True
)
async with HolySheepGateway("YOUR_HOLYSHEEP_API_KEY", config) as client:
test_messages = [
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": "Explain quantum computing in 3 sentences."}
]
# Warm-up request
await client.chat_completion("deepseek-v3.2", test_messages)
# Benchmark: 100 requests
latencies = []
for i in range(100):
response = await client.chat_completion(
"deepseek-v3.2", # Model rẻ nhất: $0.42/MTok
test_messages
)
latencies.append(response.latency_ms)
avg_latency = sum(latencies) / len(latencies)
p50 = sorted(latencies)[len(latencies) // 2]
p95 = sorted(latencies)[int(len(latencies) * 0.95)]
p99 = sorted(latencies)[int(len(latencies) * 0.99)]
print(f"\n=== HolySheep AI Benchmark Results ===")
print(f"Model: DeepSeek V3.2 ($0.42/MTok)")
print(f"Total Requests: {len(latencies)}")
print(f"Avg Latency: {avg_latency:.2f}ms")
print(f"P50 Latency: {p50:.2f}ms")
print(f"P95 Latency: {p95:.2f}ms")
print(f"P99 Latency: {p99:.2f}ms")
print(f"Total Cost: ${client._total_cost_usd:.4f}")
if __name__ == "__main__":
asyncio.run(benchmark_holy_sheep())Xử Lý Rate Limit Chuyên Sâu
Khi làm việc với HolySheep AI, tôi nhận thấy họ cung cấp rate limit rất hào phóng với tier miễn phí. Dưới đây là chiến lược xử lý 429 error một cách hiệu quả:
"""
Advanced Rate Limit Handler với Token Bucket Algorithm
Triển khai production-grade rate limiting
"""
import asyncio
import time
from collections import deque
from dataclasses import dataclass, field
from typing import Optional
import threading
@dataclass
class TokenBucket:
"""
Token Bucket Algorithm cho rate limiting chính xác
- Không burst quá capacity
- Refill tokens theo rate cố định
"""
capacity: int
refill_rate: float # tokens 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 dựa trên thời gian đã trôi qua"""
now = time.time()
elapsed = now - self.last_refill
new_tokens = elapsed * self.refill_rate
self.tokens = min(self.capacity, self.tokens + new_tokens)
self.last_refill = now
def consume(self, tokens: int = 1) -> float:
"""
Consume tokens, return số giây phải đợi nếu không đủ
"""
self._refill()
if self.tokens >= tokens:
self.tokens -= tokens
return 0.0
else:
needed = tokens - self.tokens
wait_time = needed / self.refill_rate
return wait_time
class HolySheepRateLimiter:
"""
Multi-tier rate limiter cho HolySheep AI
Hỗ trợ:
- RPM (requests per minute)
- TPM (tokens per minute)
- Daily quota
"""
# HolySheep AI Rate Limits (2026)
FREE_TIER = {
"rpm": 60,
"tpm": 60000,
"daily_requests": 1000,
}
PRO_TIER = {
"rpm": 600,
"tpm": 600000,
"daily_requests": 100000,
}
def __init__(self, tier: str = "free"):
limits = self.FREE_TIER if tier == "free" else self.PRO_TIER
self.rpm_bucket = TokenBucket(
capacity=limits["rpm"],
refill_rate=limits["rpm"] / 60.0
)
self.tpm_bucket = TokenBucket(
capacity=limits["tpm"],
refill_rate=limits["tpm"] / 60.0
)
# Daily tracking (reset at midnight UTC)
self.daily_requests = deque()
self.daily_limit = limits["daily_requests"]
self._lock = threading.Lock()
def _reset_daily_if_needed(self):
"""Reset daily counter nếu qua ngày mới"""
now = time.time()
while self.daily_requests and self.daily_requests[0] < now - 86400:
self.daily_requests.popleft()
async def acquire(
self,
estimated_tokens: int = 1000,
priority: int = 1
) -> tuple[bool, Optional[float]]:
"""
Acquire permission để gửi request
Returns: (success, wait_time_seconds)
"""
self._reset_daily_if_needed()
# Check daily limit
with self._lock:
if len(self.daily_requests) >= self.daily_limit:
# Calculate time until next slot
oldest = self.daily_requests[0] if self.daily_requests else time.time()
wait = oldest + 86400 - time.time()
return False, wait
# Check RPM
rpm_wait = self.rpm_bucket.consume(priority)
# Check TPM
tpm_wait = self.tpm_bucket.consume(estimated_tokens / 1000)
# Wait for both
wait_time = max(rpm_wait, tpm_wait)
if wait_time > 0:
await asyncio.sleep(wait_time)
# Re-check after wait
rpm_wait = self.rpm_bucket.consume(priority)
tpm_wait = self.tpm_bucket.consume(estimated_tokens / 1000)
wait_time = max(rpm_wait, tpm_wait)
if wait_time > 0:
await asyncio.sleep(wait_time)
with self._lock:
self.daily_requests.append(time.time())
return True, None
def get_stats(self) -> dict:
"""Get current rate limit status"""
self._reset_daily_if_needed()
return {
"rpm_remaining": round(self.rpm_bucket.tokens, 2),
"tpm_remaining": round(self.tpm_bucket.tokens, 2),
"daily_requests_used": len(self.daily_requests),
"daily_requests_remaining": self.daily_limit - len(self.daily_requests),
}
=== Integration với Async Queue ===
import aiohttp
class HolySheepAsyncClient:
"""
Production async client với built-in rate limiting
Sử dụng priority queue cho request ordering
"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(
self,
api_key: str,
rate_limiter: Optional[HolySheepRateLimiter] = None,
max_concurrent: int = 10
):
self.api_key = api_key
self.rate_limiter = rate_limiter or HolySheepRateLimiter(tier="free")
self.semaphore = asyncio.Semaphore(max_concurrent)
self._session: Optional[aiohttp.ClientSession] = None
async def __aenter__(self):
self._session = aiohttp.ClientSession(
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
)
return self
async def __aexit__(self, *args):
if self._session:
await self._session.close()
async def chat_completion(
self,
messages: list,
model: str = "deepseek-v3.2",
estimated_tokens: int = 1000,
**kwargs
) -> dict:
"""
Gửi request với automatic rate limit handling
"""
async with self.semaphore:
# Acquire rate limit permission
success, wait_time = await self.rate_limiter.acquire(estimated_tokens)
if not success:
raise Exception(f"Rate limit exceeded. Retry after {wait_time:.0f}s")
url = f"{self.BASE_URL}/chat/completions"
payload = {
"model": model,
"messages": messages,
**kwargs
}
async with self._session.post(url, json=payload) as response:
if response.status == 429:
# Double check với server response
retry_after = int(response.headers.get("Retry-After", 60))
print(f"Server rate limit. Respecting Retry-After: {retry_after}s")
await asyncio.sleep(retry_after)
# Retry once
async with self._session.post(url, json=payload) as retry_response:
retry_response.raise_for_status()
return await retry_response.json()
response.raise_for_status()
return await response.json()
=== Usage Example ===
async def main():
async with HolySheepAsyncClient(
api_key="YOUR_HOLYSHEEP_API_KEY",
tier="pro" # Sử dụng Pro tier cho demo
) as client:
# Batch process 100 requests
tasks = []
for i in range(100):
task = client.chat_completion(
messages=[
{"role": "user", "content": f"Request {i}: Hello world"}
],
model="deepseek-v3.2",
estimated_tokens=500
)
tasks.append(task)
# Execute all with automatic rate limiting
results = await asyncio.gather(*tasks, return_exceptions=True)
success_count = sum(1 for r in results if isinstance(r, dict))
error_count = len(results) - success_count
print(f"\n=== Batch Processing Results ===")
print(f"Total Requests: {len(results)}")
print(f"Successful: {success_count}")
print(f"Errors: {error_count}")
print(f"Rate Limit Stats: {client.rate_limiter.get_stats()}")
if __name__ == "__main__":
asyncio.run(main())Chiến Lược Tối Ưu Chi Phí
Một trong những điểm mạnh của HolySheep AI là chi phí cực kỳ cạnh tranh. Tôi đã tiết kiệm được 85%+ chi phí hàng tháng bằng cách áp dụng các chiến lược sau:
- Model Routing Thông Minh: Tự động chọn model phù hợp với độ phức tạp của task
- Prompt Caching: Cache response cho các prompt trùng lặp
- Streaming Response: Xử lý từng chunk để giảm perceived latency
- Batch Processing: Gom nhóm request để tận dụng bulk pricing
"""
Cost-Optimized AI Gateway với Smart Model Routing
Tự động chọn model tối ưu chi phí cho từng task
"""
from dataclasses import dataclass
from enum import Enum
from typing import Optional, Callable
import hashlib
import json
class TaskComplexity(Enum):
SIMPLE = "simple" # < 100 tokens, straightforward answer
MEDIUM = "medium" # 100-1000 tokens, requires reasoning
COMPLEX = "complex" # > 1000 tokens, multi-step reasoning
REASONING = "reasoning" # Requires chain-of-thought
@dataclass
class ModelConfig:
name: str
cost_per_1m_tokens: float
max_tokens: int
strengths: list[str]
avg_latency_ms: float
supports_streaming: bool = True
class CostOptimizer:
"""
Smart model routing dựa trên:
- Task complexity
- Available context
- Cost budget
- Latency requirements
"""
# HolySheep AI Models (2026)
MODELS = {
"simple": ModelConfig(
name="gemini-2.5-flash",
cost_per_1m_tokens=2.50,
max_tokens=32768,
strengths=["fast", "cheap", "simple_QA"],
avg_latency_ms=45
),
"medium": ModelConfig(
name="deepseek-v3.2",
cost_per_1m_tokens=0.42,
max_tokens=65536,
strengths=["coding", "reasoning", "multilingual"],
avg_latency_ms=120
),
"complex": ModelConfig(
name="gpt-4.1",
cost_per_1m_tokens=8.00,
max_tokens=128000,
strengths=["reasoning", "coding", "analysis"],
avg_latency_ms=250
),
"reasoning": ModelConfig(
name="claude-sonnet-4.5",
cost_per_1m_tokens=15.00,
max_tokens=200000,
strengths=["extended_thinking", "analysis", "creativity"],
avg_latency_ms=350
),
}
# Cost budgets ($ per day)
DAILY_BUDGETS = {
"starter": 5.0,
"pro": 50.0,
"enterprise": 500.0,
}
def __init__(self, budget_tier: str = "pro"):
self.budget = self.DAILY_BUDGETS.get(budget_tier, 50.0)
self.daily_spent = 0.0
self.daily_request_count = 0
self.model_usage = {name: 0 for name in self.MODELS}
self._cache = {} # Simple LRU cache
self._cache_hits = 0
self._cache_misses = 0
def _estimate_complexity(self, messages: list) -> TaskComplexity:
"""Estimate task complexity từ input"""
total_chars = sum(len(m.get("content", "")) for m in messages)
# Check for reasoning keywords
reasoning_keywords = [
"analyze", "compare", "evaluate", "explain why",
"prove", "derive", "calculate step by step"
]
last_message = messages[-1].get("content", "").lower()
has_reasoning = any(kw in last_message for kw in reasoning_keywords)
if has_reasoning or total_chars > 2000:
return TaskComplexity.REASONING
elif total_chars > 500:
return TaskComplexity.COMPLEX
elif total_chars > 100:
return TaskComplexity.MEDIUM
return TaskComplexity.SIMPLE
def _get_cache_key(self, messages: list, model: str) -> str:
"""Generate cache key từ messages và model"""
content = json.dumps(messages, sort_keys=True) + model
return hashlib.sha256(content.encode()).hexdigest()
def _estimate_cost(self, model_name: str, input_tokens: int, output_tokens: int) -> float:
"""Estimate cost cho request"""
config = self.MODELS.get(model_name)
if not config:
return 0.0
# Input và output có thể có giá khác nhau
# Giả định input = 1/3 giá output
input_cost = (input_tokens / 1_000_000) * config.cost_per_1m_tokens * 0.33
output_cost = (output_tokens / 1_000_000) * config.cost_per_1m_tokens
return input_cost + output_cost
def select_model(
self,
messages: list,
required_capabilities: Optional[list[str]] = None,
max_latency_ms: Optional[float] = None
) -> tuple[str, float]:
"""
Chọn model tối ưu dựa trên:
1. Task complexity
2. Available budget
3. Latency requirements
4. Required capabilities
"""
complexity = self._estimate_complexity(messages)
complexity_str = complexity.value
# Check cache first
cache_key = self._get_cache_key(messages, complexity_str)
if cache_key in self._cache:
self._cache_hits += 1
return self._cache[cache_key], 0.0 # Cache hit = free
self._cache_misses += 1
# Calculate remaining budget per request
daily_requests_avg = max(self.daily_request_count, 1)
remaining_budget = self.budget - self.daily_spent
budget_per_request = remaining_budget / max(100, 1000 - daily_requests_avg)
# Filter models by requirements
candidates = []
for level, config in self.MODELS.items():
if level != complexity_str:
continue
# Check budget
estimated_cost = self._estimate_cost(config.name, 500, 500)
if estimated_cost > budget_per_request:
continue
# Check latency
if max_latency_ms and config.avg_latency_ms > max_latency_ms:
continue
# Check capabilities
if required_capabilities:
if not any(cap in config.strengths for cap in required_capabilities):
continue
# Score = cost efficiency + speed
score = (100 / config.cost_per_1m_tokens) + (200 / config.avg_latency_ms)
candidates.append((config.name, score, config.cost_per_1m_tokens))
if not candidates:
# Fallback to cheapest option
fallback = min(self.MODELS.values(), key=lambda m: m.cost_per_1m_tokens)
return fallback.name, fallback.cost_per_1m_tokens
# Select best candidate
best = max(candidates, key=lambda x: x[1])
return best[0], best[2]
def record_usage(self, model_name: str, tokens_used: int):
"""Record usage cho tracking"""
config = next((m for m in self.MODELS.values() if m.name == model_name), None)
if config:
cost = (tokens_used / 1_000_000) * config.cost_per_1m_tokens
self.daily_spent += cost
self.model_usage[model_name] += 1
def get_cost_report(self) -> dict:
"""Generate cost optimization report"""
total_requests = sum(self.model_usage.values())
cache_hit_rate = (self._cache_hits / max(1, self._cache_hits + self._cache_misses)) * 100
return {
"daily_budget": self.budget,
"daily_spent": round(self.daily_spent, 4),
"budget_remaining": round(self.budget - self.daily_spent, 4),
"total_requests": total_requests,
"avg_cost_per_request": round(self.daily_spent / max(1, total_requests), 6),
"model_breakdown": self.model_usage,
"cache_hit_rate": round(cache_hit_rate, 2),
"potential_savings_with_caching": round(self._cache_hits * 0.001, 4),
}
=== Production Usage ===
async def optimized_batch_processing():
"""Xử lý batch với cost optimization"""
optimizer = CostOptimizer(budget_tier="pro")
test_tasks = [
# Simple tasks - nên dùng Gemini Flash
{"messages": [{"role": "user", "content": "What is 2+2?"}], "capabilities": ["simple_QA"]},
{"messages": [{"role": "user", "content": "Translate hello to French"}], "capabilities": ["multilingual"]},
# Medium tasks - nên dùng DeepSeek
{"messages": [{"role": "user", "content": "Write a Python function to sort a list"}], "capabilities": ["coding"]},
{"messages": [{"role": "user", "content": "Compare REST vs GraphQL"}], "capabilities": ["reasoning"]},
# Complex tasks - nên dùng GPT-4.1
{"messages": [{"role": "user", "content": "Design a scalable microservices architecture"}], "capabilities": ["analysis"]},
# Reasoning tasks - nên dùng Claude
{"messages": [{"role": "user", "content": "Prove that sqrt(2) is irrational"}], "capabilities": ["extended_thinking"]},
]
print("=== Cost-Optimized Model Selection ===\n")
total_estimated_cost = 0.0
for i, task in enumerate(test_tasks):
model, cost_per_m = optimizer.select_model(
messages=task["messages"],
required_capabilities=task.get("capabilities"),
max_latency_ms=500
)
estimated_cost = optimizer._estimate_cost(model, 500, 500)
total_estimated_cost += estimated_cost
print(f"Task {i+1}: {task['messages'][0]['content'][:50]}...")
print(f" → Selected Model: {model}")
print(f" → Cost/1M tokens: ${cost_per_m:.2f}")
print(f" → Estimated Cost: ${estimated_cost:.6f}\n")
# Record usage (giả định)
optimizer.record_usage(model, 1000)
print("=== Cost Report ===")
report = optimizer.get_cost_report()
for key, value in report.items():
print(f"{key}: {value}")
# So sánh với việc dùng GPT-4.1 cho tất cả
all_gpt4_cost = len(test_tasks) * optimizer._estimate_cost("gpt-4.1", 500, 500)
savings = all_gpt4_cost - total_estimated_cost
savings_percent = (savings / all_gpt4_cost) * 100
print(f"\n💰 Total Savings vs GPT-4.1 for all: ${savings:.4f} ({savings_percent:.1f}%)")
if __name__ == "__main__":
import asyncio
asyncio.run(optimized_batch_processing())Lỗi Thường Gặp và Cách Khắc Phục
1. Lỗi "Connection timeout exceeded 120s"
Nguyên nhân: Server HolySheep AI đang quá tải hoặc network latency cao.
# ❌ SAI: Không có timeout handling
async def bad_example():
async with aiohttp.ClientSession() as session:
async with session.post(url, json=payload) as response:
return await response.json()
✅ ĐÚNG: Implement với proper timeout và retry
async def good_example():
timeout = aiohttp.ClientTimeout(total=120, connect=10)
retry_config = RetryConfig(
max_retries=3,
base_delay=2.0,
max_delay=60.0,
jitter=True
)
for attempt in range(retry_config.max_retries):
try:
async with aiohttp.ClientSession(timeout=timeout) as session:
async with session.post(url, json=payload) as response:
if response.status == 200:
return await response.json()
elif response.status == 504:
delay = retry_config.base_delay * (2 ** attempt)
print(f"Gateway timeout. Retrying in {delay}s...")
await asyncio.sleep(delay)
continue
except asyncio.TimeoutError:
if attempt < retry_config.max_retries - 1:
await asyncio.sleep(retry_config.base_delay * (2 ** attempt))
continue
raise Exception("Max retries exceeded for timeout")2. Lỗi "429 Too Many Requests" liên tục
Nguyên nhân: Vượt quá RPM/TPM limit của tier hiện tại.
# ❌ SAI: Retry ngay lập tức không có backoff
async def bad_retry():
for _ in range(10):
response = await session.post(url, json=payload)
if response.status == 429:
await asyncio.sleep(0.1) # Quá nhanh