AI API故障应急响应：2026年企业级容灾实战手册

引言：为什么你的AI服务总是"关键时刻掉链子"

凌晨3点，你被手机警报惊醒——生产环境的AI服务彻底宕机。用户投诉如潮水般涌来，而你的团队正在黑暗中摸索。这不是科幻情节，这是无数工程师经历过的真实噩梦。

作为一名在AI基础设施领域摸爬滚打8年的老兵，我见过太多团队因为API故障处理不当而导致业务中断。今天，我将分享一套经过实战验证的应急响应方案，帮助你在AI服务出现问题时能够快速止血。

一、2026年AI API成本格局：你真的选对供应商了吗？

在深入故障处理之前，我们先看看当前的API成本结构。很多团队之所以频繁遇到服务问题，部分原因是他们被昂贵的官方API价格绑架，不得不采用不稳定的方案。

2026年主流模型成本对比

模型	输出价格(USD/MTok)	10M Token/月成本	性价比指数
GPT-4.1	$8.00	$80,000	⭐
Claude Sonnet 4.5	$15.00	$150,000	⭐
Gemini 2.5 Flash	$2.50	$25,000	⭐⭐⭐
DeepSeek V3.2	$0.42	$4,200	⭐⭐⭐⭐⭐

看到了吗？DeepSeek V3.2的成本只有GPT-4.1的1/19！这个差距意味着你可以用同样的预算支撑19倍的请求量，或者将省下的资金投入到服务稳定性建设中。

HolySheep AI的竞争优势

作为深耕亚太市场的AI API聚合平台，HolySheep AI提供了极具竞争力的定价策略：

• 极致性价比：汇率按¥1=$1计算，综合节省超过85%
• 本地化支付：支持微信、支付宝，告别国际支付烦恼
• 超低延迟：亚太区域部署，响应时间<50ms
• 开箱即用：注册即送免费额度，新用户友好

二、AI API故障分类与诊断矩阵

根据我的实战经验，AI API故障可以分为以下几大类，每类都有其特征和应对策略：

2.1 网络层故障

这是最常见的故障类型，约占所有问题的40%。特征是请求超时、连接被重置、SSL握手失败。

2.2 认证与配额故障

API Key过期、额度耗尽、并发限制触发——这类问题往往在流量高峰期集中爆发。

2.3 模型服务故障

上游模型供应商宕机、服务降级、响应质量异常。这类问题你无法直接控制，但可以通过多供应商策略规避。

2.4 应用层故障

请求格式错误、超长上下文导致OOM、流式响应中断等。

三、构建智能路由层：Python实战

解决AI API故障的根本之道是不要把所有鸡蛋放在一个篮子里。下面是一个生产级的多供应商路由实现：

# holy_sheep_router.py
多供应商AI API智能路由与故障切换系统
作者：HolySheep AI技术团队

import asyncio
import logging
from dataclasses import dataclass
from typing import Optional, Dict, List
from enum import Enum
import httpx
from datetime import datetime, timedelta

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class ProviderStatus(Enum):
    HEALTHY = "healthy"
    DEGRADED = "degraded"
    DOWN = "down"


@dataclass
class ProviderConfig:
    name: str
    base_url: str
    api_key: str
    timeout: float = 30.0
    max_retries: int = 3
    health_check_interval: int = 60


class HealthChecker:
    """供应商健康状态检查器"""
    
    def __init__(self):
        self.provider_health: Dict[str, ProviderStatus] = {}
        self.last_check: Dict[str, datetime] = {}
        self.error_counts: Dict[str, int] = {}
        self.success_counts: Dict[str, int] = {}
    
    async def check_health(self, provider: ProviderConfig) -> ProviderStatus:
        """执行健康检查"""
        try:
            async with httpx.AsyncClient(timeout=5.0) as client:
                response = await client.post(
                    f"{provider.base_url}/chat/completions",
                    headers={
                        "Authorization": f"Bearer {provider.api_key}",
                        "Content-Type": "application/json"
                    },
                    json={
                        "model": "gpt-3.5-turbo",
                        "messages": [{"role": "user", "content": "ping"}],
                        "max_tokens": 5
                    }
                )
                
                if response.status_code == 200:
                    self.provider_health[provider.name] = ProviderStatus.HEALTHY
                    self.success_counts[provider.name] = \
                        self.success_counts.get(provider.name, 0) + 1
                    self.error_counts[provider.name] = 0
                    return ProviderStatus.HEALTHY
                else:
                    raise Exception(f"Status: {response.status_code}")
                    
        except Exception as e:
            logger.error(f"Health check failed for {provider.name}: {e}")
            self.error_counts[provider.name] = \
                self.error_counts.get(provider.name, 0) + 1
            
            # 连续失败3次则标记为DOWN
            if self.error_counts[provider.name] >= 3:
                self.provider_health[provider.name] = ProviderStatus.DOWN
            else:
                self.provider_health[provider.name] = ProviderStatus.DEGRADED
                
            return self.provider_health[provider.name]


class SmartRouter:
    """智能路由引擎 - 自动故障切换"""
    
    def __init__(self):
        self.providers: List[ProviderConfig] = []
        self.health_checker = HealthChecker()
        self.current_index = 0
        self.circuit_breakers: Dict[str, int] = {}
    
    def add_provider(self, provider: ProviderConfig):
        """注册AI供应商"""
        self.providers.append(provider)
        self.circuit_breakers[provider.name] = 0
        logger.info(f"Added provider: {provider.name} @ {provider.base_url}")
    
    def get_available_provider(self) -> Optional[ProviderConfig]:
        """获取可用供应商 - 轮询+健康检查"""
        attempts = len(self.providers)
        
        for i in range(attempts):
            idx = (self.current_index + i) % len(self.providers)
            provider = self.providers[idx]
            status = self.health_checker.provider_health.get(
                provider.name, ProviderStatus.HEALTHY
            )
            
            if status != ProviderStatus.DOWN:
                self.current_index = (idx + 1) % len(self.providers)
                return provider
        
        logger.error("No available providers!")
        return None
    
    async def route_request(
        self,
        messages: List[Dict],
        model: str = "gpt-4",
        **kwargs
    ) -> Optional[Dict]:
        """路由请求并处理故障切换"""
        
        provider = self.get_available_provider()
        if not provider:
            return {"error": "All providers unavailable", "status": 503}
        
        for attempt in range(provider.max_retries):
            try:
                async with httpx.AsyncClient(
                    timeout=provider.timeout
                ) as client:
                    response = await client.post(
                        f"{provider.base_url}/chat/completions",
                        headers={
                            "Authorization": f"Bearer {provider.api_key}",
                            "Content-Type": "application/json"
                        },
                        json={
                            "model": model,
                            "messages": messages,
                            **kwargs
                        }
                    )
                    
                    if response.status_code == 200:
                        result = response.json()
                        logger.info(
                            f"Success via {provider.name}, "
                            f"latency: {result.get('response_ms', 'N/A')}ms"
                        )
                        return result
                    
                    elif response.status_code == 429:
                        # 触发限流时切换供应商
                        logger.warning(f"Rate limited by {provider.name}")
                        self.circuit_breakers[provider.name] = \
                            self.circuit_breakers.get(provider.name, 0) + 1
                        provider = self.get_available_provider()
                        if not provider:
                            break
                            
                    else:
                        logger.error(
                            f"Error from {provider.name}: "
                            f"{response.status_code}"
                        )
                        
            except httpx.TimeoutException:
                logger.error(f"Timeout from {provider.name}")
                self.circuit_breakers[provider.name] = \
                    self.circuit_breakers.get(provider.name, 0) + 1
                    
            except Exception as e:
                logger.error(f"Exception from {provider.name}: {e}")
        
        return {
            "error": "Request failed after all retries",
            "status": 500
        }


===== 使用示例 =====
async def demo():
    router = SmartRouter()
    
    # 注册多个供应商
    router.add_provider(ProviderConfig(
        name="holysheep-gpt4",
        base_url="https://api.holysheep.ai/v1",
        api_key="YOUR_HOLYSHEEP_API_KEY",  # 替换为你的密钥
        timeout=30.0
    ))
    
    router.add_provider(ProviderConfig(
        name="holysheep-deepseek",
        base_url="https://api.holysheep.ai/v1",
        api_key="YOUR_HOLYSHEEP_API_KEY",  # 替换为你的密钥
        timeout=30.0
    ))
    
    # 执行请求
    result = await router.route_request(
        messages=[{
            "role": "user",
            "content": "用Python写一个快速排序算法"
        }],
        model="gpt-4"
    )
    
    print(f"Result: {result}")


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

四、实时监控与告警系统

光有故障切换还不够，你需要一个强大的监控系统来先于用户发现问题。下面是基于Prometheus + Grafana的监控方案：

# ai_api_monitor.py
AI API实时监控系统
监控指标：延迟、错误率、成功率、成本

import time
import asyncio
from typing import Dict, Optional
from dataclasses import dataclass, field
from collections import deque
import statistics


@dataclass
class APMMetrics:
    """应用性能监控指标"""
    provider: str
    total_requests: int = 0
    successful_requests: int = 0
    failed_requests: int = 0
    timeout_requests: int = 0
    
    # 延迟指标 (毫秒)
    latencies: deque = field(default_factory=lambda: deque(maxlen=1000))
    
    # 成本追踪
    total_tokens: int = 0
    total_cost_usd: float = 0.0
    
    # 模型价格表 (USD/MTok)
    MODEL_PRICES = {
        "gpt-4": 8.0,
        "gpt-3.5-turbo": 2.0,
        "claude-3-sonnet": 15.0,
        "gemini-pro": 2.5,
        "deepseek-v3": 0.42,
        # HolySheep 2026年价格
        "holy-gpt-4": 8.0,
        "holy-claude": 15.0,
        "holy-gemini": 2.50,
        "holy-deepseek": 0.42,
    }
    
    def record_request(
        self,
        latency_ms: float,
        tokens: int = 0,
        success: bool = True,
        timeout: bool = False,
        model: str = "gpt-4"
    ):
        """记录单个请求"""
        self.total_requests += 1
        self.latencies.append(latency_ms)
        
        if timeout:
            self.timeout_requests += 1
        elif success:
            self.successful_requests += 1
            self.total_tokens += tokens
            # 计算成本
            price = self.MODEL_PRICES.get(model, 8.0)
            self.total_cost_usd += (tokens / 1_000_000) * price
        else:
            self.failed_requests += 1
    
    def get_success_rate(self) -> float:
        """计算成功率"""
        if self.total_requests == 0:
            return 0.0
        return (self.successful_requests / self.total_requests) * 100
    
    def get_latency_stats(self) -> Dict[str, float]:
        """获取延迟统计"""
        if not self.latencies:
            return {"p50": 0, "p95": 0, "p99": 0, "avg": 0}
        
        sorted_latencies = sorted(self.latencies)
        n = len(sorted_latencies)
        
        return {
            "p50": sorted_latencies[int(n * 0.50)],
            "p95": sorted_latencies[int(n * 0.95)],
            "p99": sorted_latencies[int(n * 0.99)],
            "avg": statistics.mean(sorted_latencies),
            "max": max(sorted_latencies),
            "min": min(sorted_latencies)
        }
    
    def get_cost_report(self) -> Dict:
        """获取月度成本报告"""
        return {
            "total_tokens": self.total_tokens,
            "total_cost_usd": round(self.total_cost_usd, 4),
            "cost_per_1m_tokens": round(
                self.total_cost_usd / (self.total_tokens / 1_000_000)
                if self.total_tokens > 0 else 0, 4
            )
        }


class AlertManager:
    """告警管理器 - SLA保障"""
    
    def __init__(self):
        self.rules = {
            "success_rate_min": 95.0,      # 成功率最低95%
            "latency_p99_max": 2000,        # P99延迟最高2000ms
            "error_rate_max": 5.0,          # 错误率最高5%
            "timeout_rate_max": 2.0,        # 超时率最高2%
        }
        self.alerts = []
    
    def check_metrics(self, metrics: APMMetrics) -> list:
        """检查是否触发告警"""
        triggered = []
        stats = metrics.get_latency_stats()
        success_rate = metrics.get_success_rate()
        error_rate = 100 - success_rate
        timeout_rate = (
            metrics.timeout_requests / metrics.total_requests * 100
            if metrics.total_requests > 0 else 0
        )
        
        checks = [
            ("success_rate", success_rate, "<", 
             self.rules["success_rate_min"]),
            ("latency_p99", stats["p99"], ">", 
             self.rules["latency_p99_max"]),
            ("error_rate", error_rate, ">", 
             self.rules["error_rate_max"]),
            ("timeout_rate", timeout_rate, ">", 
             self.rules["timeout_rate_max"]),
        ]
        
        for name, value, op, threshold in checks:
            should_alert = (
                (op == "<" and value < threshold) or
                (op == ">" and value > threshold)
            )
            
            if should_alert:
                triggered.append({
                    "alert": name,
                    "value": round(value, 2),
                    "threshold": threshold,
                    "severity": "critical"
                })
        
        return triggered


===== 监控使用示例 =====
def demo_monitoring():
    # 模拟供应商监控
    metrics = APMMetrics(provider="holysheep-gpt4")
    
    # 模拟1000次请求
    for i in range(1000):
        latency = abs(200 + (i % 100) * 10)  # 模拟延迟分布
        tokens = 500 + (i % 200)
        
        success = i % 50 != 0  # 2%失败率
        timeout = i % 100 == 0  # 1%超时率
        
        metrics.record_request(
            latency_ms=latency,
            tokens=tokens,
            success=success,
            timeout=timeout,
            model="deepseek-v3"  # 使用DeepSeek V3.2
        )
    
    # 输出报告
    print("=" * 50)
    print("AI API 监控报告")
    print("=" * 50)
    print(f"供应商: {metrics.provider}")
    print(f"总请求数: {metrics.total_requests}")
    print(f"成功请求: {metrics.successful_requests}")
    print(f"失败请求: {metrics.failed_requests}")
    print(f"超时请求: {metrics.timeout_requests}")
    print(f"成功率: {metrics.get_success_rate():.2f}%")
    print("\n延迟统计 (ms):")
    for k, v in metrics.get_latency_stats().items():
        print(f"  {k}: {v:.2f}")
    print("\n成本报告:")
    for k, v in metrics.get_cost_report().items():
        print(f"  {k}: {v}")
    
    # 告警检查
    alert_mgr = AlertManager()
    alerts = alert_mgr.check_metrics(metrics)
    
    print("\n告警状态:")
    if alerts:
        for alert in alerts:
            print(f"  ⚠️ {alert}")
    else:
        print("  ✅ 所有指标正常")


if __name__ == "__main__":
    demo_monitoring()

五、重试策略与熔断机制

好的重试策略可以让系统在瞬时故障中自愈，但糟糕的重试策略可能造成雪崩效应。下面是生产级的实现：

# resilient_client.py
弹性AI API客户端 - 指数退避 + 熔断器

import time
import random
from typing import Callable, Any, Optional
from dataclasses import dataclass
from enum import Enum
import asyncio


class CircuitState(Enum):
    CLOSED = "closed"      # 正常
    OPEN = "open"          # 熔断
    HALF_OPEN = "half_open"  # 半开


@dataclass
class CircuitBreaker:
    """熔断器实现"""
    name: str
    failure_threshold: int = 5      # 失败次数阈值
    recovery_timeout: float = 30.0  # 恢复等待时间(秒)
    success_threshold: int = 3       # 半开状态需要成功次数
    
    state: CircuitState = CircuitState.CLOSED
    failure_count: int = 0
    success_count: int = 0
    last_failure_time: float = 0
    
    def call(self, func: Callable, *args, **kwargs) -> Any:
        """执行带熔断保护的调用"""
        
        if self.state == CircuitState.OPEN:
            # 检查是否超时可以尝试恢复
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = CircuitState.HALF_OPEN
                self.success_count = 0
                print(f"[熔断器] {self.name} 进入半开状态")
            else:
                raise Exception(
                    f"熔断器 {self.name} 已打开，请在 "
                    f"{self.recovery_timeout}秒后重试"
                )
        
        try:
            result = func(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            self._on_failure()
            raise e
    
    def _on_success(self):
        """记录成功"""
        if self.state == CircuitState.HALF_OPEN:
            self.success_count += 1
            if self.success_count >= self.success_threshold:
                self.state = CircuitState.CLOSED
                self.failure_count = 0
                print(f"[熔断器] {self.name} 恢复正常")
        else:
            self.failure_count = 0
    
    def _on_failure(self):
        """记录失败"""
        self.failure_count += 1
        self.last_failure_time = time.time()
        
        if self.state == CircuitState.HALF_OPEN:
            self.state = CircuitState.OPEN
            print(f"[熔断器] {self.name} 重新打开")
        elif self.failure_count >= self.failure_threshold:
            self.state = CircuitState.OPEN
            print(f"[熔断器] {self.name} 打开(连续{self.failure_count}次失败)")


class ResilientAIClient:
    """弹性AI API客户端"""
    
    def __init__(self, base_url: str, api_key: str):
        self.base_url = base_url
        self.api_key = api_key
        self.circuit_breakers: dict[str, CircuitBreaker] = {}
        
        # 重试配置
        self.max_retries = 3
        self.base_delay = 1.0
        self.max_delay = 30.0
        self.exponential_base = 2
        self.jitter = True  # 添加随机抖动避免雷群效应
    
    def create_circuit_breaker(
        self,
        name: str,
        failure_threshold: int = 5
    ) -> CircuitBreaker:
        """创建熔断器"""
        cb = CircuitBreaker(
            name=name,
            failure_threshold=failure_threshold
        )
        self.circuit_breakers[name] = cb
        return cb
    
    def _calculate_delay(self, attempt: int) -> float:
        """计算退避延迟"""
        delay = min(
            self.base_delay * (self.exponential_base ** attempt),
            self.max_delay
        )
        
        if self.jitter:
            # 添加±25%随机抖动
            jitter_range = delay * 0.25
            delay += random.uniform(-jitter_range, jitter_range)
        
        return delay
    
    def request_with_retry(
        self,
        endpoint: str,
        payload: dict,
        model: str = "gpt-4"
    ) -> dict:
        """带重试和熔断的请求"""
        
        circuit_name = f"{self.base_url}:{endpoint}"
        cb = self.circuit_breakers.get(circuit_name)
        
        last_exception = None
        
        for attempt in range(self.max_retries):
            try:
                # 如果有熔断器则通过它执行
                if cb:
                    return cb.call(self._do_request, endpoint, payload)
                else:
                    return self._do_request(endpoint, payload)
                    
            except Exception as e:
                last_exception = e
                print(f"请求失败 (尝试 {attempt + 1}/{self.max_retries}): {e}")
                
                if attempt < self.max_retries - 1:
                    delay = self._calculate_delay(attempt)
                    print(f"等待 {delay:.2f}秒后重试...")
                    time.sleep(delay)
        
        raise Exception(
            f"请求在 {self.max_retries} 次尝试后失败: {last_exception}"
        )
    
    def _do_request(self, endpoint: str, payload: dict) -> dict:
        """实际执行HTTP请求"""
        import httpx
        
        # 这里简化实现，实际应该使用httpx等库
        url = f"{self.base_url}{endpoint}"
        
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        # 模拟请求
        print(f"发送请求到: {url}")
        
        # 模拟5%概率失败
        if random.random() < 0.05:
            raise Exception("模拟网络错误")
        
        return {"status": "success", "data": "response_data"}


===== 使用示例 =====
def demo_resilient_client():
    client = ResilientAIClient(
        base_url="https://api.holysheep.ai/v1",
        api_key="YOUR_HOLYSHEEP_API_KEY"
    )
    
    # 为特定端点创建熔断器
    client.create_circuit_breaker(
        name="chat/completions",
        failure_threshold=3
    )
    
    # 发送请求
    payload = {
        "model": "gpt-4",
        "messages": [{"role": "user", "content": "Hello!"}]
    }
    
    try:
        result = client.request_with_retry(
            "/chat/completions",
            payload
        )
        print(f"成功: {result}")
    except Exception as e:
        print(f"最终失败: {e}")


if __name__ == "__main__":
    demo_resilient_client()

六、故障应急响应流程

当故障发生时，时间就是生命。以下是我总结的5分钟黄金响应法则：

阶段一：发现阶段（0-30秒）

• 检查监控系统告警
• 确认影响范围（单一用户/部分用户/全部用户）
• 查看错误日志初步定位

阶段二：止血阶段（30秒-2分钟）

• 启用备用供应商路由
• 降级非核心功能
• 启动熔断器隔离故障

阶段三：恢复阶段（2-10分钟）

• 逐步切回主供应商
• 监控指标回归正常
• 确认服务SLA达标

阶段四：复盘阶段（事后）

• 生成故障报告
• 分析根本原因
• 更新应急响应文档

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 khóa API bị hết hạn, bị thu hồi hoặc bị chặn bởi nhà cung cấp.

# Kiểm tra và xử lý lỗi 401
import httpx

def validate_api_key(base_url: str, api_key: str) -> bool:
    """Kiểm tra tính hợp lệ của API key"""
    try:
        response = httpx.post(
            f"{base_url}/chat/completions",
            headers={
                "Authorization": f"Bearer {api_key}",
                "Content-Type": "application/json"
            },
            json={
                "model": "gpt-3.5-turbo",
                "messages": [{"role": "user", "content": "test"}],
                "max_tokens": 5
            },
            timeout=10.0
        )
        
        if response.status_code == 401:
            print("❌ API Key không hợp lệ hoặc đã hết hạn!")
            print("   Giải pháp: Truy cập https://www.holysheep.ai/register để lấy key mới")
            return False
        elif response.status_code == 200:
            print("✅ API Key hợp lệ")
            return True
        else:
            print(f"⚠️ Mã lỗi: {response.status_code}")
            return False
            
    except Exception as e:
        print(f"❌ Lỗi kết nối: {e}")
        return False

Sử dụng
is_valid = validate_api_key(
    base_url="https://api.holysheep.ai/v1",
    api_key="YOUR_HOLYSHEEP_API_KEY"
)

2. Lỗi 429 Rate Limit - Vượt giới hạn tốc độ

Mô tả lỗi: Gửi quá nhiều yêu cầu trong thời gian ngắn, vượt quota cho phép.

# Xử lý Rate Limit với Exponential Backoff
import time
import asyncio
from typing import Optional

class RateLimitHandler:
    """Xử lý thông minh khi gặp Rate Limit"""
    
    def __init__(self):
        self.retry_after: Optional[float] = None
        self.backoff_seconds = [1, 2, 4, 8, 16, 32]
    
    def handle_429(self, response_headers: dict) -> float:
        """Xử lý response 429 và trả về thời gian chờ"""
        
        # Ưu tiên Retry-After header
        if "retry-after" in response_headers:
            wait_time = float(response_headers["retry-after"])
            print(f"⏳ Server yêu cầu chờ {wait_time} giây")
            return wait_time
        
        # Kiểm tra x-rate-limit-reset
        if "x-rate-limit-reset" in response_headers:
            reset_time = float(response_headers["x-rate-limit-reset"])
            current_time = time.time()
            wait_time = max(reset_time - current_time, 1)
            print(f"⏳ Rate limit reset sau {wait_time:.0f} giây")
            return wait_time
        
        # Mặc định sử dụng exponential backoff
        print("⚠️ Không có thông tin retry, sử dụng backoff mặc định")
        return self.backoff_seconds[0]
    
    async def execute_with_retry(
        self,
        request_func,
        max_retries: int = 5
    ):
        """Thực thi request với tự động retry"""
        
        for attempt in range(max_retries):
            try:
                response = await request_func()
                
                if response.status_code == 429:
                    wait_time = self.handle_429(dict(response.headers))
                    print(f"🔄 Thử lại lần {attempt + 1} sau {wait_time}s...")
                    await asyncio.sleep(wait_time)
                    continue
                
                return response
                
            except Exception as e:
                if attempt < max_retries - 1:
                    wait = self.backoff_seconds[min(attempt, 5)]
                    print(f"❌ Lỗi: {e}, thử lại sau {wait}s...")
                    await asyncio.sleep(wait)
                else:
                    raise
        
        raise Exception("Đã vượt quá số lần thử lại tối đa")


Ví dụ sử dụng
async def demo_rate_limit():
    handler = RateLimitHandler()
    
    async def mock_request():
        import httpx
        return httpx.Response(
            429,
            headers={"retry-after": "5"},
            json={"error": "rate limit exceeded"}
        )
    
    try:
        response = await handler.execute_with_retry(mock_request)
        print(f"✅ Response: {response.json()}")
    except Exception as e:
        print(f"❌ Thất bại: {e}")

3. Lỗi Timeout - Yêu cầu hết thời gian chờ

Mô tả lỗi: Server phản hồi quá chậm hoặc không phản hồi.

# Xử lý Timeout với fallback đa cấp
import asyncio
import httpx
from dataclasses import dataclass
from typing import Optional, List

@dataclass
class FallbackProvider:
    """Nhà cung cấp dự phòng"""
    name: str
    base_url: str
    api_key: str
    priority: int = 1  # Ưu tiên (số càng nhỏ càng ưu tiên)
    timeout: float = 30.0

class MultiTierFallbackClient:
    """Client với fallback nhiều cấp"""
    
    def __init__(self, providers: List[FallbackProvider]):
        # Sắp xếp theo priority
        self.providers = sorted(providers, key=lambda p: p.priority)
    
    async def smart_request(
        self,
        payload: dict,
        model: str = "gpt-4"
    ) -> dict:
        """Gửi request với tự động chuyển provider"""
        
        errors = []
        
        for i, provider in enumerate(self.providers):
            try:
                print(f"🔄 Thử provider: {provider.name} (timeout: {provider.timeout}s)")
                
                async with httpx.AsyncClient(timeout=provider.timeout) as client:
                    response = await client.post(
                        f"{provider.base_url}/chat/completions",
                        headers={
                            "Authorization": f"Bearer {provider.api_key}",
                            "Content-Type": "application/json"
                        },
                        json={
                            "model": model,
                            "messages": payload.get("messages", []),
                            **payload.get("options", {})
                        }
                    )
                    
                    if response.status_code == 200:
                        result = response.json()
                        print(f"✅ Thành công với {provider.name}")
                        return {
                            "success": True,
                            "provider": provider.name,
                            "data": result
                        }
                    else:
                        errors.append(f"{provider.name}: {response.status_code}")
                        print(f"⚠️ {provider.name} trả lỗi: {response.status_code}")
                        
            except asyncio.TimeoutError:
                errors.append(f"{provider.name}: Timeout")
                print(f"⏰ {provider.name} timeout, thử provider tiếp theo...")
                
            except Exception as e:
                errors.append(f"{provider.name}: {str(e)}")
                print(f"❌ {provider.name} lỗi: {e}")
        
        # Tất cả provider đều thất bại
        return {
            "success": False,
            "error": "Tất cả providers đều không khả dụng",
            "details": errors
        }


Ví dụ sử dụng với HolySheep
async def demo_multi_tier_fallback():
    providers = [
        # Provider ưu tiên 1 - HolySheep GPT-4
        FallbackProvider(
            name="HolySheep-GPT4",
            base_url="https://api.holysheep.ai/v1",
            api_key="YOUR_HOLYSHEEP_API_KEY",
            priority=1,
            timeout=30.0
        ),
        # Provider dự phòng 2 - HolySheep DeepSeek
        FallbackProvider(
            name="HolySheep-DeepSeek",
            base_url="https://api.holysheep.ai/v1",
            api_key="YOUR_HOLYSHEEP_API_KEY",
            priority=2
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
Hướng Dẫn Toàn Diện: AI API Custom Development Với HolySheep
JWT Token Trong AI API: Hướng Dẫn Toàn Diện Từ A-Z
So Sánh Chi Tiết Chiến Lược Định Giá AI API 2026: HolySheep