在构建一个每天采集数十万条AI回答数据的系统时，我们遇到了接口限流、超时和返回格式异常等问题。简单的固定间隔重试不仅无法解决问题，反而可能加剧服务端压力。本文基于Python 3.9和tenacity、pybreaker库，分享一套完整的异常处理与重试机制设计方案，包括指数退避、熔断和降级策略，并提供可复用的代码示例和验证方法。本文不涉及具体的AI服务商API细节，重点在于通用的重试机制设计。
一、问题背景与业务约束
我们的采集系统需要定期从多个AI服务商获取回答数据，用于后续的分析和处理。业务对数据完整性和时效性有较高要求，但上游接口的稳定性不可控。具体约束如下：
- 数据量：每天需采集数十万条回答，高峰期QPS可达数百。
- 接口特性：不同服务商的接口限流策略不同，部分接口在超时后返回不完整数据。
- 成本敏感：调用大模型接口需要付费，无效重试会浪费成本。
- 监控要求：需要实时掌握采集成功率、失败原因分布，以便快速响应。
这些约束决定了重试机制必须精准、有界，并且能够快速失败。
二、异常类型与初步处理
在采集过程中，我们遇到的异常主要分为以下几类：
- 网络异常：连接超时、读取超时、DNS解析失败等。
- HTTP状态码异常：429（限流）、500（服务端错误）、503（服务不可用）等。
- 业务异常：返回数据格式错误、字段缺失、内容为空等。
针对这些异常，我们首先实现了基础的异常捕获和日志记录，但很快发现简单的重试策略（固定间隔重试3次）存在严重问题：
- 在服务端故障时，大量请求同时重试，加剧了服务端压力，导致恢复时间延长。
- 对于限流错误（429），固定间隔重试无法有效规避限流窗口。
- 对于格式错误等业务异常，重试往往无效，只会浪费资源。
因此，我们需要区分可重试和不可重试的异常，并采用更精细的重试策略。
三、重试机制设计：指数退避与抖动
为了解决上述问题，我们引入了指数退避策略。基本思想是：每次重试的间隔时间随重试次数指数增长，并加入随机抖动，避免多个请求同时重试。
核心代码示例
以下是一个使用tenacity库实现的指数退避示例（Python）：
importrandomimporttimefromtenacityimportretry,stop_after_attempt,wait_exponential,retry_if_exception_typeclassRateLimitError(Exception):passclassServerError(Exception):pass@retry(retry=retry_if_exception_type((RateLimitError,ServerError)),wait=wait_exponential(multiplier=1,min=2,max=60),stop=stop_after_attempt(5),reraise=True)deffetch_ai_answer(prompt):# 模拟请求response=call_ai_service(prompt)ifresponse.status_code==429:raiseRateLimitError("Rate limited")ifresponse.status_code>=500:raiseServerError("Server error")returnresponse.json()设计要点
- 重试条件：仅对可重试的异常（如限流、服务端错误）进行重试，对于格式错误等业务异常直接抛出。
- 退避策略：使用指数退避，初始间隔2秒，最大间隔60秒，并加入随机抖动（tenacity库默认实现）。
- 重试次数：根据业务容忍度设置为5次，避免无限重试。
选择指数退避是因为它能在短时间内快速重试，同时避免对服务端造成持续压力。固定间隔重试在服务端故障时容易导致重试风暴，而线性退避又可能等待过久。
四、熔断机制：防止雪崩
即使有了指数退避，当服务端持续故障时，大量请求仍会堆积在等待重试，导致本地资源耗尽。为此，我们引入了熔断器模式。
熔断器状态机
熔断器有三种状态：关闭、打开、半开。
- 关闭：正常调用，统计失败率。
- 打开：失败率达到阈值（如50%），直接拒绝请求，快速失败。
- 半开：经过冷却时间后，允许少量请求探测，如果成功则关闭熔断器，否则继续打开。
实现代码示例
我们使用了pybreaker库，完整配置如下：
importpybreaker breaker=pybreaker.CircuitBreaker(fail_max=5,reset_timeout=60,exclude=[RateLimitError]# 限流错误不触发熔断，因为限流是暂时的)@breaker@retry(retry=retry_if_exception_type((RateLimitError,ServerError)),wait=wait_exponential(multiplier=1,min=2,max=60),stop=stop_after_attempt(5),reraise=True)deffetch_ai_answer(prompt):# 模拟请求response=call_ai_service(prompt)ifresponse.status_code==429:raiseRateLimitError("Rate limited")ifresponse.status_code>=500:raiseServerError("Server error")returnresponse.json()设计要点
- 失败阈值：连续失败5次触发熔断，避免频繁抖动。
- 冷却时间：60秒后进入半开状态，允许探测。
- 排除特定异常：限流错误（429）不应触发熔断，因为限流是服务端主动保护，熔断反而会加重问题。
熔断机制能有效防止雪崩，但需要根据服务端的实际恢复时间调整冷却时间。
五、降级策略：保证核心流程
当熔断器打开或重试耗尽时，我们需要降级处理，避免采集任务完全失败。降级策略包括：
- 缓存降级：如果之前采集过相同或相似问题，直接使用缓存数据。
- 队列降级：将失败的任务放入待处理队列，稍后重试。
- 默认值降级：对于非关键字段，使用默认值或空值。
降级实现示例
deffetch_with_fallback(prompt):try:returnfetch_ai_answer(prompt)exceptExceptionase:# 尝试从缓存获取cached=cache.get(prompt)ifcached:returncached# 放入重试队列retry_queue.put(prompt)returnNone降级策略的选择取决于业务对数据完整性的要求。对于非关键数据，可以接受默认值；对于关键数据，则必须通过队列保证最终一致。
六、验证结果与监控
为了验证机制的有效性，我们进行了压测和故障注入实验。
压测结果
在模拟限流场景下（服务端返回429），使用指数退避后，成功率显著提升，平均响应时间明显下降。具体数据因环境而异，建议根据实际压测结果评估。
监控指标
我们通过日志和监控系统实时跟踪以下指标，具体实现可以使用Prometheus+Grafana，通过埋点采集以下指标：
- 采集成功率：成功请求数 / 总请求数。
- 重试次数分布：不同重试次数的占比。
- 熔断器状态变化：打开、关闭、半开的次数。
- 降级触发次数：使用缓存或队列降级的次数。
例如，使用Prometheus的Counter和Histogram记录请求总数、成功数、重试次数等，通过Grafana展示趋势。
七、踩坑与避坑总结
在实现过程中，我们遇到了几个典型问题：
- 重试风暴：最初没有加入抖动，导致大量请求同时重试，服务端压力骤增。加入随机抖动后问题解决。
- 熔断误判：将限流错误也计入熔断统计，导致熔断频繁触发。通过排除429错误解决。
- 重试超时：重试时未重新计算超时时间，导致整体耗时过长。在每次重试时重置超时时间。
这些问题的共同点是缺乏对异常类型的细分和全局视角，导致机制之间相互干扰。
总结
本文从AI回答采集的实际需求出发，设计并实现了一套完整的异常处理与重试机制。通过指数退避、熔断和降级策略，有效提升了采集系统的稳定性和可靠性。该方案适用于对接口稳定性要求较高、成本敏感的场景。需要注意的是，重试次数、熔断阈值等参数需要根据实际业务调整，且应结合监控系统持续优化。
