Esta página foi traduzida pela API Cloud Translation.
Switch to English

Ambientes

Ver em TensorFlow.org Executar no Google Colab Ver fonte no GitHub Download do caderno

Introdução

O objetivo do Aprendizado por Reforço (RL) é projetar agentes que aprendem interagindo com um ambiente. Na configuração RL padrão, o agente recebe uma observação a cada etapa e escolhe uma ação. A ação é aplicada ao ambiente e o ambiente retorna uma recompensa e uma nova observação. O agente treina uma política para escolher ações para maximizar a soma de recompensas, também conhecida como devolução.

Nos agentes TF, os ambientes podem ser implementados no Python ou no TensorFlow. Os ambientes Python geralmente são mais fáceis de implementar, entender e depurar, mas os ambientes TensorFlow são mais eficientes e permitem paralelização natural. O fluxo de trabalho mais comum é implementar um ambiente em Python e usar um de nossos wrappers para convertê-lo automaticamente em TensorFlow.

Vamos examinar primeiro os ambientes Python. Os ambientes TensorFlow seguem uma API muito semelhante.

Configuração

Se você ainda não instalou tf-agents ou gym, execute: 

pip install -q --pre tf-agents[reverb]
pip install -q 'gym==0.10.11'

WARNING: You are using pip version 20.1.1; however, version 20.2 is available.
You should consider upgrading via the '/tmpfs/src/tf_docs_env/bin/python -m pip install --upgrade pip' command.
WARNING: You are using pip version 20.1.1; however, version 20.2 is available.
You should consider upgrading via the '/tmpfs/src/tf_docs_env/bin/python -m pip install --upgrade pip' command.


 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

tf.compat.v1.enable_v2_behavior()

Ambientes Python

Os ambientes Python possuem um método step(action) -> next_time_step que aplica uma ação ao ambiente e retorna as seguintes informações sobre o próximo passo:

  1. observation : Esta é a parte do estado do ambiente que o agente pode observar para escolher suas ações na próxima etapa.
  2. reward : o agente está aprendendo a maximizar a soma dessas recompensas em várias etapas.
  3. step_type : as interações com o ambiente geralmente fazem parte de uma sequência / episódio. por exemplo, múltiplos movimentos em um jogo de xadrez. step_type pode ser FIRST , MID ou LAST para indicar se esse intervalo de tempo é o primeiro, o intermediário ou o último passo de uma sequência.
  4. discount : é um valor flutuante que representa o peso da recompensa na próxima etapa em relação à recompensa na etapa atual.

Eles são agrupados em uma tupla TimeStep(step_type, reward, discount, observation) nomeada TimeStep(step_type, reward, discount, observation) .

A interface que todos os ambientes python devem implementar está em environments/py_environment.PyEnvironment . Os principais métodos são: 

 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."""
    self._current_time_step = self._step(action)
    return self._current_time_step

Além do método step() , os ambientes também fornecem um método reset() que inicia uma nova sequência e fornece um TimeStep inicial. Não é necessário chamar o método de reset explicitamente. Assumimos que os ambientes são redefinidos automaticamente, quando chegam ao final de um episódio ou quando step () é chamado pela primeira vez.

Observe que as subclasses não implementam step() ou reset() diretamente. Eles substituem os _step() e _reset() . As etapas de tempo retornadas desses métodos serão armazenadas em cache e expostas por current_time_step() .

Os métodos observation_spec e action_spec retornam um ninho de (Bounded)ArraySpecs que descrevem o nome, a forma, o tipo de dados e os intervalos das observações e ações, respectivamente.

Nos agentes TF, repetidamente nos referimos a ninhos que são definidos como qualquer estrutura semelhante a uma árvore composta de listas, tuplas, tuplas nomeadas ou dicionários. Estes podem ser compostos arbitrariamente para manter a estrutura de observações e ações. Descobrimos que isso é muito útil para ambientes mais complexos, onde você tem muitas observações e ações.

Usando ambientes padrão

O TF Agents possui invólucros incorporados para muitos ambientes padrão, como o OpenAI Gym, DeepMind-control e Atari, para que eles sigam nossa interface py_environment.PyEnvironment . Esses ambientes agrupados podem ser facilmente carregados usando nossos conjuntos de ambientes. Vamos carregar o ambiente CartPole da academia OpenAI e ver a ação e 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')

Portanto, vemos que o ambiente espera ações do tipo int64 em [0, 1] e retorna TimeSteps onde as observações são um vetor float32 de comprimento 4 e o fator de desconto é um float32 em [0,0, 1,0]. Agora, vamos tentar executar uma ação fixa (1,) durante um episódio inteiro. 

 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(step_type=array(0, dtype=int32), reward=array(0., dtype=float32), discount=array(1., dtype=float32), observation=array([-0.01974552, -0.0401244 ,  0.01611689, -0.0195175 ], dtype=float32))
TimeStep(step_type=array(1, dtype=int32), reward=array(1., dtype=float32), discount=array(1., dtype=float32), observation=array([-0.020548  ,  0.15476274,  0.01572654, -0.3070721 ], dtype=float32))
TimeStep(step_type=array(1, dtype=int32), reward=array(1., dtype=float32), discount=array(1., dtype=float32), observation=array([-0.01745275,  0.34965712,  0.0095851 , -0.5947541 ], dtype=float32))
TimeStep(step_type=array(1, dtype=int32), reward=array(1., dtype=float32), discount=array(1., dtype=float32), observation=array([-0.01045961,  0.5446436 , -0.00230998, -0.8844024 ], dtype=float32))
TimeStep(step_type=array(1, dtype=int32), reward=array(1., dtype=float32), discount=array(1., dtype=float32), observation=array([ 4.3326573e-04,  7.3979688e-01, -1.9998031e-02, -1.1778107e+00],
      dtype=float32))
TimeStep(step_type=array(1, dtype=int32), reward=array(1., dtype=float32), discount=array(1., dtype=float32), observation=array([ 0.0152292 ,  0.93517274, -0.04355424, -1.476695  ], dtype=float32))
TimeStep(step_type=array(1, dtype=int32), reward=array(1., dtype=float32), discount=array(1., dtype=float32), observation=array([ 0.03393266,  1.1307988 , -0.07308814, -1.7826566 ], dtype=float32))
TimeStep(step_type=array(1, dtype=int32), reward=array(1., dtype=float32), discount=array(1., dtype=float32), observation=array([ 0.05654863,  1.3266624 , -0.10874127, -2.0971365 ], dtype=float32))
TimeStep(step_type=array(1, dtype=int32), reward=array(1., dtype=float32), discount=array(1., dtype=float32), observation=array([ 0.08308188,  1.522697  , -0.150684  , -2.4213583 ], dtype=float32))
TimeStep(step_type=array(1, dtype=int32), reward=array(1., dtype=float32), discount=array(1., dtype=float32), observation=array([ 0.11353582,  1.7187654 , -0.19911116, -2.7562623 ], dtype=float32))
TimeStep(step_type=array(2, dtype=int32), reward=array(1., dtype=float32), discount=array(0., dtype=float32), observation=array([ 0.14791113,  1.9146413 , -0.2542364 , -3.1024237 ], dtype=float32))

Criando seu próprio ambiente Python

Para muitos clientes, um caso de uso comum é aplicar um dos agentes padrão (consulte agents /) nos TF-Agents ao seu problema. Para fazer isso, eles precisam enquadrar seu problema como um ambiente. Então, vejamos como implementar um ambiente em Python.

Digamos que queremos treinar um agente para jogar o seguinte jogo de cartas (inspirado no Black Jack):

  1. O jogo é jogado usando um baralho infinito, numerado de 1 a 10.
  2. A cada passo, o agente pode fazer duas coisas: obter uma nova carta aleatória ou interromper a rodada atual.
  3. O objetivo é obter a soma de suas cartas o mais próximo possível de 21 no final da rodada, sem passar por cima.

Um ambiente que representa o jogo pode ser assim:

  1. Ações: Temos 2 ações. Ação 0: obtenha uma nova carta e Ação 1: encerre a rodada atual.
  2. Observações: Soma das cartas na rodada atual.
  3. Recompensa: O objetivo é chegar o mais próximo possível de 21 sem ultrapassar, para que possamos conseguir isso usando a seguinte recompensa no final da rodada: sum_of_cards - 21 if sum_of_cards <= 21, caso contrário -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)

Vamos garantir que fizemos tudo corretamente, definindo o ambiente acima. Ao criar seu próprio ambiente, verifique se as observações e os time_steps gerados seguem as formas e os tipos corretos, conforme definido nas suas especificações. Eles são usados ​​para gerar o gráfico TensorFlow e, como tal, podem criar problemas difíceis de depurar, se errarmos.

Para validar nosso ambiente, usaremos uma política aleatória para gerar ações e iteraremos mais de cinco episódios para garantir que as coisas estejam funcionando conforme o esperado. Um erro será gerado se recebermos um time_step que não segue as especificações do ambiente. 

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

Agora que sabemos que o ambiente está funcionando como pretendido, vamos executá-lo usando uma política fixa: solicite 3 cartões e encerre a rodada. 

 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(step_type=array(0, dtype=int32), reward=array(0., dtype=float32), discount=array(1., dtype=float32), observation=array([0], dtype=int32))
TimeStep(step_type=array(1, dtype=int32), reward=array(0., dtype=float32), discount=array(1., dtype=float32), observation=array([4], dtype=int32))
TimeStep(step_type=array(1, dtype=int32), reward=array(0., dtype=float32), discount=array(1., dtype=float32), observation=array([12], dtype=int32))
TimeStep(step_type=array(1, dtype=int32), reward=array(0., dtype=float32), discount=array(1., dtype=float32), observation=array([15], dtype=int32))
TimeStep(step_type=array(2, dtype=int32), reward=array(-6., dtype=float32), discount=array(0., dtype=float32), observation=array([15], dtype=int32))
Final Reward =  -6.0

Wrappers de ambiente

Um wrapper de ambiente pega um ambiente python e retorna uma versão modificada do ambiente. O ambiente original e o ambiente modificado são instâncias de py_environment.PyEnvironment e vários wrappers podem ser encadeados.

Alguns wrappers comuns podem ser encontrados em environments/wrappers.py . Por exemplo:

  1. ActionDiscretizeWrapper : converte um espaço de ação contínuo em um espaço de ação discreto.
  2. RunStats : captura as estatísticas de execução do ambiente, como número de etapas executadas, número de episódios concluídos etc.
  3. TimeLimit : finaliza o episódio após um número fixo de etapas.

Exemplo 1: Ação Discretizar Wrapper

InvertedPendulum é um ambiente PyBullet que aceita ações contínuas no intervalo [-2, 2] . Se queremos treinar um agente de ação discreta como o DQN nesse ambiente, precisamos discretizar (quantizar) o espaço de ação. É exatamente isso que o ActionDiscretizeWrapper faz. Compare o action_spec antes e depois da quebra: 

 env = suite_gym.load('Pendulum-v0')
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)

O discrete_action_env py_environment.PyEnvironment é uma instância de py_environment.PyEnvironment e pode ser tratado como um ambiente python comum.

Ambientes TensorFlow

A interface para ambientes TF é definida em environments/tf_environment.TFEnvironment e se parece muito com os ambientes Python. Os ambientes TF diferem dos envs python de duas maneiras:

  • Eles geram objetos tensores em vez de matrizes
  • Os ambientes TF adicionam uma dimensão de lote aos tensores gerados quando comparados às especificações.

A conversão dos ambientes python em TFEnvs permite que o fluxo tensor parellize as operações. Por exemplo, pode-se definir uma collect_experience_op que coleta dados do ambiente e os adiciona a um replay_buffer , e um train_op que lê o replay_buffer e treina o agente e executá-los em paralelo naturalmente no 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`."""

O método current_time_step() retorna o time_step atual e inicializa o ambiente, se necessário.

O método reset() força uma redefinição no ambiente e retorna o current_step.

Se a action não depender do time_step anterior, será necessário um tf.control_dependency no modo Graph .

Por enquanto, vejamos como os TFEnvironments são criados.

Criando seu próprio ambiente TensorFlow

Isso é mais complicado do que criar ambientes em Python, por isso não abordaremos nesta colab. Um exemplo está disponível aqui . O caso de uso mais comum é implementar seu ambiente em Python e envolvê-lo no TensorFlow usando nosso wrapper TFPyEnvironment (veja abaixo).

Agrupando um ambiente Python no TensorFlow

Podemos facilmente agrupar qualquer ambiente Python em um ambiente TensorFlow usando o wrapper 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(step_type=TensorSpec(shape=(), dtype=tf.int32, name='step_type'), reward=TensorSpec(shape=(), dtype=tf.float32, name='reward'), 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)))
Action Specs: BoundedTensorSpec(shape=(), dtype=tf.int64, name='action', minimum=array(0), maximum=array(1))

Observe que as especificações agora são do tipo: (Bounded)TensorSpec .

Exemplos de uso

Exemplo Simples 

 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(step_type=array([0], dtype=int32), reward=array([0.], dtype=float32), discount=array([1.], dtype=float32), observation=array([[-0.03290624, -0.0323446 ,  0.03332262, -0.00031953]],
      dtype=float32)), array([0], dtype=int32), TimeStep(step_type=array([1], dtype=int32), reward=array([1.], dtype=float32), discount=array([1.], dtype=float32), observation=array([[-0.03355313, -0.2279282 ,  0.03331622,  0.30268806]],
      dtype=float32))]
[TimeStep(step_type=array([1], dtype=int32), reward=array([1.], dtype=float32), discount=array([1.], dtype=float32), observation=array([[-0.03355313, -0.2279282 ,  0.03331622,  0.30268806]],
      dtype=float32)), array([1], dtype=int32), TimeStep(step_type=array([1], dtype=int32), reward=array([1.], dtype=float32), discount=array([1.], dtype=float32), observation=array([[-0.03811169, -0.03329653,  0.03936999,  0.02069571]],
      dtype=float32))]
[TimeStep(step_type=array([1], dtype=int32), reward=array([1.], dtype=float32), discount=array([1.], dtype=float32), observation=array([[-0.03811169, -0.03329653,  0.03936999,  0.02069571]],
      dtype=float32)), array([0], dtype=int32), TimeStep(step_type=array([1], dtype=int32), reward=array([1.], dtype=float32), discount=array([1.], dtype=float32), observation=array([[-0.03877762, -0.22896032,  0.0397839 ,  0.32553574]],
      dtype=float32))]
Total reward: [3.]

Episódios inteiros 

 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: 116
avg_length 23.2 avg_reward: 23.2