Skip to content

lingbot_video

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

Classes

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 = '')

Bases: Qwen3Attention

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

Source code in fastvideo/models/encoders/qwen3.py 
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
def __init__(
    self,
    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 = "",
) -> None:
    super().__init__()
    self.hidden_size = hidden_size
    tp_size = get_tp_world_size()
    self.total_num_heads = num_heads
    assert self.total_num_heads % tp_size == 0
    self.num_heads = self.total_num_heads // tp_size
    self.total_num_kv_heads = num_kv_heads
    if self.total_num_kv_heads >= tp_size:
        assert self.total_num_kv_heads % tp_size == 0
    else:
        assert tp_size % self.total_num_kv_heads == 0
    self.num_kv_heads = max(1, self.total_num_kv_heads // tp_size)

    self.head_dim = getattr(
        config, "head_dim", self.hidden_size // self.total_num_heads
    )
    self.rotary_dim = self.head_dim
    self.q_size = self.num_heads * self.head_dim
    self.kv_size = self.num_kv_heads * self.head_dim
    self.scaling = self.head_dim**-0.5
    self.rope_theta = rope_theta
    self.max_position_embeddings = max_position_embeddings

    self.qkv_proj = QKVParallelLinear(
        hidden_size=hidden_size,
        head_size=self.head_dim,
        total_num_heads=self.total_num_heads,
        total_num_kv_heads=self.total_num_kv_heads,
        bias=bias,
        quant_config=quant_config,
        prefix=f"{prefix}.qkv_proj",
    )
    self.o_proj = RowParallelLinear(
        input_size=self.total_num_heads * self.head_dim,
        output_size=hidden_size,
        bias=bias,
        quant_config=quant_config,
        prefix=f"{prefix}.o_proj",
    )

    rms_norm_eps = getattr(config, "rms_norm_eps", 1e-6)
    self.q_norm = RMSNorm(self.head_dim, eps=rms_norm_eps)
    self.k_norm = RMSNorm(self.head_dim, eps=rms_norm_eps)

    self.rotary_emb = get_rope(
        self.head_dim,
        rotary_dim=self.rotary_dim,
        max_position=max_position_embeddings,
        base=int(rope_theta),
        rope_scaling=rope_scaling,
        is_neox_style=True,
    )

    self.attn = LocalAttention(
        self.num_heads,
        self.head_dim,
        self.num_kv_heads,
        softmax_scale=self.scaling,
        causal=True,
        supported_attention_backends=config._supported_attention_backends,
    )

Methods:

fastvideo.models.encoders.lingbot_video.LingBotVideoQwen3VLAttention.forward 
forward(positions: Tensor, hidden_states: Tensor, attention_mask: Tensor | None = None) -> Tensor

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

Source code in fastvideo/models/encoders/lingbot_video.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
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)

Bases: Qwen3DecoderLayer

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

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

Source code in fastvideo/models/encoders/lingbot_video.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
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)

Methods:

fastvideo.models.encoders.lingbot_video.LingBotVideoQwen3VLDecoderLayer.forward 
forward(positions: Tensor, hidden_states: Tensor, attention_mask: Tensor | None = None) -> Tensor

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

Source code in fastvideo/models/encoders/lingbot_video.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
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)

Bases: Qwen3ForCausalLM

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 code in fastvideo/models/encoders/lingbot_video.py 
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
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)

Methods:

fastvideo.models.encoders.lingbot_video.LingBotVideoQwen3VLTextModel.forward 
forward(input_ids: Tensor | None = None, position_ids: Tensor | None = None, attention_mask: Tensor | None = None, inputs_embeds: 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 code in fastvideo/models/encoders/lingbot_video.py 
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
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, Tensor]]) -> set[str]

Accept either official compound keys or converted native keys.

Source code in fastvideo/models/encoders/lingbot_video.py 
212
213
214
215
216
217
218
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)