rnn-nlp-單詞預測

import reader
import numpy as np
import tensorflow as tf

# 數據參數
DATA_PATH = 'simple-examples/data/'  # 數據存放路徑
VOCAB_SIZE = 10000  # 單詞數量

# 神經網絡參數
HIDDEN_SIZE = 200       # LSTM隱藏層規模
NUM_LAYERS = 2          # LSTM結構層數
LEARNING_RATE = 1.0     # 學習速率
KEEP_PROB = 0.5         # 節點不被dropout的概率
MAX_GRAD_NORM = 5       # 用於控制梯度膨脹的參數

# 訓練參數
TRAIN_BATCH_SIZE = 20   # 訓練數據batch大小       # 圖片張數
TRAIN_NUM_STEP = 35     # 訓練數據截斷長度        # 圖片行數

# 測試參數
EVAL_BATCH_SIZE = 1     # 測試數據batch大小
EVAL_NUM_STEP = 1       # 測試數據截斷
NUM_EPOCH = 2           # 使用訓練數據的輪數


# 通過PTBModel描述模型，方便維護循環神經網絡中的狀態
class PTBModel():
    def __init__(self, is_training, batch_size, num_steps):
        # 記錄batch和截斷長度
        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])    # 注意輸入輸出shape一樣

        # 定義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)

        # 初始化state
        self.initial_state = cell.zero_state(batch_size, tf.float32)

        # 將單詞ID轉為單詞向量。每個單詞都是HIDDEN_SIZE維
        embedding = tf.get_variable('embedding', [VOCAB_SIZE, HIDDEN_SIZE])

        # 將原本batch_size*num_steps的輸入層轉化為batch_size*num_steps*HIDDEN_SIZE
        inputs = tf.nn.embedding_lookup(embedding, self.input_data)

        # 只在訓練時使用dropout
        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)  # 將當前時刻的數據和狀態傳入LSTM
                outputs.append(cell_output)  # 將當前輸出加入輸出列表

        # 將輸出列表展開成[batch,hidden_size*num_steps]
        # 再reshape成[batch*num_steps,hidden_size]
        output = tf.reshape(tf.concat(outputs, 1), [-1, HIDDEN_SIZE])

        # 將輸出傳入全連接層，每個時刻的輸出都是長度為VOCAB_SIZE的數組
        weight = tf.get_variable('weight', [HIDDEN_SIZE, VOCAB_SIZE])
        bias = tf.get_variable('bias', [VOCAB_SIZE])
        logits = tf.matmul(output, weight) + bias

        # 定義交叉熵損失函數，sequence_loss_by_example計算一個序列的交叉熵的和
        loss = tf.contrib.legacy_seq2seq.sequence_loss_by_example(
            [logits],                                               # 預測結果
            [tf.reshape(self.targets, [-1])],                       # 預期結果。將[batch_size,num_steps]壓縮成一維
            [tf.ones([batch_size * num_steps], dtype=tf.float32)]   # 損失的權重。這里所有的權重都為1，表示不同batch和不同時刻的重要程度都一樣
        )

        # 計算得到每個batch的平均損失
        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_queue, train_op, output_log, epoch_size):
    # 計算perplexity的輔助變量
    total_costs = 0.0
    iters = 0
    state = session.run(model.initial_state)

    # 使用當前數據訓練或測試模型
    for step in range(epoch_size):
        # 生成輸入和答案
        feed_dict = {}
        x, y = session.run(data_queue)
        feed_dict[model.input_data] = x
        feed_dict[model.targets] = y

        # 將狀態轉為字典
        for i, (c, h) in enumerate(model.initial_state):
            feed_dict[c] = state[i].c
            feed_dict[h] = state[i].h

        # 獲取損失值和下一個狀態
        cost, state, _ = session.run(
            [model.cost, model.final_state, train_op], feed_dict=feed_dict
        )  # 在當前batch上運行train_op並計算損失值。交叉熵損失函數計算的是下一個單詞為給定單詞的概率
        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  # batch的個數
    train_epoch_size = (train_batch_len - 1) // TRAIN_NUM_STEP  # 該epoch的訓練次數

    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)

    # 生成數據隊列，必須放在開啟多線程之前
    train_queue = reader.ptb_producer(train_data, train_model.batch_size,train_model.num_steps)
    valid_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)

    with tf.Session() as sess:
        tf.global_variables_initializer().run()

        # 開啟多線程從而支持ptb_producer()使用tf.train.range_input_producer()
        coord = tf.train.Coordinator()
        threads = tf.train.start_queue_runners(sess=sess, coord=coord)

        # 使用訓練數據訓練模型
        for i in range(NUM_EPOCH):
            print('In iteration: %d' % (i + 1))
            run_epoch(sess, train_model, train_queue, train_model.train_op, True, train_epoch_size)  # 訓練模型
            valid_perplexity = run_epoch(sess, eval_model, valid_queue, tf.no_op(), False, valid_epoch_size)  # 使用驗證數據評估模型
            print('Epoch: %d Validation Perplexity: %.3f' % (i + 1, valid_perplexity))

        # 使用測試數據測試模型
        test_perplexity = run_epoch(sess, 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__':
    tf.app.run()

理解這個代碼之前最好先理解LSTM手寫數字識別。

類比圖像解釋自然語言處理

1. 自然語言的輸入需要先把句子截成固定長度的片段，這就是一個序列，也就是一張圖片，片段中的每一個字或者詞就是圖片的一行，lstm每個時序輸入這個字或者詞。

2. 不同之處在於，自然語言每個時序都有x和y，y就是x后的那個字或者詞，而圖片是很多x一個y，這決定了兩者的網絡結構不同。

3. 所以測試時，自然語言不需要輸入完整序列，也就是不需要截斷，而圖片每次要輸入一張，即完整序列。

 

類比之后就非常容易理解了，但是代碼中有個細節還是需要思考一下的。

 # 定義輸入層
 self.input_data = tf.placeholder(tf.int32, [batch_size, num_steps])


 # 將單詞ID轉為單詞向量。每個單詞都是HIDDEN_SIZE維
 embedding = tf.get_variable('embedding', [VOCAB_SIZE, HIDDEN_SIZE])

 # 將原本batch_size*num_steps的輸入層轉化為batch_size*num_steps*HIDDEN_SIZE
 inputs = tf.nn.embedding_lookup(embedding, self.input_data)

這里官方注釋的不是很清楚，我是這么理解的。

這個input是單詞ID，embedding是所有單詞的初始化權重，embedding_lookup 操作就是wx操作，因為wx是隱層的輸入，所以維度是 HIDDEN_SIZE

假設我們只有2個單詞，隱層3個神經元

那我們手動表示應該是這樣的

embedding_lookup 是這樣的