railtracks.llm.retries

13class ExponentialRetry(RetryApproach):
14    """Exponential backoff: delay = ``base ** attempt`` seconds.
15
16    Caps ``max_tries`` at 20 to prevent runaway wait times.  Warns if the
17    theoretical maximum delay exceeds 1 000 s.
18
19    Args:
20        max_tries: Total attempts (1–20).
21        base: Backoff multiplier (>= 1.0).  Default ``2.0`` gives 1 s, 2 s, 4 s …
22        jitter: When ``True``, randomises the delay to ``uniform(0, base**attempt)``
23            to spread load across concurrent callers.
24    """
25
26    def __init__(
27        self,
28        max_tries: int,
29        base: float = 2.0,
30        jitter: bool = True,
31    ) -> None:
32        if max_tries < 1:
33            raise ValueError("max_tries must be >= 1")
34        if max_tries > _MAX_RETRY_TIMES_EXPONETIAL:
35            raise ValueError(f"max_tries must be <= {_MAX_RETRY_TIMES_EXPONETIAL}")
36        if base < 1:
37            raise ValueError("base must be >= 1")
38
39        if base**max_tries > _MAX_RECOMONDED_RETRY_TIME_EXPONETIAL:
40            warnings.warn(
41                f"With base={base} and max_tries={max_tries}, the maximum delay could exceed {_MAX_RECOMONDED_RETRY_TIME_EXPONETIAL} seconds, which may be too long for some applications."
42            )
43
44        super().__init__(
45            max_tries=max_tries,
46        )
47        self._base = base
48        self._jitter = jitter
49
50    @classmethod
51    def approach_name(cls) -> str:
52        return "exponential"
53
54    def _compute_delay(self, attempt: int) -> float:
55        delay = self._base**attempt
56        return random.uniform(0, delay) if self._jitter else delay

 7class FixedRetry(RetryApproach):
 8    """Fixed backoff: waits the same ``delay`` seconds before every retry.
 9
10    Useful when the provider's ``Retry-After`` header is unreliable or when
11    you want predictable, uniform pacing.  Caps ``max_tries`` at 100.
12
13    Args:
14        max_tries: Total attempts (1–100).
15        delay: Seconds to wait between attempts (>= 0).  Default ``1.0``.
16    """
17
18    def __init__(
19        self,
20        max_tries: int,
21        delay: float = 1.0,
22    ) -> None:
23        if max_tries < 1:
24            raise ValueError("max_tries must be >= 1")
25        if max_tries > _MAX_RETRY_TIMES_FIXED:
26            raise ValueError(f"max_tries must be <= {_MAX_RETRY_TIMES_FIXED}")
27        if delay < 0:
28            raise ValueError("delay must be >= 0")
29
30        super().__init__(max_tries=max_tries)
31        self._delay = delay
32
33    @classmethod
34    def approach_name(cls) -> str:
35        return "fixed"
36
37    def _compute_delay(self, attempt: int) -> float:
38        return self._delay

 25class RetryApproach(ABC):
 26    """Template base for retry strategies.
 27
 28    Subclasses implement ``_compute_delay`` to define the backoff schedule.
 29    The retry loop and exception handling live here so subclasses stay minimal.
 30
 31    Args:
 32        max_tries: Total attempts including the first call (not just retries).
 33    """
 34
 35    def __init__(self, max_tries: int) -> None:
 36        if max_tries < 1:
 37            raise ValueError("max_tries must be >= 1")
 38        self._max_tries = max_tries
 39
 40    @classmethod
 41    @abstractmethod
 42    def approach_name(cls) -> str:
 43        """Short identifier used in ``RetryError`` messages (e.g. ``"exponential"``)."""
 44
 45    @abstractmethod
 46    def _compute_delay(self, attempt: int) -> float:
 47        """Seconds to wait before attempt ``attempt + 1`` (0-indexed)."""
 48
 49    def call_with_retry(self, completion: Callable[[], _TResult]) -> _TResult:
 50        """Call ``completion`` up to ``max_tries`` times, sleeping between failures.
 51
 52        Raises:
 53            RetryError: All attempts failed; ``exception_list`` holds every error.
 54            Exception: Any non-retryable exception propagates immediately.
 55        """
 56        exceptions: list[Exception] = []
 57
 58        for attempt in range(self._max_tries):
 59            try:
 60                return completion()
 61            except _RETRYABLE_EXCEPTIONS as e:
 62                exceptions.append(e)
 63                if attempt == self._max_tries - 1:
 64                    raise RetryError(
 65                        self.approach_name(),
 66                        "Max retries exceeded",
 67                        ["Examine exceptions to determine the root cause."],
 68                        exceptions,
 69                    ) from e
 70
 71                time.sleep(self._compute_delay(attempt))
 72
 73        assert False, "Unreachable code"
 74
 75    async def acall_with_retry(
 76        self, completion: Callable[[], Awaitable[_TResult]]
 77    ) -> _TResult:
 78        """Async mirror of ``call_with_retry`` — awaits ``completion`` each attempt."""
 79        exceptions: list[Exception] = []
 80
 81        for attempt in range(self._max_tries):
 82            try:
 83                return await completion()
 84            except _RETRYABLE_EXCEPTIONS as e:
 85                exceptions.append(e)
 86                if attempt == self._max_tries - 1:
 87                    raise RetryError(
 88                        self.approach_name(),
 89                        "Max retries exceeded",
 90                        ["Examine exceptions to determine the root cause."],
 91                        exceptions,
 92                    ) from e
 93
 94                delay: float | None = None
 95
 96                if delay is not None:
 97                    await asyncio.sleep(delay)
 98                else:
 99                    await asyncio.sleep(self._compute_delay(attempt))
100
101        assert False, "Unreachable code"

 9class LinearRetry(RetryApproach):
10    """Linear backoff: delay = ``step * (attempt + 1)`` seconds.
11
12    Grows steadily rather than exponentially — a middle ground between fixed
13    and exponential when you expect short-lived rate limits.  Caps ``max_tries``
14    at 100.
15
16    Args:
17        max_tries: Total attempts (1–100).
18        step: Base increment in seconds (>= 0).  Default ``1.0`` gives 1 s, 2 s, 3 s …
19        jitter: When ``True``, randomises the delay to ``uniform(0, step * (attempt+1))``.
20    """
21
22    def __init__(
23        self,
24        max_tries: int,
25        step: float = 1.0,
26        jitter: bool = True,
27    ) -> None:
28        if max_tries < 1:
29            raise ValueError("max_tries must be >= 1")
30        if max_tries > _MAX_RETRY_TIMES_LINEAR:
31            raise ValueError(f"max_tries must be <= {_MAX_RETRY_TIMES_LINEAR}")
32        if step < 0:
33            raise ValueError("step must be >= 0")
34
35        super().__init__(max_tries=max_tries)
36        self._step = step
37        self._jitter = jitter
38
39    @classmethod
40    def approach_name(cls) -> str:
41        return "linear"
42
43    def _compute_delay(self, attempt: int) -> float:
44        delay = self._step * (attempt + 1)
45        return random.uniform(0, delay) if self._jitter else delay