import os
import tensorflow as tf
from tensorflow.examples.tutorials.mnist import input_data
mnist = input_data.read_data_sets(
'../datas/mnist', one_hot=True, reshape=False)
print(mnist.train.num_examples)
# 超参数设置
lr = 1e-2
epochs = 10
batch_size = 256
test_valid_size = 512 # 验证或者测试样本的数量
num_classes = 10 # 类别数量
keep_prob = 0.7 # dropout保留概率， 丢弃概率=1-0.7 = 0.3
"""
网络结构图
input [N, 28, 28, 1]
Conv1 weights=[5, 5, 1, 32] S=1 ---> [N, 28, 28, 32]
MaxPool1 kernel_size=[1, 2, 2, 1] S=2 ---> [N, 28/2, 28/2, 32]
Conv2 weights=[5, 5, 32, 64] S=1 ---> [N, 28/2, 28/2, 64]
MaxPool2 kernel_size=[1, 2, 2, 1] S=2 ---> [N, 28/4, 28/4, 64]
Flatten [N, 28/4, 28/4, 64] 4-D ---> [N, 7*7*64] 2-D
FC1 Weights=[7*7*64, 1024] ---> [N, 1024]
FC2 Weights=[1024, 10] ---> [N, 10]
"""
# 构建模型图 1、创建变量
graph = tf.Graph()
with graph.as_default():
weights = {
'conv1': tf.get_variable('w1', shape=[5, 5, 1, 32], initializer=tf.truncated_normal_initializer(stddev=0.1)),
'conv2': tf.get_variable('w2', shape=[5, 5, 32, 64], initializer=tf.truncated_normal_initializer(stddev=0.1)),
'fc1': tf.get_variable('w3', shape=[7*7*64, 1024], initializer=tf.truncated_normal_initializer(stddev=0.1)),
'fc2': tf.get_variable('w4', shape=[1024, num_classes], initializer=tf.truncated_normal_initializer(stddev=0.1))
}
biases = {
'conv1': tf.get_variable('b1', shape=[32], initializer=tf.zeros_initializer()),
'conv2': tf.get_variable('b2', shape=[64], initializer=tf.zeros_initializer()),
'fc1': tf.get_variable('b3', shape=[1024], initializer=tf.zeros_initializer()),
'fc2': tf.get_variable('b4', shape=[num_classes], initializer=tf.zeros_initializer())
}
def conv2d(input_tensor, filter_w, filter_b, strides=[1, 1, 1, 1]):
"""
实现 卷积 + 偏置项相加 + 激活函数
:param input_tensor: 输入
:param filter_w: 卷积核变量
:param filter_b: 偏置项
:param strides: 步幅
:return:
"""
# 1、卷积
conv = tf.nn.conv2d(
input=input_tensor, filter=filter_w, strides=strides, padding='SAME'
)
# 2、偏置项相加
conv = tf.nn.bias_add(conv, filter_b)
# 3、激活
conv = tf.nn.relu(conv)
return conv
def maxpool2d(input_tensor, k=2):
kernel_size = [1, k, k, 1]
strides = [1, k, k, 1]
maxpool_out = tf.nn.max_pool(
value=input_tensor, ksize=kernel_size, strides=strides, padding='SAME'
)
return maxpool_out
def fully_connect(input_tensor, weights, biases, activation=tf.nn.relu):
"""
实现全连接 + 偏置项相加 + 激活
:param input_tensor:
:param weights:
:param biases:
:param activation:
:return:
"""
fc = tf.matmul(input_tensor, weights) + biases
if activation:
fc = activation(fc)
return fc
else:
# 这里是为了返回最终输出的logits。
return fc
"""
网络结构图
input [N, 28, 28, 1]
Conv1 weights=[5, 5, 1, 32] S=1 ---> [N, 28, 28, 32]
MaxPool1 kernel_size=[1, 2, 2, 1] S=2 ---> [N, 28/2, 28/2, 32]
Conv2 weights=[5, 5, 32, 64] S=1 ---> [N, 28/2, 28/2, 64]
MaxPool2 kernel_size=[1, 2, 2, 1] S=2 ---> [N, 28/4, 28/4, 64]
Flatten [N, 28/4, 28/4, 64] 4-D ---> [N, 7*7*64] 2-D
FC1 Weights=[7*7*64, 1024] ---> [N, 1024]
FC2 Weights=[1024, 10] ---> [N, 10]
"""
def model(input_x, weights, biases, keep_prob):
"""
搭建CNN网络，返回logits
:param input_x: 模型输入，是占位符
:param weights:
:param biases:
:return:
"""
with tf.variable_scope('Network'):
# 卷积1 [N, 28, 28, 1] --> [N, 28, 28, 32]
conv1 = conv2d(
input_tensor=input_x, filter_w=weights['conv1'], filter_b=biases['conv1'])
# 池化1 [N, 28, 28, 32] -->[N, 28/2, 28/2, 32]
pool1 = maxpool2d(conv1)
# 卷积2 [N, 28/2, 28/2, 32] --> [N, 28/2, 28/2, 64]
conv2 = conv2d(
input_tensor=pool1, filter_w=weights['conv2'], filter_b=biases['conv2'])
# 池化2 [N, 28/2, 28/2, 64] -->[N, 28/4, 28/4, 64]
pool2 = maxpool2d(conv2)
# 拉平层 [N, 28/4, 28/4, 64] ---> [N, 7*7*64]
shape = pool2.get_shape() # [N, 7, 7, 64]
flatten_shape = shape[1] * shape[2] * shape[3]
flatted = tf.reshape(pool2, shape=[-1, flatten_shape])
# 全连接层1 [N, 7*7*64] ---> [N, 1024]
fc1 = fully_connect(
input_tensor=flatted, weights=weights['fc1'], biases=biases['fc1'])
fc1 = tf.nn.dropout(fc1, keep_prob=keep_prob)
# 全连接层2（输出层） [N, 1024] ---> [N, 10]
logits = fully_connect(
input_tensor=fc1, weights=weights['fc2'], biases=biases['fc2'], activation=None
)
return logits
def create_file_path(dir_path):
"""
创建文件夹函数
:param dir_path:
:return:
"""
if not os.path.exists(dir_path):
os.makedirs(dir_path)
print('创建文件夹:{}'.format(dir_path))
def train():
"""
构建模型
:return:
"""
with graph.as_default():
# 1、创建占位符
input_x = tf.placeholder(tf.float32, [None, 28, 28, 1], name='input_x')
input_y = tf.placeholder(tf.float32, [None, num_classes], name='input_y')
learning_rate = tf.placeholder(tf.float32, shape=None, name='learning_rate')
keep_probab = tf.placeholder(tf.float32, shape=None, name='keep_probab')
# 2、构建模型
logits = model(input_x, weights, biases, keep_prob=keep_probab)
# 3、构建模型损失
loss = tf.reduce_mean(tf.nn.softmax_cross_entropy_with_logits_v2(
logits=logits, labels=input_y
))
# 可视化代码
tf.summary.scalar('train_loss', loss, collections=['train'])
tf.summary.scalar('valid_loss', loss, collections=['valid'])
# 4、构建模型优化器
optimizer = tf.train.GradientDescentOptimizer(learning_rate=learning_rate)
train_opt = optimizer.minimize(loss=loss)
# 5、求模型准确率
correct_pred = tf.equal(tf.argmax(logits, 1), tf.argmax(input_y, 1))
accuracy = tf.reduce_mean(tf.cast(correct_pred, tf.float32))
# 可视化代码
tf.summary.scalar('train_accuracy', accuracy, collections=['train'])
tf.summary.scalar('valid_accuracy', accuracy, collections=['valid'])
# 6、构建持久化对象
CKECKPOINT_DIR = './model/ai13'
create_file_path(CKECKPOINT_DIR)
saver = tf.train.Saver(max_to_keep=1)
# 二、执行会话
with tf.Session(graph=graph) as sess:
sess.run(tf.global_variables_initializer())
# todo 模型可视化。
train_sum = tf.summary.merge_all('train')
valid_sum = tf.summary.merge_all('valid')
LOG_DIR = './model/graph'
train_writer = tf.summary.FileWriter(LOG_DIR+ '/train_graph', graph=sess.graph)
valid_writer = tf.summary.FileWriter(LOG_DIR+ '/valid_graph')
step = 1
for e in range(epochs):
# 构建批量数据的循环操作
for batch in range(mnist.train.num_examples // batch_size):
# 获取当前批量的数据(该数据已经归一化，且随机打乱的)
batch_x, batch_y = mnist.train.next_batch(batch_size)
feed = {input_x: batch_x, input_y:batch_y,
learning_rate:lr, keep_probab:keep_prob}
sess.run(train_opt, feed_dict=feed) # 执行模型训练
if step % 2 == 0:
feed = {input_x: batch_x, input_y: batch_y,
learning_rate: lr, keep_probab: 1.0}
train_loss, train_acc, train_sum_ = sess.run(
[loss, accuracy, train_sum], feed)
train_writer.add_summary(train_sum_, global_step=step)
val_feed = {input_x: mnist.validation.images[:test_valid_size],
input_y: mnist.validation.labels[:test_valid_size],
keep_probab: 1.0}
val_loss, val_acc, valid_sum_ = sess.run(
[loss, accuracy, valid_sum], val_feed)
valid_writer.add_summary(valid_sum_, global_step=step)
print('Epochs:{} - Step:{} - Train Loss:{:.5f} - Valid Loss:{:.5f} - Valid Acc:{:.5f}'.format(
e, step, train_loss, val_loss, val_acc
))
step += 1
# todo 模型持久化
if step % 50 == 0:
files = 'model.ckpt'
save_files = os.path.join(CKECKPOINT_DIR, files)
saver.save(sess, save_path=save_files, global_step=step)
print('模型成功持久化到文件夹:{}'.format(save_files))
# todo 测试数据集效果
test_feed = {input_x: mnist.test.images[:test_valid_size],
input_y: mnist.test.labels[:test_valid_size],
keep_probab: 1.0}
test_acc = sess.run(accuracy, test_feed)
print('Test Acc:{:.5f}'.format(test_acc))
def restore_train():
"""
恢复模型继续训练
:return:
"""
with graph.as_default():
# 1、创建占位符
input_x = tf.placeholder(tf.float32, [None, 28, 28, 1], name='input_x')
input_y = tf.placeholder(tf.float32, [None, num_classes], name='input_y')
learning_rate = tf.placeholder(tf.float32, shape=None, name='learning_rate')
keep_probab = tf.placeholder(tf.float32, shape=None, name='keep_probab')
# 2、构建模型
logits = model(input_x, weights, biases, keep_prob=keep_probab)
# 3、构建模型损失
loss = tf.reduce_mean(tf.nn.softmax_cross_entropy_with_logits_v2(
logits=logits, labels=input_y
))
# 4、构建模型优化器
optimizer = tf.train.GradientDescentOptimizer(learning_rate=learning_rate)
train_opt = optimizer.minimize(loss=loss)
# 5、求模型准确率
correct_pred = tf.equal(tf.argmax(logits, 1), tf.argmax(input_y, 1))
accuracy = tf.reduce_mean(tf.cast(correct_pred, tf.float32))
# 6、构建持久化对象
CKECKPOINT_DIR = './model/ai13'
saver = tf.train.Saver(max_to_keep=1)
# 二、执行会话
with tf.Session(graph=graph) as sess:
# fixme 恢复模型继续训练.
# 获取持久化对象的信息。
ckpt = tf.train.get_checkpoint_state(CKECKPOINT_DIR)
if ckpt is not None:
saver.restore(sess, ckpt.model_checkpoint_path)
print('成功恢复模型继续训练!')
else:
sess.run(tf.global_variables_initializer())
print('没有持久化的模型，从头随机初始化开始训练!')
step = 1
for e in range(epochs):
# 构建批量数据的循环操作
for batch in range(mnist.train.num_examples // batch_size):
# 获取当前批量的数据(该数据已经归一化，且随机打乱的)
batch_x, batch_y = mnist.train.next_batch(batch_size)
feed = {input_x: batch_x, input_y:batch_y,
learning_rate:lr, keep_probab:keep_prob}
sess.run(train_opt, feed_dict=feed) # 执行模型训练
if step % 2 == 0:
feed = {input_x: batch_x, input_y: batch_y,
learning_rate: lr, keep_probab: 1.0}
train_loss, train_acc = sess.run(
[loss, accuracy], feed)
val_feed = {input_x: mnist.validation.images[:test_valid_size],
input_y: mnist.validation.labels[:test_valid_size],
keep_probab: 1.0}
val_loss, val_acc = sess.run(
[loss, accuracy], val_feed)
print('Epochs:{} - Step:{} - Train Loss:{:.5f} - Valid Loss:{:.5f} - Valid Acc:{:.5f}'.format(
e, step, train_loss, val_loss, val_acc
))
step += 1
# todo 模型持久化
if step % 50 == 0:
files = 'model.ckpt'
save_files = os.path.join(CKECKPOINT_DIR, files)
saver.save(sess, save_path=save_files, global_step=step)
print('模型成功持久化到文件夹:{}'.format(save_files))
# todo 测试数据集效果
test_feed = {input_x: mnist.test.images[:test_valid_size],
input_y: mnist.test.labels[:test_valid_size],
keep_probab: 1.0}
test_acc = sess.run(accuracy, test_feed)
print('Test Acc:{:.5f}'.format(test_acc))
if __name__ == '__main__':
# train()
restore_train()
|
