OpenAI 客户端迁移：保留 RPM、TPM 和重试行为

一次OpenAI 客户端迁移在完成两项配置更改后看起来就已经完成了：替换 API 密钥，并将 SDK 指向新的基础 URL。第一次请求成功，响应结构看起来很熟悉，PR 似乎也已经可以合并。

这只能证明接口兼容性，并不能证明生产行为。

一次OpenAI 客户端迁移更难的部分，是在流量变得不均匀时仍然保留原有行为：请求成批到达、提示词变长、流式响应持续时间超出预期、提供方返回 429，或者响应在工作可能已经开始后才超时。如果 SDK、你的应用以及任务队列都各自独立重试，那么一次失败的调用就可能变成几次几乎同时发生的尝试。

本教程展示如何将现有的 OpenAI 风格 Python 或 TypeScript 集成迁移到统一网关，同时将限流和重试行为显式化。示例使用 Flatkey 兼容 OpenAI 的基础 URL，但这种审查方法适用于任何网关迁移。

快速回答：应该改变什么？

为了安全地进行OpenAI 客户端迁移，应将这些设置一起审查，而不是把基础 URL 当作全部变更。

如果你需要先了解缩写层面的解释，请阅读 LLM 限流解释：RPM、TPM 和重试。本指南从那篇说明文结束的地方开始：迁移差异和生产测试计划。

为什么只替换基础 URL 是必要但不充分的

Flatkey 的快速入门文档说明了最小客户端改动：保留 OpenAI SDK 的请求模式，并将基础 URL 设置为 https://router.flatkey.ai/v1。它还建议在请求后检查 Usage Logs，这样你就可以验证模型、token 数、延迟和成本。

这确实是正确的冒烟测试。生产环境中的OpenAI 客户端迁移还需要额外回答四个问题：

1. SDK 是否会自动重试 429、超时或服务器错误？
2. 是否还有另一层也会对同一个失败操作进行重试？
3. 并发是受请求速率、令牌速率，还是两者共同限制？
4. 你能否区分一个逻辑操作及其各次单独尝试？

官方 OpenAI Python 和 Node SDK 文档目前说明，默认会对选定的失败重试两次，包括 429 响应、连接错误、超时以及某些服务器错误。两个 SDK 都提供重试和超时配置。对于直接集成来说，这个默认设置很方便，但如果你自己的代码已经实现了退避机制，它可能会变成不可见的放大效应。

迁移目标不是“禁用所有重试”。目标是“知道哪一层负责重试”。

步骤 1：在更改代码之前盘点每一层重试

先画出真实的调用路径。

用户操作或任务
  -> 应用重试包装器
  -> 队列投递重试
  -> OpenAI SDK 重试
  -> 网关
  -> 提供方

对于每一层，记录：

• 哪些错误会触发下一次尝试。
• 最大尝试次数。
• 延迟是使用固定睡眠、指数退避还是抖动。
• 是否会遵守服务器提供的 Retry-After 值。
• 同一个操作标识是否会在各次尝试之间保持不变。
• 超时请求是否会被假定为在任何工作开始前就已失败。

最后一项假设是有风险的。客户端超时只说明客户端停止等待了。上游系统可能仍然已经接受或完成了请求。对于生成内容而言，因此一次重试可能会产生另一个结果和另一笔计费请求，即使你的应用只观察到一个逻辑任务。

估算最坏情况下的放大效应

假设队列可以投递一个任务三次，应用包装器允许三次尝试，而 SDK 执行初始调用加两次重试。在最坏情况下，一个逻辑任务可能触发：

3 次队列投递 × 3 次应用尝试 × 3 次 SDK 尝试 = 27 次 HTTP 尝试

你也许永远不会达到这个完整数字，但这个乘法关系解释了为什么一个短暂的 429 会演变成重试风暴。把这个数字写进迁移评审中。它能让隐藏的默认行为变得可见。

步骤 2：将端点设置移到配置中

保持迁移差异可逆。不要在整个代码库中到处替换端点字符串。

Python 迁移前后

import os
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["LLM_API_KEY"],
    base_url=os.environ.get("LLM_BASE_URL", "https://api.openai.com/v1"),
    max_retries=0,
    timeout=45.0,
)

对于 Flatkey 金丝雀环境，配置：

export LLM_API_KEY="$FLATKEY_API_KEY"
export LLM_BASE_URL="https://router.flatkey.ai/v1"

TypeScript 迁移前后

import OpenAI from "openai";

const client = new OpenAI({
  apiKey: process.env.LLM_API_KEY,
  baseURL: process.env.LLM_BASE_URL ?? "https://api.openai.com/v1",
  maxRetries: 0,
  timeout: 45_000,
});

这些示例将 SDK 重试次数设置为零，因为下一节会让应用程序明确负责重试。如果你的应用程序没有重试层，那么也可以保留受限的 SDK 重试。不要不小心两者都保留。

有关更广泛的兼容性检查清单，请参见 OpenAI Compatible API Gateway：最小代码变更的迁移检查清单。

步骤 3：让一个层明确负责重试

一个有用的重试策略包含五个部分：

1. 一小组可重试的失败。
2. 严格的尝试上限。
3. 最大的总重试时间。
4. 带抖动的指数退避。
5. 每次尝试都有结构化日志。

下面是一个用于同步聊天调用的小型 Python 封装：

import random
import time
from openai import APITimeoutError, APIConnectionError, APIStatusError, RateLimitError

RETRYABLE_STATUS_CODES = {408, 409, 429, 500, 502, 503, 504}


def create_chat_with_retry(client, *, model, messages, max_attempts=4):
    started_at = time.monotonic()

    for attempt in range(1, max_attempts + 1):
        try:
            return client.chat.completions.create(
                model=model,
                messages=messages,
            )
        except (RateLimitError, APITimeoutError, APIConnectionError) as error:
            retryable = True
            caught_error = error
        except APIStatusError as error:
            retryable = error.status_code in RETRYABLE_STATUS_CODES
            caught_error = error

        if not retryable or attempt == max_attempts:
            raise caught_error

        exponential_delay = min(2 ** (attempt - 1), 16)
        jitter = random.uniform(0, 0.5 * exponential_delay)
        sleep_seconds = exponential_delay + jitter

        print({
            "event": "llm_retry",
            "attempt": attempt,
            "next_delay_seconds": round(sleep_seconds, 2),
            "elapsed_seconds": round(time.monotonic() - started_at, 2),
            "error_type": type(caught_error).__name__,
        })
        time.sleep(sleep_seconds)

    raise RuntimeError("unreachable")

将其视为一个可审查的起点，而不是通用策略。在生产环境中，在回退到本地计算的延迟之前，先解析并遵循有效的 Retry-After 响应头。再添加一个总耗时预算，这样重试就不会超过你的产品可承受的延迟。

不要对每个错误都重试

Flatkey 的快速入门对于 429 给出了相同的高层建议：使用带抖动的指数退避进行重试。迁移相关的新增要求是，确保只有一层执行该策略。

步骤 4：同时为 RPM 和 TPM 设定并发量

OpenAI 客户端迁移可以在保持请求语法不变的同时改变容量边界。RPM 和 TPM 约束的是不同工作负载：

• 当你发送大量小请求时，RPM 会成为瓶颈。
• 当提示词、输出或并行评估规模较大时，TPM 会成为瓶颈。

应使用实际观测到的流量，而不是单一平均值。至少收集：

• 中位数和峰值时的每分钟请求数。
• p50、p95 和最大值的输入 token 数。
• p50 和 p95 的输出 token 数。
• 请求持续时间的中位数和 p95。
• 并发流的数量。

可以根据每个限制粗略估算并发上限：

基于 RPM 的并发 ≈ (RPM / 60) × 平均请求秒数

基于 TPM 的并发 ≈ (TPM / 每个请求的平均 token 数 / 60)
                        × 平均请求秒数

取较低的结果作为初始上限，然后为突发流量和重试预留余量。

示例：假设某个路由允许 600 RPM 和 300,000 TPM，平均每个请求使用 1,500 个总 token，平均时长为 3 秒。

RPM 上限：(600 / 60) × 3 = 30 个并发请求
TPM 上限：(300,000 / 1,500 / 60) × 3 = 10 个并发请求

在这个示例中，TPM 是更严格的约束。如果因为 RPM 看起来很充裕就从 30 个并发请求开始，会产生本可避免的 429 响应。

这个计算只是方向性的，并非提供方保证。提供方可能使用滚动窗口、令牌桶、输入和输出 token 分离限制、特定模型池或加速控制。测试计划必须验证所选模型和账户的真实行为。

步骤 5：单独测试流式传输和超时行为

不要把一次成功的非流式调用当作流式传输安全的证据。

对于流式请求，请测试：

• 首个 token 的到达时间。
• 分块之间的最长空闲间隔。
• 客户端读取超时。
• 当消费者断开连接时的行为。
• 你的重试封装是否会意外启动第二个流。
• 部分输出是保留、丢弃，还是展示给用户。

一个在输出部分内容后失败的流，与在任何输出之前失败的请求并不等价。自动重试可能会显示重复文本，或者产生不同的续写。应决定产品是重试、询问用户，还是直接呈现部分结果。

还要记住，SDK 超时和基础设施超时可能不同。反向代理、无服务器平台、任务 worker 或浏览器连接，可能会在客户端库达到自身超时之前就终止。在OpenAI 客户端迁移期间，记录完整请求路径中最短的超时时间。

步骤 6：在大规模流量前运行金丝雀矩阵

使用一个固定模型和一小部分流量。首次金丝雀应回答新路由是否保留行为，而不是每个模型是否都能工作。

对于每个逻辑操作，记录：

operation_id
attempt_number
base_url_name
model_requested
http_status
input_tokens
output_tokens
latency_ms
retry_delay_ms
final_outcome

然后将应用日志与 Flatkey Usage Logs 进行比较。计数应当彼此吻合。如果一个应用操作映射到多个网关请求，你的重试埋点应解释原因。

步骤 7：定义上线和回滚阈值

OpenAI 客户端迁移应在第一次金丝雀开始前就设定数值化停止条件。

示例阈值：

• 如果最终错误率上升超过约定的百分点，则回滚。
• 如果每个操作的尝试次数超过预期的重试预算，则暂停。
• 如果 p95 延迟超过产品超时预算，则暂停。
• 如果每个成功操作的 token 用量出现意外变化，则暂停。
• 仅在流式和非流式路径都通过后再扩大流量。

避免只比较原始 429 计数。良好的队列可能会在暂时增加延迟请求的同时减少最终错误。请同时跟踪尝试级和操作级结果。

迁移 pull request 检查清单

将此检查清单复制到实现 PR 中。

• Base URL 和密钥来自环境变量。
• 金丝雀使用精确且已验证的模型标识符。
• 只有一层负责重试。
• PR 中记录了 SDK 重试默认值。
• 429、超时和 5xx 行为都具有受限尝试次数。
• 退避包含抖动，并在存在 Retry-After 时遵循它。
• RPM 和 TPM 上限根据观察到的流量进行估算。
• 流式传输有单独的失败测试。
• 每次尝试共享同一个逻辑 operation_id。
• 会比较使用日志和应用日志。
• 上线和回滚阈值在启动前已写明。
• 无需再次更改代码即可恢复之前的端点。

常见迁移错误

保留 SDK 重试和应用重试，却没有计算总量

这是最重要的审查发现。默认值仍然会影响行为，即使它们在本地函数中不可见。

只测试一个很小的提示词

一条请求就能证明凭据和响应兼容性。它几乎不能说明 TPM 压力、输出限制、长流式传输或 p95 延迟。

重试身份验证和验证错误

退避无法修复无效密钥、不受支持的参数或拼写错误的模型。对未更改的负载反复重试只会浪费容量，并掩盖真正的缺陷。

把超时当作请求没有运行的证明

上游在接受调用后，客户端可能会停止等待。设计重试和计费时要把这种不确定性考虑在内。

在一个版本中同时更改端点、模型、提示词和重试策略

这会让故障很难归因。先迁移一种已知的请求形态，然后在路由可观测之后再扩展模型选择。

“OpenAI 兼容”的更安全定义

在迁移规划中，“OpenAI 兼容”应表示交互模式足够熟悉，从而减少代码变更。它不应被理解为一种承诺，即每个提供商都具有完全相同的配额、token 计数、错误语义、延迟、流式行为或参数支持。

这种区别让 OpenAI 客户端迁移 更容易审查。保留有助于稳定的接口，但在提供商和路由可能不同的地方测试操作契约。

Flatkey 在一个 OpenAI 兼容的基础 URL 之后集中管理访问和计费，这可以简化客户端 diff 和后续的模型扩展。不过，在生产流量迁移之前，工程工作仍然是把重试、吞吐量和可观测性明确化。

这就是生产级 OpenAI 客户端迁移 应该达到的标准：以明确的运行证据支撑一个小的接口变更。

在为你的金丝雀选择模型时，请查看 Flatkey 价格页面，然后只有在代码审查中的检查清单通过且路由行为已在日志中可见之后，才批准迁移。

常见问题

迁移期间我应该禁用 OpenAI SDK 重试吗？

如果你的应用或队列已经负责重试，就禁用它们。如果没有其他层负责重试，则受限的 SDK 重试可以是合理的。关键规则是避免多个独立的重试所有者。

迁移期间 RPM 和 TPM 有什么区别？

RPM 限制请求频率，而 TPM 限制 token 吞吐量。小而高频的调用可能先触发 RPM；较少但较大的提示词或输出可能先触发 TPM。两种工作负载形态都要测试。

429 应该总是重试吗？

只有在有上限的重试和延迟预算内才可以。如果可用，应遵守 Retry-After；否则使用带抖动的指数退避。如果操作已无法满足产品的延迟目标，就停止重试。

我可以安全地重试一个超时的生成吗？

不能确定。即使客户端超时，上游请求也可能已经执行。把这次重试视为可能的重复请求，并记录各次尝试之间的关系。

最小安全金丝雀是什么？

使用一个固定模型、一个请求形状、明确的重试责任、并发上限，以及针对 429、超时、流中断和回滚的测试。在扩大流量之前，将客户端尝试次数与网关使用日志进行对比。

Sources

• Flatkey quickstart
• OpenAI Python SDK：重试、超时和基础 URL 配置
• OpenAI Node SDK：重试和超时
• OpenAI Cookbook：如何处理速率限制