`fastvideo.models.encoders.qwen2_5.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.qwen2_5.Qwen2_5_VLConfig`

Qwen2_5_VLConfig(arch_config: TextEncoderArchConfig = Qwen2_5_VLArchConfig(), prefix: str = 'qwen2_5_vl', quant_config: QuantizationConfig | None = None, lora_config: Any | None = None, is_chat_model: bool = True, treat_empty_as_dot: bool = True, *, _resolved_attention_backend: AttentionBackendEnum | None = None, chat_template_enable_thinking: bool = False)

`fastvideo.models.encoders.qwen2_5.get_tp_rank`

get_tp_rank() -> int

Return my rank for the tensor model parallel group.

`fastvideo.models.encoders.qwen2_5.get_tp_world_size`

get_tp_world_size() -> int

Return world size for the tensor model parallel group.

`fastvideo.models.encoders.qwen2_5.get_act_fn`

get_act_fn(act_fn_name: str) -> nn.Module

Get an activation function by name.

`fastvideo.models.encoders.qwen2_5.SiluAndMul`

SiluAndMul()

An activation function for SwiGLU.

The function computes x -> silu(x[:d]) * x[d:] where d = x.shape[-1] // 2.

Shapes

x: (num_tokens, 2 * d) or (batch_size, seq_len, 2 * d) return: (num_tokens, d) or (batch_size, seq_len, d)

`fastvideo.models.encoders.qwen2_5.SiluAndMul.forward_native`

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

PyTorch-native implementation equivalent to forward().

`fastvideo.models.encoders.qwen2_5.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.qwen2_5.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.qwen2_5.RMSNorm.extra_repr`

extra_repr() -> str

`fastvideo.models.encoders.qwen2_5.MergedColumnParallelLinear`

MergedColumnParallelLinear(input_size: int, output_sizes: list[int], bias: bool = True, gather_output: bool = False, skip_bias_add: bool = False, params_dtype: torch.dtype | None = None, quant_config: QuantizationConfig | None = None, prefix: str = '')

Packed linear layers with column parallelism.

Similar to ColumnParallelLinear, but the weight matrix is concatenated along the output dimension. When the weight matrix is loaded, the different partitions are sharded separately.

Parameters

input_size: int: input dimension of the linear layer.
output_sizes: list[int]: list of output dimensions of the linear layer.
bias: bool: If true, add bias.
gather_output: bool: If true, call all-gather on output and make the output available to all GPUs, otherwise, every GPU will have its own output.
skip_bias_add: bool: This was added to enable performance optimizations where bias can be fused with other element-wise operations. we skip adding bias but instead return it.
params_dtype: torch.dtype | None: Data type for the parameters.
quant_config: QuantizationConfig | None: Quantization configure.
prefix: str: The name of the layer in the state dict, including all parents (e.g. model.layers.0.qkv_proj)

`fastvideo.models.encoders.qwen2_5.MergedColumnParallelLinear.weight_loader`

weight_loader(param: Parameter, loaded_weight: torch.Tensor, loaded_shard_id: int | None = None) -> None

`fastvideo.models.encoders.qwen2_5.MergedColumnParallelLinear.weight_loader_v2`

weight_loader_v2(param: BasevLLMParameter, loaded_weight: torch.Tensor, loaded_shard_id: int | None = None) -> None

`fastvideo.models.encoders.qwen2_5.QKVParallelLinear`

QKVParallelLinear(hidden_size: int, head_size: int, total_num_heads: int, total_num_kv_heads: int | None = None, bias: bool = True, skip_bias_add: bool = False, params_dtype: torch.dtype | None = None, quant_config: QuantizationConfig | None = None, prefix: str = '')

Linear layers for the attention's QKV transformation.

Linear layers for the linear transformation of the query, key, and value vectors in the attention layer. The weight matrix is concatenated along the output dimension. The layer is parallelized along the head dimension. When the number of key/value heads is smaller than the number of query heads (e.g., multi-query/grouped-query attention), the key/value head may be replicated while the query heads are partitioned.

Parameters

hidden_size: int: input hidden state size of the transformer.
head_size: int: size of each attention head.
total_num_heads: int: total number of attention query heads.
total_num_kv_heads: int | None: total number of attention key/value heads. If None, assume total_num_kv_heads = total_num_heads.
bias: bool: If true, add bias.
skip_bias_add: bool: This was added to enable performance optimizations where bias can be fused with other element-wise operations. we skip adding bias but instead return it.
params_dtype: torch.dtype | None: Data type for the parameters.
quant_config: QuantizationConfig | None: Quantization configure.
prefix: str: The name of the layer in the state dict, including all parents (e.g. model.layers.0.qkv_proj)

`fastvideo.models.encoders.qwen2_5.QKVParallelLinear.weight_loader_v2`

weight_loader_v2(param: BasevLLMParameter, loaded_weight: torch.Tensor, loaded_shard_id: str | None = None)

`fastvideo.models.encoders.qwen2_5.QKVParallelLinear.weight_loader`

weight_loader(param: Parameter, loaded_weight: torch.Tensor, loaded_shard_id: str | None = None)

`fastvideo.models.encoders.qwen2_5.RowParallelLinear`

RowParallelLinear(input_size: int, output_size: int, bias: bool = True, input_is_parallel: bool = True, skip_bias_add: bool = False, params_dtype: torch.dtype | None = None, reduce_results: bool = True, quant_config: QuantizationConfig | None = None, prefix: str = '')

Linear layer with row parallelism.

The linear layer is defined as Y = XA + b. A is parallelized along its first dimension and X along its second dimension as: - - | A_1 | | . | A = | . | X = [X_1, ..., X_p] | . | | A_p | - - Arguments: input_size: first dimension of matrix A. output_size: second dimension of matrix A. bias: If true, add bias. Note that bias is not parallelized. input_is_parallel: If true, we assume that the input is already split across the GPUs and we do not split again. skip_bias_add: This was added to enable performance optimization where bias can be fused with other element-wise operations. We skip adding bias but instead return it. params_dtype: Data type for the parameters. quant_config: Quantization configure.

`fastvideo.models.encoders.qwen2_5.RowParallelLinear.weight_loader`

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

`fastvideo.models.encoders.qwen2_5.RowParallelLinear.weight_loader_v2`

weight_loader_v2(param: BasevLLMParameter, loaded_weight: torch.Tensor)

`fastvideo.models.encoders.qwen2_5.RowParallelLinear.forward`

forward(input_) -> tuple[torch.Tensor, Parameter | None]

`fastvideo.models.encoders.qwen2_5.RowParallelLinear.extra_repr`

extra_repr() -> str

`fastvideo.models.encoders.qwen2_5.QuantizationConfig`

QuantizationConfig()

Base class for quantization configs.

`fastvideo.models.encoders.qwen2_5.QuantizationConfig.get_name`

get_name() -> QuantizationMethods

Name of the quantization method.

`fastvideo.models.encoders.qwen2_5.QuantizationConfig.get_supported_act_dtypes`

get_supported_act_dtypes() -> list[torch.dtype]

List of supported activation dtypes.

`fastvideo.models.encoders.qwen2_5.QuantizationConfig.get_min_capability`

get_min_capability() -> int

Minimum GPU capability to support the quantization method.

E.g., 70 for Volta, 75 for Turing, 80 for Ampere. This requirement is due to the custom CUDA kernels used by the quantization method.

`fastvideo.models.encoders.qwen2_5.QuantizationConfig.get_config_filenames`

get_config_filenames() -> list[str]

List of filenames to search for in the model directory.

`fastvideo.models.encoders.qwen2_5.QuantizationConfig.from_config`

from_config(config: dict[str, Any]) -> QuantizationConfig

Create a config class from the model's quantization config.

`fastvideo.models.encoders.qwen2_5.QuantizationConfig.override_quantization_method`

override_quantization_method(hf_quant_cfg, user_quant) -> QuantizationMethods | None

Detects if this quantization method can support a given checkpoint format by overriding the user specified quantization method -- this method should only be overwritten by subclasses in exceptional circumstances

`fastvideo.models.encoders.qwen2_5.QuantizationConfig.get_from_keys`

get_from_keys(config: dict[str, Any], keys: list[str]) -> Any

Get a value from the model's quantization config.

`fastvideo.models.encoders.qwen2_5.QuantizationConfig.get_from_keys_or`

get_from_keys_or(config: dict[str, Any], keys: list[str], default: Any) -> Any

Get a optional value from the model's quantization config.

`fastvideo.models.encoders.qwen2_5.QuantizationConfig.get_quant_method`

get_quant_method(layer: torch.nn.Module, prefix: str) -> QuantizeMethodBase | None

Get the quantize method to use for the quantized layer.

Parameters

layer: torch.nn.Module: The layer for the quant method.
prefix: str: The full name of the layer in the state dict

Returns: The quantize method. None if the given layer doesn't support quant method.

`fastvideo.models.encoders.qwen2_5.QuantizationConfig.validate_runtime`

validate_runtime(device: torch.device) -> None

Fail fast on a device the scheme cannot execute on.

Called once by loaders that select a serialized checkpoint scheme, before any weight is read. The default accepts every device; schemes with kernel or capability requirements override it.

`fastvideo.models.encoders.qwen2_5.QuantizationConfig.get_cache_scale`

get_cache_scale(name: str) -> str | None

`fastvideo.models.encoders.qwen2_5.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.qwen2_5.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.qwen2_5.VocabParallelEmbedding.weight_loader`

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

`fastvideo.models.encoders.qwen2_5.VocabParallelEmbedding.forward`

forward(input_)

`fastvideo.models.encoders.qwen2_5.VocabParallelEmbedding.extra_repr`

extra_repr() -> str

`fastvideo.models.encoders.qwen2_5.TextEncoder`

TextEncoder(config: TextEncoderConfig)

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

`fastvideo.models.encoders.qwen2_5.TextEncoder.forward`

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

`fastvideo.models.encoders.qwen2_5.default_weight_loader`

default_weight_loader(param: torch.Tensor, loaded_weight: torch.Tensor) -> None

Default weight loader.

`fastvideo.models.encoders.qwen2_5.sdpa_mask`

sdpa_mask(batch_size: int, cache_position: torch.Tensor, kv_length: int, kv_offset: int = 0, mask_function: Callable = causal_mask_function, attention_mask: Optional[torch.Tensor] = None, local_size: Optional[int] = None, allow_is_causal_skip: bool = True, allow_is_bidirectional_skip: bool = False, allow_torch_fix: bool = True, use_vmap: bool = False, **kwargs) -> Optional[torch.Tensor]

Create a 4D boolean mask of shape (batch_size, 1, query_length, kv_length) where a value of True indicates that the element should take part in the attention computation, and False that it should not. This function can only be used with torch>=2.5, as the context manager is otherwise not available.

Parameters

batch_size: `int`: The batch size of the input sequence.
cache_position: `torch.Tensor`: A tensor of shape (query_length,) indicating the current indices of the input sequence elements.
kv_length: `int`: The size that the key and value states will have during the attention computation.
kv_offset: `int`: An optional offset to indicate at which first position the key and values states will refer to.
mask_function: `Callable`: The mask factory function describing the mask pattern.
attention_mask: `torch.Tensor`: The 2D attention mask corresponding to padded tokens of shape (batch_size, number_of_seen_tokens+q_length)
local_size: `int`: The size of the local attention, if we do not use full attention. This is used only if allow_is_causal_skip=True to try to skip mask creation if possible.
allow_is_causal_skip: `bool`: Whether to allow to return None for the mask under conditions where we can use the is_causal argument in torch.sdpa instead. Default to True.
allow_is_bidirectional_skip: `bool`: Whether to allow to return None for the mask under conditions where we do not have to add any bias, i.e. full attention without any padding. Default to False.
allow_torch_fix: `bool`: Whether to update the mask in case a query is not attending to any tokens, to solve a bug in torch's older versions. We need an arg to skip it when using eager. By default True.
use_vmap: `bool`: Whether to use vmap during the mask construction or not. Allows powerful custom patterns that may not be index-based (for the cost of speed performance). By default False.

Creating a simple causal mask:

To create the following causal mask:

0 ■ ⬚ ⬚ ⬚ ⬚
1 ■ ■ ⬚ ⬚ ⬚
2 ■ ■ ■ ⬚ ⬚
3 ■ ■ ■ ■ ⬚
4 ■ ■ ■ ■ ■

You can do

>>> sdpa_mask(batch_size=1, cache_position=torch.arange(5), kv_length=5)
>>> tensor([[[[ True, False, False, False, False],
              [ True,  True, False, False, False],
              [ True,  True,  True, False, False],
              [ True,  True,  True,  True, False],
              [ True,  True,  True,  True,  True]]]])

Creating a sliding window mask:

To create the following sliding window mask (sliding_window=3):

0 ■ ⬚ ⬚ ⬚ ⬚
1 ■ ■ ⬚ ⬚ ⬚
2 ■ ■ ■ ⬚ ⬚
3 ⬚ ■ ■ ■ ⬚
4 ⬚ ⬚ ■ ■ ■

You can do

>>> sdpa_mask(batch_size=1, cache_position=torch.arange(5), kv_length=5, mask_function=sliding_window_causal_mask_function(3))
>>> tensor([[[[ True, False, False, False, False],
              [ True,  True, False, False, False],
              [ True,  True,  True, False, False],
              [False,  True,  True,  True, False],
              [False, False,  True,  True,  True]]]])

Creating a chunked attention mask

To create the following chunked attention mask (chunk_size=3):

0 ■ ⬚ ⬚ ⬚ ⬚
1 ■ ■ ⬚ ⬚ ⬚
2 ■ ■ ■ ⬚ ⬚
3 ⬚ ⬚ ⬚ ■ ⬚
4 ⬚ ⬚ ⬚ ■ ■

You can do

>>> sdpa_mask(batch_size=1, cache_position=torch.arange(5), kv_length=5, mask_function=chunked_causal_mask_function(3, torch.zeros(1, dtype=int)))
>>> tensor([[[[ True, False, False, False, False],
            [ True,  True, False, False, False],
            [ True,  True,  True, False, False],
            [False, False, False,  True, False],
            [False, False, False,  True,  True]]]])

`fastvideo.models.encoders.qwen2_5.init_logger`

init_logger(name: str) -> _FastvideoLogger

The main purpose of this function is to ensure that loggers are retrieved in such a way that we can be sure the root fastvideo logger has already been configured.

`fastvideo.models.encoders.qwen2_5.repeat_kv`

repeat_kv(hidden_states: torch.Tensor, n_rep: int) -> torch.Tensor

This is the equivalent of torch.repeat_interleave(x, dim=1, repeats=n_rep). The hidden states go from (batch, num_key_value_heads, seqlen, head_dim) to (batch, num_attention_heads, seqlen, head_dim)

Source

def repeat_kv(hidden_states: torch.Tensor, n_rep: int) -> torch.Tensor:
    """
    This is the equivalent of torch.repeat_interleave(x, dim=1, repeats=n_rep). The hidden states go from (batch,
    num_key_value_heads, seqlen, head_dim) to (batch, num_attention_heads, seqlen, head_dim)
    """
    batch, num_key_value_heads, slen, head_dim = hidden_states.shape
    if n_rep == 1:
        return hidden_states
    hidden_states = hidden_states[:, :, None, :, :].expand(batch, num_key_value_heads, n_rep, slen, head_dim)
    return hidden_states.reshape(batch, num_key_value_heads * n_rep, slen, head_dim)

`fastvideo.models.encoders.qwen2_5.sdpa_attention_forward`

sdpa_attention_forward(module: torch.nn.Module, query: torch.Tensor, key: torch.Tensor, value: torch.Tensor, attention_mask: Optional[torch.Tensor], dropout: float = 0.0, scaling: Optional[float] = None, is_causal: Optional[bool] = None, **kwargs) -> tuple[torch.Tensor, None]

Source

def sdpa_attention_forward(
    module: torch.nn.Module,
    query: torch.Tensor,
    key: torch.Tensor,
    value: torch.Tensor,
    attention_mask: Optional[torch.Tensor],
    dropout: float = 0.0,
    scaling: Optional[float] = None,
    is_causal: Optional[bool] = None,
    **kwargs,
) -> tuple[torch.Tensor, None]:
    if kwargs.get("output_attentions", False) or kwargs.get("head_mask") is not None:
        logger.warning_once("`sdpa` attention does not support `output_attentions=True` or `head_mask`."
                            " Please set your attention to `eager` if you want any of these features.")

    if hasattr(module, "num_key_value_groups"):
        key = repeat_kv(key, module.num_key_value_groups)
        value = repeat_kv(value, module.num_key_value_groups)

    if attention_mask is not None and attention_mask.ndim == 4:
        attention_mask = attention_mask[:, :, :, :key.shape[-2]]
    # If attention_mask is not None, convert it to boolean type
    if attention_mask is not None and attention_mask.dtype != torch.bool:
        attention_mask = attention_mask.bool()

    attn_output = torch.nn.functional.scaled_dot_product_attention(
        query,
        key,
        value,
        attn_mask=attention_mask,
        dropout_p=dropout,
        scale=scaling,
        is_causal=is_causal,
    )
    attn_output = attn_output.transpose(1, 2).contiguous()

    return attn_output, None

`fastvideo.models.encoders.qwen2_5.rotate_half`

rotate_half(x)

Rotates half the hidden dims of the input.

Source

def rotate_half(x):
    """Rotates half the hidden dims of the input."""
    x1 = x[..., :x.shape[-1] // 2]
    x2 = x[..., x.shape[-1] // 2:]
    return torch.cat((-x2, x1), dim=-1)

`fastvideo.models.encoders.qwen2_5.apply_multimodal_rotary_pos_emb`

apply_multimodal_rotary_pos_emb(q, k, cos, sin, mrope_section, unsqueeze_dim = 1)

Source

def apply_multimodal_rotary_pos_emb(q, k, cos, sin, mrope_section, unsqueeze_dim=1):
    mrope_section = [s * 2 for s in mrope_section]
    cos = torch.cat([m[i % 3] for i, m in enumerate(cos.split(mrope_section, dim=-1))], dim=-1).unsqueeze(unsqueeze_dim)
    sin = torch.cat([m[i % 3] for i, m in enumerate(sin.split(mrope_section, dim=-1))], dim=-1).unsqueeze(unsqueeze_dim)

    q_embed = (q * cos) + (rotate_half(q) * sin)
    k_embed = (k * cos) + (rotate_half(k) * sin)
    return q_embed, k_embed

`fastvideo.models.encoders.qwen2_5.Qwen2_5_VLRotaryEmbedding`

Qwen2_5_VLRotaryEmbedding(config: Qwen2_5_VLConfig, device = None)

Source

class Qwen2_5_VLRotaryEmbedding(nn.Module):

    def __init__(self, config: Qwen2_5_VLConfig, device=None):
        super().__init__()
        self.max_seq_len_cached = config.max_position_embeddings
        self.original_max_seq_len = config.max_position_embeddings
        self.config = config

        self.rope_type = config.rope_scaling.get("rope_type", "default")
        self.base = config.rope_theta

        # Simplified initialization
        head_dim = config.hidden_size // config.num_attention_heads
        dim = head_dim
        self.attention_scaling = 1.0

        inv_freq = 1.0 / (self.base**(torch.arange(0, dim, 2, dtype=torch.int64).float().to(device) / dim))
        self.register_buffer("inv_freq", inv_freq, persistent=False)
        self.original_inv_freq = inv_freq

    def forward(self, x, position_ids):
        # In contrast to other models, Qwen2_5_VL has different position ids for the grids
        # So we expand the inv_freq to shape (3, ...)
        inv_freq_expanded = self.inv_freq[None, None, :, None].float().expand(3, position_ids.shape[1], -1, 1)
        position_ids_expanded = position_ids[:, :, None, :].float()  # shape (3, bs, 1, positions)

        device_type = x.device.type if isinstance(x.device.type, str) and x.device.type != "mps" else "cpu"
        with torch.autocast(device_type=device_type, enabled=False):  # Force float32
            freqs = (inv_freq_expanded.float() @ position_ids_expanded.float()).transpose(2, 3)
            emb = torch.cat((freqs, freqs), dim=-1)
            cos = emb.cos() * self.attention_scaling
            sin = emb.sin() * self.attention_scaling

        return cos.to(dtype=x.dtype), sin.to(dtype=x.dtype)

`fastvideo.models.encoders.qwen2_5.Qwen2_5_VLRotaryEmbedding.forward`

forward(x, position_ids)

Source

def forward(self, x, position_ids):
    # In contrast to other models, Qwen2_5_VL has different position ids for the grids
    # So we expand the inv_freq to shape (3, ...)
    inv_freq_expanded = self.inv_freq[None, None, :, None].float().expand(3, position_ids.shape[1], -1, 1)
    position_ids_expanded = position_ids[:, :, None, :].float()  # shape (3, bs, 1, positions)

    device_type = x.device.type if isinstance(x.device.type, str) and x.device.type != "mps" else "cpu"
    with torch.autocast(device_type=device_type, enabled=False):  # Force float32
        freqs = (inv_freq_expanded.float() @ position_ids_expanded.float()).transpose(2, 3)
        emb = torch.cat((freqs, freqs), dim=-1)
        cos = emb.cos() * self.attention_scaling
        sin = emb.sin() * self.attention_scaling

    return cos.to(dtype=x.dtype), sin.to(dtype=x.dtype)

`fastvideo.models.encoders.qwen2_5.Qwen2_5_VLMLP`

Qwen2_5_VLMLP(config: Qwen2_5_VLConfig, quant_config: QuantizationConfig | None = None, prefix: str = '')

Source

class Qwen2_5_VLMLP(nn.Module):

    def __init__(self, config: Qwen2_5_VLConfig, quant_config: QuantizationConfig | None = None, prefix: str = ""):
        super().__init__()
        self.gate_up_proj = MergedColumnParallelLinear(
            input_size=config.hidden_size,
            output_sizes=[config.intermediate_size] * 2,
            bias=False,
            quant_config=quant_config,
            prefix=f"{prefix}.gate_up_proj",
        )
        self.down_proj = RowParallelLinear(
            input_size=config.intermediate_size,
            output_size=config.hidden_size,
            bias=False,
            quant_config=quant_config,
            prefix=f"{prefix}.down_proj",
        )
        self.act_fn = SiluAndMul()

    def forward(self, x):
        x, _ = self.gate_up_proj(x)
        x = self.act_fn(x)
        x, _ = self.down_proj(x)
        return x

`fastvideo.models.encoders.qwen2_5.Qwen2_5_VLMLP.forward`

forward(x)

Source

def forward(self, x):
    x, _ = self.gate_up_proj(x)
    x = self.act_fn(x)
    x, _ = self.down_proj(x)
    return x

`fastvideo.models.encoders.qwen2_5.Qwen2_5_VLAttention`

Qwen2_5_VLAttention(config: Qwen2_5_VLConfig, layer_idx: int, quant_config: QuantizationConfig | None = None, prefix: str = '')

Source

class Qwen2_5_VLAttention(nn.Module):

    def __init__(self,
                 config: Qwen2_5_VLConfig,
                 layer_idx: int,
                 quant_config: QuantizationConfig | None = None,
                 prefix: str = ""):
        super().__init__()
        self.config = config
        self.layer_idx = layer_idx
        self.hidden_size = config.hidden_size
        self.num_heads = config.num_attention_heads
        self.head_dim = self.hidden_size // self.num_heads
        self.num_key_value_heads = config.num_key_value_heads
        self.num_key_value_groups = self.num_heads // self.num_key_value_heads

        tp_size = get_tp_world_size()
        self.total_num_heads = self.num_heads
        assert self.total_num_heads % tp_size == 0
        self.num_heads = self.total_num_heads // tp_size

        self.total_num_kv_heads = self.num_key_value_heads
        if self.total_num_kv_heads >= tp_size:
            assert self.total_num_kv_heads % tp_size == 0
            self.num_kv_heads = self.total_num_kv_heads // tp_size
        else:
            assert tp_size % self.total_num_kv_heads == 0
            self.num_kv_heads = 1

        self.num_kv_heads = max(1, self.total_num_kv_heads // tp_size)

        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.qkv_proj = QKVParallelLinear(
            hidden_size=self.hidden_size,
            head_size=self.head_dim,
            total_num_heads=self.total_num_heads,
            total_num_kv_heads=self.total_num_kv_heads,
            bias=True,
            quant_config=quant_config,
            prefix=f"{prefix}.qkv_proj",
        )

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

        self.layer_type = config.layer_types[layer_idx] if config.layer_types else "full_attention"
        self.sliding_window = config.sliding_window if self.layer_type == "sliding_attention" else None

    def forward(
        self,
        hidden_states: torch.Tensor,
        attention_mask: Optional[torch.Tensor] = None,
        position_ids: Optional[torch.LongTensor] = None,
        position_embeddings: Optional[tuple[torch.Tensor, torch.Tensor]] = None,
        output_attentions: bool = False,
    ) -> Tuple[torch.Tensor, Optional[torch.Tensor]]:
        bsz, q_len, _ = hidden_states.size()

        qkv, _ = self.qkv_proj(hidden_states)
        query_states, key_states, value_states = qkv.split([self.q_size, self.kv_size, self.kv_size], dim=-1)

        query_states = query_states.view(bsz, q_len, self.num_heads, self.head_dim).transpose(1, 2)
        key_states = key_states.view(bsz, q_len, self.num_kv_heads, self.head_dim).transpose(1, 2)
        value_states = value_states.view(bsz, q_len, self.num_kv_heads, self.head_dim).transpose(1, 2)

        cos, sin = position_embeddings
        query_states, key_states = apply_multimodal_rotary_pos_emb(query_states, key_states, cos, sin,
                                                                   self.config.rope_scaling["mrope_section"])

        attn_output = sdpa_attention_forward(self,
                                             query_states,
                                             key_states,
                                             value_states,
                                             attention_mask,
                                             dropout=self.config.attention_dropout,
                                             scaling=self.scaling,
                                             is_causal=False)[0].reshape(bsz, q_len, -1)
        attn_output, _ = self.o_proj(attn_output)

        return attn_output

`fastvideo.models.encoders.qwen2_5.Qwen2_5_VLAttention.forward`

forward(hidden_states: torch.Tensor, attention_mask: Optional[torch.Tensor] = None, position_ids: Optional[torch.LongTensor] = None, position_embeddings: Optional[tuple[torch.Tensor, torch.Tensor]] = None, output_attentions: bool = False) -> Tuple[torch.Tensor, Optional[torch.Tensor]]

Source

def forward(
    self,
    hidden_states: torch.Tensor,
    attention_mask: Optional[torch.Tensor] = None,
    position_ids: Optional[torch.LongTensor] = None,
    position_embeddings: Optional[tuple[torch.Tensor, torch.Tensor]] = None,
    output_attentions: bool = False,
) -> Tuple[torch.Tensor, Optional[torch.Tensor]]:
    bsz, q_len, _ = hidden_states.size()

    qkv, _ = self.qkv_proj(hidden_states)
    query_states, key_states, value_states = qkv.split([self.q_size, self.kv_size, self.kv_size], dim=-1)

    query_states = query_states.view(bsz, q_len, self.num_heads, self.head_dim).transpose(1, 2)
    key_states = key_states.view(bsz, q_len, self.num_kv_heads, self.head_dim).transpose(1, 2)
    value_states = value_states.view(bsz, q_len, self.num_kv_heads, self.head_dim).transpose(1, 2)

    cos, sin = position_embeddings
    query_states, key_states = apply_multimodal_rotary_pos_emb(query_states, key_states, cos, sin,
                                                               self.config.rope_scaling["mrope_section"])

    attn_output = sdpa_attention_forward(self,
                                         query_states,
                                         key_states,
                                         value_states,
                                         attention_mask,
                                         dropout=self.config.attention_dropout,
                                         scaling=self.scaling,
                                         is_causal=False)[0].reshape(bsz, q_len, -1)
    attn_output, _ = self.o_proj(attn_output)

    return attn_output

`fastvideo.models.encoders.qwen2_5.Qwen2_5_VLDecoderLayer`

Qwen2_5_VLDecoderLayer(config: Qwen2_5_VLConfig, layer_idx: int, quant_config: QuantizationConfig | None = None, prefix: str = '')

Source

class Qwen2_5_VLDecoderLayer(nn.Module):

    def __init__(self,
                 config: Qwen2_5_VLConfig,
                 layer_idx: int,
                 quant_config: QuantizationConfig | None = None,
                 prefix: str = ""):
        super().__init__()
        self.self_attn = Qwen2_5_VLAttention(config, layer_idx, quant_config=quant_config, prefix=f"{prefix}.self_attn")
        self.mlp = Qwen2_5_VLMLP(config, quant_config=quant_config, prefix=f"{prefix}.mlp")
        self.input_layernorm = RMSNorm(config.hidden_size, eps=config.rms_norm_eps)
        self.post_attention_layernorm = RMSNorm(config.hidden_size, eps=config.rms_norm_eps)

    def forward(
        self,
        hidden_states: torch.Tensor,
        attention_mask: Optional[torch.Tensor] = None,
        position_ids: Optional[torch.LongTensor] = None,
        position_embeddings: Optional[tuple[torch.Tensor, torch.Tensor]] = None,
        output_attentions: Optional[bool] = False,
    ) -> Tuple[torch.FloatTensor, Optional[Tuple[torch.FloatTensor, torch.FloatTensor]]]:

        residual = hidden_states
        hidden_states = self.input_layernorm(hidden_states)

        hidden_states = self.self_attn(
            hidden_states=hidden_states,
            attention_mask=attention_mask,
            position_ids=position_ids,
            position_embeddings=position_embeddings,
            output_attentions=output_attentions,
        )
        hidden_states = residual + hidden_states

        residual = hidden_states
        hidden_states = self.post_attention_layernorm(hidden_states)
        hidden_states = self.mlp(hidden_states)
        hidden_states = residual + hidden_states

        outputs = (hidden_states, )

        return outputs

`fastvideo.models.encoders.qwen2_5.Qwen2_5_VLDecoderLayer.forward`

forward(hidden_states: torch.Tensor, attention_mask: Optional[torch.Tensor] = None, position_ids: Optional[torch.LongTensor] = None, position_embeddings: Optional[tuple[torch.Tensor, torch.Tensor]] = None, output_attentions: Optional[bool] = False) -> Tuple[torch.FloatTensor, Optional[Tuple[torch.FloatTensor, torch.FloatTensor]]]

Source

def forward(
    self,
    hidden_states: torch.Tensor,
    attention_mask: Optional[torch.Tensor] = None,
    position_ids: Optional[torch.LongTensor] = None,
    position_embeddings: Optional[tuple[torch.Tensor, torch.Tensor]] = None,
    output_attentions: Optional[bool] = False,
) -> Tuple[torch.FloatTensor, Optional[Tuple[torch.FloatTensor, torch.FloatTensor]]]:

    residual = hidden_states
    hidden_states = self.input_layernorm(hidden_states)

    hidden_states = self.self_attn(
        hidden_states=hidden_states,
        attention_mask=attention_mask,
        position_ids=position_ids,
        position_embeddings=position_embeddings,
        output_attentions=output_attentions,
    )
    hidden_states = residual + hidden_states

    residual = hidden_states
    hidden_states = self.post_attention_layernorm(hidden_states)
    hidden_states = self.mlp(hidden_states)
    hidden_states = residual + hidden_states

    outputs = (hidden_states, )

    return outputs

`fastvideo.models.encoders.qwen2_5.Qwen2_5_VLTextModel`

Qwen2_5_VLTextModel(config: Qwen2_5_VLConfig)

Source

class Qwen2_5_VLTextModel(TextEncoder):

    def __init__(self, config: Qwen2_5_VLConfig):
        super().__init__(config)

        quant_config = None

        self.embed_tokens = VocabParallelEmbedding(config.vocab_size,
                                                   config.hidden_size,
                                                   org_num_embeddings=config.vocab_size)

        self.layers = nn.ModuleList([
            Qwen2_5_VLDecoderLayer(config,
                                   layer_idx,
                                   quant_config=quant_config,
                                   prefix=f"{config.prefix}.layers.{layer_idx}")
            for layer_idx in range(config.num_hidden_layers)
        ])

        self.norm = RMSNorm(config.hidden_size, eps=config.rms_norm_eps)
        self.rotary_emb = Qwen2_5_VLRotaryEmbedding(config)

    def get_input_embeddings(self, input_ids: torch.Tensor) -> torch.Tensor:
        return self.embed_tokens(input_ids)

    def forward(
        self,
        input_ids: Optional[torch.LongTensor] = None,
        position_ids: Optional[torch.LongTensor] = None,
        attention_mask: Optional[torch.Tensor] = None,
        inputs_embeds: Optional[torch.FloatTensor] = None,
        output_hidden_states: Optional[bool] = None,
        **kwargs,
    ) -> BaseEncoderOutput:
        output_hidden_states = (output_hidden_states
                                if output_hidden_states is not None else self.config.output_hidden_states)

        if inputs_embeds is None:
            inputs_embeds = self.get_input_embeddings(input_ids)

        hidden_states = inputs_embeds

        if position_ids is None:
            seq_length = hidden_states.shape[1]
            cache_position = torch.arange(seq_length, device=hidden_states.device)
            position_ids = cache_position.view(1, 1, -1).expand(3, hidden_states.shape[0], -1)

        mask_kwargs = {
            "batch_size": hidden_states.shape[0],
            "cache_position": cache_position,
            "kv_length": attention_mask.shape[-1],
            "kv_offset": 0,
            "attention_mask": attention_mask
        }

        position_embeddings = self.rotary_emb(hidden_states, position_ids)

        all_hidden_states = () if output_hidden_states else None

        for decoder_layer in self.layers:
            if output_hidden_states:
                all_hidden_states += (hidden_states, )

            layer_outputs = decoder_layer(
                hidden_states,
                attention_mask=sdpa_mask(**mask_kwargs),
                position_ids=position_ids,
                position_embeddings=position_embeddings,
                output_attentions=False,
            )
            hidden_states = layer_outputs[0]

        hidden_states = self.norm(hidden_states)

        if output_hidden_states:
            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]:
        params_dict = dict(self.named_parameters())
        loaded_params: set[str] = set()
        for name, loaded_weight in weights:
            if "rotary_emb.inv_freq" in name:
                continue
            for param_name, weight_name, shard_id in self.config.stacked_params_mapping:
                if weight_name not in name:
                    continue
                name = name.replace(weight_name, param_name)
                # Skip loading extra bias for GPTQ models.
                # if name.endswith(".bias") and name not in params_dict:
                #     continue
                if name not in params_dict:
                    continue
                param = params_dict[name]
                weight_loader = param.weight_loader
                weight_loader(param, loaded_weight, shard_id)
                break
            else:
                # Skip loading extra bias for GPTQ models.
                # if name.endswith(".bias") and name not in params_dict:
                #     continue
                if name not in params_dict:
                    continue

                param = params_dict[name]
                weight_loader = getattr(param, "weight_loader", default_weight_loader)
                weight_loader(param, loaded_weight)
            loaded_params.add(name)
        return loaded_params

`fastvideo.models.encoders.qwen2_5.Qwen2_5_VLTextModel.get_input_embeddings`

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

Source

def get_input_embeddings(self, input_ids: torch.Tensor) -> torch.Tensor:
    return self.embed_tokens(input_ids)

`fastvideo.models.encoders.qwen2_5.Qwen2_5_VLTextModel.forward`

forward(input_ids: Optional[torch.LongTensor] = None, position_ids: Optional[torch.LongTensor] = None, attention_mask: Optional[torch.Tensor] = None, inputs_embeds: Optional[torch.FloatTensor] = None, output_hidden_states: Optional[bool] = None, **kwargs) -> BaseEncoderOutput

Source

def forward(
    self,
    input_ids: Optional[torch.LongTensor] = None,
    position_ids: Optional[torch.LongTensor] = None,
    attention_mask: Optional[torch.Tensor] = None,
    inputs_embeds: Optional[torch.FloatTensor] = None,
    output_hidden_states: Optional[bool] = None,
    **kwargs,
) -> BaseEncoderOutput:
    output_hidden_states = (output_hidden_states
                            if output_hidden_states is not None else self.config.output_hidden_states)

    if inputs_embeds is None:
        inputs_embeds = self.get_input_embeddings(input_ids)

    hidden_states = inputs_embeds

    if position_ids is None:
        seq_length = hidden_states.shape[1]
        cache_position = torch.arange(seq_length, device=hidden_states.device)
        position_ids = cache_position.view(1, 1, -1).expand(3, hidden_states.shape[0], -1)

    mask_kwargs = {
        "batch_size": hidden_states.shape[0],
        "cache_position": cache_position,
        "kv_length": attention_mask.shape[-1],
        "kv_offset": 0,
        "attention_mask": attention_mask
    }

    position_embeddings = self.rotary_emb(hidden_states, position_ids)

    all_hidden_states = () if output_hidden_states else None

    for decoder_layer in self.layers:
        if output_hidden_states:
            all_hidden_states += (hidden_states, )

        layer_outputs = decoder_layer(
            hidden_states,
            attention_mask=sdpa_mask(**mask_kwargs),
            position_ids=position_ids,
            position_embeddings=position_embeddings,
            output_attentions=False,
        )
        hidden_states = layer_outputs[0]

    hidden_states = self.norm(hidden_states)

    if output_hidden_states:
        all_hidden_states += (hidden_states, )

    return BaseEncoderOutput(
        last_hidden_state=hidden_states,
        hidden_states=all_hidden_states,
    )

`fastvideo.models.encoders.qwen2_5.Qwen2_5_VLTextModel.load_weights`

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

Source

def load_weights(self, weights: Iterable[tuple[str, torch.Tensor]]) -> set[str]:
    params_dict = dict(self.named_parameters())
    loaded_params: set[str] = set()
    for name, loaded_weight in weights:
        if "rotary_emb.inv_freq" in name:
            continue
        for param_name, weight_name, shard_id in self.config.stacked_params_mapping:
            if weight_name not in name:
                continue
            name = name.replace(weight_name, param_name)
            # Skip loading extra bias for GPTQ models.
            # if name.endswith(".bias") and name not in params_dict:
            #     continue
            if name not in params_dict:
                continue
            param = params_dict[name]
            weight_loader = param.weight_loader
            weight_loader(param, loaded_weight, shard_id)
            break
        else:
            # Skip loading extra bias for GPTQ models.
            # if name.endswith(".bias") and name not in params_dict:
            #     continue
            if name not in params_dict:
                continue

            param = params_dict[name]
            weight_loader = getattr(param, "weight_loader", default_weight_loader)
            weight_loader(param, loaded_weight)
        loaded_params.add(name)
    return loaded_params