환경

Copyright 2021 TF 에이전트 작성자.

TensorFlow.org에서 보기

Google Colab에서 실행

GitHub에서 소스 보기

노트북 다운로드

소개

강화 학습(RL)의 목표는 환경과 상호 작용하여 학습하는 에이전트를 설계하는 것입니다. 표준 RL 설정에서 에이전트는 모든 시간 단계에서 관찰을 수신하고 작업을 선택합니다. 작업이 환경에 적용되고 환경은 보상과 새로운 관찰을 반환합니다. 에이전트는 반환이라고도 하는 보상의 합계를 최대화하기 위한 조치를 선택하도록 정책을 훈련합니다.

TF-Agents에서 환경은 Python 또는 TensorFlow에서 구현될 수 있습니다. Python 환경은 일반적으로 구현, 이해 및 디버그하기가 더 쉽지만 TensorFlow 환경은 더 효율적이고 자연스러운 병렬화를 허용합니다. 가장 일반적인 워크플로는 Python으로 환경을 구현하고 래퍼 중 하나를 사용하여 자동으로 TensorFlow로 변환하는 것입니다.

먼저 Python 환경을 살펴보겠습니다. TensorFlow 환경은 매우 유사한 API를 따릅니다.

설정

아직 tf-agents 또는 gym을 설치하지 않았다면 다음을 실행하십시오:

pip install "gym>=0.21.0"
pip install tf-agents

from __future__ import absolute_import
from __future__ import division
from __future__ import print_function

import abc
import tensorflow as tf
import numpy as np

from tf_agents.environments import py_environment
from tf_agents.environments import tf_environment
from tf_agents.environments import tf_py_environment
from tf_agents.environments import utils
from tf_agents.specs import array_spec
from tf_agents.environments import wrappers
from tf_agents.environments import suite_gym
from tf_agents.trajectories import time_step as ts

파이썬 환경

파이썬 환경은이 step(action) -> next_time_step 환경에 조치를 적용하고, 다음 단계에 대한 다음 정보를 반환 방법 :

1. observation : 이것은 에이전트가 다음 단계의 작업을 선택하는 관찰 할 수있는 환경 상태의 일부입니다.
2. reward : 에이전트는 여러 단계에 걸쳐 이러한 보상의 합을 극대화하기 위해 배우고있다.
3. step_type : 환경과의 상호 작용은 일반적으로 순서 / 에피소드의 일부입니다. 예를 들어 체스 게임에서 여러 움직임. step_type 하나가 될 수 FIRST , MID 혹은 LAST 이때 단계는 순차적으로, 제 1 중간 또는 마지막 단계인지 여부를 나타 내기 위해.
4. discount : 이것은 현재 시간 단계에서 보상에 다음 단계 상대의 보상을 가중하는 방법을 많이 나타내는 부동입니다.

이러한 명명 된 튜플로 그룹화됩니다 TimeStep(step_type, reward, discount, observation) .

모든 파이썬 환경이 구현해야하는 인터페이스에 environments/py_environment.PyEnvironment . 주요 방법은 다음과 같습니다.

class PyEnvironment(object):

  def reset(self):
    """Return initial_time_step."""
    self._current_time_step = self._reset()
    return self._current_time_step

  def step(self, action):
    """Apply action and return new time_step."""
    if self._current_time_step is None:
        return self.reset()
    self._current_time_step = self._step(action)
    return self._current_time_step

  def current_time_step(self):
    return self._current_time_step

  def time_step_spec(self):
    """Return time_step_spec."""

  @abc.abstractmethod
  def observation_spec(self):
    """Return observation_spec."""

  @abc.abstractmethod
  def action_spec(self):
    """Return action_spec."""

  @abc.abstractmethod
  def _reset(self):
    """Return initial_time_step."""

  @abc.abstractmethod
  def _step(self, action):
    """Apply action and return new time_step."""

또한받는 step() 에있어서, 환경도 제공 reset() 새로운 시퀀스가 시작되고, 초기 제공에있어서 TimeStep . 호출 할 필요가 없습니다 reset 명시 적 방법을. 에피소드가 끝나거나 step()이 처음 호출될 때 환경이 자동으로 재설정된다고 가정합니다.

참고 서브 클래스가 구현하지 않는 step() 또는 reset() 직접. 그들은 대신 오버라이드 (override) _step() 및 _reset() 메소드. 이러한 방법에서 반환되는 시간 단계 캐시와 노출됩니다 current_time_step() .

observation_spec 과 action_spec 방법의 둥지 반환 (Bounded)ArraySpecs 각각 관찰과 행동의 이름, 모양, 데이터 유형 및 범위에 대해 설명합니다.

TF-Agents에서 우리는 목록, 튜플, 명명된 튜플 또는 사전으로 구성된 구조와 같은 트리로 정의되는 중첩을 반복적으로 참조합니다. 관찰과 행동의 구조를 유지하기 위해 임의로 구성할 수 있습니다. 우리는 이것이 많은 관찰과 행동이 있는 보다 복잡한 환경에 매우 유용하다는 것을 발견했습니다.

표준 환경 사용

TF 에이전트 내장 한 OpenAI 체육관, DeepMind 제어 아타리 같은 많은 표준 환경을위한 래퍼, 그들이 우리에 따라 그래서 py_environment.PyEnvironment 인터페이스를. 이러한 래핑된 환경은 환경 제품군을 사용하여 쉽게 로드할 수 있습니다. OpenAI 체육관에서 CartPole 환경을 로드하고 action과 time_step_spec을 살펴보겠습니다.

environment = suite_gym.load('CartPole-v0')
print('action_spec:', environment.action_spec())
print('time_step_spec.observation:', environment.time_step_spec().observation)
print('time_step_spec.step_type:', environment.time_step_spec().step_type)
print('time_step_spec.discount:', environment.time_step_spec().discount)
print('time_step_spec.reward:', environment.time_step_spec().reward)

action_spec: BoundedArraySpec(shape=(), dtype=dtype('int64'), name='action', minimum=0, maximum=1)
time_step_spec.observation: BoundedArraySpec(shape=(4,), dtype=dtype('float32'), name='observation', minimum=[-4.8000002e+00 -3.4028235e+38 -4.1887903e-01 -3.4028235e+38], maximum=[4.8000002e+00 3.4028235e+38 4.1887903e-01 3.4028235e+38])
time_step_spec.step_type: ArraySpec(shape=(), dtype=dtype('int32'), name='step_type')
time_step_spec.discount: BoundedArraySpec(shape=(), dtype=dtype('float32'), name='discount', minimum=0.0, maximum=1.0)
time_step_spec.reward: ArraySpec(shape=(), dtype=dtype('float32'), name='reward')

우리는 환경이 유형의 작업을 기대하고 있음을 볼 수 있도록 int64 에 [0, 1] 및 반환 TimeSteps 관찰이 있습니다 float32 길이 4 및 할인 요인의 벡터가있다 float32 [0.0, 1.0]입니다. 이제, 고정 작업을 수행하려고하자 (1,) 전체 에피소드를.

action = np.array(1, dtype=np.int32)
time_step = environment.reset()
print(time_step)
while not time_step.is_last():
  time_step = environment.step(action)
  print(time_step)

TimeStep(
{'discount': array(1., dtype=float32),
 'observation': array([ 0.0138565 , -0.03582913,  0.04861612, -0.03755046], dtype=float32),
 'reward': array(0., dtype=float32),
 'step_type': array(0, dtype=int32)})
TimeStep(
{'discount': array(1., dtype=float32),
 'observation': array([ 0.01313992,  0.15856317,  0.0478651 , -0.3145069 ], dtype=float32),
 'reward': array(1., dtype=float32),
 'step_type': array(1, dtype=int32)})
TimeStep(
{'discount': array(1., dtype=float32),
 'observation': array([ 0.01631118,  0.35297176,  0.04157497, -0.5917188 ], dtype=float32),
 'reward': array(1., dtype=float32),
 'step_type': array(1, dtype=int32)})
TimeStep(
{'discount': array(1., dtype=float32),
 'observation': array([ 0.02337062,  0.54748774,  0.02974059, -0.87102115], dtype=float32),
 'reward': array(1., dtype=float32),
 'step_type': array(1, dtype=int32)})
TimeStep(
{'discount': array(1., dtype=float32),
 'observation': array([ 0.03432037,  0.74219286,  0.01232017, -1.1542072 ], dtype=float32),
 'reward': array(1., dtype=float32),
 'step_type': array(1, dtype=int32)})
TimeStep(
{'discount': array(1., dtype=float32),
 'observation': array([ 0.04916423,  0.93715197, -0.01076398, -1.4430016 ], dtype=float32),
 'reward': array(1., dtype=float32),
 'step_type': array(1, dtype=int32)})
TimeStep(
{'discount': array(1., dtype=float32),
 'observation': array([ 0.06790727,  1.1324048 , -0.03962401, -1.7390285 ], dtype=float32),
 'reward': array(1., dtype=float32),
 'step_type': array(1, dtype=int32)})
TimeStep(
{'discount': array(1., dtype=float32),
 'observation': array([ 0.09055536,  1.327955  , -0.07440457, -2.04377   ], dtype=float32),
 'reward': array(1., dtype=float32),
 'step_type': array(1, dtype=int32)})
TimeStep(
{'discount': array(1., dtype=float32),
 'observation': array([ 0.11711447,  1.523758  , -0.11527998, -2.3585167 ], dtype=float32),
 'reward': array(1., dtype=float32),
 'step_type': array(1, dtype=int32)})
TimeStep(
{'discount': array(1., dtype=float32),
 'observation': array([ 0.14758962,  1.7197047 , -0.16245031, -2.6843033 ], dtype=float32),
 'reward': array(1., dtype=float32),
 'step_type': array(1, dtype=int32)})
TimeStep(
{'discount': array(0., dtype=float32),
 'observation': array([ 0.18198372,  1.9156038 , -0.21613638, -3.0218334 ], dtype=float32),
 'reward': array(1., dtype=float32),
 'step_type': array(2, dtype=int32)})

나만의 Python 환경 만들기

많은 클라이언트의 일반적인 사용 사례는 문제에 TF-Agents의 표준 에이전트(agents/ 참조) 중 하나를 적용하는 것입니다. 이렇게 하려면 문제를 환경으로 프레임화해야 합니다. 그럼 파이썬에서 환경을 구현하는 방법을 살펴보겠습니다.

에이전트가 다음(블랙 잭에서 영감을 받은) 카드 게임을 하도록 훈련시키고 싶다고 가정해 보겠습니다.

1. 이 게임은 1~10번의 무한 카드 덱을 사용하여 진행됩니다.
2. 매 턴마다 에이전트는 2가지 일을 할 수 있습니다. 새로운 무작위 카드를 얻거나 현재 라운드를 중지합니다.
3. 목표는 라운드가 끝날 때 진행하지 않고 가능한 한 21에 가까운 카드 합계를 얻는 것입니다.

게임을 나타내는 환경은 다음과 같습니다.

1. 작업: 2개의 작업이 있습니다. 액션 0: 새 카드를 받고, 액션 1: 현재 라운드를 종료합니다.
2. 관찰: 현재 라운드에 있는 카드의 합계입니다.
3. 보상: 목표는 진행하지 않고 가능한 한 21에 가까워지는 것이므로 라운드가 끝날 때 다음 보상을 사용하여 이를 달성할 수 있습니다. sum_of_cards - 21 if sum_of_cards <= 21, else -21

class CardGameEnv(py_environment.PyEnvironment):

  def __init__(self):
    self._action_spec = array_spec.BoundedArraySpec(
        shape=(), dtype=np.int32, minimum=0, maximum=1, name='action')
    self._observation_spec = array_spec.BoundedArraySpec(
        shape=(1,), dtype=np.int32, minimum=0, name='observation')
    self._state = 0
    self._episode_ended = False

  def action_spec(self):
    return self._action_spec

  def observation_spec(self):
    return self._observation_spec

  def _reset(self):
    self._state = 0
    self._episode_ended = False
    return ts.restart(np.array([self._state], dtype=np.int32))

  def _step(self, action):

    if self._episode_ended:
      # The last action ended the episode. Ignore the current action and start
      # a new episode.
      return self.reset()

    # Make sure episodes don't go on forever.
    if action == 1:
      self._episode_ended = True
    elif action == 0:
      new_card = np.random.randint(1, 11)
      self._state += new_card
    else:
      raise ValueError('`action` should be 0 or 1.')

    if self._episode_ended or self._state >= 21:
      reward = self._state - 21 if self._state <= 21 else -21
      return ts.termination(np.array([self._state], dtype=np.int32), reward)
    else:
      return ts.transition(
          np.array([self._state], dtype=np.int32), reward=0.0, discount=1.0)

위의 환경을 정의하는 모든 작업을 올바르게 수행했는지 확인합시다. 자신의 환경을 만들 때 생성된 관찰 및 time_steps가 사양에 정의된 올바른 모양과 유형을 따르는지 확인해야 합니다. 이것들은 TensorFlow 그래프를 생성하는 데 사용되며, 따라서 문제가 잘못되면 디버그하기 어려운 문제를 생성할 수 있습니다.

환경을 검증하기 위해 무작위 정책을 사용하여 작업을 생성하고 5개의 에피소드를 반복하여 의도한 대로 작동하는지 확인합니다. 환경 사양을 따르지 않는 time_step을 수신하면 오류가 발생합니다.

environment = CardGameEnv()
utils.validate_py_environment(environment, episodes=5)

이제 환경이 의도한 대로 작동한다는 것을 알았으므로 고정 정책을 사용하여 이 환경을 실행해 보겠습니다. 3장의 카드를 요청한 다음 라운드를 종료합니다.

get_new_card_action = np.array(0, dtype=np.int32)
end_round_action = np.array(1, dtype=np.int32)

environment = CardGameEnv()
time_step = environment.reset()
print(time_step)
cumulative_reward = time_step.reward

for _ in range(3):
  time_step = environment.step(get_new_card_action)
  print(time_step)
  cumulative_reward += time_step.reward

time_step = environment.step(end_round_action)
print(time_step)
cumulative_reward += time_step.reward
print('Final Reward = ', cumulative_reward)

TimeStep(
{'discount': array(1., dtype=float32),
 'observation': array([0], dtype=int32),
 'reward': array(0., dtype=float32),
 'step_type': array(0, dtype=int32)})
TimeStep(
{'discount': array(1., dtype=float32),
 'observation': array([9], dtype=int32),
 'reward': array(0., dtype=float32),
 'step_type': array(1, dtype=int32)})
TimeStep(
{'discount': array(1., dtype=float32),
 'observation': array([12], dtype=int32),
 'reward': array(0., dtype=float32),
 'step_type': array(1, dtype=int32)})
TimeStep(
{'discount': array(0., dtype=float32),
 'observation': array([21], dtype=int32),
 'reward': array(0., dtype=float32),
 'step_type': array(2, dtype=int32)})
TimeStep(
{'discount': array(0., dtype=float32),
 'observation': array([21], dtype=int32),
 'reward': array(0., dtype=float32),
 'step_type': array(2, dtype=int32)})
Final Reward =  0.0

환경 래퍼

환경 래퍼는 Python 환경을 사용하고 환경의 수정된 버전을 반환합니다. 원래의 환경과 수정 된 환경 모두의 인스턴스 py_environment.PyEnvironment , 여러 래퍼는 서로 연결 할 수 있습니다.

일반적인 래퍼는에서 찾을 수 있습니다 environments/wrappers.py . 예를 들어:

1. ActionDiscretizeWrapper : 변환 개별 활동 공간에 연속 동작 공간.
2. RunStats : 캡처 등의 스텝 수로 환경의 통계를 실행, 에피소드의 수는 완료 등
3. TimeLimit : 단계의 고정 된 수의 후 에피소드를 종료합니다.

예 1: 작업 이산화 래퍼

InvertedPendulum의 범위에서 연속 동작 받아들이 PyBullet 환경이다 [-2, 2] . 이 환경에서 DQN과 같은 이산 액션 에이전트를 훈련하려면 액션 공간을 이산화(양자화)해야 합니다. 이 정확히 무엇 ActionDiscretizeWrapper 않습니다. 비교] action_spec 전 포장 후 :

env = suite_gym.load('Pendulum-v1')
print('Action Spec:', env.action_spec())

discrete_action_env = wrappers.ActionDiscretizeWrapper(env, num_actions=5)
print('Discretized Action Spec:', discrete_action_env.action_spec())

Action Spec: BoundedArraySpec(shape=(1,), dtype=dtype('float32'), name='action', minimum=-2.0, maximum=2.0)
Discretized Action Spec: BoundedArraySpec(shape=(), dtype=dtype('int32'), name='action', minimum=0, maximum=4)

랩 discrete_action_env 의 인스턴스 py_environment.PyEnvironment 하고 정기적 파이썬 환경처럼 취급 될 수있다.

TensorFlow 환경

TF 환경의 인터페이스에 정의되어 environments/tf_environment.TFEnvironment 및 파이썬 환경과 매우 비슷합니다. TF 환경은 몇 가지 면에서 Python 환경과 다릅니다.

• 배열 대신 텐서 객체를 생성합니다.
• TF 환경은 사양과 비교할 때 생성된 텐서에 일괄 처리 차원을 추가합니다.

Python 환경을 TFEnvs로 변환하면 tensorflow가 작업을 병렬화할 수 있습니다. 예를 들어, 하나는 정의 할 수 collect_experience_op 그 수집하고 환경에서 데이터와에 추가 replay_buffer 및 train_op 로부터 읽어 replay_buffer 및 에이전트를 훈련하고 TensorFlow에서 자연적으로 병렬로 실행합니다.

class TFEnvironment(object):

  def time_step_spec(self):
    """Describes the `TimeStep` tensors returned by `step()`."""

  def observation_spec(self):
    """Defines the `TensorSpec` of observations provided by the environment."""

  def action_spec(self):
    """Describes the TensorSpecs of the action expected by `step(action)`."""

  def reset(self):
    """Returns the current `TimeStep` after resetting the Environment."""
    return self._reset()

  def current_time_step(self):
    """Returns the current `TimeStep`."""
    return self._current_time_step()

  def step(self, action):
    """Applies the action and returns the new `TimeStep`."""
    return self._step(action)

  @abc.abstractmethod
  def _reset(self):
    """Returns the current `TimeStep` after resetting the Environment."""

  @abc.abstractmethod
  def _current_time_step(self):
    """Returns the current `TimeStep`."""

  @abc.abstractmethod
  def _step(self, action):
    """Applies the action and returns the new `TimeStep`."""

current_time_step() 메서드는 현재 time_step를 반환하고 필요한 경우 환경을 초기화합니다.

reset() 메소드 세력 환경 및 반환 CURRENT_STEP에서 리셋.

는 IF action 이전에 의존하지 않는 time_step tf.control_dependency 필요하다 Graph 모드.

지금은 우리가 방법을 살펴 보자 TFEnvironments 생성됩니다.

자신만의 TensorFlow 환경 만들기

이것은 Python에서 환경을 만드는 것보다 더 복잡하므로 이 colab에서 다루지 않습니다. 예 사용할 수 있습니다 여기에 . 일반적인 사용 사례는 파이썬 환경을 구현하고 우리의 사용 TensorFlow에 포장하는 것입니다 TFPyEnvironment 래퍼 (아래 참조).

TensorFlow에서 Python 환경 래핑

우리는 쉽게 사용 TensorFlow 환경에 파이썬 환경을 래핑 할 수 TFPyEnvironment 래퍼.

env = suite_gym.load('CartPole-v0')
tf_env = tf_py_environment.TFPyEnvironment(env)

print(isinstance(tf_env, tf_environment.TFEnvironment))
print("TimeStep Specs:", tf_env.time_step_spec())
print("Action Specs:", tf_env.action_spec())

True
TimeStep Specs: TimeStep(
{'discount': BoundedTensorSpec(shape=(), dtype=tf.float32, name='discount', minimum=array(0., dtype=float32), maximum=array(1., dtype=float32)),
 'observation': BoundedTensorSpec(shape=(4,), dtype=tf.float32, name='observation', minimum=array([-4.8000002e+00, -3.4028235e+38, -4.1887903e-01, -3.4028235e+38],
      dtype=float32), maximum=array([4.8000002e+00, 3.4028235e+38, 4.1887903e-01, 3.4028235e+38],
      dtype=float32)),
 'reward': TensorSpec(shape=(), dtype=tf.float32, name='reward'),
 'step_type': TensorSpec(shape=(), dtype=tf.int32, name='step_type')})
Action Specs: BoundedTensorSpec(shape=(), dtype=tf.int64, name='action', minimum=array(0), maximum=array(1))

: 사양 유형의 지금 주 (Bounded)TensorSpec .

사용 예

간단한 예

env = suite_gym.load('CartPole-v0')

tf_env = tf_py_environment.TFPyEnvironment(env)
# reset() creates the initial time_step after resetting the environment.
time_step = tf_env.reset()
num_steps = 3
transitions = []
reward = 0
for i in range(num_steps):
  action = tf.constant([i % 2])
  # applies the action and returns the new TimeStep.
  next_time_step = tf_env.step(action)
  transitions.append([time_step, action, next_time_step])
  reward += next_time_step.reward
  time_step = next_time_step

np_transitions = tf.nest.map_structure(lambda x: x.numpy(), transitions)
print('\n'.join(map(str, np_transitions)))
print('Total reward:', reward.numpy())

[TimeStep(
{'discount': array([1.], dtype=float32),
 'observation': array([[-0.0078796 , -0.04736348, -0.04966116,  0.04563603]],
      dtype=float32),
 'reward': array([0.], dtype=float32),
 'step_type': array([0], dtype=int32)}), array([0], dtype=int32), TimeStep(
{'discount': array([1.], dtype=float32),
 'observation': array([[-0.00882687, -0.24173944, -0.04874843,  0.32224613]],
      dtype=float32),
 'reward': array([1.], dtype=float32),
 'step_type': array([1], dtype=int32)})]
[TimeStep(
{'discount': array([1.], dtype=float32),
 'observation': array([[-0.00882687, -0.24173944, -0.04874843,  0.32224613]],
      dtype=float32),
 'reward': array([1.], dtype=float32),
 'step_type': array([1], dtype=int32)}), array([1], dtype=int32), TimeStep(
{'discount': array([1.], dtype=float32),
 'observation': array([[-0.01366166, -0.04595843, -0.04230351,  0.01459712]],
      dtype=float32),
 'reward': array([1.], dtype=float32),
 'step_type': array([1], dtype=int32)})]
[TimeStep(
{'discount': array([1.], dtype=float32),
 'observation': array([[-0.01366166, -0.04595843, -0.04230351,  0.01459712]],
      dtype=float32),
 'reward': array([1.], dtype=float32),
 'step_type': array([1], dtype=int32)}), array([0], dtype=int32), TimeStep(
{'discount': array([1.], dtype=float32),
 'observation': array([[-0.01458083, -0.24044897, -0.04201157,  0.2936384 ]],
      dtype=float32),
 'reward': array([1.], dtype=float32),
 'step_type': array([1], dtype=int32)})]
Total reward: [3.]

전체 에피소드

env = suite_gym.load('CartPole-v0')
tf_env = tf_py_environment.TFPyEnvironment(env)

time_step = tf_env.reset()
rewards = []
steps = []
num_episodes = 5

for _ in range(num_episodes):
  episode_reward = 0
  episode_steps = 0
  while not time_step.is_last():
    action = tf.random.uniform([1], 0, 2, dtype=tf.int32)
    time_step = tf_env.step(action)
    episode_steps += 1
    episode_reward += time_step.reward.numpy()
  rewards.append(episode_reward)
  steps.append(episode_steps)
  time_step = tf_env.reset()

num_steps = np.sum(steps)
avg_length = np.mean(steps)
avg_reward = np.mean(rewards)

print('num_episodes:', num_episodes, 'num_steps:', num_steps)
print('avg_length', avg_length, 'avg_reward:', avg_reward)

num_episodes: 5 num_steps: 131
avg_length 26.2 avg_reward: 26.2