Theanoを使ってdeep learningしてみた

はじめに

最近chainerとcaffeに続いて第３のライブラリとしてTheanoを使い始めたので、Theanoを使った畳み込みニューラルネットワーク(CNN)の実装の解説をします。Theanoの基本的な使い方は他の記事や公式tutorialがわかりやすいのでそちらを参考にしてください。
本記事はdeep learning tutorialのDeep Convolutional Networkに沿って解説していきます。コードは解説用に変更してありますが大まかな流れは同じです。
あくまでTheanoを使ったCNNの実装の解説でDeep Learning自体の解説やtheanoの基本的な使い方の解説はないのでその辺りは他の記事を参照してください。また実装の解説と言ってもTheanoを使い始めて数日、本人の実装能力はそんなに高くない上英語も得意ではないので(公式チュートリアルは英語)至らない点も多々あります。(本人の備忘録感覚のためそこまで期待しないでください笑)

LeNet

deep learning tutorialに沿ってLeNetをベースにminstの手書き文字認識を行っていきます。LeNetは畳み込み層×2とプーリング層×2、全結合層からなる基本的なCNNです。本記事では活性化関数など細かいところは本来のものとは変えてますが基本的な構造は同じです。(詳しいLeNetの解説は元論文を読んでください)

実装

上記のLeNetに基づいたCNNを実装していきます。前提として以下をimportしているものとします。

import theano
import theano.tensor as T
import numpy as np

畳み込み層

まずはじめに畳み込み層について実装します。Theanoでは畳み込みを行うシンボルとして、T.nnet.conv.conv2dが用意されています。Theanoは重みやバイアスを自分で記述しなければならないためnumpyとtheano.sharedを用いて記述します。重みとバイアスは学習によって更新されていく値であるため、theano.sharedを用いて定義します。また、畳み込み層を追加するたび層を記述するのは面倒なのでclassとして定義します。
以下が畳み込み層classの実装例となります。

class Conv2d(object):
    def __init__(self, input, out_c, in_c, k_size)
        self._input = input #入力されるシンボル
        self._out_c = out_c #出力チャネル数
        self._in_c = in_c #入力チャネル数
        w_shp = (out_c, in_c, k_size, k_size) #重みのshape
        w_bound = np.sqrt(6. / (in_c * k_size * k_size + \
                        out_c * k_size * k_size)) #重みの制約
        #重みの定義
        self.W = theano.shared( np.asarray(
                        np.random.uniform( #乱数で初期化
                            low=-w_bound,
                            high=w_bound,
                            size=w_shp),
                        dtype=self._intype.dtype), name ='W', borrow=True)
        b_shp = out_c, #バイアスのshape
        #バイアスの定義(ゼロで初期化)
        self.b = theano.shared(np.zeros(b_shp,
                        dtype=self._input.dtype), name ='b', borrow=True)
        #畳み込みのシンボルの定義
        self.output = T.nnet.conv.conv2d(self._input, self.W) \
                        + self.b.dimshuffle('x', 0, 'x', 'x')
        #更新されるパラメータを保存
        self.params = [self.W, self.b]

dimshuffleはバイアス項の次元をvectorからT.nnet.conv.conv2dの出力であるtensor4に合わせる操作を行っています。感覚的にはreshapeとnp.transposeを組み合わせた感じです。('x', 0, 'x', 'x')の場合self.bのshapeは（1, self.b.shape[0], 1, 1）のようになります。

活性化関数

本来LeNetの活性化関数はtanhですが、今回はreluを用います。reluはmax(0, x)と言う簡単な数式で表される活性化関数です。Theanoにはreluのシンボルはないため自分で定義する必要があります。TheanoのT.max()シンボルは実数値を中に入れられない(やり方があるかもしれませんが)のとシンボルに対しif文を使えないため少し特殊な書き方をします。以下がreluの実装例です。

class relu(object):
    def __init__(self, input):
        self._input = input
        self.output  = T.switch(self._input < 0, 0, self._input)

プーリング層

プーリング層はTheanoに置いてtheano.tensor.signal.pool.pool_2dにシンボルが定義されています。プーリング層は畳み込みそうと違い重みやバイアスといった更新を行うシンボルを用意する必要がないため簡単に書けます。以下プーリング層の実装例です。

from theano.tensor.signal import pool

class Pool2d(object):
    def __init__(self, input, k_size, st, pad=0, mode='max'):
        self._input = input
        #プーリング層のシンボルの定義
        self.output = pool.pool_2d(self._input, 
                            (k_size, k_size), #カーネルサイズ
                            ignore_border=True, #端の処理(基本的にTrueでok,詳しくは公式Documentへ)
                            st=(st, st), #ストライド
                            padding=(pad, pad), #パディング
                            mode=mode) #プーリングの種類('max', 'sum', 'average_inc_pad', 'average_exc_pad')

全結合層

全結合層はtheanoでシンボルが用意されていないので自分で記述しまが、行列の内積計算で表現でき、内積のシンボルがT.dot()で与えられているため特に難しいことはありません。畳み込み層同様重みとバイアスがあるためそれぞれ定義を行います。以下全結合層の実装例です。

class FullyConnect(object):
    def __init__(self, input, inunit, outunit):
        self._input = input
        #重みの定義
        W = np.asarray(
            np.random.uniform(
            low=-np.sqrt(6. / (inunit + outunit)),
            high=np.sqrt(6. / (inunit + outunit)),
            size=(inunit, outunit)
            ),
            dtype=theano.config.floatX)
        self.W = theano.shared(value=W, name='W', borrow=True)
        #バイアスの定義
        b = np.zeros((outunit,), dtype=theano.config.floatX) #ゼロで初期化
        self.b = theano.shared(value=b, name='b', borrow=True)
        #全結合層のシンボルの定義
        self.output = T.dot(self._input, self.W) + self.b
        #更新されるパラメータを保存
        self.params = [self.W, self.b]

ロス関数

ロス関数はmnistの10クラス分類を解くためsoftmax cross entropyを用います。softmaxのシンボルはTheanoではT.nnet.softmax()に用意されているのでこれを使います。
以下実装例です。

class softmax(object):
    def __init__(self, input, y):
        self._input = input
        #softmaxのシンボル定義
        self.output = nnet.softmax(self._input)
        #cross entropyのシンボル定義(数式的にはsumですがここではmeanを用います)
        self.cost = -T.mean(T.log(self.output)[T.arange(y.shape[0]), y])

yは教師ラベルのシンボルを表します。[T.arange(y.shape[0]),y]はT.meanを行う際にy[0]からy[y.shape[0]-1]まで加算することを意味します。

LeNet

データセット

各層の定義が終わったのでここからLeNetの実装に移ります。まずはmnistのデータを用意します。pklデータがhttp://www.iro.umontreal.ca/~lisa/deep/data/mnist/mnist.pkl.gz に用意されているのでこれを適当なフォルダにダウンロードします。ここからtrainとvalidationとtestデータを取り出します。以下データのロード例。

import gzip
import cPickle

def shared_dataset(data_xy):
    data_x, data_y = data_xy
    set_x = theano.shared(np.asarray(data_x,
                  dtype=theano.config.floatX).reshape(-1,1,28,28),
                  borrow=True)
    set_y = T.cast(theano.shared(np.asarray(data_y,
                  dtype=theano.config.floatX), borrow=True), 'int32')
    return set_x, set_y

with open('/path/to/mnist.pkl.gz', 'rb') as f:
    train_set, valid_set, test_set = cPickle.load(f)

train_set_x, train_set_y = shared_dataset(train_set)
valid_set_x, valid_set_y = shared_dataset(valid_set)
test_set_x, test_set_y = shared_dataset(test_set)

ここでは実装の都合上データをtheano.sharedのシンボルとして定義していますが、入力はnumpyの配列でも問題ありません。ただ、theano.sharedで定義すると実装が幾分かすっきりします。
次に入力データと教師ラベルのシンボルを定義します。入力データは(バッチ数, チャネル数, 縦, 横)の4次元なのでT.tensor4()となり教師ラベルは1次元の整数値のベクトルなのでT.ivector()になります。

x = T.tensor4() #入力データのシンボル
y = T.ivector() #出力データのシンボル

層の定義

ここから各層の定義になります。上でそれぞれのclassを作成してあるのでいくらか楽に書けます。

conv1 = Conv2d(x, 20, 1, 5) #xを入力とし、出力が20チャネル、入力が1チャネル、カーネルサイズ5
relu1 = relu(conv1.output) #conv1の出力を入力とする
pool1 = Pool2d(relu1.output, 2, 2) #relu1の出力を入力とし, カーネルサイズ2、ストライド2

conv2 = Conv2d(pool1.output, 50, 20, 5) #poo1の出力を入力とし、出力が50チャネル、入力が20チャネル、カーネルサイズ5
relu2 = relu(conv2.output) #conv2の出力を入力とする
pool2 = Pool2d(relu2.output, 2, 2) #relu2の出力を入力とし, カーネルサイズ2、ストライド2

fc1_input = pool2.output.flatten(2) #pool2の出力のシンボルはT.tensor4のため、flatten()を使って全結合層の入力のシンボルに合わせる
fc1 = FullyConnect(fc1_input, 50*4*4, 500) #入力のユニット数が50*4*4(チャネル数*縦*横)、出力のユニット数が500
relu3 = relu(fc1.output)
fc2 = FullyConnect(relu3.output, 500, 10) #入力のユニット数が500、出力のユニット数が10(10クラス分類のため)
loss = softmax(fc2.output, y)

これでLeNetの定義が書けました。各層の出力のシンボルを次の層の入力のシンボルにすることにより全てのシンボルが繋がり、勾配計算を一度に行うことができます。すなわち、
・・・T.nnet.conv.conv2d(pool.pool2d(T.nnet.conv.conv2d()))・・・
のような長いシンボルが定義されたことになります。そのためT.grad()に最終的なシンボル(今回はloss.cost)を与えることで簡単に全ての層の勾配を計算することが可能になります。

学習

最後に学習とvalidation data、test dataによる評価のfunctionを定義します。ここまではシンボルを定義しただけなので実際の値を入力して学習などを行うことができません。そのためシンボルをtheano.functionとして定義します。その際にパラメータの更新を行うように定義することで学習を行うことができます。今回はSGDを用いてパラメータの更新を行います。以下学習のtheano.functionの実装例になります。

#学習される全パラメータをリストにまとめる
params = conv1.params + conv2.params + fc1.params + fc2.params 
#各パラメータに対する微分を計算
grads = T.grad(loss.cost, params)
#学習率の定義
learning_rate = 0.001
#更新式を定義
updates = [(param_i, param_i - learning_rate * grad_i) for param_i, grad_i in zip(params, grads)]
#学習のtheano.functionを定義
index = T.lscalar()
batch_size = 128
train_model = theano.function(inputs=[index], #入力は学習データのindex
                       outputs=loss.cost, #出力はloss.cost
                       updates=updates, #更新式
                       givens={
                            x: train_set_x[index: index + batch_size], #入力のxにtrain_set_xを与える
                            y: train_set_y[index: index + batch_size] #入力のyにtrain_set_yを与える
                       })

まず、paramsは各層の重みとバイアスのシンボルからなるリストになります(リスト同士の足し算であるため)。T.grads()は変数をリストで与えるとそれぞれの変数で微分したシンボルをリストで返すため、gradsはloss.costを各パラメータで微分したシンボルを持つリストになってます。updatesも各パラメータの更新式のリストとなります。
次にtrain_modelの定義ですが、前述したようにconv1からloss.costまでが最初に定義したxとyを入力とする一つのシンボルとなっています。train_model内でxはtrain_set_xをyがtrain_set_yをの値を受け取るようになっています。train_set_xとtrain_set_yはindexを受け取って受け取ったindexからbatch_size分のデータを参照しています。その為train_modelには引数としてindexのみを与えることで、indexからindex + batch_sizeまでのtrain_set_xとtrain_set_yの値をx, yに与えています。
あとはこのtrain_modelをfor文などで繰り返し呼ぶことで学習ができます。

for i in range(0, train_set_y.get_value().shape[0], batch_size):
    train_model(i)

評価

最後に学習モデルの精度評価を行うtheano.functionを定義します。評価ではパラメータ更新を行わないため出力はerror rateにします。loss.outputにsoftmaxのシンボルがあるためこれを使ってerror rateを算出するシンボルを定義して、error rateのシンボルを使って評価を行うtheano.functionを定義します。

pred = T.argmax(loss.output, axis=1) #予測された確率が最も高いクラスを返す
error = T.mean(T.neq(pred,y)) #予測されたクラスを正解ラベルと比較
test_model = theano.function(inputs=[index],
                             outputs=error,
                             givens={
                             x: test_set_x[index: index + batch_size],
                             y: test_set_y[index: index + batch_size]
                             })

val_model = theano.function(inputs=[index],
                             outputs=error,
                             givens={
                             x: test_set_x[index: index + batch_size],
                             y: test_set_y[index: index + batch_size]
                             })

これでvalidationとtest data用の評価functionが定義できたのでtrain_modelと同様にfor文を使うことで評価ができます。

test_losses = [test_model(i)
               for i in range(0, test_set_y.get_value().shape[0], batch_size] #バッチごとの平均のlossをリストに保存
mean_test_loss = np.mean(test_losses) #全体の平均を算出

以上でtheanoを用いたCNNの学習、評価コードが書けました。実際の学習結果は上記のコードを全てつなげて確認してみてください(自分で書いたコードをQuiita用に所々変えてデバッグしないまま書いたので動かないことがあるかも笑)。不明な点やコード、解説に誤りがある場合はコメントで知らせてください。

まとめ

Theanoを使った畳み込みニューラルネットワークの実装の解説をしました。一見コードが長く面倒に思えるかもしれませんが、自分の扱いやすいように層のクラスを一度定義してしまえばあとは楽に書けます。自分で色々と定義できるのがTheanoの利点ですね(その分面倒ですが)。LeNet以外の実装は層の定義の部分を変更すれば様々なモデルを作ることができます。またロス関数やパラメータ更新手法もシンボルを定義すれば自由に記述することが可能です。これからTheanoを使おうとしている人の役に立てば幸いです。