AI-генерация видео: интеграция API и self-hosted

Разработка систем AI-генерации видео

Студия запросила автоматическое создание 10-секундных рекламных роликов из фото продуктов. Раньше дизайнеры монтировали вручную — 2 ролика в день. Мы интегрировали API Kling 1.5 и Runway Gen-3, добавили очередь с приоритетами и постобработку. Производительность выросла до 50 роликов в день. Это типичный кейс: AI-генерация видео из текстовых или изображений-промптов уже не игрушка, а рабочий инструмент для маркетинга, продакшна, геймдева и образования.

Какие проблемы решаем

• Нестабильное качество: разные промпты дают разный результат. Наша интеграция с Kling 1.5 позволяет задавать negative prompt, CFG scale и режим pro для повышения управляемости. Для Runway подбираем соотношение 1280:768 и длительность 10 сек — этого достаточно для коротких рекламных роликов. Kling 1.5 лучше Runway по управляемости — negative prompt и cfg_scale дают на 30% больше контроля.
• Задержки генерации: при облачных API время ожидания достигает 5 минут. Мы строим асинхронные пайплайны с WebSocket-уведомлениями и очередями на Redis. В self-hosted варианте (CogVideoX на A100) latency снижается до 2 минут на 6-секундный клип.
• Стоимость на масштабе: каждый запрос к Kling стоит ~$0.10. При 5 000 роликов в месяц это $500. Self-hosted CogVideoX лучше облачных API в 2-3 раза по стоимости при больших объёмах — если нагрузка превышает 10 000 запросов/мес, рекомендуем переходить на собственные GPU.

Как интегрировать Kling API?

Для одного маркетплейса мы реализовали text-to-video и image-to-video через Kling. Код Python с httpx.AsyncClient и asyncio обрабатывает до 100 параллельных запросов. Пример класса KlingVideoGenerator ниже.

import httpx import asyncio import json class KlingVideoGenerator: def __init__(self, api_key: str): self.api_key = api_key self.base_url = "https://api.klingai.com/v1" async def text_to_video( self, prompt: str, negative_prompt: str = "", duration: int = 5, # 5 или 10 секунд aspect_ratio: str = "16:9", # 16:9, 9:16, 1:1 mode: str = "std", # std (быстрее) или pro (качество) cfg_scale: float = 0.5 ) -> str: """Создаём задачу генерации, возвращаем task_id""" async with httpx.AsyncClient() as client: resp = await client.post( f"{self.base_url}/videos/text2video", headers={"Authorization": f"Bearer {self.api_key}"}, json={ "prompt": prompt, "negative_prompt": negative_prompt, "cfg_scale": cfg_scale, "mode": mode, "duration": str(duration), "aspect_ratio": aspect_ratio } ) return resp.json()["data"]["task_id"] async def image_to_video( self, image_url: str, prompt: str = "", duration: int = 5, motion_intensity: float = 0.5 ) -> str: async with httpx.AsyncClient() as client: resp = await client.post( f"{self.base_url}/videos/image2video", headers={"Authorization": f"Bearer {self.api_key}"}, json={ "image_url": image_url, "prompt": prompt, "duration": str(duration), "cfg_scale": motion_intensity } ) return resp.json()["data"]["task_id"] async def wait_for_result(self, task_id: str, timeout: int = 300) -> str: """Ждём завершения, возвращаем URL видео""" async with httpx.AsyncClient() as client: for _ in range(timeout // 5): await asyncio.sleep(5) resp = await client.get( f"{self.base_url}/videos/text2video/{task_id}", headers={"Authorization": f"Bearer {self.api_key}"} ) data = resp.json()["data"] if data["task_status"] == "succeed": return data["task_result"]["videos"][0]["url"] elif data["task_status"] == "failed": raise RuntimeError(f"Generation failed: {data.get('task_status_msg')}") raise TimeoutError(f"Video generation timeout after {timeout}s") 

Аналогично интегрируется Runway Gen-3. Разница — в формате ответа и поддержке image-to-video с prompt_image.

Почему стоит выбрать self-hosted CogVideoX?

Отметим: когда объём генераций превышает 10 000 минут в месяц, облачные API становятся дороже GPU-сервера. CogVideoX-5b на A100 (80 GB) выдаёт 49 кадров за 2 минуты — это ~6 секунд видео при 8 FPS. Разрешение 720p, но качество приближается к Kling. Вы полностью контролируете пайплайн, можете дообучать на своих данных (LoRA) и не платить за каждый запрос.

from diffusers import CogVideoXPipeline import torch pipe = CogVideoXPipeline.from_pretrained( "THUDM/CogVideoX-5b", torch_dtype=torch.bfloat16 ) pipe.enable_model_cpu_offload() pipe.vae.enable_tiling() def generate_video_local( prompt: str, num_frames: int = 49, # ~6 сек при 8 fps guidance_scale: float = 6.0 ) -> str: video_frames = pipe( prompt=prompt, num_videos_per_prompt=1, num_inference_steps=50, num_frames=num_frames, guidance_scale=guidance_scale, generator=torch.Generator("cpu").manual_seed(42) ).frames[0] output_path = "/tmp/output_video.mp4" from diffusers.utils import export_to_video export_to_video(video_frames, output_path, fps=8) return output_path 

Как строится процесс разработки?

1. Аналитика и выбор модели — тестируем Kling, Runway, Sora (если доступен), CogVideoX на ваших данных. Собираем метрики latency p99 и стоимость.
2. Проектирование архитектуры — решаем, какой паттерн: только API (быстро) или self-hosted (дёшево на масштабе). Документируем схему.
3. Реализация интеграции — пишем асинхронные клиенты, обработчики очередей, S3-хранилище. Для self-hosted — развёртывание с Triton Inference Server и ONNX Runtime для ускорения.
4. Тестирование — A/B сравниваем с ручным монтажом. Замеряем FID/CLIP и субъективное качество.
5. Деплой и мониторинг — настраиваем CI/CD, логирование в Loki, алерты в Telegram.

Сроки ориентировочно

• Интеграция одного API (Kling/Runway): от 3 до 5 дней.
• Платформа с несколькими провайдерами, очередью и хранилищем: от 3 до 4 недель.
• Self-hosted CogVideoX + оптимизация: от 4 до 6 недель.

Что входит в работу

• Документация по API и архитектуре
• Доступ к репозиторию с кодом
• Обучение команды (2-3 сессии)
• Гарантия 3 месяца на интеграцию
• Поддержка при релизе (первые 2 недели)

Сравнение платформ

Применения по нишам

Если вам нужна AI-генерация видео, свяжитесь с нами — за 2 дня оценим проект и предложим оптимальное решение. Закажите пилотный проект и убедитесь в эффективности.