# modeling_internvideo2.py import logging import math import numpy as np import torch import torch.nn as nn import torch.nn.functional as F from transformers import PreTrainedModel from transformers.utils import logging as hf_logging from torch.utils.checkpoint import checkpoint # Correct from functools import partial from .configuration_internvideo2 import InternVideo2Config # Import the configuration try: from einops import rearrange except ImportError: raise ImportError("Please install einops to use this model.") try: from timm.models.layers import DropPath, to_2tuple except ImportError: raise ImportError("Please install timm to use this model.") logger = hf_logging.get_logger(__name__) # Position embedding functions def get_3d_sincos_pos_embed(embed_dim, grid_size, t_size, cls_token=False): assert embed_dim % 4 == 0 embed_dim_spatial = embed_dim // 4 * 3 embed_dim_temporal = embed_dim // 4 # Spatial grid_h = np.arange(grid_size, dtype=np.float32) grid_w = np.arange(grid_size, dtype=np.float32) grid = np.meshgrid(grid_w, grid_h) # W first grid = np.stack(grid, axis=0) grid = grid.reshape([2, 1, grid_size, grid_size]) pos_embed_spatial = get_2d_sincos_pos_embed_from_grid(embed_dim_spatial, grid) # Temporal grid_t = np.arange(t_size, dtype=np.float32) pos_embed_temporal = get_1d_sincos_pos_embed_from_grid(embed_dim_temporal, grid_t) # Combine spatial and temporal embeddings pos_embed_temporal = pos_embed_temporal[:, np.newaxis, :] pos_embed_temporal = np.repeat(pos_embed_temporal, grid_size**2, axis=1) pos_embed_spatial = pos_embed_spatial[np.newaxis, :, :] pos_embed_spatial = np.repeat(pos_embed_spatial, t_size, axis=0) pos_embed = np.concatenate([pos_embed_temporal, pos_embed_spatial], axis=-1) pos_embed = pos_embed.reshape([-1, embed_dim]) if cls_token: pos_embed = np.concatenate([np.zeros([1, embed_dim]), pos_embed], axis=0) return pos_embed def get_2d_sincos_pos_embed_from_grid(embed_dim, grid): assert embed_dim % 2 == 0 emb_h = get_1d_sincos_pos_embed_from_grid(embed_dim // 2, grid[0]) emb_w = get_1d_sincos_pos_embed_from_grid(embed_dim // 2, grid[1]) emb = np.concatenate([emb_h, emb_w], axis=1) return emb def get_1d_sincos_pos_embed_from_grid(embed_dim, pos): assert embed_dim % 2 == 0 omega = np.arange(embed_dim // 2, dtype=np.float32) omega /= embed_dim / 2.0 omega = 1.0 / (10000 ** omega) pos = pos.reshape(-1) out = np.einsum('m,d->md', pos, omega) emb_sin = np.sin(out) emb_cos = np.cos(out) emb = np.concatenate([emb_sin, emb_cos], axis=1) return emb # Define necessary classes: CrossAttention, AttentiveBlock, AttentionPoolingBlock, RMSNorm, LayerScale, Attention, Mlp, Block, PatchEmbed, Linear_Decoder class CrossAttention(nn.Module): def __init__( self, dim, num_heads=8, qkv_bias=False, qk_scale=None, attn_drop=0.0, proj_drop=0.0, attn_head_dim=None, out_dim=None, ): super().__init__() if out_dim is None: out_dim = dim 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 assert all_head_dim == dim self.q = nn.Linear(dim, all_head_dim, bias=False) self.k = nn.Linear(dim, all_head_dim, bias=False) self.v = nn.Linear(dim, all_head_dim, bias=False) if qkv_bias: self.q_bias = nn.Parameter(torch.zeros(all_head_dim)) self.k_bias = nn.Parameter(torch.zeros(all_head_dim)) self.v_bias = nn.Parameter(torch.zeros(all_head_dim)) else: self.q_bias = None self.k_bias = None self.v_bias = None self.attn_drop = nn.Dropout(attn_drop) self.proj = nn.Linear(all_head_dim, out_dim) self.proj_drop = nn.Dropout(proj_drop) def forward(self, x, k=None, v=None): B, N, C = x.shape N_k = k.shape[1] N_v = v.shape[1] q_bias, k_bias, v_bias = None, None, None if self.q_bias is not None: q_bias = self.q_bias k_bias = self.k_bias v_bias = self.v_bias q = F.linear(input=x, weight=self.q.weight, bias=q_bias) q = ( q.reshape(B, N, 1, self.num_heads, -1) .permute(2, 0, 3, 1, 4) .squeeze(0) ) # (B, N_head, N_q, dim) k = F.linear(input=k, weight=self.k.weight, bias=k_bias) k = ( k.reshape(B, N_k, 1, self.num_heads, -1) .permute(2, 0, 3, 1, 4) .squeeze(0) ) v = F.linear(input=v, weight=self.v.weight, bias=v_bias) v = ( v.reshape(B, N_v, 1, self.num_heads, -1) .permute(2, 0, 3, 1, 4) .squeeze(0) ) q = q * self.scale attn = q @ k.transpose(-2, -1) # (B, N_head, N_q, N_k) 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 AttentiveBlock(nn.Module): def __init__( self, dim, num_heads, qkv_bias=False, qk_scale=None, drop=0.0, attn_drop=0.0, drop_path=0.0, norm_layer=nn.LayerNorm, attn_head_dim=None, out_dim=None, ): super().__init__() self.norm1_q = norm_layer(dim) self.norm1_k = norm_layer(dim) self.norm1_v = norm_layer(dim) self.cross_attn = CrossAttention( dim, num_heads=num_heads, qkv_bias=qkv_bias, qk_scale=qk_scale, attn_drop=attn_drop, proj_drop=drop, attn_head_dim=attn_head_dim, out_dim=out_dim, ) self.drop_path = DropPath(drop_path) if drop_path > 0.0 else nn.Identity() def forward( self, x_q, x_kv, pos_q, pos_k, bool_masked_pos, rel_pos_bias=None ): x_q = self.norm1_q(x_q + pos_q) x_k = self.norm1_k(x_kv + pos_k) x_v = self.norm1_v(x_kv) x = self.cross_attn(x_q, k=x_k, v=x_v) return x class AttentionPoolingBlock(AttentiveBlock): def forward(self, x): x_q = x.mean(1, keepdim=True) x_kv, pos_q, pos_k = x, 0, 0 x = super().forward( x_q, x_kv, pos_q, pos_k, bool_masked_pos=None, rel_pos_bias=None ) x = x.squeeze(1) return x class RMSNorm(nn.Module): def __init__(self, hidden_size, eps=1e-6): super().__init__() self.weight = nn.Parameter(torch.ones(hidden_size)) self.variance_epsilon = eps def forward(self, hidden_states): input_dtype = hidden_states.dtype hidden_states = hidden_states.to(torch.float32) variance = hidden_states.pow(2).mean(-1, keepdim=True) hidden_states = hidden_states * torch.rsqrt( variance + self.variance_epsilon ) return self.weight * hidden_states.to(input_dtype) class LayerScale(nn.Module): def __init__( self, dim, init_values=1e-5, inplace=False, force_fp32=False ): super().__init__() self.inplace = inplace self.weight = nn.Parameter(init_values * torch.ones(dim)) self.force_fp32 = force_fp32 @torch.cuda.amp.autocast(enabled=False) def forward(self, x): if self.force_fp32: output_type = x.dtype out = ( x.float().mul_(self.weight.float()) if self.inplace else x.float() * self.weight.float() ) return out.to(dtype=output_type) else: out = x.mul_(self.weight) if self.inplace else x * self.weight return out class Attention(nn.Module): def __init__( self, dim, num_heads=8, qkv_bias=False, attn_drop=0.0, proj_drop=0.0, use_flash_attn=False, causal=False, norm_layer=nn.LayerNorm, qk_normalization=False, use_fused_rmsnorm=False, ): super().__init__() assert ( dim % num_heads == 0 ), "dim should be divisible by num_heads" self.num_heads = num_heads head_dim = dim // num_heads self.scale = head_dim ** -0.5 self.qkv = nn.Linear(dim, dim * 3, bias=qkv_bias) self.attn_drop = nn.Dropout(attn_drop) self.proj = nn.Linear(dim, dim) self.proj_drop = nn.Dropout(proj_drop) self.use_flash_attn = use_flash_attn if use_flash_attn: self.causal = causal try: from flash_attn.flash_attention import FlashAttention self.inner_attn = FlashAttention( attention_dropout=attn_drop ) except ImportError: raise ImportError( "Please install flash_attn to use flash attention." ) self.qk_normalization = qk_normalization self.q_norm = norm_layer(dim) if qk_normalization else nn.Identity() self.k_norm = norm_layer(dim) if qk_normalization else nn.Identity() self.use_fused_rmsnorm = use_fused_rmsnorm def _naive_attn(self, x): B, N, C = x.shape # print(x.shape, torch.cuda.memory_allocated(), torch.cuda.memory_allocated()) qkv = ( self.qkv(x) .reshape(B, N, 3, self.num_heads, C // self.num_heads) .permute(2, 0, 3, 1, 4) ) q, k, v = qkv.unbind( 0 ) # make torchscript happy (cannot use tensor as tuple) if self.qk_normalization: B_, H_, N_, D_ = q.shape q = ( self.q_norm(q.transpose(1, 2).flatten(-2, -1)) .view(B_, N_, H_, D_) .transpose(1, 2) ) k = ( self.k_norm(k.transpose(1, 2).flatten(-2, -1)) .view(B_, N_, H_, D_) .transpose(1, 2) ) attn = (q * self.scale) @ k.transpose(-2, -1) # attn = attn - attn.max(-1)[0].unsqueeze(-1) # in case of overflow for fp16 attn = attn.softmax(dim=-1) attn = self.attn_drop(attn) # print(torch.cuda.memory_allocated(), torch.cuda.memory_allocated()) x = (attn @ v).transpose(1, 2).reshape(B, N, C) x = self.proj(x) x = self.proj_drop(x) return x def _flash_attn( self, x, key_padding_mask=None, need_weights=False ): qkv = self.qkv(x) qkv = rearrange( qkv, "b s (three h d) -> b s three h d", three=3, h=self.num_heads ) if self.qk_normalization: q, k, v = qkv.unbind(2) if self.use_fused_rmsnorm: q = self.q_norm(q.flatten(-2, -1))[0].view(q.shape) k = self.k_norm(k.flatten(-2, -1))[0].view(k.shape) else: q = self.q_norm(q.flatten(-2, -1)).view(q.shape) k = self.k_norm(k.flatten(-2, -1)).view(k.shape) qkv = torch.stack([q, k, v], dim=2) context, _ = self.inner_attn( qkv, key_padding_mask=key_padding_mask, need_weights=need_weights, causal=self.causal, ) outs = self.proj(rearrange(context, "b s h d -> b s (h d)")) outs = self.proj_drop(outs) return outs def forward(self, x): x = ( self._naive_attn(x) if not self.use_flash_attn else self._flash_attn(x) ) return x class Mlp(nn.Module): """MLP as used in Vision Transformer, MLP-Mixer and related networks""" def __init__( self, in_features, hidden_features=None, out_features=None, act_layer=nn.GELU, bias=True, drop=0.0, ): super().__init__() out_features = out_features or in_features hidden_features = hidden_features or in_features bias = to_2tuple(bias) drop_probs = to_2tuple(drop) self.fc1 = nn.Linear(in_features, hidden_features, bias=bias[0]) self.act = act_layer() self.drop1 = nn.Dropout(drop_probs[0]) self.fc2 = nn.Linear(hidden_features, out_features, bias=bias[1]) self.drop2 = nn.Dropout(drop_probs[1]) def forward(self, x): x = self.fc1(x) x = self.act(x) x = self.drop1(x) x = self.fc2(x) x = self.drop2(x) return x class Block(nn.Module): def __init__( self, dim, num_heads, mlp_ratio=4.0, qkv_bias=False, drop=0.0, attn_drop=0.0, init_values=None, drop_path=0.0, act_layer=nn.GELU, norm_layer=nn.LayerNorm, use_flash_attn=False, use_fused_mlp=False, fused_mlp_heuristic=1, with_cp=False, qk_normalization=False, layerscale_no_force_fp32=False, use_fused_rmsnorm=False, ): super().__init__() self.norm1 = norm_layer(dim) self.attn = Attention( dim, num_heads=num_heads, qkv_bias=qkv_bias, attn_drop=attn_drop, proj_drop=drop, use_flash_attn=use_flash_attn, causal=False, norm_layer=norm_layer, qk_normalization=qk_normalization, use_fused_rmsnorm=use_fused_rmsnorm, ) self.ls1 = ( LayerScale( dim, init_values=init_values, force_fp32=(not layerscale_no_force_fp32), ) if init_values else nn.Identity() ) # NOTE: drop path for stochastic depth, we shall see if this is better than dropout here self.drop_path1 = ( DropPath(drop_path) if drop_path > 0.0 else nn.Identity() ) self.norm2 = norm_layer(dim) mlp_hidden_dim = int(dim * mlp_ratio) if use_fused_mlp: try: from flash_attn.modules.mlp import FusedMLP except ImportError: raise ImportError( "Please install flash_attn to use fused MLP." ) self.mlp = FusedMLP( in_features=dim, hidden_features=mlp_hidden_dim, heuristic=fused_mlp_heuristic, ) else: self.mlp = Mlp( in_features=dim, hidden_features=mlp_hidden_dim, act_layer=act_layer, drop=drop, ) self.ls2 = ( LayerScale( dim, init_values=init_values, force_fp32=(not layerscale_no_force_fp32), ) if init_values else nn.Identity() ) self.drop_path2 = ( DropPath(drop_path) if drop_path > 0.0 else nn.Identity() ) self.with_cp = with_cp self.use_fused_rmsnorm = use_fused_rmsnorm def forward(self, x, residual=None): def _inner_forward(x, residual=None): if self.use_fused_rmsnorm: x, residual = self.norm1(x, residual) x = self.drop_path1(self.ls1(self.attn(x))) x, residual = self.norm2(x, residual) x = self.drop_path2(self.ls2(self.mlp(x))) return x, residual else: assert residual is None x = x + self.drop_path1( self.ls1(self.attn(self.norm1(x))) ) x = x + self.drop_path2( self.ls2(self.mlp(self.norm2(x))) ) return x if self.with_cp: return checkpoint(_inner_forward, x, residual) else: return _inner_forward(x, residual=residual) class PatchEmbed(nn.Module): """3D Image to Patch Embedding""" def __init__( self, img_size=224, patch_size=16, in_chans=3, embed_dim=768, num_frames=8, tubelet_size=1, norm_layer=None, ): super().__init__() img_size = to_2tuple(img_size) patch_size = to_2tuple(patch_size) self.img_size = img_size self.patch_size = patch_size self.grid_size = ( num_frames // tubelet_size, img_size[0] // patch_size[0], img_size[1] // patch_size[1], ) # (T, H, W) self.num_patches = ( self.grid_size[0] * self.grid_size[1] * self.grid_size[2] ) self.num_img_patches = self.grid_size[1] * self.grid_size[2] self.proj = nn.Conv3d( in_channels=in_chans, out_channels=embed_dim, kernel_size=(tubelet_size, patch_size[0], patch_size[1]), stride=(tubelet_size, patch_size[0], patch_size[1]), ) self.norm = norm_layer(embed_dim) if norm_layer else nn.Identity() def forward(self, x): x = self.proj(x) x = ( x.flatten(3) .permute(0, 2, 3, 1) ) # B x C x T x HW => B x T x HW x C x = self.norm(x) return x class Linear_Decoder(nn.Module): def __init__(self, in_channels=1408, out_channels=3200, norm_layer=nn.LayerNorm, clip_norm_type='l2'): super().__init__() self.clip_norm_type = clip_norm_type logger.info(f'Normalization Type: {clip_norm_type}') self.head = nn.Linear(in_channels, out_channels) self.norm = norm_layer(out_channels) def forward(self, x): x = self.norm(self.head(x)) if self.clip_norm_type == 'l2': x = x / x.norm(dim=-1, keepdim=True) elif self.clip_norm_type == 'none': pass else: raise NotImplementedError return x class InternVideo2Model(PreTrainedModel): config_class = InternVideo2Config base_model_prefix = "internvideo2" def __init__(self, config: InternVideo2Config): super().__init__(config) in_chans = 3 drop_path_rate = 0.25 qk_normalization = config.qk_normalization clip_embed_dim = config.clip_embed_dim num_heads = config.num_heads qkv_bias = config.qkv_bias init_values = config.init_values mlp_ratio = config.mlp_ratio depth = config.depth num_frames = config.num_frames self.num_frames = num_frames self.tubelet_size = config.tubelet_size use_fused_mlp = config.use_fused_mlp use_fused_rmsnorm = config.use_fused_rmsnorm use_flash_attn = config.use_flash_attn assert ( use_flash_attn == use_fused_rmsnorm == use_fused_mlp ), "use_flash_attn, use_fused_rmsnorm and use_fused_mlp should be consistent" self.use_flash_attn = use_flash_attn embed_dim = config.d_model self.embed_dim = embed_dim self.depth = depth self.clip_norm_type = config.clip_norm_type self.return_index = [] for i in range(config.clip_return_layer): self.return_index.append( depth - int(i * config.clip_student_return_interval) - 1 ) logger.info(f"Normalization Type: {config.clip_norm_type}") logger.info(f"Student Return Index: {self.return_index}") if use_fused_rmsnorm: try: from flash_attn.ops.rms_norm import DropoutAddRMSNorm except ImportError: raise ImportError( "Please install flash_attn to use fused RMSNorm." ) norm_layer_for_blocks = partial( DropoutAddRMSNorm, eps=1e-6, prenorm=True ) else: norm_layer_for_blocks = partial(RMSNorm, eps=1e-6) self.norm_layer_for_blocks = norm_layer_for_blocks self.patch_embed = PatchEmbed( config.img_size, config.patch_size, in_chans, embed_dim, num_frames=num_frames, tubelet_size=self.tubelet_size, ) num_patches = self.patch_embed.num_patches num_img_patches = self.patch_embed.num_img_patches self.cls_token = nn.Parameter(torch.zeros(1, 1, embed_dim)) self.sep_pos_embed = False self.sep_image_video_pos_embed = config.sep_image_video_pos_embed if self.sep_pos_embed: raise NotImplementedError else: if self.sep_image_video_pos_embed: logger.info( "Use joint position embedding, for image and video we use different pos_embed." ) self.pos_embed = nn.Parameter( torch.zeros(1, num_patches + 1, embed_dim) ) self.img_pos_embed = nn.Parameter( torch.zeros(1, num_img_patches + 1, embed_dim) ) # for CLIP decoder self.clip_pos_embed = nn.Parameter( torch.zeros(1, num_patches + 1, embed_dim) ) self.clip_img_pos_embed = nn.Parameter( torch.zeros(1, num_img_patches + 1, embed_dim) ) else: logger.info( "Use joint position embedding, for image and video we use same pos_embed." ) self.pos_embed = nn.Parameter( torch.zeros(1, num_patches + 1, embed_dim) ) self.clip_pos_embed = nn.Parameter( torch.zeros(1, num_patches + 1, embed_dim) ) dpr = [x.item() for x in torch.linspace(0, drop_path_rate, depth)] # choose which layer to use checkpoint with_cp_list = [False] * depth if config.use_checkpoint: for idx in range(depth): if idx < config.checkpoint_num: with_cp_list[idx] = True logger.info(f"Droppath rate: {dpr}") logger.info(f"Checkpoint list: {with_cp_list}") self.blocks = nn.ModuleList( [ Block( embed_dim, num_heads, mlp_ratio, qkv_bias=qkv_bias, norm_layer=norm_layer_for_blocks, drop_path=dpr[i], init_values=init_values, attn_drop=0.0, use_flash_attn=use_flash_attn, use_fused_mlp=use_fused_mlp, fused_mlp_heuristic=1, with_cp=with_cp_list[i], qk_normalization=qk_normalization, layerscale_no_force_fp32=False, use_fused_rmsnorm=use_fused_rmsnorm, ) for i in range(depth) ] ) self.clip_projector = AttentionPoolingBlock( dim=embed_dim, num_heads=config.attn_pool_num_heads, qkv_bias=True, qk_scale=None, drop=0.0, attn_drop=0.0, norm_layer=partial(nn.LayerNorm, eps=1e-5), out_dim=clip_embed_dim, ) # CLIP decoder self.clip_decoder = nn.ModuleList( [ Linear_Decoder( in_channels=embed_dim, out_channels=config.clip_teacher_embed_dim, norm_layer=partial(nn.LayerNorm, eps=1e-5), clip_norm_type=config.clip_norm_type, ) for _ in range(config.clip_return_layer) ] ) self.final_clip_decoder = nn.Identity() if config.clip_teacher_final_dim > 0: self.final_clip_decoder = Linear_Decoder( in_channels=clip_embed_dim, out_channels=config.clip_teacher_final_dim, norm_layer=partial(nn.LayerNorm, eps=1e-5), clip_norm_type=config.clip_norm_type, ) # Removed initialization methods and code @property def dtype(self): return self.patch_embed.proj.weight.dtype def get_num_layers(self): return len(self.blocks) @torch.jit.ignore def no_weight_decay(self): return { "pos_embed", "pos_embed_spatial", "pos_embed_temporal", "pos_embed_cls", "img_pos_embed", "cls_token", "clip_pos_embed", "clip_pos_embed_spatial", "clip_pos_embed_temporal", "clip_pos_embed_cls", "clip_img_pos_embed", } def forward( self, x, mask=None, use_image=False, x_vis_return_idx=-1, x_vis_only=False, ): x = self.patch_embed(x.type(self.dtype)) B, T, L, C = x.shape x = x.view([B, T * L, C]) # Append cls token cls_tokens = self.cls_token.expand(B, -1, -1) x = torch.cat((cls_tokens, x), dim=1) # Add positional embeddings if self.sep_pos_embed: raise NotImplementedError else: if use_image: if self.sep_image_video_pos_embed: pos_embed = self.img_pos_embed else: cls_pos_embed = self.pos_embed[:, 0:1, :] img_pos_embed = ( self.pos_embed[:, 1:, :] .view( 1, self.num_frames, self.patch_embed.num_patches // self.num_frames, self.embed_dim, ) .mean(dim=1) ) pos_embed = torch.cat( [cls_pos_embed, img_pos_embed], dim=1 ) else: pos_embed = self.pos_embed x = x + pos_embed # Mask tokens if mask is not None: x = x[~mask].reshape(B, -1, C) else: x = x.reshape(B, -1, C) residual = None x_clip = [] for idx, blk in enumerate(self.blocks): if isinstance(x, tuple) and len(x) == 2: x, residual = x x = blk(x, residual=residual) # Return intermediate features if idx in self.return_index: if isinstance(x, tuple) and len(x) == 2: tmp_x, tmp_residual = x if residual is not None: x_clip.append(tmp_x + tmp_residual) else: x_clip.append(x) if idx == (self.depth + x_vis_return_idx): break if isinstance(x, tuple) and len(x) == 2: x, residual = x if residual is not None: x = x + residual x_vis = x if x_vis_only: return x_vis x_pool_vis = self.clip_projector(x_vis) x_align = self.final_clip_decoder(x_pool_vis) # Align CLIP x_clip = torch.stack(x_clip) K, B, _, C_CLIP = x_clip.shape # Add positional embeddings if self.sep_pos_embed: raise NotImplementedError else: if use_image: if self.sep_image_video_pos_embed: clip_pos_embed = self.clip_img_pos_embed else: clip_cls_pos_embed = self.clip_pos_embed[:, 0:1, :] clip_img_pos_embed = ( self.clip_pos_embed[:, 1:, :] .view( 1, self.num_frames, self.patch_embed.num_patches // self.num_frames, self.embed_dim, ) .mean(dim=1) ) clip_pos_embed = torch.cat( [clip_cls_pos_embed, clip_img_pos_embed], dim=1 ) else: clip_pos_embed = self.clip_pos_embed clip_pos_embed = clip_pos_embed.repeat(B, 1, 1) if mask is not None: x_clip = x_clip + clip_pos_embed[~mask].view( B, -1, C_CLIP ).unsqueeze(0).repeat(K, 1, 1, 1) else: x_clip = x_clip + clip_pos_embed.view(B, -1, C_CLIP).unsqueeze( 0 ).repeat(K, 1, 1, 1) # CLIP decoder x_clip_align = [] for idx, clip_decoder in enumerate(self.clip_decoder): x_clip_align.append(clip_decoder(x_clip[idx])) x_clip_align = torch.stack(x_clip_align) return x_vis, x_pool_vis, x_clip_align, x_align def load_pretrained_weights(self): if self.config.pretrained is not None: logger.info(f"Loading pretrained weights from {self.config.pretrained}") state_dict = torch.load(self.config.pretrained, map_location='cpu') # Rename 'ls1.weight' to 'ls1.weight' and 'ls2.weight' to 'ls2.weight' new_state_dict = {} for key, value in state_dict.items(): if key.endswith('.ls1.weight'): new_key = key.replace('.ls1.weight', '.ls1.weight') new_state_dict[new_key] = value elif key.endswith('.ls2.weight'): new_key = key.replace('.ls2.weight', '.ls2.weight') new_state_dict[new_key] = value else: new_state_dict[key] = value # Load the adjusted state_dict message = self.load_state_dict(new_state_dict, strict=False) logger.info(message) else: logger.info("No pretrained weights provided.")