MCPcopy Create free account
hub / github.com/Tron521/codex-console-temp / retry_with_backoff

Function retry_with_backoff

src/core/utils.py:200–253  ·  view source on GitHub ↗

带有指数退避的重试装饰器/函数 Args: func: 要重试的函数 max_retries: 最大重试次数 base_delay: 基础延迟(秒) max_delay: 最大延迟(秒) backoff_factor: 退避因子 exceptions: 要捕获的异常类型 Returns: 函数的返回值 Raises: 最后一次尝试的异常

(
    func: Callable,
    max_retries: int = 3,
    base_delay: float = 1.0,
    max_delay: float = 30.0,
    backoff_factor: float = 2.0,
    exceptions: tuple = (Exception,)
)

Source from the content-addressed store, hash-verified

198
199
200def retry_with_backoff(
201 func: Callable,
202 max_retries: int = 3,
203 base_delay: float = 1.0,
204 max_delay: float = 30.0,
205 backoff_factor: float = 2.0,
206 exceptions: tuple = (Exception,)
207) -> Any:
208 """
209 带有指数退避的重试装饰器/函数
210
211 Args:
212 func: 要重试的函数
213 max_retries: 最大重试次数
214 base_delay: 基础延迟(秒)
215 max_delay: 最大延迟(秒)
216 backoff_factor: 退避因子
217 exceptions: 要捕获的异常类型
218
219 Returns:
220 函数的返回值
221
222 Raises:
223 最后一次尝试的异常
224 """
225 last_exception = None
226
227 for attempt in range(max_retries + 1):
228 try:
229 return func()
230 except exceptions as e:
231 last_exception = e
232
233 # 如果是最后一次尝试,直接抛出异常
234 if attempt == max_retries:
235 break
236
237 # 计算延迟时间
238 delay = min(base_delay * (backoff_factor ** attempt), max_delay)
239
240 # 添加随机抖动
241 delay *= (0.5 + random.random())
242
243 # 记录日志
244 logger = logging.getLogger(__name__)
245 logger.warning(
246 f"尝试 {func.__name__} 失败 (attempt {attempt + 1}/{max_retries + 1}): {e}. "
247 f"等待 {delay:.2f} 秒后重试..."
248 )
249
250 time.sleep(delay)
251
252 # 所有重试都失败,抛出最后一个异常
253 raise last_exception
254
255
256class RetryDecorator:

Callers 1

wrapperMethod · 0.85

Calls 2

warningMethod · 0.80
sleepMethod · 0.80

Tested by

no test coverage detected