Cette page a été traduite par l'API Cloud Translation.
Switch to English

Tampons de relecture

Voir sur TensorFlow.org Exécuter dans Google Colab Afficher la source sur GitHub Télécharger le cahier

introduction

Les algorithmes d'apprentissage par renforcement utilisent des tampons de relecture pour stocker les trajectoires de l'expérience lors de l'exécution d'une politique dans un environnement. Pendant l'entraînement, les tampons de relecture sont interrogés pour un sous-ensemble des trajectoires (soit un sous-ensemble séquentiel ou un échantillon) pour «rejouer» l'expérience de l'agent.

Dans cette colab, nous explorons deux types de tampons de relecture: soutenu par python et soutenu par tensorflow, partageant une API commune. Dans les sections suivantes, nous décrivons l'API, chacune des implémentations de tampon et comment les utiliser pendant la formation à la collecte de données.

Installer

Installez tf-agents si vous ne l'avez pas déjà fait.

pip install -q --pre tf-agents[reverb]
pip install -q gym
WARNING: You are using pip version 20.1.1; however, version 20.2 is available.
You should consider upgrading via the '/tmpfs/src/tf_docs_env/bin/python -m pip install --upgrade pip' command.
WARNING: You are using pip version 20.1.1; however, version 20.2 is available.
You should consider upgrading via the '/tmpfs/src/tf_docs_env/bin/python -m pip install --upgrade pip' command.

 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()
 

Replay Buffer API

La classe Replay Buffer a la définition et les méthodes suivantes:

 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"""

 

Notez que lorsque l'objet tampon de relecture est initialisé, il nécessite la data_spec des éléments qu'il stockera. Cette spécification correspond à la TensorSpec des éléments de trajectoire qui seront ajoutés au tampon. Cette spécification est généralement acquise en examinant l'agent agent.collect_data_spec qui définit les formes, les types et les structures attendus par l'agent lors de l'entraînement (plus à ce sujet plus tard)

TFUniformReplayBuffer

TFUniformReplayBuffer est le tampon de relecture le plus couramment utilisé dans TF-Agents, nous allons donc l'utiliser dans notre tutoriel ici. Dans TFUniformReplayBuffer le stockage du tampon de sauvegarde est effectué par des variables tensorflow et fait donc partie du graphe de calcul.

Le tampon stocke des lots d'éléments et a une capacité max_length éléments max_length par segment de lot. Ainsi, la capacité totale du tampon est de batch_size x éléments max_length . Les éléments stockés dans le tampon doivent tous avoir une spécification de données correspondante. Lorsque le tampon de relecture est utilisé pour la collecte de données, la spécification est la spécification de données de collecte de l'agent.

Création du tampon:

Pour créer un TFUniformReplayBuffer nous passons:

  1. la spécification des éléments de données que le tampon stockera
  2. la batch size du batch size correspondant à la taille du lot du tampon
  3. le nombre max_length d'éléments par segment de lot

Voici un exemple de création d'un TFUniformReplayBuffer avec des exemples de spécifications de données, batch_size 32 et 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)
 

Ecriture dans le tampon:

Pour ajouter des éléments au tampon de relecture, nous utilisons la add_batch(items)items est une liste / tuple / nid de tenseurs représentant le lot d'éléments à ajouter au tampon. Chaque élément des items doit avoir une dimension extérieure égale à batch_size et les dimensions restantes doivent adhérer aux spécifications de données de l'élément (identiques aux spécifications de données transmises au constructeur de tampon de relecture).

Voici un exemple d'ajout d'un lot d'articles

 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)
 

Lecture à partir du tampon

Il existe trois façons de lire les données du TFUniformReplayBuffer :

  1. get_next() - renvoie un échantillon du tampon. La taille du lot de l'échantillon et le nombre de pas de temps renvoyés peuvent être spécifiés via des arguments de cette méthode.
  2. as_dataset() - renvoie le tampon de relecture sous la forme d'un tf.data.Dataset . On peut alors créer un itérateur de jeu de données et parcourir les échantillons des éléments dans le tampon.
  3. gather_all() - retourne tous les éléments du tampon sous forme de Tensor avec forme [batch, time, data_spec]

Vous trouverez ci-dessous des exemples de lecture à partir du tampon de relecture à l'aide de chacune de ces méthodes:

 # 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 a la même fonctionnalité que TFUniformReplayBuffer mais au lieu de variables tf, ses données sont stockées dans des tableaux numpy. Ce tampon peut être utilisé pour la collecte de données hors graphique. Le fait d'avoir le stockage de sauvegarde dans numpy peut faciliter la manipulation des données par certaines applications (comme l'indexation pour la mise à jour des priorités) sans utiliser de variables Tensorflow. Cependant, cette implémentation n'aura pas l'avantage d'optimisations de graphes avec Tensorflow.

Voici un exemple d'instanciation d'un PyUniformReplayBuffer partir des spécifications de trajectoire de stratégie de l'agent:

 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))
 

Utilisation des tampons de relecture pendant l'entraînement

Maintenant que nous savons comment créer un tampon de relecture, y écrire des éléments et y lire, nous pouvons l'utiliser pour stocker les trajectoires lors de la formation de nos agents.

Collecte de données

Voyons d'abord comment utiliser le tampon de relecture lors de la collecte de données.

Dans TF-Agents, nous utilisons un Driver (voir le didacticiel du pilote pour plus de détails) pour collecter de l'expérience dans un environnement. Pour utiliser un Driver , nous spécifions un Observer qui est une fonction que le Driver doit exécuter lorsqu'il reçoit une trajectoire.

Ainsi, pour ajouter des éléments de trajectoire au tampon de relecture, nous ajoutons un observateur qui appelle add_batch(items) pour ajouter un (lot de) éléments sur le tampon de relecture.

Voici un exemple de ceci avec TFUniformReplayBuffer . Nous créons d'abord un environnement, un réseau et un agent. Ensuite, nous créons un TFUniformReplayBuffer . Notez que les spécifications des éléments de trajectoire dans le tampon de relecture sont égales à la spécification des données de collecte de l'agent. Nous définissons ensuite sa méthode add_batch tant qu'observateur du pilote qui effectuera la collecte des données lors de notre formation:

 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))

Lecture des données d'une étape de train

Après avoir ajouté des éléments de trajectoire au tampon de relecture, nous pouvons lire des lots de trajectoires à partir du tampon de relecture à utiliser comme données d'entrée pour une étape de train.

Voici un exemple de la façon de s'entraîner sur des trajectoires à partir du tampon de relecture dans une boucle d'entraînement:

 # 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)