めもめも

このブログに記載の内容は個人の見解であり、必ずしも所属組織の立場、戦略、意見を代表するものではありません。

TensorFlow Tutorialの数学的背景 − Deep MNIST for Experts(その3)

何の話かというと

enakai00.hatenablog.com

上記の記事では、「−」「|」「+」という記号をCNN(畳み込みニューラルネットワーク)で分類することに成功しました。これを、MNISTの手書き数字画像に適用するのが今回の記事です。

使用するネットワーク

ここでは、2種類のネットワークを解説します。1つ目は、前回のネットワークを素直に拡張したものです。

前回の例では、2枚のフィルターと2個の特徴変数 (z_0,z_1) で十分でしたが、ここではより複雑な図形を扱うためにフィルターを16枚に増やして、特徴変数も16個にしています。これで、約97%の正答率が達成できます。(トレーニングセットと分離した、テストセットに対する正答率です。)

そして、Deep MNIST for Expertsの中では、さらに識別率を上げるために、次のようなネットワークを組んでいます。

ここまでくると、まさに「Deep Learning」という感じですが、各レイヤーの処理はこれまでと大きな違いはありません。まず、畳み込み層を2段に増やしています。1枚の24×24ピクセルの画像に対して、1段めの「畳み込み+プーリング」によって、32枚の14×14ピクセルの画像が得られます。これらの画像それぞれをさらに2段目の「畳み込み+プーリング」にかけます。(後で示すように、畳込みフィルターの関数は、最初のものをより一般化した形にしています。)

この時、2段目のフィルターでは、64枚ある「1つのフィルター」は内部的に32個のサブフィルターを持っており、「1段目から出てきた32枚のそれぞれを対応するサブフィルターにかけた結果を重ねあわせる」という処理をします。(つまり、サブフィルターを含めてカウントすると、全部で64*32=2048枚のフィルターがあります。)これにより、合計64枚の7×7ピクセルの画像が得られます。図にするとこんな感じです。

これにより、最終的に、1枚の画像から、64種類の特徴を取り出した画像が生成されます。これを1024個の特徴変数に変換した後に、Softmax関数で文字の種類の判定を行ないます。

Softmax関数の手前にある「ドロップアウト」は、学習処理の際に機能させる特殊なレイヤーで、1024個の特徴変数からの入力のそれぞれを1/2の確率でカットします(強制的に入力値を0にします)。つまり、学習処理を繰り返す際に、最後の特徴変数からの入力がランダムに欠落するのです。

―― そんな事したら、ちゃんと学習でけへんがな。。。

はい。その通りで、これは、ちゃんと学習させないための処理、正確に言うと、過学習を防ぐための処理になります。ネットワークが複雑になると、トレーニング用のデータ(トレーニングセット)だけが持つ特徴に特化した学習が行われてしまい、「トレーニングセットの正答率は100%なのに、トレーニングに使用しなかったデータの正答率はあまり高くならない」という現象が発生することがあります。そこで、あえて学習処理にノイズを入れて、トレーニングセットだけに特化した学習を防いでいます。

実際に機械学習を行う場合、手元にあるデータのすべてをトレーニングに使用するのではなく、一部のデータをテスト用に取り分けておき、学習結果の精度は、このテストセットで判定します。上記のネットワークを用いると、テストセットに対して、99.2%の正答率が得られます。(学習結果を用いてテストデータの判別を行う時は、ドロップアウト層は停止しておきます。)

実行結果(その1)

まず、畳み込み層が1層の最初の例です。実際のコードは下記になります。

from tensorflow.examples.tutorials.mnist import input_data
import tensorflow as tf
import matplotlib.pyplot as plt
import numpy as np

np.random.seed(1)
tf.set_random_seed(1)

num_filters = 16
num_fc = 16
mnist = input_data.read_data_sets("/tmp/data/", one_hot=True)

def visualize(sess):
  fig = plt.figure()
  C = 12

  batch = mnist.test.next_batch(C)
  res = h_pool1_flat.eval(session=sess, feed_dict={x: batch[0]})
  w = W_conv1.eval()

  for i in range(num_filters):
    subplot = fig.add_subplot(C+1, num_filters+1, i+2)
    subplot.set_xticks([])
    subplot.set_yticks([])
    subplot.imshow(w[:,:,0,i], cmap=plt.cm.gray_r, interpolation='nearest')
  for c in range(C):
    chars = res[c].reshape(14,14,num_filters)
    subplot = fig.add_subplot(C+1, num_filters+1, 1+(c+1)*(num_filters+1))
    subplot.set_xticks([])
    subplot.set_yticks([])
    subplot.imshow(batch[0][c].reshape(28,28), vmax=1, vmin=0,
                   cmap=plt.cm.gray_r, interpolation='nearest')
    for i in range(num_filters):
      subplot = fig.add_subplot(C+1, num_filters+1, (c+1)*(num_filters+1)+(i+2))
      subplot.set_xticks([])
      subplot.set_yticks([])
      subplot.imshow(chars[:,:,i], cmap=plt.cm.gray_r, interpolation='nearest')
  plt.show()

def weight_variable(shape):
  initial = tf.truncated_normal(shape, stddev=0.1, seed=1)
  return tf.Variable(initial)

def bias_variable(shape):
  initial = tf.constant(0,1, shape=shape)
  return tf.Variable(initial)

def conv2d(x, W):
  return tf.nn.conv2d(x, W, strides=[1,1,1,1], padding='SAME')

def max_pool_2x2(x):
  return tf.nn.max_pool(x, ksize=[1,2,2,1],
                        strides=[1,2,2,1], padding='SAME')

with tf.Graph().as_default():
  with tf.name_scope('input'):
    x = tf.placeholder(tf.float32, [None, 784], name='x-input')
    x_image = tf.reshape(x, [-1,28,28,1])
  
  with tf.name_scope('convolution1'):
    W_conv1 = weight_variable([5,5,1,num_filters])
    h_conv1 = tf.nn.relu(tf.abs(conv2d(x_image, W_conv1))-0.2)
    _ = tf.histogram_summary('h_conv1', h_conv1)
  
  with tf.name_scope('pooling1'):
    h_pool1 = max_pool_2x2(h_conv1)
  
  with tf.name_scope('fully-connected'):
    W_fc1 = weight_variable([14 * 14 * num_filters, num_fc])
    b_fc1 = bias_variable([num_fc])
    h_pool1_flat = tf.reshape(h_pool1, [-1, 14*14*num_filters])
    h_fc1 = tf.nn.relu(tf.matmul(h_pool1_flat, W_fc1) + b_fc1)
  
  with tf.name_scope('readout'):
    W_fc2 = weight_variable([num_fc, 10])
    b_fc2 = bias_variable([10])
    y_conv=tf.nn.softmax(tf.matmul(h_fc1, W_fc2) + b_fc2)
  
  with tf.name_scope('optimizer'):
    y_ = tf.placeholder(tf.float32, [None, 10], name='y-input')
    cross_entropy = -tf.reduce_sum(y_*tf.log(y_conv))
    optimizer = tf.train.GradientDescentOptimizer(0.001)
    train_step = optimizer.minimize(cross_entropy)
    correct_prediction = tf.equal(tf.argmax(y_conv,1), tf.argmax(y_,1))
    accuracy = tf.reduce_mean(tf.cast(correct_prediction, tf.float32))
  
  # Logging data for TensorBoard
  _ = tf.scalar_summary('cross entropy', cross_entropy)
  _ = tf.scalar_summary('accuracy', accuracy)
  
  with tf.Session() as sess:
    writer = tf.train.SummaryWriter('/tmp/cnn_single', graph_def=sess.graph_def)
    sess.run(tf.initialize_all_variables())
  
    for i in range(1001):
      batch = mnist.train.next_batch(100)
      train_step.run(feed_dict={x: batch[0], y_: batch[1]})
      if i%10 == 0:
        summary_str, acc = sess.run(
          [tf.merge_all_summaries(), accuracy],
           feed_dict={x: batch[0], y_: batch[1]})
        writer.add_summary(summary_str, i)
        print("step %d, training accuracy %g" % (i, acc))

    print("test accuracy %g" % accuracy.eval(
          feed_dict={x: mnist.test.images, y_: mnist.test.labels}))
    visualize(sess) 

これを実行すると、テストセットに対する正答率は、0.975が得られます。学習曲線は次のようになります。

学習後の32枚のフィルターは次のようになります。縦・横・斜めなどいくつかの方向を切り出すフィルターが得られているようです。


実行結果(その2)

続いて、畳み込み層が2層の場合です。実際のコードは、こちらです。

from __future__ import absolute_import
from __future__ import division
from __future__ import print_function

from tensorflow.examples.tutorials.mnist import input_data
import tensorflow as tf

mnist = input_data.read_data_sets("/tmp/data/", one_hot=True)

def weight_variable(shape):
  initial = tf.truncated_normal(shape, stddev=0.1)
  return tf.Variable(initial)

def bias_variable(shape):
  initial = tf.constant(0,1, shape=shape)
  return tf.Variable(initial)

def conv2d(x, W):
  return tf.nn.conv2d(x, W, strides=[1,1,1,1], padding='SAME')

def max_pool_2x2(x):
  return tf.nn.max_pool(x, ksize=[1,2,2,1],
                        strides=[1,2,2,1], padding='SAME')

with tf.Graph().as_default():
  with tf.name_scope('input'):
    x = tf.placeholder(tf.float32, [None, 784], name='x-input')
    x_image = tf.reshape(x, [-1,28,28,1])
  
  with tf.name_scope('convolution1'):
    W_conv1 = weight_variable([5,5,1,32])
    b_conv1 = bias_variable([32])
    h_conv1 = tf.nn.relu(conv2d(x_image, W_conv1) + b_conv1)
    _ = tf.histogram_summary('h_conv1', h_conv1)
  
  with tf.name_scope('pooling1'):
    h_pool1 = max_pool_2x2(h_conv1)
  
  with tf.name_scope('convolution2'):
    W_conv2 = weight_variable([5, 5, 32, 64])
    b_conv2 = bias_variable([64])
    h_conv2 = tf.nn.relu(conv2d(h_pool1, W_conv2) + b_conv2)
  
  with tf.name_scope('pooling2'):
    h_pool2 = max_pool_2x2(h_conv2)
  
  with tf.name_scope('fully-connected'):
    W_fc1 = weight_variable([7 * 7 * 64, 1024])
    b_fc1 = bias_variable([1024])
    h_pool2_flat = tf.reshape(h_pool2, [-1, 7*7*64])
    h_fc1 = tf.nn.relu(tf.matmul(h_pool2_flat, W_fc1) + b_fc1)
  
  with tf.name_scope('dropout'):
    keep_prob = tf.placeholder(tf.float32)
    h_fc1_drop = tf.nn.dropout(h_fc1, keep_prob)
  
  with tf.name_scope('readout'):
    W_fc2 = weight_variable([1024, 10])
    b_fc2 = bias_variable([10])
    y_conv=tf.nn.softmax(tf.matmul(h_fc1_drop, W_fc2) + b_fc2)
  
  with tf.name_scope('optimizer'):
    y_ = tf.placeholder(tf.float32, [None, 10], name='y-input')
    cross_entropy = -tf.reduce_sum(y_*tf.log(y_conv))
    train_step = tf.train.AdamOptimizer(1e-4).minimize(cross_entropy)
    correct_prediction = tf.equal(tf.argmax(y_conv,1), tf.argmax(y_,1))
    accuracy = tf.reduce_mean(tf.cast(correct_prediction, tf.float32))
  
  # Logging data for TensorBoard
  _ = tf.scalar_summary('cross entropy', cross_entropy)
  _ = tf.scalar_summary('accuracy', accuracy)
  
  with tf.Session() as sess:
    saver = tf.train.Saver()
    writer = tf.train.SummaryWriter('/tmp/cnn', graph_def=sess.graph_def)
    sess.run(tf.initialize_all_variables())
  
    for i in range(20001):
      batch = mnist.train.next_batch(50)
      train_step.run(feed_dict={x: batch[0], y_: batch[1], keep_prob: 0.5})
  
      if i%100 == 0:
        feed={x: batch[0], y_: batch[1], keep_prob: 1.0}
        summary_str, acc = sess.run(
          [tf.merge_all_summaries(), accuracy], feed_dict=feed)
        writer.add_summary(summary_str, i)
        print("step %d, training accuracy %g" % (i, acc))
        saver.save(sess, 'train_data', global_step = i)

    print("test accuracy %g"%accuracy.eval(feed_dict={
        x: mnist.test.images, y_: mnist.test.labels, keep_prob: 1.0}))

畳み込みフィルターの使い方が最初の例と違っているのは、下記の点です。

最初の例

  with tf.name_scope('convolution1'):
    W_conv1 = weight_variable([5,5,1,num_filters])
    h_conv1 = tf.nn.relu(tf.abs(conv2d(x_image, W_conv1))-0.2)
    _ = tf.histogram_summary('h_conv1', h_conv1)

今の場合

  with tf.name_scope('convolution1'):
    W_conv1 = weight_variable([5,5,1,32])
    b_conv1 = bias_variable([32])
    h_conv1 = tf.nn.relu(conv2d(x_image, W_conv1) + b_conv1)
    _ = tf.histogram_summary('h_conv1', h_conv1)

最初の例では、エッジを検出するという意図を持って、tf.nn.relu(tf.abs(conv2d(x_image, W_conv1))-0.2) という関数を使っていましたが、今の場合、エッジの検出が特徴抽出として有効かどうかはわかりませんので、tf.nn.relu(conv2d(x_image, W_conv1) + b_conv1) という一般的な線形演算にしてあります。

また、このコードは、実行に時間がかかるので、学習結果は定期的にデータファイル「train_data-」に書き出すようにしてあります。(筆者は、Xeon2.4GHz, 16core のマシンを使ってますが、これでコーヒーをゆっくり飲みながら待つ程度の時間です。)

実行が終わると、テストセットに対する正答率(99.24%)が表示されて、最終結果が「train_data-20000」に保存されます。

TensorBoardで見た学習曲線とネットワークのグラフは、こちらになります。

そして、最終結果のデータ「train_data-20000」から1段目と2段目のフィルターを可視化するコードも作っておきました。(トレーニングのコードと内容がほぼ被っています。。。冗長です。。。。すいません。)

from __future__ import absolute_import
from __future__ import division
from __future__ import print_function

from tensorflow.examples.tutorials.mnist import input_data
import tensorflow as tf
import matplotlib.pyplot as plt

mnist = input_data.read_data_sets("/tmp/data/", one_hot=True)

def visualize(sess):
  C = 10
  batch = mnist.test.next_batch(C)

  # Layer1
  fig = plt.figure()
  num_filters = 32
  h_pool1_flat = tf.reshape(h_pool1, [-1, 14*14*num_filters])
  res = h_pool1_flat.eval(session=sess, feed_dict={x: batch[0]})
  w1 = W_conv1.eval(session=sess)

  for i in range(num_filters):
    subplot = fig.add_subplot(C+1, num_filters+1, i+2)
    subplot.set_xticks([])
    subplot.set_yticks([])
    subplot.imshow(w1[:,:,0,i], cmap=plt.cm.gray_r, interpolation='nearest')
  for c in range(C):
    chars = res[c].reshape(14,14,num_filters)
    subplot = fig.add_subplot(C+1, num_filters+1, 1+(c+1)*(num_filters+1))
    subplot.set_xticks([])
    subplot.set_yticks([])
    subplot.imshow(batch[0][c].reshape(28,28), vmax=1, vmin=0,
                   cmap=plt.cm.gray_r, interpolation='nearest')
    for i in range(num_filters):
      subplot = fig.add_subplot(C+1, num_filters+1, (c+1)*(num_filters+1)+(i+2))
      subplot.set_xticks([])
      subplot.set_yticks([])
      subplot.imshow(chars[:,:,i], cmap=plt.cm.gray_r, interpolation='nearest')

  # Layer2
  fig = plt.figure()
  num_filters = 64
  res = h_pool2_flat.eval(session=sess, feed_dict={x: batch[0]})
  w2 = W_conv2.eval(session=sess)

  for i in range(int(num_filters/2)):
    subplot = fig.add_subplot(C*2+2, num_filters/2+1, i+2)
    subplot.set_xticks([])
    subplot.set_yticks([])
    subplot.imshow(w2[:,:,0,i], cmap=plt.cm.gray_r, interpolation='nearest')
  for i in range(int(num_filters/2),num_filters):
    subplot = fig.add_subplot(C*2+2, num_filters/2+1,
                              (num_filters/2+1)+(i-num_filters/2)+2)
    subplot.set_xticks([])
    subplot.set_yticks([])
    subplot.imshow(w2[:,:,0,i], cmap=plt.cm.gray_r, interpolation='nearest')

  for c in range(C):
    chars = res[c].reshape(7,7,num_filters)
    subplot = fig.add_subplot(C*2+2, num_filters/2+1,
                              1+((c+1)*2)*(num_filters/2+1))
    subplot.set_xticks([])
    subplot.set_yticks([])
    subplot.imshow(batch[0][c].reshape(28,28), vmax=1, vmin=0,
                   cmap=plt.cm.gray_r, interpolation='nearest')
    for i in range(int(num_filters/2)):
      subplot = fig.add_subplot(C*2+2, num_filters/2+1,
                                1+((c+1)*2)*(num_filters/2+1)+(i+1))
      subplot.set_xticks([])
      subplot.set_yticks([])
      subplot.imshow(chars[:,:,i], cmap=plt.cm.gray_r, interpolation='nearest')
    for i in range(int(num_filters/2),num_filters):
      subplot = fig.add_subplot(C*2+2, num_filters/2+1,
                          1+((c+1)*2+1)*(num_filters/2+1)+(i-num_filters/2+1))
      subplot.set_xticks([])
      subplot.set_yticks([])
      subplot.imshow(chars[:,:,i], cmap=plt.cm.gray_r, interpolation='nearest')

  plt.show()

def weight_variable(shape):
  initial = tf.truncated_normal(shape, stddev=0.1)
  return tf.Variable(initial)

def bias_variable(shape):
  initial = tf.constant(0,1, shape=shape)
  return tf.Variable(initial)

def conv2d(x, W):
  return tf.nn.conv2d(x, W, strides=[1,1,1,1], padding='SAME')

def max_pool_2x2(x):
  return tf.nn.max_pool(x, ksize=[1,2,2,1],
                        strides=[1,2,2,1], padding='SAME')

with tf.Graph().as_default():
  with tf.name_scope('input'):
    x = tf.placeholder(tf.float32, [None, 784], name='x-input')
    x_image = tf.reshape(x, [-1,28,28,1])
  
  with tf.name_scope('convolution1'):
    W_conv1 = weight_variable([5,5,1,32])
    b_conv1 = bias_variable([32])
    h_conv1 = tf.nn.relu(conv2d(x_image, W_conv1) + b_conv1)
    _ = tf.histogram_summary('h_conv1', h_conv1)
  
  with tf.name_scope('pooling1'):
    h_pool1 = max_pool_2x2(h_conv1)
  
  with tf.name_scope('convolution2'):
    W_conv2 = weight_variable([5, 5, 32, 64])
    b_conv2 = bias_variable([64])
    h_conv2 = tf.nn.relu(conv2d(h_pool1, W_conv2) + b_conv2)
  
  with tf.name_scope('pooling2'):
    h_pool2 = max_pool_2x2(h_conv2)
  
  with tf.name_scope('fully-connected'):
    W_fc1 = weight_variable([7 * 7 * 64, 1024])
    b_fc1 = bias_variable([1024])
    h_pool2_flat = tf.reshape(h_pool2, [-1, 7*7*64])
    h_fc1 = tf.nn.relu(tf.matmul(h_pool2_flat, W_fc1) + b_fc1)
  
  with tf.name_scope('dropout'):
    keep_prob = tf.placeholder(tf.float32)
    h_fc1_drop = tf.nn.dropout(h_fc1, keep_prob)
  
  with tf.name_scope('readout'):
    W_fc2 = weight_variable([1024, 10])
    b_fc2 = bias_variable([10])
    y_conv=tf.nn.softmax(tf.matmul(h_fc1_drop, W_fc2) + b_fc2)
  
  with tf.name_scope('optimizer'):
    y_ = tf.placeholder(tf.float32, [None, 10], name='y-input')
    cross_entropy = -tf.reduce_sum(y_*tf.log(y_conv))
    train_step = tf.train.AdamOptimizer(1e-4).minimize(cross_entropy)
    correct_prediction = tf.equal(tf.argmax(y_conv,1), tf.argmax(y_,1))
    accuracy = tf.reduce_mean(tf.cast(correct_prediction, tf.float32))
  
  # Logging data for TensorBoard
  _ = tf.scalar_summary('cross entropy', cross_entropy)
  _ = tf.scalar_summary('accuracy', accuracy)
  
  with tf.Session() as sess:
    saver = tf.train.Saver()
    saver.restore(sess, 'train_data-20000')
    visualize(sess) 

これを使うと、1段目のフィルターは次のようになります。

一番上がフィルターそのもので、その下にサンプルデータにフィルター(+プーリング)をかけた結果が並んでいます。

そして2段めのフィルターがこちらです。

それぞれの文字について、64種類の特徴が抽出されていることになります。もはや何の特徴を見ているのかはわかりませんが、これで、99.2%の識別ができるというわけです。なお、一番上は、64枚のフィルターについて、それぞれ32個あるサブフィルターのうちの1つを例として表示しています。