Skip to content

Retry policy

RetryPolicy #

Bases: Protocol

Source code in llama-index-core/llama_index/core/workflow/retry_policy.py 
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
@runtime_checkable
class RetryPolicy(Protocol):
    def next(
        self, elapsed_time: float, attempts: int, error: Exception
    ) -> Optional[float]:
        """Decides if we should make another retry, returning the number of seconds to wait before the next run.

        Args:
            elapsed_time: Time in seconds that passed since the last attempt.
            attempts: The number of attempts done so far.
            error: The last error occurred.

        Returns:
            The amount of seconds to wait before the next attempt, or None if we stop retrying.
        """

next #

next(elapsed_time: float, attempts: int, error: Exception) -> Optional[float]

Decides if we should make another retry, returning the number of seconds to wait before the next run.

Parameters:

Name Type Description Default
elapsed_time float

Time in seconds that passed since the last attempt.

 required
attempts int

The number of attempts done so far.

 required
error Exception

The last error occurred.

 required

Returns:

Type Description
Optional[float]

The amount of seconds to wait before the next attempt, or None if we stop retrying.
Source code in llama-index-core/llama_index/core/workflow/retry_policy.py 
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
def next(
    self, elapsed_time: float, attempts: int, error: Exception
) -> Optional[float]:
    """Decides if we should make another retry, returning the number of seconds to wait before the next run.

    Args:
        elapsed_time: Time in seconds that passed since the last attempt.
        attempts: The number of attempts done so far.
        error: The last error occurred.

    Returns:
        The amount of seconds to wait before the next attempt, or None if we stop retrying.
    """

ConstantDelayRetryPolicy #

A simple policy that retries a step at regular intervals for a number of times.

Source code in llama-index-core/llama_index/core/workflow/retry_policy.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
class ConstantDelayRetryPolicy:
    """A simple policy that retries a step at regular intervals for a number of times."""

    def __init__(self, maximum_attempts: int = 3, delay: float = 5) -> None:
        """Creates a ConstantDelayRetryPolicy instance.

        Args:
            maximum_attempts: How many consecutive times the workflow should try to run the step in case of an error.
            delay: how much time in seconds must pass before another attempt.
        """
        self.maximum_attempts = maximum_attempts
        self.delay = delay

    def next(
        self, elapsed_time: float, attempts: int, error: Exception
    ) -> Optional[float]:
        if attempts >= self.maximum_attempts:
            return None

        return self.delay