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

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

AI-генерация видео — создание видеоконтента из текстовых или изображений-промптов. 2024–2025 год: Sora, Kling, Runway Gen-3, Pika, Luma показывают 5–20-секундные клипы профессионального качества. Применяется в рекламе, продакшне, геймдеве, образовании.

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

Kling API интеграция

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 интеграция

import runwayml

client = runwayml.RunwayML(api_key=RUNWAY_API_KEY)

async def generate_runway_video(
    prompt: str,
    image_url: str = None,  # Для image-to-video
    duration: int = 10,
    ratio: str = "1280:768"
) -> str:
    if image_url:
        task = client.image_to_video.create(
            model="gen3a_turbo",
            prompt_image=image_url,
            prompt_text=prompt,
            duration=duration,
            ratio=ratio
        )
    else:
        task = client.text_to_video.create(
            model="gen3a_turbo",
            prompt_text=prompt,
            duration=duration,
            ratio=ratio
        )

    task_id = task.id
    while True:
        await asyncio.sleep(5)
        task = client.tasks.retrieve(task_id)
        if task.status == "SUCCEEDED":
            return task.output[0]
        elif task.status == "FAILED":
            raise RuntimeError(task.failure)

Self-hosted CogVideoX

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

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

Сроки: интеграция одного API (Kling/Runway) — 3–5 дней. Платформа с несколькими провайдерами, очередью и хранилищем — 3–4 недели.