Bufory powtórek

Zobacz na TensorFlow.org Uruchom w Google Colab Wyświetl źródło na GitHub Pobierz notatnik

Wstęp

Algorytmy uczenia się ze wzmocnieniem wykorzystują bufory odtwarzania do przechowywania trajektorii doświadczenia podczas wykonywania polityki w środowisku. Podczas uczenia bufory powtórek są odpytywane o podzbiór trajektorii (albo sekwencyjny podzbiór lub próbkę), aby „odtworzyć” doświadczenie agenta.

W tej współpracy przyjrzymy się dwóm rodzajom buforów powtórek: opartym na Pythonie i opartym na tensorflow, które mają wspólny interfejs API. W kolejnych sekcjach opisujemy API, każdą z implementacji buforów i jak z nich korzystać podczas treningu zbierania danych.

Ustawiać

Zainstaluj tf-agents, jeśli jeszcze tego nie zrobiłeś.

pip install tf-agents

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

Interfejs API bufora powtórek

Klasa Replay Buffer ma następującą definicję i metody:

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

Należy zauważyć, że kiedy przedmiot bufor odtwarzania jest inicjowany, wymaga data_spec z elementów będzie w sklepie. Spec ten odpowiada w TensorSpec elementów trajektorii zostaną dodane do bufora. Ta specyfikacja jest zazwyczaj nabywane przez patrząc na agenta agent.collect_data_spec który określa kształtów, rodzajów i struktury oczekiwanych przez agenta podczas treningu (więcej o tym później).

TFUniformReplayBufor

TFUniformReplayBuffer jest najczęściej używany bufor powtórka w TF-pełnomocników, więc będziemy go używać w naszym poradniku tutaj. W TFUniformReplayBuffer zasobnik buforowy podłoże odbywa się przez zmienne tensorflow, a tym samym stanowi część wykresu obliczeniowego.

Bufor przechowuje partii elementów i ma maksymalną pojemność max_length elementów na segmencie wsadowego. A więc całkowita pojemność buforowa jest batch_size x max_length elementów. Wszystkie elementy przechowywane w buforze muszą mieć pasującą specyfikację danych. Gdy bufor odtwarzania jest używany do zbierania danych, specyfikacja jest specyfikacją zbierania danych agenta.

Tworzenie bufora:

Aby utworzyć TFUniformReplayBuffer mijamy w:

  1. specyfikacja elementów danych, które będą przechowywane w buforze
  2. batch size odpowiada wielkości wsadu bufora
  3. max_length liczba elementów na segmencie wsadowym

Oto przykład tworzenia TFUniformReplayBuffer z próbką danych, specyfikacje batch_size 32 i 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)

Zapis do bufora:

Aby dodać elementy do bufora powtórek używamy add_batch(items) sposób, w którym items jest lista / krotki / gniazdo tensorów reprezentujących partię przedmiotów należy doliczyć do bufora. Każdy element z items może mieć zewnętrzny wymiar równy batch_size a pozostałe wymiary muszą być zgodne ze specyfikacją danych elementu (tak samo, jak dane techniczne dane przekazywane do konstruktora buforu odtwarzającego).

Oto przykład dodawania partii przedmiotów

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)

Czytanie z bufora

Istnieją trzy sposoby, aby odczytać dane z TFUniformReplayBuffer :

  1. get_next() - zwraca się próbki z buforem. Przykładowy rozmiar partii i liczbę zwróconych kroków czasowych można określić za pomocą argumentów tej metody.
  2. as_dataset() - zwraca bufor powtórka jako tf.data.Dataset . Następnie można utworzyć iterator zestawu danych i iterować próbki elementów w buforze.
  3. gather_all() - zwraca wszystkie elementy znajdujące się w buforze jako tensora o kształcie [batch, time, data_spec]

Poniżej znajdują się przykłady, jak czytać z bufora powtórek przy użyciu każdej z tych metod:

# 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 /tmp/ipykernel_15476/1348928897.py: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.
WARNING:tensorflow:From /tmpfs/src/tf_docs_env/lib/python3.7/site-packages/tensorflow/python/data/experimental/ops/counter.py:66: scan (from tensorflow.python.data.experimental.ops.scan_ops) is deprecated and will be removed in a future version.
Instructions for updating:
Use `tf.data.Dataset.scan(...) 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 /tmp/ipykernel_15476/1348928897.py: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])))

PyUniformReplayBufor

PyUniformReplayBuffer ma taką samą funkcjonalnie jako TFUniformReplayBuffer ale zamiast zmiennych TF jego dane są przechowywane w NumPy tablic. Ten bufor może być używany do zbierania danych poza wykresem. Posiadanie magazynu zapasowego w numpy może ułatwić niektórym aplikacjom manipulowanie danymi (takie jak indeksowanie w celu aktualizacji priorytetów) bez używania zmiennych Tensorflow. Jednak ta implementacja nie przyniesie korzyści w postaci optymalizacji wykresów za pomocą Tensorflow.

Poniżej jest przykład tworzenia instancji PyUniformReplayBuffer ze specyfikacją trajektorii polityka agenta:

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

Korzystanie z buforów powtórek podczas treningu

Teraz, gdy wiemy, jak utworzyć bufor powtórek, zapisywać do niego elementy i odczytywać z niego, możemy go użyć do przechowywania trajektorii podczas szkolenia naszych agentów.

Zbieranie danych

Najpierw spójrzmy, jak używać bufora powtórek podczas zbierania danych.

W TF-pełnomocników używamy Driver (patrz poradnik kierowcy więcej szczegółów) gromadzenie doświadczeń w środowisku. Aby użyć Driver , możemy określić Observer , który jest funkcją dla Driver do realizacji po otrzymaniu trajektorię.

Dlatego, aby dodać elementy trajektorii do bufora powtórek, dodajemy obserwatora, że połączenia add_batch(items) , aby dodać partię przedmiotów na buforze powtórka.

Poniżej jest przykład tego z TFUniformReplayBuffer . Najpierw tworzymy środowisko, sieć i agenta. Następnie tworzymy TFUniformReplayBuffer . Zwróć uwagę, że specyfikacje elementów trajektorii w buforze odtwarzania są takie same jak specyfikacje zebranych danych agenta. Następnie ustawić jego add_batch metodę jako obserwatora dla kierowcy, który zrobi zbierać dane podczas naszych szkoleń:

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

Odczytywanie danych dla kroku pociągu

Po dodaniu elementów trajektorii do bufora powtórek, możemy odczytać partie trajektorii z bufora powtórek, aby użyć ich jako danych wejściowych dla kroku pociągu.

Oto przykład, jak trenować na trajektoriach z bufora powtórek w pętli treningowej:

# 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.7/site-packages/tensorflow/python/util/dispatch.py:206: 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))