ai-flow-python/app/util/retry_util.py

import logging
import time
from typing import Callable, TypeVar

logger = logging.getLogger(__name__)

T = TypeVar("T")


def retry_on_failure(
    fn: Callable[[], T],
    *,
    max_retries: int = 3,
    backoff_base: float = 2.0,
    task_name: str = "",
) -> T:
    """在失败时使用指数退避进行重试，仅用于任务提交阶段。

    fn 应当在提交不成功时抛出异常，所有重试耗尽后将抛出最后一次的异常。
    """
    last_exc: Exception | None = None
    for attempt in range(1, max_retries + 1):
        try:
            return fn()
        except Exception as e:
            last_exc = e
            if attempt < max_retries:
                wait = backoff_base ** attempt
                logger.warning(
                    "[%s] Submit attempt %d/%d failed: %s, retrying in %.1fs …",
                    task_name, attempt, max_retries, e, wait,
                )
                time.sleep(wait)
            else:
                logger.error(
                    "[%s] All %d submit attempts exhausted, last error: %s",
                    task_name, max_retries, e,
                )
    raise last_exc  # type: ignore[misc]