Как использовать kling-4.0/text-to-video через HiAPI API: curl, Python и рабочий пример запроса

Возможности и сценарии применения kling-4.0/text-to-video

kling-4.0/text-to-video — это генеративная видеомодель, доступная через унифицированный API HiAPI. Вы передаёте ей текстовый промпт, а она возвращает видеоклип — движение, работа камеры и композиция сцены полностью определяются вашим описанием. Это эндпоинт преобразования текста в видео: исходного изображения для анимации нет, поэтому модель полностью самостоятельно решает, как кадрировать сцену и как выглядит субъект.

Сценарии применения:

• Вам нужно генерировать короткие кинематографичные клипы по сценарию или шаблону промпта — b-roll, трейлеры, концепт-борды, видео для соцсетей.
• Вы хотите, чтобы синхронный звук генерировался вместе с видео, а не получать беззвучный файл, которому нужна музыка на постпродакшене.
• Вы строите конвейер, где генерация видео — лишь один из множества шагов (написание сценария, подготовка материалов, публикация), поэтому вам нужен HTTP API, а не веб-интерфейс.

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

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

Предварительные требования и аутентификация в HiAPI

Прежде чем отправить первый запрос, убедитесь, что у вас есть:

• Учётная запись HiAPI и API-ключ проекта. Относитесь к нему как к секрету — никогда не коммитьте в репозиторий и не помещайте в клиентский код.
• Выбранный HTTP-клиент. В примерах ниже используется curl и requests из Python 3.
• Идентификатор модели, точно совпадающий с каталогом: kling-4.0/text-to-video. Опечатка в имени модели — самая частая причина неудачи первого вызова.

Учётные данные передаются в каждом запросе (включая запросы опроса) в виде HTTP-заголовка. Стандартный подход в HiAPI — bearer-токен:

export HIAPI_API_KEY="your-api-key-here"

Затем в запросе:

Authorization: Bearer $HIAPI_API_KEY
Content-Type: application/json

Храните базовый URL и ключ в переменных окружения или менеджере секретов, а не жёстко в коде. Для локальной разработки достаточно файла .env и чтения через os.environ в Python. Для развёрнутого сервиса внедряйте секреты во время выполнения.

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

Рабочий запрос curl

Минимальный вызов создаёт задачу и возвращает идентификатор задачи. Структура выглядит так:

curl -X POST "$HIAPI_BASE_URL/video/generations" \
  -H "Authorization: Bearer $HIAPI_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "kling-4.0/text-to-video",
    "prompt": "A slow aerial push over a rain-soaked Tokyo alley at night, neon signs reflecting in puddles, steam rising from vents, shallow depth of field, cinematic color grading",
    "generate_audio": true
  }'

Разбор тела запроса по полям:

• model — идентификатор модели. Обязательное поле.
• prompt — описание вашей сцены. Здесь сосредоточена большая часть вашего бюджета качества; см. раздел о промптах.
• generate_audio — запросить генерацию синхронного звука вместе с видео. Если нужна только визуальная дорожка, опустите или установите false.

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

Два практических совета. Во-первых, когда промпт становится длинным, сохраните JSON в файл и используйте -d @payload.json — обработка кавычек в shell для многострочных промптов — частый источник ошибок. Во-вторых, захватывайте тело ответа и HTTP-код статуса отдельно (curl -s -w "\n%{http_code}"), чтобы различать отклонённые запросы и сетевые сбои.

Перевод запроса на Python

Тот же запрос на Python, написанный так, чтобы структуру можно было сразу перенести в реальный скрипт:

import os
import requests

BASE_URL = os.environ["HIAPI_BASE_URL"].rstrip("/")
API_KEY = os.environ["HIAPI_API_KEY"]

HEADERS = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json",
}

def create_video_job(prompt: str, generate_audio: bool = True) -> dict:
    payload = {
        "model": "kling-4.0/text-to-video",
        "prompt": prompt,
        "generate_audio": generate_audio,
    }
    resp = requests.post(
        f"{BASE_URL}/video/generations",
        headers=HEADERS,
        json=payload,
        timeout=30,
    )
    resp.raise_for_status()
    return resp.json()

if __name__ == "__main__":
    job = create_video_job(
        "A slow aerial push over a rain-soaked Tokyo alley at night, "
        "neon signs reflecting in puddles, steam rising from vents, "
        "shallow depth of field, cinematic color grading"
    )
    print(job)

Почему написано именно так:

• requests.post(..., json=payload) одновременно обрабатывает сериализацию и тип содержимого, избавляя от ручного json.dumps.
• raise_for_status() превращает 4xx/5xx в перехватываемое исключение, а не молча разбирает тело ошибочного ответа как задачу.
• timeout обязателен. Без него зависшее соединение заблокирует worker на неопределённый срок.
• Конфигурация находится в модульных константах, читаемых из окружения, поэтому функцию можно тестировать и переиспользовать.

Эта функция только создаёт задачу. Она сразу возвращает дескриптор задачи — именно это вам нужно в веб-обработчике или worker'е очереди. В последующих разделах рассказывается, как дождаться результата.

Написание кинематографичных промптов, которые выдерживают проверку

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

Рабочий шаблон:

[субъект и сцена], [точка съёмки и движение],
[движение внутри кадра], [освещение], [атмосфера или цветокоррекция], [стилевой референс]

Пример применения:

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

Роль каждой части:

• Субъект и сцена — конкретные существительные лучше абстракций. «Одинокий велосипедист в жёлтом дождевике» даёт модели то, что можно отрисовать; «свобода» — нет.
• Камера — точка съёмки (низкая точка, аэросъёмка, на уровне глаз) и движение (следящая съёмка, медленный наезд, статичная) — самый мощный рычаг для ощущения продакшн-качества.
• Движение внутри кадра — плывущий туман, колышущаяся ткань, рябь на воде. Видео требует движения; укажите его явно.
• Освещение и атмосфера — время, цветовая температура, эмоциональный тон. Держите их согласованными друг с другом.
• Стиль — язык объектива или формата (анаморфная оптика, малая глубина резкости, кинематографичная цветокоррекция) надёжнее формирует текстуру, чем указание конкретного режиссёра.

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

Обработка асинхронных задач и опрос

Генерация видео не мгновенна, поэтому схема — сначала создать, затем опрашивать. Часть создания у вас уже есть; вот цикл:

import time

def wait_for_job(job_id: str, timeout: int = 600) -> dict:
    deadline = time.monotonic() + timeout
    delay = 2.0
    while time.monotonic() < deadline:
        resp = requests.get(
            f"{BASE_URL}/video/generations/{job_id}",
            headers=HEADERS,
            timeout=30,
        )
        resp.raise_for_status()
        job = resp.json()
        status = job.get("status")

        if status == "succeeded":
            return job
        if status in ("failed", "cancelled"):
            raise RuntimeError(f"Job {job_id} ended with status {status}: {job}")

        time.sleep(delay)
        delay = min(delay * 1.5, 15.0)  # back off, cap at 15s

    raise TimeoutError(f"Job {job_id} did not finish in {timeout}s")

Важные проектные решения для продакшена:

Опрос с задержкой, а не в тесном цикле

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

Всегда устанавливайте дедлайн

Каждому циклу опроса нужны максимальное астрономическое время и явная ошибка при его превышении. Иначе зависшая задача будет молча занимать worker вечно.

Считайте терминальные состояния окончательными

Различайте состояния в процессе (queued, processing) и терминальные (succeeded, failed, cancelled). Только при успехе возвращайте результат; в остальных случаях выбрасывайте исключение или переходите к повторной попытке, а не продолжайте цикл.

Отделяйте опрос от обработчика запросов

В веб-сервере не давайте HTTP-ответу блокироваться в ожидании видео. Создайте задачу, верните id клиенту и опрашивайте из фонового worker'а или потребителя очереди. Сохраняйте id задачи и её последнее известное состояние, чтобы после перезапуска не потерять отслеживание выполняющейся работы.

Скачивание и проверка результатов

Когда задача достигает succeeded, ответ содержит ссылку на сгенерированный медиафайл — обычно URL готового файла. Скачивайте его своевременно и считайте этот URL временным: получайте, сохраняйте и записывайте собственное постоянное местоположение, а не передавайте URL провайдера напрямую конечному пользователю.

import pathlib

def download_output(job: dict, dest: pathlib.Path) -> pathlib.Path:
    url = job["output"]["video_url"]
    with requests.get(url, stream=True, timeout=120) as r:
        r.raise_for_status()
        dest.parent.mkdir(parents=True, exist_ok=True)
        with open(dest, "wb") as f:
            for chunk in r.iter_content(chunk_size=1 << 20):
                f.write(chunk)
    return dest

Проверки, которые стоит сделать, прежде чем передавать файл кому-либо:

• Убедитесь, что передача завершена. Стримьте по частям, сравните записанный размер с Content-Length (если он есть), громко падайте при недокачке.
• Подтвердите наличие звуковой дорожки. Если вы запросили generate_audio, в файле должна быть аудиодорожка. Быстрая проверка через ffprobe — самый дешёвый способ:

ffprobe -v error -show_entries stream=codec_type -of csv=p=0 output.mp4

Вы должны увидеть строки video и audio. Если аудио отсутствует, перепроверьте, был ли флаг установлен в вызове создания — задним числом его добавить нельзя.

• Сделайте базовую проверку длительности и разрешения на соответствие ожиданиям, чтобы до публикации выявить обрезанный или неверный по спецификациям рендер.
• Храните с детерминированным ключом (id задачи плюс хеш), сначала записывайте во временный путь, переименовывайте после успеха и сохраняйте id задачи вместе с активом для отслеживания.

Если скачивание оборвалось на середине, повторите с задержкой; если URL устарел, вам нужно повторно получить запись задачи для новой ссылки.

Инкапсуляция в переиспользуемый клиент

Описанные части естественно складываются в класс. Собрав их вместе, вы получите нечто импортируемое между проектами без копирования:

class KlingTextToVideo:
    def __init__(self, base_url: str, api_key: str):
        self.base_url = base_url.rstrip("/")
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
        }

    def generate(self, prompt: str, generate_audio: bool = True,
                 timeout: int = 600) -> dict:
        job = create_video_job(prompt, generate_audio)  # create half
        return wait_for_job(job["id"], timeout=timeout)  # poll half

По мере взросления клиента стоит добавить несколько вещей:

• Структурированное логирование с ключом по id задачи — переходы состояний, число опросов, прошедшее время и ошибки. Когда рендер ведёт себя странно, id задачи — это ключ корреляции между логами.
• Ограниченные повторные попытки только для временных ошибок (сеть, 5xx). Никогда не повторяйте слепо отклонённые запросы; некорректное тело запроса будет падать одинаково каждый раз, лишь тратя квоту.
• Идемпотентность на вашем уровне. Если ваш сервис может отправить один и тот же промпт дважды, дедуплицируйте перед вызовом создания, а не платите за две задачи.
• Шаблоны промптов как данные, а не код. Храните кинематографичные шаблоны в конфигурационном файле, чтобы итерация промптов не требовала развёртывания.
• Учёт стоимости по секундам вывода. Биллинг основан на сгенерированном объёме вывода, поэтому отслеживайте длительность, созданную каждой задачей, чтобы относить расходы на функцию или пользователя.

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