Tutoriale

Creare un bilanciatore di carico per API AI multi-provider su hiapi

Oct 8, 2026

Cosa costruirai

Aggiungi un livello di routing leggero davanti all'endpoint chat completions di hiapi: seleziona il modello per richiesta, riprova in caso di guasti transitori, esegui il fallback su altri provider quando un provider è inattivo o soggetto a rate limit, e monitora i costi instradando le richieste economiche verso modelli economici. È lo stesso pattern che i sistemi di produzione adottano per evitare che un singolo provider upstream diventi un single point of failure—solo che qui "provider" significa semplicemente "model ID", perché hiapi mette l'accesso compatibile OpenAI a decine di provider dietro lo stesso endpoint e la stessa API key.

Questa guida riguarda solo i modelli di testo/chat. I modelli di immagini, video e audio di hiapi girano su una API separata con coda di attività asincrona (/v1/tasks), che ha il proprio meccanismo di polling/callback—se devi instradare tra i due, vedi la nota alla fine.

Prerequisiti

Una API key di hiapi. Ottienila dalla dashboard—ogni esempio qui sotto richiede solo Authorization: Bearer <key>.
Python 3.9+ con requests installato (pip install requests), oppure curl se preferisci prima testare nella shell.
Nessun SDK necessario—chat completions è un normale endpoint REST compatibile OpenAI, quindi qualsiasi strumento che supporti HTTP può essere usato.

Esempio minimo funzionante

Ogni modello di testo di hiapi segue la stessa struttura /v1/chat/completions. Ecco una singola chiamata con 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
  }'

E la versione Python equivalente, su cui si basa il resto della guida:

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

Funziona, ma non c'è failover: se deepseek-v4.1-flash è soggetto a rate limit o temporaneamente inattivo, la richiesta fallisce direttamente. Il resto della guida trasformerà questa singola chiamata in un router.

Scopri l'ambito su cui puoi instradare

Non codificare a memoria una lista di modelli—interroga GET /v1/models e filtra per supported_endpoint_types. I modelli compatibili con chat completions riportano "openai"; i modelli di immagini/video/audio di hiapi riportano "task" e non possono essere usati con il codice di questa guida:

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

Costruisci la gerarchia di routing da questa lista all'avvio (o memorizzala in cache per qualche ora), invece di congelare i model ID nel codice—i provider aggiungono, dismettono o rinominano modelli, e una gerarchia costruita su una lista statica obsoleta si degraderà silenziosamente instradando verso modelli che non esistono più.

Pattern di produzione 1: retry con backoff

Non ogni fallimento merita un retry. Un 400 con model_not_found fallirà allo stesso modo al tentativo successivo—riprovare spreca solo latenza. Un 429 o 5xx è di solito transitorio e merita un breve 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 e 403 passano direttamente attraverso raise_for_status() senza retry—un model ID errato o una chiave errata non si risolvono da soli aspettando.

Pattern di produzione 2: timeout lato client

Il requests.post(..., timeout=30) sopra non è decorativo—senza di esso, una connessione upstream bloccata bloccherà il tuo router a tempo indefinito, e nessuna logica di retry servirà a nulla perché non riprenderai mai il controllo per riprovare. Imposta un timeout abbastanza stretto da far fallire rapidamente le richieste bloccate verso il percorso di retry/failover, invece di mantenere occupato un thread. Per una singola chat completion, 20–30 secondi è ragionevole; se stai instradando richieste sensibili alla latenza e preferisci fare failover verso un modello più veloce piuttosto che attendere un modello lento, puoi abbassarlo ulteriormente.

Pattern di produzione 3: routing a livelli consapevole dei costi

Non ogni richiesta ha bisogno del tuo modello più costoso. Un pattern pratico: definisci i livelli, prova prima il livello più economico che probabilmente avrà successo, e fai escalation solo in caso di fallimento:

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

Questo prova ogni modello del livello budget prima di spendere quello premium, e sale di un livello solo dopo che ogni modello del livello corrente ha fallito—così un singolo provider inattivo non spinge il traffico verso il modello più costoso, ma un intero livello inattivo sì. Prima di decidere a quale livello appartiene un modello, controlla il prezzo attuale per modello nella pagina dei prezzi—hiapi non espone i prezzi nella risposta di /v1/models, quindi prendili da lì invece di codificare un numero che potrebbe essere obsoleto.

Pattern di produzione 4: idempotenza e sicurezza delle richieste duplicate

/v1/chat/completions è un normale POST senza chiave di idempotenza lato server—a differenza dell'API delle attività /v1/tasks di hiapi, non c'è modo di fornire al server una chiave generata dal client e fargli deduplicare le richieste di retry. Questo significa che se una richiesta va in timeout dopo che il modello ha completato la generazione ma prima che la risposta ti raggiunga, un retry ingenuo pagherà due volte la stessa generazione.

Due mitigazioni pratiche:

Mantieni il timeout lato client abbastanza stretto che "timeout" significhi quasi sempre "non ha mai raggiunto il modello", piuttosto che "la risposta è andata persa in transito".
Se il risultato della chiamata chat guida scritture downstream (ad esempio, usi l'output del modello per popolare righe di database), rendi quella scrittura idempotente—esegui un upsert su una chiave stabile invece di un insert—così le chiamate API duplicate sono sprecate ma non pericolose.

Nota su polling e callback (modelli di immagini/video)

Tutto quanto sopra vale per i modelli di testo su /v1/chat/completions, che è sincrono—ottieni la risposta completa in un singolo round-trip HTTP. I modelli di immagini, video e audio di hiapi funzionano diversamente: invii un task a /v1/tasks, poi o ne interroghi lo stato terminale o registri un URL di callback. Se il tuo router deve coprire sia modelli di testo che media, trattali come due percorsi di codice separati—gli ID dei task media non sono qualcosa che puoi riprovare con backoff come una chat completion, e applicare direttamente la logica di retry di questa guida a /v1/tasks creerà solo task di generazione duplicati.

Riferimento sulla gestione degli errori

L'endpoint chat di hiapi restituisce una busta di errore compatibile OpenAI. I due che incontrerai più spesso costruendo un router:

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

Instrada in base a error.code e non a error.type—ogni errore condivide lo stesso "type": "hiapi_error", quindi code è la vera base per distinguere "arrenditi" (model_not_found, permission_denied) da "riprova" (rate limit ed errori del server, indicati dallo stato HTTP e non da un codice diverso).

Mettere tutto insieme

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

Per una panoramica approfondita dei singoli modelli che potresti inserire in un livello, consulta le guide API di DeepSeek V4.1 Flash, Claude Sonnet 4.6 e GLM-5.3.

Domande frequenti

hiapi ha un bilanciatore di carico integrato o un failover automatico? No—hiapi ti fornisce un endpoint e una chiave per più modelli, ma la logica di routing (quale modello usare, in quale ordine, con quale strategia di retry) è una responsabilità del client, ed è esattamente ciò che implementa questa guida.

Qual è la differenza tra il "model router" di hiapi e chiamare direttamente un modello? Chiamare direttamente un modello significa single point of failure: se quel modello è soggetto a rate limit o temporaneamente inattivo, la tua richiesta fallisce. Il router prova modelli alternativi prima di arrendersi—preferibilmente su provider upstream diversi.

Posso instradare allo stesso modo tra modelli di generazione di immagini o video? Non con questo codice. Girano sull'API asincrona /v1/tasks, non su /v1/chat/completions, con semantiche diverse (invii, poi interroghi o ricevi un callback). Vedi la nota sopra.

Come faccio a sapere quali modelli sono attualmente disponibili per il routing? Chiama GET /v1/models e filtra per "openai" in supported_endpoint_types—non codificare una lista di modelli, perché la disponibilità cambia.

Cosa dovrebbe attivare l'escalation a un livello di modello più costoso? Nel pattern sopra, l'escalation avviene solo quando ogni modello del livello corrente ha fallito. Puoi estendere questo con i tuoi segnali—ad esempio, facendo escalation quando l'output del livello budget non supera il controllo qualità del tuo caso d'uso—ma quella logica è al di fuori dell'API hiapi stessa.

Ci sono rate limit attorno a cui dovrei progettare? Sì, concettualmente: la tua logica di retry dovrebbe trattare l'HTTP 429 come ritentabile con backoff (come mostrato sopra), non come fallimento terminale. Controlla i limiti specifici che hai pianificato nella dashboard.