Skip to content
FastVideo
Esc
↑↓navigate↵open⌘Jpreview
On this page

qwen2_5

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, &quot;num_key_value_groups&quot;):
    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&#91;:, :, :, :key.shape&#91;-2&#93;&#93;
# 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</code></pre></details>

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</code></pre></details>

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(&quot;rope_type&quot;, &quot;default&quot;)
    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(&quot;inv_freq&quot;, 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&#91;None, None, :, None&#93;.float().expand(3, position_ids.shape&#91;1&#93;, -1, 1)
    position_ids_expanded = position_ids&#91;:, :, None, :&#93;.float()  # shape (3, bs, 1, positions)

    device_type = x.device.type if isinstance(x.device.type, str) and x.device.type != &quot;mps&quot; else &quot;cpu&quot;
    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)</code></pre></details>

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 != &quot;mps&quot; else &quot;cpu&quot;
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)</code></pre></details>

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 = &quot;&quot;):
    super().__init__()
    self.gate_up_proj = MergedColumnParallelLinear(
        input_size=config.hidden_size,
        output_sizes=&#91;config.intermediate_size&#93; * 2,
        bias=False,
        quant_config=quant_config,
        prefix=f&quot;&#123;prefix&#125;.gate_up_proj&quot;,
    )
    self.down_proj = RowParallelLinear(
        input_size=config.intermediate_size,
        output_size=config.hidden_size,
        bias=False,
        quant_config=quant_config,
        prefix=f&quot;&#123;prefix&#125;.down_proj&quot;,
    )
    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</code></pre></details>

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 = &quot;&quot;):
    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 &gt;= 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&quot;&#123;prefix&#125;.qkv_proj&quot;,
    )

    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&quot;&#123;prefix&#125;.o_proj&quot;,
    )

    self.layer_type = config.layer_types&#91;layer_idx&#93; if config.layer_types else &quot;full_attention&quot;
    self.sliding_window = config.sliding_window if self.layer_type == &quot;sliding_attention&quot; else None

def forward(
    self,
    hidden_states: torch.Tensor,
    attention_mask: Optional&#91;torch.Tensor&#93; = None,
    position_ids: Optional&#91;torch.LongTensor&#93; = None,
    position_embeddings: Optional&#91;tuple&#91;torch.Tensor, torch.Tensor&#93;&#93; = None,
    output_attentions: bool = False,
) -&gt; Tuple&#91;torch.Tensor, Optional&#91;torch.Tensor&#93;&#93;:
    bsz, q_len, _ = hidden_states.size()

    qkv, _ = self.qkv_proj(hidden_states)
    query_states, key_states, value_states = qkv.split(&#91;self.q_size, self.kv_size, self.kv_size&#93;, 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&#91;&quot;mrope_section&quot;&#93;)

    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)&#91;0&#93;.reshape(bsz, q_len, -1)
    attn_output, _ = self.o_proj(attn_output)

    return attn_output</code></pre></details>

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(&#91;self.q_size, self.kv_size, self.kv_size&#93;, 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&#91;&quot;mrope_section&quot;&#93;)

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)&#91;0&#93;.reshape(bsz, q_len, -1)
attn_output, _ = self.o_proj(attn_output)

return attn_output</code></pre></details>

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 = &quot;&quot;):
    super().__init__()
    self.self_attn = Qwen2_5_VLAttention(config, layer_idx, quant_config=quant_config, prefix=f&quot;&#123;prefix&#125;.self_attn&quot;)
    self.mlp = Qwen2_5_VLMLP(config, quant_config=quant_config, prefix=f&quot;&#123;prefix&#125;.mlp&quot;)
    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&#91;torch.Tensor&#93; = None,
    position_ids: Optional&#91;torch.LongTensor&#93; = None,
    position_embeddings: Optional&#91;tuple&#91;torch.Tensor, torch.Tensor&#93;&#93; = None,
    output_attentions: Optional&#91;bool&#93; = False,
) -&gt; Tuple&#91;torch.FloatTensor, Optional&#91;Tuple&#91;torch.FloatTensor, torch.FloatTensor&#93;&#93;&#93;:

    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</code></pre></details>

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</code></pre></details>

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(&#91;
        Qwen2_5_VLDecoderLayer(config,
                               layer_idx,
                               quant_config=quant_config,
                               prefix=f&quot;&#123;config.prefix&#125;.layers.&#123;layer_idx&#125;&quot;)
        for layer_idx in range(config.num_hidden_layers)
    &#93;)

    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) -&gt; torch.Tensor:
    return self.embed_tokens(input_ids)

def forward(
    self,
    input_ids: Optional&#91;torch.LongTensor&#93; = None,
    position_ids: Optional&#91;torch.LongTensor&#93; = None,
    attention_mask: Optional&#91;torch.Tensor&#93; = None,
    inputs_embeds: Optional&#91;torch.FloatTensor&#93; = None,
    output_hidden_states: Optional&#91;bool&#93; = None,
    **kwargs,
) -&gt; 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&#91;1&#93;
        cache_position = torch.arange(seq_length, device=hidden_states.device)
        position_ids = cache_position.view(1, 1, -1).expand(3, hidden_states.shape&#91;0&#93;, -1)

    mask_kwargs = &#123;
        &quot;batch_size&quot;: hidden_states.shape&#91;0&#93;,
        &quot;cache_position&quot;: cache_position,
        &quot;kv_length&quot;: attention_mask.shape&#91;-1&#93;,
        &quot;kv_offset&quot;: 0,
        &quot;attention_mask&quot;: attention_mask
    &#125;

    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&#91;0&#93;

    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&#91;tuple&#91;str, torch.Tensor&#93;&#93;) -&gt; set&#91;str&#93;:
    params_dict = dict(self.named_parameters())
    loaded_params: set&#91;str&#93; = set()
    for name, loaded_weight in weights:
        if &quot;rotary_emb.inv_freq&quot; 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(&quot;.bias&quot;) and name not in params_dict:
            #     continue
            if name not in params_dict:
                continue
            param = params_dict&#91;name&#93;
            weight_loader = param.weight_loader
            weight_loader(param, loaded_weight, shard_id)
            break
        else:
            # Skip loading extra bias for GPTQ models.
            # if name.endswith(&quot;.bias&quot;) and name not in params_dict:
            #     continue
            if name not in params_dict:
                continue

            param = params_dict&#91;name&#93;
            weight_loader = getattr(param, &quot;weight_loader&quot;, default_weight_loader)
            weight_loader(param, loaded_weight)
        loaded_params.add(name)
    return loaded_params</code></pre></details>

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&#91;1&#93;
    cache_position = torch.arange(seq_length, device=hidden_states.device)
    position_ids = cache_position.view(1, 1, -1).expand(3, hidden_states.shape&#91;0&#93;, -1)

mask_kwargs = &#123;
    &quot;batch_size&quot;: hidden_states.shape&#91;0&#93;,
    &quot;cache_position&quot;: cache_position,
    &quot;kv_length&quot;: attention_mask.shape&#91;-1&#93;,
    &quot;kv_offset&quot;: 0,
    &quot;attention_mask&quot;: attention_mask
&#125;

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&#91;0&#93;

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,
)</code></pre></details>

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&#91;name&#93;
        weight_loader = getattr(param, &quot;weight_loader&quot;, default_weight_loader)
        weight_loader(param, loaded_weight)
    loaded_params.add(name)
return loaded_params</code></pre></details>

Was this page helpful?