多个 AI API 直接集成，背后藏着哪些复杂度

第一次集成 LLM 看起来简单得出奇：向 /v1/chat/completions 端点发一个 HTTP POST，不到一小时就能用上。麻烦从应用成熟时开始。为了扛住厂商故障、控制推理成本，或者让每款模型用在它最擅长的地方，团队迟早会接入多家厂商，原来的一次 API 调用就变成了一团乱麻：互不兼容的格式、各不相同的鉴权方式、退避逻辑、脆弱的流式解析器，还有分散的账单。

下面是实践中会出现的四种故障，以及能消除它们的架构模式。

1. 格式漂移与请求体归一

各家厂商的请求和响应格式并不一致。很多厂商宣称「兼容 OpenAI」，但请求体的差异是真实存在的：

• OpenAI 风格的 API 把系统指令放在 messages 数组里，写成 "role": "system"；temperature 范围是 0.0–2.0。
• Anthropic 原生 API 把系统提示词单独放到顶层的 system 参数，只接受 user / assistant 两种角色，temperature 上限是 1.0，并且必须传 max_tokens。
• Google 原生 Gemini API 用 contents 代替 messages，用 "role": "model" 代替 "role": "assistant"，文字还要嵌套在 parts 数组里。

不加一层序列化，直接把一家的请求体发给另一家，对方会返回 400 拒绝。

// OpenAI 风格的请求
{
  "model": "gpt-5.5",
  "messages": [
    {"role": "system", "content": "You are a precise JSON extractor."},
    {"role": "user", "content": "Extract data from this text..."}
  ]
}

// Anthropic 原生请求：system 提示词放在顶层，必须传 max_tokens
{
  "model": "claude-sonnet-5",
  "system": "You are a precise JSON extractor.",
  "messages": [
    {"role": "user", "content": "Extract data from this text..."}
  ],
  "max_tokens": 1024
}

工具调用的问题更深。各家对 JSON Schema 的严格程度不同：有的会因为缺一个可选字段就拒绝整个定义，有的会悄悄丢掉自己不支持的约束，于是格式错误的参数流进你的后端。没有归一层，你就得为技术栈里的每一家分别维护序列化器、解析器和 schema 校验器。

2. 限流、重试与短暂故障

传统微服务的故障是可预期的：500 是服务端错误，503 是稍后再试。LLM API 按两个相互独立的维度限流：每分钟请求数和每分钟 token 数。所以流量高峰时，即使月度预算还很充裕，端点也可能返回 429。

固定间隔的简单重试会让 429 更糟：每次重试都算进同一个时间窗，违规不断累积，最后被锁更久。标准做法是带完全随机抖动的指数退避：

import time
import random

def calculate_backoff(attempt: int, base_delay: float = 1.0, max_delay: float = 32.0) -> float:
    """Exponential backoff with full randomized jitter."""
    calculated_delay = min(max_delay, base_delay * (2 ** attempt))
    # 完全随机抖动，避免大量重试在同一时刻一起涌上
    return random.uniform(0, calculated_delay)

只靠退避扛不住长时间的故障，那需要切换到另一家厂商，而模型之间的切换并不透明：

• 上下文窗口不匹配。从 200K token 的模型切到 32K token 的模型，如果事先不检查提示词长度，就是把 429 换成了截断错误。
• 提示词敏感。为一个模型系列调好的提示词，不经调整放到另一个系列上，可能输出不成结构的内容。
• 流式中途断开。SSE 流会在回复中途断掉；和普通的 HTTP 错误不同，客户端手里留着半截回复，必须重新拼接或接着生成。

生产环境会把健康检查和熔断器配合使用：连续失败后熔断器打开，流量转到备用线路，冷却一段时间后再重新测试主线路。

3. Key 泛滥与账单分散

除了代码层面，直接集成多家厂商还是一笔运营成本。不同的 SDK 意味着每个环境里都有不同的密钥；轮换 Key 或者限定团队权限，要在每家厂商的控制台里各做一遍。每家厂商按自己的周期和预付方式计费，所以财务每个月要对四五个后台，而工程团队看不到每个功能花了多少 token。如果没有集中的预算护栏，一个没处理好的重试循环打在前沿价位的端点上，几分钟就能烧掉一大笔钱。

4. 消除这些问题的架构

可选的设计有三种：

• 直接集成各家 SDK：起步快，但厂商相关的判断会渗进业务逻辑，切换逻辑写死在代码里，每换一次模型都要改应用代码。
• 自建抽象层：把应用解耦出来，但你从此要维护一套归一和切换代码，并且一直跟着每家厂商的 API 变化。
• 统一网关：在所有厂商前面放一个 OpenAI 兼容端点，归一、切换和账单合并都在你的代码之下完成。

对大多数团队来说，网关是能随规模扩展的那种模式。把 HTTP 客户端指向 BoostRail，就能用一个 OpenAI 兼容 API 调用 15 家厂商的 73 款文本、图片和视频模型。请求格式已经归一；某款模型的一条线路出错时，请求会在任何数据返回给你之前，换到同一模型的另一条线路，不会换成别的模型，所以上面说的上下文窗口和提示词不匹配不会发生。数据开始流式返回之后，请求就停留在当前线路上。通过 BYOK，推理可以记在你自己的厂商合同上，同一模型的平台线路作为兜底。

import openai

# 换模型只改一个字符串。
client = openai.OpenAI(
    base_url="https://api.boostrail.com/v1",
    api_key="YOUR_BOOSTRAIL_API_KEY"
)

response = client.chat.completions.create(
    model="claude-sonnet-5",  # 或者 gpt-5.5、gemini-3.1-pro、deepseek-v4-pro……
    messages=[
        {"role": "user", "content": "Explain circuit breaker patterns in microservices."}
    ]
)

print(response.choices[0].message.content)

网关会多一跳网络。对多数负载来说，这和动辄几百到几千毫秒的生成时间相比很小；正式切换之前，用你自己的流量测一下。

可靠性检查清单

在跨多家厂商扩展生产环境的 AI 应用之前，确认这几点：

• 格式抽象：各厂商的请求格式隔离在一个 OpenAI 兼容层后面。
• 自适应退避：429 用带完全随机抖动的指数退避，不用固定间隔重试。
• 自动切换：健康检查和熔断器要考虑上下文窗口和提示词的差异。
• 流式恢复：SSE 解析器在连接突然断开时不抛出未处理的异常。
• 统一的监控与账单：token 花费、预算和 API 健康状况在一个地方就能看到。

完整的模型列表见 BoostRail 模型目录；在 app.boostrail.com 一分钟就能拿到 API Key。