在 hiapi 上为多提供商 AI API 构建负载均衡器

你将构建什么

在 hiapi 的 chat completions 端点前加一个轻量路由层：按请求选择模型、对瞬时故障进行重试、在某个提供商宕机或被限流时回退到其他提供商，并通过把便宜的请求路由到便宜的模型来追踪成本。这与生产系统为避免单一上游提供商成为单点故障所采用的模式相同——只不过这里的“提供商”只意味着“模型 ID”，因为 hiapi 把对数十家提供商的OpenAI 兼容访问放在同一个端点和同一个 API 密钥之后。

本指南只涉及文本/对话模型。hiapi 的图像、视频和音频模型运行在单独的异步任务队列 API（/v1/tasks）上，该 API 有自己的轮询/回调机制——如果你要跨两者路由，请参见末尾的说明。

先决条件

• 一个 hiapi API 密钥。从仪表盘获取——下面的每个示例只需要 Authorization: Bearer <key>。
• Python 3.9+ 并安装 requests（pip install requests），或者如果你更想先在 shell 中测试，可以使用 curl。
• 无需 SDK——chat completions 是普通的 OpenAI 兼容 REST 端点，因此任何支持 HTTP 的工具都可以使用。

最小可运行示例

每个 hiapi 文本模型都遵循相同的 /v1/chat/completions 结构。下面是一个使用 curl 的单次调用：

curl https://api.hiapi.ai/v1/chat/completions \
  -H "Authorization: Bearer $HIAPI_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "deepseek-v4.1-flash",
    "messages": [{"role": "user", "content": "Summarize load balancing in one sentence."}],
    "max_tokens": 100
  }'

以及等效的 Python 版本，本指南其余部分都基于它：

import os
import requests

HIAPI_BASE = "https://api.hiapi.ai/v1"
API_KEY = os.environ["HIAPI_API_KEY"]

def chat(model: str, messages: list[dict], **kwargs) -> dict:
    resp = requests.post(
        f"{HIAPI_BASE}/chat/completions",
        headers={
            "Authorization": f"Bearer {API_KEY}",
            "Content-Type": "application/json",
        },
        json={"model": model, "messages": messages, **kwargs},
        timeout=30,
    )
    resp.raise_for_status()
    return resp.json()

result = chat("deepseek-v4.1-flash", [{"role": "user", "content": "Say hi in five words."}])
print(result["choices"][0]["message"]["content"])

这样虽然能工作，但没有故障转移：如果 deepseek-v4.1-flash 被限流或临时宕机，请求就会直接失败。本指南的其余部分将把这个单独的调用变成一个路由器。

发现你可以路由的范围

不要凭记忆硬编码模型列表——查询 GET /v1/models 并按 supported_endpoint_types 过滤。兼容 chat completions 的模型会报告 "openai"；hiapi 的图像/视频/音频模型则报告 "task"，无法与本指南中的代码一起使用：

def list_chat_models() -> list[str]:
    resp = requests.get(f"{HIAPI_BASE}/models",
                         headers={"Authorization": f"Bearer {API_KEY}"}, timeout=15)
    resp.raise_for_status()
    models = resp.json().get("data", [])
    return [m["id"] for m in models if "openai" in (m.get("supported_endpoint_types") or [])]

在启动时从这个列表构建路由层级（或将其缓存几个小时），而不是在代码中冻结模型 ID——提供商会新增、下线或重命名，而基于过时硬编码列表构建的层级会悄悄退化成路由到已不存在的模型。

生产模式 1：带退避的重试

并非每个失败都值得重试。带有 model_not_found 的 400 在下一次尝试中会同样失败——重试只会浪费延迟。429 或 5xx 通常是瞬时的，值得短暂退避：

import time

RETRYABLE_STATUS = {429, 500, 502, 503, 504}

def chat_with_retry(model: str, messages: list[dict], max_attempts: int = 3, **kwargs) -> dict:
    last_exc = None
    for attempt in range(max_attempts):
        try:
            resp = requests.post(
                f"{HIAPI_BASE}/chat/completions",
                headers={"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"},
                json={"model": model, "messages": messages, **kwargs},
                timeout=30,
            )
            if resp.status_code in RETRYABLE_STATUS:
                raise requests.HTTPError(f"retryable {resp.status_code}: {resp.text}", response=resp)
            resp.raise_for_status()
            return resp.json()
        except requests.HTTPError as e:
            last_exc = e
            if attempt < max_attempts - 1:
                time.sleep(2 ** attempt)  # 1s, 2s, 4s
    raise last_exc

400、401 和 403 会直接穿过 raise_for_status() 而不重试——错误的模型 ID 或错误的密钥不会通过等待自行修复。

生产模式 2：客户端超时

上面的 requests.post(..., timeout=30) 不是装饰性的——没有它，挂起的上游连接会无限期阻塞你的路由器，再多的重试逻辑也无济于事，因为你永远不会重新获得控制权去重试。设置足够紧的超时，让卡住的请求快速失败进入重试/故障转移路径，而不是一直占着一个线程。对于单次 chat completion，20–30 秒是合理的；如果你路由的是延迟敏感的请求，并且更愿意故障转移到一个更快的模型，而不是等待一个慢模型，可以进一步调低。

生产模式 3：感知成本的分层路由

并非每个请求都需要你最昂贵的模型。一个实用的模式：定义层级，先尝试最便宜且可能成功的层级，只在失败时才升级：

TIERS = [
    ["deepseek-v4.1-flash", "deepseek-v4-flash"],  # budget tier
    ["glm-5.3", "kimi-k3"],                         # mid tier
    ["grok-4.6", "claude-opus-4-8"],                # premium tier, used last
]

def route(messages: list[dict], **kwargs) -> dict:
    last_exc = None
    for tier in TIERS:
        for model in tier:
            try:
                return chat_with_retry(model, messages, **kwargs)
            except requests.HTTPError as e:
                last_exc = e
                continue  # try next model in this tier, then next tier
    raise last_exc

这会在花费高级层级之前先尝试预算层级中的每个模型，并且只有在当前层级中的每个模型都失败后才向上移动一层——因此单个提供商宕机不会把流量推到最昂贵的模型，但整个层级宕机则会。在决定某个模型属于哪个层级之前，请在定价页面查看当前的每模型定价——hiapi 不会在 /v1/models 响应中暴露价格，因此请从那里获取，而不是硬编码一个可能过时的数字。

生产模式 4：幂等性与重复请求安全

/v1/chat/completions 是普通的 POST，没有服务器端幂等键——不同于 hiapi 的 /v1/tasks 任务 API，没有办法向服务器提供客户端生成的键并让它为你去重重试请求。这意味着如果一个请求在模型完成生成之后但在响应到达你之前超时，朴素的重试将为同一次生成支付两次费用。

两种实用缓解措施：

• 保持客户端超时足够紧，使“超时”几乎总是意味着“从未到达模型”，而不是“响应在传输中丢失”。
• 如果 chat 调用的结果驱动下游写入（例如，你使用模型输出填充数据库行），请让那个写入幂等——在稳定键上执行 upsert 而不是 insert——这样重复的 API 调用虽然浪费但不会不安全。

关于轮询与回调的说明（图像/视频模型）

以上所有内容适用于 /v1/chat/completions 上的文本模型，它是同步的——你在一次 HTTP 往返中就能获得完整响应。hiapi 的图像、视频和音频模型工作方式不同：你向 /v1/tasks 提交任务，然后要么轮询其终态，要么注册回调 URL。如果你的路由器需要同时覆盖文本和媒体模型，请将它们视为两条独立的代码路径——媒体任务 ID 不是你可以像重试 chat completion 那样带退避重试的东西，而将本指南的重试逻辑直接应用到 /v1/tasks 只会创建重复的生成任务。

错误处理参考

hiapi 的 chat 端点返回 OpenAI 兼容的错误信封。构建路由器时你最常遇到的两个：

// 400 — bad model id, don't retry
{
  "error": {
    "code": "model_not_found",
    "message": "The requested model does not exist or is not available. Check the exact model ID with GET /v1/models and try again.",
    "type": "hiapi_error"
  }
}

// 401 — bad or revoked key, don't retry
{
  "error": {
    "code": "permission_denied",
    "message": "This API key is invalid. Check that it is correct or use another API key and try again.",
    "type": "hiapi_error"
  }
}

基于 error.code 而不是 error.type 进行路由——每个错误都共享相同的 "type": "hiapi_error"，因此 code 才是真正区分“放弃”（model_not_found、permission_denied）与“重试”（限流和服务器错误，通过 HTTP 状态而不是不同的代码呈现）的依据。

整合起来

class Router:
    def __init__(self, api_key: str, tiers: list[list[str]]):
        self.api_key = api_key
        self.tiers = tiers

    def complete(self, messages: list[dict], **kwargs) -> dict:
        last_exc = None
        for tier in self.tiers:
            for model in tier:
                try:
                    return chat_with_retry(model, messages, **kwargs)
                except requests.HTTPError as e:
                    last_exc = e
        raise RuntimeError(f"all models exhausted across {len(self.tiers)} tiers") from last_exc

router = Router(API_KEY, TIERS)
out = router.complete([{"role": "user", "content": "Explain failover in one sentence."}])
print(out["choices"][0]["message"]["content"])

关于你可能插入某个层级的各个模型的深入介绍，请参阅 DeepSeek V4.1 Flash、Claude Sonnet 4.6 和 GLM-5.3 API 指南。

常见问题

hiapi 有内置的负载均衡器或自动故障转移吗？ 没有——hiapi 为你提供跨多个模型的一个端点和一把密钥，但路由逻辑（用哪个模型、按什么顺序、采用什么重试策略）是客户端关注的问题，这正是本指南所实现的。

hiapi 的“模型路由器”与直接调用一个模型有什么区别？ 直接调用一个模型意味着单点故障：如果该模型被限流或临时宕机，你的请求就会失败。路由器在放弃之前会尝试替代模型——最好跨不同的底层提供商。

我可以用同样的方式跨图像或视频生成模型路由吗？ 不能使用这段代码。它们运行在异步的 /v1/tasks API 上，而不是 /v1/chat/completions，语义不同（提交，然后轮询或接收回调）。请参见上面的说明。

我如何知道当前哪些模型可用于路由？ 调用 GET /v1/models 并在 supported_endpoint_types 中过滤 "openai"——不要硬编码模型列表，因为可用性会变化。

什么应该触发升级到更昂贵的模型层级？ 在上述模式中，只有当当前层级中的每个模型都失败时才会升级。你可以用自己的信号扩展这一点——例如，当预算层级的输出未能通过你的用例的质量检查时升级——但该逻辑超出了 hiapi API 本身的范围。

是否有我应该围绕设计的速率限制？ 有，从概念上讲：你的重试逻辑应将 HTTP 429 视为可重试并带退避（如上所示），而不是终端失败。在仪表盘中查看你计划的具体限制。