复盘：基于attention机制的多任务多模态情绪情感识别（pytorch实战），基于BERT实现文本情感分类
提示：系列被面试官问的问题，我自己当时不会，所以下来自己复盘一下，认真学习和总结，以应对未来更多的可能性
关于互联网大厂的笔试面试，都是需要细心准备的
（1）自己的科研经历， 科研内容，学习的相关领域知识，要熟悉熟透了
（2）自己的实习经历，做了 什么内容，学习的领域知识，要熟悉熟透了
（3）除了科研，实习之外，平时自己关注的 前沿知识，也不要落下，仔细了解，面试官很在乎你是否喜欢追进新科技，跟进创新概念和技术
（4）准备 数据结构与算法，有笔试的大厂，第一关就是手撕代码做算法题
面试中，实际上，你准备数据结构与算法时以备不时之需，有足够的信心面对面试官可能问的算法题，很多情况下你的科研经历和实习经历足够跟面试官聊了，就不需要考你算法了。但很多大厂就会面试问你算法题，因此不论 为了笔试面试，数据结构与算法必须熟悉熟透了
秋招提前批好多大厂不考笔试，直接面试，能否免笔试去面试，那就看你 简历实力有多强了。
文章目录
- 复盘：基于attention机制的多任务多模态情绪情感识别（pytorch实战），基于BERT实现文本情感分类
* - @[TOC](文章目录)
- 你是怎么做多模态情感检测的？
- 今天单说bert做情感分类咋搞？
- 数据集准备
- 数据预处理
- 数据处理成dataSet
- 模型的搭建
- 模型的训练
- 总结
你是怎么做多模态情感检测的？
多任务多模态整合模型
输入多模态，我用了三种模态的输入，模型是多任务模型，三个专家各自负责自己的事情
（1）一个是图像：也即人脸表情识别，专家1用常规resne或者inceptionv3提取图像表情特征，负责训练人脸情感分类模型
（2）一个是语音：直接识别用户的语气，专家2用常规的GRU模型负责提取语音MFCC特征，负责训练语音情感分类模型
（3）同时还要将语音转化为文本，用bert提取文本特征，专家3负责训练一个文本情感分类模型
（4）上面三者特征最后要做特征对其，用attention机制搞定三者的融合，最后输出一个综合多模态情绪情感分类标签
整体实现一个多任务的多模态融合模型，名字可取为：基于attention的多任务多模态实情绪情感识别
用来训练一个可以准确识别人类的情感，比单一模态情感分类要好多了
最后的效果就是这样
; 今天单说bert做情感分类咋搞？
当前，在BERT等预训练模型的基础上进行微调已经成了NLP任务的一个定式了。
为了了解BERT怎么用，在这次实践中，我实现了一个最简单的NLP任务，即文本情感分类。
基本思路如下：
所谓情感分类就是指判断 句子是积极情感还是消极情感，
例如说”今天这顿饭太美味了”是积极的情感，
“今天这顿饭简直吃不下去”是消极的情感。
基于BERT完成情感分类的基本思路如图所示。
我们知道BERT是一个预训练模型，我们把句子扔给它的时候，它对应每个字都会输出一个向量。
但是在把句子扔给BERT之前，我们会在句子最前面增加一个特殊符号[CLS]。
对应这个[CLS]，BERT也会输出一个向量，我们就是利用这个向量来进行情感分类。
为什么可以直接利用这个向量呢？这是因为BERT内部采用的是自注意力机制，
自注意力机制的特点是考虑全局又聚焦重点，
实际上[CLS]对应的向量已经嵌入了整个句子的信息，而且重点词字嵌入的信息权重要大。
所以，我们将这个向量扔给一个全连接层，就可以完成分类任务了。
由于BERT已经是一个预训练模型了， 我们在做情感分类时可以将BERT的参数固定住，不再调整，而只是调整全连接层的参数，
我在这次实践中就是这么做的。
当然也可以同时调整BERT和全连接层的参数，但是BERT模型较大，消耗的时间会长一些。
数据集准备
我在网上下了一个数据集（点击可下载，提取码为zfh3），csv格式的，
包含两列，一列是句子，一列是标签，如下图所示。
数据集应是来自大众点评。。。
数据里面标签为0的时候表示是消极的情感，
标签为1时表示的是积极的情感。
这个数据集总共有11987行，为了简单表示，可以CPU训练，当然速度慢，所以只用200条数据，其中150条数据用于训练，50条数据用于测试。
; 数据预处理
利用BERT实现情感分类的关键就是 要把数据处理成BERT需要的输入形式。
BERT的输入包括三个部分：
第一个部分 是句子中每个字对应的id，我们用input_ids表示，
这个id需要用到BERT的字库，字库里面每个字所排的次序就是id。
第二个部分 是mask，我们用input_mask表示，假设我们设置BERT输入的句子最大长度是128，
如果我们的句子长度是100，那么mask前100个填1，后面28个填0。
第三部分 是句子标识符id，我们用segment_ids表示，如果第一句全为0，如果是第二句全为1，
以此类推，由于情感分类只涉及到一个句子，所以该标识符都是0。
将一个句子处理成上面这样的输入，要经过两步，
第一步 是对句子进行分词，在英文里面叫做”tokenize”，分词后的结果称为”tokens”。
对于中文来说，分词后的结果很简单，就是一个一个的字。
完成该项工作可以使用tokenizer.tokenize(text)。下图分词的一个示例。
完成分词后，
第二步 要将tokens转换成id，对于中文来说，就是把一个一个的字转换成字对应的id。
此外呢，还要获取input_mask和segment_ids。
实现该步骤可以使用tokenizer.encode_plus(）。
下图是一个示例，第二个参数max_seq_length是指BERT输入句子的最大长度。
下面是数据预处理的代码，保存在 dataProcessor.py文件中。
import pandas as pd
import os
import logging
logging.basicConfig(format='%(asctime)s - %(levelname)s - %(name)s - %(message)s',
datefmt='%m/%d/%Y %H:%M:%S',
level=logging.INFO)
logger = logging.getLogger(__name__)
class InputExample(object):
"""A single training/test example for simple sequence classification."""
def __init__(self, text, label=None):
self.text = text
self.label = label
class InputFeatures(object):
"""A single set of features of data."""
def __init__(self, input_ids, input_mask, segment_ids, label_id):
self.input_ids = input_ids
self.input_mask = input_mask
self.segment_ids = segment_ids
self.label_id = label_id
class DataProcessor(object):
"""Base class for data converters for sequence classification data sets."""
def get_train_examples(self, data_dir):
"""Gets a collection of InputExample for the train set."""
raise NotImplementedError()
def get_dev_examples(self, data_dir):
"""Gets a collection of InputExample for the dev set."""
raise NotImplementedError()
def get_test_examples(self, data_dir):
"""Gets a collection of InputExample for the test set."""
raise NotImplementedError()
def get_labels(self):
"""Gets the list of labels for this data set."""
raise NotImplementedError()
@classmethod
def _read_csv(cls, input_file, quotechar=None):
"""Reads a tab separated value file."""
data = pd.read_csv(input_file)
return data
class MyPro(DataProcessor):
'''自定义数据读取方法，针对json文件
Returns:
examples: 数据集，包含index、中文文本、类别三个部分
'''
def get_train_examples(self, data_dir):
return self._create_examples(
self._read_csv(os.path.join(data_dir, 'train_data.csv')), 'train')
def get_dev_examples(self, data_dir):
return self._create_examples(
self._read_csv(os.path.join(data_dir, 'dev_data.csv')), 'dev')
def get_test_examples(self, data_dir):
return self._create_examples(
self._read_csv(os.path.join(data_dir, 'test_data.csv')), 'test')
def get_labels(self):
return [0, 1]
def _create_examples(self, data, set_type):
examples = []
for index, row in data.iterrows():
text = row['review']
label = row['label']
examples.append(
InputExample(text=text, label=label))
return examples
def convert_examples_to_features(examples, label_list, max_seq_length, tokenizer, show_exp=True):
'''Loads a data file into a list of InputBatch.
Args:
examples : [List] 输入样本，句子和label
label_list : [List] 所有可能的类别，0和1
max_seq_length: [int] 文本最大长度
tokenizer : [Method] 分词方法
Returns:
features:
input_ids : [ListOf] token的id，在chinese模式中就是每个分词的id，对应一个word vector
input_mask : [ListOfInt] 真实字符对应1，补全字符对应0
segment_ids: [ListOfInt] 句子标识符，第一句全为0，第二句全为1
label_id : [ListOfInt] 将Label_list转化为相应的id表示
'''
label_map = {}
for (i, label) in enumerate(label_list):
label_map[label] = i
features = []
for (ex_index, example) in enumerate(examples):
tokens = tokenizer.tokenize(example.text)
encode_dict = tokenizer.encode_plus(text=tokens,
max_length=max_seq_length,
pad_to_max_length=True,
is_pretokenized=True,
return_token_type_ids=True,
return_attention_mask=True)
input_ids = encode_dict['input_ids']
input_mask = encode_dict['attention_mask']
segment_ids = encode_dict['token_type_ids']
assert len(input_ids) == max_seq_length
assert len(input_mask) == max_seq_length
assert len(segment_ids) == max_seq_length
label_id = label_map[example.label]
if ex_index < 5 and show_exp:
logger.info("*** Example ***")
logger.info("tokens: %s" % " ".join(
[str(x) for x in tokens]))
logger.info("input_ids: %s" % " ".join([str(x) for x in input_ids]))
logger.info("input_mask: %s" % " ".join([str(x) for x in input_mask]))
logger.info(
"segment_ids: %s" % " ".join([str(x) for x in segment_ids]))
logger.info("label: %s (id = %d)" % (example.label, label_id))
features.append(
InputFeatures(input_ids=input_ids,
input_mask=input_mask,
segment_ids=segment_ids,
label_id=label_id))
return features
数据处理成dataSet
数据处理成dataSet的核心
就是把BERT模型的输入处理成tensor格式，
下面是代码，保存在 dataset.py文件中
import torch
from torch.utils.data import Dataset
class MyDataset(Dataset):
def __init__(self, features, mode):
self.nums = len(features)
self.input_ids = [torch.tensor(example.input_ids).long() for example in features]
self.input_mask = [torch.tensor(example.input_mask).float() for example in features]
self.segment_ids = [torch.tensor(example.segment_ids).long() for example in features]
self.label_id = None
if mode == 'train' or 'test':
self.label_id = [torch.tensor(example.label_id) for example in features]
def __getitem__(self, index):
data = {'input_ids': self.input_ids[index],
'input_mask': self.input_mask[index],
'segment_ids': self.segment_ids[index]}
if self.label_id is not None:
data['label_id'] = self.label_id[index]
return data
def __len__(self):
return self.nums
模型的搭建
模型的搭建很简单，
模型的第一层是BERT，将BERT输出的第一个向量，即[CLS]对应的向量，传递给一个线性层，
该线性层作为一个分类器输出维度为2的向量。【0就是消极，1就是积极，就是情感分类，很容易的】
BERT模型会有两个输出：
一个输出是序列输出，即句子的每一个字都输出一个向量，叫做seq_out，
这个输出一般用于实体识别等句子标注任务；
另一个输出是pooled_out，即[CLS]对应的向量，这个输出一般用于句子分类。
我们用pooled_out这个输出。
这里我把BERT的参数冻结住了，只调整线性层的参数，所以用x = pooled_out.detach()对反向传播进行截断。
下面是模型搭建的代码，
from torch import nn
import os
from transformers import BertModel
class ClassifierModel(nn.Module):
def __init__(self,
bert_dir,
dropout_prob=0.1):
super(ClassifierModel, self).__init__()
config_path = os.path.join(bert_dir, 'config.json')
assert os.path.exists(bert_dir) and os.path.exists(config_path), \
'pretrained bert file does not exist'
self.bert_module = BertModel.from_pretrained(bert_dir)
self.bert_config = self.bert_module.config
self.dropout_layer = nn.Dropout(dropout_prob)
out_dims = self.bert_config.hidden_size
self.obj_classifier = nn.Linear(out_dims, 2)
def forward(self,
input_ids,
input_mask,
segment_ids,
label_id=None):
bert_outputs = self.bert_module(
input_ids=input_ids,
attention_mask=input_mask,
token_type_ids=segment_ids
)
seq_out, pooled_out = bert_outputs[0], bert_outputs[1]
x = pooled_out.detach()
out = self.obj_classifier(x)
return out
模型的训练
from torch.utils.data import DataLoader
from torch.utils.tensorboard import SummaryWriter
from model import *
from dataset import *
from dataProcessor import *
import matplotlib.pyplot as plt
import time
from transformers import BertTokenizer
from transformers import logging
logging.set_verbosity_warning()
datadir = "data"
bert_dir = "bert\\bert-chinese"
my_processor = MyPro()
label_list = my_processor.get_labels()
train_data = my_processor.get_train_examples(datadir)
test_data = my_processor.get_test_examples(datadir)
tokenizer = BertTokenizer.from_pretrained(bert_dir)
train_features = convert_examples_to_features(train_data, label_list, 128, tokenizer)
test_features = convert_examples_to_features(test_data, label_list, 128, tokenizer)
train_dataset = MyDataset(train_features, 'train')
test_dataset = MyDataset(test_features, 'test')
train_data_loader = DataLoader(dataset=train_dataset, batch_size=64, shuffle=True)
test_data_loader = DataLoader(dataset=test_dataset, batch_size=64, shuffle=True)
train_data_len = len(train_dataset)
test_data_len = len(test_dataset)
print(f"训练集长度：{train_data_len}")
print(f"测试集长度：{test_data_len}")
my_model = ClassifierModel(bert_dir)
loss_fn = nn.CrossEntropyLoss()
learning_rate = 5e-3
optimizer = torch.optim.Adam(my_model.parameters(), lr=learning_rate, betas=(0.9, 0.99))
total_train_step = 0
total_test_step = 0
step = 0
epoch = 50
writer = SummaryWriter("logs")
train_loss_his = []
train_totalaccuracy_his = []
test_totalloss_his = []
test_totalaccuracy_his = []
start_time = time.time()
my_model.train()
for i in range(epoch):
print(f"-------第{i}轮训练开始-------")
train_total_accuracy = 0
for step, batch_data in enumerate(train_data_loader):
print(batch_data['input_ids'].shape)
output = my_model(**batch_data)
loss = loss_fn(output, batch_data['label_id'])
train_accuracy = (output.argmax(1) == batch_data['label_id']).sum()
train_total_accuracy = train_total_accuracy + train_accuracy
optimizer.zero_grad()
loss.backward()
optimizer.step()
total_train_step = total_train_step + 1
train_loss_his.append(loss)
writer.add_scalar("train_loss", loss.item(), total_train_step)
train_total_accuracy = train_total_accuracy / train_data_len
print(f"训练集上的准确率：{train_total_accuracy}")
train_totalaccuracy_his.append(train_total_accuracy)
total_test_loss = 0
my_model.eval()
test_total_accuracy = 0
with torch.no_grad():
for batch_data in test_data_loader:
output = my_model(**batch_data)
loss = loss_fn(output, batch_data['label_id'])
total_test_loss = total_test_loss + loss
test_accuracy = (output.argmax(1) == batch_data['label_id']).sum()
test_total_accuracy = test_total_accuracy + test_accuracy
test_total_accuracy = test_total_accuracy / test_data_len
print(f"测试集上的准确率：{test_total_accuracy}")
print(f"测试集上的loss：{total_test_loss}")
test_totalloss_his.append(total_test_loss)
test_totalaccuracy_his.append(test_total_accuracy)
writer.add_scalar("test_loss", total_test_loss.item(), i)
end_time = time.time()
total_train_time = end_time-start_time
print(f'训练时间: {total_train_time}秒')
writer.close()
plt.plot(train_loss_his, label='Train Loss')
plt.legend(loc='best')
plt.xlabel('Steps')
plt.show()
plt.plot(test_totalloss_his, label='Test Loss')
plt.legend(loc='best')
plt.xlabel('Steps')
plt.show()
plt.plot(train_totalaccuracy_his, label='Train accuracy')
plt.plot(test_totalaccuracy_his, label='Test accuracy')
plt.legend(loc='best')
plt.xlabel('Steps')
plt.show()
尽管只用了150个数据，
但是训练效果还是不错的，
训练准确度达到了90%以上，测试准确度在85%以上。
上面的代码完全可以搞一个cuda，把模型和数据部署到gpu上，把所有的数据用起来，那速度也很快的
总结
提示：重要经验：
1）BERT输出有俩，cls的一个向量作为句子的分类任务特征用，其余的特征序列是标注啥的用的特征
3）笔试求AC，可以不考虑空间复杂度，但是面试既要考虑时间复杂度最优，也要考虑空间复杂度最优。
Original: https://blog.csdn.net/weixin_46838716/article/details/126324781
Author: 冰露可乐
Title: 复盘：基于attention的多任务多模态情绪情感识别，基于BERT实现文本情感分类（pytorch实战）
原创文章受到原创版权保护。转载请注明出处：https://www.johngo689.com/707254/
转载文章受原作者版权保护。转载请注明原作者出处！