⼿把⼿教你做简单的CNN⽂本分类——基于pytorch
CNN是在图像处理领域⼤放异彩的⽹络模型，但其实在NLP领域CNN同样有许多应⽤。最近发现，在长⽂本上CNN提取特征的效果确实不错，在⽂本分类这种简单的任务上，并不需要复杂且⽆法并⾏的RNN，CNN就能搞定了。（当然，其实没必要⽤到复杂的神经⽹络，简单的机器学习模型+传统的特征，也能取得不错的效果，⽽且速度还更快）。针对⽂本分类，CNN在长⽂本上的效果很好，⽽且模型也很简单，这是我想写这篇blog的初衷。
既然是⼿把⼿教，那么洛基必须要在座的各位看完本⽂都能做出⼀个CNN⽂本分类器来！代码⾥⾯有超详细的注释，绝对是婴⼉级教学。
⾸先是深度学习环境，pytorch0.4 + python 3.6，还有其他⼀些包⽐如numpy，sklearn这些机器学习常⽤的包，⼤家可以⾃⾏pip下载，安装很⽅便。分词⼯具采⽤的是pkuseg，如果没有这个包，可以⽤jieba分词代替，不过这个包分词效果据说⽐jieba好，感兴趣的⼩伙伴可以pip install pkuseg 安装⼀下。
数据集是CNews的长⽂本分类数据，包含50000个训练样本，5000个验证样本，10000个测试样本，中⽂的新闻语料，这些新闻共分为体育、时尚、游戏等10个类别。因为是长⽂本，所以⽂本预处理的长度阈值设置为了256个token。
⼩伙伴们可以⾃⾏准备训练数据，将数据分为, , 三个⽂件，这三个⽂件每⼀⾏都是 [原始⽂本+ '\t' + 类别索引] 的格式。接着再准备⼀个⽂件，每⼀⾏是⼀个类别名，注意第⼀⾏对应的类别索引=0，第⼆⾏对应的类别索引=1，依此类推，之所以需要这个是为了后续的测试结果可视化。
模型的输⼊采⽤预训练的词向量。因为是做中⽂数据集，所以采⽤了中⽂预训练的词向量 sgns.wiki.word 。该词向量可以到⽹上下载。在训练时，需要把, , , 这四个⽂件以及词向量sgns.wiki.word⽂件放到以下⽂件⽬录：（注意，py ⽂件与CNews⽂件夹在同⼀个⽬录下）
接下来是CNN_Classifier.py的代码，内附超详细注释：
# coding: UTF-8
import os
import time
import torch
as nn
functional as F
import numpy as np
import pickle as pkl
from sklearn import metrics
from importlib import import_module
from tensorboardX import SummaryWriter
from tqdm import tqdm
from datetime import timedelta
import pkuseg # 分词⼯具包，如果没有，可以⽤jieba分词，或者⽤其他分词⼯具
>>>>>>>>>>>>>>>>#
>>>>>> 定义各种参数、路径 >>>>>>###
class Config(object):
def __init__(self, dataset_dir, embedding):
self.dev_path = dataset_dir + '/'                                    # 验证集
self.class_list = [x.strip() for x in open(
dataset_dir + '/').readlines()]                                # 类别名单
self.vocab_path = dataset_dir + '/data/vocab.pkl'                                # 词表
self.save_path = dataset_dir + del_name + '.ckpt'        # 模型训练结果
self.log_path = dataset_dir + '/log/' + del_name
np.load(dataset_dir + '/data/' + embedding)["embeddings"].astype('float32'))\
if embedding != 'random' else None                                      # 预训练词向量
self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')  # 设备
self.dropout = 0.4
self.num_classes = len(self.class_list)
self.n_vocab = 0                                                # 词表⼤⼩，在运⾏时赋值
self.num_epochs = 20
self.batch_size = 32
self.pad_size = 256                                            # 每句话处理成的长度，截长、补短
self.learning_rate = 1e-5
self.filter_sizes = (2, 3, 4)                                  # 卷积核尺⼨
self.num_filters = 256                                          # 卷积核数量(channels数)
>>>>>>>>>>>>>>>>#
>>>>>>> 定义模型结构 >>>>>>###
class Model(nn.Module):
def __init__(self, config):
super(Model, self).__init__()
bedding_pretrained is not None:
else:
# 三个卷积层分别是(1, channels=256, kernal_size=(2, 300))
#                (1, 256, (3, 300))    (1, 256, (4, 300))
# 这三个卷积层是并⾏的，同时提取2-gram、3-gram、4-gram特征
[nn.Conv2d(1, config.num_filters, (k, bed_dim)) for k in config.filter_sizes])
self.dropout = nn.Dropout(config.dropout)
self.fc = nn.Linear(config.num_filters * len(config.filter_sizes), config.num_classes)
# 假设embed_dim=300，每个卷积层的卷积核都有256个（会将⼀个输⼊seq映射到256个channel上）
# 三个卷积层分别为：(1, 256, (2, 300)), (1, 256, (3, 300)), (1, 256, (4, 300))
# x(b_size, 1, seq_len, 300)进⼊卷积层后得到 (b, 256, seq_len-1, 1), (b, 256, seq_len-2, 1), (b, 256, seq_len-3, 1)
# 卷积之后经过⼀个relu，然后把最后⼀个维度上的1去掉(squeeze)，得到x(b, 256, seq_len-1), 接着进⼊池化层
# ⼀个池化层输出⼀个(b, 256)，三个池化层输出三个(b, 256), 然后在forward⾥⾯把三个结果concat起来
def conv_and_pool(self, x, conv):
x = F.relu(conv(x)).squeeze(3)
# max_pool1d表⽰⼀维池化，⼀维的意思是，输⼊x的维度除了b_size和channel，只有⼀维，即x(b_size, channel, d1)，故池化层只需要定义⼀个宽度表⽰kern        # max_pool2d表⽰⼆维池化，x(b_size, channel, d1, d2), 所以max_pool2d定义的kernel_size是⼆维的
# max_pool1d((b, 256, seq_len-1), kernel_size = seq_len-1) -> (b, 256, 1)
# squeeze(2) 之后得到 (b, 256)
x = F.max_pool1d(x, x.size(2)).squeeze(2)
return x
"""
nn中的成员⽐如nn.Conv2d，都是类，可以提取待学习的参数。当我们在定义⽹络层的时候，层内如果有需要学习的参数，那么我们就要⽤nn组件；
nn.functional⾥的成员都是函数，只是完成⼀些功能，⽐如池化，整流线性函数，不保存参数，所以如果某⼀层只是单纯完成⼀些简单的功能，没有
待学习的参数，那么就⽤nn.funcional⾥的组件
"""
# 后续数据预处理时候，x被处理成是⼀个tuple，其形状是: (data, length).  其中data(b_size, seq_len),  length(batch_size)
# x[0]:(b_size, seq_len)
def forward(self, x):
out = bedding(x[0]) # x[0]:(b_size, seq_len, embed_dim)    x[1]是⼀维的tensor，表⽰batch_size个元素的长度        out = out.unsqueeze(1) # (b_size, 1, seq_len, embed_dim)
out = torch.cat([v_and_pool(out, conv) for conv vs], 1) # (b, channel * 3) == (b, 256 * 3)
out = self.dropout(out)
out = self.fc(out) # out(b, num_classes)
return out
# 泽维尔正态分布 xavier_normal_：均值为0，标准差为根号(2/(输⼊+输出数))的正态分布，默认gain=1
# kaiming正态分布 kaiming_normal_：均值为0，标准差为根号(2/(1+a²)f_in)的正态分布，默认a=0
# 初始化时候要避开预训练词向量
def init_network(model, method='xavier', exclude='embedding', seed=123):
for name, w in model.named_parameters():
if exclude not in name: # 对于embedding，保留预训练的embedding
if 'weight' in name:
if method == 'xavier':
nn.init.xavier_normal_(w)
elif method == 'kaiming':
nn.init.kaiming_normal_(w)
else:
al_(w)
elif 'bias' in name:
stant_(w, 0)
else:
pass
>>>>>>>>>>>>>>>>#
>>>>>>> 训练、测试过程 >>>>>>##
def train(config, model, train_iter, dev_iter, test_iter):
start_time = time.time()
optimizer = torch.optim.Adam(model.parameters(), lr=config.learning_rate)
# 学习率指数衰减，每个epoch：学习率 = gamma * 学习率
# 配合 scheduler.step() 完成学习率衰减
scheduler = torch.optim.lr_scheduler.ExponentialLR(optimizer, gamma=0.9)
total_batch = 0  # 记录进⾏到多少batch
dev_best_loss = float('inf')
last_improve = 0  # 记录上次验证集loss下降的batch数
flag = False  # 记录是否很久没有效果提升
# from tensorboardX import SummaryWriter  记录训练的⽇志
writer = SummaryWriter(log_dir=config.log_path + '/' + time.strftime('%m-%d_%H.%M', time.localtime()))
for epoch in range(config.num_epochs):
print('Epoch [{}/{}]'.format(epoch + 1, config.num_epochs))
scheduler.step() # 学习率衰减
for i, (trains, labels) in enumerate(train_iter): # 每个(train_iter)相当于 -> ((x[b, len], len[b]), labels[b])
outputs = model(trains) # trains[0]:(b_size, seq_len)保存idx的⼆维tensor,  trains[1]:(b_size)表⽰长度的⼀维tensor            # 1.清空梯度 -> 2.计算loss -> 3.反向传播 -> 4.梯度更新
<_grad()
loss = F.cross_entropy(outputs, labels) # outputs(b, num_classes), labels(b)
loss.backward()
optimizer.step()
if total_batch % 100 == 0:
# 每多少轮输出在训练集和验证集上的效果
true = labels.data.cpu()
predic = torch.max(outputs.data, 1)[1].cpu()
train_acc = metrics.accuracy_score(true, predic) # ics.accuracy_score(true, predic) 返回正确的⽐例                dev_acc, dev_loss = evaluate(config, model, dev_iter)
if dev_loss < dev_best_loss:
dev_best_loss = dev_loss
torch.save(model.state_dict(), config.save_path)
improve = '*'
improve = '*'
last_improve = total_batch
else:
improve = ''
time_dif = get_time_dif(start_time)
msg = 'Iter: {0:>6},  Train Loss: {1:>5.2},  Train Acc: {2:>6.2%},  Val Loss: {3:>5.2},  Val Acc: {4:>6.2%},  Time: {5} {6}'                print(msg.format(total_batch, loss.item(), train_acc, dev_loss, dev_acc, time_dif, improve))
writer.add_scalar("loss/train", loss.item(), total_batch)
writer.add_scalar("loss/dev", dev_loss, total_batch)
writer.add_scalar("acc/train", train_acc, total_batch)
writer.add_scalar("acc/dev", dev_acc, total_batch)
total_batch += 1
if total_batch - last_improve > quire_improvement:
# 验证集loss超过1000batch没下降，结束训练
print("No optimization for a long time, ")
flag = True
break
if flag:
break
writer.close()
test(config, model, test_iter)
def test(config, model, test_iter):
# test
model.load_state_dict(torch.load(config.save_path))
model.eval()
start_time = time.time()
test_acc, test_loss, test_report, test_confusion = evaluate(config, model, test_iter, test=True)
msg = 'Test Loss: {0:>5.2},  Test Acc: {1:>6.2%}'
print(msg.format(test_loss, test_acc))
print("Precision, Recall ")
print(test_report)
print("")
print(test_confusion)
time_dif = get_time_dif(start_time)
print("Time usage:", time_dif)
def evaluate(config, model, data_iter, test=False):
model.eval() # 关闭dropout
loss_total = 0
predict_all = np.array([], dtype=int)
labels_all = np.array([], dtype=int)
_grad(): # 将outputs从计算图中排除
for texts, labels in data_iter:
outputs = model(texts)
loss = F.cross_entropy(outputs, labels)
loss_total += loss
labels = labels.data.cpu().numpy()
predic = torch.max(outputs.data, 1)[1].cpu().numpy()
# append拼接数组和数值，也可以拼接两个数组，数组必须是np.array()类型，拼接成⼀维的np.array()
labels_all = np.append(labels_all, labels)
predict_all = np.append(predict_all, predic)
acc = metrics.accuracy_score(labels_all, predict_all)
if test:
# classification_report⽤于显⽰每个class上的各项指标结果，包括precision, recall, f1-score
report = metrics.classification_report(labels_all, predict_all, target_names=config.class_list, digits=4)
# 混淆矩阵
confusion = fusion_matrix(labels_all, predict_all)
return acc, loss_total / len(data_iter), report, confusion
return acc, loss_total / len(data_iter)
>>>>>>>>>>>>>>>>>##
>>>>>>#### 数据预处理过程 >>>>>>>###
seg = pkuseg.pkuseg() # 分词⼯具，通过 seg.cut(x) 进⾏分词。可以换成 jieba 分词
MAX_VOC_SIZE = 500000  # 词表长度限制
UNK, PAD = '<UNK>', '<PAD>'  # 未知字，padding符号
def build_vocab(file_path, tokenizer, max_size, min_freq):
vocab_dic = {}
with open(file_path, 'r', encoding='UTF-8') as f:
for line in tqdm(f):
lin = line.strip()
if not lin:
continue
content = lin.split('\t')[0]
for word in tokenizer(content):
vocab_dic[word] = (word, 0) + 1
# vocab_list = sorted([_ for _ in vocab_dic.items() if _[1] >= min_freq], key=lambda x: x[1], reverse=True)[:max_size]        vocab_list = sorted([_ for _ in vocab_dic.items() if _[1] >= min_freq], key=lambda x: x[1], reverse=True)
vocab_dic = {word_count[0]: idx for idx, word_count in enumerate(vocab_list)}
vocab_dic.update({UNK: len(vocab_dic), PAD: len(vocab_dic) + 1})
return vocab_dic
eval是做什么的
def build_dataset(config, ues_word):
if ues_word:
# tokenizer = lambda x: x.split(' ')  # 以空格隔开，word-level
tokenizer = lambda x: seg.cut(x) # 分词
else:
tokenizer = lambda x: [y for y in x]  # char-level
if ists(config.vocab_path):
vocab = pkl.load(open(config.vocab_path, 'rb'))
else:
vocab = build_ain_path, tokenizer=tokenizer, max_size=MAX_VOC_SIZE, min_freq=1)
pkl.dump(vocab, open(config.vocab_path, 'wb'))
print(f"Vocab size: {len(vocab)}")
def load_dataset(path, pad_size=32):
contents = []
with open(path, 'r', encoding='UTF-8') as f:
for line in tqdm(f):
lin = line.strip()
if not lin:
continue
content, label = lin.split('\t')
words_line = []
token = tokenizer(content)
seq_len = len(token)
if pad_size:
if len(token) < pad_size:
else:
token = token[:pad_size]
seq_len = pad_size
# word to id
for word in token:
words_line.(word, (UNK)))
contents.append((words_line, int(label), seq_len))
return contents  # [([...], 0, len), ([...], 1, len), ...]
train = load_ain_path, config.pad_size)
dev = load_dataset(config.dev_path, config.pad_size)
test = load_st_path, config.pad_size)
return vocab, train, dev, test

版权声明：本站内容均来自互联网，仅供演示用，请勿用于商业和其他非法用途。如果侵犯了您的权益请与我们联系QQ:729038198，我们将在24小时内删除。

使用PaddleNLP识别垃圾邮件(二):用BERT做中文邮件内容分类
« 上一篇
matlab如何调整输出位数,转:如何控制matlab中的数据输出格式?
下一篇 »

发表评论

Copyright © 2024 baidu.com
Powered By 688IT编程网 豫ICP备2022007602号 豫公网安备41160202000602