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

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

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

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