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

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

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

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

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} 完全不可用，无可用缓存")

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

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

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

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

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

# 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