| | |
| | |
| | |
| | |
| | |
| | |
| | |
| | |
| | |
| | |
| | |
| | |
| | |
| | |
| |
|
| | |
| | |
| | |
| | |
| |
|
| | """ |
| | EVA-CLIP backbone used in BLIP2. |
| | |
| | Code adapted from: |
| | https://github.com/salesforce/LAVIS/blob/main/lavis/models/eva_vit.py |
| | """ |
| |
|
| |
|
| | import math |
| | from functools import partial |
| | from logging import getLogger |
| | from typing import Any, Optional, Tuple, Union |
| |
|
| | import torch |
| | import torch.nn as nn |
| | import torch.nn.functional as F |
| | import torch.utils.checkpoint as checkpoint |
| |
|
| | logger = getLogger(__file__) |
| |
|
| | TRANSFORMER_ENGINE_AVAILABLE = False |
| | try: |
| | import transformer_engine.pytorch as te |
| | from transformer_engine.common.recipe import DelayedScaling, Format |
| |
|
| | TRANSFORMER_ENGINE_AVAILABLE = True |
| | logger.info("Transformer Engine is available, can set `transformer_engine=True` in config " "for faster inference.") |
| | except ImportError: |
| | pass |
| |
|
| |
|
| | def drop_path(x, drop_prob: float = 0.0, training: bool = False, scale_by_keep: bool = True): |
| | """Drop paths (Stochastic Depth) per sample (when applied in main path of residual blocks). |
| | |
| | From https://github.com/huggingface/pytorch-image-models/blob/main/timm/layers/drop.py |
| | """ |
| | if drop_prob == 0.0 or not training: |
| | return x |
| | keep_prob = 1 - drop_prob |
| | shape = (x.shape[0],) + (1,) * (x.ndim - 1) |
| | random_tensor = x.new_empty(shape).bernoulli_(keep_prob) |
| | if keep_prob > 0.0 and scale_by_keep: |
| | random_tensor.div_(keep_prob) |
| | return x * random_tensor |
| |
|
| |
|
| | class DropPath(nn.Module): |
| | """Drop paths (Stochastic Depth) per sample (when applied in main path of residual blocks).""" |
| |
|
| | def __init__(self, drop_prob: float) -> None: |
| | super().__init__() |
| | self.drop_prob = drop_prob |
| |
|
| | def forward(self, x: torch.Tensor) -> torch.Tensor: |
| | return drop_path(x, self.drop_prob, self.training) |
| |
|
| | def extra_repr(self) -> str: |
| | return "p={}".format(self.drop_prob) |
| |
|
| |
|
| | class Mlp(nn.Module): |
| | def __init__( |
| | self, |
| | in_features: int, |
| | hidden_features: Optional[int] = None, |
| | out_features: Optional[int] = None, |
| | act_layer=nn.GELU, |
| | drop: float = 0.0, |
| | transformer_engine: bool = False, |
| | ) -> None: |
| | super().__init__() |
| | out_features = out_features or in_features |
| | hidden_features = hidden_features or in_features |
| | fn = te.Linear if transformer_engine else nn.Linear |
| | self.fc1 = fn(in_features, hidden_features) |
| | self.act = act_layer() |
| | self.fc2 = fn(hidden_features, out_features) |
| | self.drop = nn.Dropout(drop) |
| |
|
| | def forward(self, x): |
| | x = self.fc1(x) |
| | x = self.act(x) |
| | x = self.fc2(x) |
| | x = self.drop(x) |
| | return x |
| |
|
| |
|
| | class Attention(nn.Module): |
| | def __init__( |
| | self, |
| | dim, |
| | num_heads=8, |
| | qkv_bias=False, |
| | qk_scale=None, |
| | attn_drop=0.0, |
| | proj_drop=0.0, |
| | window_size=None, |
| | attn_head_dim=None, |
| | **kwargs, |
| | ): |
| | super().__init__() |
| | self.num_heads = num_heads |
| | head_dim = dim // num_heads |
| | if attn_head_dim is not None: |
| | head_dim = attn_head_dim |
| | all_head_dim = head_dim * self.num_heads |
| | self.scale = qk_scale or head_dim**-0.5 |
| |
|
| | self.qkv = nn.Linear(dim, all_head_dim * 3, bias=qkv_bias) |
| |
|
| | if window_size: |
| | self.window_size = window_size |
| | self.num_relative_distance = (2 * window_size[0] - 1) * (2 * window_size[1] - 1) + 3 |
| | self.relative_position_bias_table = nn.Parameter( |
| | torch.zeros(self.num_relative_distance, num_heads) |
| | ) |
| | |
| |
|
| | |
| | coords_h = torch.arange(window_size[0]) |
| | coords_w = torch.arange(window_size[1]) |
| | coords = torch.stack(torch.meshgrid([coords_h, coords_w])) |
| | coords_flatten = torch.flatten(coords, 1) |
| | relative_coords = coords_flatten[:, :, None] - coords_flatten[:, None, :] |
| | relative_coords = relative_coords.permute(1, 2, 0).contiguous() |
| | relative_coords[:, :, 0] += window_size[0] - 1 |
| | relative_coords[:, :, 1] += window_size[1] - 1 |
| | relative_coords[:, :, 0] *= 2 * window_size[1] - 1 |
| | relative_position_index = torch.zeros( |
| | size=(window_size[0] * window_size[1] + 1,) * 2, dtype=relative_coords.dtype |
| | ) |
| | relative_position_index[1:, 1:] = relative_coords.sum(-1) |
| | relative_position_index[0, 0:] = self.num_relative_distance - 3 |
| | relative_position_index[0:, 0] = self.num_relative_distance - 2 |
| | relative_position_index[0, 0] = self.num_relative_distance - 1 |
| |
|
| | self.register_buffer("relative_position_index", relative_position_index) |
| | else: |
| | self.window_size = None |
| | self.relative_position_bias_table = None |
| | self.relative_position_index = None |
| |
|
| | self.attn_drop = nn.Dropout(attn_drop) |
| | self.proj = nn.Linear(all_head_dim, dim) |
| | self.proj_drop = nn.Dropout(proj_drop) |
| |
|
| | def forward(self, x, rel_pos_bias=None): |
| | B, N, C = x.shape |
| | qkv = self.qkv(x) |
| | qkv = qkv.reshape(B, N, 3, self.num_heads, -1).permute(2, 0, 3, 1, 4) |
| | q, k, v = qkv[0], qkv[1], qkv[2] |
| |
|
| | q = q * self.scale |
| | attn = q @ k.transpose(-2, -1) |
| |
|
| | if self.relative_position_bias_table is not None: |
| | relative_position_bias = self.relative_position_bias_table[self.relative_position_index.view(-1)].view( |
| | self.window_size[0] * self.window_size[1] + 1, self.window_size[0] * self.window_size[1] + 1, -1 |
| | ) |
| | relative_position_bias = relative_position_bias.permute(2, 0, 1).contiguous() |
| | attn = attn + relative_position_bias.unsqueeze(0) |
| |
|
| | if rel_pos_bias is not None: |
| | attn = attn + rel_pos_bias |
| |
|
| | attn = attn.softmax(dim=-1) |
| | attn = self.attn_drop(attn) |
| |
|
| | x = (attn @ v).transpose(1, 2).reshape(B, N, -1) |
| | x = self.proj(x) |
| | x = self.proj_drop(x) |
| | return x |
| |
|
| |
|
| | class TransformerEngineAttention(nn.Module): |
| | def __init__( |
| | self, |
| | dim: int, |
| | num_heads: int = 8, |
| | qkv_bias: bool = False, |
| | qk_scale: Optional[float] = None, |
| | attn_drop: float = 0.0, |
| | proj_drop: float = 0.0, |
| | window_size: Optional[int] = None, |
| | attn_head_dim: Optional[int] = None, |
| | checkpoint_attention: bool = False, |
| | ): |
| | super().__init__() |
| | self.num_heads = num_heads |
| | self.checkpoint_attention = checkpoint_attention |
| | head_dim = dim // num_heads |
| | if attn_head_dim is not None: |
| | head_dim = attn_head_dim |
| | all_head_dim = head_dim * self.num_heads |
| | self.scale = qk_scale or head_dim**-0.5 |
| |
|
| | |
| | self.qkv = te.Linear(dim, all_head_dim * 3, bias=qkv_bias) |
| |
|
| | if window_size: |
| | raise NotImplementedError("`window_size` not implemented for TE!") |
| |
|
| | self.te_attn = te.DotProductAttention( |
| | num_attention_heads=num_heads, |
| | kv_channels=head_dim, |
| | attention_dropout=attn_drop, |
| | qkv_format="bshd", |
| | softmax_scale=self.scale, |
| | attn_mask_type="no_mask", |
| | ) |
| |
|
| | |
| | self.proj = te.Linear(all_head_dim, dim) |
| | self.proj_drop = nn.Dropout(proj_drop) |
| |
|
| | def forward(self, x: torch.Tensor, rel_pos_bias: Optional[torch.Tensor] = None) -> torch.Tensor: |
| | """ |
| | x: [B, N, C] |
| | rel_pos_bias (optional): tensor of shape [num_heads, N, N] |
| | """ |
| | B, N, _ = x.shape |
| | qkv = self.qkv(x) |
| | qkv = qkv.reshape(B, N, 3, self.num_heads, -1).permute(2, 0, 1, 3, 4) |
| | q, k, v = qkv[0], qkv[1], qkv[2] |
| |
|
| | if rel_pos_bias is not None: |
| | raise NotImplementedError("`rel_pos_bias` not implemented for TE!") |
| |
|
| | |
| | y = self.te_attn(q, k, v, checkpoint_core_attention=self.checkpoint_attention) |
| |
|
| | |
| | return self.proj_drop(self.proj(y)) |
| |
|
| |
|
| | class Block(nn.Module): |
| | def __init__( |
| | self, |
| | dim, |
| | num_heads, |
| | mlp_ratio=4.0, |
| | qkv_bias=False, |
| | qk_scale=None, |
| | drop=0.0, |
| | attn_drop=0.0, |
| | drop_path=0.0, |
| | init_values=None, |
| | act_layer=nn.GELU, |
| | norm_layer=nn.LayerNorm, |
| | window_size=None, |
| | attn_head_dim=None, |
| | transformer_engine: bool = False, |
| | checkpoint_attention: bool = False, |
| | ): |
| | super().__init__() |
| | self.transformer_engine = transformer_engine |
| | self.window_size = window_size |
| | self.checkpoint_attention = checkpoint_attention |
| |
|
| | if checkpoint_attention and not transformer_engine: |
| | raise ValueError("`checkpoint_attention` needs `transformer_engine`!") |
| |
|
| | self.norm1 = norm_layer(dim) |
| | attn_fn = TransformerEngineAttention if transformer_engine else Attention |
| | self.attn = attn_fn( |
| | dim, |
| | num_heads=num_heads, |
| | qkv_bias=qkv_bias, |
| | qk_scale=qk_scale, |
| | attn_drop=attn_drop, |
| | proj_drop=drop, |
| | window_size=window_size, |
| | attn_head_dim=attn_head_dim, |
| | checkpoint_attention=checkpoint_attention, |
| | ) |
| |
|
| | |
| | self.drop_path = DropPath(drop_path) if drop_path > 0.0 else nn.Identity() |
| | self.norm2 = norm_layer(dim) |
| | mlp_hidden_dim = int(dim * mlp_ratio) |
| | self.mlp = Mlp( |
| | in_features=dim, |
| | hidden_features=mlp_hidden_dim, |
| | act_layer=act_layer, |
| | drop=drop, |
| | transformer_engine=transformer_engine, |
| | ) |
| |
|
| | if init_values is not None and init_values > 0: |
| | self.gamma_1 = nn.Parameter(init_values * torch.ones((dim)), requires_grad=True) |
| | self.gamma_2 = nn.Parameter(init_values * torch.ones((dim)), requires_grad=True) |
| | else: |
| | self.gamma_1, self.gamma_2 = None, None |
| |
|
| | def forward(self, x, rel_pos_bias=None): |
| | if self.gamma_1 is None: |
| | x = x + self.drop_path(self.attn(self.norm1(x), rel_pos_bias=rel_pos_bias)) |
| | x = x + self.drop_path(self.mlp(self.norm2(x))) |
| | else: |
| | x = x + self.drop_path(self.gamma_1 * self.attn(self.norm1(x), rel_pos_bias=rel_pos_bias)) |
| | x = x + self.drop_path(self.gamma_2 * self.mlp(self.norm2(x))) |
| | return x |
| |
|
| |
|
| | class PatchEmbed(nn.Module): |
| | """Image to Patch Embedding""" |
| |
|
| | def __init__( |
| | self, |
| | img_size: Union[int, Tuple[int, int]] = 224, |
| | patch_size: Union[int, Tuple[int, int]] = 16, |
| | in_chans: int = 3, |
| | embed_dim: int = 768, |
| | ): |
| | super().__init__() |
| | img_size = (img_size, img_size) if isinstance(img_size, int) else img_size |
| | patch_size = (patch_size, patch_size) if isinstance(patch_size, int) else patch_size |
| | num_patches = (img_size[1] // patch_size[1]) * (img_size[0] // patch_size[0]) |
| | self.patch_shape = (img_size[0] // patch_size[0], img_size[1] // patch_size[1]) |
| | self.img_size = img_size |
| | self.patch_size = patch_size |
| | self.num_patches = num_patches |
| |
|
| | self.proj = nn.Conv2d(in_chans, embed_dim, kernel_size=patch_size, stride=patch_size) |
| |
|
| | def forward(self, x, **kwargs): |
| | B, C, H, W = x.shape |
| | assert ( |
| | H == self.img_size[0] and W == self.img_size[1] |
| | ), f"Input image size ({H}*{W}) doesn't match model ({self.img_size[0]}*{self.img_size[1]})." |
| | x = self.proj(x).flatten(2).transpose(1, 2) |
| | return x |
| |
|
| |
|
| | class RelativePositionBias(nn.Module): |
| | def __init__(self, window_size, num_heads): |
| | super().__init__() |
| | self.window_size = window_size |
| | self.num_relative_distance = (2 * window_size[0] - 1) * (2 * window_size[1] - 1) + 3 |
| | self.relative_position_bias_table = nn.Parameter( |
| | torch.zeros(self.num_relative_distance, num_heads) |
| | ) |
| | |
| |
|
| | |
| | coords_h = torch.arange(window_size[0]) |
| | coords_w = torch.arange(window_size[1]) |
| | coords = torch.stack(torch.meshgrid([coords_h, coords_w])) |
| | coords_flatten = torch.flatten(coords, 1) |
| | relative_coords = coords_flatten[:, :, None] - coords_flatten[:, None, :] |
| | relative_coords = relative_coords.permute(1, 2, 0).contiguous() |
| | relative_coords[:, :, 0] += window_size[0] - 1 |
| | relative_coords[:, :, 1] += window_size[1] - 1 |
| | relative_coords[:, :, 0] *= 2 * window_size[1] - 1 |
| | relative_position_index = torch.zeros( |
| | size=(window_size[0] * window_size[1] + 1,) * 2, dtype=relative_coords.dtype |
| | ) |
| | relative_position_index[1:, 1:] = relative_coords.sum(-1) |
| | relative_position_index[0, 0:] = self.num_relative_distance - 3 |
| | relative_position_index[0:, 0] = self.num_relative_distance - 2 |
| | relative_position_index[0, 0] = self.num_relative_distance - 1 |
| |
|
| | self.register_buffer("relative_position_index", relative_position_index) |
| |
|
| | def forward(self): |
| | relative_position_bias = self.relative_position_bias_table[self.relative_position_index.view(-1)].view( |
| | self.window_size[0] * self.window_size[1] + 1, self.window_size[0] * self.window_size[1] + 1, -1 |
| | ) |
| | return relative_position_bias.permute(2, 0, 1).contiguous() |
| |
|
| |
|
| | class VisionTransformer(nn.Module): |
| | """Vision Transformer with support for patch or hybrid CNN input stage""" |
| |
|
| | def __init__( |
| | self, |
| | img_size=224, |
| | patch_size=16, |
| | in_chans=3, |
| | num_classes=1000, |
| | embed_dim=768, |
| | depth=12, |
| | num_heads=12, |
| | mlp_ratio=4.0, |
| | qkv_bias=False, |
| | qk_scale=None, |
| | drop_rate=0.0, |
| | attn_drop_rate=0.0, |
| | drop_path_rate=0.0, |
| | norm_layer=nn.LayerNorm, |
| | init_values=None, |
| | use_abs_pos_emb=True, |
| | use_rel_pos_bias=False, |
| | use_shared_rel_pos_bias=False, |
| | use_mean_pooling=True, |
| | init_scale=0.001, |
| | checkpoint_activations: bool = False, |
| | checkpoint_attention: bool = False, |
| | transformer_engine: bool = False, |
| | use_fp8: bool = False, |
| | ): |
| | super().__init__() |
| | self.image_size = img_size |
| | self.patch_size = patch_size |
| | self.num_classes = num_classes |
| | self.num_features = self.embed_dim = embed_dim |
| | self.transformer_engine = transformer_engine |
| | self.use_fp8 = use_fp8 |
| | self.fp8_recipe = None |
| |
|
| | if use_fp8 and not transformer_engine: |
| | raise ValueError("`transformer_engine` must be enabled for `use_fp8`.") |
| | if use_fp8: |
| | |
| | self.fp8_recipe = DelayedScaling(fp8_format=Format.HYBRID, amax_history_len=16, amax_compute_algo="max") |
| |
|
| | self.patch_embed = PatchEmbed(img_size=img_size, patch_size=patch_size, in_chans=in_chans, embed_dim=embed_dim) |
| | num_patches = self.patch_embed.num_patches |
| |
|
| | self.cls_token = nn.Parameter(torch.zeros(1, 1, embed_dim)) |
| | if use_abs_pos_emb: |
| | self.pos_embed = nn.Parameter(torch.zeros(1, num_patches + 1, embed_dim)) |
| | else: |
| | self.pos_embed = None |
| | self.pos_drop = nn.Dropout(p=drop_rate) |
| |
|
| | if use_shared_rel_pos_bias: |
| | self.rel_pos_bias = RelativePositionBias(window_size=self.patch_embed.patch_shape, num_heads=num_heads) |
| | else: |
| | self.rel_pos_bias = None |
| | self.checkpoint_activations = checkpoint_activations |
| | self.checkpoint_attention = checkpoint_attention |
| |
|
| | dpr = [x.item() for x in torch.linspace(0, drop_path_rate, depth)] |
| | self.use_rel_pos_bias = use_rel_pos_bias |
| | self.blocks = nn.ModuleList( |
| | [ |
| | Block( |
| | dim=embed_dim, |
| | num_heads=num_heads, |
| | mlp_ratio=mlp_ratio, |
| | qkv_bias=qkv_bias, |
| | qk_scale=qk_scale, |
| | drop=drop_rate, |
| | attn_drop=attn_drop_rate, |
| | drop_path=dpr[i], |
| | norm_layer=norm_layer, |
| | init_values=init_values, |
| | window_size=self.patch_embed.patch_shape if use_rel_pos_bias else None, |
| | transformer_engine=transformer_engine, |
| | checkpoint_attention=self.checkpoint_attention, |
| | ) |
| | for i in range(depth) |
| | ] |
| | ) |
| |
|
| | if self.pos_embed is not None: |
| | nn.init.trunc_normal_(self.pos_embed, std=0.02) |
| | nn.init.trunc_normal_(self.cls_token, std=0.02) |
| |
|
| | self.apply(self._init_weights) |
| | self.fix_init_weight() |
| |
|
| | def fix_init_weight(self): |
| | def rescale(param, layer_id): |
| | param.div_(math.sqrt(2.0 * layer_id)) |
| |
|
| | for layer_id, layer in enumerate(self.blocks): |
| | rescale(layer.attn.proj.weight.data, layer_id + 1) |
| | rescale(layer.mlp.fc2.weight.data, layer_id + 1) |
| |
|
| | def _init_weights(self, m): |
| | if isinstance(m, nn.Linear): |
| | nn.init.trunc_normal_(m.weight, std=0.02) |
| | if isinstance(m, nn.Linear) and m.bias is not None: |
| | nn.init.constant_(m.bias, 0) |
| | elif isinstance(m, nn.LayerNorm): |
| | nn.init.constant_(m.bias, 0) |
| | nn.init.constant_(m.weight, 1.0) |
| |
|
| | def get_classifier(self): |
| | return self.head |
| |
|
| | def reset_classifier(self, num_classes, global_pool=""): |
| | self.num_classes = num_classes |
| | self.head = nn.Linear(self.embed_dim, num_classes) if num_classes > 0 else nn.Identity() |
| |
|
| | def forward_features(self, x): |
| | if self.transformer_engine and self.use_fp8: |
| | with te.fp8_autocast(enabled=True, fp8_recipe=self.fp8_recipe): |
| | return self._forward_uncast(x) |
| | return self._forward_uncast(x) |
| |
|
| | def _forward_uncast(self, x): |
| | x = self.patch_embed(x) |
| | batch_size, seq_len, _ = x.size() |
| |
|
| | cls_tokens = self.cls_token.expand(batch_size, -1, -1) |
| | x = torch.cat((cls_tokens, x), dim=1) |
| | if self.pos_embed is not None: |
| | x = x + self.pos_embed |
| | x = self.pos_drop(x) |
| |
|
| | rel_pos_bias = self.rel_pos_bias() if self.rel_pos_bias is not None else None |
| | for blk in self.blocks: |
| | if self.checkpoint_activations: |
| | x = checkpoint.checkpoint(blk, x, rel_pos_bias) |
| | else: |
| | x = blk(x, rel_pos_bias) |
| | return x |
| |
|
| | def forward(self, x): |
| | x = self.forward_features(x) |
| | return x |
| |
|
| | def get_intermediate_layers(self, x): |
| | x = self.patch_embed(x) |
| | batch_size, seq_len, _ = x.size() |
| |
|
| | cls_tokens = self.cls_token.expand(batch_size, -1, -1) |
| | x = torch.cat((cls_tokens, x), dim=1) |
| | if self.pos_embed is not None: |
| | x = x + self.pos_embed |
| | x = self.pos_drop(x) |
| |
|
| | features = [] |
| | rel_pos_bias = self.rel_pos_bias() if self.rel_pos_bias is not None else None |
| | for blk in self.blocks: |
| | x = blk(x, rel_pos_bias) |
| | features.append(x) |
| |
|
| | return features |
| |
|
| | def get_num_layer(self, var_name=""): |
| | if var_name in ("cls_token", "mask_token", "pos_embed"): |
| | return 0 |
| | elif var_name.startswith("patch_embed"): |
| | return 0 |
| | elif var_name.startswith("rel_pos_bias"): |
| | return len(self.blocks) - 1 |
| | elif var_name.startswith("blocks"): |
| | layer_id = int(var_name.split(".")[1]) |
| | return layer_id + 1 |
| | else: |
| | return len(self.blocks) |
| |
|
| |
|
| | def interpolate_pos_embed( |
| | pos_embed_key: str, |
| | num_patches: int, |
| | patch_embed_shape: torch.Size, |
| | checkpoint_model: dict[str, torch.Tensor], |
| | target_h: int = None, |
| | target_w: int = None, |
| | ) -> None: |
| | if pos_embed_key in checkpoint_model: |
| | pos_embed_checkpoint = checkpoint_model[pos_embed_key].float() |
| | embedding_size = pos_embed_checkpoint.shape[-1] |
| | num_extra_tokens = patch_embed_shape - num_patches |
| | |
| | orig_size = int((pos_embed_checkpoint.shape[-2] - num_extra_tokens) ** 0.5) |
| |
|
| | |
| | if target_h is not None and target_w is not None: |
| | new_h, new_w = target_h, target_w |
| | else: |
| | |
| | new_size = int(num_patches**0.5) |
| | new_h, new_w = new_size, new_size |
| |
|
| | |
| | if orig_size * orig_size != new_h * new_w: |
| | logger.info("Positional interpolation from %dx%d to %dx%d" % (orig_size, orig_size, new_h, new_w)) |
| | extra_tokens = pos_embed_checkpoint[:, :num_extra_tokens] |
| | |
| | pos_tokens = pos_embed_checkpoint[:, num_extra_tokens:] |
| | pos_tokens = pos_tokens.reshape(-1, orig_size, orig_size, embedding_size).permute(0, 3, 1, 2) |
| | pos_tokens = torch.nn.functional.interpolate( |
| | pos_tokens, size=(new_h, new_w), mode="bicubic", align_corners=False |
| | ) |
| | pos_tokens = pos_tokens.permute(0, 2, 3, 1).flatten(1, 2) |
| | new_pos_embed = torch.cat((extra_tokens, pos_tokens), dim=1) |
| | checkpoint_model[pos_embed_key] = new_pos_embed |
| |
|
| |
|
| | class PositionalEmbeddingHook: |
| | def __init__(self, pos_embed_name, num_patches, patch_embed_shape, target_h=None, target_w=None): |
| | self.pos_embed_name = pos_embed_name |
| | self.num_patches = num_patches |
| | self.patch_embed_shape = patch_embed_shape |
| | self.target_h = target_h |
| | self.target_w = target_w |
| |
|
| | def __call__(self, state_dict, prefix, *args, **kwargs) -> None: |
| | logger.info("Calling `PositionalEmbeddingHook`") |
| | pos_embed_key = f"{prefix}{self.pos_embed_name}" |
| | interpolate_pos_embed( |
| | pos_embed_key, self.num_patches, self.patch_embed_shape, state_dict, self.target_h, self.target_w |
| | ) |
| |
|
| |
|
| | class EvaViTG(VisionTransformer): |
| | def __init__( |
| | self, |
| | img_size: Union[int, Tuple[int, int]] = 224, |
| | drop_path_rate: float = 0.4, |
| | pretrained: bool = False, |
| | checkpoint_path: Optional[str] = None, |
| | checkpoint_activations: bool = False, |
| | checkpoint_attention: bool = False, |
| | transformer_engine: bool = False, |
| | use_fp8: bool = False, |
| | **kwargs: Any, |
| | ) -> None: |
| | if not TRANSFORMER_ENGINE_AVAILABLE and transformer_engine: |
| | raise ValueError( |
| | "TransformerEngine is not available, " |
| | "please install transformer-engine or set `transformer_engine=False` in config." |
| | ) |
| | if use_fp8 and not transformer_engine: |
| | raise ValueError("`transformer_engine` must be enabled for `use_fp8`.") |
| | super().__init__( |
| | img_size=img_size, |
| | patch_size=14, |
| | use_mean_pooling=False, |
| | embed_dim=1408, |
| | depth=39, |
| | num_heads=1408 // 88, |
| | mlp_ratio=4.3637, |
| | qkv_bias=True, |
| | drop_path_rate=drop_path_rate, |
| | norm_layer=partial(nn.LayerNorm, eps=1e-6), |
| | checkpoint_activations=checkpoint_activations, |
| | checkpoint_attention=checkpoint_attention, |
| | transformer_engine=transformer_engine, |
| | use_fp8=use_fp8, |
| | ) |
| | self.checkpoint_path = checkpoint_path |
| |
|
| | |
| | self.register_pre_hooks() |
| |
|
| | |
| | if pretrained: |
| | self.load_checkpoint() |
| |
|
| | def load_checkpoint(self) -> None: |
| | logger.info(f"Loading checkpoint from {self.checkpoint_path}") |
| | state_dict = torch.load(self.checkpoint_path, map_location="cpu") |
| | incompatible_keys = self.load_state_dict(state_dict, strict=False) |
| | logger.info(f"Incompatible keys: {incompatible_keys}") |
| | logger.info(f"Loaded visual encoder {type(self)} with state dict from {self.checkpoint_path}") |
| |
|
| | def register_pre_hooks(self) -> None: |
| | """Register positional embedding interpolation when loading pre-trained checkpoints using different resolution.""" |
| | |
| | patch_h = self.patch_embed.patch_shape[0] |
| | patch_w = self.patch_embed.patch_shape[1] |
| |
|
| | embed_hook = PositionalEmbeddingHook( |
| | pos_embed_name="pos_embed", |
| | num_patches=self.patch_embed.num_patches, |
| | patch_embed_shape=self.pos_embed.shape[-2], |
| | target_h=patch_h, |
| | target_w=patch_w, |
| | ) |
| | self._register_load_state_dict_pre_hook(embed_hook) |
| |
|
| | def _initialize_weights(self, m): |
| | if isinstance(m, nn.Linear): |
| | nn.init.trunc_normal_(m.weight, std=0.02) |
| | if isinstance(m, nn.Linear) and m.bias is not None: |
| | nn.init.constant_(m.bias, 0) |
| | elif isinstance(m, nn.LayerNorm): |
| | nn.init.constant_(m.bias, 0) |
| | nn.init.constant_(m.weight, 1.0) |
| |
|