--- title: lingbot_video ---

Native LingBot-Video Qwen3-VL language model for text-only conditioning.

fastvideo.models.encoders.lingbot_video.BaseEncoderOutput

BaseEncoderOutput(last_hidden_state: torch.FloatTensor | None = None, pooler_output: torch.FloatTensor | None = None, hidden_states: tuple[torch.FloatTensor, ...] | None = None, attentions: tuple[torch.FloatTensor, ...] | None = None, attention_mask: torch.Tensor | None = None)

fastvideo.models.encoders.lingbot_video.RMSNorm

RMSNorm(hidden_size: int, eps: float = 1e-06, dtype: torch.dtype = torch.float32, var_hidden_size: int | None = None, has_weight: bool = True)

Root mean square normalization.

Computes x -> w * x / sqrt(E[x^2] + eps) where w is the learned weight. Refer to https://arxiv.org/abs/1910.07467

fastvideo.models.encoders.lingbot_video.RMSNorm.forward_native

forward_native(x: torch.Tensor, residual: torch.Tensor | None = None) -> torch.Tensor | tuple[torch.Tensor, torch.Tensor]

PyTorch-native implementation equivalent to forward().

fastvideo.models.encoders.lingbot_video.RMSNorm.extra_repr

extra_repr() -> str

fastvideo.models.encoders.lingbot_video.VocabParallelEmbedding

VocabParallelEmbedding(num_embeddings: int, embedding_dim: int, params_dtype: torch.dtype | None = None, org_num_embeddings: int | None = None, padding_size: int = DEFAULT_VOCAB_PADDING_SIZE, quant_config: QuantizationConfig | None = None, prefix: str = '')

Embedding parallelized in the vocabulary dimension.

Adapted from torch.nn.Embedding, note that we pad the vocabulary size to make sure it is divisible by the number of model parallel GPUs.

In order to support various loading methods, we ensure that LoRA-added embeddings are always at the end of TP-sharded tensors. In other words, we shard base embeddings and LoRA embeddings separately (both padded), and place them in the same tensor. In this example, we will have the original vocab size = 1010, added vocab size = 16 and padding to 64. Therefore, the total vocab size with padding will be 1088 (because we first pad 1010 to 1024, add 16, and then pad to 1088). Therefore, the tensor format looks like the following: TP1, rank 0 (no sharding): |< --------BASE-------- >|< -BASE PADDING-- >|< -----LORA------ >|< -LORA PADDING-- >| corresponding token_id: | 0 | 1 | ... | 1009 | -1 | ... | -1 | 1010 | ... | 1015 | -1 | ... | -1 | index: | 0 | 1 | ... | 1009 | 1010 | ... | 1023 | 1024 | ... | 1039 | 1040 | ... | 1087 |

TP2, rank 0: |< --------------------BASE--------------------- >|< -----LORA------ >|< -LORA PADDING- >| corresponding token_id: | 0 | 1 | 2 | ... | 497 | 498 | ... | 511 | 1000 | ... | 1015 | -1 | ... | -1 | index: | 0 | 1 | 2 | ... | 497 | 498 | ... | 511 | 512 | ... | 527 | 520 | ... | 543 | TP2, rank 1: |< -----------BASE----------- >|< -BASE PADDING- >|< -----------LORA PADDING----------- >| corresponding token_id: | 512 | 513 | 514 | ... | 1009 | -1 | ... | -1 | -1 | ... | -1 | -1 | ... | -1 | index: | 0 | 1 | 2 | ... | 497 | 498 | ... | 511 | 512 | ... | 519 | 520 | ... | 543 |

Parameters

num_embeddings: int

vocabulary size.

embedding_dim: int

size of hidden state.

params_dtype: torch.dtype | None

type of the parameters.

org_num_embeddings: int | None

original vocabulary size (without LoRA).

padding_size: int

padding size for the vocabulary.

quant_config: QuantizationConfig | None

quant config for the layer

prefix: str

full name of the layer in the state dict

fastvideo.models.encoders.lingbot_video.VocabParallelEmbedding.get_sharded_to_full_mapping

get_sharded_to_full_mapping() -> list[int] | None

Get a mapping that can be used to reindex the gathered logits for sampling.

During sampling, we gather logits from all ranks. The relationship of index->token_id will follow the same format as outlined in the class docstring. However, after the gather, we want to reindex the final logits tensor to map index->token_id one-to-one (the index is always equal the token_id it corresponds to). The indices returned by this method allow us to do that.

fastvideo.models.encoders.lingbot_video.VocabParallelEmbedding.weight_loader

weight_loader(param: Parameter, loaded_weight: torch.Tensor)

fastvideo.models.encoders.lingbot_video.VocabParallelEmbedding.forward

forward(input_)

fastvideo.models.encoders.lingbot_video.VocabParallelEmbedding.extra_repr

extra_repr() -> str

fastvideo.models.encoders.lingbot_video.TextEncoder

TextEncoder(config: TextEncoderConfig)

Base for native encoders with a model-specific forward output contract.

fastvideo.models.encoders.lingbot_video.TextEncoder.forward

forward(*args: Any, **kwargs: Any) -> TextEncoderOutputT

fastvideo.models.encoders.lingbot_video.Qwen3Attention

Qwen3Attention(config: Qwen3TextConfig, hidden_size: int, num_heads: int, num_kv_heads: int, rope_theta: float = 1000000.0, rope_scaling: dict[str, Any] | None = None, max_position_embeddings: int = 40960, quant_config: QuantizationConfig | None = None, bias: bool = False, prefix: str = '')

Qwen3 attention with QK-Norm and tensor parallelism.

Key difference from LLaMA: RMSNorm is applied to Q and K before attention.

fastvideo.models.encoders.lingbot_video.Qwen3Attention.forward

forward(positions: torch.Tensor, hidden_states: torch.Tensor, attention_mask: torch.Tensor | None = None) -> torch.Tensor

fastvideo.models.encoders.lingbot_video.Qwen3DecoderLayer

Qwen3DecoderLayer(config: Qwen3TextConfig, quant_config: QuantizationConfig | None = None, prefix: str = '')

Qwen3 transformer decoder layer.

fastvideo.models.encoders.lingbot_video.Qwen3DecoderLayer.forward

forward(positions: torch.Tensor, hidden_states: torch.Tensor, residual: torch.Tensor | None, attention_mask: torch.Tensor | None = None) -> tuple[torch.Tensor, torch.Tensor]

fastvideo.models.encoders.lingbot_video.Qwen3ForCausalLM

Qwen3ForCausalLM(config: Qwen3TextConfig)

Qwen3 causal language model for text encoding in diffusion models (e.g. Flux2 Klein).

Features: - Tensor parallelism support - FlashAttention/SDPA support via LocalAttention - QK-Norm for better training stability - output_hidden_states for Klein (layers 9, 18, 27)

fastvideo.models.encoders.lingbot_video.Qwen3ForCausalLM.from_pretrained_local

from_pretrained_local(model_path: str, model_config: Qwen3TextConfig, dtype: torch.dtype, device: torch.device) -> nn.Module

fastvideo.models.encoders.lingbot_video.Qwen3ForCausalLM.get_input_embeddings

get_input_embeddings(input_ids: torch.Tensor) -> torch.Tensor

fastvideo.models.encoders.lingbot_video.Qwen3ForCausalLM.forward

forward(input_ids: torch.Tensor | None = None, position_ids: torch.Tensor | None = None, attention_mask: torch.Tensor | None = None, inputs_embeds: torch.Tensor | None = None, output_hidden_states: bool | None = None, **kwargs: Any) -> BaseEncoderOutput

fastvideo.models.encoders.lingbot_video.Qwen3ForCausalLM.load_weights

load_weights(weights: Iterable[tuple[str, torch.Tensor]]) -> set[str]

fastvideo.models.encoders.lingbot_video.Qwen3MLP

Qwen3MLP(hidden_size: int, intermediate_size: int, hidden_act: str, quant_config: QuantizationConfig | None = None, bias: bool = False, prefix: str = '')

Qwen3 MLP with SwiGLU activation and tensor parallelism.

fastvideo.models.encoders.lingbot_video.Qwen3MLP.forward

forward(x: torch.Tensor) -> torch.Tensor

fastvideo.models.encoders.lingbot_video.LingBotVideoQwen3VLAttention

LingBotVideoQwen3VLAttention(config: Qwen3TextConfig, hidden_size: int, num_heads: int, num_kv_heads: int, rope_theta: float = 1000000.0, rope_scaling: dict[str, Any] | None = None, max_position_embeddings: int = 40960, quant_config: QuantizationConfig | None = None, bias: bool = False, prefix: str = '')

Qwen3-VL attention with the official masked repeat-K/V SDPA path.

Source
class LingBotVideoQwen3VLAttention(Qwen3Attention):
    """Qwen3-VL attention with the official masked repeat-K/V SDPA path."""

    def _apply_qwen3_vl_rope(
        self,
        positions: torch.Tensor,
        query: torch.Tensor,
        key: torch.Tensor,
    ) -> tuple[torch.Tensor, torch.Tensor]:
        """Apply NeoX RoPE with Qwen3-VL's input-dtype multiply-add ordering."""
        flat_positions = positions.flatten()
        cos_sin = self.rotary_emb.cos_sin_cache.index_select(0, flat_positions)
        cos_half, sin_half = cos_sin.chunk(2, dim=-1)
        cos = torch.cat((cos_half, cos_half), dim=-1).to(query.dtype)
        sin = torch.cat((sin_half, sin_half), dim=-1).to(query.dtype)
        if flat_positions.numel() == query.shape[1]:
            cos = cos.view(1, query.shape[1], 1, self.head_dim)
            sin = sin.view(1, query.shape[1], 1, self.head_dim)
        else:
            cos = cos.view(*query.shape[:2], 1, self.head_dim)
            sin = sin.view(*query.shape[:2], 1, self.head_dim)

        def rotate(tensor: torch.Tensor) -> torch.Tensor:
            first, second = tensor.chunk(2, dim=-1)
            rotated = torch.cat((-second, first), dim=-1)
            return tensor * cos + rotated * sin

        return rotate(query), rotate(key)

    def forward(
        self,
        positions: torch.Tensor,
        hidden_states: torch.Tensor,
        attention_mask: torch.Tensor | None = None,
    ) -> torch.Tensor:
        """Apply fused projections, QK norm, RoPE, and causal grouped attention."""
        qkv, _ = self.qkv_proj(hidden_states)
        query, key, value = qkv.split([self.q_size, self.kv_size, self.kv_size], dim=-1)
        batch_size, sequence_length = query.shape[:2]
        query = query.reshape(batch_size, sequence_length, self.num_heads, self.head_dim)
        key = key.reshape(batch_size, sequence_length, self.num_kv_heads, self.head_dim)
        value = value.reshape(batch_size, sequence_length, self.num_kv_heads, self.head_dim)
        query = self.q_norm(query)
        key = self.k_norm(key)
        query, key = self._apply_qwen3_vl_rope(positions, query, key)
        no_padding = attention_mask is None
        if no_padding:
            attention_output = torch.nn.functional.scaled_dot_product_attention(
                query.transpose(1, 2),
                key.transpose(1, 2),
                value.transpose(1, 2),
                dropout_p=0.0,
                is_causal=sequence_length > 1,
                scale=self.scaling,
                enable_gqa=self.num_heads != self.num_kv_heads,
            ).transpose(1, 2)
        else:
            groups = self.num_heads // self.num_kv_heads
            key = (key[:, :, :, None, :].expand(-1, -1, -1, groups, -1).reshape(batch_size, sequence_length,
                                                                                self.num_heads, self.head_dim))
            value = (value[:, :, :, None, :].expand(-1, -1, -1, groups, -1).reshape(batch_size, sequence_length,
                                                                                    self.num_heads, self.head_dim))
            causal_mask = torch.ones(sequence_length, sequence_length, device=query.device, dtype=torch.bool).tril()
            key_mask = attention_mask.to(device=query.device, dtype=torch.bool)
            sdpa_mask = causal_mask[None, None, :, :] & key_mask[:, None, None, :]
            attention_output = torch.nn.functional.scaled_dot_product_attention(
                query.transpose(1, 2),
                key.transpose(1, 2),
                value.transpose(1, 2),
                attn_mask=sdpa_mask,
                dropout_p=0.0,
                is_causal=False,
                scale=self.scaling,
            ).transpose(1, 2)
        output, _ = self.o_proj(attention_output.reshape(batch_size, sequence_length, -1))
        return output

fastvideo.models.encoders.lingbot_video.LingBotVideoQwen3VLAttention.forward

forward(positions: torch.Tensor, hidden_states: torch.Tensor, attention_mask: torch.Tensor | None = None) -> torch.Tensor

Apply fused projections, QK norm, RoPE, and causal grouped attention.

Source
def forward(
    self,
    positions: torch.Tensor,
    hidden_states: torch.Tensor,
    attention_mask: torch.Tensor | None = None,
) -> torch.Tensor:
    """Apply fused projections, QK norm, RoPE, and causal grouped attention."""
    qkv, _ = self.qkv_proj(hidden_states)
    query, key, value = qkv.split([self.q_size, self.kv_size, self.kv_size], dim=-1)
    batch_size, sequence_length = query.shape[:2]
    query = query.reshape(batch_size, sequence_length, self.num_heads, self.head_dim)
    key = key.reshape(batch_size, sequence_length, self.num_kv_heads, self.head_dim)
    value = value.reshape(batch_size, sequence_length, self.num_kv_heads, self.head_dim)
    query = self.q_norm(query)
    key = self.k_norm(key)
    query, key = self._apply_qwen3_vl_rope(positions, query, key)
    no_padding = attention_mask is None
    if no_padding:
        attention_output = torch.nn.functional.scaled_dot_product_attention(
            query.transpose(1, 2),
            key.transpose(1, 2),
            value.transpose(1, 2),
            dropout_p=0.0,
            is_causal=sequence_length > 1,
            scale=self.scaling,
            enable_gqa=self.num_heads != self.num_kv_heads,
        ).transpose(1, 2)
    else:
        groups = self.num_heads // self.num_kv_heads
        key = (key[:, :, :, None, :].expand(-1, -1, -1, groups, -1).reshape(batch_size, sequence_length,
                                                                            self.num_heads, self.head_dim))
        value = (value[:, :, :, None, :].expand(-1, -1, -1, groups, -1).reshape(batch_size, sequence_length,
                                                                                self.num_heads, self.head_dim))
        causal_mask = torch.ones(sequence_length, sequence_length, device=query.device, dtype=torch.bool).tril()
        key_mask = attention_mask.to(device=query.device, dtype=torch.bool)
        sdpa_mask = causal_mask[None, None, :, :] & key_mask[:, None, None, :]
        attention_output = torch.nn.functional.scaled_dot_product_attention(
            query.transpose(1, 2),
            key.transpose(1, 2),
            value.transpose(1, 2),
            attn_mask=sdpa_mask,
            dropout_p=0.0,
            is_causal=False,
            scale=self.scaling,
        ).transpose(1, 2)
    output, _ = self.o_proj(attention_output.reshape(batch_size, sequence_length, -1))
    return output

fastvideo.models.encoders.lingbot_video.LingBotVideoQwen3VLDecoderLayer

LingBotVideoQwen3VLDecoderLayer(config: Any, prefix: str)

Qwen3-VL decoder layer with explicit official residual rounding order.

Build the final Qwen3-VL attention once to avoid orphan parameters.

Source
class LingBotVideoQwen3VLDecoderLayer(Qwen3DecoderLayer):
    """Qwen3-VL decoder layer with explicit official residual rounding order."""

    def __init__(self, config: Any, prefix: str) -> None:
        """Build the final Qwen3-VL attention once to avoid orphan parameters."""
        nn.Module.__init__(self)
        self.hidden_size = config.hidden_size
        quant_config = getattr(config, "quant_config", None)
        self.self_attn = LingBotVideoQwen3VLAttention(
            config=config,
            hidden_size=self.hidden_size,
            num_heads=config.num_attention_heads,
            num_kv_heads=config.num_key_value_heads,
            rope_theta=config.rope_theta,
            rope_scaling=config.rope_scaling,
            max_position_embeddings=config.max_position_embeddings,
            quant_config=quant_config,
            bias=config.attention_bias,
            prefix=f"{prefix}.self_attn",
        )
        self.mlp = Qwen3MLP(
            hidden_size=self.hidden_size,
            intermediate_size=config.intermediate_size,
            hidden_act=config.hidden_act,
            quant_config=quant_config,
            bias=getattr(config, "mlp_bias", False),
            prefix=f"{prefix}.mlp",
        )
        self.input_layernorm = RMSNorm(self.hidden_size, eps=config.rms_norm_eps)
        self.post_attention_layernorm = RMSNorm(self.hidden_size, eps=config.rms_norm_eps)

    def forward(
        self,
        positions: torch.Tensor,
        hidden_states: torch.Tensor,
        attention_mask: torch.Tensor | None = None,
    ) -> torch.Tensor:
        """Run attention and MLP with each residual sum rounded before normalization."""
        residual = hidden_states
        hidden_states = self.input_layernorm(hidden_states)
        hidden_states = self.self_attn(positions, hidden_states, attention_mask)
        hidden_states = residual + hidden_states
        residual = hidden_states
        hidden_states = self.post_attention_layernorm(hidden_states)
        hidden_states = self.mlp(hidden_states)
        return residual + hidden_states

fastvideo.models.encoders.lingbot_video.LingBotVideoQwen3VLDecoderLayer.forward

forward(positions: torch.Tensor, hidden_states: torch.Tensor, attention_mask: torch.Tensor | None = None) -> torch.Tensor

Run attention and MLP with each residual sum rounded before normalization.

Source
def forward(
    self,
    positions: torch.Tensor,
    hidden_states: torch.Tensor,
    attention_mask: torch.Tensor | None = None,
) -> torch.Tensor:
    """Run attention and MLP with each residual sum rounded before normalization."""
    residual = hidden_states
    hidden_states = self.input_layernorm(hidden_states)
    hidden_states = self.self_attn(positions, hidden_states, attention_mask)
    hidden_states = residual + hidden_states
    residual = hidden_states
    hidden_states = self.post_attention_layernorm(hidden_states)
    hidden_states = self.mlp(hidden_states)
    return residual + hidden_states

fastvideo.models.encoders.lingbot_video.LingBotVideoQwen3VLTextModel

LingBotVideoQwen3VLTextModel(config)

Load the Qwen3-VL language-model subset without its vision tower or LM head.

Construct the exact Qwen3-VL module graph without replacing base layers.

Source
class LingBotVideoQwen3VLTextModel(Qwen3ForCausalLM):
    """Load the Qwen3-VL language-model subset without its vision tower or LM head."""

    supports_hf_from_pretrained = False

    def __init__(self, config) -> None:
        """Construct the exact Qwen3-VL module graph without replacing base layers."""
        TextEncoder.__init__(self, config)
        self.quant_config = getattr(config, "quant_config", None)
        if getattr(config, "lora_config", None) is not None:
            max_loras = getattr(config.lora_config, "max_loras", 1)
            lora_vocab_size = getattr(config.lora_config, "lora_extra_vocab_size", 1)
            lora_vocab = lora_vocab_size * max_loras
        else:
            lora_vocab = 0
        self.vocab_size = config.vocab_size + lora_vocab
        self.org_vocab_size = config.vocab_size
        self.embed_tokens = VocabParallelEmbedding(
            self.vocab_size,
            config.hidden_size,
            org_num_embeddings=config.vocab_size,
            quant_config=self.quant_config,
        )
        self.layers = nn.ModuleList(
            LingBotVideoQwen3VLDecoderLayer(config, prefix=f"{config.prefix}.layers.{index}")
            for index in range(config.num_hidden_layers))
        self.norm = RMSNorm(config.hidden_size, eps=config.rms_norm_eps)

    def forward(
        self,
        input_ids: torch.Tensor | None = None,
        position_ids: torch.Tensor | None = None,
        attention_mask: torch.Tensor | None = None,
        inputs_embeds: torch.Tensor | None = None,
        output_hidden_states: bool | None = None,
        **kwargs: Any,
    ) -> BaseEncoderOutput:
        """Run explicit Qwen3-VL layers and return the requested hidden-state tuple."""
        del kwargs
        output_hidden_states = (output_hidden_states
                                if output_hidden_states is not None else self.config.output_hidden_states)
        if inputs_embeds is None:
            if input_ids is None:
                raise ValueError("input_ids or inputs_embeds is required")
            hidden_states = self.get_input_embeddings(input_ids)
        else:
            hidden_states = inputs_embeds
        if position_ids is None:
            position_ids = torch.arange(hidden_states.shape[1], device=hidden_states.device).unsqueeze(0)
        if attention_mask is not None and bool(attention_mask.to(torch.bool).all()):
            attention_mask = None
        all_hidden_states: tuple[torch.Tensor, ...] | None = () if output_hidden_states else None
        for layer in self.layers:
            if all_hidden_states is not None:
                all_hidden_states += (hidden_states, )
            hidden_states = layer(position_ids, hidden_states, attention_mask)
        hidden_states = self.norm(hidden_states)
        if all_hidden_states is not None:
            all_hidden_states += (hidden_states, )
        return BaseEncoderOutput(
            last_hidden_state=hidden_states,
            hidden_states=all_hidden_states,
        )

    def load_weights(self, weights: Iterable[tuple[str, torch.Tensor]]) -> set[str]:
        """Accept either official compound keys or converted native keys."""
        prefix = "model.language_model."
        language_weights = ((name[len(prefix):] if name.startswith(prefix) else name, tensor)
                            for name, tensor in weights
                            if name.startswith(prefix) or not name.startswith(("model.", "lm_head.")))
        return super().load_weights(language_weights)

fastvideo.models.encoders.lingbot_video.LingBotVideoQwen3VLTextModel.forward

forward(input_ids: torch.Tensor | None = None, position_ids: torch.Tensor | None = None, attention_mask: torch.Tensor | None = None, inputs_embeds: torch.Tensor | None = None, output_hidden_states: bool | None = None, **kwargs: Any) -> BaseEncoderOutput

Run explicit Qwen3-VL layers and return the requested hidden-state tuple.

Source
def forward(
    self,
    input_ids: torch.Tensor | None = None,
    position_ids: torch.Tensor | None = None,
    attention_mask: torch.Tensor | None = None,
    inputs_embeds: torch.Tensor | None = None,
    output_hidden_states: bool | None = None,
    **kwargs: Any,
) -> BaseEncoderOutput:
    """Run explicit Qwen3-VL layers and return the requested hidden-state tuple."""
    del kwargs
    output_hidden_states = (output_hidden_states
                            if output_hidden_states is not None else self.config.output_hidden_states)
    if inputs_embeds is None:
        if input_ids is None:
            raise ValueError("input_ids or inputs_embeds is required")
        hidden_states = self.get_input_embeddings(input_ids)
    else:
        hidden_states = inputs_embeds
    if position_ids is None:
        position_ids = torch.arange(hidden_states.shape[1], device=hidden_states.device).unsqueeze(0)
    if attention_mask is not None and bool(attention_mask.to(torch.bool).all()):
        attention_mask = None
    all_hidden_states: tuple[torch.Tensor, ...] | None = () if output_hidden_states else None
    for layer in self.layers:
        if all_hidden_states is not None:
            all_hidden_states += (hidden_states, )
        hidden_states = layer(position_ids, hidden_states, attention_mask)
    hidden_states = self.norm(hidden_states)
    if all_hidden_states is not None:
        all_hidden_states += (hidden_states, )
    return BaseEncoderOutput(
        last_hidden_state=hidden_states,
        hidden_states=all_hidden_states,
    )

fastvideo.models.encoders.lingbot_video.LingBotVideoQwen3VLTextModel.load_weights

load_weights(weights: Iterable[tuple[str, torch.Tensor]]) -> set[str]

Accept either official compound keys or converted native keys.

Source
def load_weights(self, weights: Iterable[tuple[str, torch.Tensor]]) -> set[str]:
    """Accept either official compound keys or converted native keys."""
    prefix = "model.language_model."
    language_weights = ((name[len(prefix):] if name.startswith(prefix) else name, tensor)
                        for name, tensor in weights
                        if name.startswith(prefix) or not name.startswith(("model.", "lm_head.")))
    return super().load_weights(language_weights)