TensorFlow高阶API Estimator自定义模型解决

在之前的文章中，我们利用silm工具和谷歌训练好的inception-v3模型完成了一个花朵图像分类问题，但代码还是比较繁琐。为了更精简的代码和提高可读性，这一次我们利用TensorFlow提供的高阶API Estimator来解决同样的问题。同时，在最后，我们会把训练过程中的参数变化通过TensorBoard展示出来。

Estimator

Estimator是TensorFlow官方提供的一个高层API，它更好的整合了原生态TensorFlow提供的功能。它可以极大简化机器学习编程。下面来看一下TensorFlow API结构：

API Architecture

在官方文档中，有这么一句话：

We strongly recommend writing TensorFlow programs with the following APIs:

• Estimators, which represent a complete model. The Estimator API provides methods to train the model, to judge the model's accuracy, and to generate predictions.
• Datasets for Estimators, which build a data input pipeline. The Dataset API has methods to load and manipulate data, and feed it into your model. The Dataset API meshes well with the Estimators API.

可以看到Estimator和Dataset这两个API是官方强烈推荐的。Estimator提供了预创建的DNN模型，使用起来非常方便。具体怎么使用Estimator预创建模型，官方文档里面也有写，有兴趣的可以去看Estimator官方。
但是预先定义的Estimator功能有限，比如目前无法很好的实现卷积神经网络和循环神经网络，也没有办法支持自定义的损失函数，所以为了更好的使用Estimator，这篇文章会教大家怎么用Estimator自定义CNN模型，以及如何配合Dataset读取图片数据。

数据准备

在这里我们可以使用之前的谷歌提供的花朵分类数据集，也可以使用其它的。为了区分上次结果这次我们使用新的数据集。在这里我使用百度挑桃分类数据集。下载解压后可以看到是这样的目录： 数据集

数据集已经帮我们划分好了是训练还是测试。每一个文件夹代表一种桃子，总共有4种桃子（这个数据集肉眼很难辨别，可能是因为我不够专业-_-）。

数据预处理

我们还是像之前一样对数据预处理。在工程目录下新建select_peach_data.py文件。跟之前处理花朵分类的时候一样所以这里直接粘贴代码：

import glob
import os.path
import numpy as np
import tensorflow as tf

from tensorflow.python.platform import gfile

#输入图片地址
INPUT_ALL_DATA = './select_peach'
INPUT_TRAIN_DATA = './select_peach/train'
INPUT_TEST_DATA = './select_peach/test'
OUTPUT_TRAIN_FILE = './path/to/output_train.tfrecords'
OUTPUT_TEST_FILE = './path/to/output_test.tfrecords'

def _int64_feature(value):
    return tf.train.Feature(int64_list=tf.train.Int64List(value=[value]))

#生成字符串的属性
def _bytes_feature(value):
    return tf.train.Feature(bytes_list=tf.train.BytesList(value=[value]))

#检索目录并提取目录图片文件生成TFRecords
def get_img_data(sub_dirs,writer,INPUT_DATA,sess):
    current_label = 0
    is_root_dir = True
    print("文件地址: "+INPUT_DATA)
    for sub_dir in sub_dirs:
        if is_root_dir:
            is_root_dir = False
            continue
        file_list = []
        dir_name = os.path.basename(sub_dir)

        file_glob = os.path.join(INPUT_DATA, dir_name, '*.' + "png")
        # extend合并两个数组
        # glob模块的主要方法就是glob,该方法返回所有匹配的文件路径列表（list）
        # 比如：glob.glob(r’c:*.txt’) 这里就是获得C盘下的所有txt文件
        file_list.extend(glob.glob(file_glob))
        if not file_list: continue
        # print('file_list',current_label)
        # 处理图片数据
        index = 0
        for file_name in file_list:
            # 读取并解析图片 讲图片转化成299*299方便模型处理
            image_raw_data = gfile.FastGFile(file_name, 'rb').read()
            image = tf.image.decode_png(image_raw_data)
            if image.dtype != tf.float32:
                image = tf.image.convert_image_dtype(image, dtype=tf.float32)
            image = tf.image.resize_images(image, [299, 299])
            image_value = sess.run(image)
            pixels = image_value.shape[1]
            image_raw = image_value.tostring()
            # 存到features
            example = tf.train.Example(features=tf.train.Features(feature={
                'pixels': _int64_feature(pixels),
                'label': _int64_feature(current_label),
                'image_raw': _bytes_feature(image_raw)
            }))
            chance = np.random.randint(100)
            # 写入训练集
            writer.write(example.SerializeToString())
            index = index + 1
            if index == 400:
                break
            print("处理文件索引%d index%d"%(current_label,index))
        current_label += 1
#读取数据并将数据分割成训练数据、验证数据和测试数据
def create_image_lists(sess):

    #首先处理训练数据集
    sub_dirs = [x[0] for x in os.walk(INPUT_TRAIN_DATA)]
    writer_train = tf.python_io.TFRecordWriter(OUTPUT_TRAIN_FILE)
    get_img_data(sub_dirs,writer_train,INPUT_TRAIN_DATA,sess)

    sub_test_dirs = [x[0] for x in os.walk(INPUT_TEST_DATA)]
    writer_test = tf.python_io.TFRecordWriter(OUTPUT_TEST_FILE)
    get_img_data(sub_test_dirs,writer_test,INPUT_TEST_DATA,sess)

    writer_train.close()
    writer_test.close()

def main():
    with tf.Session() as sess:
        create_image_lists(sess)
        print('success')

if __name__ == '__main__':
    main()

这里因为test和train已经在文件夹上作了区分，所以这里我利用两个TFRecordWriter来把数据分别写入两个TFRecord。为了节省时间在这里我并没有利用全部的训练数据，只是加载了其中的400份。当然在真实的训练场景下你是需要加载全部的数据的。
代码没有详尽的注释，因为和之前的处理大部分都是一样的，不清楚的可以去看我之前的文章。inception-v3。

自定义Estimator

下面我们开始步入主题。先看一张Estimator类组成图。

Estimator
以下源自官方文档的一段话：
Pre-made Estimators are fully baked. Sometimes though, you need more control over an Estimator's behavior. That's where custom Estimators come in. You can create a custom Estimator to do just about anything. If you want hidden layers connected in some unusual fashion, write a custom Estimator. If you want to calculate a unique metric for your model, write a custom Estimator. Basically, if you want an Estimator optimized for your specific problem, write a custom Estimator.

A model function (or model_fn) implements the ML algorithm. The only difference between working with pre-made Estimators and custom Estimators is:

• With pre-made Estimators, someone already wrote the model function for you.
• With custom Estimators, you must write the model function.

Your model function could implement a wide range of algorithms, defining all sorts of hidden layers and metrics. Like input functions, all model functions must accept a standard group of input parameters and return a standard group of output values. Just as input functions can leverage the Dataset API, model functions can leverage the Layers API and the Metrics API.

大概意思是：预创建的 Estimator 是 tf.estimator.Estimator 基类的子类，而自定义 Estimator 是 tf.estimator.Estimator 的实例。
Pre-made Estimators和custom Estimators差异主要在于tensorflow中是否有它们可以直接使用的模型函数(model function or model_fn)的实现。对于前者，tensorflow中已经有写好的model function，因而直接调用即可；而后者的model function需要自己编写。因此，Pre-made Estimators使用方便，但使用范围小，灵活性差；custom Estimators则正好相反。

总体来说，模型是由三部分构成：Input functions、Model functions 和Estimators（评估控制器，main function)。

• Input functions：主要是由Dataset API组成，可以分为train_input_fn和eval_input_fn。前者的任务（行为）是接受参数，输出数据训练数据，后者的任务（行为）是接受参数，并输出验证数据和测试数据。
• Model functions：是由模型（the Layers API ）和监控模块（ the Metrics API）组成，主要是实现模型的训练、测试（验证）和监控显示模型参数状况的功能。
• Estimators：在模型中的作用类似于计算机中的操作系统。它将各个部分“粘合”起来，控制数据在模型中的流动与变换，同时控制模型的的各种行为（运算）。

在得知以上知识以后，我们可以开始动手编码起来。通过以上内容得知，首先我们需要先创建自定义的Model functions。下面新建my_estimator文件。
由于我们这里是实现自定义的model_fn函数，而model_fn主要功能是定义模型的结构，损失函数以及优化器。还会对预测和评测进行处理。综上我们来完成model_fn的编写。

自定义model_fn

#导入相关库
import numpy as np
import tensorflow as tf
import tensorflow.contrib.slim as slim
# 加载通过TensorFlow-Silm定义好的 inception_v3模型
import tensorflow.contrib.slim.python.slim.nets.inception_v3 as inception_v3

#图片数据地址
TRAIN_DATA = './path/to/output_train.tfrecords'
TEST_DATA = './path/to/output_test.tfrecords'

shuffle_buffer = 10000
BATCH = 64
#打开 estimator 日志
tf.logging.set_verbosity(tf.logging.INFO)

#自定义模型
#这里我们提供了两种方案。一种是直接通过slim工具定义已有模型
#另一种是通过tf.layer更加灵活地定义神经网络结构
def inception_v3_model(image,is_training):
    with slim.arg_scope(inception_v3.inception_v3_arg_scope()):
        predictions,_ = inception_v3.inception_v3(image,num_classes=5)
        return predictions
#定义lenet5模型
def lenet5(x,is_training):
    net = tf.layers.conv2d(x,32,5,activation=tf.nn.relu)
    net = tf.layers.max_pooling2d(net,2,2)
    net = tf.layers.conv2d(net,64,3,activation=tf.nn.relu)
    net = tf.layers.max_pooling2d(net,2,2)
    net = tf.contrib.layers.flatten(net)
    net = tf.layers.dense(net,1024)
    net = tf.layers.dropout(net,rate=0.4,training=is_training)
    return tf.layers.dense(net,5)
#自定义Estimator中使用的模型。定义的函数有4个收入，
#features给出在输入函数中会提供的输入层张量。这是个字典
#字典通过input_fn提供。如果是系统的输入
#系统会提供tf.estimator.inputs.numpy_input_fn中的x参数指定内容
#labels是正确答案，通过numpy_input_fn的y参数给出
#在这里我们用dataset来自定义输入函数。
#mode取值有3种可能，分别对应Estimator的train,evaluate,predict这三个函数
#mode参数可以判断当前是训练，预测还是验证模式。
#最有一个参数param也是字典，里面是有关于这个模型的相关任何超参数（学习率）
def model_fn(features,labels,mode,params):
    predict = lenet5(features,mode == tf.estimator.ModeKeys.TRAIN)
    #如果是预测模式，直接返回结果
    if mode == tf.estimator.ModeKeys.PREDICT:
        return tf.estimator.EstimatorSpec(
            mode=mode,
            predictions={"result":tf.argmax(predict,1)}
        )
  #定义损失函数，这里使用tf.losses可以直接从tf.losses.get_total_loss()拿到损失
    tf.losses.softmax_cross_entropy(tf.one_hot(labels, 5), predict, weights=1.0)

    #优化器
    optimizer = tf.train.GradientDescentOptimizer(learning_rate=params["learning_rate"])
    #定义训练过程。传入global_step的目的，为了在TensorBoard中显示图像的横坐标
    train_op = optimizer.minimize(
        loss=tf.losses.get_total_loss(),
        global_step=tf.train.get_global_step()
    )

    #定义评测标准
    #这个函数会在调用Estimator.evaluate的时候调用
    accuracy = tf.metrics.accuracy(
            predictions=tf.argmax(predict,1),
            labels=labels,
            name="acc_op"
    )
    eval_metric_ops = {
        "my_metric":accuracy
    }
    #用于向TensorBoard输出准确率图像
    #如果你不需要使用TensorBoard可以不添加这行代码
    tf.summary.scalar('accuracy', accuracy[1])
    #model_fn会返回一个EstimatorSpec
    #EstimatorSpec必须包含模型损失，训练函数。其它为可选项
    #eval_metric_ops用于定义调用Estimator.evaluate()时候所指定的函数
    return tf.estimator.EstimatorSpec(
        mode=mode,
        loss=tf.losses.get_total_loss(),
        train_op=train_op,
        eval_metric_ops=eval_metric_ops
    )

自定义Input functions

定义完了model functions接下来我们通过Dataset API来定义input functions：

#解析tfrecords
def parse(record):
    features = tf.parse_single_example(
        record,
        features={
            'image_raw': tf.FixedLenFeature([], tf.string),
            'label': tf.FixedLenFeature([], tf.int64),
            'pixels': tf.FixedLenFeature([], tf.int64)
        }
    )
    decoded_image = tf.decode_raw(features['image_raw'], tf.float16)
    label = features['label']
    return decoded_image, label
#从dataset中读取训练数据，这里和之前处理花朵分类的时候一样
def my_input_fn(file):
    dataset = tf.data.TFRecordDataset([file])
    dataset = dataset.map(parse)
    dataset = dataset.shuffle(shuffle_buffer).batch(BATCH)
    dataset = dataset.repeat(10)
    iterator = dataset.make_one_shot_iterator()
    batch_img,batch_labels = iterator.get_next()
    with tf.Session() as sess:
        batch_sess_img,batch_sess_labels = sess.run([batch_img,batch_labels])
        #这里需要特别注意 由于batch_sess_img这里是转成了string后在原有长度上增加了8倍
        #所以在这里我们要先转成numpy然后再reshape要不然会报错
        batch_sess_img = np.fromstring(batch_sess_img, dtype=np.float32)
        #numpy转换成Tensor
        batch_sess_img = tf.reshape(batch_sess_img, [BATCH, 299, 299, 3])
    return batch_sess_img,batch_sess_labels

在这里要注意，Estimator输入函数要求每次被调用可以得到一个batch的数据，包括所有的输入层数据和正确答案标注。而且my_input_fn函数并不能带有参数。稍后我们会用lambda表达式解决这个问题。

最后我们通过main函数来启动训练过程：

def main():
    #定义超参数
    model_params = {"learning_rate":0.001}
    #定义训练的相关配置参数
    #keep_checkpoint_max=1表示在只在目录下保存一份模型文件
    #log_step_count_steps=50表示每训练50次输出一次损失的值
    run_config = tf.estimator.RunConfig(keep_checkpoint_max=1,log_step_count_steps=50)
    #通过tf.estimator.Estimator来生成自定义模型
    #把我们自定义的model_fn和超参数传进去
    #这里我们还传入了持久化模型的目录
    #estimator会自动帮我们把模型持久化到这个目录下
    estimator = tf.estimator.Estimator(model_fn=model_fn,params=model_params,model_dir="./path/model",config=run_config)
    #开始训练模型，这里说一下lambda表达式
    #lambda表达式会把函数原本的输入参数变成0个或它指定的参数。可以理解为函数的默认值
    #这里传入自定义输入函数，和训练的轮数
    estimator.train(input_fn=lambda :my_input_fn(TRAIN_DATA),steps=300)
    #训练完后进行验证，这里传入我们的测试数据
    test_result = estimator.evaluate(input_fn=lambda :my_input_fn(TEST_DATA))
    #输出测试验证结果
    accuracy_score = test_result["my_metric"]
    print("\nTest accuracy:%g %%"%(accuracy_score*100))

if __name__ == '__main__':
    main()

运行程序，可以看到如下输出。因为我这里是从367步以后继续训练，所以我们在日志中看到我这里是直接加载了第367步保存的模型。
每隔一定时间，Estimator会自动创建模型文件。另外如果训练中断，下一次再启动训练的话，Estimator会自动从模型目录下加载最新的模型并且用于训练，非常方便。这就是为什么谷歌推荐我们用Estimator来训练模型，因为它封装了很多开发者并不需要关心的操作，大大提升了我们的开发效率。

INFO:tensorflow:Done calling model_fn.
INFO:tensorflow:Create CheckpointSaverHook.
INFO:tensorflow:Graph was finalized.
INFO:tensorflow:Restoring parameters from ./path/model/model.ckpt-367
INFO:tensorflow:Running local_init_op.
INFO:tensorflow:Done running local_init_op.
INFO:tensorflow:Saving checkpoints for 368 into ./path/model/model.ckpt.
INFO:tensorflow:loss = 0.2994086, step = 368
INFO:tensorflow:global_step/sec: 0.116191
INFO:tensorflow:loss = 0.2086069, step = 418 (430.326 sec)
INFO:tensorflow:Saving checkpoints for 438 into ./path/model/model.ckpt.
INFO:tensorflow:global_step/sec: 0.115405
INFO:tensorflow:loss = 0.17857286, step = 468 (433.259 sec)
INFO:tensorflow:Saving checkpoints for 506 into ./path/model/model.ckpt.
INFO:tensorflow:global_step/sec: 0.111342
INFO:tensorflow:loss = 0.107850984, step = 518 (449.065 sec)
INFO:tensorflow:global_step/sec: 0.115999
INFO:tensorflow:loss = 0.08592671, step = 568 (431.040 sec)
INFO:tensorflow:Saving checkpoints for 575 into ./path/model/model.ckpt.
INFO:tensorflow:global_step/sec: 0.112465
INFO:tensorflow:loss = 0.05861471, step = 618 (444.587 sec)
INFO:tensorflow:Saving checkpoints for 643 into ./path/model/model.ckpt.

TensorBoard

为了更加直观的看到训练过程，接下来我们将使用谷歌提供的一个工具TensorBoard来可视化我们的训练过程。
要启动TensorBoard，执行下面的命令：

#PATH替换为你模型保存的目录。要注意在这里用的是绝对路径。
tensorboard --logdir=PATH

执行命令后可以看到如下信息，说明TensorBoard已经跑起来了。

TensorBoard 1.8.0 at http://bogon:6006 (Press CTRL+C to quit)
W0817 16:14:27.129659 Reloader tf_logging.py:121] Found more than one graph event per run, or there was a metagraph containing a graph_def, as well as one or more graph events.  Overwriting the graph with the newest event.
W0817 16:14:27.650306 Reloader tf_lo

所有预创建的 Estimator 都会自动将大量信息记录到 TensorBoard 上。不过，对于自定义 Estimator，TensorBoard 只提供一个默认日志（损失图）以及您明确告知 TensorBoard 要记录的信息。对于我们刚刚创建的自定义 Estimator，并且明确说明要绘制正确率的图，所以TensorBoard 会生成以下内容：

TensorBoard.png

TensorBoard生成了三个图。分别表示正确率，训练处理的批次，训练轮数所对应的损失值

简而言之，下面是三张图显示的内容：

• global_step/sec：这是一个性能指标，显示我们在进行模型训练时每秒处理的批次数（梯度更新）。
• loss：所报告的损失。
• accuracy：准确率由下列两行记录：
  • eval_metric_ops={'my_accuracy': accuracy}（评估期间）。
  • tf.summary.scalar('accuracy', accuracy[1])（训练期间）。
    这些 Tensorboard 图是务必要将 global_step 传递给优化器的 minimize 方法的主要原因之一。如果没有它，模型就无法记录这些图的 x 坐标。

我们来看下TensorBoard的输出。可以看到随着训练步骤的增加，loss在相应的减少，accuracy也在慢慢增加。这是一个健康的训练过程。可以看到LeNet5在这个数据集上的正确率达到了95%左右。

eval

因为我自定义的Estimator在训练结束之后并没有输出正确率（暂时没找到原因），所以这里我们另外写一个程序来测试这个模型的正确率。这里我们命名为eval.py。

import tensorflow as tf
import Estimator1
import numpy as np

TEST_DATA = './path/to/output_test.tfrecords'
CKPT_PATH = './path/model'
EVAL_BATCH = 20
def getValidationData():
   dataset = tf.data.TFRecordDataset([TEST_DATA])
   dataset = dataset.map(Estimator1.parse)
   dataset = dataset.batch(EVAL_BATCH)
   iterator = dataset.make_one_shot_iterator()
   batch_img, batch_labels = iterator.get_next()

   # batch_img作处理
   return batch_img, batch_labels
def my_eval():
   #estimator的eval方法不好使 用传统方法试试
   batch_img,batch_labels = getValidationData()

   x = tf.placeholder(tf.float32, [None, 299,299,3], name='x-input')
   y_ = tf.placeholder(tf.int64, [None], name='y-input')
   y = Estimator1.lenet5(x, False)
   correct_prediction = tf.equal(tf.argmax(y, 1), y_)
   accuracy = tf.reduce_mean(tf.cast(correct_prediction, tf.float32))
   saver = tf.train.Saver()
   with tf.Session() as sess:
       while True:
           try:
               ckpt = tf.train.get_checkpoint_state(CKPT_PATH)
               if ckpt and ckpt.model_checkpoint_path:
                   saver.restore(sess,ckpt.model_checkpoint_path)
                   #通过文件名得到模型保存时迭代的轮数
                   global_step = ckpt.model_checkpoint_path.split('/')[-1].split('-')[-1]
                   batch_sess_img, batch_sess_labels = sess.run([batch_img, batch_labels])
                   batch_sess_img = np.fromstring(batch_sess_img, dtype=np.float32)
                   batch_sess_img = tf.reshape(batch_sess_img, [EVAL_BATCH, 299, 299, 3])
                   batch_sess_img = sess.run(batch_sess_img)
                   print(sess.run([tf.argmax(y,1),y_],feed_dict={x:batch_sess_img,y_:batch_sess_labels}))
                   accuracy_score = sess.run(accuracy,feed_dict={x:batch_sess_img,y_:batch_sess_labels})
                   print("After %s training step(s),validation accuracy = %g"%(global_step,accuracy_score))
               else:
                   print('No checkpoint file found')
                   return
           except tf.errors.OutOfRangeError:
               break
def main():
   my_eval()

if __name__ == '__main__':
   main()

这个程序大概的作用是：
1.读取测试数据，把测试数据打包成batch。然后定义神经网络输入变量x和正确答案的标签y_。
2.把x通过神经网络得到的前向传播结果y和y_作比较来计算正确率。
3.读取之前训练好的模型。
4.用一个while循环来输出在训练好的模型上每一个batch的正确率，直到数据读取完毕。
运行这个程序可以得到以下输出：

INFO:tensorflow:Restoring parameters from ./path/model/model.ckpt-643
[array([0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 1, 1, 1, 1]), array([0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 1, 1, 1, 1])]
After 643 training step(s),validation accuracy = 1
INFO:tensorflow:Restoring parameters from ./path/model/model.ckpt-643
[array([1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 2, 2, 2, 2, 2, 2, 2]), array([1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 2, 2, 2, 2, 2, 2, 2])]
After 643 training step(s),validation accuracy = 1
INFO:tensorflow:Restoring parameters from ./path/model/model.ckpt-643
[array([2, 2, 2, 2, 2, 2, 2, 3, 2, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3]), array([2, 2, 2, 2, 2, 2, 2, 2, 2, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3])]
After 643 training step(s),validation accuracy = 0.95
INFO:tensorflow:Restoring parameters from ./path/model/model.ckpt-643

嗯。60个数据中只有1个判断错误，也符合我们之前得到的正确率。

写在最后

Estimator是TensorFlow官方强烈推荐的API，通过上述程序大家也能看到相比传统的TensorFlow API，Estimator封装了大部分与业务逻辑无关的操作，然而通过Custom Estimator，Estimator也不失灵活性。

我们之前还通过slim定义了一个inception-v3模型，但是由于inception-v3结构比较复杂，训练的时间比较久所以这里我们就以LeNet-5作演示了。但是在复杂的图像分类问题上，比如ImageNet数据集中，LeNet-5的分类效果就不是很好。如果是复杂的图像分类问题，就要选择更加复杂的神经网络模型来训练才能达到较高的准确率。

另外这篇文章主要是以使用Estimator为主，对于其中的一些细节没有很好的阐述。之后的文章会对一些技术细节做探究。

欢迎广大喜欢AI的开发者互相交流，有问题也可以在评论区里留言，大家互相讨论，一起进步。