Tensorflow一些常用基本概念与函数（4）

摘要：本系列主要对tf的一些常用概念与方法进行描述。本文主要针对tensorflow的模型训练Training与测试Testing等相关函数进行讲解。为‘Tensorflow一些常用基本概念与函数’系列之四。

1、序言

本文所讲的内容主要为以下列表中相关函数。函数training()通过梯度下降法为最小化损失函数增加了相关的优化操作，在训练过程中，先实例化一个优化函数，比如 tf.train.GradientDescentOptimizer，并基于一定的学习率进行梯度优化训练：

optimizer = tf.train.GradientDescentOptimizer(learning_rate)

然后，可以设置 一个用于记录全局训练步骤的单值。以及使用minimize()操作，该操作不仅可以优化更新训练的模型参数，也可以为全局步骤(global step)计数。与其他tensorflow操作类似，这些训练操作都需要在tf.session会话中进行

global_step = tf.Variable(0, name='global_step', trainable=False)
train_op = optimizer.minimize(loss, global_step=global_step)

2、Tensorflow函数

2.1 训练 (Training)

一个TFRecords 文件为一个字符串序列。这种格式并非随机获取，它比较适合大规模的数据流，而不太适合需要快速分区或其他非序列获取方式。

█ 优化 (Optimizers)

tf中各种优化类提供了为损失函数计算梯度的方法，其中包含比较经典的优化算法，比如GradientDescent 和Adagrad。

class tf.train.Optimizer

▷ class tf.train.Optimizer

用法

# Create an optimizer with the desired parameters.
opt = GradientDescentOptimizer(learning_rate=0.1)
# Add Ops to the graph to minimize a cost by updating a list of variables.
# "cost" is a Tensor, and the list of variables contains tf.Variable objects.
opt_op = opt.minimize(cost, var_list=<list of variables>)
# Execute opt_op to do one step of training:
opt_op.run()

在使用它们之前处理梯度

使用minimize()操作，该操作不仅可以计算出梯度，而且还可以将梯度作用在变量上。如果想在使用它们之前处理梯度，可以按照以下三步骤使用optimizer

1、使用函数compute_gradients()计算梯度


2、按照自己的愿望处理梯度


3、使用函数apply_gradients()应用处理过后的梯度

例如：

# 创建一个optimizer.
opt = GradientDescentOptimizer(learning_rate=0.1)

# 计算<list of variables>相关的梯度
grads_and_vars = opt.compute_gradients(loss, <list of variables>)

# grads_and_vars为tuples (gradient, variable)组成的列表。
#对梯度进行想要的处理，比如cap处理
capped_grads_and_vars = [(MyCapper(gv[0]), gv[1]) for gv in grads_and_vars]

# 令optimizer运用capped的梯度(gradients)
opt.apply_gradients(capped_grads_and_vars)

选通梯度(Gating Gradients)

函数minimize() 与compute_gradients()都含有一个参数gate_gradient，用于控制在应用这些梯度时并行化的程度。

其值可以取：GATE_NONE, GATE_OP 或 GATE_GRAPH

GATE_NONE : 并行地计算和应用梯度。提供最大化的并行执行，但是会导致有的数据结果没有再现性。比如两个matmul操作的梯度依赖输入值，使用GATE_NONE可能会出现有一个梯度在其他梯度之前便应用到某个输入中，导致出现不可再现的(non-reproducible)结果

GATE_OP: 对于每个操作Op，确保每一个梯度在使用之前都已经计算完成。这种做法防止了那些具有多个输入，并且梯度计算依赖输入情形中，多输入Ops之间的竞争情况出现。
GATE_GRAPH: 确保所有的变量对应的所有梯度在他们任何一个被使用前计算完成。该方式具有最低级别的并行化程度，但是对于想要在应用它们任何一个之前处理完所有的梯度计算时很有帮助的。

█ Slots

一些optimizer的之类，比如 MomentumOptimizer 和 AdagradOptimizer 分配和管理着额外的用于训练的变量。这些变量称之为’Slots’，Slots有相应的名称，可以向optimizer访问的slots名称。有助于在log debug一个训练算法以及报告slots状态

▷ tf.train.AdamOptimizer
Adam 的基本运行方式，首先初始化：

m_0 <- 0 (Initialize initial 1st moment vector)
v_0 <- 0 (Initialize initial 2nd moment vector)
t <- 0 (Initialize timestep)

在论文中的 section2 的末尾所描述了更新规则，该规则使用梯度g来更新变量：

t <- t + 1
lr_t <- learning_rate * sqrt(1 - beta2^t) / (1 - beta1^t)

m_t <- beta1 * m_{t-1} + (1 - beta1) * g
v_t <- beta2 * v_{t-1} + (1 - beta2) * g * g
variable <- variable - lr_t * m_t / (sqrt(v_t) + epsilon)

其中epsilon 的默认值1e-8可能对于大多数情况都不是一个合适的值。例如，当在ImageNet上训练一个 Inception network时比较好的选择为1.0或者0.1。
需要注意的是，在稠密数据中即便g为0时， m_t, v_t 以及variable都将会更新。而在稀疏数据中，m_t, v_t 以及variable不被更新且值为零。

█ 梯度计算与截断(Gradient Computation and Clipping)

TensorFlow 提供了计算给定tf计算图的求导函数，并在图的基础上增加节点。优化器(optimizer )类可以自动的计算网络图的导数，但是优化器中的创建器(creators )或者专业的人员可以通过本节所述的函数调用更底层的方法。

█ 退化学习率(Decaying the learning rate)

▷ tf.train.exponential_decay

#该函数返回以下结果
decayed_learning_rate = learning_rate *
         decay_rate ^ (global_step / decay_steps)
##例： 以0.96为基数，每100000 步进行一次学习率的衰退
global_step = tf.Variable(0, trainable=False)
starter_learning_rate = 0.1
learning_rate = tf.train.exponential_decay(starter_learning_rate, global_step,
                                           100000, 0.96, staircase=True)
# Passing global_step to minimize() will increment it at each step.
learning_step = (
    tf.train.GradientDescentOptimizer(learning_rate)
    .minimize(...my loss..., global_step=global_step)
)

█ 移动平均(Moving Averages)

一些训练优化算法，比如GradientDescent 和Momentum 在优化过程中便可以使用到移动平均方法。使用移动平均常常可以较明显地改善结果。

▷ tf.train.ExponentialMovingAverage

# Example usage when creating a training model:
# Create variables.
var0 = tf.Variable(...)
var1 = tf.Variable(...)
# ... use the variables to build a training model...
...
# Create an op that applies the optimizer.  This is what we usually
# would use as a training op.
opt_op = opt.minimize(my_loss, [var0, var1])

# Create an ExponentialMovingAverage object
ema = tf.train.ExponentialMovingAverage(decay=0.9999)

# Create the shadow variables, and add ops to maintain moving averages
# of var0 and var1.
maintain_averages_op = ema.apply([var0, var1])

# Create an op that will update the moving averages after each training
# step.  This is what we will use in place of the usual training op.
with tf.control_dependencies([opt_op]):
    training_op = tf.group(maintain_averages_op)

...train the model by running training_op...

#Example of restoring the shadow variable values:
# Create a Saver that loads variables from their saved shadow values.
shadow_var0_name = ema.average_name(var0)
shadow_var1_name = ema.average_name(var1)
saver = tf.train.Saver({shadow_var0_name: var0, shadow_var1_name: var1})
saver.restore(...checkpoint filename...)
# var0 and var1 now hold the moving average values

▷ tf.train.ExponentialMovingAverage.variables_to_restore

  variables_to_restore = ema.variables_to_restore()
  saver = tf.train.Saver(variables_to_restore)

█ 协调器和队列运行器(Coordinator and QueueRunner)

查看queue中，queue相关的内容，了解tensorflow中队列的运行方式。

▷ class tf.train.Coordinator

#Coordinator的使用，用于多线程的协调
try:
  ...
  coord = Coordinator()
  # Start a number of threads, passing the coordinator to each of them.
  ...start thread 1...(coord, ...)
  ...start thread N...(coord, ...)
  # Wait for all the threads to terminate, give them 10s grace period
  coord.join(threads, stop_grace_period_secs=10)
except RuntimeException:
  ...one of the threads took more than 10s to stop after request_stop()
  ...was called.
except Exception:
  ...exception that was passed to coord.request_stop()

▷ tf.train.Coordinator.stop_on_exception()

with coord.stop_on_exception():
  # Any exception raised in the body of the with
  # clause is reported to the coordinator before terminating
  # the execution of the body.
  ...body...
#等价于
try:
  ...body...
exception Exception as ex:
  coord.request_stop(ex)

█ 布执行(Distributed execution)

可以阅读TensorFlow的分布式学习框架简介 查看更多tensorflow分布式细节。

▷ tf.train.Server

server = tf.train.Server(...)
with tf.Session(server.target):
  # ...

▷ tf.train.Supervisor

相关参数：

ready_op : 一维 字符串 tensor。该tensor是用过监视器在prepare_or_wait_for_session()计算，检查model是否准备好可以使用。如果准备好，将返回一个空阵列，如果为None，该model没有被检查。

is_chief : 如果为True，创建一个主监视器用于负责初始化与模型的恢复，若为False，则依赖主监视器。

init_op : 一个操作，用于模型不能恢复时的初始化操作。默认初始化所有操作

local_init_op : 可被所有监视器运行的初始化操作。

logdir : 设置log目录

summary_op : 一个操作(Operation )，返回Summary 和事件logs，需要设置 logdir

saver : 一个Saver对象

save_summaries_secs : 保存summaries的间隔秒数

save_model_secs : 保存model的间隔秒数

checkpoint_basename : checkpoint保存的基本名称

• 使用在单进程中

with tf.Graph().as_default():
  ...add operations to the graph...
  # Create a Supervisor that will checkpoint the model in '/tmp/mydir'.
  sv = Supervisor(logdir='/tmp/mydir')
  # Get a TensorFlow session managed by the supervisor.
  with sv.managed_session(FLAGS.master) as sess:
    # Use the session to train the graph.
    while not sv.should_stop():
      sess.run(<my_train_op>)
# 在上下文管理器with sv.managed_session()内，所有在graph的变量都被初始化。
# 或者说，一些服务器checkpoint相应模型并增加summaries至事件log中。
# 如果有例外发生，should_stop()将返回True

使用在多副本运行情况中

要使用副本训练已经部署在集群上的相同程序，必须指定其中一个task为主要，该task处理 initialization, checkpoints, summaries, 和recovery相关事物。其他task依赖该task。

# Choose a task as the chief. This could be based on server_def.task_index,
# or job_def.name, or job_def.tasks. It's entirely up to the end user.
# But there can be only one *chief*.
is_chief = (server_def.task_index == 0)
server = tf.train.Server(server_def)

with tf.Graph().as_default():
  ...add operations to the graph...
  # Create a Supervisor that uses log directory on a shared file system.
  # Indicate if you are the 'chief'
  sv = Supervisor(logdir='/shared_directory/...', is_chief=is_chief)
  # Get a Session in a TensorFlow server on the cluster.
  with sv.managed_session(server.target) as sess:
    # Use the session to train the graph.
    while not sv.should_stop():
      sess.run(<my_train_op>)

如果有task崩溃或重启，managed_session() 将检查是否Model被初始化。如果已经初始化，它只需要创建一个session并将其返回至正在训练的正常代码中。如果model需要被初始化，主task将对它进行重新初始化，而其他task将等待模型初始化完成。

注意：该程序方法一样适用于单进程的work，该单进程标注自己为主要的便行

▷ supervisor中master的字符串形式

无论运行在本机或者集群上，都可以使用以下值设定master flag：

• 定义为 ” ，要求一个进程内且没有使用RPC的session
• 定义为 ‘local’，要求一个使用基于RPC的主服务接口(“Master interface” )的session来运行tensorflow程序。更多细节可以查看 tf.train.Server.create_local_server()相关内容。
• 定义为 ‘grpc://hostname:port’，要求一个指定的RPC接口的session，同时运行内部进程的master接入远程的tensorflow workers。可用server.target返回该形式

▷ supervisor高级用法

• 启动额外的服务
  managed_session()启动了 Checkpoint 和Summary服务。如果需要运行更多的服务，可以在managed_session()控制的模块中启动他们。

#例如： 开启一个线程用于打印loss. 设置每60秒该线程运行一次，我们使用sv.loop()
 ...
  sv = Supervisor(logdir='/tmp/mydir')
  with sv.managed_session(FLAGS.master) as sess:
    sv.loop(60, print_loss, (sess))
    while not sv.should_stop():
      sess.run(my_train_op)

启动更少的的服务

managed_session() 启动了 “summary” 和 “checkpoint” 线程，这些线程通过构建器或者监督器默认自动创建了summary_op 和saver操作。如果想运行自己的 summary 和checkpointing方法，关闭这些服务，通过传递None值给summary_op 和saver参数。

在chief中每100个step，创建summaries
  # Create a Supervisor with no automatic summaries.
  sv = Supervisor(logdir='/tmp/mydir', is_chief=is_chief, summary_op=None)
  # As summary_op was None, managed_session() does not start the
  # summary thread.
  with sv.managed_session(FLAGS.master) as sess:
    for step in xrange(1000000):
      if sv.should_stop():
        break
      if is_chief and step % 100 == 0:
        # Create the summary every 100 chief steps.
        sv.summary_computed(sess, sess.run(my_summary_op))
      else:
        # Train normally
        sess.run(my_train_op)

▷ tf.train.Supervisor.managed_session

def train():
  sv = tf.train.Supervisor(...)
  with sv.managed_session(<master>) as sess:
    for step in xrange(..):
      if sv.should_stop():
        break
      sess.run(<my training op>)
      ...do other things needed at each training step...

▷ tf.train.SessionManager

with tf.Graph().as_default():
   ...add operations to the graph...
  # Create a SessionManager that will checkpoint the model in '/tmp/mydir'.
  sm = SessionManager()
  sess = sm.prepare_session(master, init_op, saver, checkpoint_dir)
  # Use the session to train the graph.
  while True:
    sess.run(<my_train_op>)
#其中prepare_session()初始化和恢复一个模型参数。 

#另一个进程将等待model准备完成，代码如下
with tf.Graph().as_default():
  ...add operations to the graph...
  # Create a SessionManager that will wait for the model to become ready.
  sm = SessionManager()
  sess = sm.wait_for_session(master)
  # Use the session to train the graph.
  while True:
    sess.run(<my_train_op>)
#wait_for_session()等待一个model被其他进程初始化

▷ tf.train.ClusterSpec

一个tf.train.ClusterSpec表示一系列的进程，这些进程都参与分布式tensorflow的计算。每一个 tf.train.Server都在一个独有的集群中构建。

创建一个具有两个jobs及其5个tasks的集群们需要定义从job名称列表到网络地址列表之间的映射。

cluster = tf.train.ClusterSpec({"worker": ["worker0.example.com:2222",
                                           "worker1.example.com:2222",
                                           "worker2.example.com:2222"],
                                "ps": ["ps0.example.com:2222",
                                       "ps1.example.com:2222"]})

▷ tf.train.replica_device_setter

# To build a cluster with two ps jobs on hosts ps0 and ps1, and 3 worker
# jobs on hosts worker0, worker1 and worker2.
cluster_spec = {
    "ps": ["ps0:2222", "ps1:2222"],
    "worker": ["worker0:2222", "worker1:2222", "worker2:2222"]}
with tf.device(tf.replica_device_setter(cluster=cluster_spec)):
  # Build your graph
  v1 = tf.Variable(...)  # assigned to /job:ps/task:0
  v2 = tf.Variable(...)  # assigned to /job:ps/task:1
  v3 = tf.Variable(...)  # assigned to /job:ps/task:0
# Run compute

█ 汇总操作(Summary Operations)

我们可以在一个session中获取summary操作的输出，并将其传输到SummaryWriter以添加至一个事件记录文件中。

将记录汇总写入文件中(Adding Summaries to Event Files)

▷ tf.train.SummaryWriter

创建一个SummaryWriter 和事件文件。如果我们传递一个Graph进入该构建器中，它将被添加到事件文件当中，这一点与使用add_graph()具有相同功能。

TensorBoard 将从事件文件中提取该graph，并将其显示。所以我们能直观地看到我们建立的graph。我们通常从我们启动的session中传递graph：

...create a graph...
# Launch the graph in a session.
sess = tf.Session()
# Create a summary writer, add the 'graph' to the event file.
writer = tf.train.SummaryWriter(<some-directory>, sess.graph)

▷ tf.train.summary_iterator

#打印时间文件中的内容
for e in tf.train.summary_iterator(path to events file):
    print(e)

#打印指定的summary值
# This example supposes that the events file contains summaries with a
# summary value tag 'loss'.  These could have been added by calling
# `add_summary()`, passing the output of a scalar summary op created with
# with: `tf.scalar_summary(['loss'], loss_tensor)`.
for e in tf.train.summary_iterator(path to events file):
    for v in e.summary.value:
        if v.tag == 'loss':
            print(v.simple_value)

█ 训练的通用函数及其他(Training utilities)

▷ tf.train.global_step

# Creates a variable to hold the global_step.
global_step_tensor = tf.Variable(10, trainable=False, name='global_step')
# Creates a session.
sess = tf.Session()
# Initializes the variable.
sess.run(global_step_tensor.initializer)
print('global_step: %s' % tf.train.global_step(sess, global_step_tensor))

global_step: 10

▷ tf.train.write_graph

v = tf.Variable(0, name='my_variable')
sess = tf.Session()
tf.train.write_graph(sess.graph_def, '/tmp/my-model', 'train.pbtxt')

▷ tf.py_func

#tf.py_func(func, inp, Tout, stateful=True, name=None)
#func：为一个python函数
#inp：为输入函数的参数，Tensor列表
#Tout： 指定func返回的输出的数据类型，是一个列表
def my_func(x):
  # x will be a numpy array with the contents of the placeholder below
  return np.sinh(x)
inp = tf.placeholder(tf.float32, [...])
y = py_func(my_func, [inp], [tf.float32])

2.2 测试 (Testing)

TensorFlow 提供了一个方便的继承unittest.TestCase类的方法，该类增加有关TensorFlow 测试的方法。如下例子：

import tensorflow as tf

class SquareTest(tf.test.TestCase):

  def testSquare(self):
    with self.test_session():
      x = tf.square([2, 3])
      self.assertAllEqual(x.eval(), [4, 9])

if __name__ == '__main__':
  tf.test.main()

█ 共用(Utilities)

█ 梯度检查(Gradient checking)

可对比compute_gradient 和compute_gradient_error函数的用法

相关链接：

[1] 安装Tensorflow（Linux ubuntu） http://blog.csdn.net/lenbow/article/details/51203526

[2] ubuntu下CUDA编译的GCC降级安装 http://blog.csdn.net/lenbow/article/details/51596706

[3] ubuntu手动安装最新Nvidia显卡驱动 http://blog.csdn.net/lenbow/article/details/51683783

[4] Tensorflow的CUDA升级，以及相关配置 http://blog.csdn.net/lenbow/article/details/52118116

[5] 基于gensim的Doc2Vec简析 http://blog.csdn.net/lenbow/article/details/52120230

[6] TensorFlow的分布式学习框架简介 http://blog.csdn.net/lenbow/article/details/52130565