Lần đầu tiên triển khai hệ thống chatbot AI cho một nền tảng thương mại điện tử quy mô 50,000 người dùng đồng thời, tôi đã gặp phải một cơn ác mộng thực sự: 429 Too Many Requests. Hệ thống hoạt động hoàn hảo trong môi trường staging, nhưng ngay khi đợt sale flash bắt đầu, API trả về lỗi hàng loạt. Đó là bài học đắt giá về việc quản lý giới hạn tỷ lệ (rate limiting) và hạn mức (quota) của Gemini 2.5 Pro API.
Tại Sao Giới Hạn API Lại Quan Trọng?
Khi xây dựng các ứng dụng AI production, giới hạn tỷ lệ không chỉ là rào cản kỹ thuật mà còn là yếu tố quyết định chi phí và trải nghiệm người dùng. Với HolySheep AI, chi phí chỉ từ $2.50/MTok cho Gemini 2.5 Flash (theo tỷ giá ¥1=$1), giúp tiết kiệm đến 85% so với các nhà cung cấp khác.
Giới Hạn Tỷ Lệ Của Gemini 2.5 Pro
Gemini 2.5 Pro có các giới hạn theo cấp độ người dùng:
- Free Tier: 15 requests/phút, 1M tokens/phút
- Standard: 60 requests/phút, 4M tokens/phút
- Advanced: 150 requests/phút, 10M tokens/phút
Triển Khai Retry Logic Với Exponential Backoff
Dưới đây là triển khai production-ready với retry thông minh:
import time
import asyncio
from typing import Optional, Dict, Any
from openai import OpenAI, RateLimitError, APITimeoutError
Khởi tạo client HolySheep AI
client = OpenAI(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1",
timeout=60.0,
max_retries=0 # Chúng ta tự xử lý retry
)
class GeminiRateLimitHandler:
"""Xử lý rate limiting với exponential backoff"""
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: Optional[int] = None) -> float:
"""Tính toán độ trễ với exponential backoff"""
if retry_after:
# Ưu tiên thời gian từ header Retry-After
return min(retry_after, self.max_delay)
# Exponential backoff: 1s, 2s, 4s, 8s, 16s...
delay = self.base_delay * (2 ** attempt)
delay = min(delay, self.max_delay)
# Thêm jitter ngẫu nhiên ±25%
if self.jitter:
import random
delay = delay * (0.75 + random.random() * 0.5)
return delay
async def call_with_retry(
self,
prompt: str,
model: str = "gemini-2.5-pro",
temperature: float = 0.7,
**kwargs
) -> Dict[str, Any]:
"""Gọi API với retry logic"""
last_error = None
for attempt in range(self.max_retries + 1):
try:
response = client.chat.completions.create(
model=model,
messages=[
{"role": "system", "content": "Bạn là trợ lý AI chuyên nghiệp."},
{"role": "user", "content": prompt}
],
temperature=temperature,
**kwargs
)
return {
"success": True,
"content": response.choices[0].message.content,
"usage": response.usage,
"attempts": attempt + 1
}
except RateLimitError as e:
last_error = e
retry_after = None
# Parse Retry-After header
if hasattr(e, 'response') and e.response:
retry_after = e.response.headers.get('Retry-After')
if retry_after:
retry_after = int(retry_after)
if attempt < self.max_retries:
delay = self.calculate_delay(attempt, retry_after)
print(f"⚠️ Rate limit hit. Retry sau {delay:.2f}s (attempt {attempt + 1}/{self.max_retries})")
await asyncio.sleep(delay)
else:
print(f"❌ Đã vượt quá số lần retry tối đa")
except APITimeoutError as e:
last_error = e
if attempt < self.max_retries:
delay = self.calculate_delay(attempt)
await asyncio.sleep(delay)
else:
break
except Exception as e:
# Các lỗi khác không retry
return {
"success": False,
"error": str(e),
"error_type": type(e).__name__
}
return {
"success": False,
"error": str(last_error),
"error_type": "MaxRetriesExceeded",
"attempts": self.max_retries + 1
}
Sử dụng
handler = GeminiRateLimitHandler(
base_delay=1.0,
max_delay=30.0,
max_retries=5
)
async def main():
result = await handler.call_with_retry(
prompt="Giải thích về RAG architecture",
model="gemini-2.5-pro",
temperature=0.7
)
if result["success"]:
print(f"✅ Thành công sau {result['attempts']} attempts")
print(f"📊 Tokens used: {result['usage']}")
else:
print(f"❌ Thất bại: {result['error']}")
asyncio.run(main())Hệ Thống Rate Limiter Tập Trung Với Token Bucket
Để quản lý rate limiting ở cấp ứng dụng (không phụ thuộc vào retry), triển khai token bucket algorithm:
import time
import threading
from collections import deque
from dataclasses import dataclass
from typing import Optional
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class RateLimitConfig:
"""Cấu hình rate limiting"""
requests_per_minute: int = 60
tokens_per_minute: int = 4_000_000
burst_size: int = 10
class TokenBucketRateLimiter:
"""
Token Bucket Algorithm cho rate limiting chính xác
- requests_per_minute: Số request cho phép mỗi phút
- tokens_per_minute: Số token cho phép mỗi phút
- burst_size: Số request có thể burst ngay lập tức
"""
def __init__(self, config: RateLimitConfig):
self.config = config
self.request_tokens = config.burst_size
self.token_tokens = 0.0
self.last_refill_time = time.time()
self.request_history = deque(maxlen=1000) # Lịch sử 1000 requests
self._lock = threading.Lock()
# Refill rates (tokens per second)
self.request_refill_rate = config.requests_per_minute / 60.0
self.token_refill_rate = config.tokens_per_minute / 60.0
def _refill(self):
"""Nạp lại tokens dựa trên thời gian trôi qua"""
now = time.time()
elapsed = now - self.last_refill_time
# Nạp request tokens
self.request_tokens = min(
self.config.burst_size,
self.request_tokens + elapsed * self.request_refill_rate
)
# Nạp token tokens
self.token_tokens = min(
self.config.tokens_per_minute,
self.token_tokens + elapsed * self.token_refill_rate
)
self.last_refill_time = now
def acquire_request(self, estimated_tokens: int = 1000) -> tuple[bool, float]:
"""
Thử acquire request và tokens cần thiết
Returns: (success, wait_time_seconds)
"""
with self._lock:
self._refill()
# Kiểm tra request tokens
if self.request_tokens < 1:
wait_for_request = (1 - self.request_tokens) / self.request_refill_rate
else:
wait_for_request = 0
# Kiểm tra token tokens
if self.token_tokens < estimated_tokens:
wait_for_token = (estimated_tokens - self.token_tokens) / self.token_refill_rate
else:
wait_for_token = 0
wait_time = max(wait_for_request, wait_for_token)
if wait_time == 0:
# Acquire thành công
self.request_tokens -= 1
self.token_tokens -= estimated_tokens
self.request_history.append(time.time())
return True, 0.0
return False, wait_time
def wait_and_acquire(self, estimated_tokens: int = 1000, timeout: float = 60.0) -> bool:
"""Chờ cho đến khi có thể acquire hoặc timeout"""
start_time = time.time()
while True:
success, wait_time = self.acquire_request(estimated_tokens)
if success:
return True
if time.time() - start_time + wait_time > timeout:
logger.error(f"⏱️ Timeout sau {timeout}s chờ rate limit")
return False
logger.info(f"⏳ Chờ {wait_time:.2f}s do rate limit...")
time.sleep(min(wait_time, 1.0)) # Không sleep quá 1s
def get_stats(self) -> dict:
"""Lấy thống kê rate limiting"""
with self._lock:
now = time.time()
recent_requests = [
t for t in self.request_history
if now - t < 60
]
return {
"available_requests": self.request_tokens,
"available_tokens": self.token_tokens,
"requests_last_60s": len(recent_requests),
"refill_rate_rpm": self.config.requests_per_minute,
"token_rate_tpm": self.config.tokens_per_minute
}
Singleton instance
_rate_limiter: Optional[TokenBucketRateLimiter] = None
def get_rate_limiter(config: Optional[RateLimitConfig] = None) -> TokenBucketRateLimiter:
global _rate_limiter
if _rate_limiter is None:
_rate_limiter = TokenBucketRateLimiter(config or RateLimitConfig())
return _rate_limiter
Sử dụng trong API calls
def call_gemini_with_rate_limit(prompt: str, max_tokens: int = 2000):
limiter = get_rate_limiter()
# Ước tính tokens cần thiết (rough estimate)
estimated_input_tokens = len(prompt.split()) * 1.3
total_tokens = estimated_input_tokens + max_tokens
if not limiter.wait_and_acquire(total_tokens):
raise Exception("Rate limit timeout - vui lòng thử lại sau")
# Gọi API
client = OpenAI(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1"
)
response = client.chat.completions.create(
model="gemini-2.5-pro",
messages=[{"role": "user", "content": prompt}],
max_tokens=max_tokens
)
return response
In ra stats
stats = get_rate_limiter().get_stats()
print(f"📊 Rate Limiter Stats: {stats}")Batch Processing Với Chunking Thông Minh
Để tối ưu hóa việc sử dụng quota, batch processing là giải pháp hiệu quả:
import asyncio
from typing import List, Dict, Any
from dataclasses import dataclass
import tiktoken # Tokenizer
@dataclass
class BatchConfig:
max_batch_size: int = 100
max_tokens_per_request: int = 200_000
concurrency_limit: int = 5
class SmartBatchProcessor:
"""Xử lý batch thông minh với token-aware chunking"""
def __init__(self, config: BatchConfig = None):
self.config = config or BatchConfig()
self.encoder = tiktoken.get_encoding("cl100k_base") # GPT-4 tokenizer
self.semaphore = asyncio.Semaphore(self.config.concurrency_limit)
def count_tokens(self, text: str) -> int:
"""Đếm tokens trong văn bản"""
return len(self.encoder.encode(text))
def smart_chunk(self, items: List[str], priority: List[int] = None) -> List[List[str]]:
"""
Chia nhỏ items thành batches dựa trên token count
- Ưu tiên items quan trọng
- Đảm bảo không vượt max_tokens_per_request
"""
if priority is None:
priority = [0] * len(items)
# Sắp xếp theo priority (cao đến thấp)
sorted_items = sorted(zip(priority, items), key=lambda x: -x[0])
batches = []
current_batch = []
current_tokens = 0
for prio, item in sorted_items:
item_tokens = self.count_tokens(item)
# Nếu item đơn lẻ vượt limit, cắt nhỏ
if item_tokens > self.config.max_tokens_per_request:
if current_batch:
batches.append(current_batch)
current_batch = []
current_tokens = 0
# Cắt item thành chunks nhỏ hơn
sub_chunks = self._split_long_text(item, self.config.max_tokens_per_request // 4)
batches.extend([[chunk] for chunk in sub_chunks])
continue
# Kiểm tra nếu thêm item sẽ vượt limit
if current_tokens + item_tokens > self.config.max_tokens_per_request:
if current_batch:
batches.append(current_batch)
current_batch = [item]
current_tokens = item_tokens
else:
current_batch.append(item)
current_tokens += item_tokens
if len(current_batch) >= self.config.max_batch_size:
batches.append(current_batch)
current_batch = []
current_tokens = 0
if current_batch:
batches.append(current_batch)
return batches
def _split_long_text(self, text: str, max_tokens: int) -> List[str]:
"""Cắt văn bản dài thành các đoạn nhỏ hơn"""
sentences = text.split('. ')
chunks = []
current_chunk = []
current_tokens = 0
for sentence in sentences:
sentence_tokens = self.count_tokens(sentence)
if current_tokens + sentence_tokens > max_tokens and current_chunk:
chunks.append('. '.join(current_chunk) + '.')
current_chunk = [sentence]
current_tokens = sentence_tokens
else:
current_chunk.append(sentence)
current_tokens += sentence_tokens
if current_chunk:
chunks.append('. '.join(current_chunk) + '.')
return chunks
Triển khai async batch processing
async def process_batch_async(
batch: List[str],
processor: SmartBatchProcessor,
client: OpenAI
) -> List[Dict[str, Any]]:
"""Xử lý một batch với concurrency control"""
async with processor.semaphore:
tasks = []
for item in batch:
task = client.chat.completions.create(
model="gemini-2.5-pro",
messages=[
{"role": "system", "content": "Phân tích và trả lời ngắn gọn."},
{"role": "user", "content": item}
],
temperature=0.3,
max_tokens=500
)
tasks.append(task)
results = await asyncio.gather(*tasks, return_exceptions=True)
processed = []
for item, result in zip(batch, results):
if isinstance(result, Exception):
processed.append({
"input": item,
"success": False,
"error": str(result)
})
else:
processed.append({
"input": item,
"success": True,
"output": result.choices[0].message.content,
"usage": result.usage
})
return processed
Sử dụng
async def main():
processor = SmartBatchProcessor(BatchConfig(
max_batch_size=50,
max_tokens_per_request=150_000,
concurrency_limit=3
))
# Dữ liệu mẫu
products = [
"Mô tả sản phẩm A..." * 100,
"Mô tả sản phẩm B..." * 100,
# ... 1000+ items
]
# Tạo batches
batches = processor.smart_chunk(products)
print(f"📦 Đã chia thành {len(batches)} batches")
# Xử lý với rate limiting
client = OpenAI(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1"
)
all_results = []
for i, batch in enumerate(batches):
print(f"🔄 Processing batch {i+1}/{len(batches)}...")
results = await process_batch_async(batch, processor, client)
all_results.extend(results)
# Delay giữa các batches để tránh rate limit
await asyncio.sleep(1)
print(f"✅ Hoàn thành {len(all_results)} items")
asyncio.run(main())Monitoring Dashboard Cho Production
Theo dõi quota usage thời gian thực là yếu tố then chốt:
import time
from datetime import datetime, timedelta
from collections import defaultdict
import threading
class QuotaMonitor:
"""Monitor quota usage và alerting"""
def __init__(self, daily_limit: int = 1_000_000):
self.daily_limit = daily_limit
self.daily_usage = defaultdict(int)
self.minute_usage = defaultdict(lambda: defaultdict(int))
self.request_timestamps = defaultdict(list)
self._lock = threading.Lock()
self._start_time = datetime.now()
def record_request(self, tokens_used: int):
"""Ghi nhận một request"""
now = datetime.now()
today = now.date().isoformat()
minute_key = now.strftime("%Y-%m-%d %H:%M")
with self._lock:
self.daily_usage[today] += tokens_used
self.minute_usage[today][minute_key] += tokens_used
self.request_timestamps[today].append(now.timestamp())
def get_usage_stats(self) -> dict:
"""Lấy thống kê sử dụng"""
now = datetime.now()
today = now.date().isoformat()
minute_key = now.strftime("%Y-%m-%d %H:%M")
with self._lock:
daily_tokens = self.daily_usage[today]
minute_tokens = self.minute_usage[today].get(minute_key, 0)
# Đếm requests trong phút hiện tại
recent_requests = [
t for t in self.request_timestamps[today]
if now.timestamp() - t < 60
]
return {
"daily_tokens": daily_tokens,
"daily_limit": self.daily_limit,
"daily_remaining": self.daily_limit - daily_tokens,
"daily_percent": (daily_tokens / self.daily_limit) * 100,
"minute_tokens": minute_tokens,
"requests_last_minute": len(recent_requests),
"estimated_daily_cost": (daily_tokens / 1_000_000) * 2.50, # $2.50/MTok
"uptime_hours": (now - self._start_time).total_seconds() / 3600
}
def check_limits(self) -> tuple[bool, str]:
"""Kiểm tra nếu sắp vượt limit"""
stats = self.get_usage_stats()
# Warning khi > 80%
if stats["daily_percent"] > 80:
return False, f"⚠️ Cảnh báo: Đã sử dụng {stats['daily_percent']:.1f}% daily quota"
# Block khi > 95%
if stats["daily_percent"] > 95:
return False, f"🚫 Dừng: Đã sử dụng {stats['daily_percent']:.1f}% daily quota"
# Warning requests/minute
if stats["requests_last_minute"] > 50:
return False, f"⚠️ Cảnh báo: {stats['requests_last_minute']} requests/minute"
return True, "OK"
Singleton
monitor = QuotaMonitor(daily_limit=5_000_000) # 5M tokens/ngày
Decorator để auto-monitor
def monitored_call(func):
"""Decorator tự động ghi nhận usage"""
def wrapper(*args, **kwargs):
result = func(*args, **kwargs)
# Estimate tokens
tokens = 1000 # Default estimate
if hasattr(result, 'usage'):
tokens = result.usage.total_tokens
monitor.record_request(tokens)
# Check limits trước khi gọi tiếp
can_continue, message = monitor.check_limits()
if not can_continue:
print(message)
# Implement fallback logic ở đây
return result
return wrapper
Dashboard
def print_dashboard():
"""In dashboard stats"""
stats = monitor.get_usage_stats()
print("\n" + "="*50)
print(f"📊 QUOTA DASHBOARD - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
print("="*50)
print(f"💰 Daily Usage: {stats['daily_tokens']:,} / {stats['daily_limit']:,} tokens")
print(f" ({stats['daily_percent']:.1f}%)")
print(f" Còn lại: {stats['daily_remaining']:,} tokens")
print(f" Chi phí ước tính: ${stats['estimated_daily_cost']:.2f}")
print(f"\n⚡ Minute Usage: {stats['minute_tokens']:,} tokens")
print(f" Requests/min: {stats['requests_last_minute']}")
print(f"\n⏱️ Uptime: {stats['uptime_hours']:.2f} giờ")
print("="*50)
Chạy dashboard loop
def dashboard_loop(interval: int = 30):
"""Loop cập nhật dashboard"""
while True:
print_dashboard()
time.sleep(interval)
Stats: 429 error rate tracking
class ErrorTracker:
"""Theo dõi error rates"""
def __init__(self):
self.errors = defaultdict(list)
self._lock = threading.Lock()
def record_error(self, error_type: str):
now = time.time()
with self._lock:
# Chỉ giữ lại errors trong 5 phút
self.errors[error_type] = [
t for t in self.errors[error_type]
if now - t < 300
] + [now]
def get_error_rate(self, error_type: str = "429", window: int = 300) -> float:
"""Tính error rate trong window (seconds)"""
now = time.time()
with self._lock:
recent = [
t for t in self.errors[error_type]
if now - t < window
]
total_requests = len(self.errors.get("total", []))
if total_requests == 0:
return 0.0
return len(recent) / (window / 60) # errors per minute
Sử dụng
error_tracker = ErrorTracker()
error_tracker.record_error("429") # Khi gặp rate limit
error_tracker.record_error("429")
print(f"📉 Error rate (429): {error_tracker.get_error_rate('429'):.2f}%")Lỗi Thường Gặp Và Cách Khắc Phục
1. Lỗi 429 Too Many Requests
Triệu chứng: API trả về lỗi 429 ngay cả khi số lượng request không nhiều.
Nguyên nhân: Có thể do burst limit hoặc quota limit theo thời gian.
# Giải pháp: Implement proper retry với check
def safe_api_call_with_quota_check(client, prompt, max_retries=3):
"""
Gọi API an toàn với kiểm tra quota trước
"""
stats = monitor.get_usage_stats()
# Nếu quota > 90%, chờ đến khi reset
if stats['daily_percent'] > 90:
print("⏳ Quota cao - chờ rate limit giảm...")
time.sleep(60) # Chờ 1 phút
for attempt in range(max_retries):
try:
response = client.chat.completions.create(
model="gemini-2.5-pro",
messages=[{"role": "user", "content": prompt}]
)
monitor.record_request(response.usage.total_tokens)
return response
except RateLimitError as e:
if attempt < max_retries - 1:
wait = int(e.headers.get('Retry-After', 5))
print(f"⏳ Chờ {wait}s theo Retry-After header...")
time.sleep(wait)
else:
raise Exception(f"Quota exceeded sau {max_retries} retries")
return None2. Lỗi Timeout Khi Xử Lý Batch Lớn
Triệu chứng: Request timeout khi gửi prompt quá dài hoặc xử lý nhiều items.
# Giải pháp: Chunking thông minh + async với timeout riêng
import asyncio
async def async_call_with_timeout(
client,
prompt: str,
timeout: float = 30.0,
max_tokens: int = 4000
):
"""Gọi async với timeout cụ thể"""
try:
loop = asyncio.get_event_loop()
response = await asyncio.wait_for(
loop.run_in_executor(
None,
lambda: client.chat.completions.create(
model="gemini-2.5-pro",
messages=[{"role": "user", "content": prompt}],
max_tokens=max_tokens,
timeout=timeout
)
),
timeout=timeout + 5 # Thêm buffer
)
return response
except asyncio.TimeoutError:
# Fallback: gọi lại với prompt ngắn hơn
print(f"⚠️ Timeout sau {timeout}s - thử lại với prompt ngắn hơn...")
shortened_prompt = truncate_prompt(prompt, max_chars=5000)
return client.chat.completions.create(
model="gemini-2.5-pro-flash", # Model nhanh hơn
messages=[{"role": "user", "content": shortened_prompt}],
max_tokens=2000,
timeout=timeout
)
def truncate_prompt(prompt: str, max_chars: int = 5000) -> str:
"""Cắt prompt nếu quá dài"""
if len(prompt) <= max_chars:
return prompt
return prompt[:max_chars] + "\n\n[Prompt đã được cắt ngắn do giới hạn độ dài]"3. Lỗi Quota Exhausted (Hết Hạn Mức)
Triệu chứng: Tất cả requests đều bị từ chối với thông báo quota exceeded.
# Giải pháp: Multi-tier fallback + caching
class TieredAPIClient:
"""
Client với fallback nhiều cấp độ
1. Gemini 2.5 Pro (ưu tiên)
2. Gemini 2.5 Flash (thay thế)
3. DeepSeek V3.2 (backup rẻ nhất - $0.42/MTok)
"""
def __init__(self):
self.tiers = [
{"model": "gemini-2.5-pro", "cost_per_mtok": 2.50, "priority": 1},
{"model": "gemini-2.5-flash", "cost_per_mtok": 2.50, "priority": 2},
{"model": "deepseek-v3.2", "cost_per_mtok": 0.42, "priority": 3},
]
self.cache = {} # Simple LRU cache
self.cache_hits = 0
self.cache_misses = 0
def get_cached_response(self, prompt_hash: str) -> Optional[str]:
"""Kiểm tra cache trước"""
if prompt_hash in self.cache:
self.cache_hits += 1
return self.cache[prompt_hash]
self.cache_misses += 1
return None
async def call_with_fallback(self, prompt: str) -> dict:
"""Gọi với fallback tự động"""
prompt_hash = str(hash(prompt))[:16]
# Check cache trước
cached = self.get_cached_response(prompt_hash)
if cached:
return {"success": True, "content": cached, "source": "cache"}
for tier in self.tiers:
try:
stats = monitor.get_usage_stats()
# Skip tier cao nếu quota còn ít
if tier["priority"] == 1 and stats["daily_percent"] > 70:
continue
client = OpenAI(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1"
)
response = client.chat.completions.create(
model=tier["model"],
messages=[{"role": "user", "content": prompt}]
)
content = response.choices[0].message.content
# Cache kết quả
self.cache[prompt_hash] = content
if len(self.cache) > 1000: # LRU eviction
oldest = list(self.cache.keys())[0]
del self.cache[oldest]
return {
"success": True,
"content": content,
"model": tier["model"],
"cost_per_mtok": tier["cost_per_mtok"]
}
except RateLimitError:
print(f"⚠️ {tier['model']} rate limited - thử tier tiếp theo...")
continue
except Exception as e:
print(f"❌ Lỗi {tier['model']}: {e}")
continue
return {"success": False, "error": "Tất cả tiers đều thất bại"}
Sử dụng
tiered_client = TieredAPIClient()
result = asyncio.run(tiered_client.call_with_fallback("Xin chào"))
print(f"📊 Cache hits: {tiered_client.cache_hits}, misses: {tiered_client.cache_misses}")4. Lỗi Authentication/Invalid API Key
Triệu chứng: Lỗi 401 Unauthorized khi gọi API.
# Giải pháp: Validation + error handling chuẩn
from pydantic import BaseModel, validator
import os
class APIConfig(BaseModel):
api_key: str
base_url: str = "https://api.holysheep.ai/v1"
@validator('api_key')
def validate_key(cls, v):
if not v or len(v) < 10:
raise ValueError("API key không hợp lệ")
if v == "YOUR_HOLYSHEEP_API_KEY":
raise ValueError("Vui lòng thay YOUR_HOLYSHEEP_API_KEY bằng API key thực")
return v
@validator('base_url')
def validate_url(cls, v):
if not v.startswith("https://"):
raise ValueError("base_url phải sử dụng HTTPS")
return v
def create