にせねこメモ

はてなダイアリーがUTF-8じゃないので移ってきました。

Chainerを触ってみる: XOR関数を訓練する

Python Chainer 機械学習

※この記事ではPython 3.5.3とChainer v1.20.0.1を使っています。

調べながら書いています。間違い等あればご指摘願います。

はじめに

『ゼロから作るDeep Learning』を一通り読んで、実際にライブラリ使って機械学習してみようという段、Chainerを触ってみたものの、deep learning界のHello worldとも呼ばれるMNISTのexampleを見ても、なにをしているのかよくわからない。

なので、もっと単純なネットワークを作って訓練してみることで、chainerの使い方を勉強して、MNISTの例で何をやってるのかが理解できるようにすることを目指した。

Chainerの基本的なオブジェクト

chainer.Variable

  • 変数に相当する。
  • 後述するFunctionやLinkは入力がVariableで、出力もVariableである。
    • 入力に関してはnumpy.ndarrayを与えると内部で変換してくれるのでどちらでも良い感じかも。
  • .dataでデータ配列numpy.ndarray (あるいはcupy.ndarray)がとれる。

chainer.Function

  • 関数に相当する。訓練すべきパラメータをもたない。
  • chainer.functions以下に(chainer.Functionを継承したものが)色々定義されている。
  • activation関数, loss関数, accuracy関数などなど

chainer.Link

  • 訓練すべきパラメータをもつ関数を表す。
  • chainer.links以下にいろいろ定義されている。例:
    • Linear (全結合)
    • Convolution2D (2次元CNN)
  • ミニバッチを入力とするので入力の次元に注意。

パーセプトロン

f:id:nixeneko:20170531234148p:plain
いくつかの入力を重みづけして足し合わせ、さらに定数(バイアス)を足し、その結果に適当な非線形関数(activation関数という)を適用したものを(単純)パーセプトロンという。

AND関数を作ってみる

ANDは単純パーセプトロンで実装できる。
f:id:nixeneko:20170601000053p:plain

  • AND関数は入力2ノード、出力1ノードの関数であり、全結合のLinearを使って実装できる。
  • initialW, initial_biasを設定することで、ネットワークのパラメータを決め打ちしている。
  • Linearはミニバッチを入力とすることに注意する。
  • activation関数にはReLUを使った。
import chainer
import chainer.links as L
import chainer.functions as F
import numpy as np

# L.Linear(in_size, out_size, wscale=1, bias=0, nobias=False, 
#          initialW=None, initial_bias=None)
W = np.array([[1,1]]) # weight
b = -1.0 # bias
a = L.Linear(2, 1, initialW=W, initial_bias=b) # full-connected layer
f = F.relu # activation function
def AND(x): # 順伝播を定義
    return f(a(x))

# use network as a function
x = np.array([[1,1],[0,1],[1,0],[0,0]], dtype=np.float32) # input data
print(AND(x).data)
#[[ 1.]
# [ 0.]
# [ 0.]
# [ 0.]]

実行すると、[1,1]の入力に対しては[1]が、それ以外には[0]が帰ってきていることがわかる。これはまさにANDの挙動である。

多層パーセプトロン(MLP)

単純パーセプトロンを複数、層状に組み合わせたものを多層パーセプトロンという。各層はベクトルを入力としベクトルを出力とする関数になっていて、全体で複雑な非線形関数を実現できる。

次は3層の例で、バイアスやactivation関数の記載は省略した。*1
f:id:nixeneko:20170531234740p:plain

XOR関数を作ってみる

同様にXOR関数を作ってみる。
XORは線形分離不可能であるので単純パーセプトロンでは再現できず、全結合層を2つ用意する必要がある。中間層(隠れ層)のノード数は2にした。
f:id:nixeneko:20170601000835p:plain

  • 全結合層(Linear)2層
  • 入力層はノード2つ, 中間層ノード2つ, 出力層ノード1つ
  • 層の重みをinitialW, initial_biasで設定。うまくいく値を決め打ちで使った。
  • activation関数として第一層にはReLUを使った。(第二層はidentity functionを使ったというか、そのままの値)
  • 前層の出力と次層の入力でノードの数を合わせる。
    • Linearの第1引数(in_size)をNoneにすると、最初に入力が与えられたときに自動的にサイズを推定してくれる。
import chainer
import chainer.links as L
import chainer.functions as F
import numpy as np

# L.Linear(in_size, out_size, wscale=1, bias=0, nobias=False, 
#          initialW=None, initial_bias=None)
# 第1層
W1 = np.array([[1,-1],[-1,1]]) # weight parameters
b1 = 0.0 # bias
a1 = L.Linear(2, 2, initialW=W1, initial_bias=b1) #入力2, 出力2
f1 = F.relu # activation function

# 第2層
W2 = np.array([[1,1]]) # weight parameters
b2 = 0.0 # bias
a2 = L.Linear(2, 1, initialW=W2, initial_bias=b2) #入力2, 出力1

# 順伝播を定義
def XOR(x):
    h1 = f1(a1(x))
    return a2(h1)

# 試してみる
x = np.array([[1,1],[1,0],[0,1],[0,0]], dtype=np.float32)
print(XOR(x).data)
#[[ 0.]
# [ 1.]
# [ 1.]
# [ 0.]]

これも実際に動かしてみると、入力の[1,1]と[0,0]には[0]を、[0,1]と[1,0]には[1]を返していて、XORを表現した関数となっている。

chainer.Chain

ChainでLinkオブジェクト(=パラメータ付き関数)をまとめることができる。
実際にはChainを継承したクラスをつくって、順伝播をその中の__call__メソッドに定義するという使い方をするようだ。
先ほどのXORを書き換えると次のようになる。

import chainer
import chainer.links as L
import chainer.functions as F
import numpy as np

# parameters
W1 = np.array([[1,-1],[-1,1]])
b1 = 0.0
W2 = np.array([[1,1]])
b2 = 0.0

class Xor(chainer.Chain):  #chainer.Chainを継承したクラスで全体の関数を表現
    def __init__(self): # constructor
        #親クラスchainer.Chainのコンストラクタでlinkをまとめる
        super(Xor, self).__init__( 
            a1 = L.Linear(2, 2, initialW=W1, initial_bias=b1),
            a2 = L.Linear(2, 1, initialW=W2, initial_bias=b2)
        )

    def __call__(self, x): # 順伝播の定義
        h1 = F.relu(self.a1(x))
        return self.a2(h1)

# 定義したものを使ってみる
XOR = Xor()
x = np.array([[1,1],[1,0],[0,1],[0,0]], dtype=np.float32)
print(XOR(x).data)
#[[ 0.]
# [ 1.]
# [ 1.]
# [ 0.]]

chainer.Chainを継承したXorクラスのインスタンスに入力を与えて呼び出すと計算結果が出力される。

さて、ここではXORネットワークのパラメータを手で予め求めた値に設定したが、これをデータから学習させてみたい。

Chainerにおける訓練の流れ

多分こんな感じ。

  1. 学習用データセットの用意
  2. lossを計算するネットワークモデルを用意(これを訓練する)
  3. Optimizerを用意してモデルをセット
  4. Updaterを用意
  5. Trainerを準備
  6. 訓練ループの実行

データセットはモデルを定義するのと同時に考えないといけないと思うので最初に入れているが、MNISTのサンプルなどではOptimizerとUpdaterの間に入っている。

これより後のコードでは

import chainer
import chainer.links as L
import chainer.functions as F
import numpy as np
from chainer.training import extensions

を前提とする。
最後の方に訓練用のコード全体を載せておいた。

1. 学習用データセットの準備

学習用データセットは、入力データと正解データの組である。
今回はXORの2入力がそれぞれ0か1のどちらかの値をとるので、入力は4種類ある。それぞれに対して対応する出力を用意する。
なお、全結合Linearが入力・出力としてnumpy.float32を要求するのでデータ形式をそのように設定している。

indata = np.array([[0,0],[0,1],[1,0],[1,1]], dtype=np.float32)
labels = np.array([ [0],  [1],  [1],  [0] ], dtype=np.float32)

dataset = chainer.datasets.TupleDataset(indata, labels)
train_iter = chainer.iterators.SerialIterator(dataset, 4) # 訓練用. batch size = 4
test_iter = chainer.iterators.SerialIterator(dataset, 4, repeat=False, shuffle=False)
                                                          # 評価用. batch size = 4
  • TupleDatasetは内部で単に2つの入力をタプルにするくらいのことしかしてなくて、あまり本質ではないっぽい。
  • 重要なのはその後のSerialIteratorなどのイテレータで、これがあるとミニバッチ単位でデータを取り出すのが簡単になる。
  • 訓練用と評価用のイテレータを用意する。
    • 今回は訓練セットと評価セットを全く同じにしているが、普通は適当に分ける。
    • どちらもミニバッチサイズを4とした。
  • 訓練用のイテレータは後のUpdaterに渡される。
  • 評価用のイテレータは特になくても訓練はできるはずだけれど、訓練中のモデルの途中経過のaccuracyとかを見るのに使われる。
    • repeat=False, shuffle=False を設定しておく(デフォルトではTrue)。特に repeat=False としないと評価が終わらなくなる。

2. lossを返すモデルの用意

訓練するモデルは、呼び出すと、つまり、 .__call__ メソッドが:

  • 用意したデータセット(のミニバッチ)を入力として受け付ける
  • lossを返す

ようなネットワークである必要がある*2。

#L.Linear(in_size, out_size, wscale=1, bias=0, nobias=False,
#         initialW=None, initial_bias=None)
class Xor(chainer.Chain):  #chainer.Chainを継承したクラスで全体の関数を表現
    def __init__(self): # constructor
        super(Xor, self).__init__( #chainer.Chainでlinkをまとめる
            l1 = L.Linear(2, 10),
            l2 = L.Linear(10, 1)
        )

    def __call__(self, x): #forward propagation definition
        h1 = F.relu(self.l1(x))
        return self.l2(h1)

my_xor = Xor() # 最終的に求めたい関数を示すモデル
accfunc = lambda x, t: F.sum(1 - abs(x-t))/x.size #ないと動かないのでとりあえず用意
model = L.Classifier(my_xor, lossfun=F.mean_squared_error, accfun=accfunc) 
                                      # datasetのデータを入力するとlossを返すモデル
  • ここで、Xorは上で定義したクラスとほとんど同じだけれど、中間層のノードを10に増やしてある*3。
  • また、initialWやinitial_biasの設定を行わず、パラメータはランダムな値で初期化される様になっている。
  • Xorのインスタンスmy_xorを作成する。
    • これは用意したデータセットのデータをそのまま入力とすることができず、さらに呼び出すとXORの計算結果を返すため、そのままではOptimizerに渡すことはできない*4。
    • そのため、用意したデータセットを入力として受け付けることができ、lossを返すようなネットワーク(ここではClassifier)でmy_xorをくるむ。
    • Classifierのデフォルトのaccuracy評価関数はnumpy.int32を入力としてとるので、代わりにnumpy.float32を入力としてとれる関数accfuncを用意して設定している。訓練には必須ではないのでダミーでもよい。

ここでは、最終的に得たい関数を表すネットワークをlossを返すネットワークでくるんでいるが(MNISTのサンプルも同様である)、他方PaintsChainerなどでは、.callメソッドで最終的に得たい関数を計算するようにして、.__call__ではlossを計算して返すようなモデルを使う実装になっていた。

3. Optimizerの用意

lossを見てモデルのパラメータを更新するのがOptimizerである。lossの計算もOptimizerの中で行うことができる*5。
これはUpdaterから呼び出される。

  • 色々な最適化手法がchainer.optimizers以下に用意されていて、適当なものを使う(ここではAdam)。
  • インスタンスを生成して.setupメソッドに2.で用意したlossを返すモデルをセットする。
  • また、必要があればOptimizerにフック関数を設定して、例えば重み減衰による正則化などが設定できる。
optimizer = chainer.optimizers.Adam()
optimizer.setup(model) #lossを返すモデルをセット
#optimizer.add_hook(chainer.optimizer.WeightDecay(0.0005))

4. Updaterの用意

Updaterの役割は、訓練用データセットのイテレータからデータのミニバッチを読み込んで、lossを計算してoptimizerでモデルのパラメータを更新することである。

今回はStandardUpdaterをそのまま使うが、複雑なモデルとかだと、StandardUpdaterを継承したUpdaterを作成してパラメータを更新するということもできる。

updater = chainer.training.StandardUpdater(train_iter, optimizer)

5. Trainerの準備

Trainerは全体の訓練ループを管理し、進捗状況表示やログ出力などを行う。

trainer = chainer.training.Trainer(updater, (3000, 'epoch'), out="test_result")

trainer.extend(extensions.Evaluator(test_iter, model))
trainer.extend(extensions.LogReport())
trainer.extend(extensions.PrintReport(['epoch', 'main/loss', 'main/accuracy']))
trainer.extend(extensions.ProgressBar())

まずTrainerを作成する。(3000, 'epoch')は終了条件で、3000エポックで終了するという意味。
その後.extendメソッドにchainer.training.extensions以下にあるようなExtensionを渡して実行させることができ、ここでは

  • Evaluatorで現在のモデルの評価
  • LogReportでログファイルへの書き出し
  • PrintReportで指定したデータの画面表示
    • 'epoch'がエポック, 'main/loss'はloss, 'main/accuracy'はaccuracyに対応する。
      • main/はおまじないみたいな感じで、Optimizerが訓練対象とするモデルを表すらしい。
  • ProgressBarで進捗状況を示すプログレスバーの表示

などを設定している。他にも、訓練中のモデルやOptimizerの状態を定期的にファイルに保存することもでき、時間のかかる訓練では、途中でやめても保存した状態から訓練を再開できるようにしておくと良さそう。

6. 訓練ループの実行

Trainerを動かして訓練を行う。

trainer.run()

lossが下がっていくのを眺めながら、訓練が終わるまで待つ。
初期値によっては局所解に囚われてしまってlossが減少しなくなったりするので、その場合は何回かやり直してみるといいかもしれない。

訓練のソース全体

import chainer
import chainer.links as L
import chainer.functions as F
import numpy as np
from chainer.training import extensions

# データセットの準備
indata = np.array([[0,0],[0,1],[1,0],[1,1]], dtype=np.float32)
labels = np.array([ [0],  [1],  [1],  [0] ], dtype=np.float32)

dataset = chainer.datasets.TupleDataset(indata, labels)
train_iter = chainer.iterators.SerialIterator(dataset, 4) # 訓練用 batch size = 4
test_iter = chainer.iterators.SerialIterator(dataset, 4, repeat=False, shuffle=False)
                                                          # 評価用 batch size = 4

#L.Linear(in_size, out_size, wscale=1, bias=0, nobias=False,
#         initialW=None, initial_bias=None)
class Xor(chainer.Chain):  #chainer.Chainを継承したクラスで全体の関数を表現
    def __init__(self): # constructor
        super(Xor, self).__init__( #chainer.Chainでlinkをまとめる
            l1 = L.Linear(2, 10),
            l2 = L.Linear(10, 1)
        )

    def __call__(self, x): #forward propagation definition
        h1 = F.relu(self.l1(x))
        return self.l2(h1)


my_xor = Xor() # 訓練したい関数を示すモデル
accfun = lambda x, t: F.sum(1 - abs(x-t))/x.size #ないと動かないのでとりあえず用意
model = L.Classifier(my_xor, lossfun=F.mean_squared_error, accfun=accfun) 
                                     # datasetのデータを入力するとlossを返すモデル

optimizer = chainer.optimizers.Adam()
optimizer.setup(model) #lossを返すモデルをセット
#optimizer.add_hook(chainer.optimizer.WeightDecay(0.0005))

# 訓練の準備
updater = chainer.training.StandardUpdater(train_iter, optimizer)
trainer = chainer.training.Trainer(updater, (3000, 'epoch'), out="test_result")
trainer.extend(extensions.Evaluator(test_iter, model))
trainer.extend(extensions.PrintReport(['epoch', 'main/loss', 'main/accuracy']))
trainer.extend(extensions.LogReport())
trainer.extend(extensions.ProgressBar())
#print('start')
trainer.run()
#print('done')

訓練したモデルを使う

my_xorが最終的に得たい関数であるXORを表すものになっているはずである。

x = np.array([[1,1],[1,0],[0,1],[0,0]], dtype=np.float32)
result = my_xor(x)
print(result.data)
#出力例:
#[[  4.03262675e-07]
# [  9.99999642e-01]
# [  9.99999642e-01]
# [  3.80910933e-07]]

訓練する度に値は変わるが、うまいこといっていれば出力に0か1に近い値がでてくる。

また、パラメータは次のようにして見ることができる。

print("l1.W:")
print(my_xor.l1.W.data)
print("l1.b:")
print(my_xor.l1.b.data)
print("l2.W:")
print(my_xor.l2.W.data)
print("l2.b:")
print(my_xor.l2.b.data)
#出力例:
#l1.W:
#[[-0.7456618   1.03302336]
# [ 1.10809541 -0.91810882]
# [ 0.47287568  0.47287571]
# [-0.73910308 -0.74221992]
# [ 0.88031632 -0.73166984]
# [-0.73261809 -1.85475707]
# [ 0.70534587  0.70534694]
# [-0.15760551 -0.42765161]
# [ 0.07897222  0.93817586]
# [ 1.17522132  0.32095835]]
#l1.b:
#[ -1.48954052e-06  -2.97201019e-09  -4.72875684e-01   0.00000000e+00
#  -5.84823523e-09   0.00000000e+00  -7.05345809e-01   0.00000000e+00
#   9.12837006e-09   6.72058091e-02]
#l2.W:
#[[ 0.49185807  0.19722475 -0.84104687  0.05196132  0.75117517 -0.09614659
#  -0.7157371   0.72252452  0.50084352  0.06860919]]
#l2.b:
#[-0.00461056]

訓練したモデルのパラメータを保存する

chainer.serializers以下に、オブジェクトをファイルに保存・読み込みする関数が用意されている。保存形式にはNPZ (numpyの形式)とHDF5の2種類があり、それぞれ別々の関数によって読み書きを行う。

#NPZ用
chainer.serializers.save_npz(filename, obj, compression=True)
chainer.serializers.load_npz(filename, obj)

#HDF5用
chainer.serializers.save_hdf5(filename, obj, compression=4)
chainer.serializers.load_hdf5(filename, obj)

訓練の途中経過のスナップショットをとってそこから再開するといった場合にはtrainerを保存・読み込みすると良いようだ。
また、さらなる訓練が要らなければ、今回はmy_xorを保存しておけばよさそうである。
つまり、保存は

chainer.serializers.save_npz('my_xor.npz', my_xor)

で、これを別のファイルで読み出すのであれば、

import chainer
import chainer.links as L
import chainer.functions as F
import numpy as np

class Xor(chainer.Chain):  #chainer.Chainを継承したクラスで全体の関数を表現
    def __init__(self): # constructor
        super(Xor, self).__init__( #chainer.Chainでlinkをまとめる
            l1 = L.Linear(2, 10),
            l2 = L.Linear(10, 1)
        )

    def __call__(self, x): #forward propagation definition
        h1 = F.relu(self.l1(x))
        return self.l2(h1)

my_xor = Xor()

chainer.serializers.load_npz('my_xor.npz', my_xor)

などとすればよい。

参考サイト

Chainerのチュートリアルやリファレンスに加え、以下の記事がChainerの理解に大いに役に立った。内容的にも大いに参考にした。

*1:outputは第3層の出力と等しい。色々図示されたものを見るに、最後のoutputは第3層と一つにまとめて書かれるのが普通な感じがある。

*2:必ずしもそうでなくても問題ないのだが、Trainerを使った訓練で扱うには色々面倒っぽい

*3:これは訓練したら極小解に陥ってしまって、何を入力しても0.5を返すようになってしまうことが多かったため。

*4:Optimizerに渡すことはできるが、独自のUpdaterを用意してlossを計算するなどしないといけない。

*5:loss関数と入力を与えて呼び出すと、内部でlossを計算し、パラメータを更新する