Python使用Tenacity一行代码实现自动重试详解

# 一个基本的 API 调用函数
from openai import OpenAI, APIConnectionError

def translate_text(text: str) -> str:
    message = [
        {'role': 'system', 'content': '您是一名顶级的字幕翻译引擎。'},
        {'role': 'user', 'content': f'<INPUT>{text}</INPUT>'},
    ]
    model = OpenAI(api_key="YOUR_API_KEY", base_url="...") 

    try:
        response = model.chat.completions.create(model="gpt-4o", messages=message)
        if response.choices:
            return response.choices[0].message.content.strip()
        raise RuntimeError("API未返回有效结果")
    except APIConnectionError as e:
        print(f"网络连接失败: {e}。需要重试...")
        raise # 程序在这里崩溃
    except Exception as e:
        print(f"发生其他错误: {e}")
        raise

# 手动实现重试
for attempt in range(3):
    try:
        # ... API 调用逻辑 ...
        return response.choices[0].message.content.strip()
    except APIConnectionError as e:
        print(f"第 {attempt + 1} 次尝试失败: {e}")
        if attempt == 2: # 检查是否是最后一次尝试
            raise
    # ... 对其他异常也要重复写相似的逻辑 ...

from tenacity import retry

@retry
def translate_text(text: str) -> str:
    # ... 内部逻辑和之前完全一样，无需任何改动 ...

from tenacity import retry, stop_after_attempt

# 总共尝试 3 次
@retry(stop=stop_after_attempt(3))
def translate_text(text: str) -> str:
    # ...

from tenacity import stop_after_delay

# 总次数达到 5 次或总耗时超过 30 秒，就停止
@retry(stop=(stop_after_attempt(5) | stop_after_delay(30)))
def translate_text(text: str) -> str:
    # ...

from tenacity import retry, wait_fixed

# 每次重试前都等待 2 秒
@retry(wait=wait_fixed(2))
def translate_text(text: str) -> str:
    # ...

from tenacity import wait_exponential

# 首次重试等 2^1=2s, 之后等 4s, 8s... 最多等到 10s
@retry(wait=wait_exponential(multiplier=1, min=2, max=10))
def translate_text(text: str) -> str:
    # ...

# 错误的方式！这无法按预期工作
@retry(retry=retry_if_not_exception_type(AuthenticationError | PermissionDeniedError))

from openai import AuthenticationError, PermissionDeniedError

# 正确的方式！使用元组
@retry(retry=retry_if_not_exception_type((AuthenticationError, PermissionDeniedError)))

from tenacity import RetryCallState

def my_error_callback(retry_state: RetryCallState):
    # retry_state 对象包含了这次重试的所有信息
    print(f"所有 {retry_state.attempt_number} 次尝试均失败！")
    return "默认的翻译结果或错误提示"

@retry(stop=stop_after_attempt(3), retry_error_callback=my_error_callback)
def translate_text(text: str) -> str:
    # ...

def return_last_value(retry_state: RetryCallState):
    # 危险！这会重新抛出异常！
    return "失败：" + retry_state.outcome.result()

def return_last_value(retry_state: RetryCallState):
    # 安全！这只会返回异常对象
    last_exception = retry_state.outcome.exception()
    return f"经过 {retry_state.attempt_number} 次尝试后失败。最后一次错误是: {last_exception}"

class TTS:
    def __init__(self, name):
        self.name = name

    def _my_callback(self, retry_state):
        print(f"实例 {self.name} 的任务失败了。")
        # ...

    # 这会失败！NameError: name 'self' is not defined
    @retry(retry_error_callback=self._my_callback)
    def run(self):
        # ...

class TTS:
    def __init__(self, name):
        self.name = name

    def run(self):
        # 在这里，self 是可用的！
        @retry(
            # 因为装饰器在 run 方法内部，它可以“捕获”到 self
            retry_error_callback=self._my_callback
        )
        def _execute_task():
            # 这里是真正需要重试的逻辑
            print(f"正在为 {self.name} 执行任务...")
            raise ValueError("任务失败")

        # 调用被装饰的内部函数
        return _execute_task()

    def _my_callback(self, retry_state: RetryCallState):
        # ...

from openai import OpenAI, AuthenticationError, PermissionDeniedError
from tenacity import retry, stop_after_attempt, wait_fixed, retry_if_not_exception_type, RetryCallState

# 1. 定义一个通用的、可复用的异常处理类
class RetryRaise:
    # 定义不应重试的致命异常
    NO_RETRY_EXCEPT = (AuthenticationError, PermissionDeniedError)

    @classmethod
    def _raise(cls, retry_state: RetryCallState):
        ex = retry_state.outcome.exception()
        if ex:
            # 根据不同异常类型，进行日志记录并抛出自定义的、更友好的 RuntimeError
            # ... 此处可以添加更复杂的异常分类逻辑 ...
            raise RuntimeError(f"重试 {retry_state.attempt_number} 次后最终失败: {ex}") from ex
        raise RuntimeError(f"重试 {retry_state.attempt_number} 次后失败，但未捕获到异常。")

# 2. 实现模板父类
class BaseProvider:
    @retry(
        stop=stop_after_attempt(3),
        wait=wait_fixed(2),
        retry=retry_if_not_exception_type(RetryRaise.NO_RETRY_EXCEPT),
        retry_error_callback=RetryRaise._raise
    )
    def _exec(self) -> str:
        """这是模板方法，负责重试。子类不应重写它。"""
        # 调用钩子方法，由子类实现
        return self._do_work()

    def _do_work(self) -> str:
        """这是钩子方法，子类必须实现它。"""
        raise NotImplementedError("子类必须实现 _do_work 方法")

# 3. 实现具体的子类
class DeepSeekProvider(BaseProvider):
    def __init__(self, api_key: str, base_url: str):
        self.api_key = api_key
        self.base_url = base_url
        self.model = OpenAI(api_key=self.api_key, base_url=self.base_url)

    def _do_work(self) -> str:
        """这里只关心核心业务逻辑，完全不用考虑重试。"""
        response = self.model.chat.completions.create(
            model="deepseek-chat",
            messages=[{'role': 'user', 'content': '你是谁？'}]
        )
        if response.choices:
            return response.choices[0].message.content.strip()
        raise RuntimeError(f"API未返回有效结果: {response}")

# --- 如何使用 ---
 provider = DeepSeekProvider(api_key="...", base_url="...")
 try:
     # 我们调用的是 _exec，它包含了重试逻辑
     result = provider._exec()
     print("执行成功:", result)
 except RuntimeError as e:
     # 如果最终失败，会捕获到 RetryRaise 抛出的友好异常
     print("执行失败:", e)