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

# -*- 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()

# -*- 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]

# -*- 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)

# -*- 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>']

# -*- 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