Spaces:
Running
on
Zero
Running
on
Zero
# Copyright 2024 The HuggingFace Inc. team. | |
# SPDX-License-Identifier: Apache-2.0 | |
"""PyTorch Siglip model.""" | |
import math | |
import warnings | |
from dataclasses import dataclass | |
from typing import Any, Optional, Tuple, Union | |
import numpy as np | |
import torch | |
import torch.utils.checkpoint | |
from torch import nn | |
from torch.nn import BCEWithLogitsLoss, CrossEntropyLoss, MSELoss | |
from torch.nn.init import _calculate_fan_in_and_fan_out | |
from transformers.activations import ACT2FN | |
from transformers.modeling_attn_mask_utils import _prepare_4d_attention_mask | |
from transformers.modeling_outputs import BaseModelOutput, BaseModelOutputWithPooling, ImageClassifierOutput | |
from transformers.modeling_utils import PreTrainedModel | |
from transformers.utils import ( | |
ModelOutput, | |
add_start_docstrings, | |
add_start_docstrings_to_model_forward, | |
is_flash_attn_2_available, | |
is_flash_attn_greater_or_equal_2_10, | |
logging, | |
replace_return_docstrings, | |
torch_int, | |
) | |
from .configuration_siglip import SiglipConfig, SiglipTextConfig, SiglipVisionConfig | |
if is_flash_attn_2_available(): | |
from transformers.modeling_flash_attention_utils import _flash_attention_forward | |
logger = logging.get_logger(__name__) | |
# General docstring | |
_CONFIG_FOR_DOC = "SiglipConfig" | |
_CHECKPOINT_FOR_DOC = "google/siglip-base-patch16-224" | |
def _trunc_normal_(tensor, mean, std, a, b): | |
# Cut & paste from PyTorch official master until it's in a few official releases - RW | |
# Method based on https://people.sc.fsu.edu/~jburkardt/presentations/truncated_normal.pdf | |
def norm_cdf(x): | |
# Computes standard normal cumulative distribution function | |
return (1.0 + math.erf(x / math.sqrt(2.0))) / 2.0 | |
if (mean < a - 2 * std) or (mean > b + 2 * std): | |
warnings.warn( | |
"mean is more than 2 std from [a, b] in nn.init.trunc_normal_. " | |
"The distribution of values may be incorrect.", | |
stacklevel=2, | |
) | |
# Values are generated by using a truncated uniform distribution and | |
# then using the inverse CDF for the normal distribution. | |
# Get upper and lower cdf values | |
l = norm_cdf((a - mean) / std) | |
u = norm_cdf((b - mean) / std) | |
# Uniformly fill tensor with values from [l, u], then translate to | |
# [2l-1, 2u-1]. | |
tensor.uniform_(2 * l - 1, 2 * u - 1) | |
# Use inverse cdf transform for normal distribution to get truncated | |
# standard normal | |
tensor.erfinv_() | |
# Transform to proper mean, std | |
tensor.mul_(std * math.sqrt(2.0)) | |
tensor.add_(mean) | |
# Clamp to ensure it's in the proper range | |
tensor.clamp_(min=a, max=b) | |
def trunc_normal_tf_( | |
tensor: torch.Tensor, mean: float = 0.0, std: float = 1.0, a: float = -2.0, b: float = 2.0 | |
) -> torch.Tensor: | |
"""Fills the input Tensor with values drawn from a truncated | |
normal distribution. The values are effectively drawn from the | |
normal distribution :math:`\\mathcal{N}(\text{mean}, \text{std}^2)` | |
with values outside :math:`[a, b]` redrawn until they are within | |
the bounds. The method used for generating the random values works | |
best when :math:`a \\leq \text{mean} \\leq b`. | |
NOTE: this 'tf' variant behaves closer to Tensorflow / JAX impl where the | |
bounds [a, b] are applied when sampling the normal distribution with mean=0, std=1.0 | |
and the result is subsequently scaled and shifted by the mean and std args. | |
Args: | |
tensor: an n-dimensional `torch.Tensor` | |
mean: the mean of the normal distribution | |
std: the standard deviation of the normal distribution | |
a: the minimum cutoff value | |
b: the maximum cutoff value | |
""" | |
with torch.no_grad(): | |
_trunc_normal_(tensor, 0, 1.0, a, b) | |
tensor.mul_(std).add_(mean) | |
def variance_scaling_(tensor, scale=1.0, mode="fan_in", distribution="normal"): | |
fan_in, fan_out = _calculate_fan_in_and_fan_out(tensor) | |
if mode == "fan_in": | |
denom = fan_in | |
elif mode == "fan_out": | |
denom = fan_out | |
elif mode == "fan_avg": | |
denom = (fan_in + fan_out) / 2 | |
variance = scale / denom | |
if distribution == "truncated_normal": | |
# constant is stddev of standard normal truncated to (-2, 2) | |
trunc_normal_tf_(tensor, std=math.sqrt(variance) / 0.87962566103423978) | |
elif distribution == "normal": | |
with torch.no_grad(): | |
tensor.normal_(std=math.sqrt(variance)) | |
elif distribution == "uniform": | |
bound = math.sqrt(3 * variance) | |
with torch.no_grad(): | |
tensor.uniform_(-bound, bound) | |
else: | |
raise ValueError(f"invalid distribution {distribution}") | |
def lecun_normal_(tensor): | |
variance_scaling_(tensor, mode="fan_in", distribution="truncated_normal") | |
def default_flax_embed_init(tensor): | |
variance_scaling_(tensor, mode="fan_in", distribution="normal") | |
# Copied from transformers.models.clip.modeling_clip.CLIPVisionModelOutput with CLIP->Siglip | |
class SiglipVisionModelOutput(ModelOutput): | |
""" | |
Base class for vision model's outputs that also contains image embeddings of the pooling of the last hidden states. | |
Args: | |
image_embeds (`torch.FloatTensor` of shape `(batch_size, output_dim)` *optional* returned when model is initialized with `with_projection=True`): | |
The image embeddings obtained by applying the projection layer to the pooler_output. | |
last_hidden_state (`torch.FloatTensor` of shape `(batch_size, sequence_length, hidden_size)`): | |
Sequence of hidden-states at the output of the last layer of the model. | |
hidden_states (`tuple(torch.FloatTensor)`, *optional*, returned when `output_hidden_states=True` is passed or when `config.output_hidden_states=True`): | |
Tuple of `torch.FloatTensor` (one for the output of the embeddings, if the model has an embedding layer, + | |
one for the output of each layer) of shape `(batch_size, sequence_length, hidden_size)`. | |
Hidden-states of the model at the output of each layer plus the optional initial embedding outputs. | |
attentions (`tuple(torch.FloatTensor)`, *optional*, returned when `output_attentions=True` is passed or when `config.output_attentions=True`): | |
Tuple of `torch.FloatTensor` (one for each layer) of shape `(batch_size, num_heads, sequence_length, | |
sequence_length)`. | |
Attentions weights after the attention softmax, used to compute the weighted average in the self-attention | |
heads. | |
""" | |
image_embeds: Optional[torch.FloatTensor] = None | |
last_hidden_state: torch.FloatTensor = None | |
hidden_states: Optional[Tuple[torch.FloatTensor, ...]] = None | |
attentions: Optional[Tuple[torch.FloatTensor, ...]] = None | |
# Copied from transformers.models.clip.modeling_clip.CLIPTextModelOutput with CLIP->Siglip | |
class SiglipTextModelOutput(ModelOutput): | |
""" | |
Base class for text model's outputs that also contains a pooling of the last hidden states. | |
Args: | |
text_embeds (`torch.FloatTensor` of shape `(batch_size, output_dim)` *optional* returned when model is initialized with `with_projection=True`): | |
The text embeddings obtained by applying the projection layer to the pooler_output. | |
last_hidden_state (`torch.FloatTensor` of shape `(batch_size, sequence_length, hidden_size)`): | |
Sequence of hidden-states at the output of the last layer of the model. | |
hidden_states (`tuple(torch.FloatTensor)`, *optional*, returned when `output_hidden_states=True` is passed or when `config.output_hidden_states=True`): | |
Tuple of `torch.FloatTensor` (one for the output of the embeddings, if the model has an embedding layer, + | |
one for the output of each layer) of shape `(batch_size, sequence_length, hidden_size)`. | |
Hidden-states of the model at the output of each layer plus the optional initial embedding outputs. | |
attentions (`tuple(torch.FloatTensor)`, *optional*, returned when `output_attentions=True` is passed or when `config.output_attentions=True`): | |
Tuple of `torch.FloatTensor` (one for each layer) of shape `(batch_size, num_heads, sequence_length, | |
sequence_length)`. | |
Attentions weights after the attention softmax, used to compute the weighted average in the self-attention | |
heads. | |
""" | |
text_embeds: Optional[torch.FloatTensor] = None | |
last_hidden_state: torch.FloatTensor = None | |
hidden_states: Optional[Tuple[torch.FloatTensor, ...]] = None | |
attentions: Optional[Tuple[torch.FloatTensor, ...]] = None | |
# Copied from transformers.models.clip.modeling_clip.CLIPOutput with CLIP->Siglip | |
class SiglipOutput(ModelOutput): | |
""" | |
Args: | |
loss (`torch.FloatTensor` of shape `(1,)`, *optional*, returned when `return_loss` is `True`): | |
Contrastive loss for image-text similarity. | |
logits_per_image (`torch.FloatTensor` of shape `(image_batch_size, text_batch_size)`): | |
The scaled dot product scores between `image_embeds` and `text_embeds`. This represents the image-text | |
similarity scores. | |
logits_per_text (`torch.FloatTensor` of shape `(text_batch_size, image_batch_size)`): | |
The scaled dot product scores between `text_embeds` and `image_embeds`. This represents the text-image | |
similarity scores. | |
text_embeds (`torch.FloatTensor` of shape `(batch_size, output_dim`): | |
The text embeddings obtained by applying the projection layer to the pooled output of [`SiglipTextModel`]. | |
image_embeds (`torch.FloatTensor` of shape `(batch_size, output_dim`): | |
The image embeddings obtained by applying the projection layer to the pooled output of [`SiglipVisionModel`]. | |
text_model_output (`BaseModelOutputWithPooling`): | |
The output of the [`SiglipTextModel`]. | |
vision_model_output (`BaseModelOutputWithPooling`): | |
The output of the [`SiglipVisionModel`]. | |
""" | |
loss: Optional[torch.FloatTensor] = None | |
logits_per_image: torch.FloatTensor = None | |
logits_per_text: torch.FloatTensor = None | |
text_embeds: torch.FloatTensor = None | |
image_embeds: torch.FloatTensor = None | |
text_model_output: BaseModelOutputWithPooling = None | |
vision_model_output: BaseModelOutputWithPooling = None | |
def to_tuple(self) -> Tuple[Any]: | |
return tuple( | |
self[k] if k not in ["text_model_output", "vision_model_output"] else getattr(self, k).to_tuple() | |
for k in self.keys() | |
) | |
class SiglipVisionEmbeddings(nn.Module): | |
def __init__(self, config: SiglipVisionConfig): | |
super().__init__() | |
self.config = config | |
self.embed_dim = config.hidden_size | |
self.image_size = config.image_size | |
self.patch_size = config.patch_size | |
self.patch_embedding = nn.Conv2d( | |
in_channels=config.num_channels, | |
out_channels=self.embed_dim, | |
kernel_size=self.patch_size, | |
stride=self.patch_size, | |
padding="valid", | |
) | |
self.num_patches = (self.image_size // self.patch_size) ** 2 | |
self.num_positions = self.num_patches | |
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 interpolate_pos_encoding(self, embeddings: torch.Tensor, height: int, width: int) -> torch.Tensor: | |
""" | |
This method allows to interpolate the pre-trained position encodings, to be able to use the model on higher resolution | |
images. This method is also adapted to support torch.jit tracing and no class embeddings. | |
Adapted from: | |
- https://github.com/facebookresearch/dino/blob/de9ee3df6cf39fac952ab558447af1fa1365362a/vision_transformer.py#L174-L194, and | |
- https://github.com/facebookresearch/dinov2/blob/e1277af2ba9496fbadf7aec6eba56e8d882d1e35/dinov2/models/vision_transformer.py#L179-L211 | |
""" | |
num_patches = embeddings.shape[1] | |
num_positions = self.position_embedding.weight.shape[0] | |
# always interpolate when tracing to ensure the exported model works for dynamic input shapes | |
if not torch.jit.is_tracing() and num_patches == num_positions and height == width: | |
return self.position_embedding(self.position_ids) | |
patch_pos_embed = self.position_embedding.weight.unsqueeze(0) | |
dim = embeddings.shape[-1] | |
new_height = height // self.patch_size | |
new_width = width // self.patch_size | |
sqrt_num_positions = torch_int(num_positions**0.5) | |
patch_pos_embed = patch_pos_embed.reshape(1, sqrt_num_positions, sqrt_num_positions, dim) | |
patch_pos_embed = patch_pos_embed.permute(0, 3, 1, 2) | |
patch_pos_embed = nn.functional.interpolate( | |
patch_pos_embed, | |
size=(new_height, new_width), | |
mode="bicubic", | |
align_corners=False, | |
) | |
patch_pos_embed = patch_pos_embed.permute(0, 2, 3, 1).view(1, -1, dim) | |
return patch_pos_embed | |
def forward(self, pixel_values: torch.FloatTensor, interpolate_pos_encoding=False) -> torch.Tensor: | |
_, _, height, width = pixel_values.shape | |
patch_embeds = self.patch_embedding(pixel_values) # shape = [*, width, grid, grid] | |
embeddings = patch_embeds.flatten(2).transpose(1, 2) | |
if interpolate_pos_encoding: | |
embeddings = embeddings + self.interpolate_pos_encoding(embeddings, height, width) | |
else: | |
embeddings = embeddings + self.position_embedding(self.position_ids) | |
return embeddings | |
# Copied from transformers.models.clip.modeling_clip.CLIPTextEmbeddings with CLIP->Siglip | |
class SiglipTextEmbeddings(nn.Module): | |
def __init__(self, config: SiglipTextConfig): | |
super().__init__() | |
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: Optional[torch.LongTensor] = None, | |
position_ids: Optional[torch.LongTensor] = None, | |
inputs_embeds: Optional[torch.FloatTensor] = None, | |
) -> torch.Tensor: | |
seq_length = input_ids.shape[-1] if input_ids is not None else inputs_embeds.shape[-2] | |
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 | |
class SiglipAttention(nn.Module): | |
"""Multi-headed attention from 'Attention Is All You Need' paper""" | |
# Copied from transformers.models.clip.modeling_clip.CLIPAttention.__init__ | |
def __init__(self, config): | |
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( | |
f"embed_dim must be divisible by num_heads (got `embed_dim`: {self.embed_dim} and `num_heads`:" | |
f" {self.num_heads})." | |
) | |
self.scale = self.head_dim**-0.5 | |
self.dropout = config.attention_dropout | |
self.k_proj = nn.Linear(self.embed_dim, self.embed_dim) | |
self.v_proj = nn.Linear(self.embed_dim, self.embed_dim) | |
self.q_proj = nn.Linear(self.embed_dim, self.embed_dim) | |
self.out_proj = nn.Linear(self.embed_dim, self.embed_dim) | |
def forward( | |
self, | |
hidden_states: torch.Tensor, | |
attention_mask: Optional[torch.Tensor] = None, | |
output_attentions: Optional[bool] = False, | |
) -> Tuple[torch.Tensor, Optional[torch.Tensor]]: | |
"""Input shape: Batch x Time x Channel""" | |
batch_size, q_len, _ = hidden_states.size() | |
query_states = self.q_proj(hidden_states) | |
key_states = self.k_proj(hidden_states) | |
value_states = self.v_proj(hidden_states) | |
query_states = query_states.view(batch_size, q_len, self.num_heads, self.head_dim).transpose(1, 2) | |
key_states = key_states.view(batch_size, q_len, self.num_heads, self.head_dim).transpose(1, 2) | |
value_states = value_states.view(batch_size, q_len, self.num_heads, self.head_dim).transpose(1, 2) | |
k_v_seq_len = key_states.shape[-2] | |
attn_weights = torch.matmul(query_states, key_states.transpose(2, 3)) * self.scale | |
if attn_weights.size() != (batch_size, self.num_heads, q_len, k_v_seq_len): | |
raise ValueError( | |
f"Attention weights should be of size {(batch_size, self.num_heads, q_len, k_v_seq_len)}, but is" | |
f" {attn_weights.size()}" | |
) | |
if attention_mask is not None: | |
if attention_mask.size() != (batch_size, 1, q_len, k_v_seq_len): | |
raise ValueError( | |
f"Attention mask should be of size {(batch_size, 1, q_len, k_v_seq_len)}, but is {attention_mask.size()}" | |
) | |
attn_weights = attn_weights + attention_mask | |
# upcast attention to fp32 | |
attn_weights = nn.functional.softmax(attn_weights, dim=-1, dtype=torch.float32).to(query_states.dtype) | |
attn_weights = nn.functional.dropout(attn_weights, p=self.dropout, training=self.training) | |
attn_output = torch.matmul(attn_weights, value_states) | |
if attn_output.size() != (batch_size, self.num_heads, q_len, self.head_dim): | |
raise ValueError( | |
f"`attn_output` should be of size {(batch_size, self.num_heads, q_len, self.head_dim)}, but is" | |
f" {attn_output.size()}" | |
) | |
attn_output = attn_output.transpose(1, 2).contiguous() | |
attn_output = attn_output.reshape(batch_size, q_len, self.embed_dim) | |
attn_output = self.out_proj(attn_output) | |
return attn_output, attn_weights | |
class SiglipFlashAttention2(SiglipAttention): | |
""" | |
SiglipAttention flash attention module. This module inherits from `SiglipAttention` as the weights of the module stays | |
untouched. The only required change would be on the forward pass where it needs to correctly call the public API of | |
flash attention and deal with padding tokens in case the input contains any of them. | |
""" | |
is_causal = False | |
# Copied from transformers.models.llama.modeling_llama.LlamaFlashAttention2.__init__ | |
def __init__(self, *args, **kwargs): | |
super().__init__(*args, **kwargs) | |
# TODO: Should be removed once Flash Attention for RoCm is bumped to 2.1. | |
# flash_attn<2.1 generates top-left aligned causal mask, while what is needed here is bottom-right alignement, that was made default for flash_attn>=2.1. This attribute is used to handle this difference. Reference: https://github.com/Dao-AILab/flash-attention/releases/tag/v2.1.0. | |
# Beware that with flash_attn<2.1, using q_seqlen != k_seqlen (except for the case q_seqlen == 1) produces a wrong mask (top-left). | |
self._flash_attn_uses_top_left_mask = not is_flash_attn_greater_or_equal_2_10() | |
# Adapted from transformers.models.llama.modeling_llama.LlamaFlashAttention2.forward | |
def forward( | |
self, | |
hidden_states: torch.Tensor, | |
attention_mask: Optional[torch.LongTensor] = None, | |
output_attentions: bool = False, | |
) -> Tuple[torch.Tensor, Optional[torch.Tensor], Optional[Tuple[torch.Tensor]]]: | |
output_attentions = False | |
batch_size, q_len, _ = hidden_states.size() | |
query_states = self.q_proj(hidden_states) | |
key_states = self.k_proj(hidden_states) | |
value_states = self.v_proj(hidden_states) | |
# Flash attention requires the input to have the shape | |
# batch_size x seq_length x head_dim x hidden_dim | |
# therefore we just need to keep the original shape | |
query_states = query_states.view(batch_size, q_len, self.num_heads, self.head_dim).transpose(1, 2) | |
key_states = key_states.view(batch_size, q_len, self.num_heads, self.head_dim).transpose(1, 2) | |
value_states = value_states.view(batch_size, q_len, self.num_heads, self.head_dim).transpose(1, 2) | |
# TODO: These transpose are quite inefficient but Flash Attention requires the layout [batch_size, sequence_length, num_heads, head_dim]. We would need to refactor the KV cache | |
# to be able to avoid many of these transpose/reshape/view. | |
query_states = query_states.transpose(1, 2) | |
key_states = key_states.transpose(1, 2) | |
value_states = value_states.transpose(1, 2) | |
dropout_rate = self.dropout if self.training else 0.0 | |
# In PEFT, usually we cast the layer norms in float32 for training stability reasons | |
# therefore the input hidden states gets silently casted in float32. Hence, we need | |
# cast them back in the correct dtype just to be sure everything works as expected. | |
# This might slowdown training & inference so it is recommended to not cast the LayerNorms | |
# in fp32. | |
input_dtype = query_states.dtype | |
if input_dtype == torch.float32: | |
if torch.is_autocast_enabled(): | |
target_dtype = torch.get_autocast_gpu_dtype() | |
# Handle the case where the model is quantized | |
elif hasattr(self.config, "_pre_quantization_dtype"): | |
target_dtype = self.config._pre_quantization_dtype | |
else: | |
target_dtype = self.q_proj.weight.dtype | |
logger.warning_once( | |
f"The input hidden states seems to be silently casted in float32, this might be related to" | |
f" the fact you have upcasted embedding or layer norm layers in float32. We will cast back the input in" | |
f" {target_dtype}." | |
) | |
query_states = query_states.to(target_dtype) | |
key_states = key_states.to(target_dtype) | |
value_states = value_states.to(target_dtype) | |
attn_output = _flash_attention_forward( | |
query_states, | |
key_states, | |
value_states, | |
attention_mask, | |
q_len, | |
dropout=dropout_rate, | |
is_causal=self.is_causal, | |
use_top_left_mask=self._flash_attn_uses_top_left_mask, | |
) | |
attn_output = attn_output.reshape(batch_size, q_len, self.embed_dim).contiguous() | |
attn_output = self.out_proj(attn_output) | |
if not output_attentions: | |
attn_weights = None | |
return attn_output, attn_weights | |
class SiglipSdpaAttention(SiglipAttention): | |
""" | |
Siglip attention module using torch.nn.functional.scaled_dot_product_attention. This module inherits from | |
`SiglipAttention` as the weights of the module stays untouched. The only changes are on the forward pass to adapt to | |
SDPA API. | |
""" | |
is_causal = False | |
# Adapted from SiglipAttention.forward and transformers.models.llama.modeling_llama.LlamaSdpaAttention.forward | |
def forward( | |
self, | |
hidden_states: torch.Tensor, | |
attention_mask: Optional[torch.Tensor] = None, | |
output_attentions: Optional[bool] = False, | |
) -> Tuple[torch.Tensor, Optional[torch.Tensor]]: | |
if output_attentions: | |
# TODO: Improve this warning with e.g. `model.config.attn_implementation = "manual"` once this is implemented. | |
logger.warning_once( | |
"SiglipModel is using SiglipSdpaAttention, but `torch.nn.functional.scaled_dot_product_attention` does not support `output_attentions=True`. Falling back to the manual attention implementation, " | |
'but specifying the manual implementation will be required from Transformers version v5.0.0 onwards. This warning can be removed using the argument `attn_implementation="eager"` when loading the model.' | |
) | |
return super().forward( | |
hidden_states=hidden_states, | |
attention_mask=attention_mask, | |
output_attentions=output_attentions, | |
) | |
batch_size, q_len, _ = hidden_states.size() | |
query_states = self.q_proj(hidden_states) | |
key_states = self.k_proj(hidden_states) | |
value_states = self.v_proj(hidden_states) | |
query_states = query_states.view(batch_size, q_len, self.num_heads, self.head_dim).transpose(1, 2) | |
key_states = key_states.view(batch_size, q_len, self.num_heads, self.head_dim).transpose(1, 2) | |
value_states = value_states.view(batch_size, q_len, self.num_heads, self.head_dim).transpose(1, 2) | |
# SDPA with memory-efficient backend is currently (torch==2.1.2) bugged with non-contiguous inputs with custom attn_mask, | |
# Reference: https://github.com/pytorch/pytorch/issues/112577. | |
if query_states.device.type == "cuda" and attention_mask is not None: | |
query_states = query_states.contiguous() | |
key_states = key_states.contiguous() | |
value_states = value_states.contiguous() | |
# We dispatch to SDPA's Flash Attention or Efficient kernels via this `is_causal` if statement instead of an inline conditional assignment | |
# in SDPA to support both torch.compile's dynamic shapes and full graph options. An inline conditional prevents dynamic shapes from compiling. | |
is_causal = True if self.is_causal and q_len > 1 else False | |
attn_output = torch.nn.functional.scaled_dot_product_attention( | |
query_states, | |
key_states, | |
value_states, | |
attn_mask=attention_mask, | |
dropout_p=self.dropout if self.training else 0.0, | |
is_causal=is_causal, | |
) | |
attn_output = attn_output.transpose(1, 2).contiguous() | |
attn_output = attn_output.view(batch_size, q_len, self.embed_dim) | |
attn_output = self.out_proj(attn_output) | |
return attn_output, None | |
SIGLIP_ATTENTION_CLASSES = { | |
"eager": SiglipAttention, | |
"flash_attention_2": SiglipFlashAttention2, | |
"sdpa": SiglipSdpaAttention, | |
} | |
# Copied from transformers.models.clip.modeling_clip.CLIPMLP with CLIP->Siglip | |
class SiglipMLP(nn.Module): | |
def __init__(self, config): | |
super().__init__() | |
self.config = config | |
self.activation_fn = ACT2FN[config.hidden_act] | |
self.fc1 = nn.Linear(config.hidden_size, config.intermediate_size) | |
self.fc2 = nn.Linear(config.intermediate_size, config.hidden_size) | |
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 | |
class SiglipEncoderLayer(nn.Module): | |
def __init__(self, config: SiglipConfig): | |
super().__init__() | |
self.embed_dim = config.hidden_size | |
self.self_attn = SIGLIP_ATTENTION_CLASSES[config._attn_implementation](config=config) | |
self.layer_norm1 = nn.LayerNorm(self.embed_dim, eps=config.layer_norm_eps) | |
self.mlp = SiglipMLP(config) | |
self.layer_norm2 = nn.LayerNorm(self.embed_dim, eps=config.layer_norm_eps) | |
# Ignore copy | |
def forward( | |
self, | |
hidden_states: torch.Tensor, | |
attention_mask: torch.Tensor, | |
output_attentions: Optional[bool] = False, | |
) -> Tuple[torch.FloatTensor]: | |
""" | |
Args: | |
hidden_states (`torch.FloatTensor`): | |
Input to the layer of shape `(batch, seq_len, embed_dim)`. | |
attention_mask (`torch.FloatTensor`): | |
Attention mask of shape `(batch, 1, q_len, k_v_seq_len)` where padding elements are indicated by very large negative values. | |
output_attentions (`bool`, *optional*, defaults to `False`): | |
Whether or not to return the attentions tensors of all attention layers. See `attentions` under | |
returned tensors for more detail. | |
""" | |
residual = hidden_states | |
hidden_states = self.layer_norm1(hidden_states) | |
hidden_states, attn_weights = self.self_attn( | |
hidden_states=hidden_states, | |
attention_mask=attention_mask, | |
output_attentions=output_attentions, | |
) | |
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 | |
outputs = (hidden_states,) | |
if output_attentions: | |
outputs += (attn_weights,) | |
return outputs | |
class SiglipPreTrainedModel(PreTrainedModel): | |
""" | |
An abstract class to handle weights initialization and a simple interface for downloading and loading pretrained | |
models. | |
""" | |
config_class = SiglipConfig | |
base_model_prefix = "siglip" | |
supports_gradient_checkpointing = True | |
_no_split_modules = [ | |
"SiglipTextEmbeddings", | |
"SiglipEncoderLayer", | |
"SiglipVisionEmbeddings", | |
"SiglipEncoderLayer", | |
"SiglipMultiheadAttentionPoolingHead", | |
] | |
_supports_flash_attn_2 = True | |
_supports_sdpa = True | |
def _init_weights(self, module): | |
"""Initialize the weights""" | |
if isinstance(module, SiglipVisionEmbeddings): | |
width = ( | |
self.config.vision_config.hidden_size | |
if isinstance(self.config, SiglipConfig) | |
else self.config.hidden_size | |
) | |
nn.init.normal_(module.position_embedding.weight, std=1 / np.sqrt(width)) | |
elif isinstance(module, nn.Embedding): | |
default_flax_embed_init(module.weight) | |
elif isinstance(module, SiglipAttention): | |
nn.init.xavier_uniform_(module.q_proj.weight) | |
nn.init.xavier_uniform_(module.k_proj.weight) | |
nn.init.xavier_uniform_(module.v_proj.weight) | |
nn.init.xavier_uniform_(module.out_proj.weight) | |
nn.init.zeros_(module.q_proj.bias) | |
nn.init.zeros_(module.k_proj.bias) | |
nn.init.zeros_(module.v_proj.bias) | |
nn.init.zeros_(module.out_proj.bias) | |
elif isinstance(module, SiglipMLP): | |
nn.init.xavier_uniform_(module.fc1.weight) | |
nn.init.xavier_uniform_(module.fc2.weight) | |
nn.init.normal_(module.fc1.bias, std=1e-6) | |
nn.init.normal_(module.fc2.bias, std=1e-6) | |
elif isinstance(module, SiglipMultiheadAttentionPoolingHead): | |
nn.init.xavier_uniform_(module.probe.data) | |
nn.init.xavier_uniform_(module.attention.in_proj_weight.data) | |
nn.init.zeros_(module.attention.in_proj_bias.data) | |
elif isinstance(module, SiglipModel): | |
logit_scale_init = torch.log(torch.tensor(1.0)) | |
module.logit_scale.data.fill_(logit_scale_init) | |
module.logit_bias.data.zero_() | |
elif isinstance(module, SiglipForImageClassification): | |
nn.init.normal_( | |
module.classifier.weight, | |
std=self.config.vision_config.hidden_size**-0.5 * self.config.initializer_factor, | |
) | |
elif isinstance(module, (nn.Linear, nn.Conv2d)): | |
lecun_normal_(module.weight) | |
if module.bias is not None: | |
nn.init.zeros_(module.bias) | |
elif isinstance(module, nn.LayerNorm): | |
module.bias.data.zero_() | |
module.weight.data.fill_(1.0) | |
SIGLIP_START_DOCSTRING = r""" | |
This model inherits from [`PreTrainedModel`]. Check the superclass documentation for the generic methods the | |
library implements for all its model (such as downloading or saving, resizing the input embeddings, pruning heads | |
etc.) | |
This model is also a PyTorch [torch.nn.Module](https://pytorch.org/docs/stable/nn.html#torch.nn.Module) subclass. | |
Use it as a regular PyTorch Module and refer to the PyTorch documentation for all matter related to general usage | |
and behavior. | |
Parameters: | |
config ([`SiglipConfig`]): Model configuration class with all the parameters of the model. | |
Initializing with a config file does not load the weights associated with the model, only the | |
configuration. Check out the [`~PreTrainedModel.from_pretrained`] method to load the model weights. | |
""" | |
SIGLIP_TEXT_INPUTS_DOCSTRING = r""" | |
Args: | |
input_ids (`torch.LongTensor` of shape `(batch_size, sequence_length)`): | |
Indices of input sequence tokens in the vocabulary. Padding will be ignored by default should you provide | |
it. | |
Indices can be obtained using [`AutoTokenizer`]. See [`PreTrainedTokenizer.encode`] and | |
[`PreTrainedTokenizer.__call__`] for details. | |
[What are input IDs?](../glossary#input-ids) | |
attention_mask (`torch.Tensor` of shape `(batch_size, sequence_length)`, *optional*): | |
Mask to avoid performing attention on padding token indices. Mask values selected in `[0, 1]`: | |
- 1 for tokens that are **not masked**, | |
- 0 for tokens that are **masked**. | |
[What are attention masks?](../glossary#attention-mask) | |
position_ids (`torch.LongTensor` of shape `(batch_size, sequence_length)`, *optional*): | |
Indices of positions of each input sequence tokens in the position embeddings. Selected in the range `[0, | |
config.max_position_embeddings - 1]`. | |
[What are position IDs?](../glossary#position-ids) | |
output_attentions (`bool`, *optional*): | |
Whether or not to return the attentions tensors of all attention layers. See `attentions` under returned | |
tensors for more detail. | |
output_hidden_states (`bool`, *optional*): | |
Whether or not to return the hidden states of all layers. See `hidden_states` under returned tensors for | |
more detail. | |
return_dict (`bool`, *optional*): | |
Whether or not to return a [`~utils.ModelOutput`] instead of a plain tuple. | |
""" | |
SIGLIP_VISION_INPUTS_DOCSTRING = r""" | |
Args: | |
pixel_values (`torch.FloatTensor` of shape `(batch_size, num_channels, height, width)`): | |
Pixel values. Padding will be ignored by default should you provide it. Pixel values can be obtained using | |
[`AutoImageProcessor`]. See [`CLIPImageProcessor.__call__`] for details. | |
output_attentions (`bool`, *optional*): | |
Whether or not to return the attentions tensors of all attention layers. See `attentions` under returned | |
tensors for more detail. | |
output_hidden_states (`bool`, *optional*): | |
Whether or not to return the hidden states of all layers. See `hidden_states` under returned tensors for | |
more detail. | |
interpolate_pos_encoding (`bool`, *optional*, defaults to `False`): | |
Whether to interpolate the pre-trained position encodings. | |
return_dict (`bool`, *optional*): | |
Whether or not to return a [`~utils.ModelOutput`] instead of a plain tuple. | |
""" | |
SIGLIP_INPUTS_DOCSTRING = r""" | |
Args: | |
input_ids (`torch.LongTensor` of shape `(batch_size, sequence_length)`): | |
Indices of input sequence tokens in the vocabulary. Padding will be ignored by default should you provide | |
it. | |
Indices can be obtained using [`AutoTokenizer`]. See [`PreTrainedTokenizer.encode`] and | |
[`PreTrainedTokenizer.__call__`] for details. | |
[What are input IDs?](../glossary#input-ids) | |
attention_mask (`torch.Tensor` of shape `(batch_size, sequence_length)`, *optional*): | |
Mask to avoid performing attention on padding token indices. Mask values selected in `[0, 1]`: | |
- 1 for tokens that are **not masked**, | |
- 0 for tokens that are **masked**. | |
[What are attention masks?](../glossary#attention-mask) | |
position_ids (`torch.LongTensor` of shape `(batch_size, sequence_length)`, *optional*): | |
Indices of positions of each input sequence tokens in the position embeddings. Selected in the range `[0, | |
config.max_position_embeddings - 1]`. | |
[What are position IDs?](../glossary#position-ids) | |
pixel_values (`torch.FloatTensor` of shape `(batch_size, num_channels, height, width)`): | |
Pixel values. Padding will be ignored by default should you provide it. Pixel values can be obtained using | |
[`AutoImageProcessor`]. See [`CLIPImageProcessor.__call__`] for details. | |
return_loss (`bool`, *optional*): | |
Whether or not to return the contrastive loss. | |
output_attentions (`bool`, *optional*): | |
Whether or not to return the attentions tensors of all attention layers. See `attentions` under returned | |
tensors for more detail. | |
output_hidden_states (`bool`, *optional*): | |
Whether or not to return the hidden states of all layers. See `hidden_states` under returned tensors for | |
more detail. | |
interpolate_pos_encoding (`bool`, *optional*, defaults to `False`): | |
Whether to interpolate the pre-trained position encodings. | |
return_dict (`bool`, *optional*): | |
Whether or not to return a [`~utils.ModelOutput`] instead of a plain tuple. | |
""" | |
# Copied from transformers.models.altclip.modeling_altclip.AltCLIPEncoder with AltCLIP->Siglip | |
class SiglipEncoder(nn.Module): | |
""" | |
Transformer encoder consisting of `config.num_hidden_layers` self attention layers. Each layer is a | |
[`SiglipEncoderLayer`]. | |
Args: | |
config: SiglipConfig | |
""" | |
def __init__(self, config: SiglipConfig): | |
super().__init__() | |
self.config = config | |
self.layers = nn.ModuleList([SiglipEncoderLayer(config) for _ in range(config.num_hidden_layers)]) | |
self.gradient_checkpointing = False | |
# Ignore copy | |
def forward( | |
self, | |
inputs_embeds, | |
attention_mask: Optional[torch.Tensor] = None, | |
output_attentions: Optional[bool] = None, | |
output_hidden_states: Optional[bool] = None, | |
return_dict: Optional[bool] = None, | |
) -> Union[Tuple, BaseModelOutput]: | |
r""" | |
Args: | |
inputs_embeds (`torch.FloatTensor` of shape `(batch_size, sequence_length, hidden_size)`): | |
Optionally, instead of passing `input_ids` you can choose to directly pass an embedded representation. | |
This is useful if you want more control over how to convert `input_ids` indices into associated vectors | |
than the model's internal embedding lookup matrix. | |
attention_mask (`torch.Tensor` of shape `(batch_size, sequence_length)`, *optional*): | |
Mask to avoid performing attention on padding token indices. Mask values selected in `[0, 1]`: | |
- 1 for tokens that are **not masked**, | |
- 0 for tokens that are **masked**. | |
[What are attention masks?](../glossary#attention-mask) | |
output_attentions (`bool`, *optional*): | |
Whether or not to return the attentions tensors of all attention layers. See `attentions` under | |
returned tensors for more detail. | |
output_hidden_states (`bool`, *optional*): | |
Whether or not to return the hidden states of all layers. See `hidden_states` under returned tensors | |
for more detail. | |
return_dict (`bool`, *optional*): | |
Whether or not to return a [`~utils.ModelOutput`] instead of a plain tuple. | |
""" | |
output_attentions = output_attentions if output_attentions is not None else self.config.output_attentions | |
output_hidden_states = ( | |
output_hidden_states if output_hidden_states is not None else self.config.output_hidden_states | |
) | |
return_dict = return_dict if return_dict is not None else self.config.use_return_dict | |
encoder_states = () if output_hidden_states else None | |
all_attentions = () if output_attentions else None | |
hidden_states = inputs_embeds | |
for encoder_layer in self.layers: | |
if output_hidden_states: | |
encoder_states = encoder_states + (hidden_states,) | |
if self.gradient_checkpointing and self.training: | |
layer_outputs = self._gradient_checkpointing_func( | |
encoder_layer.__call__, | |
hidden_states, | |
attention_mask, | |
output_attentions, | |
) | |
else: | |
layer_outputs = encoder_layer( | |
hidden_states, | |
attention_mask, | |
output_attentions=output_attentions, | |
) | |
hidden_states = layer_outputs[0] | |
if output_attentions: | |
all_attentions = all_attentions + (layer_outputs[1],) | |
if output_hidden_states: | |
encoder_states = encoder_states + (hidden_states,) | |
if not return_dict: | |
return tuple(v for v in [hidden_states, encoder_states, all_attentions] if v is not None) | |
return BaseModelOutput( | |
last_hidden_state=hidden_states, hidden_states=encoder_states, attentions=all_attentions | |
) | |
class SiglipTextTransformer(nn.Module): | |
def __init__(self, config: SiglipTextConfig): | |
super().__init__() | |
self.config = config | |
embed_dim = config.hidden_size | |
self.embeddings = SiglipTextEmbeddings(config) | |
self.encoder = SiglipEncoder(config) | |
self.final_layer_norm = nn.LayerNorm(embed_dim, eps=config.layer_norm_eps) | |
self.head = nn.Linear(embed_dim, embed_dim) | |
self._use_flash_attention_2 = config._attn_implementation == "flash_attention_2" | |
def forward( | |
self, | |
input_ids: Optional[torch.Tensor] = None, | |
attention_mask: Optional[torch.Tensor] = None, | |
position_ids: Optional[torch.Tensor] = None, | |
output_attentions: Optional[bool] = None, | |
output_hidden_states: Optional[bool] = None, | |
return_dict: Optional[bool] = None, | |
) -> Union[Tuple, BaseModelOutputWithPooling]: | |
r""" | |
Returns: | |
""" | |
output_attentions = output_attentions if output_attentions is not None else self.config.output_attentions | |
output_hidden_states = ( | |
output_hidden_states if output_hidden_states is not None else self.config.output_hidden_states | |
) | |
return_dict = return_dict if return_dict is not None else self.config.use_return_dict | |
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) | |
# note: SigLIP's text model does not use a causal mask, unlike the original CLIP model. | |
# expand attention_mask | |
if attention_mask is not None and not self._use_flash_attention_2: | |
# [batch_size, seq_len] -> [batch_size, 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, | |
output_attentions=output_attentions, | |
output_hidden_states=output_hidden_states, | |
return_dict=return_dict, | |
) | |
last_hidden_state = encoder_outputs[0] | |
last_hidden_state = self.final_layer_norm(last_hidden_state) | |
# Assuming "sticky" EOS tokenization, last token is always EOS. | |
pooled_output = last_hidden_state[:, -1, :] | |
pooled_output = self.head(pooled_output) | |
if not return_dict: | |
return (last_hidden_state, pooled_output) + encoder_outputs[1:] | |
return BaseModelOutputWithPooling( | |
last_hidden_state=last_hidden_state, | |
pooler_output=pooled_output, | |
hidden_states=encoder_outputs.hidden_states, | |
attentions=encoder_outputs.attentions, | |
) | |
class SiglipTextModel(SiglipPreTrainedModel): | |
config_class = SiglipTextConfig | |
def __init__(self, config: SiglipTextConfig): | |
super().__init__(config) | |
self.text_model = SiglipTextTransformer(config) | |
# Initialize weights and apply final processing | |
self.post_init() | |
def get_input_embeddings(self) -> nn.Module: | |
return self.text_model.embeddings.token_embedding | |
def set_input_embeddings(self, value): | |
self.text_model.embeddings.token_embedding = value | |
def forward( | |
self, | |
input_ids: Optional[torch.Tensor] = None, | |
attention_mask: Optional[torch.Tensor] = None, | |
position_ids: Optional[torch.Tensor] = None, | |
output_attentions: Optional[bool] = None, | |
output_hidden_states: Optional[bool] = None, | |
return_dict: Optional[bool] = None, | |
) -> Union[Tuple, BaseModelOutputWithPooling]: | |
r""" | |
Returns: | |
Examples: | |
```python | |
>>> from transformers import AutoTokenizer, SiglipTextModel | |
>>> model = SiglipTextModel.from_pretrained("google/siglip-base-patch16-224") | |
>>> tokenizer = AutoTokenizer.from_pretrained("google/siglip-base-patch16-224") | |
>>> # important: make sure to set padding="max_length" as that's how the model was trained | |
>>> inputs = tokenizer(["a photo of a cat", "a photo of a dog"], padding="max_length", return_tensors="pt") | |
>>> outputs = model(**inputs) | |
>>> last_hidden_state = outputs.last_hidden_state | |
>>> pooled_output = outputs.pooler_output # pooled (EOS token) states | |
```""" | |
return_dict = return_dict if return_dict is not None else self.config.use_return_dict | |
return self.text_model( | |
input_ids=input_ids, | |
attention_mask=attention_mask, | |
position_ids=position_ids, | |
output_attentions=output_attentions, | |
output_hidden_states=output_hidden_states, | |
return_dict=return_dict, | |
) | |
class SiglipVisionTransformer(nn.Module): | |
def __init__(self, config: SiglipVisionConfig): | |
super().__init__() | |
self.config = config | |
embed_dim = config.hidden_size | |
self.embeddings = SiglipVisionEmbeddings(config) | |
self.encoder = SiglipEncoder(config) | |
self.post_layernorm = nn.LayerNorm(embed_dim, eps=config.layer_norm_eps) | |
self.use_head = True if not hasattr(config, "vision_use_head") else config.vision_use_head | |
if self.use_head: | |
self.head = SiglipMultiheadAttentionPoolingHead(config) | |
def forward( | |
self, | |
pixel_values, | |
output_attentions: Optional[bool] = None, | |
output_hidden_states: Optional[bool] = None, | |
return_dict: Optional[bool] = None, | |
interpolate_pos_encoding: Optional[bool] = False, | |
) -> Union[Tuple, BaseModelOutputWithPooling]: | |
r""" | |
Returns: | |
""" | |
output_attentions = output_attentions if output_attentions is not None else self.config.output_attentions | |
output_hidden_states = ( | |
output_hidden_states if output_hidden_states is not None else self.config.output_hidden_states | |
) | |
return_dict = return_dict if return_dict is not None else self.config.use_return_dict | |
hidden_states = self.embeddings(pixel_values, interpolate_pos_encoding=interpolate_pos_encoding) | |
encoder_outputs = self.encoder( | |
inputs_embeds=hidden_states, | |
output_attentions=output_attentions, | |
output_hidden_states=output_hidden_states, | |
return_dict=return_dict, | |
) | |
last_hidden_state = encoder_outputs[0] | |
last_hidden_state = self.post_layernorm(last_hidden_state) | |
pooler_output = self.head(last_hidden_state) if self.use_head else None | |
if not return_dict: | |
return (last_hidden_state, pooler_output) + encoder_outputs[1:] | |
return BaseModelOutputWithPooling( | |
last_hidden_state=last_hidden_state, | |
pooler_output=pooler_output, | |
hidden_states=encoder_outputs.hidden_states, | |
attentions=encoder_outputs.attentions, | |
) | |
class SiglipMultiheadAttentionPoolingHead(nn.Module): | |
"""Multihead Attention Pooling.""" | |
def __init__(self, config: SiglipVisionConfig): | |
super().__init__() | |
self.probe = nn.Parameter(torch.randn(1, 1, config.hidden_size)) | |
self.attention = torch.nn.MultiheadAttention(config.hidden_size, config.num_attention_heads, batch_first=True) | |
self.layernorm = nn.LayerNorm(config.hidden_size, eps=config.layer_norm_eps) | |
self.mlp = SiglipMLP(config) | |
def forward(self, hidden_state): | |
batch_size = hidden_state.shape[0] | |
probe = self.probe.repeat(batch_size, 1, 1) | |
hidden_state = self.attention(probe, hidden_state, hidden_state)[0] | |
residual = hidden_state | |
hidden_state = self.layernorm(hidden_state) | |
hidden_state = residual + self.mlp(hidden_state) | |
return hidden_state[:, 0] | |
class SiglipVisionModel(SiglipPreTrainedModel): | |
config_class = SiglipVisionConfig | |
main_input_name = "pixel_values" | |
def __init__(self, config: SiglipVisionConfig): | |
super().__init__(config) | |
self.vision_model = SiglipVisionTransformer(config) | |
# Initialize weights and apply final processing | |
self.post_init() | |
def get_input_embeddings(self) -> nn.Module: | |
return self.vision_model.embeddings.patch_embedding | |
def forward( | |
self, | |
pixel_values, | |
output_attentions: Optional[bool] = None, | |
output_hidden_states: Optional[bool] = None, | |
return_dict: Optional[bool] = None, | |
interpolate_pos_encoding: bool = False, | |
) -> Union[Tuple, BaseModelOutputWithPooling]: | |
r""" | |
Returns: | |
Examples: | |
```python | |
>>> from PIL import Image | |
>>> import requests | |
>>> from transformers import AutoProcessor, SiglipVisionModel | |
>>> model = SiglipVisionModel.from_pretrained("google/siglip-base-patch16-224") | |
>>> processor = AutoProcessor.from_pretrained("google/siglip-base-patch16-224") | |
>>> url = "http://images.cocodataset.org/val2017/000000039769.jpg" | |
>>> image = Image.open(requests.get(url, stream=True).raw) | |
>>> inputs = processor(images=image, return_tensors="pt") | |
>>> outputs = model(**inputs) | |
>>> last_hidden_state = outputs.last_hidden_state | |
>>> pooled_output = outputs.pooler_output # pooled features | |
```""" | |
return_dict = return_dict if return_dict is not None else self.config.use_return_dict | |
return self.vision_model( | |
pixel_values=pixel_values, | |
output_attentions=output_attentions, | |
output_hidden_states=output_hidden_states, | |
return_dict=return_dict, | |
interpolate_pos_encoding=interpolate_pos_encoding, | |
) | |
class SiglipModel(SiglipPreTrainedModel): | |
config_class = SiglipConfig | |
def __init__(self, config: SiglipConfig): | |
super().__init__(config) | |
if not isinstance(config.text_config, SiglipTextConfig): | |
raise TypeError( | |
"config.text_config is expected to be of type SiglipTextConfig but is of type" | |
f" {type(config.text_config)}." | |
) | |
if not isinstance(config.vision_config, SiglipVisionConfig): | |
raise TypeError( | |
"config.vision_config is expected to be of type SiglipVisionConfig but is of type" | |
f" {type(config.vision_config)}." | |
) | |
text_config = config.text_config | |
vision_config = config.vision_config | |
# First, initialize the text and vision models with proper attention implementation | |
text_model = SiglipTextModel._from_config(text_config) | |
vision_model = SiglipVisionModel._from_config(vision_config) | |
# Second, get the text and vision submodules (for backward compatibility) | |
self.text_model = text_model.text_model | |
self.vision_model = vision_model.vision_model | |
self.logit_scale = nn.Parameter(torch.randn(1)) | |
self.logit_bias = nn.Parameter(torch.randn(1)) | |
# Initialize weights and apply final processing | |
self.post_init() | |
def get_text_features( | |
self, | |
input_ids: Optional[torch.Tensor] = None, | |
attention_mask: Optional[torch.Tensor] = None, | |
position_ids: Optional[torch.Tensor] = None, | |
output_attentions: Optional[bool] = None, | |
output_hidden_states: Optional[bool] = None, | |
return_dict: Optional[bool] = None, | |
) -> torch.FloatTensor: | |
r""" | |
Returns: | |
text_features (`torch.FloatTensor` of shape `(batch_size, output_dim`): The text embeddings obtained by | |
applying the projection layer to the pooled output of [`SiglipTextModel`]. | |
Examples: | |
```python | |
>>> from transformers import AutoTokenizer, AutoModel | |
>>> import torch | |
>>> model = AutoModel.from_pretrained("google/siglip-base-patch16-224") | |
>>> tokenizer = AutoTokenizer.from_pretrained("google/siglip-base-patch16-224") | |
>>> # important: make sure to set padding="max_length" as that's how the model was trained | |
>>> inputs = tokenizer(["a photo of a cat", "a photo of a dog"], padding="max_length", return_tensors="pt") | |
>>> with torch.no_grad(): | |
... text_features = model.get_text_features(**inputs) | |
```""" | |
# Use SigLIP model's config for some fields (if specified) instead of those of vision & text components. | |
output_attentions = output_attentions if output_attentions is not None else self.config.output_attentions | |
output_hidden_states = ( | |
output_hidden_states if output_hidden_states is not None else self.config.output_hidden_states | |
) | |
return_dict = return_dict if return_dict is not None else self.config.use_return_dict | |
text_outputs = self.text_model( | |
input_ids=input_ids, | |
attention_mask=attention_mask, | |
position_ids=position_ids, | |
output_attentions=output_attentions, | |
output_hidden_states=output_hidden_states, | |
return_dict=return_dict, | |
) | |
pooled_output = text_outputs[1] | |
return pooled_output | |
def get_image_features( | |
self, | |
pixel_values: Optional[torch.FloatTensor] = None, | |
output_attentions: Optional[bool] = None, | |
output_hidden_states: Optional[bool] = None, | |
return_dict: Optional[bool] = None, | |
interpolate_pos_encoding: bool = False, | |
) -> torch.FloatTensor: | |
r""" | |
Returns: | |
image_features (`torch.FloatTensor` of shape `(batch_size, output_dim`): The image embeddings obtained by | |
applying the projection layer to the pooled output of [`SiglipVisionModel`]. | |
Examples: | |
```python | |
>>> from PIL import Image | |
>>> import requests | |
>>> from transformers import AutoProcessor, AutoModel | |
>>> import torch | |
>>> model = AutoModel.from_pretrained("google/siglip-base-patch16-224") | |
>>> processor = AutoProcessor.from_pretrained("google/siglip-base-patch16-224") | |
>>> url = "http://images.cocodataset.org/val2017/000000039769.jpg" | |
>>> image = Image.open(requests.get(url, stream=True).raw) | |
>>> inputs = processor(images=image, return_tensors="pt") | |
>>> with torch.no_grad(): | |
... image_features = model.get_image_features(**inputs) | |
```""" | |
# Use SiglipModel's config for some fields (if specified) instead of those of vision & text components. | |
output_attentions = output_attentions if output_attentions is not None else self.config.output_attentions | |
output_hidden_states = ( | |
output_hidden_states if output_hidden_states is not None else self.config.output_hidden_states | |
) | |
return_dict = return_dict if return_dict is not None else self.config.use_return_dict | |
vision_outputs = self.vision_model( | |
pixel_values=pixel_values, | |
output_attentions=output_attentions, | |
output_hidden_states=output_hidden_states, | |
return_dict=return_dict, | |
interpolate_pos_encoding=interpolate_pos_encoding, | |
) | |
pooled_output = vision_outputs[1] | |
return pooled_output | |
def forward( | |
self, | |
input_ids: Optional[torch.LongTensor] = None, | |
pixel_values: Optional[torch.FloatTensor] = None, | |
attention_mask: Optional[torch.Tensor] = None, | |
position_ids: Optional[torch.LongTensor] = None, | |
return_loss: Optional[bool] = None, | |
output_attentions: Optional[bool] = None, | |
output_hidden_states: Optional[bool] = None, | |
return_dict: Optional[bool] = None, | |
interpolate_pos_encoding: bool = False, | |
) -> Union[Tuple, SiglipOutput]: | |
r""" | |
Returns: | |
Examples: | |
```python | |
>>> from PIL import Image | |
>>> import requests | |
>>> from transformers import AutoProcessor, AutoModel | |
>>> import torch | |
>>> model = AutoModel.from_pretrained("google/siglip-base-patch16-224") | |
>>> processor = AutoProcessor.from_pretrained("google/siglip-base-patch16-224") | |
>>> url = "http://images.cocodataset.org/val2017/000000039769.jpg" | |
>>> image = Image.open(requests.get(url, stream=True).raw) | |
>>> texts = ["a photo of 2 cats", "a photo of 2 dogs"] | |
>>> # important: we pass `padding=max_length` since the model was trained with this | |
>>> inputs = processor(text=texts, images=image, padding="max_length", return_tensors="pt") | |
>>> with torch.no_grad(): | |
... outputs = model(**inputs) | |
>>> logits_per_image = outputs.logits_per_image | |
>>> probs = torch.sigmoid(logits_per_image) # these are the probabilities | |
>>> print(f"{probs[0][0]:.1%} that image 0 is '{texts[0]}'") | |
31.9% that image 0 is 'a photo of 2 cats' | |
```""" | |
# Use SigLIP model's config for some fields (if specified) instead of those of vision & text components. | |
output_attentions = output_attentions if output_attentions is not None else self.config.output_attentions | |
output_hidden_states = ( | |
output_hidden_states if output_hidden_states is not None else self.config.output_hidden_states | |
) | |
return_dict = return_dict if return_dict is not None else self.config.use_return_dict | |
vision_outputs = self.vision_model( | |
pixel_values=pixel_values, | |
output_attentions=output_attentions, | |
output_hidden_states=output_hidden_states, | |
return_dict=return_dict, | |
interpolate_pos_encoding=interpolate_pos_encoding, | |
) | |
text_outputs = self.text_model( | |
input_ids=input_ids, | |
attention_mask=attention_mask, | |
position_ids=position_ids, | |
output_attentions=output_attentions, | |
output_hidden_states=output_hidden_states, | |
return_dict=return_dict, | |
) | |
image_embeds = vision_outputs[1] | |
text_embeds = text_outputs[1] | |
# normalized features | |
image_embeds = image_embeds / image_embeds.norm(p=2, dim=-1, keepdim=True) | |
text_embeds = text_embeds / text_embeds.norm(p=2, dim=-1, keepdim=True) | |
# cosine similarity as logits | |
logits_per_text = ( | |
torch.matmul(text_embeds, image_embeds.t().to(text_embeds.device)) * self.logit_scale.exp() | |
+ self.logit_bias | |
) | |
logits_per_image = logits_per_text.t() | |
loss = None | |
if return_loss: | |
# Adapted from https://github.com/google-research/big_vision/blob/01edb81a4716f93a48be43b3a4af14e29cdb3a7f/big_vision/trainers/proj/image_text/siglip.py#L287 | |
eye = torch.eye(logits_per_text.size(0), device=logits_per_text.device) | |
m1_diag1 = -torch.ones_like(logits_per_text) + 2 * eye | |
loglik = torch.nn.functional.logsigmoid(m1_diag1 * logits_per_text) | |
nll = -torch.sum(loglik, dim=-1) | |
loss = nll.mean() | |
if not return_dict: | |
output = (logits_per_image, logits_per_text, text_embeds, image_embeds, text_outputs, vision_outputs) | |
return ((loss,) + output) if loss is not None else output | |
return SiglipOutput( | |
loss=loss, | |
logits_per_image=logits_per_image, | |
logits_per_text=logits_per_text, | |
text_embeds=text_embeds, | |
image_embeds=image_embeds, | |
text_model_output=text_outputs, | |
vision_model_output=vision_outputs, | |
) | |
class SiglipForImageClassification(SiglipPreTrainedModel): | |
main_input_name = "pixel_values" | |
def __init__(self, config: SiglipConfig) -> None: | |
super().__init__(config) | |
self.num_labels = config.num_labels | |
# Create the vision model with proper attention | |
# and take only vision_model submodule (for backward compatibility) | |
vision_model = SiglipVisionModel._from_config(config.vision_config) | |
self.vision_model = vision_model.vision_model | |
# Classifier head | |
self.classifier = ( | |
nn.Linear(config.vision_config.hidden_size, config.num_labels) if config.num_labels > 0 else nn.Identity() | |
) | |
# Initialize weights and apply final processing | |
self.post_init() | |
def forward( | |
self, | |
pixel_values: Optional[torch.Tensor] = None, | |
labels: Optional[torch.Tensor] = None, | |
output_attentions: Optional[bool] = None, | |
output_hidden_states: Optional[bool] = None, | |
return_dict: Optional[bool] = None, | |
interpolate_pos_encoding: bool = False, | |
) -> Union[tuple, ImageClassifierOutput]: | |
r""" | |
labels (`torch.LongTensor` of shape `(batch_size,)`, *optional*): | |
Labels for computing the image classification/regression loss. Indices should be in `[0, ..., | |
config.num_labels - 1]`. If `config.num_labels == 1` a regression loss is computed (Mean-Square loss), If | |
`config.num_labels > 1` a classification loss is computed (Cross-Entropy). | |
Returns: | |
Examples: | |
```python | |
>>> from transformers import AutoImageProcessor, SiglipForImageClassification | |
>>> import torch | |
>>> from PIL import Image | |
>>> import requests | |
>>> torch.manual_seed(3) # doctest: +IGNORE_RESULT | |
>>> url = "http://images.cocodataset.org/val2017/000000039769.jpg" | |
>>> image = Image.open(requests.get(url, stream=True).raw) | |
>>> # note: we are loading a `SiglipModel` from the hub here, | |
>>> # so the head will be randomly initialized, hence the predictions will be random if seed is not set above. | |
>>> image_processor = AutoImageProcessor.from_pretrained("google/siglip-base-patch16-224") | |
>>> model = SiglipForImageClassification.from_pretrained("google/siglip-base-patch16-224") | |
>>> inputs = image_processor(images=image, return_tensors="pt") | |
>>> outputs = model(**inputs) | |
>>> logits = outputs.logits | |
>>> # model predicts one of the two classes | |
>>> predicted_class_idx = logits.argmax(-1).item() | |
>>> print("Predicted class:", model.config.id2label[predicted_class_idx]) | |
Predicted class: LABEL_1 | |
```""" | |
output_attentions = output_attentions if output_attentions is not None else self.config.output_attentions | |
output_hidden_states = ( | |
output_hidden_states if output_hidden_states is not None else self.config.output_hidden_states | |
) | |
return_dict = return_dict if return_dict is not None else self.config.use_return_dict | |
outputs = self.vision_model( | |
pixel_values, | |
output_attentions=output_attentions, | |
output_hidden_states=output_hidden_states, | |
return_dict=return_dict, | |
interpolate_pos_encoding=interpolate_pos_encoding, | |
) | |
sequence_output = outputs[0] | |
# average pool the patch tokens | |
sequence_output = torch.mean(sequence_output, dim=1) | |
# apply classifier | |
logits = self.classifier(sequence_output) | |
loss = None | |
if labels is not None: | |
# move labels to correct device to enable model parallelism | |
labels = labels.to(logits.device) | |
if self.config.problem_type is None: | |
if self.num_labels == 1: | |
self.config.problem_type = "regression" | |
elif self.num_labels > 1 and (labels.dtype == torch.long or labels.dtype == torch.int): | |
self.config.problem_type = "single_label_classification" | |
else: | |
self.config.problem_type = "multi_label_classification" | |
if self.config.problem_type == "regression": | |
loss_fct = MSELoss() | |
if self.num_labels == 1: | |
loss = loss_fct(logits.squeeze(), labels.squeeze()) | |
else: | |
loss = loss_fct(logits, labels) | |
elif self.config.problem_type == "single_label_classification": | |
loss_fct = CrossEntropyLoss() | |
loss = loss_fct(logits.view(-1, self.num_labels), labels.view(-1)) | |
elif self.config.problem_type == "multi_label_classification": | |
loss_fct = BCEWithLogitsLoss() | |
loss = loss_fct(logits, labels) | |
if not return_dict: | |
output = (logits,) + outputs[2:] | |
return ((loss,) + output) if loss is not None else output | |
return ImageClassifierOutput( | |
loss=loss, | |
logits=logits, | |
hidden_states=outputs.hidden_states, | |
attentions=outputs.attentions, | |
) | |