Reborn 的技术博客

Agent 容错机制 —— 重试与超时，让 Agent 更扛造

2026-06-23·AI, Agent, Retry, Timeout

核心增量：RetryFunc.py（88 行）+ CallFunc.py（40 行）+ 主循环 2 处调用点替换

一、问题：Agent 太脆弱了

场景 A: LLM API 偶尔 429（限流）→ Agent 直接崩溃
场景 B: LLM API 返回 503 → 已经跑了 4 轮的对话全部白费
场景 C: 工具函数卡死 60 秒 → Agent 永远等不到响应

生产环境黄金法则：任何一次远程调用都可能失败，任何一次工具执行都可能超时。

两个正交的容错需求：

  • 重试：LLM API 瞬时故障时自动恢复
  • 超时：工具函数卡死时强制中断

二、两层容错职责划分

LLM 调用:  stream_llm_call → with_retry（指数退避重试）
工具执行:  active_tool_map → call_with_timeout（线程池超时）

错误分类:
  429/5xx/网络超时 → 重试
  400/401/403     → 不重试
  工具卡死         → 超时中断

两条线互不干扰，不共享任何状态或逻辑。

三、重试机制：with_retry

3.1 核心原则：只重试该重试的

RETRYABLE_STATUSES = {429, 500, 502, 503, 504}
RETRYABLE_ERROR_CODES = {"rate_limit_exceeded", "server_error", "internal_server_error",
                         "service_unavailable", "api_connection_error", "api_timeout"}

def is_retryable(error: Exception) -> bool:
    if hasattr(error, "status_code"):
        if error.status_code in RETRYABLE_STATUSES:
            return True
        if error.status_code in (400, 401, 403):
            return False
    if hasattr(error, "code") and error.code in RETRYABLE_ERROR_CODES:
        return True
    if isinstance(error, (ConnectionError, TimeoutError)):
        return True
    return False

| 策略 | 场景 | 理由 | | --- | --- | --- | | 重试 | 429 / 5xx / 网络超时 | 瞬时故障，下次大概率成功 | | 不重试 | 400 / 401 / 403 | 请求本身有问题 | | 保守不重试 | 未知错误 | 不知道是什么，不冒险 |

3.2 指数退避

def with_retry(fn, max_retries=3, base_delay=1.0, max_delay=30.0, label=""):
    last_error = None

    for attempt in range(max_retries + 1):      # 1 次原始 + 3 次重试
        try:
            return fn()
        except Exception as e:
            last_error = e
            if not is_retryable(e):
                raise
            if attempt == max_retries:
                break

            delay = min(base_delay * (2 ** attempt), max_delay)
            time.sleep(delay)

    raise last_error

退避序列（base_delay=1.0）：原始 0s → 重试1 1s → 重试2 2s → 重试3 4s，上限 30s。

四、工具超时：call_with_timeout

为什么用线程池而不是 asyncio

工具函数是同步的（requests.get、sqlite3.connect），强行改 async 需要用户重写所有工具。

DEFAULT_TOOL_TIMEOUT = 30  # 秒

def call_with_timeout(func, args=(), kwargs=None, timeout=DEFAULT_TOOL_TIMEOUT):
    kwargs = kwargs or {}

    with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
        future = executor.submit(func, *args, **kwargs)
        try:
            result = future.result(timeout=timeout)
            return str(result)
        except concurrent.futures.TimeoutError:
            future.cancel()
            return f"工具执行超时（{timeout}秒），已取消执行"
        except Exception as e:
            return f"工具执行错误: {e}"

超时后的行为

  1. future.result(timeout) 抛 TimeoutError
  2. future.cancel() 尝试取消
  3. 返回字符串格式错误信息，Agent 看到后可以调整策略
🔧 http_request({"url":"https://slow-api.com"})
  → 工具执行超时（30秒），已取消执行
🧠 工具超时了，我换个方式试试...

五、设计精要

重试和超时是正交的

LLM 的瞬时故障靠重试兜底，工具的长时间阻塞靠超时兜底，互不干扰。

错误返回是字符串而非异常

Agent 收到的是一条普通文本（就像工具正常返回一样），可以基于它自主决策。

保守优于激进

未知错误不重试，避免放大问题。

#AI#Agent#Retry#Timeout