https://huggingface.co/spaces/yuntian-deng/ChatGPT
Chatbot
将# 将 4输入分开，构建新的相同模态结合的2输入，2分支
import math
import logging
from functools import partial
from collections import OrderedDict
from copy import deepcopy

import torch
import torch.nn as nn
import torch.nn.functional as F

from timm.models.layers import to_2tuple

from lib.models.layers.patch_embed import PatchEmbed, PatchEmbed_event, xcorr_depthwise
from .utils import combine_tokens, recover_tokens
from .vit import VisionTransformer
from …layers.attn_blocks import CEBlock
from .new_counter_guide import Counter_Guide
from .ad_counter_guide import Counter_Guide_Enhanced
from …mamba_fetrack.models_mamba import VisionMamba, vim_small_patch16_224_bimambav2_final_pool_mean_abs_pos_embed_with_midclstok_div2
# 集成CrossMamba模块进行不同模态特征交互
from …mamba_fetrack.mamba_cross import CrossMamba
_logger = logging.getLogger(name)


class VisionTransformerCE(VisionTransformer):
“”" Vision Transformer with candidate elimination (CE) module

A PyTorch impl of : An Image is Worth 16x16 Words: Transformers for Image Recognition at Scale
- https://arxiv.org/abs/2010.11929

Includes distillation token & head support for DeiT: Data-efficient Image Transformers
- https://arxiv.org/abs/2012.12877
“”“
def init(self, img_size=224, patch_size=16, in_chans=3, num_classes=1000, embed_dim=768, depth=12,
num_heads=12, mlp_ratio=4., qkv_bias=True, representation_size=None, distilled=False,
drop_rate=0., attn_drop_rate=0., drop_path_rate=0., embed_layer=PatchEmbed, norm_layer=None,
act_layer=None, weight_init=‘’,
ce_loc=None, ce_keep_ratio=None):
super().init()
if isinstance(img_size, tuple):
self.img_size = img_size
else:
self.img_size = to_2tuple(img_size)
self.patch_size = patch_size
self.in_chans = in_chans

self.num_classes = num_classes
self.num_features = self.embed_dim = embed_dim # num_features for consistency with other models
self.num_tokens = 2 if distilled else 1
norm_layer = norm_layer or partial(nn.LayerNorm, eps=1e-6)
act_layer = act_layer or nn.GELU

self.patch_embed = embed_layer(
img_size=img_size, patch_size=patch_size, in_chans=in_chans, embed_dim=embed_dim)
num_patches = self.patch_embed.num_patches

self.cls_token = nn.Parameter(torch.zeros(1, 1, embed_dim))
self.dist_token = nn.Parameter(torch.zeros(1, 1, embed_dim)) if distilled else None
self.pos_embed = nn.Parameter(torch.zeros(1, num_patches + self.num_tokens, embed_dim))
self.pos_drop = nn.Dropout(p=drop_rate)
self.pos_embed_event = PatchEmbed_event(in_chans=32, embed_dim=768, kernel_size=4, stride=4)
dpr = [x.item() for x in torch.linspace(0, drop_path_rate, depth)] # stochastic depth decay rule
blocks = []
ce_index = 0
self.ce_loc = ce_loc
for i in range(depth):
ce_keep_ratio_i = 1.0
if ce_loc is not None and i in ce_loc:
ce_keep_ratio_i = ce_keep_ratio[ce_index]
ce_index += 1

blocks.append(
CEBlock(
dim=embed_dim, num_heads=num_heads, mlp_ratio=mlp_ratio, qkv_bias=qkv_bias, drop=drop_rate,
attn_drop=attn_drop_rate, drop_path=dpr[i], norm_layer=norm_layer, act_layer=act_layer,
keep_ratio_search=ce_keep_ratio_i)
)

self.blocks = nn.Sequential(blocks)
self.norm = norm_layer(embed_dim)

self.init_weights(weight_init)

# 添加交互模块counter_guide
self.counter_guide = Counter_Guide_Enhanced(768, 768)
# 初始化CrossMamba模块
self.cross_mamba = CrossMamba(dim=768)
#########################################

def forward_features(self, z, x, event_z, event_x,
mask_z=None, mask_x=None,
ce_template_mask=None, ce_keep_rate=None,
return_last_attn=False
):
# 分支1 处理流程
B, H, W = x.shape[0], x.shape[2], x.shape[3]

x = self.patch_embed(x)
z = self.patch_embed(z)
z += self.pos_embed_z
x += self.pos_embed_x

if mask_z is not None and mask_x is not None:
mask_z = F.interpolate(mask_z[None].float(), scale_factor=1. / self.patch_size).to(torch.bool)[0]
mask_z = mask_z.flatten(1).unsqueeze(-1)

mask_x = F.interpolate(mask_x[None].float(), scale_factor=1. / self.patch_size).to(torch.bool)[0]
mask_x = mask_x.flatten(1).unsqueeze(-1)

mask_x = combine_tokens(mask_z, mask_x, mode=self.cat_mode)
mask_x = mask_x.squeeze(-1)

if self.add_cls_token:
cls_tokens = self.cls_token.expand(B, -1, -1)
cls_tokens = cls_tokens + self.cls_pos_embed
if self.add_sep_seg:
x += self.search_segment_pos_embed
z += self.template_segment_pos_embed
x = combine_tokens(z, x, mode=self.cat_mode)
if self.add_cls_token:
x = torch.cat([cls_tokens, x], dim=1)
x = self.pos_drop(x)

lens_z = self.pos_embed_z.shape[1]
lens_x = self.pos_embed_x.shape[1]
global_index_t = torch.linspace(0, lens_z - 1, lens_z).to(x.device)
global_index_t = global_index_t.repeat(B, 1)
global_index_s = torch.linspace(0, lens_x - 1, lens_x).to(x.device)
global_index_s = global_index_s.repeat(B, 1)
removed_indexes_s = []
# # 分支2 处理流程
event_x = self.pos_embed_event(event_x)
event_z = self.pos_embed_event(event_z)
event_x += self.pos_embed_x
event_z += self.pos_embed_z

event_x = combine_tokens(event_z, event_x, mode=self.cat_mode)

if self.add_cls_token:
event_x = torch.cat([cls_tokens, event_x], dim=1)

lens_z = self.pos_embed_z.shape[1]
lens_x = self.pos_embed_x.shape[1]

global_index_t1 = torch.linspace(0, lens_z - 1, lens_z).to(event_x.device)
global_index_t1 = global_index_t1.repeat(B, 1)
global_index_s1 = torch.linspace(0, lens_x - 1, lens_x).to(event_x.device)
global_index_s1 = global_index_s1.repeat(B, 1)

removed_indexes_s1 = []

for i, blk in enumerate(self.blocks):
# 第一个分支处理
x, global_index_t, global_index_s, removed_index_s, attn = <br/> blk(x, global_index_t, global_index_s, mask_x, ce_template_mask, ce_keep_rate)
# 第二个分支处理
event_x, global_index_t1, global_index_s1, removed_index_s1, attn1 = <br/> blk(event_x, global_index_t1, global_index_s1, mask_x, ce_template_mask, ce_keep_rate)


if self.ce_loc is not None and i in self.ce_loc:
removed_indexes_s.append(removed_index_s)
removed_indexes_s1.append(removed_index_s1)

# 在第1层增加counter_guide模块,验证早期融合效果
if i == 0 :
# 引入counter_guide进行特征增强
enhanced_x, enhanced_event_x = self.counter_guide(x, event_x)
# 将增强后的特征与原特征相加
x = x + enhanced_x
event_x = event_x + enhanced_event_x
##############
lens_x_new = global_index_s.shape[1]
lens_z_new = global_index_t.shape[1]
z = x[:, :lens_z_new2]
x = x[:, lens_z_new2:]
lens_eventx_new = global_index_s1.shape[1]
lens_eventz_new = global_index_t1.shape[1]
event_z = event_x[:, :lens_eventx_new2]
event_x = event_x[:, lens_eventx_new*2:]
## 恢复原始的 token 顺序 #################################
if removed_indexes_s and removed_indexes_s[0] is not None:
removed_indexes_cat = torch.cat(removed_indexes_s, dim=1)
pruned_lens_x = lens_x - lens_x_new
pad_x = torch.zeros([B, pruned_lens_x, x.shape[2]], device=x.device)
x = torch.cat([x, pad_x], dim=1)
index_all = torch.cat([global_index_s, removed_indexes_cat], dim=1)
# 恢复原始顺序
C = x.shape[-1]
x = torch.zeros_like(x).scatter(dim=1, index=index_all.unsqueeze(-1).expand(B, -1, C).to(torch.int64), src=x)
x = recover_tokens(x, lens_z_new, lens_x, mode=self.cat_mode)

if removed_indexes_s1 and removed_indexes_s1[0] is not None:
removed_indexes_cat1 = torch.cat(removed_indexes_s1, dim=1)
pruned_lens_x1 = lens_x - lens_eventx_new
pad_x1 = torch.zeros([B, pruned_lens_x1, event_x.shape[2]], device=event_x.device)
event_x = torch.cat([event_x, pad_x1], dim=1)
index_all1 = torch.cat([global_index_s1, removed_indexes_cat1], dim=1)
# 恢复原始顺序
C = event_x.shape[-1]
event_x = torch.zeros_like(event_x).scatter(dim=1, index=index_all1.unsqueeze(-1).expand(B, -1, C).to(torch.int64), src=event_x)
event_x = recover_tokens(event_x, lens_eventz_new, lens_x, mode=self.cat_mode)
################################################################

## 引入Crossmamba进行事件和图像特征融合,不采用concatenate
x = self.cross_mamba(x, event_x)
# 应用LayerNorm归一化处理
x = self.norm(x)

aux_dict = {
“attn”: attn,
‘attn1’: attn1,
“removed_indexes_s”: removed_indexes_s, # used for visualization
‘removed_indexes_s1’: removed_indexes_s1,
}

return x, aux_dict

def forward(self, z, x, event_z, event_x,
ce_template_mask=None, ce_keep_rate=None,
tnc_keep_rate=None,
return_last_attn=False):

x, aux_dict = self.forward_features(z, x, event_z, event_x, ce_template_mask=ce_template_mask, ce_keep_rate=ce_keep_rate,)

return x, aux_dict


def _create_vision_transformer(pretrained=False, **kwargs):
model = VisionTransformerCE(**kwargs)

if pretrained:
if ‘npz’ in pretrained:
model.load_pretrained(pretrained, prefix=‘’)
else:
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


def vit_base_patch16_224_ce(pretrained=False, **kwargs):
“”” ViT-Base model (ViT-B/16) from original paper (https://arxiv.org/abs/2010.11929).
“”“
model_kwargs = dict(
patch_size=16, embed_dim=768, depth=12, num_heads=12, **kwargs)
model = _create_vision_transformer(pretrained=pretrained, **model_kwargs)
return model


def vit_large_patch16_224_ce(pretrained=False, **kwargs):
“”” ViT-Large model (ViT-L/16) from original paper (https://arxiv.org/abs/2010.11929).
“”"
model_kwargs = dict(
patch_size=16, embed_dim=1024, depth=24, num_heads=16, **kwargs)
model = _create_vision_transformer(pretrained=pretrained, **model_kwargs)
return model中的x = recover_tokens(x, lens_z_new, lens_x, mode=self.cat_mode)和event_x = recover_tokens(event_x, lens_eventz_new, lens_x, mode=self.cat_mode)的lens_z_new，lens_eventz_new进行检查。
在给定的代码片段中，用于恢复原始 token 顺序的 lens_z_new 和 lens_eventz_new 变量应当代表在某个处理中实际保留的 token 的实际长度（数量），这些变量的正确性对于模型的正常工作至关重要。

需要检查以下代码段：

lens_z_new = global_index_t.shape[1]
lens_eventx_new = global_index_s1.shape[1]
lens_eventz_new = global_index_t1.shape[1]


确保 lens_z_new 和 lens_eventz_new 正确地反映了保留的 token 数量。从代码片段来看，变量 global_index_t 和 global_index_t1 代表了在特定层中实际参与计算的 token 索引，因此这些变量的长度应当就是需要的长度。

lens_z_new = global_index_t.shape[1]
lens_eventx_new = global_index_s1.shape[1]
lens_eventz_new = global_index_t1.shape[1]


代码中 lens_z_new2 和 lens_eventx_new2 似乎可能是打印或运行时变量的命名问题，这两个变量名在其他部分没有看到定义及其用法。

此外，在 recover_tokens 的地方，也要确保传递的 lens_z_new 和 lens_eventz_new 表示的是正确的长度：

x = recover_tokens(x, lens_z_new, lens_x, mode=self.cat_mode)
event_x = recover_tokens(event_x, lens_eventz_new, lens_x, mode=self.cat_mode)


- lens_z_new 应该对应的是 z 的保留长度。
- lens_eventz_new 应该对应的是 event_z 的保留长度。

最后，检查代码中 global_index_t，global_index_s1 这些变量形状是否正确，确保在循环 for i, blk in enumerate(self.blocks) 中，每次循环结束后这些变量都有正确的形状和长度，来正确计数。

例如，确保 global_index_t 在合并和削减操作后的形状符合预期：

global_index_t = torch.linspace(0, lens_z - 1, lens_z).to(x.device)


及后续计算处理中的每一步都准确反映全局索引的数量。

通用步骤如下：

1. 初始长度设定检查: 确保 global_index_t 和 global_index_t1 等变量初始设定的长度是合理的。
2. 中间处理检查: 检查在 self.blocks 循环中的每一个中间处理步骤（如 removed_indexes_s 和 removed_indexes_s1 是否正确添加和计算）。
3. 恢复 Token 检查: 确认在恢复 token 的步骤中传入的长度参数是之前中间处理步骤合乎逻辑推导出来的值。

确保这些步骤覆盖完整，输出 token 序列的复原应该是准确的。
参考import math
import logging
from functools import partial
from collections import OrderedDict
from copy import deepcopy

import torch
import torch.nn as nn
import torch.nn.functional as F

from timm.models.layers import to_2tuple

from lib.models.layers.patch_embed import PatchEmbed, PatchEmbed_event, xcorr_depthwise
from .utils import combine_tokens, recover_tokens
from .vit import VisionTransformer
from …layers.attn_blocks import CEBlock

logger = logging.getLogger(name)


class VisionTransformerCE(VisionTransformer):
“”" Vision Transformer with candidate elimination (CE) module

A PyTorch impl of : An Image is Worth 16x16 Words: Transformers for Image Recognition at Scale
- https://arxiv.org/abs/2010.11929

Includes distillation token & head support for DeiT: Data-efficient Image Transformers
- https://arxiv.org/abs/2012.12877
“”“

def init(self, img_size=224, patch_size=16, in_chans=3, num_classes=1000, embed_dim=768, depth=12,
num_heads=12, mlp_ratio=4., qkv_bias=True, representation_size=None, distilled=False,
drop_rate=0., attn_drop_rate=0., drop_path_rate=0., embed_layer=PatchEmbed, norm_layer=None,
act_layer=None, weight_init=‘’,
ce_loc=None, ce_keep_ratio=None):
“””
Args:
img_size (int, tuple): input image size
patch_size (int, tuple): patch size
in_chans (int): number of input channels
num_classes (int): number of classes for classification head
embed_dim (int): embedding dimension
depth (int): depth of transformer
num_heads (int): number of attention heads
mlp_ratio (int): ratio of mlp hidden dim to embedding dim
qkv_bias (bool): enable bias for qkv if True
representation_size (Optional[int]): enable and set representation layer (pre-logits) to this value if set
distilled (bool): model includes a distillation token and head as in DeiT models
drop_rate (float): dropout rate
attn_drop_rate (float): attention dropout rate
drop_path_rate (float): stochastic depth rate
embed_layer (nn.Module): patch embedding layer
norm_layer: (nn.Module): normalization layer
weight_init: (str): weight init scheme
“”"
# super().init()
super().init()
if isinstance(img_size, tuple):
self.img_size = img_size
else:
self.img_size = to_2tuple(img_size)
self.patch_size = patch_size
self.in_chans = in_chans

self.num_classes = num_classes
self.num_features = self.embed_dim = embed_dim # num_features for consistency with other models
self.num_tokens = 2 if distilled else 1
norm_layer = norm_layer or partial(nn.LayerNorm, eps=1e-6)
act_layer = act_layer or nn.GELU

self.patch_embed = embed_layer(
img_size=img_size, patch_size=patch_size, in_chans=in_chans, embed_dim=embed_dim)
num_patches = self.patch_embed.num_patches

self.cls_token = nn.Parameter(torch.zeros(1, 1, embed_dim))
self.dist_token = nn.Parameter(torch.zeros(1, 1, embed_dim)) if distilled else None
self.pos_embed = nn.Parameter(torch.zeros(1, num_patches + self.num_tokens, embed_dim))
self.pos_drop = nn.Dropout(p=drop_rate)
self.pos_embed_event = PatchEmbed_event(in_chans=32, embed_dim=768, kernel_size=4, stride=4)
# self.pos_embed_event = PatchEmbed_event(in_chans=32, embed_dim=768, kernel_size=4, stride=4)
# self.pos_embed_event_z = PatchEmbed_event(in_chans=32, embed_dim=768, kernel_size=3, stride=1)
# attn = CrossAttn(768, 4, 3072, 0.1, ‘relu’)
# self.cross_attn = Iter_attn(attn, 2)

dpr = [x.item() for x in torch.linspace(0, drop_path_rate, depth)] # stochastic depth decay rule
blocks = []
ce_index = 0
self.ce_loc = ce_loc
for i in range(depth):
ce_keep_ratio_i = 1.0
if ce_loc is not None and i in ce_loc:
ce_keep_ratio_i = ce_keep_ratio[ce_index]
ce_index += 1

blocks.append(
CEBlock(
dim=embed_dim, num_heads=num_heads, mlp_ratio=mlp_ratio, qkv_bias=qkv_bias, drop=drop_rate,
attn_drop=attn_drop_rate, drop_path=dpr[i], norm_layer=norm_layer, act_layer=act_layer,
keep_ratio_search=ce_keep_ratio_i)
)

self.blocks = nn.Sequential(blocks)
self.norm = norm_layer(embed_dim)

self.init_weights(weight_init)

def forward_features(self, z, x, event_z, event_x,
mask_z=None, mask_x=None,
ce_template_mask=None, ce_keep_rate=None,
return_last_attn=False
):
B, H, W = x.shape[0], x.shape[2], x.shape[3]

event_z = self.pos_embed_event(event_z) # [:,:,:,:1000]
event_x = self.pos_embed_event(event_x) # B 768 1024
x = self.patch_embed(x)
z = self.patch_embed(z)

event_z += self.pos_embed_z
event_x += self.pos_embed_x
z += self.pos_embed_z
x += self.pos_embed_x

# attention mask handling # B, H, W
if mask_z is not None and mask_x is not None:
mask_z = F.interpolate(mask_z[None].float(), scale_factor=1. / self.patch_size).to(torch.bool)[0]
mask_z = mask_z.flatten(1).unsqueeze(-1)

mask_x = F.interpolate(mask_x[None].float(), scale_factor=1. / self.patch_size).to(torch.bool)[0]
mask_x = mask_x.flatten(1).unsqueeze(-1)

mask_x = combine_tokens(mask_z, mask_x, mode=self.cat_mode)
mask_x = mask_x.squeeze(-1)

if self.add_cls_token:
cls_tokens = self.cls_token.expand(B, -1, -1)
cls_tokens = cls_tokens + self.cls_pos_embed

if self.add_sep_seg:
x += self.search_segment_pos_embed
z += self.template_segment_pos_embed

x = combine_tokens(z, event_z, x, event_x, mode=self.cat_mode) # 64+64+256+256=640
# x = combine_tokens(z, x, event_z, event_x, mode=self.cat_mode) # 64+64+256+256=640
if self.add_cls_token:
x = torch.cat([cls_tokens, x], dim=1)

x = self.pos_drop(x)
lens_z = self.pos_embed_z.shape[1]
lens_x = self.pos_embed_x.shape[1]

global_index_t = torch.linspace(0, lens_z - 1, lens_z).to(x.device)
global_index_t = global_index_t.repeat(B, 1)

global_index_s = torch.linspace(0, lens_x - 1, lens_x).to(x.device)
global_index_s = global_index_s.repeat(B, 1)
removed_indexes_s = []
for i, blk in enumerate(self.blocks):
x, global_index_t, global_index_s, removed_index_s, attn = <br/> blk(x, global_index_t, global_index_s, mask_x, ce_template_mask, ce_keep_rate)

if self.ce_loc is not None and i in self.ce_loc:
removed_indexes_s.append(removed_index_s)

x = self.norm(x)
lens_x_new = global_index_s.shape[1]
lens_z_new = global_index_t.shape[1]

z = x[:, :lens_z_new2]
x = x[:, lens_z_new*2:]

if removed_indexes_s and removed_indexes_s[0] is not None:
removed_indexes_cat = torch.cat(removed_indexes_s, dim=1)

pruned_lens_x = lens_x - lens_x_new
pad_x = torch.zeros([B, pruned_lens_x, x.shape[2]], device=x.device)
x = torch.cat([x, pad_x], dim=1)
index_all = torch.cat([global_index_s, removed_indexes_cat], dim=1)
# recover original token order
C = x.shape[-1]
x = torch.zeros_like(x).scatter(dim=1, index=index_all.unsqueeze(-1).expand(B, -1, C).to(torch.int64), src=x)

x = recover_tokens(x, lens_z_new, lens_x, mode=self.cat_mode)
x = x[:, :lens_x] # RGB head
x = torch.cat([event_x, x], dim=1)
# x = x[:, lens_x//2:] # event head
# x = torch.cat([z, x], dim=1)
# re-concatenate with the template, which may be further used by other modules
# x, event_x = x[:, :lens_x//2], x[:, lens_x//2:]
# x = x[:, -lens_x//2:]
aux_dict = {
“attn”: attn,
“removed_indexes_s”: removed_indexes_s, # used for visualization
}

return x, aux_dict

def forward(self, z, x, event_z, event_x,
ce_template_mask=None, ce_keep_rate=None,
tnc_keep_rate=None,
return_last_attn=False):

x, aux_dict = self.forward_features(z, x, event_z, event_x, ce_template_mask=ce_template_mask, ce_keep_rate=ce_keep_rate,)

return x, aux_dict


def _create_vision_transformer(pretrained=False, **kwargs):
model = VisionTransformerCE(**kwargs)

if pretrained:
if ‘npz’ in pretrained:
model.load_pretrained(pretrained, prefix=‘’)
else:
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


def vit_base_patch16_224_ce(pretrained=False, **kwargs):
“”" ViT-Base model (ViT-B/16) from original paper (https://arxiv.org/abs/2010.11929).
“”“
model_kwargs = dict(
patch_size=16, embed_dim=768, depth=12, num_heads=12, **kwargs)
model = _create_vision_transformer(pretrained=pretrained, **model_kwargs)
return model


def vit_large_patch16_224_ce(pretrained=False, **kwargs):
“”” ViT-Large model (ViT-L/16) from original paper (https://arxiv.org/abs/2010.11929).
“”"
model_kwargs = dict(
patch_size=16, embed_dim=1024, depth=24, num_heads=16, **kwargs)
model = _create_vision_transformer(pretrained=pretrained, **model_kwargs)
return model中关于 lens_x_new 和 lens_z_new的定义，检查上段代码