Создание балансировщика нагрузки для API мультипровайдерных ИИ на hiapi

Что вы создадите

Добавьте лёгкий слой маршрутизации перед конечной точкой chat completions в hiapi: выбирайте модель по запросу, повторяйте попытки при временных сбоях, переключайтесь на других провайдеров, когда один из них недоступен или ограничен по скорости, и отслеживайте затраты, направляя дешёвые запросы к дешёвым моделям. Это тот же шаблон, который используют производственные системы, чтобы избежать единой точки отказа у одного вышестоящего провайдера, — только здесь «провайдер» означает просто «ID модели», поскольку hiapi объединяет OpenAI-совместимый доступ к десяткам провайдеров за одной конечной точкой и одним ключом API.

Это руководство охватывает только текстовые/чат-модели. Модели изображений, видео и аудио в hiapi работают через отдельный API очереди асинхронных задач (/v1/tasks) с собственным механизмом опроса/обратных вызовов — если вам нужно маршрутизировать между ними, см. примечание в конце.

Предварительные требования

• Ключ API hiapi. Получите его в панели управления — во всех примерах ниже требуется только Authorization: Bearer <key>.
• Python 3.9+ с установленным requests (pip install requests) или curl, если вы предпочитаете сначала протестировать в оболочке.
• SDK не нужен — chat completions — это обычная конечная точка REST, совместимая с OpenAI, поэтому подойдёт любой инструмент, поддерживающий HTTP.

Минимальный рабочий пример

Каждая текстовая модель hiapi следует одной и той же структуре /v1/chat/completions. Вот одиночный вызов с помощью 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
  }'

И эквивалентная версия на Python, на которой основана остальная часть руководства:

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

Это работает, но без отказоустойчивости: если deepseek-v4.1-flash ограничен по скорости или временно недоступен, запрос просто завершится ошибкой. Остальная часть руководства превратит этот одиночный вызов в маршрутизатор.

Определение того, что можно маршрутизировать

Не кодируйте список моделей жёстко по памяти — запросите GET /v1/models и отфильтруйте по supported_endpoint_types. Модели, совместимые с chat completions, сообщают "openai"; модели изображений/видео/аудио hiapi сообщают "task" и не могут использоваться с кодом из этого руководства:

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

Стройте уровни маршрутизации из этого списка при запуске (или кешируйте его на несколько часов), а не фиксируйте ID моделей в коде — провайдеры добавляют, снимают с обслуживания или переименовывают модели, и уровень, построенный на устаревшем жёстко заданном списке, будет незаметно деградировать до маршрутизации к несуществующим моделям.

Производственный шаблон 1: повторные попытки с экспоненциальной задержкой

Не каждый сбой стоит повторять. 400 с model_not_found снова провалится при следующей попытке — повтор только тратит задержку. 429 или 5xx обычно временные и заслуживают короткой задержки:

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 и 403 проходят через raise_for_status() без повторных попыток — неправильный ID модели или неправильный ключ не исправятся сами собой от ожидания.

Производственный шаблон 2: тайм-ауты на стороне клиента

Приведённый выше requests.post(..., timeout=30) не для украшения — без него зависшее вышестоящее соединение будет бесконечно блокировать ваш маршрутизатор, и никакая логика повторных попыток не поможет, потому что вы никогда не вернёте контроль, чтобы повторить попытку. Установите достаточно короткий тайм-аут, чтобы зависшие запросы быстро переходили в путь повторной попытки/отказоустойчивости, а не занимали поток. Для одного chat completion разумны 20–30 секунд; если вы маршрутизируете запросы, чувствительные к задержке, и предпочитаете переключиться на более быструю модель, а не ждать медленную, можно снизить ещё больше.

Производственный шаблон 3: маршрутизация с учётом стоимости по уровням

Не каждому запросу нужна ваша самая дорогая модель. Практичный шаблон: определите уровни, сначала попробуйте самый дешёвый уровень, который может успешно справиться, и повышайте только при сбое:

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

Это попробует каждую модель бюджетного уровня перед тем, как потратить деньги на премиум-уровень, и перейдёт на следующий уровень только после того, как каждая модель в текущем уровне потерпит неудачу — поэтому сбой одного провайдера не направит трафик к самой дорогой модели, но сбой всего уровня — направит. Прежде чем решать, к какому уровню относится модель, проверьте текущие цены на модель на странице цен — hiapi не раскрывает цены в ответе /v1/models, поэтому берите их оттуда, а не жёстко задавайте потенциально устаревшее число.

Производственный шаблон 4: идемпотентность и безопасность повторных запросов

/v1/chat/completions — это обычный POST без серверного ключа идемпотентности — в отличие от API задач /v1/tasks в hiapi, здесь нет способа передать серверу сгенерированный клиентом ключ, чтобы он дедуплицировал повторные попытки запроса. Это означает, что если запрос истекает по тайм-ауту после того, как модель завершила генерацию, но до того, как ответ дошёл до вас, наивная повторная попытка оплатит одну и ту же генерацию дважды.

Две практичные меры смягчения:

• Держите тайм-аут клиента достаточно коротким, чтобы «тайм-аут» почти всегда означал «запрос никогда не дошёл до модели», а не «ответ потерялся в пути».
• Если результат вызова chat управляет нижестоящей записью (например, вы заполняете строку базы данных выводом модели), сделайте идемпотентной эту запись — выполните upsert по стабильному ключу вместо insert — чтобы повторные вызовы API были расточительными, но не небезопасными.

Примечание об опросе и обратных вызовах (модели изображений/видео)

Всё вышесказанное относится к текстовым моделям на /v1/chat/completions, которые синхронны — вы получаете полный ответ за один HTTP-обмен. Модели изображений, видео и аудио в hiapi работают иначе: вы отправляете задачу в /v1/tasks, а затем либо опрашиваете её конечное состояние, либо регистрируете URL обратного вызова. Если вашему маршрутизатору нужно охватывать и текстовые, и медиамодели, рассматривайте их как два независимых пути кода — ID медиазадачи не то, что можно повторять с задержкой, как chat completion, и прямое применение логики повторных попыток из этого руководства к /v1/tasks просто создаст дублирующиеся задачи генерации.

Справочник по обработке ошибок

Конечная точка chat в hiapi возвращает конверт ошибок, совместимый с OpenAI. Две, с которыми вы будете сталкиваться чаще всего при построении маршрутизатора:

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

Маршрутизируйте на основе error.code, а не error.type — каждая ошибка имеет одинаковое "type": "hiapi_error", поэтому именно code является реальным основанием для различения «сдаться» (model_not_found, permission_denied) и «повторить» (ограничение скорости и серверные ошибки, представленные через HTTP-статус, а не отдельным кодом).

Собираем всё вместе

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

Для более подробного знакомства с отдельными моделями, которые вы можете включить в уровень, см. руководства по API DeepSeek V4.1 Flash, Claude Sonnet 4.6 и GLM-5.3.

Часто задаваемые вопросы

Есть ли в hiapi встроенный балансировщик нагрузки или автоматическая отказоустойчивость? Нет — hiapi предоставляет вам одну конечную точку и один ключ для множества моделей, но логика маршрутизации (какую модель использовать, в каком порядке, какую стратегию повторных попыток применять) — это забота клиента, что и реализует данное руководство.

В чём разница между «маршрутизатором моделей» hiapi и прямым вызовом одной модели? Прямой вызов одной модели означает единую точку отказа: если эта модель ограничена по скорости или временно недоступна, ваш запрос завершится ошибкой. Маршрутизатор попробует альтернативные модели, прежде чем сдаться — в идеале от разных нижестоящих провайдеров.

Могу ли я маршрутизировать между моделями генерации изображений или видео таким же образом? Не с помощью этого кода. Они работают через асинхронный API /v1/tasks, а не /v1/chat/completions, с другой семантикой (отправка, затем опрос или получение обратного вызова). См. примечание выше.

Как узнать, какие модели сейчас доступны для маршрутизации? Вызовите GET /v1/models и отфильтруйте "openai" в supported_endpoint_types — не жёстко задавайте список моделей, поскольку доступность меняется.

Что должно вызывать повышение до более дорогого уровня модели? В шаблоне выше повышение происходит только тогда, когда каждая модель в текущем уровне потерпела неудачу. Вы можете расширить это собственными сигналами — например, повысить уровень, когда вывод бюджетного уровня не проходит проверку качества для вашего случая использования, — но эта логика выходит за рамки самого API hiapi.

Есть ли ограничения скорости, которые мне следует учитывать при проектировании? Да, концептуально: ваша логика повторных попыток должна рассматривать HTTP 429 как повторяемую с задержкой (как показано выше), а не как окончательный сбой. Смотрите конкретные ограничения вашего тарифа в панели управления.