溫馨提示×
您好，登錄后才能下訂單哦！
點擊 登錄注冊 即表示同意《億速云用戶服務條款》
您好，登錄后才能下訂單哦！
使用循環神經網絡(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.com
import sys
reload(sys)
sys.setdefaultencoding('utf8')
import collections
import settings
import utils
import numpy as np
def 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,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, ]
index = int(np.searchsorted(cumsum_rate, 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.com
import tensorflow as tf
import functools
import settings
HIDDEN_SIZE = 128
NUM_LAYERS = 2
def 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
@doublewrap
def define_scope(function, scope=None, *args, **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, *args, **kwargs):
setattr(self, attribute, function(self))
return getattr(self, attribute)
return decorator
class 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, dtype=tf.float32, 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, 0.5), tf.equal(self.label, 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.com import numpy as np import settings class 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.com
import settings
import tensorflow as tf
import models
import dataset
import os
BATCH_SIZE = settings.BATCH_SIZE
# 數據
x = tf.placeholder(tf.int32, [None, None])
# 標簽
y = tf.placeholder(tf.float32, [None, 1])
# emb層的dropout保留率
emb_keep = tf.placeholder(tf.float32)
# rnn層的dropout保留率
rnn_keep = tf.placeholder(tf.float32)
# 創建一個模型
model = models.Model(x, y, emb_keep, 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.com
import settings
import tensorflow as tf
import models
import dataset
import os
import time
# 為了在使用GPU訓練的同時，使用CPU進行驗證
os.environ['CUDA_VISIBLE_DEVICES'] = ''
BATCH_SIZE = settings.BATCH_SIZE
# 數據
x = tf.placeholder(tf.int32, [None, None])
# 標簽
y = tf.placeholder(tf.float32, [None, 1])
# emb層的dropout保留率
emb_keep = tf.placeholder(tf.float32)
# rnn層的dropout保留率
rnn_keep = tf.placeholder(tf.float32)
# 創建一個模型
model = models.Model(x, y, emb_keep, 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.com
import settings
def read_vocab_list():
"""
讀取詞匯表
:return:由詞匯表中所有單詞組成的列表
"""
with open(settings.VOCAB_PATH, 'r') as f:
vocab_list = f.read().strip().split('\n')
return vocab_list
def read_word_to_id_dict():
"""
生成一個單詞到編號的映射
:return:單詞到編號的字典
"""
vocab_list = read_vocab_list()
word2id = dict(zip(vocab_list, range(len(vocab_list))))
return word2id
def read_id_to_word_dict():
"""
生成一個編號到單詞的映射
:return:編號到單詞的字典
"""
vocab_list = read_vocab_list()
id2word = dict(zip(range(len(vocab_list)), vocab_list))
return id2word
def 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,0.1]，訓練2000個batch_size=64的mini_batch時，模型在各數據集上的acc表現大致如下：
訓練集 0.95
開發集 0.79
測試集 0.80
更多
轉行做機器學習，要學的還很多，文中如有錯誤紕漏之處，懇請諸位大佬拍磚指教…
項目GitHub地址：https://github.com/AaronJny/emotional_classification_with_rnn
以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支持億速云。
免責聲明：本站發布的內容（圖片、視頻和文字）以原創、轉載和分享為主，文章觀點不代表本網站立場，如果涉及侵權請聯系站長郵箱：is@yisu.com進行舉報，并提供相關證據，一經查實，將立刻刪除涉嫌侵權內容。
