إعادة المخازن المؤقتة

عرض على TensorFlow.org تشغيل في Google Colab عرض المصدر على جيثب تحميل دفتر

مقدمة

تستخدم خوارزميات التعلم المعزز مخازن إعادة التشغيل المؤقتة لتخزين مسارات الخبرة عند تنفيذ سياسة في بيئة ما. أثناء التدريب ، يتم الاستعلام عن المخازن المؤقتة لإعادة التشغيل لمجموعة فرعية من المسارات (إما مجموعة فرعية متسلسلة أو عينة) "لإعادة تشغيل" تجربة الوكيل.

في هذا colab ، نستكشف نوعين من المخازن المؤقتة لإعادة التشغيل: المدعومة من Python والمدعومة بالتدفق المتدفق ، والتي تشترك في واجهة برمجة تطبيقات مشتركة. في الأقسام التالية ، نصف API ، كل من تطبيقات المخزن المؤقت وكيفية استخدامها أثناء التدريب على جمع البيانات.

اقامة

قم بتثبيت وكلاء tf إذا لم تكن قد قمت بذلك بالفعل.

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

إعادة تشغيل عازلة 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 س 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 ولكن بدلا من المتغيرات فريق العمل، يتم تخزين البيانات في صفائف نمباي. يمكن استخدام هذا المخزن المؤقت لجمع البيانات خارج الرسم البياني. قد يؤدي وجود تخزين النسخ الاحتياطي في numpy إلى تسهيل قيام بعض التطبيقات بمعالجة البيانات (مثل الفهرسة لتحديث الأولويات) دون استخدام متغيرات Tensorflow. ومع ذلك ، لن يستفيد هذا التنفيذ من تحسينات الرسم البياني باستخدام Tensorflow.

وفيما يلي مثال على instantiating ل 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))