Conozca lo último en aprendizaje automático, IA generativa y más en el Simposio WiML 2023.

Se usó la API de Cloud Translation para traducir esta página.

Ambientes

Copyright 2021 Los autores de TF-Agents.

Ver en TensorFlow.org

Ejecutar en Google Colab

Ver fuente en GitHub

Descargar cuaderno

Introducción

El objetivo del aprendizaje por refuerzo (RL) es diseñar agentes que aprendan interactuando con un entorno. En la configuración estándar de RL, el agente recibe una observación en cada paso de tiempo y elige una acción. La acción se aplica al entorno y el entorno devuelve una recompensa y una nueva observación. El agente entrena una política para elegir acciones para maximizar la suma de recompensas, también conocido como retorno.

En TF-Agents, los entornos se pueden implementar en Python o TensorFlow. Los entornos de Python suelen ser más fáciles de implementar, comprender y depurar, pero los entornos de TensorFlow son más eficientes y permiten la paralelización natural. El flujo de trabajo más común es implementar un entorno en Python y usar uno de nuestros contenedores para convertirlo automáticamente en TensorFlow.

Veamos primero los entornos de Python. Los entornos de TensorFlow siguen una API muy similar.

Configuración

Si aún no ha instalado tf-agents o gym, ejecute:

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

Entornos de Python

Entornos Python tienen un step(action) -> next_time_step método que se aplica una acción para el medio ambiente, y devuelve la siguiente información sobre el siguiente paso:

observation : Esta es la parte del Estado ambiente que el agente puede observar que elegir sus acciones en el siguiente paso.
reward : El agente es aprender a maximizar la suma de estas recompensas a través de múltiples pasos.
step_type : Las interacciones con el medio ambiente son generalmente parte de una secuencia / episodio. por ejemplo, múltiples movimientos en una partida de ajedrez. step_type puede ser o bien FIRST , MID o LAST para indicar si este paso de tiempo es el primero, intermedio o el último paso en una secuencia.
discount : Se trata de un flotador que representa la cantidad de ponderar la recompensa a la próxima hora de pasos relativos a la recompensa en el paso de tiempo actual.

Estos se agrupan en una tupla llamado TimeStep(step_type, reward, discount, observation) .

La interfaz que todos los entornos Python deben implementar en environments/py_environment.PyEnvironment . Los principales métodos son:

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."""

Además de la step() método, entornos también proporcionan un reset() método que se inicia una nueva secuencia y proporciona una inicial TimeStep . No es necesario llamar al reset método explícitamente. Suponemos que los entornos se restablecen automáticamente, ya sea cuando llegan al final de un episodio o cuando se llama a step () por primera vez.

Tenga en cuenta que las subclases no implementan step() o reset() directamente. En su lugar, anulan la _step() y _reset() métodos. Los pasos de tiempo de volver de estos métodos se almacenan en caché y se expusieron a través de current_time_step() .

El observation_spec y los action_spec métodos devuelven un nido de (Bounded)ArraySpecs que describen el nombre, forma, tipo de datos y los rangos de las observaciones y las acciones, respectivamente.

En TF-Agents nos referimos repetidamente a nidos que se definen como cualquier estructura de árbol compuesta de listas, tuplas, tuplas con nombre o diccionarios. Estos se pueden componer arbitrariamente para mantener la estructura de observaciones y acciones. Hemos descubierto que esto es muy útil para entornos más complejos donde tiene muchas observaciones y acciones.

Usar entornos estándar

Agentes TF se ha incorporado en contenedores para muchos entornos estándar como el OpenAI gimnasia, DeepMind-control y Atari, para que sigan nuestra py_environment.PyEnvironment interfaz. Estos entornos envueltos se pueden cargar fácilmente utilizando nuestras suites de entornos. Carguemos el entorno CartPole desde el gimnasio OpenAI y veamos la acción y 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')

Así vemos que el medio ambiente espera que las acciones del tipo int64 en [0, 1] y devuelve TimeSteps donde las observaciones son una float32 vector de longitud 4 y el factor de descuento es un float32 en [0,0, 1,0]. Ahora, vamos a tratar de tomar una acción fija (1,) de un episodio entero.

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

Creando su propio entorno Python

Para muchos clientes, un caso de uso común es aplicar uno de los agentes estándar (ver agentes /) en TF-Agents a su problema. Para hacer esto, deben enmarcar su problema como un entorno. Así que veamos cómo implementar un entorno en Python.

Digamos que queremos entrenar a un agente para que juegue el siguiente juego de cartas (inspirado en Black Jack):

El juego se juega usando una baraja infinita de cartas numeradas del 1 al 10.
En cada turno, el agente puede hacer 2 cosas: obtener una nueva carta aleatoria o detener la ronda actual.
El objetivo es conseguir que la suma de tus cartas sea lo más cercana posible a 21 al final de la ronda, sin pasarse.

Un entorno que representa el juego podría verse así:

Acciones: Tenemos 2 acciones. Acción 0: obtener una nueva carta y Acción 1: terminar la ronda actual.
Observaciones: Suma de las cartas de la ronda actual.
Recompensa: El objetivo es acercarse lo más posible a 21 sin pasarse, por lo que podemos lograrlo usando la siguiente recompensa al final de la ronda: suma_de_tarjetas - 21 si suma_de_tarjetas <= 21, de lo contrario -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)

Asegurémonos de que hicimos todo correctamente definiendo el entorno anterior. Al crear su propio entorno, debe asegurarse de que las observaciones y los intervalos de tiempo generados sigan las formas y tipos correctos según se definen en sus especificaciones. Estos se utilizan para generar el gráfico de TensorFlow y, como tales, pueden crear problemas difíciles de depurar si los hacemos mal.

Para validar nuestro entorno, usaremos una política aleatoria para generar acciones e iteraremos más de 5 episodios para asegurarnos de que las cosas funcionen según lo previsto. Se genera un error si recibimos un time_step que no sigue las especificaciones del entorno.

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

Ahora que sabemos que el entorno funciona según lo previsto, ejecutemos este entorno con una política fija: solicite 3 cartas y luego finalice la ronda.

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

Envoltorios ambientales

Un contenedor de entorno toma un entorno de Python y devuelve una versión modificada del entorno. Tanto el entorno original y el entorno modificado son instancias de py_environment.PyEnvironment , y múltiples envolturas se pueden encadenar juntos.

Algunas envolturas comunes se pueden encontrar en environments/wrappers.py . Por ejemplo:

ActionDiscretizeWrapper : convierte un espacio de acción continua a un espacio de funcionamiento discreto.
RunStats : Captura corren estadísticas del medio ambiente, como el número de pasos dados, número de episodios completó etc.
TimeLimit : Termina el episodio después de un número fijo de pasos.

Ejemplo 1: Envoltura discretizada de acción

InvertedPendulum es un entorno PyBullet que acepta acciones continuas en el intervalo [-2, 2] . Si queremos entrenar a un agente de acción discreto como DQN en este entorno, tenemos que discretizar (cuantificar) el espacio de acción. Esto es exactamente lo que el ActionDiscretizeWrapper hace. Comparar la action_spec antes y después de la envoltura:

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)

La envuelta discrete_action_env es una instancia de py_environment.PyEnvironment y puede ser tratada como un ambiente regular Python.

Entornos de TensorFlow

La interfaz para entornos TF se define en environments/tf_environment.TFEnvironment y es muy similar a los entornos de Python. Los entornos TF se diferencian de los entornos de Python en un par de formas:

Generan objetos tensores en lugar de matrices.
Los entornos TF agregan una dimensión de lote a los tensores generados en comparación con las especificaciones.

La conversión de los entornos de Python en TFEnvs permite que tensorflow paralelice las operaciones. Por ejemplo, se podría definir un collect_experience_op que recoge datos desde el medio ambiente y se suma a una replay_buffer , y un train_op que lee de la replay_buffer y entrena el agente, y las dirigen en paralelo de forma natural en 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`."""

El current_time_step() método devuelve el time_step actual e inicializa el entorno, si es necesario.

Los reset() fuerzas método retroceden en el medio ambiente y devuelve el CURRENT_STEP.

Si la action no depende de la anterior time_step un tf.control_dependency es necesario en Graph de modo.

Por ahora, vamos a ver cómo TFEnvironments se crean.

Creación de su propio entorno de TensorFlow

Esto es más complicado que crear entornos en Python, por lo que no lo cubriremos en esta colab. Un ejemplo se encuentra disponible aquí . El caso de uso más común es poner en práctica su entorno en Python y se envuelve en TensorFlow utilizando nuestro TFPyEnvironment envoltorio (ver más abajo).

Envolviendo un entorno de Python en TensorFlow

Podemos concluir fácilmente a cualquier entorno de Python en un entorno TensorFlow utilizando el TFPyEnvironment envoltura.

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

Tenga en cuenta las especificaciones son ahora de tipo: (Bounded)TensorSpec .

Ejemplos de uso

Ejemplo simple

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.]

Episodios completos

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