# Motion Stream Real Time Interactive Video > Generate videos at 29 FPS with interactive motion control through teacher-student distillation of motion-conditioned video models, using sliding-window causal attention and attention sinks to maintain constant latency for indefinite-length generation. - Skill: `adu2021/motion-stream-real-time-interactive-video` (Agent Skill) - Install (CLI): `npx skillmds@latest add adu2021/motion-stream-real-time-interactive-video` - Raw SKILL.md: https://api.skillmd.com/api/skills/adu2021/motion-stream-real-time-interactive-video/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: Coding & Dev Tools - License: MIT - Author: adu2021 (https://skillmd.com/u/adu2021) - Updated: 2026-09-17 - Page: https://skillmd.com/skills/adu2021/motion-stream-real-time-interactive-video --- # 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. ```python 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. ```python 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. ```python 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. ```python 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. ```python 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. ## Reference arXiv: https://arxiv.org/abs/2511.01266