多模态图像分类案例

当然可以以下是使用CIFAR-10数据集进行多模态图像识别分类的完整代码。由于CIFAR-10只有图像没有文本描述我会模拟生成对应的文本描述来实现多模态实验import torch import torch.nn as nn import torch.optim as optim from torch.utils.data import Dataset, DataLoader import torchvision import torchvision.transforms as transforms import numpy as np import matplotlib.pyplot as plt from transformers import BertTokenizer, BertModel import warnings warnings.filterwarnings(ignore) # 设置设备 device torch.device(cuda if torch.cuda.is_available() else cpu) print(fUsing device: {device}) # 1. CIFAR-10类别对应的文本描述 CIFAR10_CLASS_DESCRIPTIONS { 0: [This is an airplane flying in the sky, A commercial aircraft with wings and engines, Airplane soaring through clouds], 1: [A modern automobile on the road, Car driving on highway, Vehicle with four wheels and doors], 2: [A beautiful bird perched on a branch, Small bird with colorful feathers, Bird flying in the air], 3: [A cute cat sitting on furniture, Domestic cat with soft fur, Cat playing with a toy], 4: [A graceful deer standing in forest, Wild deer with antlers, Deer grazing in meadow], 5: [A friendly dog running in park, Dog wagging its tail happily, Pet dog playing fetch], 6: [A green frog sitting on lily pad, Small amphibian frog jumping, Frog near pond water], 7: [A strong horse galloping in field, Horse running freely on grassland, Beautiful horse with mane], 8: [A large ship sailing on ocean, Cargo ship crossing sea, Boat floating on water], 9: [A heavy truck transporting goods, Large truck on highway, Delivery truck carrying cargo] } # 2. 多模态CIFAR-10数据集 class MultimodalCIFAR10(Dataset): def __init__(self, cifar_dataset, tokenizerNone, max_length64, is_trainTrue): self.cifar_dataset cifar_dataset self.tokenizer tokenizer self.max_length max_length self.is_train is_train def __len__(self): return len(self.cifar_dataset) def __getitem__(self, idx): image, label self.cifar_dataset[idx] # 根据类别随机选择一个文本描述 descriptions CIFAR10_CLASS_DESCRIPTIONS[label] text np.random.choice(descriptions) # 对文本进行编码 if self.tokenizer: encoding self.tokenizer( text, truncationTrue, paddingmax_length, max_lengthself.max_length, return_tensorspt ) input_ids encoding[input_ids].squeeze(0) attention_mask encoding[attention_mask].squeeze(0) else: input_ids None attention_mask None return { image: image, text_input_ids: input_ids, text_attention_mask: attention_mask, label: torch.tensor(label, dtypetorch.long), text: text # 保留原始文本用于展示 } # 3. 图像编码器轻量级版本 class LightweightImageEncoder(nn.Module): def __init__(self, embed_dim256): super(LightweightImageEncoder, self).__init__() # 使用简单的CNN架构 self.features nn.Sequential( # Conv Block 1 nn.Conv2d(3, 64, 3, padding1), nn.BatchNorm2d(64), nn.ReLU(), nn.Conv2d(64, 64, 3, padding1), nn.BatchNorm2d(64), nn.ReLU(), nn.MaxPool2d(2, 2), # 16x16 # Conv Block 2 nn.Conv2d(64, 128, 3, padding1), nn.BatchNorm2d(128), nn.ReLU(), nn.Conv2d(128, 128, 3, padding1), nn.BatchNorm2d(128), nn.ReLU(), nn.MaxPool2d(2, 2), # 8x8 # Conv Block 3 nn.Conv2d(128, 256, 3, padding1), nn.BatchNorm2d(256), nn.ReLU(), nn.Conv2d(256, 256, 3, padding1), nn.BatchNorm2d(256), nn.ReLU(), nn.MaxPool2d(2, 2), # 4x4 ) self.global_pool nn.AdaptiveAvgPool2d((1, 1)) self.projection nn.Sequential( nn.Linear(256, embed_dim), nn.BatchNorm1d(embed_dim), nn.ReLU(), nn.Dropout(0.2) ) def forward(self, x): features self.features(x) features self.global_pool(features) features features.view(features.size(0), -1) embeddings self.projection(features) return embeddings # 4. 文本编码器轻量级版本 class LightweightTextEncoder(nn.Module): def __init__(self, vocab_size30522, embed_dim256, max_length64): super(LightweightTextEncoder, self).__init__() self.embedding nn.Embedding(vocab_size, 128) self.position_encoding nn.Parameter(torch.randn(1, max_length, 128)) # 简单的Transformer编码器 encoder_layer nn.TransformerEncoderLayer( d_model128, nhead4, dim_feedforward512, dropout0.1, batch_firstTrue ) self.transformer_encoder nn.TransformerEncoder(encoder_layer, num_layers2) self.projection nn.Sequential( nn.Linear(128, embed_dim), nn.BatchNorm1d(embed_dim), nn.ReLU(), nn.Dropout(0.2) ) def forward(self, input_ids, attention_mask): # Token embedding Position encoding embeddings self.embedding(input_ids) seq_len input_ids.size(1) embeddings embeddings self.position_encoding[:, :seq_len, :] # Transformer编码 encoded self.transformer_encoder(embeddings) # 使用第一个token[CLS]的输出 cls_output encoded[:, 0, :] # 投影到embedding空间 embeddings self.projection(cls_output) return embeddings # 5. 跨模态注意力融合 class CrossModalFusion(nn.Module): def __init__(self, embed_dim256, num_heads4): super(CrossModalFusion, self).__init__() self.cross_attention nn.MultiheadAttention( embed_dim, num_heads, dropout0.1, batch_firstTrue ) self.self_attention nn.MultiheadAttention( embed_dim, num_heads, dropout0.1, batch_firstTrue ) self.norm1 nn.LayerNorm(embed_dim) self.norm2 nn.LayerNorm(embed_dim) self.norm3 nn.LayerNorm(embed_dim) self.ffn nn.Sequential( nn.Linear(embed_dim, embed_dim * 2), nn.GELU(), nn.Dropout(0.1), nn.Linear(embed_dim * 2, embed_dim), nn.Dropout(0.1) ) def forward(self, image_features, text_features): # 增加序列维度 image_seq image_features.unsqueeze(1) # [batch, 1, embed_dim] text_seq text_features.unsqueeze(1) # [batch, 1, embed_dim] # 拼接形成序列 combined torch.cat([image_seq, text_seq], dim1) # [batch, 2, embed_dim] # 自注意力 attn_output, _ self.self_attention(combined, combined, combined) combined self.norm1(combined attn_output) # 跨模态注意力 cross_attn_output, _ self.cross_attention( querycombined, keycombined, valuecombined ) combined self.norm2(combined cross_attn_output) # FFN ffn_output self.ffn(combined) fused self.norm3(combined ffn_output) # 全局池化得到最终特征 fused fused.mean(dim1) # [batch, embed_dim] return fused # 6. 多模态分类模型CIFAR-10版本 class MultimodalCIFAR10Classifier(nn.Module): def __init__(self, num_classes10, embed_dim256, vocab_size30522): super(MultimodalCIFAR10Classifier, self).__init__() self.image_encoder LightweightImageEncoder(embed_dimembed_dim) self.text_encoder LightweightTextEncoder( vocab_sizevocab_size, embed_dimembed_dim ) self.fusion CrossModalFusion(embed_dimembed_dim) # 自适应门控机制 self.gate_network nn.Sequential( nn.Linear(embed_dim * 2, embed_dim), nn.Sigmoid() ) # 分类器 self.classifier nn.Sequential( nn.Linear(embed_dim, 128), nn.BatchNorm1d(128), nn.ReLU(), nn.Dropout(0.3), nn.Linear(128, 64), nn.BatchNorm1d(64), nn.ReLU(), nn.Dropout(0.2), nn.Linear(64, num_classes) ) def forward(self, images, text_input_ids, text_attention_mask): # 编码 image_features self.image_encoder(images) text_features self.text_encoder(text_input_ids, text_attention_mask) # 融合 fused_features self.fusion(image_features, text_features) # 门控机制 combined torch.cat([image_features, text_features], dim-1) gate self.gate_network(combined) final_features gate * fused_features (1 - gate) * image_features # 分类 output self.classifier(final_features) return output # 7. 训练函数 def train_epoch(model, dataloader, criterion, optimizer, device): model.train() running_loss 0.0 correct 0 total 0 for batch_idx, batch in enumerate(dataloader): images batch[image].to(device) text_input_ids batch[text_input_ids].to(device) text_attention_mask batch[text_attention_mask].to(device) labels batch[label].to(device) optimizer.zero_grad() outputs model(images, text_input_ids, text_attention_mask) loss criterion(outputs, labels) loss.backward() optimizer.step() running_loss loss.item() _, predicted outputs.max(1) total labels.size(0) correct predicted.eq(labels).sum().item() if batch_idx % 50 49: print(fBatch [{batch_idx1}/{len(dataloader)}], fLoss: {running_loss/50:.4f}, fAcc: {100.*correct/total:.2f}%) running_loss 0.0 return correct / total # 8. 验证函数 def validate(model, dataloader, criterion, device): model.eval() test_loss 0.0 correct 0 total 0 all_preds [] all_labels [] with torch.no_grad(): for batch in dataloader: images batch[image].to(device) text_input_ids batch[text_input_ids].to(device) text_attention_mask batch[text_attention_mask].to(device) labels batch[label].to(device) outputs model(images, text_input_ids, text_attention_mask) loss criterion(outputs, labels) test_loss loss.item() _, predicted outputs.max(1) total labels.size(0) correct predicted.eq(labels).sum().item() all_preds.extend(predicted.cpu().numpy()) all_labels.extend(labels.cpu().numpy()) accuracy 100. * correct / total print(fValidation Loss: {test_loss/len(dataloader):.4f}, fAccuracy: {accuracy:.2f}%) return accuracy, all_preds, all_labels # 9. 单模态基线模型 class SingleModalBaseline(nn.Module): def __init__(self, modal_typeimage, num_classes10, embed_dim256): super(SingleModalBaseline, self).__init__() self.modal_type modal_type if modal_type image: self.encoder LightweightImageEncoder(embed_dimembed_dim) elif modal_type text: self.encoder LightweightTextEncoder(embed_dimembed_dim) self.classifier nn.Sequential( nn.Linear(embed_dim, 128), nn.BatchNorm1d(128), nn.ReLU(), nn.Dropout(0.3), nn.Linear(128, 64), nn.BatchNorm1d(64), nn.ReLU(), nn.Dropout(0.2), nn.Linear(64, num_classes) ) def forward(self, imagesNone, text_input_idsNone, text_attention_maskNone): if self.modal_type image: features self.encoder(images) else: features self.encoder(text_input_ids, text_attention_mask) output self.classifier(features) return output # 10. 可视化结果 def visualize_results(model, testloader, classes, device, num_images10): model.eval() # 获取一批测试数据 dataiter iter(testloader) batch next(dataiter) images batch[image][:num_images] texts batch[text][:num_images] labels batch[label][:num_images] # 预测 with torch.no_grad(): outputs model( images.to(device), batch[text_input_ids][:num_images].to(device), batch[text_attention_mask][:num_images].to(device) ) _, predicted torch.max(outputs, 1) # 反归一化 mean torch.tensor([0.4914, 0.4822, 0.4465]).view(3, 1, 1) std torch.tensor([0.2023, 0.1994, 0.2010]).view(3, 1, 1) # 显示图像 fig, axes plt.subplots(2, 5, figsize(20, 8)) axes axes.ravel() for idx in range(min(num_images, len(images))): img images[idx] * std mean img torch.clamp(img, 0, 1) img img.permute(1, 2, 0).numpy() axes[idx].imshow(img) title_color green if predicted[idx] labels[idx] else red axes[idx].set_title( fTrue: {classes[labels[idx]]}\n fPred: {classes[predicted[idx]]}\n fText: {texts[idx][:30]}..., colortitle_color, fontsize10 ) axes[idx].axis(off) plt.tight_layout() plt.show() # 11. 混淆矩阵 def plot_confusion_matrix(all_labels, all_preds, classes): from sklearn.metrics import confusion_matrix import seaborn as sns cm confusion_matrix(all_labels, all_preds) plt.figure(figsize(10, 8)) sns.heatmap(cm, annotTrue, fmtd, cmapBlues, xticklabelsclasses, yticklabelsclasses) plt.xlabel(Predicted Label) plt.ylabel(True Label) plt.title(Confusion Matrix) plt.tight_layout() plt.show() # 12. 主程序 def main(): # 超参数 batch_size 64 num_epochs 25 learning_rate 1e-3 embed_dim 256 print(*60) print(CIFAR-10 多模态图像识别分类实验) print(*60) # CIFAR-10类别名称 classes (plane, car, bird, cat, deer, dog, frog, horse, ship, truck) # 数据预处理 transform_train transforms.Compose([ transforms.RandomCrop(32, padding4), transforms.RandomHorizontalFlip(), transforms.ColorJitter(brightness0.2, contrast0.2), transforms.ToTensor(), transforms.Normalize((0.4914, 0.4822, 0.4465), (0.2023, 0.1994, 0.2010)) ]) transform_test transforms.Compose([ transforms.ToTensor(), transforms.Normalize((0.4914, 0.4822, 0.4465), (0.2023, 0.1994, 0.2010)) ]) # 加载CIFAR-10数据集 print(\n加载CIFAR-10数据集...) trainset torchvision.datasets.CIFAR10( root./data, trainTrue, downloadTrue, transformtransform_train ) testset torchvision.datasets.CIFAR10( root./data, trainFalse, downloadTrue, transformtransform_test ) # 初始化tokenizer tokenizer BertTokenizer.from_pretrained(bert-base-uncased) # 创建多模态数据集 print(创建多模态数据集...) multimodal_trainset MultimodalCIFAR10(trainset, tokenizer, is_trainTrue) multimodal_testset MultimodalCIFAR10(testset, tokenizer, is_trainFalse) train_loader DataLoader( multimodal_trainset, batch_sizebatch_size, shuffleTrue, num_workers2 ) test_loader DataLoader( multimodal_testset, batch_sizebatch_size, shuffleFalse, num_workers2 ) # 创建多模态模型 print(\n创建多模态分类模型...) multimodal_model MultimodalCIFAR10Classifier( num_classes10, embed_dimembed_dim ).to(device) # 计算模型参数量 total_params sum(p.numel() for p in multimodal_model.parameters()) trainable_params sum(p.numel() for p in multimodal_model.parameters() if p.requires_grad) print(f总参数量: {total_params:,}) print(f可训练参数量: {trainable_params:,}) # 损失函数和优化器 criterion nn.CrossEntropyLoss() optimizer optim.AdamW(multimodal_model.parameters(), lrlearning_rate, weight_decay5e-4) scheduler optim.lr_scheduler.CosineAnnealingLR(optimizer, T_maxnum_epochs) # 训练多模态模型 print(\n开始训练多模态模型...) best_acc 0.0 train_acc_history [] test_acc_history [] for epoch in range(1, num_epochs 1): print(f\nEpoch {epoch}/{num_epochs}) print(-*40) train_acc train_epoch(multimodal_model, train_loader, criterion, optimizer, device) test_acc, _, _ validate(multimodal_model, test_loader, criterion, device) train_acc_history.append(train_acc) test_acc_history.append(test_acc) scheduler.step() if test_acc best_acc: best_acc test_acc torch.save(multimodal_model.state_dict(), best_multimodal_cifar10.pth) print(f保存最佳模型准确率: {best_acc:.2f}%) print(f\n多模态模型最佳准确率: {best_acc:.2f}%) # 消融实验 print(\n *60) print(消融实验对比不同模态的性能) print(*60) ablation_results {} for modal_type in [image, text]: print(f\n训练{modal_type}单模态模型...) single_model SingleModalBaseline( modal_typemodal_type, num_classes10, embed_dimembed_dim ).to(device) single_optimizer optim.AdamW(single_model.parameters(), lrlearning_rate) best_single_acc 0.0 for epoch in range(15): # 简化的训练 single_model.train() for batch_idx, batch in enumerate(train_loader): single_optimizer.zero_grad() if modal_type image: outputs single_model(imagesbatch[image].to(device)) else: outputs single_model( text_input_idsbatch[text_input_ids].to(device), text_attention_maskbatch[text_attention_mask].to(device) ) loss criterion(outputs, batch[label].to(device)) loss.backward() single_optimizer.step() # 评估 single_model.eval() correct 0 total 0 with torch.no_grad(): for batch in test_loader: if modal_type image: outputs single_model(imagesbatch[image].to(device)) else: outputs single_model( text_input_idsbatch[text_input_ids].to(device), text_attention_maskbatch[text_attention_mask].to(device) ) _, predicted outputs.max(1) total batch[label].size(0) correct predicted.eq(batch[label].to(device)).sum().item() accuracy 100. * correct / total ablation_results[modal_type] accuracy print(f{modal_type}模型准确率: {accuracy:.2f}%) # 最终评估和可视化 print(\n *60) print(最终评估) print(*60) # 加载最佳模型 multimodal_model.load_state_dict(torch.load(best_multimodal_cifar10.pth)) final_acc, all_preds, all_labels validate(multimodal_model, test_loader, criterion, device) # 可视化结果 print(\n可视化预测结果...) visualize_results(multimodal_model, test_loader, classes, device) # 绘制混淆矩阵 print(\n绘制混淆矩阵...) plot_confusion_matrix(all_labels, all_preds, classes) # 结果汇总 print(\n *60) print(实验结果汇总) print(*60) print(f多模态模型准确率: {final_acc:.2f}%) print(f仅图像模型准确率: {ablation_results.get(image, 0):.2f}%) print(f仅文本模型准确率: {ablation_results.get(text, 0):.2f}%) # 绘制训练曲线 plt.figure(figsize(10, 5)) plt.plot(train_acc_history, labelTrain Accuracy, linewidth2) plt.plot(test_acc_history, labelTest Accuracy, linewidth2) plt.xlabel(Epoch) plt.ylabel(Accuracy (%)) plt.title(Multimodal Model Training Progress on CIFAR-10) plt.legend() plt.grid(True) plt.tight_layout() plt.show() print(\n实验完成) if __name__ __main__: main()代码特点1.CIFAR-10适配为每个CIFAR-10类别生成多个文本描述使用轻量级网络适应32x32的小图像尺寸2.多模态设计图像编码器轻量级CNN适合小尺寸图像文本编码器简单Transformer处理文本描述跨模态融合自注意力交叉注意力机制3.实验功能完整的训练和验证流程消融实验对比单模态vs多模态混淆矩阵分析预测结果可视化4.性能优化模型轻量化设计学习率余弦退火权重衰减正则化安装依赖pip install torch torchvision transformers matplotlib seaborn scikit-learn预期效果多模态模型通常比单模态模型高3-8%图像模态贡献主要性能文本模态提供辅助信息这个实验展示了如何利用CIFAR-10进行多模态学习通过添加文本描述来提升分类性能。