⽂本分类（六）：pytorch实现DPCNN
⼀、简介
ACL2017年中，腾讯AI-lab提出了Deep Pyramid Convolutional Neural Networks for Text Categorization(DPCNN)。论⽂中提出了⼀种基于word-level级别的⽹络-DPCNN，由于上⼀篇⽂章介绍的TextCNN 不能通过卷积获得⽂本的长距离依赖关系，⽽论⽂中DPCNN通过不断加深⽹络，可以抽取长距离的⽂本依赖关系。实验证明在不增加太多计算成本的情况下，增加⽹络深度就可以获得最佳的准确率。
DPCNN结构
究竟是多么⽜逼的⽹络呢？我们下⾯来窥探⼀下模型的芳容。
DPCNN结构细节
模型是如何通过加深⽹络来捕捉⽂本的长距离依赖关系的呢？下⾯我们来⼀⼀道来。为了更加简单的解释DPCNN，这⾥我先不解释是什么是Region embedding，我们先把它当作word embedding。
等长卷积
⾸先交代⼀下卷积的的⼀个基本概念。⼀般常⽤的卷积有以下三类：
假设输⼊的序列长度为n，卷积核⼤⼩为m，步长(stride)为s，输⼊序列两端各填补p个零(zero padding)，那么该卷积层的输出序列为(n-
m+2p)/s+1。
(1) 窄卷积(narrow convolution): 步长s=1，两端不补零，即p=0，卷积后输出长度为n-m+1。
(2) 宽卷积(wide onvolution) ：步长s=1，两端补零p=m-1，卷积后输出长度 n+m-1。人教版七年级下英语
(3) 等长卷积(equal-width convolution): 步长s=1，两端补零p=(m-1)/2，卷积后输出长度为n。如下图所⽰，左右两端同时补零p=1,s=3。
池化
那么DPCNN是如何捕捉长距离依赖的呢？这⾥我直接引⽤⽂章的⼩标题——Downsampling with the number of feature maps fixed。
作者选择了适当的两层等长卷积来提⾼词位embedding的表⽰的丰富性。然后接下来就开始 Downsampling (池化)。再每⼀个卷积块(两层的等长卷积)后，使⽤⼀个size=3和stride=2进⾏maxpooling进⾏池化。序列的长度就被压缩成了原来的⼀半。其能够感知到的⽂本⽚段就⽐之前长了⼀倍。
例如之前是只能感知3个词位长度的信息，经过1/2池化层后就能感知6个词位长度的信息啦，这时把1/2池化层和size=3的卷积层组合起来如图所⽰
固定feature maps(filters)的数量
为什么要固定feature maps的数量呢?许多模型每当执⾏池化操作时，增加feature maps的数量，导致总计算复杂度是深度的函数。与此相反，作者对feature map的数量进⾏了修正，他们实验发现增加feature map的数量只会⼤⼤增加计算时间，⽽没有提⾼精度。
另外，⼣⼩瑶⼩姐姐在知乎也详细的解释了为什么要固定feature maps的数量。有兴趣的可以去知乎搜⼀搜，讲的⾮常透彻。
固定了feature map的数量，每当使⽤⼀个size=3和stride=2进⾏maxpooling进⾏池化时，每个卷积层的计算时间减半(数据⼤⼩减半)，从⽽形成⼀个⾦字塔。
这就是论⽂题⽬所谓的 Pyramid。
cri是什么意思好啦，看似问题都解决了，⽬标成功达成。剩下的我们就只需要重复的进⾏等长卷积+等长卷积+使⽤⼀个size=3和stride=2进⾏maxpooling 进⾏池化就可以啦，DPCNN就可以捕捉⽂本的长距离依赖啦！
Shortcut connections with pre-activation
但是！如果问题真的这么简单的话，深度学习就⼀下⼦少了超级多的难点了。
(1) 初始化CNN的时，往往各层权重都初始化为很⼩的值，这导致了最开始的⽹络中，后续⼏乎每层的输⼊都是接近0，这时的⽹络输出没有意义；
(2) ⼩权重阻碍了梯度的传播，使得⽹络的初始训练阶段往往要迭代好久才能启动；
(3) 就算⽹络启动完成，由于深度⽹络中仿射矩阵（每两层间的连接边）近似连乘，训练过程中⽹络也⾮常容易发⽣梯度爆炸或弥散问题。
当然，上述这⼏点问题本质就是梯度弥散问题。那么如何解决深度CNN⽹络的梯度弥散问题呢？当然是膜⼀下何恺明⼤神，然后把ResNet 的精华拿来⽤啦！ ResNet中提出的shortcut-connection/ skip-connection/ residual-connection（残差连接）就是⼀种⾮常简单、合理、有效的解决⽅案。
类似地，为了使深度⽹络的训练成为可能，作者为了恒等映射，所以使⽤加法进⾏shortcut connections，即z+f(z)，其中 f ⽤的是两层的等长卷积。这样就可以极⼤的缓解了梯度消失问题。
另外，作者也使⽤了 pre-activation，这个最初在何凯明的“Identity Mappings in Deep Residual Networks上提及，有兴趣的⼤家可以看看这个的原理。直观上，这种“线性”简化了深度⽹络的训练，类似于LSTM中constant error carouls的作⽤。⽽且实验证明 pre-activation优于post-activation。
整体来说，巧妙的结构设计，使得这个模型不需要为了维度匹配问题⽽担忧。
Region embedding
同时DPCNN的底层貌似保持了跟TextCNN⼀样的结构，这⾥作者将TextCNN的包含多尺⼨卷积滤波器的卷积层的卷积结果称之为Region embedding，意思就是对⼀个⽂本区域/⽚段（⽐如3gram）进⾏⼀组卷积操作后⽣成的embedding。
另外，作者为了进⼀步提⾼性能，还使⽤了tv-embedding (two-views embedding)进⼀步提⾼DPCNN
的accuracy。
上述介绍了DPCNN的整体架构，可见DPCNN的架构之精美。本⽂是在原始论⽂以及知乎上的⼀篇⽂章的基础上进⾏整理。本⽂可能也会有很多错误，如果有错误，欢迎⼤家指出来！建议⼤家为了更好的理解DPCNN ，看⼀下原始论⽂和参考⾥⾯的知乎。
⼆、pytorch实现
1、DPCNN.py
# coding: UTF-8
import torch
as nn
functional as F
import numpy as np
class Config(object):
"""配置参数"""
def__init__(lf, datat, embedding):
lf.dev_path = datat + '/'# 验证集
lf.class_list = [x.strip() for x in open(
datat + '/', encoding='utf-8').readlines()] # 类别名单
lf.vocab_path = datat + '/data/vocab.pkl'# 词表
lf.save_path = datat + '/saved_dict/' + lf.model_name + '.ckpt'# 模型训练结果
lf.log_path = datat + '/log/' + lf.model_name
山东外事翻译学院lf.embedding_pretrained = sor(
np.load(datat + '/data/' + embedding)["embeddings"].astype('float32'))\
if embedding != 'random'el None # 预训练词向量
lf.device = torch.device('cuda'if torch.cuda.is_available() el'cpu') # 设备
lf.dropout = 0.2 # 随机失活
lf.n_vocab = 0 # 词表⼤⼩，在运⾏时赋值
lf.num_epochs = 20 # epoch数
lf.batch_size = 128 # mini-batch⼤⼩
lf.pad_size = 32 # 每句话处理成的长度(短填长切)
lf.learning_rate = 1e-3 # 学习率
bedding_pretrained is not None el 300 # 字向量维度
lf.num_filters = 250 # 卷积核数量(channels数)
'''Deep Pyramid Convolutional Neural Networks for Text Categorization'''
class Model(nn.Module):
def__init__(lf, config):
super(Model, lf).__init__()
bedding_pretrained is not None:
lf.max_pool = nn.MaxPool2d(kernel_size=(3, 1), stride=2)
lf.padding1 = nn.ZeroPad2d((0, 0, 1, 1)) # top bottom
lf.padding2 = nn.ZeroPad2d((0, 0, 0, 1)) # bottom
lf.fc = nn.Linear(config.num_filters, config.num_class)
def forward(lf, x):
x = x[0]
x = lf.embedding(x)
x = x.unsqueeze(1) # [batch_size, 250, q_len, 1]
x = lf.conv_region(x) # [batch_size, 250, q_len-3+1, 1]
x = lf.padding1(x) # [batch_size, 250, q_len, 1]
x = lf.relu(x)
x = lf.conv(x) # [batch_size, 250, q_len-3+1, 1]
x = lf.padding1(x) # [batch_size, 250, q_len, 1]
x = lf.relu(x)
x = lf.conv(x) # [batch_size, 250, q_len-3+1, 1]
while x.size()[2] > 2:
x = lf._block(x)
x = x.squeeze() # [batch_size, num_filters(250)]
x = lf.fc(x)
return x
def _block(lf, x):
x = lf.padding2(x)
px = lf.max_pool(x)
x = lf.padding1(px)
x = F.relu(x)
statisticsx = lf.conv(x)
x = lf.padding1(x)
x = F.relu(x)国家励志奖学金范文
x = lf.conv(x)
x = x + px
return x
2、run.py
# coding: UTF-8
import time
import torch
import numpy as np
from importlib import import_module
from utils import build_datat, build_iterator, get_time_dif
from train_eval import train, init_network
import argpar
parr = argpar.ArgumentParr(description='Text Classification')
parr.add_argument('--model', default = "DPCNN", type=str, help='choo a model: DPCNN, BERT')
parr.add_argument('--embedding', default='pre_trained', type=str, help='random or pre_trained') parr.add_argument('--word', default=Fal, type=bool, help='True for word, Fal for char')
args = parr.par_args()
if__name__ == '__main__':
datat = 'gongqing'# 数据集
# 搜狗新闻:embedding_SougouNews.npz, 腾讯:embedding_Tencent.npz, 随机初始化:random
embedding = 'embedding_SougouNews.npz'
bedding == 'random':
embedding = 'random'
model_name = del # DPCNN, Transformer
x = import_module('models.' + model_name)
config = x.Config(datat, embedding)
np.random.ed(1)
torch.manual_ed(1)
torch.cuda.manual_ed_all(1)
torch.backends.cudnn.deterministic = True # 保证每次结果⼀样
start_time = time.time()
print("")
vocab, train_data, dev_data, test_data = build_datat(config, args.word)
train_iter = build_iterator(train_data, config)
dev_iter = build_iterator(dev_data, config)
test_iter = build_iterator(test_data, config)
time_dif = get_time_dif(start_time)
print("Time usage:", time_dif)
# train
config.n_vocab = len(vocab)
model = x.Model(config).to(config.device)
init_network(model)
print(model.parameters)
train(config, model, train_iter, dev_iter, test_iter)
3、train_eval.py
# coding: UTF-8
kelly hu
import numpy as np
import torch安吉里之战
as nn
functional as F
from sklearn import metrics
import time
from utils import get_time_dif
from tensorboardX import SummaryWriter
# 权重初始化，默认xavier
def init_network(model, method='xavier', exclude='embedding', ed=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)
el:
al_(w)
elif'bias'in name:
stant_(w, 0)
el:
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 = torch.optim.lr_scheduler.ExponentialLR(optimizer, gamma=0.9)
total_batch = 0 # 记录进⾏到多少batch
dev_best_loss = float('inf')
last_improve = 0 # 记录上次验证集loss下降的batch数
flag = Fal # 记录是否很久没有效果提升
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))
vtech# scheduler.step() # 学习率衰减
for i, (trains, labels) in enumerate(train_iter):
outputs = model(trains)
<_grad()
loss = F.cross_entropy(outputs, labels)
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)
dev_acc, dev_loss = evaluate(config, model, dev_iter)
if dev_loss < dev_best_loss:
dev_best_loss = dev_losssuffer from
torch.save(model.state_dict(), config.save_path)
improve = '*'
last_improve = total_batch
el:
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.clo()
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=Fal):
model.eval()
loss_total = 0
predict_all = np.array([], dtype=int)
labels_all = np.array([], dtype=int)
_grad():
for texts, labels in data_iter:
outputs = model(texts)
