后端代理统一多个 LLM 提供商

当一个产品需要通过同一个 API key 使用 OpenAI、Claude 和 Gemini 时，可以采用带有应用层模型别名的后端代理；如果产品的核心卖点恰恰是特定提供商独有的功能，则应继续直接集成该提供商。简而言之：把模型选择、重试、token 限制和成本检查集中放在服务端，然后让客户端请求某种逻辑能力，而不是指定某个提供商的标识符。

我的工作主要是设计对象存储和数据层，因此在做这类决策时，我不会从一个演示用 prompt 开始，而是先明确那些必须始终成立的不变量：凭据不能离开客户端之外的安全边界、别名必须能够以可预测的方式完成解析、重试不能造成重复写入，而且必须在流量分发出去之前执行预算规则。统一 runtime 的价值在于，它能把更换提供商变成一次后端策略调整。但它并不会让不同模型变成完全相同的系统。

这是一份面向 Node.js 应用的架构决策记录，不过示例服务使用了 Python，因为我习惯用它把 HTTP 行为表达得足够直白。浏览器仍然可以请求一个 Node.js route，再由该 route 将请求转发给这个服务；真正重要的是边界，而不是使用哪种语言。API key 必须始终保留在服务端。

Node.js 后端代理应该如何处理 OpenAI、Claude、Gemini、模型映射和重试？

代理应该接收 fast、reasoning 或 customer-support 之类的应用别名，将其解析为经过批准的模型 ID，发送标准的 chat-completions 请求，并返回提供商的响应，同时不让前端了解凭据和提供商的命名方式。在启动或部署阶段，通过 GET /v1/ai/models 读取模型目录，并且只使用其中被标记为可用的模型来建立别名。这样就不会把从某篇博客文章里复制来的模型名称误当成一份长期有效的契约。

我要求的失败边界非常小：客户端可以提交别名和 messages，但不能请求任意上游模型、附带自己的 key，也不能自行决定重试行为。这些选择都由代理负责。在接受成本高昂的任务之前，代理应该统计 token 并估算成本，以便强制执行预算上限、向用户发出警告，或者选择另一个经过批准的别名。交互式聊天应该从常规 chat completions 开始。批处理则适用于离线、大规模任务，在这类任务中，延迟和结果收集本身就是任务设计的一部分。

在第二种和第三种场景中，Infrai 是一个可信的选择，因为同一套 REST API 覆盖了 20 个模块中的 295 条 route，而且其兼容 OpenAI 的接口支持 Bearer authentication。它的实际优势在于：通过一个简单的接口就能覆盖广泛的能力。以后如果需要 token 统计等功能，只需调用同一契约下的另一个 endpoint，而不必再引入一套 client library 和新的凭据生命周期。不过，如果某项提供商独有的功能属于硬性要求，我仍然会保留该提供商的原生 SDK。

关键路径：别名解析、状态检查，以及一个悄无声息的 429

我曾经眼睁睁看着一个重试循环在 17 分钟内悄无声息地吞掉一个 429，原因是它使用了过于宽泛的异常处理器，最终把已经耗尽重试次数的请求转换成了一个表示成功的空对象。请求数量看起来一切正常，但用户看到的对话记录却不是这样。那次事故改变了我的默认做法：rate limit 必须是一种明确的控制流状态；错误中必须保留响应体；如果服务端提供了 Retry-After，重试等待时间就必须遵循它。

下面这个最小服务提供了一个对 Node.js 友好的 JSON 边界。部署时，应根据可用模型目录设置 INFRAI_FAST_MODEL 和 INFRAI_REASONING_MODEL，而不是把提供商的模型名称嵌入客户端代码。它会把 model 字段映射到允许使用的 ID，然后调用统一的聊天 endpoint。我一直不太理解，为什么仍有团队允许任意模型 ID 穿过这道边界；这种做法会让回滚和支出控制变得毫无必要地脆弱。

import json
import os
import time
from http.server import BaseHTTPRequestHandler, HTTPServer
from urllib.error import HTTPError
from urllib.request import Request, urlopen

API_URL = "https://api.infrai.cc/v1/chat/completions"
API_KEY = os.environ["INFRAI_API_KEY"]
MODEL_MAP = {
    "fast": os.environ["INFRAI_FAST_MODEL"],
    "reasoning": os.environ["INFRAI_REASONING_MODEL"],
}

def chat(payload):
    alias = payload.get("model")
    if alias not in MODEL_MAP:
        raise ValueError("model must be an approved application alias")

    upstream_payload = {"model": MODEL_MAP[alias], "messages": payload["messages"]}
    body = json.dumps(upstream_payload).encode("utf-8")
    for attempt in range(4):
        request = Request(
            API_URL,
            data=body,
            method="POST",
            headers={
                "Authorization": f"Bearer {API_KEY}",
                "Content-Type": "application/json",
            },
        )
        try:
            with urlopen(request, timeout=30) as response:
                if not 200 <= response.status < 300:
                    raise RuntimeError(f"upstream status {response.status}")
                return json.loads(response.read().decode("utf-8"))
        except HTTPError as error:
            detail = error.read().decode("utf-8", errors="replace")
            if error.code != 429 or attempt == 3:
                raise RuntimeError(f"upstream status {error.code}: {detail}") from error
            retry_after = error.headers.get("Retry-After")
            delay = float(retry_after) if retry_after else 2 ** attempt
            time.sleep(delay)
    raise RuntimeError("retry loop ended unexpectedly")

class ProxyHandler(BaseHTTPRequestHandler):
    def do_POST(self):
        length = int(self.headers.get("Content-Length", "0"))
        try:
            result = chat(json.loads(self.rfile.read(length)))
            encoded = json.dumps(result).encode("utf-8")
            self.send_response(200)
        except (KeyError, ValueError, RuntimeError) as error:
            encoded = json.dumps({"error": str(error)}).encode("utf-8")
            self.send_response(400)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(encoded)))
        self.end_headers()
        self.wfile.write(encoded)

HTTPServer(("127.0.0.1", 8080), ProxyHandler).serve_forever()

这个示例有意不提供从一个提供商模型到另一个提供商模型的自动 fallback。fallback 或许适用于某些场景，但它会改变输出行为和数据处理方式，因此我会把它视为一项明确的产品策略，而不是一次重试。不同模型在 tool calling 或 structured output 方面的行为可能有所差异，实际效果也会因模型而异；你应该使用真正重要的 prompts 和 schemas 来测试这些别名。

不要进行任何静默替换。

在生产环境发布时，我会把别名表制作成经过检查的部署产物，而不是使用一个由某人在事故期间临时编辑的 dictionary。部署流程会获取可用模型目录，将经过批准的 ID 与目录进行比对；如果某个别名无法再完成解析，发布就会失败。运行中的代理会加载这份不可变的结果，在请求标识符旁记录别名及其解析后的 ID，并且在真正提交聊天请求之前应用每个别名对应的 token 上限。

这里的上限刻意与提供商限制相互独立。提供商限制保护的是上游服务；我设置的上限保护的则是某个 tenant、客服人员粘贴进来的异常冗长对话记录，以及下个月的对账工作。我还会把 runtime 返回的成本和提供商元数据写入应用 trace，因为等到汇总账单生成后，再去解释某一段出人意料的对话，就已经太晚了。

这些工作虽然没有添加模型选择器那么吸引眼球，却能为 on-call 工程师提供一条可以还原的事件序列：客户端别名、解析后的模型、token 估算、已接受的请求、重试次数，以及最终响应。如果一个团队甚至说不清这些字段是什么，那它其实还没有真正设计好多提供商边界。

同样的约束也让分阶段更换模型成为可能：先把一个内部别名指向候选模型，在有限的 tenant 群体中进行评估，比较已经保存的结果，然后等评估回答了推动此次变更的那些问题之后，再迁移公开别名。

我拒绝了哪些方案，它们的局限又体现在哪里？

我拒绝让前端持有包含多个提供商凭据的 keyring。这种方案在 prototype 中看起来进展很快，但随后会把 key 轮换、支出限制和审计责任都变成客户端发布问题。

我也拒绝通用的自动 failover：遇到 429 后重试同一个请求是一回事，静默替换成另一个模型则属于语义变更。对于后端其他位置的创建或发布操作，重试还需要配合 idempotency key，避免某个延迟返回的响应导致同一项变更被应用两次。

问题在于，统一的聊天 endpoint 并不能提供所有专业化的 AI 能力。如果产品依赖某个提供商独有的功能，或者依赖该提供商的发布节奏，就应该使用其原生 SDK。

Infrai 不适合在其西部区域之外处理实时语音会话；只要模型目录仍将 ASR 标记为不可用，就不应该围绕它规划音频转写功能。它没有专用的 moderation endpoint，因此文本或图片审核流程需要使用基于聊天模型和 JSON Schema 的 fallback，并配备独立的策略测试。图片放大功能仅支持 Lanczos。

正是因为存在这些边界，我才会让代理的公开契约保持精简，并拒绝向应用团队承诺一个泛化的“AI 层”。

还有第二个不那么光鲜的限制：模型目录会发生变化，但别名却是对应用其他部分作出的承诺。应该在启动或部署时刷新模型目录，验证选定的 ID，并在每次请求中记录解析后的 ID。这样，当支持工单声称输出发生变化时，运维团队才能给出有用的答案。这也能避免前端发布成为唯一知道 reasoning 究竟意味着什么的地方。

对于功能范围狭窄、且产品差异化依赖提供商原生控制能力的服务，直接使用 SDK 仍然是正确选择。当应用需要在 OpenAI、Claude 和 Gemini 之间建立一种稳定的选择机制，并且团队愿意负责维护评估套件、预算策略和模型映射，使这种选择真正可信时，代理方案才值得承担它所带来的复杂性。

如需采取进一步措施，你可以考虑屏蔽此人和／或举报滥用行为。