پخش مجدد بافرها

مشاهده در TensorFlow.org در Google Colab اجرا شود مشاهده منبع در GitHub دانلود دفترچه یادداشت

معرفی

الگوریتم‌های یادگیری تقویتی از بافرهای پخش مجدد برای ذخیره مسیرهای تجربه هنگام اجرای سیاست در یک محیط استفاده می‌کنند. در طول آموزش، بافرهای پخش مجدد برای زیرمجموعه ای از مسیرها (یا یک زیرمجموعه متوالی یا یک نمونه) جستجو می شوند تا تجربه عامل "بازپخش" شود.

در این colab، ما دو نوع بافر پخش مجدد را بررسی می‌کنیم: پشتیبان پایتون و پشتوانه تنسورفلو، که یک API مشترک دارند. در بخش‌های بعدی، API، هر یک از پیاده‌سازی‌های بافر و نحوه استفاده از آنها در طول آموزش جمع‌آوری داده‌ها را توضیح می‌دهیم.

برپایی

اگر قبلاً نصب نکرده اید، tf-agents را نصب کنید.

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

Replay Buffer API

کلاس Replay Buffer دارای تعریف و روش های زیر است:

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-نمایندگی است، بنابراین ما آن را در آموزش ما در اینجا استفاده خواهد کرد. در TFUniformReplayBuffer ذخیره سازی بافر حمایت توسط متغیرهای tensorflow انجام می شود و در نتیجه بخشی از نمودار محاسبه است.

فروشگاه بافر دسته از عناصر و دارای حداکثر ظرفیت max_length عناصر در بخش دسته ای. بنابراین، ظرفیت بافر کل است batch_size X max_length عناصر. عناصر ذخیره شده در بافر همگی باید دارای مشخصات داده منطبق باشند. هنگامی که بافر پخش مجدد برای جمع آوری داده ها استفاده می شود، مشخصات، مشخصات جمع آوری داده های عامل است.

ایجاد بافر:

برای ایجاد یک TFUniformReplayBuffer ما در عبور:

  1. مشخصات عناصر داده ای که بافر ذخیره می کند
  2. batch size مربوط به اندازه دسته ای از بافر
  3. max_length تعدادی از عناصر در بخش دسته ای

اینجا یک مثال از ایجاد یک است TFUniformReplayBuffer با مشخصات داده های نمونه، batch_size 32 و 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)

نوشتن در بافر:

برای اضافه کردن عناصر به بافر پخش، ما استفاده از 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 /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])))

PyUniformReplayBuffer

PyUniformReplayBuffer است functionaly همان TFUniformReplayBuffer اما به جای متغیرهای TF، اطلاعات آن در آرایه نامپای ذخیره می شود. این بافر می تواند برای جمع آوری داده های خارج از نمودار استفاده شود. داشتن حافظه پشتیبان در 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-نمایندگی ما با استفاده از 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()

خواندن داده ها برای یک پله قطار

پس از افزودن عناصر مسیر به بافر پخش مجدد، می‌توانیم دسته‌هایی از مسیرها را از بافر پخش مجدد بخوانیم تا به عنوان داده ورودی برای مرحله قطار استفاده کنیم.

در اینجا مثالی از نحوه تمرین روی مسیرها از بافر پخش مجدد در یک حلقه آموزشی آورده شده است:

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