interactive-video-generation

name	interactive-video-generation
title	Autoregressive Adversarial Post-Training for Real-Time Interactive Video Generation
version	0.0.2
engine	skillxiv-v0.0.2-claude-opus-4.6
license	MIT
url	https://arxiv.org/abs/2506.09350
keywords	["video generation","adversarial training","real-time inference","interactive"]
description	Convert pre-trained latent video diffusion into real-time autoregressive generators using adversarial post-training, achieving 24fps streaming on single H100.

Real-Time Interactive Video Generation

Core Concept

This work transforms pre-trained latent video diffusion models into real-time interactive generators via adversarial post-training. Rather than iterative refinement, the model generates complete frames autoregressively (1 NFE per frame), enabling genuine real-time streaming while accepting user controls. Adversarial training proves more efficient than diffusion for autoregressive generation.

Architecture Overview

• Latent autoregressive generation: 1 neural function evaluation per frame
• Adversarial training framework: Discriminator ensures quality via adversarial loss
• Student forcing: Mitigates error accumulation in long sequences
• Interactive control: Accept user inputs during generation
• Full KV caching: Computational efficiency for long sequences
• Real-time performance: 24fps at 736x416, 1280x720 on 8xH100

Implementation

Step 1: Design Latent Frame Generator

Create autoregressive decoder in latent space:

class LatentFrameGenerator(torch.nn.Module):
    def __init__(self, latent_dim: int = 8,
                 context_dim: int = 768,
                 num_layers: int = 12):
        super().__init__()
        self.latent_dim = latent_dim

        # Latent encoder/decoder (from pretrained VAE)
        self.vae = LatentVAE.from_pretrained()

        # Autoregressive transformer for latent generation
        self.transformer = torch.nn.TransformerDecoder(
            torch.nn.TransformerDecoderLayer(
                d_model=context_dim,
                nhead=8,
                dim_feedforward=2048,
                batch_first=True
            ),
            num_layers=num_layers
        )

        # Generate next frame latents
        self.latent_head = torch.nn.Linear(
            context_dim,
            latent_dim
        )

        # Context encoder (e.g., text to image prompt)
        self.context_encoder = torch.nn.Sequential(
            torch.nn.Linear(768, context_dim),
            torch.nn.ReLU()
        )

    def encode_frames_to_latents(self, frames: torch.Tensor
                                ) -> torch.Tensor:
        """Convert RGB frames to latent vectors."""

        # Use VAE encoder
        with torch.no_grad():
            latents = self.vae.encode(frames)

        return latents  # (batch, num_frames, latent_dim)

    def generate_frame_latent() -> :
        

        
        context_features = .context_encoder(context)

        
         kv_cache   :
            
            decoder_input = prev_latents[-:].unsqueeze()
            hidden = .transformer(
                decoder_input,
                memory=context_features,
                memory_key_padding_mask=
            )
            
            kv_cache = hidden
        :
            
            hidden = .transformer(
                prev_latents,
                memory=context_features
            )

        
        next_latent = .latent_head(hidden[-])

         next_latent, kv_cache

     () -> torch.Tensor:
        

         torch.no_grad():
            frame = .vae.decode(latent)

         frame

Step 2: Implement Adversarial Training

Train discriminator to ensure autoregressive quality:

class VideoDiscriminator(torch.nn.Module):
    def __init__(self, num_frames: int = 8):
        super().__init__()
        self.num_frames = num_frames

        # 3D CNN for spatiotemporal discrimination
        self.spatial_temporal_blocks = torch.nn.Sequential(
            torch.nn.Conv3d(3, 64, kernel_size=(3, 4, 4),
                           stride=(1, 2, 2), padding=(1, 1, 1)),
            torch.nn.LeakyReLU(0.2),
            torch.nn.Conv3d(64, 128, kernel_size=(3, 4, 4),
                           stride=(1, 2, 2), padding=(1, 1, 1)),
            torch.nn.LeakyReLU(0.2),
            torch.nn.Conv3d(128, 256, kernel_size=(3, 4, 4),
                           stride=(1, 2, 2), padding=(1, 1, 1)),
            torch.nn.LeakyReLU(0.2)
        )

        # Classification head
        self.classifier = torch.nn.Sequential(
            torch.nn.AdaptiveAvgPool3d(),
            torch.nn.Flatten(),
            torch.nn.Linear(, ),
            torch.nn.LeakyReLU(),
            torch.nn.Linear(, )
        )

     () -> torch.Tensor:
        

        
        video = video.permute(, , , , )

        features = .spatial_temporal_blocks(video)
        score = .classifier(features)

         score

Step 3: Adversarial Post-Training Loop

Train generator and discriminator with student forcing:

class AdversarialPostTrainer:
    def __init__(self, generator: LatentFrameGenerator,
                 discriminator: VideoDiscriminator):
        self.generator = generator
        self.discriminator = discriminator

        self.g_optimizer = torch.optim.Adam(
            generator.parameters(),
            lr=1e-4,
            betas=(0.5, 0.999)
        )
        self.d_optimizer = torch.optim.Adam(
            discriminator.parameters(),
            lr=1e-4,
            betas=(0.5, 0.999)
        )

    def student_forcing_rollout(self, initial_latent,
                               context, num_frames,
                               teacher_forcing_prob=0.5):
        """Rollout with scheduled teacher forcing."""

        generated_latents = [initial_latent]

        for frame_idx in range(1, num_frames):
            # With probability p, use teacher (real) latent
            if torch.rand(1).item() < teacher_forcing_prob:
                # During training, mix real and generated
                prev_latent = initial_latent  # Would use real
            else:
                prev_latent = generated_latents[-1]

            # Generate next frame
            next_latent, _ = self.generator.generate_frame_latent(
                torch.stack(generated_latents),
                context
            )

            generated_latents.append(next_latent)

        return torch.stack(generated_latents)

    def () -> :
        

        batch_size = real_video.shape[]
        num_frames = real_video.shape[]

        
        real_latents = .generator.encode_frames_to_latents(
            real_video
        )

        
        fake_latents = .student_forcing_rollout(
            real_latents[:, :],
            context,
            num_frames,
            teacher_forcing_prob=
        )

        
        fake_frames = torch.stack([
            .generator.decode_latent_to_frame(lat)
             lat  fake_latents
        ], dim=)

        
        real_scores = .discriminator(real_video)
        fake_scores = .discriminator(fake_frames.detach())

        d_loss_real = torch.nn.functional.softplus(-real_scores).mean()
        d_loss_fake = torch.nn.functional.softplus(fake_scores).mean()
        d_loss = d_loss_real + d_loss_fake

        .d_optimizer.zero_grad()
        d_loss.backward()
        .d_optimizer.step()

        
        fake_scores = .discriminator(fake_frames)
        g_loss_adv = torch.nn.functional.softplus(
            -fake_scores
        ).mean()

        
        g_loss_recon = torch.nn.functional.mse_loss(
            fake_frames,
            real_video
        )

        g_loss = g_loss_adv +  * g_loss_recon

        .g_optimizer.zero_grad()
        g_loss.backward()
        .g_optimizer.step()

         {
            : d_loss.item(),
            : g_loss.item(),
            : g_loss_adv.item(),
            : g_loss_recon.item()
        }

Step 4: Real-Time Streaming Inference

Generate and stream video with user controls:

class RealtimeVideoStreamer:
    def __init__(self, generator: LatentFrameGenerator,
                 target_fps: int = 24):
        self.generator = generator
        self.target_fps = target_fps
        self.frame_time = 1.0 / target_fps

    def stream_video(self, prompt: str,
                    user_control_stream = None,
                    output_path: str = None,
                    duration_seconds: float = 10.0):
        """Stream video generation in real-time."""

        import time
        import cv2

        total_frames = int(self.target_fps * duration_seconds)

        # Initialize video writer
        if output_path:
            fourcc = cv2.VideoWriter_fourcc(*'mp4v')
            writer = cv2.VideoWriter(
                output_path, fourcc,
                self.target_fps,
                (1280, 720)
            )

        # Encode context once
        context = self.generator.context_encoder(
            encode_text(prompt)
        )

        # Generate initial frame
        initial_latent = torch.randn(1, self.generator.latent_dim)
        prev_frames = [initial_latent]

        start_time = time.time()
        kv_cache = None

        for frame_idx in range(, total_frames):
            frame_start = time.time()

            
             user_control_stream   :
                control = user_control_stream.get_latest()
                 control   :
                    
                    context = update_context(context, control)

            
             torch.no_grad():
                next_latent, kv_cache = (
                    .generator.generate_frame_latent(
                        torch.stack(prev_frames),
                        context,
                        kv_cache
                    )
                )

                
                frame = .generator.decode_latent_to_frame(
                    next_latent
                )

            prev_frames.append(next_latent)

            
            frame_elapsed = time.time() - frame_start
             frame_elapsed < .frame_time:
                time.sleep(.frame_time - frame_elapsed)

            
             output_path:
                frame_np = frame[].cpu().numpy()
                frame_np = (frame_np * ).astype()
                frame_np = cv2.cvtColor(frame_np, cv2.COLOR_RGB2BGR)
                writer.write(frame_np)

            
             frame_idx %  == :
                elapsed = time.time() - start_time
                fps = frame_idx / elapsed
                ()

         output_path:
            writer.release()

Practical Guidance

Student Forcing: Critical for preventing error accumulation. Start with high teacher probability (0.5), decay during training.

Adversarial vs Diffusion: Adversarial training is more efficient than iterative diffusion for autoregressive generation. Single forward pass per frame enables real-time performance.

KV Caching: Essential for real-time inference. Cache attention states to avoid recomputation.

Interactive Control: Accept user inputs (text prompts, control signals) during streaming by updating context features without resetting generation.

Performance Targets: 24fps at 736x416 on single H100 is achievable; 1280x720 requires multi-GPU. Optimize for your hardware.

When to Apply: Use for interactive video generation systems, real-time content creation, or streaming applications where latency matters.

Reference

Adversarial post-training enables efficient real-time video generation by converting diffusion models to autoregressive form. Key innovations: (1) single NFE per frame, (2) student forcing to prevent error accumulation, (3) full KV cache utilization. Together these achieve 24fps performance while maintaining quality.