AI Agent Engineer

AI Agent Engineer

异常、超时、重试与资源管理

9 min read

异常、超时、重试与资源管理

本节目标

把失败当作接口的一部分：能区分失败类型，设置总时限，只重试可能恢复且允许重复执行的操作，并保证文件、连接和任务在成功或失败后都被收尾。

先画出失败边界

Agent 工具不是“调用一个函数就结束”。一次操作可能经过输入校验、读取本地文件、访问外部服务、解析结果和持久化输出。每层应只处理自己真正能恢复的异常。

失败类型示例通常处理方式
输入/契约错误缺字段、状态值非法立即拒绝；告诉调用方如何修正
瞬时外部失败短暂连接中断、服务临时不可用在满足重试条件时有限重试
永久外部失败无权限、资源不存在、参数非法不重试；返回稳定错误类别
程序缺陷未覆盖的状态、断言失败记录证据并失败，不能伪装成输入错误
取消/总时限耗尽用户停止、上游 deadline 到期尽快传播并清理资源

异常类应表达调用者可以采取的动作，而不只是底层库名：

class TaskError(Exception):
    """任务处理失败的基类。"""
 
 
class InvalidTask(TaskError):
    """调用者修正输入后才可能成功。"""
 
 
class TemporaryDependencyFailure(TaskError):
    """外部依赖短暂失败，可能允许重试。"""

转换异常时保留因果链：

try:
    text = path.read_text(encoding="utf-8")
except OSError as exc:
    raise TaskError(f"无法读取输入文件: {path.name}") from exc

raise ... from exc 让日志保留根因；面向用户的消息仍可隐藏本机绝对路径或敏感响应。

try 的范围要小

只包住你预期失败的语句。范围过大会把自身 bug 误判成可恢复错误。

try:
    raw = path.read_text(encoding="utf-8")
except FileNotFoundError as exc:
    raise InvalidTask("输入文件不存在") from exc
else:
    return parse_document(raw)  # 解析缺陷不会被错当成文件不存在
  • except：只捕获能解释或恢复的具体异常；
  • else：只在没有异常时执行；
  • finally：无论成功失败都必须做的收尾；
  • 上下文管理器 with / async with：优先用于文件、锁、连接和临时资源。

不要使用 except Exception: pass。如果顶层为了生成统一错误响应而捕获宽异常，也必须记录根因、返回失败状态并让监控可见。

超时是预算，不是一串独立数字

连接超时、单次请求超时和整个任务的总 deadline 不同。如果每次重试都重新获得完整超时，三次“10 秒请求”加退避可能让用户等待远超预期。

工程上先定义总预算，再为每一步分配剩余时间：

任务总时限 20 s
├─ 输入校验 ≤ 1 s
├─ 外部调用（含所有重试）≤ 15 s
└─ 解析、写入与收尾预留 4 s

标准库没有一个能强制中断任意同步函数的通用安全开关。真正的网络、数据库或 SDK 调用必须使用其明确的超时参数；对异步代码使用 asyncio.timeout()。不要以为 Future.result(timeout=...) 会自动终止后台正在执行的阻塞工作。

何时允许重试

同时满足以下条件才重试：

  1. 失败被明确归类为瞬时失败；
  2. 操作本身是幂等的，或使用幂等键避免重复副作用；
  3. 尚未超过最大尝试次数和总 deadline；
  4. 上游没有取消；
  5. 每次失败都可观察，并最终返回最后根因。

读取通常比“扣款、发邮件、创建工单”更容易安全重试。后者即使客户端看到超时，服务端也可能已经成功；没有幂等键时盲目重试会重复执行。

下面是可测试的同步重试骨架。sleep 被注入，测试时无需真的等待：

from collections.abc import Callable
from typing import TypeVar
 
T = TypeVar("T")
 
 
def retry_transient(
    operation: Callable[[], T],
    *,
    attempts: int,
    sleep: Callable[[float], None],
    base_delay: float = 0.2,
) -> T:
    if attempts < 1:
        raise ValueError("attempts 必须至少为 1")
 
    for attempt in range(1, attempts + 1):
        try:
            return operation()
        except TemporaryDependencyFailure:
            if attempt == attempts:
                raise
            sleep(base_delay * (2 ** (attempt - 1)))
    raise AssertionError("循环应当已经返回或抛出")

生产系统通常还加入随机抖动（jitter），避免大量客户端同时重试。随机源也应可注入，以便测试确定性。HTTP 的具体状态码、Retry-After、认证和幂等语义在 API 中学习；本节只负责 Python 控制流。

这个骨架没有实现总 deadline

它只演示“有限次数 + 可测试退避”。真实外部调用还要把剩余时间传给底层超时参数，并在每次调用和休眠前用单调时钟检查总 deadline；同步函数若不支持中断，外层计时不能强制停止它。取消传播也必须由实际同步/异步运行模型单独设计。

收尾与取消

  • 文件优先 with path.open(...) as file:；
  • 临时目录使用 tempfile.TemporaryDirectory()；
  • 先写临时文件、校验成功后再原子替换，避免半成品；
  • 不在 finally 中抛出新的无关异常覆盖原始失败；
  • 异步任务收到 CancelledError 时先做必要清理，再重新抛出，不能吞掉取消。

练习

  1. 为上面的 retry_transient 写测试：前两次失败、第三次成功；断言调用次数和退避序列为 0.2, 0.4。
  2. 再写“永久错误不重试”和“尝试次数耗尽”的测试。
  3. 选择一个有副作用的操作，说明它需要什么幂等键或补偿机制后才能重试。
  4. 为一个 10 秒总预算画出单次调用、退避和收尾的最坏时间线。

自测

  • 能解释输入错误、瞬时失败、永久失败和程序缺陷为什么不能统一重试。
  • 能说明单次超时与任务总 deadline 的区别。
  • 能证明某个操作是否幂等，而不是只说“应该没问题”。
  • 能保留异常因果链，同时避免向调用方泄露敏感细节。
  • 能指出资源在成功、异常和取消三条路径上的关闭位置。

相关概念与下一步

参考资料

获取日期：2026-07-14。

Backlinks