Keras（四）CNN 卷積神經網絡 RNN 循環神經網絡 原理及實例

原文鏈接：http://www.one2know.cn/keras5/

CNN 卷積神經網絡

• 卷積 池化
  https://www.cnblogs.com/peng8098/p/nlp_16.html 中有介紹
• 以數據集MNIST構建一個卷積神經網路

from keras.layers import Dense,Activation,Conv2D,MaxPooling2D,Flatten
from keras.models import Model,Sequential
from keras.datasets import mnist
from keras.utils import np_utils

# 構建數據集
(x_train,y_train),(x_test,y_test) = mnist.load_data()
x_train = x_train.reshape(x_train.shape[0],1,28,28)/255
x_test = x_test.reshape(x_test.shape[0],1,28,28)/255
y_train = np_utils.to_categorical(y_train,num_classes=10)
y_test = np_utils.to_categorical(y_test,num_classes=10)
print(x_train[0].shape)
print(y_train[:3])

## 構建模型
model = Sequential()

# 第一層 卷積層
model.add(Conv2D(
    # input_shape=(60000,1,28,28),
    batch_input_shape=(32,1,28,28), # 輸入數據的shape
    filters=32, # 濾波器數量為32
    kernel_size=5,
    strides=1,
    padding='same', # same即不改變原來數據的長度和寬度
    data_format='channels_first'
))
model.add(Activation('relu')) # 激勵函數為relu

# 第二層 池化層
model.add(MaxPooling2D(
    pool_size=2, # 分辨率長寬各降低一半，輸出數據shape為（32，14，14）
    strides=2,
    padding='same',
    data_format='channels_first'
))

# 再加一遍卷積層和池化層 輸出數據shape為（64，7，7）
model.add(Conv2D(64, 5, strides=1, padding='same', data_format='channels_first'))
model.add(Activation('relu'))
model.add(MaxPooling2D(2, 2, 'same', data_format='channels_first'))

# 將數據抹平 再加一層全連接層
model.add(Flatten())
model.add(Dense(1024))
model.add(Activation('relu'))

# 再加一層全連接層 作為輸出層
model.add(Dense(10))
model.add(Activation('softmax'))

# 設置adam優化方法，loss函數, metrics方法來觀察輸出結果
model.compile(optimizer='adam',loss='categorical_crossentropy',metrics=['accuracy'])

# 訓練模型
model.fit(x_train, y_train, epochs=1, batch_size=32)

# 預測
loss,accuracy = model.evaluate(x_test,y_test)
print('test loss:',loss)
print('test accuracy:',accuracy)

輸出：

Epoch 1/1

   32/60000 [..............................] - ETA: 31:05 - loss: 2.2981 - acc: 0.1562
   64/60000 [..............................] - ETA: 19:05 - loss: 2.2658 - acc: 0.2344

   32/10000 [..............................] - ETA: 35s
   96/10000 [..............................] - ETA: 21s
   
test loss: 0.03328929296457209
test accuracy: 0.9897

RNN 循環神經網絡

• 序列數據
  
  我們想象現在有一組序列數據 data 0,1,2,3. 在當預測 result0 的時候,我們基於的是 data0, 同樣在預測其他數據的時候, 我們也都只單單基於單個的數據. 每次使用的神經網絡都是同一個 NN. 不過這些數據是有關聯 順序的 , 就像在廚房做菜, 醬料 A要比醬料 B 早放, 不然就串味了. 所以普通的神經網絡結構並不能讓 NN 了解這些數據之間的關聯
• 處理序列數據的神經網路
  最基本的方式,就是記住之前發生的事情. 那我們讓神經網絡也具備這種記住之前發生的事的能力. 再分析 Data0 的時候, 我們把分析結果存入記憶. 然后當分析 data1的時候, NN會產生新的記憶, 但是新記憶和老記憶是沒有聯系的. 我們就簡單的把老記憶調用過來, 一起分析. 如果繼續分析更多的有序數據 , RNN就會把之前的記憶都累積起來, 一起分析
  
  每次 RNN 運算完之后都會產生一個對於當前狀態的描述 , state. 我們用簡寫 S( t) 代替, 然后這個 RNN開始分析 x(t+1) , 他會根據 x(t+1)產生s(t+1), 不過此時 y(t+1) 是由 s(t) 和 s(t+1) 共同創造的. 所以我們通常看到的 RNN 也可以表達成這種樣子

RNN Classifier 實例

• 依然使用MNIST數據集

import numpy as np
np.random.seed(1)

from keras.datasets import mnist
from keras.utils import np_utils
from keras.models import Sequential
from keras.layers import SimpleRNN, Activation, Dense
from keras.optimizers import Adam

# 超參數
TIME_STEPS = 28
INPUT_SIZE = 28
BATCH_INDEX = 0 # 從第0個開始訓練
BATCH_SIZE = 50 # 一個batch50個數據
CELL_SIZE = 50 # 輸出50個神經元
OUTPUT_SIZE = 10 # 輸出10個類：0~9
LR = 0.001 # 學習速度

(x_train, y_train), (x_test, y_test) = mnist.load_data()

# data pre-processing
x_train = x_train.reshape(-1, 28, 28) / 255.      # 標准化
x_test = x_test.reshape(-1, 28, 28) / 255.
y_train = np_utils.to_categorical(y_train, num_classes=10)
y_test = np_utils.to_categorical(y_test, num_classes=10)

## 搭建模型
model = Sequential()

# 添加RNN層
model.add(SimpleRNN(
    batch_input_shape=(None, TIME_STEPS, INPUT_SIZE),
    output_dim=CELL_SIZE,
    unroll=True,
))

# 添加輸出層
model.add(Dense(OUTPUT_SIZE))
model.add(Activation('softmax'))

# 設置優化器
adam = Adam(LR)
model.compile(optimizer=adam,loss='categorical_crossentropy',metrics=['accuracy'])

# 訓練
for step in range(40001):
    X_batch = x_train[BATCH_INDEX: BATCH_INDEX+BATCH_SIZE, :, :]
    Y_batch = y_train[BATCH_INDEX: BATCH_INDEX+BATCH_SIZE, :]
    cost = model.train_on_batch(X_batch, Y_batch)
    BATCH_INDEX += BATCH_SIZE
    BATCH_INDEX = 0 if BATCH_INDEX >= x_train.shape[0] else BATCH_INDEX

    if step % 500 == 0: # 每訓練500進行一次測試
        cost, accuracy = model.evaluate(x_test, y_test, batch_size=y_test.shape[0], verbose=False)
        print('test cost: ', cost, 'test accuracy: ', accuracy)

輸出：

test cost:  2.3316211700439453 test accuracy:  0.12210000306367874
test cost:  0.5586103200912476 test accuracy:  0.8342999815940857
test cost:  0.4080776870250702 test accuracy:  0.8806999921798706
。。。。。。
test cost:  0.12420056015253067 test accuracy:  0.9653000235557556
test cost:  0.13435833156108856 test accuracy:  0.9632999897003174
test cost:  0.12595564126968384 test accuracy:  0.9653000235557556

RNN Regressor 實例

import numpy as np
np.random.seed(1)
from keras.models import Sequential
from keras.layers import Dense,TimeDistributed,SimpleRNN
from keras.optimizers import Adam
import matplotlib.pyplot as plt

# 超參數
BATCH_START = 0
TIME_STEPS = 20 # 時間步長 前面20個數據對下一個有影響
BATCH_SIZE = 50
INPUT_SIZE = 1
OUTPUT_SIZE = 1
CELL_SIZE = 20
LR = 0.01

# 生成數據
def get_batch():
    global BATCH_START, TIME_STEPS
    xs = np.arange(BATCH_START, BATCH_START+TIME_STEPS*BATCH_SIZE).reshape((BATCH_SIZE, TIME_STEPS)) / (10*np.pi)
    seq = np.sin(xs)
    res = np.cos(xs)
    BATCH_START += TIME_STEPS
    return [seq[:, :, np.newaxis], res[:, :, np.newaxis], xs]

# 查看數據
# get_batch()
# exit()

## 搭建網絡
model = Sequential()
# 添加RNN層
model.add(SimpleRNN(
    batch_input_shape=(BATCH_SIZE, TIME_STEPS, INPUT_SIZE),
    output_dim=CELL_SIZE,
    return_sequences=True,  # 對於每一個時間點需不需要輸出對應的output,True每個時刻都輸出，False最后的輸出output
    stateful=True,  # batch與batch之間是否有聯系，需不需要將狀態進行傳遞
))
# 添加輸出層
model.add(TimeDistributed(Dense(OUTPUT_SIZE))) # TimeDistributed：對每一個output進行全連接的計算

# 優化器
adam = Adam()
model.compile(
    optimizer=adam,
    loss='mse',
)

# 訓練
print('Training ------------')
for step in range(501):
    # data shape = (batch_num, steps, inputs/outputs)
    X_batch, Y_batch, xs = get_batch()
    cost = model.train_on_batch(X_batch, Y_batch)
    pred = model.predict(X_batch, BATCH_SIZE)
    plt.plot(xs[0, :], Y_batch[0].flatten(), 'r', xs[0, :], pred.flatten()[:TIME_STEPS], 'b--')
    plt.ylim((-1.2, 1.2))
    plt.draw()
    plt.pause(0.1)
    if step % 10 == 0:
        print('train cost: ', cost)