--- title: clip ---

Minimal implementation of CLIPVisionModel intended to be only used within a vision language model.

fastvideo.models.encoders.clip.LocalAttention

LocalAttention(num_heads: int, head_size: int, num_kv_heads: int | None = None, softmax_scale: float | None = None, causal: bool = False, supported_attention_backends: tuple[AttentionBackendEnum, ...] | None = None, default_backend: AttentionBackendEnum | None = None, **extra_impl_args)

Attention layer.

fastvideo.models.encoders.clip.LocalAttention.forward

forward(q: torch.Tensor, k: torch.Tensor, v: torch.Tensor, freqs_cis: tuple[torch.Tensor, torch.Tensor] | None = None) -> torch.Tensor

Apply local attention between query, key and value tensors.

Parameters

q: torch.Tensor

Query tensor of shape [batch_size, seq_len, num_heads, head_dim]

k: torch.Tensor

Key tensor of shape [batch_size, seq_len, num_heads, head_dim]

v: torch.Tensor

Value tensor of shape [batch_size, seq_len, num_heads, head_dim]

Returns

torch.Tensor

torch.Tensor: Output tensor after local attention

fastvideo.models.encoders.clip.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.clip.CLIPTextConfig

CLIPTextConfig(arch_config: TextEncoderArchConfig = CLIPTextArchConfig(), prefix: str = 'clip', quant_config: QuantizationConfig | None = None, lora_config: Any | None = None, is_chat_model: bool = False, treat_empty_as_dot: bool = False, num_hidden_layers_override: int | None = None, require_post_norm: bool | None = None, enable_scale: bool = True, is_causal: bool = True, *, _resolved_attention_backend: AttentionBackendEnum | None = None, chat_template_enable_thinking: bool = False)

fastvideo.models.encoders.clip.CLIPVisionConfig

CLIPVisionConfig(arch_config: ImageEncoderArchConfig = CLIPVisionArchConfig(), prefix: str = 'clip', quant_config: QuantizationConfig | None = None, lora_config: Any | None = None, num_hidden_layers_override: int | None = 31, require_post_norm: bool | None = None, enable_scale: bool = False, is_causal: bool = False, *, _resolved_attention_backend: AttentionBackendEnum | None = None)

fastvideo.models.encoders.clip.divide

divide(numerator: int, denominator: int) -> int

Ensure that numerator is divisible by the denominator and return the division value.

fastvideo.models.encoders.clip.get_tp_world_size

get_tp_world_size() -> int

Return world size for the tensor model parallel group.

fastvideo.models.encoders.clip.get_act_fn

get_act_fn(act_fn_name: str) -> nn.Module

Get an activation function by name.

fastvideo.models.encoders.clip.ColumnParallelLinear

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

Linear layer with column parallelism.

The linear layer is defined as Y = XA + b. A is parallelized along its second dimension as A = [A_1, ..., A_p].

Parameters

input_size: int

first dimension of matrix A.

output_size: int

second dimension of matrix A.

bias: bool

If true, add bias.

gather_output: bool

If true, call all-gather on output and make Y available to all GPUs, otherwise, every GPU will have its output which is Y_i = XA_i

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.

output_sizes: list[int] | None

list of output sizes packed into one output, like for QKV the list would be size 3.

prefix: str

The name of the layer in the state dict, including all parents (e.g. model.layers.0.qkv_proj)

fastvideo.models.encoders.clip.ColumnParallelLinear.weight_loader

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

fastvideo.models.encoders.clip.ColumnParallelLinear.weight_loader_v2

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

fastvideo.models.encoders.clip.ColumnParallelLinear.forward

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

fastvideo.models.encoders.clip.ColumnParallelLinear.extra_repr

extra_repr() -> str

fastvideo.models.encoders.clip.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.clip.QKVParallelLinear.weight_loader_v2

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

fastvideo.models.encoders.clip.QKVParallelLinear.weight_loader

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

fastvideo.models.encoders.clip.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.clip.RowParallelLinear.weight_loader

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

fastvideo.models.encoders.clip.RowParallelLinear.weight_loader_v2

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

fastvideo.models.encoders.clip.RowParallelLinear.forward

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

fastvideo.models.encoders.clip.RowParallelLinear.extra_repr

extra_repr() -> str

fastvideo.models.encoders.clip.QuantizationConfig

QuantizationConfig()

Base class for quantization configs.

fastvideo.models.encoders.clip.QuantizationConfig.get_name

get_name() -> QuantizationMethods

Name of the quantization method.

fastvideo.models.encoders.clip.QuantizationConfig.get_supported_act_dtypes

get_supported_act_dtypes() -> list[torch.dtype]

List of supported activation dtypes.

fastvideo.models.encoders.clip.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.clip.QuantizationConfig.get_config_filenames

get_config_filenames() -> list[str]

List of filenames to search for in the model directory.

fastvideo.models.encoders.clip.QuantizationConfig.from_config

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

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

fastvideo.models.encoders.clip.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.clip.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.clip.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.clip.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.clip.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.clip.QuantizationConfig.get_cache_scale

get_cache_scale(name: str) -> str | None

fastvideo.models.encoders.clip.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.clip.ImageEncoder

ImageEncoder(config: ImageEncoderConfig)

fastvideo.models.encoders.clip.ImageEncoder.forward

forward(pixel_values: torch.Tensor, **kwargs) -> BaseEncoderOutput

fastvideo.models.encoders.clip.TextEncoder

TextEncoder(config: TextEncoderConfig)

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

fastvideo.models.encoders.clip.TextEncoder.forward

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

fastvideo.models.encoders.clip.resolve_visual_encoder_outputs

resolve_visual_encoder_outputs(encoder_outputs: torch.Tensor | list[torch.Tensor], feature_sample_layers: list[int] | None, post_layer_norm: torch.nn.LayerNorm | None, max_possible_layers: int) -> torch.Tensor

Given the outputs a visual encoder module that may correspond to the output of the last layer, or a list of hidden states to be stacked, handle post normalization and resolve it into a single output tensor.

Parameters

encoder_outputs: torch.Tensor | list[torch.Tensor]

Output of encoder's last layer or all hidden states.

feature_sample_layers: list[int] | None

Optional layer indices to grab from the encoder outputs; if provided, encoder outputs must be a list.

post_layer_norm: torch.nn.LayerNorm | None

Post norm to apply to the output of the encoder.

max_possible_layers: int

Total layers in the fully loaded visual encoder.

fastvideo.models.encoders.clip.default_weight_loader

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

Default weight loader.

fastvideo.models.encoders.clip.CLIPVisionEmbeddings

CLIPVisionEmbeddings(config: CLIPVisionConfig)
Source
class CLIPVisionEmbeddings(nn.Module):

    def __init__(self, config: CLIPVisionConfig):
        super().__init__()
        self.config = config
        self.embed_dim = config.hidden_size
        self.image_size = config.image_size
        self.patch_size = config.patch_size
        assert self.image_size % self.patch_size == 0

        self.class_embedding = nn.Parameter(torch.randn(self.embed_dim))

        self.patch_embedding = nn.Conv2d(
            in_channels=config.num_channels,
            out_channels=self.embed_dim,
            kernel_size=self.patch_size,
            stride=self.patch_size,
            bias=False,
        )

        self.num_patches = (self.image_size // self.patch_size)**2
        self.num_positions = self.num_patches + 1
        self.position_embedding = nn.Embedding(self.num_positions, self.embed_dim)
        self.register_buffer("position_ids", torch.arange(self.num_positions).expand((1, -1)), persistent=False)

    def forward(self, pixel_values: torch.Tensor) -> torch.Tensor:
        batch_size = pixel_values.shape[0]
        target_dtype = self.patch_embedding.weight.dtype
        patch_embeds = self.patch_embedding(pixel_values.to(dtype=target_dtype))  # shape = [*, width, grid, grid]
        patch_embeds = patch_embeds.flatten(2).transpose(1, 2)

        class_embeds = self.class_embedding.expand(batch_size, 1, -1)
        embeddings = torch.cat([class_embeds, patch_embeds], dim=1)
        embeddings = embeddings + self.position_embedding(self.position_ids)

        return embeddings

fastvideo.models.encoders.clip.CLIPVisionEmbeddings.forward

forward(pixel_values: torch.Tensor) -> torch.Tensor
Source
def forward(self, pixel_values: torch.Tensor) -> torch.Tensor:
    batch_size = pixel_values.shape[0]
    target_dtype = self.patch_embedding.weight.dtype
    patch_embeds = self.patch_embedding(pixel_values.to(dtype=target_dtype))  # shape = [*, width, grid, grid]
    patch_embeds = patch_embeds.flatten(2).transpose(1, 2)

    class_embeds = self.class_embedding.expand(batch_size, 1, -1)
    embeddings = torch.cat([class_embeds, patch_embeds], dim=1)
    embeddings = embeddings + self.position_embedding(self.position_ids)

    return embeddings

fastvideo.models.encoders.clip.CLIPTextEmbeddings

CLIPTextEmbeddings(config: CLIPTextConfig)
Source
class CLIPTextEmbeddings(nn.Module):

    def __init__(self, config: CLIPTextConfig):
        super().__init__()
        self.config = config
        embed_dim = config.hidden_size

        self.token_embedding = nn.Embedding(config.vocab_size, embed_dim)
        self.position_embedding = nn.Embedding(config.max_position_embeddings, embed_dim)

        # position_ids (1, len position emb) is contiguous in memory and exported when serialized
        self.register_buffer("position_ids",
                             torch.arange(config.max_position_embeddings).expand((1, -1)),
                             persistent=False)

    def forward(
        self,
        input_ids: torch.LongTensor | None = None,
        position_ids: torch.LongTensor | None = None,
        inputs_embeds: torch.FloatTensor | None = None,
    ) -> torch.Tensor:
        if input_ids is not None:
            seq_length = input_ids.shape[-1]
        elif inputs_embeds is not None:
            seq_length = inputs_embeds.shape[-2]
        else:
            raise ValueError("Either input_ids or inputs_embeds must be provided.")

        max_position_embedding = self.position_embedding.weight.shape[0]

        if seq_length > max_position_embedding:
            raise ValueError(f"Sequence length must be less than max_position_embeddings (got `sequence length`: "
                             f"{seq_length} and max_position_embeddings: {max_position_embedding}")

        if position_ids is None:
            position_ids = self.position_ids[:, :seq_length]

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

        position_embeddings = self.position_embedding(position_ids)
        embeddings = inputs_embeds + position_embeddings

        return embeddings

fastvideo.models.encoders.clip.CLIPTextEmbeddings.forward

forward(input_ids: torch.LongTensor | None = None, position_ids: torch.LongTensor | None = None, inputs_embeds: torch.FloatTensor | None = None) -> torch.Tensor
Source
def forward(
    self,
    input_ids: torch.LongTensor | None = None,
    position_ids: torch.LongTensor | None = None,
    inputs_embeds: torch.FloatTensor | None = None,
) -> torch.Tensor:
    if input_ids is not None:
        seq_length = input_ids.shape[-1]
    elif inputs_embeds is not None:
        seq_length = inputs_embeds.shape[-2]
    else:
        raise ValueError("Either input_ids or inputs_embeds must be provided.")

    max_position_embedding = self.position_embedding.weight.shape[0]

    if seq_length > max_position_embedding:
        raise ValueError(f"Sequence length must be less than max_position_embeddings (got `sequence length`: "
                         f"{seq_length} and max_position_embeddings: {max_position_embedding}")

    if position_ids is None:
        position_ids = self.position_ids[:, :seq_length]

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

    position_embeddings = self.position_embedding(position_ids)
    embeddings = inputs_embeds + position_embeddings

    return embeddings

fastvideo.models.encoders.clip.CLIPAttention

CLIPAttention(config: CLIPVisionConfig | CLIPTextConfig, quant_config: QuantizationConfig | None = None, prefix: str = '')

Multi-headed attention from 'Attention Is All You Need' paper

Source
class CLIPAttention(nn.Module):
    """Multi-headed attention from 'Attention Is All You Need' paper"""

    def __init__(
        self,
        config: CLIPVisionConfig | CLIPTextConfig,
        quant_config: QuantizationConfig | None = None,
        prefix: str = "",
    ):
        super().__init__()
        self.config = config
        self.embed_dim = config.hidden_size
        self.num_heads = config.num_attention_heads
        self.head_dim = self.embed_dim // self.num_heads
        if self.head_dim * self.num_heads != self.embed_dim:
            raise ValueError("embed_dim must be divisible by num_heads "
                             f"(got `embed_dim`: {self.embed_dim} and `num_heads`:"
                             f" {self.num_heads}).")
        self.scale = self.head_dim**-0.5 if config.enable_scale else None
        self.dropout = config.attention_dropout

        self.qkv_proj = QKVParallelLinear(
            hidden_size=self.embed_dim,
            head_size=self.head_dim,
            total_num_heads=self.num_heads,
            quant_config=quant_config,
            prefix=f"{prefix}.qkv_proj",
        )

        self.out_proj = RowParallelLinear(
            input_size=self.embed_dim,
            output_size=self.embed_dim,
            quant_config=quant_config,
            prefix=f"{prefix}.out_proj",
        )

        self.tp_size = get_tp_world_size()
        self.num_heads_per_partition = divide(self.num_heads, self.tp_size)

        self.attn = LocalAttention(self.num_heads_per_partition,
                                   self.head_dim,
                                   self.num_heads_per_partition,
                                   softmax_scale=self.scale,
                                   causal=config.is_causal,
                                   supported_attention_backends=config._supported_attention_backends)

    def _shape(self, tensor: torch.Tensor, seq_len: int, bsz: int):
        return tensor.view(bsz, seq_len, self.num_heads, self.head_dim).transpose(1, 2).contiguous()

    def forward(
        self,
        hidden_states: torch.Tensor,
    ):
        """Input shape: Batch x Time x Channel"""

        qkv_states, _ = self.qkv_proj(hidden_states)
        query_states, key_states, value_states = qkv_states.chunk(3, dim=-1)
        # use flash_attn_func
        query_states = query_states.reshape(query_states.shape[0], query_states.shape[1], self.num_heads_per_partition,
                                            self.head_dim)
        key_states = key_states.reshape(key_states.shape[0], key_states.shape[1], self.num_heads_per_partition,
                                        self.head_dim)
        value_states = value_states.reshape(value_states.shape[0], value_states.shape[1], self.num_heads_per_partition,
                                            self.head_dim)
        attn_output = self.attn(query_states, key_states, value_states)
        attn_output = attn_output.reshape(attn_output.shape[0], attn_output.shape[1],
                                          self.num_heads_per_partition * self.head_dim)
        attn_output, _ = self.out_proj(attn_output)

        return attn_output, None

fastvideo.models.encoders.clip.CLIPAttention.forward

forward(hidden_states: torch.Tensor)

Input shape: Batch x Time x Channel

Source
def forward(
    self,
    hidden_states: torch.Tensor,
):
    """Input shape: Batch x Time x Channel"""

    qkv_states, _ = self.qkv_proj(hidden_states)
    query_states, key_states, value_states = qkv_states.chunk(3, dim=-1)
    # use flash_attn_func
    query_states = query_states.reshape(query_states.shape[0], query_states.shape[1], self.num_heads_per_partition,
                                        self.head_dim)
    key_states = key_states.reshape(key_states.shape[0], key_states.shape[1], self.num_heads_per_partition,
                                    self.head_dim)
    value_states = value_states.reshape(value_states.shape[0], value_states.shape[1], self.num_heads_per_partition,
                                        self.head_dim)
    attn_output = self.attn(query_states, key_states, value_states)
    attn_output = attn_output.reshape(attn_output.shape[0], attn_output.shape[1],
                                      self.num_heads_per_partition * self.head_dim)
    attn_output, _ = self.out_proj(attn_output)

    return attn_output, None

fastvideo.models.encoders.clip.CLIPMLP

CLIPMLP(config: CLIPVisionConfig | CLIPTextConfig, quant_config: QuantizationConfig | None = None, prefix: str = '')
Source
class CLIPMLP(nn.Module):

    def __init__(
        self,
        config: CLIPVisionConfig | CLIPTextConfig,
        quant_config: QuantizationConfig | None = None,
        prefix: str = "",
    ) -> None:
        super().__init__()
        self.config = config
        self.activation_fn = get_act_fn(config.hidden_act)
        self.fc1 = ColumnParallelLinear(config.hidden_size,
                                        config.intermediate_size,
                                        bias=True,
                                        quant_config=quant_config,
                                        prefix=f"{prefix}.fc1")
        self.fc2 = RowParallelLinear(config.intermediate_size,
                                     config.hidden_size,
                                     bias=True,
                                     quant_config=quant_config,
                                     prefix=f"{prefix}.fc2")

    def forward(self, hidden_states: torch.Tensor) -> torch.Tensor:
        hidden_states, _ = self.fc1(hidden_states)
        hidden_states = self.activation_fn(hidden_states)
        hidden_states, _ = self.fc2(hidden_states)

        return hidden_states

fastvideo.models.encoders.clip.CLIPMLP.forward

forward(hidden_states: torch.Tensor) -> torch.Tensor
Source
def forward(self, hidden_states: torch.Tensor) -> torch.Tensor:
    hidden_states, _ = self.fc1(hidden_states)
    hidden_states = self.activation_fn(hidden_states)
    hidden_states, _ = self.fc2(hidden_states)

    return hidden_states

fastvideo.models.encoders.clip.CLIPEncoderLayer

CLIPEncoderLayer(config: CLIPTextConfig | CLIPVisionConfig, quant_config: QuantizationConfig | None = None, prefix: str = '')
Source
class CLIPEncoderLayer(nn.Module):

    def __init__(
        self,
        config: CLIPTextConfig | CLIPVisionConfig,
        quant_config: QuantizationConfig | None = None,
        prefix: str = "",
    ) -> None:
        super().__init__()
        self.self_attn = CLIPAttention(
            config,
            quant_config=quant_config,
            prefix=f"{prefix}.self_attn",
        )
        self.layer_norm1 = nn.LayerNorm(config.hidden_size, eps=config.layer_norm_eps)
        self.mlp = CLIPMLP(config, quant_config=quant_config, prefix=f"{prefix}.mlp")
        self.layer_norm2 = nn.LayerNorm(config.hidden_size, eps=config.layer_norm_eps)

    def forward(self, hidden_states: torch.Tensor) -> torch.Tensor:

        residual = hidden_states

        hidden_states = self.layer_norm1(hidden_states)
        hidden_states, _ = self.self_attn(hidden_states=hidden_states)
        hidden_states = residual + hidden_states

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

        return hidden_states

fastvideo.models.encoders.clip.CLIPEncoderLayer.forward

forward(hidden_states: torch.Tensor) -> torch.Tensor
Source
def forward(self, hidden_states: torch.Tensor) -> torch.Tensor:

    residual = hidden_states

    hidden_states = self.layer_norm1(hidden_states)
    hidden_states, _ = self.self_attn(hidden_states=hidden_states)
    hidden_states = residual + hidden_states

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

    return hidden_states

fastvideo.models.encoders.clip.CLIPEncoder

CLIPEncoder(config: CLIPVisionConfig | CLIPTextConfig, quant_config: QuantizationConfig | None = None, num_hidden_layers_override: int | None = None, prefix: str = '')

Transformer encoder consisting of config.num_hidden_layers self attention layers. Each layer is a [CLIPEncoderLayer].

Parameters

config: CLIPVisionConfig | CLIPTextConfig

CLIPConfig

Source
class CLIPEncoder(nn.Module):
    """
    Transformer encoder consisting of `config.num_hidden_layers` self
    attention layers. Each layer is a [`CLIPEncoderLayer`].

    Args:
        config: CLIPConfig
    """

    def __init__(
        self,
        config: CLIPVisionConfig | CLIPTextConfig,
        quant_config: QuantizationConfig | None = None,
        num_hidden_layers_override: int | None = None,
        prefix: str = "",
    ) -> None:
        super().__init__()

        self.config = config

        if num_hidden_layers_override is None:
            num_hidden_layers = config.num_hidden_layers
        else:
            num_hidden_layers = num_hidden_layers_override
        self.layers = nn.ModuleList([
            CLIPEncoderLayer(config=config, quant_config=quant_config, prefix=f"{prefix}.layers.{layer_idx}")
            for layer_idx in range(num_hidden_layers)
        ])

    def forward(self, inputs_embeds: torch.Tensor, return_all_hidden_states: bool) -> torch.Tensor | list[torch.Tensor]:
        hidden_states_pool = [inputs_embeds]
        hidden_states = inputs_embeds

        for encoder_layer in self.layers:
            hidden_states = encoder_layer(hidden_states)
            if return_all_hidden_states:
                hidden_states_pool.append(hidden_states)
        # If we have multiple feature sample layers, we return all hidden
        # states in order and grab the ones we need by index.
        if return_all_hidden_states:
            return hidden_states_pool
        return [hidden_states]

fastvideo.models.encoders.clip.CLIPEncoder.forward

forward(inputs_embeds: torch.Tensor, return_all_hidden_states: bool) -> torch.Tensor | list[torch.Tensor]
Source
def forward(self, inputs_embeds: torch.Tensor, return_all_hidden_states: bool) -> torch.Tensor | list[torch.Tensor]:
    hidden_states_pool = [inputs_embeds]
    hidden_states = inputs_embeds

    for encoder_layer in self.layers:
        hidden_states = encoder_layer(hidden_states)
        if return_all_hidden_states:
            hidden_states_pool.append(hidden_states)
    # If we have multiple feature sample layers, we return all hidden
    # states in order and grab the ones we need by index.
    if return_all_hidden_states:
        return hidden_states_pool
    return [hidden_states]

fastvideo.models.encoders.clip.CLIPTextTransformer

CLIPTextTransformer(config: CLIPTextConfig, quant_config: QuantizationConfig | None = None, num_hidden_layers_override: int | None = None, prefix: str = '')
Source
class CLIPTextTransformer(nn.Module):

    def __init__(self,
                 config: CLIPTextConfig,
                 quant_config: QuantizationConfig | None = None,
                 num_hidden_layers_override: int | None = None,
                 prefix: str = ""):
        super().__init__()
        self.config = config
        embed_dim = config.hidden_size

        self.embeddings = CLIPTextEmbeddings(config)

        self.encoder = CLIPEncoder(config,
                                   quant_config=quant_config,
                                   num_hidden_layers_override=num_hidden_layers_override,
                                   prefix=prefix)

        self.final_layer_norm = nn.LayerNorm(embed_dim, eps=config.layer_norm_eps)

        # For `pooled_output` computation
        self.eos_token_id = config.eos_token_id

    def forward(
        self,
        input_ids: torch.Tensor | 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,
    ) -> BaseEncoderOutput:
        r"""
        Returns:

        """
        output_hidden_states = (output_hidden_states
                                if output_hidden_states is not None else self.config.output_hidden_states)

        if input_ids is None:
            raise ValueError("You have to specify input_ids")

        input_shape = input_ids.size()
        input_ids = input_ids.view(-1, input_shape[-1])

        hidden_states = self.embeddings(input_ids=input_ids, position_ids=position_ids)

        # CLIP's text model uses causal mask, prepare it here.
        # https://github.com/openai/CLIP/blob/cfcffb90e69f37bf2ff1e988237a0fbe41f33c04/clip/model.py#L324
        # causal_attention_mask = _create_4d_causal_attention_mask(
        #     input_shape, hidden_states.dtype, device=hidden_states.device
        # )

        # # expand attention_mask
        # if attention_mask is not None and not self._use_flash_attention_2:
        #     raise NotImplementedError("attention_mask is not supported for CLIPTextTransformer")
        #     # [bsz, seq_len] -> [bsz, 1, tgt_seq_len, src_seq_len]
        #     attention_mask = _prepare_4d_attention_mask(attention_mask, hidden_states.dtype)

        encoder_outputs = self.encoder(
            inputs_embeds=hidden_states,
            # attention_mask=attention_mask,
            # causal_attention_mask=causal_attention_mask,
            # output_attentions=output_attentions,
            return_all_hidden_states=output_hidden_states,
            # return_dict=return_dict,
        )

        last_hidden_state = encoder_outputs[-1]
        last_hidden_state = self.final_layer_norm(last_hidden_state)

        if self.eos_token_id == 2:
            # The `eos_token_id` was incorrect before PR #24773: Let's keep what have been done here.
            # A CLIP model with such `eos_token_id` in the config can't work correctly with extra new tokens added
            # ------------------------------------------------------------
            # text_embeds.shape = [batch_size, sequence_length, transformer.width]
            # take features from the eot embedding (eot_token is the highest number in each sequence)
            # casting to torch.int for onnx compatibility: argmax doesn't support int64 inputs with opset 14
            pooled_output = last_hidden_state[
                torch.arange(last_hidden_state.shape[0], device=last_hidden_state.device),
                input_ids.to(dtype=torch.int, device=last_hidden_state.device).argmax(dim=-1),
            ]
        else:
            # The config gets updated `eos_token_id` from PR #24773 (so the use of exta new tokens is possible)
            pooled_output = last_hidden_state[
                torch.arange(last_hidden_state.shape[0], device=last_hidden_state.device),
                # We need to get the first position of `eos_token_id` value (`pad_token_ids` might equal to `eos_token_id`)
                # Note: we assume each sequence (along batch dim.) contains an  `eos_token_id` (e.g. prepared by the tokenizer)
                (input_ids.to(dtype=torch.int, device=last_hidden_state.device) == self.eos_token_id).int().argmax(
                    dim=-1),
            ]

        return BaseEncoderOutput(
            last_hidden_state=last_hidden_state,
            pooler_output=pooled_output,
            hidden_states=encoder_outputs,
            # attentions=encoder_outputs.attentions,
        )

fastvideo.models.encoders.clip.CLIPTextTransformer.forward

forward(input_ids: torch.Tensor | 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) -> BaseEncoderOutput

Returns:

Source
def forward(
    self,
    input_ids: torch.Tensor | 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,
) -> BaseEncoderOutput:
    r"""
    Returns:

    """
    output_hidden_states = (output_hidden_states
                            if output_hidden_states is not None else self.config.output_hidden_states)

    if input_ids is None:
        raise ValueError("You have to specify input_ids")

    input_shape = input_ids.size()
    input_ids = input_ids.view(-1, input_shape[-1])

    hidden_states = self.embeddings(input_ids=input_ids, position_ids=position_ids)

    # CLIP's text model uses causal mask, prepare it here.
    # https://github.com/openai/CLIP/blob/cfcffb90e69f37bf2ff1e988237a0fbe41f33c04/clip/model.py#L324
    # causal_attention_mask = _create_4d_causal_attention_mask(
    #     input_shape, hidden_states.dtype, device=hidden_states.device
    # )

    # # expand attention_mask
    # if attention_mask is not None and not self._use_flash_attention_2:
    #     raise NotImplementedError("attention_mask is not supported for CLIPTextTransformer")
    #     # [bsz, seq_len] -> [bsz, 1, tgt_seq_len, src_seq_len]
    #     attention_mask = _prepare_4d_attention_mask(attention_mask, hidden_states.dtype)

    encoder_outputs = self.encoder(
        inputs_embeds=hidden_states,
        # attention_mask=attention_mask,
        # causal_attention_mask=causal_attention_mask,
        # output_attentions=output_attentions,
        return_all_hidden_states=output_hidden_states,
        # return_dict=return_dict,
    )

    last_hidden_state = encoder_outputs[-1]
    last_hidden_state = self.final_layer_norm(last_hidden_state)

    if self.eos_token_id == 2:
        # The `eos_token_id` was incorrect before PR #24773: Let's keep what have been done here.
        # A CLIP model with such `eos_token_id` in the config can't work correctly with extra new tokens added
        # ------------------------------------------------------------
        # text_embeds.shape = [batch_size, sequence_length, transformer.width]
        # take features from the eot embedding (eot_token is the highest number in each sequence)
        # casting to torch.int for onnx compatibility: argmax doesn't support int64 inputs with opset 14
        pooled_output = last_hidden_state[
            torch.arange(last_hidden_state.shape[0], device=last_hidden_state.device),
            input_ids.to(dtype=torch.int, device=last_hidden_state.device).argmax(dim=-1),
        ]
    else:
        # The config gets updated `eos_token_id` from PR #24773 (so the use of exta new tokens is possible)
        pooled_output = last_hidden_state[
            torch.arange(last_hidden_state.shape[0], device=last_hidden_state.device),
            # We need to get the first position of `eos_token_id` value (`pad_token_ids` might equal to `eos_token_id`)
            # Note: we assume each sequence (along batch dim.) contains an  `eos_token_id` (e.g. prepared by the tokenizer)
            (input_ids.to(dtype=torch.int, device=last_hidden_state.device) == self.eos_token_id).int().argmax(
                dim=-1),
        ]

    return BaseEncoderOutput(
        last_hidden_state=last_hidden_state,
        pooler_output=pooled_output,
        hidden_states=encoder_outputs,
        # attentions=encoder_outputs.attentions,
    )

fastvideo.models.encoders.clip.CLIPTextModel

CLIPTextModel(config: CLIPTextConfig)
Source
class CLIPTextModel(TextEncoder):

    def __init__(
        self,
        config: CLIPTextConfig,
    ) -> None:
        super().__init__(config)
        self.text_model = CLIPTextTransformer(config=config, quant_config=config.quant_config, prefix=config.prefix)

    def forward(
        self,
        input_ids: torch.Tensor | 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,
    ) -> BaseEncoderOutput:

        outputs: BaseEncoderOutput = self.text_model(
            input_ids=input_ids,
            attention_mask=attention_mask,
            position_ids=position_ids,
            output_hidden_states=output_hidden_states,
        )
        return outputs

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

        # Define mapping for stacked parameters
        stacked_params_mapping = [
            # (param_name, shard_name, shard_id)
            ("qkv_proj", "q_proj", "q"),
            ("qkv_proj", "k_proj", "k"),
            ("qkv_proj", "v_proj", "v"),
        ]
        params_dict = dict(self.named_parameters())
        loaded_params: set[str] = set()

        weights = list(weights)
        # Some checkpoints (Kandinsky6's, for one) save CLIP's
        # inner CLIPTextTransformer's own state_dict() directly, without the
        # `text_model.` prefix this class's `self.text_model` wrapper puts on
        # every parameter -- unlike other checkpoints already loaded through
        # this class (e.g. Kandinsky5's), whose keys already include it.
        # Detect and normalize once, up front, rather than guessing per key.
        needs_text_model_prefix = bool(weights) and not any(name.startswith("text_model.") for name, _ in weights)

        for name, loaded_weight in weights:
            if needs_text_model_prefix:
                name = f"text_model.{name}"
            # Handle q_proj, k_proj, v_proj -> qkv_proj mapping
            for param_name, weight_name, shard_id in stacked_params_mapping:
                if weight_name in name:
                    # Replace the weight name with the parameter name
                    model_param_name = name.replace(weight_name, param_name)

                    if model_param_name in params_dict:
                        param = params_dict[model_param_name]
                        weight_loader = param.weight_loader
                        weight_loader(param, loaded_weight, shard_id)
                        loaded_params.add(model_param_name)
                    break
            else:
                # Use default weight loader for all other parameters
                if name in params_dict:
                    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.clip.CLIPTextModel.forward

forward(input_ids: torch.Tensor | 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) -> BaseEncoderOutput
Source
def forward(
    self,
    input_ids: torch.Tensor | 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,
) -> BaseEncoderOutput:

    outputs: BaseEncoderOutput = self.text_model(
        input_ids=input_ids,
        attention_mask=attention_mask,
        position_ids=position_ids,
        output_hidden_states=output_hidden_states,
    )
    return outputs

fastvideo.models.encoders.clip.CLIPTextModel.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]:

    # Define mapping for stacked parameters
    stacked_params_mapping = [
        # (param_name, shard_name, shard_id)
        ("qkv_proj", "q_proj", "q"),
        ("qkv_proj", "k_proj", "k"),
        ("qkv_proj", "v_proj", "v"),
    ]
    params_dict = dict(self.named_parameters())
    loaded_params: set[str] = set()

    weights = list(weights)
    # Some checkpoints (Kandinsky6's, for one) save CLIP's
    # inner CLIPTextTransformer's own state_dict() directly, without the
    # `text_model.` prefix this class's `self.text_model` wrapper puts on
    # every parameter -- unlike other checkpoints already loaded through
    # this class (e.g. Kandinsky5's), whose keys already include it.
    # Detect and normalize once, up front, rather than guessing per key.
    needs_text_model_prefix = bool(weights) and not any(name.startswith("text_model.") for name, _ in weights)

    for name, loaded_weight in weights:
        if needs_text_model_prefix:
            name = f"text_model.{name}"
        # Handle q_proj, k_proj, v_proj -> qkv_proj mapping
        for param_name, weight_name, shard_id in stacked_params_mapping:
            if weight_name in name:
                # Replace the weight name with the parameter name
                model_param_name = name.replace(weight_name, param_name)

                if model_param_name in params_dict:
                    param = params_dict[model_param_name]
                    weight_loader = param.weight_loader
                    weight_loader(param, loaded_weight, shard_id)
                    loaded_params.add(model_param_name)
                break
        else:
            # Use default weight loader for all other parameters
            if name in params_dict:
                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.clip.CLIPTextModelWithProjection

CLIPTextModelWithProjection(config: CLIPTextConfig)
Source
class CLIPTextModelWithProjection(CLIPTextModel):

    def __init__(self, config: CLIPTextConfig) -> None:
        super().__init__(config)
        self.text_projection = nn.Linear(config.hidden_size, config.projection_dim, bias=False)

    def forward(
        self,
        input_ids: torch.Tensor | 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,
    ) -> BaseEncoderOutput:
        outputs = super().forward(
            input_ids=input_ids,
            position_ids=position_ids,
            attention_mask=attention_mask,
            inputs_embeds=inputs_embeds,
            output_hidden_states=output_hidden_states,
            **kwargs,
        )
        pooled = outputs.pooler_output
        if pooled is not None:
            pooled = self.text_projection(pooled)
        return BaseEncoderOutput(
            last_hidden_state=outputs.last_hidden_state,
            pooler_output=pooled,
            hidden_states=outputs.hidden_states,
            attentions=outputs.attentions,
            attention_mask=outputs.attention_mask,
        )

fastvideo.models.encoders.clip.CLIPTextModelWithProjection.forward

forward(input_ids: torch.Tensor | 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) -> BaseEncoderOutput
Source
def forward(
    self,
    input_ids: torch.Tensor | 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,
) -> BaseEncoderOutput:
    outputs = super().forward(
        input_ids=input_ids,
        position_ids=position_ids,
        attention_mask=attention_mask,
        inputs_embeds=inputs_embeds,
        output_hidden_states=output_hidden_states,
        **kwargs,
    )
    pooled = outputs.pooler_output
    if pooled is not None:
        pooled = self.text_projection(pooled)
    return BaseEncoderOutput(
        last_hidden_state=outputs.last_hidden_state,
        pooler_output=pooled,
        hidden_states=outputs.hidden_states,
        attentions=outputs.attentions,
        attention_mask=outputs.attention_mask,
    )

fastvideo.models.encoders.clip.CLIPVisionTransformer

CLIPVisionTransformer(config: CLIPVisionConfig, quant_config: QuantizationConfig | None = None, num_hidden_layers_override: int | None = None, require_post_norm: bool | None = None, prefix: str = '')
Source
class CLIPVisionTransformer(nn.Module):

    def __init__(
        self,
        config: CLIPVisionConfig,
        quant_config: QuantizationConfig | None = None,
        num_hidden_layers_override: int | None = None,
        require_post_norm: bool | None = None,
        prefix: str = "",
    ) -> None:
        super().__init__()

        self.config = config
        embed_dim = config.hidden_size

        self.embeddings = CLIPVisionEmbeddings(config)

        # NOTE: This typo of "layrnorm" is not fixed on purpose to match
        # the original transformers code and name of the model weights.
        self.pre_layrnorm = nn.LayerNorm(embed_dim, eps=config.layer_norm_eps)

        self.encoder = CLIPEncoder(
            config=config,
            quant_config=quant_config,
            num_hidden_layers_override=num_hidden_layers_override,
            prefix=f"{prefix}.encoder",
        )

        num_hidden_layers = config.num_hidden_layers
        if len(self.encoder.layers) > config.num_hidden_layers:
            raise ValueError(f"The original encoder only has {num_hidden_layers} "
                             f"layers, but you requested {len(self.encoder.layers)} layers.")

        # If possible, skip post_layernorm to conserve memory
        if require_post_norm is None:
            require_post_norm = len(self.encoder.layers) == num_hidden_layers

        if require_post_norm:
            self.post_layernorm = nn.LayerNorm(embed_dim, eps=config.layer_norm_eps)
        else:
            self.post_layernorm = None

    def forward(
        self,
        pixel_values: torch.Tensor,
        feature_sample_layers: list[int] | None = None,
    ) -> torch.Tensor:

        hidden_states = self.embeddings(pixel_values)
        hidden_states = self.pre_layrnorm(hidden_states)

        return_all_hidden_states = feature_sample_layers is not None

        # Produces either the last layer output or all of the hidden states,
        # depending on if we have feature_sample_layers or not
        encoder_outputs = self.encoder(inputs_embeds=hidden_states, return_all_hidden_states=return_all_hidden_states)

        if not return_all_hidden_states:
            encoder_outputs = encoder_outputs[0]

        # Handle post-norm (if applicable) and stacks feature layers if needed
        encoder_outputs = resolve_visual_encoder_outputs(encoder_outputs, feature_sample_layers, self.post_layernorm,
                                                         self.config.num_hidden_layers)

        return encoder_outputs

fastvideo.models.encoders.clip.CLIPVisionTransformer.forward

forward(pixel_values: torch.Tensor, feature_sample_layers: list[int] | None = None) -> torch.Tensor
Source
def forward(
    self,
    pixel_values: torch.Tensor,
    feature_sample_layers: list[int] | None = None,
) -> torch.Tensor:

    hidden_states = self.embeddings(pixel_values)
    hidden_states = self.pre_layrnorm(hidden_states)

    return_all_hidden_states = feature_sample_layers is not None

    # Produces either the last layer output or all of the hidden states,
    # depending on if we have feature_sample_layers or not
    encoder_outputs = self.encoder(inputs_embeds=hidden_states, return_all_hidden_states=return_all_hidden_states)

    if not return_all_hidden_states:
        encoder_outputs = encoder_outputs[0]

    # Handle post-norm (if applicable) and stacks feature layers if needed
    encoder_outputs = resolve_visual_encoder_outputs(encoder_outputs, feature_sample_layers, self.post_layernorm,
                                                     self.config.num_hidden_layers)

    return encoder_outputs

fastvideo.models.encoders.clip.CLIPVisionModel

CLIPVisionModel(config: CLIPVisionConfig)
Source
class CLIPVisionModel(ImageEncoder):
    config_class = CLIPVisionConfig
    main_input_name = "pixel_values"
    packed_modules_mapping = {"qkv_proj": ["q_proj", "k_proj", "v_proj"]}

    def __init__(self, config: CLIPVisionConfig) -> None:
        super().__init__(config)
        self.vision_model = CLIPVisionTransformer(config=config,
                                                  quant_config=config.quant_config,
                                                  num_hidden_layers_override=config.num_hidden_layers_override,
                                                  require_post_norm=config.require_post_norm,
                                                  prefix=f"{config.prefix}.vision_model")

    def forward(
        self,
        pixel_values: torch.Tensor,
        feature_sample_layers: list[int] | None = None,
        **kwargs,
    ) -> BaseEncoderOutput:
        last_hidden_state = self.vision_model(pixel_values, feature_sample_layers)
        return BaseEncoderOutput(last_hidden_state=last_hidden_state)

    @property
    def device(self):
        return next(self.parameters()).device

    # (TODO) Add prefix argument for filtering out weights to be loaded
    #        ref: https://github.com/vllm-project/vllm/pull/7186#discussion_r1734163986
    def load_weights(self, weights: Iterable[tuple[str, torch.Tensor]]) -> set[str]:

        params_dict = dict(self.named_parameters())
        loaded_params: set[str] = set()
        layer_count = len(self.vision_model.encoder.layers)

        for name, loaded_weight in weights:
            if name.startswith("visual_projection"):
                continue
            # post_layernorm is not needed in CLIPVisionModel
            if (name.startswith("vision_model.post_layernorm") and self.vision_model.post_layernorm is None):
                continue

            # omit layers when num_hidden_layers_override is set
            if name.startswith("vision_model.encoder.layers"):
                layer_idx = int(name.split(".")[3])
                if layer_idx >= layer_count:
                    continue

            for (param_name, weight_name, shard_id) in self.config.arch_config.stacked_params_mapping:
                if weight_name not in name:
                    continue
                name = name.replace(weight_name, param_name)

                param = params_dict[name]
                weight_loader = param.weight_loader
                weight_loader(param, loaded_weight, shard_id)
                break
            else:
                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.clip.CLIPVisionModel.forward

forward(pixel_values: torch.Tensor, feature_sample_layers: list[int] | None = None, **kwargs) -> BaseEncoderOutput
Source
def forward(
    self,
    pixel_values: torch.Tensor,
    feature_sample_layers: list[int] | None = None,
    **kwargs,
) -> BaseEncoderOutput:
    last_hidden_state = self.vision_model(pixel_values, feature_sample_layers)
    return BaseEncoderOutput(last_hidden_state=last_hidden_state)

fastvideo.models.encoders.clip.CLIPVisionModel.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()
    layer_count = len(self.vision_model.encoder.layers)

    for name, loaded_weight in weights:
        if name.startswith("visual_projection"):
            continue
        # post_layernorm is not needed in CLIPVisionModel
        if (name.startswith("vision_model.post_layernorm") and self.vision_model.post_layernorm is None):
            continue

        # omit layers when num_hidden_layers_override is set
        if name.startswith("vision_model.encoder.layers"):
            layer_idx = int(name.split(".")[3])
            if layer_idx >= layer_count:
                continue

        for (param_name, weight_name, shard_id) in self.config.arch_config.stacked_params_mapping:
            if weight_name not in name:
                continue
            name = name.replace(weight_name, param_name)

            param = params_dict[name]
            weight_loader = param.weight_loader
            weight_loader(param, loaded_weight, shard_id)
            break
        else:
            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