Spaces:
Running
Running
File size: 1,486 Bytes
09f70c3 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 | """
Utility functions for RAG pipeline.
Provides common utilities like retry logic, validation helpers, etc.
"""
import time
import logging
from typing import Callable, TypeVar, Any
logger = logging.getLogger(__name__)
T = TypeVar("T")
def retry_with_backoff(
func: Callable[..., T],
max_retries: int = 3,
base_delay: float = 1.0,
max_delay: float = 10.0,
backoff_multiplier: float = 2.0,
) -> T:
"""
Retry a function with exponential backoff.
Args:
func: Function to retry
max_retries: Maximum number of retry attempts
base_delay: Initial delay in seconds
max_delay: Maximum delay in seconds
backoff_multiplier: Multiplier for exponential backoff
Returns:
Result of the function
Raises:
Exception: The last exception if all retries fail
"""
last_exception = None
delay = base_delay
for attempt in range(max_retries):
try:
return func()
except Exception as e:
last_exception = e
if attempt < max_retries - 1:
logger.warning(
f"Attempt {attempt + 1}/{max_retries} failed: {e}. "
f"Retrying in {delay:.1f}s..."
)
time.sleep(delay)
delay = min(delay * backoff_multiplier, max_delay)
else:
logger.error(f"All {max_retries} attempts failed. Last error: {e}")
raise last_exception
|