AI API 重试与熔断机制最佳实践：构建弹性可靠的系统

AI API 重试与熔断机制最佳实践：构建弹性可靠的系统

AI API 天生不稳定：网络超时、服务端错误、限速限制……任何一个环节出问题都会导致应用失败。本文介绍 AI API 的重试与熔断机制最佳实践，帮你构建弹性可靠的系统。

为什么需要重试和熔断？

重试机制设计

#### 1. 哪些错误应该重试？

代码示例（Python）：

def should_retry(error):
    """判断是否应该重试"""
    if isinstance(error, (ConnectionError, TimeoutError)):
        return True
    
    if hasattr(error, 'status_code'):
        status_code = error.status_code
        if status_code == 429:
            return True
        if 500 <= status_code < 600:
            return True
    
    return False

#### 2. 重试策略

指数退避（Exponential Backoff）：

第1次重试：等待 1s
第2次重试：等待 2s
第3次重试：等待 4s
第4次重试：等待 8s
第5次重试：等待 16s
...

带抖动的指数退避（Exponential Backoff with Jitter）：

避免多个客户端同时重试导致的「惊群效应」：

import random
import time

def exponential_backoff_with_jitter(
    attempt, 
    base_delay=1, 
    max_delay=60, 
    jitter_factor=0.5
):
    """
    带抖动的指数退避
    """
    # 指数退避
    delay = min(base_delay * (2 ** attempt), max_delay)
    
    # 添加抖动
    jitter = random.uniform(-jitter_factor * delay, jitter_factor * delay)
    delay = max(0, delay + jitter)
    
    return delay

# 使用
for attempt in range(5):
    try:
        response = client.chat.completions.create(...)
        break
    except Exception as e:
        if should_retry(e) and attempt < 4:
            delay = exponential_backoff_with_jitter(attempt)
            time.sleep(delay)
        else:
            raise

#### 3. 429 限速错误的特殊处理

429 错误通常包含 Retry-After 响应头，告诉我们多久后可以重试：

def get_retry_after(error):
    """从 429 响应中获取重试时间"""
    if hasattr(error, 'response') and error.response:
        retry_after = error.response.headers.get('Retry-After')
        if retry_after:
            return int(retry_after)
    
    # 如果没有 Retry-After，使用默认值
    return 10

#### 4. 重试次数限制

熔断机制设计

#### 什么是熔断器？

熔断器有三种状态：

Closed（关闭）→ 正常放行请求
    ↓（失败率超过阈值）
Open（打开）→ 直接拒绝请求
    ↓（等待一段时间后）
Half-Open（半打开）→ 尝试少量请求
    ↓（成功）→ 回到 Closed
    ↓（失败）→ 回到 Open

#### 熔断器参数

#### Python 实现熔断器

import time
from collections import deque
from enum import Enum

class CircuitState(Enum):
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"

class CircuitBreaker:
    def __init__(
        self,
        failure_threshold=0.5,
        window_size=60,
        min_requests=10,
        open_timeout=30,
        half_open_max_calls=3
    ):
        self.failure_threshold = failure_threshold
        self.window_size = window_size
        self.min_requests = min_requests
        self.open_timeout = open_timeout
        self.half_open_max_calls = half_open_max_calls
        
        self.state = CircuitState.CLOSED
        self.failure_times = deque()
        self.success_times = deque()
        self.open_at = None
        self.half_open_calls = 0
    
    def _cleanup_old_entries(self):
        """清理窗口外的记录"""
        now = time.time()
        cutoff = now - self.window_size
        
        while self.failure_times and self.failure_times[0] < cutoff:
            self.failure_times.popleft()
        
        while self.success_times and self.success_times[0] < cutoff:
            self.success_times.popleft()
    
    def _get_failure_rate(self):
        """计算失败率"""
        self._cleanup_old_entries()
        total = len(self.failure_times) + len(self.success_times)
        if total < self.min_requests:
            return 0.0
        return len(self.failure_times) / total
    
    def _can_attempt_call(self):
        """判断是否可以尝试调用"""
        if self.state == CircuitState.CLOSED:
            return True
        
        if self.state == CircuitState.OPEN:
            now = time.time()
            if now - self.open_at >= self.open_timeout:
                self.state = CircuitState.HALF_OPEN
                self.half_open_calls = 0
                return True
            return False
        
        if self.state == CircuitState.HALF_OPEN:
            return self.half_open_calls < self.half_open_max_calls
        
        return False
    
    def _on_success(self):
        """成功回调"""
        now = time.time()
        
        if self.state == CircuitState.HALF_OPEN:
            self.half_open_calls += 1
            if self.half_open_calls >= self.half_open_max_calls:
                self.state = CircuitState.CLOSED
                self.failure_times.clear()
                self.success_times.clear()
        else:
            self.success_times.append(now)
    
    def _on_failure(self):
        """失败回调"""
        now = time.time()
        
        if self.state == CircuitState.HALF_OPEN:
            self.state = CircuitState.OPEN
            self.open_at = now
        else:
            self.failure_times.append(now)
            failure_rate = self._get_failure_rate()
            if failure_rate >= self.failure_threshold:
                self.state = CircuitState.OPEN
                self.open_at = now
    
    def execute(self, func, fallback=None):
        """执行函数"""
        if not self._can_attempt_call():
            if fallback is not None:
                return fallback()
            raise Exception("Circuit breaker is OPEN")
        
        try:
            result = func()
            self._on_success()
            return result
        except Exception as e:
            self._on_failure()
            raise

# 使用
breaker = CircuitBreaker(
    failure_threshold=0.5,
    window_size=60,
    min_requests=10,
    open_timeout=30
)

def call_ai_api():
    return client.chat.completions.create(...)

try:
    response = breaker.execute(call_ai_api)
except Exception as e:
    print(f"请求失败: {e}")

完整的弹性客户端

#### 结合重试 + 熔断 + 退避

import time
import random
from functools import wraps

def resilient_api_call(
    max_retries=3,
    base_delay=1,
    max_delay=30,
    circuit_breaker=None
):
    """
    弹性 API 调用装饰器
    """
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            last_exception = None
            
            for attempt in range(max_retries + 1):
                try:
                    # 检查熔断器
                    if circuit_breaker:
                        return circuit_breaker.execute(
                            lambda: func(*args, **kwargs)
                        )
                    else:
                        return func(*args, **kwargs)
                    
                except Exception as e:
                    last_exception = e
                    
                    # 判断是否应该重试
                    if attempt >= max_retries:
                        break
                    
                    if not should_retry(e):
                        break
                    
                    # 计算延迟
                    delay = exponential_backoff_with_jitter(
                        attempt,
                        base_delay=base_delay,
                        max_delay=max_delay
                    )
                    
                    # 429 特殊处理
                    if hasattr(e, 'status_code') and e.status_code == 429:
                        retry_after = get_retry_after(e)
                        delay = max(delay, retry_after)
                    
                    time.sleep(delay)
            
            raise last_exception
        
        return wrapper
    return decorator

# 使用
breaker = CircuitBreaker()

@resilient_api_call(
    max_retries=3,
    circuit_breaker=breaker
)
def call_ai_with_resilience(model, messages):
    return client.chat.completions.create(
        model=model,
        messages=messages
    )

try:
    response = call_ai_with_resilience("gpt-4o-mini", [...])
except Exception as e:
    print(f"最终失败: {e}")

多供应商容错

当一个供应商失败时，自动切换到另一个：

class MultiProviderClient:
    def __init__(self, providers):
        self.providers = providers  # [{"name": "openai", "client": ..., "priority": 1}]
        self.circuit_breakers = {
            p["name"]: CircuitBreaker() for p in providers
        }
    
    def call(self, *args, **kwargs):
        # 按优先级排序
        sorted_providers = sorted(
            self.providers,
            key=lambda p: p["priority"]
        )
        
        last_exception = None
        
        for provider in sorted_providers:
            name = provider["name"]
            client = provider["client"]
            breaker = self.circuit_breakers[name]
            
            if not breaker._can_attempt_call():
                continue
            
            try:
                return breaker.execute(
                    lambda: client.chat.completions.create(*args, **kwargs)
                )
            except Exception as e:
                last_exception = e
                continue
        
        raise last_exception or Exception("All providers failed")

最佳实践

#### 1. 重试 + 熔断配合使用

#### 2. 监控重试和熔断

关键指标：

• 重试次数/比例
• 熔断状态变化
• 失败率趋势
• 多供应商切换次数

#### 3. 合理设置参数

• 重试次数不要太多（2-5 次）
• 熔断阈值不要太敏感（50% 比较合理）
• 窗口大小适中（60s-5min）

#### 4. 提供降级方案

当所有重试都失败时：

• 返回缓存的结果
• 返回默认值
• 提示用户稍后重试

总结

AI API 的重试与熔断机制是构建弹性系统的关键：

• ✅ 只对可重试的错误进行重试
• ✅ 使用带抖动的指数退避
• ✅ 熔断器防止级联故障
• ✅ 多供应商容错提高可用性
• ✅ 监控指标持续优化

建议：

1. 先从重试开始，逐步添加熔断

2. 使用成熟的库（tenacity、pybreaker）

3. 持续监控，根据数据调整参数

4. 考虑多供应商架构，避免单点故障

可在本站查看更多 AI API 中转平台，找到更稳定可靠的 AI API 服务。