以前遇到要重试执行的函数时，要么直接调用两遍，要么写一个简单的类似于下面的装饰器。

def retry_when_except(func):
    def wrapper(*args, **kwargs):
        try:
            print("首次执行")
            res = func(*args, **kwargs)
            return res
        except:
            try:
                print("再次执行")
                res = func(*args, **kwargs)
                return res
            except:
                raise Exception("执行失败")

    return wrapper()


@retry_when_except
def func1():
    res = 5 / 0
    # raise Exception("hello")
    return res


if __name__ == '__main__':
    res = func1()
    print(res)

后面发现这样确实太 low 了，一搜索发现有很多现成的重试模块，比如 retrying, tenacity 等。其中的 retrying 简洁好用，针对其示例学习一下功能用法。

retrying

安装

在开始使用 retrying 模块之前，需要先安装它。可以通过以下命令安装：

pip install retrying

默认无限重试

retrying 模块提供了一个装饰器 @retry，可以轻松地为函数添加重试逻辑。以下是一个简单的示例（异常后会无限重试）：

from retrying import retry

@retry
def might_fail():
    print("Retrying...")
    raise Exception("This function fails!")

try:
    might_fail()
except Exception as e:
    print(f"Caught exception: {e}")

Retrying...
Retrying...
Retrying...
...

最大重试次数

可以通过 stop_max_attempt_number 参数设置最大重试次数，超过最大重试次数后仍失败则正常报错：

@retry(stop_max_attempt_number=3)
def might_fail():
    print("Retrying...")
    raise Exception("This function fails!")

Retrying...
Retrying...
Retrying...
Exception: This function fails!

设置重试间隔

使用 wait_fixed 参数可以设置每次重试之间的固定间隔（毫秒）：

import time

from retrying import retry


@retry(wait_fixed=2000, stop_max_attempt_number=3)
def might_fail():
    print(f"{time.time()} Retrying...")
    raise Exception("This function fails!")


might_fail()

1776664234.2184367 Retrying...
1776664236.2195823 Retrying...
1776664238.2207365 Retrying...
Exception: This function fails!

根据异常类型重试

通过 retry_on_exception 参数可以指定仅在特定异常发生时重试：

from retrying import retry


def is_io_error(exception):
    return isinstance(exception, IOError)


@retry(retry_on_exception=is_io_error, stop_max_attempt_number=3)
def might_fail():
    print("Retrying...")
    raise IOError("This is an IOError!")


might_fail()

Retrying...
Retrying...
Retrying...
OSError: This is an IOError!

根据返回值重试

使用 retry_on_result 参数可以根据函数的返回值决定是否重试：

import random
from retrying import retry


def is_not_ok(result):
    return result != "OK"


@retry(retry_on_result=is_not_ok, stop_max_attempt_number=3)
def might_fail():
    print("Retrying...")
    return random.choice(["OK", "FAIL"])


might_fail()

组合多个条件

可以组合多个条件来实现更复杂的重试逻辑：

@retry(
    stop_max_attempt_number=3,
    wait_fixed=1000,
    retry_on_exception=lambda e: isinstance(e, IOError)
)
def might_fail():
    print("Retrying...")
    raise IOError("This is an IOError!")

随机化重试间隔

通过 wait_random_min 和 wait_random_max 参数可以设置随机化的重试间隔：

@retry(
    wait_random_min=1000,
    wait_random_max=5000,
    stop_max_attempt_number=3
)
def might_fail():
    print("Retrying...")
    raise Exception("This function fails!")

指数退避策略

使用 wait_exponential_multiplier 和 wait_exponential_max 参数可以实现指数退避策略。

import time

from retrying import retry


@retry(
    wait_exponential_multiplier=1000,
    stop_max_attempt_number=5
)
def might_fail():
    print(f"{time.time()} Retrying...")
    raise Exception("This function fails!")


might_fail()

1777456304.9110534 Retrying...
1777456306.9129317 Retrying...
1777456310.9137948 Retrying...
1777456318.9147003 Retrying...
1777456334.915977 Retrying...

tenacity

tenacity 实际上是 retrying 库的一个fork。主要是因为 retrying 早已无人维护。因此，tenacity 基本上是目前 Python 生态中实现重试逻辑的事实标准。很多大型开源项目基本上都用 tenacity 来作为重试处理模块。

安装

Tenacity 是一个 Python 重试库，用于优雅地处理可能失败的操作。安装命令如下：

pip install tenacity

默认无限重试

使用 @retry 装饰器自动重试函数：

from tenacity import retry


@retry
def might_fail():
    print("Retrying...")
    raise Exception("This function fails!")


try:
    might_fail()
except Exception as e:
    print(f"Caught exception: {e}")

最大重试次数

通过 stop 参数控制停止条件，例如最多重试 5 次：

from tenacity import retry, stop_after_attempt


@retry(stop=stop_after_attempt(3))
def might_fail():
    print("Retrying...")
    raise Exception("This function fails!")


try:
    might_fail()
except Exception as e:
    print(f"Caught exception: {e}")

Retrying...
Retrying...
Retrying...
Caught exception: RetryError[<Future at 0x1aeb25edf90 state=finished raised Exception>]

设置重试间隔

使用 wait 参数配置等待策略，如固定间隔（秒），指数退避：

import time

from tenacity import retry, wait_fixed, stop_after_attempt


@retry(wait=wait_fixed(3), stop=stop_after_attempt(3))
def might_fail():
    print(f"{time.time()} Retrying...")
    raise Exception("This function fails!")


might_fail()

1777453915.9286058 Retrying...
1777453918.9300616 Retrying...
1777453921.9304826 Retrying...
Exception: This function fails!

根据异常类型重试

通过 retry_if_exception_type 指定仅在特定异常时重试：

from tenacity import retry, retry_if_exception_type, stop_after_attempt


@retry(retry=retry_if_exception_type(IOError), stop=stop_after_attempt(3))
def might_fail():
    print("Retrying...")
    raise IOError("This is an IOError!")


might_fail()

Retrying...
Retrying...
Retrying...
OSError: This is an IOError!

组合多个条件

同时配置停止条件、等待策略和重试条件：

from tenacity import retry, stop_after_attempt, wait_fixed

@retry(stop=stop_after_attempt(3), wait=wait_fixed(2))
def might_fail():
    print("Retry with fixed wait and max attempts")
    raise Exception

回调函数

使用 after 回调在每次重试后执行操作：

import logging

from tenacity import retry, after_log, stop_after_attempt

logging.basicConfig(level=logging.INFO)


@retry(after=after_log(logging.getLogger(), logging.INFO), stop=stop_after_attempt(3))
def might_fail():
    print("Log after each retry")
    raise Exception


might_fail()

Log after each retry
Log after each retry
Log after each retry
INFO:root:Finished call to '__main__.might_fail' after 0(s), this was the 1st time calling it.
INFO:root:Finished call to '__main__.might_fail' after 0(s), this was the 2nd time calling it.
INFO:root:Finished call to '__main__.might_fail' after 0(s), this was the 3rd time calling it.

当然，除了使用 after_log，我们也可以仿照去使用自定义的函数。

from tenacity import retry, stop_after_attempt


def log_it(retry_state) -> None:
    print(f"retry state info : {retry_state}")


@retry(after=log_it, stop=stop_after_attempt(3))
def might_fail():
    print("Log after each retry")
    raise Exception


might_fail()

Log after each retry
retry state info : <RetryCallState 1885969272208: attempt #1; slept for 0.0; last result: failed (Exception )>
Log after each retry
retry state info : <RetryCallState 1885969272208: attempt #2; slept for 0.0; last result: failed (Exception )>
Log after each retry
retry state info : <RetryCallState 1885969272208: attempt #3; slept for 0.0; last result: failed (Exception )>

指数退避策略

比如使用指数为 2 的等待策略，后续每次重试的等待间隔会是前一次的 2 倍时间。

import time

from tenacity import retry, wait_exponential, stop_after_attempt


@retry(wait=wait_exponential(2), stop=stop_after_attempt(5))
def might_fail():
    print(f"{time.time()} Retrying...")
    raise Exception("This function fails!")


might_fail()

1777456029.1546159 Retrying...
1777456031.1555977 Retrying...
1777456035.156208 Retrying...
1777456043.15678 Retrying...
1777456059.1572459 Retrying...

异步函数支持

Tenacity 也支持异步函数的重试：

import asyncio

from tenacity import retry, stop_after_attempt


@retry(stop=stop_after_attempt(3))
async def async_might_fail():
    print("Async retry")
    raise Exception


asyncio.run(async_might_fail())

Async retry
Async retry
Async retry
Logo
AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

cover

我测试了一个0.9B模型，它竟然能自动区分多人讲话）

avatar AI Agent技术社区
cover

从一张商品图到正式上架：实测店匠 Shoplazza Athena，AI Agent 不只是聊天框

avatar AI Agent技术社区
cover

【AI Agent】《2026年9月 AI Agent全栈开发技术选型专项面试宝典》

avatar AI Agent技术社区