Spaces:
Running
Running
File size: 2,010 Bytes
5f3e9f5 | 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 59 60 | """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):
@wraps(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
|