以前に、私的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_train、x_valid、x_test、y_train、y_valid、y_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_test、y_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()
損失の方のグラフはこんな感じ。
学習の進捗をグラフに表示する
結果を表示するのではなくて、学習している最中に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ごとのaccuracyやlossについても保存しておきたいと思ったら、keras.callbacksにCSVLoggerがあるので、これで引数としてファイル名を指定しておくと、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])
ちなみに、実行中のグラフはこんな感じ。
本日のコード
というワケで、本日のコードの全貌。
ついでというワケではないけれど、コマンドライン引数で、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)
