python - 批处理迭代如何在 Tensorflow 中工作？

Question

我正在尝试重用 PTB language model在我的数据上，但缺乏 Tensorflow 的知识来理解它如何处理训练数据的批量迭代。以下是我在训练期间对批量迭代的理解:

while epoch <= maxepoch do
  for minibatch in data_iterator() do
    model.forward(minibatch)
    (...)
  end
end

再简单不过了，不是吗？ 类似的事情在许多其他框架中完成，但在 Tensorflow 中没有:) 下面是官方 PTB 语言模型教程中的 minibatch 函数示例:

def ptb_producer(raw_data, batch_size, num_steps, name=None):
    with tf.name_scope(name, "PTBProducer", [raw_data, batch_size, num_steps]):
        raw_data = tf.convert_to_tensor(raw_data, name="raw_data", dtype=tf.int32)

        data_len = tf.size(raw_data)
        batch_len = data_len // batch_size
        data = tf.reshape(raw_data[0 : batch_size * batch_len],
                                            [batch_size, batch_len])

        epoch_size = (batch_len - 1) // num_steps
        assertion = tf.assert_positive(
                epoch_size,
                message="epoch_size == 0, decrease batch_size or num_steps")
        with tf.control_dependencies([assertion]):
            epoch_size = tf.identity(epoch_size, name="epoch_size")

        i = tf.train.range_input_producer(epoch_size, shuffle=False).dequeue()
        x = tf.strided_slice(data, [0, i * num_steps], [batch_size, (i + 1) * num_steps])
        x.set_shape([batch_size, num_steps])
        y = tf.strided_slice(data, [0, i * num_steps + 1], [batch_size, (i + 1) * num_steps + 1])
        y.set_shape([batch_size, num_steps])
        return x, y

此函数在调用后返回 x 输入和 y 目标。我在这里看不到 Python 迭代器的迹象，但是有一个对 tf.strided_slice 的调用，它使用 tf.train.range_input_producer 生成的 i 索引，所以这应该在数据上模拟一个滑动窗口。但是，该函数在训练前仅被调用一次，那么它如何遍历我的数据呢？这还不清楚。有人可以解释这种“神奇”且完全模糊的 Tensorflow 机制吗？

Answer 1

“魔法”隐藏在调用 tf.train.range_input_producer 的行中:

i = tf.train.range_input_producer(epoch_size, shuffle=False).dequeue()

... 创建一个操作，该操作从队列中弹出值 持有 0..epoch_size-1 整数。换句话说，它在 0..epoch_size-1 范围内迭代。

---

是的，这似乎违反直觉。所以这是一个在 tensorflow 中使用队列的简单可运行示例:

index = tf.train.range_input_producer(10, shuffle=False).dequeue()

with tf.Session() as sess:
  coord = tf.train.Coordinator()
  threads = tf.train.start_queue_runners(coord=coord)

  for i in range(15):
    print(sess.run(index))

  coord.request_stop()
  coord.join(threads)

运行时，您应该看到从 0 到 9 的值，然后是从 0 到 4 的 5 个值>。请注意，sess.run 计算相同的张量 index，但它每次都得到一个不同的值。可以添加更多依赖于 index 的操作，它们将使用 index 的新值进行评估。

另请注意，队列在另一个线程中运行，因此为了使用tf.train.range_input_producer，必须启动一个Coordinator 并产生一些线程(并最终停止它们)。如果您尝试在没有 Coordinator 的情况下运行相同的示例，sess.run(index) 将阻止 脚本执行。

你可以试试这个例子，例如，设置 shuffle=True 等

---

回到 PTB 生产者片段:

i = tf.train.range_input_producer(epoch_size, shuffle=False).dequeue()
x = tf.strided_slice(data, [0, i*num_steps], [batch_size, (i+1)*num_steps])
x.set_shape([batch_size, num_steps])
y = tf.strided_slice(data, [0, i*num_steps+1], [batch_size, (i+1)*num_steps+1])
y.set_shape([batch_size, num_steps])

现在应该清楚了，尽管 x 和 y 被定义为简单的张量，但它们实际上是 迭代器 数据。所有线程工作都由 tf.train.Supervisor 负责.因此调用优化操作(取决于 x 和 y)将自动获取新批处理。

---

推荐阅读:

• tensorflow "Reading data"教程