Реализация Text-to-Video генерации

Разработка систем Text-to-Video генерации

Text-to-video создаёт видео из текстового описания без исходного изображения. Задача сложнее image generation: нужна временная согласованность между кадрами, понимание движения, физики. 2024–2025: коммерческие API достигли уровня, пригодного для рекламного контента.

Мультипровайдерный сервис

from abc import ABC, abstractmethod
import asyncio
import httpx
from enum import Enum

class VideoProvider(Enum):
    KLING = "kling"
    RUNWAY = "runway"
    LUMA = "luma"
    COGVIDEOX = "cogvideox"

class BaseVideoGenerator(ABC):
    @abstractmethod
    async def generate(self, prompt: str, **kwargs) -> bytes: ...

class MultiProviderVideoService:
    def __init__(self, providers: dict[VideoProvider, BaseVideoGenerator]):
        self.providers = providers

    async def generate(
        self,
        prompt: str,
        provider: VideoProvider = VideoProvider.KLING,
        fallback: VideoProvider = VideoProvider.RUNWAY,
        **kwargs
    ) -> bytes:
        try:
            return await self.providers[provider].generate(prompt, **kwargs)
        except Exception as e:
            if fallback and fallback != provider:
                return await self.providers[fallback].generate(prompt, **kwargs)
            raise

    async def generate_batch(
        self,
        prompts: list[str],
        provider: VideoProvider = VideoProvider.KLING,
        max_concurrent: int = 5
    ) -> list[bytes]:
        semaphore = asyncio.Semaphore(max_concurrent)

        async def generate_one(prompt):
            async with semaphore:
                return await self.generate(prompt, provider=provider)

        return await asyncio.gather(*[generate_one(p) for p in prompts])

Оптимизация промптов для t2v

from openai import AsyncOpenAI

client = AsyncOpenAI()

async def enhance_video_prompt(
    user_prompt: str,
    style: str = "cinematic",
    duration: int = 5
) -> str:
    """Расширяем короткий промпт до полного для видеогенерации"""
    response = await client.chat.completions.create(
        model="gpt-4o",
        messages=[{
            "role": "system",
            "content": f"""Расширь промпт для AI-видеогенерации.
            Добавь:
            - Движение камеры (slow pan, dolly in, aerial shot и т.д.)
            - Атмосферу и освещение
            - Динамику сцены (что движется, как)
            - Стиль: {style}
            Длина видео: {duration} секунд.
            Только промпт, на английском языке."""
        }, {
            "role": "user",
            "content": user_prompt
        }]
    )
    return response.choices[0].message.content.strip()

CogVideoX — self-hosted SOTA

from diffusers import CogVideoXPipeline, CogVideoXDDIMScheduler
import torch

class CogVideoXGenerator(BaseVideoGenerator):
    def __init__(self):
        self.pipe = CogVideoXPipeline.from_pretrained(
            "THUDM/CogVideoX-5b",
            torch_dtype=torch.bfloat16
        )
        self.pipe.scheduler = CogVideoXDDIMScheduler.from_config(
            self.pipe.scheduler.config, timestep_spacing="trailing"
        )
        self.pipe.enable_model_cpu_offload()
        self.pipe.vae.enable_tiling()

    async def generate(self, prompt: str, num_frames: int = 49, **kwargs) -> bytes:
        from diffusers.utils import export_to_video
        import tempfile

        video_frames = self.pipe(
            prompt=prompt,
            num_videos_per_prompt=1,
            num_inference_steps=50,
            num_frames=num_frames,
            guidance_scale=6.0,
            generator=torch.Generator("cpu").manual_seed(42)
        ).frames[0]

        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f:
            export_to_video(video_frames, f.name, fps=8)
            return open(f.name, "rb").read()

Стоимость и производительность

При объёме > 1000 видео/месяц CogVideoX self-hosted становится выгоднее API. Сроки: мультипровайдерный сервис с очередью — 1 неделя.