textCNN模型学习及使用

大家好，欢迎来到IT知识分享网。

Yoon Kim在论文(2014 EMNLP) Convolutional Neural Networks for Sentence Classification提出TextCNN。

  将卷积神经网络CNN应用到文本分类任务，利用多个不同size的kernel来提取句子中的关键信息（类似于多窗口大小的ngram），从而能够更好地捕捉局部相关性。

一、textCNN模型结构

  textCNN的详细的结构如下图所示

• Embedding：第一层是图中最左边的7×5的句子矩阵，每行是词向量，词向量维度=5。
• Convolution：然后经过 kernel_sizes=(2,3,4) 的一维卷积层，每个kernel_size 的out_channel=2。
• MaxPolling：第三层是一个1-max pooling层，这样不同长度句子经过pooling层之后都能变成定长的表示。然后将这些定长的特征表示进行concatenate。
• FullConnection and Softmax：最后接一层全连接的 softmax 层，输出每个类别的概率。

二、textCNN与用于图像的CNN的不同

  由于该模型是用于文本的(而非CNN的传统处理对象：图像)，因此在cnn的操作上相对应地做了一些小调整：

• 对于文本任务，输入层自然使用了word embedding来做input data representation。
• 接下来是卷积层，大家在图像处理中经常看到的卷积核都是正方形的，比如4*4，然后在整张image上沿宽和高逐步移动进行卷积操作。但是NLP中输入的“image”是一个词矩阵，比如n个words，每个word用200维的vector表示的话，这个”image”就是n*200的矩阵，卷积核只在高度上已经滑动，在宽度上和word vector的维度一致（=200），也就是说每次窗口滑动过的位置都是完整的单词，不会将几个单词的一部分“vector”进行卷积，这也保证了word作为语言中最小粒度的合理性。（当然，如果研究的粒度是character-level而不是word-level，需要另外的方式处理）
• 由于卷积核和word embedding的宽度一致，一个卷积核对于一个sentence，卷积后得到的结果是一个vector， shape=（sentence_len – filter_window + 1, 1），那么，在max-pooling后得到的就是一个Scalar。所以，这点也是和图像卷积的不同之处，需要注意一下。
• 正是由于max-pooling后只是得到一个scalar，在nlp中，会实施多个filter_window_size（比如3,4,5个words的宽度分别作为卷积的窗口大小），每个window_size又有num_filters个（比如64个）卷积核。一个卷积核得到的只是一个scalar太孤单了，智慧的人们就将相同window_size卷积出来的num_filter个scalar组合在一起，组成这个window_size下的feature_vector。
• 最后再将所有window_size下的feature_vector也组合成一个single vector，作为最后一层softmax的输入。

三、论文中的参数

关于model的参数

• filter windows: [3,4,5]
• filter maps: 100 for each filter window
• dropout rate: 0.5
• l2 constraint: 3
• randomly select 10% of training data as dev set(early stopping)
• word2vec(google news) as initial input, dim = 300
• sentence of length: n, padding where necessary
• number of target classes
• dataset size
• vocabulary size

关于training的参数

• mini batch size: 50
• shuffuled mini batch
• Adadelta update rule: similar results to Adagrad but required fewer epochs
• Test method: standard train/test split ot CV

Dropout策略：

• 在训练阶段，对max-pooling layer的输出实行一些dropout，以概率p激活，激活的部分传递给softmax层。
• 在测试阶段，w已经学好了，但是不能直接用于unseen sentences，要乘以p之后再用，这个阶段没有dropout了全部输出给softmax层。

四、实验设置

  文中做了四个对比的实验，都是围绕词向量展开的。

• CNN-rand： 词向量随机初始化，同时当作训练过程中优化的参数
• CNN-static：词向量使用word2vec，同时固定不变。
• CNN-non-static：词向量使用word2vec，但是在训练过程中进行微调。
• CNN-multichannel：CNN-static和CNN-non-static的混合版本，即输入这两种类型。

五、利用textCNN进行中文文本分类

1、数据获取

2、数据预处理

（1）选择样本
  百科问答版中数据类别非常多，为了简化，从中筛选了少量的样本进行训练学习。选择了标题前2个字为教育、健康、生活、娱乐和游戏五个类别，同时每个类别各5000，共50005条数据进行训练。同时从验证集中筛选同样5个类别的数据各1000条（共10005）来作为验证数据。新建get_my_tain_data.py文件，代码如下：

# -*- coding: utf-8 -*- """ 从原数据中选取部分数据 选取数据的类别存放在： WantedClass字典中 选取每个类别的数量为：5000条 """ import json TrainJsonFile = '../data/baike_qa2019/baike_qa_train.json' MyTrainJsonFile = '../data/baike_qa2019/my_traindata.json' ValJsonFile = '../data/baike_qa2019/baike_qa_valid.json' MyValJsonFile = '../data/baike_qa2019/my_valdata.json' WantedClass = { 
   '教育': 0, '健康': 0, '生活': 0, '娱乐': 0, '游戏': 0} WantedNum = 5000 numWantedAll = WantedNum * 5 def main(inFile, MyFile): Datas = open(inFile, 'r', encoding='utf_8').readlines() f = open(MyFile, 'w', encoding='utf_8') num = 0 for line in Datas: data = json.loads(line) # 读取一行数据,并且返回一个字典 data cla = data['category'][0:2] # 提取类别中的前两个字符 if cla in WantedClass and WantedClass[cla] < WantedNum: json_data = json.dumps(data, ensure_ascii=False) # 输出真正的中文需要指定ensure_ascii=False,否则输出的是中文的ascii f.write(json_data) # 将该行数据写入文件 f.write('\n') WantedClass[cla] += 1 num += 1 if num // 500: print("processed %s row" % num) if num >= numWantedAll: print("over") break if __name__ == '__main__': main(TrainJsonFile, MyTrainJsonFile) # main(ValJsonFile, MyValJsonFile) 

上述代码中文件的存放路径要搞清楚了。

（2）生成词表

  在有了训练数据之后，我们需要得到训练数据中所有的“title”对应的词表。也就是说我们首先对每个标题使用jieba分词工具进行分词，之后去除停用词，剩下的就构成了我们的词表。新建get_wordlist.py文件，具体代码如下：

# -*- coding: utf_8 -*- """ 主要是将词向量转换为对应的id 另外，统计了不同长度的句向量所占比例,方便设置最大的句子长度 """ import jieba import json trainFile = '../data/baike_qa2019/my_traindata.json' StopWordFile = 'stopwords.txt' word2idFile = 'word2id.txt' lengthFile = 'sen_length.txt' def read_stopword(file): data = open(file, 'r', encoding='utf_8').read().split('\n') print(data[0:5]) return data def main(): worddict = { 
   } stoplist = read_stopword(StopWordFile) datas = open(trainFile, 'r', encoding='utf-8').read().split('\n') datas = list(filter(None, datas)) data_num = len(datas) # 训练句子总数 len_dic = { 
   } # 统计句子长度 for line in datas: line = json.loads(line) title = line['title'] title_seg = jieba.cut(title, cut_all=False) length = 0 for w in title_seg: if w in stoplist: # 去除停用词 continue length += 1 if w in worddict: # 该词存在于字典中，数量+1 worddict[w] += 1 else: worddict[w] = 1 if length in len_dic: # 该长度存在于字典中 len_dic[length] += 1 else: len_dic[length] = 1 wordlist = sorted(worddict.items(), key=lambda item: item[1], reverse=True) # 将worddict按照数量进行逆序排列 f = open(word2idFile, 'w', encoding='utf-8') ind = 0 for w in wordlist: line = w[0] + ' ' + str(ind) + ' ' + str(w[1]) + '\n' # 词 、id 和 数量 ind += 1 f.write(line) for k, v in len_dic.items(): len_dic[k] = round(v * 1.0 / data_num, 3) len_list = sorted(len_dic.items(), key=lambda item: item[0], reverse=True) # 按照数量逆序排列 f = open(lengthFile, 'w') for t in len_list: d = str(t[0]) + ' ' + str(t[1]) + '\n' f.write(d) if __name__ == '__main__': main() 

#-*- coding: utf_8 -*- """ 主要是将 title 的文本内容转换为向量id的形式 """ import json import jieba import random trainFile = '../data/baike_qa2019/my_traindata.json' valFile = '../data/baike_qa2019/my_valdata.json' stopwordFile = 'stopwords.txt' word2idFile = 'word2id.txt' trainDataVecFile = 'traindata_vec.txt' valDataVecFile = 'valdata_vec.txt' maxLen = 20 labelFile = 'labelFile.txt' def read_labelFile(file): data = open(file, 'r', encoding='utf-8').read().split('\n') label2id = { 
   } id2label = { 
   } for line in data: line = line.split(' ') label = line[0] id = int(line[1]) label2id[label] = id id2label[id] = label return label2id, id2label def read_stopword(file): data = open(file, 'r', encoding='utf_8').read().split('\n') return data def get_worddict(file): """ 文件word2id.txt中的第一、二列存放的就是word和其id 该函数就是将他们读入字典 word2id和id2word中 """ datas = open(file, 'r', encoding='utf_8').read().split('\n') datas = list(filter(None, datas)) word2id = { 
   } for line in datas: line = line.split(' ') word2id[line[0]] = int(line[1]) id2word = { 
   word2id[w]: w for w in word2id} return word2id, id2word def json2vec(inFile, outFile): label2id, id2label = read_labelFile(labelFile) word2id, id2word = get_worddict(word2idFile) dataVec = open(outFile, 'w') # 输出文件 stoplist = read_stopword(stopwordFile) # 读入停用词表 datas = open(inFile, 'r', encoding='utf-8').read().split('\n') # 读入训练数据 datas = list(filter(None, datas)) random.shuffle(datas) # 将数据进行洗牌 for line in datas: line = json.loads(line) title = line['title'] # 标题内容 cla = line['category'][0:2] # 类别 cla_id = label2id[cla] # 类别对应的id title_seg = jieba.cut(title, cut_all=False) # 将 title内容分词 title_vec = [cla_id] # title向量的第一个位置存放了类别id for w in title_seg: # 生成 title的句向量 if w in stoplist: continue if w in word2id: title_vec.append(word2id[w]) length = len(title_vec) # 句向量的长度 if length > maxLen+1: # 句向量大于20，截断 title_vec = title_vec[0:21] if length < maxLen + 1: # 句向量小于20，补0 title_vec.extend([0]*(maxLen - length + 1)) # 将一个句向量写入文件 for n in title_vec: dataVec.write(str(n) + ',') dataVec.write('\n') def main(): #json2vec(trainFile, trainDataVecFile) json2vec(valFile, valDataVecFile) if __name__ == '__main__': main() 

3、模型搭建

 import torch import torch.nn as nn from torch.nn import functional as F import math class textCNN(nn.Module): def __init__(self, param): super(textCNN, self).__init__() ci = 1 kernel_num = param['kernel_num'] kernel_size = param['kernel_size'] vocab_size = param['vocab_size'] embed_dim = param['embed_dim'] dropout = param['dropout'] class_num = param['class_num'] self.param = param self.embed = nn.Embedding(vocab_size, embed_dim, padding_idx=1) self.conv11 = nn.Conv1d(in_channels=embed_dim, out_channels=kernel_num, kernel_size=3) self.conv12 = nn.Conv1d(in_channels=embed_dim, out_channels=kernel_num, kernel_size=4) self.conv13 = nn.Conv1d(in_channels=embed_dim, out_channels=kernel_num, kernel_size=5) self.dropout = nn.Dropout(dropout) self.fc = nn.Linear(len(kernel_size) * kernel_num, class_num) def init_embed(self, embed_matrix): self.embed.weight = nn.Parameter(torch.Tensor(embed_matrix)) @staticmethod def conv_and_pool(x, conv): # x: batch_size * embed_dim * sentence_length x = conv(x) # x: batch_size * kernel_num * (sentence_length-kernel_size+1) x = F.relu(x) x = F.max_pool1d(x, x.size(2)).squeeze(2) # 在最后一个维度上进行最大池化，输出 x： batch_size * kernel_num return x def forward(self, x): # x: (batch, sentence_length) x = self.embed(x) # x: (batch, sentence_length, embed_dim) # TODO init embed matrix with pre-trained x = x.permute(0, 2, 1) # x: (batch, embed_dim, sentence_length) x1 = self.conv_and_pool(x, self.conv11) # (batch, kernel_num) x2 = self.conv_and_pool(x, self.conv12) # (batch, kernel_num) x3 = self.conv_and_pool(x, self.conv13) # (batch, kernel_num) x = torch.cat((x1, x2, x3), 1) # (batch, 3 * kernel_num) x = self.dropout(x) logit = F.log_softmax(self.fc(x), dim=1) return logit def init_weight(self): for m in self.modules(): if isinstance(m, nn.Conv2d): n = m.kernel_size[0] * m.kernel_size[1] * m.out_channels m.weight.data.normal_(0, math.sqrt(2. / n)) if m.bias is not None: m.bias.data.zero_() elif isinstance(m, nn.BatchNorm2d): m.weight.data.fill_(1) m.bias.data.zero_() elif isinstance(m, nn.Linear): m.weight.data.normal_(0, 0.01) m.bias.data.zero_() 

  其中卷积层用的是一维卷积函数Conv1d()，这里一定要好好学习一下该函数作为文本任务时与二维卷积函数Conv2d()的区别，如果不懂的可以移步这里

4、训练

  接下来就是训练了，新建 train.py文件，代码如下：

import torch import os import torch.nn as nn import numpy as np import time from model import textCNN import sen2id import textCNN_data word2id, id2word = sen2id.get_worddict('word2id.txt') label2id, id2label = sen2id.read_labelFile('labelFile.txt') textCNN_param = { 
    'vocab_size': len(word2id), 'embed_dim': 60, 'class_num': len(label2id), "kernel_num": 16, "kernel_size": [3, 4, 5], "dropout": 0.5, } dataLoader_param = { 
    'batch_size': 128, 'shuffle': True } def train(): print("init net...") net = textCNN(textCNN_param) weightFile = 'weight.pkl' if os.path.exists(weightFile): print('load weight') net.load_state_dict(torch.load(weightFile)) else: net.init_weight() print(net) #net.cuda() # init dataset print('init dataset...') dataLoader = textCNN_data.textCNN_dataLoader(dataLoader_param) valdata = textCNN_data.get_valdata() optimizer = torch.optim.Adam(net.parameters(), lr=0.01) criterion = nn.NLLLoss() log = open('log_{}.txt'.format(time.strftime('%y%m%d%H')), 'w') log.write('epoch step loss\n') log_test = open('log_test_{}.txt'.format(time.strftime('%y%m%d%H')), 'w') log_test.write('epoch step test_acc\n') print("training...") for epoch in range(100): for i, (clas, sentences) in enumerate(dataLoader): optimizer.zero_grad() sentences = sentences.type(torch.LongTensor) clas = clas.type(torch.LongTensor) out = net(sentences) loss = criterion(out, clas) loss.backward() optimizer.step() if (i+1) % 1 == 0: print("epoch:", epoch + 1, "step:", i + 1, "loss:", loss.item()) data = str(epoch + 1) + ' '+ str(i + 1) + ' ' + str(loss.item()) + '\n' log.write(data) print("save model...") torch.save(net.state_dict(), weightFile) torch.save(net.state_dict(), "model\{}_model_iter_{}_{}_loss_{:.2f}.pkl".format(time.strftime('%y%m%d%H'), epoch, i, loss.item())) print("epoch:", epoch + 1, "step:", i+1, "loss:", loss.item()) if __name__ == "__main__": train() 

5、测试

  新建test.py文件，具体代码如下：

import torch import os import torch.nn as nn import numpy as np import time from model import textCNN import sen2id word2id, id2word = sen2id.get_worddict("word2id.txt") label2id, id2label = sen2id.read_labelFile("labelFile.txt") textCNN_param = { 
    'vocab_size': len(word2id), 'embed_dim': 60, 'class_num': len(label2id), "kernel_num": 16, "kernel_size": [3, 4, 5], "dropout": 0.5, } def get_valData(file): datas = open(file, 'r').read().split('\n') datas = list(filter(None, datas)) return datas def parse_net_result(out): score = max(out) label = np.where(out==score)[0][0] return label, score def test(): # init net print('init net...') net = textCNN(textCNN_param) weightFile = 'weight.pkl' if os.path.exists(weightFile): print('load weight') net.load_state_dict(torch.load(weightFile)) else: print('No weight file') exit() print(net) net.eval() numAll = 0 numRight = 0 testData = get_valData('valdata_vec.txt') for data in testData: numAll += 1 data = data.split(',') label = int(data[0]) sentence = np.array([int(x) for x in data[1:21]]) sentence = torch.from_numpy(sentence) # sentence:[20] predict = net(sentence.unsqueeze(0).type(torch.LongTensor)).cpu().detach().numpy()[0] # predict是小于0的数 label_pre, score = parse_net_result(predict) if label_pre == label and score > -100.: numRight += 1 if numAll % 100 == 0: print('acc:{}({}/{})'.format(numRight/numAll, numRight, numAll)) if __name__ == '__main__': test() 

参考资料;

• 博客：http://www.wildml.com/2015/11/understanding-convolutional-neural-networks-for-nlp/
• 中文文本语料：https://github.com/brightmart/nlp_chinese_corpus
• 博客：textCNN模型学习及使用