이 페이지는 Cloud Translation API를 통해 번역되었습니다.
Switch to English

재생 버퍼

TensorFlow.org에서보기 Google Colab에서 실행 GitHub에서 소스보기 노트북 다운로드

소개

강화 학습 알고리즘은 환경에서 정책을 실행할 때 경험 궤적을 저장하기 위해 재생 버퍼를 사용합니다. 훈련 중에 에이전트의 경험을 "재생"하기 위해 궤적의 하위 집합 (순차 하위 집합 또는 샘플)에 대해 재생 버퍼를 쿼리합니다.

이 colab에서는 공통 API를 공유하는 python 지원 및 tensorflow 지원의 두 가지 유형의 재생 버퍼를 탐색합니다. 다음 섹션에서는 API, 각 버퍼 구현 및 데이터 수집 훈련 중에이를 사용하는 방법에 대해 설명합니다.

설정

아직 설치하지 않았다면 tf-agents를 설치하십시오.

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 을 확인하여 agent.collect_data_spec (나중에 자세히 설명).

TFUniformReplayBuffer

TFUniformReplayBuffer 는 TF-Agents에서 가장 일반적으로 사용되는 재생 버퍼이므로 여기 튜토리얼에서 사용할 것입니다. TFUniformReplayBuffer 에서 백업 버퍼 저장은 tensorflow 변수에 의해 수행되므로 계산 그래프의 일부입니다.

버퍼는 요소 배치를 저장하고 배치 세그먼트 당 최대 용량 max_length 요소를 갖습니다. 따라서 총 버퍼 용량은 batch_size x max_length 요소입니다. 버퍼에 저장된 요소는 모두 일치하는 데이터 사양을 가져야합니다. 재생 버퍼가 데이터 수집에 사용되는 경우 사양은 에이전트의 수집 데이터 사양입니다.

버퍼 만들기 :

TFUniformReplayBuffer 를 생성하려면 TFUniformReplayBuffer 을 전달합니다.

  1. 버퍼가 저장할 데이터 요소의 사양
  2. 버퍼의 batch size 해당하는 배치 크기
  3. 배치 세그먼트 당 요소의 max_length

다음은 샘플 데이터 사양 batch_size 32 및 max_length 1000으로 TFUniformReplayBuffer 를 생성하는 예입니다.

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] 모양의 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-1-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-1-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 같은 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))

훈련 중 재생 버퍼 사용

이제 리플레이 버퍼를 만들고, 항목을 쓰고, 읽는 방법을 알았으므로 에이전트를 훈련하는 동안 궤적을 저장하는 데 사용할 수 있습니다.

데이터 수집

먼저 데이터 수집 중에 재생 버퍼를 사용하는 방법을 살펴 보겠습니다.

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