pytorch学习（七）——文本分类实例_pytorch文本分类

1. 整体思路

在这篇文章里，我们要试着用pytorch对文本进行分类，我来叙述下这个实例的基本思路。文本分类不像图像分类，图像读入计算机就是一个个的像素点，就已经是数值类型了，但是文本不同，文本是一个个的文字组成起来的，但是神经网络中能够接受训练的是一个个的数字，那么就要想办法将文字转为数值，一个可行的办法是使用 One-hot 编码，但是One-hot编码的冗余程度太高了，这里我们使用更加有效的方法，词向量的方法，即将文本转为一个个的向量，用多维向量进行表示。本例中下载了搜狗以及腾讯预训练的词向量模型，可以选择一个进行使用。

首先，我们确定文本分类中使用的模型，这里我们使用RNN模型中的双向LSTM构建模型。然后，对数据进行预处理，将数据分为文本以及标签两个部分，由于模型是预先搭建好的，所以这里要确定每句话最多处理多少个字数，即 $seq\_size$ ，多于 $seq\_size$ 则将其截断，少于 $seq\_size$ 则进行填充。最后，确定好损失函数和优化器，对模型的参数进行训练即可。

2. 工具设置

这里我们使用搜狗的词向量模型，并设置随机数种子，设置随机数种子能够帮助我们在每一次运行代码时都得到相似的结果，方便对代码进行复现。相关设置如下代码所示

import torch
import numpy as np

# 存放数据的文件夹
dataset = 'text_classify_data'

# 搜狗新闻:embedding_SougouNews.npz, 腾讯:embedding_Tencent.npz
embedding = 'embedding_SougouNews.npz'

# 设置随机数种子，保证每次运行结果一致，不至于不能复现模型
np.random.seed(1)
torch.manual_seed(1)
torch.cuda.manual_seed_all(1)
torch.backends.cudnn.deterministic = True  # 保证每次结果一样
1
2
3
4
5
6
7
8
9
10
11
12
13
14

3. 超参数设置

本例中采用另外一种设置超参数的方法，将所有的超参数都写在一个类中，这样需要传递超参数时只需要传递该类的对象即可，十分方便。而且如果需要对超参数进行修改，可以只在该类中进行修改，避免了修改超参数全篇找名称的麻烦，该例用到的超参数及设置如下所示：

import numpy as np

class Config(object):

    """配置参数"""
    def __init__(self, dataset, embedding):
        '''
        :param dataset: 数据所在的文件夹路径
        :param embedding: 使用的词嵌入文件名称
        '''
        self.model_name = 'TextRNN'
        self.train_path = dataset + '/data/train.txt'                                # 训练集
        self.dev_path = dataset + '/data/dev.txt'                                    # 验证集
        self.test_path = dataset + '/data/test.txt'                                  # 测试集
        self.class_list = [x.strip() for x in open(
            dataset + '/data/class.txt').readlines()]                                # 类别列表
        self.vocab_path = dataset + '/data/vocab.pkl'                                # 词表
        self.save_path = dataset + '/saved_dict/' + self.model_name + '.ckpt'        # 模型训练结果
        self.embedding_pretrained = torch.tensor(
            np.load(dataset + '/data/' + embedding)["embeddings"].astype('float32')) # 预训练词向量
        self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')   # 设备

        self.dropout = 0.5                                              # 随机失活
        self.require_improvement = 1000                                 # 若超过1000batch效果还没提升，则提前结束训练
        self.num_classes = len(self.class_list)                         # 类别数
        self.num_epochs = 10                                            # epoch数
        self.batch_size = 128                                           # mini-batch大小
        self.pad_size = 32                                              # 每句话处理成的长度(短填长切)
        self.learning_rate = 1e-3                                       # 学习率
        self.embed = self.embedding_pretrained.size(1)                  # 词向量维度, 若使用了预训练词向量，则维度统一
        self.hidden_size = 128                                          # lstm隐藏层
        self.num_layers = 2                                             # lstm层数


# 超参数的设置
config = Config(dataset, embedding)
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

3. 数据处理

数据已经传到了百度网盘，大家可以从这里获取（提取码：1234）。由于得到的数据中的格式如下所示

上述即一段文本，一段文本所处的分类，所以在处理数据的时候，需要将文本与标签分别读出来存放在不同变量中。除此之外，还需要载入词嵌入模型，将每个词读出成为一个词向量。本例中采用更加简单的字向量，将一段话分成一个字一个字的列表。

下面我简单说说预训练的词嵌入模块。词嵌入模块中对每一个词都进行了一个编号，读入词嵌入模型后，能够读出编号所对应的词向量，除此之外，词嵌入中还有两种特殊情况：

1. 为了防止文本中出现词向量中不存在的词，词嵌入中还对这些不知道的词设置了一个编号，可以使用 '<UNK>' 字符得到相应的编号；
2. 为了防止一些词语没有达到规定的最大长度，词嵌入允许对其进行填充，填充的字符为 '<PAD>' ，可以使用该字符得到编号。

所以，数据预处理的任务就很明确了，将文本与标签分开，再将文本中的每一个字分开，长度多于 $seq\_size$ 则将其截断，少于 $seq\_size$ 则进行填充，最后将各字转为词嵌入模型中的标号返回，方便模型得到对象的向量。

具体的代码实现如下所示：

import pickle as pkl
from tqdm import tqdm

UNK, PAD = '<UNK>', '<PAD>'  # 未知字，padding符号


def get_data(config):
    tokenizer = lambda x: [y for y in x]  # 字级别

    vocab = pkl.load(open(config.vocab_path, 'rb'))

    print(f"Vocab size: {len(vocab)}")

    train = load_dataset(config.train_path, config.pad_size, tokenizer, vocab)
    dev = load_dataset(config.dev_path, config.pad_size, tokenizer, vocab)
    test = load_dataset(config.test_path, config.pad_size, tokenizer, vocab)
    return vocab, train, dev, test


def load_dataset(path, pad_size, tokenizer, vocab):
    '''
    将路径文本文件分词并转为三元组返回
    :param path: 文件路径
    :param pad_size: 每个序列的大小
    :param tokenizer: 转为词级别或字级别
    :param vocab: 词向量模型
    :return: 二元组，含有字ID，标签
    '''
    contents = []
    with open(path, 'r', encoding='UTF-8') as f:
        # tqdm可以看进度条
        for line in tqdm(f):
            lin = line.strip()
            if not lin:
                continue
            content, label = lin.split('\t')

            # word_line存储每个字的id
            words_line = []
            # 分割器，分成每个字
            token = tokenizer(content)
            # 字的长度
            seq_len = len(token)
            if pad_size:
                # 如果字长度小于指定长度，则填充，否则截断
                if len(token) < pad_size:
                    token.extend([vocab.get(PAD)] * (pad_size - len(token)))
                else:
                    token = token[:pad_size]
                    seq_len = pad_size
            # 将每个字映射为ID
            for word in token:
                words_line.append(vocab.get(word, vocab.get(UNK)))
            contents.append((words_line, int(label)))
    return contents


vocab, train_data, dev_data, test_data = get_data(config)
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

4. 制作数据管道

数据已经处理成为了词嵌入模型中的编号，剩下的就是制作出数据管道了，方便模型在训练数据时能够一批一批的拿到数据，我前面对pytorch自定义数据管道进行了讲解，如果不清楚的可以看前面的博客。下面进行自定义数据管道：

class TextDataset(Dataset):
    def __init__(self, data, config):
        self.device = config.device
        # 将传入的文本存到一起
        self.x = torch.LongTensor([x[0] for x in data]).to(self.device)
        # 将传入的标签存到一起
        self.y = torch.LongTensor([x[1] for x in data]).to(self.device)


    def __getitem__(self,index):
    	# 拿出文本中的一个
        self.text = self.x[index]
        # 拿出对应标签中的一个
        self.label = self.y[index]
        return self.text, self.label


    def __len__(self):
        return len(self.x)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

最后将数据管道实例化，注意这里的数据分为训练集、开发集、测试集，都要进行实例化：

dataloaders = {
    'train': DataLoader(TextDataset(train_data, config), 128, shuffle=True),
    'dev': DataLoader(TextDataset(dev_data, config), 128, shuffle=True),
    'test': DataLoader(TextDataset(test_data, config), 128, shuffle=True)
}
1
2
3
4
5

5. 构建模型

文本分类使用RNN模型的效果更加好，这里使用的是RNN的一种变种，双向LSTM模型，模型的构建如下所示：

import torch
import torch.nn as nn

class RNNModel(nn.Module):
    def __init__(self, config):
        super(RNNModel, self).__init__()
        # 使用预训练的词向量模型，freeze=False 表示允许参数在训练中更新
        self.embedding = nn.Embedding.from_pretrained(config.embedding_pretrained, freeze=False)
        # bidirectional=True表示使用的是双向LSTM
        self.lstm = nn.LSTM(config.embed, config.hidden_size, config.num_layers,
                            bidirectional=True, batch_first=True, dropout=config.dropout)
        # 因为是双向LSTM，所以层数为config.hidden_size * 2
        self.fc = nn.Linear(config.hidden_size * 2, config.num_classes)

    def forward(self, x):
        out = self.embedding(x)  # [batch_size, seq_len, embeding]=[128, 32, 300]
        # lstm 的input为[batchsize, max_length, embedding_size]，输出表示为 output,(h_n,c_n),
        # 保存了每个时间步的输出，如果想要获取最后一个时间步的输出，则可以这么获取：output_last = output[:,-1,:]
        out, _ = self.lstm(out)
        out = self.fc(out[:, -1, :])  # 句子最后时刻的 hidden state
        return out
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

这里需要注意的是 nn.lstm 的输入以及输出的格式。

nn.lstm 要求输入数据的维度格式为 [batchsize, max_length, embedding_size] ，batchsize为每一个batch得到的数据的大小，max_length 为允许的最大词长度，embedding_size 为词向量的维度。

nn.lstm 的输出形式为 output, (h_n,c_n)=nn.lstm(input) ，output 保存了每个个时间步的输出，如果想要获取最后一个时间步的输出，则可以这么获取：output_last = output[:,-1,:]，h_n,c_n 为LSTM内部的参数。

6. 初始化模型参数

模型参数的初始化对模型的学习也有一定的影响，详细的初始化方法可以查看这篇文章，这里我们可以实现以下对模型进行 $Xavier$ 以及 $Kaiming$ 初始化的方法，代码如下所示：

import torch.nn as nn

# 权重初始化，默认xavier
def init_network(model, method='xavier', exclude='embedding', seed=123):
    for name, w in model.named_parameters():
    	# 不对词嵌入的层的参数进行初始化，因为我们使用的是预训练的模型
        if exclude not in name:
            if 'weight' in name:
                if method == 'xavier':
                    nn.init.xavier_normal_(w)
                elif method == 'kaiming':
                    nn.init.kaiming_normal_(w)
                else:
                    nn.init.normal_(w)
            elif 'bias' in name:
                nn.init.constant_(w, 0)
            else:
                pass
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

7. 训练模型

接下来就是对模型进行训练了，由于这里是对文本进行分类，速度肯定比对图像分类要快得多，所以，我们采用迭代100个batch就保存一个最优模型，并在开发集进行测试的方法，而不是一个epoch才进行一次保存模型测试模型，这里设置一个早停机制，如果连续1000个batch模型性能都没有提升，那么就提前结束模型训练。

import torch.nn as nn
import time
import torch
import copy
import pandas as pd
import datetime
from sklearn import metrics
import numpy as np

def train_best(config, model, dataloaders, log_step=100):
    '''
    训练模型
    :param config: 超参数
    :param model: 模型
    :param dataloaders: 处理后的数据，包含trian,dev,test
    :param log_step: 每隔多少个batch打印一次数据，默认100
    :return: 训练的指标
    '''

    optimizer = torch.optim.Adam(model.parameters(), lr=config.learning_rate)
    loss_function = torch.nn.CrossEntropyLoss()
    best_acc = 0
    # 最优模型
    best_model = copy.deepcopy(model.state_dict())

    total_step = 0  # 记录进行到多少batch
    dev_best_loss = float('inf')
    last_improve = 0  # 记录上次验证集loss下降的batch数
    flag = False  # 记录是否很久没有效果提升

    # 保存每一个100个batch的信息
    dfhistory = pd.DataFrame(columns=["epoch", "train_loss", "train_acc", "dev_loss", "dev_acc"])

    device = config.device

    print("Start Training...\n")
    nowtime = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    print("==========" * 8 + "%s\n" % nowtime)

    for i in range(config.num_epochs):
        # 1，训练循环----------------------------------------------------------------

        # 记录每一个batch
        step = 0

        print('Epoch [{}/{}]\n'.format(i + 1, config.num_epochs))

        for inputs, labels in dataloaders['train']:
            # 训练模式，可以更新参数
            model.train()

            inputs = inputs.to(device)
            labels = labels.to(device)
            # 梯度清零，防止累加
            optimizer.zero_grad()

            outputs = model(inputs)
            loss = loss_function(outputs, labels)
            loss.backward()
            optimizer.step()

            total_step += 1
            step += 1

            if step % log_step == 0:
                true = labels.data.cpu()
                # torch.max返回的值最大的值以及最大值的索引，这里只要[1]维度的索引，不要值
                predic = torch.max(outputs.data, 1)[1].cpu()
                train_loss = loss.item()
                train_acc = metrics.accuracy_score(true, predic)

                # 2，开发集验证----------------------------------------------------------------
                dev_acc, dev_loss = dev_eval(model, dataloaders['dev'], loss_function)

                dfhistory.loc[i] = (i, train_loss, train_acc, dev_loss, dev_acc)

                if dev_loss < dev_best_loss:
                    dev_best_loss = dev_loss
                    torch.save(model.state_dict(), config.save_path)
                    last_improve = total_step

                print("[step = {} batch]  train_loss = {:.3f}, train_acc = {:.2%}, dev_loss = {:.3f}, dev_acc = {:.2%}".
                      format(step, train_loss, train_acc, dev_loss, dev_acc))

            if total_step - last_improve > config.require_improvement:
                # 验证集loss超过1000batch没下降，结束训练
                print("No optimization for a long time, auto-stopping...")
                flag = True
                break
        if flag:
            break

    # 3，验证循环----------------------------------------------------------------
    model.load_state_dict(torch.load(config.save_path))
    model.eval()
    start_time = time.time()
    test_acc, test_loss = dev_eval(model, dataloaders['test'], loss_function)
    print('================'*8)
    print('test_loss: {:.3f}      test_acc: {:.2%}'.format(test_loss, test_acc))

    return dfhistory


def dev_eval(model, data, loss_function):
    '''
    得到开发集和测试集的准确率和loss
    :param model: 模型
    :param data: 测试集集和开发集的数据
    :param loss_function: 损失函数
    :return: 损失和准确率
    '''
    model.eval()
    loss_total = 0
    predict_all = np.array([], dtype=int)
    labels_all = np.array([], dtype=int)
    with torch.no_grad():
        for texts, labels in data:
            outputs = model(texts)
            loss = loss_function(outputs, labels)
            loss_total += loss.item()
            labels = labels.data.cpu().numpy()
            predic = torch.max(outputs.data, 1)[1].cpu().numpy()
            labels_all = np.append(labels_all, labels)
            predict_all = np.append(predict_all, predic)

    acc = metrics.accuracy_score(labels_all, predict_all)
    return acc, loss_total / len(data)
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

全部的代码可以在GitHub仓库进行查看。