Mở Đầu: Bài Học Đắt Giá Từ Thảm Họa Chi Phí
Tôi vẫn nhớ như in ngày đầu tiên triển khai hệ thống RAG cho một doanh nghiệp thương mại điện tử quy mô vừa. Đêm hôm đó, sau khi hệ thống "lên production" thành công, tôi mở dashboard billing của nhà cung cấp API cũ — con số $847 cho 4 tiếng đồng hồ khiến tôi suýt ngã khỏi ghế. Đó là lúc tôi hiểu ra: **monitoring không phải là tùy chọn, mà là sinh mạng của dự án AI**. Sau 3 năm tối ưu chi phí cho hơn 50 dự án sử dụng HolyShehep AI, tôi chia sẻ với bạn playbook hoàn chỉnh về usage tracking và cost attribution.Tại Sao Monitoring AI API Lại Quan Trọng?
Trước khi đi vào chi tiết kỹ thuật, hãy xem lý do thực tế:
- Chi phí bùng nổ không kiểm soát: GPT-4.1 @ $8/MTok nghe có vẻ rẻ, nhưng với 10 triệu token/ngày, bạn sẽ mất $80/ngày = $2,400/tháng
- Không phân biệt được endpoint: Biết tổng chi phí $5,000 nhưng không biết chatbot ngốn bao nhiêu, RAG ngốn bao nhiêu
- Performance degradation: Không có metrics, bạn không biết response time tăng hay giảm
- Khó debug production: Khi user report lỗi, bạn không có data để reproduce
Kiến Trúc Monitoring Hoàn Chỉnh
1. Cấu Trúc Database Cho Usage Tracking
Tôi đã thử nhiều schema, và đây là cấu trúc tối ưu nhất cho việc tracking chi tiết:
-- Schema PostgreSQL cho AI API Monitoring
CREATE TABLE ai_api_requests (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
request_id UUID NOT NULL UNIQUE,
provider VARCHAR(50) NOT NULL, -- 'holysheep', 'openai', etc.
model VARCHAR(100) NOT NULL,
endpoint VARCHAR(100) NOT NULL,
operation VARCHAR(50) NOT NULL, -- 'chat', 'embedding', 'completion'
-- Token metrics
prompt_tokens INTEGER NOT NULL,
completion_tokens INTEGER NOT NULL,
total_tokens INTEGER GENERATED ALWAYS AS (prompt_tokens + completion_tokens) STORED,
-- Cost tracking (USD)
input_cost DECIMAL(10, 6) NOT NULL,
output_cost DECIMAL(10, 6) NOT NULL,
total_cost DECIMAL(10, 6) NOT NULL,
-- Performance metrics
latency_ms INTEGER NOT NULL,
time_to_first_token_ms INTEGER,
-- Context
user_id VARCHAR(100),
session_id VARCHAR(100),
request_metadata JSONB,
response_metadata JSONB,
-- Timestamps
created_at TIMESTAMPTZ DEFAULT NOW(),
-- Indexes for efficient querying
CONSTRAINT valid_tokens CHECK (prompt_tokens >= 0 AND completion_tokens >= 0)
);
-- Bảng tổng hợp theo ngày ( materialized view )
CREATE MATERIALIZED VIEW daily_usage_summary AS
SELECT
DATE(created_at) as usage_date,
provider,
model,
operation,
COUNT(*) as request_count,
SUM(prompt_tokens) as total_prompt_tokens,
SUM(completion_tokens) as total_completion_tokens,
SUM(total_tokens) as total_tokens,
ROUND(SUM(total_cost)::numeric, 2) as total_cost_usd,
AVG(latency_ms)::integer as avg_latency_ms,
PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY latency_ms)::integer as p95_latency_ms
FROM ai_api_requests
GROUP BY DATE(created_at), provider, model, operation
WITH DATA;
CREATE UNIQUE INDEX idx_daily_summary ON daily_usage_summary(usage_date, provider, model, operation);
-- Bảng budget alerts
CREATE TABLE budget_alerts (
id SERIAL PRIMARY KEY,
budget_name VARCHAR(100) NOT NULL,
limit_usd DECIMAL(10, 2) NOT NULL,
alert_threshold DECIMAL(3, 2) DEFAULT 0.8, -- 80%
current_spend DECIMAL(10, 2) DEFAULT 0,
last_alert_at TIMESTAMPTZ,
created_at TIMESTAMPTZ DEFAULT NOW()
);2. Python Client Wrapper Với Automatic Logging
Đây là wrapper mà tôi dùng trong mọi dự án — nó tự động log mọi request mà không cần thay đổi code existing:
# monitoring_client.py
import time
import uuid
import psycopg2
from psycopg2.extras import Json
from typing import Optional, Dict, Any, Callable
from functools import wraps
import httpx
HolySheep AI Configuration
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_PRICING = {
"gpt-4.1": {"input": 8.0, "output": 8.0}, # $8/MTok
"claude-sonnet-4.5": {"input": 15.0, "output": 15.0}, # $15/MTok
"gemini-2.5-flash": {"input": 2.50, "output": 2.50}, # $2.50/MTok
"deepseek-v3.2": {"input": 0.42, "output": 0.42}, # $0.42/MTok
}
class AIMonitoringClient:
def __init__(self, api_key: str, db_config: dict):
self.api_key = api_key
self.db_conn = psycopg2.connect(**db_config)
self.base_url = HOLYSHEEP_BASE_URL
self.client = httpx.Client(timeout=120.0)
def calculate_cost(self, model: str, prompt_tokens: int, completion_tokens: int) -> tuple[float, float, float]:
"""Tính chi phí theo model - đơn vị USD"""
if model not in HOLYSHEEP_PRICING:
# Default fallback
input_cost = (prompt_tokens / 1_000_000) * 8.0
output_cost = (completion_tokens / 1_000_000) * 8.0
else:
prices = HOLYSHEEP_PRICING[model]
input_cost = (prompt_tokens / 1_000_000) * prices["input"]
output_cost = (completion_tokens / 1_000_000) * prices["output"]
return input_cost, output_cost, input_cost + output_cost
def log_request(
self,
request_id: str,
model: str,
endpoint: str,
operation: str,
prompt_tokens: int,
completion_tokens: int,
latency_ms: int,
time_to_first_token: Optional[int],
user_id: Optional[str],
session_id: Optional[str],
request_metadata: Dict[str, Any],
response_metadata: Dict[str, Any]
):
"""Log request vào database"""
input_cost, output_cost, total_cost = self.calculate_cost(
model, prompt_tokens, completion_tokens
)
with self.db_conn.cursor() as cur:
cur.execute("""
INSERT INTO ai_api_requests (
request_id, provider, model, endpoint, operation,
prompt_tokens, completion_tokens,
input_cost, output_cost, total_cost,
latency_ms, time_to_first_token_ms,
user_id, session_id, request_metadata, response_metadata
) VALUES (
%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s
)
""", (
request_id, "holysheep", model, endpoint, operation,
prompt_tokens, completion_tokens,
input_cost, output_cost, total_cost,
latency_ms, time_to_first_token,
user_id, session_id, Json(request_metadata), Json(response_metadata)
))
self.db_conn.commit()
def chat_completion(
self,
model: str,
messages: list,
user_id: Optional[str] = None,
session_id: Optional[str] = None,
metadata: Optional[Dict] = None,
**kwargs
) -> Dict[str, Any]:
"""Gọi chat completion với automatic monitoring"""
request_id = str(uuid.uuid4())
start_time = time.time()
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
"X-Request-ID": request_id
}
payload = {
"model": model,
"messages": messages,
**kwargs
}
response = self.client.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload
)
latency_ms = int((time.time() - start_time) * 1000)
response_data = response.json()
if response.status_code == 200:
usage = response_data.get("usage", {})
prompt_tokens = usage.get("prompt_tokens", 0)
completion_tokens = usage.get("completion_tokens", 0)
self.log_request(
request_id=request_id,
model=model,
endpoint="/chat/completions",
operation="chat",
prompt_tokens=prompt_tokens,
completion_tokens=completion_tokens,
latency_ms=latency_ms,
time_to_first_token=None,
user_id=user_id,
session_id=session_id,
request_metadata=metadata or {},
response_metadata={"model_used": response_data.get("model")}
)
return response_data
def embeddings(
self,
model: str,
input_text: str,
user_id: Optional[str] = None,
metadata: Optional[Dict] = None
) -> Dict[str, Any]:
"""Gọi embeddings với automatic monitoring"""
request_id = str(uuid.uuid4())
start_time = time.time()
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
"X-Request-ID": request_id
}
payload = {"model": model, "input": input_text}
response = self.client.post(
f"{self.base_url}/embeddings",
headers=headers,
json=payload
)
latency_ms = int((time.time() - start_time) * 1000)
if response.status_code == 200:
response_data = response.json()
usage = response_data.get("usage", {})
prompt_tokens = usage.get("prompt_tokens", len(input_text) // 4)
self.log_request(
request_id=request_id,
model=model,
endpoint="/embeddings",
operation="embedding",
prompt_tokens=prompt_tokens,
completion_tokens=0,
latency_ms=latency_ms,
time_to_first_token=None,
user_id=user_id,
session_id=None,
request_metadata=metadata or {},
response_metadata={"dimensions": len(response_data.get("data", [{}])[0].get("embedding", []))}
)
return response_data
return response.json()
Khởi tạo client
def get_monitoring_client():
return AIMonitoringClient(
api_key="YOUR_HOLYSHEEP_API_KEY",
db_config={
"host": "localhost",
"database": "ai_monitoring",
"user": "postgres",
"password": "your_password"
}
)3. Real-time Dashboard Metrics
Với schema trên, đây là các query analytics quan trọng:
# analytics_queries.py
from datetime import datetime, timedelta
class UsageAnalytics:
def __init__(self, db_conn):
self.db_conn = db_conn
def get_cost_by_service(self, days: int = 30) -> list:
"""Chi phí theo từng service/endpoint"""
query = """
SELECT
operation,
model,
COUNT(*) as total_requests,
SUM(total_tokens) as total_tokens,
ROUND(SUM(total_cost)::numeric, 2) as total_cost_usd,
ROUND(AVG(total_cost)::numeric, 4) as avg_cost_per_request,
AVG(latency_ms)::integer as avg_latency_ms
FROM ai_api_requests
WHERE created_at >= NOW() - INTERVAL '%s days'
GROUP BY operation, model
ORDER BY total_cost_usd DESC
"""
with self.db_conn.cursor() as cur:
cur.execute(query, (days,))
return cur.fetchall()
def get_user_cost_breakdown(self, days: int = 30) -> list:
"""Chi phí theo từng user - critical cho multi-tenant apps"""
query = """
SELECT
user_id,
COUNT(*) as requests,
SUM(total_tokens) as tokens,
ROUND(SUM(total_cost)::numeric, 2) as cost_usd,
AVG(latency_ms)::integer as avg_latency
FROM ai_api_requests
WHERE user_id IS NOT NULL
AND created_at >= NOW() - INTERVAL '%s days'
GROUP BY user_id
HAVING SUM(total_cost) > 0
ORDER BY cost_usd DESC
LIMIT 100
"""
with self.db_conn.cursor() as cur:
cur.execute(query, (days,))
return cur.fetchall()
def get_top_consumers(self, limit: int = 10) -> list:
"""Top users tiêu thụ nhiều nhất - phát hiện anomalous usage"""
query = """
WITH hourly_spending AS (
SELECT
user_id,
DATE_TRUNC('hour', created_at) as hour,
SUM(total_cost) as hour_cost,
SUM(total_tokens) as hour_tokens
FROM ai_api_requests
WHERE created_at >= NOW() - INTERVAL '24 hours'
AND user_id IS NOT NULL
GROUP BY user_id, DATE_TRUNC('hour', created_at)
)
SELECT
user_id,
COUNT(DISTINCT hour) as active_hours,
SUM(hour_cost) as total_cost,
SUM(hour_tokens) as total_tokens,
ROUND(AVG(hour_cost)::numeric, 4) as avg_hourly_cost,
MAX(hour_cost) as peak_hour_cost
FROM hourly_spending
GROUP BY user_id
ORDER BY total_cost DESC
LIMIT %s
"""
with self.db_conn.cursor() as cur:
cur.execute(query, (limit,))
return cur.fetchall()
def detect_anomalies(self, std_dev_threshold: float = 3.0) -> list:
"""Phát hiện anomalous usage - chi phí vượt ngưỡng bất thường"""
query = """
WITH stats AS (
SELECT
user_id,
AVG(total_cost) as mean_cost,
STDDEV(total_cost) as std_cost
FROM ai_api_requests
WHERE created_at >= NOW() - INTERVAL '7 days'
AND user_id IS NOT NULL
GROUP BY user_id
)
SELECT
ar.user_id,
ar.request_id,
ar.total_cost,
s.mean_cost,
s.std_cost,
ROUND((ar.total_cost - s.mean_cost) / NULLIF(s.std_cost, 0), 2) as z_score,
ar.created_at
FROM ai_api_requests ar
JOIN stats s ON ar.user_id = s.user_id
WHERE ar.created_at >= NOW() - INTERVAL '24 hours'
AND ar.total_cost > s.mean_cost + (%s * s.std_cost)
ORDER BY ar.total_cost DESC
"""
with self.db_conn.cursor() as cur:
cur.execute(query, (std_dev_threshold,))
return cur.fetchall()
def get_performance_trends(self, model: str, days: int = 7) -> list:
"""Xu hướng performance theo thời gian"""
query = """
SELECT
DATE_TRUNC('hour', created_at) as hour,
COUNT(*) as requests,
AVG(latency_ms)::integer as avg_latency,
PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY latency_ms)::integer as p95,
PERCENTILE_CONT(0.99) WITHIN GROUP (ORDER BY latency_ms)::integer as p99
FROM ai_api_requests
WHERE model = %s
AND created_at >= NOW() - INTERVAL '%s days'
GROUP BY DATE_TRUNC('hour', created_at)
ORDER BY hour
"""
with self.db_conn.cursor() as cur:
cur.execute(query, (model, days))
return cur.fetchall()
def generate_cost_report(self, start_date: datetime, end_date: datetime) -> dict:
"""Tạo báo cáo chi phí chi tiết"""
query = """
SELECT
provider,
model,
operation,
COUNT(*) as total_requests,
SUM(prompt_tokens) as input_tokens,
SUM(completion_tokens) as output_tokens,
SUM(total_tokens) as total_tokens,
ROUND(SUM(input_cost)::numeric, 4) as input_cost,
ROUND(SUM(output_cost)::numeric, 4) as output_cost,
ROUND(SUM(total_cost)::numeric, 2) as total_cost,
AVG(latency_ms)::integer as avg_latency
FROM ai_api_requests
WHERE created_at BETWEEN %s AND %s
GROUP BY provider, model, operation
ORDER BY total_cost DESC
"""
with self.db_conn.cursor() as cur:
cur.execute(query, (start_date, end_date))
rows = cur.fetchall()
return {
"period": {"start": start_date.isoformat(), "end": end_date.isoformat()},
"breakdown": [
{
"provider": r[0], "model": r[1], "operation": r[2],
"requests": r[3], "input_tokens": r[4], "output_tokens": r[5],
"total_tokens": r[6], "input_cost": r[7], "output_cost": r[8],
"total_cost": r[9], "avg_latency": r[10]
}
for r in rows
],
"summary": {
"total_requests": sum(r[3] for r in rows),
"total_tokens": sum(r[6] for r in rows),
"total_cost": sum(r[9] for r in rows)
}
}So Sánh Chi Phí: HolySheep AI vs Providers Khác
Khi tôi chuyển từ provider cũ sang HolyShehep AI, khoản tiết kiệm thực tế đã thay đổi cách tôi nhìn về AI infrastructure:
| Model | Provider Cũ ($/MTok) | HolySheep AI ($/MTok) | Tiết Kiệm |
|---|---|---|---|
| GPT-4.1 | $60 | $8 | 86.7% |
| Claude Sonnet 4.5 | $90 | $15 | 83.3% |
| Gemini 2.5 Flash | $15 | $2.50 | 83.3% |
| DeepSeek V3.2 | $2.80 | $0.42 | 85% |
Thực tế từ dự án thương mại điện tử của tôi: Với 50 triệu token/tháng, chi phí giảm từ ~$2,800 xuống còn ~$380 — tiết kiệm $2,420/tháng = $29,040/năm.
HolySheep AI còn hỗ trợ thanh toán qua WeChat Pay và Alipay — rất thuận tiện cho các đối tác Trung Quốc hoặc developers có tài khoản Trung Quốc. Latency trung bình dưới 50ms, đảm bảo trải nghiệm người dùng mượt mà.
Triển Khai Production: Ví Dụ Thực Tế
Đây là cách tôi triển khai monitoring cho hệ thống RAG của khách hàng thương mại điện tử:
# production_example.py
from monitoring_client import get_monitoring_client
from analytics_queries import UsageAnalytics
import psycopg2
from datetime import datetime, timedelta
Khởi tạo monitoring
client = get_monitoring_client()
def handle_customer_query(user_id: str, query: str, session_id: str):
"""Xử lý query với full monitoring"""
# 1. Embed query để search
embed_response = client.embeddings(
model="text-embedding-3-small",
input_text=query,
user_id=user_id,
metadata={"type": "query_embedding", "query_length": len(query)}
)
# 2. Search trong vector database (giả lập)
retrieved_context = search_vector_db(embed_response["data"][0]["embedding"])
# 3. Gọi LLM với context
messages = [
{"role": "system", "content": "Bạn là trợ lý tư vấn sản phẩm..."},
{"role": "user", "content": f"Context: {retrieved_context}\n\nQuestion: {query}"}
]
chat_response = client.chat_completion(
model="deepseek-v3.2", # Model giá rẻ, hiệu quả cao
messages=messages,
user_id=user_id,
session_id=session_id,
metadata={
"type": "customer_support",
"context_length": len(retrieved_context),
"retrieval_method": "vector_search"
},
temperature=0.7,
max_tokens=500
)
return chat_response["choices"][0]["message"]["content"]
def search_vector_db(embedding: list) -> str:
"""Simulated vector search - thay bằng Pinecone/Chroma thực tế"""
return "Sản phẩm A giá 299k, sản phẩm B giá 499k..."
Analytics Dashboard
def generate_daily_report():
db_conn = psycopg2.connect(
host="localhost",
database="ai_monitoring",
user="postgres",
password="your_password"
)
analytics = UsageAnalytics(db_conn)
print("=" * 60)
print("BÁO CÁO CHI PHÍ AI - 30 NGÀY GẦN NHẤT")
print("=" * 60)
# Chi phí theo service
print("\n📊 Chi phí theo Service:")
print("-" * 60)
for op, model, reqs, tokens, cost, avg_req, avg_lat in analytics.get_cost_by_service(30):
print(f" {op:20} | {model:20} | {reqs:6} reqs | ${cost:8.2f} | {avg_lat}ms")
# Top consumers
print("\n👥 Top 10 Users:")
print("-" * 60)
for user, hours, cost, tokens, avg_hour, peak_hour in analytics.get_top_consumers(10):
print(f" {user:15} | ${cost:8.2f} | {tokens:10} tokens | Peak: ${peak_hour:.2f}/hr")
# Anomaly detection
print("\n⚠️ Phát hiện bất thường:")
anomalies = analytics.detect_anomalies()
if anomalies:
for user, req_id, cost, mean, std, z, time in anomalies:
print(f" User {user}: ${cost:.4f} (z-score: {z:.1f})")
else:
print(" Không có bất thường được phát hiện")
db_conn.close()
Chạy report hàng ngày
if __name__ == "__main__":
generate_daily_report()Lỗi Thường Gặp Và Cách Khắc Phục
1. Lỗi "Connection timeout" Hoặc Latency Cao
Nguyên nhân: Network route không tối ưu, queue bị congestion hoặc request quá lớn.
# Trước tiên, kiểm tra latency thực tế
import httpx
import time
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
def diagnose_latency():
"""Chẩn đoán vấn đề latency"""
# Test 1: Simple health check
client = httpx.Client(timeout=10.0)
start = time.time()
response = client.get(f"{HOLYSHEEP_BASE_URL}/models")
health_latency = int((time.time() - start) * 1000)
print(f"Health check latency: {health_latency}ms")
# Test 2: Small completion
headers = {"Authorization": f"Bearer {API_KEY}"}
start = time.time()
response = client.post(
f"{HOLYSHEEP_BASE_URL}/chat/completions",
headers=headers,
json={
"model": "deepseek-v3.2",
"messages": [{"role": "user", "content": "Hi"}],
"max_tokens": 10
}
)
small_latency = int((time.time() - start) * 1000)
print(f"Small request latency: {small_latency}ms")
# Test 3: Large completion (stress test)
large_prompt = "Xin chào " * 1000 # ~9KB
start = time.time()
response = client.post(
f"{HOLYSHEEP_BASE_URL}/chat/completions",
headers=headers,
json={
"model": "deepseek-v3.2",
"messages": [{"role": "user", "content": large_prompt}],
"max_tokens": 100
}
)
large_latency = int((time.time() - start) * 1000)
print(f"Large request latency: {large_latency}ms")
# Diagnosis
if health_latency > 1000:
print("❌ Vấn đề: Network route không tối ưu")
print(" Giải pháp: Thử proxy hoặc CDN gần server nhất")
elif large_latency > 5000:
print("⚠️ Cảnh báo: Request quá lớn")
print(" Giải pháp: Giảm max_tokens hoặc split request")
else:
print("✅ Latency bình thường")
Nếu latency vẫn cao, thử connection pooling
def create_optimized_client():
"""Client với connection pooling cho latency thấp hơn"""
return httpx.Client(
timeout=30.0,
limits=httpx.Limits(
max_connections=100,
max_keepalive_connections=20,
keepalive_expiry=30.0
),
http2=True # HTTP/2 cho parallel requests
)2. Lỗi "Invalid API Key" Hoặc Authentication Failed
Nguyên nhân: Key sai format, đã bị revoke, hoặc không đủ permissions.
# Kiểm tra và validate API key
def validate_api_key(api_key: str) -> dict:
"""Validate API key với HolySheep"""
import httpx
client = httpx.Client(timeout=10.0)
# Method 1: Test với models endpoint
try:
response = client.get(
"https://api.holysheep.ai/v1/models",
headers={"Authorization": f"Bearer {api_key}"}
)
if response.status_code == 200:
models = response.json()
return {
"valid": True,
"models_available": len(models.get("data", [])),
"message": "API key hợp lệ"
}
elif response.status_code == 401:
return {
"valid": False,
"error": "INVALID_KEY",
"message": "API key không hợp lệ hoặc đã bị revoke"
}
except httpx.ConnectError:
return {
"valid": False,
"error": "CONNECTION_ERROR",
"message": "Không thể kết nối - kiểm tra network"
}
# Method 2: Test với dummy request
response = client.post(
"https://api.holysheep.ai/v1/chat/completions",
headers={"Authorization": f"Bearer {api_key}"},
json={
"model": "deepseek-v3.2",
"messages": [{"role": "user", "content": "test"}],
"max_tokens": 1
}
)
if response.status_code == 200:
return {"valid": True, "message": "API key hoạt động tốt"}
elif response.status_code == 401:
return {"valid": False, "error": "UNAUTHORIZED", "message": "Authentication failed"}
elif response.status_code == 429:
return {"valid": True, "warning": "RATE_LIMITED", "message": "Key hợp lệ nhưng bị rate limit"}
return {"valid": False, "error": "UNKNOWN", "message": response.text}
Cách lấy API key mới nếu cần
def get_new_api_key_instructions():
return """
Cách lấy API key mới từ HolySheep AI:
1. Truy cập: https://www.holysheep.ai/register
2. Đăng nhập hoặc tạo tài khoản mới
3. Vào Dashboard -> API Keys
4. Click "Create New Key"
5. Copy key (bắt đầu bằng 'hs_' hoặc 'sk-')
⚠️ Lưu ý:
- Key chỉ hiển thị 1 lần duy nhất
- Không commit key vào source code
- Sử dụng environment variable thay vì hardcode
"""3. Lỗi "Rate Limit Exceeded"
Nguyên nhân: Gửi quá nhiều request trong thời gian ngắn, vượt quota.
# Implement retry logic với exponential backoff
import time
import asyncio
from typing import Callable, Any
from functools import wraps
class RateLimitHandler:
def __init__(self, max_retries: int = 3, base_delay: float = 1.0):
self.max_retries = max_retries
self.base_delay = base_delay
def retry_with_backoff(self, func: Callable) -> Callable:
"""Decorator cho retry logic"""
@wraps(func)
def wrapper(*args, **kwargs) -> Any:
last_exception = None
for attempt in range(self.max_retries + 1):
try:
return func(*args, **kwargs)
except httpx.HTTPStatusError as e:
if e.response.status_code == 429:
# Rate limited - extract retry delay từ header
retry_after = e.response.headers.get("Retry-After", "60")
try:
delay = int(retry_after)
except ValueError:
delay = int(self.base_delay * (2 ** attempt))
print(f"⚠️ Rate limited. Retry sau {delay