מאגר חוזר

הצג באתר TensorFlow.org הפעל בגוגל קולאב צפה במקור ב-GitHub הורד מחברת

מבוא

אלגוריתמי למידה לחיזוק משתמשים ב-replay buffers כדי לאחסן מסלולי ניסיון בעת ​​ביצוע מדיניות בסביבה. במהלך האימון, מאגרי הפעלה חוזרים נשאלים עבור תת-קבוצה של המסלולים (או תת-קבוצה רציפה או מדגם) כדי "לשחק מחדש" את חוויית הסוכן.

בקולאב זה, אנו בוחנים שני סוגים של מאגרי הפעלה חוזרים: מגובים בפיתון ו-tensorflow, חולקים 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 רשימה / tuple / קן של טנסור המייצג את קבוצת פריטים יתווסף למאגר. כול רכיב של 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. ניתן להשתמש במאגר זה לאיסוף נתונים מחוץ לגרף. אחסון הגיבוי ב-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))

שימוש במאגרים חוזרים במהלך האימון

כעת, כשאנחנו יודעים ליצור חיץ חוזר, לכתוב לו פריטים ולקרוא ממנו, אנחנו יכולים להשתמש בו כדי לאחסן מסלולים במהלך אימון הסוכנים שלנו.

איסוף נתונים

ראשית, הבה נבחן כיצד להשתמש במאגר ההשמעה החוזר במהלך איסוף הנתונים.

In-סוכני TF נשתמש Driver (ראה הדרכת 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))