Tôi đã từng mất 3 ngày debug một lỗi không ngờ tới. Hệ thống chatbot doanh nghiệp của một khách hàng bỗng dưng trả về toàn "Server Error" vào giờ cao điểm. Sau khi đào sâu vào logs, tôi thấy điều này xuất hiện lặp đi lặp lại:
Exception in thread Thread-42:
httpx.HTTPStatusError: Client response '429 Too Many Requests'
{"error":{"code":"rate_limit_exceeded","message":"Request rate limit exceeded.
Please retry after 30 seconds.","retry_after":30}}Đó là lúc tôi nhận ra: 429 không chỉ là lỗi — nó là tín hiệu cho thấy kiến trúc xử lý request của bạn đang thất bại. Trong bài viết này, tôi sẽ chia sẻ chiến lược xử lý rate limit hiệu quả, áp dụng được cho cả OpenAI lẫn HolySheep AI với chi phí tiết kiệm đến 85%.
Mục Lục
- Tại Sao 429 Error Xảy Ra?
- Kiến Trúc Xử Lý Rate Limit Tổng Thể
- HolySheep Queue — Hàng Đợi Thông Minh
- Retry Logic Với Exponential Backoff
- Circuit Breaker Pattern
- Bảo Vệ Ngân Sách Doanh Nghiệp
- Bảng Giá và So Sánh
- Lỗi Thường Gặp và Cách Khắc Phục
- Phù Hợp / Không Phù Hợp Với Ai
- Đăng Ký HolySheep AI
Tại Sao 429 Error Xảy Ra Trong AI API?
HTTP 429 (Too Many Requests) là response từ server khi client gửi quá nhiều request trong một khoảng thời gian nhất định. Với các AI API như GPT-4.1 hay Claude Sonnet 4.5, có 3 loại giới hạn chính:
- RPM (Requests Per Minute): Số request mỗi phút, thường từ 60-500 tùy gói subscription
- TPM (Tokens Per Minute): Giới hạn token đầu vào + đầu ra mỗi phút
- RPD (Requests Per Day): Giới hạn tổng request hàng ngày
Thực tế tôi đã gặp trường hợp: một ứng dụng RAG (Retrieval-Augmented Generation) gửi 200 requests/giây vào 9:00 sáng, ngay lập tức nhận 50 lỗi 429 liên tiếp, khiến batch processing 10,000 tài liệu trì hoãn 6 tiếng đồng hồ.
Kiến Trúc Xử Lý Rate Limit Tổng Thể
Trước khi code, hãy hiểu kiến trúc tổng thể mà tôi áp dụng cho các dự án enterprise:
+------------------------+
| Client Application |
+-----------+------------+
|
v
+-----------+------------+
| Load Balancer |
+-----------+------------+
|
+-------+-------+
| |
v v
+-------+-----+ +---+--------+
| Rate Limiter | | Queue |
| (Token | | Manager |
| Bucket) | | |
+-------+-----+ +---+--------+
| |
v v
+---+--------+ +----+-----+
| Retry | | Circuit |
| Engine | | Breaker |
+---+--------+ +----+-----+
| |
v v
+-----------------------+
| AI API |
| HolySheep / OpenAI |
+-----------------------+Điểm mấu chốt: không bao giờ gửi request trực tiếp đến API. Luôn đặt một lớp xử lý trung gian để quản lý luồng request.
HolySheep Queue — Triển Khai Hàng Đợi Thông Minh
Tôi khuyên dùng HolySheep AI vì hỗ trợ concurrency cao và latency chỉ dưới 50ms. Dưới đây là implementation queue manager hoàn chỉnh:
import asyncio
import time
from collections import deque
from dataclasses import dataclass, field
from typing import Optional, Callable, Any
from threading import Lock
import httpx
@dataclass
class QueueConfig:
"""Cấu hình cho HolySheep AI Queue Manager"""
base_url: str = "https://api.holysheep.ai/v1"
api_key: str = "YOUR_HOLYSHEEP_API_KEY"
max_concurrent: int = 10
max_queue_size: int = 1000
default_timeout: float = 30.0
retry_attempts: int = 3
retry_base_delay: float = 1.0
class HolySheepQueueManager:
"""
Queue Manager cho HolySheep AI API
- Tự động quản lý concurrency
- Retry với exponential backoff
- Theo dõi budget theo thời gian thực
"""
def __init__(self, config: QueueConfig):
self.config = config
self.queue = deque()
self.active_requests = 0
self.lock = Lock()
self.budget_spent = 0.0
self.budget_limit = float('inf')
self.request_stats = {
'total': 0,
'success': 0,
'rate_limited': 0,
'failed': 0
}
self.client = httpx.AsyncClient(
base_url=config.base_url,
headers={
"Authorization": f"Bearer {config.api_key}",
"Content-Type": "application/json"
},
timeout=config.default_timeout
)
async def enqueue(self, task: Callable, *args, **kwargs) -> Any:
"""Thêm task vào queue với kiểm tra rate limit"""
with self.lock:
if len(self.queue) >= self.config.max_queue_size:
raise QueueFullError(f"Queue đã đầy: {self.config.max_queue_size}")
self.queue.append((task, args, kwargs))
self.request_stats['total'] += 1
return await self._process_queue()
async def _process_queue(self) -> Any:
"""Xử lý queue với concurrency control"""
while True:
with self.lock:
if not self.queue:
return None
if self.active_requests >= self.config.max_concurrent:
await asyncio.sleep(0.1)
continue
task, args, kwargs = self.queue.popleft()
self.active_requests += 1
try:
result = await self._execute_with_retry(task, *args, **kwargs)
with self.lock:
self.request_stats['success'] += 1
return result
except Exception as e:
with self.lock:
self.request_stats['failed'] += 1
raise
finally:
with self.lock:
self.active_requests -= 1
async def _execute_with_retry(self, task: Callable, *args, **kwargs) -> Any:
"""Thực thi task với retry logic"""
last_exception = None
for attempt in range(self.config.retry_attempts):
try:
result = await task(*args, **kwargs)
return result
except httpx.HTTPStatusError as e:
last_exception = e
if e.response.status_code == 429:
with self.lock:
self.request_stats['rate_limited'] += 1
retry_after = int(e.response.headers.get('retry-after', 30))
wait_time = retry_after * (2 ** attempt)
print(f"[HolySheep] Rate limited! Chờ {wait_time}s (attempt {attempt + 1})")
await asyncio.sleep(wait_time)
else:
raise
except Exception as e:
last_exception = e
await asyncio.sleep(self.config.retry_base_delay * (2 ** attempt))
raise last_exception
Sử dụng
async def call_holysheep_chat(manager: HolySheepQueueManager, prompt: str):
"""Gọi HolySheep Chat Completion API"""
response = await manager.client.post(
"/chat/completions",
json={
"model": "gpt-4.1",
"messages": [{"role": "user", "content": prompt}],
"max_tokens": 1000
}
)
response.raise_for_status()
data = response.json()
# Theo dõi chi phí (ước tính)
usage = data.get('usage', {})
tokens = usage.get('total_tokens', 0)
estimated_cost = (tokens / 1_000_000) * 8 # $8/MTok cho GPT-4.1
with manager.lock:
manager.budget_spent += estimated_cost
return data['choices'][0]['message']['content']
Khởi tạo và sử dụng
config = QueueConfig(
api_key="YOUR_HOLYSHEEP_API_KEY",
max_concurrent=10,
max_queue_size=500
)
manager = HolySheepQueueManager(config)
async def main():
# Xử lý 100 requests với concurrency limit
tasks = [
manager.enqueue(call_holysheep_chat, manager, f"Tạo nội dung {i}")
for i in range(100)
]
results = await asyncio.gather(*tasks, return_exceptions=True)
print(f"Tổng chi phí ước tính: ${manager.budget_spent:.4f}")
print(f"Stats: {manager.request_stats}")
asyncio.run(main())Retry Logic Với Exponential Backoff Chi Tiết
Exponential backoff là chiến lược chờ lâu hơn sau mỗi lần thất bại. Công thức tôi sử dụng:
wait_time = base_delay * (2 ** attempt) + jitter
Ví dụ:
Attempt 1: 1 * 2^0 + random(0-1) = 1-2s
Attempt 2: 1 * 2^1 + random(0-1) = 2-3s
Attempt 3: 1 * 2^2 + random(0-1) = 4-5s
Attempt 4: 1 * 2^3 + random(0-1) = 8-9s
Đây là implementation đầy đủ cho HolySheep API:
import random
import asyncio
from datetime import datetime, timedelta
from enum import Enum
class RateLimitStrategy(Enum):
"""Chiến lược xử lý rate limit"""
RETRY_IMMEDIATE = "immediate"
RETRY_EXPONENTIAL = "exponential"
QUEUE_AND_RETRY = "queue_and_retry"
FALLBACK_MODEL = "fallback"
class HolySheepRetryHandler:
"""
Retry Handler chuyên biệt cho HolySheep AI
- Exponential backoff với jitter
- Fallback giữa các models
- Budget-aware retry
"""
# Model fallback order: đắt -> rẻ
MODEL_FALLBACK = {
"gpt-4.1": {"fallback": "claude-sonnet-4.5", "cost_ratio": 8/15},
"claude-sonnet-4.5": {"fallback": "gemini-2.5-flash", "cost_ratio": 15/2.5},
"gemini-2.5-flash": {"fallback": "deepseek-v3.2", "cost_ratio": 2.5/0.42},
"deepseek-v3.2": {"fallback": None, "cost_ratio": 1.0}
}
def __init__(self, api_key: str):
self.api_key = api_key
self.rate_limit_until: Optional[datetime] = None
self.request_history: list[datetime] = []
self.circuit_open = False
self.circuit_open_time: Optional[datetime] = None
self.circuit_timeout = timedelta(minutes=5)
def _calculate_backoff(self, attempt: int, retry_after: int = None) -> float:
"""Tính toán thời gian chờ với exponential backoff + jitter"""
if retry_after:
base = retry_after
else:
base = 1.0
# Exponential backoff: base * 2^attempt
exponential = base * (2 ** attempt)
# Jitter ngẫu nhiên: tránh thundering herd
jitter = random.uniform(0, 0.5 * exponential)
return exponential + jitter
def _should_retry(self, attempt: int, max_attempts: int, error: Exception) -> bool:
"""Quyết định có nên retry không"""
if attempt >= max_attempts:
return False
if self.circuit_open:
if datetime.now() - self.circuit_open_time > self.circuit_timeout:
self.circuit_open = False
return True
return False
# Retry cho các lỗi có thể phục hồi
retryable_codes = {429, 500, 502, 503, 504}
if hasattr(error, 'response') and hasattr(error.response, 'status_code'):
return error.response.status_code in retryable_codes
return isinstance(error, (ConnectionError, TimeoutError, httpx.TimeoutException))
def _should_open_circuit(self, error: Exception) -> bool:
"""Quyết định có nên mở circuit breaker không"""
if hasattr(error, 'response'):
status = error.response.status_code
# Mở circuit nếu liên tục rate limited
if status == 429:
return True
# Đếm số lỗi gần đây
recent_errors = [
dt for dt in self.request_history
if dt > datetime.now() - timedelta(minutes=1)
]
if len(recent_errors) >= 10:
return True
return False
async def execute_with_retry(
self,
prompt: str,
model: str = "gpt-4.1",
max_attempts: int = 4,
strategy: RateLimitStrategy = RateLimitStrategy.QUEUE_AND_RETRY
) -> dict:
"""Thực thi request với đầy đủ retry logic"""
if self.circuit_open:
# Circuit breaker đang mở -> thử model rẻ hơn
fallback_config = self.MODEL_FALLBACK.get(model, {})
if fallback_config.get("fallback"):
model = fallback_config["fallback"]
print(f"[Circuit Breaker] Chuyển sang model fallback: {model}")
else:
raise CircuitBreakerOpenError("Tất cả models đều không khả dụng")
attempt = 0
last_error = None
while attempt < max_attempts:
try:
# Kiểm tra rate limit trước khi gửi
if self.rate_limit_until and datetime.now() < self.rate_limit_until:
wait_seconds = (self.rate_limit_until - datetime.now()).total_seconds()
print(f"[Rate Limit] Đợi {wait_seconds:.1f}s trước khi gửi...")
await asyncio.sleep(wait_seconds)
# Gửi request
response = await self._call_holysheep(prompt, model)
self.request_history.append(datetime.now())
return response
except httpx.HTTPStatusError as e:
last_error = e
if e.response.status_code == 429:
# Parse retry-after header
retry_after = int(e.response.headers.get('retry-after', 30))
self.rate_limit_until = datetime.now() + timedelta(seconds=retry_after)
if self._should_open_circuit(e):
self.circuit_open = True
self.circuit_open_time = datetime.now()
raise CircuitBreakerOpenError("Circuit breaker đã mở do rate limit")
wait_time = self._calculate_backoff(attempt, retry_after)
print(f"[429] Attempt {attempt + 1}/{max_attempts}, đợi {wait_time:.1f}s")
await asyncio.sleep(wait_time)
attempt += 1
elif e.response.status_code >= 500:
wait_time = self._calculate_backoff(attempt)
print(f"[5xx] Attempt {attempt + 1}/{max_attempts}, đợi {wait_time:.1f}s")
await asyncio.sleep(wait_time)
attempt += 1
else:
raise
except (ConnectionError, httpx.TimeoutException) as e:
last_error = e
wait_time = self._calculate_backoff(attempt)
print(f"[Network] Attempt {attempt + 1}/{max_attempts}, đợi {wait_time:.1f}s")
await asyncio.sleep(wait_time)
attempt += 1
# Thử fallback model nếu retry hết
fallback_config = self.MODEL_FALLBACK.get(model, {})
if fallback_config.get("fallback"):
fallback_model = fallback_config["fallback"]
print(f"[Fallback] Chuyển sang {fallback_model}")
return await self.execute_with_retry(
prompt,
model=fallback_model,
max_attempts=max_attempts
)
raise RetryExhaustedError(f"Hết số lần retry: {last_error}")
async def _call_holysheep(self, prompt: str, model: str) -> dict:
"""Gọi HolySheep API thực tế"""
async with httpx.AsyncClient(
base_url="https://api.holysheep.ai/v1",
headers={"Authorization": f"Bearer {self.api_key}"},
timeout=60.0
) as client:
response = await client.post(
"/chat/completions",
json={
"model": model,
"messages": [{"role": "user", "content": prompt}]
}
)
response.raise_for_status()
return response.json()
Custom exceptions
class RateLimitError(Exception): pass
class CircuitBreakerOpenError(Exception): pass
class RetryExhaustedError(Exception): pass
class QueueFullError(Exception): pass
Sử dụng
async def main():
handler = HolySheepRetryHandler(api_key="YOUR_HOLYSHEEP_API_KEY")
try:
result = await handler.execute_with_retry(
prompt="Phân tích dữ liệu bán hàng Q1 2026",
model="gpt-4.1",
max_attempts=3
)
print(f"Kết quả: {result}")
except CircuitBreakerOpenError as e:
print(f"Circuit breaker mở: {e}")
except RetryExhaustedError as e:
print(f"Hết retry: {e}")Circuit Breaker Pattern Cho AI API
Circuit breaker ngăn hệ thống gọi liên tục vào API đang có vấn đề. Tôi triển khai 3 trạng thái:
CLOSED ──(failures > threshold)──> OPEN
^ │
│ │ (timeout)
│ v
└──────(success)──────────── HALF-OPENTriển khai đầy đủ:
from enum import Enum
from dataclasses import dataclass
import time
class CircuitState(Enum):
CLOSED = "closed"
OPEN = "open"
HALF_OPEN = "half_open"
@dataclass
class CircuitBreakerConfig:
failure_threshold: int = 5 # Số lỗi để mở circuit
success_threshold: int = 3 # Số thành công để đóng circuit
timeout: float = 60.0 # Thời gian chờ (giây)
half_open_max_calls: int = 3 # Số calls trong half-open
class AICircuitBreaker:
"""
Circuit Breaker cho AI API
Bảo vệ hệ thống khỏi cascading failures
"""
def __init__(self, name: str, config: CircuitBreakerConfig = None):
self.name = name
self.config = config or CircuitBreakerConfig()
self.state = CircuitState.CLOSED
self.failure_count = 0
self.success_count = 0
self.last_failure_time: float = 0
self.half_open_calls = 0
def record_success(self):
"""Ghi nhận thành công"""
if self.state == CircuitState.HALF_OPEN:
self.success_count += 1
if self.success_count >= self.config.success_threshold:
self._transition_to(CircuitState.CLOSED)
print(f"[CircuitBreaker:{self.name}] Đã đóng circuit!")
else:
self.failure_count = 0
def record_failure(self):
"""Ghi nhận thất bại"""
self.failure_count += 1
self.last_failure_time = time.time()
if self.state == CircuitState.HALF_OPEN:
self._transition_to(CircuitState.OPEN)
elif (self.failure_count >= self.config.failure_threshold and
self.state == CircuitState.CLOSED):
self._transition_to(CircuitState.OPEN)
def can_execute(self) -> bool:
"""Kiểm tra có được phép thực thi không"""
if self.state == CircuitState.CLOSED:
return True
if self.state == CircuitState.OPEN:
elapsed = time.time() - self.last_failure_time
if elapsed >= self.config.timeout:
self._transition_to(CircuitState.HALF_OPEN)
return True
return False
if self.state == CircuitState.HALF_OPEN:
return self.half_open_calls < self.config.half_open_max_calls
return False
def _transition_to(self, new_state: CircuitState):
"""Chuyển trạng thái circuit"""
print(f"[CircuitBreaker:{self.name}] {self.state.value} -> {new_state.value}")
self.state = new_state
if new_state == CircuitState.CLOSED:
self.failure_count = 0
self.success_count = 0
elif new_state == CircuitState.HALF_OPEN:
self.half_open_calls = 0
self.success_count = 0
elif new_state == CircuitState.OPEN:
self.last_failure_time = time.time()
def get_status(self) -> dict:
"""Lấy trạng thái hiện tại"""
return {
"name": self.name,
"state": self.state.value,
"failure_count": self.failure_count,
"last_failure": self.last_failure_time,
"can_execute": self.can_execute()
}
Triển khai với HolySheep
class HolySheepWithCircuitBreaker:
"""HolySheep API client với Circuit Breaker tích hợp"""
def __init__(self, api_key: str):
self.api_key = api_key
self.circuit_breaker = AICircuitBreaker("holysheep-gpt4")
self.fallback_circuit = AICircuitBreaker("holysheep-deepseek")
self.client = httpx.AsyncClient(
base_url="https://api.holysheep.ai/v1",
headers={"Authorization": f"Bearer {api_key}"},
timeout=60.0
)
async def chat(self, prompt: str, model: str = "gpt-4.1") -> dict:
"""Gọi API với circuit breaker protection"""
# Kiểm tra circuit breaker
circuit = self.circuit_breaker if "gpt-4.1" in model else self.fallback_circuit
if not circuit.can_execute():
# Thử model fallback
if "gpt-4.1" in model:
print(f"[CircuitBreaker] GPT-4.1 unavailable, thử DeepSeek V3.2")
return await self.chat(prompt, "deepseek-v3.2")
raise CircuitBreakerOpenError(f"Circuit breaker OPEN cho {model}")
try:
circuit.half_open_calls += 1
response = await self.client.post(
"/chat/completions",
json={
"model": model,
"messages": [{"role": "user", "content": prompt}]
}
)
if response.status_code == 200:
circuit.record_success()
return response.json()
else:
circuit.record_failure()
response.raise_for_status()
except httpx.HTTPStatusError as e:
circuit.record_failure()
raise
except Exception as e:
circuit.record_failure()
raise
async def close(self):
await self.client.aclose()
Sử dụng
async def main():
client = HolySheepWithCircuitBreaker("YOUR_HOLYSHEEP_API_KEY")
# Test circuit breaker
for i in range(20):
try:
result = await client.chat(f"Test request {i}")
print(f"Request {i}: SUCCESS")
except CircuitBreakerOpenError as e:
print(f"Request {i}: BLOCKED - {e}")
await asyncio.sleep(5) # Đợi một chút
except Exception as e:
print(f"Request {i}: ERROR - {e}")
# In trạng thái circuit breaker
print(f"Circuit status: {client.circuit_breaker.get_status()}")
asyncio.run(main())Bảo Vệ Ngân Sách Doanh Nghiệp
Đây là phần quan trọng nhất mà nhiều developer bỏ qua. Tôi đã chứng kiến hóa đơn $10,000/tháng chỉ vì không giới hạn budget. Giải pháp:
from dataclasses import dataclass, field
from typing import Optional
from datetime import datetime, timedelta
import threading
@dataclass
class BudgetConfig:
daily_limit: float = 100.0 # Giới hạn ngân sách hàng ngày (USD)
monthly_limit: float = 2000.0 # Giới hạn hàng tháng
alert_threshold: float = 0.8 # Cảnh báo khi đạt 80%
per_request_max: float = 0.50 # Giới hạn cho 1 request
@dataclass
class BudgetTracker:
"""Theo dõi và giới hạn chi phí API"""
config: BudgetConfig
daily_spent: float = 0.0
monthly_spent: float = 0.0
daily_reset: datetime = field(default_factory=datetime.now)
monthly_reset: datetime = field(default_factory=datetime.now)
_lock: threading.Lock = field(default_factory=threading.Lock)
# Callbacks cho alerts
on_budget_warning: Optional[callable] = None
on_budget_exceeded: Optional[callable] = None
def __post_init__(self):
self._reset_if_needed()
def _reset_if_needed(self):
"""Reset counters nếu đến ngày mới/tháng mới"""
now = datetime.now()
if now.date() > self.daily_reset.date():
self.daily_spent = 0
self.daily_reset = now
print(f"[Budget] Đã reset chi phí hàng ngày")
if now.month != self.monthly_reset.month:
self.monthly_spent = 0
self.monthly_reset = now
print(f"[Budget] Đã reset chi phí hàng tháng")
def check_and_update(self, cost: float, request_id: str = "") -> bool:
"""
Kiểm tra và cập nhật chi phí
Returns: True nếu request được phép, False nếu vượt budget
"""
with self._lock:
self._reset_if_needed()
# Kiểm tra các giới hạn
checks = [
("Per-request", cost <= self.config.per_request_max),
("Daily", self.daily_spent + cost <= self.config.daily_limit),
("Monthly", self.monthly_spent + cost <= self.config.monthly_limit)
]
failed_checks = [name for name, passed in checks if not passed]
if failed_checks:
print(f"[Budget] BLOCKED - Vượt giới hạn: {failed_checks}")
if self.on_budget_exceeded:
self.on_budget_exceeded(failed_checks, cost)
return False
# Cập nhật chi phí
self.daily_spent += cost
self.monthly_spent += cost
# Kiểm tra ngưỡng cảnh báo
daily_pct = self.daily_spent / self.config.daily_limit
monthly_pct = self.monthly_spent / self.config.monthly_limit
if daily_pct >= self.config.alert_threshold:
print(f"[Budget] WARNING: Đã dùng {daily_pct*100:.0f}% budget hàng ngày")
if self.on_budget_warning:
self.on_budget_warning("daily", daily_pct)
if monthly_pct >= self.config.alert_threshold:
print(f"[Budget] WARNING: Đã dùng {monthly_pct*100:.0f}% budget hàng tháng")
if self.on_budget_warning:
self.on_budget_warning("monthly", monthly_pct)
return True
def get_status(self) -> dict:
"""Lấy trạng thái budget hiện tại"""
return {
"daily_spent": self.daily_spent,
"daily_limit": self.config.daily_limit,
"daily_remaining": self.config.daily_limit - self.daily_spent,
"monthly_spent": self.monthly_spent,
"monthly_limit": self.config.monthly_limit,
"monthly_remaining": self.config.monthly_limit - self.monthly_spent,
"next_daily_reset": self.daily_reset + timedelta(days=1),
"next_monthly_reset": self.monthly_reset + timedelta(days=30)
}
Triển khai với HolySheep
class HolySheepBudgetManager:
"""Budget-aware HolySheep API Client"""
# Bảng giá HolySheep 2026 (USD/MTok)
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.5, "output": 2.5},
"deepseek-v3.2": {"input": 0.42, "output": 0.42}
}
def __init__(self, api_key: str, budget_config: BudgetConfig = None):
self.api_key = api_key
self.budget = BudgetTracker(budget_config or BudgetConfig())
self.client = httpx.AsyncClient(
base_url="https://api.holysheep.ai/v1",
headers={"Authorization": f"Bearer {api_key}"}
)
# Setup alerts
self.budget.on_budget_warning = self._handle_warning
self.budget.on_budget_exceeded = self._handle_exceeded
def _estimate_cost(self, model: str, tokens: int) -> float:
"""Ước tính chi phí dựa trên model và số tokens"""
pricing = self.HOLYSHEEP_PRICING.get(model, {"input": 8.0, "output": 8.0})
return (tokens / 1_000_000) * ((pricing["input"] + pricing["output"]) / 2)
def _handle_warning(self, period: str, percentage: float):
"""Xử lý cảnh báo budget"""
print(f"🚨 ALERT: Budget {period} đã đạt {percentage*100:.0f}%!")
# Gử