凌晨两点订单服务的压测还没结束日志里HTTP 429已经从零星的几条变成每秒上百条。我们这次把 DeepSeek-V4.1-Flash 接进高并发链路入口统一走 TaoTokenhttps://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_contentretry429 Key 从同一个地方领请求地址统一设成 https://taotoken.net/api。第一轮排查其实跑偏了。一开始以为是 Key 本身有问题换了两三次 Key429 曲线纹丝不动又把客户端并发从 64 压到 32也只是把峰值往后推了十几分钟。真正有用的线索在响应里——retry-after、上游请求 ID以及我们自己日志里那条被截断的响应体。把这三样对齐之后才发现问题不是额度不够而是重试风暴叠加熔断器缺失上游一次正常的限流信号被客户端放大成了四到六倍的无效请求越重试越堵越堵越重试。这篇文章把复盘过程按可跟做的顺序写下来产出三样东西一套能直接落地的重试退避参数、一个三态熔断器实现、一份能用来定位限流根因的 429 日志字段清单。它不解决模型推理速度问题只解决请求发出去的这一层是否稳。1. 先固定故障面一条能定位问题的 429 日志高并发下最怕的不是报错而是报错没有上下文。429本身只说明这一秒不该发这个请求但到底是并发上限、单位时间请求数还是请求体太大导致排队全靠日志里那几个字段区分。我们最终统一的日志结构是这样{ ts: 2025-06-11T02:14:37.41208:00, trace_id: b7c1e2f0a94d, upstream_request_id: req_9f2a7c31, endpoint: chat.completions, model: deepseek-v4.1-flash, attempt: 3, status: 429, retry_after_ms: 1000, breaker_state: CLOSED, inflight: 78, stream: true, elapsed_ms: 812, error_code: rate_limit_exceeded }几个字段的作用要分清attempt记录这是第几次尝试。如果日志里attempt长期大于 2说明不是上游在限你是你在放大流量。retry_after_ms直接来自响应头。有它就用它不要自己拍一个固定值。breaker_state记录熔断器当时的状态。没有这一列你无法区分上游拒绝和我们主动拒绝。inflight是发出请求那一刻的在途数量。限流和并发数几乎总是正相关这个值能把两者对上。upstream_request_id是跟平台侧对账的唯一凭据出问题时要能报出来。写完日志结构之后要做的第一件事是把它接进采集。我们只用一个很小的字典计数器就够了不需要引入重型依赖import logging from collections import Counter log logging.getLogger(taotoken.retry) STATS Counter() def note_retry(status: int, attempt: int, breaker_state: str) - None: STATS[requests_total] 1 if status 429: STATS[http_429_total] 1 STATS[retry_amplification] attempt - 1 if breaker_state OPEN: STATS[breaker_reject_total] 1 if STATS[requests_total] % 1000 0: log.warning(retry stats snapshot: %s, dict(STATS))压测跑完如果retry_amplification / http_429_total大于 2基本可以判定重试策略过激。2. 把入口收敛到 TaoTokenBase URL 与 Key 的最小改动故障面固定之后第二步是把请求地址和密钥来源统一。这一步越简单越好因为改动越小越容易把限流问题和配置问题区分开。Key 在 TaoToken 控制台领取具体入口是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_contentbaseurl 申请完成后在 API Keys 页面创建即可。请求地址是固定的export TAOTOKEN_BASE_URLhttps://taotoken.net/api export TAOTOKEN_API_KEYYOUR_API_KEY调用侧用 OpenAI 兼容客户端时只需要把base_url指过去import httpx from openai import OpenAI client OpenAI( base_urlhttps://taotoken.net/api, api_keyYOUR_API_KEY, timeouthttpx.Timeout(connect5.0, write30.0, pool2.0, read180.0), ) resp client.chat.completions.create( modeldeepseek-v4.1-flash, messages[{role: user, content: 连通性自检}], max_tokens32, ) print(resp.choices[0].message.content)超时这里值得单独说一句。长上下文请求的 prefill 阶段耗时明显高于短请求如果你沿用短请求那套 30 秒总超时会出现大量客户端主动断开、但上游仍在计算的幽灵请求。这些请求照样占并发名额表现就是客户端看到的是超时上游看到的是并发没降下来。所以拆成connect / write / pool / read四段超时是长上下文链路的标配不是可选项。配置改完之后先跑连通性自检再跑并发压测两者之间要有明确的日志分界点方便回滚。3. 重试策略区分 429、5xx 和超时最常见的错误写法是只要失败就重试三次间隔 1 秒。在限流场景里这等于自己给自己叠了三倍流量。正确的做法是先分类再决定要不要重试429可重试但必须尊重retry-after且要计入熔断器失败计数。500 / 502 / 503 / 504可重试用指数退避加抖动。连接超时、读超时可重试但要区分是否是长上下文导致的正常慢。400 / 401 / 403 / 404不可重试重试只会浪费配额。422或参数校验类错误不可重试直接落到业务异常。下面这段是可以直接用的重试包装退避参数按我们压测后的取值给import asyncio import logging import random from dataclasses import dataclass import httpx log logging.getLogger(taotoken.retry) RETRYABLE_STATUS {429, 500, 502, 503, 504} RETRYABLE_EXC ( httpx.ConnectTimeout, httpx.ReadTimeout, httpx.WriteTimeout, httpx.RemoteProtocolError, ) dataclass class RetryPolicy: max_attempts: int 4 base_delay: float 0.35 max_delay: float 8.0 jitter_ratio: float 0.3 def next_delay(self, attempt: int, retry_after_s: float | None) - float: if retry_after_s is not None: return min(retry_after_s, self.max_delay) raw min(self.base_delay * (2 ** (attempt - 1)), self.max_delay) return raw * (1 random.uniform(-self.jitter_ratio, self.jitter_ratio)) def parse_retry_after(resp: httpx.Response) - float | None: value resp.headers.get(retry-after) if not value: return None try: return float(value) except ValueError: return None async def post_json( client: httpx.AsyncClient, path: str, payload: dict, policy: RetryPolicy, on_retryNone, ) - httpx.Response: last_exc None for attempt in range(1, policy.max_attempts 1): try: resp await client.post(path, jsonpayload) if resp.status_code not in RETRYABLE_STATUS: return resp if attempt policy.max_attempts: return resp delay policy.next_delay(attempt, parse_retry_after(resp)) log.warning( retryable status%s attempt%s sleep%.2fs body%s, resp.status_code, attempt, delay, resp.text[:200], ) if on_retry: on_retry(resp.status_code, attempt) await asyncio.sleep(delay) except RETRYABLE_EXC as exc: last_exc exc if attempt policy.max_attempts: raise delay policy.next_delay(attempt, None) log.warning(retryable exc%r attempt%s sleep%.2fs, exc, attempt, delay) if on_retry: on_retry(-1, attempt) await asyncio.sleep(delay) raise last_exc # pragma: no cover两个细节容易被忽略。第一抖动必须有否则所有客户端会在同一个毫秒点同时回来形成二次尖峰。第二max_attempts4是上限不是目标如果日志里三次重试的占比超过 5%说明入口并发就该往下调而不是继续加尝试次数。4. 熔断器让停下来成为一种主动决策重试解决的是偶发失败熔断解决的是持续性失败。两者的边界很清楚上游偶发 429重试能救回来上游连续 429重试只会把连接池耗尽把整个服务拖垮。我们用的是一个标准三态机import time from enum import Enum class State(str, Enum): CLOSED CLOSED OPEN OPEN HALF_OPEN HALF_OPEN class CircuitBreaker: def __init__( self, failure_threshold: int 20, window_s: float 10.0, open_hold_s: float 15.0, half_open_probes: int 3, half_open_success_needed: int 3, ): self.failure_threshold failure_threshold self.window_s window_s self.open_hold_s open_hold_s self.half_open_probes half_open_probes self.half_open_success_needed half_open_success_needed self.state State.CLOSED self._failures: list[float] [] self._opened_at 0.0 self._probes 0 self._successes 0 self._probe_inflight 0 def _prune(self, now: float) - None: cutoff now - self.window_s self._failures [t for t in self._failures if t cutoff] def allow(self) - bool: now time.monotonic() if self.state is State.OPEN: if now - self._opened_at self.open_hold_s: self.state State.HALF_OPEN self._probes 0 self._successes 0 self._probe_inflight 0 else: return False if self.state is State.HALF_OPEN: if self._probe_inflight self.half_open_probes: return False self._probe_inflight 1 return True return True def on_success(self) - None: if self.state is State.HALF_OPEN: self._probe_inflight max(0, self._probe_inflight - 1) self._successes 1 if self._successes self.half_open_success_needed: self.state State.CLOSED self._failures.clear() else: self._prune(time.monotonic()) def on_failure(self) - None: now time.monotonic() if self.state is State.HALF_OPEN: self._probe_inflight max(0, self._probe_inflight - 1) self._opened_at now self.state State.OPEN return self._failures.append(now) self._prune(now) if len(self._failures) self.failure_threshold: self._opened_at now self.state State.OPEN几个参数是压测出来的不是拍脑袋failure_threshold20配合window_s10意味着 10 秒内 20 次失败就开闸open_hold_s15大约是长上下文请求平均完成时间的 3 倍保证半开探测发出去之前老请求已经落地半开只放 3 个探测且需要连续 3 次成功才闭合避免上游刚恢复就被打回原形。半开阶段是最容易写错的地方如果探测请求不限流几十个协程会同时挤进HALF_OPEN等于开闸瞬间又来一次尖峰。所以allow()里的_probe_inflight计数不能省。5. 并发闸门客户端自己也要有上限重试和熔断都是事后控制真正的第一道闸门是并发数。我们的做法是把请求分成两档import asyncio class TwoTierGate: 短请求与长上下文请求分开限流避免长请求把短请求的连接占满。 def __init__(self, short_limit: int 48, long_limit: int 12): self.short_sem asyncio.Semaphore(short_limit) self.long_sem asyncio.Semaphore(long_limit) def pick(self, prompt_tokens: int) - asyncio.Semaphore: return self.long_sem if prompt_tokens 32_000 else self.short_sem async def run(self, prompt_tokens: int, coro): sem self.pick(prompt_tokens) async with sem: return await coro这样拆的原因是长上下文请求的 read 时间可能是短请求的几十倍混在同一个信号量里长请求会把并发名额占住短请求排队到超时。分开之后两档各自拥塞互不牵连。闸门数值的确定方法很简单从单档 8 并发开始每轮加 8直到 429 占比超过 2% 就回退一档。这个值在不同业务里差别很大别抄别人的数字。6. 客户端侧配置Claude Code、Codex 与 CC Switch除了自研调用链路很多团队的稳定性排查还涉及命令行工具。这三套配置的字段完全不同混用是最常见的踩坑点。Claude Code 走的是settings.json环境变量前缀是ANTHROPIC_{ env: { ANTHROPIC_BASE_URL: https://taotoken.net/api, ANTHROPIC_AUTH_TOKEN: YOUR_API_KEY, ANTHROPIC_MODEL: deepseek-v4.1-flash } }Codex 走的是config.toml字段名和 Claude Code 没有一处相同不要照搬ANTHROPIC_*model deepseek-v4.1-flash model_provider taotoken [model_providers.taotoken] name TaoToken base_url https://taotoken.net/api env_key TAOTOKEN_API_KEY wire_api chatenv_key只写变量名真正的值放在 shell 环境里export TAOTOKEN_API_KEYYOUR_API_KEY如果团队里有人用 CC Switch 做多供应商切换那就当成三件套来配供应商名称、Base URL、模型名。三件套里任何一项填错表现都是 401 或者 404而不是 429所以遇到 429 时先别急着怀疑这里。配完之后建议先做一次单请求验证再进压测环境。验证阶段如果就报错说明是配置问题不该进入稳定性排查流程。7. 长上下文请求的专属注意事项这一版的模型在长上下文上做了不少优化KV 缓存占用下降、跨层复用这类机制让长文本更省资源但对客户端来说有两点必须自己处理第一首字节超时和总超时要分开。长 prompt 的 prefill 阶段可能几十秒都不返回第一个 token如果你设的总超时是 60 秒那么大量请求会在还在算的时候被客户端判定失败。失败的请求被重试重试又占并发这是典型的自伤。# 流式场景读超时放宽但对首包单独计时 first_token_timeout_s 45.0 stream client.chat.completions.create( modeldeepseek-v4.1-flash, messages[{role: user, content: long_prompt}], streamTrue, )第二长上下文请求不要和短请求共用一个熔断器失败窗口。长请求更容易因为超时被记为失败从而误触发熔断把本来健康的短请求链路一起切断。我们按prompt_tokens分了两条独立计数路径逻辑和上面的双档闸门一致。8. 把 429 变成可执行的结论日志、指标、闸门都到位之后最后一步是把结论固定成一张表。我们用的判定规则是这样的现象大概率原因动作429 占比 1%重试 1 次后成功正常波动不动429 占比 1%~5%attempt 集中在 2并发略高并发下调 1 档429 占比 5%retry_amplification 2重试放大流量先降重试次数再降并发熔断器频繁 OPEN恢复后很快再 OPEN半开探测过宽收紧half_open_probes客户端超时多但上游无对应记录读超时过短按 prefill 耗时重设读超时401/404 与 429 同时出现配置问题混入先修配置再谈限流这张表的价值在于它把要不要加机器要不要换 Key这类拍脑袋的讨论换成了一次数据比对。9. 上线前的验收清单压测通过不等于线上稳定下面这几项是我们每次改配置都会重新跑一遍的单请求连通性验证通过且日志里能看到上游请求 ID。稳态压测 10 分钟429 占比稳定在 2% 以内且不随时间上升。手动注入 429把上游响应伪造为限流确认熔断器在 10 秒内开闸且开闸后不再发出无效请求。熔断后等待open_hold_s确认半开探测只放行约定数量的请求。长上下文请求与短请求同时打确认两档互不影响。关闭全部重试观察 429 基线再逐级打开确认放大系数可控。检查日志里attempt、breaker_state、inflight三个字段无缺失。其中第 3 项最容易被跳过也最有用。很多团队的熔断器写了但从来没被触发过真正出问题时才发现状态机有 bug。整套流程里唯一需要改地址的地方只有一处把请求地址指向 https://taotoken.net/api Key 用自己的。剩下的重试、熔断、闸门、日志全是客户端自己的工程问题和模型本身无关。把这一层做扎实换模型、换版本、换并发档位都不会让链路重新抖一次。如果要从零开始按这个顺序走最省事先去 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_contentflow 领 Key创建好之后用模型对话页做一次单请求验证https://taotoken.net/models/detail/chat?utm_sourcetaotoken_aicg_blog_endutm_contentflow 确认链路通了再决定是走 Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentflow 还是直接建 Key 接入https://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentflow 。命令行工具侧的完整字段说明在 Claude Code 文档里https://taotoken.net/doc/ClaudeCodeAnthropic?utm_sourcetaotoken_aicg_blog_endutm_contentflow 配置前先对一遍字段名比事后翻 401 日志快得多。