Spaces:
Running
Running
| """Retry mechanism with exponential backoff for AI requests.""" | |
| import time | |
| from functools import wraps | |
| # Exceptions whose name matches one of these should propagate immediately | |
| # instead of being retried. Matching by class name (rather than importing | |
| # the classes directly) avoids circular imports with modules like | |
| # core.ai_client that depend on this decorator. | |
| _NON_RETRYABLE_NAMES = frozenset({ | |
| 'CancelledError', | |
| 'KeyboardInterrupt', | |
| 'SystemExit', | |
| 'GeneratorExit', | |
| }) | |
| def retry_with_backoff(max_retries=3, base_delay=1, max_delay=30, exceptions_to_skip=()): | |
| """ | |
| Decorator for retrying failed AI requests with exponential backoff. | |
| Args: | |
| max_retries: Maximum number of retry attempts | |
| base_delay: Initial delay in seconds | |
| max_delay: Maximum delay between retries | |
| exceptions_to_skip: Extra exception classes to re-raise without retry | |
| """ | |
| def decorator(func): | |
| def wrapper(*args, **kwargs): | |
| retries = 0 | |
| while retries <= max_retries: | |
| try: | |
| return func(*args, **kwargs) | |
| except exceptions_to_skip: | |
| raise | |
| except Exception as e: | |
| # Let cancellations and shutdowns propagate instantly. | |
| if type(e).__name__ in _NON_RETRYABLE_NAMES: | |
| raise | |
| retries += 1 | |
| if retries > max_retries: | |
| print(f"❌ Max retries ({max_retries}) exceeded") | |
| raise | |
| # Calculate delay with exponential backoff | |
| delay = min(base_delay * (2 ** (retries - 1)), max_delay) | |
| print(f"⚠️ Attempt {retries} failed: {str(e)}") | |
| print(f"🔄 Retrying in {delay} seconds... ({retries}/{max_retries})") | |
| time.sleep(delay) | |
| return None | |
| return wrapper | |
| return decorator | |