Tối Ưu Hóa Concurrent Requests và Throughput Cho AI Model API

Tại Sao Concurrent Requests Quan Trọng?

Với mức giá HolySheep AI (tỷ giá ¥1 = $1, tiết kiệm 85%+ so với các provider khác):

• DeepSeek V3.2: $0.42/MTok - Model tiết kiệm cho các tác vụ thông thường
• Gemini 2.5 Flash: $2.50/MTok - Cân bằng giữa tốc độ và chất lượng
• GPT-4.1: $8/MTok - Model cao cấp cho tác vụ phức tạp
• Claude Sonnet 4.5: $15/MTok - Chất lượng cao nhất

Triển Khai Semaphore-Based Rate Limiting

import asyncio
import httpx
from dataclasses import dataclass
from typing import Optional
import time

@dataclass
class RateLimitConfig:
    max_concurrent: int = 10
    requests_per_minute: int = 60
    retry_attempts: int = 3
    retry_delay: float = 1.0

class HolySheepAIClient:
    def __init__(
        self,
        api_key: str,
        base_url: str = "https://api.holysheep.ai/v1",
        config: Optional[RateLimitConfig] = None
    ):
        self.api_key = api_key
        self.base_url = base_url
        self.config = config or RateLimitConfig()
        
        # Semaphore để kiểm soát số lượng request đồng thời
        self._semaphore = asyncio.Semaphore(self.config.max_concurrent)
        
        # Rate limiter thời gian
        self._request_times: list = []
        self._lock = asyncio.Lock()
        
        self._client = httpx.AsyncClient(
            timeout=httpx.Timeout(60.0, connect=10.0),
            limits=httpx.Limits(max_connections=100, max_keepalive_connections=20)
        )
    
    async def _check_rate_limit(self):
        """Kiểm tra và áp dụng rate limit theo thời gian"""
        async with self._lock:
            now = time.time()
            # Xóa các request cũ hơn 60 giây
            self._request_times = [t for t in self._request_times if now - t < 60]
            
            if len(self._request_times) >= self.config.requests_per_minute:
                # Tính thời gian chờ
                wait_time = 60 - (now - self._request_times[0]) + 0.5
                if wait_time > 0:
                    await asyncio.sleep(wait_time)
                    # Cập nhật lại sau khi sleep
                    self._request_times = self._request_times[1:]
            
            self._request_times.append(time.time())
    
    async def chat_completion(
        self,
        messages: list,
        model: str = "deepseek-chat",
        temperature: float = 0.7,
        max_tokens: int = 2048
    ) -> dict:
        """Gửi request với semaphore và retry logic"""
        
        async with self._semaphore:
            await self._check_rate_limit()
            
            for attempt in range(self.config.retry_attempts):
                try:
                    response = await self._client.post(
                        f"{self.base_url}/chat/completions",
                        headers={
                            "Authorization": f"Bearer {self.api_key}",
                            "Content-Type": "application/json"
                        },
                        json={
                            "model": model,
                            "messages": messages,
                            "temperature": temperature,
                            "max_tokens": max_tokens
                        }
                    )
                    
                    if response.status_code == 200:
                        return response.json()
                    elif response.status_code == 429:
                        # Rate limit - chờ và thử lại
                        retry_after = int(response.headers.get("Retry-After", 5))
                        await asyncio.sleep(retry_after)
                        continue
                    elif response.status_code == 401:
                        raise Exception("API key không hợp lệ - kiểm tra HolySheep dashboard")
                    else:
                        response.raise_for_status()
                        
                except (httpx.TimeoutException, httpx.ConnectError) as e:
                    if attempt < self.config.retry_attempts - 1:
                        wait = self.config.retry_delay * (2 ** attempt)
                        print(f"Retry {attempt + 1} sau {wait}s - Error: {e}")
                        await asyncio.sleep(wait)
                    else:
                        raise
        
        raise Exception("Đã vượt quá số lần retry")

Sử dụng
async def main():
    client = HolySheepAIClient(
        api_key="YOUR_HOLYSHEEP_API_KEY",
        config=RateLimitConfig(
            max_concurrent=20,
            requests_per_minute=300
        )
    )
    
    tasks = [
        client.chat_completion(
            messages=[{"role": "user", "content": f"Tính toán batch {i}"}],
            model="deepseek-chat"
        )
        for i in range(100)
    ]
    
    start = time.time()
    results = await asyncio.gather(*tasks, return_exceptions=True)
    elapsed = time.time() - start
    
    success = sum(1 for r in results if isinstance(r, dict))
    print(f"Hoàn thành {success}/100 requests trong {elapsed:.2f}s")
    print(f"Throughput: {success/elapsed:.2f} requests/giây")

asyncio.run(main())

Batch Processing Với Exponential Backoff

import asyncio
import httpx
from typing import List, Dict, Any
from collections import deque
import time

class BatchProcessor:
    def __init__(
        self,
        api_key: str,
        batch_size: int = 50,
        max_concurrent_batches: int = 3,
        base_delay: float = 1.0,
        max_delay: float = 60.0
    ):
        self.api_key = api_key
        self.base_url = "https://api.holysheep.ai/v1"
        self.batch_size = batch_size
        self.max_concurrent_batches = max_concurrent_batches
        self.base_delay = base_delay
        self.max_delay = max_delay
        
        self._batch_semaphore = asyncio.Semaphore(max_concurrent_batches)
        self._request_queue: deque = deque()
        self._processing = False
    
    async def _send_batch(
        self,
        requests: List[Dict[str, Any]],
        delay: float
    ) -> List[Dict]:
        """Gửi một batch requests với exponential backoff"""
        
        await asyncio.sleep(delay)
        
        async with self._batch_semaphore:
            tasks = []
            for req in requests:
                task = self._send_single_request(req)
                tasks.append(task)
            
            results = await asyncio.gather(*tasks, return_exceptions=True)
            return results
    
    async def _send_single_request(self, request: Dict) -> Dict:
        """Gửi một request đơn lẻ"""
        
        async with httpx.AsyncClient(timeout=30.0) as client:
            try:
                response = await client.post(
                    f"{self.base_url}/chat/completions",
                    headers={
                        "Authorization": f"Bearer {self.api_key}",
                        "Content-Type": "application/json"
                    },
                    json=request
                )
                
                if response.status_code == 200:
                    return {"success": True, "data": response.json()}
                elif response.status_code == 429:
                    return {"success": False, "error": "rate_limited", "request": request}
                else:
                    return {"success": False, "error": f"http_{response.status_code}"}
                    
            except Exception as e:
                return {"success": False, "error": str(e), "request": request}
    
    async def process_all(
        self,
        requests: List[Dict[str, Any]],
        progress_callback=None
    ) -> Dict[str, Any]:
        """Xử lý tất cả requests với batch processing thông minh"""
        
        total = len(requests)
        completed = 0
        failed = []
        results = []
        
        # Chia thành các batch
        batches = [
            requests[i:i + self.batch_size]
            for i in range(0, total, self.batch_size)
        ]
        
        delay = self.base_delay
        
        for idx, batch in enumerate(batches):
            batch_results = await self._send_batch(batch, delay)
            
            for result in batch_results:
                if isinstance(result, dict):
                    if result.get("success"):
                        results.append(result.get("data"))
                        completed += 1
                    else:
                        error_type = result.get("error")
                        if error_type == "rate_limited":
                            # Tăng delay khi bị rate limit
                            delay = min(delay * 1.5, self.max_delay)
                            failed.append(result.get("request"))
                        else:
                            failed.append(result.get("request"))
                
                if progress_callback:
                    progress_callback(completed, total)
            
            # Reset delay sau mỗi batch thành công
            if all(r.get("success") for r in batch_results if isinstance(r, dict)):
                delay = self.base_delay
            
            print(f"Batch {idx + 1}/{len(batches)}: {len(batch_results)} requests")
        
        return {
            "total": total,
            "completed": completed,
            "failed": len(failed),
            "success_rate": completed / total * 100,
            "results": results,
            "failed_requests": failed
        }

Sử dụng với ví dụ thực tế
async def process_documents():
    processor = BatchProcessor(
        api_key="YOUR_HOLYSHEEP_API_KEY",
        batch_size=20,
        max_concurrent_batches=2,
        base_delay=0.5
    )
    
    # Tạo 500 requests mẫu
    requests = [
        {
            "model": "deepseek-chat",
            "messages": [{"role": "user", "content": f"Tóm tắt tài liệu #{i}"}],
            "max_tokens": 500
        }
        for i in range(500)
    ]
    
    def progress(current, total):
        if current % 50 == 0:
            print(f"Tiến độ: {current}/{total} ({current/total*100:.1f}%)")
    
    start_time = time.time()
    result = await processor.process_all(requests, progress_callback=progress)
    elapsed = time.time() - start_time
    
    print(f"\n=== KẾT QUẢ ===")
    print(f"Tổng requests: {result['total']}")
    print(f"Thành công: {result['completed']}")
    print(f"Thất bại: {result['failed']}")
    print(f"Tỷ lệ thành công: {result['success_rate']:.2f}%")
    print(f"Thời gian xử lý: {elapsed:.2f}s")
    print(f"Throughput trung bình: {result['completed']/elapsed:.2f} req/s")

asyncio.run(process_documents())

Monitoring và Metrics Dashboard

import asyncio
import time
from dataclasses import dataclass, field
from typing import Dict, List
from collections import defaultdict
import statistics

@dataclass
class RequestMetrics:
    latency_ms: List[float] = field(default_factory=list)
    errors: List[Dict] = field(default_factory=list)
    status_codes: Dict[int, int] = field(default_factory=lambda: defaultdict(int))
    
    def record_request(self, latency: float, status_code: int, error: str = None):
        self.latency_ms.append(latency)
        self.status_codes[status_code] += 1
        if error:
            self.errors.append({"time": time.time(), "error": error})
    
    def get_stats(self) -> Dict:
        if not self.latency_ms:
            return {}
        
        return {
            "total_requests": len(self.latency_ms),
            "avg_latency_ms": statistics.mean(self.latency_ms),
            "p50_latency_ms": statistics.median(self.latency_ms),
            "p95_latency_ms": sorted(self.latency_ms)[int(len(self.latency_ms) * 0.95)],
            "p99_latency_ms": sorted(self.latency_ms)[int(len(self.latency_ms) * 0.99)],
            "min_latency_ms": min(self.latency_ms),
            "max_latency_ms": max(self.latency_ms),
            "error_rate": len(self.errors) / len(self.latency_ms) * 100,
            "status_breakdown": dict(self.status_codes)
        }

class PerformanceMonitor:
    def __init__(self, window_seconds: int = 60):
        self.window_seconds = window_seconds
        self.metrics: Dict[str, RequestMetrics] = defaultdict(RequestMetrics)
        self._cleanup_task: asyncio.Task = None
    
    async def start(self):
        """Bắt đầu cleanup task định kỳ"""
        self._cleanup_task = asyncio.create_task(self._cleanup_loop())
    
    async def stop(self):
        """Dừng monitoring"""
        if self._cleanup_task:
            self._cleanup_task.cancel()
    
    async def _cleanup_loop(self):
        """Xóa metrics cũ định kỳ"""
        while True:
            await asyncio.sleep(self.window_seconds)
            self._cleanup_old_metrics()
    
    def _cleanup_old_metrics(self):
        """Xóa metrics cũ hơn window"""
        cutoff = time.time() - self.window_seconds
        # Implementation phụ thuộc vào storage
    
    def record(self, model: str, latency: float, status_code: int, error: str = None):
        """Ghi nhận một request"""
        self.metrics[model].record_request(latency, status_code, error)
    
    async def record_async(self, model: str, request_func, *args, **kwargs):
        """Ghi nhận request async với timing"""
        start = time.time()
        error = None
        status_code = 200
        
        try:
            result = await request_func(*args, **kwargs)
            return result
        except Exception as e:
            error = str(e)
            status_code = 500
            raise
        finally:
            latency = (time.time() - start) * 1000
            self.record(model, latency, status_code, error)
    
    def get_dashboard(self) -> Dict:
        """Lấy dashboard data"""
        dashboard = {}
        for model, metrics in self.metrics.items():
            stats = metrics.get_stats()
            if stats:
                dashboard[model] = stats
        
        return {
            "timestamp": time.time(),
            "models": dashboard,
            "total_throughput": sum(
                m.get_stats().get("total_requests", 0) 
                for m in self.metrics.values()
            )
        }

Sử dụng monitoring
async def monitored_example():
    monitor = PerformanceMonitor(window_seconds=300)
    await monitor.start()
    
    try:
        # Monitor các model khác nhau
        models = ["deepseek-chat", "gpt-4", "claude-sonnet"]
        
        for model in models:
            for i in range(10):
                start = time.time()
                
                # Gọi API (thay bằng client thực tế)
                await asyncio.sleep(0.1)  # Simulate API call
                
                latency = (time.time() - start) * 1000
                monitor.record(model, latency, 200)
        
        # Lấy dashboard
        dashboard = monitor.get_dashboard()
        
        print("=== PERFORMANCE DASHBOARD ===")
        for model, stats in dashboard["models"].items():
            print(f"\n{model.upper()}:")
            print(f"  Total Requests: {stats['total_requests']}")
            print(f"  Avg Latency: {stats['avg_latency_ms']:.2f}ms")
            print(f"  P95 Latency: {stats['p95_latency_ms']:.2f}ms")
            print(f"  P99 Latency: {stats['p99_latency_ms']:.2f}ms")
            print(f"  Error Rate: {stats['error_rate']:.2f}%")
            print(f"  Status Codes: {stats['status_breakdown']}")
    
    finally:
        await monitor.stop()

asyncio.run(monitored_example())

Lỗi Thường Gặp và Cách Khắc Phục

1. Lỗi Connection Timeout

# ❌ SAI: Timeout quá ngắn cho các model nặng
client = httpx.Client(timeout=5.0)

✅ ĐÚNG: Timeout linh hoạt theo loại request
client = httpx.AsyncClient(
    timeout=httpx.Timeout(
        connect=10.0,      # Connection timeout
        read=120.0,        # Read timeout cho model lớn (GPT-4.1, Claude Sonnet)
        write=30.0,       # Write timeout
        pool=10.0         # Pool timeout
    )
)

2. Lỗi Rate Limit 429

# ❌ SAI: Retry ngay lập tức không có backoff
for attempt in range(10):
    try:
        response = await client.post(url, json=data)
    except RateLimitError:
        await asyncio.sleep(0.1)  # Quá nhanh!

✅ ĐÚNG: Exponential backoff với jitter
import random

async def smart_retry(request_func, max_attempts=5):
    delay = 1.0
    
    for attempt in range(max_attempts):
        try:
            return await request_func()
        except RateLimitError as e:
            if attempt == max_attempts - 1:
                raise
            
            # Exponential backoff với random jitter (±25%)
            jitter = delay * 0.25 * (2 * random.random() - 1)
            wait_time = delay + jitter
            
            print(f"Rate limited - chờ {wait_time:.2f}s trước retry {attempt + 1}")
            await asyncio.sleep(wait_time)
            
            delay *= 2  # Tăng gấp đôi mỗi lần

3. Lỗi Authentication 401

# ❌ SAI: Hardcode API key trong code
API_KEY = "sk-xxxxx"  # Nguy hiểm!

✅ ĐÚNG: Sử dụng environment variables
import os
from dotenv import load_dotenv

load_dotenv()  # Load từ .env file

API_KEY = os.environ.get("HOLYSHEEP_API_KEY")
if not API_KEY:
    raise ValueError("HOLYSHEEP_API_KEY không được set!")

Hoặc sử dụng secret manager (AWS Secrets, HashiCorp Vault, etc.)
from keyring import get_password
API_KEY = get_password("holysheep", "api_key")

4. Memory Leak Khi Xử Lý Batch Lớn

# ❌ SAI: Giữ tất cả results trong memory
all_results = []
for batch in large_batches:
    results = await process_batch(batch)
    all_results.extend(results)  # Memory leak khi batch lớn!

✅ ĐÚNG: Stream results ra file/database
async def process_and_stream(client, batches, output_file):
    with open(output_file, 'a') as f:
        for batch in batches:
            results = await client.process_batch(batch)
            
            for result in results:
                f.write(json.dumps(result) + '\n')
            
            # Flush sau mỗi batch để giải phóng memory
            f.flush()
            
            # Force garbage collection định kỳ
            gc.collect()

Hoặc sử dụng async generator
async def generate_results(batches):
    for batch in batches:
        results = await process_batch(batch)
        for result in results:
            yield result

Bảng So Sánh Chi Phí và Hiệu Suất

Với HolySheep AI, tiết kiệm có thể lên đến 85%+ khi sử dụng DeepSeek V3.2 ($0.42/MTok) cho các tác vụ thông thường, đồng thời tận dụng độ trễ thấp dưới 50ms để xử lý concurrent requests hiệu quả hơn.

Kết Luận

• Sử dụng Semaphore để kiểm soát số lượng request đồng thời
• Triển khai Exponential Backoff để xử lý rate limit thông minh
• Batch Processing với Queue Management hiệu quả
• Monitoring với Metrics Collection real-time