python - TensorFlow:如何在培训期间多次评估验证数据队列？

Question

DR
如何在每次k训练迭代后评估验证集，使用单独的训练和验证数据队列，而不在多个流程中使用单独的tf.Sessions考虑到我的特殊问题，似乎没有一个干净的方法来实现这一点，而我目前的解决方法（我认为可行）给了我未定义的行为。救命！
整个故事
我想在每k次训练迭代中评估一个验证集，但我无法理解如何在TensorFlow中正确地实现这个验证集。这应该是最常见的操作之一，但它认为TensorFlow的API/体系结构在这里对我不利，或者至少使事情变得不必要的困难。
我的假设是：
[A1]此处所述的用于培训/验证的多进程模型不适用于我的问题，因为我必须假设没有足够的GPU内存来加载两次变量。
[A2]我想每k次训练迭代对验证集进行评估。
[A3]培训和验证数据不能简单地从磁盘读取，而是动态生成。这使得不可能预先可靠地预先计算验证集的大小。
[A4]验证集太大，无法预先计算并存储到磁盘上。
[A5]有效验证集大小不一定是批大小的倍数。
培训输入管道设置如下：
tf.train.slice_input_producer()生成文件名（无序排列）列表，每个文件名都引用原始输入数据。
自定义数据生成函数从每个原始输入数据块生成可变数量的培训示例/标签。
生成的培训示例/标签在送入网络之前通过tf.train.shuffle_batch()排队。
由于[a3]、[a4]、[a5]，验证输入管道的设置方式几乎相同，但最终输入队列是通过tf.train.batch()生成的，因为不需要进行洗牌。由于上述假设，基于feed-dict的方法也不可行，而且似乎与使用更高级别的函数（如tf.train.batch）不兼容。
但是，使用两组不同的队列进行培训和验证的简单实现不起作用。据我所知，我有两个选择：
[b1]将验证的num_epochs参数设置为tf.train.slice_input_producer。
在这种情况下，验证集是无限循环的，但我需要提前知道验证集的大小，以明确限制每次运行验证集时要评估的批数。此外，如果验证集的大小不能被批大小整除，那么在最后一批中，我将总是多拉一点。因为这会改变每次验证数据的评估顺序，所以这是不可接受的。
[b2]将validation的None参数设置为num_epochs，另外将tf.train.slice_input_producer函数的1参数设置为allow_smaller_final_batch。
在这种情况下，验证集只循环一次，之后将永远关闭相应的队列。默认情况下，这将使无法对验证集进行两次或更多次评估。因为我不知道在TensorFlow中重新打开队列的好方法，所以我需要解决这个限制。
由于选项[b1]的更大限制，我选择了解决选项[b2]的问题。概述我当前方法的（伪）代码如下：
训练循环应该相当规范。每k次迭代，都会调用一个函数来评估验证集。
请注意，我只启动名称以“train”开头的队列；这是为收集生成的培训数据而设置的队列。为了做到这一点，我创建了两个助手函数，tf.train.batch和True。

def train_loop(train_ops, vali_ops, ...):
    with tf.Session() as sess:
        coord = tf.train.Coordinator()
        sess.run([tf.initialize_all_variables(), tf.initialize_local_variables()])
        load_latest_snapshot(sess, loader, snapshot_file)

        # Launch the queue runners
        queues = get_queues_by_name("train")
        threads = start_queue_runners(sess, coord, queues)

        try:
            for step in range(start_iteration, num_train_iterations):
                # Runs the session on validation set
                if step % K == 0:
                    validation_results = run_validation(vali_ops, snapshot_file)

                # TRAINING:
                # ...

        except Exception as e:
            coord.request_stop(e)
        finally:
            coord.request_stop()
            coord.join(threads)

helper函数如下所示：

def get_queues_by_name(name):
    """Retrieves all queues that contain the string given by 'name'"""
    all_queues = tf.get_collection(tf.GraphKeys.QUEUE_RUNNERS)
    return [q for q in all_queues if name in q.name]


def start_queue_runners(session, coordinator, queues):
    """Similar to tf.train.start_queue_runners but now accepts a list of queues instead of a graph collection"""
    with session.graph.as_default():
        threads = []
        for queue in queues:
            log("Queue", "Starting queue '%s'" % queue.name, level=2)
            threads.extend(queue.create_threads(session, coordinator, daemon=True, start=True))
    return threads

在 get_queues_by_name函数中，针对封闭队列的问题，我选择的解决方法是创建一个新的 start_queue_runners。我还只启动与收集验证集数据的队列相关联的线程。

def run_validation(ops, snapshot_file):  # Called inside train_loop()
    results = None
    loader = tf.train.Saver()

    with tf.Session() as sess:
        coord = tf.train.Coordinator()
        sess.run([tf.initialize_local_variables()])
        load_latest_snapshot(sess, loader, snapshot_file)

        # Launch the queue runners
        queues = get_queues_by_name("eval")
        threads = start_queue_runners(sess, coord, queues)

        # Performs the inference in batches
        try:
            # Evaluate validation set:
            results = eval_in_batches(ops, sess)
        except Exception as e:
            coord.request_stop(e)
        finally:
            coord.request_stop()
            coord.join(threads)

    return results

我不知道在这里创建一个新的 run_validation是否是一个好主意，但这似乎是完成重新启动验证队列的唯一方法。理想情况下，我也不想重新加载模型快照，因为这在概念上似乎是不必要的。
此代码的问题是，我在运行期间看到不稳定/未定义的行为，例如在验证集评估期间网络中出现的NAN或INF。这似乎主要发生在验证集队列与仍在填充的培训集队列同时填充的情况下（因为在验证集评估期间培训队列是打开的）。例如，如果我在迭代0中评估验证集（当两个队列仍然需要填充时），那么这种情况很常见。尽管培训/验证队列在不同的会话中运行，但它们似乎共享一些全局状态。
有人能解释为什么会发生这种情况，以及我如何在考虑到我的上述假设[A1]-[A5]的情况下更清晰地解决这一问题吗？

Answer 1

我现在也面临着类似的问题。到目前为止，我完全避免了任何队列，只是通过feed_dict输入数据，但是我显然因为不使用队列和并行性而失去了一些性能（尽管我仍然对当前的速度很满意，就像我之前在Theano中所做的一样）。现在我想重新设计这个并使用队列，然后偶然发现了这个问题。有一些相关的问题。
我目前正在考虑这样做：
在培训中，我想使用一个RandomShuffleQueue使其更加复杂。我想我会忽略这个问题，一旦把张量排成队列的读者线程结束，我会让训练停止，所以我会释放这个时期的剩余capacity项，并将其用于下一个时期。也许为了使它具有确定性，我签入了我仍然从队列中读取的列车线程，直到只剩下min_after_dequeue个项目为止。
在评估中，我想使用相同的图表和相同的会话。我可以使用tf.cond从另一个单独的队列中读取，而不是使用RandomShuffleQueue。或者我可以在评估中使用feed_dict。如果我使用一个单独的队列，我将使用一个FIFOQueue并仔细跟踪我所做的步骤的数量。我还可以在队列中引入另一个虚拟张量，它给我一个end_of_epoch标志，这样我就知道在eval线程中什么时候停止。
在TensorFlow 1.2中，将有tf.contrib.data接口（this、this、this），它提供了tf.contrib.data.DatasetAPI，该API还支持类似于tf.RandomShuffleQueue和批处理和循环多个时期的洗牌。另外，通过在数据上创建迭代器来访问数据，并且可以重置迭代器。一些相关的stackoverflow问题是issue comment和documentation overview。