python - train - TensorFlow: ¿Cómo puedo evaluar una cola de datos de validación varias veces durante el entrenamiento?

tensorflow tutorial pdf

Question

tl; dr

¿Cómo puedo evaluar un conjunto de validación después de cada K iteraciones de entrenamiento, utilizando colas separadas para los datos de capacitación y validación, sin recurrir a tf.Sessions separadas en múltiples procesos? No parece haber una manera limpia de lograr esto, dado mi problema particular, y mi solución actual (que pensé que funcionaría) me da un comportamiento indefinido. ¡Ayuda!

La historia completa

Quiero evaluar un conjunto de validación en cada iteración de entrenamiento K, y no puedo descubrir cómo implementar esto correctamente en TensorFlow. Esta debería ser una de las operaciones más comunes, sin embargo, parece que la arquitectura / API de TensorFlow está trabajando en mi contra aquí o al menos hace las cosas innecesariamente difíciles.

Mis suposiciones son:

• [A1] El modelo de proceso múltiple para entrenamiento / validación como se describe aquí https://www.tensorflow.org/how_tos/reading_data/#multiple_input_pipelines no es aplicable a mi problema, ya que tengo que asumir que no hay suficiente memoria de GPU disponible para cargar las variables dos veces.
• [A2] Quiero evaluar el conjunto de validación de cada K iteraciones de entrenamiento.
• [A3] Los datos de entrenamiento y validación no se pueden leer simplemente desde el disco, sino que se generan sobre la marcha. Esto hace que sea imposible pre-calcular de manera confiable el tamaño del conjunto de validación por adelantado.
• [A4] El conjunto de validación es demasiado grande para calcularlo previamente y almacenarlo en el disco.
• [A5] El tamaño efectivo del conjunto de validación no es necesariamente un múltiplo del tamaño del lote.

El canal de entrada de entrenamiento se configura de la siguiente manera:

• Un tf.train.slice_input_producer() genera una lista tf.train.slice_input_producer() de nombres de archivos, cada uno de los cuales se refiere a datos de entrada sin procesar.
• Una función de generación de datos personalizada genera un número variable de ejemplares / etiquetas de entrenamiento de cada fragmento de datos de entrada sin procesar.
• Los ejemplares / etiquetas de entrenamiento generados se ponen en cola a través de tf.train.shuffle_batch() antes de ser alimentados a la red.

Debido a [A3], [A4], [A5], el canal de entrada de validación se configura de una forma casi idéntica, excepto que la cola de entrada final se genera a través de tf.train.batch() , ya que no es conveniente la tf.train.batch() . Debido a las suposiciones anteriores, un enfoque basado en feed_dict también es inviable, y también parece ser incompatible con el uso de una función de nivel superior como tf.train.batch .

Sin embargo, una implementación sencilla que utiliza dos conjuntos diferentes de colas para la capacitación y la validación no funciona. Por lo que entiendo, tengo dos opciones:

• [B1] Establezca el argumento num_epochs de la validación tf.train.slice_input_producer en None .

  En este caso, el conjunto de validación se realiza un ciclo interminable, pero necesitaría saber el tamaño del conjunto de validación por adelantado para limitar explícitamente el número de lotes para evaluar por ejecución a través del conjunto de validación. Además, si el tamaño del conjunto de validación no es divisible por el tamaño del lote, siempre extraeré un poco más en el último lote. Como esto cambiaría el orden de evaluación de los datos de validación cada vez, esto no es aceptable.

• [B2] Establezca el argumento num_epochs de la validación tf.train.slice_input_producer en 1 , y además configure el argumento allow_smaller_final_batch de la función tf.train.batch en True .

  En este caso, el conjunto de validación se realiza un ciclo exactamente una vez, después de lo cual la cola respectiva se cierra para siempre. De forma predeterminada, esto hará que la evaluación del conjunto de validación sea dos o más veces imposible. Como no conozco una buena forma de reabrir una cola en TensorFlow, tengo que solucionar esta limitación.

Debido a las mayores limitaciones de la opción [B1], opté por solucionar los problemas de la opción [B2]. El código (pseudo) que describe mi enfoque actual es el siguiente:

El bucle de entrenamiento debe ser bastante canónico. Cada iteraciones K, se llama una función para evaluar el conjunto de validación. Tenga en cuenta que solo comienzo las colas que tienen un nombre que comienza con "train_"; esta es la cola configurada para recopilar los datos de entrenamiento generados. Para hacer esto, creé dos funciones auxiliares, get_queues_by_name y start_queue_runners .

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)

Las funciones de ayuda se ven así:

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

En la función run_validation , mi solución alternativa elegida contra el problema de una cola cerrada es crear una nueva tf.Session . También solo comienzo los subprocesos asociados con la cola que recopila datos del conjunto de validación.

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

No sé si crear una nueva tf.Session aquí es una buena idea, pero parece ser la única forma de reiniciar la cola de validación. Idealmente, tampoco me gustaría volver a cargar la instantánea del modelo, ya que esto parece conceptualmente innecesario.

El problema con este código es que veo un comportamiento errático / indefinido durante la ejecución, como la aparición de NaN o Inf dentro de la red durante la evaluación del conjunto de validación. Esto parece ocurrir predominantemente cuando la cola del conjunto de validación se llena al mismo tiempo que la cola del conjunto de capacitación aún se está llenando (ya que la cola de capacitación está abierta durante la evaluación del conjunto de validación). Por ejemplo, esto sucede muy a menudo si evalúo el conjunto de validación en la iteración 0 (cuando aún deben completarse ambas colas). Casi parece que las colas de capacitación / validación comparten algún estado global, aunque se ejecutan en una sesión diferente.

¿Alguien puede explicar por qué sucede esto y cómo puedo resolverlo de manera más limpia al tomar en cuenta mis suposiciones anteriores [A1] - [A5]?

Answer 1

Actualmente estoy enfrentando un problema similar. Hasta ahora, evité las colas y simplemente feed_dict los datos a través de feed_dict pero obviamente estoy perdiendo algo de rendimiento al no usar colas y paralelismo (aunque todavía estoy contento con la velocidad actual como lo hice en Theano anteriormente ). Ahora quiero rediseñar esto y usar colas y me encontré con este problema. Hay this , this , this temas relacionados.

Actualmente estoy pensando en hacerlo de esta manera:

• En el entrenamiento, quiero usar un RandomShuffleQueue que lo hace aún más complicado. Creo que simplemente ignoraré el problema y una vez que el hilo del lector que pone en tensión los tensores en la cola finalice, dejaré que se detenga el entrenamiento, así que pierdo los elementos restantes de capacity para esta época y simplemente los uso para la próxima época. Tal vez para que sea determinista, compruebo el hilo del tren que todavía leo de la cola hasta que solo min_after_dequeue elementos min_after_dequeue .

• En la evaluación, quiero usar el mismo gráfico y la misma sesión. Puedo usar tf.cond para leer desde otra cola separada en lugar de RandomShuffleQueue . O podría usar feed_dict en la evaluación. Si usara una cola separada, usaría un FIFOQueue y realizaría un seguimiento cuidadoso de que hago la cantidad correcta de pasos. También podría introducir otro tensor ficticio que pongo en cola en la cola, lo que me da una end_of_epoch o algo así, por lo que sé en el hilo de evaluación cuándo detenerme.

En TensorFlow 1.2, estará la interfaz tf.contrib.data ( comentario del problema , descripción general de la documentación , documentación de la API ), que proporciona la API del conjunto de datos tf.contrib.data.Dataset que también admite la tf.RandomShuffleQueue similar a tf.RandomShuffleQueue y el tf.RandomShuffleQueue lotes y el bucle Múltiples épocas. Además, puede acceder a los datos creando un iterador sobre ellos y puede restablecer el iterador. Algunas preguntas relacionadas sobre StackOverflow están here y here .