深度学习框架-Tensorflow基础学习

Tensorflow简介

1. Tensorflow是一个编程系统，使用图来表示计算任务。使用图来表示计算任务. 图中的节点被称之为 op (operation 的缩写). 一个 op 获得 0 个或多个 Tensor, 执行计算, 产生 0 个或多个 Tensor. 每个 Tensor 是一个类型化的多维数组. TensorFlow 程序通常被组织成一个构建阶段和一个执行阶段. 在构建阶段, op 的执行步骤 被描述成一个图. 在执行阶段, 使用会话执行执行图中的 op。

2. Tensorflow作用

• 使用图来表示计算任务
• 在Session的上下文中执行图
• 使用tensor表示数据
• 通过Variable维护状态
• 使用feed和fetch可以任意地操作赋值或者从中获取数据

3.运作方式

• inference()函数会尽可能地构建图表，做到返回包含了预测结果（output prediction）的Tensor。它接受图像占位符为输入，在此基础上借助ReLu(Rectified Linear Units)激活函数，构建一对完全连接（layers），以及一个有着十个节点（node）、指明了输出logtis模型的线性层。

• 输入与占位符（Inputs and Placeholders）：placeholder_inputs()函数将生成两个tf.placeholder操作，定义传入图表中的shape参数，shape参数中包括batch_size值。

  在训练循环（training loop）的后续步骤中，传入的整个图像和标签数据集会被切片，以符合每一个操作所设置的batch_size值，占位符操作将会填补以符合这个batch_size值。然后使用feed_dict参数，将数据传入sess.run()函数。

• 构建图表（Build the Graph）

变量：创建、初始化、保存和加载

当训练的时候，用变量来保存和更新参数。变量包含张量（Tensor）存放于内存的缓冲区。

创建

• tf.Variable

  import tensorflow as tf
  weights = tf.Variable(tf.random_normal([784,200], stddev=0.35), name = "weights") # 权重
  

初始化

• 有时候需要由另一个变量的初始化值给当前变量初始化，tf.initialize_all_variables()是并行地初始化所有变量，所以在有这种需求的情况下需要小心!用其它变量的值初始化一个新的变量时，使用其它变量的initialized_value()属性。你可以直接把已初始化的值作为新变量的初始值，或者把它当做tensor计算得到一个值赋予新变量。

  # 承接上面创建的变量weights
  w2 = tf.Variable(weights.initialized_value(), name = 'w2')
  

• 自定义初始化。tf.initialized_all_variables()函数添加一个op来初始化变量模型的所有变量。

保存和加载

最简单的保存和恢复模型的方法是使用tf.train.Saver对象。构造器给graph的所有变量，或是定义在列表里的变量，添加save和restoreops。saver对象提供了方法来运行这些ops，定义检查点文件的读写路径。

• 检查点文件

  变量存储在二进制文件里，主要包含从变量名到tensor值的映射关系。当创建一个Saver对象时，默认情况下，将每个变量Variables.name属性的值。

• 保存变量

  tf.train.Saver()创建一个saver对象来管理所有的变量

  # Create all variables
  v1 = tf.Variable(..., name= 'v1')
  v2 = tf.Variable(..., name='v2')
  # Add an op to initialize the variables
  init_op = initialized_all_variables()
  # Add ops to save and restore all the variables
  saver() = tf.train.Saver()
  # Later, launch the model, initialize the variables, do some work,save the variables to disk.
  with tf.Session() as sess:
      sess.run()
      #..do some work with the model..
      save_path = saver.save(sess, '/tmp/model.ckpt')
      print('Model saved in file:', save_path)
  

• 恢复变量

  恢复变量使用同一个Saver对象来恢复对象。saver.restore()恢复变量。

数据读取

Feeding Data

Tensorflow的feeding机制允许在tensorflow运算图中将数据注入到任意一个张量中。通过run()或者eval()函数输入feed_dict参数

with tf.Session() as tf:
    input = tf.placeholder(tf.float32)
    classifer = ...
    print(classifer.eval(feed_dict={input:my_python_preprocessing_fn()}))

从文件中读取数据的步骤

• 1. 文件名列表：可以使用字符串张量(比如["file0", "file1"], [("file%d" % i) for i in range(2)]， [("file%d" % i) for i in range(2)]) 或者tf.train.match_filenames_once 函数来产生文件名列表。
• 1. 可配置的文件名乱序（shuffling）：string_input_producer 提供的可配置参数来设置文件名乱序和最大的训练迭代数， QueueRunner会为每次迭代(epoch)将所有的文件名加入文件名队列中， 如果shuffle=True的话， 会对文件名进行乱序处理。这一过程是比较均匀的，因此它可以产生均衡的文件名队列。
  2. 可配置的最大训练迭代次数（epoch limit）
  3. 文件名队列
  4. 针对输入文件格式的阅读器
  5. 解析器
  6. 可配置的预处理器
  7. 样本队列

不同文件格式的数据读取

不同的文件格式，需要选择不同的文件阅读器，然后将文件名队列提供给阅读器的read()方法。

• CSV文件

  需要用到textLineReader()、decode_csv()

  每次read的执行都会从文件中读取一行内容， decode_csv 操作会解析这一行内容并将其转为张量列表。如果输入的参数有缺失，record_default参数可以根据张量的类型来设置默认值。

  在调用run或者eval去执行read之前， 你必须调用tf.train.start_queue_runners来将文件名填充到队列。否则read操作会被阻塞到文件名队列中有值为止。

  # 使用string_input_producer来生成一个先入先出的队列
  filename_queue = tf.train.string_input_producer(['file0.csv', 'file1.csv'])
  # 使用阅读器读取
  reader = tf.TextLineReader()
  key,value = reader.read(filename_queue)
  # decord result
  recorde_defaults = [[1], [1], [1], [1], [1]]
  col1,col2,col3,col4,col5 = tf.decode_csv(value, record_defaults=record_defaults)
  features = tf.concat(0, [col1,col2,col3,col4])
  # start popularing the filename queue
  with tf.Session() as sess:
      coord = tf.train.Coordinator()
      threads = tf.train.start_queue_runners(coord=coord)
   	for i in range(1200):
          example,label = sess.run([features, cols])
      coord.request_stop()
      coord.join(threads) # join the queue
  

• 从二进制文件中读取固定长度的data

  从二进制文件中读取固定长度的data，可以使用tf.FixedLengthRecordReader的tf.decode_raw操作。decode_raw操作可以将一个字符串转换成一个unit8的张量。

  如果文件格式定义为：每条记录的长度都是固定的，一个字节的的标签，后面的3072是图像的数据。uint8张量可以从中获取图片并根据需要进行重组。

• 标准的TFRecode格式

  这种方法可以允许你讲任意的数据转换为TensorFlow所支持的格式， 这种方法可以使TensorFlow的数据集更容易与网络应用架构相匹配。这种建议的方法就是使用TFRecords文件。

  TFRecode文件包含了tf.train.Example协议内存块(protocol buffer),这个协议内存块包含了字段Features特征值。可以写一段代码获取数据，并将数据填入Example协议内存块中，并将内存块序列化成一个字符串，并且通过tf.python_io_TFRecordWriterclass写入TFRecords文件。

  从TFRecords文件中读取数据， 可以使用tf.TFRecordReader的tf.parse_single_example解析器。这个parse_single_example操作可以将Example协议内存块(protocol buffer)解析为张量。

预处理

可以在tensorflow/models/image/cifar10/cifar10.py找到数据归一化， 提取随机数据片，增加噪声或失真等等预处理的例子。

批处理

在数据输入管线的末端， 我们需要有另一个队列来执行输入样本的训练，评价和推理。因此我们使用tf.train.shuffle_batch 函数来对队列中的样本进行乱序处理。

def read_file(filename_queue):
    reader = tf.SomeReader()
    key,record_string = reader.read(filename_queue)
    example,label = tf.some_decoder(record_string) # 解码器
    processed_example = some_processing(example)
return processed_example,label

def input_pipeline(filenames, batch_size, num_epochs=None):
    filename_queue = tf.train.string_input_producer(
        filenames,num_epochs=num_epochs,shuffle=True
    )
    example,label = read_file(filename_queue)
    # min_after_dequeue defines how big a buffer we will randomly sample
  #   from -- bigger means better shuffling but slower start up and more
  #   memory used.
  # capacity must be larger than min_after_dequeue and the amount larger
  #   determines the maximum we will prefetch.  Recommendation:
  #   min_after_dequeue + (num_threads + a small safety margin) * batch_size
  min_after_dequeue = 10000
    capacity = min_after_dequeue + 3*batch_size
    example_batch, label_batch = tf.train.shuffle_batch(
    	[example, label], batch_size = batch_size,capacity=capacity,
        min_after_dequeue=min_after_dequeue
    )
    return example_batch,label_batch

如果你需要对不同文件中的样子有更强的乱序和并行处理，可以使用tf.train.shuffle_batch_join 函数. 示例:

def read_my_file_format(filename_queue):
  # Same as above

def input_pipeline(filenames, batch_size, read_threads, num_epochs=None):
  filename_queue = tf.train.string_input_producer(
      filenames, num_epochs=num_epochs, shuffle=True)
  example_list = [read_my_file_format(filename_queue)
                  for _ in range(read_threads)]
  min_after_dequeue = 10000
  capacity = min_after_dequeue + 3 * batch_size
  example_batch, label_batch = tf.train.shuffle_batch_join(
      example_list, batch_size=batch_size, capacity=capacity,
      min_after_dequeue=min_after_dequeue)
  return example_batch, label_batch

另一种替代方案是： 使用tf.train.shuffle_batch 函数,设置num_threads的值大于1。 这种方案可以保证同一时刻只在一个文件中进行读取操作(但是读取速度依然优于单线程)，而不是之前的同时读取多个文件。这种方案的优点是：

• 避免了两个不同的线程从同一个文件中读取同一个样本。
• 避免了过多的磁盘搜索操作。

创建线程并使用QueueRunner对象来预取

使用上面列出的许多tf.train函数添加QueueRunner到你的数据流图中。在你运行任何训练步骤之前，需要调用tf.train.start_queue_runners函数，否则数据流图将一直挂起。tf.train.start_queue_runners 这个函数将会启动输入管道的线程，填充样本到队列中，以便出队操作可以从队列中拿到样本。这种情况下最好配合使用一个tf.train.Coordinator，这样可以在发生错误的情况下正确地关闭这些线程。如果你对训练迭代数做了限制，那么需要使用一个训练迭代数计数器，并且需要被初始化。

推荐：

# Create the graph, etc.
init_op = tf.initialize_all_variables()

# Create a session for running operations in the Graph.
sess = tf.Session()

# Initialize the variables (like the epoch counter).
sess.run(init_op)

# Start input enqueue threads.
coord = tf.train.Coordinator()
threads = tf.train.start_queue_runners(sess=sess, coord=coord)

try:
    while not coord.should_stop():
        # Run training steps or whatever
        sess.run(train_op)

except tf.errors.OutOfRangeError:
    print 'Done training -- epoch limit reached'
finally:
    # When done, ask the threads to stop.
    coord.request_stop()

# Wait for threads to finish.
coord.join(threads)
sess.close()

线程和队列

在使用TensorFlow进行异步计算时，队列是一种强大的机制。

• Coordinator：用来协同多个线程工作，同步终止，主要方法如下：

  should_stop():如果线程应该停止则返回True。
  request_stop(<exception>): 请求该线程停止。
  join(<list of threads>):等待被指定的线程终止。
      
  # 线程体：循环执行，直到`Coordinator`收到了停止请求。
  # 如果某些条件为真，请求`Coordinator`去停止其他线程。
  def MyLoop(coord):
    while not coord.should_stop():
      ...do something...
      if ...some condition...:
        coord.request_stop()
  
  # Main code: create a coordinator.
  coord = Coordinator()
  
  # Create 10 threads that run 'MyLoop()'
  threads = [threading.Thread(target=MyLoop, args=(coord)) for i in xrange(10)]
  
  # Start the threads and wait for all of them to stop.
  for t in threads: t.start()
  coord.join(threads)
  

• QueueRunner：QueueRunner会创建一组线程，这些线程可以执行Enqueue操作，使用同一个Coordinator协同处理。

  # Create a queue runner that will run 4 threads in parallel to enqueue
  # examples.
  qr = tf.train.QueueRunner(queue, [enqueue_op] * 4)
  
  # Launch the graph.
  sess = tf.Session()
  # Create a coordinator, launch the queue runner threads.
  coord = tf.train.Coordinator()
  enqueue_threads = qr.create_threads(sess, coord=coord, start=True)
  # Run the training loop, controlling termination with the coordinator.
  for step in xrange(1000000):
      if coord.should_stop():
          break
      sess.run(train_op)
  # When done, ask the threads to stop.
  coord.request_stop()
  # And wait for them to actually do it.
  coord.join(threads)
  

OP接口

定义OP接口

向 TensorFlow 系统注册来定义 Op 的接口. 在注册时, 指定 Op 的名称, 它的输入(类型和名称) 和输出(类型和名称), 和所需要任何 属性的文档说明.

 #include "tensorflow/core/framework/op.h"
REGISTER_OP("ZeroOut")
    .Input("to_zero: int32")
    .Output("zeroed: int32");

为op实现kernel

在定义接口之后, 提供一个或多个 Op 的实现. 为这些 kernel 的每一个创建一个对应的类, 继承 OpKernel, 覆盖 Compute 方法. Compute 方法提供一个类型为 OpKernelContext* 的参数 context, 用于访问一些有用的信息。

例如：

#include "tensorflow/core/framework/op_kernel.h"
using namespace tensorflow;
class ZeroOutOp : public OpKernel {
 public:
  explicit ZeroOutOp(OpKernelConstruction* context) : OpKernel(context) {}
  void Compute(OpKernelContext* context) override {
    // 获取输入 tensor.
    const Tensor& input_tensor = context->input(0);
    auto input = input_tensor.flat<int32>();
   // 创建一个输出 tensor.
    Tensor* output_tensor = NULL;
    OP_REQUIRES_OK(context, context->allocate_output(0, input_tensor.shape(),
                                                     &output_tensor));
    auto output = output_tensor->template flat<int32>();
    // 设置 tensor 除第一个之外的元素均设为 0.
    const int N = input.size();
    for (int i = 1; i < N; i++) {
      output(i) = 0;
    }
    // 尽可能地保留第一个元素的值.
    if (N > 0) output(0) = input(0);
  }
};

实现 kernel 后, 将其注册到 TensorFlow 系统中. 注册时, 可以指定该 kernel 运行时的多个约束 条件. 例如可以指定一个 kernel 在 CPU 上运行, 另一个在 GPU 上运行.

# 将下列代码加入到 zero_out.cc 中, 注册 ZeroOut op:
REGISTER_KERNEL_BUILDER(Name("ZeroOut").Device(DEVICE_CPU), ZeroOutOp);

自定义数据读取

编写一个文件格式读写器

1.Reader 是专门用来读取文件中的记录的。TensorFlow中内建了一些读写器Op的实例：

• tf.TFRecordReader (代码位于kernels/tf_record_reader_op.cc)
• tf.FixedLengthRecordReader (代码位于 kernels/fixed_length_record_reader_op.cc)
• tf.TextLineReader (代码位于 kernels/text_line_reader_op.cc)

这些读写器的界面是一样的，唯一的差异是在它们的构造函数中。最重要的方法是 Read。 它需要一个行列参数，通过这个行列参数，可以在需要的时候随时读取文件名 (例如： 当 Read Op首次运行，或者前一个 Read` 从一个文件中读取最后一条记录时)。它将会生成两个标量张量： 一个字符串和一个字符串关键值。

新创建一个名为 SomeReader 的读写器，需要以下步骤：

1. 在 C++ 中, 定义一个 tensorflow::ReaderBase的子类，命名为 "SomeReader".
2. 在 C++ 中，注册一个新的读写器Op和Kernel，命名为 "SomeReader"。
3. 在 Python 中, 定义一个 tf.ReaderBase 的子类，命名为 "SomeReader"。
   可以执行以下方法：

• OnWorkStartedLocked：打开下一个文件
• ReadLocked：读取一个记录或报告 EOF/error
• OnWorkFinishedLocked：关闭当前文件
• ResetLocked：清空记录，例如：一个错误记录

2.注册Op。需要用到一个调用指令定义在 tensorflow/core/framework/op.h中的REGISTER_OP。

#include "tensorflow/core/framework/op.h"
REGISTER_OP("TextLineReader")
    .Output("reader_handle: Ref(string)")
    .Attr("skip_header_lines: int = 0")
    .Attr("container: string = ''")
    .Attr("shared_name: string = ''")
    .SetIsStateful()
    .Doc(R"doc(
A Reader that outputs the lines of a file delimited by '
'.
)doc");

3.定义并注册 OpKernel。要定义一个 OpKernel， 读写器可以使用定义在tensorflow/core/framework/reader_op_kernel.h中的 ReaderOpKernel 的递减快捷方式，并运行一个叫 SetReaderFactory 的构造函数。 定义所需要的类之后，你需要通过 REGISTER_KERNEL_BUILDER(...) 注册这个类。

 #include "tensorflow/core/framework/reader_op_kernel.h"
class TextLineReaderOp : public ReaderOpKernel {
 public:
  explicit TextLineReaderOp(OpKernelConstruction* context)
      : ReaderOpKernel(context) {
    int skip_header_lines = -1;
    OP_REQUIRES_OK(context,
                   context->GetAttr("skip_header_lines", &skip_header_lines));
    OP_REQUIRES(context, skip_header_lines >= 0,
                errors::InvalidArgument("skip_header_lines must be >= 0 not ",
                                        skip_header_lines));
    Env* env = context->env();
    SetReaderFactory([this, skip_header_lines, env]() {
      return new TextLineReader(name(), skip_header_lines, env);
    });
  }
};
REGISTER_KERNEL_BUILDER(Name("TextLineReader").Device(DEVICE_CPU),
                        TextLineReaderOp);

1. 添加 Python 包装器，你需要将 tensorflow.python.ops.io_ops 导入到tensorflow/python/user_ops/user_ops.py，并添加一个 io_ops.ReaderBase的衍生函数。

   from tensorflow.python.framework import ops
   from tensorflow.python.ops import common_shapes
   from tensorflow.python.ops import io_ops
   class SomeReader(io_ops.ReaderBase):
       def __init__(self, name=None):
           rr = gen_user_ops.some_reader(name=name)
           super(SomeReader, self).__init__(rr)
   ops.NoGradient("SomeReader")
   ops.RegisterShape("SomeReader")(common_shapes.scalar_shape)
   

   你可以在 tensorflow/python/ops/io_ops.py中查看一些范例。

   编写一个记录格式op

   一个普通的Op， 需要一个标量字符串记录作为输入， 因此遵循 添加Op的说明。 你可以选择一个标量字符串作为输入， 并包含在错误消息中报告不正确的格式化数据。

   用于解码记录的运算实例：

   • tf.parse_single_example (and tf.parse_example)
   • tf.decode_csv
   • tf.decode_raw

请注意，使用多个Op 来解码某个特定的记录格式也是有效的。 例如，你有一张以字符串格式保存在tf.train.Example 协议缓冲区的图像文件。 根据该图像的格式， 你可能从 tf.parse_single_example 的Op 读取响应输出并调用 tf.decode_jpeg， tf.decode_png， 或者 tf.decode_raw。通过读取 tf.decode_raw 的响应输出并使用tf.slice 和 tf.reshape 来提取数据是通用的方法。

使用GPU

• "/cpu:0": 机器中的 CPU
• "/gpu:0": 机器中的 GPU, 如果你有一个的话.
• "/gpu:1": 机器中的第二个 GPU, 以此类推...

如果一个 TensorFlow 的 operation 中兼有 CPU 和 GPU 的实现, 当这个算子被指派设备时, GPU 有优先权. 比如matmul中 CPU 和 GPU kernel 函数都存在. 那么在 cpu:0 和 gpu:0 中, matmul operation 会被指派给 gpu:0 。

指派某个GPU执行运行：

# 新建一个 graph.
with tf.device('/gpu:2'):
  a = tf.constant([1.0, 2.0, 3.0, 4.0, 5.0, 6.0], shape=[2, 3], name='a')
  b = tf.constant([1.0, 2.0, 3.0, 4.0, 5.0, 6.0], shape=[3, 2], name='b')
  c = tf.matmul(a, b)
# 新建 session with log_device_placement 并设置为 True.
sess = tf.Session(config=tf.ConfigProto(
      allow_soft_placement=True, log_device_placement=True))
# 运行这个 op.
print sess.run(c)

共享变量

共享变量的方法就是在单独的代码块中来创建他们并且通过使用他们的函数。

variables_dict = {
    "conv1_weights": tf.Variable(tf.random_normal([5,5,32,32]), name="conv1_weights"),
    "conv1_biases":tf.Variable(tf.zeros([32]), name="conv1_biases"),
    ...
}

def my_image_filter(input_images, variables_dict):
    conv1 = tf.nn.conv2d(input_images, variables_dict["conv1_weights"],
        strides=[1, 1, 1, 1], padding='SAME')
    relu1 = tf.nn.relu(conv1 + variables_dict["conv1_biases"])

    conv2 = tf.nn.conv2d(relu1, variables_dict["conv2_weights"],
        strides=[1, 1, 1, 1], padding='SAME')
    return tf.nn.relu(conv2 + variables_dict["conv2_biases"])

# The 2 calls to my_image_filter() now use the same variables
result1 = my_image_filter(image1, variables_dict)
result2 = my_image_filter(image2, variables_dict)

虽然使用上面的方式创建变量是很方便的，但是在这个模块代码之外却破坏了其封装性：

• 在构建试图的代码中标明变量的名字，类型，形状来创建.
• 当代码改变了，调用的地方也许就会产生或多或少或不同类型的变量.

解决此类问题的方法之一就是使用类来创建模块，在需要的地方使用类来小心地管理他们需要的变量. 一个更高明的做法,不用调用类，而是利用TensorFlow 提供了变量作用域 机制，当构建一个视图时,很容易就可以共享命名过的变量.

变量作用域

变量作用域机制在TensorFlow中主要由两部分组成：

• tf.get_variable(<name>, <shape>, <initializer>): 通过所给的名字创建或是返回一个变量.
• tf.variable_scope(<scope_name>): 通过 tf.get_variable()为变量名指定命名空间.

方法 tf.get_variable() 用来获取或创建一个变量，而不是直接调用tf.Variable.它采用的不是像`tf.Variable这样直接获取值来初始化的方法.一个初始化就是一个方法，创建其形状并且为这个形状提供一个张量.这里有一些在TensorFlow中使用的初始化变量：

• tf.constant_initializer(value) 初始化一切所提供的值,

• tf.random_uniform_initializer(a, b)从a到b均匀初始化,

• tf.random_normal_initializer(mean, stddev) 用所给平均值和标准差初始化均匀分布.

  # 单独创建一个卷积的函数，共享变量权重和偏置封装
  def conv_relu(input, kernel_shape, bias_shape):
      # Create variable named "weights".
      weights = tf.get_variable("weights", kernel_shape,
          initializer=tf.random_normal_initializer())
      # Create variable named "biases".
      biases = tf.get_variable("biases", bias_shape,
          initializer=tf.constant_intializer(0.0))
      conv = tf.nn.conv2d(input, weights,
          strides=[1, 1, 1, 1], padding='SAME')
      return tf.nn.relu(conv + biases)
  

  tf.variable_scope() 为变量指定了相应的命名空间。

  def my_image_filter(input_images):
      with tf.variable_scope("conv1"):
          # Variables created here will be named "conv1/weights", "conv1/biases".
          relu1 = conv_relu(input_images, [5, 5, 32, 32], [32])
      with tf.variable_scope("conv2"):
          # Variables created here will be named "conv2/weights", "conv2/biases".
          return conv_relu(relu1, [5, 5, 32, 32], [32])
  # 开始调用
  result1 = my_image_filter(image1)
  result2 = my_image_filter(image2)
  # Raises ValueError(... conv1/weights already exists ...)
  

  就像你看见的一样，tf.get_variable()会检测已经存在的变量是否已经共享.如果你想共享他们，你需要像下面使用的一样，通过reuse_variables()这个方法来指定.

  with tf.variable_scope("image_filters") as scope:
      result1 = my_image_filter(image1)
      scope.reuse_variables()
      result2 = my_image_filter(image2)
  

具体机制见Tensorflow教程