Recurrent Neural Network

September 26, 2024

Recurrent Neural Network (RNN) 是類神經網絡架構其一,它的特性是:

  1. RNN 的內在層 (Hidden layers) 有層內迴路 (見下文解釋),
  2. RNN 多用來處理具時序性的資料,例如語言翻譯,聲音資料等。

相對於線性的類神經網路(如 Perceptron),訊息傳遞僅限於類神經層之間,同一層的類神經元不會互相影響。RNN使用同一神經層的不同神經元來儲存不同時間點的資料,且這些有時序關係的神經元會相互影響。

單層Hidden layer的例子

此範例使用單一內在層,但相似的方法可延伸至多個內在層。和Perceptron類似,輸入層(Input layer) xt 的每一個特徵值各乘上權重值, wwh,就是影響第一層 Hidden layer 的數值。

import torch
import torch.nn as nn


# 假設輸入的五個特徵值是1, 1, 1, 1, 1,xt是 1 x 5 的矩陣。
xt = torch.tensor([[1, 1, 1, 1, 1]])

# wxh  5 x 2 的矩陣 (二個類神經元) xh代表自輸入層到第一內部層。
wxh = torch.rand(5, 2)

# bxh  1 x 2 的矩陣。第一內部層設置二個節點
bxh = torch.rand(1, 2)

# hh 代表第一內部層至第二內部層。
whh = torch.rand(2, 2)
bhh = torch.rand(1, 2)

ht = xt × wxh + bxh

上列方程式右側的矩陣運算是:(1 x 5) × (5 x 2) + (1 x 2),所以 ht 是 1 x 2 的矩陣。ht即輸入層傳至第一層Hidden layer的 影響。

到此為止的計算方式只是一般迴歸方程式。RNN的不同處是輸入層的每一個特徵值之間有時序上的關係。假設上例的 xt 的五個特徵值只是時間點一的數值,且 xt 有三個時間點,完整的輸入資料將會是:

x_seq = torch.tensor([[1.0]*5, [2.0]*5, [3.0]*5]).float()
# tensor([[1., 1., 1., 1., 1.],  # <- 時間點一
#         [2., 2., 2., 2., 2.],  # <- 時間點二
#         [3., 3., 3., 3., 3.]]) # <- 時間點三

所以輸入層有三個時間點,五個特徵值。在RNN中, Hidden layer的每一時間點會用一個node表示,而每一個node除了接受來自於輸入層的影響之外,也會受前一個時間點的Hidden node影響。

此時序影響的實現方式稱為Memory Cell。其計算公式如下:

下方的方程式右側的矩陣運算是(1 x 2) × (2 x 2) + (1 x 2),所以 output 是 一個 1 x 2 的矩陣。

output = tanh(ht × whh + bhh)

在內部層,我們通常會加上一個 Activation Function 轉換迴歸方程式的輸出值,常用的轉換方程式有ReLU及tanh。綜言之,RNN 比 Perceptron 多了時間的向度。Perceptron 計算有類神經層、資料特徵數及樣本數的三個向度,RNN則加上時序的向度,共四個向度,除此之外,有時還須加上協助計算的批次(Batch)向度,因此RNN通常使用五個向度的tensor。

out_man = []
for t in range(3):
    xt = torch.reshape(x_seq[t], (1, 5))
    print(f'Time step {t} =>')
    print('   Input           :', xt.numpy())
    
    # 1 x 5 %*% 5 x 2 + 1 x 2 = 1 x 2 
    ht = torch.matmul(xt, w_xh) + b_xh    
    print(' Go to Hidden      :', ht.detach().numpy())
    
    if t>0:
        prev_h = out_man[t-1]
    else:
        prev_h = torch.zeros((ht.shape))
    
    # # 1 x 2 %*% 2 x 2 + 1 x 2 = 1 x 2
    print('   ht          :', ht.detach().numpy())
    print('   prev_h      :', prev_h.detach().numpy())

    ot = ht + torch.matmul(prev_h, w_hh) + b_hh
    ot = torch.tanh(ot)
    out_man.append(ot)
    print('   Output (manual) :', ot.detach().numpy())
    # print('   RNN output      :', out_man[:, t].detach().numpy())
    print()

# Time step 0 =>
#    Input           : [[1. 1. 1. 1. 1.]]
#  Go to Hidden      : [[2.591517  3.7888808]]
#    ht          : [[2.591517  3.7888808]]
#    prev_h      : [[0. 0.]]
#    Output (manual) : [[0.9983715 0.9994927]]
# 
# Time step 1 =>
#    Input           : [[2. 2. 2. 2. 2.]]
#  Go to Hidden      : [[5.178768 6.611533]]
#    ht          : [[5.178768 6.611533]]
#    prev_h      : [[0.9983715 0.9994927]]
#    Output (manual) : [[0.9999987  0.99999994]]
# 
# Time step 2 =>
#    Input           : [[3. 3. 3. 3. 3.]]
#  Go to Hidden      : [[7.7660193 9.434184 ]]
#    ht          : [[7.7660193 9.434184 ]]
#    prev_h      : [[0.9999987  0.99999994]]
#    Output (manual) : [[1. 1.]]
# 

上列程式碼顯示手動計算上述例子的過程。假設輸入資料僅有三個時間點,設置單層內在層,此內在層有二個節點,內在層的輸出值是三個時間點的資料逐次經過迴歸方程式之後的累加結果(因此,在內在層裡,每一時間點影響下一個時間點的數值)。

如此時序影響的計算可以用For loop來了解,實作上,機器學習套件,例如 PyTorch ,是使用 nn.RNN 完成。下列程式碼指出輸入層的每筆一資料有五個特徵值, Hidden層有二個nodes,此網絡只有一層內在層,batch_first=True 將批次層置於tensor的第一個向度。

import torch
import torch.nn as nn
import tensorflow as tf

# PyTorch
torch.manual_seed(1)
torch_model = nn.RNN(input_size=5, hidden_size=2, num_layers=1, 
                   batch_first=True) 


# Tensorflow
tensor_model = tf.keras.layers.SimpleRNN(
    units=2,
    input_shape=(None, 5),
    return_sequences=True,
    stateful=False
)

在RNN的類神經網路中,除了如上述單純 Memory cell的實現方式外,為了解決實際計算時的諸多問題,RNN的Memory cells還包含了如LSTM,GRU以及最近非常流行的Transformer,這些不同的Memory cells實現了認知機制數學建模中的許多概念,例如遺忘、選擇性注意力、有限注意力,記憶形態的不同(長期、短期記憶 vs. 工作記憶)等概念。

巴赫眾讚歌的例子

JSB-Chorales是一組巴赫眾讚歌的資料集,它收集了382首在公領域的巴赫眾讚歌,這些歌曲是用零以及整數36至81儲存的,這些整數數字代表音符在音譜上的位置,例如 36 代表 C1 音階,81 代表A5音階,而 0 代表休止符。

每一首歌曲以這些整數數值表示,整理成如下矩陣 (Python 原生的 list資料型態)。每一行代表一個時間點,而每一欄代表一個音符,一個時間點,彈奏四個音符。

# [[74, 70, 65, 58],  # 時間點一
#  [74, 70, 65, 58],  # 時間點二
#  [74, 70, 65, 58],  # 時間點三
#  [74, 70, 65, 58],  # ...
#  [75, 70, 58, 55],
#  [75, 70, 58, 55],
#  [75, 70, 60, 55],
#  ...
#  [70, 65, 62, 46],
#  [70, 65, 62, 46],
#  [70, 65, 62, 46]]

它聽起來大致是如此:

上面的矩陣將每一個時間點儲存為一個Python list,如[70, 65, 62, 46]。加上上例的歌曲有有192個時間點,所以它有192 個 list ,這些 list 則是被儲存在另一個外層的 list 之中。

因為多數機器學習套件希望自由地在單獨CPU,單獨GPU(或TPU),或者同時在CPU及GPU兩者上運算,加上CPU及GPU所能直接使用的記憶體是不同的硬體單元,Python的原生資料型態預設使用CPU及其記憶體來運算,所以我們必須將 list 的資料型態轉換為tensor,如此方能自由地在CPU及GPU的運算空間運作。下面用train_chorales作為範別說明。

讀取巴赫眾讚歌資料

在下載檔案後,我們可以使用一個簡單的自訂函數,將分別位於三個資料夾中的音樂檔讀入。

import pandas as pd


def load_chorales(files, path):
    filepaths = [path + file for file in files]  
    out = [pd.read_csv(filepath).values.tolist() for filepath in filepaths]
    return out


train_path = '../../data/jsb_chorales/train/'
valid_path = '../../data/jsb_chorales/valid/'
test_path = '../../data/jsb_chorales/test/'
train_files =  sorted( os.listdir(train_path) )
valid_files =  sorted( os.listdir(valid_path) )
test_files  =  sorted( os.listdir(test_path) )

train_chorales = load_chorales(train_files, train_path)
valid_chorales = load_chorales(valid_files, valid_path)
test_chorales = load_chorales(test_files, test_path)

我們可以使用 Python 中的 set 資料型態將訓練組,驗證組及測試組的資料檔都收集在一個set之中,並找出總共使用了多了音符,最高音及最低音。

notes = set()
for chorales in (train_chorales, valid_chorales, test_chorales):
    for chorale in chorales:
        for chord in chorale:
            # This line converts the chord into a set and performs a union operation (|=) with the notes set. This means it adds all unique elements from chord to notes.
            notes |= set(chord)

n_notes = len(notes) # 47
min_note = min(notes - {0})
max_note = max(notes)

如何決定時序資料裡的「特徵值」及「目標值」?

時序資料與非時序資料略為不同的一點在於,它通常沒有顯而易見的特徵值及目標值的分別。比方說,在笑容分類作業裡,我們可以定義目標值為一張照片裡的人臉是否展現笑容,若是則記錄為1,若否則記錄為0,如此毎張照片有其像素值做為特徵值,而照片是否呈現笑容則為目標值。此原則無法直接適用在時序資料。

以下解釋一項稱為Sequence to Sequence的時序資料整理方法,此方法的原則是希望有較早時間點的一小段(例如5個時間點)資料作為特徵值,而此段資料所對應的目標值則是後移一小段時間,等長的資料段落,且此資料段落有部份資料是和特徵值的資料有時間上的重疊。以下我們使用眾讚歌範例來說明。

  1. 首先,tf.ragged.constant函數將上面的 nested list 的 Python資料型態轉換為tensor。 chorales tensor是一個 229 x 1 的列向量,包含229首歌曲(讀取為229個csv檔案),此tensor裡面使用一個 numpy 向量記錄每一首歌的時間長度,例如第一個csv檔案有192個時間點,而第二個檔案有228個時間點。

chorales的每一個元素,例如chorales[0]代表一首歌曲,也同時被轉換為tensor的資料型態。

import tensorflow as tf


chorales = tf.ragged.constant(train_chorales, ragged_rank=1)
# chorales.row_lengths()
# <tf.Tensor: shape=(229,), dtype=int64, numpy=
# array([192, 228, 208, 432, 260, ... 228, 164])>

chorales[0]
# <tf.Tensor: shape=(192, 4), dtype=int32, numpy=
# array([[74, 70, 65, 58],
#        [74, 70, 65, 58],
#        [74, 70, 65, 58],
#        [74, 70, 65, 58],
#    ...       ])
  1. 接下來我們用 tf.data.Dataset.from_tensor_slices函數將 chorales tensor轉換為 tensorflow 套件中的 Dataset 類別 (Class) ,如此我們可以使用此類別裡的函數來方便處理資料。
dataset = tf.data.Dataset.from_tensor_slices(chorales)

Dataset類別裡的mapflat_map 函數會將Dataset裡的每一首歌曲作為輸入,送進to_windows的自定函數中處理。mapflat_map的區別在於是否將輸出的資料的向度平面化。map不改變輸出向度,flat_map會改變輸出向度。

dataset.flat_map(to_windows)
  1. to_windows函數中,我們再一次使用tf.data.Dataset.from_tensor_slices將個別的歌曲轉換為tensorflow的Dataset類別,一樣地,如此做是為了方便使用此類別的函數處理每一首歌曲的內容。

  2. 比方說,我們決定視六個連續的時間點為一筆樣本,移動三個時間點,再取五個連續時間點,作為下一筆樣本,以此類推,到了資料的尾端,如果剩下的資料不足五個時間點,則捨棄剩餘資料。

window_size = 5
window_shift = 3

train_dataset0 = tf.data.Dataset.from_tensor_slices(chorales[0])
train_dataset0 = train_dataset0.window(
    window_size + 1, window_shift, drop_remainder=True
)

# for i, item in enumerate(train_dataset0.take(3)):
#     print(i, item)
# 0 tf.Tensor(
# [[74 70 65 58]
#  [74 70 65 58]
#  [74 70 65 58]
#  [74 70 65 58]
#  [75 70 58 55]
#  [75 70 58 55]], shape=(6, 4), dtype=int32)
# 1 tf.Tensor(
# [[74 70 65 58]
#  [75 70 58 55]
#  [75 70 58 55]
#  [75 70 60 55]
#  [75 70 60 55]
#  [77 69 62 50]], shape=(6, 4), dtype=int32)
# 2 tf.Tensor(
# [[75 70 60 55]
#  [75 70 60 55]
#  [77 69 62 50]
#  [77 69 62 50]
#  [77 69 62 50]
#  [77 69 62 50]], shape=(6, 4), dtype=int32)

使用flat_map的函數,我們可以將資料整理成上面所示。每一筆樣本是一個 6 x 4的tensor。

  1. 最後,我們將每一筆樣本用tf.where轉換。此函數如同一個ifesle語法,若遭遇到0,則維持原樣,若遭遇到其他數值,則減去35 (window - min_note + 1),所以原來代表C1音符的36,變為1,而原來代表A5音符的81變為46。

  2. tf.reshape(window, [-1])將上示的6 x 4矩陣換為為 1 x 24的列向量。

def preprocess(window):
    window = tf.where(window == 0, window, window - min_note + 1) 
    return tf.reshape(window, [-1]) 
for i, item in enumerate(chorales.take(3)):
    print(i, item)

# 0 tf.Tensor([39 35 30 23 39 35 30 23 39 35 30 23 39 35 30 23 40 35 23 20 40 35 23 20], shape=(24,), dtype=int32)
# 1 tf.Tensor([39 35 30 23 40 35 23 20 40 35 23 20 40 35 25 20 40 35 25 20 42 34 27 15], shape=(24,), dtype=int32)
# 2 tf.Tensor([40 35 25 20 40 35 25 20 42 34 27 15 42 34 27 15 42 34 27 15 42 34 27 15], shape=(24,), dtype=int32)
  1. 最後我們使用Dataset類別裡的 batch函數將每兩筆樣本收集為一個批次處理單元。
# 每一個批次有兩筆樣本
batch_size = 2
chorales = chorales.batch(batch_size)

# 將每一筆樣本,整理為(特徵,目標)的資料集
def create_target(batch):
    X = batch[:, :-1]  # all columns except the last one
    Y = batch[:, 1:]   # starting from the 2nd column to the last.
    return X, Y
chorales= chorales.map(create_target)

爾後再次使用map函數,將chorales Dataset裡每一個元素中第一個至第23個數值取出為樣本特徵值,[:, :-1]代表除了最後一欄的所有的欄位。以上面的 第一筆 Tensor (即0 tf.Tensor) 為例,即,

# 第一批次中的第一筆樣本
[39 35 30 23 39 35 30 23 39 35 30 23 39 35 30 23 40 35 23 20 40 35 23]

第二個數值至最後一欄的數值則為目標值,[:, 1:]代表從第二欄開始至最後一欄的。因為Python以0開始計次,所以1代表第二個數值。加上之前的tf.reshape(window, [-1])函數已經將每一個樣本整理為行向量,所以每一筆樣本只有一行。

# 第一批次中的第一筆目標
[35 30 23 39 35 30 23 39 35 30 23 39 35 30 23 40 35 23 20 40 35 23 20]

下面列出處理完後的例子。

# 處理前
# 0 tf.Tensor(
#   --------------------第一欄至倒數第二欄 (特徵值)-------------------------
#   |   --------------------第二欄至最後 (目標值)-------------------------|---
#   |   |                                                              |  |
# [[39 35 30 23 39 35 30 23 39 35 30 23 39 35 30 23 40 35 23 20 40 35 23 20]
#  [39 35 30 23 40 35 23 20 40 35 23 20 40 35 25 20 40 35 25 20 42 34 27 15]], shape=(2, 24), dtype=int32)
# 1 tf.Tensor(
# [[40 35 25 20 40 35 25 20 42 34 27 15 42 34 27 15 42 34 27 15 42 34 27 15]
#  [42 34 27 15 42 34 27 15 42 34 27 15 42 35 27 20 42 35 27 20 42 34 27 20]], shape=(2, 24), dtype=int32)

# 處理後 
#---------------------------------------------------------------------------------------------------------
# 0 (樣本一)
# 特徵值
# tf.Tensor(
# [[39 35 30 23 39 35 30 23 39 35 30 23 39 35 30 23 40 35 23 20 40 35 23]
#  [39 35 30 23 40 35 23 20 40 35 23 20 40 35 25 20 40 35 25 20 42 34 27]], shape=(2, 23), dtype=int32) 
# 目標值
# tf.Tensor(
# [[35 30 23 39 35 30 23 39 35 30 23 39 35 30 23 40 35 23 20 40 35 23 20]
#  [35 30 23 40 35 23 20 40 35 23 20 40 35 25 20 40 35 25 20 42 34 27 15]], shape=(2, 23), dtype=int32)
# 1 (樣本二)
# 特徵值
# tf.Tensor(
# [[40 35 25 20 40 35 25 20 42 34 27 15 42 34 27 15 42 34 27 15 42 34 27]
#  [42 34 27 15 42 34 27 15 42 34 27 15 42 35 27 20 42 35 27 20 42 34 27]], shape=(2, 23), dtype=int32) 
# 目標值
# tf.Tensor(
# [[35 25 20 40 35 25 20 42 34 27 15 42 34 27 15 42 34 27 15 42 34 27 15]
#  [34 27 15 42 34 27 15 42 34 27 15 42 35 27 20 42 35 27 20 42 34 27 20]], shape=(2, 23), dtype=int32)    

小結

本文解釋RNN的基本概念記憶細胞(Memory cell),並簡介了最簡單的記憶細胞。它的主要功能是為了使類神經層可以記憶時序關係。本文並提及多種進階的記憶細胞,如LSTM,GRU及Transformer,這些不同的記憶細胞,如同不同類別的神經細胞,有它們特殊的功能,例如長期、短期記憶,選擇性地遺忘以及選擇性注意。這些不同的細胞型態的出現是為了應付複雜的時序資料,例如大型語言模型;我們還提及了時序資料的一項重要概念:如何定義特徵值及目標值,這裡我們僅展示了「段落特徵預測段落目標」(Sequence to sequence)的方法,其他方法還有「段落特徵預測單一目標值」等,我們在接下來的文章在分項介紹。

最後,讓我們來聽看看,我們所訓練的模型模仿上述資料所產生巴赫眾(仿)讚歌: