AI Agent容错设计：挂了也能继续跑

容错的三层体系：让AI Agent在生产环境稳定运行

为什么Agent需要容错？

在生产环境中，AI Agent会频繁遇到API超时、模型返回异常、工具调用失败等问题。没有容错设计，一个错误就会导致整个工作流崩溃。

第一层：调用容错（指数退避重试）

import time
import random
from functools import wraps

def retry_with_backoff(max_retries=3, base_delay=2):
    """指数退避重试装饰器"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_retries - 1:
                        raise
                    delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
                    print(f"第{attempt+1}次重试，等待{delay:.1f}秒... 错误: {e}")
                    time.sleep(delay)
            return None
        return wrapper
    return decorator

@retry_with_backoff(max_retries=3, base_delay=2)
def call_llm_api(prompt):
    # 假设API可能超时
    resp = openai_client.chat.completions.create(
        model="gpt-4o", messages=[{"role": "user", "content": prompt}],
        timeout=30
    )
    return resp.choices[0].message.content

# 效果：API临时故障时自动重试，2s→4s→8s递增等待
# 成功率从单次95%提升到3次重试后的99.99%

第二层：降级策略（模型降级）

class ModelFallback:
    """多级模型降级：主模型→备选→本地模型"""
    MODELS = [
        {"provider": "openai", "model": "gpt-4o", "priority": 1},
        {"provider": "deepseek", "model": "deepseek-chat", "priority": 2},
        {"provider": "local", "model": "qwen2.5:7b", "priority": 3},
    ]

    def query(self, prompt, max_attempts=3):
        errors = []
        for model in self.MODELS[:max_attempts]:
            try:
                if model['provider'] == 'openai':
                    return call_openai(prompt)
                elif model['provider'] == 'deepseek':
                    return call_deepseek(prompt)
                else:
                    return call_local_ollama(prompt)
            except Exception as e:
                errors.append(f"{model['model']}失败: {e}")
                print(f"降级到{model['model']}...")
                continue
        raise Exception(f"所有模型均失败: {errors}")

# 降级延迟对比
# | 层级 | 模型 | 延迟 | 质量 |
# |:----|:----|:----|:----|
# | 主 | GPT-4o | 1-2s | 100% |
# | 备选 | DeepSeek | 0.8-1.5s | 95% |
# | 兜底 | Qwen2.5:7B | 3-5s | 80% |

第三层：超时保护和断路保护

import asyncio

class CircuitBreaker:
    """断路器模式：连续失败N次后熔断一段时间"""
    def __init__(self, failure_threshold=3, recovery_timeout=60):
        self.failure_count = 0
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.last_failure_time = 0
        self.state = "CLOSED"  # CLOSED / OPEN / HALF_OPEN

    async def call(self, func, *args, **kwargs):
        if self.state == "OPEN":
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = "HALF_OPEN"
            else:
                raise Exception("熔断中，请求被拒绝")

        try:
            result = await asyncio.wait_for(func(*args, **kwargs), timeout=30)
            if self.state == "HALF_OPEN":
                self.state = "CLOSED"
                self.failure_count = 0
            return result
        except Exception as e:
            self.failure_count += 1
            self.last_failure_time = time.time()
            if self.failure_count >= self.failure_threshold:
                self.state = "OPEN"
                print(f"熔断器打开！连续{self.failure_count}次失败")
            raise e

常见故障场景数据

最佳实践

1. 超时时间不要超过30秒，太长会阻塞后续任务
2. 重试要加抖动（jitter），避免所有请求同时重试造成雪崩
3. 熔断恢复后首次请求只放行部分流量（半开模式）