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)