今さら TensorFlow のディープラーニングに入門してみた～MNISTのサンプルプログラムを自分用に書き直す編～

ディープラーニングは Chainer から入ったのですが、わけあって TensorFlow を使いたくなったので、勉強することにしました。www.tensorflow.org

動機

動機というのはICML2017のベストペーパーである以下ですが、

[1703.04730] Understanding Black-box Predictions via Influence Functions
qiita.com

影響関数を使って、評価データの分類に対する helpful or harmful 教師データを抽出したということ。この理論がそれらしい結果まで出してるところに衝撃を受けてる。

この中で “Efficiently Calculating Influence” における計算の肝は、評価データに対する $\tilde{H}^{-1}_{\hat{\theta}} \nabla_{\theta} L (z_{test}, \hat{\theta})$ を、

$\tilde{H}^{-1}_j v = v + [{I - \nabla^{2}_{\theta} L (z_i, \hat{\theta})}] \tilde{H}^{-1}_{j-1} v$

という漸化式で近似することにある(多分)。もちろん一番大変なのが、損失関数の $\theta$ 方向への二階微分であり、つまりはニューラルネットの係数を適宜参照しながら二階微分できるように仕込んでおく必要があるということになる。
著者による実装もあるが、そうするとやはり TensorFlow を使うしかないように思う。

実装

まず手始めに MNIST を動かしてみればいいだろうと考えてサンプルプログラムを見ても、このとっかかりにくさ… 初学者には Chainer の方がすっきりしてて見やすいですね。

TensorFlow でディープラーニングをする場合、おおまかな流れとして、まず変数の箱を用意して、その変数がニューラルネットの中をどう流れるか定義する、そして学習時は変数の箱にバッチをフィードするということをしてやる必要があるらしい。

まずネットワークはなんかいちいち書くのがめんどくさくなってきたのと、Chainer のコードと一緒に管理しておいた方がいいと思うので、外部から記述出来るようにするのが今回の設計思想に。

# ニューラルネットを定義
Network_layers = [{'layer':'CNN', 'ch':8, 'ksize':(3,3), 'stride':(2,2), 'activation':'LRelu'},
                   {'layer':'CNN', 'ch':16, 'ksize':(3,3), 'stride':(2,2), 'activation':'LRelu'},
                   {'layer':'CNN', 'ch':32, 'ksize':(3,3), 'stride':(2,2), 'activation':'LRelu'},
                   {'layer':'Linear', 'ch':32, 'dropout':True, 'activation':'LRelu'},
                   {'layer':'Linear', 'ch':10, 'dropout':True, 'activation':None},
                   ]

とりあえず、CNN(+ksize,stride,pad) か Linear を指定して、活性化関数(Relu or LeakyRelu)とドロップアウトをするかを定義してやる。BatchNormalization ぐらい用意しとけって感じだけど、めんどくさそうだったのでもう少し理解を深めてからにする。

ネットワーク定義を見て変数の流れと損失関数を定義する tfnet.MTTFNet という class を用意して、main では以下のように記述するようにする。

import tensorflow as tf
import tfnet

learning_rate = 1e-3
x = tf.placeholder(tf.float32, shape=[None, 784])
t = tf.placeholder(tf.float32, shape=[None, 10])
keep_prob = tf.placeholder(tf.float32)
        
x_image = tf.reshape(x, [-1,28,28,1])
        
# ニューラルネットを構築
net = tfnet.MTTFNet(Network_layers)
net.set_keep_prob(self.keep_prob)
y = net.make(x_image)
loss = net.loss_func(y,t)
    
# Optimizer(=Adam)
train_step = tf.train.AdamOptimizer(learning_rate).minimize(loss)

tf.placeholder が上述の変数の入れ物で、keep_prob をTensorFlowの定義に従ってドロップアウトしない率とする。
このへんは Chainer ライクにしておくと分かり良いような気がしたので。

tfnet.MTTFNet の中では、loss_func でソフトマックス交差エントロピーの流れを以下のように定義し、

def loss_func(self, y, t):
    loss = tf.nn.softmax_cross_entropy_with_logits(logits=y, labels=t)
    return tf.reduce_mean(loss)

ネットワークの変数の流れは簡潔に以下。

def make(self, x):
    if len(x.shape) == 4:
        img_x = int(x.shape[1])
        img_y = int(x.shape[2])
        img_ch = int(x.shape[3])
    else:
        img_x = 1
        img_y = 1
        img_ch = int(x.shape[1])
    
    h = x
    n_layer = 0
    ## ニューラルネット構成
    for hierarchy in self.Network_layers:
        layer_name = 'layer' + str(n_layer)
        n_layer += 1
        with tf.variable_scope(layer_name):
            layer = hierarchy['layer']
            ch = hierarchy['ch']
            activation = hierarchy['activation']
            total_binding_ch = img_x*img_y*img_ch
            
            b = tf.Variable(tf.constant(self.initial_W, shape=[ch]), name='b')
            if layer == 'CNN':
                ksize = hierarchy['ksize']
                stride = hierarchy['stride']
                
                #### 重みの初期化
                kernel_ch = ksize[0]*ksize[1]*img_ch
                W = tf.Variable(tf.truncated_normal([kernel_ch*ch], stddev=self.initial_W), name='W')
                W_conv = tf.reshape(W, [ksize[0],ksize[1],img_ch,ch])
                
                #### 畳み込み層
                h = tf.nn.conv2d(h, W_conv, strides=[1, stride[0], stride[1], 1], padding='SAME') + b
                img_x = ceil(float(img_x) / float(stride[0]))
                img_y = ceil(float(img_y) / float(stride[1]))
                img_ch = ch
                
            elif layer == 'Linear':
                use_dropout = hierarchy['dropout']
                
                #### 重みの初期化
                W = tf.Variable(tf.truncated_normal([total_binding_ch*ch], stddev=self.initial_W), name='W')
                W_lin = tf.reshape(W, [total_binding_ch,ch])
                
                ### 全結合層
                h = tf.reshape(h, [-1, total_binding_ch])
                h = tf.matmul(h, W_lin) + b
                img_x = 1
                img_y = 1
                img_ch = ch
                
                ### ドロップアウト
                if use_dropout:
                    h = tf.nn.dropout(h, self.keep_prob)
                
            ### 活性化関数
            if activation == 'Relu':
                h = tf.nn.relu(h)
            elif activation == 'LRelu':
                # leaky_relu
                h = tf.maximum(h, self.LRelu_alpha*h)
            
    print('MTINFO: successfully constructed the neural network')
    return h

x が入ってきて、係数との掛け算をして活性化関数を通って h で出るという道を定義している。

ここで、自動生成したネットワークに対して layer**/W、layer**/b という名前空間から $\theta$ にアクセスできる状態にしてみた。これでうまくいかなかったら、うまくいかなかった時に考えよう(笑)

実際にMNISTを呼んできてバッチ処理を行う関数が以下。

def run(self):
    # MNIST を取得
    from tensorflow.examples.tutorials.mnist import input_data
    data = input_data.read_data_sets('MNIST_data/', one_hot=True)
    
    saver = tf.train.Saver()
    with tf.Session() as sess:
        init = tf.global_variables_initializer()
        sess.run(init)
        
        # 保存モデルの読み込み
        ckpt = tf.train.get_checkpoint_state('model/')
        if ckpt and self.flag_resum:
            try:
                last_model = ckpt.model_checkpoint_path
                saver.restore(sess, last_model)
            except:
                print('MTINFO: The restored model is invalid')
        else:
            print('MTINFO: All paramteres are initialized')
        
        # 学習フェイズ
        if self.flag_train:
            epoch = 1
            epochs_completed = 0
            batch_count = 0
            tot_acc = 0
            
            while(epochs_completed < self.n_epoch):
                batch = data.train.next_batch(self.batchsize, shuffle=True)
                if epoch == epochs_completed:
                    print('MTINFO: train epoch %d > accuracy %g'%(epoch, tot_acc/batch_count))
                    epoch += 1
                    batch_count = 0
                    tot_acc = 0

                self.train_step.run(feed_dict={self.x: batch[0],self.t: batch[1],self.keep_prob: 0.5})
                epochs_completed = data.train._epochs_completed
                tot_acc += self.get_acc(batch[0],batch[1])
                batch_count += 1
            else:
                print('MTINFO: train epoch %d > accuracy %g'%(epoch, tot_acc/batch_count))
        
        # テストフェイズ
        tot_acc = self.get_acc(data.test.images, data.test.labels)
        print('MTINFO: test accuracy %g'%(tot_acc))
            
        saver.save(sess, 'model/model.ckpt')

tf.global_variables_initializer で変数を初期化して、tf.train.Saver()くんにモデルを管理してもらう。

デフォルトで epoch で回さないのはよく分からないけど、train._epochs_completed という変数を参照するとちゃんと epoch 数を教えてくれることが分かった。

data.train.next_batch(self.batchsize) でバッチを取得して、 train_step.run(feed_dict=~) という関数に通すと、先程の変数の入れ物にバッチを突っ込んで定義したニューラルネットを通って学習が進むらしい。

# Minibatch-size: 256
# epoch: 10
MTINFO: successfully constructed the neural network
Extracting MNIST_data/train-images-idx3-ubyte.gz
Extracting MNIST_data/train-labels-idx1-ubyte.gz
Extracting MNIST_data/t10k-images-idx3-ubyte.gz
Extracting MNIST_data/t10k-labels-idx1-ubyte.gz
MTINFO: All paramteres are initialized
MTINFO: train epoch 1 > accuracy 0.691297
MTINFO: train epoch 2 > accuracy 0.913899
MTINFO: train epoch 3 > accuracy 0.9371
MTINFO: train epoch 4 > accuracy 0.949564
MTINFO: train epoch 5 > accuracy 0.957031
MTINFO: train epoch 6 > accuracy 0.961791
MTINFO: train epoch 7 > accuracy 0.964771
MTINFO: train epoch 8 > accuracy 0.967624
MTINFO: train epoch 9 > accuracy 0.970749
MTINFO: train epoch 10 > accuracy 0.972129
MTINFO: test accuracy 0.9721

一応学習は進んでいるみたいなので、partially success.