Search

Search HiAPI models, tools, and resources.

Back to blog
Tutorial

Building a Load Balancer for Multi-Provider AI APIs on hiapi

hiapitutorialapi-integrationroutingreliability

What you'll build

A thin routing layer in front of hiapi's chat completions endpoint that picks a model per request, retries transient failures, falls back to a different provider when one is down or rate-limited, and tracks cost by routing cheap requests to cheap models. This is the same pattern production systems use to avoid a single upstream provider becoming a single point of failure — except here "provider" just means "model id," because hiapi puts OpenAI-compatible access to dozens of providers behind one endpoint and one API key.

This guide covers text/chat models only. hiapi's image, video, and audio models run on a separate asynchronous job-queue API (/v1/tasks), which has its own polling/callback story — see the note near the end if you're routing across both.

Prerequisites

  • A hiapi API key. Grab one from the dashboard — every example below just needs Authorization: Bearer <key>.
  • Python 3.9+ with requests installed (pip install requests), or curl if you'd rather test from the shell first.
  • No SDK required — chat completions is a plain OpenAI-compatible REST endpoint, so anything that speaks HTTP works.

Minimal working example

Every hiapi text model speaks the same /v1/chat/completions shape. Here's a single call with 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
  }'

And the Python equivalent, which is what the rest of this guide builds on:

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

That works, but it has no failover: if deepseek-v4.1-flash is rate-limited or temporarily down, the request just fails. The rest of this guide turns that single call into a router.

Discover what you can route across

Don't hardcode a model list from memory — query GET /v1/models and filter on supported_endpoint_types. Chat-completions-compatible models report "openai"; hiapi's image/video/audio models report "task" instead and won't work with the code in this 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 [])]

Build your routing tiers from this list at startup (or cache it for a few hours) rather than freezing model ids in code — providers get added, retired, or renamed, and a tier built from a stale hardcoded list quietly degrades into routing to a model that no longer exists.

Production pattern 1: retries with backoff

Not every failure deserves a retry. A 400 with model_not_found will fail identically on the next attempt — retrying just burns latency. A 429 or 5xx is usually transient and worth a short 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, and 403 fall straight through raise_for_status() without a retry — a bad model id or a bad key won't fix itself by waiting.

Production pattern 2: client-side timeouts

requests.post(..., timeout=30) above isn't decorative — without it, a hung upstream connection blocks your router indefinitely and no amount of retry logic helps, because you never get back control to retry. Set a timeout tight enough that a stuck request fails fast into your retry/failover path instead of holding a thread open. 20–30s is reasonable for a single chat completion; tune it down further if you're routing latency-sensitive requests and would rather fail over to a second model than wait out a slow one.

Production pattern 3: cost-aware tiered routing

Not every request needs your most expensive model. A practical pattern: define tiers, try the cheapest tier that's likely to succeed, and escalate only on failure:

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

This tries every model in the budget tier before spending on the premium tier, and only moves up a tier after every model in the current one has failed — so a single provider outage doesn't push traffic to your most expensive models, but a tier-wide outage does. Check current per-model pricing on the pricing page before deciding where a model belongs in your tiers — hiapi doesn't expose price in the /v1/models response, so pull it from there rather than hardcoding a number that can go stale.

Production pattern 4: idempotency and duplicate-request safety

/v1/chat/completions is a plain POST with no server-side idempotency key — unlike hiapi's /v1/tasks job API, there's no way to hand the server a client-generated key and have it dedupe a retried request for you. That means if a request times out after the model finished generating but before the response reached you, a naive retry will pay for the same generation twice.

Two practical mitigations:

  • Keep your client-side timeout tight enough that "timed out" almost always means "never reached the model," not "response lost in transit."
  • If the result of a chat call drives a downstream write (e.g., you're using the model's output to populate a database row), make that write idempotent — upsert on a stable key instead of insert — so a duplicate API call is wasteful but not unsafe.

A note on polling vs. callbacks (image/video models)

Everything above applies to text models on /v1/chat/completions, which is synchronous — you get the full response in one HTTP round trip. hiapi's image, video, and audio models work differently: you submit a job to /v1/tasks, then either poll for its terminal state or register a callback URL. If your router needs to span both text and media models, treat them as two separate code paths — a media task id is not something you retry with backoff the way you'd retry a chat completion, and applying this guide's retry logic directly to /v1/tasks will just create duplicate generation jobs.

Error handling reference

hiapi's chat endpoint returns OpenAI-compatible error envelopes. The two you'll hit most often when building a 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"
  }
}

Route on error.code, not error.type — every error shares the same "type": "hiapi_error", so code is what actually distinguishes "give up" (model_not_found, permission_denied) from "retry" (rate-limit and server errors, surfaced via HTTP status rather than a distinct code).

Putting it together

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

For deeper dives on individual models you might slot into a tier, see the DeepSeek V4.1 Flash, Claude Sonnet 4.6, and GLM-5.3 API guides.

FAQ

Does hiapi have a built-in load balancer or automatic failover? No — hiapi gives you one endpoint and one key across many models, but routing logic (which model, in what order, with what retry policy) is a client-side concern, which is exactly what this guide implements.

What's the difference between a hiapi "model router" and calling one model directly? Calling one model directly means a single point of failure: if that model is rate-limited or temporarily down, your request fails. A router tries alternate models — ideally across different underlying providers — before giving up.

Can I route across image or video generation models the same way? Not with this code. Those run on the asynchronous /v1/tasks API instead of /v1/chat/completions, with different semantics (submit, then poll or receive a callback). See the note above.

How do I know which models are currently available to route across? Call GET /v1/models and filter for "openai" in supported_endpoint_types — don't hardcode a model list, since availability changes.

What should trigger escalation to a more expensive model tier? In the pattern above, escalation happens only when every model in the current tier has failed. You can extend this with your own signal — e.g., escalate when the budget tier's output fails a quality check for your use case — but that logic is outside the scope of hiapi's API itself.

Is there a rate limit I should design around? Yes, conceptually: your retry logic should treat HTTP 429 as retryable with backoff (as shown above), not as a terminal failure. Check your plan's specific limits in the dashboard.

Generate it with HiAPI

Choose a model, enter your prompt, and see the result.

HiAPI Blog

Related articles

HiAPI

Generate it with HiAPI

Start generating