ترجمت واجهة Cloud Translation API‏ هذه الصفحة.
Switch to English

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

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

المقدمة

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

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

اقامة

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

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

تحتوي فئة 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-Agents ، وبالتالي TFUniformReplayBuffer في برنامجنا التعليمي هنا. في 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 عبارة عن قائمة / مجموعة / عش من الموترات التي تمثل مجموعة العناصر المراد إضافتها إلى المخزن المؤقت. يجب أن يكون لكل عنصر من 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() - إرجاع جميع العناصر الموجودة في المخزن المؤقت على شكل Tensor بالشكل [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 له نفس الوظيفة التي يتمتع بها 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-Agents ، نستخدم 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))