Khi tôi lần đầu deploy hệ thống xử lý ngôn ngữ tự nhiên cho startup của mình vào tháng 3/2025, mọi thứ tưởng chừng hoàn hảo cho đến khi dashboard billing hiển thị con số $2,847.50 cho một ngày. Đó là lúc tôi nhận ra mình đã không hiểu cách Gemini API tính phí batch request và cách tối ưu hóa nó. Bài viết này là tổng kết 6 tháng thực chiến, từ những lỗi đau thương nhất đến giải pháp đưa chi phí xuống chỉ còn $127.80/ngày — tiết kiệm 95% chi phí.
Kịch bản lỗi thực tế: Khi账单 trở thành cơn ác mộng
03:47 sáng, Slack alert reo liên tục. Monitoring dashboard hiển thị:
ERROR: RateLimitError: 429 Too Many Requests
DETAIL: Quota exceeded for dimension 'GenerateContentRequestsPerMinute'
RESOLUTION: Retry-After: 45 seconds
TIMESTAMP: 2025-03-15T03:47:22.847Z
SESSION_ID: batch_proc_a4f8b2c1
BATCH_SIZE: 500 requests
ACTUAL_COST: $0.00 (rate limited)
Tưởng chừng vô hại, nhưng phía sau alert này là cả một chuỗi lỗi domino: rate limit触发 → retry exponential backoff → tăng request volume → quota exhaustion toàn bộ ngày → cuối tháng账单 tăng vọt. Hãy cùng tôi phân tích từng nguyên nhân và cách khắc phục.
Kiến trúc Batch Request tối ưu với HolySheep AI
Trước khi đi vào chi tiết, bạn cần hiểu rằng Gemini API qua HolySheep AI có cấu trúc pricing khác biệt đáng kể so với官方 Google:
- Gemini 2.5 Flash: $2.50/1M tokens (so với Google $7.50/1M tokens — tiết kiệm 66%)
- Độ trễ trung bình: <50ms cho batch 100 requests
- Tín dụng miễn phí: Đăng ký nhận ngay credits để test không giới hạn
- Thanh toán: Hỗ trợ WeChat Pay, Alipay, Visa — không cần thẻ quốc tế
Cấu hình Batch Request: Mô hình Producer-Consumer
Đây là kiến trúc production-ready mà tôi sử dụng cho hệ thống xử lý 10 triệu tokens/ngày:
# config.py - Cấu hình tối ưu cho batch processing
import os
from dataclasses import dataclass
from typing import Optional
@dataclass
class HolySheepConfig:
"""Cấu hình HolySheep AI - Gemini API"""
base_url: str = "https://api.holysheep.ai/v1"
api_key: str = os.getenv("HOLYSHEEP_API_KEY")
# Tối ưu batch parameters
batch_size: int = 100 # Số requests/batch
max_concurrent_batches: int = 5 # Concurrent batches
rate_limit_rpm: int = 600 # Requests per minute (60% của giới hạn)
retry_attempts: int = 3
retry_backoff_base: float = 2.0 # Exponential backoff base
max_retry_delay: float = 60.0 # Max delay seconds
# Timeout settings
connect_timeout: float = 10.0
read_timeout: float = 30.0
# Cost optimization
model: str = "gemini-2.5-flash" # Model rẻ nhất, nhanh nhất
max_tokens_per_request: int = 8192 # Giới hạn output để tiết kiệm
config = HolySheepConfig()# batch_processor.py - Xử lý batch với concurrency control
import asyncio
import aiohttp
import time
from typing import List, Dict, Any
from dataclasses import dataclass
from .config import HolySheepConfig
@dataclass
class BatchRequest:
prompt: str
metadata: Dict[str, Any]
@dataclass
class BatchResponse:
text: str
usage: Dict[str, int]
latency_ms: float
cost_usd: float
error: Optional[str] = None
class HolySheepBatchProcessor:
"""
HolySheep AI Batch Processor - Tối ưu cho high-volume requests
Sử dụng semaphore để kiểm soát concurrency
"""
def __init__(self, config: HolySheepConfig):
self.config = config
self.semaphore = asyncio.Semaphore(config.max_concurrent_batches)
self.request_count = 0
self.total_cost = 0.0
async def process_single(
self,
session: aiohttp.ClientSession,
request: BatchRequest
) -> BatchResponse:
"""Xử lý một request đơn lẻ với retry logic"""
start_time = time.time()
for attempt in range(self.config.retry_attempts):
try:
async with session.post(
f"{self.config.base_url}/chat/completions",
json={
"model": self.config.model,
"messages": [{"role": "user", "content": request.prompt}],
"max_tokens": self.config.max_tokens_per_request,
"temperature": 0.7
},
headers={
"Authorization": f"Bearer {self.config.api_key}",
"Content-Type": "application/json"
},
timeout=aiohttp.ClientTimeout(
total=self.config.read_timeout,
connect=self.config.connect_timeout
)
) as response:
if response.status == 200:
data = await response.json()
latency = (time.time() - start_time) * 1000
# Tính chi phí
usage = data.get("usage", {})
input_tokens = usage.get("prompt_tokens", 0)
output_tokens = usage.get("completion_tokens", 0)
# HolySheep pricing: $2.50/1M tokens
cost = (input_tokens + output_tokens) / 1_000_000 * 2.50
self.request_count += 1
self.total_cost += cost
return BatchResponse(
text=data["choices"][0]["message"]["content"],
usage=usage,
latency_ms=latency,
cost_usd=cost
)
elif response.status == 429:
# Rate limit - exponential backoff
retry_after = self.config.retry_backoff_base ** attempt
retry_after = min(retry_after, self.config.max_retry_delay)
await asyncio.sleep(retry_after)
continue
elif response.status == 401:
return BatchResponse(
text="",
usage={},
latency_ms=0,
cost_usd=0,
error="Unauthorized: Kiểm tra API key"
)
else:
error_text = await response.text()
return BatchResponse(
text="",
usage={},
latency_ms=0,
cost_usd=0,
error=f"HTTP {response.status}: {error_text}"
)
except aiohttp.ClientError as e:
if attempt == self.config.retry_attempts - 1:
return BatchResponse(
text="", usage={}, latency_ms=0, cost_usd=0,
error=f"ConnectionError: {str(e)}"
)
await asyncio.sleep(self.config.retry_backoff_base ** attempt)
return BatchResponse(
text="", usage={}, latency_ms=0, cost_usd=0,
error="Max retries exceeded"
)
async def process_batch(
self,
requests: List[BatchRequest]
) -> List[BatchResponse]:
"""Xử lý batch với concurrency control"""
async with aiohttp.ClientSession() as session:
tasks = []
for req in requests:
async with self.semaphore:
task = self.process_single(session, req)
tasks.append(task)
responses = await asyncio.gather(*tasks)
# Thêm delay giữa các batch để tránh rate limit
await asyncio.sleep(60.0 / self.config.rate_limit_rpm * len(requests))
return responses
Sử dụng
async def main():
config = HolySheepConfig()
processor = HolySheepBatchProcessor(config)
test_requests = [
BatchRequest(
prompt=f"Phân tích dữ liệu #{i}: cho startup {i}",
metadata={"index": i, "type": "analysis"}
)
for i in range(100)
]
results = await processor.process_batch(test_requests)
print(f"Processed: {processor.request_count} requests")
print(f"Total cost: ${processor.total_cost:.4f}")
successful = [r for r in results if not r.error]
failed = [r for r in results if r.error]
print(f"Success rate: {len(successful)/len(results)*100:.1f}%")
if __name__ == "__main__":
asyncio.run(main())Tối ưu hóa chi phí: Chiến lược 3 lớp
Qua 6 tháng thực chiến, tôi đã phát triển chiến lược tiết kiệm chi phí 3 lớp:
Lớp 1: Tối ưu Model Selection
Không phải lúc nào cũng cần Gemini 2.5 Pro. Bảng so sánh chi phí thực tế:
# model_selector.py - Chọn model tối ưu chi phí
from enum import Enum
from typing import Optional, List, Dict
from dataclasses import dataclass
class TaskComplexity(Enum):
SIMPLE = "simple" # Trả lời ngắn, classification
MEDIUM = "medium" # Tóm tắt, translation
COMPLEX = "complex" # Phân tích sâu, reasoning
@dataclass
class ModelPricing:
name: str
input_cost_per_million: float # $/1M tokens
output_cost_per_million: float
avg_latency_ms: float
best_for: List[str]
HolySheep AI Pricing 2026
HOLYSHEEP_MODELS = {
"gemini-2.5-flash": ModelPricing(
name="gemini-2.5-flash",
input_cost_per_million=2.50,
output_cost_per_million=2.50,
avg_latency_ms=45,
best_for=["batch_processing", "high_volume", "simple_tasks"]
),
"gpt-4.1": ModelPricing(
name="gpt-4.1",
input_cost_per_million=8.00,
output_cost_per_million=8.00,
avg_latency_ms=120,
best_for=["complex_reasoning", "long_context"]
),
"claude-sonnet-4.5": ModelPricing(
name="claude-sonnet-4.5",
input_cost_per_million=15.00,
output_cost_per_million=15.00,
avg_latency_ms=150,
best_for=["creative_writing", "analysis"]
),
"deepseek-v3.2": ModelPricing(
name="deepseek-v3.2",
input_cost_per_million=0.42,
output_cost_per_million=0.42,
avg_latency_ms=80,
best_for=["code_generation", "budget_tasks"]
)
}
class ModelSelector:
"""Chọn model tối ưu chi phí dựa trên task complexity"""
def select_model(
self,
task_type: TaskComplexity,
context_length: int
) -> str:
"""
Selection logic:
- SIMPLE task → Gemini 2.5 Flash (tiết kiệm 95% so với Claude)
- MEDIUM task → DeepSeek V3.2 (rẻ nhất cho translation/summary)
- COMPLEX task → GPT-4.1 (balance giữa capability và cost)
"""
if task_type == TaskComplexity.SIMPLE:
return "gemini-2.5-flash"
elif task_type == TaskComplexity.MEDIUM:
if context_length < 32000:
return "deepseek-v3.2" # Rẻ hơn 83% so với Flash
return "gemini-2.5-flash"
else: # COMPLEX
return "gpt-4.1"
def estimate_cost(
self,
model: str,
input_tokens: int,
output_tokens: int
) -> Dict[str, float]:
"""Ước tính chi phí cho một request"""
pricing = HOLYSHEEP_MODELS.get(model)
if not pricing:
raise ValueError(f"Model {model} không được hỗ trợ")
input_cost = (input_tokens / 1_000_000) * pricing.input_cost_per_million
output_cost = (output_tokens / 1_000_000) * pricing.output_cost_per_million
total = input_cost + output_cost
return {
"input_cost": round(input_cost, 6),
"output_cost": round(output_cost, 6),
"total_cost": round(total, 6),
"savings_vs_claude": round(total * 14.58, 6) # Claude = $15/M
}
def optimize_batch(self, requests: List[Dict]) -> List[str]:
"""Phân loại và tối ưu batch requests"""
optimized = []
for req in requests:
task = self._classify_task(req)
model = self.select_model(task, req.get("context_length", 0))
optimized.append(model)
return optimized
Ví dụ sử dụng
selector = ModelSelector()
Test cost estimation
test_cases = [
{"model": "gemini-2.5-flash", "input": 10000, "output": 5000},
{"model": "deepseek-v3.2", "input": 10000, "output": 5000},
{"model": "claude-sonnet-4.5", "input": 10000, "output": 5000},
]
print("So sánh chi phí cho 15,000 tokens:")
for tc in test_cases:
cost = selector.estimate_cost(tc["model"], tc["input"], tc["output"])
print(f"{tc['model']:20} - ${cost['total_cost']:.4f}")Lớp 2: Caching và Deduplication
# smart_cache.py - Cache thông minh giảm 40% chi phí
import hashlib
import json
import redis
from typing import Optional, Any, Callable
from functools import wraps
import time
class SemanticCache:
"""
Cache với semantic similarity - giảm chi phí đáng kể
Vì nhiều prompt giống nhau sẽ được serve từ cache
"""
def __init__(self, redis_client: redis.Redis, ttl_seconds: int = 3600):
self.redis = redis_client
self.ttl = ttl_seconds
self.hit_count = 0
self.miss_count = 0
self.total_savings = 0.0
def _normalize_prompt(self, prompt: str) -> str:
"""Chuẩn hóa prompt để tăng cache hit rate"""
return prompt.strip().lower().replace("\n", " ").replace(" ", " ")
def _get_cache_key(self, prompt: str, model: str) -> str:
"""Tạo cache key từ prompt hash"""
normalized = self._normalize_prompt(prompt)
hash_val = hashlib.sha256(f"{normalized}:{model}".encode()).hexdigest()[:16]
return f"sem_cache:{hash_val}"
def get(self, prompt: str, model: str) -> Optional[dict]:
"""Lấy kết quả từ cache"""
key = self._get_cache_key(prompt, model)
cached = self.redis.get(key)
if cached:
self.hit_count += 1
data = json.loads(cached)
# Tính savings (giả định cached response là 5000 tokens)
self.total_savings += (5000 / 1_000_000) * 2.50
return data
else:
self.miss_count += 1
return None
def set(self, prompt: str, model: str, response: dict):
"""Lưu kết quả vào cache"""
key = self._get_cache_key(prompt, model)
self.redis.setex(key, self.ttl, json.dumps(response))
def get_stats(self) -> dict:
"""Thống kê cache performance"""
total = self.hit_count + self.miss_count
hit_rate = (self.hit_count / total * 100) if total > 0 else 0
return {
"hits": self.hit_count,
"misses": self.miss_count,
"hit_rate": f"{hit_rate:.1f}%",
"total_savings_usd": round(self.total_savings, 2)
}
def cached_api_call(cache: SemanticCache, model: str):
"""Decorator cho API calls với caching"""
def decorator(func: Callable):
@wraps(func)
async def wrapper(prompt: str, *args, **kwargs):
# Check cache first
cached = cache.get(prompt, model)
if cached:
return {**cached, "cached": True}
# Call API
result = await func(prompt, *args, **kwargs)
# Store in cache
cache.set(prompt, model, result)
return {**result, "cached": False}
return wrapper
return decorator
Sử dụng trong batch processing
class OptimizedBatchProcessor:
def __init__(self, api_processor, cache: SemanticCache):
self.processor = api_processor
self.cache = cache
async def process_with_cache(self, prompts: List[str]) -> List[dict]:
"""Xử lý batch với caching thông minh"""
results = []
for prompt in prompts:
# Check cache
cached = self.cache.get(prompt, "gemini-2.5-flash")
if cached:
results.append({**cached, "source": "cache"})
else:
# Call API
response = await self.processor.call_api(prompt)
self.cache.set(prompt, "gemini-2.5-flash", response)
results.append({**response, "source": "api"})
return results
Ví dụ: Batch 1000 requests, giả định 40% cache hit
Without cache: 1000 * $0.0375 = $37.50
With 40% cache: 600 * $0.0375 = $22.50
Savings: $15.00 (40%)
Lớp 3: Request Batching và Token Optimization
# token_optimizer.py - Tối ưu token usage
import re
from typing import List, Dict, Tuple
class TokenOptimizer:
"""
Tối ưu token usage qua:
1. Prompt compression
2. Batch multiple requests
3. Truncate thông minh
"""
# Patterns có thể loại bỏ
REMOVABLE_PATTERNS = [
r'\s+', # Multiple spaces
r'\n{3,}', # Multiple newlines
r'^\s*', # Leading spaces
r'\s*$', # Trailing spaces
r'[?!.]{2,}', # Multiple punctuation
]
def compress_prompt(self, prompt: str, max_tokens: int = 4096) -> str:
"""Nén prompt giữ lại thông tin quan trọng"""
# Loại bỏ patterns không cần thiết
compressed = prompt
for pattern in self.REMOVABLE_PATTERNS:
compressed = re.sub(pattern, ' ', compressed)
# Ước tính token count (giả định 1 token ≈ 4 chars)
estimated_tokens = len(compressed) / 4
if estimated_tokens > max_tokens:
# Truncate từ cuối, giữ phần đầu quan trọng
compressed = compressed[:max_tokens * 4]
# Cắt ngang câu
last_period = compressed.rfind('.')
if last_period > len(compressed) * 0.7:
compressed = compressed[:last_period + 1]
return compressed.strip()
def batch_similar_requests(
self,
requests: List[Dict]
) -> List[Dict]:
"""
Gộp các request tương tự thành batch để:
- Giảm số API calls
- Tận dụng batch pricing
"""
# Nhóm theo prefix
groups: Dict[str, List[Dict]] = {}
for req in requests:
prefix = req["prompt"][:50].lower()
if prefix not in groups:
groups[prefix] = []
groups[prefix].append(req)
# Tạo batched requests
batched = []
for group_prompt, group_reqs in groups.items():
if len(group_reqs) > 1:
# Gộp thành single batch request
combined_prompt = self._create_combined_prompt(group_reqs)
batched.append({
"type": "batch",
"prompt": combined_prompt,
"original_count": len(group_reqs),
"ids": [r["id"] for r in group_reqs]
})
else:
batched.append({
"type": "single",
**group_reqs[0]
})
return batched
def _create_combined_prompt(self, requests: List[Dict]) -> str:
"""Tạo prompt gộp từ nhiều requests"""
items = []
for i, req in enumerate(requests, 1):
items.append(f"Task {i}: {req['prompt']}")
return "\n".join(items)
def estimate_savings(
self,
original_count: int,
original_avg_tokens: int,
batched_count: int
) -> Dict[str, any]:
"""
Ước tính savings khi batch:
Ví dụ: 100 requests → 20 batched calls
Original: 100 * 5000 tokens = 500,000 tokens
Batched: 20 * 5000 tokens = 100,000 tokens (do gộp context)
Savings: 80%
"""
original_cost = (original_count * original_avg_tokens / 1_000_000) * 2.50
batched_cost = (batched_count * original_avg_tokens / 1_000_000) * 2.50
return {
"original_calls": original_count,
"batched_calls": batched_count,
"original_cost_usd": round(original_cost, 4),
"batched_cost_usd": round(batched_cost, 4),
"savings_percent": round((1 - batched_cost/original_cost) * 100, 1),
"savings_usd": round(original_cost - batched_cost, 4)
}
Test optimizer
optimizer = TokenOptimizer()
Case 1: Compress long prompt
long_prompt = """
Xin chào! Tôi cần bạn phân tích dữ liệu sau đây...
Ngày: 15/03/2025
Nội dung: Phân tích doanh thu tháng 3 năm 2025
Dữ liệu: 1,234,567 VND
Yêu cầu: Tính toán tăng trưởng so với tháng trước
!!! Vui lòng phân tích kỹ lưỡng và đưa ra insights chi tiết nhất có thể...
"""
compressed = optimizer.compress_prompt(long_prompt, max_tokens=200)
print(f"Original: {len(long_prompt)} chars")
print(f"Compressed: {len(compressed)} chars")
print(f"Compression: {(1-len(compressed)/len(long_prompt))*100:.1f}%")
Case 2: Estimate batch savings
savings = optimizer.estimate_savings(100, 5000, 20)
print(f"\nBatch savings: {savings['savings_percent']}% (${savings['savings_usd']})")Lỗi thường gặp và cách khắc phục
Trong quá trình vận hành hệ thống xử lý hàng triệu tokens mỗi ngày, tôi đã gặp và xử lý hàng trăm lỗi khác nhau. Dưới đây là 5 lỗi phổ biến nhất kèm giải pháp đã được test và verify:
1. Lỗi 401 Unauthorized - API Key không hợp lệ
# ❌ SAI: Hardcode API key trực tiếp
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "sk-holysheep-abc123..." # KHÔNG BAO GIỜ làm thế này!
✅ ĐÚNG: Sử dụng environment variable
import os
from dotenv import load_dotenv
load_dotenv() # Load .env file
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = os.getenv("HOLYSHEEP_API_KEY") # Hoặc os.environ["HOLYSHEEP_API_KEY"]
if not API_KEY:
raise ValueError("HOLYSHEEP_API_KEY not found in environment variables")
Validate API key format
def validate_api_key(key: str) -> bool:
"""Kiểm tra format API key"""
if not key or len(key) < 20:
return False
if not key.startswith(("sk-", "hs-")):
return False
return True
if not validate_api_key(API_KEY):
raise ValueError("Invalid API key format. Please check your HolySheep AI dashboard.")
Sử dụng key với headers
headers = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}2. Lỗi 429 Rate Limit - Quá nhiều requests
# ❌ SAI: Gửi request liên tục không kiểm soát
async def bad_batch_processing(requests):
results = []
for req in requests: # 1000 requests liên tục
result = await call_api(req) # Sẽ bị rate limit ngay!
results.append(result)
return results
✅ ĐÚNG: Sử dụng token bucket algorithm
import asyncio
import time
from collections import deque
class TokenBucketRateLimiter:
"""
Token Bucket Rate Limiter - Kiểm soát rate limit hiệu quả
Hoạt động:
- Bucket chứa tối đa capacity tokens
- Refill refill_rate tokens mỗi giây
- Mỗi request tiêu tốn 1 token
"""
def __init__(self, capacity: int = 600, refill_rate: float = 10.0):
self.capacity = capacity
self.tokens = capacity
self.refill_rate = refill_rate
self.last_refill = time.time()
self._lock = asyncio.Lock()
self.request_times = deque(maxlen=100)
async def acquire(self, tokens: int = 1):
"""Acquire tokens, wait if necessary"""
async with self._lock:
while True:
self._refill()
if self.tokens >= tokens:
self.tokens -= tokens
self.request_times.append(time.time())
return
# Calculate wait time
deficit = tokens - self.tokens
wait_time = deficit / self.refill_rate
await asyncio.sleep(wait_time)
def _refill(self):
"""Refill tokens based on elapsed time"""
now = time.time()
elapsed = now - self.last_refill
new_tokens = elapsed * self.refill_rate
self.tokens = min(self.capacity, self.tokens + new_tokens)
self.last_refill = now
def get_stats(self) -> dict:
"""Thống kê rate limiting"""
return {
"current_tokens": round(self.tokens, 2),
"capacity": self.capacity,
"utilization": f"{(1-self.tokens/self.capacity)*100:.1f}%",
"requests_last_60s": len([t for t in self.request_times if time.time()-t < 60])
}
Sử dụng rate limiter
rate_limiter = TokenBucketRateLimiter(capacity=600, refill_rate=10.0)
async def good_batch_processing(requests: List[Dict]) -> List[Dict]:
"""Xử lý batch với rate limiting hiệu quả"""
results = []
for req in requests:
# Chờ đến khi có token
await rate_limiter.acquire()
# Gửi request
result = await call_api(req)
results.append(result)
# Log stats định kỳ
if len(results) % 100 == 0:
print(f"Progress: {len(results)}/{len(requests)}")
print(f"Rate limit stats: {rate_limiter.get_stats()}")
return results
Test: 1000 requests với rate limit 600/minute
Without limiter: ~30 seconds, then 429 error
With limiter: ~100 seconds, 0 errors, smooth processing
3. Lỗi Connection Timeout - Network instability
# ❌ SAI: Timeout quá ngắn hoặc không có retry
async def bad_api_call(session, prompt):
async with session.post(url, json=data) as response:
return await response.json() # Không có timeout, không có retry
✅ ĐÚNG: Exponential backoff với jitter
import random
import asyncio
class ResilientAPIClient:
"""
API Client với retry logic thông minh:
- Exponential backoff: 1s, 2s, 4s, 8s...
- Jitter: Thêm randomness để tránh thundering herd
- Timeout: Cấu hình được
"""
def __init__(
self,
base_url: str,
api_key: str,
max_retries: int = 5,
base_delay: float = 1.0,
max_delay: float = 60.0,
timeout: float = 30.0
):
self.base_url = base_url
self.api_key = api_key
self.max_retries = max_retries
self.base_delay = base_delay
self.max_delay = max_delay
self.timeout = timeout
self.errors_log = []
def _calculate_delay(self, attempt: int) -> float:
"""
Tính delay với exponential backoff + jitter
Formula: min(max_delay, base_delay * 2^attempt) + random(0, 1)
Ví dụ:
- Attempt 0: 1 + 0.5 = 1.5s
- Attempt 1: 2 + 0.3 = 2.3s
- Attempt 2: 4 + 0.7 = 4.7s
- Attempt 3: 8 + 0.1 = 8.1s
- Attempt 4: 16 + 0.9 = 16.9s
"""
exponential_delay = self.base_delay * (2 ** attempt)
jitter = random.uniform(0, 1)
return min(self.max_delay, exponential_delay + jitter)
async def call_with_retry(self, prompt: str) -> Dict:
"""Gọi API với retry logic"""
timeout = aiohttp.ClientTimeout(total=self.timeout)
for attempt in range(self.max_retries):
try:
async with aiohttp.ClientSession(timeout=timeout) as session:
async with session.post(
f"{self.base_url}/chat/completions",
json={
"model": "gemini-2.5-flash",
"messages": [{"role": "user", "content": prompt}]
},
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
) as response:
if response.status == 200:
return await response.json()
elif response.status == 429:
# Rate limit - retry
delay = self._calculate_delay(attempt)
print(f"Rate limited