Đêm mưa tháng 11/2025, hệ thống chatbot của một startup EdTech Việt Nam đột nhiên chậm như rùa bò. Đội kỹ thuật nhận được hàng trăm alert: ConnectionError: timeout after 30000ms, 429 Too Many Requests, 504 Gateway Timeout. Kỹ sư trực đêm — một thanh niên 26 tuổi mới vào nghề — loay hoay không biết tắc ở đâu: mạng? server? hay chính cách gọi API của mình?
Câu chuyện này — thật ra — xảy ra với rất nhiều dev team Việt Nam. Và giải pháp không nằm ở việc thêm server hay tăng timeout. Mà nằm ở một thứ tưởng như nhàm chán nhưng cực kỳ quyền năng: log phân tích.
Bài viết này sẽ hướng dẫn bạn cách build một hệ thống phân tích log API call hoàn chỉnh, giúp bạn không chỉ debug lỗi mà còn phát hiện bottleneck trước khi nó trở thành sự cố.
Tại sao log là "bản đồ vàng" cho performance?
Khi bạn gọi API AI model qua HolySheep AI, mỗi request đều tạo ra metadata quý giá: thời gian phản hồi, HTTP status code, token usage, round-trip time (RTT). Chỉ cần thu thập và phân tích đúng cách, bạn có thể:
- Phát hiện latency spike trước khi user phàn nàn
- Tối ưu batch size và concurrent requests
- Giảm chi phí API call đến 40% bằng cách loại bỏ request thừa
- Đối phó với rate limit bằng chiến lược exponential backoff thông minh
Kiến trúc hệ thống thu thập log
Trước khi code, hãy hiểu luồng dữ liệu:
+----------------+ +------------------+ +------------------+
| Application | --> | Log Collector | --> | Analysis Engine |
| (Your API) | | (Structured) | | (Aggregations) |
+----------------+ +------------------+ +------------------+
| | |
JSON logs Timestamp + Metrics:
- request_id metadata - p50/p95/p99 latency
- duration_ms - request_id - error rate
- status_code - trace_id - token distribution
- tokens_used - cost estimation
Setup project và cài đặt dependencies
mkdir ai-log-analyzer
cd ai-log-analyzer
python3 -m venv venv
source venv/bin/activate # Linux/Mac
Hoặc: venv\Scripts\activate # Windows
pip install requests pandas matplotlib seaborn python-json-logger
requests: gọi API
pandas: phân tích dữ liệu
matplotlib/seaborn: visualize
python-json-logger: structured logging
1. Structured Logging — Nền tảng của mọi thứ
Đây là điều tôi đã rút ra sau 3 năm debug hệ thống AI: unstructured log là thảm họa. Bạn không thể search, filter, hay aggregate text thuần túy. Luôn dùng structured JSON log.
# logger_setup.py
import logging
import json
import sys
from datetime import datetime
from pythonjsonlogger import jsonlogger
class HolySheepLogFormatter(jsonlogger.JsonFormatter):
"""Custom formatter cho HolySheep API calls"""
def add_fields(self, log_record, record, message_dict):
super().add_fields(log_record, record, message_dict)
log_record['@timestamp'] = datetime.utcnow().isoformat() + 'Z'
log_record['service'] = 'ai-api-gateway'
log_record['environment'] = 'production'
log_record['provider'] = 'holysheep'
def setup_logger(name: str = 'ai_logger', level=logging.INFO):
logger = logging.getLogger(name)
logger.setLevel(level)
# Console handler với JSON format
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(HolySheepLogFormatter(
'%(message)s',
rename_fields={'levelname': 'level', 'name': 'logger'}
))
logger.addHandler(handler)
# File handler cho production
file_handler = logging.FileHandler('api_calls.jsonl')
file_handler.setFormatter(HolySheepLogFormatter('%(message)s'))
logger.addHandler(file_handler)
return logger
ai_logger = setup_logger()
2. Wrapper class — Centralize mọi API calls
Đây là pattern quan trọng nhất. Thay vì scatter requests khắp nơi, bạn wrap tất cả trong một class. Mọi call đều đi qua đây → mọi log đều structured.
# holysheep_client.py
import requests
import time
import json
from datetime import datetime
from typing import Optional, Dict, Any, List
from logger_setup import ai_logger
class HolySheepAIClient:
"""Wrapper cho HolySheep AI API với structured logging tích hợp"""
BASE_URL = "https://api.holysheep.ai/v1"
# Pricing reference 2026 (USD per 1M tokens)
PRICING = {
'gpt-4.1': {'input': 8.00, 'output': 8.00},
'claude-sonnet-4.5': {'input': 15.00, 'output': 15.00},
'gemini-2.5-flash': {'input': 2.50, 'output': 2.50},
'deepseek-v3.2': {'input': 0.42, 'output': 0.42},
}
def __init__(self, api_key: str):
self.api_key = api_key
self.session = requests.Session()
self.session.headers.update({
'Authorization': f'Bearer {api_key}',
'Content-Type': 'application/json'
})
self.request_count = 0
self.total_cost = 0.0
def _log_request(self, model: str, payload: Dict, start_time: float):
"""Log mọi request với timing và metadata"""
duration_ms = (time.time() - start_time) * 1000
# Estimate tokens
input_tokens = len(json.dumps(payload)) // 4 # rough estimate
log_data = {
'event': 'api_request',
'request_id': f'req_{self.request_count}_{int(start_time*1000)}',
'model': model,
'input_tokens_estimate': input_tokens,
'duration_ms': round(duration_ms, 2),
'endpoint': f'{self.BASE_URL}/chat/completions',
'payload_size_bytes': len(json.dumps(payload).encode()),
}
ai_logger.info('API request initiated', extra=log_data)
return log_data['request_id']
def _log_response(self, request_id: str, status_code: int,
duration_ms: float, response_data: Optional[Dict] = None,
error: Optional[str] = None):
"""Log mọi response với status và metrics"""
log_data = {
'event': 'api_response',
'request_id': request_id,
'status_code': status_code,
'duration_ms': round(duration_ms, 2),
'success': 200 <= status_code < 300,
}
if response_data:
usage = response_data.get('usage', {})
log_data.update({
'input_tokens': usage.get('prompt_tokens', 0),
'output_tokens': usage.get('completion_tokens', 0),
'total_tokens': usage.get('total_tokens', 0),
})
# Calculate cost
model = response_data.get('model', 'unknown')
if model in self.PRICING:
cost = (log_data['input_tokens'] * self.PRICING[model]['input'] +
log_data['output_tokens'] * self.PRICING[model]['output']) / 1_000_000
log_data['estimated_cost_usd'] = round(cost, 6)
self.total_cost += cost
if error:
log_data['error'] = error
log_data['error_type'] = error.__class__.__name__
ai_logger.info('API response received', extra=log_data)
def chat_completions(self, model: str, messages: List[Dict],
temperature: float = 0.7, max_tokens: int = 1000,
retry_count: int = 3) -> Dict[str, Any]:
"""
Gọi Chat Completions API với automatic logging và retry logic
"""
self.request_count += 1
start_time = time.time()
payload = {
'model': model,
'messages': messages,
'temperature': temperature,
'max_tokens': max_tokens,
}
request_id = self._log_request(model, payload, start_time)
for attempt in range(retry_count):
try:
response = self.session.post(
f'{self.BASE_URL}/chat/completions',
json=payload,
timeout=30
)
duration_ms = (time.time() - start_time) * 1000
if response.status_code == 200:
data = response.json()
self._log_response(request_id, 200, duration_ms, data)
return data
elif response.status_code == 429:
# Rate limit - exponential backoff
retry_after = int(response.headers.get('Retry-After', 2 ** attempt))
ai_logger.warning('Rate limited', extra={
'request_id': request_id,
'attempt': attempt + 1,
'retry_after_seconds': retry_after
})
if attempt < retry_count - 1:
time.sleep(retry_after)
continue
elif response.status_code == 401:
self._log_response(request_id, 401, duration_ms, error=Exception('Invalid API key'))
raise Exception('401 Unauthorized: Kiểm tra API key của bạn tại holysheep.ai')
elif response.status_code == 500:
if attempt < retry_count - 1:
time.sleep(2 ** attempt)
continue
# Log other errors
self._log_response(request_id, response.status_code, duration_ms,
error=Exception(response.text[:200]))
return {'error': response.text, 'status_code': response.status_code}
except requests.exceptions.Timeout as e:
duration_ms = (time.time() - start_time) * 1000
self._log_response(request_id, 408, duration_ms, error=e)
if attempt == retry_count - 1:
raise
except requests.exceptions.ConnectionError as e:
duration_ms = (time.time() - start_time) * 1000
self._log_response(request_id, 503, duration_ms, error=e)
ai_logger.error('Connection failed', extra={
'request_id': request_id,
'error': str(e)
})
time.sleep(1)
return {'error': 'Max retries exceeded'}
def get_stats(self) -> Dict[str, Any]:
"""Trả về thống kê session hiện tại"""
return {
'total_requests': self.request_count,
'total_cost_usd': round(self.total_cost, 6),
'avg_cost_per_request': round(self.total_cost / max(self.request_count, 1), 6)
}
3. Sử dụng client — Ví dụ thực tế
# example_usage.py
from holysheep_client import HolySheepAIClient
Khởi tạo client với API key của bạn
Đăng ký tại: https://www.holysheep.ai/register
client = HolySheepAIClient(api_key='YOUR_HOLYSHEEP_API_KEY')
Ví dụ 1: Simple chat
messages = [
{'role': 'system', 'content': 'Bạn là trợ lý phân tích log chuyên nghiệp.'},
{'role': 'user', 'content': 'Phân tích log này và đưa ra recommendations: [dữ liệu log thực tế]'}
]
response = client.chat_completions(
model='deepseek-v3.2', # Model giá rẻ, hiệu năng cao - chỉ $0.42/1M tokens
messages=messages,
temperature=0.3,
max_tokens=500
)
if 'error' not in response:
print(f"Response: {response['choices'][0]['message']['content']}")
print(f"Tokens used: {response['usage']['total_tokens']}")
else:
print(f"Lỗi: {response['error']}")
Ví dụ 2: Batch processing với progress tracking
import time
test_queries = [
"Phân tích latency spike trên server A",
"Nguyên nhân 429 rate limit",
"Tối ưu batch size cho 1000 requests",
"So sánh chi phí giữa các provider",
]
results = []
for i, query in enumerate(test_queries):
print(f"Processing {i+1}/{len(test_queries)}: {query[:30]}...")
resp = client.chat_completions(
model='deepseek-v3.2',
messages=[{'role': 'user', 'content': query}]
)
results.append(resp)
time.sleep(0.1) # Tránh burst
In thống kê
stats = client.get_stats()
print(f"\n=== Session Stats ===")
print(f"Total requests: {stats['total_requests']}")
print(f"Total cost: ${stats['total_cost_usd']}")
print(f"Avg cost: ${stats['avg_cost_per_request']}")
4. Phân tích log và detect bottleneck
Sau khi thu thập đủ log (tối thiểu 24h), đây là script phân tích để tìm bottleneck.
# log_analyzer.py
import json
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
from datetime import datetime, timedelta
from collections import defaultdict
import numpy as np
class APILogAnalyzer:
"""Phân tích log để tìm performance bottlenecks"""
def __init__(self, log_file: str = 'api_calls.jsonl'):
self.log_file = log_file
self.df = None
self.load_logs()
def load_logs(self):
"""Load và parse JSON logs"""
records = []
with open(self.log_file, 'r') as f:
for line in f:
try:
records.append(json.loads(line.strip()))
except json.JSONDecodeError:
continue
self.df = pd.DataFrame(records)
# Parse timestamp
if '@timestamp' in self.df.columns:
self.df['timestamp'] = pd.to_datetime(self.df['timestamp'])
# Filter chỉ events liên quan
self.df = self.df[self.df['event'].isin(['api_request', 'api_response'])]
def analyze_latency(self) -> dict:
"""Phân tích latency distribution"""
responses = self.df[self.df['event'] == 'api_response'].copy()
if responses.empty:
return {'error': 'No response data'}
latency_stats = {
'count': len(responses),
'mean_ms': responses['duration_ms'].mean(),
'median_ms': responses['duration_ms'].median(),
'p50_ms': responses['duration_ms'].quantile(0.50),
'p95_ms': responses['duration_ms'].quantile(0.95),
'p99_ms': responses['duration_ms'].quantile(0.99),
'max_ms': responses['duration_ms'].max(),
'min_ms': responses['duration_ms'].min(),
'std_ms': responses['duration_ms'].std(),
}
return latency_stats
def detect_latency_spikes(self, threshold_pct: float = 95) -> pd.DataFrame:
"""Phát hiện các request có latency bất thường"""
responses = self.df[self.df['event'] == 'api_response'].copy()
if responses.empty:
return pd.DataFrame()
threshold = responses['duration_ms'].quantile(threshold_pct / 100)
spikes = responses[responses['duration_ms'] > threshold]
return spikes[['@timestamp', 'request_id', 'duration_ms', 'status_code',
'model', 'input_tokens', 'output_tokens']].sort_values('duration_ms', ascending=False)
def analyze_errors(self) -> dict:
"""Phân tích error distribution"""
responses = self.df[self.df['event'] == 'api_response'].copy()
errors = responses[responses['status_code'] >= 400]
error_stats = {
'total_requests': len(responses),
'total_errors': len(errors),
'error_rate_pct': round(len(errors) / max(len(responses), 1) * 100, 2),
'errors_by_code': errors['status_code'].value_counts().to_dict() if not errors.empty else {},
'errors_by_model': errors.groupby('model').size().to_dict() if not errors.empty else {},
}
# Error type analysis
if 'error_type' in errors.columns:
error_stats['errors_by_type'] = errors['error_type'].value_counts().head(10).to_dict()
return error_stats
def analyze_cost(self) -> dict:
"""Phân tích chi phí theo model và thời gian"""
responses = self.df[self.df['event'] == 'api_response'].copy()
responses_with_cost = responses[responses.get('estimated_cost_usd', pd.Series()).notna()]
if responses_with_cost.empty:
return {'message': 'No cost data available'}
cost_by_model = responses_with_cost.groupby('model').agg({
'estimated_cost_usd': 'sum',
'input_tokens': 'sum',
'output_tokens': 'sum',
'request_id': 'count'
}).rename(columns={'request_id': 'request_count'})
return {
'total_cost_usd': round(responses_with_cost['estimated_cost_usd'].sum(), 6),
'cost_by_model': cost_by_model.to_dict(),
'avg_cost_per_request': round(responses_with_cost['estimated_cost_usd'].mean(), 6)
}
def identify_bottlenecks(self) -> list:
"""Tổng hợp các bottleneck đã phát hiện"""
bottlenecks = []
# Check 1: High latency
latency = self.analyze_latency()
if latency.get('p99_ms', 0) > 5000:
bottlenecks.append({
'type': 'HIGH_LATENCY',
'severity': 'HIGH',
'description': f'p99 latency = {latency["p99_ms"]:.0f}ms (target: <5000ms)',
'recommendation': 'Xem xét batch requests hoặc switch sang model có lower latency'
})
# Check 2: High error rate
errors = self.analyze_errors()
if errors.get('error_rate_pct', 0) > 5:
bottlenecks.append({
'type': 'HIGH_ERROR_RATE',
'severity': 'CRITICAL',
'description': f'Error rate = {errors["error_rate_pct"]}% (threshold: 5%)',
'recommendation': 'Kiểm tra API key, network connectivity, và rate limits'
})
# Check 3: Cost optimization
cost = self.analyze_cost()
if 'total_cost_usd' in cost and cost['total_cost_usd'] > 0:
# Check if using expensive models for simple tasks
responses = self.df[self.df['event'] == 'api_response']
expensive_model_usage = responses[responses['model'] == 'claude-sonnet-4.5']
if len(expensive_model_usage) > 0:
cheap_model_usage = responses[responses['model'] == 'deepseek-v3.2']
if len(cheap_model_usage) < len(expensive_model_usage):
bottlenecks.append({
'type': 'COST_OPTIMIZATION',
'severity': 'MEDIUM',
'description': f'Using expensive models (Claude Sonnet $15/MTok) more than cheap ones (DeepSeek $0.42/MTok)',
'recommendation': 'Switch simple tasks sang deepseek-v3.2 để tiết kiệm 97% chi phí'
})
# Check 4: Rate limit issues
rate_limit_errors = errors.get('errors_by_code', {}).get(429, 0)
if rate_limit_errors > 10:
bottlenecks.append({
'type': 'RATE_LIMIT',
'severity': 'HIGH',
'description': f'{rate_limit_errors} requests bị 429 rate limit',
'recommendation': 'Implement exponential backoff và request queuing'
})
return bottlenecks
def generate_report(self) -> str:
"""Generate báo cáo phân tích đầy đủ"""
report_lines = [
"=" * 60,
"API CALL ANALYSIS REPORT",
"=" * 60,
f"Generated at: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
f"Total events analyzed: {len(self.df)}",
"",
]
# Latency section
latency = self.analyze_latency()
report_lines.extend([
"-" * 40,
"LATENCY ANALYSIS",
"-" * 40,
f"Total responses: {latency.get('count', 0)}",
f"Mean: {latency.get('mean_ms', 0):.2f}ms",
f"Median (p50): {latency.get('median_ms', 0):.2f}ms",
f"p95: {latency.get('p95_ms', 0):.2f}ms",
f"p99: {latency.get('p99_ms', 0):.2f}ms",
f"Max: {latency.get('max_ms', 0):.2f}ms",
"",
])
# Errors section
errors = self.analyze_errors()
report_lines.extend([
"-" * 40,
"ERROR ANALYSIS",
"-" * 40,
f"Error rate: {errors.get('error_rate_pct', 0)}%",
f"Errors by HTTP code: {errors.get('errors_by_code', {})}",
"",
])
# Cost section
cost = self.analyze_cost()
if 'total_cost_usd' in cost:
report_lines.extend([
"-" * 40,
"COST ANALYSIS",
"-" * 40,
f"Total cost: ${cost['total_cost_usd']:.6f}",
f"Avg per request: ${cost.get('avg_cost_per_request', 0):.6f}",
"",
])
# Bottlenecks section
bottlenecks = self.identify_bottlenecks()
report_lines.extend([
"-" * 40,
"IDENTIFIED BOTTLENECKS",
"-" * 40,
])
if bottlenecks:
for i, b in enumerate(bottlenecks, 1):
report_lines.extend([
f"{i}. [{b['severity']}] {b['type']}",
f" {b['description']}",
f" → {b['recommendation']}",
"",
])
else:
report_lines.append("No critical bottlenecks detected.")
report_lines.append("")
report_lines.append("=" * 60)
return "\n".join(report_lines)
def visualize_latency(self, output_file: str = 'latency_distribution.png'):
"""Tạo biểu đồ latency distribution"""
responses = self.df[self.df['event'] == 'api_response'].copy()
if responses.empty:
print("No data to visualize")
return
fig, axes = plt.subplots(2, 2, figsize=(14, 10))
# 1. Latency histogram
axes[0, 0].hist(responses['duration_ms'], bins=50, edgecolor='black', alpha=0.7)
axes[0, 0].set_xlabel('Latency (ms)')
axes[0, 0].set_ylabel('Frequency')
axes[0, 0].set_title('Latency Distribution')
axes[0, 0].axvline(responses['duration_ms'].quantile(0.95), color='red',
linestyle='--', label='p95')
axes[0, 0].legend()
# 2. Latency over time
if 'timestamp' in responses.columns:
responses_sorted = responses.sort_values('timestamp')
axes[0, 1].plot(responses_sorted['timestamp'], responses_sorted['duration_ms'],
alpha=0.6, marker='.')
axes[0, 1].set_xlabel('Time')
axes[0, 1].set_ylabel('Latency (ms)')
axes[0, 1].set_title('Latency Over Time')
axes[0, 1].tick_params(axis='x', rotation=45)
# 3. Latency by model
if 'model' in responses.columns:
model_latency = responses.groupby('model')['duration_ms'].mean().sort_values()
axes[1, 0].barh(model_latency.index, model_latency.values)
axes[1, 0].set_xlabel('Mean Latency (ms)')
axes[1, 0].set_title('Average Latency by Model')
# 4. Error rate pie chart
error_responses = responses[responses['status_code'] >= 400]
if not error_responses.empty:
error_counts = error_responses['status_code'].value_counts()
axes[1, 1].pie(error_counts.values, labels=error_counts.index, autopct='%1.1f%%')
axes[1, 1].set_title('Error Distribution by Status Code')
plt.tight_layout()
plt.savefig(output_file, dpi=150)
print(f"Chart saved to {output_file}")
Main execution
if __name__ == '__main__':
analyzer = APILogAnalyzer('api_calls.jsonl')
# Generate text report
print(analyzer.generate_report())
# Generate visualization
analyzer.visualize_latency()
# Save bottlenecks to JSON for automation
bottlenecks = analyzer.identify_bottlenecks()
with open('bottlenecks_detected.json', 'w') as f:
json.dump(bottlenecks, f, indent=2)
5. Chiến lược tối ưu hóa dựa trên log
5.1 Tối ưu batch size
Phân tích log cho thấy nhiều dev gọi từng request riêng lẻ thay vì batch. Với HolySheep AI, batch request có thể giảm RTT overhead đáng kể.
# batch_optimizer.py
from holysheep_client import HolySheepAIClient
from concurrent.futures import ThreadPoolExecutor, as_completed
import time
class BatchOptimizer:
"""Tối ưu hóa batch requests dựa trên latency analysis"""
def __init__(self, client: HolySheepAIClient):
self.client = client
def sequential_batch(self, queries: list, model: str = 'deepseek-v3.2') -> list:
"""Sequential processing - baseline"""
results = []
start = time.time()
for q in queries:
resp = self.client.chat_completions(model=model, messages=[{'role': 'user', 'content': q}])
results.append(resp)
total_time = time.time() - start
return {'results': results, 'total_time': total_time, 'avg_per_request': total_time / len(queries)}
def parallel_batch(self, queries: list, model: str = 'deepseek-v3.2',
max_workers: int = 5) -> list:
"""Parallel processing với thread pool"""
results = []
start = time.time()
with ThreadPoolExecutor(max_workers=max_workers) as executor:
futures = {
executor.submit(
self.client.chat_completions,
model=model,
messages=[{'role': 'user', 'content': q}]
): i for i, q in enumerate(queries)
}
results = [None] * len(queries)
for future in as_completed(futures):
idx = futures[future]
try:
results[idx] = future.result()
except Exception as e:
results[idx] = {'error': str(e)}
total_time = time.time() - start
return {'results': results, 'total_time': total_time, 'avg_per_request': total_time / len(queries)}
def smart_batch(self, queries: list, model: str = 'deepseek-v3.2') -> dict:
"""
Smart batching: combine multiple queries into single API call
Chỉ hoạt động với models hỗ trợ system prompt mạnh
"""
# Phân nhóm queries có pattern tương tự
grouped_queries = []
batch_size = 10
for i in range(0, len(queries), batch_size):
batch = queries[i:i+batch_size]
# Format as numbered list
combined_prompt = "Phân tích lần lượt các yêu cầu sau:\n\n"
for idx, q in enumerate(batch, 1):
combined_prompt += f"{idx}. {q}\n"
resp = self.client.chat_completions(
model=model,
messages=[
{'role': 'system', 'content': 'Bạn là chuyên gia phân tích. Trả lời ngắn gọn, đánh số theo thứ tự.'},
{'role': 'user', 'content': combined_prompt}
],
max_tokens=2000
)
grouped_queries.append(resp)
return {'results': grouped_queries, 'batches': len(grouped_queries)}
Demo optimization comparison
if __name__ == '__main__':
client = HolySheepAIClient(api_key='YOUR_HOLYSHEEP_API_KEY')
optimizer = BatchOptimizer(client)
test_queries = [
"Phân tích log error rate",
"Tối ưu batch size",
"So sánh chi phí API",
"Xử lý rate limit",
"Caching strategy",
] * 2 # 10 queries
# Sequential baseline
seq_result = optimizer.sequential_batch(test_queries[:5])
print(f"Sequential: {seq_result['total_time']:.2f}s total, {seq_result['avg_per_request']:.2f}s per request")
# Parallel
par_result = optimizer.parallel_batch(test_queries[:5], max_workers=5)
print(f"Parallel: {par_result['total_time']:.2f}s total, {par_result['avg_per_request']:.2f}s per request")
# Smart batch
smart_result = optimizer.smart_batch(test_queries[:5])
print(f"Smart Batch: {len(smart_result['results'])} API calls cho {len(test_queries[:5])} queries")
5.2 Exponential Backoff thông minh
Từ log thực tế, tôi thấy nhiều team retry ngay lập tức khi gặp 429, dẫn đến thundering herd. Đây là implementation đúng:
# smart_retry.py
import time
import random
from functools import wraps
from typing import Callable, Any
class SmartRetry:
"""Exponential backoff với jitter - tránh thundering herd"""
def __init__(self, base_delay: float = 1.0, max_delay: float = 60.0,
max_retries: int = 5, jitter: bool = True):
self.base_delay = base_delay
self.max_delay = max_delay
self.max_retries = max_retries
self.jitter = jitter
def calculate_delay(self, attempt: int, retry_after: int = None) -> float:
"""Tính delay với exponential backoff + optional jitter"""
# Nếu server gửi Retry-After header, ưu tiên dùng nó
if retry_after:
return min(retry_after, self.max_delay)
# Exponential backoff: base * 2^attempt
delay = self.base_delay * (2 ** attempt)
delay = min(delay, self.max_delay)
# Thêm jitter để tránh synchronized retries
if self.jitter:
# Full jitter: random between 0 and calculated delay
delay = random.uniform(0, delay)
return 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