在Python中调用大模型API时经常遇到网络超时、限流429或服务端5xx错误。如果直接失败采集任务可能中断如果盲目重试可能加重服务压力。本文以requests库为例实现指数退避重试、熔断和降级机制确保采集任务稳定完成。一、问题现象与业务约束假设我们有一个定时任务需要从多个大模型API获取回答用于后续分析。实际运行中我们遇到以下问题网络抖动导致连接超时任务失败。调用频率过高触发限流HTTP 429。服务端偶尔返回500、502、503。批量任务中单个失败导致整个批次失败。业务约束实时性要求不高但数据完整性要求高。需要控制请求速率避免被封。系统需要可观测便于排查。二、异常分类与处理策略根据异常类型我们采取不同策略网络异常超时、连接错误瞬时故障重试。HTTP 5xx500、502、503服务端暂时不可用重试。HTTP 429限流等待后重试并考虑熔断。HTTP 4xx400、401等客户端错误不重试。业务异常返回内容格式错误不重试记录并跳过。三、重试机制设计1. 重试条件只对以下情况重试requests.exceptions.Timeoutrequests.exceptions.ConnectionErrorHTTP 状态码为429、500、502、503其他异常直接抛出或记录。2. 指数退避与抖动重试间隔采用指数退避并加入随机抖动避免多个请求同时重试造成雪崩。importrandomimporttimedefget_backoff_time(attempt,base_delay1,max_delay60):计算重试等待时间指数退避加抖动delaymin(max_delay,base_delay*(2**attempt))jitterrandom.uniform(0,delay*0.1)returndelayjitter3. 最大重试次数设置最大重试次数如3次超过后放弃并记录失败日志。4. 重试装饰器实现以下是一个重试装饰器用于包裹API调用函数importfunctoolsimporttimeimportloggingimportrequests loggerlogging.getLogger(__name__)defretry_on_failure(max_retries3,base_delay1,max_delay60):defdecorator(func):functools.wraps(func)defwrapper(*args,**kwargs):forattemptinrange(max_retries):try:returnfunc(*args,**kwargs)except(requests.exceptions.Timeout,requests.exceptions.ConnectionError)ase:ifattemptmax_retries-1:raisedelayget_backoff_time(attempt,base_delay,max_delay)logger.warning(fAttempt{attempt1}failed:{e}, retrying in{delay:.2f}s)time.sleep(delay)exceptrequests.exceptions.HTTPErrorase:ife.response.status_codein(429,500,502,503):ifattemptmax_retries-1:raisedelayget_backoff_time(attempt,base_delay,max_delay)logger.warning(fHTTP{e.response.status_code}on attempt{attempt1}, retrying in{delay:.2f}s)time.sleep(delay)else:raisereturnNonereturnwrapperreturndecorator使用示例retry_on_failure(max_retries3,base_delay1,max_delay60)defcall_ai_service(prompt):responserequests.post(https://api.example.com/v1/completions,json{prompt:prompt},timeout5)response.raise_for_status()returnresponse.json()四、熔断机制当某个服务持续失败时继续重试只会浪费资源。引入熔断器当失败率达到阈值时暂时停止调用快速失败。1. 熔断器状态关闭正常调用。打开熔断直接抛出异常。半开允许少量请求试探成功则关闭失败则重新打开。2. 使用pybreaker实现pybreaker是一个Python熔断器库安装pipinstallpybreaker配置示例importpybreaker breakerpybreaker.CircuitBreaker(fail_max5,reset_timeout60)breakerretry_on_failure()defcall_ai_service(prompt):# 实际调用代码pass当连续失败5次后熔断器打开后续调用直接抛出CircuitBreakerError。60秒后进入半开状态允许少量请求试探成功则关闭熔断器。注意fail_max和reset_timeout需根据业务调整。五、降级策略当重试和熔断都无法解决问题时需要降级处理备用模型主模型不可用时切换到备用模型。缓存结果相同问题直接使用缓存。延迟处理将任务放入队列稍后重试。在我们的系统中当熔断器打开时将任务标记为“待重试”由定时任务在稍后重新执行。验证结果为了验证重试机制可以模拟超时和500错误。正常情况下应当看到以下现象模拟超时设置超时时间为0.1秒服务端延迟0.5秒重试3次后成功。模拟500错误服务端返回500重试3次后成功。重试日志会显示每次重试的等待时间最终成功。验证方法检查日志中是否有重试记录。确认最终调用成功。统计重试次数是否符合预期。常见问题与避坑1. 重试导致请求重复如果API不是幂等的重试可能产生重复请求。解决方案使用请求ID去重确保同一请求只被处理一次。2. 退避时间过长如果基础延迟设置过大重试间隔可能过长影响效率。需要根据业务容忍度调整参数。3. 熔断误判如果某个服务偶尔超时但整体健康熔断可能误判。通过设置合理的失败阈值和采样窗口来减少误判。总结本文解决了Python调用大模型API时的异常处理问题。根因是网络和服务端不稳定最终采用指数退避重试、熔断和降级策略。方案适用于对实时性要求不高的采集场景但需要根据业务调整参数。当前限制未考虑动态超时和基于历史数据的退避调整未来可进一步优化。