retry_utils

Retry utilities for AI client implementations.

This module provides retry decorators and utilities for handling transient failures like rate limits, service unavailability, and network errors.

Retry utilities for AI client implementations.

This module provides retry decorators and utilities for handling transient failures like rate limits, service unavailability, and network errors.

exception RateLimitError[source]

Bases: Exception

Base exception for rate limit errors.

exception ServiceUnavailableError[source]

Bases: Exception

Exception for service unavailable errors (503).

get_retry_decorator(max_attempts=3, min_wait=1, max_wait=60, exponential_base=2, jitter=True, retry_exceptions=(<class 'ffai.retry_utils.RateLimitError'>, <class 'ffai.retry_utils.ServiceUnavailableError'>, <class 'ConnectionError'>, <class 'TimeoutError'>), log_level=20)[source]

Create a retry decorator with configurable parameters.

Parameters:

  • max_attempts (int) – Maximum number of retry attempts

  • min_wait (float) – Minimum wait time in seconds

  • max_wait (float) – Maximum wait time in seconds

  • exponential_base (float) – Base for exponential backoff

  • jitter (bool) – Whether to add jitter to wait times

  • retry_exceptions (tuple) – Tuple of exception types to retry on

  • log_level (int) – Logging level for retry messages

Returns:

Configured retry decorator

Example

>>> @get_retry_decorator(max_attempts=5, min_wait=2)
... def my_api_call():
...     return client.generate_response("Hello")
get_configured_retry_decorator()[source]

Get retry decorator configured from global config, with fallback to defaults.

Reads retry settings from the application config (config/main.yaml). Falls back to get_retry_decorator() defaults when config is unavailable.

Returns:

Configured retry decorator

should_retry_exception(exception)[source]

Determine if an exception should trigger a retry.

Parameters:

exception (Exception) – The exception to check

Returns:

True if the exception should trigger a retry

Return type:

bool

extract_retry_after(exception)[source]

Extract retry-after delay from exception if available.

Parameters:

exception (Exception) – The exception to extract retry-after from

Returns:

Retry delay in seconds, or None if not available

Return type:

float | None

retry_with_backoff(func, max_attempts=3, min_wait=1, max_wait=60, exponential_base=2, jitter=True, *args, **kwargs)[source]

Execute a function with retry and exponential backoff.

Parameters:

  • func (Any) – Function to execute

  • max_attempts (int) – Maximum number of retry attempts

  • min_wait (float) – Minimum wait time in seconds

  • max_wait (float) – Maximum wait time in seconds

  • exponential_base (float) – Base for exponential backoff

  • jitter (bool) – Whether to add jitter to wait times

  • *args – Positional arguments for func

  • **kwargs – Keyword arguments for func

Returns:

Result of func if successful

Raises:

  • RetryError – If all retry attempts fail

  • Exception – Original exception if not retryable

Return type:

Any

Example

>>> result = retry_with_backoff(
...     client.generate_response,
...     max_attempts=5,
...     prompt="Hello"
... )
create_rate_limit_error(exception)[source]

Create a RateLimitError from another exception.

Parameters:

exception (Exception) – Original exception

Returns:

RateLimitError with information from original exception

Return type:

RateLimitError

Classes

class RateLimitError[source]

Bases: Exception

Base exception for rate limit errors.

add_note()

Exception.add_note(note) – add a note to the exception

args
with_traceback()

Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

class ServiceUnavailableError[source]

Bases: Exception

Exception for service unavailable errors (503).

add_note()

Exception.add_note(note) – add a note to the exception

args
with_traceback()

Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

Functions

create_rate_limit_error(exception)[source]

Create a RateLimitError from another exception.

Parameters:

exception (Exception) – Original exception

Returns:

RateLimitError with information from original exception

Return type:

RateLimitError

extract_retry_after(exception)[source]

Extract retry-after delay from exception if available.

Parameters:

exception (Exception) – The exception to extract retry-after from

Returns:

Retry delay in seconds, or None if not available

Return type:

float | None

get_configured_retry_decorator()[source]

Get retry decorator configured from global config, with fallback to defaults.

Reads retry settings from the application config (config/main.yaml). Falls back to get_retry_decorator() defaults when config is unavailable.

Returns:

Configured retry decorator

get_retry_decorator(max_attempts=3, min_wait=1, max_wait=60, exponential_base=2, jitter=True, retry_exceptions=(<class 'ffai.retry_utils.RateLimitError'>, <class 'ffai.retry_utils.ServiceUnavailableError'>, <class 'ConnectionError'>, <class 'TimeoutError'>), log_level=20)[source]

Create a retry decorator with configurable parameters.

Parameters:

  • max_attempts (int) – Maximum number of retry attempts

  • min_wait (float) – Minimum wait time in seconds

  • max_wait (float) – Maximum wait time in seconds

  • exponential_base (float) – Base for exponential backoff

  • jitter (bool) – Whether to add jitter to wait times

  • retry_exceptions (tuple) – Tuple of exception types to retry on

  • log_level (int) – Logging level for retry messages

Returns:

Configured retry decorator

Example

>>> @get_retry_decorator(max_attempts=5, min_wait=2)
... def my_api_call():
...     return client.generate_response("Hello")
retry_with_backoff(func, max_attempts=3, min_wait=1, max_wait=60, exponential_base=2, jitter=True, *args, **kwargs)[source]

Execute a function with retry and exponential backoff.

Parameters:

  • func (Any) – Function to execute

  • max_attempts (int) – Maximum number of retry attempts

  • min_wait (float) – Minimum wait time in seconds

  • max_wait (float) – Maximum wait time in seconds

  • exponential_base (float) – Base for exponential backoff

  • jitter (bool) – Whether to add jitter to wait times

  • *args – Positional arguments for func

  • **kwargs – Keyword arguments for func

Returns:

Result of func if successful

Raises:

  • RetryError – If all retry attempts fail

  • Exception – Original exception if not retryable

Return type:

Any

Example

>>> result = retry_with_backoff(
...     client.generate_response,
...     max_attempts=5,
...     prompt="Hello"
... )
should_retry_exception(exception)[source]

Determine if an exception should trigger a retry.

Parameters:

exception (Exception) – The exception to check

Returns:

True if the exception should trigger a retry

Return type:

bool