Streaming Video Generation

StreamDiT: Real-Time Streaming Video Generation with Diffusion Transformers

Generating videos in real-time for interactive applications is fundamentally challenging because traditional diffusion models generate entire videos at once, creating latency. StreamDiT redesigns the generation process for streaming: instead of denoising a full video tensor, the model processes a moving buffer of frames continuously, generating new frames one at a time. This allows frames to be displayed immediately without waiting for the entire video, achieving 16 FPS on a single GPU—enabling interactive video generation applications.

The key innovation is modifying flow matching (a diffusion variant) with a moving buffer mechanism where frames enter and exit the buffer dynamically. Rather than predicting frame T given frames 1..T-1, the model works within a local window, reducing memory and computation. Coupled with efficient window attention and distillation, this architecture enables real-time performance without sacrificing quality.

Core Concept

StreamDiT reframes video generation from monolithic denoising to streaming frame synthesis. A moving buffer maintains the last N frames; as the model generates the next frame, the oldest frame drops from the buffer. This local context suffices for coherence because humans perceive temporal continuity at short timescales. The model predicts the noise distribution for the next frame given its local temporal context, not the entire history.

The architecture uses varying time embeddings that differ per frame within the buffer, enabling the model to learn different behaviors for frames at different positions in the buffer. This allows the model to condition predictions on relative temporal position, improving consistency and enabling longer-coherent sequences via shifting windows.

Architecture Overview

The system comprises:

• Base Transformer: Adaptive layer-normalized Diffusion Transformer (adaLN DiT) serving as the core generative model
• Time Embeddings: Frame-dependent sequences rather than scalars, enabling per-frame noise level specification
• Window Attention: Local attention within spatial-temporal windows, with periodic shifting for global communication
• Latent Processing: Video compressed into latent space via temporal (4×) and spatial (8×) auto-encoder
• Training Strategy: Mixed training across different buffer partitioning schemes balancing quality and consistency

Implementation

Start with the streaming buffer and window attention mechanism:

import torch
import torch.nn as nn
from typing import Optional, Tuple

class StreamingBuffer:
    """
    Maintains moving buffer of frames for streaming generation.

    Frames enter at one end, exit at the other, enabling efficient
    local-context video generation without storing the entire sequence.
    """

    def __init__(self, buffer_size: int = 8, frame_dim: int = 64):
        self.buffer_size = buffer_size
        self.frame_dim = frame_dim
        self.frames = None

    def add_frame(self, new_frame: torch.Tensor) -> Optional[torch.Tensor]:
        """
        Add new frame to buffer, return frame exiting if buffer full.

        Args:
            new_frame: (batch, channels, height, width)

        Returns:
            exiting_frame: frame that left buffer, or None if not full yet
        """
        if self.frames is None:
            # Initialize buffer
            self.frames = new_frame.unsqueeze(1)
            return None

        # Add frame
        self.frames = torch.cat([self.frames, new_frame.unsqueeze(1)], dim=1)

        # Remove oldest if buffer exceeds size
        exiting = None
        if self.frames.shape[1] > self.buffer_size:
            exiting = self.frames[:, 0]
            self.frames = self.frames[:, 1:]

        return exiting

    def get_context(self) -> torch.Tensor:
        """Return current buffer frames as context for generation."""
        return self.frames if self.frames is not None else torch.empty(0)

class WindowAttention(nn.Module):
    """
    Efficient attention limited to local spatial-temporal windows.

    Prevents quadratic complexity in sequence length by attending only
    within local regions, with periodic shifting for global information flow.
    """

    def __init__(self, hidden_dim: int, num_heads: int = 12,
                 window_size: int = 8):
        super().__init__()
        self.hidden_dim = hidden_dim
        self.num_heads = num_heads
        self.window_size = window_size

        self.attention = nn.MultiheadAttention(
            hidden_dim, num_heads, batch_first=True
        )
        self.norm = nn.LayerNorm(hidden_dim)

    def create_window_mask(self, seq_len: int, shift: bool = False) -> torch.Tensor:
        """
        Create attention mask for local windows.

        If shift=True, shifts window boundaries for cross-window information.

        Returns: mask of shape (seq_len, seq_len) where True blocks attention
        """
        mask = torch.ones(seq_len, seq_len, dtype=torch.bool)

        if shift:
            # Shift windows by half window size
            shift_size = self.window_size // 2
        else:
            shift_size = 0

        # Allow attention within windows
        for i in range(seq_len):
            window_start = ((i - shift_size) // self.window_size) * self.window_size
            window_end = min(window_start + self.window_size, seq_len)
            mask[i, window_start:window_end] = False

        return mask

    def forward(self, x: torch.Tensor, shift: bool = False) -> torch.Tensor:
        """
        Apply windowed attention to sequence.

        Args:
            x: (batch, seq_len, hidden_dim)
            shift: whether to shift window boundaries for next layer

        Returns:
            attended: (batch, seq_len, hidden_dim)
        """
        mask = self.create_window_mask(x.shape[1], shift=shift)
        mask = mask.to(x.device)

        attn_out, _ = self.attention(x, x, x, attn_mask=mask)
        return self.norm(x + attn_out)

Implement adaptive time embeddings for per-frame noise specification:

class AdaptiveTimeEmbedding(nn.Module):
    """
    Generate per-frame time embeddings enabling frame-specific noise levels.

    Rather than a scalar timestep, produces a sequence of embeddings,
    one per frame, allowing different denoising stages per position.
    """

    def __init__(self, hidden_dim: int = 768, max_frames: int = 32):
        super().__init__()
        self.hidden_dim = hidden_dim
        self.max_frames = max_frames

        # Sinusoidal position encoding
        self.register_buffer('position_encoding',
                            self._create_position_encoding(max_frames, hidden_dim))

        # Time-to-embedding projection
        self.time_proj = nn.Sequential(
            nn.Linear(1, hidden_dim),
            nn.SiLU(),
            nn.Linear(hidden_dim, hidden_dim)
        )

    def _create_position_encoding(self, seq_len: int, d_model: int) -> torch.Tensor:
        """Create sinusoidal positional encodings."""
        position = torch.arange(seq_len).unsqueeze(1)
        div_term = torch.exp(torch.arange(0, d_model, 2) * -(torch.log(torch.tensor(10000.0)) / d_model))

        pe = torch.zeros(seq_len, d_model)
        pe[:, 0::2] = torch.sin(position * div_term)
        pe[:, 1::2] = torch.cos(position * div_term)

        return pe

    def forward(self, t: torch.Tensor, num_frames: int) -> torch.Tensor:
        """
        Generate time embeddings for frame sequence.

        Args:
            t: timestep scalar (0 to 1, where 1 is unnoised)
            num_frames: number of frames in buffer

        Returns:
            embeddings: (num_frames, hidden_dim) time embeddings
        """
        # Get position encodings for frames
        pos_enc = self.position_encoding[:num_frames]  # (num_frames, hidden_dim)

        # Get base time embedding
        t_embed = self.time_proj(t.unsqueeze(0).unsqueeze(0))  # (1, 1, hidden_dim)

        # Combine time and position
        frame_embeddings = pos_enc + t_embed  # (num_frames, hidden_dim)

        return frame_embeddings

Implement the streaming diffusion model:

from diffusers import DDPMScheduler

class StreamingDiT(nn.Module):
    """
    Diffusion Transformer for streaming video generation.

    Generates frames one at a time within a moving buffer, enabling
    real-time streaming output without waiting for full videos.
    """

    def __init__(self, latent_dim: int = 8, hidden_dim: int = 768,
                 num_layers: int = 16, buffer_size: int = 8):
        super().__init__()
        self.latent_dim = latent_dim
        self.hidden_dim = hidden_dim
        self.buffer_size = buffer_size

        # Input projection
        self.input_proj = nn.Linear(latent_dim, hidden_dim)

        # Transformer blocks with window attention
        self.layers = nn.ModuleList([
            nn.TransformerEncoderLayer(
                d_model=hidden_dim,
                nhead=12,
                dim_feedforward=2048,
                batch_first=True,
                activation='gelu'
            )
            for _ in range(num_layers)
        ])

        # Window attention alternates shifts
        self.window_attentions = nn.ModuleList([
            WindowAttention(hidden_dim, window_size=buffer_size)
            for _ in range(num_layers)
        ])

        # Time embeddings
        self.time_embedding = AdaptiveTimeEmbedding(hidden_dim, max_frames=buffer_size)

        # Output projection for noise prediction
        self.output_proj = nn.Linear(hidden_dim, latent_dim)

        # Scheduler for diffusion
        self.scheduler = DDPMScheduler(num_train_timesteps=1000)

    def forward(self, buffer: torch.Tensor, t: torch.Tensor,
                text_embed: Optional[torch.Tensor] = None) -> torch.Tensor:
        """
        Predict noise for next frame in buffer.

        Args:
            buffer: (batch, num_frames, latent_dim) current frame buffer
            t: timestep (0 to 1)
            text_embed: (batch, hidden_dim) optional text conditioning

        Returns:
            noise_pred: (batch, 1, latent_dim) predicted noise for next frame
        """
        batch_size, num_frames, _ = buffer.shape

        # Project input
        x = self.input_proj(buffer)  # (batch, num_frames, hidden_dim)

        # Add time embeddings
        time_emb = self.time_embedding(t, num_frames)  # (num_frames, hidden_dim)
        x = x + time_emb.unsqueeze(0)

        # Add text conditioning if provided
        if text_embed is not None:
            x = x + text_embed.unsqueeze(1)

        # Apply transformer layers with alternating window attention
        for i, layer in enumerate(self.layers):
            # Standard transformer block
            x = layer(x)

            # Windowed attention with alternating shifts
            shift = (i % 2) == 1
            x = self.window_attentions[i % len(self.window_attentions)](x, shift=shift)

        # Predict noise for next frame (use last frame position)
        noise_pred = self.output_proj(x[:, -1:, :])  # (batch, 1, latent_dim)

        return noise_pred

    def generate_streaming(self, prompt: str, num_frames: int = 128,
                          num_inference_steps: int = 8) -> torch.Tensor:
        """
        Generate video frame by frame in streaming fashion.

        Yields frames as they're generated rather than returning full video.

        Args:
            prompt: text description of video
            num_frames: total frames to generate
            num_inference_steps: denoising steps (low for speed)

        Yields:
            frame: single generated frame (batch, 3, 512, 512)
        """
        buffer = StreamingBuffer(buffer_size=self.buffer_size)

        # Encode prompt (placeholder)
        text_embed = self._encode_text(prompt)

        # Initialize with noise
        current_frame = torch.randn(1, self.latent_dim)

        for frame_idx in range(num_frames):
            # Add frame to buffer
            buffer.add_frame(current_frame)

            # Get buffer context
            context = buffer.get_context()

            # Denoise frame
            self.scheduler.set_timesteps(num_inference_steps)
            for t_idx, t in enumerate(self.scheduler.timesteps):
                # Normalize timestep
                t_norm = t.float() / self.scheduler.config.num_train_timesteps

                # Predict noise
                noise_pred = self.forward(context, t_norm, text_embed)

                # Denoise
                current_frame = self.scheduler.step(
                    noise_pred, t, current_frame
                ).prev_sample

            # Decode to image
            image = self._decode_latent(current_frame)
            yield image

            # Update progress
            if (frame_idx + 1) % 16 == 0:
                print(f"Generated {frame_idx + 1}/{num_frames} frames")

    def _encode_text(self, prompt: str) -> torch.Tensor:
        """Encode text prompt to embedding."""
        # Placeholder for CLIP/T5 encoding
        return torch.randn(1, self.hidden_dim)

    def _decode_latent(self, latent: torch.Tensor) -> torch.Tensor:
        """Decode latent to image space."""
        # Placeholder for VAE decoder
        return torch.randn(1, 3, 512, 512)

Practical Guidance

Hyperparameter Table:

When to Use:

• You need interactive video generation (streaming video, real-time applications)
• You want 16+ FPS on single GPU with good visual quality
• You're generating long-form videos (100+ frames) where latency compounds
• You can tolerate 8-step denoising per frame
• You need to display frames immediately without waiting for full video

When NOT to Use:

• You need highest quality output (full-length diffusion steps produce better results)
• You need single-shot generation without streaming (monolithic models faster)
• You're generating very short clips (<16 frames) where streaming overhead dominates
• You need fine-grained control over entire video coherence
• Your hardware has very limited VRAM (<8GB GPU memory)

Common Pitfalls:

• Buffer too small: Small buffers lose long-range temporal coherence. Minimum 4 frames; 8+ recommended.
• Too few diffusion steps: <4 steps produce severe artifacts. 8 is minimum for acceptable quality.
• Inference steps too high: More than 20 steps negates real-time benefit. Profile your hardware for optimal step count.
• Text encoding overhead: CLIP encoding can bottleneck. Cache or use lightweight models like DistilBERT.
• Memory spikes: Buffer residuals and activation caching can cause unexpected VRAM spikes. Monitor carefully.
• Temporal flickering: If buffer doesn't overlap frames properly, adjacent frames may flicker. Test with static scenes.

Reference

Authors (2025). StreamDiT: Real-Time Streaming Text-to-Video Generation. arXiv preprint arXiv:2507.03745. https://arxiv.org/abs/2507.03745