AI Video Generation System Development

AI Video Generation Systems Development

AI video generation — creating video content from text or image prompts. 2024–2025: Sora, Kling, Runway Gen-3, Pika, Luma show 5–20-second professional quality clips. Applied in advertising, production, game development, education.

Platform Comparison

Kling API Integration

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 or 10 seconds
        aspect_ratio: str = "16:9",  # 16:9, 9:16, 1:1
        mode: str = "std",  # std (faster) or pro (quality)
        cfg_scale: float = 0.5
    ) -> str:
        """Create generation task, return 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:
        """Wait for completion, return video 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 Integration

import runwayml

client = runwayml.RunwayML(api_key=RUNWAY_API_KEY)

async def generate_runway_video(
    prompt: str,
    image_url: str = None,  # For 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 sec at 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

Applications by Niche

Timeline: integrating one API (Kling/Runway) — 3–5 days. Platform with multiple providers, queue, and storage — 3–4 weeks.