Tối hôm qua, mình đang chạy một batch job xử lý 500 tài liệu tiếng Việt bằng Claude API. Đến phút thứ 47, terminal bỗng nhiên hiển thị dòng chữ đỏ lòm: ConnectionError: timeout after 30s. Toàn bộ job bị hủy, 47 phút CPU time đã tiêu tốn — trôi theo mây.
Sau sự cố đó, mình quyết định xây dựng một hệ thống script hóa hoàn chỉnh với khả năng retry thông minh, rate limiting tự động, và quan trọng nhất — tiết kiệm 85% chi phí với HolySheep AI.
Tại Sao Cần Script Hóa AI Tasks?
Trong thực chiến production, bạn sẽ gặp những vấn đề kinh điển:
- Rate Limit exceeded — API provider chặn request liên tục
- Partial failure — 1/10 tasks thất bại, phải chạy lại toàn bộ
- Cost explosion — Không kiểm soát được budget khi scale
- No idempotency — Retry sai cách tạo duplicate data
Bài viết này sẽ hướng dẫn bạn xây dựng hệ thống batch processing với error handling chuyên nghiệp, tiết kiệm chi phí đáng kể.
Kiến Trúc Tổng Quan
Mình thiết kế hệ thống gồm 4 layers:
- Batch Controller — Quản lý queue và parallel execution
- Rate Limiter — Kiểm soát RPM/TPM tự động
- Retry Engine — Exponential backoff với jitter
- Result Aggregator — Tổng hợp và lưu trữ kết quả
Cài Đặt Môi Trường
# Tạo virtual environment
python3 -m venv cline-batch-env
source cline-batch-env/bin/activate
Cài đặt dependencies
pip install httpx aiofiles pydantic python-dotenv tqdm
Cấu trúc project
mkdir -p cline-batch/{src,config,data,logs}
cd cline-batchScript Cốt Lõi: Batch Processor
#!/usr/bin/env python3
"""
HolySheep AI Batch Processing Script
Author: HolySheep AI Team
Version: 1.0.0
"""
import asyncio
import json
import time
import hashlib
from datetime import datetime
from pathlib import Path
from typing import List, Dict, Optional
import httpx
from tqdm import tqdm
=== CẤU HÌNH HOLYSHEEP AI ===
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY" # Thay bằng API key của bạn
Rate limiting config
MAX_REQUESTS_PER_MINUTE = 60
MAX_TOKENS_PER_MINUTE = 150_000
Retry config
MAX_RETRIES = 3
INITIAL_BACKOFF = 1.0 # Giây
MAX_BACKOFF = 32.0
class HolySheepBatchProcessor:
"""Batch processor với error handling và retry logic"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = BASE_URL
self.request_times: List[float] = []
self.token_counts: List[tuple] = [] # (timestamp, tokens)
self.results: List[Dict] = []
self.errors: List[Dict] = []
async def chat_completion(
self,
messages: List[Dict],
model: str = "gpt-4.1",
temperature: float = 0.7,
max_tokens: int = 2048
) -> Dict:
"""Gửi request lên HolySheep AI với rate limiting và retry"""
endpoint = f"{self.base_url}/chat/completions"
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
# Tính toán estimated tokens cho rate limiting
estimated_tokens = sum(
len(str(m.get("content", ""))) // 4
for m in messages
) + max_tokens
# Chờ nếu cần thiết (rate limiting)
await self._wait_for_rate_limit(estimated_tokens)
# Retry loop với exponential backoff
last_error = None
for attempt in range(MAX_RETRIES):
try:
async with httpx.AsyncClient(timeout=60.0) as client:
response = await client.post(
endpoint,
headers=headers,
json=payload
)
if response.status_code == 200:
return response.json()
elif response.status_code == 429:
# Rate limit - tăng backoff
wait_time = min(INITIAL_BACKOFF * (2 ** attempt), MAX_BACKOFF)
wait_time += asyncio.random() * 0.5 # Jitter
print(f"⚠️ Rate limited. Chờ {wait_time:.1f}s...")
await asyncio.sleep(wait_time)
continue
elif response.status_code == 401:
raise Exception("❌ API Key không hợp lệ. Kiểm tra YOUR_HOLYSHEEP_API_KEY")
elif response.status_code >= 500:
# Server error - retry
wait_time = INITIAL_BACKOFF * (2 ** attempt)
print(f"⚠️ Server error {response.status_code}. Retry sau {wait_time}s...")
await asyncio.sleep(wait_time)
continue
else:
error_detail = response.json().get("error", {})
raise Exception(f"Lỗi {response.status_code}: {error_detail}")
except httpx.TimeoutException as e:
last_error = e
wait_time = INITIAL_BACKOFF * (2 ** attempt)
print(f"⏱️ Timeout. Retry attempt {attempt + 1}/{MAX_RETRIES}...")
await asyncio.sleep(wait_time)
except httpx.ConnectError as e:
last_error = e
wait_time = 5 + attempt * 2 # Base 5s + delay
print(f"🔌 Connection error. Chờ {wait_time}s...")
await asyncio.sleep(wait_time)
raise Exception(f"Failed sau {MAX_RETRIES} attempts: {last_error}")
async def _wait_for_rate_limit(self, estimated_tokens: int):
"""Đảm bảo không vượt rate limit"""
current_time = time.time()
# Clean up old entries (> 1 phút)
self.request_times = [t for t in self.request_times if current_time - t < 60]
self.token_counts = [
(t, tokens) for t, tokens in self.token_counts
if current_time - t < 60
]
# Check requests per minute
if len(self.request_times) >= MAX_REQUESTS_PER_MINUTE:
oldest = self.request_times[0]
wait_time = 60 - (current_time - oldest)
if wait_time > 0:
print(f"📊 RPM limit. Chờ {wait_time:.1f}s...")
await asyncio.sleep(wait_time)
# Check tokens per minute
current_tokens = sum(tokens for _, tokens in self.token_counts)
if current_tokens + estimated_tokens > MAX_TOKENS_PER_MINUTE:
oldest = self.token_counts[0][0] if self.token_counts else current_time
wait_time = 60 - (current_time - oldest)
if wait_time > 0:
print(f"📊 TPM limit. Chờ {wait_time:.1f}s...")
await asyncio.sleep(wait_time)
# Ghi nhận request
self.request_times.append(time.time())
self.token_counts.append((time.time(), estimated_tokens))
async def process_batch(
self,
tasks: List[Dict],
model: str = "gpt-4.1",
max_concurrent: int = 5
) -> Dict:
"""Xử lý batch với concurrency control"""
print(f"🚀 Bắt đầu batch: {len(tasks)} tasks")
print(f"💰 Model: {model} | Concurrency: {max_concurrent}")
semaphore = asyncio.Semaphore(max_concurrent)
async def process_single(task: Dict, index: int) -> Dict:
async with semaphore:
task_id = task.get("id", f"task_{index}")
start_time = time.time()
try:
messages = task.get("messages", [])
result = await self.chat_completion(messages, model)
processing_time = time.time() - start_time
return {
"id": task_id,
"status": "success",
"result": result,
"processing_time": processing_time,
"timestamp": datetime.now().isoformat()
}
except Exception as e:
processing_time = time.time() - start_time
error_entry = {
"id": task_id,
"status": "failed",
"error": str(e),
"processing_time": processing_time,
"timestamp": datetime.now().isoformat()
}
self.errors.append(error_entry)
return error_entry
# Execute với progress bar
with tqdm(total=len(tasks), desc="Processing") as pbar:
async def update_pbar(coro):
result = await coro
pbar.update(1)
return result
results = await asyncio.gather(*[
update_pbar(process_single(task, i))
for i, task in enumerate(tasks)
])
# Thống kê
success_count = sum(1 for r in results if r["status"] == "success")
failed_count = len(results) - success_count
summary = {
"total": len(tasks),
"success": success_count,
"failed": failed_count,
"success_rate": f"{success_count/len(tasks)*100:.1f}%",
"total_time": time.time() - self.start_time if hasattr(self, 'start_time') else 0,
"results": results
}
print(f"\n✅ Hoàn thành: {success_count}/{len(tasks)} thành công ({summary['success_rate']})")
return summary
async def main():
"""Ví dụ sử dụng"""
# Khởi tạo processor
processor = HolySheepBatchProcessor(API_KEY)
processor.start_time = time.time()
# Tạo sample tasks
sample_tasks = [
{
"id": f"doc_{i}",
"messages": [
{"role": "system", "content": "Bạn là trợ lý phân tích văn bản tiếng Việt."},
{"role": "user", "content": f"Tóm tắt nội dung sau: Tài liệu số {i} - Nội dung mẫu về công nghệ AI và ứng dụng trong doanh nghiệp."}
]
}
for i in range(10)
]
# Xử lý batch
results = await processor.process_batch(
tasks=sample_tasks,
model="gpt-4.1",
max_concurrent=3
)
# Lưu kết quả
output_file = f"results_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
with open(output_file, "w", encoding="utf-8") as f:
json.dump(results, f, ensure_ascii=False, indent=2)
print(f"💾 Kết quả lưu tại: {output_file}")
if __name__ == "__main__":
asyncio.run(main())Script Quản Lý Chi Phí
#!/usr/bin/env python3
"""
Cost Tracker cho HolySheep AI
Theo dõi chi phí theo thời gian thực
"""
import json
from datetime import datetime
from dataclasses import dataclass, asdict
from typing import List, Dict, Optional
=== BẢNG GIÁ HOLYSHEEP AI 2026 (USD per 1M tokens) ===
HOLYSHEEP_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},
}
=== SO SÁNH VỚI OPENAI ===
OPENAI_PRICING = {
"gpt-4.1": {"input": 15.00, "output": 60.00}, # Output gấp 7.5x!
}
@dataclass
class TokenUsage:
"""Theo dõi usage của một request"""
timestamp: str
model: str
input_tokens: int
output_tokens: int
cost: float
request_id: str
class CostTracker:
"""Tracker chi phí với báo cáo chi tiết"""
def __init__(self):
self.usages: List[TokenUsage] = []
self.daily_budget = 100.0 # Ngân sách hàng ngày
self.monthly_budget = 2000.0
def add_usage(
self,
model: str,
input_tokens: int,
output_tokens: int,
request_id: str = ""
) -> TokenUsage:
"""Tính và ghi nhận chi phí"""
pricing = HOLYSHEEP_PRICING.get(model, {"input": 0, "output": 0})
input_cost = (input_tokens / 1_000_000) * pricing["input"]
output_cost = (output_tokens / 1_000_000) * pricing["output"]
total_cost = input_cost + output_cost
usage = TokenUsage(
timestamp=datetime.now().isoformat(),
model=model,
input_tokens=input_tokens,
output_tokens=output_tokens,
cost=total_cost,
request_id=request_id
)
self.usages.append(usage)
return usage
def get_total_cost(self, days: Optional[int] = None) -> float:
"""Tính tổng chi phí"""
if days is None:
return sum(u.cost for u in self.usages)
cutoff = datetime.now().timestamp() - (days * 86400)
return sum(
u.cost for u in self.usages
if datetime.fromisoformat(u.timestamp).timestamp() > cutoff
)
def generate_report(self) -> Dict:
"""Tạo báo cáo chi phí chi tiết"""
total_cost = self.get_total_cost()
total_input = sum(u.input_tokens for u in self.usages)
total_output = sum(u.output_tokens for u in self.usages)
# Chi phí nếu dùng OpenAI
openai_cost = sum(
(u.input_tokens / 1_000_000) * OPENAI_PRICING.get(u.model, {}).get("input", 0) +
(u.output_tokens / 1_000_000) * OPENAI_PRICING.get(u.model, {}).get("output", 0)
for u in self.usages
)
savings = openai_cost - total_cost
savings_percent = (savings / openai_cost * 100) if openai_cost > 0 else 0
# Theo model
by_model = {}
for usage in self.usages:
model = usage.model
if model not in by_model:
by_model[model] = {"count": 0, "cost": 0, "tokens": 0}
by_model[model]["count"] += 1
by_model[model]["cost"] += usage.cost
by_model[model]["tokens"] += usage.input_tokens + usage.output_tokens
return {
"report_time": datetime.now().isoformat(),
"total_requests": len(self.usages),
"total_input_tokens": total_input,
"total_output_tokens": total_output,
"total_cost_usd": round(total_cost, 4),
"openai_equivalent_cost": round(openai_cost, 4),
"savings_usd": round(savings, 4),
"savings_percent": round(savings_percent, 1),
"budget_usage": {
"daily": f"{self.get_total_cost(1)/self.daily_budget*100:.1f}%",
"monthly": f"{total_cost/self.monthly_budget*100:.1f}%"
},
"by_model": by_model
}
def print_report(self):
"""In báo cáo đẹp mắt"""
report = self.generate_report()
print("=" * 60)
print("💰 BÁO CÁO CHI PHÍ HOLYSHEEP AI")
print("=" * 60)
print(f"📊 Tổng requests: {report['total_requests']:,}")
print(f"📥 Tổng input tokens: {report['total_input_tokens']:,}")
print(f"📤 Tổng output tokens: {report['total_output_tokens']:,}")
print("-" * 60)
print(f"💵 Chi phí HolySheep: ${report['total_cost_usd']:.4f}")
print(f"💵 Chi phí OpenAI: ${report['openai_equivalent_cost']:.4f}")
print(f"✨ TIẾT KIỆM: ${report['savings_usd']:.4f} ({report['savings_percent']}%)")
print("-" * 60)
print("📈 Theo Model:")
for model, data in report['by_model'].items():
print(f" • {model}: {data['count']} requests, ${data['cost']:.4f}")
print("=" * 60)
=== DEMO ===
if __name__ == "__main__":
tracker = CostTracker()
# Giả lập usage
for i in range(100):
tracker.add_usage(
model="gpt-4.1",
input_tokens=1500,
output_tokens=500,
request_id=f"req_{i}"
)
for i in range(50):
tracker.add_usage(
model="deepseek-v3.2",
input_tokens=2000,
output_tokens=800,
request_id=f"req_ds_{i}"
)
tracker.print_report()So Sánh Chi Phí Thực Tế
| Model | HolySheep Input | OpenAI Input | Tiết kiệm |
|---|---|---|---|
| GPT-4.1 | $8.00 | $15.00 | 46.7% |
| Claude Sonnet 4.5 | $15.00 | $15.00 | Ngang nhau |
| Gemini 2.5 Flash | $2.50 | $0.30 | Model khác |
| DeepSeek V3.2 | $0.42 | N/A | Model siêu rẻ |
Lưu ý quan trọng: Với GPT-4.1, output token của OpenAI là $60/M (gấp 7.5x input), trong khi HolySheep chỉ $8/M cho cả input lẫn output. Điều này tạo ra sự khác biệt chi phí khổng lồ cho các tác vụ sinh text dài.
Xử Lý 1000 Tasks Với Progress Tracking
#!/usr/bin/env python3
"""
Production Batch Processor - Xử lý 1000+ tasks
Features: Checkpoint, Progress persistence, Graceful shutdown
"""
import asyncio
import aiofiles
import json
import signal
import sys
from pathlib import Path
from datetime import datetime
from collections import deque
CHECKPOINT_INTERVAL = 50 # Lưu checkpoint mỗi 50 tasks
BATCH_SIZE = 100
class ProductionBatchProcessor:
"""Processor cho production với fault tolerance cao"""
def __init__(self, api_key: str, checkpoint_file: str = "checkpoint.json"):
self.api_key = api_key
self.checkpoint_file = Path(checkpoint_file)
self.completed_ids = set()
self.failed_tasks = deque(maxlen=100) # Giữ 100 lỗi gần nhất
self.stats = {
"total": 0,
"completed": 0,
"failed": 0,
"start_time": None,
"last_update": None
}
# Graceful shutdown handler
signal.signal(signal.SIGINT, self._handle_shutdown)
signal.signal(signal.SIGTERM, self._handle_shutdown)
self._shutdown_requested = False
def _handle_shutdown(self, signum, frame):
"""Xử lý shutdown graceful"""
print("\n⚠️ Nhận signal shutdown. Đang lưu checkpoint...")
self._shutdown_requested = True
self._save_checkpoint()
print("💾 Checkpoint đã lưu. Thoát an toàn.")
sys.exit(0)
async def _load_checkpoint(self):
"""Load checkpoint nếu có"""
if self.checkpoint_file.exists():
async with aiofiles.open(self.checkpoint_file, 'r') as f:
data = json.loads(await f.read())
self.completed_ids = set(data.get("completed_ids", []))
self.stats.update(data.get("stats", {}))
print(f"📂 Đã load checkpoint: {len(self.completed_ids)} tasks đã hoàn thành")
def _save_checkpoint(self):
"""Lưu checkpoint"""
data = {
"completed_ids": list(self.completed_ids),
"stats": self.stats,
"timestamp": datetime.now().isoformat()
}
with open(self.checkpoint_file, 'w') as f:
json.dump(data, f, indent=2)
async def process_large_batch(
self,
all_tasks: list,
model: str = "gpt-4.1"
) -> dict:
"""Xử lý batch lớn với checkpointing"""
await self._load_checkpoint()
self.stats["start_time"] = datetime.now().isoformat()
# Lọc bỏ tasks đã hoàn thành
remaining_tasks = [
(i, task) for i, task in enumerate(all_tasks)
if str(task.get("id", i)) not in self.completed_ids
]
self.stats["total"] = len(all_tasks)
print(f"🎯 Cần xử lý: {len(remaining_tasks)}/{len(all_tasks)} tasks")
batch_start = 0
while batch_start < len(remaining_tasks) and not self._shutdown_requested:
batch_end = min(batch_start + BATCH_SIZE, len(remaining_tasks))
batch = remaining_tasks[batch_start:batch_end]
print(f"\n📦 Batch {batch_start//BATCH_SIZE + 1}: {len(batch)} tasks")
results = await self._process_batch(batch, model)
# Cập nhật stats
for task_id, status, result in results:
if status == "success":
self.completed_ids.add(str(task_id))
self.stats["completed"] += 1
else:
self.failed_tasks.append({"id": task_id, "error": result})
self.stats["failed"] += 1
# Save checkpoint
self._save_checkpoint()
self.stats["last_update"] = datetime.now().isoformat()
progress = self.stats["completed"] / self.stats["total"] * 100
print(f"📊 Progress: {self.stats['completed']}/{self.stats['total']} ({progress:.1f}%)")
batch_start = batch_end
return self._generate_final_report()
async def _process_batch(self, tasks: list, model: str) -> list:
"""Xử lý một batch tasks"""
results = []
for idx, task in tasks:
if self._shutdown_requested:
break
try:
# Gọi API (implement tương tự phần trên)
result = await self._call_api(task, model)
results.append((task.get("id", idx), "success", result))
except Exception as e:
results.append((task.get("id", idx), "failed", str(e)))
# Rate limiting nhẹ giữa các request
await asyncio.sleep(0.1)
return results
async def _call_api(self, task: dict, model: str) -> dict:
"""Gọi HolySheep API"""
import httpx
async with httpx.AsyncClient() as client:
response = await client.post(
f"https://api.holysheep.ai/v1/chat/completions",
headers={"Authorization": f"Bearer {self.api_key}"},
json={
"model": model,
"messages": task.get("messages", []),
"max_tokens": task.get("max_tokens", 2048)
},
timeout=60.0
)
if response.status_code == 200:
return response.json()
else:
raise Exception(f"API error: {response.status_code}")
def _generate_final_report(self) -> dict:
"""Tạo báo cáo cuối cùng"""
return {
"total_tasks": self.stats["total"],
"completed": self.stats["completed"],
"failed": self.stats["failed"],
"success_rate": f"{self.stats['completed']/self.stats['total']*100:.1f}%",
"failed_tasks_sample": list(self.failed_tasks)[:10] # Top 10 lỗi
}
=== SỬ DỤNG ===
async def main():
processor = ProductionBatchProcessor(
api_key="YOUR_HOLYSHEEP_API_KEY",
checkpoint_file="batch_checkpoint.json"
)
# Load tasks từ file
with open("input_tasks.json", "r") as f:
tasks = json.load(f)
report = await processor.process_large_batch(tasks, model="gpt-4.1")
print("\n" + "=" * 50)
print("📋 BÁO CÁO CUỐI CÙNG")
print(json.dumps(report, indent=2, ensure_ascii=False))
if __name__ == "__main__":
asyncio.run(main())Lỗi Thường Gặp và Cách Khắc Phục
1. Lỗi "ConnectionError: timeout after 30s"
Nguyên nhân: Network instability hoặc server HolySheep AI đang bảo trì.
# KHẮC PHỤC: Implement retry với timeout tăng dần
import httpx
import asyncio
async def robust_request(url: str, payload: dict, headers: dict):
"""Request với timeout tăng dần và retry thông minh"""
timeouts = [30, 60, 120] # Tăng dần theo attempt
for attempt, timeout in enumerate(timeouts):
try:
async with httpx.AsyncClient(timeout=timeout) as client:
response = await client.post(url, json=payload, headers=headers)
return response.json()
except httpx.TimeoutException:
print(f"⚠️ Attempt {attempt + 1} timeout sau {timeout}s")
if attempt < len(timeouts) - 1:
await asyncio.sleep(2 ** attempt) # Backoff
except httpx.ConnectError as e:
print(f"🔌 Connection failed: {e}")
await asyncio.sleep(5) # Chờ network ổn định
# Fallback: Trả về mock data để không block pipeline
return {"error": "max_retries_exceeded", "fallback": True}2. Lỗi "401 Unauthorized"
Nguyên nhân: API key không hợp lệ, hết hạn, hoặc chưa kích hoạt.
# KHẮC PHỤC: Validate API key trước khi bắt đầu batch
import httpx
async def validate_api_key(api_key: str) -> dict:
"""Validate API key và lấy thông tin quota"""
try:
async with httpx.AsyncClient(timeout=10.0) as client:
response = await client.get(
"https://api.holysheep.ai/v1/models",
headers={"Authorization": f"Bearer {api_key}"}
)
if response.status_code == 200:
return {"valid": True, "models": response.json()}
elif response.status_code == 401:
return {
"valid": False,
"error": "API key không hợp lệ. Kiểm tra lại tại https://www.holysheep.ai/register"
}
elif response.status_code == 403:
return {
"valid": False,
"error": "API key chưa được kích hoạt. Vui lòng xác thực email."
}
else:
return {"valid": False, "error": f"Lỗi {response.status_code}"}
except Exception as e:
return {"valid": False, "error": f"Không thể kết nối: {e}"}
Sử dụng
api_key = "YOUR_HOLYSHEEP_API_KEY"
validation = await validate_api_key(api_key)
if not validation["valid"]:
print(f"❌ {validation['error']}")
exit(1)
else:
print("✅ API key hợp lệ! Bắt đầu batch...")3. Lỗi "429 Rate Limit Exceeded"
Nguyên nhân: Vượt quá RPM hoặc TPM limit của tài khoản.
# KHẮC PHỤC: Adaptive rate limiting với token bucket
import time
import asyncio
from threading import Lock
class AdaptiveRateLimiter:
"""Rate limiter tự động điều chỉnh theo response headers"""
def __init__(self):
self.rpm_limit = 60
self.tpm_limit = 150000
self.current_rpm = 0
self.current_tpm = 0
self.window_start = time.time()
self.lock = Lock()
def update_from_response(self, headers: dict):
"""Cập nhật limits từ response headers nếu có"""
if "x-ratelimit-limit-requests" in headers:
self.rpm_limit = int(headers["x-ratelimit-limit-requests"])
if "x-ratelimit-limit-tokens" in headers:
self.tpm_limit = int(headers["x-ratelimit-limit-tokens"])
async def acquire(self, tokens_needed: int):
"""Acquire permission với automatic backoff"""
while True:
with self.lock:
current_time = time.time()
# Reset window nếu đã qua 1 phút
if current_time - self.window_start >= 60:
self.current_rpm = 0
self.current_tpm = 0
self.window_start = current_time
# Check limits
can_proceed = (
self.current_rpm < self.rpm_limit and
self.current_tpm + tokens_needed <= self.tpm_limit
)
if can_proceed:
self.current_rpm += 1
self.current_tpm += tokens_needed
return True
# Tính wait time
time_in_window = current_time - self.window_start
wait_time = 60 - time_in_window + 1 # Buffer 1s
print(f"⏳ Rate limit. Chờ {wait_time:.1f}s...")
await asyncio.sleep(min(wait_time, 5)) # Max wait 5s mỗi lần
def get_stats(self) -> dict:
"""Lấy thống kê hiện tại"""
return {
"rpm_usage": f"{self.current_rpm}/{self.rpm_limit}",
"tpm_usage": f"{self.current_tpm}/{self.tpm_limit}"
}4. Lỗi "Quota Exceeded" Hoặc Hết Credit
Nguyên nhân: Tài khoản hết credit hoặc vượt monthly limit.
# KHẮC PHỤC: Kiểm tra quota trước batch và budget alert
import httpx
from datetime import datetime
class BudgetController:
"""Kiểm soát ngân sách với alerts"""
def __init__(self, api_key: str, max_daily_spend