在生产环境接入大模型 API 的三个月里我们从调通就行一路踩坑到半夜被报警叫醒再也不用慌。本文复盘实际遇到的限流、重试炸弹、SSE 流式响应超时等问题以及最终沉淀下来的一套策略。一、从一次凌晨 3 点的报警说起事情的起因很简单业务需要同时接入 DeepSeek、通义千问、豆包等多家大模型 API。第一阶段我们快速实现了统一调用层跑通了 demo然后自信地上了生产。第一周风平浪静。第二周某个凌晨 3 点PagerDuty 响了——下游模型调用大面积超时重试机制触发后又造成了雪崩整个链路的延迟从 200ms 飙到 30 秒以上。事后复盘踩了三个坑坑根因后果无差别重试所有错误类型都重试 3 次429限流和 503服务不可用被反复重试加剧下游负载重试无退避失败立即重试短时间内对同一模型厂商发起大量重复请求触发更严格的限流无降级路径主模型挂了只能报错业务中断 40 分钟二、踩坑一限流 —— 不是加 retry 就能解决的2.1 踩坑现场最早的重试逻辑非常简单# ❌ 最初的反面教材importtimeforattemptinrange(3):try:responsecall_model_api(prompt)breakexceptException:time.sleep(1)# 固定等待 1 秒continue这个逻辑在生产上跑了三天就暴露了问题429 状态码也被重试。下游模型厂商返回 429Rate Limit Exceeded说明我们已经触发了限流此时立刻重试只会让情况更糟——厂商的限流算法会认为你在持续高频请求限流窗口越拉越长。固定 1 秒等待没有意义。有些模型厂商的限流周期是 1 分钟1 秒后重试等于白给。2.2 修复后的限流处理策略# ✅ 区分错误类型针对性处理importtimeimportrandomfromenumimportEnumclassRetryDecision(Enum):RETRY_IMMEDIATELYretry_immediately# 立即重试RETRY_WITH_BACKOFFretry_with_backoff# 退避重试DO_NOT_RETRYdo_not_retry# 不重试直接降级defclassify_error(status_code:int,error_type:str)-RetryDecision: 错误分类 —— 这是限流策略的核心。 不是所有错误都值得重试。 # 429: 限流 —— 等待后重试ifstatus_code429:returnRetryDecision.RETRY_WITH_BACKOFF# 5xx: 服务端临时故障 —— 可以重试但要退避ifstatus_codein(500,502,503):returnRetryDecision.RETRY_WITH_BACKOFF# 4xx: 客户端错误401 未授权、403 禁止、404 不存在—— 不重试if400status_code500andstatus_code!429:returnRetryDecision.DO_NOT_RETRY# 网络超时 / 连接错误 —— 退避重试iferror_typein(timeout,connection_error):returnRetryDecision.RETRY_WITH_BACKOFFreturnRetryDecision.DO_NOT_RETRY2.3 指数退避 抖动判断可能是什么也不做重新发真正的关键在怎么等。我们采用了指数退避 随机抖动defcalculate_backoff(attempt:int,base_delay:float2.0,max_delay:float60.0)-float: 指数退避2^attempt * base_delay 加上随机抖动±25% 的随机偏移避免惊群效应 退避时间线示例base_delay2s 第 1 次重试: ~2s 第 2 次重试: ~4s 第 3 次重试: ~8s 上限: 60s exponentialmin(base_delay*(2**attempt),max_delay)jitterrandom.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 风格limitresponse_headers.get(x-ratelimit-limit-requests)remainingresponse_headers.get(x-ratelimit-remaining-requests)# Anthropic 风格ifnotlimit:limitresponse_headers.get(anthropic-ratelimit-requests-limit)remainingresponse_headers.get(anthropic-ratelimit-requests-remaining)# 通用风格ifnotlimit:limitresponse_headers.get(x-ratelimit-limit)remainingresponse_headers.get(x-ratelimit-remaining)iflimit:info[limit]int(limit)ifremaining:info[remaining]int(remaining)# Reset 时间戳resetresponse_headers.get(x-ratelimit-reset)ifnotreset:resetresponse_headers.get(anthropic-ratelimit-requests-reset)ifreset:info[reset_at]resetreturninfo这样在 remaining 接近 0 时提前减速比等到 429 再反应要好得多。三、踩坑二重试炸弹 —— 每层都在重试最终炸了3.1 谁在帮我重试我们的调用链是这样的客户端 (axios, timeout60s, retry3) ↓ API 网关 (Spring Cloud Gateway, retry3) ↓ 业务服务 (HTTP Client, connect timeout10s, read timeout30s) ↓ 下游模型 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:int2)-dict: 统一的重试入口。 整个调用链只在这里做重试其他层都是直通。 forattemptinrange(max_retries1):decisionclassify_error(status_code,error_type)ifdecisionRetryDecision.DO_NOT_RETRY:raiseNonRetryableError(f不可重试的错误:{status_code})ifdecisionRetryDecision.RETRY_WITH_BACKOFF:delaycalculate_backoff(attempt)print(f[重试{attempt1}/{max_retries}] 等待{delay:.1f}s)time.sleep(delay)3.3 熔断器防止重试拖垮上游重试策略收敛到一层后还需要加熔断器。当某个下游模型持续不可用熔断器快速失败避免上游请求堆积fromdataclassesimportdataclassimporttimedataclassclassCircuitBreaker: 简单的滑动窗口熔断器。 逻辑 - 30 秒内失败超过 5 次 → 进入熔断状态30 秒 - 熔断状态下所有请求直接拒绝不调用下游 - 30 秒后进入半开状态尝试恢复 failure_threshold:int5recovery_timeout:float30.0window_duration:float30.0def__post_init__(self):self.failure_count0self.last_failure_time0.0self.stateclosed# closed → open → half_open → closedself.opened_at0.0defcall(self,func,*args,**kwargs):nowtime.time()# 熔断状态下直接拒绝ifself.stateopen:ifnow-self.opened_atself.recovery_timeout:self.statehalf_openprint([熔断器] 进入半开状态尝试恢复...)else:raiseCircuitBreakerOpenError(熔断器已打开拒绝请求)try:resultfunc(*args,**kwargs)# 成功如果之前是半开状态恢复正常ifself.statehalf_open:self.stateclosedself.failure_count0print([熔断器] 恢复成功关闭熔断)returnresultexceptExceptionase:self.failure_count1self.last_failure_timenowifself.failure_countself.failure_threshold:self.stateopenself.opened_atnowprint(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 手动解析clienthttpx.Client(timeouthttpx.Timeout(connectSSE_CONFIG[connect_timeout],readSSE_CONFIG[read_timeout],poolSSE_CONFIG[total_timeout],))last_token_timetime.time()withclient.stream(POST,url,jsonpayload,headersheaders)asresponse:forlineinresponse.iter_lines():ifline.startswith(data:):last_token_timetime.time()dataline[5:].strip()ifdata[DONE]:breakyieldjson.loads(data)else:# 服务端发送的保活 comment如 : heartbeatelapsedtime.time()-last_token_timeifelapsedSSE_CONFIG[heartbeat_interval]*2:print(f[SSE] 警告:{elapsed:.0f}s 未收到 token)4.4 客户端也要防 hang前端也需要类似策略不是永远等待而是有超时兜底同时给用户合理的提示。// 前端 SSE 读取constcontrollernewAbortController();consttimeoutIdsetTimeout((){controller.abort();showToast(回复生成超时请重试或换一个更简单的提问方式);},300_000);// 5 分钟兜底fetch(sseUrl,{signal:controller.signal}).then(async(res){// ... 处理流式数据}).catch((err){if(err.nameAbortError){// 超时处理可以降级到上一轮回答或提示用户缩短 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:strchat)-dict: 按降级链依次尝试全部失败则返回兜底回答。 modelsFALLBACK_CHAIN.get(scenario,FALLBACK_CHAIN[chat])formodel_idinmodels:try:resultcall_model(model_id,prompt,max_retries1)# 成功记录指标便于后续优化降级链metrics.increment(fmodel.{model_id}.success)returnresultexcept(TemporaryFailure,TimeoutError)ase:# 临时故障记录并尝试下一个metrics.increment(fmodel.{model_id}.failure)print(f[降级]{model_id}不可用 ({e})尝试下一个...)continueexceptNonRetryableError:# 不可重试错误如 401不降级直接抛raise# 全部模型不可用metrics.increment(fallback.exhausted)returnfallback_response(scenario)5.4 兜底回答的质量兜底回答尽量有上下文关联而不是冷冰冰的系统错误FALLBACK_TEMPLATES{code:当前代码助手服务繁忙。你可以先尝试以下方式\n1. 在已有代码中搜索类似实现\n2. 访问我们的开发者文档[链接]\n3. 稍后重试问题会自动恢复,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(rate10,burst20)formodel_idinALL_MODELS}defcall(self,prompt:str,scenario:strchat)-dict:modelsFALLBACK_CHAIN.get(scenario,FALLBACK_CHAIN[chat])formodel_idinmodels:# 1. 熔断检查cbself.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_RETRIES1):try:responseself._http_post(model_id,prompt)returnresponseexceptRateLimitedase:delaycalculate_backoff(attempt)time.sleep(delay)continueraiseTemporaryFailure(f{model_id}重试{MAX_RETRIES}次后仍失败)七、接入星枢无极后这些策略变成了内置能力踩完这些坑后我们把这些策略沉淀到了星枢无极平台中你原来需要自己做的接入星枢无极后逐个申请各厂商 API Key一个 Key调用 40 模型处理各厂商不同格式的限流头平台统一限流信息透传写降级链逻辑内置模型降级链一个模型挂了自动切处理不同协议的 SSE 流式统一的 OpenAI / Anthropic 协议流式输出配置熔断器和重试策略平台侧已实现开箱即用监控各模型的可用性和延迟统一 Dashboard 可视化对于不想重复踩坑的团队直接用平台 API 替代原始厂商 API就能免费获得以上所有策略。5 分钟接入# 只需改一个 base_urlcurlhttp://ai.591ll.com/api/v1/chat/completions\-HAuthorization: Bearer YOUR_KEY\-HContent-Type: application/json\-d{ model: deepseek-v3.1, messages: [{role: user, content: 用 Python 写一个快速排序}] }八、总结策略一句话要点影响最大的场景错误分类429 和 401 的处理方式完全不同永远不要无差别重试限流恢复指数退避 抖动2^n × base random_jitter避免惊群并发重试熔断器下游持续故障时快速失败保护上游资源级联故障只在最外层重试多层重试会放大请求量到不可控请求风暴降级链一个模型不可用自动切到能力相近的备用模型业务连续性SSE 超时控制读超时和总超时分开设置长文本生成不容易 hang流式对话兜底回答全部降级链耗尽时有体面的退路用户体验底限本文由星枢无极团队在生产环境中实战总结。关联阅读5 分钟接入星枢无极 API支持 40 大模型 | 国内大模型 API 价格一览