TL;DR: Bài viết này là playbook di chuyển hoàn chỉnh giúp đội ngũ kỹ thuật chuyển từ API chính hãng hoặc relay khác sang HolySheep AI, biến các lỗi model, trạng thái retry và chiến lược bồi thường thành thông báo có thể giải thích cho khách hàng. Bao gồm code mẫu production-ready, template giao tiếp, và phân tích ROI chi tiết với con số tiết kiệm cụ thể.

Vấn Đề Thực Tế: Khi API Trả Lỗi, Khách Hàng Chỉ Thấy "Something Went Wrong"

Là developer, tôi đã trải qua cảnh khách hàng gửi email với nội dung: "Tôi bị lỗi khi sử dụng AI chatbot. Trang web hiển thị 'Internal Server Error'. Đề nghị hoàn tiền." Thay vì một thông báo rõ ràng như "Model đang quá tải, chúng tôi đã tự động chuyển sang server dự phòng và gửi bạn 500 tín dụng miễn phí", họ nhận được HTTP 500 và một email hủy订阅.

Bài học xương máu: 73% người dùng rời bỏ sau một trải nghiệm lỗi kém — theo nghiên cứu của Zendesk năm 2025. Vấn đề không nằm ở API gốc mà ở cách chúng ta diễn giải lỗi và truyền tải đến end-user.

Tại Sao HolySheep Là Giải Pháp Thay Thế Tối Ưu

Sau 6 tháng test và đánh giá, đội ngũ của tôi đã chuyển hoàn toàn từ API chính hãng sang HolySheep AI vì những lý do cụ thể:

Playbook Di Chuyển: Từ Relay Khác Sang HolySheep Trong 48 Giờ

Bước 1: Audit Code Hiện Tại — Xác Định Tất Cả Điểm Gọi API

# 1. Tìm tất cả file chứa API calls

Chạy command sau để scan:

find . -type f -name "*.py" -o -name "*.js" -o -name "*.ts" | xargs grep -l "api.openai.com\|api.anthropic.com\|RELAY_URL" > api_calls_audit.txt


2. Đếm số lượng API calls

wc -l api_calls_audit.txt


3. Tạo mapping file cũ → API mới

cat > api_mapping.json << 'EOF'
{
  "api.openai.com/v1/chat/completions": "https://api.holysheep.ai/v1/chat/completions",
  "api.anthropic.com/v1/messages": "https://api.holysheep.ai/v1/messages",
  "RELAY_CUSTOM_URL": "https://api.holysheep.ai/v1"
}
EOF

Bước 2: Cập Nhật Base Configuration

# File: config/api_config.py
import os
from dataclasses import dataclass
from typing import Optional

@dataclass
class HolySheepConfig:
    # ✅ Endpoint chính thức của HolySheep
    base_url: str = "https://api.holysheep.ai/v1"
    
    # ✅ API Key từ HolySheep Dashboard
    api_key: str = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
    
    # Cấu hình retry strategy
    max_retries: int = 3
    retry_delay: float = 1.0  # seconds
    timeout: int = 60  # seconds
    
    # Model mapping (tương thích với code cũ)
    model_map: dict = None
    
    def __post_init__(self):
        self.model_map = {
            "gpt-4": "gpt-4-turbo",
            "gpt-4o": "gpt-4o",
            "claude-3-5-sonnet": "claude-sonnet-4-20250514",
            "gpt-3.5-turbo": "gpt-3.5-turbo"
        }


Sử dụng singleton pattern

config = HolySheepConfig()

Bước 3: Triển Khai Error Handler Wrapper

# File: services/holysheep_client.py
import requests
import time
import json
from typing import Dict, Any, Optional
from dataclasses import dataclass, field
from enum import Enum
from datetime import datetime

class ErrorSeverity(Enum):
    LOW = "low"           # Retry thành công, không cần thông báo
    MEDIUM = "medium"     # Đã retry, cần log nhưng không ảnh hưởng user
    HIGH = "high"         # Cần fallback, thông báo nhẹ cho user
    CRITICAL = "critical" # Hệ thống fail, cần escalation

@dataclass
class APIError:
    code: str
    message: str
    severity: ErrorSeverity
    retry_after: Optional[int] = None
    model: Optional[str] = None
    timestamp: datetime = field(default_factory=datetime.utcnow)
    metadata: Dict[str, Any] = field(default_factory=dict)
    
    def to_customer_message(self, lang: str = "vi") -> str:
        """Chuyển đổi error thành message thân thiện cho customer"""
        messages = {
            "vi": {
                "rate_limit": "🔄 Server đang bận, chúng tôi đang thử lại tự động. Bạn vẫn được ưu tiên xử lý.",
                "model_unavailable": "🤖 Model hiện tại đang bảo trì, hệ thống tự động chuyển sang model thay thế.",
                "timeout": "⏱️ Yêu cầu mất nhiều thời gian hơn dự kiến. Chúng tôi đang xử lý, vui lòng đợi.",
                "auth_error": "🔐 Phiên làm việc đã hết hạn. Vui lòng đăng nhập lại.",
                "server_error": "⚠️ Hệ thống AI tạm thời quá tải. Đội ngũ kỹ thuật đã được thông báo.",
                "invalid_request": "📝 Dữ liệu đầu vào không hợp lệ. Vui lòng kiểm tra lại."
            },
            "en": {
                "rate_limit": "🔄 Server is busy, we're retrying automatically.",
                "model_unavailable": "🤖 Current model is under maintenance, switching to alternative.",
                "timeout": "⏱️ Request is taking longer than expected. Please wait.",
                "auth_error": "🔐 Session expired. Please log in again.",
                "server_error": "⚠️ AI system temporarily overloaded.",
                "invalid_request": "📝 Invalid input data. Please check and try again."
            }
        }
        return messages.get(lang, messages["vi"]).get(self.code, self.message)

class HolySheepClient:
    """Production-ready client với comprehensive error handling"""
    
    def __init__(self, config):
        self.base_url = config.base_url
        self.api_key = config.api_key
        self.max_retries = config.max_retries
        self.retry_delay = config.retry_delay
        self.timeout = config.timeout
        self.model_map = config.model_map
        self.session = requests.Session()
        
        # Headers mặc định
        self.session.headers.update({
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        })
        
        # Event callbacks cho monitoring
        self.error_callbacks = []
        self.compensation_callbacks = []
    
    def register_error_callback(self, callback):
        """Đăng ký callback khi có error"""
        self.error_callbacks.append(callback)
    
    def register_compensation_callback(self, callback):
        """Đăng ký callback để apply compensation (credits, refund)"""
        self.compensation_callbacks.append(callback)
    
    def _map_error_to_customer_facing(self, error_response: Dict) -> APIError:
        """Parse error response từ HolySheep API thành structured error"""
        error_mapping = {
            429: ("rate_limit", ErrorSeverity.MEDIUM, 60),
            500: ("server_error", ErrorSeverity.HIGH, None),
            503: ("model_unavailable", ErrorSeverity.HIGH, 30),
            408: ("timeout", ErrorSeverity.LOW, 5),
            401: ("auth_error", ErrorSeverity.CRITICAL, None),
            400: ("invalid_request", ErrorSeverity.MEDIUM, None),
        }
        
        status_code = error_response.get("status_code", 500)
        error_info = error_mapping.get(status_code, ("server_error", ErrorSeverity.HIGH, None))
        
        return APIError(
            code=error_info[0],
            message=error_response.get("error", {}).get("message", "Unknown error"),
            severity=error_info[1],
            retry_after=error_info[2],
            model=error_response.get("model"),
            metadata={
                "raw_error": error_response,
                "status_code": status_code,
                "response_time_ms": error_response.get("response_time_ms", 0)
            }
        )
    
    def chat_completions(self, messages: list, model: str = "gpt-4o", 
                         temperature: float = 0.7, **kwargs) -> Dict[str, Any]:
        """
        Wrapper cho /chat/completions endpoint với error handling hoàn chỉnh
        
        Args:
            messages: List of message objects
            model: Model name (sẽ được map nếu cần)
            temperature: Sampling temperature
            **kwargs: Các parameters khác (max_tokens, etc.)
        
        Returns:
            Dict chứa response và metadata
        """
        # Map model name nếu cần
        mapped_model = self.model_map.get(model, model)
        
        payload = {
            "model": mapped_model,
            "messages": messages,
            "temperature": temperature,
            **kwargs
        }
        
        start_time = time.time()
        last_error = None
        
        for attempt in range(self.max_retries):
            try:
                response = self.session.post(
                    f"{self.base_url}/chat/completions",
                    json=payload,
                    timeout=self.timeout
                )
                
                if response.status_code == 200:
                    result = response.json()
                    return {
                        "success": True,
                        "data": result,
                        "latency_ms": int((time.time() - start_time) * 1000),
                        "model": mapped_model,
                        "attempt": attempt + 1
                    }
                
                # Parse error
                error_response = {
                    "status_code": response.status_code,
                    "error": response.json() if response.text else {},
                    "response_time_ms": int((time.time() - start_time) * 1000)
                }
                
                api_error = self._map_error_to_customer_facing(error_response)
                
                # Retry logic
                if api_error.severity in [ErrorSeverity.LOW, ErrorSeverity.MEDIUM]:
                    if api_error.retry_after:
                        time.sleep(min(api_error.retry_after, self.retry_delay * (attempt + 1)))
                        continue
                
                # Trigger error callbacks
                for callback in self.error_callbacks:
                    callback(api_error, attempt, payload)
                
                # Apply compensation cho HIGH/CRITICAL errors
                if api_error.severity in [ErrorSeverity.HIGH, ErrorSeverity.CRITICAL]:
                    self._apply_compensation(api_error, payload)
                
                last_error = api_error
                break
                
            except requests.exceptions.Timeout:
                last_error = APIError(
                    code="timeout",
                    message="Request timed out",
                    severity=ErrorSeverity.LOW,
                    retry_after=5
                )
                time.sleep(self.retry_delay)
                
            except requests.exceptions.ConnectionError:
                last_error = APIError(
                    code="connection_error",
                    message="Cannot connect to HolySheep API",
                    severity=ErrorSeverity.HIGH,
                    retry_after=10
                )
                time.sleep(self.retry_delay * 2)
                
            except Exception as e:
                last_error = APIError(
                    code="unexpected_error",
                    message=str(e),
                    severity=ErrorSeverity.CRITICAL
                )
                break
        
        # Return error response sau khi exhausted retries
        return {
            "success": False,
            "error": last_error,
            "customer_message": last_error.to_customer_message("vi"),
            "latency_ms": int((time.time() - start_time) * 1000),
            "total_attempts": self.max_retries
        }
    
    def _apply_compensation(self, error: APIError, original_request: Dict):
        """Apply compensation (credits, discounts) dựa trên error type"""
        compensation_rules = {
            "rate_limit": {"credits": 100, "discount_code": "SORRY10"},
            "server_error": {"credits": 500, "discount_code": "TECHISSUE"},
            "model_unavailable": {"credits": 200, "discount_code": "MODELSWITCH"},
            "timeout": {"credits": 50, "discount_code": None},
        }
        
        compensation = compensation_rules.get(error.code, {"credits": 100, "discount_code": None})
        
        for callback in self.compensation_callbacks:
            callback(
                user_id=original_request.get("user_id"),
                error_type=error.code,
                compensation=compensation
            )


Khởi tạo global client

from config.api_config import config
client = HolySheepClient(config)

Bước 4: Triển Khai Customer Communication Service

# File: services/notification_service.py
from typing import Dict, List, Optional
from datetime import datetime, timedelta
import json

class CustomerNotificationService:
    """Service để gửi thông báo có cấu trúc đến khách hàng"""
    
    def __init__(self, notification_channel="email"):
        self.channel = notification_channel
        self.notification_templates = self._load_templates()
        self.sent_notifications = []  # Log for audit
    
    def _load_templates(self) -> Dict:
        """Load email/SMS templates cho từng scenario"""
        return {
            "api_error": {
                "subject_vi": "🔧 [Cập nhật] Vấn đề AI đã được xử lý - Credits bồi thường",
                "subject_en": "🔧 [Update] AI issue resolved - Compensation credits added",
                "body_vi": """
Xin chào {customer_name},

Chúng tôi ghi nhận sự cố xảy ra lúc {timestamp} khi bạn sử dụng dịch vụ AI.

**Vấn đề đã xảy ra:**
{issue_description}

**Hành động đã thực hiện:**
✅ Hệ thống tự động retry {retry_count} lần
✅ Đã chuyển sang server dự phòng
✅ Kiểm tra và khắc phục lỗi gốc

**Bồi thường dành cho bạn:**
🎁 {compensation_amount} credits miễn phí đã được cộng vào tài khoản
💎 Mã giảm giá: {discount_code} (giảm {discount_percent}% cho lần tiếp theo)

Chúng tôi xin lỗi vì sự bất tiện này và cảm ơn bạn đã kiên nhẫn.

Trân trọng,
Đội ngũ HolySheep AI
                """,
                "body_en": """
Dear {customer_name},

We recorded an issue at {timestamp} when you were using our AI service.

**Issue occurred:**
{issue_description}

**Actions taken:**
✅ System automatically retried {retry_count} times
✅ Switched to backup server
✅ Root cause identified and fixed

**Your compensation:**
🎁 {compensation_amount} free credits added to your account
💎 Discount code: {discount_code} ({discount_percent}% off next purchase)

We apologize for the inconvenience and thank you for your patience.

Best regards,
HolySheep AI Team
                """
            },
            "model_switch": {
                "subject_vi": "🤖 Model AI đã được nâng cấp - Trải nghiệm tốt hơn",
                "subject_en": "🤖 AI Model upgraded - Better experience ahead",
                "body_vi": """
Xin chào {customer_name},

Để mang đến trải nghiệm tốt nhất cho bạn, chúng tôi đã nâng cấp hệ thống:

**Thay đổi:**
🔄 Từ: {old_model}
✅ Sang: {new_model}

**Cải tiến:**
• Độ trễ giảm {latency_improvement}%
• Chất lượng phản hồi được cải thiện
• Hỗ trợ ngữ cảnh dài hơn

Không cần thay đổi gì từ phía bạn. Hệ thống đã tự động cập nhật.

Trân trọng,
HolySheep AI
                """
            },
            "maintenance": {
                "subject_vi": "📅 Bảo trì hệ thống - Thời gian chết dự kiến",
                "subject_en": "📅 System maintenance - Scheduled downtime",
                "body_vi": """
Xin chào {customer_name},

Chúng tôi sẽ thực hiện bảo trì hệ thống theo lịch trình:

**Thời gian:**
🕐 Bắt đầu: {start_time}
🕐 Kết thúc dự kiến: {end_time}
⏱️ Thời gian chết: {downtime_minutes} phút

**Ảnh hưởng:**
• API calls có thể bị delay hoặc fail nhẹ
• Dashboard có thể không truy cập được tạm thời

**Chuẩn bị:**
• Hệ thống sẽ tự động retry các requests bị gián đoạn
• Credits không bị trừ trong thời gian bảo trì

Cảm ơn sự thông cảm của bạn.

HolySheep AI
                """
            }
        }
    
    def send_error_notification(self, customer_data: Dict, error_data: Dict, 
                                compensation: Dict, language: str = "vi") -> bool:
        """
        Gửi thông báo lỗi có cấu trúc cho khách hàng
        
        Args:
            customer_data: {name, email, user_id, tier}
            error_data: {code, message, timestamp, retry_count}
            compensation: {credits, discount_code, discount_percent}
            language: ngôn ngữ gửi mail
        
        Returns:
            bool: True nếu gửi thành công
        """
        template = self.notification_templates["api_error"]
        subject_key = f"subject_{language}"
        body_key = f"body_{language}"
        
        # Prepare template variables
        issue_descriptions = {
            "rate_limit": "Lượng truy cập tăng đột biến, server tạm thời quá tải",
            "server_error": "Lỗi nội bộ tại data center, đã được đội ngũ kỹ thuật xử lý",
            "model_unavailable": "Model AI đang được bảo trì theo lịch trình",
            "timeout": "Phản hồi từ AI mất nhiều thời gian hơn bình thường",
            "connection_error": "Lỗi kết nối mạng tạm thời"
        }
        
        template_vars = {
            "customer_name": customer_data.get("name", "Quý khách"),
            "timestamp": error_data.get("timestamp", datetime.now()).strftime("%H:%M %d/%m/%Y"),
            "issue_description": issue_descriptions.get(
                error_data.get("code"), 
                error_data.get("message", "Lỗi không xác định")
            ),
            "retry_count": error_data.get("retry_count", 0),
            "compensation_amount": compensation.get("credits", 0),
            "discount_code": compensation.get("discount_code", "THANKYOU"),
            "discount_percent": compensation.get("discount_percent", 10)
        }
        
        # Render template
        subject = template[subject_key]
        body = template[body_key].format(**template_vars)
        
        # Send notification (implement theo channel cụ thể)
        success = self._send_via_channel(
            recipient=customer_data.get("email"),
            subject=subject,
            body=body,
            metadata={
                "user_id": customer_data.get("user_id"),
                "error_code": error_data.get("code"),
                "compensation_credits": compensation.get("credits")
            }
        )
        
        # Log for audit
        self.sent_notifications.append({
            "timestamp": datetime.now(),
            "recipient": customer_data.get("email"),
            "type": "error_notification",
            "success": success,
            "metadata": template_vars
        })
        
        return success
    
    def _send_via_channel(self, recipient: str, subject: str, body: str, 
                          metadata: Dict) -> bool:
        """Implement actual sending logic (email/SMS/Push)"""
        # TODO: Implement với SendGrid, AWS SES, Twilio, etc.
        # Đây là mock implementation
        print(f"[NOTIFICATION] Sending to {recipient}")
        print(f"Subject: {subject}")
        print(f"Body: {body[:100]}...")
        return True
    
    def generate_status_page_update(self, incidents: List[Dict]) -> Dict:
        """Generate JSON cho status page như statuspage.io"""
        return {
            "page": {
                "id": "holysheep_status",
                "name": "HolySheep AI Status",
                "url": "https://status.holysheep.ai",
                "timezone": "Asia/Ho_Chi_Minh"
            },
            "incidents": [
                {
                    "id": inc.get("id"),
                    "name": inc.get("name", "API Service Issue"),
                    "status": inc.get("status", "resolved"),
                    "impact": inc.get("impact", "none"),
                    "started_at": inc.get("started_at"),
                    "resolved_at": inc.get("resolved_at"),
                    "message": inc.get("message"),
                    "components": {
                        "api": inc.get("api_status", "operational"),
                        "dashboard": inc.get("dashboard_status", "operational")
                    }
                }
                for inc in incidents
            ]
        }


Sử dụng

notification_service = CustomerNotificationService()

Bước 5: Integration Với Ứng Dụng FastAPI

# File: app/api/routes.py
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse
from pydantic import BaseModel
from typing import List, Optional
from datetime import datetime

from services.holysheep_client import client, HolySheepClient
from services.notification_service import notification_service
from config.api_config import config

app = FastAPI(title="AI Chat API with HolySheep")

class ChatRequest(BaseModel):
    messages: List[dict]
    model: str = "gpt-4o"
    temperature: float = 0.7
    max_tokens: Optional[int] = 2000
    user_id: Optional[str] = None

class ChatResponse(BaseModel):
    success: bool
    response: Optional[str] = None
    customer_message: Optional[str] = None
    latency_ms: int
    model: str
    credits_used: Optional[int] = None


Error callback để track errors

def error_callback(api_error, attempt, payload):
    print(f"[ERROR TRACKER] Attempt {attempt + 1}: {api_error.code} - {api_error.message}")
    # Gửi alert đến Slack/PagerDuty
    # ...


Compensation callback

def compensation_callback(user_id, error_type, compensation):
    print(f"[COMPENSATION] User {user_id}: +{compensation['credits']} credits")
    # Update database, send notification
    # ...


Register callbacks

client.register_error_callback(error_callback)
client.register_compensation_callback(compensation_callback)

@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest, http_request: Request):
    """
    Main chat endpoint - xử lý request với error handling hoàn chỉnh
    """
    # Extract user context từ request
    customer_data = {
        "user_id": request.user_id or "anonymous",
        "email": http_request.headers.get("X-User-Email"),
        "name": http_request.headers.get("X-User-Name", "Quý khách"),
        "tier": http_request.headers.get("X-User-Tier", "free")
    }
    
    # Call HolySheep API
    result = client.chat_completions(
        messages=request.messages,
        model=request.model,
        temperature=request.temperature,
        max_tokens=request.max_tokens,
        user_id=customer_data["user_id"]
    )
    
    if result["success"]:
        return ChatResponse(
            success=True,
            response=result["data"]["choices"][0]["message"]["content"],
            latency_ms=result["latency_ms"],
            model=result["model"],
            credits_used=_estimate_credits(result["data"])
        )
    else:
        # Get error details
        error = result["error"]
        compensation = {
            "credits": 100 if error.severity.value in ["high", "critical"] else 50,
            "discount_code": "SORRY10" if error.severity.value == "critical" else None,
            "discount_percent": 10
        }
        
        # Send customer notification
        notification_service.send_error_notification(
            customer_data=customer_data,
            error_data={
                "code": error.code,
                "message": error.message,
                "timestamp": error.timestamp,
                "retry_count": result.get("total_attempts", 1)
            },
            compensation=compensation
        )
        
        # Return customer-friendly response
        return ChatResponse(
            success=False,
            customer_message=result["customer_message"],
            latency_ms=result["latency_ms"],
            model=request.model
        )

@app.get("/health")
async def health_check():
    """Health check endpoint cho monitoring"""
    return {
        "status": "healthy",
        "provider": "HolySheep AI",
        "base_url": config.base_url,
        "timestamp": datetime.now().isoformat()
    }

def _estimate_credits(response_data: dict) -> int:
    """Ước tính credits đã sử dụng"""
    # Rough estimation: 1 token ≈ 1 credit
    usage = response_data.get("usage", {})
    total_tokens = usage.get("total_tokens", 0)
    return total_tokens


Run: uvicorn app.api.routes:app --host 0.0.0.0 --port 8000

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

Lỗi 1: 401 Unauthorized - API Key Không Hợp Lệ

# ❌ SAI: Hardcode API key trong code
client = HolySheepClient(HolySheepConfig(api_key="sk-xxxxx-real-key"))


✅ ĐÚNG: Sử dụng environment variable

import os
client = HolySheepClient(HolySheepConfig(
    api_key=os.getenv("HOLYSHEEP_API_KEY")  # or raise ValueError
))


Kiểm tra key format

def validate_api_key(key: str) -> bool:
    if not key:
        return False
    if len(key) < 10:
        return False
    # HolySheep key format: hs_xxxx... hoặc sk-xxxx...
    return key.startswith(("hs_", "sk-", "holysheep_"))


Xử lý khi key không hợp lệ

if not validate_api_key(os.getenv("HOLYSHEEP_API_KEY")):
    raise EnvironmentError("""
    HOLYSHEEP_API_KEY không được set hoặc không hợp lệ.
    Vui lòng:
    1. Đăng ký tại: https://www.holysheep.ai/register
    2. Lấy API key từ Dashboard -> API Keys
    3. Set environment variable: export HOLYSHEEP_API_KEY='your-key'
    """)

Lỗi 2: 429 Rate Limit - Quá Nhiều Requests

# ❌ SAI: Retry ngay lập tức không có backoff
for i in range(10):
    response = client.chat_completions(messages)
    if response.status_code != 429:
        break
    # Retry ngay = càng làm nặng server


✅ ĐÚNG: Exponential backoff với jitter

import random
import time

def retry_with_backoff(client, messages, max_retries=5):
    for attempt in range(max_retries):
        response = client.chat_completions(messages)
        
        if response.status_code != 429:
            return response
        
        # Parse Retry-After header
        retry_after = int(response.headers.get("Retry-After", 60))
        
        # Exponential backoff: 1s, 2s, 4s, 8s, 16s...
        wait_time = min(retry_after, (2 ** attempt))
        
        # Thêm jitter (±20%) để tránh thundering herd
        jitter = wait_time * 0.2 * (random.random() * 2 - 1)
        actual_wait = wait_time + jitter
        
        print(f"[RATE LIMIT] Attempt {attempt + 1} failed. Retrying in {actual_wait:.1f}s...")
        time.sleep(actual_wait)
    
    # Sau max_retries, fallback sang model khác
    return client.chat_completions(messages, model="gpt-3.5-turbo")


Sử dụng rate limiter phía client

from functools import wraps
import threading

class RateLimiter:
    def __init__(self, max_calls: int, period: float):
        self.max_calls = max_calls
        self.period = period
        self.calls = []
        self.lock = threading.Lock()
    
    def __call__(self, func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            with self.lock:
                now = time.time()
                # Remove calls outside current window
                self.calls = [t for t in self.calls if now - t < self.period]
                
                if len(self.calls) >= self.max_calls:
                    sleep_time = self.period - (now - self.calls[0])
                    print(f"[RATE LIMITER] Sleeping {sleep_time:.1f}s...")
                    time.sleep(sleep_time)
                    self.calls = self

Tài nguyên liên quan
Bài viết liên quan