AI API 接入踩坑记录：限流、重试、降级策略

在生产环境接入大模型 API 的三个月里，我们从"调通就行"一路踩坑到"半夜被报警叫醒再也不用慌"。本文复盘实际遇到的限流、重试炸弹、SSE 流式响应超时等问题，以及最终沉淀下来的一套策略。

一、从一次凌晨 3 点的报警说起

事情的起因很简单：业务需要同时接入 DeepSeek、通义千问、豆包等多家大模型 API。第一阶段我们快速实现了统一调用层，跑通了 demo，然后自信地上了生产。

第一周风平浪静。第二周某个凌晨 3 点，PagerDuty 响了——下游模型调用大面积超时，重试机制触发后又造成了雪崩，整个链路的延迟从 200ms 飙到 30 秒以上。

事后复盘，踩了三个坑：

二、踩坑一：限流 —— 不是加 retry 就能解决的

2.1 踩坑现场

最早的重试逻辑非常简单：

# ❌ 最初的反面教材importtimeforattemptinrange(3):try:response=call_model_api(prompt)breakexceptException:time.sleep(1)# 固定等待 1 秒continue

这个逻辑在生产上跑了三天就暴露了问题：

1. 429 状态码也被重试。下游模型厂商返回 429（Rate Limit Exceeded）说明我们已经触发了限流，此时立刻重试只会让情况更糟——厂商的限流算法会认为你在持续高频请求，限流窗口越拉越长。
2. 固定 1 秒等待没有意义。有些模型厂商的限流周期是 1 分钟，1 秒后重试等于白给。

2.2 修复后的限流处理策略

# ✅ 区分错误类型，针对性处理importtimeimportrandomfromenumimportEnumclassRetryDecision(Enum):RETRY_IMMEDIATELY="retry_immediately"# 立即重试RETRY_WITH_BACKOFF="retry_with_backoff"# 退避重试DO_NOT_RETRY="do_not_retry"# 不重试，直接降级defclassify_error(status_code:int,error_type:str)->RetryDecision:""" 错误分类 —— 这是限流策略的核心。 不是所有错误都值得重试。 """# 429: 限流 —— 等待后重试ifstatus_code==429:returnRetryDecision.RETRY_WITH_BACKOFF# 5xx: 服务端临时故障 —— 可以重试，但要退避ifstatus_codein(500,502,503):returnRetryDecision.RETRY_WITH_BACKOFF# 4xx: 客户端错误（401 未授权、403 禁止、404 不存在）—— 不重试if400<=status_code<500andstatus_code!=429:returnRetryDecision.DO_NOT_RETRY# 网络超时 / 连接错误 —— 退避重试iferror_typein("timeout","connection_error"):returnRetryDecision.RETRY_WITH_BACKOFFreturnRetryDecision.DO_NOT_RETRY

2.3 指数退避 + 抖动

判断可能是什么也不做重新发，真正的关键在怎么等。我们采用了指数退避 + 随机抖动：

defcalculate_backoff(attempt:int,base_delay:float=2.0,max_delay:float=60.0)->float:""" 指数退避：2^attempt * base_delay 加上随机抖动：±25% 的随机偏移，避免"惊群效应" 退避时间线示例（base_delay=2s）： 第 1 次重试: ~2s 第 2 次重试: ~4s 第 3 次重试: ~8s 上限: 60s """exponential=min(base_delay*(2**attempt),max_delay)jitter=random.uniform(0.75,1.25)# ±25% 抖动returnexponential*jitter

为什么需要抖动？假设 10 个并发请求同时触发重试，如果没有随机抖动，它们会在完全相同的时刻发起第二次请求，在模型厂商眼里就是一个瞬时流量尖峰，再次触发限流。

2.4 读取厂商的限流头信息

很多大模型厂商在响应头中会返回限流信息，读这些信息比盲目等待更可靠：

defextract_rate_limit_info(response_headers:dict)->dict:""" 从响应头中提取限流信息。 不同厂商的头字段名不同，这里以常见的几种为例： - OpenAI: x-ratelimit-limit-requests, x-ratelimit-remaining-requests - Anthropic: anthropic-ratelimit-requests-limit/remaining/reset - 部分国内厂商: x-ratelimit-limit, x-ratelimit-remaining, x-ratelimit-reset """info={"limit":None,"remaining":None,"reset_at":None,}# OpenAI 风格limit=response_headers.get("x-ratelimit-limit-requests")remaining=response_headers.get("x-ratelimit-remaining-requests")# Anthropic 风格ifnotlimit:limit=response_headers.get("anthropic-ratelimit-requests-limit")remaining=response_headers.get("anthropic-ratelimit-requests-remaining")# 通用风格ifnotlimit:limit=response_headers.get("x-ratelimit-limit")remaining=response_headers.get("x-ratelimit-remaining")iflimit:info["limit"]=int(limit)ifremaining:info["remaining"]=int(remaining)# Reset 时间戳reset=response_headers.get("x-ratelimit-reset")ifnotreset:reset=response_headers.get("anthropic-ratelimit-requests-reset")ifreset:info["reset_at"]=resetreturninfo

这样，在 remaining 接近 0 时提前减速，比等到 429 再反应要好得多。

三、踩坑二：重试炸弹 —— 每层都在重试，最终炸了

3.1 谁在帮我重试？

我们的调用链是这样的：

客户端 (axios, timeout=60s, retry=3) ↓ API 网关 (Spring Cloud Gateway, retry=3) ↓ 业务服务 (HTTP Client, connect timeout=10s, read timeout=30s) ↓ 下游模型 API

每一层都配置了自己的超时和重试。一个请求在最坏情况下：

客户端超时 → 触发网关重试 → 每次网关重试又触发业务服务重试

理论最坏重试次数：3 × 3 = 9 次重试，加上初始请求总共 10 次调用。下游直接被打崩。

3.2 解决方案：只在最外层重试

# 规则：重试只在一层做，其余层设足够大的超时但不重试# 网关层: 不重试spring:cloud:gateway:routes:-id:ai-callfilters:-name:Retryargs:retries:0# ← 网关层不重试# 业务服务层: 不重试feign:client:config:default:retryer:feign.Retryer.NEVER_RETRY# ← Feign 不重试

只在最上层（业务代码中的调用器）控制重试逻辑，确保重试次数可控：

defcall_with_retry(prompt:str,max_retries:int=2)->dict:""" 统一的重试入口。 整个调用链只在这里做重试，其他层都是直通。 """forattemptinrange(max_retries+1):decision=classify_error(status_code,error_type)ifdecision==RetryDecision.DO_NOT_RETRY:raiseNonRetryableError(f"不可重试的错误:{status_code}")ifdecision==RetryDecision.RETRY_WITH_BACKOFF:delay=calculate_backoff(attempt)print(f"[重试{attempt+1}/{max_retries}] 等待{delay:.1f}s")time.sleep(delay)

3.3 熔断器：防止重试拖垮上游

重试策略收敛到一层后，还需要加熔断器。当某个下游模型持续不可用，熔断器快速失败，避免上游请求堆积：

fromdataclassesimportdataclassimporttime@dataclassclassCircuitBreaker:""" 简单的滑动窗口熔断器。 逻辑： - 30 秒内失败超过 5 次 → 进入熔断状态（30 秒） - 熔断状态下所有请求直接拒绝，不调用下游 - 30 秒后进入半开状态，尝试恢复 """failure_threshold:int=5recovery_timeout:float=30.0window_duration:float=30.0def__post_init__(self):self.failure_count=0self.last_failure_time=0.0self.state="closed"# closed → open → half_open → closedself.opened_at=0.0defcall(self,func,*args,**kwargs):now=time.time()# 熔断状态下：直接拒绝ifself.state=="open":ifnow-self.opened_at>self.recovery_timeout:self.state="half_open"print("[熔断器] 进入半开状态，尝试恢复...")else:raiseCircuitBreakerOpenError("熔断器已打开，拒绝请求")try:result=func(*args,**kwargs)# 成功：如果之前是半开状态，恢复正常ifself.state=="half_open":self.state="closed"self.failure_count=0print("[熔断器] 恢复成功，关闭熔断")returnresultexceptExceptionase:self.failure_count+=1self.last_failure_time=nowifself.failure_count>=self.failure_threshold:self.state="open"self.opened_at=nowprint(f"[熔断器] 失败{self.failure_count}次，熔断打开")raisee

四、踩坑三：SSE 流式响应的静默中断

4.1 现象

流式对话（Server-Sent Events）是 AI 模型调用中最常见的场景。我们遇到过一种奇怪的现象：前端收到一半的回答突然停了，没有错误，没有 close 事件，就只是"卡住"。

4.2 根因

模型厂商在处理长文本生成（尤其是 3000+ token 的回答）时，中间会有较长的安静期——token 之间间隔可能长达 10~20 秒。我们的 HTTP 连接池默认设置如下：

connect_timeout = 10s read_timeout = 30s

在安静期内，read_timeout触发器超时，连接被中断，但客户端没有收到明确的 FIN/RST 包，导致 hang 住。

4.3 解决方案

# ✅ SSE 流式调用的正确超时配置SSE_CONFIG={"connect_timeout":10,# 建连超时：建立 TCP 连接的最长时间"read_timeout":300,# 读取超时：两次 token 之间的最大间隔（5 分钟）"total_timeout":600,# 总超时：整个生成过程的上限（10 分钟）"heartbeat_interval":15,# 心跳间隔：服务端定期发 comment 行保活}defsse_stream_call(prompt:str):""" SSE 流式调用，加入了保活心跳和读取超时处理。 """importsseclient# 或者用 httpx + 手动解析client=httpx.Client(timeout=httpx.Timeout(connect=SSE_CONFIG["connect_timeout"],read=SSE_CONFIG["read_timeout"],pool=SSE_CONFIG["total_timeout"],))last_token_time=time.time()withclient.stream("POST",url,json=payload,headers=headers)asresponse:forlineinresponse.iter_lines():ifline.startswith("data:"):last_token_time=time.time()data=line[5:].strip()ifdata=="[DONE]":breakyieldjson.loads(data)else:# 服务端发送的保活 comment（如 ": heartbeat"）elapsed=time.time()-last_token_timeifelapsed>SSE_CONFIG["heartbeat_interval"]*2:print(f"[SSE] 警告:{elapsed:.0f}s 未收到 token")

4.4 客户端也要防 hang

前端也需要类似策略：不是永远等待，而是有超时兜底，同时给用户合理的提示。

// 前端 SSE 读取constcontroller=newAbortController();consttimeoutId=setTimeout(()=>{controller.abort();showToast("回复生成超时，请重试或换一个更简单的提问方式");},300_000);// 5 分钟兜底fetch(sseUrl,{signal:controller.signal}).then(async(res)=>{// ... 处理流式数据}).catch((err)=>{if(err.name==="AbortError"){// 超时处理：可以降级到上一轮回答，或提示用户缩短 prompt}}).finally(()=>clearTimeout(timeoutId));

五、踩坑四：模型突然不可用 —— 降级策略

5.1 为什么需要降级

5xx 和 503 不一定意味着模型"坏了"，可能只是暂时负载高。但如果某个模型持续不可用，用户不可能一直等着。我们需要的不是"一个模型挂了全业务停摆"，而是无声切换。

5.2 降级层级

用户请求 "帮我写一段 Python 代码" ↓ 首选: DeepSeek V3.1 (reasoning/code 场景最优) ↓ 失败（503 或超时） 降级 1: 通义千问 Qwen3 (同场景，能力接近) ↓ 失败 降级 2: 文心一言 ERNIE 4.5 ↓ 全部失败 兜底: 返回预生成的通用回答 + 提示"当前服务繁忙"

5.3 代码实现

# 模型降级链配置FALLBACK_CHAIN={"code":["deepseek-v3.1","qwen3-max","ernie-4.5"],"chat":["qwen3-max","doubao-pro-32k","chatglm4"],"translation":["qwen3-max","doubao-pro-32k"],"video":["doubao-seedance","kling-v1.5"],}defcall_with_fallback(prompt:str,scenario:str="chat")->dict:""" 按降级链依次尝试，全部失败则返回兜底回答。 """models=FALLBACK_CHAIN.get(scenario,FALLBACK_CHAIN["chat"])formodel_idinmodels:try:result=call_model(model_id,prompt,max_retries=1)# 成功，记录指标便于后续优化降级链metrics.increment(f"model.{model_id}.success")returnresultexcept(TemporaryFailure,TimeoutError)ase:# 临时故障，记录并尝试下一个metrics.increment(f"model.{model_id}.failure")print(f"[降级]{model_id}不可用 ({e})，尝试下一个...")continueexceptNonRetryableError:# 不可重试错误（如 401），不降级，直接抛raise# 全部模型不可用metrics.increment("fallback.exhausted")returnfallback_response(scenario)

5.4 兜底回答的质量

兜底回答尽量有上下文关联，而不是冷冰冰的"系统错误"：

FALLBACK_TEMPLATES={"code":"当前代码助手服务繁忙。你可以先尝试以下方式：\n""1. 在已有代码中搜索类似实现\n""2. 访问我们的开发者文档：[链接]\n""3. 稍后重试，问题会自动恢复","chat":"AI 服务暂时繁忙，预计 {eta} 分钟内恢复。\n""在此期间，你可以先浏览我们的模型库和价格对比：[链接]","translation":"翻译服务暂时不可用，请稍后重试。",}

六、最终方案：完整的调用器架构

把限流、重试、熔断、降级串起来，形成一套完整的健壮调用器：

请求进入 │ ▼ ┌─────────────────┐ │ 1. 熔断器检查 │ ←─ 如果熔断打开，直接走降级链 └────────┬────────┘ │ closed / half_open ▼ ┌─────────────────┐ │ 2. 模型选择器 │ ←─ 根据场景选首选 + 降级链 └────────┬────────┘ │ ▼ ┌─────────────────┐ │ 3. 限流检查器 │ ←─ 本地令牌桶，QPS 上限保护 └────────┬────────┘ │ 通过 ▼ ┌─────────────────┐ │ 4. 调用远端 API │ └────────┬────────┘ │ ┌─────┴──────┐ │ │ 成功 失败 │ │ ▼ ▼ 返回结果 ┌──────────────┐ │ 5. 错误分类器 │ └──────┬─────────┘ │ ┌────────┼──────────┐ │ │ │ 可重试 不可重试 熔断触发 │ │ │ ▼ ▼ ▼ 指数退避 抛异常 熔断打开 重试 走降级链 走降级链

对应的核心代码骨架：

classRobustAICaller:"""健壮的 AI 模型调用器 —— 集成了熔断、限流、重试、降级"""def__init__(self):# 每个模型独立熔断器self.circuit_breakers={model_id:CircuitBreaker()formodel_idinALL_MODELS}# 本地令牌桶限流（每模型 QPS 上限）self.rate_limiters={model_id:TokenBucket(rate=10,burst=20)formodel_idinALL_MODELS}defcall(self,prompt:str,scenario:str="chat")->dict:models=FALLBACK_CHAIN.get(scenario,FALLBACK_CHAIN["chat"])formodel_idinmodels:# 1. 熔断检查cb=self.circuit_breakers[model_id]# 2. 限流检查ifnotself.rate_limiters[model_id].consume():continue# 当前模型 QPS 用尽，试下一个try:returncb.call(self._do_call,model_id,prompt)exceptCircuitBreakerOpenError:# 熔断打开，降级到下一个模型continueexceptTemporaryFailure:# 临时故障，降级到下一个模型continueexceptNonRetryableError:# 认证错误、权限错误等，这些不应该降级raisereturnfallback_response(scenario)def_do_call(self,model_id:str,prompt:str)->dict:"""实际发起 HTTP 请求，内含指数退避重试"""forattemptinrange(MAX_RETRIES+1):try:response=self._http_post(model_id,prompt)returnresponseexceptRateLimitedase:delay=calculate_backoff(attempt)time.sleep(delay)continueraiseTemporaryFailure(f"{model_id}重试{MAX_RETRIES}次后仍失败")

七、接入星枢无极后，这些策略变成了内置能力

踩完这些坑后，我们把这些策略沉淀到了星枢无极平台中：

对于不想重复踩坑的团队，直接用平台 API 替代原始厂商 API，就能免费获得以上所有策略。5 分钟接入：

# 只需改一个 base_urlcurlhttp://ai.591ll.com/api/v1/chat/completions\-H"Authorization: Bearer YOUR_KEY"\-H"Content-Type: application/json"\-d'{ "model": "deepseek-v3.1", "messages": [{"role": "user", "content": "用 Python 写一个快速排序"}] }'

八、总结

本文由星枢无极团队在生产环境中实战总结。关联阅读：5 分钟接入星枢无极 API，支持 40+ 大模型 | 国内大模型 API 价格一览