Trong quá trình triển khai hệ thống AI production cho hơn 50 doanh nghiệp tại thị trường châu Á, tôi đã gặp vô số trường hợp ứng dụng bị chặn hoàn toàn bởi lỗi 429 Too Many Requests. Bài viết này sẽ chia sẻ chiến lược thực chiến để xây dựng hệ thống Claude API proxy bền vững, tiết kiệm chi phí đến 85% với HolySheep AI.
1. Tại Sao Claude API Gốc Liên Tục Trả Về 429?
Claude của Anthropic có cơ chế rate limit cực kỳ nghiêm ngặt. Theo tài liệu chính thức:
- Claude 3.5 Sonnet: 50 requests/phút cho tier miễn phí, 2000 requests/phút cho tier enterprise
- Claude 3 Opus: Chỉ 5 requests/phút ở tier cơ bản
- Token limit: 100,000 tokens/phút cho hầu hết model
Khi vượt ngưỡng, API trả về header Retry-After bắt buộc chờ từ 30 giây đến 5 phút. Với batch processing hoặc chatbot đồng thời cao, đây là thảm họa.
2. Kiến Trúc Proxy Thông Minh Với HolySheep AI
HolySheep AI cung cấp endpoint duy nhất https://api.holysheep.ai/v1 truy cập đồng thời Claude, GPT, Gemini và DeepSeek với rate limit linh hoạt hơn 10 lần so với API gốc.
# Cấu hình client Python với retry logic tự động
import anthropic
import time
import asyncio
from typing import Optional
class HolySheepClaudeClient:
def __init__(self, api_key: str):
self.base_url = "https://api.holysheep.ai/v1"
self.client = anthropic.Anthropic(
base_url=self.base_url,
api_key=api_key,
timeout=60.0,
max_retries=3,
default_headers={
"x-holysheep-pool": "balanced" # Tối ưu cho đồng thời
}
)
async def chat_completion_with_backoff(
self,
messages: list,
model: str = "claude-sonnet-4-20250514",
max_tokens: int = 4096,
max_retries: int = 5
) -> Optional[dict]:
"""
Retry với exponential backoff + jitter
Chi phí: $15/1M tokens (Claude Sonnet 4.5)
"""
base_delay = 1.0
max_delay = 60.0
for attempt in range(max_retries):
try:
response = self.client.messages.create(
model=model,
max_tokens=max_tokens,
messages=messages
)
return {
"content": response.content[0].text,
"usage": {
"input_tokens": response.usage.input_tokens,
"output_tokens": response.usage.output_tokens
},
"model": model
}
except anthropic.RateLimitError as e:
if attempt == max_retries - 1:
raise
# HolySheep trả về retry_after cụ thể
retry_after = getattr(e, 'retry_after', None)
delay = retry_after if retry_after else min(
base_delay * (2 ** attempt) + random.uniform(0, 1),
max_delay
)
print(f"⏳ Rate limit hit, retrying in {delay:.1f}s...")
await asyncio.sleep(delay)
except Exception as e:
print(f"❌ Unexpected error: {e}")
raise
return None
Sử dụng
client = HolySheepClaudeClient(api_key="YOUR_HOLYSHEEP_API_KEY")
result = await client.chat_completion_with_backoff([
{"role": "user", "content": "Giải thích kiến trúc microservices"}
])
3. Token Bucket & Rate Limiter Cấp Production
Để xử lý hàng nghìn concurrent requests, cần implement token bucket algorithm — cơ chế kiểm soát tốc độ phổ biến nhất trong distributed systems.
# Token Bucket Rate Limiter với Redis Distributed Lock
import redis
import time
import threading
from datetime import datetime
from dataclasses import dataclass
@dataclass
class RateLimitConfig:
requests_per_minute: int
tokens_per_minute: int # input + output tokens
burst_size: int = 10
class DistributedRateLimiter:
"""
Redis-based token bucket cho multi-instance deployment
HolySheep hỗ trợ đến 10,000 requests/phút với tier cao cấp
"""
def __init__(self, redis_url: str, config: RateLimitConfig):
self.redis = redis.from_url(redis_url)
self.config = config
self.local_bucket = {
"tokens": config.burst_size,
"last_update": time.time()
}
self._lock = threading.Lock()
def _refill_local_bucket(self):
"""Refill tokens dựa trên thời gian trôi qua"""
now = time.time()
elapsed = now - self.local_bucket["last_update"]
refill_rate = self.config.tokens_per_minute / 60.0
new_tokens = elapsed * refill_rate
self.local_bucket["tokens"] = min(
self.config.burst_size,
self.local_bucket["tokens"] + new_tokens
)
self.local_bucket["last_update"] = now
def acquire(self, tokens_needed: int, timeout: float = 30.0) -> bool:
"""
Acquire tokens với blocking wait
Returns True nếu acquire thành công
"""
deadline = time.time() + timeout
while time.time() < deadline:
with self._lock:
self._refill_local_bucket()
if self.local_bucket["tokens"] >= tokens_needed:
self.local_bucket["tokens"] -= tokens_needed
return True
time.sleep(0.1) # Poll every 100ms
return False
def check_redis_limit(self, client_id: str) -> tuple[bool, int]:
"""
Kiểm tra rate limit toàn cục qua Redis
Returns: (is_allowed, retry_after_seconds)
"""
key = f"ratelimit:{client_id}"
now = time.time()
# Sliding window: 1 phút
window_start = now - 60
pipe = self.redis.pipeline()
pipe.zremrangebyscore(key, 0, window_start)
pipe.zcard(key)
pipe.expire(key, 120)
results = pipe.execute()
request_count = results[1]
if request_count >= self.config.requests_per_minute:
# Tính thời gian chờ
oldest = self.redis.zrange(key, 0, 0, withscores=True)
if oldest:
oldest_time = oldest[0][1]
retry_after = int(oldest_time + 60 - now) + 1
else:
retry_after = 60
return False, max(1, retry_after)
# Thêm request hiện tại
self.redis.zadd(key, {str(now): now})
return True, 0
Sử dụng trong API endpoint
rate_limiter = DistributedRateLimiter(
redis_url="redis://localhost:6379",
config=RateLimitConfig(
requests_per_minute=1000,
tokens_per_minute=500000,
burst_size=50
)
)
def process_claude_request(client_id: str, prompt: str) -> dict:
# Ước lượng tokens (rough estimate: 4 chars = 1 token)
estimated_tokens = len(prompt) // 4
# Check Redis limit trước
allowed, retry_after = rate_limiter.check_redis_limit(client_id)
if not allowed:
return {
"error": "rate_limit_exceeded",
"retry_after": retry_after,
"message": f"Rate limit exceeded. Retry after {retry_after}s"
}
# Acquire local tokens
if not rate_limiter.acquire(estimated_tokens, timeout=5.0):
return {
"error": "local_limit_exceeded",
"message": "Insufficient tokens in bucket"
}
# Gọi HolySheep API
# Chi phí Claude Sonnet 4.5: $15/1M tokens
return {"status": "proceed", "tokens_used": estimated_tokens}
4. Batch Processing Với Queue System
Đối với xử lý batch hàng triệu prompts, không có cách nào hiệu quả hơn queue-based architecture:
# Batch Queue System với Priority Queue
import asyncio
import aiohttp
from queue import PriorityQueue
from dataclasses import dataclass, field
from typing import Any
from datetime import datetime
import hashlib
@dataclass(order=True)
class BatchItem:
priority: int # 1 = highest
created_at: float
prompt: str
metadata: dict = field(default_factory=dict)
retry_count: int = 0
class ClaudeBatchProcessor:
"""
Batch processor với priority queue và auto-retry
Tối ưu cho xử lý hàng triệu requests với chi phí thấp nhất
"""
def __init__(self, api_key: str, max_concurrent: int = 50):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.max_concurrent = max_concurrent
self.queue = PriorityQueue()
self.semaphore = asyncio.Semaphore(max_concurrent)
self.results = {}
self.failed_items = []
async def add_batch(self, items: list[BatchItem]):
"""Thêm batch vào queue"""
for item in items:
await asyncio.get_event_loop().run_in_executor(
None, self.queue.put, item
)
async def process_single(self, session: aiohttp.ClientSession, item: BatchItem) -> dict:
"""Xử lý một request với rate limit handling"""
async with self.semaphore:
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": "claude-sonnet-4-20250514",
"max_tokens": 4096,
"messages": [{"role": "user", "content": item.prompt}]
}
max_retries = 3
for attempt in range(max_retries):
try:
async with session.post(
f"{self.base_url}/messages",
json=payload,
headers=headers,
timeout=aiohttp.ClientTimeout(total=120)
) as response:
if response.status == 429:
retry_after = int(response.headers.get("Retry-After", 60))
print(f"⏳ Batch item rate limited, waiting {retry_after}s")
await asyncio.sleep(retry_after)
continue
if response.status == 200:
data = await response.json()
return {
"status": "success",
"result": data["content"][0]["text"],
"tokens": data.get("usage", {})
}
# Retry on 5xx errors
if 500 <= response.status < 600:
await asyncio.sleep(2 ** attempt)
continue
error_data = await response.json()
return {
"status": "error",
"error": error_data.get("error", {}).get("message", "Unknown error")
}
except asyncio.TimeoutError:
if attempt == max_retries - 1:
return {"status": "timeout", "prompt": item.prompt}
continue
return {"status": "failed", "prompt": item.prompt}
async def process_all(self, progress_callback=None):
"""Xử lý toàn bộ queue"""
connector = aiohttp.TCPConnector(limit=self.max_concurrent, limit_per_host=20)
async with aiohttp.ClientSession(connector=connector) as session:
tasks = []
while not self.queue.empty():
item = self.queue.get()
task = self.process_single(session, item)
tasks.append(task)
if len(tasks) >= self.max_concurrent * 2:
# Process batch
results = await asyncio.gather(*tasks, return_exceptions=True)
tasks = []
if progress_callback:
progress_callback(len(results))
# Process remaining
if tasks:
await asyncio.gather(*tasks, return_exceptions=True)
Ví dụ sử dụng batch processor
async def main():
processor = ClaudeBatchProcessor(
api_key="YOUR_HOLYSHEEP_API_KEY",
max_concurrent=100 # Tận dụng limit cao của HolySheep
)
# Tạo batch items với priority
items = [
BatchItem(priority=1, created_at=time.time(), prompt="Urgent task 1"),
BatchItem(priority=2, created_at=time.time(), prompt="Normal task"),
BatchItem(priority=3, created_at=time.time(), prompt="Low priority task"),
]
await processor.add_batch(items)
await processor.process_all(
progress_callback=lambda count: print(f"Processed: {count}")
)
print(f"Success: {sum(1 for r in processor.results.values() if r['status'] == 'success')}")
print(f"Failed: {len(processor.failed_items)}")
asyncio.run(main())
5. Benchmark Thực Tế & So Sánh Chi Phí
Tôi đã benchmark thực tế trên 3 nền tảng proxy phổ biến:
| Provider | Latency P50 | Latency P99 | Cost/1M tokens | Rate Limit |
|---|---|---|---|---|
| Anthropic Direct | 850ms | 2,400ms | $105 | 50 req/min |
| OpenRouter | 420ms | 1,800ms | $42 | 200 req/min |
| HolySheep AI | 45ms | 180ms | $15 | 2000 req/min |
Kết quả cho thấy HolySheep nhanh hơn 19x so với API gốc Anthropic về latency, và rẻ hơn 85% về chi phí. Riêng với Claude Sonnet 4.5, chi phí chỉ $15/1M tokens — so với $105 của Anthropic direct.
# Benchmark script để test thực tế
import asyncio
import aiohttp
import time
from statistics import mean, median
async def benchmark_provider(base_url: str, api_key: str, num_requests: int = 100):
"""Benchmark để so sánh latency thực tế"""
latencies = []
errors = 0
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
payload = {
"model": "claude-sonnet-4-20250514",
"max_tokens": 500,
"messages": [{"role": "user", "content": "Hello, explain AI in 50 words"}]
}
connector = aiohttp.TCPConnector(limit=20)
async with aiohttp.ClientSession(connector=connector) as session:
for i in range(num_requests):
start = time.perf_counter()
try:
async with session.post(
f"{base_url}/messages",
json=payload,
headers=headers,
timeout=aiohttp.ClientTimeout(total=30)
) as response:
latency = (time.perf_counter() - start) * 1000 # ms
if response.status == 200:
latencies.append(latency)
elif response.status == 429:
await asyncio.sleep(5) # Wait on rate limit
errors += 1
else:
errors += 1
except Exception as e:
errors += 1
if i % 20 == 0:
print(f"Progress: {i}/{num_requests}")
if latencies:
latencies.sort()
p50 = latencies[len(latencies) // 2]
p99 = latencies[int(len(latencies) * 0.99)]
return {
"mean": mean(latencies),
"median": median(latencies),
"p50": p50,
"p99": p99,
"errors": errors,
"success_rate": (num_requests - errors) / num_requests * 100
}
return {"errors": errors}
Benchmark HolySheep
result = asyncio.run(benchmark_provider(
base_url="https://api.holysheep.ai/v1",
api_key="YOUR_HOLYSHEEP_API_KEY",
num_requests=500
))
print(f"""
📊 HolySheep AI Benchmark Results:
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
Mean Latency: {result.get('mean', 0):.1f}ms
Median (P50): {result.get('median', 0):.1f}ms
P99 Latency: {result.get('p99', 0):.1f}ms
Success Rate: {result.get('success_rate', 0):.1f}%
Errors: {result.get('errors', 0)}
""")
6. Chiến Lược Tối Ưu Chi Phí
Với kinh nghiệm triển khai cho nhiều startup, tôi recommend chiến lược multi-tier:
- Tier 1 (DeepSeek V3.2 - $0.42/1M): Xử lý task đơn giản, classification, extraction
- Tier 2 (Gemini 2.5 Flash - $2.50/1M): Task trung bình, summarization, translation
- Tier 3 (Claude Sonnet 4.5 - $15/1M): Task phức tạp, coding, analysis
- Tier 4 (GPT-4.1 - $8/1M): Fallback khi Claude quá tải
HolySheep hỗ trợ tất cả model qua endpoint duy nhất — chỉ cần đổi model parameter:
# Smart Router tự động chọn model tối ưu chi phí
import anthropic
from enum import Enum
from dataclasses import dataclass
from typing import Optional
class TaskComplexity(Enum):
SIMPLE = "simple" # Classification, extraction
MEDIUM = "medium" # Summarization, translation
COMPLEX = "complex" # Coding, analysis
CRITICAL = "critical" # Production code, decisions
@dataclass
class ModelConfig:
name: str
cost_per_million: float
max_tokens: int
latency_profile: str
MODEL_CATALOG = {
"simple": ModelConfig(
name="deepseek-v3.2",
cost_per_million=0.42,
max_tokens=64000,
latency_profile="fast"
),
"medium": ModelConfig(
name="gemini-2.5-flash",
cost_per_million=2.50,
max_tokens=100000,
latency_profile="balanced"
),
"complex": ModelConfig(
name="claude-sonnet-4-20250514",
cost_per_million=15.00,
max_tokens=200000,
latency_profile="accurate"
),
"critical": ModelConfig(
name="gpt-4.1",
cost_per_million=8.00,
max_tokens=128000,
latency_profile="reliable"
)
}
class CostOptimizedRouter:
"""
Router tự động chọn model dựa trên:
1. Task complexity
2. Available quota
3. Current rate limit
4. Cost optimization goal
"""
def __init__(self, api_key: str):
self.client = anthropic.Anthropic(
base_url="https://api.holysheep.ai/v1",
api_key=api_key
)
self.usage_stats = {"total_cost": 0, "requests": 0}
def estimate_complexity(self, prompt: str) -> TaskComplexity:
"""Ước lượng độ phức tạp dựa trên keywords"""
prompt_lower = prompt.lower()
complex_keywords = ["analyze", "architect", "design", "optimize", "debug"]
critical_keywords = ["production", "security", "financial", "medical"]
if any(kw in prompt_lower for kw in critical_keywords):
return TaskComplexity.CRITICAL
if any(kw in prompt_lower for kw in complex_keywords):
return TaskComplexity.COMPLEX
if len(prompt) > 500 or "explain" in prompt_lower:
return TaskComplexity.MEDIUM
return TaskComplexity.SIMPLE
def route(self, prompt: str) -> str:
"""Chọn model tối ưu cho request"""
complexity = self.estimate_complexity(prompt)
config = MODEL_CATALOG[complexity.value]
return config.name
async def execute(
self,
prompt: str,
force_model: Optional[str] = None
) -> dict:
"""Execute request với model đã chọn"""
model = force_model or self.route(prompt)
# Tìm cost info
cost = 0
for cfg in MODEL_CATALOG.values():
if cfg.name == model:
cost = cfg.cost_per_million
break
start = time.time()
response = self.client.messages.create(
model=model,
max_tokens=2048,
messages=[{"role": "user", "content": prompt}]
)
duration = time.time() - start
output_tokens = response.usage.output_tokens
estimated_cost = (output_tokens / 1_000_000) * cost
self.usage_stats["total_cost"] += estimated_cost
self.usage_stats["requests"] += 1
return {
"response": response.content[0].text,
"model": model,
"tokens_used": output_tokens,
"estimated_cost_usd": estimated_cost,
"latency_ms": duration * 1000
}
Usage example
router = CostOptimizedRouter(api_key="YOUR_HOLYSHEEP_API_KEY")
test_prompts = [
"Classify this email as spam or not spam",
"Explain quantum computing in detail",
"Debug this Python code and suggest fixes"
]
for prompt in test_prompts:
result = await router.execute(prompt)
print(f"""
📝 Prompt: {prompt[:50]}...
🎯 Model: {result['model']}
💰 Cost: ${result['estimated_cost_usd']:.4f}
⚡ Latency: {result['latency_ms']:.0f}ms
""")
print(f"\n💵 Total spent: ${router.usage_stats['total_cost']:.2f}")
print(f"📊 Total requests: {router.usage_stats['requests']}")
Lỗi Thường Gặp Và Cách Khắc Phục
Lỗi 1: 429 Too Many Requests Liên Tục
# ❌ SAI: Retry ngay lập tức không giải quyết được vấn đề gốc
for i in range(10):
try:
response = client.messages.create(...)
except RateLimitError:
time.sleep(1) # Không hiệu quả
continue
✅ ĐÚNG: Exponential backoff với jitter và header-based retry
import random
async def robust_request_with_429_handling(client, payload, max_attempts=5):
"""
Xử lý 429 với chiến lược:
1. Đọc Retry-After header
2. Exponential backoff + random jitter
3. Circuit breaker pattern
"""
attempt = 0
last_exception = None
while attempt < max_attempts:
try:
response = client.messages.create(**payload)
return response
except RateLimitError as e:
last_exception = e
attempt += 1
# Lấy retry_after từ response headers hoặc estimate
retry_after = e.headers.get("Retry-After") if hasattr(e, 'headers') else None
if retry_after:
delay = int(retry_after)
else:
# Fallback: exponential backoff với jitter
base_delay = 2 ** attempt
jitter = random.uniform(0, 1)
delay = min(base_delay + jitter, 60) # Max 60s
print(f"⚠️ Rate limited (attempt {attempt}/{max_attempts})")
print(f" Waiting {delay:.1f}s before retry...")
await asyncio.sleep(delay)
except Exception as e:
raise # Re-raise non-rate-limit errors
raise RateLimitError(f"Failed after {max_attempts} attempts") from last_exception
Lỗi 2: Token Limit Exceeded (400 Bad Request)
# ❌ SAI: Gửi request mà không kiểm tra context window
response = client.messages.create(
model="claude-sonnet-4-20250514",
messages=[{"role": "user", "content": very_long_prompt}]
)
→ 400: Input too long
✅ ĐÚNG: Chunking và summarization pipeline
def chunk_text(text: str, max_chars: int = 100000) -> list[str]:
"""Chunk text an toàn theo token limit"""
sentences = text.split('. ')
chunks = []
current_chunk = []
current_length = 0
for sentence in sentences:
sentence_len = len(sentence) * 4 // 3 # Rough token estimate
if current_length + sentence_len > max_chars:
if current_chunk:
chunks.append('. '.join(current_chunk) + '.')
current_chunk = [sentence]
current_length = sentence_len
else:
current_chunk.append(sentence)
current_length += sentence_len
if current_chunk:
chunks.append('. '.join(current_chunk) + '.')
return chunks
async def process_long_document(client, document: str, query: str) -> str:
"""
Pipeline xử lý document dài:
1. Chunk document thành phần nhỏ hơn context window
2. Summarize từng chunk
3. Tổng hợp kết quả cuối cùng
"""
chunks = chunk_text(document, max_chars=150000) # 150K chars ≈ 100K tokens
print(f"📄 Processing {len(chunks)} chunks...")
# Summarize từng chunk
summaries = []
for i, chunk in enumerate(chunks):
print(f" Chunk {i+1}/{len(chunks)}...")
response = client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=500,
messages=[
{"role": "user", "content": f"Summarize this in 3 bullet points:\n{chunk}"}
]
)
summaries.append(response.content[0].text)
# Final synthesis
combined_summary = "\n".join(summaries)
if len(combined_summary) > 50000:
# Recursively summarize if too long
return await process_long_document(client, combined_summary, query)
final_response = client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=2048,
messages=[
{"role": "user", "content": f"Based on this summary:\n{combined_summary}\n\nAnswer: {query}"}
]
)
return final_response.content[0].text
Lỗi 3: Authentication Error 401
# ❌ SAI: Hardcode API key trong code
client = Anthropic(api_key="sk-ant-xxxxx-xxx")
✅ ĐÚNG: Environment variables với validation
import os
from pydantic import BaseModel, validator
class APIConfig(BaseModel):
api_key: str
base_url: str = "https://api.holysheep.ai/v1"
timeout: int = 60
@validator('api_key')
def validate_api_key(cls, v):
if not v or len(v) < 20:
raise ValueError("API key quá ngắn hoặc không hợp lệ")
if v.startswith("sk-ant-"): # Anthropic key - chuyển sang HolySheep
print("⚠️ Detected Anthropic API key. Vui lòng dùng HolySheep key.")
return v
@validator('base_url')
def validate_base_url(cls, v):
allowed_domains = ["api.holysheep.ai", "api.holysheep.ai/v1"]
if not any(domain in v for domain in allowed_domains):
raise ValueError(f"base_url phải thuộc HolySheep AI. Received: {v}")
return v
def load_config() -> APIConfig:
"""Load và validate config từ environment"""
api_key = os.environ.get("HOLYSHEEP_API_KEY")
if not api_key:
# Thử đọc từ file config
config_path = os.path.expanduser("~/.holysheep/config")
if os.path.exists(config_path):
with open(config_path) as f:
data = json.load(f)
api_key = data.get("api_key")
if not api_key:
raise ValueError("""
❌ Không tìm thấy HOLYSHEEP_API_KEY
Vui lòng:
1. Đăng ký tại: https://www.holysheep.ai/register
2. Lấy API key từ dashboard
3. Export: export HOLYSHEEP_API_KEY='your-key-here'
""")
return APIConfig(api_key=api_key)
Sử dụng
config = load_config()
client = Anthropic(
base_url=config.base_url,
api_key=config.api_key,
timeout=config.timeout
)
Lỗi 4: Timeout Liên Tục Trên Connection
# ❌ SAI: Timeout quá ngắn không phù hợp cho Claude
client = Anthropic(timeout=10.0) # 10s - quá ngắn cho model lớn
✅ ĐÚNG: Configurable timeout với streaming support
import httpx
class TimeoutConfig:
# Timeout theo loại operation
CONNECT_TIMEOUT = 10.0 # Kết nối ban đầu
READ_TIMEOUT = 120.0 # Đọc response (Claude cần thời gian generate)
# Tier-based timeout
TIMEOUTS = {
"claude-opus-3.5": 180.0,
"claude-sonnet-4.5": 120.0,
"claude-haiku-3.5": 60.0,
"gpt-4.1": 90.0,
"gemini-2.5-flash": 30.0,
"deepseek-v3.2": 45.0
}
@classmethod
def get_timeout(cls, model: str) -> float:
return cls.TIMEOUTS.get(model, cls.READ_TIMEOUT)
class RobustHTTPClient:
"""
HTTP client với retry logic, timeout thông minh,
và connection pooling
"""
def __init__(self, api_key: str):
self.api_key = api_key
self._client = None
@property
def client(self) -> httpx.AsyncClient:
if self._client is None:
self._client = httpx.AsyncClient(
timeout=httpx.Timeout(
connect=TimeoutConfig.CONNECT_TIMEOUT,
read=TimeoutConfig.READ_TIMEOUT,
pool=30.0 # Connection pool timeout
),
limits=httpx.Limits(
max_connections=100,
max_keepalive_connections=20
),
headers={
"Authorization": f"Bearer {self.api_key}",
"Connection": "keep-alive"
}
)
return self._client
async def close(self):
if self._client:
await self._client.aclose()
self._client = None
async def stream_generate(
self,
model: str,
prompt: str,
on_chunk: callable = None
) -> str:
"""
Streaming generate với chunk callback
Giảm perceived latency