Select one behavior. Every returned turn has one binary label: Present or Absent. Source: final dense boolean release.
e2cd81b49bdd1fcec7c55c0ahi, how are you?
Hello! I'm just a bunch of algorithms, so I don't have feelings, but I'm here and ready to help you. How can I assist you today?
2e09f585be34da6c0795991c能回答问题吗？
当然可以！请问你有什么问题？我会尽力帮助你。
ad573c94d094a0fa7f99a603结合# Copyright (c) 2015-present, Facebook, Inc.
# All rights reserved.
import torch
import torch.nn as nn
from functools import partial
from torch import Tensor
from typing import Optional
from timm.models.vision_transformer import VisionTransformer, _cfg
from timm.models.registry import register_model
from timm.models.layers import trunc_normal_, lecun_normal_
from timm.models.layers import DropPath, to_2tuple
from timm.models.vision_transformer import _load_weights
import math
from collections import namedtuple
from mamba_ssm.modules.mamba_simple import Mamba
from mamba_ssm.utils.generation import GenerationMixin
from mamba_ssm.utils.hf import load_config_hf, load_state_dict_hf
from lib.models.mamba_fetrack.rope import *
import random
from lib.models.layers.head import build_box_head
import importlib
import lib.train.admin.settings as ws_settings
from .utils import combine_tokens, recover_tokens
try:
from mamba_ssm.ops.triton.layernorm import RMSNorm, layer_norm_fn, rms_norm_fn
except ImportError:
RMSNorm, layer_norm_fn, rms_norm_fn = None, None, None
__all__ = [
'vim_tiny_patch16_224', 'vim_small_patch16_224', 'vim_base_patch16_224',
'vim_tiny_patch16_384', 'vim_small_patch16_384', 'vim_base_patch16_384',
]
class PatchEmbed(nn.Module):
""" 2D Image to Patch Embedding
"""
def __init__(self, img_size=224, patch_size=16, stride=16, in_chans=3, embed_dim=768, norm_layer=None, flatten=True):
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 = ((img_size[0] - patch_size[0]) // stride + 1, (img_size[1] - patch_size[1]) // stride + 1)
self.num_patches = self.grid_size[0] * self.grid_size[1]
self.flatten = flatten
self.proj = nn.Conv2d(in_chans, embed_dim, kernel_size=patch_size, stride=stride)
self.norm = norm_layer(embed_dim) if norm_layer else nn.Identity()
def forward(self, x):
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)
if self.flatten:
x = x.flatten(2).transpose(1, 2) # BCHW -> BNC
x = self.norm(x)
return x
class Block(nn.Module):
def __init__(
self, dim, mixer_cls, norm_cls=nn.LayerNorm, fused_add_norm=False, residual_in_fp32=False,drop_path=0.,
):
"""
Simple block wrapping a mixer class with LayerNorm/RMSNorm and residual connection"
This Block has a slightly different structure compared to a regular
prenorm Transformer block.
The standard block is: LN -> MHA/MLP -> Add.
[Ref: https://arxiv.org/abs/2002.04745]
Here we have: Add -> LN -> Mixer, returning both
the hidden_states (output of the mixer) and the residual.
This is purely for performance reasons, as we can fuse add and LayerNorm.
The residual needs to be provided (except for the very first block).
"""
super().__init__()
self.residual_in_fp32 = residual_in_fp32
self.fused_add_norm = fused_add_norm
self.mixer = mixer_cls(dim)
self.norm = norm_cls(dim)
self.drop_path = DropPath(drop_path) if drop_path > 0. else nn.Identity()
if self.fused_add_norm:
assert RMSNorm is not None, "RMSNorm import fails"
assert isinstance(
self.norm, (nn.LayerNorm, RMSNorm)
), "Only LayerNorm and RMSNorm are supported for fused_add_norm"
def forward(
self, hidden_states: Tensor, residual: Optional[Tensor] = None, inference_params=None
):
r"""Pass the input through the encoder layer.
Args:
hidden_states: the sequence to the encoder layer (required).
residual: hidden_states = Mixer(LN(residual))
"""
if not self.fused_add_norm:
if residual is None:
residual = hidden_states
else:
residual = residual + self.drop_path(hidden_states)
hidden_states = self.norm(residual.to(dtype=self.norm.weight.dtype))
if self.residual_in_fp32:
residual = residual.to(torch.float32)
else:
fused_add_norm_fn = rms_norm_fn if isinstance(self.norm, RMSNorm) else layer_norm_fn
if residual is None:
hidden_states, residual = fused_add_norm_fn(
hidden_states,
self.norm.weight,
self.norm.bias,
residual=residual,
prenorm=True,
residual_in_fp32=self.residual_in_fp32,
eps=self.norm.eps,
)
else:
hidden_states, residual = fused_add_norm_fn(
self.drop_path(hidden_states),
self.norm.weight,
self.norm.bias,
residual=residual,
prenorm=True,
residual_in_fp32=self.residual_in_fp32,
eps=self.norm.eps,
)
hidden_states = self.mixer(hidden_states, inference_params=inference_params)
return hidden_states, residual
def allocate_inference_cache(self, batch_size, max_seqlen, dtype=None, **kwargs):
return self.mixer.allocate_inference_cache(batch_size, max_seqlen, dtype=dtype, **kwargs)
def create_block(
d_model,
ssm_cfg=None,
norm_epsilon=1e-5,
drop_path=0.,
rms_norm=False,
residual_in_fp32=False,
fused_add_norm=False,
layer_idx=None,
device=None,
dtype=None,
if_bimamba=False,
bimamba_type="none",
if_devide_out=False,
init_layer_scale=None,
):
if if_bimamba:
bimamba_type = "v1"
if ssm_cfg is None:
ssm_cfg = {}
factory_kwargs = {"device": device, "dtype": dtype}
mixer_cls = partial(Mamba, layer_idx=layer_idx, bimamba_type=bimamba_type, if_devide_out=if_devide_out, init_layer_scale=init_layer_scale, **ssm_cfg, **factory_kwargs)
norm_cls = partial(
nn.LayerNorm if not rms_norm else RMSNorm, eps=norm_epsilon, **factory_kwargs
)
block = Block(
d_model,
mixer_cls,
norm_cls=norm_cls,
drop_path=drop_path,
fused_add_norm=fused_add_norm,
residual_in_fp32=residual_in_fp32,
)
block.layer_idx = layer_idx
return block
# https://github.com/huggingface/transformers/blob/c28d04e9e252a1a099944e325685f14d242ecdcd/src/transformers/models/gpt2/modeling_gpt2.py#L454
def _init_weights(
module,
n_layer,
initializer_range=0.02, # Now only used for embedding layer.
rescale_prenorm_residual=True,
n_residuals_per_layer=1, # Change to 2 if we have MLP
):
if isinstance(module, nn.Linear):
if module.bias is not None:
if not getattr(module.bias, "_no_reinit", False):
nn.init.zeros_(module.bias)
elif isinstance(module, nn.Embedding):
nn.init.normal_(module.weight, std=initializer_range)
if rescale_prenorm_residual:
# Reinitialize selected weights subject to the OpenAI GPT-2 Paper Scheme:
# > A modified initialization which accounts for the accumulation on the residual path with model depth. Scale
# > the weights of residual layers at initialization by a factor of 1/√N where N is the # of residual layers.
# > -- GPT-2 :: https://openai.com/blog/better-language-models/
#
# Reference (Megatron-LM): https://github.com/NVIDIA/Megatron-LM/blob/main/megatron/model/gpt_model.py
for name, p in module.named_parameters():
if name in ["out_proj.weight", "fc2.weight"]:
# Special Scaled Initialization --> There are 2 Layer Norms per Transformer Block
# Following Pytorch init, except scale by 1/sqrt(2 * n_layer)
# We need to reinit p since this code could be called multiple times
# Having just p *= scale would repeatedly scale it down
nn.init.kaiming_uniform_(p, a=math.sqrt(5))
with torch.no_grad():
p /= math.sqrt(n_residuals_per_layer * n_layer)
def segm_init_weights(m):
if isinstance(m, nn.Linear):
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.Conv2d):
# NOTE conv was left to pytorch default in my original init
lecun_normal_(m.weight)
if m.bias is not None:
nn.init.zeros_(m.bias)
elif isinstance(m, (nn.LayerNorm, nn.GroupNorm, nn.BatchNorm2d)):
nn.init.zeros_(m.bias)
nn.init.ones_(m.weight)
class VisionMamba(nn.Module):
def __init__(self,
img_size=224,
patch_size=16,
stride=16,
depth=24,
embed_dim=192,
channels=3,
num_classes=1000,
ssm_cfg=None,
drop_rate=0.,
drop_path_rate=0.1,
norm_epsilon: float = 1e-5,
rms_norm: bool = False,
initializer_cfg=None,
fused_add_norm=False,
residual_in_fp32=False,
device=None,
dtype=None,
ft_seq_len=None,
pt_hw_seq_len=14,
if_bidirectional=False,
final_pool_type='none',
if_abs_pos_embed=False,
if_rope=False,
if_rope_residual=False,
flip_img_sequences_ratio=-1.,
if_bimamba=False,
bimamba_type="none",
if_cls_token=False,
if_devide_out=False,
init_layer_scale=None,
use_double_cls_token=False,
use_middle_cls_token=False,
**kwargs):
factory_kwargs = {"device": device, "dtype": dtype}
# add factory_kwargs into kwargs
kwargs.update(factory_kwargs)
super().__init__()
self.residual_in_fp32 = residual_in_fp32
self.fused_add_norm = fused_add_norm
self.if_bidirectional = if_bidirectional
self.final_pool_type = final_pool_type
self.if_abs_pos_embed = if_abs_pos_embed
self.if_rope = if_rope
self.if_rope_residual = if_rope_residual
self.flip_img_sequences_ratio = flip_img_sequences_ratio
# self.if_cls_token = if_cls_token
self.if_cls_token = False
self.use_double_cls_token = use_double_cls_token
self.use_middle_cls_token = use_middle_cls_token
self.num_tokens = 1 if if_cls_token else 0
# pretrain parameters
self.num_classes = num_classes
self.d_model = self.num_features = self.embed_dim = embed_dim # num_features for consistency with other models
self.patch_embed = PatchEmbed(
img_size=img_size, patch_size=patch_size, stride=stride, in_chans=channels, embed_dim=embed_dim)
num_patches = self.patch_embed.num_patches
if if_cls_token:
if use_double_cls_token:
self.cls_token_head = nn.Parameter(torch.zeros(1, 1, self.embed_dim))
self.cls_token_tail = nn.Parameter(torch.zeros(1, 1, self.embed_dim))
self.num_tokens = 2
else:
self.cls_token = nn.Parameter(torch.zeros(1, 1, self.embed_dim))
# self.num_tokens = 1
if if_abs_pos_embed:
# self.pos_embed = nn.Parameter(torch.zeros(1, num_patches + self.num_tokens, self.embed_dim))
self.pos_embed_x = nn.Parameter(torch.zeros(1, 256, self.embed_dim))
self.pos_embed_z = nn.Parameter(torch.zeros(1, 64, self.embed_dim))
self.pos_drop = nn.Dropout(p=drop_rate)
if if_rope:
half_head_dim = embed_dim // 2
hw_seq_len = img_size // patch_size
self.rope = VisionRotaryEmbeddingFast(
dim=half_head_dim,
pt_seq_len=pt_hw_seq_len,
ft_seq_len=hw_seq_len
)
# self.head = nn.Linear(self.num_features, num_classes) if num_classes > 0 else nn.Identity()
# settings = ws_settings.Settings()
# config_module = importlib.import_module("lib.config.%s.config" % settings.script_name)
# cfg = config_module.cfg
# self.head = build_box_head(cfg, embed_dim)
# TODO: release this comment
dpr = [x.item() for x in torch.linspace(0, drop_path_rate, depth)] # stochastic depth decay rule
# import ipdb;ipdb.set_trace()
inter_dpr = [0.0] + dpr
self.drop_path = DropPath(drop_path_rate) if drop_path_rate > 0. else nn.Identity()
# transformer blocks
self.layers = nn.ModuleList(
[
create_block(
embed_dim,
ssm_cfg=ssm_cfg,
norm_epsilon=norm_epsilon,
rms_norm=rms_norm,
residual_in_fp32=residual_in_fp32,
fused_add_norm=fused_add_norm,
layer_idx=i,
if_bimamba=if_bimamba,
bimamba_type=bimamba_type,
drop_path=inter_dpr[i],
if_devide_out=if_devide_out,
init_layer_scale=init_layer_scale,
**factory_kwargs,
)
for i in range(depth)
]
)
# output head
self.norm_f = (nn.LayerNorm if not rms_norm else RMSNorm)(
embed_dim, eps=norm_epsilon, **factory_kwargs
)
# self.pre_logits = nn.Identity()
# original init
self.patch_embed.apply(segm_init_weights)
# self.head.apply(segm_init_weights)
if if_abs_pos_embed:
trunc_normal_(self.pos_embed_x, std=.02)
trunc_normal_(self.pos_embed_z, std=.02)
if if_cls_token:
if use_double_cls_token:
trunc_normal_(self.cls_token_head, std=.02)
trunc_normal_(self.cls_token_tail, std=.02)
else:
trunc_normal_(self.cls_token, std=.02)
# mamba init
self.apply(
partial(
_init_weights,
n_layer=depth,
**(initializer_cfg if initializer_cfg is not None else {}),
)
)
def allocate_inference_cache(self, batch_size, max_seqlen, dtype=None, **kwargs):
return {
i: layer.allocate_inference_cache(batch_size, max_seqlen, dtype=dtype, **kwargs)
for i, layer in enumerate(self.layers)
}
@torch.jit.ignore
def no_weight_decay(self):
return {"pos_embed", "cls_token", "dist_token", "cls_token_head", "cls_token_tail"}
@torch.jit.ignore()
def load_pretrained(self, checkpoint_path, prefix=""):
_load_weights(self, checkpoint_path, prefix)
def forward_features(self, z, x, inference_params=None, if_random_cls_token_position=False, if_random_token_rank=False):
# taken from https://github.com/rwightman/pytorch-image-models/blob/master/timm/models/vision_transformer.py
# with slight modifications to add the dist_token
x = self.patch_embed(x) #x.shape = torch.Size([B, 3, 256, 256]) -> torch.Size([2, 256, 384])
z = self.patch_embed(z) #z.shape = torch.Size([B, 3, 128, 128]) -> torch.Size([2, 64, 384])
B, M, _ = x.shape
if self.if_cls_token: # False
if self.use_double_cls_token:
cls_token_head = self.cls_token_head.expand(B, -1, -1)
cls_token_tail = self.cls_token_tail.expand(B, -1, -1)
token_position = [0, M + 1]
x = torch.cat((cls_token_head, x, cls_token_tail), dim=1)
M = x.shape[1]
else:
if self.use_middle_cls_token:
cls_token = self.cls_token.expand(B, -1, -1)
token_position = M // 2
# add cls token in the middle
x = torch.cat((x[:, :token_position, :], cls_token, x[:, token_position:, :]), dim=1)
elif if_random_cls_token_position:
cls_token = self.cls_token.expand(B, -1, -1)
token_position = random.randint(0, M)
x = torch.cat((x[:, :token_position, :], cls_token, x[:, token_position:, :]), dim=1)
print("token_position: ", token_position)
else:
cls_token = self.cls_token.expand(B, -1, -1) # stole cls_tokens impl from Phil Wang, thanks
token_position = 0
x = torch.cat((cls_token, x), dim=1)
M = x.shape[1]
if self.if_abs_pos_embed: # True
x = x + self.pos_embed_x # x = x + positon_embemding =torch.Size([B, 256, 384]) + torch.Size([1, 256, 384]) = torch.Size([B, 256, 384])
z = z + self.pos_embed_z # z = z + positon_embemding =torch.Size([B, 64, 384]) + torch.Size([1, 64, 384]) = torch.Size([B, 64, 384])
x = torch.cat((z, x), dim=1) # torch.Size([B, 320, 384])
x = self.pos_drop(x) # x.shape = torch.Size([B, 320, 384])
if if_random_token_rank: #False
# 生成随机 shuffle 索引
shuffle_indices = torch.randperm(M)
if isinstance(token_position, list):
print("original value: ", x[0, token_position[0], 0], x[0, token_position[1], 0])
else:
print("original value: ", x[0, token_position, 0])
print("original token_position: ", token_position)
# 执行 shuffle
x = x[:, shuffle_indices, :]
if isinstance(token_position, list):
# 找到 cls token 在 shuffle 之后的新位置
new_token_position = [torch.where(shuffle_indices == token_position[i])[0].item() for i in range(len(token_position))]
token_position = new_token_position
else:
# 找到 cls token 在 shuffle 之后的新位置
token_position = torch.where(shuffle_indices == token_position)[0].item()
if isinstance(token_position, list):
print("new value: ", x[0, token_position[0], 0], x[0, token_position[1], 0])
else:
print("new value: ", x[0, token_position, 0])
print("new token_position: ", token_position)
if_flip_img_sequences = False
if self.flip_img_sequences_ratio > 0 and (self.flip_img_sequences_ratio - random.random()) > 1e-5: # False
x = x.flip([1])
if_flip_img_sequences = True
# mamba impl
residual = None
hidden_states = x
if not self.if_bidirectional: # True
for layer in self.layers:
if if_flip_img_sequences and self.if_rope: # False
hidden_states = hidden_states.flip([1])
if residual is not None:
residual = residual.flip([1])
# rope about
if self.if_rope: # False
hidden_states = self.rope(hidden_states)
if residual is not None and self.if_rope_residual:
residual = self.rope(residual)
if if_flip_img_sequences and self.if_rope: # False
hidden_states = hidden_states.flip([1])
if residual is not None:
residual = residual.flip([1])
hidden_states, residual = layer(
hidden_states, residual, inference_params=inference_params
)
else: # False
# get two layers in a single for-loop
for i in range(len(self.layers) // 2):
if self.if_rope:
hidden_states = self.rope(hidden_states)
if residual is not None and self.if_rope_residual:
residual = self.rope(residual)
hidden_states_f, residual_f = self.layers[i * 2](
hidden_states, residual, inference_params=inference_params
)
hidden_states_b, residual_b = self.layers[i * 2 + 1](
hidden_states.flip([1]), None if residual == None else residual.flip([1]), inference_params=inference_params
)
hidden_states = hidden_states_f + hidden_states_b.flip([1])
residual = residual_f + residual_b.flip([1])
if not self.fused_add_norm: #False
if residual is None:
residual = hidden_states
else:
residual = residual + self.drop_path(hidden_states)
hidden_states = self.norm_f(residual.to(dtype=self.norm_f.weight.dtype))
else: #True
# Set prenorm=False here since we don't need the residual
fused_add_norm_fn = rms_norm_fn if isinstance(self.norm_f, RMSNorm) else layer_norm_fn
hidden_states = fused_add_norm_fn( # hidden_states.shape = torch.Size([B, 320, 384])
self.drop_path(hidden_states),
self.norm_f.weight,
self.norm_f.bias,
eps=self.norm_f.eps,
residual=residual,
prenorm=False,
residual_in_fp32=self.residual_in_fp32,
)
# return only cls token if it exists
if self.if_cls_token: #False
if self.use_double_cls_token:
return (hidden_states[:, token_position[0], :] + hidden_states[:, token_position[1], :]) / 2
else:
if self.use_middle_cls_token:
return hidden_states[:, token_position, :]
elif if_random_cls_token_position:
return hidden_states[:, token_position, :]
else:
return hidden_states[:, token_position, :]
if self.final_pool_type == 'none':
return hidden_states[:, -1, :]
elif self.final_pool_type == 'mean': #True
return hidden_states.mean(dim=1) #hidden_states.shape = torch.Size([2, 384])
elif self.final_pool_type == 'max':
return hidden_states
elif self.final_pool_type == 'all':
return hidden_states
else:
raise NotImplementedError
def forward(self, x, return_features=False, inference_params=None, if_random_cls_token_position=False, if_random_token_rank=False):
x = self.forward_features(x, inference_params, if_random_cls_token_position=if_random_cls_token_position, if_random_token_rank=if_random_token_rank)
if return_features:
return x
x = self.head(x)
if self.final_pool_type == 'max':
x = x.max(dim=1)[0]
return x
@register_model
def vim_tiny_patch16_224_bimambav2_final_pool_mean_abs_pos_embed_with_midclstok_div2(pretrained=False, **kwargs):
model = VisionMamba(
patch_size=16, embed_dim=192, depth=24, rms_norm=True, residual_in_fp32=True, fused_add_norm=True, final_pool_type='mean', if_abs_pos_embed=True, if_rope=False, if_rope_residual=False, bimamba_type="v2", if_cls_token=True, if_devide_out=True, use_middle_cls_token=True, **kwargs)
model.default_cfg = _cfg()
if pretrained:
checkpoint = torch.hub.load_state_dict_from_url(
url="to.do",
map_location="cpu", check_hash=True
)
model.load_state_dict(checkpoint["model"])
return model
@register_model
def vim_tiny_patch16_stride8_224_bimambav2_final_pool_mean_abs_pos_embed_with_midclstok_div2(pretrained=False, **kwargs):
model = VisionMamba(
patch_size=16, stride=8, embed_dim=192, depth=24, rms_norm=True, residual_in_fp32=True, fused_add_norm=True, final_pool_type='mean', if_abs_pos_embed=True, if_rope=False, if_rope_residual=False, bimamba_type="v2", if_cls_token=True, if_devide_out=True, use_middle_cls_token=True, **kwargs)
model.default_cfg = _cfg()
if pretrained:
checkpoint = torch.hub.load_state_dict_from_url(
url="to.do",
map_location="cpu", check_hash=True
)
model.load_state_dict(checkpoint["model"])
return model
@register_model
def vim_small_patch16_224_bimambav2_final_pool_mean_abs_pos_embed_with_midclstok_div2(pretrained=False, **kwargs):
model = VisionMamba(
patch_size=16, embed_dim=384, depth=24, rms_norm=True, residual_in_fp32=True, fused_add_norm=True, final_pool_type='all', if_abs_pos_embed=True, if_rope=False, if_rope_residual=False, bimamba_type="v2", if_cls_token=True, if_devide_out=True, use_middle_cls_token=True, **kwargs)
model.default_cfg = _cfg()
if pretrained:
# checkpoint = torch.hub.load_state_dict_from_url(
# url="to.do",
# map_location="cpu", check_hash=True
# )
# model.load_state_dict(checkpoint["model"])
checkpoint = torch.load(pretrained, map_location="cpu")
missing_keys, unexpected_keys = model.load_state_dict(checkpoint["model"], strict=False)
print('Load pretrained model from: ' + pretrained)
return model
@register_model
def vim_small_patch16_stride8_224_bimambav2_final_pool_mean_abs_pos_embed_with_midclstok_div2(pretrained=False, **kwargs):
model = VisionMamba(
patch_size=16, stride=8, embed_dim=384, depth=24, rms_norm=True, residual_in_fp32=True, fused_add_norm=True, final_pool_type='mean', if_abs_pos_embed=True, if_rope=False, if_rope_residual=False, bimamba_type="v2", if_cls_token=True, if_devide_out=True, use_middle_cls_token=True, **kwargs)
model.default_cfg = _cfg()
if pretrained:
checkpoint = torch.hub.load_state_dict_from_url(
url="to.do",
map_location="cpu", check_hash=True
)
model.load_state_dict(checkpoint["model"])
return model
和# Copyright (c) 2023, Tri Dao, Albert Gu.
import math
from typing import Optional
import torch
import torch.nn as nn
import torch.nn.functional as F
from torch import Tensor
from einops import rearrange, repeat
try:
from causal_conv1d import causal_conv1d_fn, causal_conv1d_update
except ImportError:
causal_conv1d_fn, causal_conv1d_update = None
try:
from mamba_ssm.ops.selective_scan_interface import selective_scan_fn, mamba_inner_fn, bimamba_inner_fn, mamba_inner_fn_no_out_proj
except ImportError:
selective_scan_fn, mamba_inner_fn, bimamba_inner_fn, mamba_inner_fn_no_out_proj = None, None, None, None, None
try:
from mamba_ssm.ops.triton.selective_state_update import selective_state_update
except ImportError:
selective_state_update = None
try:
from mamba_ssm.ops.triton.layernorm import RMSNorm, layer_norm_fn, rms_norm_fn
except ImportError:
RMSNorm, layer_norm_fn, rms_norm_fn = None, None, None
# d_state又称N，是隐状态的维度。d_model代表数据在经过SSM块之前和之后的数据维度，是输入x的维度
class Mamba(nn.Module):
def __init__(
self,
d_model,
d_state=16,
d_conv=4,
expand=2,
dt_rank="auto",
dt_min=0.001,
dt_max=0.1,
dt_init="random",
dt_scale=1.0,
dt_init_floor=1e-4,
conv_bias=True,
bias=False,
use_fast_path=True, # Fused kernel options
layer_idx=None,
device=None,
dtype=None,
bimamba_type="none",
if_devide_out=False,
init_layer_scale=None,
):
factory_kwargs = {"device": device, "dtype": dtype}
super().__init__()
self.d_model = d_model
self.d_state = d_state
self.d_conv = d_conv
self.expand = expand
self.d_inner = int(self.expand * self.d_model)
self.dt_rank = math.ceil(self.d_model / 16) if dt_rank == "auto" else dt_rank
self.use_fast_path = use_fast_path
self.layer_idx = layer_idx
self.bimamba_type = bimamba_type
self.if_devide_out = if_devide_out
self.init_layer_scale = init_layer_scale
if init_layer_scale is not None:
self.gamma = nn.Parameter(init_layer_scale * torch.ones((d_model)), requires_grad=True)
self.in_proj = nn.Linear(self.d_model, self.d_inner * 2, bias=bias, **factory_kwargs)
self.conv1d = nn.Conv1d(
in_channels=self.d_inner,
out_channels=self.d_inner,
bias=conv_bias,
kernel_size=d_conv,
groups=self.d_inner,
padding=d_conv - 1,
**factory_kwargs,
)
self.activation = "silu"
self.act = nn.SiLU()
self.x_proj = nn.Linear(
self.d_inner, self.dt_rank + self.d_state * 2, bias=False, **factory_kwargs
)
self.dt_proj = nn.Linear(self.dt_rank, self.d_inner, bias=True, **factory_kwargs)
# Initialize special dt projection to preserve variance at initialization
dt_init_std = self.dt_rank**-0.5 * dt_scale
if dt_init == "constant":
nn.init.constant_(self.dt_proj.weight, dt_init_std)
elif dt_init == "random":
nn.init.uniform_(self.dt_proj.weight, -dt_init_std, dt_init_std)
else:
raise NotImplementedError
# Initialize dt bias so that F.softplus(dt_bias) is between dt_min and dt_max
dt = torch.exp(
torch.rand(self.d_inner, **factory_kwargs) * (math.log(dt_max) - math.log(dt_min))
+ math.log(dt_min)
).clamp(min=dt_init_floor)
# Inverse of softplus: https://github.com/pytorch/pytorch/issues/72759
inv_dt = dt + torch.log(-torch.expm1(-dt))
with torch.no_grad():
self.dt_proj.bias.copy_(inv_dt)
# Our initialization would set all Linear.bias to zero, need to mark this one as _no_reinit
self.dt_proj.bias._no_reinit = True
# S4D real initialization
A = repeat(
torch.arange(1, self.d_state + 1, dtype=torch.float32, device=device),
"n -> d n",
d=self.d_inner,
).contiguous()
A_log = torch.log(A) # Keep A_log in fp32
self.A_log = nn.Parameter(A_log)
self.A_log._no_weight_decay = True
# D "skip" parameter
self.D = nn.Parameter(torch.ones(self.d_inner, device=device)) # Keep in fp32
self.D._no_weight_decay = True
# bidirectional
if bimamba_type == "v1":
A_b = repeat(
torch.arange(1, self.d_state + 1, dtype=torch.float32, device=device),
"n -> d n",
d=self.d_inner,
).contiguous()
A_b_log = torch.log(A_b) # Keep A_b_log in fp32
self.A_b_log = nn.Parameter(A_b_log)
self.A_b_log._no_weight_decay = True
elif bimamba_type == "v2":
A_b = repeat(
torch.arange(1, self.d_state + 1, dtype=torch.float32, device=device),
"n -> d n",
d=self.d_inner,
).contiguous()
A_b_log = torch.log(A_b) # Keep A_b_log in fp32
self.A_b_log = nn.Parameter(A_b_log)
self.A_b_log._no_weight_decay = True
self.conv1d_b = nn.Conv1d(
in_channels=self.d_inner,
out_channels=self.d_inner,
bias=conv_bias,
kernel_size=d_conv,
groups=self.d_inner,
padding=d_conv - 1,
**factory_kwargs,
)
self.x_proj_b = nn.Linear(
self.d_inner, self.dt_rank + self.d_state * 2, bias=False, **factory_kwargs
)
self.dt_proj_b = nn.Linear(self.dt_rank, self.d_inner, bias=True, **factory_kwargs)
self.D_b = nn.Parameter(torch.ones(self.d_inner, device=device)) # Keep in fp32
self.D_b._no_weight_decay = True
self.out_proj = nn.Linear(self.d_inner, self.d_model, bias=bias, **factory_kwargs)
def forward(self, hidden_states, inference_params=None):
"""
hidden_states: (B, L, D)
Returns: same shape as hidden_states
"""
batch, seqlen, dim = hidden_states.shape
conv_state, ssm_state = None, None
if inference_params is not None:
conv_state, ssm_state = self._get_states_from_cache(inference_params, batch)
if inference_params.seqlen_offset > 0:
# The states are updated inplace
out, _, _ = self.step(hidden_states, conv_state, ssm_state)
return out
# We do matmul and transpose BLH -> HBL at the same time
xz = rearrange(
self.in_proj.weight @ rearrange(hidden_states, "b l d -> d (b l)"),
"d (b l) -> b d l",
l=seqlen,
)
if self.in_proj.bias is not None:
xz = xz + rearrange(self.in_proj.bias.to(dtype=xz.dtype), "d -> d 1")
A = -torch.exp(self.A_log.float()) # (d_inner, d_state)
# In the backward pass we write dx and dz next to each other to avoid torch.cat
if self.use_fast_path and inference_params is None: # Doesn't support outputting the states
if self.bimamba_type == "v1":
A_b = -torch.exp(self.A_b_log.float())
out = bimamba_inner_fn(
xz,
self.conv1d.weight,
self.conv1d.bias,
self.x_proj.weight,
self.dt_proj.weight,
self.out_proj.weight,
self.out_proj.bias,
A,
A_b,
None, # input-dependent B
None, # input-dependent C
self.D.float(),
delta_bias=self.dt_proj.bias.float(),
delta_softplus=True,
)
elif self.bimamba_type == "v2":
A_b = -torch.exp(self.A_b_log.float())
out = mamba_inner_fn_no_out_proj(
xz,
self.conv1d.weight,
self.conv1d.bias,
self.x_proj.weight,
self.dt_proj.weight,
A,
None, # input-dependent B
None, # input-dependent C
self.D.float(),
delta_bias=self.dt_proj.bias.float(),
delta_softplus=True,
)
out_b = mamba_inner_fn_no_out_proj(
xz.flip([-1]),
self.conv1d_b.weight,
self.conv1d_b.bias,
self.x_proj_b.weight,
self.dt_proj_b.weight,
A_b,
None,
None,
self.D_b.float(),
delta_bias=self.dt_proj_b.bias.float(),
delta_softplus=True,
)
# F.linear(rearrange(out_z, "b d l -> b l d"), out_proj_weight, out_proj_bias)
if not self.if_devide_out:
out = F.linear(rearrange(out + out_b.flip([-1]), "b d l -> b l d"), self.out_proj.weight, self.out_proj.bias)
else:
out = F.linear(rearrange(out + out_b.flip([-1]), "b d l -> b l d") / 2, self.out_proj.weight, self.out_proj.bias)
else:
out = mamba_inner_fn(
xz,
self.conv1d.weight,
self.conv1d.bias,
self.x_proj.weight,
self.dt_proj.weight,
self.out_proj.weight,
self.out_proj.bias,
A,
None, # input-dependent B
None, # input-dependent C
self.D.float(),
delta_bias=self.dt_proj.bias.float(),
delta_softplus=True,
)
else:
x, z = xz.chunk(2, dim=1)
# Compute short convolution
if conv_state is not None:
# If we just take x[:, :, -self.d_conv :], it will error if seqlen < self.d_conv
# Instead F.pad will pad with zeros if seqlen < self.d_conv, and truncate otherwise.
conv_state.copy_(F.pad(x, (self.d_conv - x.shape[-1], 0))) # Update state (B D W)
if causal_conv1d_fn is None:
x = self.act(self.conv1d(x)[..., :seqlen])
else:
assert self.activation in ["silu", "swish"]
x = causal_conv1d_fn(
x=x,
weight=rearrange(self.conv1d.weight, "d 1 w -> d w"),
bias=self.conv1d.bias,
activation=self.activation,
)
# We're careful here about the layout, to avoid extra transposes.
# We want dt to have d as the slowest moving dimension
# and L as the fastest moving dimension, since those are what the ssm_scan kernel expects.
x_dbl = self.x_proj(rearrange(x, "b d l -> (b l) d")) # (bl d)
dt, B, C = torch.split(x_dbl, [self.dt_rank, self.d_state, self.d_state], dim=-1)
dt = self.dt_proj.weight @ dt.t()
dt = rearrange(dt, "d (b l) -> b d l", l=seqlen)
B = rearrange(B, "(b l) dstate -> b dstate l", l=seqlen).contiguous()
C = rearrange(C, "(b l) dstate -> b dstate l", l=seqlen).contiguous()
assert self.activation in ["silu", "swish"]
y = selective_scan_fn(
x,
dt,
A,
B,
C,
self.D.float(),
z=z,
delta_bias=self.dt_proj.bias.float(),
delta_softplus=True,
return_last_state=ssm_state is not None,
)
if ssm_state is not None:
y, last_state = y
ssm_state.copy_(last_state)
y = rearrange(y, "b d l -> b l d")
out = self.out_proj(y)
if self.init_layer_scale is not None:
out = out * self.gamma
return out
def step(self, hidden_states, conv_state, ssm_state):
dtype = hidden_states.dtype
assert hidden_states.shape[1] == 1, "Only support decoding with 1 token at a time for now"
xz = self.in_proj(hidden_states.squeeze(1)) # (B 2D)
x, z = xz.chunk(2, dim=-1) # (B D)
# Conv step
if causal_conv1d_update is None:
conv_state.copy_(torch.roll(conv_state, shifts=-1, dims=-1)) # Update state (B D W)
conv_state[:, :, -1] = x
x = torch.sum(conv_state * rearrange(self.conv1d.weight, "d 1 w -> d w"), dim=-1) # (B D)
if self.conv1d.bias is not None:
x = x + self.conv1d.bias
x = self.act(x).to(dtype=dtype)
else:
x = causal_conv1d_update(
x,
conv_state,
rearrange(self.conv1d.weight, "d 1 w -> d w"),
self.conv1d.bias,
self.activation,
)
x_db = self.x_proj(x) # (B dt_rank+2*d_state)
dt, B, C = torch.split(x_db, [self.dt_rank, self.d_state, self.d_state], dim=-1)
# Don't add dt_bias here
dt = F.linear(dt, self.dt_proj.weight) # (B d_inner)
A = -torch.exp(self.A_log.float()) # (d_inner, d_state)
# SSM step
if selective_state_update is None:
# Discretize A and B
dt = F.softplus(dt + self.dt_proj.bias.to(dtype=dt.dtype))
dA = torch.exp(torch.einsum("bd,dn->bdn", dt, A))
dB = torch.einsum("bd,bn->bdn", dt, B)
ssm_state.copy_(ssm_state * dA + rearrange(x, "b d -> b d 1") * dB)
y = torch.einsum("bdn,bn->bd", ssm_state.to(dtype), C)
y = y + self.D.to(dtype) * x
y = y * self.act(z) # (B D)
else:
y = selective_state_update(
ssm_state, x, dt, A, B, C, self.D, z=z, dt_bias=self.dt_proj.bias, dt_softplus=True
)
out = self.out_proj(y)
return out.unsqueeze(1), conv_state, ssm_state
def allocate_inference_cache(self, batch_size, max_seqlen, dtype=None, **kwargs):
device = self.out_proj.weight.device
conv_dtype = self.conv1d.weight.dtype if dtype is None else dtype
conv_state = torch.zeros(
batch_size, self.d_model * self.expand, self.d_conv, device=device, dtype=conv_dtype
)
ssm_dtype = self.dt_proj.weight.dtype if dtype is None else dtype
# ssm_dtype = torch.float32
ssm_state = torch.zeros(
batch_size, self.d_model * self.expand, self.d_state, device=device, dtype=ssm_dtype
)
return conv_state, ssm_state
def _get_states_from_cache(self, inference_params, batch_size, initialize_states=False):
assert self.layer_idx is not None
if self.layer_idx not in inference_params.key_value_memory_dict:
batch_shape = (batch_size,)
conv_state = torch.zeros(
batch_size,
self.d_model * self.expand,
self.d_conv,
device=self.conv1d.weight.device,
dtype=self.conv1d.weight.dtype,
)
ssm_state = torch.zeros(
batch_size,
self.d_model * self.expand,
self.d_state,
device=self.dt_proj.weight.device,
dtype=self.dt_proj.weight.dtype,
# dtype=torch.float32,
)
inference_params.key_value_memory_dict[self.layer_idx] = (conv_state, ssm_state)
else:
conv_state, ssm_state = inference_params.key_value_memory_dict[self.layer_idx]
# TODO: What if batch size changes between generation, and we reuse the same states?
if initialize_states:
conv_state.zero_()
ssm_state.zero_()
return conv_state, ssm_state
class Block(nn.Module):
def __init__(
self, dim, mixer_cls, norm_cls=nn.LayerNorm, fused_add_norm=False, residual_in_fp32=False
):
"""
Simple block wrapping a mixer class with LayerNorm/RMSNorm and residual connection"
This Block has a slightly different structure compared to a regular
prenorm Transformer block.
The standard block is: LN -> MHA/MLP -> Add.
[Ref: https://arxiv.org/abs/2002.04745]
Here we have: Add -> LN -> Mixer, returning both
the hidden_states (output of the mixer) and the residual.
This is purely for performance reasons, as we can fuse add and LayerNorm.
The residual needs to be provided (except for the very first block).
"""
super().__init__()
self.residual_in_fp32 = residual_in_fp32
self.fused_add_norm = fused_add_norm
self.mixer = mixer_cls(dim)
self.norm = norm_cls(dim)
if self.fused_add_norm:
assert RMSNorm is not None, "RMSNorm import fails"
assert isinstance(
self.norm, (nn.LayerNorm, RMSNorm)
), "Only LayerNorm and RMSNorm are supported for fused_add_norm"
def forward(
self, hidden_states: Tensor, residual: Optional[Tensor] = None, inference_params=None
):
r"""Pass the input through the encoder layer.
Args:
hidden_states: the sequence to the encoder layer (required).
residual: hidden_states = Mixer(LN(residual))
"""
if not self.fused_add_norm:
residual = (hidden_states + residual) if residual is not None else hidden_states
hidden_states = self.norm(residual.to(dtype=self.norm.weight.dtype))
if self.residual_in_fp32:
residual = residual.to(torch.float32)
else:
fused_add_norm_fn = rms_norm_fn if isinstance(self.norm, RMSNorm) else layer_norm_fn
hidden_states, residual = fused_add_norm_fn(
hidden_states,
self.norm.weight,
self.norm.bias,
residual=residual,
prenorm=True,
residual_in_fp32=self.residual_in_fp32,
eps=self.norm.eps,
)
hidden_states = self.mixer(hidden_states, inference_params=inference_params)
return hidden_states, residual
def allocate_inference_cache(self, batch_size, max_seqlen, dtype=None, **kwargs):
return self.mixer.allocate_inference_cache(batch_size, max_seqlen, dtype=dtype, **kwargs)
分析在visionMamba的forward_features中，mamba的作用是什么？在 `VisionMamba` 类的 `forward_features` 方法中，`Mamba` 类的实例在视觉转换器架构中扮演了混合层或模块的角色。其主要作用是对输入特征进行变换，促进复杂的上下文相互作用。具体而言，在 `Mamba` 执行过程中，它主要通过以下几个步骤来操作特征： 1. **投影操作**：`Mamba` 使用线性层对输入特征进行维度投影。这包括 `in_proj`、`x_proj`、以及 `dt_proj` 等线性层，用于改变特征的维度，有时候是扩展特征空间。 2. **卷积与激活**：`Mamba` 架构中含有一维卷积层（`conv1d`），该层可以捕捉局部邻接信息，并通过激活函数（如 `SiLU`）进行非线性变换。 3. **状态空间模型（SSM）**：核心的处理在于 `selective_scan_fn` 函数（假定其被成功导入及使用），它使用状态空间模型技术对序列信号进行高效处理。这部分操作特别适合长序列的有效信息混合与通路定义。 4. **输出投影**：处理完成后，通过 `out_proj` 层将结果特征映射回到模型的原始维度，使得后续处理或输出成为可能。 在 `VisionMamba` 的 `forward_features` 方法中，`Mamba` 类的作用是提供基于状态空间的处理方法，使得模型可以有效地结合长距上下文和局部特征，从而帮助模型更好地识别和分析输入图像序列中的复杂模式。通过这种复杂而细致的特征混合，`Mamba` 促进了图像特征的深层次提取和语义理解，是增强模型性能的关键构建模块。
700d098128da95069ee0862d将forward_features中涉及到mamba的部分列出来。
在 `VisionMamba` 类的 `forward_features` 方法中涉及到 `Mamba` 的部分主要是在循环中对各个层进行迭代处理时调用的。以下是与 `Mamba` 类相关的部分代码：
```python
# 初始化 residual 和 hidden_states
residual = None
hidden_states = x
# 遍历模型的层
if not self.if_bidirectional: # 若模型不是双向的
for layer in self.layers:
if if_flip_img_sequences and self.if_rope:
hidden_states = hidden_states.flip([1])
if residual is not None:
residual = residual.flip([1])
# 使用 rotary embedding (若启用) 处理特征
if self.if_rope:
hidden_states = self.rope(hidden_states)
if residual is not None and self.if_rope_residual:
residual = self.rope(residual)
if if_flip_img_sequences and self.if_rope:
hidden_states = hidden_states.flip([1])
if residual is not None:
residual = residual.flip([1])
# 调用 Mamba 类在当前层的实例
hidden_states, residual = layer(
hidden_states, residual, inference_params=inference_params
)
```
在循环中，每个 `layer` 实际上是 `Block` 的实例，而 `Block` 类中包含了一个 `Mamba` 的实例。在 `layer` 调用时（这也是每个 `Block` 实例的调用），实际上调用了 `Block` 类的 `forward` 方法：
```python
hidden_states = self.mixer(hidden_states, inference_params=inference_params)
```
在这里，`self.mixer` 是 `Mamba` 类的实例。因此，在循环中调用每个 `layer` 时，就等于调用了 `Mamba` 的 `forward` 方法。这一调用负责处理输入的特征，执行状态空间模型和一系列投影、卷积操作，然后返回处理后的特征。通过这些处理，`Mamba` 可以在特征的维度、上下文混合和序列建模上发挥关键作用。25d2bb9847b868b3b0d09806现在基于上文，我们期望将class LinearAttention(nn.Module):
r""" Linear Attention with LePE and RoPE.
Args:
dim (int): Number of input channels.
num_heads (int): Number of attention heads.
qkv_bias (bool, optional): If True, add a learnable bias to query, key, value. Default: True
"""
def __init__(self, dim, input_resolution, num_heads, qkv_bias=True, **kwargs):
super().__init__()
self.dim = dim
self.input_resolution = input_resolution
self.num_heads = num_heads
self.qk = nn.Linear(dim, dim * 2, bias=qkv_bias)
self.elu = nn.ELU()
self.lepe = nn.Conv2d(dim, dim, 3, padding=1, groups=dim)
self.rope = RoPE(shape=(input_resolution[0], input_resolution[1], dim))
def forward(self, x):
"""
Args:
x: input features with shape of (B, N, C)
"""
b, n, c = x.shape
h = int(n ** 0.5)
w = int(n ** 0.5)
num_heads = self.num_heads
head_dim = c // num_heads
qk = self.qk(x).reshape(b, n, 2, c).permute(2, 0, 1, 3)
q, k, v = qk[0], qk[1], x
# q, k, v: b, n, c
q = self.elu(q) + 1.0
k = self.elu(k) + 1.0
q_rope = self.rope(q.reshape(b, h, w, c)).reshape(b, n, num_heads, head_dim).permute(0, 2, 1, 3)
k_rope = self.rope(k.reshape(b, h, w, c)).reshape(b, n, num_heads, head_dim).permute(0, 2, 1, 3)
q = q.reshape(b, n, num_heads, head_dim).permute(0, 2, 1, 3)
k = k.reshape(b, n, num_heads, head_dim).permute(0, 2, 1, 3)
v = v.reshape(b, n, num_heads, head_dim).permute(0, 2, 1, 3)
z = 1 / (q @ k.mean(dim=-2, keepdim=True).transpose(-2, -1) + 1e-6)
kv = (k_rope.transpose(-2, -1) * (n ** -0.5)) @ (v * (n ** -0.5))
x = q_rope @ kv * z
x = x.transpose(1, 2).reshape(b, n, c)
v = v.transpose(1, 2).reshape(b, h, w, c).permute(0, 3, 1, 2)
x = x + self.lepe(v).permute(0, 2, 3, 1).reshape(b, n, c)
return x
集成进去，给出集成位置。要将 `LinearAttention` 集成到给定的 `VisionMamba` 网络中，您需要一个清晰的目标和适当的位置。通常，我们希望在特定任务或模型需求的背景下整合注意力机制，这可能会在块级别（block-level）或网络级别（network-level）进行实现。
考虑到 `VisionMamba` 的现有架构设计，您可能想要将 `LinearAttention` 替代或添加到 Transformer 中的某些位置。以下是一些潜在的集成点：
1. **替换 `Mamba` 模块**：
- 如果 `LinearAttention` 被认为是模型的一种核心注意力机制，您可以在 `create_block` 函数中创建区块（block）的过程中，用 `LinearAttention` 替代 `Mamba`。
- 您需要确保 `LinearAttention` 和 `Mamba` 具有兼容的输入输出规模。
2. **作为 `Block` 的额外组件**：
- 作为一种更复杂的集成方法，您也可以设计 `Block` 来同时使用 `Mamba` 和 `LinearAttention`。这样您可以通过将它们的输出组合在一起或应用某种通路选择机制来利用这两者的优势。
- 在这种方法中，您需要调整 `Block` 的 `forward` 方法逻辑，以便处理来自 `LinearAttention` 和 `Mamba` 的不同信号。
3. **独立层**：
- 在 `VisionMamba` 网络中，可以考虑在 `PatchEmbed` 和 `Block` 之间或者是某些其他位置单独插入一个 `LinearAttention` 层。
- 这种方法适用于在增强长距离特征交互时希望引入额外的注意力模块，而不干扰 `Mamba` 的核心功能。
以下是可能的代码集成示例，假设我们将 `LinearAttention` 视为替代一种或添加到某种存在的机制中，如：
```python
class Block(nn.Module):
def __init__(self, dim, mixer_cls, attention_cls=None, norm_cls=nn.LayerNorm, ...):
super().__init__()
self.attention = attention_cls(dim, input_resolution=(14, 14), num_heads=8) if attention_cls else None
self.mixer = mixer_cls(dim)
self.norm = norm_cls(dim)
...
def forward(self, hidden_states: Tensor, residual: Optional[Tensor] = None, inference_params=None):
if self.attention:
hidden_states = self.attention(hidden_states)
# Existing logic
hidden_states, residual = self.mixer(hidden_states, residual, inference_params=inference_params)
return hidden_states, residual
def create_block(...):
...
# Create a block with both Mamba and LinearAttention
block = Block(
d_model,
mixer_cls=mixer_cls,
attention_cls=LinearAttention, # Add this
norm_cls=norm_cls,
...
)
...
```
请根据需要调整和扩展代码示例，以确保输入/输出维度的兼容性，并进行适当的训练和测试以验证集成的效果。4e25993cdf89fadebd22c128我们期望能够组合mamba和LinearAttention，而不是取代mamba
要组合 `Mamba` 和 `LinearAttention`，同时利用它们各自的特性，可以在 `Block` 中构建一个新的结构，使两者能够协作工作。这种组合可以为模型提供更多样化的信息抽象和更强的表现能力。以下是一个可能的组合实现方案：
### 设计思路
1. **添加 `LinearAttention` 到 `Block`**：
- 通过在 `Block` 中引入 `LinearAttention` 层，您可以在 `Mamba` 的处理逻辑之前或之后进行特征交互。
- 这将使得注意力机制与状态空间模型的能力相结合。
2. **组合信息**：
- 可以通过简单的加法、拼接或加权求和等策略将 `LinearAttention` 和 `Mamba` 的输出组合成一个统一的信息表示。
- 使用适当的归一化或门控机制来调节这两者的影响。
### 代码示例
结合两种机制后，`Block` 类可能会类似于以下代码：
```python
class Block(nn.Module):
def __init__(self, dim, mixer_cls, attention_cls, norm_cls=nn.LayerNorm, method='add', **kwargs):
"""
method: 指定组合两者输出的方式，例如 'add', 'concatenate'
"""
super().__init__()
self.attention = attention_cls(dim, input_resolution=(14, 14), num_heads=8, **kwargs)
self.mixer = mixer_cls(dim)
self.norm = norm_cls(dim)
self.method = method
def forward(self, hidden_states: Tensor, residual: Optional[Tensor] = None, inference_params=None):
attention_output = self.attention(hidden_states)
mixer_output, residual = self.mixer(hidden_states, residual, inference_params=inference_params)
if self.method == 'add':
combined_output = attention_output + mixer_output
elif self.method == 'concatenate':
combined_output = torch.cat((attention_output, mixer_output), dim=-1)
# If using concatenation, you may need an additional projection layer
combined_output = self.norm(combined_output)
return combined_output, residual
def create_block(...):
...
# Instantiate Block with both Mamba and LinearAttention
block = Block(
dim=d_model,
mixer_cls=mixer_cls,
attention_cls=LinearAttention,
norm_cls=norm_cls,
method='add', # 或 'concatenate'
...
)
...
```
### 注意事项
- **兼容性**：请确保 `attention_output` 和 `mixer_output` 在相加或拼接时具有相同或兼容的特征维度，尤其是当合并方法是 `"concatenate"` 时。
- **性能调优**：可以通过添加参数调节机制（例如 `gate` 机制）来精细控制 `LinearAttention` 和 `Mamba` 的影响程度。
- **实验验证**：记得在训练和测试阶段小心验证和调优组合模型，以确保该组合在特定任务中带来性能提升。
通过这种方式，您既利用了 `LinearAttention` 的强大局部特征提取能力，又结合了 `Mamba` 在处理长序列或复杂时序信号中的优势。2bcc5495fffca559fe99abce那么能否串联两种操作？
当然可以将 `Mamba` 和 `LinearAttention` 串联，以便在特征处理中依次应用这两种不同的操作。这种串联方式允许模型首先利用一种方法对输入进行初步处理后，再通过另一种方法进行进一步抽象和提取。
下面是如何在 `Block` 类中实现串联操作的示例：
### 编码示例
在这个示例中，我们假设先进行 `LinearAttention` 处理，然后再进行 `Mamba` 处理：
```python
class Block(nn.Module):
def __init__(self, dim, mixer_cls, attention_cls, norm_cls=nn.LayerNorm, **kwargs):
super().__init__()
# 初始化 LinearAttention 和 Mamba 两个对象
self.attention = attention_cls(dim, input_resolution=(14, 14), num_heads=8, **kwargs)
self.mixer = mixer_cls(dim)
self.norm = norm_cls(dim)
def forward(self, hidden_states: Tensor, residual: Optional[Tensor] = None, inference_params=None):
# 将输入特征经过 LinearAttention 处理
attention_output = self.attention(hidden_states)
# 使用处理后的特征经过 Mamba
mixer_output, residual = self.mixer(attention_output, residual, inference_params=inference_params)
# 将最终输出进行归一化或其他后处理
mixed_output = self.norm(mixer_output)
return mixed_output, residual
def create_block(...):
...
# 创建 Block 对象，包含串联的 LinearAttention 和 Mamba
block = Block(
dim=d_model,
mixer_cls=mixer_cls,
attention_cls=LinearAttention,
norm_cls=norm_cls,
...
)
...
```
### 逻辑说明
1. **线性注意力（LinearAttention）**：
- 首先对输入特征进行线性自注意力操作。这种方法适用于从输入中提取精细的局部关系。
2. **状态空间模型（Mamba）**：
- 在经历自注意力处理后，结果被传递到 `Mamba`，利用其在长序列建模和复杂时序处理中的优势。
3. **归一化**：
- 可以选择在最后加上归一化（或其他后处理），确保数据范围的稳定性和提高训练的收敛性。
### 使用注意
- **顺序**：注意 `Mamba` 和 `LinearAttention` 的顺序可以根据任务需求调整。
- **性能**：通过训练和验证确保这种串联带来的性能提升，并根据实验结果对模块的顺序或超参数进行调优。
- **输入输出维度**：确保每步处理的输入输出维度是匹配的。必要时，在不同模块之间加入额外的投影层以调整特征维度。9f218a436a8aab922b4d2a0a那么现在需要设置的是传参的问题.self.attention = attention_cls(dim, input_resolution=(14, 14), num_heads=8, **kwargs)如何设置？
要设置 `LinearAttention` 的参数，您需要确保在 `Block` 类和 `create_block` 函数中正确地传递这些参数。这里是如何组织和设置参数的一些建议：
### 参数设置
1. **确定 `LinearAttention` 的输入参数**：
- `dim`: 输入通道的数量通常对应于模型的嵌入维度。
- `input_resolution`: 这是一个二元组，表示输入特征图的长和宽。这可能需要从外部传入，或根据模型特定需求设置。
- `num_heads`: 注意力头的数量；可以是固定值或作为参数传入。
2. **利用 `**kwargs` 扩展传入的其他参数**：
- 如果 `LinearAttention` 类有其他可选或额外的参数，可以通过 `**kwargs` 传临时参数。
### 示例代码
以下是如何在 `Block` 类中传递参数并在 `create_block` 中调用：
```python
class Block(nn.Module):
def __init__(self, dim, mixer_cls, attention_cls, input_resolution, num_heads=8, norm_cls=nn.LayerNorm, **kwargs):
super().__init__()
# 初始化 LinearAttention 和 Mamba
self.attention = attention_cls(dim, input_resolution=input_resolution, num_heads=num_heads, **kwargs)
self.mixer = mixer_cls(dim)
self.norm = norm_cls(dim)
def forward(self, hidden_states: Tensor, residual: Optional[Tensor] = None, inference_params=None):
# 通过 LinearAttention 处理
attention_output = self.attention(hidden_states)
# 经过 Mamba 处理
mixer_output, residual = self.mixer(attention_output, residual, inference_params=inference_params)
# 最后的处理
mixed_output = self.norm(mixer_output)
return mixed_output, residual
def create_block(d_model, ssm_cfg=None, norm_epsilon=1e-5, drop_path=0., rms_norm=False, input_resolution=(14, 14), num_heads=8, **kwargs):
# 设置工厂参数
factory_kwargs = {"device": device, "dtype": dtype}
# 定义 mixer (Mamba) 类
mixer_cls = partial(Mamba, **ssm_cfg, **factory_kwargs)
# 定义 norm 类
norm_cls = partial(nn.LayerNorm if not rms_norm else RMSNorm, eps=norm_epsilon, **factory_kwargs)
# 创建 Block 对象
block = Block(
dim=d_model,
mixer_cls=mixer_cls,
attention_cls=LinearAttention,
input_resolution=input_resolution,
num_heads=num_heads,
norm_cls=norm_cls,
**kwargs
)
return block
```
### 注意事项
- **参数化**：确保 `input_resolution` 和 `num_heads` 是与任务相关的适当值，您可以根据特定的模型和数据集需求调整这些参数。
- **扩展性**：通过使用 `**kwargs`，可以方便地向 `LinearAttention` 或 `Block` 类传递额外或新的参数，而无需修改现有函数签名。
- **调试**：验证参数的正确传递和运行为保证模型在初期阶段正确设置的关键。c35c3ce9944d0da3f57aaba4这样吧，把二者add起来
为了将 `Mamba` 和 `LinearAttention` 的输出相加，您可以在 `Block` 类中定义一种组合方法，在将输入传递通过这两个模块后，将它们的输出相加。以下是实现这一逻辑的示例代码：
### 编码步骤
1. **在 `Block` 中定义两个模块**：一个用于 `LinearAttention`，另一个用于 `Mamba`。
2. **在 `forward` 方法中处理输入**：将输入依次传递给 `LinearAttention` 和 `Mamba`。
3. **组合操作**：将这两者的输出添加在一起，形成最终的输出。
### 示例代码
以下是如何实现这一方案的示例：
```python
class Block(nn.Module):
def __init__(self, dim, mixer_cls, attention_cls, input_resolution, num_heads=8, norm_cls=nn.LayerNorm, **kwargs):
super().__init__()
# 初始化 LinearAttention 和 Mamba 模块
self.attention = attention_cls(dim, input_resolution=input_resolution, num_heads=num_heads, **kwargs)
self.mixer = mixer_cls(dim)
self.norm = norm_cls(dim)
def forward(self, hidden_states: Tensor, residual: Optional[Tensor] = None, inference_params=None):
# 经过 LinearAttention 处理
attention_output = self.attention(hidden_states)
# 经过 Mamba 处理
mamba_output, residual = self.mixer(hidden_states, residual, inference_params=inference_params)
# 将两者的输出相加
combined_output = attention_output + mamba_output
# 进行归一化
normalized_output = self.norm(combined_output)
return normalized_output, residual
def create_block(d_model, ssm_cfg=None, norm_epsilon=1e-5, drop_path=0., rms_norm=False, input_resolution=(14, 14), num_heads=8, **kwargs):
# 设置工厂参数
factory_kwargs = {"device": kwargs.get('device'), "dtype": kwargs.get('dtype')}
# 定义 mixer (Mamba) 类
mixer_cls = partial(Mamba, **ssm_cfg, **factory_kwargs)
# 定义 norm 类
norm_cls = partial(nn.LayerNorm if not rms_norm else RMSNorm, eps=norm_epsilon, **factory_kwargs)
# 创建 Block 对象
block = Block(
dim=d_model,
mixer_cls=mixer_cls,
attention_cls=LinearAttention,
input_resolution=input_resolution,
num_heads=num_heads,
norm_cls=norm_cls,
**kwargs
)
return block
```
### 重要注意事项
- **输入维度匹配**：确保输入到 `attention` 和 `mamba` 的特征维度是匹配的。这种相加操作要求这两者输出特征具有相同的形状。
- **调试与验证**：在模型部署到实验中前，仔细验证网络的初始化和前向传播，以确认组合后的模型在您的训练任务中表现良好。可能需要对超参数进行调试以达到最佳效果。
通过这种方法，您可以有效地结合 `Mamba` 和 `LinearAttention` 的优势，两者的共同作用有望提升特征提取和建模能力。949e375ee9149f0f0e4a45fd现在分析import torch
import torch.nn as nn
from functools import partial
from torch import Tensor
from typing import Optional
from timm.models.vision_transformer import VisionTransformer, _cfg
from timm.models.registry import register_model
from timm.models.layers import trunc_normal_, lecun_normal_
from timm.models.layers import DropPath, to_2tuple
from timm.models.vision_transformer import _load_weights
import math
from collections import namedtuple
from mamba_ssm.modules.mamba_simple import Mamba
from mamba_ssm.utils.generation import GenerationMixin
from mamba_ssm.utils.hf import load_config_hf, load_state_dict_hf
from lib.models.mamba_fetrack.rope import *
import random
from lib.models.layers.head import build_box_head
import importlib
import lib.train.admin.settings as ws_settings
from .utils import combine_tokens, recover_tokens
#
from .HinBlock import HinResBlock
from .ResGLU import ResGLUAdapter, ModelArgs
from .Attn import LinearAttention
try:
from mamba_ssm.ops.triton.layernorm import RMSNorm, layer_norm_fn, rms_norm_fn
except ImportError:
RMSNorm, layer_norm_fn, rms_norm_fn = None, None, None
__all__ = [
'vim_tiny_patch16_224', 'vim_small_patch16_224', 'vim_base_patch16_224',
'vim_tiny_patch16_384', 'vim_small_patch16_384', 'vim_base_patch16_384',
]
class PatchEmbed(nn.Module):
""" 2D Image to Patch Embedding
"""
def __init__(self, img_size=224, patch_size=16, stride=16, in_chans=3, embed_dim=768, norm_layer=None, flatten=True):
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 = ((img_size[0] - patch_size[0]) // stride + 1, (img_size[1] - patch_size[1]) // stride + 1)
self.num_patches = self.grid_size[0] * self.grid_size[1]
self.flatten = flatten
self.proj = nn.Conv2d(in_chans, embed_dim, kernel_size=patch_size, stride=stride)
self.norm = norm_layer(embed_dim) if norm_layer else nn.Identity()
def forward(self, x):
B, C, H, W = x.shape
x = self.proj(x)
if self.flatten:
x = x.flatten(2).transpose(1, 2) # BCHW -> BNC
x = self.norm(x)
return x
class Block(nn.Module):
def __init__(
self, dim, mixer_cls, attention_cls, input_resolution, num_heads=8, norm_cls=nn.LayerNorm, fused_add_norm=False, residual_in_fp32=False,drop_path=0., integrate_glu=False
):
"""
Simple block wrapping a mixer class with LayerNorm/RMSNorm and residual connection"
This Block has a slightly different structure compared to a regular
prenorm Transformer block.
The standard block is: LN -> MHA/MLP -> Add.
[Ref: https://arxiv.org/abs/2002.04745]
Here we have: Add -> LN -> Mixer, returning both
the hidden_states (output of the mixer) and the residual.
This is purely for performance reasons, as we can fuse add and LayerNorm.
The residual needs to be provided (except for the very first block).
"""
super().__init__()
self.residual_in_fp32 = residual_in_fp32
self.fused_add_norm = fused_add_norm
#
self.attention = attention_cls(dim, input_resolution=input_resolution, num_heads=num_heads, **kwargs)
self.mixer = mixer_cls(dim)
self.norm = norm_cls(dim)
#
self.drop_path = DropPath(drop_path) if drop_path > 0. else nn.Identity()
if self.fused_add_norm:
assert RMSNorm is not None, "RMSNorm import fails"
assert isinstance(
self.norm, (nn.LayerNorm, RMSNorm)
), "Only LayerNorm and RMSNorm are supported for fused_add_norm"
# 引入门控机制
model_args = ModelArgs()
# 如果需要集成 ResGLUAdapter，则初始化
if integrate_glu:
model_args = ModelArgs()
self.glu_adapter = ResGLUAdapter(model_args)
else:
self.glu_adapter = None
# self.glu_adapter = ResGLUAdapter(model_args)
def forward(
self, hidden_states: Tensor, residual: Optional[Tensor] = None, inference_params=None
):
r"""Pass the input through the encoder layer.
Args:
hidden_states: the sequence to the encoder layer (required).
residual: hidden_states = Mixer(LN(residual))
"""
if not self.fused_add_norm:
if residual is None:
residual = hidden_states
else:
residual = residual + self.drop_path(hidden_states)
hidden_states = self.norm(residual.to(dtype=self.norm.weight.dtype))
if self.residual_in_fp32:
residual = residual.to(torch.float32)
else:
fused_add_norm_fn = rms_norm_fn if isinstance(self.norm, RMSNorm) else layer_norm_fn
if residual is None:
hidden_states, residual = fused_add_norm_fn(
hidden_states,
self.norm.weight,
self.norm.bias,
residual=residual,
prenorm=True,
residual_in_fp32=self.residual_in_fp32,
eps=self.norm.eps,
)
else:
hidden_states, residual = fused_add_norm_fn(
self.drop_path(hidden_states),
self.norm.weight,
self.norm.bias,
residual=residual,
prenorm=True,
residual_in_fp32=self.residual_in_fp32,
eps=self.norm.eps,
)
attention_out = self.attention(hidden_states)
hidden_states = self.mixer(hidden_states, inference_params=inference_params)
# 增加门控机制, 消除冗余和不相关的特征，保留最具表现力的输出
if self.glu_adapter is not None:
hidden_states = self.glu_adapter(hidden_states)
# hidden_states = self.glu_adapter(hidden_states)
hidden_states = attention_out + hidden_states
return hidden_states, residual
def allocate_inference_cache(self, batch_size, max_seqlen, dtype=None, **kwargs):
return self.mixer.allocate_inference_cache(batch_size, max_seqlen, dtype=dtype, **kwargs)
def create_block(
d_model,
ssm_cfg=None,
norm_epsilon=1e-5,
drop_path=0.,
rms_norm=False,
input_resolution=(14, 14),
num_heads=8,
residual_in_fp32=False,
fused_add_norm=False,
layer_idx=None,
integrate_glu=False, # 新增参数
device=None,
dtype=None,
if_bimamba=False,
bimamba_type="none",
if_devide_out=False,
init_layer_scale=None,
):
if if_bimamba:
bimamba_type = "v1"
if ssm_cfg is None:
ssm_cfg = {}
factory_kwargs = {"device": device, "dtype": dtype}
mixer_cls = partial(Mamba, layer_idx=layer_idx, bimamba_type=bimamba_type, if_devide_out=if_devide_out, init_layer_scale=init_layer_scale, **ssm_cfg, **factory_kwargs)
norm_cls = partial(
nn.LayerNorm if not rms_norm else RMSNorm, eps=norm_epsilon, **factory_kwargs
)
block = Block(
d_model,
mixer_cls,
attention_cls=LinearAttention,
input_resolution= input_resolution,
num_heads= num_heads,
norm_cls=norm_cls,
drop_path=drop_path,
fused_add_norm=fused_add_norm,
residual_in_fp32=residual_in_fp32,
integrate_glu=integrate_glu, # 传递参数
)
block.layer_idx = layer_idx
return block
# https://github.com/huggingface/transformers/blob/c28d04e9e252a1a099944e325685f14d242ecdcd/src/transformers/models/gpt2/modeling_gpt2.py#L454
def _init_weights(
module,
n_layer,
initializer_range=0.02, # Now only used for embedding layer.
rescale_prenorm_residual=True,
n_residuals_per_layer=1, # Change to 2 if we have MLP
):
if isinstance(module, nn.Linear):
if module.bias is not None:
if not getattr(module.bias, "_no_reinit", False):
nn.init.zeros_(module.bias)
elif isinstance(module, nn.Embedding):
nn.init.normal_(module.weight, std=initializer_range)
if rescale_prenorm_residual:
for name, p in module.named_parameters():
if name in ["out_proj.weight", "fc2.weight"]:
nn.init.kaiming_uniform_(p, a=math.sqrt(5))
with torch.no_grad():
p /= math.sqrt(n_residuals_per_layer * n_layer)
def segm_init_weights(m):
if isinstance(m, nn.Linear):
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.Conv2d):
# NOTE conv was left to pytorch default in my original init
lecun_normal_(m.weight)
if m.bias is not None:
nn.init.zeros_(m.bias)
elif isinstance(m, (nn.LayerNorm, nn.GroupNorm, nn.BatchNorm2d)):
nn.init.zeros_(m.bias)
nn.init.ones_(m.weight)
class VisionMamba(nn.Module):
def __init__(self,
img_size=224,
patch_size=16,
stride=16,
depth=24,
embed_dim=192,
channels=3,
num_classes=1000,
ssm_cfg=None,
drop_rate=0.,
drop_path_rate=0.1,
norm_epsilon: float = 1e-5,
rms_norm: bool = False,
initializer_cfg=None,
fused_add_norm=False,
residual_in_fp32=False,
device=None,
dtype=None,
ft_seq_len=None,
pt_hw_seq_len=14,
if_bidirectional=False,
final_pool_type='none',
if_abs_pos_embed=False,
if_rope=False,
if_rope_residual=False,
flip_img_sequences_ratio=-1.,
if_bimamba=False,
bimamba_type="none",
if_cls_token=False,
if_devide_out=False,
init_layer_scale=None,
use_double_cls_token=False,
use_middle_cls_token=False,
**kwargs):
factory_kwargs = {"device": device, "dtype": dtype}
# add factory_kwargs into kwargs
kwargs.update(factory_kwargs)
super().__init__()
self.residual_in_fp32 = residual_in_fp32
self.fused_add_norm = fused_add_norm
self.if_bidirectional = if_bidirectional
self.final_pool_type = final_pool_type
self.if_abs_pos_embed = if_abs_pos_embed
self.if_rope = if_rope
self.if_rope_residual = if_rope_residual
self.flip_img_sequences_ratio = flip_img_sequences_ratio
# self.if_cls_token = if_cls_token
self.if_cls_token = False
self.use_double_cls_token = use_double_cls_token
self.use_middle_cls_token = use_middle_cls_token
self.num_tokens = 1 if if_cls_token else 0
# pretrain parameters
self.num_classes = num_classes
self.d_model = self.num_features = self.embed_dim = embed_dim # num_features for consistency with other models
base_filter=32
self.hin_block = nn.Sequential(
nn.Conv2d(channels, base_filter, kernel_size=3, stride=1, padding=1),
HinResBlock(base_filter, base_filter), # 第一个 HinResBlock
HinResBlock(base_filter, base_filter), # 第二个 HinResBlock
HinResBlock(base_filter, base_filter), # 第三个 HinResBlock
nn.Conv2d(base_filter, channels, kernel_size=1) # 将通道数从32转换为3
)
##############################
self.patch_embed = PatchEmbed(
img_size=img_size, patch_size=patch_size, stride=stride, in_chans=channels, embed_dim=embed_dim)
num_patches = self.patch_embed.num_patches
if if_cls_token:
if use_double_cls_token:
self.cls_token_head = nn.Parameter(torch.zeros(1, 1, self.embed_dim))
self.cls_token_tail = nn.Parameter(torch.zeros(1, 1, self.embed_dim))
self.num_tokens = 2
else:
self.cls_token = nn.Parameter(torch.zeros(1, 1, self.embed_dim))
# self.num_tokens = 1
if if_abs_pos_embed:
# self.pos_embed = nn.Parameter(torch.zeros(1, num_patches + self.num_tokens, self.embed_dim))
self.pos_embed_x = nn.Parameter(torch.zeros(1, 256, self.embed_dim))
self.pos_embed_z = nn.Parameter(torch.zeros(1, 64, self.embed_dim))
self.pos_drop = nn.Dropout(p=drop_rate)
if if_rope:
half_head_dim = embed_dim // 2
hw_seq_len = img_size // patch_size
self.rope = VisionRotaryEmbeddingFast(
dim=half_head_dim,
pt_seq_len=pt_hw_seq_len,
ft_seq_len=hw_seq_len
)
# TODO: release this comment
dpr = [x.item() for x in torch.linspace(0, drop_path_rate, depth)] # stochastic depth decay rule
# import ipdb;ipdb.set_trace()
inter_dpr = [0.0] + dpr
self.drop_path = DropPath(drop_path_rate) if drop_path_rate > 0. else nn.Identity()
# transformer blocks
self.layers = nn.ModuleList(
[
create_block(
embed_dim,
ssm_cfg=ssm_cfg,
norm_epsilon=norm_epsilon,
rms_norm=rms_norm,
residual_in_fp32=residual_in_fp32,
fused_add_norm=fused_add_norm,
layer_idx=i,
integrate_glu=(i == 0), # 仅在第一个块中集成 ResGLUAdapter
if_bimamba=if_bimamba,
bimamba_type=bimamba_type,
drop_path=inter_dpr[i],
if_devide_out=if_devide_out,
init_layer_scale=init_layer_scale,
**factory_kwargs,
)
for i in range(depth)
]
)
# output head
self.norm_f = (nn.LayerNorm if not rms_norm else RMSNorm)(
embed_dim, eps=norm_epsilon, **factory_kwargs
)
# self.pre_logits = nn.Identity()
# original init
self.patch_embed.apply(segm_init_weights)
# self.head.apply(segm_init_weights)
if if_abs_pos_embed:
trunc_normal_(self.pos_embed_x, std=.02)
trunc_normal_(self.pos_embed_z, std=.02)
if if_cls_token:
if use_double_cls_token:
trunc_normal_(self.cls_token_head, std=.02)
trunc_normal_(self.cls_token_tail, std=.02)
else:
trunc_normal_(self.cls_token, std=.02)
# mamba init
self.apply(
partial(
_init_weights,
n_layer=depth,
**(initializer_cfg if initializer_cfg is not None else {}),
)
)
def allocate_inference_cache(self, batch_size, max_seqlen, dtype=None, **kwargs):
return {
i: layer.allocate_inference_cache(batch_size, max_seqlen, dtype=dtype, **kwargs)
for i, layer in enumerate(self.layers)
}
@torch.jit.ignore
def no_weight_decay(self):
return {"pos_embed", "cls_token", "dist_token", "cls_token_head", "cls_token_tail"}
@torch.jit.ignore()
def load_pretrained(self, checkpoint_path, prefix=""):
_load_weights(self, checkpoint_path, prefix)
def forward_features(self, z, x, inference_params=None, if_random_cls_token_position=False, if_random_token_rank=False, target_layers=None):
# print(f"Shape of x before hin_block: {x.shape if x is not None else 'None'}")
x = self.hin_block(x)
z = self.hin_block(z)
x = self.patch_embed(x) #x.shape = torch.Size([B, 3, 256, 256]) -> torch.Size([2, 256, 384])
z = self.patch_embed(z) #z.shape = torch.Size([B, 3, 128, 128]) -> torch.Size([2, 64, 384])
B, M, _ = x.shape
if self.if_cls_token: # False
if self.use_double_cls_token:
cls_token_head = self.cls_token_head.expand(B, -1, -1)
cls_token_tail = self.cls_token_tail.expand(B, -1, -1)
token_position = [0, M + 1]
x = torch.cat((cls_token_head, x, cls_token_tail), dim=1)
M = x.shape[1]
else:
if self.use_middle_cls_token:
cls_token = self.cls_token.expand(B, -1, -1)
token_position = M // 2
# add cls token in the middle
x = torch.cat((x[:, :token_position, :], cls_token, x[:, token_position:, :]), dim=1)
elif if_random_cls_token_position:
cls_token = self.cls_token.expand(B, -1, -1)
token_position = random.randint(0, M)
x = torch.cat((x[:, :token_position, :], cls_token, x[:, token_position:, :]), dim=1)
print("token_position: ", token_position)
else:
cls_token = self.cls_token.expand(B, -1, -1) # stole cls_tokens impl from Phil Wang, thanks
token_position = 0
x = torch.cat((cls_token, x), dim=1)
M = x.shape[1]
if self.if_abs_pos_embed: # True
x = x + self.pos_embed_x # x = x + positon_embemding =torch.Size([B, 256, 384]) + torch.Size([1, 256, 384]) = torch.Size([B, 256, 384])
z = z + self.pos_embed_z # z = z + positon_embemding =torch.Size([B, 64, 384]) + torch.Size([1, 64, 384]) = torch.Size([B, 64, 384])
x = torch.cat((z, x), dim=1) # torch.Size([B, 320, 384])
x = self.pos_drop(x) # x.shape = torch.Size([B, 320, 384])
if if_random_token_rank: #False
# 生成随机 shuffle 索引
shuffle_indices = torch.randperm(M)
if isinstance(token_position, list):
print("original value: ", x[0, token_position[0], 0], x[0, token_position[1], 0])
else:
print("original value: ", x[0, token_position, 0])
print("original token_position: ", token_position)
# 执行 shuffle
x = x[:, shuffle_indices, :]
if isinstance(token_position, list):
# 找到 cls token 在 shuffle 之后的新位置
new_token_position = [torch.where(shuffle_indices == token_position[i])[0].item() for i in range(len(token_position))]
token_position = new_token_position
else:
# 找到 cls token 在 shuffle 之后的新位置
token_position = torch.where(shuffle_indices == token_position)[0].item()
if isinstance(token_position, list):
print("new value: ", x[0, token_position[0], 0], x[0, token_position[1], 0])
else:
print("new value: ", x[0, token_position, 0])
print("new token_position: ", token_position)
if_flip_img_sequences = False
if self.flip_img_sequences_ratio > 0 and (self.flip_img_sequences_ratio - random.random()) > 1e-5: # False
x = x.flip([1])
if_flip_img_sequences = True
# mamba impl
residual = None
hidden_states = x
outputs = {}
i = None
if not self.if_bidirectional: # True
for i, layer in enumerate(self.layers):
if if_flip_img_sequences and self.if_rope: # False
hidden_states = hidden_states.flip([1])
if residual is not None:
residual = residual.flip([1])
# rope about
if self.if_rope: # False
hidden_states = self.rope(hidden_states)
if residual is not None and self.if_rope_residual:
residual = self.rope(residual)
if if_flip_img_sequences and self.if_rope: # False
hidden_states = hidden_states.flip([1])
if residual is not None:
residual = residual.flip([1])
hidden_states, residual = layer(
hidden_states, residual, inference_params=inference_params
)
# 检查是否是目标层
if target_layers is not None and i in target_layers:
outputs[i] = hidden_states
# print(f"Stored layer {i} output, hidden_states shape: {hidden_states.shape}")
else: # False
# get two layers in a single for-loop
for i in range(len(self.layers) // 2):
if self.if_rope:
hidden_states = self.rope(hidden_states)
if residual is not None and self.if_rope_residual:
residual = self.rope(residual)
hidden_states_f, residual_f = self.layers[i * 2](
hidden_states, residual, inference_params=inference_params
)
hidden_states_b, residual_b = self.layers[i * 2 + 1](
hidden_states.flip([1]), None if residual == None else residual.flip([1]), inference_params=inference_params
)
hidden_states = hidden_states_f + hidden_states_b.flip([1])
residual = residual_f + residual_b.flip([1])
if not self.fused_add_norm: #False
if residual is None:
residual = hidden_states
else:
residual = residual + self.drop_path(hidden_states)
hidden_states = self.norm_f(residual.to(dtype=self.norm_f.weight.dtype))
else: #True
# Set prenorm=False here since we don't need the residual
fused_add_norm_fn = rms_norm_fn if isinstance(self.norm_f, RMSNorm) else layer_norm_fn
hidden_states = fused_add_norm_fn( # hidden_states.shape = torch.Size([B, 320, 384])
self.drop_path(hidden_states),
self.norm_f.weight,
self.norm_f.bias,
eps=self.norm_f.eps,
residual=residual,
prenorm=False,
residual_in_fp32=self.residual_in_fp32,
)
# return only cls token if it exists
if self.if_cls_token: #False
if self.use_double_cls_token:
return (hidden_states[:, token_position[0], :] + hidden_states[:, token_position[1], :]) / 2
else:
if self.use_middle_cls_token:
return hidden_states[:, token_position, :]
elif if_random_cls_token_position:
return hidden_states[:, token_position, :]
else:
return hidden_states[:, token_position, :]
if self.final_pool_type == 'none':
final_output = hidden_states[:, -1, :]
elif self.final_pool_type == 'mean':
final_output = hidden_states.mean(dim=1)
elif self.final_pool_type == 'max':
final_output = hidden_states
elif self.final_pool_type == 'all':
final_output = hidden_states
else:
raise NotImplementedError
outputs['final'] = hidden_states
# 如果没有指定 target_layers，只返回最终特征
if not target_layers:
return final_output
# 有指定 target_layers，返回包含指定层和最终特征的字典
return outputs
def forward(self, x, return_features=False, inference_params=None, if_random_cls_token_position=False, if_random_token_rank=False):
x = self.forward_features(x, inference_params, if_random_cls_token_position=if_random_cls_token_position, if_random_token_rank=if_random_token_rank)
if return_features:
return x
x = self.head(x)
if self.final_pool_type == 'max':
x = x.max(dim=1)[0]
return x
@register_model
def vim_tiny_patch16_224_bimambav2_final_pool_mean_abs_pos_embed_with_midclstok_div2(pretrained=False, **kwargs):
model = VisionMamba(
patch_size=16, embed_dim=192, depth=24, rms_norm=True, residual_in_fp32=True, fused_add_norm=True, final_pool_type='mean', if_abs_pos_embed=True, if_rope=False, if_rope_residual=False, bimamba_type="v2", if_cls_token=True, if_devide_out=True, use_middle_cls_token=True, **kwargs)
model.default_cfg = _cfg()
if pretrained:
checkpoint = torch.hub.load_state_dict_from_url(
url="to.do",
map_location="cpu", check_hash=True
)
model.load_state_dict(checkpoint["model"])
return model
@register_model
def vim_tiny_patch16_stride8_224_bimambav2_final_pool_mean_abs_pos_embed_with_midclstok_div2(pretrained=False, **kwargs):
model = VisionMamba(
patch_size=16, stride=8, embed_dim=192, depth=24, rms_norm=True, residual_in_fp32=True, fused_add_norm=True, final_pool_type='mean', if_abs_pos_embed=True, if_rope=False, if_rope_residual=False, bimamba_type="v2", if_cls_token=True, if_devide_out=True, use_middle_cls_token=True, **kwargs)
model.default_cfg = _cfg()
if pretrained:
checkpoint = torch.hub.load_state_dict_from_url(
url="to.do",
map_location="cpu", check_hash=True
)
model.load_state_dict(checkpoint["model"])
return model
@register_model
def vim_small_patch16_224_bimambav2_final_pool_mean_abs_pos_embed_with_midclstok_div2(pretrained=False, **kwargs):
model = VisionMamba(
patch_size=16, embed_dim=384, depth=24, rms_norm=True, residual_in_fp32=True, fused_add_norm=True, final_pool_type='all', if_abs_pos_embed=True, if_rope=False, if_rope_residual=False, bimamba_type="v2", if_cls_token=True, if_devide_out=True, use_middle_cls_token=True, **kwargs)
model.default_cfg = _cfg()
if pretrained:
checkpoint = torch.load(pretrained, map_location="cpu")
missing_keys, unexpected_keys = model.load_state_dict(checkpoint["model"], strict=False)
print('Load pretrained model from: ' + pretrained)
return model
@register_model
def vim_small_patch16_stride8_224_bimambav2_final_pool_mean_abs_pos_embed_with_midclstok_div2(pretrained=False, **kwargs):
model = VisionMamba(
patch_size=16, stride=8, embed_dim=384, depth=24, rms_norm=True, residual_in_fp32=True, fused_add_norm=True, final_pool_type='mean', if_abs_pos_embed=True, if_rope=False, if_rope_residual=False, bimamba_type="v2", if_cls_token=True, if_devide_out=True, use_middle_cls_token=True, **kwargs)
model.default_cfg = _cfg()
if pretrained:
checkpoint = torch.hub.load_state_dict_from_url(
url="to.do",
map_location="cpu", check_hash=True
)
model.load_state_dict(checkpoint["model"])
return model
中关于LinearAttention的集成。分析哪里有问题？在给定代码中，`LinearAttention` 已被集成到 `Block` 类中。我将注意到几个可能的问题和需要修正的地方，以及需要注意的方面：
### 1. **缺失的参数传递**
在 `Block` 类的构造函数中，`LinearAttention` 被初始化时使用了 `**kwargs`：
```python
self.attention = attention_cls(dim, input_resolution=input_resolution, num_heads=num_heads, **kwargs)
```
不过，这个 `kwargs` 在 `Block` 的初始化中似乎没有被显式传递，因此传入的参数必须在 `create_block` 函数或 `Block` 初始化时明确传递。
### 2. **输入/输出维度**
确保 `LinearAttention` 和 `Mamba` 的输入和输出维度是相匹配的。在前向传播中：
```python
attention_out = self.attention(hidden_states)
hidden_states = self.mixer(hidden_states, inference_params=inference_params)
```
请确保 `hidden_states` 的维度对于这两个模块都是相同的，以使得它们能够顺利处理。如果 `hidden_states` 的维度在 `attention` 或 `mixer` 处理后发生改变（通常不会，因为它们都为线性投影），可能需要额外的线性层来匹配这个变化。
### 3. **GLU 集成逻辑**
`ResGLUAdapter` 可能需要明确的初始化，它在代码中被条件性引入：
```python
if integrate_glu:
self.glu_adapter = ResGLUAdapter(model_args)
else:
self.glu_adapter = None
```
确保 `ResGLUAdapter` 的集成逻辑清晰可验证，按您所需的条件自动初始化。此外，需要验证的是 `ResGLUAdapter` 的输入和输出维度合乎上下文逻辑。
### 4. **forward_features 方法**
在使用 `LinearAttention`，`Mamba` 和可能的 `GLU` 时：
```python
# In Block.forward
attention_out = self.attention(hidden_states)
hidden_states = self.mixer(hidden_states, inference_params=inference_params)
if self.glu_adapter is not None:
hidden_states = self.glu_adapter(hidden_states)
hidden_states = attention_out + hidden_states
```
确保 `attention_out` 和 `hidden_states` 的相加操作在形状和渠道数量上是相容的。通常，在 Transformer 架构中这会很自然地得到满足，然而，必须确认这一点。
### 5. **参数初始化和权重共享**
任何需要共享的参数或权重需要确保在多个 `Block` 或整个 `VisionMamba` 模型中让权重之间保持一致。
### 推荐操作
- **测试**：通过单元测试确保每个块以及组合模块功能的相互转换。
- **日志记录**：增加调试信息以帮助确诊问题。可以在 forward 函数中通过打印张量的尺寸来验证。
通过这些调试和改进措施，可以帮助确保 `LinearAttention` 的正确集成，同时有效实现其特性提供的增强效果。