[人工智能]CNN实现文本情感分类

前言

作为一个OIer，原本以为入门一些语言和框架不是一件难事，然而真正去实践了才知道python以及一系列库的函数和库实在是用处繁多，每次看到一句话就要在网上查很久它的用处，并且描述的语言通常有点晦涩，尤其是对于只是自学了一下python的我来说，在这里通过举一个例子来简单实现一份深度学习的代码，并且用并不严谨但易懂的语言解释，同时记录一下实验结果和感想。

门槛要求，基本学习过神经网络的原理（指上了不明所以的AI导论的课），对于实现方法有所困惑（指什么都不会写）

问题

这个任务来自THU人工智能导论第二次大作业，将一个已经完成了分词的中文句子进行分析，将它分类为正面或负面的情感

下发文件包括：

• train.txt,test.txt,validation.txt，每个文件代表一组数据，每个文件内由若干行组成

  • 每行开头是0/1表示负面或正面，之后是分词后的一句话，用空格分词
  • train.txt有接近20000句，validation.txt有接近5000句，test.txt有接近400句

• wiki_word2vec_50.bin，词向量包，输入后可以得到每个词的长度为50的向量，可以用gensim导入，详见代码

  词向量：我们要判别两个词语是否相似，可以通过训练出词向量，对于每个词用向量表示，如果两个向量差越小，就代表着两个词越接近。

  wiki_word2vec_50.bin里包含了十几万个词语，虽然并不完整，但只有少部分的词语没有，大体上并不影响训练的结果

  当然，也可以有更长的词向量，能够更加精确地描述一个词语的信息

方法

• 考虑使用CNN模型

• 对于一个句子，大小为$L\times 50$（50是词向量大小），有若干个大小为$k\times 50$的卷积核，对于一个卷积核可以将句子变成$(L?k+1)\times 1$的矩阵，再进行$max\_pooling$（最大池化，就是将这个矩阵变成一个值，为矩阵中的最大值）以及激活函数$relu$（对0取max），这样就能得到一个大小为卷积核个数$c$的向量，最后进行$c\to 2$的一个全连接，以及softmax激活函数得到$0,1$的概率

• 好的，你已经知道CNN是什么了，你可以轻松写出代码了！（）

• 然而，如果你不想在BP算法的时候手算梯度，你应该至少要会一种深度学习框架，通俗的说就是一个你能想到的常见的操作都帮你封装好了的模板库，你只需要import一下它就可以帮你写好代码（bushi）！

• 我用的是pytorch

Pytorch

一个流行的模板库（深度学习框架），里面包含了非常多的函数，如果你不想将pytorch官方的手册看一遍，你最好找一份朴素的实现从最简单的开始做起

在这里，仅仅介绍在本次实验之前你需要知道的关于Pytorch的知识

1. Pytorch中已经封装好的操作都是对于其自带的变量类型torch.tensor来做的，tensor是一个类似numpy的数组，你可以通过tensor.ones(),tensor.zeros(),tensor.randn()来新建一个 tensor，你甚至可以将一个numpy或list直接变成tensor
2. 如何构建神经网络？让我们用代码来一点点解释！

实现

import torch #你已经下载好了pytorch,import它就完事了
import gensim #便于我们导入word2vec50.bin词向量
import math 
import torchvision #在某些时候你可用它来指表，直观显示你的训练情况，不过由于这里是入门教程就不用了
import torch.nn as nn #便于简化代码，下同
import torch.nn.functional as F
import torch.optim as optim 
from torch.utils.data import DataLoader,Dataset

train_input_name = 'train_shuffle.txt' 
test_input_name = 'test.txt'
all_input_name = 'validation.txt'
# train是我们的训练数据（很大），test是一组小数据，跑起来比较快，便于我们实时观测当前训练结果，validation是一组不大不小的数据，相对test运行次数更少，它可以帮助我们更加精确地得到训练的结果。需要注意的是test和validation是不参与训练的，避免出现过拟合的情况（即网络仅仅在训练数据下优秀），仅仅作为测试使用

# 读取word2vec
print("Loading word2vec...")
model = gensim.models.KeyedVectors.load_word2vec_format('wiki_word2vec_50.bin', binary=True)
words = model.key_to_index.keys() # gensim用法，model是一个dictionary，这样就成功提取出了词语库


train_batch_size = 64 # 批量训练，这个变量表示一批的大小，提高训练的效率
test_batch_size = 1000 # 测试数据一批的大小
 
vec_size = 50 # 词向量的大小
learning_rate = 0.005 # 训练率，是重要的超参数之一，过高时容易左右横跳，过低时训练太慢
n_epoch = 100000 # 将train训练多少次，这里设了个很大的数（图一乐），使得它停不下来
momentum = 0.5 # 表示对于一次梯度下降，要继承上一次下降方向的多少倍
random_seed = 114514 
torch.manual_seed(random_seed) # 随机种子，生成初始网络

limit_size = 1e9

# 读入部分
def readdata(name):
    file = open(name,'r',encoding='utf-8') #由于是中文数据，要加utf-8
    data = file.readlines() # data是一个字符串构成的list，分别对应每一行
    List = [] # 第i个元素储存第i个句子的词向量矩阵
    Target = [] # 第i个元素储存第i个句子的结果
    cnt = 0
    # 这三个不重要
    log_cnt = 1000 
    
    for line in data:
        L = line[2:].split(' ') 
        L[len(L)-1] = L[len(L)-1].strip() # 忽略这个字符串后的空格
        sentence = torch.zeros(len(L),vec_size,dtype=torch.float)
        # 构建了一个len(L)*vec_size大小，数据类型为实数的tensor
        for i in range(len(L)) : # 枚举每一个单词
            if L[i] in words: # 如果单词在单词表里有
                sentence[i:] =torch.tensor(model.get_vector(L[i]))
                # model.get_vector(word)得到word的词向量
        List.append(sentence),Target.append(int(line[0]))

        if cnt % log_cnt == 0 :
            print("Reading Sentence of {}".format(name),cnt,)
        cnt += 1
        if cnt >= limit_size :
            break
    return  List,torch.tensor(Target)

class Mydataset(Dataset):
    def __init__(self, List, Target):
        self.List = List
        self.Target = Target
    def __len__(self):
        return len(self.List)
    def __getitem__(self, index):
        return self.List[index], self.Target[index]

def my_collate(batch):
    data = [item[0] for item in batch]
    target = [item[1] for item in batch]
    target = torch.tensor(target)
    return [data, target]

train_x, train_y = readdata(train_input_name) 
train_ds = Mydataset(train_x, train_y) 
# train_ds是一个list,每一个元素是一个(train_x[i],train_y[i]),Mydataset主要是将长度不同的train_x合起来，详情略，不过只需要超上面的代码即可（比较通用）
train_dataloader = DataLoader(train_ds, batch_size = train_batch_size, shuffle = True, collate_fn = my_collate)
# 含义为将train_ds这个list按照batch_size一组打包，并且随机排列,collate_fn主要是由于上一句train_ds自定义了，所以这里也要自定义（用于将一个batch里的data和target分开），详情略了

#同上，分别输入三组数据
test_x, test_y = readdata(test_input_name)
test_ds = Mydataset(test_x, test_y)
test_dataloader = DataLoader(test_ds, batch_size = test_batch_size, shuffle = True, collate_fn= my_collate)

all_x, all_y = readdata(all_input_name)
all_ds = Mydataset(all_x, all_y)
all_dataloader = DataLoader(all_ds, batch_size = test_batch_size, shuffle = True, collate_fn= my_collate)

# 卷积核的个数（重要的指标，需要自己定义，对网络的训练效果有不小的影响）
ConvList = []
for i in range(100):
    ConvList.append(3)
for i in range(100):
    ConvList.append(4)
for i in range(100):
    ConvList.append(5)

# 构造神经网络Net
class Net(nn.Module): # 从nn.Module继承子类（相当于引用已经给你封装好的东西）
    # 在Net里必须要定义__init__(self)和forward(self,...)
    # __init__是初始化一个类的时候会运行的部分，你需要将需要训练的网络权值提前定义好，通过pytorch已经封装好的一些函数
    # forward(self,...)需要定义如果...作为输入，网络的输出是什么,network(x)就能返回network.forward(x)的结果
    
    def __init__(self):
        super(Net, self).__init__() # 我也不知道有什么用，反正都要加
        self.conv = [] #一组卷积核，用list存
        for size in ConvList : 
            self.conv.append(nn.Conv2d(1, 1,  kernel_size=(size, vec_size)))
            # nn.Conv2d()是一个卷积核，分别表示，输入通道数，输出通道数，卷积核的尺寸为size*vec_size，
            # 定义的时候自动随机初始化
        self.fc2 = nn.Linear(len(ConvList), 2)
        # 定义卷积层，是从卷积核个数到2的卷积层

    def forward(self, x):
        y = torch.zeros(x.size(0),1,len(ConvList))
        # batch_size * 1 * 卷积核个数，表示卷积、池化后排成一排的结果
        for i in range(len(ConvList)):
            z = (self.conv[i])(x) # Conv2d可以看成一个函数，给一个tensor就能出一个卷积后的tensor
            z = F.relu(F.max_pool2d(z, kernel_size=(z.size(2),1))) 
            # max_pool2d表示二维池化，对z池化，池化的核的大小为 句子长度*1，相当于取最大值
            # 得到[batch_size,1,1,1]的tensor
            z = z.view(-1)
            # 拍扁成一维数组
            y[:,0,i] = z
        y = F.softmax(self.fc2(y),dim=2)
        # 必须要加dim=2，表示对第2维softmax(维数从0开始)
        return y.view(-1,2)
    	# 第1维是大小2，第0维是size/2，也是batch_size
network = Net()
optimizer = optim.SGD(network.parameters(), lr=learning_rate, momentum=momentum)
# 对于类Net的实例化，optimizer用来自动算梯度，自动更新网络

# 从目录中提取已经训练过的网络，用于存档和读档
network_state_dict = torch.load('model.pth')
network.load_state_dict(network_state_dict)
optimizer_state_dict = torch.load('optimizer.pth')
optimizer.load_state_dict(optimizer_state_dict)

# 将batch(一个list)变成一个句子长度一样的tensor，这样才能喂到网络里去计算
def expand(batch):
    max_len = 0
    for sen in batch :
        max_len =max(max_len, sen.size(0))
    out = torch.zeros(len(batch), 1, max_len, vec_size)
    for i in range(len(batch)):
        for j in range(batch[i].size(0)):
            out[i, 0, j] = batch[i][j]
    return out

# 定义损失函数为二分类的交叉熵
def loss_fun(pred, tar):
    loss = 0
    for i in range(tar.size(0)):
        loss -= torch.log(pred[i][tar[i]])
        # 注意这里不能用math.log，要用torch里带的才能求梯度
    return loss


log_interval = 30
def train(epoch):
    print('----Epoch {} Start---'.format(epoch))
    loss_sum = 0
    # 这里就是train_dataloader的具体用法了，enumerate可以理解为它的指针列表
    for batch_idx, (x, y) in enumerate(train_dataloader):
        network.train() # 告诉神经网络它正在训练
        optimizer.zero_grad() # 清空梯度（上一次训练的梯度可能会残留）
        x = expand(x) # 将x扩展为一个tensor，由于之前dataloader的定义这里x还是一个list(tensor)
        pred = network(x) # 网络输出结果
        loss = loss_fun(pred, y)/train_batch_size
        loss.backward() # 自动BP，torch好评
        optimizer.step() # 更新网络，torch好评

        # 按时输出以及存档、测试
        if batch_idx % log_interval == 0:
            print('Train Epoch: {} [{:.0f}%]'.format(epoch, 100.0 * batch_idx / len(train_dataloader)))
            torch.save(network.state_dict(), './model.pth')
            torch.save(optimizer.state_dict(), './optimizer.pth')
            test(test_dataloader)
	# 存档
    torch.save(network.state_dict(), './model.pth')
    torch.save(optimizer.state_dict(), './optimizer.pth')

def Accurate(pred, y):
    cnt = 0
    for i in range(y.size(0)):
        if pred[i][0]>pred[i][1]:
            k = 0
        else:
            k = 1
        if k == y[i]:
            cnt += 1
    return cnt

# 测试一下精确度
def test(test_dataloader):
    network.eval() #告诉网络你正在测试
    loss_sum = 0
    AC = 0
    Total = 0
    with torch.no_grad(): # 『请』不要计算梯度
        for batch_idx, (x, y) in enumerate(test_dataloader):
            x = expand(x)
            pred = network(x)
            loss_sum += loss_fun(pred, y).item()
            AC += Accurate(pred, y) # 准确率
            Total += y.size(0)
    print("Test Output : {}    AC rate : {}%".format(loss_sum,100.0*AC/Total))

test(test_dataloader) # 事前测一测
for epoch in range(n_epoch): 
    train(epoch)
    # 历史版本存档
    torch.save(network.state_dict(), './model/model{}.pth'.format(epoch))
    torch.save(optimizer.state_dict(), './model/optimizer{}.pth'.format(epoch))
    print('Result of Epoch ',epoch)
    test(all_dataloader)
    print('\n')

结论

• 卷积核的个数对实验的结果有着至关重要的影响

  1. 在比较少的时候（20个$3\times 3$，30个$2\times 2$，20个$1\times 1$），一开始只能训练到65（正确率），经过50轮epoch才能到73，最后在74收敛
  2. 在比较多的时候（100个$3\times 3$，100个$4\times 4$，100个$5\times 5$），很快就能跑到70，经过70轮之后最好能到77，但是在epoch不断变大，甚至到200轮的时候一度下降到了74、73，出现了过拟合的情况，尽管lost仍然在缓慢降低，但准确率却变低了
  3. 在不多不少的时候也能跑出2的成绩
  4. 在卷积核更多的时候并没有明显的优化，反而是训练速度更慢了

• learning_rate是一个比较重要的参数

  1. 一开始设置到了0.5，网络的训练就完全没有效果了，曾让我一度怀疑是不是哪里写错了

  2. 不同的问题要具体分析，但是0.01和0.001的在正确率上的区别并不明显

• 由于是白嫖机房的机子，并没有NvidaGPU，所以只能暂且不用GPU了

• RNN优化并不明显，速度会更慢

• 问了一下别的同学似乎也就做到80-82的样子，似乎被词向量和数据限制住了上限

感想

• 这是我第一次训练神经网络，从连python都不会到自己写出来（当然少不了查找资料），花费了一个星期的时间，虽然连续开十几个标签页递归式查找资料真的很痛苦，但是当你看到屏幕上不断跳动的Loss Output在不断变低时的喜悦是难以言表的。
• 当然，作为菜鸟在这个方面还有很多不会的东西，上文里甚至可能会有写错了的东西，但不管怎么样这都是一个不断进步的过程。
• 与此同时也是一个我对于人工智能怯魅的过程，其实也没有那么神奇嘛，都是计算机一点点统计出来、算出来的，不过也常常难免为这种统计的有效性而惊讶。

开发者涨薪指南

48位大咖的思考法则、工作方式、逻辑体系