初心者による機械学習メモ

PythonでChainerを使った機械学習を始めました。今まで、Webアプリケーションなどを作ったことはあったけど、Pythonや機械学習は扱ったことがありませんでした。とりあえず、当ページにメモを残しながら勉強を進めたいと思います。

前提

・macOS 10.12、Python 2.7を使っています。
・Chainerのバージョンは、1.18.0あたりです。
・Chainerは、CPUのみで使っています。（GPU未使用）

Python

テンプレート

私がPythonスクリプトを書き始める時のテンプレートです。

template.py

# coding:utf-8

import argparse
import logging
import os
import sys

logger = logging.getLogger(__name__)

class App:
    def run(self, args):
        logger.info(args)

if __name__ == '__main__':
    logging.basicConfig(format='%(asctime)s %(name)s %(levelname)s %(message)s')
    logger.setLevel(logging.INFO)

    parser = argparse.ArgumentParser()
    parser.add_argument('--debug', action='store_true', help='デバッグログを出力する。')
    parser.add_argument('arg1', help='arg1 help')
    args = parser.parse_args()
    if args.debug:
        logger.setLevel(logging.DEBUG)
    logger.debug('Start')

    App().run(args)

    logger.debug('End')

パス操作

パス（ディレクトリ名、ファイル名）を分割したり、連結したりします。

path01.py

import os
# パスからディレクトリ、ファイル名、拡張子を分割
print(__file__) #'parentdir/path01.py'
print(os.path.dirname(__file__)) #'parentdir'
print(os.path.basename(__file__)) #'path01.py'
print(os.path.splitext(os.path.basename(__file__))) #'path01'
# パス連結
print(os.path.join('parentdir', 'subdir')) #'parentdir/subdir'
# 特殊記号を展開
print(os.path.expanduser('~'))

パスの存在チェックなどをします。

path02.py

import os
if os.path.exists('dir1/file1.txt'):
    print('指定パスが存在します')
if os.path.isdir('dir1/dir2'):
    print('ディレクトリです')
if os.path.isfile('dir1/file1.txt'):
    print('ファイルです')

日時関連

日付、時刻、日時を取得します。datetimeモジュールのdate, datetime, timeクラスを使って取得できました。

datetime01.py

import datetime
# 現在日時
date1a     = datetime.date.today()
datetime1a = datetime.datetime.now()
time1a     = datetime.datetime.now().time()
# 指定日時
date1b     = datetime.date(2017, 12, 31)
datetime1b = datetime.datetime(2017, 12, 31, 10, 30, 12)
datetime1b = datetime.datetime.combine(datetime.date(2017, 12, 31), datetime.time(10, 30, 12))
time1b     = datetime.time(10, 30, 12)
# 文字列から解析
# datetimeクラスでは、strptimeメソッドで解析できる。date, timeクラスでは、strptimeメソッドがないので工夫する。
date1c     = datetime.datetime.strptime('2017/3/12' + ' 00:00:00', '%Y/%m/%d %H:%M:%S').date()
datetime1c = datetime.datetime.strptime('2017/3/12 10:10:44', '%Y/%m/%d %H:%M:%S')
time1c     = datetime.datetime.strptime('2000/1/1' + ' 10:30:12', '%Y/%m/%d %H:%M:%S').time()

日付、時刻、日時を計算します。基本的には、timedeltaクラスを使って計算するみたいです。timedeltaクラスで扱える日時の単位は、days, seconds, microseconds, milliseconds, minutes, hours, weeksであり、1ヶ月後とか1年後などは工夫が必要になりました。python-dateutilというライブラリを使えば楽になるらしいですが、まだ使ってみてないです。

datetime02.py

import datetime
# 各クラスのreplaceメソッドを使う方法
date2a     = date1a.replace(year=date1a.year + 1)
datetime2a = datetime1a.replace(year=datetime1a.year + 1, hour=datetime1a.hour + 2)
time2a     = time1a.replace(hour=time1a.hour + 2)
# timedeltaクラスを使う方法。timeクラスに対しては使えない模様。
date2b     = date1a + datetime.timedelta(days=1)
datetime2b = datetime1a + datetime.timedelta(days=1, hours=2)

日付、日時、時刻の文字列を取得します。

datetime03.py

import datetime
datetime.date.today().isoformat() #'2017-03-12'
datetime.date.today().strftime('%Y/%m/%d') #'2017/03/12'
datetime.datetime.now().isoformat() #'2017-03-12T10:10:44.303608'
datetime.datetime.now().strftime('%Y%m%d-%H%M%S') #'20170312-101044'
datetime.datetime.now().time().isoformat() #'10:10:44.303608'
datetime.datetime.now().time().strftime('%H%M%S') #'101044'

列挙型（enum）

Python3.4からは、列挙型（enum）が使えるようになっています。
Python2.4以降でも利用可能で、「pip install enum34」でインストールすれば使えました。

簡単な使い方です。

python

from enum import Enum

class Color(Enum):
    RED = 1
    GREEN = 2
    BLUE = 3

# コード値と名前の相互変換
Color(2).name        #'GREEN'
Color['GREEN'].value #2

# 比較
Color(3) is Color.BLUE
Color(3) == Color.BLUE

# 要素数
len(Color) #3

# 全要素アクセス
for color in Color:
    print(color)

列挙型には独自メソッドも追加できました。

enum02.py

class Color2(Enum):
    RED = 1
    GREEN = 2
    BLUE = 3

    def to_jp(self):
        return ['赤', '緑', '青'][self.value - 1]

参考サイト
- Python 3.4 から標準ライブラリに入る Enum 型が今からでも便利 - Qiita
- 8.13. enum — 列挙型のサポート — Python 3.6.1 ドキュメント

テキストファイル

次のようにしてテキストファイルを読み書きできました。

textfile01.py

# 書き込み
with open('a.txt', 'w') as f:
    f.write('line1\n')
with open('b.txt', 'w') as f:
    f.writelines(['line1\n','line2\n','line3\n'])
# 読み込み
with open('a.txt', 'r') as f:
    for line in f:
        print(line)
with open('b.txt', 'r') as f:
    arr = f.readlines()
    print(arr)

JSON

Pythonの標準機能でJSONを簡単に扱えました。

json01.py

import json

# エンコード
json.dumps(['foo', {'bar': ('baz', None, 1.0, 2)}])
#'["foo", {"bar": ["baz", null, 1.0, 2]}]'
json.dumps({"c": 0, "b": 0, "a": 0}, sort_keys=True)
#'{"a": 0, "b": 0, "c": 0}'
# エンコードしてファイルへ保存
f = open('test.json', 'w')
json.dump({"c": 0, "b": 0, "a": 0}, f, sort_keys=True)
f.close()

# デコード
json.loads('["foo", {"bar":["baz", null, 1.0, 2]}]')
#[u'foo', {u'bar': [u'baz', None, 1.0, 2]}]
# ファイルから読み込んでデコード
f = open('test.json')
json_obj = json.load(f)
f.close()

単体テスト

Pythonの標準機能で簡単な単体テストライブラリが用意されていました。開発規模が大きくなってきても、単体テストケースを作っておくと安心して開発できますよね。

テストケースの書き方は、次のような感じ。ファイルの名前は、"test_"+テスト対象クラス名+".py"。テストクラスの名前は、"Test"+テスト対象クラス名 となるのが多い感じがする。でも、"Test"を頭につけるのと後ろにつけるのとどっちがいいのかな。

unittest01.py

import unittest

class TestString(unittest.TestCase):
    def test_upper(self):
        # 準備
        str = 'foo'
        # テスト実行
        result = str.upper()
        # 検証
        self.assertEqual(result, 'FOO')

if __name__ == '__main__':
    unittest.main()

コマンドプロンプトから、次の2種類の呼び出し方ができます。

python -m unittest unittest01 #方法1 この場合はテストケースに「unittest.main()」が不要となる。
python unittest01.py          #方法2

テストケースの前後に、初期化処理や後処理が必要な場合は、setUp(self), tearDown(self), setUpClass(cls), tearDownClass(cls)メソッドをオーバーライドすることで実装できるようです。
アサーションメソッドは、色々用意されていました。よく使いそうなのは、assertEqual(a,b,msg=None), assertNotEqual(a,b,msg=None), assertTrue(expr,msg=None), assertFalse(expr,msg=None), assertIsNone(expr,msg=None), assertIsNotNone(expr,msg=None), assertIn(a,b,msg=None), assertNotIn(a,b,msg=None)かな。その他にも、数値の大小比較や、文字列の正規表現マッチングなどもありました。
テストスイートもありました。

参考サイト
- 25.3. unittest — ユニットテストフレームワーク — Python 2.7.13 ドキュメント

その他

連携して動作するPythonスクリプトが増えてきた場合、ディレクトリを分けて管理したくなりました。
その場合は、次の手順を行えば対応可能でした。
なお、ディレクトリ名とスクリプト名を同じにするとうまく呼び出せないようです。

• Pythonスクリプトをグルーピングするためのディレクトリを作成する。
• 作成したディレクトリにPythonスクリプトを移動する。
• 作成したディレクトリに、「__init__.py」という名前で0バイトファイルを作成する。
• 呼び出し元となるPythonスクリプトに記述しているimport文を、移動先のディレクトリ名に合わせて変更する。（例）「import foo」→「import dir.foo」

NumPy

numpy01.py

import numpy as np

# タプルやリストをnumpy配列に変換する。
arr = (1, 2, 3)
numpy_arr1 = np.array(arr)
numpy_arr2 = np.array(arr, dtype=np.float32)

# 分類器が出力した各要素の確信度の配列から、
# 最も確信度が高い要素のインデックスを取得する。
y_data = np.array([[0.10, 0.20, 0.70]])
index = y_data.argmax(axis=1)

Chainer

import文です。Chainerのサイトで次のように書くのが推奨されていました。

chainer_import01.py

import numpy as np
import chainer
from chainer import cuda, Function, gradient_check, report, training, utils, Variable
from chainer import datasets, iterators, optimizers, serializers
from chainer import Link, Chain, ChainList
import chainer.functions as F
import chainer.links as L
from chainer.training import extensions

モデル定義

モデルは次のように定義するようです。最適なモデルを定義するためには、試行錯誤やノウハウが必要そうなので、有識者が定義したモデルを使わせてもらうのが良さそうです。Chainerのexampleにもいくつかモデルが定義されているので、これも使えそうです。

chainer_mlp.py

# chainer-1.18.0/examples/mnist/train_mnist.pyを参考

# Network definition
# 多層パーセプトロン（Multi Layer Perceptron）
class MLP(Chain):
    insize = 28

    def __init__(self, n_in=28*28, n_units=100, n_out=10):
        super(MLP, self).__init__(
            l1=L.Linear(n_in, n_units),  # n_in -> n_units
            l2=L.Linear(n_units, n_units),  # n_units -> n_units
            l3=L.Linear(n_units, n_out),  # n_units -> n_out
        )

    def __call__(self, x):
        h1 = F.relu(self.l1(x))
        h2 = F.relu(self.l2(h1))
        h3 = self.l3(h2)
        return h3

モデルを保存、読み込みする処理です。Chainerではnpzやhdf5形式でファイル保存/読み込みする処理が用意されています。これを使うのが基本だとは思いますが、読み込み時に事前にモデルオブジェクトを生成しておく必要があるので、少々使いづらく思いました。Pythonで用意されているpickleを使うと改善できましたが、ディープなモデルだとファイルサイズがかなり大きくなってしまいました。

chainer_model_repository01.py

class ModelRepository:
    def save_npz(self, path, model):
        assert(os.path.splitext(path)[1] == '.npz')
        serializers.save_npz(path, model)
    def load_npz(self, path, model):
        assert(os.path.splitext(path)[1] == '.npz')
        serializers.load_npz(path, model)
        return model

    def save_pkl(self, path, model):
        assert(os.path.splitext(path)[1] == '.pkl')
        pickle.dump(model, open(path, 'wb'), pickle.HIGHEST_PROTOCOL)
    def load_pkl(self, path):
        assert(os.path.splitext(path)[1] == '.pkl')
        model = pickle.load(open(path, 'rb'))
        return model

学習

機械学習する処理です。Chainer 1.11.0で導入されたtrainerというオブジェクトで学習をさせるようです。モデルオブジェクトに対しては、Classifierクラスでラップし、lossを計算する必要があります。

chainer_train01.py

# coding:utf-8
# MNIST手書き数字を学習する。

# from __future__ import print_function
import argparse
import logging
import os
import sys

import numpy as np
import chainer
from chainer import cuda, Function, gradient_check, report, training, utils, Variable
from chainer import datasets, iterators, optimizers, serializers
from chainer import Link, Chain, ChainList
import chainer.functions as F
import chainer.links as L
from chainer.training import extensions

import mlp
import neural_network as nn

logger = logging.getLogger(__name__)

class App:
    def __init__(self):
        self.args = None

    def run(self, args):
        self.args = args

        model = L.Classifier(self.get_model())
        optimizer = self.get_optimizer(model)
        train_iter, test_iter = self.get_dataset_iter()
        trainer = self.get_trainer(model, optimizer, train_iter, test_iter)
        if args.resume:
            chainer.serializers.load_npz(args.resume, trainer)

        print('# Train-size: {}'.format(len(train_iter.dataset)))
        print('# Test-size: {}'.format(len(test_iter.dataset)))
        print('# Minibatch-size: {}'.format(args.batchsize))
        print('# epoch: {}'.format(args.epoch))
        trainer.run()

        self.save_model(model.predictor)

    def get_model(self):
        model = ModelFactory().create(self.args.arch)
        return model

    def get_optimizer(self, model):
        optimizer = optimizers.Adam()
        optimizer.setup(model)
        return optimizer

    def get_dataset_iter(self):
        train, test = self.get_dataset()
        train_iter = chainer.iterators.SerialIterator(train, args.batchsize)
        test_iter = chainer.iterators.SerialIterator(test, args.batchsize, repeat=False, shuffle=False)
        return train_iter, test_iter

    def get_dataset(self):
        train, test = DataFactory().create(self.args.debug_smalldata)
        return train, test

    def get_trainer(self, model, optimizer, train_iter, test_iter):
        updater = training.StandardUpdater(train_iter, optimizer)
        trainer = training.Trainer(updater, (args.epoch, 'epoch'), out=args.out)
        trainer.extend(extensions.Evaluator(test_iter, model))
        trainer.extend(extensions.dump_graph('main/loss'))
        trainer.extend(extensions.snapshot(), trigger=(args.epoch, 'epoch'))
        trainer.extend(extensions.LogReport())
        trainer.extend(extensions.PrintReport(
            ['epoch', 'main/loss', 'validation/main/loss',
             'main/accuracy', 'validation/main/accuracy', 'elapsed_time']))
        trainer.extend(extensions.ProgressBar())
        return trainer

    def save_model(self, model):
        nn.ModelRepository().save_npz(os.path.join(args.out, self.args.arch + '.npz'), model)
        logger.debug('Saved the model to %s' % args.out)

class ModelFactory:
    archs = {
        'mlp': mlp.MLP,
    }

    def create(self, arch):
        return ModelFactory.archs[arch]()

class DataFactory:
    def create(self, debug_smalldata=False):
        train, test = chainer.datasets.get_mnist()
        if debug_smalldata:
            train = train[0:10]
            test = test[0:10]
        return train, test

if __name__ == '__main__':
    logging.basicConfig(format='%(asctime)s %(name)s %(levelname)s %(message)s')
    logger.setLevel(logging.INFO)

    parser = argparse.ArgumentParser()
    parser.add_argument('--debug', action='store_true',
                        help='デバッグログを出力する。')
    parser.add_argument('--debug-smalldata', action='store_true',
                        help='少ないデータで疎通確認する。')
    parser.add_argument('--arch', '-a', choices=ModelFactory().archs.keys(), default='nin',
                        help='Convnet architecture')
    parser.add_argument('--batchsize', '-b', type=int, default=100,
                        help='Number of images in each mini-batch')
    parser.add_argument('--epoch', '-e', type=int, default=20,
                        help='Number of sweeps over the dataset to train')
    parser.add_argument('--out', '-o', default='result',
                        help='Directory to output the result')
    parser.add_argument('--resume', '-r', default='',
                        help='Resume the training from snapshot')
    args = parser.parse_args()
    if args.debug:
        logger.setLevel(logging.DEBUG)
    logger.debug('Start')

    App().run(args)

    logger.debug('End')

予測

学習したモデルを使って、予測をしてみます。学習するときは、Classifierクラスを使っていましたが、予測するときは使いません。

chainer_predict01.py

# coding:utf-8
# MNIST手書き数字を予測する。

import argparse
import logging
import os
import sys

import numpy as np
import chainer
from chainer import cuda, Function, gradient_check, report, training, utils, Variable
from chainer import datasets, iterators, optimizers, serializers
from chainer import Link, Chain, ChainList
import chainer.functions as F
import chainer.links as L
from chainer.training import extensions

import util

logger = logging.getLogger(__name__)

class App:
    def __init__(self):
        self.args = None

    def run(self, args):
        self.args = args

        model = self.load_model()
        train, test = self.get_dataset()
        x_arr, t_arr = self.split_xt(test[0:10])
        x_arr = np.array(x_arr)
        t_arr = np.array(t_arr)

        print('t %s' % str(t_arr))
        y_arr = model(Variable(x_arr)).data.argmax(axis=1)
        print('y %s' % str(y_arr))

    def load_model(self):
        model = nn.ModelRepository().load(args.model, 'model')
        logger.debug('Loaded the model from %s' % args.model)
        return model

    def get_dataset(self):
        train, test = chainer.datasets.get_mnist()
        return train, test

    def split_xt(self, xt_arr):
        x_arr = []
        t_arr = []
        for x, t in xt_arr:
            x_arr.append(x)
            t_arr.append(t)
        return x_arr, t_arr

if __name__ == '__main__':
    logging.basicConfig(format='%(asctime)s %(name)s %(levelname)s %(message)s')
    logger.setLevel(logging.INFO)

    parser = argparse.ArgumentParser()
    parser.add_argument('--debug', action='store_true', help='デバッグログを出力する。')
    parser.add_argument('--model',
                        help='Directory to input the model')
    args = parser.parse_args()
    if args.debug:
        logger.setLevel(logging.DEBUG)
    logger.debug('Start')

    App().run(args)

    logger.debug('End')

用語

機械学習に関連して頻出する用語をメモしておきます。

• 機械学習（machine learning）
• ニューラルネットワーク（neural network, NN）：脳機能を計算機上に表現するモデル。Wikipediaリンク
• パーセプトロン（perceptron）：1957年に考案されたニューラルネットワークの一種。単純なパーセプトロンでは、AND,OR計算を表現できるが、XOR計算は表現できない。パーセプトロンを2層に重ねるとXORも表現できる。複数の層を持つパーセプトロンを多層パーセプトロン（multi-layered perceptron, MLP）と呼ぶ。Wikipediaリンク
• 畳み込みニューラルネットワーク（convolutional neural network, CNN）：畳み込み層（Convolutionレイヤ）、プーリング層（Poolingレイヤ）を組み合わせて構築するニューラルネットワーク。畳み込み層では、カーネル、パディング（padding）、ストライド（stride）という考え方がある。代表的なCNNとしては、LeNet（1998年考案、最も初期のCNN）、AlexNetなどがある。
• ディープラーニング：ニューラルネットワークの層を深くしたネットワーク。なぜ層を深くすると認識率が上がるのか、感覚的には説明できても理論的には解明されていないらしい。主なネットワークとしては、VGG（VGG16, VGG19）、GoogLeNet、ResNetなどがある。
• 全結合層（full-connected layer）
• 自己符号化器（autoencoder）

• 再帰型ニューラルネット（recurrent neural network, RNN）

• 活性化関数（activation function）：ニューラルネットワークで、入力信号の総和を出力信号に変換する関数。単純な活性化関数としてはステップ関数がある。その他に、シグモイド関数（sigmoid function）、正規化線形関数（rectified linear unit, ReLU）、ソフトマックス関数などがある。ReLUは、他の活性化関数に比べてシンプルで計算が早いため、近年広く使われているらしい。Wikipediaリンク

• ソフトマックス関数（softmax function）：活性化関数の一種。当関数の出力は、0.0〜1.0の実数となり、総和が1になる。そのため、当関数の出力を「確率」として解釈することができる。分類問題でニューラルネットワーク学習時の出力層に多く使われる。

• 最適化関数（optimization function）：ニューラルネットワーク学習時に、ネットワークの重みを更新する量を決めるために使う関数。確率的勾配効果法（stochastic gradient descent, SGD）、モーメンタム（momentum）、AdaGrad、Adamなどがある。学習対象の問題によって、得手、不得手があるらしいが、SGDやAdamがよく使われている模様。

• 順伝播（foward propagation）

• 逆伝播（backward propagation）

• 過学習（overfitting）

• 訓練データ（教師データ）、テストデータ

• 損失関数（loss function）：ニューラルネットワークの性能の悪さを表す指標。2乗和誤差（mean squared error）、交差エントロピー誤差（cross entropy error）などがある。

• ミニバッチ学習

• 学習率（learning rate）

• 勾配降下法（gradient descent method）

• ハイパーパラメータ

• エポック（epoch）：学習回数を表す単位。全ての訓練データを1回使ったら1エポック。

• 勾配消失（gradient vanishing）

• 重みの初期値：ニューラルネットワークの重みに適切な初期値を与えると、安定して効率的な学習が行えるようになる。「Xavierの初期値」、「Heの初期値」などがある。

• Batch Normalization：各層における入力データの分布を正規化する手法。学習を早くできて、初期値にあまり依存しなくなり、過学習を抑制する効果が望める。

• Dropout

• データ拡張（data augmentation）：訓練データを加工してデータ数を増やす手法。crop処理したり、flip処理（左右反転）したり、輝度変更、拡大・縮小したりする。

• MNISTデータセット：手書き数字の画像セット。機械学習の分野で有名なデータセット。

• CIFAR-10データセット：物体カテゴリ認識のためのデータセット。サイズ32x32pxのカラー画像、10クラス。リンク

• よく出てくる英単語：予測する（predict）。正確さ、精度（accuracy）。正規化（normalization）。微分（differentiation）。勾配（gradient）。

参考サイト

Python 2.7 ドキュメント
Chainer