Construa um balanceador de carga para APIs de IA multicliente no hiapi

O que você vai construir

Adicione uma camada de roteamento leve na frente do endpoint de chat completions do hiapi: selecione o modelo por requisição, tente novamente em falhas transitórias, faça fallback para outros provedores quando um provedor cair ou for limitado por taxa e acompanhe custos roteando requisições baratas para modelos baratos. Este é o mesmo padrão que sistemas de produção adotam para evitar que um único provedor upstream se torne um ponto único de falha — exceto que aqui "provedor" significa apenas "ID de modelo", porque o hiapi coloca o acesso compatível com OpenAI a dezenas de provedores atrás do mesmo endpoint e da mesma chave de API.

Este guia aborda apenas modelos de texto/chat. Os modelos de imagem, vídeo e áudio do hiapi rodam em uma API separada de fila de tarefas assíncronas (/v1/tasks), que tem seu próprio mecanismo de polling/callback — se você precisar rotear entre os dois, veja a observação no final.

Pré-requisitos

• Uma chave de API do hiapi. Obtenha-a no painel — todos os exemplos abaixo precisam apenas de Authorization: Bearer <key>.
• Python 3.9+ com requests instalado (pip install requests), ou curl se você preferir testar no shell primeiro.
• Nenhum SDK necessário — chat completions é um endpoint REST comum compatível com OpenAI, então qualquer ferramenta que suporte HTTP funciona.

Exemplo mínimo executável

Todo modelo de texto do hiapi segue a mesma estrutura /v1/chat/completions. Aqui está uma chamada única com 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 a versão equivalente em Python, na qual o restante deste guia se baseia:

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

Isso funciona, mas não tem failover: se deepseek-v4.1-flash for limitado por taxa ou cair temporariamente, a requisição simplesmente falha. O restante deste guia transforma essa chamada única em um roteador.

Descubra o que você pode rotear

Não codifique uma lista de modelos de memória — consulte GET /v1/models e filtre por supported_endpoint_types. Modelos compatíveis com chat completions reportam "openai"; os modelos de imagem/vídeo/áudio do hiapi reportam "task" e não funcionam com o código deste guia:

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

Construa suas camadas de roteamento a partir dessa lista na inicialização (ou armazene-a em cache por algumas horas), em vez de congelar IDs de modelo no código — os provedores adicionam, descontinuam ou renomeiam modelos, e camadas construídas sobre uma lista codificada desatualizada degradam silenciosamente para rotear para modelos que não existem mais.

Padrão de produção 1: retentativas com backoff

Nem toda falha merece nova tentativa. Um 400 com model_not_found falhará da mesma forma na próxima tentativa — a retentativa só desperdiça latência. 429 ou 5xx geralmente são transitórios e merecem um 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 passam direto por raise_for_status() sem retentativa — um ID de modelo errado ou uma chave errada não se corrigem esperando.

Padrão de produção 2: timeouts no cliente

O requests.post(..., timeout=30) acima não é decorativo — sem ele, uma conexão upstream travada bloqueia seu roteador indefinidamente, e nenhuma lógica de retentativa ajuda, porque você nunca recupera o controle para tentar de novo. Defina timeouts suficientemente curtos para que requisições travadas falhem rápido e sigam para o caminho de retentativa/failover, em vez de ocupar uma thread indefinidamente. Para um único chat completion, 20–30 segundos é razoável; se você estiver roteando requisições sensíveis a latência e preferir fazer failover para um modelo mais rápido em vez de esperar um modelo lento, reduza ainda mais.

Padrão de produção 3: roteamento em camadas com consciência de custo

Nem toda requisição precisa do seu modelo mais caro. Um padrão prático: defina camadas, tente primeiro a camada mais barata com chance de sucesso e só escale em caso de falha:

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

Isso tenta todos os modelos da camada econômica antes de gastar na camada premium e só sobe uma camada depois que todos os modelos da camada atual falham — assim a queda de um único provedor não empurra o tráfego para o modelo mais caro, mas a queda da camada inteira sim. Antes de decidir em qual camada um modelo se encaixa, verifique o preço atual por modelo na página de preços — o hiapi não expõe preços na resposta de /v1/models, então obtenha-os de lá em vez de codificar um valor que pode estar desatualizado.

Padrão de produção 4: idempotência e segurança contra requisições duplicadas

/v1/chat/completions é um POST comum, sem chave de idempotência no lado do servidor — diferente da API de tarefas /v1/tasks do hiapi, não há como fornecer ao servidor uma chave gerada pelo cliente e fazer com que ele deduplique requisições repetidas. Isso significa que, se uma requisição expirar depois que o modelo terminou de gerar mas antes que a resposta chegue até você, uma retentativa ingênua pagará duas vezes pela mesma geração.

Duas mitigações práticas:

• Mantenha o timeout do cliente curto o suficiente para que "timeout" quase sempre signifique "nunca chegou ao modelo" em vez de "a resposta se perdeu no trânsito".
• Se o resultado da chamada de chat alimenta uma escrita downstream (por exemplo, você preenche uma linha do banco de dados com a saída do modelo), torne essa escrita idempotente — faça upsert em uma chave estável em vez de insert — para que chamadas de API duplicadas sejam desperdício, mas não inseguras.

Observação sobre polling e callbacks (modelos de imagem/vídeo)

Tudo acima se aplica a modelos de texto em /v1/chat/completions, que é síncrono — você obtém a resposta completa em uma única ida e volta HTTP. Os modelos de imagem, vídeo e áudio do hiapi funcionam de forma diferente: você envia uma tarefa para /v1/tasks e então ou consulta seu estado final ou registra uma URL de callback. Se seu roteador precisar cobrir modelos de texto e de mídia, trate-os como dois caminhos de código separados — um ID de tarefa de mídia não é algo que você possa tentar novamente com backoff como um chat completion, e aplicar a lógica de retentativa deste guia diretamente a /v1/tasks só criaria tarefas de geração duplicadas.

Referência de tratamento de erros

O endpoint de chat do hiapi retorna um envelope de erro compatível com OpenAI. Os dois que você encontrará com mais frequência ao construir um roteador:

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

Roteie com base em error.code em vez de error.type — todo erro compartilha o mesmo "type": "hiapi_error", então code é o que realmente distingue "desistir" (model_not_found, permission_denied) de "tentar novamente" (limitação de taxa e erros de servidor, apresentados via status HTTP e não por códigos distintos).

Juntando tudo

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

Para uma introdução aprofundada aos modelos individuais que você pode inserir em uma camada, consulte os guias de API do DeepSeek V4.1 Flash, Claude Sonnet 4.6 e GLM-5.3.

Perguntas frequentes

O hiapi tem um balanceador de carga embutido ou failover automático? Não — o hiapi fornece um endpoint e uma chave para vários modelos, mas a lógica de roteamento (qual modelo usar, em que ordem, com que estratégia de retentativa) é uma preocupação do cliente, e é exatamente isso que este guia implementa.

Qual é a diferença entre o "roteador de modelos" do hiapi e chamar um modelo diretamente? Chamar um modelo diretamente significa ponto único de falha: se esse modelo for limitado por taxa ou cair temporariamente, sua requisição falha. O roteador tenta modelos alternativos antes de desistir — de preferência em provedores subjacentes diferentes.

Posso rotear entre modelos de geração de imagem ou vídeo da mesma forma? Não com este código. Eles rodam na API assíncrona /v1/tasks, não em /v1/chat/completions, e têm semântica diferente (enviar, depois consultar ou receber callback). Veja a observação acima.

Como sei quais modelos estão disponíveis para roteamento agora? Chame GET /v1/models e filtre por "openai" em supported_endpoint_types — não codifique a lista de modelos, pois a disponibilidade muda.

O que deve acionar a escalada para uma camada de modelo mais caro? No padrão acima, a escalada só ocorre quando todos os modelos da camada atual falham. Você pode estender isso com seus próprios sinais — por exemplo, escalar quando a saída da camada econômica não passa no seu controle de qualidade para o caso de uso — mas essa lógica está fora do escopo da API do hiapi em si.

Existem limites de taxa em torno dos quais devo projetar? Sim, conceitualmente: sua lógica de retentativa deve tratar HTTP 429 como retentável com backoff (como mostrado acima), não como falha terminal. Consulte os limites específicos do seu plano no painel.