tf.compat.v1.train.Supervisor

Помощник для обучения, который создаёт контрольные точки моделей и вычисляет сводки.

tf.compat.v1.train.Supervisor(
    graph=None,
    ready_op=USE_DEFAULT,
    ready_for_local_init_op=USE_DEFAULT,
    is_chief=True,
    init_op=USE_DEFAULT,
    init_feed_dict=None,
    local_init_op=USE_DEFAULT,
    logdir=None,
    summary_op=USE_DEFAULT,
    saver=USE_DEFAULT,
    global_step=USE_DEFAULT,
    save_summaries_secs=120,
    save_model_secs=600,
    recovery_wait_secs=30,
    stop_grace_secs=120,
    checkpoint_basename='model.ckpt',
    session_manager=None,
    summary_writer=USE_DEFAULT,
    init_fn=None,
    local_init_run_options=None
)

Этот класс устарел. Пожалуйста, используйте tf.compat.v1.train.MonitoredTrainingSession вместо него.

Supervisor — это небольшой оболочка вокруг Coordinator, Saver, и SessionManager, который обрабатывает общие потребности программ обучения TensorFlow.

Использование для одной программы

with tf.Graph().as_default():
  ...add operations to the graph...
  # Create a Supervisor that will checkpoint the model in '/tmp/mydir'.
  sv = Supervisor(logdir='/tmp/mydir')
  # Get a TensorFlow session managed by the supervisor.
  with sv.managed_session(FLAGS.master) as sess:
    # Use the session to train the graph.
    while not sv.should_stop():
      sess.run(<my_train_op>)

Внутри блока with sv.managed_session() все переменные в графе были инициализированы. Кроме того, несколько служб были запущены для создания контрольных точек модели и добавления сводок в журнал событий.

Если программа аварийно завершается и перезапускается, управляемая сессия автоматически повторно инициализирует переменные из последней контрольной точки.

Supervisor уведомляется о любых исключениях, сгенерированных одной из служб. После возникновения исключения, should_stop() возвращает True. В этом случае цикл обучения также должен быть остановлен. Вот почему цикл обучения должен проверять на наличие sv.should_stop().

Исключения, которые указывают на то, что входные данные для обучения исчерпаны, tf.errors.OutOfRangeError, также заставляют sv.should_stop() возвращать True, но не повторно поднимать их из блока with: они указывают на нормальное завершение.

Использование для нескольких реплик

Для обучения с репликами вы развертываете одну и ту же программу в Cluster. Одна из задач должна быть идентифицирована как главная: задача, которая обрабатывает инициализацию, контрольные точки, сводки и восстановление. Другие задачи зависят от главной задачи для этих служб.

Единственное изменение, которое необходимо внести в код программы для одной задачи, — указать, работает ли программа в качестве главной.

# Choose a task as the chief. This could be based on server_def.task_index,
# or job_def.name, or job_def.tasks. It's entirely up to the end user.
# But there can be only one *chief*.
is_chief = (server_def.task_index == 0)
server = tf.distribute.Server(server_def)

with tf.Graph().as_default():
  ...add operations to the graph...
  # Create a Supervisor that uses log directory on a shared file system.
  # Indicate if you are the 'chief'
  sv = Supervisor(logdir='/shared_directory/...', is_chief=is_chief)
  # Get a Session in a TensorFlow server on the cluster.
  with sv.managed_session(server.target) as sess:
    # Use the session to train the graph.
    while not sv.should_stop():
      sess.run(<my_train_op>)

В задаче главной Supervisor работает точно так же, как в первом примере выше. В других задачах sv.managed_session() ожидает, пока модель не будет инициализирована, прежде чем возвращать сессию коду обучения. Задачи, не являющиеся главными, зависят от главной задачи для инициализации модели.

Если одна из задач аварийно завершается и перезапускается, managed_session() проверяет, инициализирована ли модель. Если да, то она просто создаёт сессию и возвращает её коду обучения, который продолжает работу нормально. Если модель нужно инициализировать, главная задача позаботится об её повторной инициализации; другие задачи просто ожидают инициализации модели.

Примечание: Эта изменённая программа по-прежнему работает нормально как программа для одной задачи. Программа для одной задачи отмечает себя как главную.

Какой использовать строку master

Как на вашей машине, так и в кластере вы можете использовать следующие значения для флага --master:

• Указание '' запрашивает сессию в процессе, которая не использует RPC.

• Указание 'local' запрашивает сессию, которая использует RPC-интерфейс «Интерфейс мастера» для запуска программ TensorFlow. Подробнее см. tf.train.Server.create_local_server.

• Указание 'grpc://hostname:port' запрашивает сессию, которая использует интерфейс RPC к определённому хосту, а также позволяет мастеру в процессе получать доступ к удалённым рабочим узлам TensorFlow. Часто уместно передать server.target (для некоторого tf.distribute.Server с именем `server`).

Расширенное использование

Запуск дополнительных служб

managed_session() запускает службы «Контрольные точки» и «Сводки» (потоки). Если вам необходимо запустить больше служб, вы можете просто запустить их в блоке, контролируемом managed_session().

Пример: Запустить поток для печати потерь. Мы хотим, чтобы этот поток выполнялся каждые 60 секунд, поэтому мы запускаем его с sv.loop().

...
sv = Supervisor(logdir='/tmp/mydir')
with sv.managed_session(FLAGS.master) as sess:
  sv.loop(60, print_loss, (sess, ))
  while not sv.should_stop():
    sess.run(my_train_op)

Запуск меньшего количества служб

managed_session() запускает потоки «сводка» и «контрольные точки», которые используют необязательные summary_op и saver, переданные в конструктор, или созданные автоматически Supervisor. Если вы хотите запустить собственную логику сводок и создания контрольных точек, отключите эти службы, передав None в параметры summary_op и saver.

Пример: Создать сводки вручную каждые 100 шагов в главной задаче.

# Create a Supervisor with no automatic summaries.
sv = Supervisor(logdir='/tmp/mydir', is_chief=is_chief, summary_op=None)
# As summary_op was None, managed_session() does not start the
# summary thread.
with sv.managed_session(FLAGS.master) as sess:
  for step in range(1000000):
    if sv.should_stop():
      break
    if is_chief and step % 100 == 0:
      # Create the summary every 100 chief steps.
      sv.summary_computed(sess, sess.run(my_summary_op))
    else:
      # Train normally
      sess.run(my_train_op)

Настраиваемая инициализация модели

managed_session() поддерживает только инициализацию модели путём запуска init_op или восстановления из последней контрольной точки. Если у вас есть особые потребности в инициализации, см. способ указания local_init_op при создании Supervisor. Вы также можете использовать SessionManager напрямую для создания сессии и проверки возможности автоматической инициализации.

Методы

Loop

Просмотреть исходный код

Loop(
    timer_interval_secs, target, args=None, kwargs=None
)

Запускает LooperThread, который периодически вызывает функцию.

Если timer_interval_secs равно None, поток вызывает target(*args, **kwargs) многократно. В противном случае он вызывает его каждые timer_interval_secs секунды. Поток завершается при запросе остановки.

Запущенный поток добавляется в список потоков, управляемых надзирателем, поэтому его не нужно передавать методу stop().

PrepareSession

Просмотреть исходный код

PrepareSession(
    master='',
    config=None,
    wait_for_checkpoint=False,
    max_wait_secs=7200,
    start_standard_services=True
)

Убедитесь, что модель готова к использованию.

Создаёт сессию на 'master', восстанавливая или инициализируя модель по мере необходимости или ожидая готовности сессии. Если выполняется как главный и start_standard_service установлено в True, также вызовите менеджер сессии для запуска стандартных служб.

RequestStop

Просмотреть исходный код

RequestStop(
    ex=None
)

Запросить остановку потоков координатором.

См. Coordinator.request_stop().

ShouldStop

Просмотреть исходный код

ShouldStop()

Проверить, было ли указано координатору остановиться.

См. Coordinator.should_stop().

StartQueueRunners

Просмотреть исходный код

StartQueueRunners(
    sess, queue_runners=None
)

Запустить потоки для QueueRunners.

Обратите внимание, что исполнители очередей, собранные в ключе графа QUEUE_RUNNERS, уже автоматически запускаются при создании сессии с надзирателем, поэтому если у вас нет дополнительных исполнителей очередей для запуска, вам не нужно вызывать это явно.

Очереди несовместимы с eager выполнением. Для получения данных при включённом eager выполнении используйте API tf.data.

StartStandardServices

Просмотреть исходный код

StartStandardServices(
    sess
)

Запустить стандартные службы для 'sess'.

Это запускает службы в фоновом режиме. Запускаемые службы зависят от параметров конструктора и могут включать:

• Поток сводок, вычисляющий сводки каждые save_summaries_secs.
• Поток контрольных точек, сохраняющий модель каждые save_model_secs.
• Поток StepCounter, измеряющий время шага.

Stop

Просмотреть исходный код

Stop(
    threads=None, close_summary_writer=True, ignore_live_threads=False
)

Остановить службы и координатор.

Это не закрывает сессию.

StopOnException

Просмотреть исходный код

StopOnException()

Обработчик контекста для остановки надзирателя при возникновении исключения.

См. Coordinator.stop_on_exception().

SummaryComputed

Просмотреть исходный код

SummaryComputed(
    sess, summary, global_step=None
)

Указывает, что сводка была вычислена.

WaitForStop

Просмотреть исходный код

WaitForStop()

Ожидание остановки координатора.

loop

Просмотреть исходный код

loop(
    timer_interval_secs, target, args=None, kwargs=None
)

Запуск потока LooperThread, который периодически вызывает функцию.

Если timer_interval_secs равно None, поток вызывает target(*args, **kwargs) многократно. В противном случае он вызывает её каждые timer_interval_secs секунды. Поток завершается при запросе остановки.

Запущенный поток добавляется в список потоков, управляемых Supervisor, поэтому его не нужно передавать методу stop().

managed_session

Просмотреть исходный код

@contextlib.contextmanager
managed_session(
    master='',
    config=None,
    start_standard_services=True,
    close_summary_writer=True
)

Возвращает менеджер контекста для управляемой сессии.

Этот менеджер контекста создаёт и автоматически восстанавливает сессию. Он необязательно запускает стандартные службы, которые обрабатывают контрольные точки и сводки. Он отслеживает исключения, возникающие в блоке with или в службах, и останавливает Supervisor по мере необходимости.

Менеджер контекста обычно используется следующим образом:

def train():
  sv = tf.compat.v1.train.Supervisor(...)
  with sv.managed_session(<master>) as sess:
    for step in range(..):
      if sv.should_stop():
        break
      sess.run(<my training op>)
      ...do other things needed at each training step...

Исключение, возникшее в блоке with или в одном из потоков служб, снова возникает при выходе из блока. Это делается после остановки всех потоков и закрытия сессии. Например, исключение AbortedError, возникающее в случае прерывания одного из рабочих процессов в распределённой модели, снова возникает при выходе из блока.

Если вы хотите повторить цикл обучения в случае прерывания, вы можете сделать это следующим образом:

def main(...):
  while True
    try:
      train()
    except tf.errors.Aborted:
      pass

В качестве специального случая исключения, используемые для управления потоком, например, OutOfRangeError, которые сообщают о том, что очереди ввода исчерпаны, не возникают снова из блока with: они указывают на чистое завершение цикла обучения и считаются нормальным завершением.

prepare_or_wait_for_session

Просмотреть исходный код

prepare_or_wait_for_session(
    master='',
    config=None,
    wait_for_checkpoint=False,
    max_wait_secs=7200,
    start_standard_services=True
)

Убедитесь, что модель готова к использованию.

Создайте сессию на 'master', восстанавливая или инициализируя модель по мере необходимости, или подождите, пока сессия не будет готова. Если выполняется как главный и start_standard_service установлено в True, также вызовите менеджер сессии для запуска стандартных служб.

request_stop

Просмотреть исходный код

request_stop(
    ex=None
)

Запрос на остановку потоков координатором.

См. Coordinator.request_stop().

should_stop

Просмотреть исходный код

should_stop()

Проверка, запросил ли координатор остановку.

См. Coordinator.should_stop().

start_queue_runners

Просмотреть исходный код

start_queue_runners(
    sess, queue_runners=None
)

Запуск потоков для QueueRunners.

Обратите внимание, что исполнители очередей, собранные в ключе графа QUEUE_RUNNERS, автоматически запускаются при создании сессии с Supervisor, поэтому, если у вас нет не собранных исполнителей очередей для запуска, нет необходимости вызывать это явно.

Очереди не совместимы с выполнением eager. Для приема данных при включенном выполнении eager используйте API tf.data.

start_standard_services

Просмотреть исходный код

start_standard_services(
    sess
)

Запуск стандартных служб для 'sess'.

Это запускает службы в фоновом режиме. Запускаемые службы зависят от параметров конструктора и могут включать:

• Поток Summary, вычисляющий сводки каждые save_summaries_secs.
• Поток Checkpoint, сохраняющий модель каждые save_model_secs.
• Поток StepCounter, измеряющий время шага.

stop

Просмотр исходного кода

stop(
    threads=None, close_summary_writer=True, ignore_live_threads=False
)

Останавливает сервисы и координатор.

Это не закрывает сессию.

stop_on_exception

Просмотр исходного кода

stop_on_exception()

Обработчик контекста для остановки супервайзера при возникновении исключения.

См. Coordinator.stop_on_exception().

summary_computed

Просмотр исходного кода

summary_computed(
    sess, summary, global_step=None
)

Указывает, что была вычислена сводка.

wait_for_stop

Просмотр исходного кода

wait_for_stop()

Ожидание остановки координатора.