复盘：基于attention的多任务多模态情绪情感识别，基于BERT实现文本情感分类（pytorch实战）_多模态情绪识别

复盘：基于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`s for the train set."""
        raise NotImplementedError()

    def get_dev_examples(self, data_dir):
        """Gets a collection of `InputExample`s for the dev set."""
        raise NotImplementedError()

    def get_test_examples(self, data_dir):
        """Gets a collection of `InputExample`s 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."""
        # dicts = []
        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():
            # guid = "%s-%s" % (set_type, i)
            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`s.

    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)
        # tokens进行编码
        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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146

数据处理成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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

模型的搭建

模型的搭建很简单，

模型的第一层是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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

模型的训练

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.SGD(my_model.parameters(), lr=learning_rate)
#  Adam 参数betas=(0.9, 0.99)
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")
# writer.add_graph(myModel, input_to_model=myTrainDataLoader[1], verbose=False)
# writer.add_graph(myModel)
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):
        # writer.add_images("tarin_data", imgs, total_train_step)
        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)
# for parameters in myModel.parameters():
#    print(parameters)
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()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124

尽管只用了150个数据，
但是训练效果还是不错的，
训练准确度达到了90%以上，测试准确度在85%以上。



上面的代码完全可以搞一个cuda，把模型和数据部署到gpu上，把所有的数据用起来，那速度也很快的

---

总结

提示：重要经验：

1）BERT输出有俩，cls的一个向量作为句子的分类任务特征用，其余的特征序列是标注啥的用的特征
3）笔试求AC，可以不考虑空间复杂度，但是面试既要考虑时间复杂度最优，也要考虑空间复杂度最优。