※この記事ではPython 3.5.3とChainer v1.20.0.1を使っています。
調べながら書いています。間違い等あればご指摘願います。
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)
- ミニバッチを入力とするので入力の次元に注意。
いくつかの入力を重みづけして足し合わせ、さらに定数(バイアス)を足し、その結果に適当な非線形関数(activation関数という)を適用したものを(単純)パーセプトロンという。
AND関数を作ってみる
ANDは単純パーセプトロンで実装できる。
- 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
W = np.array([[1,1]])
b = -1.0
a = L.Linear(2, 1, initialW=W, initial_bias=b)
f = F.relu
def AND(x):
return f(a(x))
x = np.array([[1,1],[0,1],[1,0],[0,0]], dtype=np.float32)
print(AND(x).data)
実行すると、[1,1]の入力に対しては[1]が、それ以外には[0]が帰ってきていることがわかる。これはまさにANDの挙動である。
単純パーセプトロンを複数、層状に組み合わせたものを多層パーセプトロンという。各層はベクトルを入力としベクトルを出力とする関数になっていて、全体で複雑な非線形関数を実現できる。
次は3層の例で、バイアスやactivation関数の記載は省略した。*1
XOR関数を作ってみる
同様にXOR関数を作ってみる。
XORは線形分離不可能であるので単純パーセプトロンでは再現できず、全結合層を2つ用意する必要がある。中間層(隠れ層)のノード数は2にした。
- 全結合層(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
W1 = np.array([[1,-1],[-1,1]])
b1 = 0.0
a1 = L.Linear(2, 2, initialW=W1, initial_bias=b1)
f1 = F.relu
W2 = np.array([[1,1]])
b2 = 0.0
a2 = L.Linear(2, 1, initialW=W2, initial_bias=b2)
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)
これも実際に動かしてみると、入力の[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
W1 = np.array([[1,-1],[-1,1]])
b1 = 0.0
W2 = np.array([[1,1]])
b2 = 0.0
class Xor(chainer.Chain):
def __init__(self):
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)
chainer.Chainを継承したXorクラスのインスタンスに入力を与えて呼び出すと計算結果が出力される。
さて、ここではXORネットワークのパラメータを手で予め求めた値に設定したが、これをデータから学習させてみたい。
Chainerにおける訓練の流れ
多分こんな感じ。
- 学習用データセットの用意
- lossを計算するネットワークモデルを用意(これを訓練する)
- Optimizerを用意してモデルをセット
- Updaterを用意
- Trainerを準備
- 訓練ループの実行
データセットはモデルを定義するのと同時に考えないといけないと思うので最初に入れているが、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)
test_iter = chainer.iterators.SerialIterator(dataset, 4, repeat=False, shuffle=False)
- TupleDatasetは内部で単に2つの入力をタプルにするくらいのことしかしてなくて、あまり本質ではないっぽい。
- 重要なのはその後のSerialIteratorなどのイテレータで、これがあるとミニバッチ単位でデータを取り出すのが簡単になる。
- 訓練用と評価用のイテレータを用意する。
- 今回は訓練セットと評価セットを全く同じにしているが、普通は適当に分ける。
- どちらもミニバッチサイズを4とした。
- 訓練用のイテレータは後のUpdaterに渡される。
- 評価用のイテレータは特になくても訓練はできるはずだけれど、訓練中のモデルの途中経過のaccuracyとかを見るのに使われる。
- repeat=False, shuffle=False を設定しておく(デフォルトではTrue)。特に repeat=False としないと評価が終わらなくなる。
2. lossを返すモデルの用意
訓練するモデルは、呼び出すと、つまり、 .__call__ メソッドが:
- 用意したデータセット(のミニバッチ)を入力として受け付ける
- lossを返す
ようなネットワークである必要がある*2。
class Xor(chainer.Chain):
def __init__(self):
super(Xor, self).__init__(
l1 = L.Linear(2, 10),
l2 = L.Linear(10, 1)
)
def __call__(self, x):
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)
- ここで、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)
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)
test_iter = chainer.iterators.SerialIterator(dataset, 4, repeat=False, shuffle=False)
class Xor(chainer.Chain):
def __init__(self):
super(Xor, self).__init__(
l1 = L.Linear(2, 10),
l2 = L.Linear(10, 1)
)
def __call__(self, x):
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)
optimizer = chainer.optimizers.Adam()
optimizer.setup(model)
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())
trainer.run()
訓練したモデルを使う
my_xorが最終的に得たい関数であるXORを表すものになっているはずである。
x = np.array([[1,1],[1,0],[0,1],[0,0]], dtype=np.float32)
result = my_xor(x)
print(result.data)
訓練する度に値は変わるが、うまいこといっていれば出力に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)
訓練したモデルのパラメータを保存する
chainer.serializers以下に、オブジェクトをファイルに保存・読み込みする関数が用意されている。保存形式にはNPZ (numpyの形式)とHDF5の2種類があり、それぞれ別々の関数によって読み書きを行う。
chainer.serializers.save_npz(filename, obj, compression=True)
chainer.serializers.load_npz(filename, obj)
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):
def __init__(self):
super(Xor, self).__init__(
l1 = L.Linear(2, 10),
l2 = L.Linear(10, 1)
)
def __call__(self, x):
h1 = F.relu(self.l1(x))
return self.l2(h1)
my_xor = Xor()
chainer.serializers.load_npz('my_xor.npz', my_xor)
などとすればよい。
