基于循环神经网络(RNN)实现影评情感分类

概述使用循环神经网络(RNN)实现影评情感分类作为对循环神经网络的实践，我用循环神经网络做了个影评情感的分类，即判断影评的感情色彩是正面的，还是负面的。

使用循环神经网络(RNN)实现影评情感分类

作为对循环神经网络的实践，我用循环神经网络做了个影评情感的分类，即判断影评的感情色彩是正面的，还是负面的。

选择使用RNN来做情感分类，主要是因为影评是一段文字，是序列的，而RNN对序列的支持比较好，能够“记忆”前文。虽然可以提取特征词向量，然后交给传统机器学习模型或全连接神经网络去做，也能取得很好的效果，但只从端对端的角度来看的话，RNN无疑是最合适的。

以下介绍实现过程。

一、数据预处理

本文中使用的训练数据集为https://www.cs.cornell.edu/people/pabo/movie-review-data/上的sentence polarity dataset v1.0，包含正负面评论各5331条。可以点击进行下载。

数据下载下来之后需要进行解压，得到rt-polarity.neg和rt-polarity.pos文件，这两个文件是windows-1252编码的，先将它转成unicode处理起来会更方便。

补充一下小知识，当我们打开一个文件，发现乱码，却又不知道该文件的编码是什么的时候，可以使用python的chardet类库进行判断，这里的windows-1252就是使用该类库检测出来的。

在数据预处理部分，我们要完成如下处理过程：

1.转码

即将文件转为unicode编码，方便我们后续 *** 作。读取文件，转换编码，重新写入到新文件即可。不存在技术难点。

2.生成词汇表

读取训练文件，提取出所有的单词，并统计各个单词出现的次数。为了避免低频词的干扰，同时减少模型参数，我们只保留部分高频词，比如这里我只保存出现次数前9999个，同时将低频词标识符<unkown>加入到词汇表中。

3.借助词汇表将影评转化为词向量

单词是没法直接输入给模型的，所以我们需要将词汇表中的每个单词对应于一个编号，将影评数据转化成词向量。方便后面生成词嵌入矩阵。

4.填充词向量并转化为np数组

因为不同评论的长度是不同的，我们要组成batch进行训练，就需要先将其长度统一。这里我选择以最长的影评为标准，对其他较短的影评的空白部分进行填充。然后将其转化成numpy的数组。

5.按比例划分数据集

按照机器学习的惯例，数据集应被划分为三份，即训练集、开发集和测试集。当然，有时也会只划分两份，即只包括训练集和开发集。

这里我划分成三份，训练集、开发集和测试集的占比为[0.8,0.1,0.1]。划分的方式为轮盘赌法，在numpy中可以使用cumsum和searchsorted来简洁地实现轮盘赌法。

6.打乱数据集，写入文件

为了取得更好的训练效果，将数据集随机打乱。为了保证在训练和模型调整的过程中训练集、开发集、测试集不发生改变，将三个数据集写入到文件中，使用的时候从文件中读取。

下面贴上数据预处理的代码，注释写的很细，就不多说了。

# -*- Coding: utf-8 -*-# @Time : 18-3-14 下午2:28# @Author : AaronJny# @Email : Aaron__7@163.comimport sysreload(sys)sys.setdefaultencoding('utf8')import collectionsimport settingsimport utilsimport numpy as npdef create_vocab(): """ 创建词汇表，写入文件中 :return: """ # 存放出现的所有单词 word_List = [] # 从文件中读取数据，拆分单词 with open(settings.NEG_TXT,'r') as f: f_lines = f.readlines() for line in f_lines: words = line.strip().split() word_List.extend(words) with open(settings.POS_TXT,'r') as f: f_lines = f.readlines() for line in f_lines: words = line.strip().split() word_List.extend(words) # 统计单词出现的次数 counter = collections.Counter(word_List) sorted_words = sorted(counter.items(),key=lambda x: x[1],reverse=True) # 选取高频词 word_List = [word[0] for word in sorted_words] word_List = ['<unkown>'] + word_List[:settings.VOCAB_SIZE - 1] # 将词汇表写入文件中 with open(settings.VOCAB_PATH,'w') as f: for word in word_List: f.write(word + '\n')def create_vec(txt_path,vec_path): """ 根据词汇表生成词向量 :param txt_path: 影评文件路径 :param vec_path: 输出词向量路径 :return: """ # 获取单词到编号的映射 word2ID = utils.read_word_to_ID_dict() # 将语句转化成向量 vec = [] with open(txt_path,'r') as f: f_lines = f.readlines() for line in f_lines: tmp_vec = [str(utils.get_ID_by_word(word,word2ID)) for word in line.strip().split()] vec.append(tmp_vec) # 写入文件中 with open(vec_path,'w') as f: for tmp_vec in vec: f.write(' '.join(tmp_vec) + '\n')def cut_train_dev_test(): """ 使用轮盘赌法，划分训练集、开发集和测试集 打乱，并写入不同文件中 :return: """ # 三个位置分别存放训练、开发、测试 data = [[],[],[]] labels = [[],[]] # 累加概率 rate [0.8,0.1] cumsum_rate [0.8,0.9,1.0] rate = np.array([settings.TRAIN_RATE,settings.DEV_RATE,settings.TEST_RATE]) cumsum_rate = np.cumsum(rate) # 使用轮盘赌法划分数据集 with open(settings.POS_VEC,'r') as f: f_lines = f.readlines() for line in f_lines: tmp_data = [int(word) for word in line.strip().split()] tmp_label = [1,] index = int(np.searchsorted(cumsum_rate,np.random.rand(1) * 1.0)) data[index].append(tmp_data) labels[index].append(tmp_label) with open(settings.NEG_VEC,'r') as f: f_lines = f.readlines() for line in f_lines: tmp_data = [int(word) for word in line.strip().split()] tmp_label = [0,np.random.rand(1) * 1.0)) data[index].append(tmp_data) labels[index].append(tmp_label) # 计算一下实际上分割出来的比例 print '最终分割比例',np.array([map(len,data)],dtype=np.float32) / sum(map(len,data)) # 打乱数据，写入到文件中 shuffle_data(data[0],labels[0],settings.TRAIN_DATA) shuffle_data(data[1],labels[1],settings.DEV_DATA) shuffle_data(data[2],labels[2],settings.TEST_DATA)def shuffle_data(x,y,path): """ 填充数据，生成np数组 打乱数据，写入文件中 :param x: 数据 :param y: 标签 :param path: 保存路径 :return: """ # 计算影评的最大长度 maxlen = max(map(len,x)) # 填充数据 data = np.zeros([len(x),maxlen],dtype=np.int32) for row in range(len(x)): data[row,:len(x[row])] = x[row] label = np.array(y) # 打乱数据 state = np.random.get_state() np.random.shuffle(data) np.random.set_state(state) np.random.shuffle(label) # 保存数据 np.save(path + '_data',data) np.save(path + '_labels',label)def decode_file(infile,outfile): """ 将文件的编码从'windows-1252'转为Unicode :param infile: 输入文件路径 :param outfile: 输出文件路径 :return: """ with open(infile,'r') as f: txt = f.read().decode('windows-1252') with open(outfile,'w') as f: f.write(txt)if __name__ == '__main__': # 解码文件 decode_file(settings.ORIGIN_POS,settings.POS_TXT) decode_file(settings.ORIGIN_NEG,settings.NEG_TXT) # 创建词汇表 create_vocab() # 生成词向量 create_vec(settings.NEG_TXT,settings.NEG_VEC) create_vec(settings.POS_TXT,settings.POS_VEC) # 划分数据集 cut_train_dev_test()

二、模型编写

数据处理好之后，开始模型的编写。这里选用循环神经网络，建模过程大致如下：

1.使用embedding构建词嵌入矩阵

在数据预处理中，我们将影评处理成了一个个单词编号构成的向量，也就是说，一条影评，对应于一个由单词编号构成的向量。

将这样的向量进行embedding，即可构建出词嵌入矩阵。在词嵌入矩阵中，每个词由一个向量表示，矩阵中不同向量之间的差异对应于它们表示的词之间的差异。

2.使用LSTM作为循环神经网络的基本单元

长短时记忆网络(LSTM)能够自动完成前文信息的“记忆”和“遗忘”，在循环神经网络中表现良好，已经成为在循环神经网络中大部分人的首选。这里我选择使用LSTM作为循环神经网络的基本单元。

3.对embedding和LSTM进行随机失活(dropout)

为了提高模型的泛化能力，并减少参数，我对embedding层和LSTM单元进行dropout。

4.建立深度为2的深度循环神经网络

为了提高模型的拟合能力，使用深度循环神经网络，我选择的深度为2。

5.给出二分类概率

对深度循环神经网络的最后节点的输出做逻辑回归，通过sigmoID使结果落到0-1之间，代表结果是正类的概率。

损失函数使用交叉熵，优化器选择Adam。

此部分代码如下(注：代码中装饰器的作用为划分命名空间以及保证张量运算只被定义一次)：

# -*- Coding: utf-8 -*-# @Time : 18-3-14 下午2:57# @Author : AaronJny# @Email : Aaron__7@163.comimport tensorflow as tfimport functoolsimport settingsHIDDEN_SIZE = 128NUM_LAYERS = 2def doublewrap(function): @functools.wraps(function) def decorator(*args,**kwargs): if len(args) == 1 and len(kwargs) == 0 and callable(args[0]): return function(args[0]) else: return lambda wrapee: function(wrapee,*args,**kwargs) return decorator@doublewrapdef define_scope(function,scope=None,**kwargs): attribute = '_cache_' + function.__name__ name = scope or function.__name__ @property @functools.wraps(function) def decorator(self): if not hasattr(self,attribute): with tf.variable_scope(name,**kwargs): setattr(self,attribute,function(self)) return getattr(self,attribute) return decoratorclass Model(object): def __init__(self,data,lables,emb_keep,rnn_keep): """ 神经网络模型 :param data:数据 :param lables: 标签 :param emb_keep: emb层保留率 :param rnn_keep: rnn层保留率 """ self.data = data self.label = lables self.emb_keep = emb_keep self.rnn_keep = rnn_keep self.predict self.loss self.global_step self.ema self.optimize self.acc @define_scope def predict(self): """ 定义前向传播过程 :return: """ # 词嵌入矩阵权重 embedding = tf.get_variable('embedding',[settings.VOCAB_SIZE,HIDDEN_SIZE]) # 使用dropout的LSTM lstm_cell = [tf.nn.rnn_cell.DropoutWrapper(tf.nn.rnn_cell.BasicLSTMCell(HIDDEN_SIZE),self.rnn_keep) for _ in  range(NUM_LAYERS)] # 构建循环神经网络 cell = tf.nn.rnn_cell.MultiRNNCell(lstm_cell) # 生成词嵌入矩阵，并进行dropout input = tf.nn.embedding_lookup(embedding,self.data) dropout_input = tf.nn.dropout(input,self.emb_keep) # 计算rnn的输出 outputs,last_state = tf.nn.dynamic_rnn(cell,dropout_input,dtype=tf.float32) # 做二分类问题，这里只需要最后一个节点的输出 last_output = outputs[:,-1,:] # 求最后节点输出的线性加权和 weights = tf.Variable(tf.truncated_normal([HIDDEN_SIZE,1]),dtype=tf.float32,name='weights') bias = tf.Variable(0,name='bias') logits = tf.matmul(last_output,weights) + bias return logits @define_scope def ema(self): """ 定义移动平均 :return: """ ema = tf.train.ExponentialMovingAverage(settings.EMA_RATE,self.global_step) return ema @define_scope def loss(self): """ 定义损失函数，这里使用交叉熵 :return: """ loss = tf.nn.sigmoID_cross_entropy_with_logits(labels=self.label,logits=self.predict) loss = tf.reduce_mean(loss) return loss @define_scope def global_step(self): """ step,没什么好说的，注意指定trainable=False :return: """ global_step = tf.Variable(0,trainable=False) return global_step @define_scope def optimize(self): """ 定义反向传播过程 :return: """ # 学习率衰减 learn_rate = tf.train.exponential_decay(settings.LEARN_RATE,self.global_step,settings.LR_DECAY_STEP,settings.LR_DECAY) # 反向传播优化器 optimizer = tf.train.AdamOptimizer(learn_rate).minimize(self.loss,global_step=self.global_step) # 移动平均 *** 作 ave_op = self.ema.apply(tf.trainable_variables()) # 组合构成训练op with tf.control_dependencIEs([optimizer,ave_op]): train_op = tf.no_op('train') return train_op @define_scope def acc(self): """ 定义模型acc计算过程 :return: """ # 对前向传播的结果求sigmoID output = tf.nn.sigmoID(self.predict) # 真负类 ok0 = tf.logical_and(tf.less_equal(output,0.5),tf.equal(self.label,0)) # 真正类 ok1 = tf.logical_and(tf.greater(output,1)) # 一个数组，所有预测正确的都为True,否则False ok = tf.logical_or(ok0,ok1) # 先转化成浮点型，再通过求平均来计算acc acc = tf.reduce_mean(tf.cast(ok,dtype=tf.float32)) return acc

三、组织数据集

我编写了一个类用于组织数据，方便训练和验证使用。代码很简单，就不多说了，直接贴代码：

# -*- Coding: utf-8 -*-# @Time : 18-3-14 下午3:33# @Author : AaronJny# @Email : Aaron__7@163.comimport numpy as npimport settingsclass Dataset(object): def __init__(self,data_kind=0): """ 生成一个数据集对象 :param data_kind: 决定了使用哪种数据集 0-训练集 1-开发集 2-测试集 """ self.data,self.labels = self.read_data(data_kind) self.start = 0 # 记录当前batch位置 self.data_size = len(self.data) # 样例数 def read_data(self,data_kind): """ 从文件中加载数据 :param data_kind:数据集种类 0-训练集 1-开发集 2-测试集 :return: """ # 获取数据集路径 data_path = [settings.TRAIN_DATA,settings.DEV_DATA,settings.TEST_DATA][data_kind] # 加载 data = np.load(data_path + '_data.npy') labels = np.load(data_path + '_labels.npy') return data,labels def next_batch(self,batch_size): """ 获取一个大小为batch_size的batch :param batch_size: batch大小 :return: """ start = self.start end = min(start + batch_size,self.data_size) self.start = end # 当遍历完成后回到起点 if self.start >= self.data_size: self.start = 0 # 返回一个batch的数据和标签 return self.data[start:end],self.labels[start:end]

四、模型训练

训练过程中，额外 *** 作主要有两个：

1.使用移动平均

我使用移动平均的主要目的是使loss曲线尽量平滑，以及提升模型的泛化能力。

2.使用学习率指数衰减

目的是保证前期学习率足够大，能够快速降低loss，后期学习率变小，能更好地逼近最优解。

当然，就是说说而已，这次的训练数据比较简单，学习率衰减发挥的作用不大。

训练过程中，定期保存模型，以及checkpoint。这样可以在训练的同时，在验证脚本中读取最新模型进行验证。

此部分具体代码如下：

# -*- Coding: utf-8 -*-# @Time : 18-3-14 下午4:41# @Author : AaronJny# @Email : Aaron__7@163.comimport settingsimport tensorflow as tfimport modelsimport datasetimport osBATCH_SIZE = settings.BATCH_SIZE# 数据x = tf.placeholder(tf.int32,[None,None])# 标签y = tf.placeholder(tf.float32,1])# emb层的dropout保留率emb_keep = tf.placeholder(tf.float32)# rnn层的dropout保留率rnn_keep = tf.placeholder(tf.float32)# 创建一个模型model = models.Model(x,rnn_keep)# 创建数据集对象data = dataset.Dataset(0)saver = tf.train.Saver()with tf.Session() as sess: # 全局初始化 sess.run(tf.global_variables_initializer()) # 迭代训练 for step in range(settings.TRAIN_TIMES): # 获取一个batch进行训练 x,y = data.next_batch(BATCH_SIZE) loss,_ = sess.run([model.loss,model.optimize],{model.data: x,model.label: y,model.emb_keep: settings.EMB_KEEP_PROB,model.rnn_keep: settings.RNN_KEEP_PROB}) # 输出loss if step % settings.SHOW_STEP == 0: print 'step {},loss is {}'.format(step,loss) # 保存模型 if step % settings.SAVE_STEP == 0: saver.save(sess,os.path.join(settings.CKPT_PATH,settings.MODEL_name),model.global_step)

五、验证模型

加载最新模型进行验证，通过修改数据集对象的参数可以制定训练/开发/测试集进行验证。

加载模型的时候，使用移动平均的影子变量覆盖对应变量。

代码如下：

# -*- Coding: utf-8 -*-# @Time : 18-3-14 下午5:09# @Author : AaronJny# @Email : Aaron__7@163.comimport settingsimport tensorflow as tfimport modelsimport datasetimport osimport time# 为了在使用GPU训练的同时，使用cpu进行验证os.environ['CUDA_VISIBLE_DEVICES'] = ''BATCH_SIZE = settings.BATCH_SIZE# 数据x = tf.placeholder(tf.int32,rnn_keep)# 创建一个数据集对象data = dataset.Dataset(1) # 0-训练集 1-开发集 2-测试集# 移动平均变量restore_variables = model.ema.variables_to_restore()# 使用移动平均变量进行覆盖saver = tf.train.Saver(restore_variables)with tf.Session() as sess: while True: # 加载最新的模型 ckpt = tf.train.get_checkpoint_state(settings.CKPT_PATH) saver.restore(sess,ckpt.model_checkpoint_path) # 计算并输出acc acc = sess.run([model.acc],{model.data: data.data,model.label: data.labels,model.emb_keep: 1.0,model.rnn_keep: 1.0}) print 'acc is ',acc time.sleep(1)

六、对词汇表进行 *** 作的几个方法

把对词汇表进行 *** 作的几个方法提取出来了，放到了utils.py文件中。

# -*- Coding: utf-8 -*-# @Time : 18-3-14 下午2:44# @Author : AaronJny# @Email : Aaron__7@163.comimport settingsdef read_vocab_List(): """ 读取词汇表 :return:由词汇表中所有单词组成的列表 """ with open(settings.VOCAB_PATH,'r') as f: vocab_List = f.read().strip().split('\n') return vocab_Listdef read_word_to_ID_dict(): """ 生成一个单词到编号的映射 :return:单词到编号的字典 """ vocab_List = read_vocab_List() word2ID = dict(zip(vocab_List,range(len(vocab_List)))) return word2IDdef read_ID_to_word_dict(): """ 生成一个编号到单词的映射 :return:编号到单词的字典 """ vocab_List = read_vocab_List() ID2word = dict(zip(range(len(vocab_List)),vocab_List)) return ID2worddef get_ID_by_word(word,word2ID): """ 给定一个单词和字典，获得单词在字典中的编号 :param word: 给定单词 :param word2ID: 单词到编号的映射 :return: 若单词在字典中，返回对应的编号 否则，返回word2ID['<unkown>'] """ if word in word2ID: return word2ID[word] else: return word2ID['<unkown>']

七、对模型进行配置

模型的配置参数大多数都被提取出来，单独放到了settings.py文件中，可以在这里对模型进行配置。

# -*- Coding: utf-8 -*-# @Time : 18-3-14 下午2:44# @Author : AaronJny# @Email : Aaron__7@163.com# 源数据路径ORIGIN_NEG = 'data/rt-polarity.neg'ORIGIN_POS = 'data/rt-polarity.pos'# 转码后的数据路径NEG_TXT = 'data/neg.txt'POS_TXT = 'data/pos.txt'# 词汇表路径VOCAB_PATH = 'data/vocab.txt'# 词向量路径NEG_VEC = 'data/neg.vec'POS_VEC = 'data/pos.vec'# 训练集路径TRAIN_DATA = 'data/train'# 开发集路径DEV_DATA = 'data/dev'# 测试集路径TEST_DATA = 'data/test'# 模型保存路径CKPT_PATH = 'ckpt'# 模型名称MODEL_name = 'model'# 词汇表大小VOCAB_SIZE = 10000# 初始学习率LEARN_RATE = 0.0001# 学习率衰减LR_DECAY = 0.99# 衰减频率LR_DECAY_STEP = 1000# 总训练次数TRAIN_TIMES = 2000# 显示训练loss的频率SHOW_STEP = 10# 保存训练模型的频率SAVE_STEP = 100# 训练集占比TRAIN_RATE = 0.8# 开发集占比DEV_RATE = 0.1# 测试集占比TEST_RATE = 0.1# BATCH大小BATCH_SIZE = 64# emb层dropout保留率EMB_KEEP_PROB = 0.5# rnn层dropout保留率RNN_KEEP_PROB = 0.5# 移动平均衰减率EMA_RATE = 0.99

八、运行模型

至此，模型构建完成。模型的运行步骤大致如下：

1.确保数据文件放在了对应路径中，运行python process_data对数据进行预处理。

2.运行python train.py对模型进行训练，训练好的模型会自动保存到对应的路径中。

3.运行python eval.py读取保存的最新模型，对训练/开发/测试集进行验证。

我简单跑了一下，由于数据集较小，模型的泛化能力不是很好。

当训练集、开发集、测试集的分布为[0.8,0.1]，训练2000个batch_size=64的mini_batch时，模型在各数据集上的acc表现大致如下：

训练集 0.95

开发集 0.79

测试集 0.80

更多

转行做机器学习，要学的还很多，文中如有错误纰漏之处，恳请诸位大佬拍砖指教…

项目GitHub地址：https://github.com/AaronJny/emotional_classification_with_rnn

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持编程小技巧。

您可能感兴趣的文章:基于循环神经网络(RNN)的古诗生成器Python实现的递归神经网络简单示例Python基于numpy灵活定义神经网络结构的方法Python实现的人工神经网络算法示例【基于反向传播算法】python机器学习之神经网络（一） 总结

以上是内存溢出为你收集整理的基于循环神经网络(RNN)实现影评情感分类全部内容，希望文章能够帮你解决基于循环神经网络(RNN)实现影评情感分类所遇到的程序开发问题。

如果觉得内存溢出网站内容还不错，欢迎将内存溢出网站推荐给程序员好友。