# Interactive Video Generation > Convert pre-trained latent video diffusion into real-time autoregressive generators using adversarial post-training, achieving 24fps streaming on single H100. - Skill: `adu2021/interactive-video-generation` (Agent Skill) - Install (CLI): `npx skillmds@latest add adu2021/interactive-video-generation` - Raw SKILL.md: https://api.skillmd.com/api/skills/adu2021/interactive-video-generation/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: AI & ML - License: MIT - Author: adu2021 (https://skillmd.com/u/adu2021) - Updated: 2026-09-17 - Page: https://skillmd.com/skills/adu2021/interactive-video-generation --- # 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: ```python 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(self, prev_latents: torch.Tensor, context: torch.Tensor, kv_cache = None ) -> tuple: """Generate next frame in latent space.""" # Encode context context_features = self.context_encoder(context) # Autoregressive generation if kv_cache is not None: # Use KV cache for efficiency decoder_input = prev_latents[-1:].unsqueeze(0) hidden = self.transformer( decoder_input, memory=context_features, memory_key_padding_mask=None ) # Update cache (simplified) kv_cache = hidden else: # Full autoregressive hidden = self.transformer( prev_latents, memory=context_features ) # Predict next latent next_latent = self.latent_head(hidden[-1]) return next_latent, kv_cache def decode_latent_to_frame(self, latent: torch.Tensor ) -> torch.Tensor: """Convert latent vector back to RGB frame.""" with torch.no_grad(): frame = self.vae.decode(latent) return frame ``` ### Step 2: Implement Adversarial Training Train discriminator to ensure autoregressive quality: ```python 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(1), torch.nn.Flatten(), torch.nn.Linear(256, 128), torch.nn.LeakyReLU(0.2), torch.nn.Linear(128, 1) ) def forward(self, video: torch.Tensor) -> torch.Tensor: """Discriminate real vs generated video.""" # Expects (batch, num_frames, 3, height, width) video = video.permute(0, 2, 1, 3, 4) features = self.spatial_temporal_blocks(video) score = self.classifier(features) return score ``` ### Step 3: Adversarial Post-Training Loop Train generator and discriminator with student forcing: ```python 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 train_step(self, real_video: torch.Tensor, context: torch.Tensor) -> dict: """Single adversarial training step.""" batch_size = real_video.shape[0] num_frames = real_video.shape[1] # Encode real video to latents real_latents = self.generator.encode_frames_to_latents( real_video ) # Generate fake video with student forcing fake_latents = self.student_forcing_rollout( real_latents[:, 0:1], context, num_frames, teacher_forcing_prob=0.3 ) # Decode fake latents to frames fake_frames = torch.stack([ self.generator.decode_latent_to_frame(lat) for lat in fake_latents ], dim=1) # Discriminator loss real_scores = self.discriminator(real_video) fake_scores = self.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 self.d_optimizer.zero_grad() d_loss.backward() self.d_optimizer.step() # Generator loss fake_scores = self.discriminator(fake_frames) g_loss_adv = torch.nn.functional.softplus( -fake_scores ).mean() # Reconstruction loss g_loss_recon = torch.nn.functional.mse_loss( fake_frames, real_video ) g_loss = g_loss_adv + 10.0 * g_loss_recon self.g_optimizer.zero_grad() g_loss.backward() self.g_optimizer.step() return { "d_loss": d_loss.item(), "g_loss": g_loss.item(), "g_loss_adv": g_loss_adv.item(), "g_loss_recon": g_loss_recon.item() } ``` ### Step 4: Real-Time Streaming Inference Generate and stream video with user controls: ```python 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(1, total_frames): frame_start = time.time() # Check for user controls if user_control_stream is not None: control = user_control_stream.get_latest() if control is not None: # Modify context based on user input context = update_context(context, control) # Generate next frame with torch.no_grad(): next_latent, kv_cache = ( self.generator.generate_frame_latent( torch.stack(prev_frames), context, kv_cache ) ) # Decode to RGB frame = self.generator.decode_latent_to_frame( next_latent ) prev_frames.append(next_latent) # Maintain real-time speed frame_elapsed = time.time() - frame_start if frame_elapsed < self.frame_time: time.sleep(self.frame_time - frame_elapsed) # Save frame if output_path: frame_np = frame[0].cpu().numpy() frame_np = (frame_np * 255).astype('uint8') frame_np = cv2.cvtColor(frame_np, cv2.COLOR_RGB2BGR) writer.write(frame_np) # Status if frame_idx % 12 == 0: elapsed = time.time() - start_time fps = frame_idx / elapsed print(f"Frame {frame_idx}: {fps:.1f} FPS") if 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.