AI API 请求限速与队列设计：令牌桶算法 Python 实现

Khi làm việc với AI API, một trong những thách thức lớn nhất mà developer gặp phải là quản lý rate limiting (giới hạn tốc độ yêu cầu). Bài viết này sẽ hướng dẫn bạn thiết kế hệ thống queue thông minh sử dụng 令牌桶算法 (Token Bucket Algorithm) với Python, đồng thời so sánh các giải pháp API relay phổ biến.

Bảng so sánh: HolySheep vs API chính thức vs Dịch vụ Relay khác

Như bạn thấy, HolySheep AI cung cấp mức giá cạnh tranh với chi phí thấp hơn đáng kể so với API chính thức, đặc biệt khi tỷ giá chỉ còn ¥1=$1. Đăng ký tại đây để nhận tín dụng miễn phí khi bắt đầu.

令牌桶算法 (Token Bucket Algorithm) là gì?

令牌桶算法 là một thuật toán kiểm soát tốc độ (rate limiting) phổ biến, hoạt động theo nguyên lý:

• Hệ thống có một "thùng chứa" chứa các "thẻ bài" (token)
• Mỗi yêu cầu API cần lấy một token từ thùng
• Nếu thùng trống, yêu cầu phải đợi hoặc bị từ chối
• Token được thêm vào thùng với tốc độ cố định (refill rate)
• Thùng có容量 tối đa (bucket capacity)

Triển khai Token Bucket với Python

Cài đặt thư viện cần thiết

pip install httpx aiofiles asyncio-limiter

Triển khai Token Bucket cơ bản

import time
import asyncio
from threading import Lock
from typing import Optional
from dataclasses import dataclass

@dataclass
class TokenBucket:
    """
    Token Bucket Algorithm Implementation
    - capacity: Số token tối đa trong thùng
    - refill_rate: Số token được thêm mỗi giây
    - tokens: Số token hiện tại
    - last_refill: Thời điểm refill cuối cùng
    """
    capacity: float
    refill_rate: float
    tokens: float
    last_refill: float
    
    @classmethod
    def create(cls, capacity: float, requests_per_second: float):
        """Tạo token bucket mới"""
        return cls(
            capacity=capacity,
            refill_rate=requests_per_second,
            tokens=capacity,
            last_refill=time.time()
        )
    
    def _refill(self):
        """Tự động refill token dựa trên thời gian trôi qua"""
        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 acquire(self, tokens: float = 1.0, blocking: bool = True, timeout: Optional[float] = None) -> bool:
        """
        Lấy token từ bucket
        
        Args:
            tokens: Số token cần lấy
            blocking: True = đợi nếu không đủ token, False = trả về ngay
            timeout: Thời gian chờ tối đa (giây)
        
        Returns:
            True nếu lấy được token, False nếu không
        """
        start_time = time.time()
        
        while True:
            with Lock():
                self._refill()
                
                if self.tokens >= tokens:
                    self.tokens -= tokens
                    return True
            
            if not blocking:
                return False
            
            if timeout is not None and (time.time() - start_time) >= timeout:
                return False
            
            # Đợi một chút trước khi thử lại
            wait_time = tokens / self.refill_rate
            time.sleep(min(wait_time, 0.1))
    
    def available_tokens(self) -> float:
        """Kiểm tra số token hiện có"""
        with Lock():
            self._refill()
            return self.tokens

Ví dụ sử dụng
bucket = TokenBucket.create(capacity=100, requests_per_second=10)
print(f"Token hiện có: {bucket.available_tokens():.2f}")

Triển khai Async API Client với Rate Limiting

import httpx
import asyncio
from typing import Dict, Any, Optional, List
import time

class HolySheepAIClient:
    """
    HolySheep AI API Client với Token Bucket Rate Limiting
    
    Ưu điểm:
    - Tự động quản lý rate limit
    - Retry thông minh khi gặp lỗi 429
    - Hỗ trợ async/await
    - Queue management thông minh
    """
    
    def __init__(
        self,
        api_key: str,
        base_url: str = "https://api.holysheep.ai/v1",
        requests_per_second: float = 10.0,
        max_tokens_per_request: int = 100,
        max_retries: int = 3
    ):
        self.api_key = api_key
        self.base_url = base_url
        self.max_retries = max_retries
        
        # Token Bucket cho rate limiting
        self.rate_limiter = TokenBucket.create(
            capacity=max_tokens_per_request,
            requests_per_second=requests_per_second
        )
        
        # HTTP Client
        self.client = httpx.AsyncClient(
            timeout=httpx.Timeout(60.0, connect=10.0),
            headers={
                "Authorization": f"Bearer {api_key}",
                "Content-Type": "application/json"
            }
        )
        
        # Queue để quản lý các yêu cầu
        self._request_queue: asyncio.Queue = asyncio.Queue()
        self._is_processing = False
    
    async def chat_completions(
        self,
        model: str = "gpt-4.1",
        messages: List[Dict[str, str]],
        temperature: float = 0.7,
        max_tokens: Optional[int] = None,
        **kwargs
    ) -> Dict[str, Any]:
        """
        Gọi API chat completions với rate limiting
        
        Args:
            model: Model sử dụng (gpt-4.1, claude-sonnet-4.5, gemini-2.5-flash, deepseek-v3.2)
            messages: Danh sách messages theo format OpenAI
            temperature: Độ ngẫu nhiên (0-2)
            max_tokens: Số token tối đa trong response
        
        Returns:
            Response từ API
        """
        payload = {
            "model": model,
            "messages": messages,
            "temperature": temperature,
        }
        
        if max_tokens:
            payload["max_tokens"] = max_tokens
        
        payload.update(kwargs)
        
        # Thử gọi API với retry logic
        for attempt in range(self.max_retries):
            try:
                # Chờ đến khi có token available
                await self._wait_for_token()
                
                # Gọi API
                response = await self.client.post(
                    f"{self.base_url}/chat/completions",
                    json=payload
                )
                
                # Xử lý response
                if response.status_code == 200:
                    return response.json()
                
                elif response.status_code == 429:
                    # Rate limited - đợi và thử lại
                    retry_after = int(response.headers.get("retry-after", 60))
                    print(f"Rate limited! Đợi {retry_after}s trước khi thử lại...")
                    await asyncio.sleep(retry_after)
                    continue
                
                elif response.status_code == 401:
                    raise ValueError("API Key không hợp lệ. Vui lòng kiểm tra YOUR_HOLYSHEEP_API_KEY")
                
                else:
                    response.raise_for_status()
            
            except httpx.HTTPError as e:
                if attempt == self.max_retries - 1:
                    raise
                wait_time = 2 ** attempt
                print(f"Lỗi: {e}. Thử lại sau {wait_time}s...")
                await asyncio.sleep(wait_time)
        
        raise Exception("Đã hết số lần thử lại")
    
    async def _wait_for_token(self):
        """Đợi cho đến khi có token available"""
        while True:
            tokens = self.rate_limiter.available_tokens()
            if tokens >= 1.0:
                self.rate_limiter.acquire(1.0, blocking=False)
                break
            # Đợi một chút trước khi kiểm tra lại
            await asyncio.sleep(0.1)
    
    async def batch_process(self, prompts: List[str], model: str = "gpt-4.1") -> List[Dict[str, Any]]:
        """
        Xử lý nhiều prompts cùng lúc với queue management
        
        Args:
            prompts: Danh sách prompts cần xử lý
            model: Model sử dụng
        
        Returns:
            Danh sách responses
        """
        messages_list = [{"role": "user", "content": prompt} for prompt in prompts]
        
        tasks = [
            self.chat_completions(model=model, messages=messages)
            for message in messages_list
        ]
        
        # Xử lý với concurrency limit
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        return results
    
    async def close(self):
        """Đóng HTTP client"""
        await self.client.aclose()

============== VÍ DỤ SỬ DỤNG ==============

async def main():
    # Khởi tạo client với API key của bạn
    client = HolySheepAIClient(
        api_key="YOUR_HOLYSHEEP_API_KEY",
        requests_per_second=5,  # 5 requests/giây
        max_tokens_per_request=50
    )
    
    try:
        # Gọi một request đơn
        response = await client.chat_completions(
            model="gpt-4.1",
            messages=[
                {"role": "system", "content": "Bạn là trợ lý AI hữu ích"},
                {"role": "user", "content": "Giải thích Token Bucket Algorithm"}
            ],
            max_tokens=500
        )
        print(f"Response: {response['choices'][0]['message']['content']}")
        
        # Xử lý batch với nhiều prompts
        prompts = [
            "Token Bucket là gì?",
            "Rate Limiting có vai trò gì?",
            "Tại sao cần quản lý API requests?"
        ]
        
        batch_results = await client.batch_process(prompts, model="deepseek-v3.2")
        
        for i, result in enumerate(batch_results):
            if isinstance(result, dict):
                print(f"Prompt {i+1}: {result['choices'][0]['message']['content'][:100]}...")
            else:
                print(f"Prompt {i+1} lỗi: {result}")
    
    finally:
        await client.close()

if __name__ == "__main__":
    asyncio.run(main())

Queue Design cho High-Traffic Applications

Đối với các ứng dụng có lưu lượng lớn, bạn cần một hệ thống queue phức tạp hơn để quản lý:

import asyncio
import heapq
from typing import Callable, Any, Dict, List
from dataclasses import dataclass, field
from enum import Enum
import uuid
from datetime import datetime

class Priority(Enum):
    HIGH = 1
    NORMAL = 2
    LOW = 3

@dataclass(order=True)
class QueuedRequest:
    """Request được đưa vào queue"""
    priority: int
    created_at: float = field(compare=False)
    request_id: str = field(compare=False)
    payload: Dict[str, Any] = field(compare=False)
    callback: Callable = field(compare=False, default=None)
    max_retries: int = field(compare=False, default=3)
    current_retry: int = field(compare=False, default=0)

class SmartRequestQueue:
    """
    Hệ thống Queue thông minh với:
    - Priority queuing (ưu tiên request quan trọng)
    - Batch processing (gom nhóm request để xử lý hiệu quả)
    - Automatic retry (tự động thử lại khi thất bại)
    - Dead letter queue (lưu request thất bại để xem xét sau)
    """
    
    def __init__(
        self,
        rate_limiter: Token
Tài nguyên liên quan
📚 Hướng dẫn AI API
💰 Xem giá
📖 Tài liệu nhà phát triển
🚀 Đăng ký miễn phí
Bài viết liên quan
vi graphrag vs chuantong ragzhishitupuzengqiangjiansu 2026 0
vi structured output json modeai api jiegouhuashuchuw 2026 0