TensorFlow学习手册（四）

前言

本节将描述一下循环神经网络与长短期记忆网络两种模型。

循环神经网络（RNN）

循环神经网络主要用于处理和预测序列化数据的。循环神经网络的来源就是为了刻画一个序列当前的输出和之前信息的关系，它的隐藏层之间的结点是有连接的，隐藏层的输入不仅包括输入层的输出，还包括上一时刻隐藏层的输出。

循环神经网络的经典形式

图1 循环神经网络经典结构示意图

图1展示的是一个典型的循环神经网络，在每一时刻t，循环神经网络会针对该时刻的输入结合当前模型的状态给出一个输出，并更新模型的状态。在每一个时刻，循环神经网络的模块A在读取了 $x_t$ 和 $h_{t-1}$ 之后会产生本时刻的输出 $O_t$ 。由于模块A中的运算和变量在不同时刻是相同的，因此循环神经网络理论上可以被看作是同一神经网络结构被无限复制的结果。循环神经网络是在不同时间位置共享参数，从而能够使用有限的参数处理任意长度的序列。

循环神经网络的展开形式

如果将完整的输入输出序列展开，就可以看到如图2所示的结构。从下图可以看出循环神经网络当前状态 $h_t$ 和当前的输入 $x_t$ 共同决定的。在时刻t，状态 $h_{t-1}$ 浓缩了前面序列 $x_0,x_1,…,x_{t-1}$ 的信息，用于作为输出 $o_t$ 的参考。由于序列的长度可以无限延长，维度有限的h状态不可能将序列的全部信息都保存下来，因此模型必须学习只保留与后面任务 $o_t,o_{t+1},…$ 相关的最重要的信息。

图2 循环神经网络按时间展开后的结构

循环神经网络对长度为N的序列展开之后，可以视为一个有N个中间层的前馈神经网络。这个前馈神经网络没有训练链接，可以使用反向传播算法进行训练。循环神经网络要求每一个时刻都有一个输入，但是不一定每个时刻都需要有输出。

在循环神经网络中，被复制多次的结构称为循环体。图3是一个最简单的循环体结构。循环神经网络中的状态是通过一个向量来表示的，这个向量的维度也称为循环神经网络隐藏层的大小，假设其为n。从下图可以看出循环体中的神经网络的输入有两部分，一部分为上一时刻的状态，一部分为当前时刻的输入样本。对于时间序列数据来说，每一时刻的输入样例可以是该时刻的数值或者是单词向量。

图3 使用单层全连接神经网络作为循环体的循环神经网络结构图

循环神经网络唯一的区别在于因为它每个时刻都有一个输出，所以循环神经网络的总损失为所有时刻（或者部分时刻上）的损失总和。

长短期记忆网络（LSTM）

与单一tanh循环体结构不同，LSTM是一种拥有三个“门”结构的特殊网络结构，如图4所示。为了使循环神经网络更有效的保存长期记忆，“遗忘门”和“输入门”至关重要，它们是LSTM结构的核心。“遗忘门”会根据当前的输入 $x_t$ 和上一时刻输出 $h_{t-1}$ 决定哪一部分记忆需要被遗忘。“输入门”会根据 $x_t$ 和 $h_{t-1}$ 决定哪些信息加入到状态 $c_{t-1}$ 中生成新的状态 $c_t$ 。

图4 LSTM单元结构示意图

图5是用流程图形式表示了LSTM单元的细节。

图5 LSTM单元细节图

循环神经网络的变种

双向循环神经网络

在之前介绍的经典的循环神经网络中，状态的传输都是从前往后单向的，然而在有些问题中，当前时刻的输出不仅和之前的状态有关，也和之后的状态有关。双向循环神经网络是由两个独立的循环神经网络叠加在一起组成的。输出由这两个循环神经网络的输出拼接而成。如下图所示：

图6 双向循环神经网络

深层循环神经网络

深层循环神经网络是循环神经网络的另外一种变种。为了增强模型的表达能力，可以在网络中设置多个循环层，将每层循环网络的输出传给下一层进行处理。

图7 深层循环神经网络

循环神经网络样例程序

以时序预测为例，利用循环神经网络实现对函数sin x取值的预测。在以下程序中每隔SAMPLE_ITERVAL对sin函数进行一次采样，采样得到的序列就是sin函数离散化之后的结果。

# -*- coding: utf-8 -*-
# @Author   : netycc
# @Time     : 2018/11/16 8:44 PM
# @File     : 循环神经网络样例.py
# @Software : PyCharm Community Edition

import numpy as np
import tensorflow as tf

import matplotlib as mpl
# mpl.use("Agg")
from matplotlib import pyplot as plt

HIDDEN_SIZE=30  #LSTM中的隐藏节点数
NUM_LAYERS=2    #LSTM的层数

TIMESTEPS=10    #循环神经网络的训练长度
TRAINING_STEPS=10000    #训练轮数
BATCH_SIZE=32   #batch大小

TRAINING_EXAMPLES=10000 #训练数据个数
TESTING_EXAMPLES=1000   #测试数据个数
SAMPLE_GAP=0.01         #采样间隔

def generate_data(seq):
    X=[]
    y=[]

    for i in range(len(seq)-TIMESTEPS):
        X.append([seq[i:i+TIMESTEPS]])  #i:i+TIMESTEPS这组数据作为训练数据
        y.append([seq[i+TIMESTEPS]])    #为了预测第i+TIMESTEPS这一数据
    return np.array(X,dtype=np.float32),np.array(y,dtype=np.float32)

def lstm_model(X,y,is_training):
    # 多层LSTM
    cell = tf.nn.rnn_cell.MultiRNNCell([
        tf.nn.rnn_cell.LSTMCell(HIDDEN_SIZE)   for _ in range(NUM_LAYERS)])

    outputs,_=tf.nn.dynamic_rnn(cell,X,dtype=tf.float32)
    output=outputs[:,-1,:]

    # 对LSTM网络的输出再加一层全连接层并计算损失
    predictions = tf.contrib.layers.fully_connected(output,1,activation_fn=None)

    if not is_training:
        return predictions,None,None

    loss=tf.losses.mean_squared_error(labels=y,predictions=predictions)

    train_op=tf.contrib.layers.optimize_loss(
        loss,tf.train.get_global_step(),optimizer="Adagrad",learning_rate=0.1)

    return predictions,loss,train_op

def train(sess,train_X,train_y):
    ds=tf.data.Dataset.from_tensor_slices((train_X,train_y))
    ds=ds.repeat().shuffle(1000).batch(BATCH_SIZE)
    X,y=ds.make_one_shot_iterator().get_next()
    # 调用模型
    with tf.variable_scope("model"):
        predictions,loss,train_op=lstm_model(X,y,True)

    sess.run(tf.global_variables_initializer())
    for i in range(TRAINING_STEPS):
        _,l=sess.run([train_op,loss])
        if i %100==0:
            print("train step:"+str(i),", loss:"+str(l))


def run_eval(sess,test_X,test_y):
    ds=tf.data.Dataset.from_tensor_slices((test_X,test_y))
    ds=ds.batch(1)
    X,y=ds.make_one_shot_iterator().get_next()
    # 调用模型计算结果
    with tf.variable_scope("model",reuse=True):
        prediction,_,_=lstm_model(X,[0.0],False)

    predictions=[]
    labels=[]
    for i in range(TESTING_EXAMPLES):
        p,l=sess.run([prediction,y])
        predictions.append(p)
        labels.append(l)

    predictions=np.array(predictions).squeeze()
    labels=np.array(labels).squeeze()
    rmse=np.sqrt(((predictions-labels)**2).mean(axis=0))
    print("Mean Square Error is :%.2f"%rmse)

    return predictions,labels

if __name__ == '__main__':
    test_start=(TRAINING_EXAMPLES+TIMESTEPS)*SAMPLE_GAP
    test_end=test_start+(TESTING_EXAMPLES+TIMESTEPS)*SAMPLE_GAP
    train_X,train_y=generate_data(np.sin(np.linspace(
        0,test_start,TRAINING_EXAMPLES+TIMESTEPS,dtype=np.float32
    )))
    test_X,test_y=generate_data(np.sin(np.linspace(
        test_start,test_end,TESTING_EXAMPLES+TIMESTEPS,dtype=np.float32
    )))

    with tf.Session() as sess:
        train(sess,train_X,train_y)
        predictions,labels=run_eval(sess,test_X,test_y)
        # plt.figure()
        plt.plot(predictions, label="predictions")
        plt.plot(labels, label='real_sin')
        plt.legend()
        plt.show()

预测结果