-
Notifications
You must be signed in to change notification settings - Fork 1
/
Copy pathretry_decorator.py
30 lines (26 loc) · 1.07 KB
/
retry_decorator.py
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
# retry_decorator.py
import time
import openai
from functools import wraps
def retry_on_service_unavailable(max_retries=5, backoff_factor=0.5):
"""A decorator for retrying a function call with exponential backoff.
Args:
max_retries (int): Maximum number of retries before giving up. Default is 5.
backoff_factor (float): Multiplier for the delay between retries. Default is 0.5.
Returns:
Callable: Decorated function that will be retried on `openai.error.ServiceUnavailableError`.
"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
retries = 0
while retries < max_retries:
try:
return func(*args, **kwargs)
except openai.error.ServiceUnavailableError:
sleep_time = backoff_factor * (2 ** retries)
time.sleep(sleep_time)
retries += 1
return func(*args, **kwargs) # Final attempt, let exception propagate if this fails
return wrapper
return decorator