หน้านี้ได้รับการแปลโดย Cloud Translation API
Switch to English

เล่นซ้ำบัฟเฟอร์

ดูใน TensorFlow.org เรียกใช้ใน Google Colab ดูแหล่งที่มาบน GitHub ดาวน์โหลดสมุดบันทึก

บทนำ

อัลกอริธึมการเรียนรู้การเสริมกำลังใช้บัฟเฟอร์การเล่นซ้ำเพื่อจัดเก็บวิถีแห่งประสบการณ์เมื่อดำเนินนโยบายในสภาพแวดล้อม ในระหว่างการฝึกอบรมบัฟเฟอร์การเล่นซ้ำจะถูกสอบถามสำหรับชุดย่อยของวิถี (ไม่ว่าจะเป็นชุดย่อยตามลำดับหรือกลุ่มตัวอย่าง) เพื่อ "เล่นซ้ำ" ประสบการณ์ของตัวแทน

ใน colab นี้เราจะสำรวจบัฟเฟอร์การเล่นซ้ำสองประเภท ได้แก่ python-backed และ tensorflow-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()

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-Agents ดังนั้นเราจะใช้มันในบทช่วยสอนของเราที่นี่ ใน 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 คือ list / tuple / nest ของ tensors ที่แสดงถึงชุดของไอเท็มที่จะเพิ่มลงในบัฟเฟอร์ แต่ละองค์ประกอบของ 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]

ด้านล่างนี้เป็นตัวอย่างวิธีอ่านจาก replay buffer โดยใช้แต่ละวิธีต่อไปนี้:

# 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 ข้อมูลจะถูกเก็บไว้ในอาร์เรย์จำนวนนับ บัฟเฟอร์นี้สามารถใช้สำหรับการรวบรวมข้อมูลที่ไม่อยู่ในกราฟ การมีพื้นที่จัดเก็บข้อมูลสำรองเป็นตัวเลขอาจทำให้บางแอปพลิเคชันสามารถจัดการข้อมูลได้ง่ายขึ้น (เช่นการจัดทำดัชนีเพื่ออัปเดตลำดับความสำคัญ) โดยไม่ต้องใช้ตัวแปร 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))

ใช้รีเพลย์บัฟเฟอร์ระหว่างการฝึก

ตอนนี้เรารู้วิธีสร้างบัฟเฟอร์รีเพลย์เขียนรายการและอ่านจากมันเราสามารถใช้มันเพื่อเก็บวิถีระหว่างการฝึกอบรมตัวแทนของเรา

การเก็บรวบรวมข้อมูล

ก่อนอื่นเรามาดูวิธีใช้ replay buffer ระหว่างการรวบรวมข้อมูล

ใน TF-Agents เราใช้ Driver (ดูรายละเอียดเพิ่มเติมในบทช่วยสอนไดร์เวอร์) เพื่อรวบรวมประสบการณ์ในสภาพแวดล้อม ในการใช้ Driver เราระบุ Observer ซึ่งเป็นฟังก์ชันสำหรับ Driver จะดำเนินการเมื่อได้รับวิถี

ดังนั้นในการเพิ่มองค์ประกอบวิถีให้กับบัฟเฟอร์การเล่นซ้ำเราจึงเพิ่มผู้สังเกตการณ์ที่เรียก add_batch(items) เพื่อเพิ่ม (batch of) รายการใน replay buffer

ด้านล่างนี้คือตัวอย่างของสิ่งนี้กับ 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))