基于Pytorch的NLP入门任务思想及代码实现：判断文本中是否出现指定字

今天学了第一个基于Pytorch框架的NLP任务：

判断文本中是否出现指定字

思路：（注意：这是基于字的算法）

任务：判断文本中是否出现“xyz”，出现其中之一即可

训练部分：

一，我们要先设计一个模型去训练数据。
这个Pytorch的模型：
首先通过embedding层：将字符转化为离散数值（矩阵）
通过线性层：设置网络的连接层，进行映射
通过dropout层：将一部分输入设为0（可去掉）
通过激活层：sigmoid激活
通过一个pooling层：降维，将矩阵->向量
通过另一个输出线性层：使输出是一维（1或0）
通过一个激活层*：sigmoid激活。

二，设置一个函数：这个函数能将设定的字符变成字符集，将每一个字符设定一个代号，比如说：“我爱你”-> 我：1，爱：2，你：3。当出现"你爱我"时，计算机接受的是：3，2，1。这样方便计算机处理字符。

三，因为我们没有训练样本和测试样本，所以我们要自己生成一些随机样本。通过random.choice在字符集中随机顺序输出字符作为输入，并将输入中含有"xyz"的样本的输出值为“1”，反之为“0”

四，设置一个函数，将随机得到的样本，放入数据集中（列表），便于运算。

五，设置测试函数：随机建立一些样本，根据样本的输出来设定有多少个正样本，多少个负样本，再将预测的样本输出来与样本输出对比，得到正确率。

六，最后的main函数：按照训练轮数和训练组数，通过BP反向传播更新权重进行训练。然后调取测试函数得到acc等数据。将loss和acc的数值绘制下来，保存模型和词表。

预测部分

将保存的词表和模型加载进来，将输入的字符转化为列表，然后进入模型forward函数进行预测，最后打印出结果。

代码实现：

import torch
import torch.nn as nn
import numpy as np
import random
import json
import matplotlib.pyplot as plt

"""
基于pytorch的网络编写
实现一个网络完成一个简单nlp任务
判断文本中是否有某些特定字符出现
"""

class TorchModel(nn.Module):
    def __init__(self, input_dim, sentence_length, vocab):
        super(TorchModel, self).__init__()
        self.embedding = nn.Embedding(len(vocab) + 1, input_dim)    #embedding层：将字符转化为离散数值
        self.layer = nn.Linear(input_dim, input_dim)    #对输入数据做线性变换
        self.classify = nn.Linear(input_dim, 1)     #映射到一维
        self.pool = nn.AvgPool1d(sentence_length)   #pooling层：降维
        self.activation = torch.sigmoid     #sigmoid做激活函数
        self.dropout = nn.Dropout(0.1)  #一部分输入为0
        self.loss = nn.functional.mse_loss  #loss采用均方差损失

    #当输入真实标签，返回loss值；无真实标签，返回预测值
    def forward(self, x, y=None):
        x = self.embedding(x)
        #输入维度:(batch_size, sen_len)输出维度：(batch_size, sen_len, input_dim)将文本->矩阵
        x = self.layer(x)
        #输入维度:(batch_size, sen_len, input_dim)输出维度：(batch_size, sen_len, input_dim)
        x = self.dropout(x)
        #输入维度:(batch_size, sen_len, input_dim)输出维度：(batch_size, sen_len, input_dim)
        x = self.activation(x)
        #输入维度:(batch_size, sen_len, input_dim)输出维度：(batch_size, sen_len, input_dim)
        x = self.pool(x.transpose(1,2)).squeeze()
        #输入维度:(batch_size, sen_len, input_dim)输出维度：(batch_size, input_dim)将矩阵->向量
        x = self.classify(x)
        #输入维度:(batch_size, input_dim)输出维度：(batch_size, 1)
        y_pred = self.activation(x)
        #输入维度:(batch_size, 1)输出维度：(batch_size, 1)
        if y is not None:
            return self.loss(y_pred, y)
        else:
            return y_pred

#字符集随便挑了一些汉字，实际上还可以扩充
#为每个汉字生成一个标号
#{"不":1, "东":2, "个":3...}
#不东个->[1,2,3]
def build_vocab():
    chars = "不东个么买五你儿几发可同名呢方人上额旅法xyz"  #随便设置一个字符集
    vocab = {}
    for index, char in enumerate(chars):
        vocab[char] = index + 1   #每个字对应一个序号,+1是序号从1开始
    vocab['unk'] = len(vocab)+1   #不在表中的值设为前一个+1
    return vocab

#随机生成一个样本
#从所有字中选取sentence_length个字
#如果vocab中的xyz出现在样本中，则为正样本
#反之为负样本
def build_sample(vocab, sentence_length):
    #将vacab转化为字表，随机从字表选取sentence_length个字，可能重复
    x = [random.choice(list(vocab.keys())) for _ in range(sentence_length)]
    #指定哪些字必须在正样本出现
    if set("xyz") & set(x):     #若xyz与x中的字符相匹配，则为1，为正样本
        y = 1
    else:
        y = 0
    x = [vocab.get(word,vocab['unk']) for word in x]   #将字转换成序号
    return x, y

#建立数据集
#输入需要的样本数量。需要多少生成多少
def build_dataset(sample_length,vocab, sentence_length):
    dataset_x = []
    dataset_y = []
    for i in range(sample_length):
        x, y = build_sample(vocab, sentence_length)
        dataset_x.append(x)
        dataset_y.append([y])
    return torch.LongTensor(dataset_x), torch.FloatTensor(dataset_y)

#建立模型
def build_model(vocab, char_dim, sentence_length):
    model = TorchModel(char_dim, sentence_length, vocab)
    return model

#测试代码
#用来测试每轮模型的准确率
def evaluate(model, vocab, sample_length):
    model.eval()
    x, y = build_dataset(200, vocab, sample_length)
    #建立200个用于测试的样本（因为测试样本是随机生成的，所以不存在过拟合）
    print("本次预测集中共有%d个正样本，%d个负样本"%(sum(y), 200 - sum(y)))
    correct, wrong = 0, 0
    with torch.no_grad():
        y_pred = model(x)      #调用Pytorch模型预测
        for y_p, y_t in zip(y_pred, y):  #与真实标签进行对比
            if float(y_p) < 0.5 and int(y_t) == 0:
                correct += 1   #负样本判断正确
            elif float(y_p) >= 0.5 and int(y_t) == 1:
                correct += 1   #正样本判断正确
            else:
                wrong += 1
    print("正确预测个数：%d, 正确率：%f"%(correct, correct/(correct+wrong)))
    return correct/(correct+wrong)


def main():
    epoch_num = 10        #训练轮数
    batch_size = 20       #每次训练样本个数
    train_sample = 1000   #每轮训练总共训练的样本总数
    char_dim = 20         #每个字的维度
    sentence_length = 10   #样本文本长度
    vocab = build_vocab()       #建立字表
    model = build_model(vocab, char_dim, sentence_length)    #建立模型
    optim = torch.optim.Adam(model.parameters(), lr=0.005)   #建立优化器
    log = []
    for epoch in range(epoch_num):
        model.train()
        watch_loss = []
        for batch in range(int(train_sample / batch_size)):
            x, y = build_dataset(batch_size, vocab, sentence_length) #每次训练构建一组训练样本
            optim.zero_grad()    #梯度归零
            loss = model(x, y)   #计算loss
            watch_loss.append(loss.item())  #将loss存下来，方便画图
            loss.backward()      #计算梯度
            optim.step()         #更新权重
        print("=========\n第%d轮平均loss:%f" % (epoch + 1, np.mean(watch_loss)))
        acc = evaluate(model, vocab, sentence_length)   #测试本轮模型结果
        log.append([acc, np.mean(watch_loss)])
    plt.plot(range(len(log)), [l[0] for l in log])  #画acc曲线：蓝色的
    plt.plot(range(len(log)), [l[1] for l in log])  #画loss曲线：黄色的
    plt.show()
    #保存模型
    torch.save(model.state_dict(), "model.pth")
    writer = open("vocab.json", "w", encoding="utf8")
    #保存词表
    writer.write(json.dumps(vocab, ensure_ascii=False, indent=2))
    writer.close()
    return

#最终预测
def predict(model_path, vocab_path, input_strings):
    char_dim = 20  # 每个字的维度
    sentence_length = 10  # 样本文本长度
    vocab = json.load(open(vocab_path, "r", encoding="utf8"))
    model = build_model(vocab, char_dim, sentence_length)    #建立模型
    model.load_state_dict(torch.load(model_path))   #将模型文件加载进来
    x = []
    for input_string in input_strings:  #转化输入
        x.append([vocab[char] for char in input_string])
    model.eval()    #在torch中预测注意这个：停止dropout
    with torch.no_grad():   #在torch中预测注意这个：停止梯度
        result = model.forward(torch.LongTensor(x)) #根据自己设计的函数定义，只输入x就会输出预测值
    for i, input_string in enumerate(input_strings):
        print(round(float(result[i])), input_string, result[i])
        #round(float(result))是将预测结果四舍五入得到0或1的预测值



if __name__ == "__main__":
    main()
    #如果是进行预测，将下面两行解除注释，将main()注释掉，即可调用最终预测函数进行预测
    # test_strings = ["个么买不东五你儿x发", "不东东么儿几买五你发", "不东个么买五你个么买", "不z个五么买你儿几发"]
    # predict("model.pth", "vocab.json", test_strings)
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
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167

运行结果展示：

训练部分：

（蓝色是acc曲线，黄色是loss曲线）

预测部分：

一些补充：

一：model.eval()或者model.train()的作用

如果模型中有BN层(Batch Normalization）和Dropout，需要在训练时添加model.train()，在测试时添加model.eval()。其中model.train()是保证BN层用每一批数据的均值和方差，而model.eval()是保证BN用全部训练数据的均值和方差；而对于Dropout，model.train()是随机取一部分网络连接来训练更新参数，而model.eval()是利用到了所有网络连接。

二：Pytorch模型中用了两次激活函数

在每一个网络层后使用一个激活层是一种比较常见的模型搭建方式，但不是必要的。这个只是举例，去掉也是可行的。在具体任务中，带着好还是不带好也跟数据和任务本身有关，没有确定答案（如果在代码 中把第一个激活层注释掉 反而性能更好）

三：对x = self.pool(x.transpose(1,2)).squeeze()代码的解读

通过shape方法我们能知道，在pool前，x的维度输出是[20,10,20]，代表20个10×20的矩阵，代表着[这一批的个数，样本文本长度，输入维度]，transpose(1,2)是将x中行和列调换（转置），然后通过pooling层将[20,20,10]->[20,20,1]，最后通过squeeze()进行降维变成[20,20]。
（池化层的作用及理解）

四：embedding层的理解

embedding层并不是单纯的单词映射，而是将单词表中每个单词的数值与权重相乘。在第一次时有默认权重，然后在接下来的训练中，embedding层的权重与分类权重一起经过训练。