Tutorial

Einen Load Balancer für Multi-Provider-KI-APIs auf hiapi bauen

Oct 8, 2026

Was du baust

Lege eine schlanke Routing-Schicht vor den chat-completions-Endpunkt von hiapi: Wähle das Modell pro Anfrage aus, wiederhole transiente Fehler, falle auf andere Anbieter zurück, wenn ein Anbieter ausfällt oder gedrosselt wird, und verfolge die Kosten, indem du günstige Anfragen an günstige Modelle leitest. Das ist dasselbe Muster, das Produktionssysteme einsetzen, um zu verhindern, dass ein einzelner Upstream-Anbieter zum Single Point of Failure wird – nur dass hier „Anbieter“ einfach „Modell-ID“ bedeutet, da hiapi den OpenAI-kompatiblen Zugang zu Dutzenden von Anbietern hinter einem einzigen Endpunkt und einem einzigen API-Schlüssel bündelt.

Diese Anleitung behandelt nur Text-/Chat-Modelle. Die Bild-, Video- und Audiomodelle von hiapi laufen über eine separate asynchrone Task-Queue-API (/v1/tasks) mit eigenem Polling-/Callback-Mechanismus – siehe den Hinweis am Ende, wenn du über beide hinweg routen möchtest.

Voraussetzungen

Ein hiapi-API-Schlüssel. Hole ihn dir vom Dashboard – jedes Beispiel unten benötigt nur Authorization: Bearer <key>.
Python 3.9+ mit installiertem requests (pip install requests), oder curl, wenn du lieber zuerst in der Shell testen möchtest.
Kein SDK erforderlich – chat completions ist ein einfacher OpenAI-kompatibler REST-Endpunkt, sodass jedes Tool, das HTTP spricht, funktioniert.

Minimales lauffähiges Beispiel

Jedes hiapi-Textmodell folgt derselben /v1/chat/completions-Struktur. Hier ein einzelner Aufruf mit 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
  }'

Und die äquivalente Python-Version, auf der der Rest dieser Anleitung aufbaut:

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

Das funktioniert, hat aber kein Failover: Wenn deepseek-v4.1-flash gedrosselt wird oder vorübergehend ausfällt, schlägt die Anfrage einfach fehl. Der Rest dieser Anleitung verwandelt diesen einzelnen Aufruf in einen Router.

Entdecken, was du routen kannst

Hardcode keine Modellliste aus dem Gedächtnis – frage GET /v1/models ab und filtere nach supported_endpoint_types. Modelle, die chat completions unterstützen, melden "openai"; die Bild-/Video-/Audiomodelle von hiapi melden "task" und können nicht mit dem Code in dieser Anleitung verwendet werden:

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

Baue die Routing-Hierarchie beim Start aus dieser Liste (oder cache sie für ein paar Stunden), anstatt Modell-IDs im Code einzufrieren – Anbieter fügen Modelle hinzu, nehmen sie offline oder benennen sie um, und eine Hierarchie, die auf einer veralteten hartcodierten Liste basiert, degradiert stillschweigend dazu, an nicht mehr existierende Modelle zu routen.

Produktionsmuster 1: Wiederholung mit Backoff

Nicht jeder Fehler ist eine Wiederholung wert. Ein 400 mit model_not_found wird beim nächsten Versuch genauso fehlschlagen – eine Wiederholung verschwendet nur Latenz. 429 oder 5xx sind in der Regel transient und ein kurzes Backoff wert:

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 und 403 laufen direkt durch raise_for_status(), ohne wiederholt zu werden – eine falsche Modell-ID oder ein falscher Schlüssel behebt sich nicht durch Warten von selbst.

Produktionsmuster 2: Client-Timeouts

Das requests.post(..., timeout=30) oben ist nicht dekorativ – ohne es blockiert eine hängende Upstream-Verbindung deinen Router unbegrenzt, und noch so viel Wiederholungslogik hilft nichts, weil du nie wieder die Kontrolle erlangst, um es erneut zu versuchen. Setze ein ausreichend knappes Timeout, damit hängende Anfragen schnell in den Retry-/Failover-Pfad fallen, anstatt einen Thread zu belegen. Für eine einzelne Chat-Completion sind 20–30 Sekunden sinnvoll; wenn du latenzempfindliche Anfragen routest und lieber auf ein schnelleres Modell ausweichst als auf ein langsames zu warten, kannst du es weiter senken.

Produktionsmuster 3: Kostenbewusstes hierarchisches Routing

Nicht jede Anfrage braucht dein teuerstes Modell. Ein praktisches Muster: Definiere Ebenen, versuche zuerst die günstigste Ebene, die wahrscheinlich erfolgreich ist, und steige nur bei Fehlschlag auf:

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

Dies versucht zuerst jedes Modell in der Budget-Ebene, bevor die Premium-Ebene Kosten verursacht, und steigt erst dann eine Ebene höher, wenn jedes Modell der aktuellen Ebene fehlgeschlagen ist – ein einzelner Anbieterausfall schiebt also keinen Traffic zum teuersten Modell, ein Ausfall einer gesamten Ebene hingegen schon. Bevor du entscheidest, zu welcher Ebene ein Modell gehört, prüfe die aktuellen Preise pro Modell auf der Preisseite – hiapi gibt Preise nicht in der /v1/models-Antwort an, hole sie also von dort, anstatt eine möglicherweise veraltete Zahl hart zu codieren.

Produktionsmuster 4: Idempotenz und Sicherheit bei doppelten Anfragen

/v1/chat/completions ist ein gewöhnliches POST ohne serverseitige Idempotenzschlüssel – anders als die Task-API /v1/tasks von hiapi gibt es keine Möglichkeit, dem Server einen clientseitig generierten Schlüssel mitzugeben und Wiederholungsanfragen deduplizieren zu lassen. Das bedeutet: Wenn eine Anfrage nachdem das Modell die Generierung abgeschlossen hat, aber bevor die Antwort dich erreicht, ein Timeout bekommt, bezahlt eine naive Wiederholung dieselbe Generierung zweimal.

Zwei praktische Gegenmaßnahmen:

Halte das Client-Timeout knapp genug, sodass „Timeout“ fast immer „das Modell wurde nie erreicht“ bedeutet und nicht „die Antwort ging auf dem Transportweg verloren“.
Wenn das Ergebnis des Chat-Aufrufs einen nachgelagerten Schreibvorgang auslöst (z. B. du befüllst eine Datenbankzeile mit der Modellausgabe), gestalte diesen Schreibvorgang idempotent – ein Upsert auf einem stabilen Schlüssel statt eines Insert –, sodass doppelte API-Aufrufe zwar verschwenderisch, aber nicht unsicher sind.

Hinweis zu Polling und Callbacks (Bild-/Video-Modelle)

Alles oben gilt für Textmodelle über /v1/chat/completions, die synchron sind – du erhältst die vollständige Antwort in einem einzigen HTTP-Roundtrip. Die Bild-, Video- und Audiomodelle von hiapi funktionieren anders: Du reichst einen Task bei /v1/tasks ein und pollst dann entweder dessen Endzustand oder registrierst eine Callback-URL. Wenn dein Router sowohl Text- als auch Medienmodelle abdecken soll, behandle sie als zwei getrennte Codepfade – Medien-Task-IDs sind nichts, was du mit Backoff erneut versuchen kannst, wie du eine Chat-Completion wiederholen würdest, und die Wiederholungslogik dieser Anleitung direkt auf /v1/tasks anzuwenden, würde nur doppelte Generierungsaufträge erzeugen.

Referenz zur Fehlerbehandlung

Der Chat-Endpunkt von hiapi gibt einen OpenAI-kompatiblen Fehlerumschlag zurück. Die beiden, auf die du beim Bauen eines Routers am häufigsten stößt:

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

Route anhand von error.code statt error.type – jeder Fehler teilt denselben "type": "hiapi_error", daher ist code der eigentliche Unterscheider zwischen „aufgeben“ (model_not_found, permission_denied) und „erneut versuchen“ (Ratenbegrenzung und Serverfehler, die über den HTTP-Status und nicht über einen eigenen Code signalisiert werden).

Alles zusammenführen

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

Für tiefere Einblicke in die einzelnen Modelle, die du in eine Ebene einklinken könntest, lies die API-Anleitungen zu DeepSeek V4.1 Flash, Claude Sonnet 4.6 und GLM-5.3.

Häufig gestellte Fragen

Hat hiapi einen eingebauten Load Balancer oder automatisches Failover? Nein – hiapi gibt dir einen Endpunkt und einen Schlüssel für mehrere Modelle, aber die Routing-Logik (welches Modell, in welcher Reihenfolge, mit welcher Retry-Strategie) ist eine Client-Aufgabe, und genau die implementiert diese Anleitung.

Was ist der Unterschied zwischen dem „Model Router“ von hiapi und dem direkten Aufruf eines Modells? Ein Modell direkt aufzurufen bedeutet Single Point of Failure: Wird dieses Modell gedrosselt oder fällt es vorübergehend aus, schlägt deine Anfrage fehl. Ein Router versucht alternative Modelle – am besten über verschiedene zugrunde liegende Anbieter hinweg –, bevor er aufgibt.

Kann ich auf dieselbe Weise über Bild- oder Videogenerierungsmodelle hinweg routen? Nicht mit diesem Code. Sie laufen über die asynchrone /v1/tasks-API statt über /v1/chat/completions und haben eine andere Semantik (einreichen und dann pollen oder einen Callback erhalten). Siehe den Hinweis oben.

Woher weiß ich, welche Modelle gerade für das Routing verfügbar sind? Rufe GET /v1/models auf und filtere supported_endpoint_types nach "openai" – hardcode keine Modellliste, da die Verfügbarkeit sich ändert.

Was sollte das Hochstufen auf eine teurere Modellebene auslösen? Im obigen Muster wird erst hochgestuft, wenn jedes Modell der aktuellen Ebene fehlgeschlagen ist. Du kannst das mit eigenen Signalen erweitern – z. B. hochstufen, wenn die Ausgabe der Budget-Ebene die Qualitätsprüfung deines Anwendungsfalls nicht besteht –, aber diese Logik liegt außerhalb der hiapi-API selbst.

Gibt es Ratenbegrenzungen, um die ich herum designen sollte? Ja, konzeptionell: Deine Wiederholungslogik sollte HTTP 429 wie oben gezeigt als wiederholbar mit Backoff behandeln und nicht als endgültigen Fehler. Prüfe deine konkreten Limits im Dashboard.