笙亿网络策划

先把超时、重试和降级写进配置——GPT-6 Sol 接入前要定的三件事

约 12 分钟读完 1 阅
文章导读
准备把 GPT-6 Sol 接进现有服务时，最该先定下来的不是提示词，而是配置边界：模型名、密钥、请求地址、单次超时、重试上限和降级开关。这几项里只要有两三项写死在代码里，之后换模型、调超时或者关掉重试，都得改代码重新发版。可行的做法是把它们全部抽到环境变量或 YAML 里，代码只读配置、不认字面值，然后拿「只改配置、不改代码」来验证抽象是否真的到位。
📋 目录
  1. A 把模型名、密钥和请求地址抽成配置项
  2. B 给单次请求设超时，并区分连接超时与读取超时
  3. C 重试只放在可重复执行的请求上，并限制次数与退避
  4. D 写一层降级：连续失败或超限时切备用模型或排队
  5. E 在日志里确认一次失败请求的完整链路
A A

准备把 GPT-6 Sol 接进现有服务时，最该先定下来的不是提示词，而是配置边界：模型名、密钥、请求地址、单次超时、重试上限和降级开关。这几项里只要有两三项写死在代码里，之后换模型、调超时或者关掉重试，都得改代码重新发版。可行的做法是把它们全部抽到环境变量或 YAML 里，代码只读配置、不认字面值，然后拿「只改配置、不改代码」来验证抽象是否真的到位。

接入前优先把模型名、密钥、请求地址、连接超时、读取超时、重试上限和降级开关放进配置，而不是写死在调用代码里。判断标准很直接：只改配置、不改代码，就能换模型名、压超时、观察重试与降级是否生效；如果必须动代码才能调整，说明这层抽象没做到位。边界是这些默认值都要结合自身网络与业务确认，不要照搬别人的取值。

把模型名、密钥和请求地址抽成配置项

容易被硬编码的三个字段通常是 base_url、api_key、model。把它们集中到一处，切换和回滚就都变成一次配置变更，配合环境变量还能让本地、预发、线上用不同取值。下面是一份通用骨架，字段名可以按自己项目习惯改，结构照搬即可。

llm:
  sol:
    base_url: ${SOL_BASE_URL}        # 请求地址，路径由客户端拼接
    api_key: ${SOL_API_KEY}          # 只从环境变量注入，不进代码库
    model: ${SOL_MODEL}              # 例如 gpt-6-sol
    connect_timeout_ms: 2000         # 建立连接的最长等待
    read_timeout_ms: 30000           # 连接建立后等待响应体的最长等待
    total_timeout_ms: 45000          # 单次调用的整体上限
    retry:
      max_attempts: 3                # 含首次，通常不超过 3
      backoff_base_ms: 300
      backoff_max_ms: 3000
      jitter: true
    fallback:
      enabled: true
      model: ${SOL_FALLBACK_MODEL}
      consecutive_failures: 5
      window_seconds: 60
      cooldown_seconds: 120
      queue_on_exhausted: true

逐字段看：base_url 和 model 决定打到哪个端点，api_key 只做鉴权；connect_timeout_ms 与 read_timeout_ms 是两个独立的等待上限，建议不要合成一个；total_timeout_ms 用来兜住「重试之后仍然很久」的情况；retry 和 fallback 两组是开关，压测或排查时可以先关掉。

验证方式很直接：服务启动或热加载时，把最终生效的 model 和 base_url 主机名打印到启动日志；然后把 SOL_MODEL 换成另一个值重启，确认日志里的模型名跟着变了、代码一行没动。密钥建议只走环境变量或密钥管理组件，配置文件里出现明文 api_key 通常意味着迟早会被提交进仓库，可以加一条检查，在配置文件里搜一下常见的密钥前缀。

给单次请求设超时，并区分连接超时与读取超时

只记一句「请求失败」是没法排查的。设置两类超时之后，日志里能一眼区分是网络没通，还是服务端已经在处理但没在期限内把响应体给回来。

try:
    resp = session.post(
        url,
        json=payload,
        headers={'Authorization': 'Bearer %s' % api_key},
        timeout=(connect_timeout, read_timeout),  # 元组：先连接，后读取
    )
except ConnectTimeout:
    log_fail(error_type='connect_timeout')   # 连接没建起来
except ReadTimeout:
    log_fail(error_type='read_timeout')      # 连上了，响应体没等到

对应现象：connect_timeout 通常指向 DNS 解析、出口网络、端口不通或 TLS 握手卡住，这类问题重试往往也救不回来；read_timeout 通常指向服务端排队或生成内容过长，重试有意义但要限次数。要注意流式输出场景下，读取超时指的是「两次数据块之间的间隔」还是「整体时长」，取决于客户端实现，接入前需要先确认，否则容易把正常的长回答误判成超时。

验证方式：把 connect_timeout 压到极小值，或者把 read_timeout 压到 1 秒，主动触发一次，再到日志里确认打印的是哪一类超时、elapsed 字段是否与设置接近。这一步做完再调回正常值。

先把超时、重试和降级写进配置——GPT-6 Sol 接入前要定的三件事

重试只放在可重复执行的请求上，并限制次数与退避

重试放错地方，故障时会把压力放大。判断标准是这次调用重复执行会不会带来副作用：单纯文本生成、查询类请求一般可以重试；下单、扣费、发消息这类有写副作用的请求不建议自动重试，应交由业务层决定。已经收到部分流式内容的请求也不适合重试。

适合重试的错误通常包括连接失败、连接超时、未收到任何内容时的读取超时，以及 429、502、503、504 这类过载或网关错误。参数错误（400、422）、鉴权失败（401、403）、资源不存在（404）重试没有意义，应该直接失败并暴露出来。

def call_with_retry(req, cfg):
    last_err = None
    for attempt in range(1, cfg.max_attempts + 1):
        try:
            return do_call(req, timeout=(cfg.connect_timeout, cfg.read_timeout))
        except RetryableError as e:
            last_err = e
            if attempt == cfg.max_attempts:
                raise
            delay = min(cfg.backoff_base_ms * (2 ** (attempt - 1)), cfg.backoff_max_ms)
            if cfg.jitter:
                delay = delay * random.uniform(0.5, 1.0)   # 抖动，避免同时重试
            time.sleep(delay / 1000.0)
    raise last_err

退避用指数增长并加抖动，是为了避免多个实例在同一时刻一起重试。max_attempts 建议不超过 3，每次尝试都要记录 attempt 序号和本次实际等待时长，排查时才有依据。

验证方式：本地起一个 mock 服务固定返回 503，观察客户端日志里出现了几次调用、时间戳之间的间隔是否按预期增长；再把 mock 改成返回 400，确认这次没有重试、直接失败。

写一层降级：连续失败或超限时切备用模型或排队

降级的目标是主链路不因为单一模型不可用而整体挂掉，而不是把失败藏起来。触发条件建议用「窗口内连续失败次数」或「窗口内失败比例」之一，再配一个错误码白名单。触发后按主模型 → 备用模型 → 排队（异步稍后重试）→ 明确快速失败并返回可识别错误码的顺序处理。

先把超时、重试和降级写进配置——GPT-6 Sol 接入前要定的三件事
class Fallback:
    def call(self, req):
        if self.breaker.is_open():                 # 熔断未恢复
            return self.route(req)
        for i in range(self.cfg.max_attempts):
            try:
                return self.call_primary(req)
            except RetryableError:
                pass
        self.breaker.trip(reason='consecutive_failures')
        log('degraded=true from=%s to=%s reason=%s'
            % (self.cfg.primary_model, self.cfg.fallback_model, 'consecutive'))
        return self.route(req)                     # 备用模型或排队

    def route(self, req):
        if self.cfg.queue_on_exhausted:
            return self.enqueue(req, max_wait_ms=self.cfg.max_queue_wait_ms)
        raise UpstreamUnavailable('all models unavailable')

切换动作必须落日志和告警：至少记录 degraded 标记、原模型、备用模型、触发原因和触发时的 request_id。告警建议只在「刚进入降级」和「冷却期结束恢复」各发一次，避免每次请求都发。冷却期内不要频繁试探主模型，否则容易在对方还没恢复时反复失败；恢复策略可以先做半开试探，成功几次后再完全切回。

验证方式：让 mock 连续返回失败，确认日志里出现一次切换记录、后续请求带上了备用模型名；再把 mock 恢复，观察冷却期结束后是否回到主模型。

在日志里确认一次失败请求的完整链路

排查应该是查字段，不是猜。一条请求相关的结构化日志建议包含这些字段：

  • request_id：贯穿全链路的唯一标识，入口生成并透传到每次调用
  • model、base_url_host：本次实际用的模型与端点主机
  • attempt、max_attempts：第几次尝试、上限是多少
  • connect_timeout_ms、read_timeout_ms：本次生效的两类超时值
  • elapsed_ms：本次尝试的耗时
  • status_code、error_type：HTTP 状态码与超时/连接错误分类
  • retry_count：这条请求累计重试次数
  • degraded、fallback_model：是否走了降级、切到了哪个模型
  • queue_wait_ms（可选）：进入排队后的等待时长

拿到 request_id 后按它过滤，就能看到同一条请求的全部尝试：

grep 'request_id=7f3a9c21' app.log
grep 'request_id=7f3a9c21' app.log | grep -E 'attempt|degraded' | wc -l
grep '"degraded":true' app.log | tail -50

如果日志是 JSON 行格式，用 jq 更省事：

jq -c 'select(.request_id=="7f3a9c21")' app.log
jq -c 'select(.request_id=="7f3a9c21") | {attempt, error_type, status_code, elapsed_ms, degraded}' app.log

排查路径按这个顺序走：先看 error_type 是 connect_timeout 还是 read_timeout，确定问题在网络侧还是服务端；再看 status_code 是 429 还是 5xx，判断是否落在可重试范围；然后看 retry_count 是否已到上限，区分「一次就挂」和「重试后仍挂」；最后看 degraded 是否为 true 以及 fallback_model 的值，确认降级是否按预期生效。这四步走完，问题基本能收敛到某个配置项或网络链路上，而不是停在「模型不可用」这种结论上。