传知代码-CENet及多模态情感计算实战（论文复现）

代码以及视频讲解

本文所涉及所有资源均在传知代码平台可获取

一、概述

本文对 “Cross-Modal Enhancement Network for Multimodal Sentiment Analysis” 论文进行讲解和手把手复现教学，解决当下热门的多模态情感计算问题，并展示在MOSI和MOSEI两个数据集上的效果

二、论文地址

DOI: 10.1109/TMM.2022.3183830

三、研究背景

情感分析在人工智能向情感智能发展中起着重要作用。早期的情感分析研究主要集中在分析单模态数据上，包括文本情感分析、图像情感分析、音频情感分析等。然而，人类的情感是通过人体的多种感官来传达的。因此，单模态情感分析忽略了人类情感的多维性。相比之下，多模态情感分析通过结合文本、视觉和音频等多模态数据来推断一个人的情感状态。与单模态情感分析相比，多模态数据包含多样化的情感信息，具有更高的预测精度。目前，多模态情感分析已被广泛应用于视频理解、人机交互、政治活动等领域。近年来，随着互联网和各种多媒体平台的快速发展，通过互联网表达情感的载体和方式也变得越来越多样化。这导致了多媒体数据的快速增长，为多模态情感分析提供了大量的数据源。下图展示了多模态在情感计算任务中的优势。
在这里插入图片描述

四、主要贡献

提出了一种跨模态增强网络，通过融入长范围非文本情感语境来增强预训练语言模型中的文本表示;
提出一种特征转换策略，通过减小文本模态和非文本模态的初始表示之间的分布差异，促进了不同模态的融合；
融合了新的预训练语言模型SentiLARE来提高模型对情感词的提取效率，从而提升对情感计算的准确性。

五、论文思路

作者提出的跨模态增强网络(CENet)模型通过将视觉和声学信息集成到语言模型中来增强文本表示。在基于transformer的预训练语言模型中嵌入跨模态增强(CE)模块，根据非对齐非文本数据中隐含的长程情感线索增强每个单词的表示。此外，针对声学模态和视觉模态，提出了一种特征转换策略，以减少语言模态和非语言模态的初始表示之间的分布差异，从而促进不同模态的融合。

六、主要内容和网络架构

首先我们展示一下CENet的总体网络架构
在这里插入图片描述

通过该图，我们可以看出该模型主要有以下几部分组成：1.非文本模态特征转化；2.跨模态增强；3.预训练语言模型输出；接下来将对他们分别进行讲解：
1. 非文本模态转换
针对预训练语言模型，初始文本表示是基于词汇表的单词索引序列，而视觉和听觉的表示则是实值向量序列。为了缩小这些异质模态之间的初始分布差异，进而减少在融合过程中非文本特征和文本特征之间的分布差距，本文提出了一种将非文本向量转换为索引的特征转换策略。这种策略有助于促进文本表征与非文本情感语境的有机融合。

具体而言，特征转换策略利用无监督聚类算法分别构建了“声学词汇表”和“视觉词汇表”。通过查询这些非语言词汇表，可以将原始的非语言特征序列转换为索引序列。下图展示了特征转换过程的具体步骤。考虑到k-means方法具有计算复杂度低和实现简单等优点，作者选择使用k-means算法来学习非语言模态的词汇。
在这里插入图片描述

2. 跨模态增强模块
本文提出的CE模块旨在将长程视觉和声学信息集成到预训练语言模型中，以增强文本的表示能力。CE模块的核心组件是跨模态嵌入单元，其结构如下图所示。该单元利用跨模态注意力机制捕捉长程非文本情感信息，并生成基于文本的非语言嵌入。嵌入层的参数可学习，用于将经过特征转换策略处理后得到的非文本索引向量映射到高维空间，然后生成文本模态对非文本模态的注意力权重矩阵。
在这里插入图片描述

在初始训练阶段，由于语言表征和非语言表征处于不同的特征空间，它们之间的相关性通常较低。因此，注意力权重矩阵中的元素可能较小。为了更有效地学习模型参数，研究者在应用softmax之前使用超参数来缩放这些注意力权重矩阵。

基于注意力权重矩阵，可以生成基于文本的非语言向量。将基于文本的声学嵌入和基于文本的视觉嵌入结合起来，形成非语言增强嵌入。最后，通过整合非语言增强嵌入来更新文本的表示。因此，CE模块的提出旨在为文本提供非语言上下文信息，通过增加非语言增强嵌入来调整文本表示，从而使其在语义上更加准确和丰富。

3. 预训练语言模型输出
作者采用SentiLARE作为语言模型，其利用包括词性和单词情感极性在内的语言知识来学习情感感知的语言表示。CE模块被集成到预训练语言模型的第i层中。值得注意的是，任何基于Transformer的预训练语言模型都可以与CE模块集成。下面是作者根据SentiLARE的设置进行的步骤：

给定一个单词序列，首先通过Stanford Log-Linear词性(POS)标记器学习其词性序列，并通过SentiwordNet学习单词级情感极性序列。
然后，使用预训练语言模型的分词器获取词标索引序列。这个序列作为输入，产生一个初步的增强语言知识表示。
更新后的文本表示将作为第(i+1)层的输入，并通过SentiLARE中的剩余层进行处理。
每一层的输出将是具有视觉和听觉信息的文本主导的高级情感表示。
最后，将这些文本表示输入到分类头中，以获取情感强度。

因此，CE模块通过将非语言增强嵌入集成到预训练语言模型中，有助于生成更富有情感感知的语言表示。这种方法能够在文本表示中有效地整合视觉和听觉信息，从而提升情感分析等任务的性能。

七、数据集介绍

1. CMU-MOSI: 它是一个多模态数据集，包括文本、视觉和声学模态。它来自Youtube上的93个电影评论视频。这些视频被剪辑成2199个片段。每个片段都标注了[-3,3]范围内的情感强度。该数据集分为三个部分，训练集(1,284段)、验证集(229段)和测试集(686段)。
2. CMU-MOSEI: 它类似于CMU-MOSI，但规模更大。它包含了来自在线视频网站的23,453个注释视频片段，涵盖了250个不同的主题和1000个不同的演讲者。CMU-MOSEI中的样本被标记为[-3,3]范围内的情感强度和6种基本情绪。因此，CMU-MOSEI可用于情感分析和情感识别任务。

八、性能对比

有下图可以观察到，该论文提出的CENet与其他SOTA模型对比性能有明显提升：
在这里插入图片描述

九、复现过程（重要）

1. 数据集准备
下载MOSI和MOSEI数据集已提取好的特征文件(.pkl)。把它放在"./dataset”目录。

2. 下载预训练语言模型
下载SentiLARE语言模型文件，然后将它们放入"/pretrained-model / sentilare_model”目录。

3. 下载需要的包

pip install -r requirements.txt

4. 搭建CENet模块
利用pytorch框架对CENet模块进行搭建：

class CE(nn.Module):def __init__(self, beta_shift_a=0.5, beta_shift_v=0.5, dropout_prob=0.2):super(CE, self).__init__()self.visual_embedding = nn.Embedding(label_size + 1, TEXT_DIM, padding_idx=label_size)self.acoustic_embedding = nn.Embedding(label_size + 1, TEXT_DIM, padding_idx=label_size)self.hv = SelfAttention(TEXT_DIM)self.ha = SelfAttention(TEXT_DIM)self.cat_connect = nn.Linear(2 * TEXT_DIM, TEXT_DIM)def forward(self, text_embedding, visual=None, acoustic=None, visual_ids=None, acoustic_ids=None):visual_ = self.visual_embedding(visual_ids)acoustic_ = self.acoustic_embedding(acoustic_ids)visual_ = self.hv(text_embedding, visual_)acoustic_ = self.ha(text_embedding, acoustic_) visual_acoustic = torch.cat((visual_, acoustic_), dim=-1)shift = self.cat_connect(visual_acoustic)embedding_shift = shift + text_embeddingreturn embedding_shift

5. 将CE模块与预训练语言模型融合

class BertEncoder(nn.Module):def __init__(self, config):super(BertEncoder, self).__init__()self.output_attentions = config.output_attentionsself.output_hidden_states = config.output_hidden_statesself.layer = nn.ModuleList([BertLayer(config) for _ in range(config.num_hidden_layers)])self.CE = CE()def forward(self, hidden_states, visual=None, acoustic=None, visual_ids=None, acoustic_ids=None, attention_mask=None, head_mask=None):all_hidden_states = ()all_attentions = ()for i, layer_module in enumerate(self.layer):if self.output_hidden_states:all_hidden_states = all_hidden_states + (hidden_states,)if i == ROBERTA_INJECTION_INDEX:hidden_states = self.CE(hidden_states, visual=visual, acoustic=acoustic, visual_ids=visual_ids, acoustic_ids=acoustic_ids)

6. 训练代码编写

定义一个整体的训练过程 train()函数，它负责训练模型多个 epoch，并在每个 epoch 结束后评估模型在验证集和测试集上的性能，并记录相关的指标和损失；并在训练最后一轮输出所有测试集id，true label 和 predicted label。

def train(args,model,train_dataloader,validation_dataloader,test_data_loader,optimizer,scheduler,
):valid_losses = []test_accuracies = []for epoch_i in range(int(args.n_epochs)):train_loss, train_pre, train_label = train_epoch(args, model, train_dataloader, optimizer, scheduler)valid_loss, valid_pre, valid_label = evaluate_epoch(args, model, validation_dataloader)test_loss, test_pre, test_label = evaluate_epoch(args, model, test_data_loader)train_acc, train_mae, train_corr, train_f_score = score_model(train_pre, train_label)test_acc, test_mae, test_corr, test_f_score = score_model(test_pre, test_label)non0_test_acc, _, _, non0_test_f_score = score_model(test_pre, test_label, use_zero=True)valid_acc, valid_mae, valid_corr, valid_f_score = score_model(valid_pre, valid_label)print("epoch:{}, train_loss:{}, train_acc:{}, valid_loss:{}, valid_acc:{}, test_loss:{}, test_acc:{}".format(epoch_i, train_loss, train_acc, valid_loss, valid_acc, test_loss, test_acc))valid_losses.append(valid_loss)test_accuracies.append(test_acc)wandb.log(({"train_loss": train_loss,"valid_loss": valid_loss,"train_acc": train_acc,"train_corr": train_corr,"valid_acc": valid_acc,"valid_corr": valid_corr,"test_loss": test_loss,"test_acc": test_acc,"test_mae": test_mae,"test_corr": test_corr,"test_f_score": test_f_score,"non0_test_acc": non0_test_acc,"non0_test_f_score": non0_test_f_score,"best_valid_loss": min(valid_losses),"best_test_acc": max(test_accuracies),}))# 输出测试集的 id、真实标签和预测标签with torch.no_grad():for step, batch in enumerate(test_data_loader):batch = tuple(t.to(DEVICE) for t in batch)input_ids, visual_ids, acoustic_ids, pos_ids, senti_ids, polarity_ids, visual, acoustic, input_mask, segment_ids, label_ids = batchvisual = torch.squeeze(visual, 1)outputs = model(input_ids,visual,acoustic,visual_ids,acoustic_ids,pos_ids, senti_ids, polarity_ids,token_type_ids=segment_ids,attention_mask=input_mask,labels=None,)logits = outputs[0]logits = logits.detach().cpu().numpy()label_ids = label_ids.detach().cpu().numpy()logits = np.squeeze(logits).tolist()label_ids = np.squeeze(label_ids).tolist()# 假设您从 label_ids 中获取 idsids = [f"sample_{idx}" for idx in range(len(label_ids))]  # 这里是示例，您可以根据实际情况生成合适的 ids# 输出所有测试样本的 id、真实标签和预测标签值for i in range(len(ids)):print(f"id: {ids[i]}, true label: {label_ids[i]}, predicted label: {logits[i]}")