本頁面由 Cloud Translation API 翻譯而成。
Switch to English

重播緩衝區

在TensorFlow.org上查看 在Google Colab中運行 在GitHub上查看源代碼 下載筆記本

介紹

強化學習算法使用重播緩衝區來存儲在環境中執行策略時的體驗軌跡。在訓練過程中,查詢軌蹟的子集(順序子集或樣本)的重播緩衝區,以“重播”代理的經驗。

在本次合作中,我們探索了兩種類型的重播緩衝區:python-backed和ten​​sorflow-backed,它們共享一個公共API。在以下各節中,我們將介紹API,每種緩衝區實現以及如何在數據收集培訓期間使用它們。

建立

如果尚未安裝tf-agent,請先安裝。

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

重播緩衝區類具有以下定義和方法:

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

請注意,在初始化重播緩衝區對象時,它需要將存儲的元素的data_spec 。此規範對應於將添加到緩衝區的軌跡元素的TensorSpec 。通常通過查看代理的agent.collect_data_spec來獲取此規範,該代理定義了培訓時代理期望的形狀,類型和結構(稍後會詳細介紹)

TFUniformReplayBuffer

TFUniformReplayBuffer是TF-Agent中最常用的重播緩衝區,因此我們將在本教程中使用它。在TFUniformReplayBuffer ,後備緩衝區存儲由tensorflow變量完成,因此是計算圖的一部分。

緩衝區存儲元素批次,每個批次段具有最大容量max_length元素。因此,總緩衝區容量為batch_size x max_length元素。緩衝區中存儲的元素必須全部具有匹配的數據規範。當將重放緩衝區用於數據收集時,該規範就是代理的收集數據規範。

創建緩衝區:

要創建TFUniformReplayBuffer我們傳入:

  1. 緩衝區將存儲的數據元素的規範
  2. 與緩衝區的batch size相對應的批量大小
  3. 每個批處理段的max_length元素max_length

這是一個使用示例數據規範( batch_size 32和max_length 1000)創建TFUniformReplayBuffer的示例。

0

寫入緩衝區:

要將元素添加到重播緩衝區,我們使用add_batch(items)方法,其中items是張量的列表/元組/嵌套的張量,表示要添加到緩衝區的一批商品。 items每個元素都必須具有等於batch_size的外部尺寸,其餘尺寸必須符合項目的數據規格(與傳遞給重放緩衝區構造函數的數據規格相同)。

這是添加一批物品的示例

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)

從緩衝區讀取

有三種方法可以從TFUniformReplayBuffer讀取數據:

  1. get_next() -從緩衝區返回一個樣本。可以通過此方法的參數指定樣本的批次大小和返回的時間步數。
  2. as_dataset() -將重播緩衝區作為tf.data.Dataset返回。然後可以創建一個數據集迭代器,並迭代緩衝區中各項的樣本。
  3. gather_all() -以形狀為[batch, time, data_spec]的張量返回緩衝區中的所有項目

下面是如何使用這些方法從重播緩衝區中進行讀取的示例:

# 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具有相同的functionaly作為TFUniformReplayBuffer但不是TF變量,它的數據存儲在numpy的陣列。該緩衝區可用於圖形外數據收集。將備份存儲空間設置為numpy可能會使某些應用程序更容易在不使用Tensorflow變量的情況下進行數據操作(例如為更新優先級建立索引)。但是,此實現不會具有使用Tensorflow進行圖形優化的好處。

以下是從代理的策略軌跡規範實例化PyUniformReplayBuffer的示例:

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

訓練期間使用重播緩衝區

現在,我們知道瞭如何創建重播緩衝區,向其中寫入項目並從中讀取內容,我們可以在訓練代理程序時使用它來存儲軌跡。

數據採集

首先,讓我們看一下如何在數據收集期間使用重放緩衝區。

在TF-Agent中,我們使用Driver (有關更多詳細信息,請參見驅動程序教程)來收集環境中的經驗。要使用Driver ,我們指定一個Observer ,它是Driver在收到軌跡時執行的功能。

因此,要將軌跡元素添加到重播緩衝區中,我們添加了一個調用add_batch(items)的觀察者,以在重播緩衝區中添加(批量)項。

以下是使用TFUniformReplayBuffer的示例。我們首先創建一個環境,​​一個網絡和一個代理。然後,我們創建一個TFUniformReplayBuffer 。請注意,重播緩衝區中的軌跡元素的規格與代理的收集數據規格相同。然後,將其add_batch方法設置為驅動程序的觀察者,該驅動程序將在培訓期間進行數據收集:

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

讀取火車步驟的數據

在將軌跡元素添加到重播緩衝區後,我們可以從重播緩衝區讀取一批軌跡,以用作訓練步驟的輸入數據。

這是一個如何在訓練循環中從重播緩衝區訓練軌蹟的示例:

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