用Tensorflow搭建CNN卷积神经网络，实现MNIST手写数字识别写在前面的话训练数据本程序用到的第三方库干货，代码及讲解

1
2
3
4
TRAIN_LABEL_PATH = "/path/to/train.txt"
TRAIN_IMAGE_PATH = "/path/to/train/"
VAL_LABEL_PATH = "/path/to/val.txt"
VAL_IMAGE_PATH = "/path/to/val/"

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
def weigth_variable(shape, name):
    # 这里使用截断的正态分布，标准差为0.1
    initial = tf.truncated_normal(shape, stddev=0.1)
    return tf.Variable(initial, name = name)

def bias_variable(shape):
    # bias初始化为0.1避免死亡节点
    initial = tf.constant(0.1, shape=shape)
    return initial

def conv2d(x, W):
    # 参数中x是输入，W是卷积的参数，比如[5,5,1,32]：前面两个数字代表卷积核的尺寸；第三个数字代表有多少个channel。因为我们只有灰度单色，所以是1，如果是RGB彩色图片，这里应该是3。
    # 最后一个数字代表卷积核的数量，也就是这个卷积层会提取多少类的特征。
    # Strides代表卷积模板移动的步长，都是1代表会不遗漏地划过图片的每一个点。
    # Padding代表边界的处理方式，这里的SAME代表给边界加上Padding让卷积的输出和输入保持同样(SAME)是尺寸。
    return tf.nn.conv2d(x, W, strides=[1, 1, 1, 1], padding="SAME")

def max_pool_2x2(x):
    # tf.nn.max_pool是tf中的最大池化函数，我们这里用2*2的最大池化，即将2*2像素块降为1*1的像素。
    # 因为希望整体上缩小图片尺寸，因此池化层的strides也设为横竖两个方向以2为步长。如果步长还是1，那么我们会得到一个尺寸不变的图片。
    return tf.nn.max_pool(x, ksize=[1, 2, 2, 1], strides=[1, 2, 2, 1], padding="SAME")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
TRAIN_LABEL_DICT = {}
VAL_LABEL_DICT = {}
with open(TRAIN_LABEL_PATH, "r") as f:
    lines = f.readlines()
for line in lines:
    arr = line.strip().split(" ")
    TRAIN_LABEL_DICT[arr[0]] = int(arr[1])

with open(VAL_LABEL_PATH, "r") as f:
    lines = f.readlines()
for line in lines:
    arr = line.strip().split(" ")
    VAL_LABEL_DICT[arr[0]] = int(arr[1])

train_image_list = []
train_label_list = []
val_image_list = []
val_label_list = []
for filename, label in TRAIN_LABEL_DICT.items():
    if os.path.isfile(os.path.join(TRAIN_IMAGE_PATH, filename)):
        train_image_list.append(os.path.join(TRAIN_IMAGE_PATH, filename))
        train_label_list.append(label)

for filename, label in VAL_LABEL_DICT.items():
    if os.path.isfile(os.path.join(VAL_IMAGE_PATH, filename)):
        val_image_list.append(os.path.join(VAL_IMAGE_PATH, filename))
        val_label_list.append(label)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
CAPACITY = len(train_label_list)
VAL_COUNT = len(val_label_list)
# 转换类型 list => tensor
train_image_list_tensor = tf.convert_to_tensor(train_image_list)
train_label_list_tensor = tf.convert_to_tensor(train_label_list)
val_image_list_tensor = tf.convert_to_tensor(val_image_list)
val_label_list_tensor = tf.convert_to_tensor(val_label_list)
# 转换标签类型至OneHot向量
# 如：数字2 => [0, 0, 1, 0, 0, 0, 0, 0, 0, 0] 
train_label_list_tensor = tf.one_hot(train_label_list_tensor,10)
val_label_list_tensor = tf.one_hot(val_label_list_tensor, 10)
# 创建管道
train_input_queue = tf.train.slice_input_producer(
    [train_image_list_tensor, train_label_list_tensor],
    shuffle=False)
val_input_queue = tf.train.slice_input_producer(
    [val_image_list_tensor, val_label_list_tensor],
    shuffle=False)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
# 训练数据读取
train_file_content = tf.read_file(train_input_queue[0])
train_image = tf.image.decode_jpeg(train_file_content, channels=1)
train_image = tf.to_float(train_image)
# 训练数据预处理，归一化
tmp = tf.reshape(train_image, [784])
tmp = (tmp - 0.0) / (255.0)
train_image = tf.reshape(tmp, [28, 28, 1])
train_image.set_shape((28, 28, 1))

# 训练标签读取
train_label = train_input_queue[1]

# 创建训练batch对象
batch_size = 50
num_preprocess_threads = 1
min_queue_examples = 256
images = tf.train.shuffle_batch(
    [train_image, train_label],
    batch_size=batch_size,
    num_threads=num_preprocess_threads,
    capacity=CAPACITY,
    min_after_dequeue=min_queue_examples)


# 验证数据读取及预处理
val_file_content = tf.read_file(val_input_queue[0])
val_image = tf.image.decode_jpeg(val_file_content, channels=1)
val_image = tf.to_float(val_image)

tmp = tf.reshape(val_image, [784])
tmp = (tmp - 0.0) / (255.0)
val_image = tf.reshape(tmp, [28, 28, 1])
val_image.set_shape((28, 28, 1))

val_label = val_input_queue[1]

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
# 定义占位符
x = tf.placeholder(tf.float32, [None, 28, 28, 1])
y_ = tf.placeholder(tf.float32, [None, 10])

# 第一层卷积池化
W_conv1 = weigth_variable([5, 5, 1, 32], "w_conv1")
b_conv1 = bias_variable([32])
h_conv1 = tf.nn.relu(conv2d(x, W_conv1) + b_conv1)
h_pool1 = max_pool_2x2(h_conv1)

# 第二层卷积池化
W_conv2 = weigth_variable([5, 5, 32, 64], "w_conv2")
b_conv2 = bias_variable([64])
h_conv2 = tf.nn.relu(conv2d(h_pool1, W_conv2) + b_conv2)
h_pool2 = max_pool_2x2(h_conv2)

# 全连接隐藏层，激活函数为ReLU
W_fc1 = weigth_variable([7 * 7 * 64, 1024], "w_fc1")
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)

keep_prob = tf.placeholder(tf.float32)
h_fc1_drop = tf.nn.dropout(h_fc1, keep_prob)

# 全连接输出层，使用softmax分类器
W_fc2 = weigth_variable([1024, 10], "w_fc2")
b_fc2 = bias_variable([10])
y_conv = tf.nn.softmax(tf.matmul(h_fc1_drop, W_fc2) + b_fc2)

# 平均信息量
cross_entropy = tf.reduce_mean(-tf.reduce_sum(y_ * tf.log(y_conv), reduction_indices=[1]))
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))
tf.summary.scalar('accuracy', accuracy)

1
2
3
init_op = tf.global_variables_initializer()
saver = tf.train.Saver()
merged = tf.summary.merge_all()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
with tf.Session() as sess:

    # tensorboard   
    summary_writer = tf.summary.FileWriter('/tmp/tensorboard_logs', sess.graph)

    # 如果有已训练好的模型，就加载已训练好的
    # 否则初始化一个新参数模型
    try:
        saver.restore(sess, "/tmp/imageModel-0")
        print("model restored.")
    except Exception as e:
        sess.run(init_op)

    # 启动文件加载队列
    coord = tf.train.Coordinator()
    threads = tf.train.start_queue_runners(coord=coord)

    # 训练迭代
    for i in range(2000):
        sample, label = sess.run([images[0], images[1]])
        # 每100次迭代打印一次训练进度
        if i % 100 == 0:
            #### 这段代码可以绘制出当前图像以肉眼观测
            # tmp = sample[0].reshape([28, 28])
            # print(label[0])
            # im = Image.fromarray(np.uint8(tmp))
            # plt.imshow(im)
            # plt.show()
            ####
            train_accuracy, summary = sess.run([accuracy, merged], feed_dict={x: sample, y_: label, keep_prob:1.0})

            print("step %d, training accuracy %g" % (i, train_accuracy))
            summary_writer.add_summary(summary, i)

        train_step.run(feed_dict={x: sample, y_: label, keep_prob:0.8})

    # 评估准确率
    print("start evaluate")
    sum_total = 0.0
    for i in range(VAL_COUNT):
        image, label = sess.run([val_image, val_label])
        sum_total += sess.run(accuracy, feed_dict={x: [image], y_: [label], keep_prob:1.0})

    print("eval precision:")
    print(sum_total / VAL_COUNT)

    # 保存当前神经网络参数模型
    save_path = saver.save(sess, "/tmp/imageModel", global_step=0)
    print("save path:")
    print(save_path)
    print("###############################")

    coord.request_stop()
    coord.join(threads)