hiapi 上でマルチプロバイダー AI API 向けのロードバランサーを構築する

構築するもの

hiapi の chat completions エンドポイントの前に軽量なルーティング層を追加します。リクエストごとにモデルを選択し、一時的な障害に対してリトライし、あるプロバイダーがダウンまたはレート制限された場合に他のプロバイダーへフォールバックし、安価なリクエストを安価なモデルにルーティングすることでコストを追跡します。これは、単一の上流プロバイダーが単一障害点になるのを避けるために本番システムが採用しているのと同じパターンです。ただしここでの「プロバイダー」は「モデル ID」を意味するだけです。hiapi は数十社のプロバイダーへのOpenAI 互換アクセスを、同じエンドポイントと同じ API キーの背後に置いているからです。

本ガイドはテキスト/チャットモデルのみを扱います。hiapi の画像、動画、音声モデルは別個の非同期タスクキュー API（/v1/tasks）で動作し、独自のポーリング/コールバック機構を持ちます。両方をまたいでルーティングする場合は、末尾の注意事項を参照してください。

前提条件

• hiapi API キー。ダッシュボードから取得してください。以下の各例では Authorization: Bearer <key> のみが必要です。
• Python 3.9+ と requests がインストール済み（pip install requests）。または、まずシェルでテストしたい場合は 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

これはプレミアム階層にコストをかける前に、予算階層の各モデルを試し、現在の階層のすべてのモデルが失敗した場合にのみ 1 つ上の階層に移動します。そのため、単一のプロバイダーがダウンしてもトラフィックが最も高価なモデルに押し出されることはありませんが、階層全体がダウンすればそうなります。あるモデルをどの階層に置くかを決める前に、料金ページで現在のモデルごとの価格を確認してください。hiapi は /v1/models のレスポンスで価格を公開しないため、古くなっている可能性のある数値をハードコードするのではなく、そこから取得してください。

本番パターン 4：冪等性と重複リクエストの安全性

/v1/chat/completions は通常の POST であり、サーバー側の冪等キーはありません。hiapi の /v1/tasks タスク API とは異なり、クライアント生成のキーをサーバーに渡してリトライリクエストを重複排除させる方法はありません。つまり、モデルが生成を完了した後、レスポンスがあなたに届く前にリクエストがタイムアウトした場合、素朴なリトライは同じ生成に対して二重に課金されます。

実用的な緩和策は 2 つあります：

• クライアントタイムアウトを十分にタイトに保ち、「タイムアウト」がほぼ常に「モデルに到達しなかった」を意味するようにし、「レスポンスが転送中に失われた」を意味しないようにする。
• chat 呼び出しの結果が下流の書き込みを駆動する場合（たとえば、モデル出力を使ってデータベースの行を埋める場合）、その書き込みを冪等にする——安定したキーで insert ではなく upsert を実行する——そうすれば重複した API 呼び出しは無駄ではあっても安全でなくなることはありません。

ポーリングとコールバックに関する注意（画像/動画モデル）

上記のすべては /v1/chat/completions 上のテキストモデルに適用され、同期的です。1 回の HTTP ラウンドトリップで完全なレスポンスを取得します。hiapi の画像、動画、音声モデルは動作が異なります。/v1/tasks にタスクを送信し、その終端状態をポーリングするか、コールバック URL を登録します。ルーターがテキストとメディアモデルの両方をカバーする必要がある場合は、それらを 2 つの独立したコードパスとして扱ってください。メディアのタスク ID は、chat completion をバックオフ付きでリトライするようにはリトライできるものではなく、本ガイドのリトライロジックを /v1/tasks に直接適用すると、重複した生成タスクを作成するだけです。

エラー処理リファレンス

hiapi の chat エンドポイントは OpenAI 互換のエラーエンベロープを返します。ルーターを構築する際に最もよく遭遇する 2 つ：

// 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.type ではなく error.code に基づいてルーティングしてください。すべてのエラーが同じ "type": "hiapi_error" を共有するため、「諦める」（model_not_found、permission_denied）と「リトライする」（レート制限やサーバーエラーは、異なるコードではなく HTTP ステータスで示される）を実際に区別するのは code です。

まとめ

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 は複数のモデルをまたぐ 1 つのエンドポイントと 1 つのキーを提供しますが、ルーティングロジック（どのモデルを、どの順序で、どのリトライ戦略で使うか）はクライアント側の関心事であり、まさに本ガイドで実装しているものです。

hiapi の「モデルルーター」と単一モデルの直接呼び出しの違いは何ですか？ 単一モデルの直接呼び出しは単一障害点を意味します。そのモデルがレート制限されたり一時的にダウンしたりすると、リクエストは失敗します。ルーターは諦める前に代替モデルを試します。理想的には異なる基盤プロバイダーをまたいで。

同じ方法で画像や動画生成モデルをまたいでルーティングできますか？ このコードではできません。それらは /v1/chat/completions ではなく非同期の /v1/tasks API で動作し、セマンティクスが異なります（送信してから、ポーリングするかコールバックを受け取る）。上記の注意事項を参照してください。

現在どのモデルがルーティングに利用可能かを知るにはどうすればよいですか？ GET /v1/models を呼び出し、supported_endpoint_types で "openai" をフィルタリングしてください。利用可能性は変化するため、モデルリストをハードコードしないでください。

何がより高価なモデル階層への昇格をトリガーすべきですか？ 上記のパターンでは、現在の階層のすべてのモデルが失敗した場合にのみ昇格します。独自のシグナルでこれを拡張できます。たとえば、予算階層の出力がユースケースの品質チェックに合格しなかったときに昇格するなど。ただし、そのロジックは hiapi API 自体の範囲外です。

設計時に考慮すべきレート制限はありますか？ 概念的にはあります。リトライロジックは HTTP 429 を終端の失敗としてではなく、バックオフ付きでリトライ可能として扱うべきです（上記のとおり）。ダッシュボードで、計画している具体的な制限を確認してください。