Deep Learning CNN’s in Tensorflow with GPUs
Taylor Guo, 2017年6月11日
本文讲解卷积神经网络架构，如何在tensorflow中创建卷积神经网络，如何在图像标注上做预测。最后是如何在GPU上运行模型，这样就可以节省时间来创建更好的模型，而不是等待它们收敛。
卷积神经网络介绍
创建卷积神经网络 并 在CPU上训练
GPU上训练
基本的机器学习知识
基本的TensorFlow知识
AWS账号 （GPU）
卷积神经网络是目前图像分类的最新架构。广泛地应用于人脸识别，自动驾驶，物体识别。
卷积神经网络的基本架构由3部分组成：卷积，池化 和 全连接。这些组件一起作用，学习输入图像的稠密特征表示。
卷积主要是由卷积核组成（最上面的绿色部分），也称为滤波器，是用一个滑动窗口从输入图像中提取特征。滤波器在输入图像上平移的位移称为步长。每次移动都会计算卷积核和当前输入图像区域的矩阵乘法。可以用多个滤波器堆叠来创建输入图像的高维表示。
如果滤波器没有均匀地与输入图像大小相映射会出现什么状况？
有两种方法处理滤波器大小和输入图像大小尺寸不同，same padding 和 valid padding。Same padding在输入图像的边界外补零（如上图所示）来保留输入图像的长宽。Valid padding不补边缘，在卷积时不会超过输入图像的边界。
通常都会使用same padding 或者 快速降低输入图像的维度。
最后是激活函数（通常是ReLU），对卷积后的结果进行非线性处理。ReLU和其他的激活函数不同，比如sigmoid和tanh，ReLU只有一边，是非对称的。这种一边的形式的激活特性可以让网络创建稀疏表示（隐藏单元是0值），可以增加计算效率。
池化操作为了降低维度。它用一个函数对相邻元素数值进行求和。两个常用的函数是最大池化和平均池化。通过计算输入图像区域的最大值，输出图像将这个区域内的亮度值求和，（进行最大或平均）。
池化层也有一个内核，padding沿着步长移动。为了计算池化操作的输出图像的大小，可以用以下公式计算：
（输入图像宽度 - 内核宽度 + 2*padding） / 步长 + 1
全连接层与神经网络比较类似。输入部分的每个神经元都和输出部分的每个神经元相连接；完全连接起来。因为这种连接，输出部分的每个神经元都只使用一次。
在卷积神经网络中，输入层通过池化后填入全连接层。根据不同的任务，用回归或分类算法创建想要的输出。
上面是卷积神经网络的构建。将输入图像传入卷积网络，提取高维特征。池化总结了图像中的空间特征，减小了维度。最后，这个特征表示传入全连接层给 分类器 或者 回归器。
现在已经了解了卷积神经网络，可以在TensorFlow中创建卷积神经网络了。
创建卷积神经网络训练 MNIST 数据集（手写数字图像集）。训练之后，可以获得~98.0%精确度，大概10K次迭代。
Anaconda
非Anaconda
$ pip install tensorflow
需要创建3个分离的输入；训练集，验证集，测试集。
验证集提供了更多的数据来调参，可以更好地训练模型。
数据可以用如下命令下载：
$ curl https://pjreddie.com/media/files/mnist_train.csv -o data/mnist_train.csv # 104 MB
$ curl https://pjreddie.com/media/files/mnist_test.csv -o data/mnist_test.csv # 17.4 MB
Python Code
import numpy as np
IMAGE_SIZE = 28
def load_train_data(data_path, validation_size=500):
"""
Load mnist data. Each row in csv is formatted (label, input)
:return: 3D Tensor input of train and validation set with 2D Tensor of one hot encoded image labels
"""
# Data format: 1 byte label, 28 * 28 input
train_data = np.genfromtxt(data_path, delimiter=',', dtype=np.float32)
x_train = train_data[:, 1:]
# Get label and one-hot encode
y_train = train_data[:, 0]
y_train = (np.arange(10) == y_train[:, None]).astype(np.float32)
# get a validation set and remove it from the train set
x_train, x_val, y_train, y_val = x_train[0:(len(x_train) - validation_size), :], x_train[(
len(x_train) - validation_size):len(x_train), :], \
y_train[0:(len(y_train) - validation_size), :], y_train[(
len(y_train) - validation_size):len(y_train), :]
# reformat the data so it's not flat
x_train = x_train.reshape(len(x_train), IMAGE_SIZE, IMAGE_SIZE, 1)
x_val = x_val.reshape(len(x_val), IMAGE_SIZE, IMAGE_SIZE, 1)
return x_train, x_val, y_train, y_val
def load_test_data(data_path):
"""
Load mnist test data
:return: 3D Tensor input of train and validation set with 2D Tensor of one hot encoded image labels
"""
test_data = np.genfromtxt(data_path, delimiter=',', dtype=np.float32)
x_test = test_data[:, 1:]
y_test = np.array(test_data[:, 0])
y_test = (np.arange(10) == y_test[:, None]).astype(np.float32)
x_test = x_test.reshape(len(x_test), IMAGE_SIZE, IMAGE_SIZE, 1)
return x_test, y_test可以用更有帮助的函数创建网络。这些函数用以创建之前讨论的单个组件。
模型定义：
import tensorflow as tf
class Model(object):
def __init__(self, batch_size=128, learning_rate=1e-4, num_labels=10):
self._batch_size = batch_size
self._learning_rate = learning_rate
self._num_labels = num_labels
def inference(self, images, keep_prob):
pass
def train(self, loss, global_step):
pass
def loss(self, logits, labels):
pass
def accuracy(self, logits, labels):
pass
def _create_conv2d(self, x, W):
return tf.nn.conv2d(input=x,
filter=W,
strides=[1, 1, 1, 1],
padding='SAME')
def _create_max_pool_2x2(self, input):
return tf.nn.max_pool(value=input,
ksize=[1, 2, 2, 1],
strides=[1, 2, 2, 1],
padding='SAME')
def _create_weights(self, shape):
return tf.Variable(tf.truncated_normal(shape=shape, stddev=0.1, dtype=tf.float32))
def _create_bias(self, shape):
return tf.Variable(tf.constant(1., shape=shape, dtype=tf.float32))
def _activation_summary(self, x):
tensor_name = x.op.name
tf.summary.histogram(tensor_name + '/activations', x)
tf.summary.scalar(tensor_name + '/sparsity', tf.nn.zero_fraction(x))下面是训练模型的代码。有3个主要的函数：
import tensorflow as tf
class Model(object):
def __init__(self, batch_size=128, learning_rate=1e-4, num_labels=10):
self._batch_size = batch_size
self._learning_rate = learning_rate
self._num_labels = num_labels
def inference(self, images, keep_prob):
with tf.variable_scope('conv1') as scope:
kernel = self._create_weights([5, 5, 1, 32])
conv = self._create_conv2d(images, kernel)
bias = self._create_bias([32])
preactivation = tf.nn.bias_add(conv, bias)
conv1 = tf.nn.relu(preactivation, name=scope.name)
self._activation_summary(conv1)
# pool 1
h_pool1 = self._create_max_pool_2x2(conv1)
with tf.variable_scope('conv2') as scope:
kernel = self._create_weights([5, 5, 32, 64])
conv = self._create_conv2d(h_pool1, kernel)
bias = self._create_bias([64])
preactivation = tf.nn.bias_add(conv, bias)
conv2 = tf.nn.relu(preactivation, name=scope.name)
self._activation_summary(conv2)
# pool 2
h_pool2 = self._create_max_pool_2x2(conv2)
with tf.variable_scope('local1') as scope:
reshape = tf.reshape(h_pool2, [-1, 7 * 7 * 64])
W_fc1 = self._create_weights([7 * 7 * 64, 1024])
b_fc1 = self._create_bias([1024])
local1 = tf.nn.relu(tf.matmul(reshape, W_fc1) + b_fc1, name=scope.name)
self._activation_summary(local1)
with tf.variable_scope('local2_linear') as scope:
W_fc2 = self._create_weights([1024, self._num_labels])
b_fc2 = self._create_bias([self._num_labels])
local1_drop = tf.nn.dropout(local1, keep_prob)
local2 = tf.nn.bias_add(tf.matmul(local1_drop, W_fc2), b_fc2, name=scope.name)
self._activation_summary(local2)
return local2
def train(self, loss, global_step):
tf.summary.scalar('learning_rate', self._learning_rate)
train_op = tf.train.AdamOptimizer(self._learning_rate).minimize(loss, global_step=global_step)
return train_op
def loss(self, logits, labels):
with tf.variable_scope('loss') as scope:
cross_entropy = tf.nn.softmax_cross_entropy_with_logits(logits=logits, labels=labels)
cost = tf.reduce_mean(cross_entropy, name=scope.name)
tf.summary.scalar('cost', cost)
return cost
def accuracy(self, logits, labels):
with tf.variable_scope('accuracy') as scope:
accuracy = tf.reduce_mean(tf.cast(tf.equal(tf.argmax(logits, 1), tf.argmax(labels, 1)),
dtype=tf.float32), name=scope.name)
tf.summary.scalar('accuracy', accuracy)
return accuracy
def _create_conv2d(self, x, W):
return tf.nn.conv2d(input=x,
filter=W,
strides=[1, 1, 1, 1],
padding='SAME')
def _create_max_pool_2x2(self, input):
return tf.nn.max_pool(value=input,
ksize=[1, 2, 2, 1],
strides=[1, 2, 2, 1],
padding='SAME')
def _create_weights(self, shape):
return tf.Variable(tf.truncated_normal(shape=shape, stddev=0.1, dtype=tf.float32))
def _create_bias(self, shape):
return tf.Variable(tf.constant(1., shape=shape, dtype=tf.float32))
def _activation_summary(self, x):
tensor_name = x.op.name
tf.summary.histogram(tensor_name + '/activations', x)
tf.summary.scalar(tensor_name + '/sparsity', tf.nn.zero_fraction(x))代码在这里
Inference 这个函数创建对输入表示创建一个预测。它对每个输入返回一个 1×10的张量。张量中的值会传递给损失函数，计算这个预测与基准相差多少。
根据参数batch_size，每次需要处理128个图像。这个技术称为mini-batch。
在更小的数据集上batch上处理输入数据，而不是整个数据集，这样就可以有效使用内存。
因为是在每个小型数据集上更新权重，而不是处理所有样本，模型也能够更快收敛。
Loss 可以用softmax cross entropy 函数执行N路分类。Softmax可以归一化（将tensor加成1个数值）从inference函数产生的输入数据。
张量归一化后，Cross entropy 计算编码标注。cross entropy计算预测值与基准值之间的差异。每次迭代，优化器都最小化cross entropy的值。
Cross Entropy ： ∑−y∗log（pred）
10k次迭代训练模型。
每1000次迭代，测试验证集上的模型，设置理想的精度。
最后，在测试集上评估训练好的模型，得到测量精度。
经过10k次迭代，精度大概在98.0%。
运行下面的命令，执行代码：
$ python3 mnist_conv2d_medium_tutorial/train.py
import tensorflow as tf
import mnist_conv2d_medium_tutorial.mnist as mnist
from mnist_conv2d_medium_tutorial.model import Model
FLAGS = tf.app.flags.FLAGS
NUM_LABELS = 10
def train():
model = Model()
with tf.Graph().as_default():
images, val_images, labels, val_labels = mnist.load_train_data(FLAGS.train_data)
x = tf.placeholder(shape=[None, mnist.IMAGE_SIZE, mnist.IMAGE_SIZE, 1], dtype=tf.float32, name='x')
y = tf.placeholder(shape=[None, NUM_LABELS], dtype=tf.float32, name='y')
keep_prob = tf.placeholder(tf.float32, name='dropout_prob')
global_step = tf.contrib.framework.get_or_create_global_step()
logits = model.inference(x, keep_prob=keep_prob)
loss = model.loss(logits=logits, labels=y)
accuracy = model.accuracy(logits, y)
summary_op = tf.summary.merge_all()
train_op = model.train(loss, global_step=global_step)
init = tf.global_variables_initializer()
saver = tf.train.Saver()
with tf.Session(config=tf.ConfigProto(log_device_placement=True)) as sess:
writer = tf.summary.FileWriter(FLAGS.summary_dir, sess.graph)
sess.run(init)
for i in range(FLAGS.num_iter):
offset = (i * FLAGS.batch_size) % (len(images) - FLAGS.batch_size)
batch_x, batch_y = images[offset:(offset + FLAGS.batch_size), :], labels[
offset:(offset + FLAGS.batch_size), :]
_, cur_loss, summary = sess.run([train_op, loss, summary_op],
feed_dict={x: batch_x, y: batch_y, keep_prob: 0.5})
writer.add_summary(summary, i)
print(i, cur_loss)
if i % 1000 == 0:
validation_accuracy = accuracy.eval(feed_dict={x: val_images, y: val_labels, keep_prob: 1.0})
print('Iter {} Accuracy: {}'.format(i, validation_accuracy))
if i == FLAGS.num_iter - 1:
saver.save(sess, FLAGS.checkpoint_file_path, global_step)
def main(argv=None):
train()
if __name__ == '__main__':
tf.app.flags.DEFINE_integer('batch_size', 128, 'size of training batches')
tf.app.flags.DEFINE_integer('num_iter', 10000, 'number of training iterations')
tf.app.flags.DEFINE_string('checkpoint_file_path', 'checkpoints/model.ckpt-10000', 'path to checkpoint file')
tf.app.flags.DEFINE_string('train_data', 'data/mnist_train.csv', 'path to train and test data')
tf.app.flags.DEFINE_string('summary_dir', 'graphs', 'path to directory for storing summaries')
tf.app.run()模型训练好后，就可以在测试集上评估了。
import tensorflow as tf
from mnist_conv2d_medium_tutorial import mnist
from mnist_conv2d_medium_tutorial.model import Model
FLAGS = tf.app.flags.FLAGS
def evaluate():
with tf.Graph().as_default():
images, labels = mnist.load_test_data(FLAGS.test_data)
model = Model()
logits = model.inference(images, keep_prob=1.0)
accuracy = model.accuracy(logits, labels)
saver = tf.train.Saver()
with tf.Session() as sess:
tf.global_variables_initializer().run()
saver.restore(sess, FLAGS.checkpoint_file_path)
total_accuracy = sess.run([accuracy])
print('Test accuracy: {}'.format(total_accuracy))
def main(argv=None):
evaluate()
if __name__ == '__main__':
tf.app.flags.DEFINE_string('checkpoint_file_path', 'checkpoints/model.ckpt-10000', 'path to checkpoint file')
tf.app.flags.DEFINE_string('test_data', 'data/mnist_test.csv', 'path to test data')
tf.app.run()$ python3 mnist_conv2d_medium_tutorial/evaluate.py
运行以下命令，可视化结果：
$ tensorboard –logdir=graphs/ –port=6006 navigate in browser: localhost:6006