Skip to content

kandinsky5

Classes

fastvideo.pipelines.stages.kandinsky5.Kandinsky5DenoisingStage

Kandinsky5DenoisingStage(transformer, scheduler)

Bases: PipelineStage

Source code in fastvideo/pipelines/stages/kandinsky5.py
def __init__(self, transformer, scheduler) -> None:
    super().__init__()
    self.transformer = transformer
    self.scheduler = scheduler

Methods:

fastvideo.pipelines.stages.kandinsky5.Kandinsky5DenoisingStage.fast_sta_nabla staticmethod
fast_sta_nabla(T: int, H: int, W: int, wT: int = 3, wH: int = 3, wW: int = 3, device: device | str = 'cuda') -> Tensor

Create a sparse temporal attention (STA) mask for efficient video generation.

This method generates a mask that limits attention to nearby frames and spatial positions, reducing computational complexity for video generation.

Parameters:

Name Type Description Default
T int

Number of temporal frames

required
H int

Height in latent space

required
W int

Width in latent space

required
wT int

Temporal attention window size

3
wH int

Height attention window size

3
wW int

Width attention window size

3
device str

Device to create tensor on

'cuda'

Returns:

Type Description
Tensor

torch.Tensor: Sparse attention mask of shape (THW, THW)

Source code in fastvideo/pipelines/stages/kandinsky5.py
@staticmethod
def fast_sta_nabla(
    T: int,
    H: int,
    W: int,
    wT: int = 3,
    wH: int = 3,
    wW: int = 3,
    device: torch.device | str = "cuda",
) -> torch.Tensor:
    """
    Create a sparse temporal attention (STA) mask for efficient video generation.

    This method generates a mask that limits attention to nearby frames and spatial positions, reducing
    computational complexity for video generation.

    Args:
        T (int): Number of temporal frames
        H (int): Height in latent space
        W (int): Width in latent space
        wT (int): Temporal attention window size
        wH (int): Height attention window size
        wW (int): Width attention window size
        device (str): Device to create tensor on

    Returns:
        torch.Tensor: Sparse attention mask of shape (T*H*W, T*H*W)
    """
    max_extent = int(torch.tensor([T, H, W], device=device).amax().item())
    r = torch.arange(0, max_extent, 1, dtype=torch.int16, device=device)
    mat = (r.unsqueeze(1) - r.unsqueeze(0)).abs()
    sta_t, sta_h, sta_w = (
        mat[:T, :T].flatten(),
        mat[:H, :H].flatten(),
        mat[:W, :W].flatten(),
    )
    sta_t = sta_t <= wT // 2
    sta_h = sta_h <= wH // 2
    sta_w = sta_w <= wW // 2
    sta_hw = (sta_h.unsqueeze(1) * sta_w.unsqueeze(0)).reshape(H, H, W, W).transpose(1, 2).flatten()
    sta = (sta_t.unsqueeze(1) * sta_hw.unsqueeze(0)).reshape(T, T, H * W, H * W).transpose(1, 2)
    return sta.reshape(T * H * W, T * H * W)
fastvideo.pipelines.stages.kandinsky5.Kandinsky5DenoisingStage.get_sparse_params
get_sparse_params(sample: Tensor, device: device) -> dict[str, Any] | None

Generate sparse attention parameters for the transformer based on sample dimensions.

This method computes the sparse attention configuration needed for efficient video processing in the transformer model.

Parameters:

Name Type Description Default
sample Tensor

Input sample tensor

required
device device

Device to place tensors on

required

Returns:

Name Type Description
Dict dict[str, Any] | None

Dictionary containing sparse attention parameters

Source code in fastvideo/pipelines/stages/kandinsky5.py
def get_sparse_params(self, sample: torch.Tensor, device: torch.device) -> dict[str, Any] | None:
    """
    Generate sparse attention parameters for the transformer based on sample dimensions.

    This method computes the sparse attention configuration needed for efficient video processing in the
    transformer model.

    Args:
        sample (torch.Tensor): Input sample tensor
        device (torch.device): Device to place tensors on

    Returns:
        Dict: Dictionary containing sparse attention parameters
    """
    assert self.transformer.config.patch_size[0] == 1
    _, T, H, W, _ = sample.shape
    T, H, W = (
        T // self.transformer.config.patch_size[0],
        H // self.transformer.config.patch_size[1],
        W // self.transformer.config.patch_size[2],
    )
    if self.transformer.config.attention_type == "nabla":
        sta_mask = self.fast_sta_nabla(
            T,
            H // 8,
            W // 8,
            self.transformer.config.attention_wT,
            self.transformer.config.attention_wH,
            self.transformer.config.attention_wW,
            device=device,
        )

        sparse_params = {
            "sta_mask": sta_mask.unsqueeze_(0).unsqueeze_(0),
            "attention_type": self.transformer.config.attention_type,
            "to_fractal": True,
            "P": self.transformer.config.attention_P,
            "wT": self.transformer.config.attention_wT,
            "wW": self.transformer.config.attention_wW,
            "wH": self.transformer.config.attention_wH,
            "add_sta": self.transformer.config.attention_add_sta,
            "visual_shape": (T, H, W),
            "method": self.transformer.config.attention_method,
        }
    else:
        sparse_params = None

    return sparse_params

fastvideo.pipelines.stages.kandinsky5.Kandinsky5DmdDenoisingStage

Kandinsky5DmdDenoisingStage(transformer, scheduler)

Bases: Kandinsky5DenoisingStage

DMD (few fixed steps, no CFG) variant of Kandinsky5DenoisingStage.

Reuses the parent's RoPE/scale_factor/sparse-params helpers; only the denoising loop differs: a fixed short timestep schedule (pipeline_config.dmd_denoising_steps) with a single forward pass per step and no classifier-free-guidance branch, matching DMD's distilled few-step generator.

dmd_denoising_steps values (e.g. [1000, 750, 500, 250]) are literal final target timesteps -- that's how DMD2Method on the training side resolves them: nearest-sigma lookup against a scheduler that has never had set_timesteps called on it, so e.g. "750" maps to sigma 0.75. This stage therefore keeps a private FlowMatchEulerDiscreteScheduler (same shift as the pipeline scheduler) and drives it directly via predict-x0 + re-noise, exactly mirroring DMD2Method._student_rollout's "simulate" branch and Wan's own DmdDenoisingStage. It must NOT reuse the pipeline's shared scheduler object through scheduler.step(): by the time this stage runs, TimestepPreparationStage has already called scheduler.set_timesteps(timesteps=dmd_denoising_steps) on it, which re-applies the flow-match shift warp on top of values that are already final (e.g. sigma 0.75 -> 0.9375 at shift=5) -- a double shift that leaves every step far noisier than the student was trained for, so the sampled video is still mostly noise after the last step instead of converged.

Source code in fastvideo/pipelines/stages/kandinsky5.py
def __init__(self, transformer, scheduler) -> None:
    super().__init__(transformer, scheduler)
    self._sample_scheduler = FlowMatchEulerDiscreteScheduler(shift=scheduler.shift)

fastvideo.pipelines.stages.kandinsky5.Kandinsky5ImageEncodingStage

Kandinsky5ImageEncodingStage(vae: ParallelTiledVAE, pipeline=None)

Bases: EncodingStage

Encode the conditioning image into a VAE latent for I2V.

Source code in fastvideo/pipelines/stages/kandinsky5.py
def __init__(self, vae: ParallelTiledVAE, pipeline=None) -> None:
    super().__init__(vae=vae)

fastvideo.pipelines.stages.kandinsky5.Kandinsky5NormalizationStage

Bases: PipelineStage

Normalize the first latent frames to reduce I2V conditioning artifacts.

Functions: