Créer un équilibreur de charge pour les API d'IA multi-fournisseurs sur hiapi

Ce que vous allez construire

Ajoutez une couche de routage légère devant le point de terminaison chat completions de hiapi : sélectionnez le modèle par requête, réessayez les défaillances transitoires, basculez vers d'autres fournisseurs lorsqu'un fournisseur est indisponible ou limité en débit, et suivez les coûts en routant les requêtes bon marché vers les modèles bon marché. C'est le même schéma que celui utilisé par les systèmes de production pour éviter qu'un fournisseur amont unique devienne un point de défaillance unique — sauf qu'ici « fournisseur » signifie simplement « ID de modèle », car hiapi place l'accès compatible OpenAI à des dizaines de fournisseurs derrière le même point de terminaison et la même clé API.

Ce guide ne concerne que les modèles de texte/chat. Les modèles d'image, de vidéo et d'audio de hiapi fonctionnent sur une API de file de tâches asynchrone séparée (/v1/tasks), qui possède son propre mécanisme d'interrogation/rappel — si vous devez router entre les deux, consultez la note à la fin.

Prérequis

• Une clé API hiapi. Obtenez-la depuis le tableau de bord — chaque exemple ci-dessous nécessite uniquement Authorization: Bearer <key>.
• Python 3.9+ avec requests installé (pip install requests), ou curl si vous préférez tester d'abord dans le shell.
• Aucun SDK nécessaire — chat completions est un simple point de terminaison REST compatible OpenAI, donc tout outil prenant en charge HTTP peut être utilisé.

Exemple minimal fonctionnel

Chaque modèle texte hiapi suit la même structure /v1/chat/completions. Voici un appel unique avec 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
  }'

Et la version Python équivalente, sur laquelle repose le reste de ce guide :

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"])

Cela fonctionne, mais sans basculement : si deepseek-v4.1-flash est limité en débit ou temporairement indisponible, la requête échoue directement. Le reste de ce guide transforme cet appel unique en un routeur.

Découvrir l'étendue de ce que vous pouvez router

Ne codez pas en dur une liste de modèles de mémoire — interrogez GET /v1/models et filtrez par supported_endpoint_types. Les modèles compatibles avec chat completions rapportent "openai" ; les modèles d'image/vidéo/audio de hiapi rapportent "task" et ne peuvent pas être utilisés avec le code de ce guide :

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 [])]

Construisez votre hiérarchie de routage à partir de cette liste au démarrage (ou mettez-la en cache pendant quelques heures), plutôt que de figer les IDs de modèle dans le code — les fournisseurs ajoutent, retirent ou renomment des modèles, et une hiérarchie construite sur une liste codée en dur obsolète dégénérera silencieusement en routage vers des modèles qui n'existent plus.

Modèle de production 1 : nouvelles tentatives avec backoff

Toutes les erreurs ne méritent pas une nouvelle tentative. Un 400 avec model_not_found échouera de la même manière à la tentative suivante — réessayer ne fait que gaspiller de la latence. Un 429 ou 5xx est généralement transitoire et mérite un bref backoff :

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 et 403 traversent raise_for_status() sans nouvelle tentative — un ID de modèle erroné ou une clé incorrecte ne se réparera pas en attendant.

Modèle de production 2 : délais d'attente côté client

Le requests.post(..., timeout=30) ci-dessus n'est pas décoratif — sans lui, une connexion amont suspendue bloquerait indéfiniment votre routeur, et aucune logique de nouvelle tentative ne pourrait aider car vous ne reprendriez jamais le contrôle pour réessayer. Définissez un délai suffisamment court pour que les requêtes bloquées échouent rapidement vers le chemin de nouvelle tentative/basculement plutôt que de retenir un thread indéfiniment. Pour un seul chat completion, 20 à 30 secondes est raisonnable ; si vous routez des requêtes sensibles à la latence et préférez basculer vers un modèle plus rapide plutôt que d'attendre un modèle lent, réduisez encore.

Modèle de production 3 : routage hiérarchique sensible aux coûts

Toutes les requêtes n'ont pas besoin de votre modèle le plus cher. Un schéma pratique : définissez des niveaux, essayez d'abord le niveau le moins cher susceptible de réussir, et ne montez qu'en cas d'échec :

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

Cela essaie chaque modèle du niveau économique avant de dépenser pour le niveau premium, et ne passe au niveau supérieur qu'après l'échec de tous les modèles du niveau actuel — ainsi une panne d'un seul fournisseur ne pousse pas le trafic vers les modèles les plus chers, mais une panne de tout un niveau le fait. Avant de décider à quel niveau appartient un modèle, consultez les tarifs actuels par modèle sur la page de tarification — hiapi n'expose pas les prix dans la réponse /v1/models, donc récupérez-les là plutôt que de coder en dur un chiffre potentiellement obsolète.

Modèle de production 4 : idempotence et sécurité des requêtes en double

/v1/chat/completions est un POST simple, sans clé d'idempotence côté serveur — contrairement à l'API de tâches /v1/tasks de hiapi, il n'y a aucun moyen de fournir une clé générée par le client au serveur et de le laisser dédupliquer les nouvelles tentatives pour vous. Cela signifie que si une requête expire après que le modèle a terminé sa génération mais avant que la réponse ne vous parvienne, une nouvelle tentative naïve paiera deux fois pour la même génération.

Deux atténuations pratiques :

• Gardez des délais d'attente côté client suffisamment courts pour qu'un « timeout » signifie presque toujours « jamais atteint le modèle », et non « réponse perdue en transit ».
• Si le résultat de l'appel chat pilote une écriture en aval (par exemple, vous remplissez une ligne de base de données avec la sortie du modèle), rendez cette écriture idempotente — faites un upsert sur une clé stable plutôt qu'un insert — afin qu'un appel API en double soit inutile mais pas dangereux.

Note sur l'interrogation et les rappels (modèles d'image/vidéo)

Tout ce qui précède s'applique aux modèles texte sur /v1/chat/completions, qui est synchrone — vous obtenez la réponse complète en un aller-retour HTTP. Les modèles d'image, de vidéo et d'audio de hiapi fonctionnent différemment : vous soumettez une tâche à /v1/tasks, puis soit vous interrogez son état final, soit vous enregistrez une URL de rappel. Si votre routeur doit couvrir à la fois les modèles texte et multimédias, traitez-les comme deux chemins de code séparés — un ID de tâche multimédia n'est pas quelque chose que vous pouvez réessayer avec backoff comme un chat completion, et appliquer directement la logique de nouvelle tentative de ce guide à /v1/tasks ne ferait que créer des tâches de génération en double.

Référence de gestion des erreurs

Le point de terminaison chat de hiapi renvoie une enveloppe d'erreur compatible OpenAI. Les deux que vous rencontrerez le plus souvent en construisant un routeur :

// 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"
  }
}

Routez en fonction de error.code plutôt que de error.type — chaque erreur partage le même "type": "hiapi_error", donc code est ce qui distingue réellement « abandonner » (model_not_found, permission_denied) de « réessayer » (limitation de débit et erreurs serveur, présentées via le statut HTTP plutôt que par un code distinct).

Mise en œuvre complète

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"])

Pour une présentation approfondie des différents modèles que vous pourriez insérer dans un niveau, consultez les guides API DeepSeek V4.1 Flash, Claude Sonnet 4.6 et GLM-5.3.

FAQ

hiapi dispose-t-il d'un équilibreur de charge intégré ou d'un basculement automatique ? Non — hiapi vous fournit un point de terminaison et une clé pour plusieurs modèles, mais la logique de routage (quel modèle utiliser, dans quel ordre, avec quelle stratégie de nouvelle tentative) est une préoccupation côté client, ce que ce guide implémente.

Quelle est la différence entre le « routeur de modèles » de hiapi et l'appel direct à un seul modèle ? L'appel direct à un seul modèle signifie un point de défaillance unique : si ce modèle est limité en débit ou temporairement indisponible, votre requête échoue. Le routeur essaie des modèles alternatifs avant d'abandonner — idéalement à travers différents fournisseurs sous-jacents.

Puis-je router de la même manière entre les modèles de génération d'image ou de vidéo ? Pas avec ce code. Ils fonctionnent sur l'API asynchrone /v1/tasks, et non sur /v1/chat/completions, avec une sémantique différente (soumettre, puis interroger ou recevoir un rappel). Voir la note ci-dessus.

Comment savoir quels modèles sont actuellement disponibles pour le routage ? Appelez GET /v1/models et filtrez sur "openai" dans supported_endpoint_types — ne codez pas en dur la liste des modèles, car la disponibilité change.

Qu'est-ce qui doit déclencher une montée vers un niveau de modèle plus cher ? Dans le schéma ci-dessus, la montée ne se produit que lorsque tous les modèles du niveau actuel ont échoué. Vous pouvez étendre cela avec vos propres signaux — par exemple, monter lorsque la sortie du niveau économique échoue à votre contrôle qualité pour votre cas d'usage — mais cette logique dépasse le cadre de l'API hiapi elle-même.

Y a-t-il des limites de débit autour desquelles je dois concevoir ? Oui, conceptuellement : votre logique de nouvelle tentative doit traiter un 429 HTTP comme réessayable avec backoff (comme montré ci-dessus), et non comme un échec terminal. Consultez les limites spécifiques que vous prévoyez dans le tableau de bord.