1.4 Keras MNIST Playground

1.4.1 Keras 基础介绍

Keras 是一个用 Python 编写的高级神经网络 API，它能够以 TensorFlow，CNTK 或者 Theano 作为后端运行。

Keras 具有如下优点：

• 由于用户友好，高度模块化，可扩展性，可以简单而快速的进行原型设计。
• 同时支持卷积神经网络和循环神经网络，以及两者的组合。
• 在 CPU 和 GPU 上无缝运行。

Keras 的核心数据结构是 model，一种组织网络层的方式。最简单的模型是 Sequential 顺序模型，它把多个网络层线性堆叠起来。

Sequential 模型如下所示：

from keras.models import Sequential model = Sequential()

可以简单地使用 .add() 来堆叠模型：

from keras.layers import Dense model.add(Dense(units=64, activation='relu', input_dim=100)) model.add(Dense(units=10, activation='softmax'))

在完成了模型的构建后, 可以使用 .compile() 来配置学习过程：

model.compile(loss='categorical_crossentropy', optimizer='sgd', metrics=['accuracy'])

然后，就可以批量地在训练数据上进行迭代了。

# x_train 和 y_train 是 Numpy 数组 -- 就像在 Scikit-Learn API 中一样。 model.fit(x_train, y_train, epochs=5, batch_size=32)

只需一行代码就能评估模型性能：

loss_and_metrics = model.evaluate(x_test, y_test, batch_size=128)

或者对新的数据生成预测：

classes = model.predict(x_test, batch_size=128)

1.4.2 MNIST Playground

欢迎来到 MINST Playgound，在这里，你将了解到数据增强技术，并可以通过拖拽一个控制面板来创建深度学习模型，识别手写数字。无需编写任何代码，快来试试吧。

你需要按顺序运行下方每个 cell，直到看到控制面板，然后即可点击或拖拽来创建模型并训练。

# 导入必要的包

!mkdir -p ~/.keras/datasets
!cp ./mnist.npz ~/.keras/datasets/mnist.npz


%matplotlib inline
import time
import matplotlib.pyplot as plt
import numpy as np
import imgaug
import imgaug.augmenters as iaa
from ipywidgets import interact, interactive, fixed, interact_manual
from ipywidgets import IntSlider, FloatSlider, Dropdown, Checkbox, Button, Output, \
                       SelectMultiple,Image, HBox, VBox,SelectionSlider, HTML
from IPython import display
import tensorflow as tf
from tensorflow.keras.datasets import mnist
from tensorflow import keras
from tensorflow.python.util import deprecation
from keras.layers import Input, Dense,Flatten
from keras.models import Model
from keras.callbacks import Callback
from keras.utils.np_utils import to_categorical
from keras.utils import model_to_dot
deprecation._PRINT_DEPRECATION_WARNINGS = False

处理数据

首先我们定义一个 MNIST 类，它将负责管理 MNIST 数据集

class MNIST:
    """ 用来管理 MNIST 数据集的类
    
        Attributes:
            x_train:  MNIST 训练集的特征数据
            y_train:  MNIST 训练集的标签数据
            x_test:  MNIST 测试集的特征数据
            y_test:  MNIST 测试集的标签数据
            train_size: MNIST 训练集的样本数
            test_size: MNIST 测试集的样本数
            augmented: 是否对训练集数据进行数据增强
            seq: 进行数据增强的 pipline
    """
    def __init__(self):
        """初始化 MNIST 类
        """
        (self.x_train, self.y_train), (
        self.x_test, self.y_test) = mnist.load_data()
        self.x_train = self.x_train.reshape(self.x_train.shape[0], 28, 28, 1)
        self.x_test = self.x_test.reshape(self.x_test.shape[0], 28, 28, 1)
        self.x_train = self.x_train.astype('float32')
        self.x_test = self.x_test.astype('float32')
        self.train_size = self.x_train.shape[0]
        self.test_size = self.x_test.shape[0]
        self.y_train = to_categorical(self.y_train, 10)
        self.y_test = to_categorical(self.y_test, 10)
        self.noise = 0
        self.rotation = 0
        self.aug_possibility = 0 
        self.seq = None
        
    def update_noise(self, noise):
        """数据增强中是否增加 noise
        :param noise: 噪声比例
        :return:
        """
        self.noise = noise
        self.update_aug_seq()
        
    def update_rotation(self, rotation):
        """数据增强中图片旋转的最大角度
        :param rotation: 图片旋转的最大角度值
        :return:
        """
        self.rotation = rotation
        self.update_aug_seq()
    
    def update_aug_possibility(self, possibility):
        """数据增强中图片旋转的最大角度
        :param rotation: 图片旋转的最大角度值
        :return:
        """
        self.aug_possibility = possibility
        self.update_aug_seq()
        
    def update_aug_seq(self):
        """更新数据增强的 pipline
        :return:None
        """
        if self.aug_possibility:
            aug_seq = []
            sometimes = lambda aug: iaa.Sometimes(self.aug_possibility, aug)
            if self.noise:
                aug_seq.append(sometimes(iaa.Salt(self.noise))) 
            if self.rotation:
                aug_seq.append(sometimes(iaa.Affine(
                        rotate=(-self.rotation, self.rotation), 
                    )))
            self.seq = iaa.Sequential(aug_seq)
        else:
            self.seq = None

    def get_batch(self, batch_size, get_sample=False):
        """获取一个 batch 的训练数据（原始数据或者经过数据增强后的数据）
        :param batch_size: 一个 batch 所包含的图片数目
        :param get_sample: 是否返回固定样本来做展示
        :return:
        """
        while True:
            if get_sample:
                randidx = [i for i in range(batch_size)]
            else:
                randidx = np.random.randint(self.train_size, size=batch_size)
            epoch_x = self.x_train[randidx]
            epoch_y = self.y_train[randidx]
            if self.aug_possibility and (self.noise or self.rotation):
                epoch_x = self.seq(images=epoch_x)
            epoch_x /= 255.0
            yield epoch_x, epoch_y

    def get_batch_test(self, batch_size):
        """获取一个 batch 的测试数据
        :param batch_size: 一个 batch 所包含的图片数目
        :return:
        """
        while True:
            randidx = np.random.randint(self.test_size, size=batch_size)
            epoch_x = self.x_test[randidx]
            epoch_y = self.y_test[randidx]
            epoch_x /= 255.0
            yield epoch_x, epoch_y

    def plot_images(self, show=True):
        """绘制几个样本图片
        :param show: 是否显示绘图
        :return:
        """
        sample_num = 9
        imgs, _ = next(self.get_batch(sample_num,get_sample=True))
        img_figure = plt.figure(1)
        img_figure.set_figwidth(5)
        img_figure.set_figheight(5)
        for index in range(0, sample_num):
            ax = plt.subplot(3, 3, index + 1)
            ax.imshow(imgs[index].reshape(28, 28), cmap='gray')
        plt.margins(0, 0)
        img_figure.savefig('data_sample.png')
        if not show:
            plt.close(fig=img_figure)

数据增强技术

# 查看原始图片
mnist_data = MNIST()
mnist_data.plot_images()

数据增强技术就是利用有限的数据产生更多的等价数据。我们可以对原始的图片进行平移、旋转、缩放、加噪声等技术处理，来产生新的图片。 比如我们对一张小狗的图片进行旋转，加噪声，缩放等处理后，可以得到 6 张新的图片。

使用数据增强技术可以拓展我们的数据集，提高模型的识别性能和泛化能力。

下面我们对 mnist 数据集进行加噪声和旋转的处理。

mnist_data.update_noise(0.2)
mnist_data.update_rotation(15)
mnist_data.update_aug_possibility(0.5)
mnist_data.plot_images()

定义模型

下面我们先定义一个 DrawCallback 类，用来在每个训练 epoch 结束后，打印出训练曲线。

class DrawCallback(Callback):
    """ 进行模型训练的 callback 类，用来绘制训练图

    Attributes:
        init_loss:  初始的 loss 值
        epoch_data:  已经完成的 epoch 数的列表
        loss_data:  每个 epoch 的训练数据的 loss 值的列表
        val_loss_data:  每个 epoch 的测试数据的 loss 值的列表
        accuracy_data: 每个 epoch 的训练数据的 acc 值的列表
        val_accuracy_data: 每个 epoch 的测试数据的 acc 值的列表
        best_val_acc: 测试数据的最佳的 acc

    """
    def __init__(self):
        """初始化 DrawCallback 类
        """
        super().__init__()
        self.init_loss = None
        self.epoch_data = []
        self.loss_data = []
        self.val_loss_data = []
        self.accuracy_data = []
        self.val_accuracy_data = []
        self.best_val_acc = 0

    def runtime_plot(self, epoch=None):
        """绘制训练曲线
        :param epoch: 当前进行到的 epoch 数
        :return: None
        """
        # 总的 epoch 数
        epochs = self.params.get("epochs")
        # 定义图片尺寸
        img_figure = plt.figure(1)
        img_figure.set_figwidth(16)
        img_figure.set_figheight(5)
        # 绘制 loss 曲线
        ax1 = plt.subplot(1, 2, 1)
        ax1.set_ylim(0, int(self.init_loss * 3))
        ax1.set_xlim(1, epochs)
        ax1.plot(self.epoch_data, self.loss_data, 'b', label='loss_train')
        ax1.plot(self.epoch_data, self.val_loss_data, 'r', label='loss_val')
        ax1.set_xlabel('Epoch {}/{}'.format(epoch, epochs))
        ax1.set_ylabel('Loss')
        ax1.legend()
        # 绘制 acc 曲线
        ax2 = plt.subplot(1, 2, 2)
        ax2.set_ylim(0, 1)
        ax2.set_xlim(1, epochs)
        ax2.plot(self.epoch_data, self.accuracy_data, 'b', label='acc_train')
        ax2.plot(self.epoch_data, self.val_accuracy_data, 'r', label='acc_val')
        ax2.set_xlabel('Epoch {}/{}'.format(epoch, epochs))
        ax2.set_ylabel('ACC')
        ax2.legend()
        # 清除历史图片
        display.clear_output(wait=True)
        # 展示新图片
        plt.show()

    def on_epoch_end(self, epoch, logs=None):
        """在每个 epoch 结束后，存储 loss 值和 acc 值，并更新训练曲线图
        :param epoch: 当前进行到的 epoch 数
        :param logs: 当前 epoch 返回的 log
        :return: None
        """
        # 从 logs 中获取 loss 和 acc 的值并存到对应的 list 里
        epoch = epoch + 1
        logs = logs or {}
        loss = logs.get("loss")
        val_loss = logs.get("val_loss")
        accuracy = logs.get("accuracy")
        val_accuracy = logs.get("val_accuracy")
        if val_accuracy > self.best_val_acc:
            self.best_val_acc = val_accuracy
        if self.init_loss is None:
            self.init_loss = loss
        self.epoch_data.append(epoch)
        self.loss_data.append(loss)
        self.val_loss_data.append(val_loss)
        self.accuracy_data.append(accuracy)
        self.val_accuracy_data.append(val_accuracy)
        # 绘制训练图
        self.runtime_plot(epoch)
        # 如果训练完成，打印最终的结果
        if epoch == self.params.get("epochs"):
            print('训练完成')
            print('训练集准确率: {:.2%}'.format(accuracy))
            print('测试集准确率: {:.2%}'.format(val_accuracy))
            print('最高测试集准确率:{:.2%}'.format(self.best_val_acc))

然后我们定义一个 NN 类，我们将基于此类来构建和管理模型。

class NN(object):
    def __init__(self):
        """初始化 NN 类，构建一个多层全连接神经网络
        Attributes:
            layers: 隐藏层的数目
            neurons: 每一层隐藏层的神经元的个数
            epochs: 模型训练的轮数
            model: 构建后的模型
        """
        self.layers = 1
        self.neurons = [8]
        self.epochs = 5
        self.model = None

    def build_model(self):
        """构建并编译模型
        :return: None
        """
        # 定义输入层和 flatten 层
        inputs = Input(shape=(28, 28, 1,))
        output = Flatten()(inputs)
        # 定义隐藏层
        for neuron in self.neurons:
            output = Dense(neuron, activation='relu')(output)
        # 定义输出层
        predictions = Dense(10, activation='softmax')(output)
        # 构建模型
        self.model = Model(inputs=inputs, outputs=predictions)
        # 编译模型
        self.model.compile(loss='categorical_crossentropy',
                           optimizer='rmsprop',
                           metrics=['accuracy'])

    def start_train(self, data, validation_data):
        """开始训练模型
        :param data: 训练数据
        :param validation_data: 测试数据
        :return: None
        """
        plot_callback = DrawCallback()
        self.model.fit_generator(data,
                                 validation_data=validation_data,
                                 validation_steps=300,
                                 steps_per_epoch=500, 
                                 epochs=self.epochs, 
                                 verbose=1,
                                 callbacks=[plot_callback])

    def plot_model(self):
        """保存模型结构图
        :return: None
        """
        keras.utils.plot_model(
            self.model,
            to_file='model.png',
            show_shapes=True,
            show_layer_names=True)

操作面板

下面的代码将创建出一个可交互的操作面板，你可以通过此面板来进行数据增强，模型定义和训练的操作，并查看训练曲线图。

mlp = NN()
mlp.build_model()
mnist_data = MNIST()
## 定义操作界面
# 隐藏层数目
hidden_layers_widget = IntSlider(min=0, 
                                 max=3,
                                 value=1,
                                 description='隐藏层数目:',
                                 layout={'width': '400px'},
                                 style={'description_width': 'initial'})
# 每层隐藏层包含的神经元数目
neuron_widgets = VBox([SelectionSlider(options=[8,16, 32, 64,128,256,512],
    value=8,
    description='隐藏层1的神经元数目:',
    continuous_update=False,
    orientation='horizontal',
    readout=True,
    layout={'width': '400px'},
    style={'description_width': 'initial'})])

# 开始训练的按钮
start_train_widget = Button(
    description='开始训练',
    button_style='info',
    tooltip='点击开始训练',
)

output_widget = Output()

data_head_widget = HTML(
    value="<h1>准备数据</h1>",
    placeholder='',
)

data_aug_possibility_widget = FloatSlider(
    value=0,
    min=0,
    max=1.0,
    step=0.1,
    description='每个 batch 内增强图片的比例:',
    disabled=False,
    continuous_update=False,
    orientation='horizontal',
    readout=True,
    readout_format='.1f',
    layout={'width': '400px'},
    style={'description_width': 'initial'}
)


data_noise_widget = FloatSlider(
    value=0,
    min=0,
    max=0.2,
    step=0.05,
    description='为图片增加白噪声点:',
    disabled=False,
    continuous_update=False,
    orientation='horizontal',
    readout=True,
    readout_format='.2f',
    layout={'width': '400px'},
    style={'description_width': 'initial'}
)

data_rotate_widget = SelectionSlider(options=[0, 10, 20, 30],
                                     value=0,
                                     description='图片左右旋转的最大角度:',
                                     continuous_update=False,
                                     orientation='horizontal',
                                     readout=True,
                                     layout={'width': '400px'},
                                     style={'description_width': 'initial'})

def plot_sample_data():
    mnist_data.plot_images(show=False)
    file = open("data_sample.png", "rb")
    image = file.read()
    show_data_sample_widget.value = image

def update_data_noise(*args):
    mnist_data.update_noise(data_noise_widget.value)
    plot_sample_data()

def update_data_rotation(*args):
    mnist_data.update_rotation(data_rotate_widget.value)
    plot_sample_data() 
    
def update_data_aug_possibility(*args):
    mnist_data.update_aug_possibility(data_aug_possibility_widget.value)
    plot_sample_data()   

data_noise_widget.observe(update_data_noise, 'value') 
data_rotate_widget.observe(update_data_rotation, 'value') 
data_aug_possibility_widget.observe(update_data_aug_possibility, 'value') 

show_data_sample_widget = Image(
    value=b'',
    format='png',
    width=200,
)

plot_sample_data()

model_head_widget = HTML(
    value="<h1>构建模型</h1>",
    placeholder='',
)

show_model_widget = Image(
    value=b'',
    format='jpg',
    width=250,
)

def build_plot_model(*args):
    keras.backend.clear_session()
    mlp.build_model()
    mlp.plot_model()
    file = open("model.png", "rb")
    image = file.read()
    show_model_widget.value = image

build_plot_model()


# 更改隐藏层神经元数量时，更新模型实例的参数
def update_neuron_number(*args):
    mlp.neurons = [c.value for c in neuron_widgets.children]
    
for e in neuron_widgets.children:
    e.observe(update_neuron_number, 'value')
    e.observe(build_plot_model, 'value')
    

# 更改隐藏层数目时，动态增加或减小控制每一层神经元数量的按钮，同时更新模型实例的参数
def update_layers(*args):
    if hidden_layers_widget.value > mlp.layers:
        neuron_widgets.children += (SelectionSlider(options=[8,16, 32, 64,128,256,512],
                                                    value=8,
                                                    description='隐藏层{}的神经元数目:'.format(hidden_layers_widget.value),
                                                    continuous_update=False,
                                                    orientation='horizontal',
                                                    readout=True,
                                                   layout={'width': '400px'},
                                  style={'description_width': 'initial'}),)
    else:
        neuron_widgets.children = neuron_widgets.children[:-1]
    mlp.layers = hidden_layers_widget.value
    mlp.neurons = [c.value for c in neuron_widgets.children]
    for e in neuron_widgets.children:
        e.observe(update_neuron_number, 'value')
        e.observe(build_plot_model, 'value')
    build_plot_model()

hidden_layers_widget.observe(update_layers, 'value')    


data_H_widget = HBox([VBox([data_aug_possibility_widget, 
                            data_noise_widget, 
                            data_rotate_widget]), 
                      show_data_sample_widget])

model_H_widget = HBox([VBox([hidden_layers_widget ,
                             neuron_widgets]), 
                       show_model_widget])

train_head_widget = HTML(
    value="<h1>训练模型</h1>",
    placeholder='',
)

epochs_widget = SelectionSlider(options=[5, 10, 20, 50, 100],
                                     value=5,
                                     description='训练轮数:',
                                     continuous_update=False,
                                     orientation='horizontal',
                                     readout=True,
                                     layout={'width': '400px'},
                                     style={'description_width': 'initial'})

def update_epochs(*args):
    mlp.epochs = epochs_widget.value

epochs_widget.observe(update_epochs, 'value') 


    
def on_button_clicked(b):
    with output_widget:
        display.clear_output(wait=True)
        mlp.start_train(mnist_data.get_batch(32),
                      validation_data=mnist_data.get_batch_test(32))
        
start_train_widget.on_click(on_button_clicked) 

display.display(data_head_widget, 
                data_H_widget, 
                model_head_widget, 
                model_H_widget , 
                train_head_widget, 
                epochs_widget, 
                start_train_widget, 
                output_widget)