教程 | 维度、广播操作与可视化：如何高效使用TensorFlow

选自GitHub

机器之心编译

参与：Nurhachu Null、李泽南

本文从 Tensorflow 基础、理解静态维度和动态维度、广播操作（Broadingcast 的好处和坏处）、使用 Python 操作的原型内核和先进的可视化等几个方面详细梳理如何高效使用TensorFlow。

Tensorflow 基础

TensorFlow 和其他诸如 numpy 之类的数学计算库的根本区别在于：在 TensorFlow 中，运算操作是符号化的。这是一个强大的思想，它能够让 TensorFlow 做任何事情（例如，自动求微分），而这些事情对于 numpy 等命令式的库而言是不可能完成的。但是这也导致了随之而来的代价，就是掌握这个库会相对困难一些。在本文中，作者会尝试揭开 TensorFlow 的神秘面纱，并提供一些关于高效使用 TensorFlow 的指南和实践例子。

让我们以一个简单的例子开始，我们想让两个随机矩阵相乘。首先我们看一下用 numpy 来实现这个例子：

import numpy as np

x = np.random.normal(size=[10, 10])
y = np.random.normal(size=[10, 10])
z = np.dot(x, y)print(z)

这样的计算在 TensorFlow 中会是什么样子？

结果如下：

import tensorflow as tf

x = tf.random_normal([10, 10])
y = tf.random_normal([10, 10])
z = tf.matmul(x, y)

sess = tf.Session()
z_val = sess.run(z)print(z_val)

与 numpy 直接执行计算并将结果复制到变量 z 中的思路不同的是，TensorFlow 仅仅给图中代表结果的节点给提供了一个操作。如果我们直接打印 z 的值，我们会得到下面的信息：

Tensor("MatMul:0", shape=(10, 10), dtype=float32)

由于两个输入矩阵都有被完全定义的维度，TensorFlow 还能够在指定张量的维度的同时指定它的数据类型。为了计算出张量的值，我们需要使用 Session.run() 这个函数来创建一个会话。

Tip：在使用 Jupyter notebook 的时候，要确保在开始的时候调用一下 tf.reset_default() 函数，以在定义新节点之前清空符号图。

为了理解符号计算有多么强大，让我们来看一下另一个例子。假设我们有一些一条曲线上的样本点（例如曲线是 f(x) = 5x^2 + 3），但是我们想要在不知道参数的情况下来估计这个函数 f(x)。我们定义一个含参数的函数 g(x, w) = w0 x^2 + w1 x + w2，它是关于输入数据 x 和隐藏参数 w 的函数，我们的目标就是找到这组隐藏参数，使得 g(x, w) ≈ f(x)。我们可以通过最小化下面的损失函数 L(w) 来实现：L(w) = (f(x) - g(x, w))^2。尽管这个简单的问题已经有一个闭合的解决方法了，但我们还是选择使用一个更加通用的方法，这个方法能够被应用在任何可微分的函数中，它使用了随机梯度降的方法。我们简单地计算损失函数 L(w) 在一组样本点上关于 w 的平均梯度，然后朝着梯度的反方向变化参数 w。

下面展示了这个方法在 TensorFlow 中是如何实现的：

import numpy as np import tensorflow as tf # 使用占位符从python向TensorFlow运算符中传递参数值。我们在这里定义了两个占位符，其中一个用来存放输入特征x，另一个用来存放输出y. x = tf.placeholder(tf.float32) y = tf.placeholder(tf.float32) # 假设我们已经知道了期望的函数是一个二次多项式，我们就会分配一个具有3个元素的向量来代表这些参数。这些变量会被以随机地进行初始化。 w = tf.get_variable("w", shape=[3, 1]) # 我们定义yhat为我们对y的估计值 f = tf.stack([tf.square(x), x, tf.ones_like(x)], 1) yhat = tf.squeeze(tf.matmul(f, w), 1) # 损失函数被定义为y的估计值和真实值之间地l2距离。我们还附加了一个收缩项，以确保结果得到的权值会比较小。 loss = tf.nn.l2_loss(yhat - y) + 0.1 * tf.nn.l2_loss(w) # 我们以0.1的学习率使用Adam优化器来最小化损失函数。 train_op = tf.train.AdamOptimizer(0.1).minimize(loss) def generate_data(): x_val = np.random.uniform(-10.0, 10.0, size=100) y_val = 5 * np.square(x_val) + 3 return x_val, y_val sess = tf.Session() # 因为我们要使用这些变量，所以我们需要先将它们初始化 sess.run(tf.global_variables_initializer()) for _ in range(1000): x_val, y_val = generate_data() _, loss_val = sess.run([train_op, loss], {x: x_val, y: y_val}) print(loss_val) print(sess.run([w]))

运行完这段代码之后，我得到的参数结果是：

[4.98605919,-0.00187828875e-04,3.8395009]

上面是编辑运行完之后的结果，它对应的损失值是 17.6175. 每一次的具体结果都会不同，但是最终结果都很接近期望的函数值。下面是原文作者提供的值。

[4.9924135,0.00040895029, 3.4504161]

这是与期望参数相当接近的近似。

这只是 TensorFlow 能够做到的事情的冰山一角而已。很多类似于优化具有上百万个参数的大型神经网络的问题都能够用 TensorFlow 以很少量的代码来高效地实现。与此同时，TensorFlow 的开发团队还致力于在多种设备、多线程以及支持多平台等问题上更进一步。

简单起见，在绝大多数例子中我们都手动地创建了会话，我们并没有保存和加载 checkpoint，但是这却是我们在实战中经常需要做的事情。你很可能想着使用估计 API 来进行会话管理以及做日志。我们在 code/framework 路径下提供了一个简单的可扩展架构，作为使用 TensorFlow 来训练神经网络的一个实际架构的例子。

理解静态维度和动态维度

TensorFlow 中的张量具有静态维度的属性，它在构建图的时候就被确定好了。静态维度也有可能是不确定的。举个例子，我们也许会定义一个维度为 [None，128] 的张量。

import tensorflow as tf

a = tf.placeholder([None, 128])

这意味着第一个维度可以是任意大小，会在 Session.run() 的过程中被动态地决定。在表现静态张量的时候，TensorFlow 有着相当丑的 API：

static_shape = a.get_shape().as_list()  # returns [None, 128]

（这个应该写成 a,shape() 的形式，但是这里有人把它定义得太不方便了。）

为获得张量的动态形式，你可以调用 tf.shape 功能，它会返回一个表示给定张量的形状的张量：

dynamic_shape = tf.shape(a)

一个张量的静态维度可以使用 Tensor.set_shape() 函数来进行设置。

a.set_shape([32, 128])

仅当你知道自己在做什么的时候再使用这个函数，事实上使用 tf.reshape() 会更加安全。

a =  tf.reshape(a, [32, 128])

如果有一个函数能在方便的时候返回静态维度，在可用的时候返回动态维度，那将会很方便。下面就定义了这样一个函数：

def get_shape(tensor):
  static_shape = tensor.get_shape().as_list()
  dynamic_shape = tf.unstack(tf.shape(tensor))
  dims = [s[1] if s[0] is None else s[0]
          for s in zip(static_shape, dynamic_shape)]  
  return dims

现在设想：我们想通过折叠第二维和第三维来把一个 3 维的矩阵转换成一个 2 维的矩阵。我们可以使用上述的 get_shape() 函数来完成这件事：

b = placeholder([None, 10, 32])
shape = get_shape(tensor)
b = tf.reshape(b, [shape[0], shape[1] * shape[2]])

值得注意的是，这里无论矩阵是不是静态的，这个方法都能奏效。

事实上我们可以写一个更加具有通用目标的函数来折叠任何几个维度：

import tensorflow as tfimport numpy as np
def reshape(tensor, dims_list):
  shape = get_shape(tensor)
  dims_prod = []
  for dims in dims_list:
      if isinstance(dims, int):
      dims_prod.append(shape[dims])    
      elif all([isinstance(shape[d], int) for d in dims]):
      dims_prod.append(np.prod([shape[d] for d in dims]))    
      else:
      dims_prod.append(tf.prod([shape[d] for d in dims]))
  tensor = tf.reshape(tensor, dims_prod)  
  return tensor

然后折叠第二个维度就变得非常容易了。

b = placeholder([None, 10, 32])
b = tf.reshape(b, [0, [1, 2]])

广播操作

TensorFlow 支持广播逐个元素的操作。正常情况下，当你想执行类似于加法和乘法的操作时，你需要确保算子的维度是匹配的。例如，你不能把一个维度为 [3,2] 的张量与一个维度为 [3,4] 的张量相加。但是在一个特殊的情况下你可以使用异常的维度。TensorFlow 会隐式地把一个张量的异常维度调整到与另一个算子相匹配的维度以实现维度兼容。所以将一个维度为 [3,2] 的张量与一个维度为 [3,1] 的张量相加是合法的。

import tensorflow as tf

a = tf.constant([[1., 2.], [3., 4.]])
b = tf.constant([[1.], [2.]])# c = a + tf.tile(a, [1, 2])c = a + b 

广播允许我们执行隐式调整，这能够让代码更短，更加高效地使用内存，因为我们不需要存储调整操作中间结果的内存开销。这个方法可以被用在一个场景中：那就是结合不同长度的特征。为了连接不同长度的特征，我们通常会把输入张量进行调整，然后把结果连接起来并应用一些非线性处理方法。这是很多神经网络中的常用方法。

a = tf.random_uniform([5, 3, 5])
b = tf.random_uniform([5, 1, 6])

# concat a and b and apply nonlinearity

tiled_b = tf.tile(b, [1, 3, 1])
c = tf.concat([a, tiled_b], 2)
d = tf.layers.dense(c, 10, activation=tf.nn.relu)

但是这个可以用广播的方法做得更加有效。我们可以利用 f(m(x + y)) 等价于 f(mx + my) 这一事实。所以我们可以将线性操作分开处理，然后使用广播的方法去做隐式的连接。

pa = tf.layers.dense(a, 10, activation=None)
pb = tf.layers.dense(b, 10, activation=None)
d = tf.nn.relu(pa + pb)

事实上这段代码是相当通用的，只要张量之间能够进行广播操作，它就能够被用于任何维度的张量上。

def tile_concat_dense(a, b, units, activation=tf.nn.relu):
    pa = tf.layers.dense(a, units, activation=None)
    pb = tf.layers.dense(b, units, activation=None)
    c = pa + pb    if activation is not None:
        c = activation(c)    
return c

到目前为止，我们讨论了广播操作的好的一面。你会问，那么它不好的一面是什么呢？隐式的假设总会让调试变得更加难。看一下下面的例子：

a = tf.constant([[1.], [2.]])
b = tf.constant([1., 2.])
c = tf.reduce_sum(a + b)

你认为 c 的值会是多少呢？如果你说是 6，那你就错了。结果会是 12。这是因为当两个张量的秩不匹配的时候，TensorFlow 就会自动地以较低的维度来扩展第一维的大小，所以加法的结果会变成 [[2,3],[3,4]]，所以在全体参数上的求和操作会给出 12 的结果。

避免这个问题的办法就是尽可能地显示化。如果我们显示地指定了要将哪个维度进行求和，解决这个问题就会变得很容易了。

a = tf.constant([[1.], [2.]])
b = tf.constant([1., 2.])
c = tf.reduce_sum(a + b, 0)

现在 c 的值会是 [5,7]，考虑到输出结果的维度，我们会立即猜想是不是哪里出了错。一般的经验法则就是在求和操作以及使用 tf.squeeze() 的时候总要指定具体的维度。

原型内核与 Python 操作下的高度可视化

为了更高的效率，TensorFlow 的运算内核是用 C++编写的。但是用 C++写 TensorFlow 内核是一件痛苦的事情。所以，在你实现内核之前，你也许会想着快速地实现一个原型系统。借助于 tf.py_func() 函数，你可以将任何一段 Python 代码转化成 TensorFlow 操作。

例如，下面的例子展示了如何在 TensorFlow 中使用 Python 操作来实现一个简单的 ReLU 非线性核。

import numpy as np
import tensorflow as tf
import uuiddef relu(inputs):    
# Define the op in python
    def _relu(x):        
            return np.maximum(x, 0.)    
          
    # Define the op's gradient in python
    def _relu_grad(x):            return np.float32(x > 0)    

    # An adapter that defines a gradient op compatible with Tensorflow
    def _relu_grad_op(op, grad):
        x = op.inputs[0]
        x_grad = grad * tf.py_func(_relu_grad, [x], tf.float32)        
    return x_grad    

    
    # Register the gradient with a unique id
    grad_name = "MyReluGrad_" + str(uuid.uuid4())
    tf.RegisterGradient(grad_name)(_relu_grad_op)    


    # Override the gradient of the custom op
    g = tf.get_default_graph()    
    with g.gradient_override_map({"PyFunc": grad_name}):
        output = tf.py_func(_relu, [inputs], tf.float32)   
    return output

你可以使用 TensorFlow 的梯度检查器来验证梯度是否正确：

x = tf.random_normal([10])
y = relu(x * x)with tf.Session():
    diff = tf.test.compute_gradient_error(x, [10], y, [10])    
print(diff)

函数 compute_gradient_error() 会计算出梯度的数值，并且返回与给定梯度相比的差别。我所期望的是一个很小的差距。

需要注意的是，这个实现是相当低效的，并且仅对原型开发有用，因为 Python 代码并不是能够并行的，也无法在 GPU 上运行。一旦你验证了自己的思想，你肯定会想着把它写成一个 c++内核。

在实践中，我们通常会在 Tensorboard 上使用 Python 操作来实现可视化。假设你在构建一个图像分类的模型，并且想要在训练的过程中可视化模型的预测结果。TensorFlow 允许使用 tf.summary.image() 函数来做可视化。

image = tf.placeholder(tf.float32)
tf.summary.image("image", image)

但是这仅仅会可视化输入图像。为了可视化预测结果，你必须寻求一种能够做图像注解的方式，这种方式几乎在现有的操作中根本就不存在。一种比较容易的方法就是在 Python 中画图，然后用 Python 操作将其封装起来。

import io
import matplotlib.pyplot as plt
import numpy as npimport PIL
import tensorflow as tf
  def visualize_labeled_images(images, labels, max_outputs=3, name='image'):    
  def _visualize_image(image, label):   
     
        # Do the actual drawing in python
        fig = plt.figure(figsize=(3, 3), dpi=80)
        ax = fig.add_subplot(111)
        ax.imshow(image[::-1,...])
        ax.text(0, 0, str(label), 
          horizontalalignment='left', 
          verticalalignment='top')
        fig.canvas.draw()        

        # Write the plot as a memory file.
        buf = io.BytesIO()
        data = fig.savefig(buf, format='png')
        buf.seek(0)   
             # Read the image and convert to numpy array
        img = PIL.Image.open(buf)        
        return np.array(img.getdata()).reshape(img.size[0], img.size[1], -1)    
   def _visualize_images(images, labels):   
             
        # Only display the given number of examples in the batch
        outputs = []        
        for i in range(max_outputs):
            output = _visualize_image(images[i], labels[i])
            outputs.append(output)        
        return np.array(outputs, dtype=np.uint8)    
        
   # Run the python op.
   figs = tf.py_func(_visualize_images, [images, labels], tf.uint8)    
   return tf.summary.image(name, figs)

要注意，因为这里的 summary 通常都会隔一段时间才评估一次（并不是每一步都有），所以这个方法是实用的，而不用担心由此引发的效率问题。

原文地址：https://github.com/vahidk/EffectiveTensorflow

本文为机器之心编译，转载请联系本公众号获得授权。