Crear un balanceador de carga para API de IA multiproveedor en hiapi

Qué vas a construir

Coloca una capa de enrutamiento ligera delante del endpoint de chat completions de hiapi: selecciona el modelo por solicitud, reintenta ante fallos transitorios, recurre a otros proveedores cuando uno está caído o limitado, y rastrea los costos enrutando solicitudes económicas a modelos económicos. Es el mismo patrón que usan los sistemas de producción para evitar que un único proveedor upstream se convierta en un punto único de fallo; solo que aquí «proveedor» significa simplemente «ID de modelo», porque hiapi expone el acceso compatible con OpenAI a decenas de proveedores detrás de un mismo endpoint y una misma clave de API.

Esta guía solo cubre modelos de texto/chat. Los modelos de imagen, vídeo y audio de hiapi se ejecutan en una API de cola de tareas asíncrona independiente (/v1/tasks), con su propio mecanismo de sondeo/retrollamada; si necesitas enrutar entre ambos, consulta la nota al final.

Requisitos previos

• Una clave de API de hiapi. Obtén una desde el panel; cada uno de los ejemplos siguientes solo necesita Authorization: Bearer <key>.
• Python 3.9+ con requests instalado (pip install requests), o curl si prefieres probar primero en el shell.
• No se necesita SDK: chat completions es un endpoint REST normal compatible con OpenAI, así que cualquier herramienta que soporte HTTP sirve.

Ejemplo mínimo funcional

Todos los modelos de texto de hiapi siguen la misma estructura /v1/chat/completions. Aquí tienes una llamada única 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
  }'

Y la versión equivalente en Python, sobre la que se basa el resto de esta guía:

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

Esto funciona, pero no hay conmutación por error: si deepseek-v4.1-flash está limitado o temporalmente caído, la solicitud falla directamente. El resto de esta guía convierte esa llamada única en un enrutador.

Descubre el alcance de lo que puedes enrutar

No codifiques a mano una lista de modelos de memoria; consulta GET /v1/models y filtra por supported_endpoint_types. Los modelos compatibles con chat completions informan "openai"; los modelos de imagen/vídeo/audio de hiapi informan "task" y no funcionarán con el código de esta guía:

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

Construye tu jerarquía de enrutamiento a partir de esta lista al iniciar (o cachéala durante unas horas), en lugar de congelar IDs de modelo en el código: los proveedores añaden, retiran o renombran modelos, y una jerarquía construida sobre una lista codificada obsoleta se degradará silenciosamente a enrutar hacia modelos que ya no existen.

Patrón de producción 1: reintentos con retroceso exponencial

No todo fallo merece un reintento. Un 400 con model_not_found fallará igual en el siguiente intento; reintentar solo desperdicia latencia. Un 429 o un 5xx suelen ser transitorios y merecen un breve retroceso:

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 y 403 atraviesan raise_for_status() sin reintentarse: un ID de modelo incorrecto o una clave incorrecta no se arreglan solos esperando.

Patrón de producción 2: tiempos de espera en el cliente

El requests.post(..., timeout=30) anterior no es decorativo: sin él, una conexión upstream colgada bloquea tu enrutador indefinidamente, y por mucha lógica de reintentos que tengas no servirá de nada, porque nunca recuperarás el control para reintentar. Establece un tiempo de espera lo suficientemente ajustado como para que las solicitudes atascadas fallen rápido y pasen a la ruta de reintento/conmutación por error, en lugar de ocupar un hilo indefinidamente. Para un único chat completion, 20–30 segundos es razonable; si enrutas solicitudes sensibles a la latencia y prefieres conmutar por error a un modelo más rápido antes que esperar a uno lento, redúcelo aún más.

Patrón de producción 3: enrutamiento por niveles consciente del costo

No todas las solicitudes necesitan tu modelo más caro. Un patrón práctico: define niveles, intenta primero el nivel más económico que probablemente tenga éxito y escala solo si falla:

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

Esto intenta cada modelo del nivel económico antes de gastar en el nivel premium, y solo sube un nivel después de que falle cada modelo del nivel actual; así, la caída de un único proveedor no empuja el tráfico hacia el modelo más caro, pero la caída de un nivel completo sí lo hace. Antes de decidir a qué nivel pertenece un modelo, consulta el precio actual por modelo en la página de precios; hiapi no expone precios en la respuesta de /v1/models, así que obténlos de ahí en lugar de codificar un número que podría quedar obsoleto.

Patrón de producción 4: idempotencia y seguridad ante solicitudes duplicadas

/v1/chat/completions es un POST normal sin claves de idempotencia del lado del servidor; a diferencia de la API de tareas /v1/tasks de hiapi, no hay forma de proporcionar al servidor una clave generada por el cliente y hacer que deduplique solicitudes reintentadas. Esto significa que si una solicitud se agota después de que el modelo termine de generar pero antes de que llegue la respuesta, un reintento ingenuo pagará dos veces por la misma generación.

Dos mitigaciones prácticas:

• Mantén el tiempo de espera del cliente lo suficientemente ajustado como para que «timeout» casi siempre signifique «nunca llegó al modelo» y no «la respuesta se perdió en tránsito».
• Si el resultado de la llamada de chat impulsa una escritura posterior (por ejemplo, usas la salida del modelo para rellenar una fila de base de datos), haz que esa escritura sea idempotente: upsert sobre una clave estable en lugar de insert, de modo que las llamadas duplicadas a la API, aunque derrochadoras, no sean inseguras.

Nota sobre sondeo y retrollamadas (modelos de imagen/vídeo)

Todo lo anterior se aplica a los modelos de texto en /v1/chat/completions, que es síncrono: obtienes la respuesta completa en un único viaje de ida y vuelta HTTP. Los modelos de imagen, vídeo y audio de hiapi funcionan de forma distinta: envías una tarea a /v1/tasks y luego sondeas su estado final o registras una URL de retrollamada. Si tu enrutador necesita cubrir tanto modelos de texto como de medios, trátalos como dos rutas de código separadas: un ID de tarea de medios no es algo que puedas reintentar con retroceso como un chat completion, y aplicar directamente la lógica de reintentos de esta guía a /v1/tasks solo crearía tareas de generación duplicadas.

Referencia de manejo de errores

El endpoint de chat de hiapi devuelve un sobre de error compatible con OpenAI. Los dos que más encontrarás al construir un enrutador:

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

Enruta basándote en error.code, no en error.type: todos los errores comparten el mismo "type": "hiapi_error", así que code es lo que realmente distingue «abandonar» (model_not_found, permission_denied) de «reintentar» (los límites de tasa y los errores del servidor se manifiestan mediante el estado HTTP, no mediante códigos distintos).

Poniéndolo todo junto

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 una introducción en profundidad a los distintos modelos que podrías insertar en un nivel, consulta las guías de API de DeepSeek V4.1 Flash, Claude Sonnet 4.6 y GLM-5.3.

Preguntas frecuentes

¿hiapi tiene un balanceador de carga integrado o conmutación por error automática? No: hiapi te ofrece un endpoint y una clave para múltiples modelos, pero la lógica de enrutamiento (qué modelo usar, en qué orden, con qué política de reintentos) es una preocupación del cliente, que es exactamente lo que implementa esta guía.

¿Cuál es la diferencia entre el «enrutador de modelos» de hiapi y llamar directamente a un modelo? Llamar directamente a un modelo significa un punto único de fallo: si ese modelo está limitado o temporalmente caído, tu solicitud falla. El enrutador prueba modelos alternativos antes de rendirse, preferiblemente de distintos proveedores subyacentes.

¿Puedo enrutar entre modelos de generación de imágenes o vídeo de la misma manera? No con este código. Se ejecutan en la API asíncrona /v1/tasks, no en /v1/chat/completions, y tienen una semántica diferente (enviar y luego sondear o recibir una retrollamada). Consulta la nota anterior.

¿Cómo sé qué modelos están disponibles actualmente para enrutar? Llama a GET /v1/models y filtra por "openai" en supported_endpoint_types; no codifiques la lista de modelos, porque la disponibilidad cambia.

¿Qué debería provocar el escalado a un nivel de modelo más caro? En el patrón anterior, solo se escala cuando falla cada modelo del nivel actual. Puedes ampliar esto con tus propias señales, por ejemplo, escalando cuando la salida del nivel económico no supera la comprobación de calidad de tu caso de uso; pero esa lógica queda fuera del alcance de la propia API de hiapi.

¿Hay límites de tasa en torno a los que deba diseñar? Sí, conceptualmente: tu lógica de reintentos debe tratar el 429 HTTP como reintentable con retroceso (como se muestra arriba), no como un fallo terminal. Consulta los límites específicos de tu plan en el panel.