Python
Keras
TensorFlow

Keras(+Tensorflow)でMNISTしてみる

以前に、私的TensorFlow入門でも書いたんだけれど、MNISTをまたTensorFlowで書いてみる。今度は、Kerasを使ってみる。

多階層のニューラルネットでmodelを作成しようとすると、TensorFlowでは層を追加していくのってどうやってやるの?とか、直観的に分かりづらいんだけれど、Kerasはそのあたりを上手くやってくれてる感じ。

Keras(とTensorFlow)のインストール

kerasやtensorflowのインストールごとき、特に解説する必要もない気がするけれど、一応。

% pip install -U keras tensorflow

Kerasのソースをダウンロードしておく

kerasのソースをダウンロードしておく。

% git clone https://github.com/keras-team/keras.git

なぜなら、examplesがあるから。

MNISTのデータをダウンロードしておく

kearsではデータセットとしてMNISTが公開されている

とりあえず、一旦、データセットだけをダウンロードしておきたいので、ipythonで以下のように実行すると、~/.keras/datasets/にmnist.npzというファイルが保存される。

from keras.datasets import mnist
(x_train, y_train), (x_test, y_test) = mnist.load_data()

これにより、x_trainには60000個の学習用データ、y_trainにはそれに対応するラベル、x_testには10000個のテスト用データと、ソレに対応するラベルであるy_testという変数が、それぞれ作成される。

機械学習の勉強をしていると、必ず「trainingセットとtestセットとvalidationセットは別にしておけよ」と言われると思うんだけれど、これだとtrainingセットとtestセットしかないじゃん。というワケで、trainingセットから、10000個くらいをランダムに選んで、validationセットも作っておくことにする。

from sklearn.model_selection import train_test_split
x_train1, x_valid, y_train1, y_valid = train_test_split(x_train, y_train, test_size=0.175)

これで、validationセットが10500個できた。

MNISTをConvolutional Neural Networkで解く

modelの作成

とりあえず、modelを作る。

from keras.models import Sequential
from keras.layers import Dense, Dropout, Flatten
from keras.layers import Conv2D, MaxPooling2D

model = Sequential()
model.add(Conv2D(32, kernel_size=(3, 3), activation='relu', input_shape=(28, 28, 1)))
model.add(Conv2D(64, (3, 3), activation='relu'))
model.add(MaxPooling2D(pool_size=(2, 2)))
model.add(Dropout(0.25))
model.add(Flatten())
model.add(Dense(128, activation='relu'))
model.add(Dropout(0.5))
model.add(Dense(10, activation='softmax'))

ちゃんと過学習しないように、Dropout()しておく。

Layer (type) Output Shape Param #
conv2d_1 (Conv2D) (None, 26, 26, 32) 320
conv2d_2 (Conv2D) (None, 24, 24, 64) 18496
max_pooling2d_1 (MaxPooling2 (None, 12, 12, 64) 0
dropout_1 (Dropout) (None, 12, 12, 64) 0
flatten_1 (Flatten) (None, 9216) 0
dense_1 (Dense) (None, 128) 1179776
dropout_2 (Dropout) (None, 128) 0
dense_2 (Dense) (None, 10) 1290

学習

学習する。最適化には、RMSprop()を使う。examplesには、Adadelta()を使うって書いてあるんだけれど、なんとなくRMSprop()で。
こういうのの選び方って、鉄則みたいなのあるんですかね。

学習前に、x_trainx_validx_testy_trainy_validy_testの前処理をする。

# 28 x 28の画像がgrayscaleで1chなので、28, 28, 1にreshapeする
x_train = x_train.reshape(x_train.shape[0], 28, 28, 1)
x_valid = x_valid.reshape(x_valid.shape[0], 28, 28, 1)
x_test = x_test.reshape(x_test.shape[0], 28, 28, 1)

# 0-255の整数値を0〜1の小数に変換する
# MNISTって必ずこの処理入るけれど、意味あるのかな
x_train = x_train.astype('float32')
x_valid = x_valid.astype('float32')
x_test = x_test.astype('float32')
x_train /= 255
x_valid /= 255
x_test /= 255

# one-hot vector形式に変換する
y_train = keras.utils.to_categorical(y_train, 10)
y_valid = keras.utils.to_categorical(y_valid, 10)
y_test = keras.utils.to_categorical(y_test, 10)

前処理が済んだら、学習する。

from keras.optimizers import RMSprop

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

history = model.fit(x_train, y_train,
                    batch_size=128,
                    epochs=10,
                    verbose=1,
                    validation_data=(x_valid, y_valid))

学習した結果は、次のようにすると表示できた。
評価は、学習に使っていないx_testy_testで行う。

score = model.evaluate(x_test, y_test, verbose=0)
print('Test loss:', score[0])
print('Test accuracy:', score[1])

我が家の2012年型のMacBook Proで結構な時間をかけて学習した結果が、コレ。

Test loss: 0.046481294665449875
Test accuracy: 0.9889

なかなか良い。昨年の今頃にCNNを使わずにやった結果(0.9184) より、ずっと良い。

学習結果をグラフにしてみる

学習結果は、historyに格納されている。ドキュメントによると、以下のように書いてある。

Historyインスタンス.本インスタンスのhistory属性は訓練時に得られた全ての情報を含みます.

で、Historyってどういう構造になっているのか?っていうのが、Kerasのドキュメントを読んでもよく分からないんだけれど、examplesをいろいろ漁ると分かる。

グラフを描くときは、もちろんmatplotlibを使う。

from matplotlib import pyplot as plt

# 精度のplot
plt.plot(history.history['acc'], marker='.', label='acc')
plt.plot(history.history['val_acc'], marker='.', label='val_acc')
plt.title('model accuracy')
plt.grid()
plt.xlabel('epoch')
plt.ylabel('accuracy')
plt.legend(loc='best')
plt.show()

# 損失のplot
plt.plot(history.history['loss'], marker='.', label='loss')
plt.plot(history.history['val_loss'], marker='.', label='val_loss')
plt.title('model loss')
plt.grid()
plt.xlabel('epoch')
plt.ylabel('loss')
plt.legend(loc='best')
plt.show()

損失の方のグラフはこんな感じ。

loss_plot.png

学習の進捗をグラフに表示する

結果を表示するのではなくて、学習している最中にaccuracyとかlossとかをplotできたら、「ちゃんと動いている」という実感が得られそうだ。そういうときは、matplotlib.pyplotをinteractiveモードで動かしながら、epochごとに結果をplotすれば良い。

epochごとに処理するために、Callback関数を使う。Pythonで最適化問題を解くでも使った手法だ。

具体的には、model.fit()の時に、callbacksとして指定する。詳しくは、ドキュメントを。

class PlotLosses(Callback):
    '''
    学習中のlossについてlive plotする
    '''

    def on_train_begin(self, logs={}):
        '''
        訓練開始時に実施
        '''
        self.epoch_cnt = 0      # epochの回数を初期化
        plt.axis([0, self.epochs, 0, 0.25])
        plt.ion()               # pyplotをinteractive modeにする

    def on_train_end(self, logs={}):
        '''
        訓練修了時に実施
        '''
        plt.ioff()              # pyplotのinteractive modeをoffにする
        plt.legend(['loss', 'val_loss'], loc='best')
        plt.show()

    def on_epoch_end(self, epoch, logs={}):
        '''
        epochごとに実行する処理
        '''
        loss = logs.get('loss')
        val_loss = logs.get('val_loss')
        x = self.epoch_cnt
        # epochごとのlossとval_lossをplotする
        plt.scatter(x, loss, c='b', label='loss')
        plt.scatter(x, val_loss, c='r', label='val_loss')
        plt.pause(0.05)
        # epoch回数をcount up
        self.epoch_cnt += 1

基本的には、on_train_begin()にて、plt.ion()して、on_train_end()で、plt.ioff()する。

そして、on_epoch_end()で、epochの終わりごとにplt.scatter()するだけの、簡単なclassを定義して、そのinstanceをmodel.fit()callbacks=[]に指定するだけ。

ついでにデータも保存しておく

進捗状況をグラフに表示するついでに、epochごとのaccuracylossについても保存しておきたいと思ったら、keras.callbacksCSVLoggerがあるので、これで引数としてファイル名を指定しておくと、CSVファイルにepochごとの結果を保存してくれるので、後からExcel等でグラフを描き直してもイイ。

というワケで、次のようなコードを書くと、進捗状況をグラフ表示しながら、CSVに保存することができるようになる。

plot_losses = PlotLosses()
csv_logger = CSVLogger('trainlog.csv')

history = model.fit(x_train, y_train,
                    batch_size=batch_size, epochs=epochs,
                    verbose=1,
                    validation_data=(x_test, y_test),
                    callbacks=[plot_losses, csv_logger])

ちなみに、実行中のグラフはこんな感じ。

Figure_1.png


本日のコード

というワケで、本日のコードの全貌。
ついでというワケではないけれど、コマンドライン引数で、epochの数とbatch_sizeを指定するようにした。

'''
Keras(+Tensorflow)でMNISTを実施する
学習中の進捗をグラフで表示する
'''

import keras
from keras.datasets import mnist
from keras.models import Sequential
from keras.layers import Dense, Dropout, Flatten, Conv2D, MaxPooling2D
from keras.optimizers import RMSprop
from keras.callbacks import Callback, CSVLogger
from matplotlib import pyplot as plt
from sklearn.model_selection import train_test_split
import argparse


class PlotLosses(Callback):
    '''
    学習中のlossについてlive plotする
    '''

    def on_train_begin(self, logs={}):
        '''
        訓練開始時に実施
        '''
        self.epoch_cnt = 0      # epochの回数を初期化
        plt.axis([0, self.epochs, 0, 0.25])
        plt.ion()               # pyplotをinteractive modeにする

    def on_train_end(self, logs={}):
        '''
        訓練修了時に実施
        '''
        plt.ioff()              # pyplotのinteractive modeをoffにする
        plt.legend(['loss', 'val_loss'], loc='best')
        plt.show()

    def on_epoch_end(self, epoch, logs={}):
        '''
        epochごとに実行する処理
        '''
        loss = logs.get('loss')
        val_loss = logs.get('val_loss')
        x = self.epoch_cnt
        # epochごとのlossとval_lossをplotする
        plt.scatter(x, loss, c='b', label='loss')
        plt.scatter(x, val_loss, c='r', label='val_loss')
        plt.pause(0.05)
        # epoch回数をcount up
        self.epoch_cnt += 1


def plot_result(history):
    '''
    plot result
    全ての学習が終了した後に、historyを参照して、accuracyとlossをそれぞれplotする
    '''

    # accuracy
    plt.figure()
    plt.plot(history.history['acc'], label='acc', marker='.')
    plt.plot(history.history['val_acc'], label='val_acc', marker='.')
    plt.grid()
    plt.legend(loc='best')
    plt.title('accuracy')
    plt.savefig('graph_accuracy.png')
    plt.show()

    # loss
    plt.figure()
    plt.plot(history.history['loss'], label='loss', marker='.')
    plt.plot(history.history['val_loss'], label='val_loss', marker='.')
    plt.grid()
    plt.legend(loc='best')
    plt.title('loss')
    plt.savefig('graph_loss.png')
    plt.show()


def main(epochs=5, batch_size=128):
    '''
    MNISTの学習とその結果の表示
    @args:
        epochs: epochの回数
        batch_size: ミニバッチのサイズ
    '''

    # load MNIST data
    (x_train, y_train), (x_test, y_test) = mnist.load_data()
    x_train1, x_valid, y_train1, y_valid = train_test_split(x_train, y_train, test_size=0.175)
    x_train = x_train1
    y_train = y_train1

    x_train = x_train.reshape(x_train.shape[0], 28, 28, 1).astype('float32')/255
    x_valid = x_valid.reshape(x_valid.shape[0], 28, 28, 1).astype('float32')/255
    x_test = x_test.reshape(x_test.shape[0], 28, 28, 1).astype('float32')/255

    # convert one-hot vector
    y_train = keras.utils.to_categorical(y_train, 10)
    y_valid = keras.utils.to_categorical(y_valid, 10)
    y_test = keras.utils.to_categorical(y_test, 10)

    # create model
    model = Sequential()
    model.add(Conv2D(32, kernel_size=(3, 3), activation='relu', input_shape=(28, 28, 1)))
    model.add(Conv2D(64, (3, 3), activation='relu'))
    model.add(MaxPooling2D(pool_size=(2, 2)))
    model.add(Dropout(0.25))
    model.add(Flatten())
    model.add(Dense(128, activation='relu'))
    model.add(Dropout(0.5))
    model.add(Dense(10, activation='softmax'))

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

    print(model.summary())

    # callback function
    plot_losses = PlotLosses()      # グラフ表示(live plot)
    plot_losses.epochs = epochs
    csv_logger = CSVLogger('trainlog.csv')

    # train
    history = model.fit(x_train, y_train,
                        batch_size=batch_size, epochs=epochs,
                        verbose=1,
                        validation_data=(x_valid, y_valid),
                        callbacks=[plot_losses, csv_logger])

    # result
    score = model.evaluate(x_test, y_test, verbose=0)
    print('Test loss: {0}'.format(score[0]))
    print('Test accuracy: {0}'.format(score[1]))

    plot_result(history)


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='MNIST')
    parser.add_argument('--epochs', dest='epochs', type=int, help='size of epochs')
    parser.add_argument('--batch_size', dest='batch_size', type=int, help='size of batch')
    args = parser.parse_args()
    if args.epochs:
        epochs = args.epochs
    else:
        epochs = 5
    if args.batch_size:
        batch_size = args.batch_size
    else:
        batch_size = 128

    main(epochs, batch_size)

graph_accuracy.png

graph_loss.png