论文速读|PAR: A New Benchmark Dataset and A Large Language Model Augmented Framework.AAAI25

    @misc{jin2024pedestrianattributerecognitionnew,
      title={Pedestrian Attribute Recognition: A New Benchmark Dataset and A Large Language Model Augmented Framework}, 
      author={Jiandong Jin and Xiao Wang and Qian Zhu and Haiyang Wang and Chenglong Li},
      year={2024},
      eprint={2408.09720},
      archivePrefix={arXiv},
      primaryClass={cs.CV},
      url={https://arxiv.org/abs/2408.09720}, 
    }
    
    
      
      
      
      
      
      
      
      
      
    

    class CBAM(nn.Module):
    def __init__(self, in_channels, output_dim=1, reduction=16, kernel_size=7, hidden_dims=None):
        super(CBAM, self).__init__()
        self.channel_attention = ChannelAttention(in_channels, reduction=reduction)
        self.spatial_attention = SpatialAttention(kernel_size=kernel_size)
        self.relu = nn.ReLU()
        self.fc = nn.Linear(in_channels, output_dim)
    
        # 初始化线性层参数
        for m in self.fc.modules():
            if isinstance(m, nn.Linear):
                nn.init.kaiming_normal_(m.weight)
                if m.bias is not None:
                    nn.init.zeros_(m.bias)
    
    def forward(self, x):
        x = x.permute(0, 2, 1)  # [B, L, D] -> [B, D, L]
        x_out = x * self.channel_attention(x)
        x_out = x_out.permute(0, 2, 1)  # [B, D, L] -> [B, L, D]
        x_out = x_out * self.spatial_attention(x_out.permute(0, 2, 1)).permute(0, 2, 1)  # [B, L, D]
        x_out = torch.mean(x_out, dim=1)  # 对 L 维度进行全局平均池化，得到 [B, D]
        x_out = self.fc(self.relu(x_out))
        return x_out
    ...
    class ChannelAttention(nn.Module):
    def __init__(self, in_channels, reduction=16):
        super(ChannelAttention, self).__init__()
        self.avg_pool = nn.AdaptiveAvgPool1d(1)
        self.max_pool = nn.AdaptiveMaxPool1d(1)
        self.fc = nn.Sequential(
            nn.Conv1d(in_channels, in_channels // reduction, 1, bias=False),
            nn.ReLU(),
            nn.Conv1d(in_channels // reduction, in_channels, 1, bias=False)
        )
        self.sigmoid = nn.Sigmoid()
    
        # 初始化卷积和线性层参数
        for m in self.modules():
            if isinstance(m, nn.Conv1d) or isinstance(m, nn.Linear):
                nn.init.kaiming_normal_(m.weight)
                if m.bias is not None:
                    nn.init.zeros_(m.bias)
    
    def forward(self, x):
        avg_out = self.fc(self.avg_pool(x))
        max_out = self.fc(self.max_pool(x))
        out = avg_out + max_out
        return self.sigmoid(out)
    
    
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
    

     <Img><ImageHere_Head></Img>
     What are wearing on their head?
    What are wearing on top? ... <Img><ImageHere Topwear></Img>
    Assistant:
    
    
      
      
      
      
    

    class CEL_Sigmoid(nn.Module):
    def __init__(self, sample_weight=None, size_average=True, attr_idx=None):
        super(CEL_Sigmoid, self).__init__()
    
        self.sample_weight = sample_weight
        self.size_average = size_average
        self.attr_idx = attr_idx
    
    def forward(self, logits, targets):
        batch_size = logits.shape[0]
    
        loss = F.binary_cross_entropy_with_logits(logits, targets, reduction='none')
    
        targets_mask = torch.where(targets.detach().cpu() > 0.5, torch.ones(1), torch.zeros(1))
        if self.sample_weight is not None:
            if self.attr_idx is not None and targets_mask.shape[1] != self.sample_weight.shape[0]:
                weight = ratio2weight(targets_mask[:, self.attr_idx], self.sample_weight)
                loss = loss[:, self.attr_idx]
            else:
                weight = ratio2weight(targets_mask, self.sample_weight)
            # import pdb;pdb.set_trace()
            loss = (loss * weight.cuda())
    
        loss = loss.sum() / batch_size if self.size_average else loss.sum()
    
        return loss
    
    def ratio2weight(targets, ratio):
    ratio = torch.from_numpy(ratio).type_as(targets)
    pos_weights = targets * (1 - ratio)
    neg_weights = (1 - targets) * ratio
    weights = torch.exp(neg_weights + pos_weights)
    
    # for RAP dataloader, targets element may be 2, with or without smooth, some element must great than 1
    weights[targets > 1] = 0.0
    
    return weights
    
    
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
    

    #for training
    def forward(self, im, simple_answer, random_sentences, gt_label=None):
        b_s=im.shape[0]
        # 获取im embed & ims_mask  from eva vit-g&Qformer
        vit_img_embeds, inputs_llama, atts_img = self.encode_img(im)
        #随机选取指令
        instruction = self.custom_insturction
        #将im embed 填入指令 更新后的im embed & imgae mask after llm_token_embedding
        img_embeds, atts_img = self.prompt_wrap(inputs_llama, atts_img, instruction)
        #处理真值
        self.llama_tokenizer.padding_side = "right"
        # text = [t + self.end_sym for t in answer]
        text = [t + self.end_sym for t in simple_answer]
        #真值token
        to_regress_tokens = self.llama_tokenizer(
            text,
            return_tensors="pt",
            padding="longest",
            truncation=True,
            max_length=self.max_txt_len,
            add_special_tokens=False
        ).to(im.device)
        
        random_maxlength=to_regress_tokens.input_ids.size(1)
        random_text = [t + self.end_sym for t in random_sentences]
        random_tokens = self.llama_tokenizer(
            random_text,
            return_tensors="pt",
            padding="max_length",
            truncation=True,
            max_length=random_maxlength,
            add_special_tokens=False
        ).to(im.device)     
        to_regress_embeds = self.embed_tokens(random_tokens.input_ids)
        
        batch_size = img_embeds.shape[0]
        bos = torch.ones([batch_size, 1],
                         dtype=to_regress_tokens.input_ids.dtype,
                         device=to_regress_tokens.input_ids.device) * self.llama_tokenizer.bos_token_id
        bos_embeds = self.embed_tokens(bos)
        atts_bos = atts_img[:, :1]
        IB,IL,ID=img_embeds.size()
        LB,LL,LD = to_regress_embeds.size()
        #回答这里的问题
        inputs_embeds, attention_mask, input_lens = \
            self.concat_emb_input_output(img_embeds, atts_img, to_regress_embeds, to_regress_tokens.attention_mask)#这里应该使用to_regress_token还是random_tokens的attention_mask
    
        inputs_embeds = torch.cat([bos_embeds, inputs_embeds], dim=1)
        attention_mask = torch.cat([atts_bos, attention_mask], dim=1)
        
        #填充真值
        part_targets = to_regress_tokens.input_ids.masked_fill(
            to_regress_tokens.input_ids == self.llama_tokenizer.pad_token_id, -100
        )
        labels = (
            torch.ones([inputs_embeds.shape[0], inputs_embeds.shape[1]],
                       dtype=torch.long).to(im.device).fill_(-100)
        )
        #input_lens : instruction长度
        for i, target in enumerate(part_targets):
            labels[i, input_lens[i] + 1:input_lens[i] + len(target) + 1] = target  # plus 1 for bos
        
        with self.maybe_autocast():
            outputs = self.llama_model(
                inputs_embeds=inputs_embeds,
                attention_mask=attention_mask,
                return_dict=True,
                output_hidden_states=True,
                labels=labels,
            )
        
        #添加BLEU CIDEr指标
        hidden_states = outputs.hidden_states[-1]
        hidden_states = self.llm_relu(self.llm_proj(hidden_states[:,-LL:]))
        llm_logits = torch.cat([self.cbam_llm[i](hidden_states) for i in range(self.attr_num)], dim=1)
        
        B,K,L,D = vit_img_embeds.size()
        
        logits_class = torch.cat([self.cbam_class[j + sum(self.group_details[:i])](vit_img_embeds[:,i]) for i in range(self.attr_group_num) for j in range(0, self.group_details[i])], dim=1)
        
        instance_vit_embeds = self.instance_relu(self.instance_proj(vit_img_embeds))
        logits_instance = torch.cat([self.cbam_instance[i](instance_vit_embeds[:,i]) for i in range(self.attr_group_num)], dim=1)
        return outputs.loss, self.bn_class(logits_class), self.bn_instance(logits_instance),  self.bn_llm(llm_logits)
    
    
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
    

    def create_eva_vit_g(img_size=224,drop_path_rate=0.4,use_checkpoint=False,precision="fp16", cross_layer_num=3, vit_model_path=''):
    model = VisionTransformer(
        img_size=img_size,
        patch_size=14,
        use_mean_pooling=False,
        embed_dim=1408,
        depth=39,
        num_heads=1408//88,
        mlp_ratio=4.3637,
        qkv_bias=True,
        drop_path_rate=drop_path_rate,
        norm_layer=partial(nn.LayerNorm, eps=1e-6),
        use_checkpoint=use_checkpoint,
    )  
    
    state_dict = torch.load(vit_model_path, map_location="cpu")    
    interpolate_pos_embed(model,state_dict)
    
    incompatible_keys = model.load_state_dict(state_dict, strict=False)
    dpr = [x.item() for x in torch.linspace(0, 0.4, 39)][-cross_layer_num:]
    cross_layers = nn.ModuleList([Block(dim=1408, num_heads=1408//88, mlp_ratio=4.3637, qkv_bias=True, qk_scale=None,
                drop=0., attn_drop=0., drop_path=0.4, norm_layer=partial(nn.LayerNorm, eps=1e-6),
                init_values=None, window_size=None) for i in range(cross_layer_num)])
    for i in range(cross_layer_num):
        model_block = model.blocks[-(cross_layer_num - i)]
        model_block_state_dict = model_block.state_dict()
        cross_layer_state_dict = cross_layers[i].state_dict()
        for name, param in model_block_state_dict.items():
            if name in cross_layer_state_dict:
                cross_layer_state_dict[name].copy_(param)
        cross_layers[i].load_state_dict(cross_layer_state_dict)
    print(apply_cross(cross_layers)[0])
    print(apply_lora(model)[0])
    mark_only_lora_as_trainable(model)
    return model, cross_layers
    ....
    class Custom_Attention(nn.Module):
    def __init__(self, existing_mha, lora_r=args.v_lora_r, lora_alpha=args.v_lora_alpha, lora_dropout=args.v_lora_dropout, enable_lora=args.v_lora_target_modules):
        super(Custom_Attention, self).__init__()
        self.num_heads = existing_mha.num_heads
        dim = existing_mha.dim 
        head_dim = existing_mha.dim // self.num_heads 
        if existing_mha.attn_head_dim is not None:
            head_dim = existing_mha.attn_head_dim
        all_head_dim = head_dim * self.num_heads
        
        self.scale = existing_mha.scale
    
        qkv = existing_mha.qkv 
        self.q_bias = existing_mha.q_bias
        self.v_bias = existing_mha.v_bias
        
        self.q_proj = nn.Linear(dim, all_head_dim, bias=existing_mha.q_bias is not None)
        self.k_proj = nn.Linear(dim, all_head_dim, bias=existing_mha.q_bias is not None)
        self.v_proj = nn.Linear(dim, all_head_dim, bias=existing_mha.q_bias is not None)
        self.proj = existing_mha.proj
        with torch.no_grad():  
            # Initialize q_proj
            self.q_proj.weight.data.copy_(qkv.weight.data[:all_head_dim, :])
            # Initialize k_proj
            self.k_proj.weight.data.copy_(qkv.weight.data[all_head_dim:2*all_head_dim, :])
            # Initialize v_proj
            self.v_proj.weight.data.copy_(qkv.weight.data[2*all_head_dim:, :])
            if self.q_bias is not None:
                self.q_proj.bias.data.copy_(self.q_bias)
                self.k_proj.bias.data.copy_(torch.zeros_like(self.v_bias))
                self.v_proj.bias.data.copy_(self.v_bias)
        LoRALayer.__init__(self, r=lora_r, lora_alpha=lora_alpha, dropout_rate=lora_dropout)
        for item in enable_lora:
            if item == 'q':
                self.q_proj = LinearLoRA(self.q_proj,
                                         r=lora_r,
                                         lora_alpha=lora_alpha,
                                         fan_in_fan_out=False,
                                         dropout_rate = lora_dropout)
            elif item == 'k':
                self.k_proj = LinearLoRA(self.k_proj,
                                         r=lora_r,
                                         lora_alpha=lora_alpha,
                                         fan_in_fan_out=False,
                                         dropout_rate = lora_dropout)
            elif item == 'v':
                self.v_proj = LinearLoRA(self.v_proj,
                                         r=lora_r,
                                         lora_alpha=lora_alpha,
                                         fan_in_fan_out=False,
                                         dropout_rate = lora_dropout)
            elif item == 'o':
                self.proj = LinearLoRA(self.proj,
                                         r=lora_r,
                                         lora_alpha=lora_alpha,
                                         fan_in_fan_out=False,
                                         dropout_rate = lora_dropout)
                
        self.window_size = existing_mha.window_size
    
        if self.window_size:
            self.num_relative_distance = existing_mha.num_relative_distance
            self.relative_position_bias_table = existing_mha.relative_position_bias_table
            self.register_buffer("relative_position_index", existing_mha.relative_position_index)
        else:
            self.window_size = None
            self.relative_position_bias_table = None
            self.relative_position_index = None
    
        self.attn_drop = existing_mha.attn_drop
        self.proj_drop = existing_mha.proj_drop
    
        
    def forward(self, x, rel_pos_bias=None,**kwargs):
        B, N, C = x.shape
        q = self.q_proj(x).reshape(B, N, self.num_heads, -1).permute(0, 2, 1, 3)
        k = self.k_proj(x).reshape(B, N, self.num_heads, -1).permute(0, 2, 1, 3)
        v = self.v_proj(x).reshape(B, N, self.num_heads, -1).permute(0, 2, 1, 3)
        
        
        q = q * self.scale
        attn = (q @ k.transpose(-2, -1))
    
        if self.relative_position_bias_table is not None:
            relative_position_bias = \
                self.relative_position_bias_table[self.relative_position_index.view(-1)].view(
                    self.window_size[0] * self.window_size[1] + 1,
                    self.window_size[0] * self.window_size[1] + 1, -1)  # Wh*Ww,Wh*Ww,nH
            relative_position_bias = relative_position_bias.permute(2, 0, 1).contiguous()  # nH, Wh*Ww, Wh*Ww
            attn = attn + relative_position_bias.unsqueeze(0)
    
        if rel_pos_bias is not None:
            attn = attn + rel_pos_bias
        
        attn = attn.softmax(dim=-1)
        attn = self.attn_drop(attn)
    
        x = (attn @ v).transpose(1, 2).reshape(B, N, -1)
        x = self.proj(x)
        x = self.proj_drop(x)
        
        return x
    ...
    def apply_lora(clip_model):
    list_lora_layers = []
    for i, block in enumerate(clip_model.blocks):
        for name, submodule in block.named_children():
            if isinstance(submodule, Attention):
                new_multi_head_lora = Custom_Attention(submodule)
                setattr(block, name, new_multi_head_lora)
                list_lora_layers.append(new_multi_head_lora)
    return list_lora_layers