Skip to content

magi_human

daVinci-MagiHuman DiT (base variant).

Ported from https://github.com/GAIR-NLP/daVinci-MagiHuman (inference/model/dit/dit_module.py, ~950 lines in the reference).

Architecture summary (verified against GAIR/daVinci-MagiHuman/base/ weights):

- 40 transformer layers, hidden 5120, head_dim 128. - GQA with 40 query heads and 8 KV heads. - Multi-modality "sandwich": layers 0..3 and 36..39 use 3-way modality experts (video/audio/text) packed inside each linear as weight[..., out * 3, in]. Middle layers share a single expert. - Per-head attention gating: the QKV projection emits an extra num_heads_q channels that are sigmoid-gated onto the attention output. - Activation is GELU7 on layers 0..3 (non-gated, intermediate=4hidden) and SwiGLU7 elsewhere (gated, intermediate=int(hidden4⅔)//44). - Position encoding is an element-wise Fourier embedding over 9-column coords (t,h,w + original TxHxW + reference TxHxW), not a standard 1D/3D RoPE. - Forward takes a flat concatenated token stream (video first, then audio, then text) plus a modality map; the internal ModalityDispatcher permutes by modality before each linear so per-expert chunks line up.

Deviations from the "use fastvideo.layers primitives everywhere" guideline in the add-model skill:

- The packed-expert linears store weight as [out * num_experts, in]. FastVideo's ReplicatedLinear does not model this layout; we use raw nn.Parameter with a small wrapper below. This is deliberate and scoped to this DiT: ReplicatedLinear still handles the adapter.* embedders and final_linear_{video,audio} (single-expert) here. - Self-attention is full-sequence and crosses modalities inside the flat concat stream; DistributedAttention assumes a clean spatial-sequence layout, so for the first port we use torch SDPA. Multi-GPU sequence parallelism is a follow-up. - torch.compile via magi_compiler is replaced with a plain nn.Module.

For the full history and shape-by-shape verification notes, see .claude/skills/add-model/SKILL.md and the scaffold PR description.

Classes

fastvideo.models.dits.magi_human.ElementWiseFourierEmbed

ElementWiseFourierEmbed(dim: int, temperature: float = 10000.0, dtype: dtype = float32)

Bases: Module

Element-wise Fourier embedding over 9-column coords (t, h, w, T, H, W, ref_T, ref_H, ref_W). Produces a per-token positional embedding that acts as the RoPE angle input for attention.

Weight: bands of shape [dim // 8] (fixed at init via freq_bands).

Source code in fastvideo/models/dits/magi_human.py
def __init__(
    self,
    dim: int,
    temperature: float = 10000.0,
    dtype: torch.dtype = torch.float32,
):
    super().__init__()
    self.dim = dim
    bands = _freq_bands(dim // 8, temperature=temperature).to(dtype)
    # `register_buffer` so state_dict keeps it, matching upstream naming.
    self.register_buffer("bands", bands)

fastvideo.models.dits.magi_human.MagiAttention

MagiAttention(cfg: AttentionSubConfig)

Bases: Module

Self-attention with GQA + optional per-head sigmoid gating.

Source code in fastvideo/models/dits/magi_human.py
def __init__(self, cfg: AttentionSubConfig):
    super().__init__()
    self.cfg = cfg
    self.gating_size = cfg.num_heads_q if cfg.enable_attn_gating else 0
    qkv_out = (
        cfg.num_heads_q * cfg.head_dim
        + 2 * cfg.num_heads_kv * cfg.head_dim
        + self.gating_size
    )
    self.pre_norm = MultiModalityRMSNorm(cfg.hidden_size, num_modality=cfg.num_modality)
    self.linear_qkv = PackedExpertLinear(
        cfg.hidden_size, qkv_out, num_experts=cfg.num_modality, bias=False,
    )
    self.linear_proj = PackedExpertLinear(
        cfg.num_heads_q * cfg.head_dim, cfg.hidden_size,
        num_experts=cfg.num_modality, bias=False,
    )
    self.q_norm = MultiModalityRMSNorm(cfg.head_dim, num_modality=cfg.num_modality)
    self.k_norm = MultiModalityRMSNorm(cfg.head_dim, num_modality=cfg.num_modality)

    self.q_size = cfg.num_heads_q * cfg.head_dim
    self.kv_size = cfg.num_heads_kv * cfg.head_dim

    self.attn = LocalAttention(
        num_heads=cfg.num_heads_q,
        head_size=cfg.head_dim,
        num_kv_heads=cfg.num_heads_kv,
        causal=False,
        supported_attention_backends=(
            AttentionBackendEnum.FLASH_ATTN,
            AttentionBackendEnum.TORCH_SDPA,
        ),
    )

fastvideo.models.dits.magi_human.MagiHumanDiT

MagiHumanDiT(config: MagiHumanVideoConfig, hf_config: dict | None = None, **kwargs)

Bases: BaseDiT

Top-level DiT for daVinci-MagiHuman (base).

Forward signature mirrors the reference DiTModel.forward: it takes a flat token stream, its per-token coords and modality mapping, and returns per-modality outputs packed into a max-channel-width tensor.

This scaffold is single-GPU only; the ulysses_scheduler().dispatch(...) sequence-parallel wrapping in the reference has no equivalent here yet.

Source code in fastvideo/models/dits/magi_human.py
def __init__(self, config: MagiHumanVideoConfig, hf_config: dict | None = None, **kwargs):
    super().__init__(config=config, hf_config=hf_config or {})
    arch: MagiHumanArchConfig = getattr(config, "arch_config", config)
    self.arch = arch

    # BaseDiT contract instance vars.
    self.hidden_size = arch.hidden_size
    self.num_attention_heads = arch.num_attention_heads
    self.num_channels_latents = arch.num_channels_latents

    self.adapter = MagiAdapter(arch)
    self.block = _TransformerBlock(arch)
    self.final_norm_video = MultiModalityRMSNorm(arch.hidden_size)
    self.final_norm_audio = MultiModalityRMSNorm(arch.hidden_size)
    self.final_linear_video = nn.Linear(
        arch.hidden_size, arch.video_in_channels, bias=False, dtype=torch.float32,
    )
    self.final_linear_audio = nn.Linear(
        arch.hidden_size, arch.audio_in_channels, bias=False, dtype=torch.float32,
    )
    # Dispatcher + mask cache: modality_mapping is the same tensor object
    # across all timesteps in the denoising loop; data_ptr()+shape+dtype+device
    # is a fast, collision-free key that avoids rebuilding ModalityDispatcher
    # (which calls argsort + bincount) on every forward call.
    self._cached_dispatcher: Optional[ModalityDispatcher] = None
    self._cached_video_mask: Optional[torch.Tensor] = None
    self._cached_audio_mask: Optional[torch.Tensor] = None
    self._cached_text_mask: Optional[torch.Tensor] = None
    self._cached_modality_key: Optional[tuple] = None

Methods:

fastvideo.models.dits.magi_human.MagiHumanDiT.forward
forward(x: Tensor, coords_mapping: Tensor, modality_mapping: Tensor) -> Tensor

Parameters:

Name Type Description Default
x Tensor

[L, max(V_ch, A_ch, T_ch)]

required
coords_mapping Tensor

[L, 9]

required
modality_mapping Tensor

[L] (int in {VIDEO, AUDIO, TEXT})

required

Returns: out: [L, max(V_ch, A_ch)] with video channels in video slots and audio channels in audio slots; text slots are zero.

Source code in fastvideo/models/dits/magi_human.py
def forward(
    self,
    x: torch.Tensor,
    coords_mapping: torch.Tensor,
    modality_mapping: torch.Tensor,
) -> torch.Tensor:
    """
    Args:
        x:                [L, max(V_ch, A_ch, T_ch)]
        coords_mapping:   [L, 9]
        modality_mapping: [L]  (int in {VIDEO, AUDIO, TEXT})
    Returns:
        out: [L, max(V_ch, A_ch)] with video channels in video slots and
             audio channels in audio slots; text slots are zero.
    """
    key = self._modality_cache_key(modality_mapping)
    if key != self._cached_modality_key:
        self._cached_dispatcher = ModalityDispatcher(modality_mapping, num_modalities=3)
        self._cached_video_mask = modality_mapping == Modality.VIDEO
        self._cached_audio_mask = modality_mapping == Modality.AUDIO
        self._cached_text_mask = modality_mapping == Modality.TEXT
        self._cached_modality_key = key
    dispatcher = self._cached_dispatcher
    video_mask = self._cached_video_mask
    audio_mask = self._cached_audio_mask
    text_mask = self._cached_text_mask
    num_video_tokens = int(video_mask.sum().item())
    if num_video_tokens:
        num_frames = int(coords_mapping[:num_video_tokens, 0].max().item()) + 1
    else:
        num_frames = 0

    x, rope = self.adapter(x, coords_mapping, video_mask, audio_mask, text_mask)
    # Keep the residual stream in adapter dtype (fp32) entering the block.
    # Upstream daVinci-MagiHuman dit_module.py:923 casts to params_dtype,
    # which is fp32 by default; each layer's pre_norm.to(bf16) handles
    # the bf16 internal-compute boundary, and linear_proj outputs bf16
    # which gets promoted back to fp32 by the residual addition. Casting
    # the residual to bf16 here degrades the cross-layer accumulator and
    # compounds visibly over 40 layers in pipeline parity.
    x = ModalityDispatcher.permute(x, dispatcher.permute_mapping)

    x = self.block(
        x, rope,
        permute_mapping=dispatcher.permute_mapping,
        inv_permute_mapping=dispatcher.inv_permute_mapping,
        modality_dispatcher=dispatcher,
        num_video_tokens=num_video_tokens,
        num_frames=num_frames,
    )
    x = ModalityDispatcher.inv_permute(x, dispatcher.inv_permute_mapping)

    x_video = x[video_mask].to(self.final_norm_video.weight.dtype)
    x_video = self.final_norm_video(x_video)
    x_video = self.final_linear_video(x_video)

    x_audio = x[audio_mask].to(self.final_norm_audio.weight.dtype)
    x_audio = self.final_norm_audio(x_audio)
    x_audio = self.final_linear_audio(x_audio)

    max_ch = max(self.arch.video_in_channels, self.arch.audio_in_channels)
    out = torch.zeros(x.shape[0], max_ch, device=x.device, dtype=x.dtype)
    out[video_mask, : self.arch.video_in_channels] = x_video.to(out.dtype)
    out[audio_mask, : self.arch.audio_in_channels] = x_audio.to(out.dtype)
    return out

fastvideo.models.dits.magi_human.ModalityDispatcher

ModalityDispatcher(modality_mapping: Tensor, num_modalities: int)

Permute a flat token stream so same-modality tokens are contiguous.

The DiT's multi-expert linears apply a different weight chunk per modality. Instead of carrying a branch inside each Linear, we pre-permute tokens so each chunk sees a contiguous slice, then un-permute before computing RoPE/attention across the full sequence.

Source code in fastvideo/models/dits/magi_human.py
def __init__(self, modality_mapping: torch.Tensor, num_modalities: int):
    self.modality_mapping = modality_mapping
    self.num_modalities = num_modalities
    self.permute_mapping = torch.argsort(modality_mapping)
    self.inv_permute_mapping = torch.argsort(self.permute_mapping)
    permuted = modality_mapping[self.permute_mapping]
    self.group_size = torch.bincount(permuted, minlength=num_modalities).to(torch.int32)
    self.group_size_cpu: list[int] = [int(x) for x in self.group_size.cpu().tolist()]

fastvideo.models.dits.magi_human.MultiModalityRMSNorm

MultiModalityRMSNorm(dim: int, eps: float = 1e-06, num_modality: int = 1)

Bases: Module

RMSNorm with optional per-modality scale.

When num_modality == 1, behaves identically to a standard RMSNorm with weight initialized to zero (effective weight is 1 + weight, hence the learnable +1 offset baked into the forward path). When num_modality > 1, the weight tensor packs per-modality scales along its flat axis and the dispatcher selects the right chunk per modality.

Source code in fastvideo/models/dits/magi_human.py
def __init__(self, dim: int, eps: float = 1e-6, num_modality: int = 1):
    super().__init__()
    self.dim = dim
    self.eps = eps
    self.num_modality = num_modality
    # Always stored in fp32; matches the reference initialization.
    self.weight = nn.Parameter(torch.zeros(dim * num_modality, dtype=torch.float32))

fastvideo.models.dits.magi_human.PackedExpertLinear

PackedExpertLinear(in_features: int, out_features: int, num_experts: int = 1, bias: bool = False, dtype: dtype = bfloat16)

Bases: Module

Linear where the weight is packed per-modality along the output axis.

Shapes

weight: [out_features * num_experts, in_features] bias: [out_features * num_experts] (optional)

When num_experts == 1, behaves exactly like nn.Linear. When num_experts > 1, forward dispatches the input via the supplied ModalityDispatcher, applies the per-modality weight/bias chunk, and gathers the outputs in original order.

Why not use ReplicatedLinear? Because the packed-expert layout is not what ReplicatedLinear (or any other fastvideo.layers.linear) is wired for. Using raw nn.Parameter keeps weight loading trivial (names map 1:1 to the upstream checkpoint) and avoids quantization-path assumptions that don't match this layout.

Source code in fastvideo/models/dits/magi_human.py
def __init__(
    self,
    in_features: int,
    out_features: int,
    num_experts: int = 1,
    bias: bool = False,
    dtype: torch.dtype = torch.bfloat16,
):
    super().__init__()
    self.in_features = in_features
    self.out_features = out_features
    self.num_experts = num_experts
    self.use_bias = bias
    self.weight = nn.Parameter(
        torch.empty(out_features * num_experts, in_features, dtype=dtype)
    )
    if bias:
        self.bias = nn.Parameter(
            torch.empty(out_features * num_experts, dtype=dtype)
        )
    else:
        self.register_parameter("bias", None)

Functions:

fastvideo.models.dits.magi_human.swiglu7

swiglu7(x: Tensor, alpha: float = 1.702, limit: float = 7.0) -> Tensor

Gated swish-GLU with OpenAI-OSS-style limits and +1 linear bias.

Source code in fastvideo/models/dits/magi_human.py
def swiglu7(x: torch.Tensor, alpha: float = 1.702, limit: float = 7.0) -> torch.Tensor:
    """Gated swish-GLU with OpenAI-OSS-style limits and +1 linear bias."""
    in_dtype = x.dtype
    x = x.to(torch.float32)
    x_glu, x_linear = x[..., ::2], x[..., 1::2]
    x_glu = x_glu.clamp(max=limit)
    x_linear = x_linear.clamp(min=-limit, max=limit)
    out_glu = x_glu * torch.sigmoid(alpha * x_glu)
    return (out_glu * (x_linear + 1)).to(in_dtype)