Video Generation Foundation Models: 3D DiT Architectures, Flow Matching, and Spatiotemporal Attention

Step 1: HunyuanVideo / 3D DiT Pipeline Wrapper

from __future__ import annotations

from typing import Optional
import torch
from diffusers import HunyuanVideoPipeline, HunyuanVideoTransformer3DModel
from diffusers.utils import export_to_video


class VideoGenerationEngine:
    """Production wrapper for open-source 13B 3D DiT models."""

    def __init__(
        self,
        model_id: str = "tencent/HunyuanVideo",
        device: str = "cuda",
        dtype: torch.dtype = torch.bfloat16,
        enable_cpu_offload: bool = True,
    ):
        self.device = device
        self.dtype = dtype

        self.pipeline = HunyuanVideoPipeline.from_pretrained(
            model_id,
            torch_dtype=dtype,
        ).to(device)

        # 3D VAE tiling & slicing to conserve GPU VRAM during 720p/1080p decoding
        self.pipeline.vae.enable_tiling()
        self.pipeline.vae.enable_slicing()

        if enable_cpu_offload:
            self.pipeline.enable_model_cpu_offload()

    def generate_text_to_video(
        self,
        prompt: str,
        negative_prompt: str = "blurry, low quality, distorted, jitter, artifact",
        height: int = 720,
        width: int = 1280,
        num_frames: int = 49,  # ~5s at 10 fps (causal 3D VAE 4x temporal compression)
        num_inference_steps: int = 40,
        guidance_scale: float = 6.0,
        output_path: str = "output_video.mp4",
        seed: Optional[int] = None,
    ) -> str:
        generator = torch.Generator(device=self.device).manual_seed(seed) if seed is not None else None

        video_frames = self.pipeline(
            prompt=prompt,
            negative_prompt=negative_prompt,
            height=height,
            width=width,
            num_frames=num_frames,
            num_inference_steps=num_inference_steps,
            guidance_scale=guidance_scale,
            generator=generator,
        ).frames[0]

        export_to_video(video_frames, output_path, fps=24)
        return output_path

Step 2: Causal Video Dataset Loader with Decord

from __future__ import annotations

from pathlib import Path
from typing import Dict, Any
import decord
import torch
import torch.nn.functional as F
from torch.utils.data import Dataset

decord.bridge.set_bridge("torch")


class SpatiotemporalVideoDataset(Dataset):
    """Loads and preprocesses video clips into [T, C, H, W] tensors."""

    def __init__(
        self,
        data_dir: str,
        num_frames: int = 49,
        height: int = 480,
        width: int = 720,
    ):
        self.data_dir = Path(data_dir)
        self.video_files = sorted(list((self.data_dir / "videos").glob("*.mp4")))
        self.num_frames = num_frames
        self.height = height
        self.width = width

    def __len__(self) -> int:
        return len(self.video_files)

    def __getitem__(self, idx: int) -> Dict[str, Any]:
        video_path = self.video_files[idx]
        caption_path = self.data_dir / "captions" / f"{video_path.stem}.txt"

        caption = caption_path.read_text(encoding="utf-8").strip() if caption_path.exists() else ""

        vr = decord.VideoReader(str(video_path))
        total_frames = len(vr)

        # Uniform temporal sampling
        frame_indices = torch.linspace(0, total_frames - 1, self.num_frames).long()
        raw_frames = vr.get_batch(frame_indices.tolist())  # [T, H, W, C] in uint8

        # Convert to [T, C, H, W] in [-1.0, 1.0]
        frames = raw_frames.permute(0, 3, 1, 2).float() / 127.5 - 1.0
        frames = F.interpolate(frames, size=(self.height, self.width), mode="bilinear", align_corners=False)

        return {"frames": frames, "caption": caption}

Step 3: LoRA Fine-Tuning Pipeline with Flow Matching

from __future__ import annotations

import torch
import torch.nn.functional as F
from peft import LoraConfig, get_peft_model
from torch.utils.data import DataLoader
from diffusers import CogVideoXTransformer3DModel
from video_dataset import SpatiotemporalVideoDataset


def train_video_lora(
    model_id: str = "THUDM/CogVideoX-5b",
    data_dir: str = "datasets/action_dataset",
    output_dir: str = "checkpoints/video_lora",
    epochs: int = 50,
    lr: float = 1e-4,
    batch_size: int = 1,
    gradient_accumulation_steps: int = 4,
):
    device = torch.device("cuda")

    # 1. Initialize Transformer with LoRA
    transformer = CogVideoXTransformer3DModel.from_pretrained(
        model_id, subfolder="transformer", torch_dtype=torch.bfloat16
    ).to(device)

    lora_config = LoraConfig(
        r=32,
        lora_alpha=32,
        target_modules=["to_q", "to_k", "to_v", "to_out.0"],
        lora_dropout=0.05,
    )
    transformer = get_peft_model(transformer, lora_config)
    transformer.enable_gradient_checkpointing()

    dataset = SpatiotemporalVideoDataset(data_dir=data_dir)
    dataloader = DataLoader(dataset, batch_size=batch_size, shuffle=True, num_workers=4)

    optimizer = torch.optim.AdamW(
        [p for p in transformer.parameters() if p.requires_grad], lr=lr, weight_decay=1e-2
    )

    transformer.train()
    step = 0

    for epoch in range(epochs):
        for batch in dataloader:
            # frames: [B, T, C, H, W] -> simulate pre-encoded latents z_0: [B, T', C', H', W']
            latents = batch["frames"].to(device, dtype=torch.bfloat16)

            # Sample continuous flow matching timestep t in [0, 1]
            noise = torch.randn_like(latents)
            t = torch.rand(latents.shape[0], device=device, dtype=torch.bfloat16)

            # Linear interpolation trajectory
            t_expand = t.view(-1, 1, 1, 1, 1)
            noisy_latents = (1.0 - t_expand) * latents + t_expand * noise
            velocity_target = noise - latents

            # Dummy text embeddings (simulate T5 encoding)
            text_context = torch.zeros((latents.shape[0], 77, 4096), device=device, dtype=torch.bfloat16)

            # Predict velocity
            pred_velocity = transformer(
                hidden_states=noisy_latents,
                timestep=t * 1000.0,
                encoder_hidden_states=text_context,
            ).sample

            loss = F.mse_loss(pred_velocity, velocity_target) / gradient_accumulation_steps
            loss.backward()

            if (step + 1) % gradient_accumulation_steps == 0:
                torch.nn.utils.clip_grad_norm_(transformer.parameters(), 1.0)
                optimizer.step()
                optimizer.zero_grad()

            step += 1

    transformer.save_pretrained(output_dir)

Empirical Benchmark Evaluation

We evaluated open-weight and closed-weight video foundation models across the VBench text-to-video benchmark:

Troubleshooting Common Synthesis Artifacts

1. Frame-to-Frame Temporal Jitter & Strobing

• Symptom: Rapid micro-vibrations across static background geometry.
• Remedy: Increase sampling steps from $30\to 50$ and ensure the causal 3D VAE uses temporal sliding-window slicing during decoding.

2. Physical Inconsistency in Object Interpenetration

• Symptom: Solid objects pass through one another or liquid splashes defy gravity.
• Remedy: Condition the model with explicit depth maps or integrate trajectory-conditioned control signals.

3. VRAM OOM During 1080p Video Generation

• Symptom: CUDA out-of-memory error during VAE upsampling.
• Remedy: Invoke pipeline.vae.enable_tiling() and pipeline.vae.enable_slicing() to stream 2D/3D convolution tiles sequentially.