openai-agents-python 沙箱重试机制全解析：深入 `agents.sandbox.util.retry` 模块源码与实战

本文以 openai-agents-python 仓库中沙箱（sandbox）子系统的重试工具模块 agents.sandbox.util.retry 为核心，完整讲解其设计动机、核心 API、退避（backoff）算法、异常链判定工具，以及它在 Docker、Blaxel、E2B、Modal、Vercel 等多家沙箱提供方实现中的真实用法。读完本文，你将能够理解该框架如何处理瞬时故障（transient failure），并能在自己的异步代码中复用它提供的 retry_async 装饰器与异常判定工具。

该模块的 API 参考文档位于 docs/ref/sandbox/util/retry.md，其源码实现位于 src/agents/sandbox/util/retry.py，对应的单元测试位于 tests/sandbox/test_retry.py。文档参考页由 docs/scripts/generate_ref_files.py 自动生成（即 # \Retry`标题加::: agents.sandbox.util.retry` 的 mkdocstrings 指令），真正的技术细节全部沉淀在源码模块中，本文即以此为据展开。

为什么沙箱子系统需要一个独立的重试模块

在 openai-agents-python 中，沙箱（Sandbox）用于为 Agent 提供隔离、可复现、可快照的执行环境，涉及 Docker 本地沙箱以及 Blaxel、Cloudflare、Daytona、E2B、Modal、Runloop、Vercel 等远程托管沙箱提供方（对应 src/agents/extensions/sandbox/ 下的各子目录）。这类场景有两个显著特点：

1. 外部依赖多：创建沙箱、持久化工作区（persist_workspace）、打包上传 tar 归档等操作都依赖网络与远端服务；
2. 瞬时故障不可避免：HTTP 5xx、网关超时、asyncio.TimeoutError、SDK 抛出的 aiohttp.ClientError 等，往往是短暂的、重试即可恢复的。

如果每次遇到这类故障就直接失败，Agent 的执行体验会很不稳定。因此框架把一套"可配置的重试逻辑 + 异常链判定工具"抽成独立模块 src/agents/sandbox/util/retry.py，供各沙箱提供方复用。从源码结构看，重试策略（什么时候重试、间隔多久、最多几次、退避方式）被设计为可插拔的装饰器参数，而"异常到底算不算瞬时故障"则由各提供方通过 lambda 自由定义。

核心 API 一览

整个模块只依赖标准库（asyncio、functools、inspect、enum），不引入任何第三方依赖，所有公开符号如下：

符号	类型	作用
BackoffStrategy	str, Enum	退避策略枚举：FIXED / LINEAR / EXPONENTIAL
DEFAULT_TRANSIENT_RETRY_INTERVAL_S	float	默认重试间隔，0.25 秒
DEFAULT_TRANSIENT_RETRY_MAX_ATTEMPT	int	默认最大尝试次数，3
DEFAULT_TRANSIENT_RETRY_BACKOFF	BackoffStrategy	默认退避策略，EXPONENTIAL
TRANSIENT_HTTP_STATUS_CODES	frozenset[int]	视为瞬时故障的 HTTP 状态码集合：{500, 502, 503, 504}
iter_exception_chain(exc)	生成器	沿 __cause__ / __context__ 遍历整条异常链（防环）
exception_chain_contains_type(exc, types)	bool	异常链中是否存在指定类型的异常
exception_chain_has_status_code(exc, codes)	bool	异常链中是否存在携带指定 HTTP 状态码的异常
retry_async(...)	装饰器	为异步函数注入重试逻辑

其中 BackoffStrategy 继承自 str, Enum 并重写了 __str__ 返回枚举值本身（src/agents/sandbox/util/retry.py#L14-L20），因此 str(BackoffStrategy.EXPONENTIAL) 得到字符串 "exponential"，方便序列化与日志输出——这一点在单元测试 test_retry_async_retries_with_expected_backoff_and_async_hook 中也有断言覆盖。

retry_async 装饰器：参数语义与校验规则

retry_async 的完整签名（src/agents/sandbox/util/retry.py#L65-L75）如下：

def retry_async(
    *,
    interval: float = DEFAULT_TRANSIENT_RETRY_INTERVAL_S,
    max_attempt: int = DEFAULT_TRANSIENT_RETRY_MAX_ATTEMPT,
    backoff: BackoffStrategy = DEFAULT_TRANSIENT_RETRY_BACKOFF,
    retry_if: Callable[..., bool],
    on_retry: Callable[..., object] | None = None,
) -> ...

注意所有参数都是关键字参数（*），其中 retry_if 为必填。各参数语义：

• interval（默认 0.25 秒）：基础重试间隔，也是 FIXED 策略下的固定等待时长；
• max_attempt（默认 3）：最大尝试次数（含首次调用），即最多失败 max_attempt - 1 次；
• backoff（默认 EXPONENTIAL）：退避策略，决定每次重试前的等待时长如何随尝试次数增长；
• retry_if：判定函数，形如 retry_if(exc, *args, **kwargs)，接收捕获到的异常以及被装饰函数的原始参数，返回 True 表示该异常属于"可重试的瞬时故障"；
• on_retry：可选回调，在每次决定重试之后、asyncio.sleep 之前被调用，形如 on_retry(exc, attempt, max_attempt, delay_s, *args, **kwargs)；它可以是普通函数，也可以是协程函数（框架通过 inspect.isawaitable 识别并 await）。

参数校验

装饰器在创建阶段即做三组校验（src/agents/sandbox/util/retry.py#L83-L95），非法配置直接抛 ValueError：

if max_attempt < 1:
    raise ValueError("max_attempt must be >= 1")
if interval < 0:
    raise ValueError("interval must be >= 0")
if backoff not in {FIXED, LINEAR, EXPONENTIAL}:
    raise ValueError("backoff must be BackoffStrategy.FIXED, ...")

对应的测试 test_retry_async_validates_configuration（tests/sandbox/test_retry.py）分别验证了 max_attempt=0、interval=-1 以及传入非法枚举值 "quadratic" 三种情况都会抛出带相应消息的 ValueError。

重试循环与退避算法

被装饰的函数被替换为如下循环逻辑（src/agents/sandbox/util/retry.py#L100-L125）：

for attempt in range(1, max_attempt + 1):
    try:
        return await fn(*args, **kwargs)
    except Exception as exc:
        if attempt >= max_attempt or not retry_if(exc, *args, **kwargs):
            raise
        if backoff is BackoffStrategy.EXPONENTIAL:
            delay_s = interval * (2 ** (attempt - 1))
        elif backoff is BackoffStrategy.LINEAR:
            delay_s = interval * attempt
        else:
            delay_s = interval
        if on_retry is not None:
            hook_result = on_retry(exc, attempt, max_attempt, delay_s, *args, **kwargs)
            if inspect.isawaitable(hook_result):
                await hook_result
        await asyncio.sleep(delay_s)

三种退避策略的计算方式（模块 docstring 与实现一致）：

策略	第 attempt 次重试前等待时长	说明
FIXED	interval	恒定延迟，每次都等同样长的时间
LINEAR	interval * attempt	线性增长，第一次重试等 1 倍间隔，第二次等 2 倍，依此类推
EXPONENTIAL	interval * 2 ** (attempt - 1)	指数翻倍，第 n 次重试等待 2^(n-1) 倍间隔，对网络抖动最友好

以 interval=0.5、max_attempt=3 为例，三次尝试之间两次等待的时长，测试 test_retry_async_retries_with_expected_backoff_and_async_hook 给出了精确断言（tests/sandbox/test_retry.py）：

• FIXED：[0.5, 0.5]
• LINEAR：[0.5, 1.0]
• EXPONENTIAL：[0.5, 1.0]

该测试用 monkeypatch.setattr(asyncio, "sleep", fake_sleep) 把真实休眠替换成记录延迟的假函数，从而在毫秒级验证了三种策略的等待序列；同时验证了 on_retry 异步钩子被调用两次、参数为 (attempt, max_attempt, delay_s)，且最终装饰器返回成功结果 "ok:sandbox"。

值得注意的细节：

• retry_if 返回 False 时立即 raise，不会调用 asyncio.sleep。测试 test_retry_async_stops_without_sleep_when_retry_is_rejected 用"一旦 sleep 就抛 AssertionError"的假函数验证了这一点——函数只执行一次，重试被判定拒绝后原异常原样抛出（tests/sandbox/test_retry.py）；
• 循环外的 raise AssertionError("unreachable") 是类型系统的收尾保护，正常情况下不会执行；
• 装饰器使用 functools.wraps(fn)，保留被装饰函数的元信息（__name__、__doc__ 等）；
• 采用 ParamSpec 与 TypeVar 做类型标注，保证装饰器不会破坏被装饰函数的调用签名类型检查。

异常链判定工具：穿透 __cause__ / __context__

真实故障往往不是孤立异常：SDK 抛出的 aiohttp.ClientError 可能被包装成带 __cause__ 的 SandboxError，而 HTTP 状态码可能藏在异常的 status_code、http_code 或 response.status_code 属性里。因此模块提供了三个配套工具。

iter_exception_chain：安全的异常链遍历

def iter_exception_chain(exc: BaseException) -> Iterable[BaseException]:
    seen: set[int] = set()
    current: BaseException | None = exc
    while current is not None and id(current) not in seen:
        yield current
        seen.add(id(current))
        current = getattr(current, "__cause__", None) or getattr(current, "__context__", None)

它沿 __cause__（优先）或 __context__ 逐层向上遍历整条异常链，并用 id() 集合防止异常链出现环导致死循环。测试 test_iter_exception_chain_supports_context_and_stops_on_cycles（tests/sandbox/test_retry.py）验证了两点：外层异常的 __context__ 指向内层异常时能依次遍历到两者；构造"互为 cause"的环形链时遍历能正确终止且不重复。

exception_chain_contains_type：按类型判定

def exception_chain_contains_type(exc, error_types) -> bool:
    if not error_types:
        return False
    return any(isinstance(candidate, error_types) for candidate in iter_exception_chain(exc))

空元组直接返回 False；否则对链上每个异常做 isinstance 判定。这解决了"包装层异常类型不对、但根源异常类型正确"的常见问题。

exception_chain_has_status_code：按 HTTP 状态码判定

def exception_chain_has_status_code(exc, status_codes) -> bool:
    for candidate in iter_exception_chain(exc):
        for value in (
            getattr(candidate, "status_code", None),
            getattr(candidate, "http_code", None),
            getattr(getattr(candidate, "response", None), "status_code", None),
        ):
            if isinstance(value, int) and value in status_codes:
                return True
    return False

它同时探测三种常见的状态码存放位置：异常自身的 status_code 属性、http_code 属性，以及 response.status_code（兼容封装了 HTTP 响应对象的 SDK 异常）。测试用自定义的 _ErrorWithHttpMetadata 分别构造了三种携带方式，并验证 500、502、504 均能被识别、503 不会被误判（tests/sandbox/test_retry.py）。

实战用法：如何在自己的异步代码中使用

参考各沙箱提供方源码中的真实用法，复用一个最小示例。假设你的函数会调用一个不稳定的远端接口：

import asyncio
from agents.sandbox.util.retry import (
    BackoffStrategy,
    exception_chain_contains_type,
    exception_chain_has_status_code,
    retry_async,
    TRANSIENT_HTTP_STATUS_CODES,
)

@retry_async(
    interval=0.25,
    max_attempt=3,
    backoff=BackoffStrategy.EXPONENTIAL,
    retry_if=lambda exc, *args, **kwargs: (
        exception_chain_contains_type(exc, (asyncio.TimeoutError,))
        or exception_chain_has_status_code(exc, TRANSIENT_HTTP_STATUS_CODES)
    ),
    on_retry=lambda exc, attempt, max_attempt, delay_s, *args, **kwargs: (
        print(f"attempt {attempt}/{max_attempt} failed: {exc!r}, retrying in {delay_s}s")
    ),
)
async def persist_snapshot(remote_url: str) -> bytes:
    # ... 调用远端接口，可能抛 TimeoutError 或携带 5xx 状态码的异常
    ...

要点归纳：

1. retry_if 是"要不要重试"的唯一决策入口——它既决定是否重试，也天然过滤掉业务性错误（如 4xx、校验失败），这类错误应直接上抛，不做无意义的重试；
2. retry_if 会收到被装饰函数的原始参数，因此可以结合参数做更精细的判断（例如根据目标环境决定是否重试）；
3. on_retry 支持同步与异步两种形态，适合在重试前做日志、指标上报或清理工作；
4. 默认值就是为"瞬时故障"调好的：间隔 0.25 秒、最多 3 次、指数退避，多数场景可直接采用。

仓库中的真实应用场景

该模块并非孤立的工具代码，而是被沙箱子系统的多家提供方广泛复用，可作为学习"何时该重试"的最佳范本：

• Docker 本地沙箱：src/agents/sandbox/sandboxes/docker.py 中的 persist_workspace 方法使用 retry_async，retry_if 仅依据 exception_chain_has_status_code(exc, TRANSIENT_HTTP_STATUS_CODES) 判定是否重试（docker.py）——即工作区持久化遇到 5xx 这类瞬时 HTTP 错误时自动重试；
• Blaxel：src/agents/extensions/sandbox/blaxel/sandbox.py 的 persist_workspace 将 asyncio.TimeoutError 与瞬时 HTTP 状态码合并作为重试条件（blaxel/sandbox.py）；
• Daytona：src/agents/extensions/sandbox/daytona/sandbox.py 对持久化命令 _run_persist_workspace_command 同时检查"可重试的提供方错误类型"与瞬时 HTTP 状态码（daytona/sandbox.py）；
• E2B：src/agents/extensions/sandbox/e2b/sandbox.py 在错误分类逻辑中直接用 TRANSIENT_HTTP_STATUS_CODES 判定 "transient_http_status"，并结合 exception_chain_contains_type 识别可重试的提供方超时；
• Modal：src/agents/extensions/sandbox/modal/sandbox.py 用 iter_exception_chain 逐层检查 SandboxError.retryable 标志，再结合 ExecTransportError 类型与瞬时 HTTP 状态码综合判定（modal/sandbox.py），并用 retry_async 包装 _persist_workspace_via_tar；
• Vercel：src/agents/extensions/sandbox/vercel/sandbox.py 定义 _vercel_provider_retryability 对提供方错误分类后，用 retry_async 包装 _create_sandbox_with_retry 与文件写入操作（vercel/sandbox.py）；
• Runloop：src/agents/extensions/sandbox/runloop/sandbox.py 使用 iter_exception_chain 遍历异常链并匹配"可重试错误类型"集合。

可以看到，各家提供方的共同模式是：把"分类判定"与"重试执行"分离——分类逻辑（retry_if lambda）可以非常复杂、因提供方而异，而重试的节流、次数控制与退避计算统一由 retry_async 完成，职责清晰、无重复代码。

测试覆盖与质量保障

tests/sandbox/test_retry.py 是理解该模块行为契约的最佳入口，覆盖了四类场景：

1. 异常链遍历：__context__ 链遍历、环形链防死循环（test_iter_exception_chain_supports_context_and_stops_on_cycles）；
2. 判定工具：类型匹配、三种 HTTP 状态码属性位置探测、负例不误判（test_exception_chain_helpers_detect_types_and_status_codes）；
3. 配置校验：三种非法参数均抛 ValueError（test_retry_async_validates_configuration）；
4. 重试行为：三种退避策略的精确等待序列、异步 on_retry 钩子参数、重试被拒绝时不 sleep 且原样抛错（两个 @pytest.mark.asyncio 测试）。

这套测试使用 monkeypatch 替换 asyncio.sleep，使重试等待在测试中被"短路"，既保证了测试速度，又精确断言了延迟计算逻辑——这也是异步重试类代码值得借鉴的测试手法。

总结与使用建议

agents.sandbox.util.retry 是 openai-agents-python 沙箱子系统的"容错基础设施"：它以极小的 API 表面积（一个枚举、三个默认常量、三个工具函数、一个装饰器），为所有沙箱提供方统一解决了"瞬时故障自动重试"这一横切关注点。使用时的关键决策可以概括为三点：

• 选对重试条件：用 exception_chain_contains_type 匹配超时/传输类异常，用 exception_chain_has_status_code + TRANSIENT_HTTP_STATUS_CODES 匹配 5xx 瞬时错误，不要对业务性错误重试；
• 选对退避策略：网络抖动场景优先 EXPONENTIAL（默认），需要稳定节奏可改用 FIXED 或 LINEAR；
• 善用 on_retry 钩子：在重试间隙输出日志或上报指标，能显著提升故障可观测性。

如果你正在为 Agent 沙箱、远端 SDK 调用或任何不可靠的外部依赖编写容错逻辑，这个模块的源码与测试本身就是一份高质量的实现参考。