RSVP para seu evento TensorFlow Everywhere hoje mesmo!
Esta página foi traduzida pela API Cloud Translation.
Switch to English

Ambientes

Ver no TensorFlow.org Executar no Google Colab Ver fonte no GitHub Baixar caderno

Introdução

O objetivo do Reinforcement Learning (RL) é projetar agentes que aprendem interagindo com um ambiente. Na configuração RL padrão, o agente recebe uma observação a cada passo de tempo e escolhe uma ação. A ação é aplicada ao meio ambiente e o meio ambiente retorna uma recompensa e uma nova observação. O agente treina uma política de escolha de ações para maximizar a soma das recompensas, também conhecido como retorno.

Em TF-Agents, os ambientes podem ser implementados em Python ou 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 a 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.

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

Estabelecer

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

pip install -q 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

tf.compat.v1.enable_v2_behavior()

Ambientes Python

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

  1. observation : é 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, vários movimentos em um jogo de xadrez. step_type pode ser FIRST , MID ou LAST para indicar se esta etapa de tempo é a primeira, intermediária ou última etapa em uma sequência.
  4. discount : é um valor flutuante que representa quanto pesar a recompensa na próxima etapa de tempo em relação à recompensa na etapa de tempo atual.

Eles são agrupados em uma tupla 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."""

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 reset explicitamente. Assumimos que os ambientes são redefinidos automaticamente, seja quando chegam ao final de um episódio ou quando a etapa () é chamada pela primeira vez.

Observe que as subclasses não implementam step() ou reset() diretamente. Em vez disso, eles substituem os _step() e _reset() . Os intervalos de tempo retornados desses métodos serão armazenados em cache e expostos por meio de 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.

Em TF-Agents, 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

TF Agents tem wrappers integrados para muitos ambientes padrão, como OpenAI Gym, DeepMind-control e Atari, para que sigam nossa interface py_environment.PyEnvironment . Esses ambientes agrupados podem ser carregados facilmente usando nossos pacotes de ambiente. Vamos carregar o ambiente CartPole do ginásio OpenAI e olhar para a ação e o 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 realizar uma ação fixa (1,) para 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.01285449,  0.04769544,  0.01983412, -0.00245379], dtype=float32))
TimeStep(step_type=array(1, dtype=int32), reward=array(1., dtype=float32), discount=array(1., dtype=float32), observation=array([ 0.0138084 ,  0.24252741,  0.01978504, -0.2888134 ], dtype=float32))
TimeStep(step_type=array(1, dtype=int32), reward=array(1., dtype=float32), discount=array(1., dtype=float32), observation=array([ 0.01865895,  0.43736172,  0.01400878, -0.57519126], dtype=float32))
TimeStep(step_type=array(1, dtype=int32), reward=array(1., dtype=float32), discount=array(1., dtype=float32), observation=array([ 0.02740618,  0.6322845 ,  0.00250495, -0.8634283 ], dtype=float32))
TimeStep(step_type=array(1, dtype=int32), reward=array(1., dtype=float32), discount=array(1., dtype=float32), observation=array([ 0.04005187,  0.82737225, -0.01476362, -1.1553226 ], dtype=float32))
TimeStep(step_type=array(1, dtype=int32), reward=array(1., dtype=float32), discount=array(1., dtype=float32), observation=array([ 0.05659932,  1.0226836 , -0.03787007, -1.452598  ], dtype=float32))
TimeStep(step_type=array(1, dtype=int32), reward=array(1., dtype=float32), discount=array(1., dtype=float32), observation=array([ 0.07705299,  1.2182497 , -0.06692202, -1.7568679 ], dtype=float32))
TimeStep(step_type=array(1, dtype=int32), reward=array(1., dtype=float32), discount=array(1., dtype=float32), observation=array([ 0.10141798,  1.4140631 , -0.10205939, -2.069591  ], dtype=float32))
TimeStep(step_type=array(1, dtype=int32), reward=array(1., dtype=float32), discount=array(1., dtype=float32), observation=array([ 0.12969925,  1.6100639 , -0.1434512 , -2.3920157 ], dtype=float32))
TimeStep(step_type=array(1, dtype=int32), reward=array(1., dtype=float32), discount=array(1., dtype=float32), observation=array([ 0.16190052,  1.8061239 , -0.19129153, -2.725115  ], dtype=float32))
TimeStep(step_type=array(2, dtype=int32), reward=array(1., dtype=float32), discount=array(0., dtype=float32), observation=array([ 0.198023  ,  2.002027  , -0.24579382, -3.0695074 ], dtype=float32))

Criação de seu próprio ambiente Python

Para muitos clientes, um caso de uso comum é aplicar um dos agentes padrão (consulte agentes /) em Agentes TF ao problema. Para fazer isso, eles devem enquadrar seu problema como um ambiente. Então, vamos ver 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 com um baralho infinito de cartas numeradas de 1 ... 10.
  2. A cada jogada, o agente pode fazer 2 coisas: obter uma nova carta aleatória ou interromper a rodada atual.
  3. O objetivo é fazer com que a soma de suas cartas seja o mais próximo possível de 21 no final da rodada, sem ultrapassar.

Um ambiente que representa o jogo pode ser assim:

  1. Ações: temos 2 ações. Ação 0: pegue 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, então podemos conseguir isso usando a seguinte recompensa no final da rodada: soma_de_cartas - 21 se soma_de_cartas <= 21, senão -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 nos certificar de que fizemos tudo corretamente definindo o ambiente acima. Ao criar seu próprio ambiente, você deve se certificar de que as observações e time_steps gerados seguem as formas e tipos corretos, conforme definido em suas especificações. Eles são usados ​​para gerar o gráfico do 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 faremos uma iteração em mais de 5 episódios para garantir que as coisas estejam funcionando como pretendido. Um erro é 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 conforme o esperado, vamos executá-lo usando uma política fixa: peça 3 cartas 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([2], dtype=int32))
TimeStep(step_type=array(1, dtype=int32), reward=array(0., dtype=float32), discount=array(1., dtype=float32), observation=array([7], dtype=int32))
TimeStep(step_type=array(1, dtype=int32), reward=array(0., dtype=float32), discount=array(1., dtype=float32), observation=array([8], dtype=int32))
TimeStep(step_type=array(2, dtype=int32), reward=array(-13., dtype=float32), discount=array(0., dtype=float32), observation=array([8], dtype=int32))
Final Reward =  -13.0

Envoltórios de ambiente

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

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

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

Exemplo 1: Wrapper de Discretização de Ação

InvertedPendulum é um ambiente PyBullet que aceita ações contínuas no intervalo [-2, 2] . Se quisermos treinar um agente de ação discreto como DQN neste ambiente, temos que discretizar (quantizar) o espaço de ação. Isso é exatamente o que ActionDiscretizeWrapper faz. Compare o action_spec antes e depois do empacotamento:

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 empacotado é uma instância de py_environment.PyEnvironment e pode ser tratado como um ambiente Python normal.

Ambientes TensorFlow

A interface para ambientes TF é definida em environments/tf_environment.TFEnvironment e é muito semelhante aos ambientes Python. Os ambientes TF diferem dos ambientes 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.

Converter os ambientes Python em TFEnvs permite que o tensorflow paralelize as operações. Por exemplo, pode-se definir um collect_experience_op que coleta dados do ambiente e adiciona a um replay_buffer , e um train_op que lê de replay_buffer e treina o agente, e os executa 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 um reset no ambiente e retorna current_step.

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

Por enquanto, vamos ver como os ambientes TFEnvironments são criados.

Criar seu próprio ambiente TensorFlow

Isso é mais complicado do que criar ambientes em Python, então não vamos cobrir isso 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).

Envolvendo um ambiente Python no TensorFlow

Podemos facilmente envolver 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.03501577, -0.04957427,  0.00623939,  0.03762257]],
      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.03600726, -0.24478514,  0.00699184,  0.33226755]],
      dtype=float32))]
[TimeStep(step_type=array([1], dtype=int32), reward=array([1.], dtype=float32), discount=array([1.], dtype=float32), observation=array([[-0.03600726, -0.24478514,  0.00699184,  0.33226755]],
      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.04090296, -0.0497634 ,  0.01363719,  0.04179767]],
      dtype=float32))]
[TimeStep(step_type=array([1], dtype=int32), reward=array([1.], dtype=float32), discount=array([1.], dtype=float32), observation=array([[-0.04090296, -0.0497634 ,  0.01363719,  0.04179767]],
      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.04189822, -0.24507822,  0.01447314,  0.33875188]],
      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: 138
avg_length 27.6 avg_reward: 27.6