TensorFlow实现RNN循环神经网络

import numpy as np 
 
# 定义RNN的参数。 
X = [1,2] 
state = [0.0, 0.0] 
w_cell_state = np.asarray([[0.1, 0.2], [0.3, 0.4]]) 
w_cell_input = np.asarray([0.5, 0.6]) 
b_cell = np.asarray([0.1, -0.1]) 
w_output = np.asarray([[1.0], [2.0]]) 
b_output = 0.1 
 
# 执行前向传播过程。 
for i in range(len(X)): 
  before_activation = np.dot(state, w_cell_state) + X[i] * w_cell_input + b_cell 
  state = np.tanh(before_activation) 
  final_output = np.dot(state, w_output) + b_output 
  print ("before activation: ", before_activation) 
  print ("state: ", state) 
  print ("output: ", final_output) 

import tensorflow as tf 
 
 
# 定义一个LSTM结构 
lstm = rnn_cell.BasicLSTMCell(lstm_hidden_size) 
 
# 将LSTM中的状态初始化为全0数组，每次使用一个batch的训练样本 
state = lstm.zero_state(batch_size,tf.float32) 
 
# 定义损失函数 
loss = 0.0 
 
# 规定一个最大序列长度 
for i in range(num_steps): 
  # 复用之前定义的变量 
  if i > 0: 
    tf.get_variable_scope().reuse_variables() 
  # 将当前输入和前一时刻的状态传入定义的LSTM结构，得到输出和更新后的状态 
  lstm_output, state = lstm(current_input,state) 
   
  # 将当前时刻的LSTM结构的输出传入一个全连接层得到最后的输出。 
  final_output = fully_connectd(lstm_output) 
   
  # 计算当前时刻输出的损失 
  loss += calc_loss(final_output,expected_output) 

import numpy as np 
import tensorflow as tf 
import reader 
  
DATA_PATH = "../datasets/PTB/data" 
HIDDEN_SIZE = 200 # 隐藏层规模 
NUM_LAYERS = 2 # 深层RNN中的LSTM结构的层数 
VOCAB_SIZE = 10000 # 单词标识符个数 
 
LEARNING_RATE = 1.0 # 学习速率 
TRAIN_BATCH_SIZE = 20 # 训练数据大小 
TRAIN_NUM_STEP = 35 # 训练数据截断长度 
 
# 测试时不需要截断 
EVAL_BATCH_SIZE = 1 # 测试数据大小 
EVAL_NUM_STEP = 1 # 测试数据截断长度 
NUM_EPOCH = 2 # 使用训练数据轮数 
KEEP_PROB = 0.5 # 节点不被dropout 
MAX_GRAD_NORM = 5 # 控制梯度膨胀参数 
 
 
# 定义一个类来描述模型结构。 
class PTBModel (object): 
  def __init__(self, is_training, batch_size, num_steps): 
 
    self.batch_size = batch_size 
    self.num_steps = num_steps 
 
    # 定义输入层。 
    self.input_data = tf.placeholder (tf.int32, [batch_size, num_steps]) 
    self.targets = tf.placeholder (tf.int32, [batch_size, num_steps]) 
 
    # 定义使用LSTM结构及训练时使用dropout。 
    lstm_cell = tf.contrib.rnn.BasicLSTMCell (HIDDEN_SIZE) 
    if is_training: 
      lstm_cell = tf.contrib.rnn.DropoutWrapper (lstm_cell, output_keep_prob=KEEP_PROB) 
    cell = tf.contrib.rnn.MultiRNNCell ([lstm_cell] * NUM_LAYERS) 
 
    # 初始化最初的状态。 
    self.initial_state = cell.zero_state (batch_size, tf.float32) 
    embedding = tf.get_variable ("embedding", [VOCAB_SIZE, HIDDEN_SIZE]) 
 
    # 将原本单词ID转为单词向量。 
    inputs = tf.nn.embedding_lookup (embedding, self.input_data) 
 
    if is_training: 
      inputs = tf.nn.dropout (inputs, KEEP_PROB) 
 
    # 定义输出列表。 
    outputs = [] 
    state = self.initial_state 
    with tf.variable_scope ("RNN"): 
      for time_step in range (num_steps): 
        if time_step > 0: tf.get_variable_scope ().reuse_variables () 
        cell_output, state = cell (inputs[:, time_step, :], state) 
        outputs.append (cell_output) 
    output = tf.reshape (tf.concat (outputs, 1), [-1, HIDDEN_SIZE]) 
    weight = tf.get_variable ("weight", [HIDDEN_SIZE, VOCAB_SIZE]) 
    bias = tf.get_variable ("bias", [VOCAB_SIZE]) 
    logits = tf.matmul (output, weight) + bias 
 
    # 定义交叉熵损失函数和平均损失。 
    loss = tf.contrib.legacy_seq2seq.sequence_loss_by_example ( 
      [logits], 
      [tf.reshape (self.targets, [-1])], 
      [tf.ones ([batch_size * num_steps], dtype=tf.float32)]) 
    self.cost = tf.reduce_sum (loss) / batch_size 
    self.final_state = state 
 
    # 只在训练模型时定义反向传播操作。 
    if not is_training: return 
    trainable_variables = tf.trainable_variables () 
 
    # 控制梯度大小，定义优化方法和训练步骤。 
    grads, _ = tf.clip_by_global_norm (tf.gradients (self.cost, trainable_variables), MAX_GRAD_NORM) 
    optimizer = tf.train.GradientDescentOptimizer (LEARNING_RATE) 
    self.train_op = optimizer.apply_gradients (zip (grads, trainable_variables)) 
 
 
 
# 使用给定的模型model在数据data上运行train_op并返回在全部数据上的perplexity值 
def run_epoch(session, model, data, train_op, output_log, epoch_size): 
  total_costs = 0.0 
  iters = 0 
  state = session.run(model.initial_state) 
 
  # 训练一个epoch。 
  for step in range(epoch_size): 
    x, y = session.run(data) 
    cost, state, _ = session.run([model.cost, model.final_state, train_op], 
                    {model.input_data: x, model.targets: y, model.initial_state: state}) 
    total_costs += cost 
    iters += model.num_steps 
 
    if output_log and step % 100 == 0: 
      print("After %d steps, perplexity is %.3f" % (step, np.exp(total_costs / iters))) 
  return np.exp(total_costs / iters) 
 
 
# 定义主函数并执行 
def main(): 
  train_data, valid_data, test_data, _ = reader.ptb_raw_data(DATA_PATH) 
 
  # 计算一个epoch需要训练的次数 
  train_data_len = len(train_data) 
  train_batch_len = train_data_len // TRAIN_BATCH_SIZE 
  train_epoch_size = (train_batch_len - 1) // TRAIN_NUM_STEP 
 
  valid_data_len = len(valid_data) 
  valid_batch_len = valid_data_len // EVAL_BATCH_SIZE 
  valid_epoch_size = (valid_batch_len - 1) // EVAL_NUM_STEP 
 
  test_data_len = len(test_data) 
  test_batch_len = test_data_len // EVAL_BATCH_SIZE 
  test_epoch_size = (test_batch_len - 1) // EVAL_NUM_STEP 
 
  initializer = tf.random_uniform_initializer(-0.05, 0.05) 
  with tf.variable_scope("language_model", reuse=None, initializer=initializer): 
    train_model = PTBModel(True, TRAIN_BATCH_SIZE, TRAIN_NUM_STEP) 
 
  with tf.variable_scope("language_model", reuse=True, initializer=initializer): 
    eval_model = PTBModel(False, EVAL_BATCH_SIZE, EVAL_NUM_STEP) 
 
  # 训练模型。 
  with tf.Session() as session: 
    tf.global_variables_initializer().run() 
 
    train_queue = reader.ptb_producer(train_data, train_model.batch_size, train_model.num_steps) 
    eval_queue = reader.ptb_producer(valid_data, eval_model.batch_size, eval_model.num_steps) 
    test_queue = reader.ptb_producer(test_data, eval_model.batch_size, eval_model.num_steps) 
 
    coord = tf.train.Coordinator() 
    threads = tf.train.start_queue_runners(sess=session, coord=coord) 
 
    for i in range(NUM_EPOCH): 
      print("In iteration: %d" % (i + 1)) 
      run_epoch(session, train_model, train_queue, train_model.train_op, True, train_epoch_size) 
 
      valid_perplexity = run_epoch(session, eval_model, eval_queue, tf.no_op(), False, valid_epoch_size) 
      print("Epoch: %d Validation Perplexity: %.3f" % (i + 1, valid_perplexity)) 
 
    test_perplexity = run_epoch(session, eval_model, test_queue, tf.no_op(), False, test_epoch_size) 
    print("Test Perplexity: %.3f" % test_perplexity) 
 
    coord.request_stop() 
    coord.join(threads) 
 
if __name__ == "__main__": 
  main()