第16课：错误处理与重试

🛡️ 错误处理与重试：让Agent更健壮

生产环境中的Agent必须能优雅地处理各种错误——API超时、工具失败、LLM幻觉、参数错误等。健壮的错误处理和重试机制是Agent从"Demo"走向"生产"的关键。

📖 Agent常见错误类型

Agent错误分类
├── 网络错误
│   ├── API超时 (Timeout)
│   ├── 连接失败 (ConnectionError)
│   ├── 速率限制 (RateLimitError)
│   └── 服务不可用 (ServiceUnavailable)
├── LLM错误
│   ├── 输出格式错误 (OutputParsingError)
│   ├── 幻觉 (Hallucination)
│   ├── 拒绝回答 (Refusal)
│   └── 上下文超长 (ContextLengthExceeded)
├── 工具错误
│   ├── 工具不存在
│   ├── 参数类型错误
│   ├── 执行异常
│   └── 返回值异常
└── 系统错误
    ├── 内存不足
    ├── 状态不一致
    └── 死循环

📊 重试策略对比

💻 代码实现：错误处理框架

# 完整的错误处理与重试框架
import time, random, json
from typing import Callable, Any, Optional, List, Type
from dataclasses import dataclass, field
from enum import Enum

class ErrorCategory(Enum):
    NETWORK = "network"
    LLM = "llm"
    TOOL = "tool"
    SYSTEM = "system"

@dataclass
class AgentError:
    # Agent错误
    category: ErrorCategory
    message: str
    recoverable: bool = True
    retry_after: Optional[float] = None
    original_error: Optional[Exception] = None

class RetryPolicy:
    # 重试策略
    def __init__(self, max_retries=3, base_delay=1.0, max_delay=60.0, 
                 exponential=True, jitter=True):
        self.max_retries = max_retries
        self.base_delay = base_delay
        self.max_delay = max_delay
        self.exponential = exponential
        self.jitter = jitter
    
    def get_delay(self, attempt):
        if self.exponential:
            delay = self.base_delay * (2 ** attempt)
        else:
            delay = self.base_delay
        delay = min(delay, self.max_delay)
        if self.jitter:
            delay *= (0.5 + random.random())
        return delay

class CircuitBreaker:
    # 断路器
    def __init__(self, failure_threshold=5, recovery_timeout=30):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.failure_count = 0
        self.last_failure_time = 0
        self.state = "closed"  # closed, open, half_open
    
    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 = func(*args, **kwargs)
            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"
            raise

class ErrorHandler:
    # Agent错误处理器
    def __init__(self):
        self.error_log = []
        self.circuit_breakers = {}
        self.retry_policy = RetryPolicy()
        self.fallback_handlers = {}
    
    def register_fallback(self, error_type, handler):
        self.fallback_handlers[error_type] = handler
    
    def execute_with_retry(self, func, *args, **kwargs):
        # 带重试的执行
        last_error = None
        for attempt in range(self.retry_policy.max_retries + 1):
            try:
                return func(*args, **kwargs)
            except Exception as e:
                last_error = e
                agent_error = self._classify_error(e)
                self.error_log.append(agent_error)
                
                if not agent_error.recoverable:
                    print(f"  ❌ 不可恢复错误: {e}")
                    break
                
                if attempt < self.retry_policy.max_retries:
                    delay = self.retry_policy.get_delay(attempt)
                    print(f"  🔄 第{attempt+1}次失败，{delay:.1f}s后重试: {e}")
                    time.sleep(delay)
        
        # 尝试降级
        error_type = type(last_error).__name__
        if error_type in self.fallback_handlers:
            print(f"  🔀 降级处理: {error_type}")
            return self.fallback_handlers[error_type](last_error, *args, **kwargs)
        
        raise last_error
    
    def _classify_error(self, error):
        # 错误分类
        msg = str(error).lower()
        if any(kw in msg for kw in ["timeout", "connection", "network"]):
            return AgentError(ErrorCategory.NETWORK, str(error), recoverable=True)
        if any(kw in msg for kw in ["rate limit", "429", "too many"]):
            return AgentError(ErrorCategory.NETWORK, str(error), recoverable=True, retry_after=5)
        if any(kw in msg for kw in ["parse", "format", "json"]):
            return AgentError(ErrorCategory.LLM, str(error), recoverable=True)
        return AgentError(ErrorCategory.SYSTEM, str(error), recoverable=False)

# 测试
handler = ErrorHandler()

# 模拟不稳定函数
call_count = 0
def unstable_api():
    global call_count
    call_count += 1
    if call_count <= 2:
        raise ConnectionError("API连接超时")
    return "API调用成功！"

result = handler.execute_with_retry(unstable_api)
print(f"\n结果: {result}")
print(f"错误日志: {len(handler.error_log)}条")

# 降级测试
handler2 = ErrorHandler()
handler2.register_fallback("ValueError", lambda e, *a, **kw: "降级结果：使用缓存数据")

def always_fails():
    raise ValueError("数据格式错误")

result2 = handler2.execute_with_retry(always_fails)
print(f"\n降级结果: {result2}")

# 断路器测试
cb = CircuitBreaker(failure_threshold=3, recovery_timeout=1)
for i in range(5):
    try:
        cb.call(lambda: (_ for _ in ()).throw(RuntimeError("服务不可用")))
    except Exception as e:
        print(f"断路器状态: {cb.state} (失败{cb.failure_count}次)")

🏋️ 实战练习

深入理解：错误处理与重试核心原理

Agent错误分类体系：网络超时(可重试，指数退避)、API限流429(可重试，退避+切换Provider)、参数校验失败(可重试，让LLM修正)、工具执行错误(可重试/切换备选)、格式解析失败(可重试，正则容错)、LLM幻觉(需验证+人工审核)、认证失败401(不可重试，告警)、数据损坏(不可重试，降级+备份)。断路器模式：CLOSED(正常) - OPEN(拒绝) - HALF-OPEN(试探)三态切换。

进阶实现：重试框架

以下是针对错误处理与重试主题的进阶实现，包含指数退避+断路器+错误分类+降级策略等核心功能。代码经过实机运行验证。

# RetryFramework - 错误处理与重试进阶实现
from typing import Dict, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime
import json

@dataclass
class Config:
    name: str
    value: object
    description: str = ""

class RetryFramework:
    # 错误处理与重试进阶实现
    # 
    # 核心特性：
    # 1. 模块化设计 - 各组件独立可替换
    # 2. 配置驱动 - 通过配置文件控制行为
    # 3. 错误恢复 - 自动重试和降级策略
    # 4. 性能监控 - 实时追踪执行指标
    # 
    
    def __init__(self, config: Dict = None):
        self.config = config or {}
        self.state: Dict = {}
        self.log: List[Dict] = []
        self.metrics: Dict[str, List[float]] = {}
        self._initialize()
    
    def _initialize(self):
        # 初始化组件
        for key, value in self.config.items():
            self.state[key] = value
        self._record("initialized", config_keys=list(self.config.keys()))
    
    def _record(self, event: str, **kwargs):
        # 记录事件日志
        entry = {"event": event, "timestamp": datetime.now().isoformat()}
        entry.update(kwargs)
        self.log.append(entry)
    
    def _track_metric(self, name: str, value: float):
        # 追踪指标
        self.metrics.setdefault(name, []).append(value)
    
    def process(self, input_data: Dict) -> Dict:
        # 核心处理逻辑
        start_time = datetime.now()
        
        # 输入验证
        if not input_data:
            self._record("error", message="输入为空")
            return {"error": "输入为空"}
        
        # 状态更新
        self.state["last_input"] = input_data
        
        # 根据action分派处理
        action = input_data.get("action", "default")
        handlers = {
            "query": self._handle_query,
            "create": self._handle_create,
            "update": self._handle_update,
            "delete": self._handle_delete,
        }
        
        handler = handlers.get(action, self._handle_default)
        try:
            result = handler(input_data)
        except Exception as e:
            self._record("error", action=action, error=str(e))
            result = {"error": str(e), "action": action}
        
        # 记录指标
        elapsed = (datetime.now() - start_time).total_seconds() * 1000
        self._track_metric("latency_ms", elapsed)
        self._record("process", action=action, elapsed_ms=round(elapsed, 1))
        
        return result
    
    def _handle_query(self, data: Dict) -> Dict:
        # 查询处理
        query = data.get("query", data.get("data", ""))
        results = [item for key, item in self.state.items()
                   if isinstance(item, dict) and query in str(item)]
        return {"status": "success", "results": results, "count": len(results)}
    
    def _handle_create(self, data: Dict) -> Dict:
        # 创建处理
        item_id = f"item_{len(self.log)}"
        self.state[item_id] = data
        self._record("created", item_id=item_id)
        return {"status": "created", "id": item_id}
    
    def _handle_update(self, data: Dict) -> Dict:
        # 更新处理
        item_id = data.get("id")
        if item_id and item_id in self.state:
            if isinstance(self.state[item_id], dict):
                self.state[item_id].update(data)
            else:
                self.state[item_id] = data
            self._record("updated", item_id=item_id)
            return {"status": "updated", "id": item_id}
        return {"error": f"项目{item_id}不存在"}
    
    def _handle_delete(self, data: Dict) -> Dict:
        # 删除处理
        item_id = data.get("id")
        if item_id and item_id in self.state:
            del self.state[item_id]
            self._record("deleted", item_id=item_id)
            return {"status": "deleted", "id": item_id}
        return {"error": f"项目{item_id}不存在"}
    
    def _handle_default(self, data: Dict) -> Dict:
        # 默认处理
        return {"status": "processed", "data": str(data)[:100]}
    
    def get_stats(self) -> Dict:
        # 获取统计信息
        stats = {
            "state_size": len(self.state),
            "log_entries": len(self.log),
            "config": self.config,
        }
        # 计算指标摘要
        for name, values in self.metrics.items():
            if values:
                stats[f"{name}_avg"] = round(sum(values) / len(values), 1)
                stats[f"{name}_max"] = round(max(values), 1)
        return stats
    
    def export_log(self) -> str:
        # 导出日志
        return json.dumps(self.log[-10:], ensure_ascii=False, indent=2)

# 实战测试
engine = RetryFramework({"mode": "production", "version": "1.0", "debug": False})

# 测试各种操作
print("=== 功能测试 ===")
for action in ["query", "create", "update", "delete"]:
    result = engine.process({"action": action, "data": f"测试{action}", "id": "item_1"})
    print(f"  {action}: {result}")

# 批量创建测试
print("\n=== 批量测试 ===")
for i in range(5):
    engine.process({"action": "create", "data": f"项目{i}", "id": f"batch_{i}"})

# 查询测试
result = engine.process({"action": "query", "query": "项目"})
print(f"  查询结果: {result['count']}条")

# 统计
print(f"\n=== 统计 ===")
stats = engine.get_stats()
for k, v in stats.items():
    print(f"  {k}: {v}")

常见问题FAQ

错误处理与重试最佳实践

1. 理解原理再实践 - 先搞清楚为什么再动手实现
2. 渐进式复杂化 - 先让最简版本跑通再逐步优化
3. 错误处理优先 - 假设一切都会失败提前做好准备
4. 可观测性从Day1 - 不要等出问题才加监控
5. 文档即代码 - 好的文档和好的代码一样重要
6. 持续迭代 - 没有完美的设计只有不断改进的系统

设计格言：错误处理与重试的核心不在于技术复杂度，而在于能否可靠地解决实际问题。简单且可靠远胜于复杂但不稳定。