Интеграция API генерации видео с ИИ в Python: опрос, Webhook и контроль затрат

Выбор режима получения результата

По сравнению с большинством вызовов API генерация видео идёт медленно. Запрос на создание нескольких секунд видео может занимать от десятков секунд до нескольких минут — это означает, что HTTP-ответ на создание задачи почти никогда не содержит само видео. Прежде чем написать хотя бы строку кода на Python, решите, как вы будете получать результат. Есть три практичных режима.

Режим 1: Отправка и опрос

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

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

Режим 2: Callback через Webhook

Вы отправляете задачу с URL обратного вызова, возвращаете управление вызывающей стороне и позволяете платформе отправить результат POST-запросом на ваш эндпоинт по завершении задачи. Это полностью устраняет опрос и даёт почти мгновенное уведомление. Он также продолжает работать после перезапуска процесса, отправившего задачу, поскольку доставка управляется извне.

Цена — эксплуатационная: вам нужен общедоступный HTTPS-эндпоинт, нужно проверять, что обратный вызов действительно пришёл от провайдера, и нужно обрабатывать случаи, когда обратный вызов так и не приходит — потому что сети дают сбои, эндпоинты настраиваются с ошибками, а полезные нагрузки теряются.

Webhook подходит для длительных или пакетных нагрузок: большие очереди задач, асинхронные конвейеры и любые сценарии, где удерживать соединение или крутить цикл вхолостую в течение нескольких минут — это расточительство.

Режим 3: Гибридный режим с подстраховкой

Режим, к которому в конечном итоге приходят большинство производственных систем: Webhook как основной сигнал, но с сохранением медленной периодической сверки, опрашивающей задачи, которые всё ещё помечены как ожидающие после порога устаревания. Если обратный вызов пропущен, его поймает опросчик. Если опросчик запаздывает, обратный вызов обычно приходит первым и помечает задачу выполненной.

Выбирайте по нагрузке, а не по кодовой базе. Интерактивные эндпоинты могут опрашивать; фоновые очереди могут использовать обратные вызовы; общее хранилище задач и логика терминальных состояний должны быть одинаковыми для обоих, чтобы стратегию получения можно было изменить, не затрагивая остальную систему.

Настройка клиента и отправка задачи

Скелет отправки должен оставаться компактным. Всё ниже предполагает, что есть одно место для хранения вашего API-ключа и базового URL и функция, превращающая промпт с параметрами в дескриптор задачи.

Конфигурация и обработка ключей

Никогда не хардкодьте ключи. Читайте их из переменных окружения и громко падайте при запуске, если они отсутствуют, чтобы не обнаружить проблему при первом запросе клиента.

import os

BASE_URL = os.environ.get("HIAPI_BASE_URL", "https://api.hiapi.ai")
API_KEY = os.environ["HIAPI_API_KEY"]  # 启动时的 KeyError 是故意的

Держите ключи подальше от журналов. Если вы логируете запросы для отладки, логируйте промпты и параметры, но маскируйте заголовок авторизации.

Одиночный вызов отправки

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

import requests

def submit_video_job(prompt: str, *, idempotency_key: str, **params) -> dict:
    payload = {"prompt": prompt, **params}
    resp = requests.post(
        f"{BASE_URL}/v1/video/generations",
        headers={
            "Authorization": f"Bearer {API_KEY}",
            "Content-Type": "application/json",
            "Idempotency-Key": idempotency_key,
        },
        json=payload,
        timeout=(5, 30),  # соединение, чтение
    )
    resp.raise_for_status()
    return resp.json()

Ключи идемпотентности важны. Если ваш вызов отправки завершился по тайм-ауту, вы фактически не знаете, была ли задача создана. Повтор с тем же ключом безопасно переотправит запрос, а не создаст — и не заставит вас платить за — дублирующую задачу.

Что следует немедленно сохранить

В момент, когда отправка возвращает результат, прежде чем делать что-либо ещё, запишите строку в свою собственную базу данных:

• ID задачи и ключ идемпотентности
• ваш собственный внутренний ID запроса и пользователя или пакет, к которому он относится
• точные параметры запроса, чтобы вы могли его воспроизвести или проверить
• состояние pending и временную метку

Если ваш процесс упадёт после отправки, но до сохранения, задача существует на стороне провайдера, но без локальной записи. Сохранение сначала, а затем продолжение, делает вашу базу данных источником истины для того, что вы должны сверять.

Опрос результатов правильным способом

Цикл опроса легко написать плохо. Наивная версия — while True без паузы — расходует квоту, бомбардирует API и делает ваши журналы нечитаемыми. Правильный цикл имеет четыре свойства: интервал, экспоненциальная задержка, жёсткий предел и проверка терминального состояния.

Цикл

import time, random

def poll_job(job_id: str, *, max_attempts: int = 60, base: float = 2.0, cap: float = 20.0):
    for attempt in range(1, max_attempts + 1):
        resp = requests.get(
            f"{BASE_URL}/v1/video/generations/{job_id}",
            headers={"Authorization": f"Bearer {API_KEY}"},
            timeout=(5, 15),
        )
        resp.raise_for_status()
        body = resp.json()
        status = body["status"]

        if status == "succeeded":
            return body
        if status in {"failed", "cancelled"}:
            raise RuntimeError(f"任务 {job_id} 以 {status} 结束：{body.get('error')}")

        delay = min(cap, base * (2 ** (attempt - 1)))
        time.sleep(delay + random.uniform(0, 1.0))

    raise TimeoutError(f"任务 {job_id} 在 {max_attempts} 次尝试后仍为待处理")

Почему важна каждая часть

Экспоненциальная задержка. Задачи генерации продвигаются по этапам: постановка в очередь, выполнение, рендеринг. Опрос каждую секунду на ранней стадии постановки в очередь двадцать раз вернёт один и тот же ответ. Начинать с нескольких секунд и удваивать до верхнего предела позволяет объёму вызовов быть пропорциональным длительности задачи, а не вашему терпению.

Джиттер. Добавление случайных десятых долей секунды предотвращает синхронный опрос всеми запущенными клиентами после развёртывания или шторма повторов.

Максимальное число попыток и бюджет по настенному времени. Экспоненциальная задержка без ограничения числа попыток может выполняться часами. Выберите верхний предел исходя из ожидаемой худшей длительности задачи и рассматривайте превышение как сбой, который нужно выявить, а не как цикл, который продолжает тихо работать. Также отслеживайте прошедшее время — как только вступает в силу ограничение интервала, одно лишь число попыток вводит в заблуждение.

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

Ещё одно правило: не опрашивайте внутри обработчика веб-запросов. Переместите цикл в фоновый обработчик, чтобы медленные задачи не занимали пул потоков запросов.

Получение результатов через callback Webhook

Эндпоинт Webhook — это публичный, по умолчанию неаутентифицированный HTTP-маршрут, на который кто угодно может отправить POST. Относитесь к нему соответственно.

Проверка отправителя

Если платформа подписывает обратные вызовы, проверяйте подпись по исходному телу запроса до разбора JSON — проверка подписи выполняется над байтами, а повторная сериализация полезной нагрузки может их изменить. Сравнивайте с помощью функции с постоянным временем выполнения и отклоняйте несовпадения с не-2xx статусом. Если подпись недоступна, используйте хотя бы секретный сегмент пути плюс список разрешённых или общий токен в заголовке и никогда не доверяйте содержимому полезной нагрузки, пока не сопоставите её с созданной вами задачей.

Быстро отвечайте, работайте позже

Единственная задача вашего эндпоинта — подтвердить получение. Разобрать, проверить, поставить в очередь, ответить. Всё медленное — скачивание видео, вызов базы данных под блокировкой, вызов нижестоящего сервиса — происходит в обработчике, а не в эндпоинте. Провайдер повторяет обратные вызовы, не получившие быстрого успешного ответа, а медленный обработчик превращает одно событие в несколько.

from flask import Flask, request, abort

app = Flask(__name__)

@app.post("/callbacks/video")
def video_callback():
    raw = request.get_data()
    if not verify_signature(raw, request.headers):
        abort(401)
    event = request.get_json(force=True)
    job_id = event.get("job_id")
    if not job_id:
        abort(400)
    queue.enqueue("process_video_event", event)  # 立即返回
    return {"ok": True}, 200

Сверка с вашими записями

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

• Неизвестный ID задачи: это может быть устаревшее событие удалённой задачи или поддельный запрос. Зарегистрируйте и отбросьте его. Не создавайте запись из входящей полезной нагрузки.
• Задача уже помечена выполненной: это повторная доставка. Подтвердите её и не делайте ничего другого. Повторное скачивание и повторное начисление по одной и той же задаче — классический дефект Webhook.
• Задача ожидает: пометьте её выполненной (или неудачной), сохраните ссылку на ресурс и передайте шагу скачивания.

Сделайте так, чтобы переход состояния был обусловлен текущим состоянием — то есть обновление успешно только если задача всё ещё ожидает, — чтобы две одновременные доставки не могли обе победить. В сочетании с описанным выше гибридным опросчиком ни одна задача не будет потеряна, даже если обратный вызов никогда не придёт.

Тайм-ауты, повторы и обработка сбоев

Повторы — это место, где интеграция тихо ломается. Следующие правила применимы к отправке, опросу и обратным вызовам, но безопасные границы различаются.

Сначала классифицируйте ошибки

Временные: сброс соединения, ответы 5xx, ответы об ограничении скорости, тайм-ауты чтения. Они могут успешно пройти со второй попытки.

Терминальные: ошибки валидации 4xx, неподдерживаемые параметры, сбои аутентификации и сбои уровня задачи (например, отклонённый промпт). Повторять их — значит просто повторять тот же сбой, а для отправленных задач — повторять затраты.

Если сомневаетесь, проверяйте код статуса и тело ошибки, а не повторяйте всё по умолчанию.

Задержка с джиттером и жёсткая остановка

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

def with_retries(fn, *, attempts: int = 4, base: float = 0.5, cap: float = 8.0):
    for i in range(attempts):
        try:
            return fn()
        except TransientError:
            if i == attempts - 1:
                raise
            delay = min(cap, base * (2 ** i))
            time.sleep(delay + random.uniform(0, delay / 2))

Идемпотентность на границах

• Отправка: при повторе отправки всегда отправляйте тот же ключ идемпотентности. Это единственный безопасный способ восстановиться после тайм-аута запроса, который мог создать задачу. Сгенерируйте ключ один раз, сохраните его вместе с предполагаемым запросом и переиспользуйте.
• Опрос: естественно идемпотентен — только чтение. Можно свободно повторять в пределах вашего бюджета времени.
• Обратный вызов: доставка по своей природе как минимум однократна. Дедуплицируйте по ID задачи плюс тип события и сделайте переходы состояния обработчика условными, чтобы повторная доставка была безоперационной.

Выявляйте терминальное состояние

Каждый путь задачи должен в конечном итоге сохранять ровно один результат: успех со ссылкой на ресурс или сбой с причиной и историей попыток. Никогда не оставляйте задачу в состоянии pending на неопределённый срок — ваш гибридный опросчик существует именно для того, чтобы принудительно принять решение после порога устаревания. Выявляйте это состояние стороне, потребляющей задачу, чтобы вышестоящий код мог решить переотправить, оповестить или вернуть внимание, а не ждать вечно.

Скачивание и хранение готового видео

URL в полезной нагрузке завершения является временным. Он может быть подписан, может истечь, и его хост — не то, на что ваше приложение должно полагаться при чтении. Последний шаг интеграции — переместить артефакт в хранилище, которое вы контролируете.

Получение ресурса

Скачивайте потоковым запросом, чтобы большие файлы не попадали в память как единый объект bytes, и устанавливайте тайм-ауты, отражающие размер файла, а не типичный вызов API.

def download_asset(url: str, dest_path: str) -> None:
    with requests.get(url, stream=True, timeout=(5, 300)) as r:
        r.raise_for_status()
        with open(dest_path, "wb") as f:
            for chunk in r.iter_content(chunk_size=1 << 20):
                f.write(chunk)

Если ваш бэкенд хранилища поддерживает потоковую запись, передайте ему итератор ответа напрямую, а не сохраняйте во временный локальный файл.

Проверяйте до того, как доверять

Как минимум проверьте успешный статус ответа, соответствие длины содержимого любому заявленному размеру и нетривиальный размер файла. Если полезная нагрузка завершения содержит метаданные длительности или формата, выборочно проверьте, что скачанный артефакт действительно может быть видео, а не страницей ошибки или усечённой частью файла. Повреждённый ресурс, доставленный пользователю, хуже, чем задача, которую вы пометили как неудачную и переотправили.

Перемещение в ваше собственное хранилище

Запишите объект в ваш бакет или источник CDN, с ключом, производным от вашего внутреннего ID задачи, и зафиксируйте этот ключ в строке задачи. Отдавайте пользователям со своего URL. Выгоды накапливаются:

• Временные URL провайдера никогда не попадают в ваш фронтенд, письма или журналы.
• Истечение срока больше не имеет значения; вы можете перегенерировать производные или перекодировать по своему графику.
• Хранение, контроль доступа и удаление становятся вашей политикой, а не провайдера.

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

Контроль затрат на генерацию

Генерация тарифицируется по объёму работы на запрос — обычно по длительности вывода или эквивалентным токенам, точные тарифы отображаются на панели цен платформы. Поскольку каждый вызов намного дороже типичного текстового запроса, контроль затрат должен быть в точке вызова, а не в ежемесячном обзоре.

Ограничение параллелизма

Самый эффективный единый ограничитель — ограничение числа одновременных задач генерации. Без него штормы повторов, вышедшие из-под контроля циклы или всплески пользовательских запросов могут многократно увеличить расходы быстрее, чем любой дашборд успеет это показать. Поместите семафор или пул обработчиков с ограничением перед отправкой и задайте его размер осознанно.

from threading import Semaphore

_submit_gate = Semaphore(5)  # 根据你的预算和提供方限制调整

def guarded_submit(prompt, *, idempotency_key, **params):
    with _submit_gate:
        return submit_video_job(prompt, idempotency_key=idempotency_key, **params)

Валидация перед отправкой

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

Записывайте объём задач, а не только ошибки

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

Не платите дважды за одну и ту же задачу

У режима сбоя повторной генерации есть три распространённых источника, у каждого есть защита:

• Повтор отправки при тайм-ауте без ключа идемпотентности. Переиспользуйте этот ключ.
• Повторная доставка Webhook запускает новое скачивание или новую задачу. Дедуплицируйте по ID задачи и делайте переходы условными.
• Опросчик сдаётся и переотправляет, пока исходная задача всё ещё выполняется. Проверьте состояние существующей задачи перед созданием замены и отслеживайте отношение замещения в таблице задач.

Рассматривайте каждую переотправку как решение, требующее записанной причины, а не как путь восстановления по умолчанию.

Контрольный список перед запуском

Перед запуском интеграции проверьте сквозным образом эти решения и режимы сбоя.

Получение и жизненный цикл

• Режим получения выбран по нагрузке, с гибридной подстраховкой, чтобы пропущенный обратный вызов не терял задачу.
• Строка задачи записывается после отправки, но до любых других побочных эффектов, с параметрами, ключом идемпотентности и владельцем.
• Каждая задача достигает ровно одного терминального состояния; порог устаревания принудительно принимает решение по любому всё ещё ожидающему элементу.
• Переходы состояния обусловлены текущим состоянием, поэтому повторные доставки и параллельный опрос безвредны.

Сетевое поведение

• Для каждого запроса установлены тайм-ауты соединения и чтения, включая скачивания.
• Временные и терминальные ошибки классифицируются, повторяются только временные.
• Повторы используют экспоненциальную задержку с джиттером и жёсткие ограничения на число попыток и затраченное время.
• Задержка опроса имеет как ограничение интервала, так и бюджет по настенному времени; цикл опроса выполняется в обработчике, а не в обработчике запросов.

Обратные вызовы и хранилище

• Эндпоинт обратного вызова проверяет отправителя по исходному телу, быстро отвечает и ставит работу в очередь.
• Каждый обратный вызов сверяется с вашими собственными записями задач перед любыми действиями.
• Завершённые ресурсы потоково передаются в ваше собственное хранилище, проходят проверку и отдаются только с вашего URL; шаг скачивания идемпотентен.

Затраты и эксплуатация

• Отправка ограничена параллелизмом и проверкой параметров длительности, разрешения и входных данных.
• Объём задач записывается по функции и по пользователю, поэтому расходы видны как тенденция, а не сюрприз.
• Пути повторной генерации — повторы отправки, повторные доставки Webhook, переотправки опросчиком — все обрабатываются явно.
• Есть оповещения о задачах, застрявших в ожидании сверх порога, и о постоянном достижении лимита параллелизма.

Прогоните реальную задачу по каждому пути — успех, сбой, пропущенный обратный вызов, повторный обратный вызов — и убедитесь, что терминальные состояния, записи в хранилище и счётчики выглядят корректно. Если каждый пункт выше выполняется в этом упражнении, интеграция готова к запуску с API генерации видео HiAPI.ai.