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 дні оцінимо проєкт і запропонуємо оптимальне рішення. Замовте пілотний проєкт і переконайтеся в ефективності.