Cách Debug Lỗi AI API Response Với Custom Error Handlers

Mở đầu: Khi API Trả Về "ConnectionError: timeout" Lúc 2 Giờ Sáng

Tôi vẫn nhớ rất rõ cái đêm tháng 6 năm 2024 — hệ thống AI chatbot của khách hàng báo lỗi liên tục, log đầy những dòng ConnectionError: timeout và 401 Unauthorized. Đó là lúc tôi nhận ra: nếu không có custom error handler tốt, bạn sẽ mất hàng giờ để debug thay vì fix trong 5 phút.

Bài viết này là tổng hợp kinh nghiệm thực chiến của tôi trong 3 năm làm việc với các AI API — từ OpenAI, Anthropic, đến HolySheheep AI. Tôi sẽ chia sẻ cách xây dựng hệ thống xử lý lỗi chuyên nghiệp, tiết kiệm 85%+ chi phí với mức giá chỉ ¥1=$1.

Tại Sao Cần Custom Error Handler?

Default error handling của thư viện HTTP thường không đủ cho production. Bạn cần:

• Xử lý retry tự động với exponential backoff
• Phân biệt lỗi tạm thời (timeout, rate limit) vs lỗi cố định (auth, invalid request)
• Ghi log chi tiết để trace lỗi nhanh chóng
• Tự động fallback sang provider dự phòng

Triển Khai Custom Error Handler Toàn Diện

1. Cấu Trúc Error Classes

"""
Custom Error Handler cho HolySheep AI API
Author: HolySheep AI Technical Team
"""

import time
import asyncio
import logging
from typing import Optional, Dict, Any
from dataclasses import dataclass, field
from enum import Enum
from datetime import datetime

class ErrorSeverity(Enum):
    LOW = "low"           # Timeout nhẹ, retry được
    MEDIUM = "medium"     # Rate limit, cần delay
    HIGH = "high"         # Auth error, config error
    CRITICAL = "critical" # Payment, quota exceeded

@dataclass
class APIError(Exception):
    """Base exception cho tất cả API errors"""
    message: str
    status_code: Optional[int] = None
    error_code: Optional[str] = None
    severity: ErrorSeverity = ErrorSeverity.MEDIUM
    retry_after: Optional[int] = None
    metadata: Dict[str, Any] = field(default_factory=dict)
    timestamp: datetime = field(default_factory=datetime.now)
    
    def __str__(self):
        return f"[{self.severity.value.upper()}] {self.message} (HTTP {self.status_code})"

class RateLimitError(APIError):
    """Khi vượt quota - thường có retry-after header"""
    def __init__(self, message: str, retry_after: int = 60, **kwargs):
        super().__init__(message, status_code=429, severity=ErrorSeverity.MEDIUM, **kwargs)
        self.retry_after = retry_after

class AuthenticationError(APIError):
    """401 Unauthorized - API key không hợp lệ hoặc hết hạn"""
    def __init__(self, message: str, **kwargs):
        super().__init__(message, status_code=401, severity=ErrorSeverity.HIGH, **kwargs)

class InvalidRequestError(APIError):
    """400 Bad Request - Payload không hợp lệ"""
    def __init__(self, message: str, validation_errors: list = None, **kwargs):
        super().__init__(message, status_code=400, severity=ErrorSeverity.MEDIUM, **kwargs)
        self.validation_errors = validation_errors or []

class ServerError(APIError):
    """500/502/503 - Lỗi phía server HolySheep AI"""
    def __init__(self, message: str, **kwargs):
        super().__init__(message, status_code=500, severity=ErrorSeverity.CRITICAL, **kwargs)

class TimeoutError(APIError):
    """Request timeout - thường retry được"""
    def __init__(self, message: str, timeout_ms: int = 30000, **kwargs):
        super().__init__(message, status_code=408, severity=ErrorSeverity.LOW, **kwargs)
        self.timeout_ms = timeout_ms

2. Retry Engine Với Exponential Backoff

"""
HolySheep AI Client với built-in retry logic
Pricing 2026: GPT-4.1 $8/MTok, Claude Sonnet 4.5 $15/MTok
Special: ¥1=$1 - tiết kiệm 85%+ vs competitors
"""

import aiohttp
import asyncio
import logging
from typing import Optional, Callable, Any
import json

logger = logging.getLogger(__name__)

class HolySheepRetryClient:
    """Client với intelligent retry và error handling"""
    
    BASE_URL = "https://api.holysheep.ai/v1"
    MAX_RETRIES = 3
    INITIAL_DELAY = 1.0  # Giây
    MAX_DELAY = 60.0     # Giây
    
    # Error codes chuẩn của HolySheep API
    RETRYABLE_ERROR_CODES = {
        "rate_limit_exceeded",
        "server_overloaded", 
        "temporary_unavailable",
        "timeout",
        "connection_reset"
    }
    
    def __init__(self, api_key: str, default_model: str = "gpt-4.1"):
        self.api_key = api_key
        self.default_model = default_model
        self.session: Optional[aiohttp.ClientSession] = None
        
    async def __aenter__(self):
        timeout = aiohttp.ClientTimeout(total=60, connect=10)
        self.session = aiohttp.ClientSession(timeout=timeout)
        return self
        
    async def __aexit__(self, *args):
        if self.session:
            await self.session.close()
    
    def _should_retry(self, status_code: int, error_code: str = None) -> bool:
        """Xác định có nên retry không"""
        if status_code == 429:  # Rate limit
            return True
        if 500 <= status_code < 600:  # Server error
            return True
        if status_code == 408:  # Timeout
            return True
        if error_code and error_code in self.RETRYABLE_ERROR_CODES:
            return True
        return False
    
    def _calculate_delay(self, attempt: int, retry_after: int = None) -> float:
        """Tính delay với exponential backoff + jitter"""
        if retry_after:
            return min(retry_after, self.MAX_DELAY)
        
        delay = self.INITIAL_DELAY * (2 ** attempt)
        # Thêm jitter 0-25% để tránh thundering herd
        import random
        jitter = delay * random.uniform(0, 0.25)
        return min(delay + jitter, self.MAX_DELAY)
    
    async def _handle_response(self, response: aiohttp.ClientResponse) -> Dict[str, Any]:
        """Parse và xử lý response từ HolySheep API"""
        status = response.status
        data = await response.json() if response.content_type == 'application/json' else {}
        
        # Lấy retry-after từ header nếu có
        retry_after = response.headers.get('Retry-After')
        if retry_after:
            retry_after = int(retry_after)
        
        if status == 200:
            return {"success": True, "data": data}
        
        # Parse error response
        error_code = data.get("error", {}).get("code", "")
        error_message = data.get("error", {}).get("message", "Unknown error")
        
        # Map sang custom exceptions
        if status == 401:
            raise AuthenticationError(
                f"Authentication failed: {error_message}",
                metadata={"error_code": error_code}
            )
        elif status == 400:
            raise InvalidRequestError(
                f"Invalid request: {error_message}",
                validation_errors=data.get("error", {}).get("details", [])
            )
        elif status == 429:
            raise RateLimitError(
                f"Rate limit exceeded: {error_message}",
                retry_after=retry_after or 60
            )
        elif 500 <= status < 600:
            raise ServerError(
                f"Server error: {error_message}",
                metadata={"status": status, "error_code": error_code}
            )
        else:
            raise APIError(
                f"Unexpected error: {error_message}",
                status_code=status,
                severity=ErrorSeverity.HIGH,
                metadata={"error_code": error_code}
            )
    
    async def chat_completion(
        self,
        messages: list,
        model: str = None,
        temperature: float = 0.7,
        max_tokens: int = 2048,
        retry_count: int = 0
    ) -> Dict[str, Any]:
        """
        Gọi Chat Completion API với automatic retry
        
        Model pricing 2026:
        - gpt-4.1: $8/MTok (input) - $8/MTok (output)
        - claude-sonnet-4.5: $15/MTok (input) - $15/MTok (output)  
        - gemini-2.5-flash: $2.50/MTok (input) - $10/MTok (output)
        - deepseek-v3.2: $0.42/MTok (input) - $0.42/MTok (output)
        """
        if not self.session:
            raise RuntimeError("Client must be used as async context manager")
        
        endpoint = f"{self.BASE_URL}/chat/completions"
        model = model or self.default_model
        
        payload = {
            "model": model,
            "messages": messages,
            "temperature": temperature,
            "max_tokens": max_tokens
        }
        
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        try:
            logger.info(f"Request to {endpoint} with model {model}")
            async with self.session.post(endpoint, json=payload, headers=headers) as resp:
                result = await self._handle_response(resp)
                logger.info(f"Success: {model} response received")
                return result
                
        except RateLimitError as e:
            logger.warning(f"Rate limit hit, retrying after {e.retry_after}s...")
            await asyncio.sleep(e.retry_after)
            return await self.chat_completion(messages, model, temperature, max_tokens, retry_count + 1)
            
        except (TimeoutError, aiohttp.ClientError) as e:
            if retry_count < self.MAX_RETRIES:
                delay = self._calculate_delay(retry_count)
                logger.warning(f"Retry {retry_count + 1}/{self.MAX_RETRIES} after {delay:.1f}s: {e}")
                await asyncio.sleep(delay)
                return await self.chat_completion(messages, model, temperature, max_tokens, retry_count + 1)
            raise TimeoutError(f"Max retries exceeded: {str(e)}")
            
        except ServerError as e:
            if retry_count < self.MAX_RETRIES:
                delay = self._calculate_delay(retry_count)
                logger.error(f"Server error, retrying: {e}")
                await asyncio.sleep(delay)
                return await self.chat_completion(messages, model, temperature, max_tokens, retry_count + 1)
            raise

============== SỬ DỤNG ==============

async def main():
    async with HolySheepRetryClient(api_key="YOUR_HOLYSHEEP_API_KEY") as client:
        try:
            response = await client.chat_completion(
                messages=[
                    {"role": "system", "content": "Bạn là trợ lý AI tiếng Việt."},
                    {"role": "user", "content": "Xin chào, hãy giới thiệu về HolySheep AI"}
                ],
                model="deepseek-v3.2",  # Model rẻ nhất: $0.42/MTok
                temperature=0.7,
                max_tokens=500
            )
            print(f"Response: {response['data']['choices'][0]['message']['content']}")
            
        except AuthenticationError as e:
            print(f"❌ Auth Error: {e}")
            # Kiểm tra API key hoặc upgrade plan
        except RateLimitError as e:
            print(f"⏳ Rate limited, retry after {e.retry_after}s")
        except APIError as e:
            print(f"❌ API Error: {e}")
            # Log chi tiết để debug

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

3. Error Monitoring Dashboard

"""
Error Monitoring System - Theo dõi và alert khi có lỗi
Tích hợp với logging, metrics, và alerting
"""

import logging
from datetime import datetime, timedelta
from collections import defaultdict
from typing import Dict, List, Optional
import json

class ErrorMonitor:
    """Monitor và phân tích lỗi API"""
    
    def __init__(self, alert_threshold: int = 10):
        self.alert_threshold = alert_threshold
        self.error_counts: Dict[str, int] = defaultdict(int)
        self.error_log: List[Dict] = []
        self.error_by_hour: Dict[int, Dict[str, int]] = defaultdict(lambda: defaultdict(int))
        
    def log_error(self, error: APIError, context: Dict = None):
        """Ghi log error với context"""
        error_entry = {
            "timestamp": datetime.now().isoformat(),
            "type": type(error).__name__,
            "message": str(error),
            "status_code": error.status_code,
            "severity": error.severity.value,
            "retry_after": error.retry_after,
            "metadata": error.metadata,
            "context": context or {}
        }
        
        self.error_log.append(error_entry)
        self.error_counts[type(error).__name__] += 1
        
        # Đếm theo giờ
        hour = datetime.now().hour
        self.error_by_hour[hour][type(error).__name__] += 1
        
        # Alert nếu vượt threshold
        if self.error_counts[type(error).__name__] >= self.alert_threshold:
            self._send_alert(error)
    
    def _send_alert(self, error: APIError):
        """Gửi alert khi có vấn đề nghiêm trọng"""
        # Có thể tích hợp với Slack, PagerDuty, email...
        logging.critical(
            f"🚨 ALERT: {type(error).__name__} threshold exceeded! "
            f"Count: {self.error_counts[type(error).__name__]}"
        )
    
    def get_error_stats(self, hours: int = 24) -> Dict:
        """Lấy thống kê lỗi trong N giờ qua"""
        cutoff = datetime.now() - timedelta(hours=hours)
        recent_errors = [
            e for e in self.error_log 
            if datetime.fromisoformat(e["timestamp"]) > cutoff
        ]
        
        return {
            "total_errors": len(recent_errors),
            "by_type": defaultdict(int, {
                e["type"]: recent_errors.count(e) 
                for e in recent_errors
            }),
            "by_severity": defaultdict(int, {
                e["severity"]: recent_errors.count(e)
                for e in recent_errors
            }),
            "most_common": sorted(
                self.error_counts.items(), 
                key=lambda x: x[1], 
                reverse=True
            )[:5]
        }
    
    def generate_report(self) -> str:
        """Tạo báo cáo lỗi định kỳ"""
        stats = self.get_error_stats()
        
        report = f"""
📊 HOLYSHEP AI ERROR REPORT
Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
================================

Total Errors (24h): {stats['total_errors']}

By Type:
{chr(10).join(f"  - {k}: {v}" for k, v in stats['by_type'].items())}

By Severity:
{chr(10).join(f"  - {k}: {v}" for k, v in stats['by_severity'].items())}

Top 5 Most Common:
{chr(10).join(f"  {i+1}. {k}: {v} times" for i, (k, v) in enumerate(stats['most_common']))}
"""
        return report

============== INTEGRATION ==============

Tích hợp với FastAPI
from fastapi import FastAPI, HTTPException

app = FastAPI()
error_monitor = ErrorMonitor(alert_threshold=5)

@app.post("/v1/chat")
async def chat_endpoint(request: ChatRequest):
    async with HolySheepRetryClient(api_key="YOUR_HOLYSHEEP_API_KEY") as client:
        try:
            result = await client.chat_completion(
                messages=request.messages,
                model=request.model
            )
            return result
        except APIError as e:
            error_monitor.log_error(e, {"endpoint": "/v1/chat", "model": request.model})
            raise HTTPException(status_code=e.status_code or 500, detail=str(e))

@app.get("/admin/errors")
async def get_error_stats():
    """Endpoint để xem thống kê lỗi"""
    return error_monitor.get_error_stats()

@app.get("/admin/errors/report")
async def get_error_report():
    """Endpoint để lấy báo cáo lỗi"""
    return {"report": error_monitor.generate_report()}

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

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

Mô tả lỗi: Khi gọi API mà nhận được response HTTP 401.

# Error Response:
{
  "error": {
    "code": "invalid_api_key",
    "message": "The API key provided is invalid or has been revoked"
  }
}

Cách khắc phục:
1. Kiểm tra API key trong dashboard HolySheep AI
https://www.holysheep.ai/dashboard/api-keys

2. Verify API key format (phải bắt đầu bằng "hss_")
YOUR_API_KEY = "hss_xxxxxxxxxxxxxxxxxxxx"

3. Kiểm tra key còn hạn không
async def verify_api_key(api_key: str) -> bool:
    async with HolySheepRetryClient(api_key=api_key) as client:
        try:
            await client.chat_completion(
                messages=[{"role": "user", "content": "test"}],
                max_tokens=1
            )
            return True
        except AuthenticationError:
            return False

4. Nếu key hết hạn, tạo key mới tại:
https://www.holysheep.ai/dashboard/api-keys

2. Lỗi "429 Rate Limit Exceeded" - Vượt Quá Giới Hạn Request

Mô tả lỗi: Gửi quá nhiều request trong thời gian ngắn.

# Error Response:
{
  "error": {
    "code": "rate_limit_exceeded",
    "message": "You have exceeded your API rate limit"
  }
}
Response Header: Retry-After: 60

Cách khắc phục:
1. Implement rate limiter phía client
import asyncio
from datetime import datetime, timedelta

class RateLimiter:
    def __init__(self, max_requests: int = 60, window_seconds: int = 60):
        self.max_requests = max_requests
        self.window = timedelta(seconds=window_seconds)
        self.requests = []
        
    async def acquire(self):
        now = datetime.now()
        # Loại bỏ request cũ
        self.requests = [r for r in self.requests if now - r < self.window]
        
        if len(self.requests) >= self.max_requests:
            sleep_time = (self.requests[0] + self.window - now).total_seconds()
            await asyncio.sleep(sleep_time)
            self.requests = [r for r in self.requests if datetime.now() - r < self.window]
        
        self.requests.append(datetime.now())

2. Sử dụng batch processing thay vì gọi tuần tự
async def batch_chat_completion(messages_list: list, batch_size: int = 10):
    rate_limiter = RateLimiter(max_requests=50, window_seconds=60)
    results = []
    
    for i in range(0, len(messages_list), batch_size):
        batch = messages_list[i:i+batch_size]
        tasks = []
        
        for msg in batch:
            await rate_limiter.acquire()
            async with HolySheepRetryClient(api_key="YOUR_HOLYSHEEP_API_KEY") as client:
                tasks.append(client.chat_completion(msg))
        
        batch_results = await asyncio.gather(*tasks, return_exceptions=True)
        results.extend(batch_results)
        
        # Delay giữa các batch
        await asyncio.sleep(1)
    
    return results

3. Upgrade plan nếu cần throughput cao hơn
HolySheep AI có các gói: Free (100 req/min), Pro (500 req/min), Enterprise (unlimited)

3. Lỗi "ConnectionError: timeout" - Request Timeout

Mô tả lỗi: Server không phản hồi trong thời gian chờ.

# Error: asyncio.exceptions.TimeoutError hoặc
aiohttp.client_exceptions.ServerTimeoutError

Cách khắc phục:
1. Tăng timeout cho requests lớn
async def chat_with_extended_timeout():
    timeout = aiohttp.ClientTimeout(total=120, connect=30)  # 120s total, 30s connect
    
    async with aiohttp.ClientSession(timeout=timeout) as session:
        # Retry với backoff
        for attempt in range(3):
            try:
                async with session.post(
                    "https://api.holysheep.ai/v1/chat/completions",
                    json={"model": "gpt-4.1", "messages": [...], "max_tokens": 4096},
                    headers={"Authorization": f"Bearer YOUR_API_KEY"}
                ) as resp:
                    return await resp.json()
            except asyncio.TimeoutError:
                wait = 2 ** attempt
                print(f"Timeout, retrying in {wait}s...")
                await asyncio.sleep(wait)
                
2. Split large requests thành chunks
async def chat_with_streaming(messages: list, chunk_size: int = 10):
    """Xử lý message dài bằng cách split thành chunks"""
    chunks = [messages[i:i+chunk_size] for i in range(0, len(messages), chunk_size)]
    responses = []
    
    async with HolySheepRetryClient(api_key="YOUR_API_KEY") as client:
        for i, chunk in enumerate(chunks):
            try:
                result = await client.chat_completion(
                    messages=[{"role": "system", "content": f"Part {i+1}/{len(chunks)}"}] + chunk,
                    max_tokens=2000  # Giới hạn output để tránh timeout
                )
                responses.append(result)
            except TimeoutError:
                # Fallback: gửi lại với model nhẹ hơn
                result = await client.chat_completion(
                    messages=chunk,
                    model="deepseek-v3.2",  # Model nhanh hơn, rẻ hơn
                    max_tokens=1000
                )
                responses.append(result)
                
    return responses

3. Kiểm tra network với ping test
import subprocess

def check_api_health():
    result = subprocess.run(
        ["ping", "-c", "4", "api.holysheep.ai"],
        capture_output=True, text=True
    )
    if result.returncode != 0:
        print("⚠️ Network issue detected!")
        # Thử reconnect hoặc switch backup provider
    else:
        print("✅ API endpoint reachable")
        print(result.stdout)

4. Lỗi "400 Bad Request" - Request Format Không Hợp Lệ

# Error Response:
{
  "error": {
    "code": "invalid_request",
    "message": "Invalid request parameters",
    "details": [
      {"field": "messages", "error": "messages cannot be empty"},
      {"field": "temperature", "error": "temperature must be between 0 and 2"}
    ]
  }
}

Cách khắc phục:
from pydantic import BaseModel, validator
from typing import List, Optional

class ChatRequest(BaseModel):
    messages: List[dict]
    model: str = "gpt-4.1"
    temperature: float = 0.7
    max_tokens: int = 2048
    top_p: Optional[float] = 1.0
    stream: bool = False
    
    @validator('messages')
    def messages_not_empty(cls, v):
        if not v:
            raise ValueError("messages cannot be empty")
        for msg in v:
            if 'role' not in msg or 'content' not in msg:
                raise ValueError("Each message must have 'role' and 'content'")
        return v
    
    @validator('temperature')
    def temperature_range(cls, v):
        if not 0 <= v <= 2:
            raise ValueError("temperature must be between 0 and 2")
        return v
    
    @validator('max_tokens')
    def max_tokens_range(cls, v):
        if v < 1 or v > 32000:
            raise ValueError("max_tokens must be between 1 and 32000")
        return v

Sử dụng validation trước khi gọi API
async def safe_chat(request_data: dict):
    try:
        validated = ChatRequest(**request_data)
        async with HolySheepRetryClient(api_key="YOUR_API_KEY") as client:
            return await client.chat_completion(
                messages=validated.messages,
                model=validated.model,
                temperature=validated.temperature,
                max_tokens=validated.max_tokens
            )
    except Exception as e:
        print(f"Validation error: {e}")
        raise

Bảng So Sánh Error Codes

Error Code	HTTP Status	Severity	Retry?	Giải Pháp
invalid_api_key	401	CRITICAL	❌	Kiểm tra/tạo API key mới
rate_limit_exceeded	429	MEDIUM	✅	Chờ theo Retry-After
server_overloaded	503	MEDIUM	✅	Retry với exponential backoff
invalid_request	400	HIGH	❌	Sửa request payload
context_length_exceeded	400	HIGH	❌	Giảm message history
quota_exceeded	429	CRITICAL	❌	Nâng cấp plan

Kinh Nghiệm Thực Chiến

Trong quá trình vận hành hệ thống AI cho nhiều khách hàng, tôi đã rút ra được vài bài học quan trọng:

1. Luôn có fallback plan: Một lần hệ thống của tôi bị downtime 2 tiếng vì HolySheep API có maintenance không được thông báo trước. Từ đó tôi luôn setup backup với provider thứ 2. Với HolySheep AI, tỷ giá ¥1=$1 giúp việc duy trì backup không tốn nhiều chi phí.

2. Monitoring là chìa khóa: Error không được log = error không tồn tại. Đầu tư vào monitoring sẽ giúp bạn phát hiện vấn đề trước khi khách hàng phàn nàn.

3. Retry nhưng có giới hạn: Đặt MAX_RETRIES hợp lý (3-5 lần). Retry vô hạn sẽ gây tích lũy request và tăng chi phí không cần thiết.

4. Chọn model phù hợp: Không phải lúc nào cũng cần GPT-4.1 ($8/MTok). Với các task đơn giản, DeepSeek V3.2 ($0.42/MTok) hoàn toàn đủ. Tiết kiệm 95% chi phí cho những use case không đòi hỏi model mạnh nhất.

Tổng Kết

Custom error handler là phần không thể thiếu khi làm việc với AI API. Với những gì đã chia sẻ, bạn có thể:

• Xử lý mọi loại lỗi từ HolySheep AI một cách chuyên nghiệp
• Implement retry logic với exponential backoff
• Monitor và alert khi có vấn đề
• Tiết kiệm đến 85%+ chi phí với mức giá ¥1=$1

Hãy bắt đầu bằng cách clone code từ bài viết này và tích hợp vào project của bạn. Đừng quên đăng ký tài khoản HolySheep AI để nhận tín dụng miễn phí khi bắt đầu!

👉 Đăng ký HolySheep AI — nhận tín dụng miễn phí khi đăng ký