y君用tensorflow跑MNIST的一个代码示例（带中文注释） - 机器人与人工智能爱好者论坛

# -*- coding: utf-8 -*-
"""Functions for downloading and reading MNIST data."""
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
import gzip
import os
import numpy
from six.moves import urllib
from six.moves import xrange # pylint: disable=redefined-builtin
SOURCE_URL = 'http://yann.lecun.com/exdb/mnist/'
#DataSet对象
'''
self 类似 C++ this
'''
class DataSet(object):
def __init__(self, images, labels, fake_data=False):
if fake_data:
self._num_examples = 10000
else:
assert images.shape[0] == labels.shape[0], (
"images.shape: %s labels.shape: %s" % (images.shape,
labels.shape))
#保存数据长度到num_examples
self._num_examples = images.shape[0]
print ("self._num_examples : ", (self._num_examples))
# Convert shape from [num examples, rows, columns, depth]
# to [num examples, rows*columns] (assuming depth == 1)
print ("num ",images.shape[0])
print ("rows",images.shape[1])
print ("cols",images.shape[2])
assert images.shape[3] == 1
images = images.reshape(images.shape[0],
images.shape[1] * images.shape[2])
# Convert from [0, 255] -> [0.0, 1.0].
images = images.astype(numpy.float32)
images = numpy.multiply(images, 1.0 / 255.0)
print ("-------")
self._images = images
self._labels = labels
self._epochs_completed = 0
self._index_in_epoch = 0
print ("__init__ end")
@property
def images(self):
return self._images
@property
def labels(self):
return self._labels
@property
def num_examples(self):
return self._num_examples
@property
def epochs_completed(self):
return self._epochs_completed
#作用为切换样本
def next_batch(self, batch_size, fake_data=False):
"""Return the next `batch_size` examples from this data set."""
if fake_data:
fake_image = [1.0 for _ in xrange(784)]
fake_label = 0
return [fake_image for _ in xrange(batch_size)], [
fake_label for _ in xrange(batch_size)]
start = self._index_in_epoch
self._index_in_epoch += batch_size
#判断index是否超过最大样本数量超过的话就是训练完了
if self._index_in_epoch > self._num_examples:
# Finished epoch
self._epochs_completed += 1
# Shuffle the data
perm = numpy.arange(self._num_examples)
numpy.random.shuffle(perm)
self._images = self._images[perm]
self._labels = self._labels[perm]
# Start next epoch
start = 0
self._index_in_epoch = batch_size
assert batch_size <= self._num_examples
#计算切换样本后的index
end = self._index_in_epoch
#根据计算后的index 在iamges 和label 中读取数据
return self._images[start:end], self._labels[start:end]
#读取一个int32类型的数据
def _read32(bytestream):
dt = numpy.dtype(numpy.uint32).newbyteorder('>')
result = numpy.frombuffer(bytestream.read(4), dtype=dt)
return result
def dense_to_one_hot(labels_dense, num_classes=10):
"""Convert class labels from scalars to one-hot vectors."""
#获取标签数量
num_labels = labels_dense.shape[0]
index_offset = numpy.arange(num_labels) * num_classes
#清零
labels_one_hot = numpy.zeros((num_labels, num_classes))
#末尾填1 ?
labels_one_hot.flat[index_offset + labels_dense.ravel()] = 1
return labels_one_hot
#函数功能是下载文件如果有文件就不下载了
def maybe_download(filename, work_directory):
if not os.path.exists(work_directory):
#如果不存在目录就创建一个
os.mkdir(work_directory)
#把文件目录和文件名合成
filepath = os.path.join(work_directory, filename)
print('filename : ', filename)
if not os.path.exists(filepath):
#如果不存在文件就开始下载
print('SOURCE_URL + filename : ', SOURCE_URL + filename)
filepath, _ = urllib.request.urlretrieve(SOURCE_URL + filename, filepath)
#获取文件状态
statinfo = os.stat(filepath)
#输出下载成功
print('Successfully downloaded', filename, statinfo.st_size, 'bytes.')
return filepath
def getfiledata(filepath):
print('mapfile : ', filepath)
#打开目标文件
with gzip.open(filepath) as bytestream:
magic = _read32(bytestream)
if magic != 2051:
raise ValueError(
'Invalid magic number %d in MNIST image file: %s' %
(magic, filepath))
num_images = _read32(bytestream)
rows = _read32(bytestream)
cols = _read32(bytestream)
buf = bytestream.read(rows * cols * num_images)
data = numpy.frombuffer(buf, dtype=numpy.uint8)
#扩展为 data[num_images][rows][cols][1]
data = data.reshape(num_images, rows, cols,1)
return data
def getlabeldata(filepath, one_hot=False):
"""Extract the labels into a 1D uint8 numpy array [index]."""
print('mapfile : ', filepath)
with gzip.open(filepath) as bytestream:
magic = _read32(bytestream)
if magic != 2049:
raise ValueError(
'Invalid magic number %d in MNIST label file: %s' %
(magic, filepath))
num_items = _read32(bytestream)
buf = bytestream.read(num_items)
labels = numpy.frombuffer(buf, dtype=numpy.uint8)
if one_hot:
return dense_to_one_hot(labels)
return labels
def downloadimagefile(train_dir,fake_data=False, one_hot=False):
class DataSets(object):
pass
data_sets = DataSets()
#55000训练图像，5000验证图片训练图像，5000验证图片
TRAIN_IMAGES = 'train-images-idx3-ubyte.gz'
#训练集标签
TRAIN_LABELS = 'train-labels-idx1-ubyte.gz'
#测试集图像 - 10000图片
TEST_IMAGES = 't10k-images-idx3-ubyte.gz'
#测试集标签
TEST_LABELS = 't10k-labels-idx1-ubyte.gz'
VALIDATION_SIZE = 5000
print ("load file")
#下载训练集图像 local_file 存放的是下载下来的文件路径
file_path = maybe_download(TRAIN_IMAGES, train_dir)
train_images = getfiledata(file_path)
#训练集标签
file_path = maybe_download(TRAIN_LABELS, train_dir)
train_labels = getlabeldata(file_path, one_hot=one_hot)
#测试集图像
file_path = maybe_download(TEST_IMAGES, train_dir)
test_images = getfiledata(file_path)
#测试集标签
file_path = maybe_download(TEST_LABELS, train_dir)
test_labels = getlabeldata(file_path, one_hot=one_hot)
validation_images = train_images[:VALIDATION_SIZE]
validation_labels = train_labels[:VALIDATION_SIZE]
train_images = train_images[VALIDATION_SIZE:]
train_labels = train_labels[VALIDATION_SIZE:]
#作为主要训练集。
data_sets.train = DataSet(train_images, train_labels)
#用于迭代验证训练准确度。
data_sets.validation = DataSet(validation_images, validation_labels)
#用于最终测试训练准确度
data_sets.test = DataSet(test_images, test_labels)
return data_sets

复制代码

# -*- coding: utf-8 -*-
import tensorflow as tf
import get_data
import numpy
#下载数据
import math
import numpy as np
from array import array
'''
x = np.arange(1, 1001).reshape(20, 50)
print x
print x.flat[3]
a = numpy.array([[1,2,3],[4,5,6],[7,8,9]])
print a.shape
'''
mnist = get_data.downloadimagefile("MNIST_data/",one_hot=True)
#创建一个session 方便在shell中用
sess = tf.InteractiveSession()
#开始构建回归模型
#先创建占位符占位符并不能直接输出因为这个格式只有tf自己能够识别这是一个二维数组一维任意长度
#一维为batch大小二维是图片所有像素点 28*28=784
x = tf.placeholder("float", [None, 784])
y_ = tf.placeholder("float", shape=[None, 10])
#定义权重W和偏置b 全部初始化为0
#w是一个28*28*10的矩阵因为有28*28个像素点和10个输出值
W = tf.Variable(tf.zeros([784,10]))
#因为有10个分类所以b 长度为10
b = tf.Variable(tf.zeros([10]))
#Variable 定义的变量需要在session之前调用initialize_all_variables完成初始化才能载session中使用
sess.run(tf.initialize_all_variables())
#计算每个分类的softmax概率值
y = tf.nn.softmax(tf.matmul(x,W) + b)
#计算交叉熵
#tf.reduce_sum把minibatch里的每张图片的交叉熵值都加起来了。我们计算的交叉熵是指整个minibatch的。
cross_entropy = -tf.reduce_sum(y_*tf.log(y))
#我们已经定义好了模型和训练的时候用的损失函数，接下来使用TensorFlow来训练。
#因为TensorFlow知道整个计算图，它会用自动微分法来找到损失函数对于各个变量的梯度。
#TensorFlow有大量内置的优化算法. 这个例子中，我们用最速下降法让交叉熵下降，步长为0.01.
train_step = tf.train.GradientDescentOptimizer(0.01).minimize(cross_entropy)
#这一行代码实际上是用来往计算图上添加一个新操作，其中包括计算梯度，计算每个参数的步长变化，并且计算出新的参数值。
#train_step这个操作，用梯度下降来更新权值。因此，整个模型的训练可以通过反复地运行train_step来完成。
#每次迭代加载50个样本
rows = 28
cols = 28
num_images = 1
for i in range(1000):
#next_batch会自动把数据转换成tf能认的
batch = mnist.train.next_batch(50)
train_step.run(feed_dict={x: batch[0], y_: batch[1]})
correct_prediction = tf.equal(tf.argmax(y,1), tf.argmax(y_,1))
accuracy = tf.reduce_mean(tf.cast(correct_prediction, "float"))
print accuracy.eval(feed_dict={x: mnist.test.images, y_: mnist.test.labels})
#下面构建一个多层卷积网络上面的代码只有%91准确率
#在创建模型之前，我们先来创建权重和偏置。一般来说，初始化时应加入轻微噪声，来打破对称性，防止零梯度的问题。
#因为我们用的是ReLU，所以用稍大于0的值来初始化偏置能够避免节点输出恒为0的问题（dead neurons）。
#为了不在建立模型的时候反复做初始化操作，我们定义两个函数用于初始化。
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)
#卷积和池化
#TensorFlow在卷积和池化上有很强的灵活性。
#我们怎么处理边界？步长应该设多大？在这个实例里，我们会一直使用vanilla版本。
#我们的卷积使用1步长（stride size），0边距（padding size）的模板，保证输出和输入是同一个大小。
#我们的池化用简单传统的2x2大小的模板做max pooling。为了代码更简洁，我们把这部分抽象成一个函数。
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')
#第一层卷积
#现在我们可以开始实现第一层了。它由一个卷积接一个max pooling完成。
#卷积在每个5x5的patch中算出32个特征。权重是一个[5, 5, 1, 32]的张量，
#前两个维度是patch的大小，接着是输入的通道数目，最后是输出的通道数目。输出对应一个同样大小的偏置向量。
W_conv1 = weight_variable([5, 5, 1, 32])
b_conv1 = bias_variable([32])
#为了用这一层，我们把x变成一个4d向量，第2、3维对应图片的宽高，最后一维代表颜色通道。
x_image = tf.reshape(x, [-1,28,28,1])
#我们把x_image和权值向量进行卷积相乘，加上偏置，使用ReLU激活函数，最后max pooling。
h_conv1 = tf.nn.relu(conv2d(x_image, W_conv1) + b_conv1)
h_pool1 = max_pool_2x2(h_conv1)
#第二层卷积
#为了构建一个更深的网络，我们会把几个类似的层堆叠起来。第二层中，每个5x5的patch会得到64个特征。
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)
h_pool2 = max_pool_2x2(h_conv2)
#密集连接层
#现在，图片降维到7x7，我们加入一个有1024个神经元的全连接层，用于处理整个图片。
#我们把池化层输出的张量reshape成一些向量，乘上权重矩阵，加上偏置，使用ReLU激活。
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)
#Dropout
#为了减少过拟合，我们在输出层之前加入dropout。我们用一个placeholder来代表一个神经元在dropout中被保留的概率。
#这样我们可以在训练过程中启用dropout，在测试过程中关闭dropout。
#TensorFlow的tf.nn.dropout操作会自动处理神经元输出值的scale。所以用dropout的时候可以不用考虑scale。
keep_prob = tf.placeholder("float")
h_fc1_drop = tf.nn.dropout(h_fc1, keep_prob)
#输出层
#最后，我们添加一个softmax层，就像前面的单层softmax regression一样。
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)
#训练和评估模型
#这次效果又有多好呢？我们用前面几乎一样的代码来测测看。
#只是我们会用更加复杂的ADAM优化器来做梯度最速下降，在feed_dict中加入额外的参数keep_prob来控制dropout比例。
#然后每100次迭代输出一次日志。
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, "float"))
sess.run(tf.initialize_all_variables())
for i in range(20000):
batch = mnist.train.next_batch(50)
if i%100 == 0:
train_accuracy = accuracy.eval(feed_dict={
x:batch[0], y_: batch[1], keep_prob: 1.0})
print "step %d, training accuracy %g"%(i, train_accuracy)
train_step.run(feed_dict={x: batch[0], y_: batch[1], keep_prob: 0.5})
print "test accuracy %g"%accuracy.eval(feed_dict={
x: mnist.test.images, y_: mnist.test.labels, keep_prob: 1.0})

复制代码