OpenClaw错误处理实战：优雅降级、重试机制与熔断器设计

import asyncio
import random
from typing import Callable, TypeVar, Optional
T = TypeVar("T")
# 通用重试函数：支持指数退避 + 随机抖动
async def retry_with_backoff(
    fn: Callable[..., T],            # 被重试的异步函数
    max_retries: int = 3,            # 最大重试次数
    base_delay: float = 1.0,         # 初始延迟（秒）
    max_delay: float = 60.0,         # 最大延迟上限（秒）
    jitter: float = 0.4,             # 抖动幅度，0.0-1.0
    retryable: tuple = (Exception,), # 可重试的异常类型
) -> Optional[T]:
    """带指数退避与抖动的通用重试函数。"""
    for attempt in range(max_retries + 1):
        try:
            return await fn()
        except retryable as e:
            if attempt == max_retries:
                print(f"[Retry] 达到最大重试次数 {max_retries}，放弃执行")
                return None
            raw = min(base_delay * (2 ** attempt), max_delay)
            delta = random.uniform(-jitter, jitter) * raw
            delay = max(0.1, raw + delta)
            print(f"[Retry] 第 {attempt + 1} 次失败: {e}，{delay:.1f}s 后重试")
            await asyncio.sleep(delay)
    return None

from dataclasses import dataclass, field
from typing import Any, Callable, Optional
from enum import Enum
class ToolStatus(Enum):
    HEALTHY = "healthy"
    DEGRADED = "degraded"
    UNAVAILABLE = "unavailable"
# 工具三级降级：主工具 -> 降级方案 -> 缓存
@dataclass
class ToolWithFallback:
    """带三级降级的工具包装器：主工具 → 降级方案 → 缓存。"""
    name: str
    primary_fn: Callable
    fallback_fn: Optional[Callable] = None
    cache_fn: Optional[Callable] = None
    status: ToolStatus = ToolStatus.HEALTHY
    failure_count: int = 0
    degradation_threshold: int = 3
    async def execute(self, **kwargs) -> Any:
        if self.status == ToolStatus.UNAVAILABLE:
            return await self._try_cache(**kwargs)
        try:
            result = await self.primary_fn(**kwargs)
            self.failure_count = 0
            self.status = ToolStatus.HEALTHY
            return result
        except Exception as e:
            self.failure_count += 1
            if self.failure_count >= self.degradation_threshold:
                self.status = ToolStatus.DEGRADED
            print(f"[Tool:{self.name}] 主工具失败: {e}")
        if self.status == ToolStatus.DEGRADED and self.fallback_fn:
            try:
                return await self.fallback_fn(**kwargs)
            except Exception as e:
                print(f"[Tool:{self.name}] 降级失败: {e}")
                self.status = ToolStatus.UNAVAILABLE
        return await self._try_cache(**kwargs)
    async def _try_cache(self, **kwargs) -> Any:
        if self.cache_fn:
            result = await self.cache_fn(**kwargs)
            if result:
                return result
        raise RuntimeError(f"工具 {self.name} 完全不可用，无可用缓存")

import time, threading
from enum import Enum
from dataclasses import dataclass, field
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  # OPEN 后等待多久进入 HALF_OPEN
    half_open_max_calls: int = 3    # 半开状态允许的试探请求数
    success_threshold: int = 2      # 半开状态连续成功多少次恢复 CLOSED
    state: CircuitState = CircuitState.CLOSED
    failure_count: int = 0
    success_count: int = 0
    half_open_calls: int = 0
    last_failure_time: float = 0.0
    _lock: threading.Lock = field(default_factory=threading.Lock)
    def can_execute(self) -> bool:
        with self._lock:
            if self.state == CircuitState.CLOSED: return True
            if self.state == CircuitState.OPEN:
                if time.time() - self.last_failure_time >= self.recovery_timeout:
                    self.state, self.half_open_calls, self.success_count = CircuitState.HALF_OPEN, 0, 0
                    return True
                return False
            if self.half_open_calls < self.half_open_max_calls:
                self.half_open_calls += 1
                return True
            return False
    def record(self, success: bool):
        with self._lock:
            if success:
                if self.state == CircuitState.HALF_OPEN:
                    self.success_count += 1
                    if self.success_count >= self.success_threshold:
                        self.state, self.failure_count = CircuitState.CLOSED, 0
                else:
                    self.failure_count = 0
            else:
                self.failure_count += 1
                self.last_failure_time = time.time()
                self.state = CircuitState.OPEN

import json
import time
import traceback
from dataclasses import dataclass, asdict, field
from typing import Optional
@dataclass
class ErrorLog:
    """结构化错误日志，便于后续检索与聚合分析。"""
    timestamp: float
    error_type: str
    error_code: str
    component: str
    message: str
    stack_trace: Optional[str] = None
    retry_count: int = 0
    recovery_action: str = ""
    recovery_result: str = ""
    context: dict = field(default_factory=dict)
    def to_json(self) -> str:
        return json.dumps(asdict(self), ensure_ascii=False, indent=2)
def log_error(
    error: Exception,
    error_type: str,
    component: str,
    recovery_action: str = "",
    retry_count: int = 0,
    **context
):
    log = ErrorLog(
        timestamp=time.time(),
        error_type=error_type,
        error_code=getattr(error, "code", type(error).__name__),
        component=component,
        message=str(error),
        stack_trace=traceback.format_exc(),
        retry_count=retry_count,
        recovery_action=recovery_action,
        context=context,
    )
    print(log.to_json())
    return log

# openclaw-error-handling.yaml
retry:
  default:
    max_retries: 3
    base_delay: 1.0
    max_delay: 60.0
    jitter: 0.4
    retryable_errors: ["429", "503", "timeout", "connection_reset"]
  overrides:
    openai_api: { max_retries: 5, base_delay: 2.0, jitter: 0.5 }
    search_tool: { max_retries: 2, base_delay: 0.5, jitter: 0.3 }
model_fallback:
  chain:
    - model: gpt-4o
      trigger: "error_rate > 0.3 OR latency_p95 > 10s"
    - model: gpt-4o-mini
      trigger: "error_rate > 0.5 OR latency_p95 > 15s"
    - model: claude-3-5-haiku
      trigger: "consecutive_errors >= 5"
    - model: local-qwen2.5-7b
      trigger: "all_remote_unavailable"
circuit_breakers:
  openai_api: { failure_threshold: 5, recovery_timeout: 30, half_open_max_calls: 2 }
  search_tool: { failure_threshold: 3, recovery_timeout: 60, half_open_max_calls: 1 }
  feishu_api: { failure_threshold: 10, recovery_timeout: 15, half_open_max_calls: 3 }
feature_degradation:
  levels:
    - name: full
      max_priority: 3
      description: "所有功能可用"
    - name: reduced
      max_priority: 1
      trigger: "system_load > 0.8"
      description: "关闭语音和图片生成"
    - name: minimal
      max_priority: 0
      trigger: "system_load > 0.95 OR error_rate > 0.5"
      description: "仅保留核心对话"
alerts:
  feishu_webhook: "${FEISHU_ALERT_WEBHOOK}"
  rules:
    - condition: "circuit_state == 'open'"
      severity: critical
      cooldown: 60
    - condition: "model_degraded"
      severity: high
      cooldown: 300
    - condition: "consecutive_failures >= 5"
      severity: medium
      cooldown: 600