adu2021 Title: Enable Sub-Second, Infinitely-Long Video Generation With Motion Control
Traditional video generation models either lack motion control or sacrifice speed for quality. MotionStream achieves both through a two-stage approach: a bidirectional teacher model with motion guidance is distilled into a fast causal student. The student uses sliding-window attention to handle arbitrary lengths at constant latency, with attention sinks preventing quality degradation during extended generation.
The result is interactive video generation where users paint trajectories and see results in real-time.
Core Concept
Real-Time Motion-Controlled Video Generation:
- Teacher Model: Bidirectional, motion-guided, high-quality video generation
- Student Model: Causal, single-forward-pass, fast inference
- Distribution Matching Distillation: Bake teacher guidance into student learning
- Sliding-Window Attention: Fixed context window prevents O(n²) memory growth
- Attention Sinks: Preserve image coherence by pinning initial frame tokens
Architecture Overview
- Motion Conditioning: Lightweight sinusoidal embeddings for trajectory inputs
- Teacher Architecture: Text + motion encoder producing guidance signals
- Student Architecture: Causal video generation with sliding-window attention
- Attention Sink Mechanism: Fixed anchor tokens for initialization
- Training: Distillation with self-rollout validation during training
Implementation Steps
1. Design Motion Conditioning System
Encode user trajectories into guidance signals for video generation.
class MotionConditioner(nn.Module):
def __init__(self, hidden_dim=512):
# Sinusoidal embeddings for trajectory points
self.position_encoding = PositionalEncoding(hidden_dim)
# Process trajectories: list of (x, y, frame) points
self.trajectory_encoder = nn.LSTM(2, hidden_dim, batch_first=True)
# Project to hidden dimension for fusion with video
self.projection = nn.Linear(hidden_dim, hidden_dim)
def encode_trajectory(self, trajectory_points):
# trajectory_points: [batch, num_points, 2] (x, y coordinates)
# Add sinusoidal position encoding
encoded = self.position_encoding(trajectory_points)
# LSTM to capture motion dynamics
lstm_out, _ = self.trajectory_encoder(encoded)
# Project for fusion
motion_guidance = self.projection(lstm_out) # [batch, num_points, hidden_dim]
return motion_guidance
def forward(self, text, trajectory=None):
# Encode text (text encoder not shown)
text_features = self.encode_text(text)
if trajectory is not None:
motion_guidance = self.encode_trajectory(trajectory)
# Fuse text and motion
combined = text_features + motion_guidance
else:
combined = text_features
return combined
2. Implement Teacher-Student Distillation
Train student to replicate teacher guidance with single forward pass.
class TeacherStudentDistillation(nn.Module):
def __init__(self, teacher_model, student_model):
self.teacher = teacher_model
self.student = student_model
def forward_teacher(self, condition, num_frames=16):
# Teacher processes bidirectionally for high quality
# Can look ahead and backward in sequence
video = self.teacher(condition, bidirectional=True)
return video
def forward_student(self, condition, num_frames=16):
# Student processes causally (left-to-right only)
video = self.student(condition, bidirectional=False)
return video
def compute_distillation_loss(self, condition, num_frames=16):
# Generate videos from both models
teacher_video = self.forward_teacher(condition, num_frames)
student_video = self.forward_student(condition, num_frames)
# Distribution matching distillation
# Minimize difference in frame-by-frame distributions
teacher_feat = self.teacher.encode(teacher_video) # [batch, frames, dim]
student_feat = self.student.encode(student_video)
# MSE on features
feature_loss = F.mse_loss(student_feat, teacher_feat)
# Perceptual loss on frames
teacher_frames = teacher_video
student_frames = student_video
perceptual_loss = self.compute_perceptual_loss(teacher_frames, student_frames)
return feature_loss + perceptual_loss
def compute_perceptual_loss(self, teacher_frames, student_frames):
# Use pretrained VGG for perceptual distance
teacher_perc = self.vgg(teacher_frames)
student_perc = self.vgg(student_frames)
return F.mse_loss(teacher_perc, student_perc)
3. Implement Sliding-Window Causal Attention
Enable constant-latency inference for arbitrary-length generation.
class SlidingWindowCausalAttention(nn.Module):
def __init__(self, hidden_dim, window_size=8, num_heads=8):
self.window_size = window_size
self.hidden_dim = hidden_dim
self.query_projection = nn.Linear(hidden_dim, hidden_dim)
self.key_projection = nn.Linear(hidden_dim, hidden_dim)
self.value_projection = nn.Linear(hidden_dim, hidden_dim)
self.output_projection = nn.Linear(hidden_dim, hidden_dim)
def forward(self, x, cache=None):
# x: [batch, 1, hidden_dim] (single frame during generation)
batch_size, seq_len, hidden_dim = x.shape
Q = self.query_projection(x) # [batch, 1, hidden_dim]
K = self.key_projection(x)
V = self.value_projection(x)
if cache is not None:
# Retrieve previous window of keys/values
past_K, past_V = cache
# Concatenate with current
K = torch.cat([past_K, K], dim=1) # [batch, window_size + 1, hidden_dim]
V = torch.cat([past_V, V], dim=1)
# Keep only last window_size frames
if K.shape[1] > self.window_size:
K = K[:, -self.window_size:, :]
V = V[:, -self.window_size:, :]
else:
K = K[:, :self.window_size, :]
V = V[:, :self.window_size, :]
# Scaled dot-product attention (no masking needed, only past is available)
scores = torch.matmul(Q, K.transpose(-2, -1)) / np.sqrt(hidden_dim)
weights = F.softmax(scores, dim=-1)
output = torch.matmul(weights, V)
output = self.output_projection(output)
# Cache for next iteration
cache = (K, V)
return output, cache
4. Implement Attention Sinks for Temporal Coherence
Prevent quality degradation during extended generation by anchoring to initial frames.
class AttentionSinkAttention(nn.Module):
def __init__(self, hidden_dim, num_sink_tokens=4):
self.num_sinks = num_sink_tokens
self.hidden_dim = hidden_dim
# Learnable sink tokens initialized from first frame
self.sink_tokens = nn.Parameter(torch.randn(1, num_sink_tokens, hidden_dim))
def forward(self, x, initial_frame=None):
# x: current video latents
# initial_frame: first frame (used as anchor)
batch_size, seq_len, hidden_dim = x.shape
# Initialize sinks from initial frame if provided
if initial_frame is not None:
self.sink_tokens.data = initial_frame[:, :self.num_sinks, :]
# Attention mechanism that always includes sinks
# sinks act as "memory anchors" preventing drift
# Compute attention over (current sequence + sinks)
combined = torch.cat([self.sink_tokens.expand(batch_size, -1, -1), x], dim=1)
# Attention computation (details omitted for brevity)
# Key: attention always sees sinks, preventing divergence from initial frame
return combined
5. Train with Self-Rollout During Training
Simulate inference-time behavior (rolling KV cache) during training.
def train_streaming_video_model(student_model, teacher_model, num_steps=10000):
optimizer = torch.optim.Adam(student_model.parameters(), lr=1e-4)
for step in range(num_steps):
condition = sample_condition_batch()
num_frames = 16
# Training with self-rollout (simulate streaming inference)
cache = None
student_loss = 0
for frame_idx in range(num_frames):
# Student forward (causal, streaming)
frame_student, cache = student_model.generate_frame(
condition, frame_idx, cache
)
# Teacher forward (for supervision)
full_video_teacher = teacher_model(condition)
frame_teacher = full_video_teacher[:, frame_idx, :, :, :]
# Distillation loss on this frame
loss_frame = F.mse_loss(frame_student, frame_teacher)
student_loss += loss_frame
optimizer.zero_grad()
student_loss.backward()
optimizer.step()
if step % 100 == 0:
print(f"Step {step}: Loss {student_loss.item():.4f}")
Practical Guidance
When to Use:
- Real-time interactive video generation
- Streaming video applications
- Motion-conditioned synthesis (UI automation, animation)
Hyperparameters:
- window_size: 8 (balance latency vs. coherence)
- num_sink_tokens: 4 (anchor quality)
- distillation_weight: 0.8 (relative to perception loss)
When NOT to Use:
- Applications requiring frame-by-frame editing flexibility
- Scenarios needing post-hoc video modifications
- Very long sequences (beyond 1-2 minutes)
Pitfalls:
- KV cache overflow: Window must fit in GPU memory; adjust for your hardware
- Sink token initialization: Poor initialization causes early convergence to wrong attractor
- Self-rollout mismatch: Training with rolling cache but evaluating differently causes distribution mismatch
Integration Point: Deploy as interactive layer in video editing/creation tools.