大模型 API 的生产可靠性:限流、重试与故障切换

在 notebook 里调一次模型很简单。跑真实流量才见可靠性:上游有瞬时错误、限流,偶尔还会慢响应。下面这些模式能让你的应用稳住,而且不管你调 OpenAI 还是通过 OpenAI 兼容网关调国产模型,套路都一样。

1. 指数退避重试

瞬时的 429(限流)和 5xx 错误应当重试,并逐次拉大延迟,免得猛锤一个本就吃力的上游。

import time, os
from openai import OpenAI, APIError, RateLimitError

client = OpenAI(
    api_key=os.environ["TURILOOP_API_KEY"],
    base_url="https://api.turiloop.com/v1",
)

def call_with_retry(model, messages, retries=4):
    for attempt in range(retries):
        try:
            return client.chat.completions.create(
                model=model, messages=messages, timeout=60,
            )
        except (RateLimitError, APIError) as e:
            if attempt == retries - 1:
                raise
            time.sleep(2 ** attempt)   # 1s, 2s, 4s, 8s

2. 永远设超时

没有超时的请求可能永远挂住、堵死一个 worker。设上(对话比如 60s,图像生成更长),并把超时当作可重试错误。

3. 兜底到第二个模型

如果主模型在失败或饱和,优雅降级到另一个。每个模型只差一个 model 字符串,故障切换很简单:

def robust_call(messages):
    for model in ["deepseek-v4-pro", "glm-5.1", "deepseek-v4-flash"]:
        try:
            return call_with_retry(model, messages)
        except Exception:
            continue   # 试下一个模型
    raise RuntimeError("所有模型都失败了")

一个 OpenAI 兼容的 Key 同时够到 DeepSeek、GLM、Kimi 和 Claude,让这成为真实、廉价的安全网,而不是又一次接入。

4. 幂等与部分失败处理

• agent 循环要设迭代上限,免得一个卡住的会话把账单跑飞。
• 每次调用记下 model 和 token 用量,便于对账和发现异常。
• 非流式批处理任务,让重试幂等(同输入、同去重键),重试不会重复处理。

5. 让网关帮你分担

好的中转本身已经替你做了一部分:上游层的多通道路由和故障切换,单个供应商抖动不会在你的应用里冒成错误。Turiloop 在一个 Key 背后跑多上游冗余 + 自动切换。你仍然要加应用层重试和兜底模型,但下限更高。

把这四个模式做一次、封进一个 helper,你的大模型调用就能在真实负载下稳住。国际卡、按量付费。