Se usó la API de Cloud Translation para traducir esta página.
Switch to English

Búferes de reproducción

Ver en TensorFlow.org Ejecutar en Google Colab Ver fuente en GitHub Descargar cuaderno

Introducción

Los algoritmos de aprendizaje por refuerzo utilizan búferes de reproducción para almacenar trayectorias de experiencia al ejecutar una política en un entorno. Durante el entrenamiento, se consultan los búferes de reproducción para un subconjunto de las trayectorias (ya sea un subconjunto secuencial o una muestra) para "reproducir" la experiencia del agente.

En esta colab, exploramos dos tipos de búferes de reproducción: respaldados por python y respaldados por tensorflow, que comparten una API común. En las siguientes secciones, describimos la API, cada una de las implementaciones de búfer y cómo usarlas durante el entrenamiento de recolección de datos.

Preparar

Instale tf-agents si aún no lo ha hecho.

pip install -q tf-agents
pip install -q gym
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function

import tensorflow as tf
import numpy as np

from tf_agents import specs
from tf_agents.agents.dqn import dqn_agent
from tf_agents.drivers import dynamic_step_driver
from tf_agents.environments import suite_gym
from tf_agents.environments import tf_py_environment
from tf_agents.networks import q_network
from tf_agents.replay_buffers import py_uniform_replay_buffer
from tf_agents.replay_buffers import tf_uniform_replay_buffer
from tf_agents.specs import tensor_spec
from tf_agents.trajectories import time_step

tf.compat.v1.enable_v2_behavior()

API de búfer de reproducción

La clase Replay Buffer tiene la siguiente definición y métodos:

class ReplayBuffer(tf.Module):
  """Abstract base class for TF-Agents replay buffer."""

  def __init__(self, data_spec, capacity):
    """Initializes the replay buffer.

    Args:
      data_spec: A spec or a list/tuple/nest of specs describing
        a single item that can be stored in this buffer
      capacity: number of elements that the replay buffer can hold.
    """

  @property
  def data_spec(self):
    """Returns the spec for items in the replay buffer."""

  @property
  def capacity(self):
    """Returns the capacity of the replay buffer."""

  def add_batch(self, items):
    """Adds a batch of items to the replay buffer."""

  def get_next(self,
               sample_batch_size=None,
               num_steps=None,
               time_stacked=True):
    """Returns an item or batch of items from the buffer."""

  def as_dataset(self,
                 sample_batch_size=None,
                 num_steps=None,
                 num_parallel_calls=None):
    """Creates and returns a dataset that returns entries from the buffer."""


  def gather_all(self):
    """Returns all the items in buffer."""
    return self._gather_all()

  def clear(self):
    """Resets the contents of replay buffer"""

Tenga en cuenta que cuando se inicializa el objeto de búfer de reproducción, requiere la data_spec de data_spec de los elementos que almacenará. Esta especificación corresponde al TensorSpec de los elementos de trayectoria que se agregarán al búfer. Esta especificación generalmente se adquiere mirando el agent.collect_data_spec un agente, que define las formas, tipos y estructuras que espera el agente durante el entrenamiento (más sobre esto más adelante)

TFUniformReplayBuffer

TFUniformReplayBuffer es el búfer de reproducción más utilizado en TF-Agents, por lo que lo usaremos en nuestro tutorial aquí. En TFUniformReplayBuffer el almacenamiento del búfer de respaldo se realiza mediante variables de flujo tensorial y, por lo tanto, es parte del gráfico de cálculo.

El búfer almacena lotes de elementos y tiene una capacidad máxima de elementos max_length por segmento de lote. Por lo tanto, la capacidad total del búfer es batch_size x max_length elementos. Todos los elementos almacenados en el búfer deben tener una especificación de datos coincidente. Cuando el búfer de reproducción se utiliza para la recopilación de datos, la especificación es la especificación de recopilación de datos del agente.

Creando el búfer:

Para crear un TFUniformReplayBuffer pasamos:

  1. la especificación de los elementos de datos que almacenará el búfer
  2. el batch size del batch size correspondiente al tamaño del lote del búfer
  3. el número max_length de elementos por segmento de lote

Aquí hay un ejemplo de TFUniformReplayBuffer crear un TFUniformReplayBuffer con especificaciones de datos de muestra, batch_size 32 y max_length 1000.

data_spec =  (
        tf.TensorSpec([3], tf.float32, 'action'),
        (
            tf.TensorSpec([5], tf.float32, 'lidar'),
            tf.TensorSpec([3, 2], tf.float32, 'camera')
        )
)

batch_size = 32
max_length = 1000

replay_buffer = tf_uniform_replay_buffer.TFUniformReplayBuffer(
    data_spec,
    batch_size=batch_size,
    max_length=max_length)

Escribiendo en el búfer:

Para agregar elementos al búfer de reproducción, usamos el add_batch(items) donde items es una lista / tupla / nido de tensores que representan el lote de elementos que se agregarán al búfer. Cada elemento de los items debe tener una dimensión externa igual a batch_size y las dimensiones restantes deben cumplir con la especificación de datos del elemento (igual que las especificaciones de datos pasadas al constructor del búfer de reproducción).

Aquí tienes un ejemplo de cómo agregar un lote de artículos

action = tf.constant(1 * np.ones(
    data_spec[0].shape.as_list(), dtype=np.float32))
lidar = tf.constant(
    2 * np.ones(data_spec[1][0].shape.as_list(), dtype=np.float32))
camera = tf.constant(
    3 * np.ones(data_spec[1][1].shape.as_list(), dtype=np.float32))
  
values = (action, (lidar, camera))
values_batched = tf.nest.map_structure(lambda t: tf.stack([t] * batch_size),
                                       values)
  
replay_buffer.add_batch(values_batched)

Leyendo del búfer

Hay tres formas de leer datos de TFUniformReplayBuffer :

  1. get_next() : devuelve una muestra del búfer. El tamaño del lote de muestra y el número de pasos de tiempo devueltos se pueden especificar mediante argumentos a este método.
  2. as_dataset() : devuelve el búfer de reproducción como un tf.data.Dataset . Luego, se puede crear un iterador de conjunto de datos e iterar a través de las muestras de los elementos en el búfer.
  3. gather_all() : devuelve todos los elementos en el búfer como un tensor con forma [batch, time, data_spec]

A continuación se muestran ejemplos de cómo leer desde el búfer de reproducción utilizando cada uno de estos métodos:

# add more items to the buffer before reading
for _ in range(5):
  replay_buffer.add_batch(values_batched)

# Get one sample from the replay buffer with batch size 10 and 1 timestep:

sample = replay_buffer.get_next(sample_batch_size=10, num_steps=1)

# Convert the replay buffer to a tf.data.Dataset and iterate through it
dataset = replay_buffer.as_dataset(
    sample_batch_size=4,
    num_steps=2)

iterator = iter(dataset)
print("Iterator trajectories:")
trajectories = []
for _ in range(3):
  t, _ = next(iterator)
  trajectories.append(t)
  
print(tf.nest.map_structure(lambda t: t.shape, trajectories))

# Read all elements in the replay buffer:
trajectories = replay_buffer.gather_all()

print("Trajectories from gather all:")
print(tf.nest.map_structure(lambda t: t.shape, trajectories))

WARNING:tensorflow:From <ipython-input-6-1f9907631cb9>:7: ReplayBuffer.get_next (from tf_agents.replay_buffers.replay_buffer) is deprecated and will be removed in a future version.
Instructions for updating:
Use `as_dataset(..., single_deterministic_pass=False) instead.
Iterator trajectories:
[(TensorShape([4, 2, 3]), (TensorShape([4, 2, 5]), TensorShape([4, 2, 3, 2]))), (TensorShape([4, 2, 3]), (TensorShape([4, 2, 5]), TensorShape([4, 2, 3, 2]))), (TensorShape([4, 2, 3]), (TensorShape([4, 2, 5]), TensorShape([4, 2, 3, 2])))]
WARNING:tensorflow:From <ipython-input-6-1f9907631cb9>:24: ReplayBuffer.gather_all (from tf_agents.replay_buffers.replay_buffer) is deprecated and will be removed in a future version.
Instructions for updating:
Use `as_dataset(..., single_deterministic_pass=True)` instead.
Trajectories from gather all:
(TensorShape([32, 6, 3]), (TensorShape([32, 6, 5]), TensorShape([32, 6, 3, 2])))

PyUniformReplayBuffer

PyUniformReplayBuffer tiene la misma funcionalidad que TFUniformReplayBuffer pero en lugar de variables tf, sus datos se almacenan en matrices numerosas. Este búfer se puede utilizar para la recopilación de datos fuera del gráfico. Tener el almacenamiento de respaldo en numpy puede facilitar que algunas aplicaciones realicen la manipulación de datos (como la indexación para actualizar las prioridades) sin usar variables de Tensorflow. Sin embargo, esta implementación no tendrá el beneficio de las optimizaciones de gráficos con Tensorflow.

A continuación, se muestra un ejemplo de PyUniformReplayBuffer una instancia de PyUniformReplayBuffer partir de las especificaciones de trayectoria de la política del agente:

replay_buffer_capacity = 1000*32 # same capacity as the TFUniformReplayBuffer

py_replay_buffer = py_uniform_replay_buffer.PyUniformReplayBuffer(
    capacity=replay_buffer_capacity,
    data_spec=tensor_spec.to_nest_array_spec(data_spec))

Usar búferes de reproducción durante el entrenamiento

Ahora que sabemos cómo crear un búfer de reproducción, escribir elementos en él y leer de él, podemos usarlo para almacenar trayectorias durante el entrenamiento de nuestros agentes.

Recopilación de datos

Primero, veamos cómo usar el búfer de reproducción durante la recolección de datos.

En TF-Agents usamos un Driver (consulte el tutorial del controlador para obtener más detalles) para recopilar experiencia en un entorno. Para usar un Driver , especificamos un Observer que es una función que el Driver debe ejecutar cuando recibe una trayectoria.

Por lo tanto, para agregar elementos de trayectoria al búfer de reproducción, agregamos un observador que llama a add_batch(items) para agregar un (lote de) elementos en el búfer de reproducción.

A continuación se muestra un ejemplo de esto con TFUniformReplayBuffer . Primero creamos un entorno, una red y un agente. Luego creamos un TFUniformReplayBuffer . Tenga en cuenta que las especificaciones de los elementos de trayectoria en el búfer de reproducción son iguales a las especificaciones de recopilación de datos del agente. Luego configuramos su método add_batch como el observador del controlador que recopilará los datos durante nuestro entrenamiento:

env = suite_gym.load('CartPole-v0')
tf_env = tf_py_environment.TFPyEnvironment(env)

q_net = q_network.QNetwork(
    tf_env.time_step_spec().observation,
    tf_env.action_spec(),
    fc_layer_params=(100,))

agent = dqn_agent.DqnAgent(
    tf_env.time_step_spec(),
    tf_env.action_spec(),
    q_network=q_net,
    optimizer=tf.compat.v1.train.AdamOptimizer(0.001))

replay_buffer_capacity = 1000

replay_buffer = tf_uniform_replay_buffer.TFUniformReplayBuffer(
    agent.collect_data_spec,
    batch_size=tf_env.batch_size,
    max_length=replay_buffer_capacity)

# Add an observer that adds to the replay buffer:
replay_observer = [replay_buffer.add_batch]

collect_steps_per_iteration = 10
collect_op = dynamic_step_driver.DynamicStepDriver(
  tf_env,
  agent.collect_policy,
  observers=replay_observer,
  num_steps=collect_steps_per_iteration).run()
WARNING:tensorflow:From /tmpfs/src/tf_docs_env/lib/python3.6/site-packages/tf_agents/drivers/dynamic_step_driver.py:203: calling while_loop_v2 (from tensorflow.python.ops.control_flow_ops) with back_prop=False is deprecated and will be removed in a future version.
Instructions for updating:
back_prop=False is deprecated. Consider using tf.stop_gradient instead.
Instead of:
results = tf.while_loop(c, b, vars, back_prop=False)
Use:
results = tf.nest.map_structure(tf.stop_gradient, tf.while_loop(c, b, vars))

Leer datos para un paso de tren

Después de agregar elementos de trayectoria al búfer de reproducción, podemos leer lotes de trayectorias del búfer de reproducción para usarlos como datos de entrada para un paso de tren.

A continuación, se muestra un ejemplo de cómo entrenar en trayectorias desde el búfer de reproducción en un ciclo de entrenamiento:

# Read the replay buffer as a Dataset,
# read batches of 4 elements, each with 2 timesteps:
dataset = replay_buffer.as_dataset(
    sample_batch_size=4,
    num_steps=2)

iterator = iter(dataset)

num_train_steps = 10

for _ in range(num_train_steps):
  trajectories, _ = next(iterator)
  loss = agent.train(experience=trajectories)

WARNING:tensorflow:From /tmpfs/src/tf_docs_env/lib/python3.6/site-packages/tensorflow/python/util/dispatch.py:201: calling foldr_v2 (from tensorflow.python.ops.functional_ops) with back_prop=False is deprecated and will be removed in a future version.
Instructions for updating:
back_prop=False is deprecated. Consider using tf.stop_gradient instead.
Instead of:
results = tf.foldr(fn, elems, back_prop=False)
Use:
results = tf.nest.map_structure(tf.stop_gradient, tf.foldr(fn, elems))