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

Playing CartPole con el método actor-crítico

Ver en TensorFlow.org Ejecutar en Google Colab Ver fuente en GitHub Descargar cuaderno

Este instructivo demuestra cómo implementar el método Actor-Critic con TensorFlow para entrenar a un agente en el entorno Open AI Gym CartPole-V0. Se supone que el lector está familiarizado con los métodos de aprendizaje reforzado en gradiente de políticas .

Métodos actor-crítico

Los métodos de actor crítico son métodos de aprendizaje de diferencia temporal (TD) que representan la función de política independiente de la función de valor.

Una función de política (o política) devuelve una distribución de probabilidad sobre las acciones que el agente puede realizar en función del estado dado. Una función de valor determina el rendimiento esperado para un agente que comienza en un estado dado y actúa de acuerdo con una política particular para siempre.

En el método Actor-Critic, la política se conoce como el actor que propone un conjunto de posibles acciones dado un estado, y la función de valor estimado se denomina crítica , que evalúa las acciones tomadas por el actor en función de la política dada. .

En este tutorial, tanto el actor como el crítico estarán representados utilizando una red neuronal con dos salidas.

CartPole-v0

En el entorno CartPole-v0 , un poste está conectado a un carro que se mueve a lo largo de una pista sin fricción. El poste comienza en posición vertical y el objetivo del agente es evitar que se caiga aplicando una fuerza de -1 o +1 al carro. Se otorga una recompensa de +1 por cada paso que el poste permanece en posición vertical. Un episodio termina cuando (1) el poste está a más de 15 grados de la vertical o (2) el carro se mueve más de 2.4 unidades desde el centro.

Modelo actor-crítico entrenado en ambiente Cartpole-v0

El problema se considera "resuelto" cuando la recompensa total promedio por el episodio llega a 195 en 100 intentos consecutivos.

Preparar

Importe los paquetes necesarios y configure los ajustes globales.

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

# Install additional packages for visualization
sudo apt-get install -y xvfb python-opengl > /dev/null 2>&1
pip install -q pyvirtualdisplay > /dev/null 2>&1
pip install -q git+https://github.com/tensorflow/docs > /dev/null 2>&1
import collections
import gym
import numpy as np
import tensorflow as tf
import tqdm

from matplotlib import pyplot as plt
from tensorflow.keras import layers
from typing import Any, List, Sequence, Tuple


# Create the environment
env = gym.make("CartPole-v0")

# Set seed for experiment reproducibility
seed = 42
env.seed(seed)
tf.random.set_seed(seed)
np.random.seed(seed)

# Small epsilon value for stabilizing division operations
eps = np.finfo(np.float32).eps.item()

Modelo

El actor y el crítico se modelarán utilizando una red neuronal que genera las probabilidades de acción y el valor crítico, respectivamente. Usamos subclases de modelos para definir el modelo.

Durante el pase hacia adelante, el modelo tomará el estado como entrada y generará tanto las probabilidades de acción como el valor crítico $ V $, que modela la función de valor dependiente del estado. El objetivo es entrenar un modelo que elija acciones basadas en una política $ \ pi $ que maximice el rendimiento esperado.

Para Cartpole-v0, hay cuatro valores que representan el estado: posición del carro, velocidad del carro, ángulo del polo y velocidad del polo, respectivamente. El agente puede realizar dos acciones para empujar el carrito hacia la izquierda (0) y hacia la derecha (1) respectivamente.

Consulte la página wiki CartPole-v0 de OpenAI Gym para obtener más información.

class ActorCritic(tf.keras.Model):
  """Combined actor-critic network."""

  def __init__(
      self, 
      num_actions: int, 
      num_hidden_units: int):
    """Initialize."""
    super().__init__()

    self.common = layers.Dense(num_hidden_units, activation="relu")
    self.actor = layers.Dense(num_actions)
    self.critic = layers.Dense(1)

  def call(self, inputs: tf.Tensor) -> Tuple[tf.Tensor, tf.Tensor]:
    x = self.common(inputs)
    return self.actor(x), self.critic(x)
num_actions = env.action_space.n  # 2
num_hidden_units = 128

model = ActorCritic(num_actions, num_hidden_units)

Formación

Para capacitar al agente, deberá seguir estos pasos:

  1. Ejecute el agente en el entorno para recopilar datos de entrenamiento por episodio.
  2. Calcule el rendimiento esperado en cada paso de tiempo.
  3. Calcule la pérdida para el modelo combinado actor-crítico.
  4. Calcule gradientes y actualice los parámetros de la red.
  5. Repita 1-4 hasta alcanzar el criterio de éxito o el número máximo de episodios.

1. Recopilación de datos de entrenamiento

Al igual que en el aprendizaje supervisado, para formar el modelo actor-crítico, necesitamos tener datos de formación. Sin embargo, para recopilar dichos datos, el modelo debería "ejecutarse" en el entorno.

Recopilamos datos de entrenamiento para cada episodio. Luego, en cada paso de tiempo, el pase hacia adelante del modelo se ejecutará en el estado del entorno para generar probabilidades de acción y el valor crítico basado en la política actual parametrizada por los pesos del modelo.

La siguiente acción será muestreada a partir de las probabilidades de acción generadas por el modelo, que luego se aplicarán al entorno, provocando que se genere el siguiente estado y recompensa.

Este proceso se implementa en la función run_episode , que usa operaciones de TensorFlow para que luego se pueda compilar en un gráfico de TensorFlow para un entrenamiento más rápido. Tenga en cuenta que tf.TensorArray s se utilizaron para admitir la iteración de Tensor en matrices de longitud variable.

# Wrap OpenAI Gym's `env.step` call as an operation in a TensorFlow function.
# This would allow it to be included in a callable TensorFlow graph.

def env_step(action: np.ndarray) -> Tuple[np.ndarray, np.ndarray, np.ndarray]:
  """Returns state, reward and done flag given an action."""

  state, reward, done, _ = env.step(action)
  return (state.astype(np.float32), 
          np.array(reward, np.int32), 
          np.array(done, np.int32))


def tf_env_step(action: tf.Tensor) -> List[tf.Tensor]:
  return tf.numpy_function(env_step, [action], 
                           [tf.float32, tf.int32, tf.int32])
def run_episode(
    initial_state: tf.Tensor,  
    model: tf.keras.Model, 
    max_steps: int) -> List[tf.Tensor]:
  """Runs a single episode to collect training data."""

  action_probs = tf.TensorArray(dtype=tf.float32, size=0, dynamic_size=True)
  values = tf.TensorArray(dtype=tf.float32, size=0, dynamic_size=True)
  rewards = tf.TensorArray(dtype=tf.int32, size=0, dynamic_size=True)

  initial_state_shape = initial_state.shape
  state = initial_state

  for t in tf.range(max_steps):
    # Convert state into a batched tensor (batch size = 1)
    state = tf.expand_dims(state, 0)
  
    # Run the model and to get action probabilities and critic value
    action_logits_t, value = model(state)
  
    # Sample next action from the action probability distribution
    action = tf.random.categorical(action_logits_t, 1)[0, 0]
    action_probs_t = tf.nn.softmax(action_logits_t)

    # Store critic values
    values = values.write(t, tf.squeeze(value))

    # Store log probability of the action chosen
    action_probs = action_probs.write(t, action_probs_t[0, action])
  
    # Apply action to the environment to get next state and reward
    state, reward, done = tf_env_step(action)
    state.set_shape(initial_state_shape)
  
    # Store reward
    rewards = rewards.write(t, reward)

    if tf.cast(done, tf.bool):
      break

  action_probs = action_probs.stack()
  values = values.stack()
  rewards = rewards.stack()
  
  return action_probs, values, rewards

2. Calcular los rendimientos esperados

Convertimos la secuencia de recompensas por cada paso de tiempo $ t $, $ {r_ {t}} ^ {T} _ {t = 1} $ recolectados durante un episodio en una secuencia de retornos esperados $ {G_ {t}} ^ { T} _ {t = 1} $ en el que la suma de las recompensas se toma del paso de tiempo actual $ t $ a $ T $ y cada recompensa se multiplica por un factor de descuento que disminuye exponencialmente $ \ gamma $:

$$G_{t} = \sum^{T}_{t'=t} \gamma^{t'-t}r_{t'}$$

Dado que $ \ gamma \ in (0,1) $, las recompensas más alejadas del paso de tiempo actual reciben menos peso.

Intuitivamente, el rendimiento esperado simplemente implica que las recompensas ahora son mejores que las recompensas posteriores. En un sentido matemático, se trata de asegurar que la suma de las recompensas converja.

Para estabilizar el entrenamiento, también estandarizamos la secuencia de retornos resultante (es decir, tener una media cero y una desviación estándar unitaria).

def get_expected_return(
    rewards: tf.Tensor, 
    gamma: float, 
    standardize: bool = True) -> tf.Tensor:
  """Compute expected returns per timestep."""

  n = tf.shape(rewards)[0]
  returns = tf.TensorArray(dtype=tf.float32, size=n)

  # Start from the end of `rewards` and accumulate reward sums
  # into the `returns` array
  rewards = tf.cast(rewards[::-1], dtype=tf.float32)
  discounted_sum = tf.constant(0.0)
  discounted_sum_shape = discounted_sum.shape
  for i in tf.range(n):
    reward = rewards[i]
    discounted_sum = reward + gamma * discounted_sum
    discounted_sum.set_shape(discounted_sum_shape)
    returns = returns.write(i, discounted_sum)
  returns = returns.stack()[::-1]

  if standardize:
    returns = ((returns - tf.math.reduce_mean(returns)) / 
               (tf.math.reduce_std(returns) + eps))

  return returns

3. La pérdida del actor crítico

Dado que estamos usando un modelo híbrido actor-crítico, usaremos la función de pérdida que es una combinación de pérdidas de actor y crítico para el entrenamiento, como se muestra a continuación:

$$L = L_{actor} + L_{critic}$$

Pérdida de actor

Formulamos la pérdida del actor en función de los gradientes de política con el crítico como una línea de base dependiente del estado y calculamos estimaciones de una sola muestra (por episodio).

$$L_{actor} = -\sum^{T}_{t=1} log\pi_{\theta}(a_{t} | s_{t})[G(s_{t}, a_{t}) - V^{\pi}_{\theta}(s_{t})]$$

dónde:

  • $ T $: el número de pasos de tiempo por episodio, que puede variar por episodio
  • $ s_ {t} $: el estado en el paso temporal $ t $
  • $ a_ {t} $: acción elegida en el paso de tiempo $ t $ estado dado $ s $
  • $ \ pi _ {\ theta} $: es la política (actor) parametrizada por $ \ theta $
  • $ V ^ {\ pi} _ {\ theta} $: es la función de valor (crítico) también parametrizada por $ \ theta $
  • $ G = G_ {t} $: el rendimiento esperado para un estado dado, par de acciones en el paso de tiempo $ t $

Agregamos un término negativo a la suma, ya que queremos maximizar las probabilidades de acciones que produzcan recompensas más altas minimizando la pérdida combinada.


Ventaja

El término $ G - V $ en nuestra formulación $ L_ {actor} $ se llama ventaja , que indica cuánto mejor se le da a una acción en un estado particular sobre una acción aleatoria seleccionada de acuerdo con la política $ \ pi $ para ese estado.

Si bien es posible excluir una línea de base, esto puede resultar en una gran variación durante el entrenamiento. Y lo bueno de elegir el crítico $ V $ como línea de base es que se entrenó para estar lo más cerca posible de $ G $, lo que lleva a una variación menor.

Además, sin el crítico, el algoritmo trataría de aumentar las probabilidades de las acciones tomadas en un estado en particular en función del rendimiento esperado, lo que puede no marcar una gran diferencia si las probabilidades relativas entre acciones siguen siendo las mismas.

Por ejemplo, suponga que dos acciones para un estado dado producirían el mismo rendimiento esperado. Sin la crítica, el algoritmo intentaría aumentar la probabilidad de estas acciones en función del objetivo $ J $. Con el crítico, puede resultar que no hay ventaja ($ G - V = 0 $) y, por lo tanto, no se obtiene ningún beneficio al aumentar las probabilidades de las acciones y el algoritmo establecería los gradientes en cero.


Pérdida crítica

El entrenamiento de $ V $ para que esté lo más cerca posible de $ G $ se puede configurar como un problema de regresión con la siguiente función de pérdida:

$$L_{critic} = L_{\delta}(G, V^{\pi}_{\theta})$$

donde $ L _ {\ delta} $ es la pérdida de Huber , que es menos sensible a los valores atípicos en los datos que la pérdida por error al cuadrado.

huber_loss = tf.keras.losses.Huber(reduction=tf.keras.losses.Reduction.SUM)

def compute_loss(
    action_probs: tf.Tensor,  
    values: tf.Tensor,  
    returns: tf.Tensor) -> tf.Tensor:
  """Computes the combined actor-critic loss."""

  advantage = returns - values

  action_log_probs = tf.math.log(action_probs)
  actor_loss = -tf.math.reduce_sum(action_log_probs * advantage)

  critic_loss = huber_loss(values, returns)

  return actor_loss + critic_loss

4. Definición del paso de entrenamiento para actualizar los parámetros

Combinamos todos los pasos anteriores en un paso de entrenamiento que se ejecuta en cada episodio. Todos los pasos que conducen a la función de pérdida se ejecutan con el contexto tf.GradientTape para permitir la diferenciación automática.

Usamos el optimizador de Adam para aplicar los degradados a los parámetros del modelo.

También calculamos la suma de las recompensas sin episode_reward , episode_reward , en este paso que se usaría más adelante para evaluar si cumplimos con el criterio de éxito.

Aplicamos el contexto tf.function a la función train_step para que pueda compilarse en un gráfico de TensorFlow invocable, lo que puede llevar a una aceleración 10 veces mayor en el entrenamiento.

optimizer = tf.keras.optimizers.Adam(learning_rate=0.01)


@tf.function
def train_step(
    initial_state: tf.Tensor, 
    model: tf.keras.Model, 
    optimizer: tf.keras.optimizers.Optimizer, 
    gamma: float, 
    max_steps_per_episode: int) -> tf.Tensor:
  """Runs a model training step."""

  with tf.GradientTape() as tape:

    # Run the model for one episode to collect training data
    action_probs, values, rewards = run_episode(
        initial_state, model, max_steps_per_episode) 

    # Calculate expected returns
    returns = get_expected_return(rewards, gamma)

    # Convert training data to appropriate TF tensor shapes
    action_probs, values, returns = [
        tf.expand_dims(x, 1) for x in [action_probs, values, returns]] 

    # Calculating loss values to update our network
    loss = compute_loss(action_probs, values, returns)

  # Compute the gradients from the loss
  grads = tape.gradient(loss, model.trainable_variables)

  # Apply the gradients to the model's parameters
  optimizer.apply_gradients(zip(grads, model.trainable_variables))

  episode_reward = tf.math.reduce_sum(rewards)

  return episode_reward

5. Ejecute el ciclo de entrenamiento

Ejecutamos el entrenamiento ejecutando el paso de entrenamiento hasta que se alcanza el criterio de éxito o el número máximo de episodios.

Mantenemos un registro continuo de las recompensas de los episodios usando una cola. Una vez que se alcanzan las 100 pruebas, la recompensa más antigua se elimina en el extremo izquierdo (final) de la cola y la más nueva se agrega al principio (derecha). También se mantiene una suma acumulada de las recompensas por eficiencia computacional.

Dependiendo de su tiempo de ejecución, el entrenamiento puede terminar en menos de un minuto.

%%time

max_episodes = 10000
max_steps_per_episode = 1000

# Cartpole-v0 is considered solved if average reward is >= 195 over 100 
# consecutive trials
reward_threshold = 195
running_reward = 0

# Discount factor for future rewards
gamma = 0.99

with tqdm.trange(max_episodes) as t:
  for i in t:
    initial_state = tf.constant(env.reset(), dtype=tf.float32)
    episode_reward = int(train_step(
        initial_state, model, optimizer, gamma, max_steps_per_episode))

    running_reward = episode_reward*0.01 + running_reward*.99
  
    t.set_description(f'Episode {i}')
    t.set_postfix(
        episode_reward=episode_reward, running_reward=running_reward)
  
    # Show average episode reward every 10 episodes
    if i % 10 == 0:
      pass # print(f'Episode {i}: average reward: {avg_reward}')
  
    if running_reward > reward_threshold:  
        break

print(f'\nSolved at episode {i}: average reward: {running_reward:.2f}!')
Episode 1524:  15%|█▌        | 1524/10000 [08:16<46:00,  3.07it/s, episode_reward=200, running_reward=195]

Solved at episode 1524: average reward: 195.03!
CPU times: user 20min 43s, sys: 4min 52s, total: 25min 35s
Wall time: 8min 16s



Visualización

Después del entrenamiento, sería bueno visualizar cómo se comporta el modelo en el entorno. Puede ejecutar las celdas a continuación para generar una animación GIF de la ejecución de un episodio del modelo. Tenga en cuenta que es necesario instalar paquetes adicionales para que OpenAI Gym represente correctamente las imágenes del entorno en Colab.

# Render an episode and save as a GIF file

from IPython import display as ipythondisplay
from PIL import Image
from pyvirtualdisplay import Display


display = Display(visible=0, size=(400, 300))
display.start()


def render_episode(env: gym.Env, model: tf.keras.Model, max_steps: int): 
  screen = env.render(mode='rgb_array')
  im = Image.fromarray(screen)

  images = [im]
  
  state = tf.constant(env.reset(), dtype=tf.float32)
  for i in range(1, max_steps + 1):
    state = tf.expand_dims(state, 0)
    action_probs, _ = model(state)
    action = np.argmax(np.squeeze(action_probs))

    state, _, done, _ = env.step(action)
    state = tf.constant(state, dtype=tf.float32)

    # Render screen every 10 steps
    if i % 10 == 0:
      screen = env.render(mode='rgb_array')
      images.append(Image.fromarray(screen))
  
    if done:
      break
  
  return images


# Save GIF image
images = render_episode(env, model, max_steps_per_episode)
image_file = 'cartpole-v0.gif'
# loop=0: loop forever, duration=1: play each frame for 1ms
images[0].save(
    image_file, save_all=True, append_images=images[1:], loop=0, duration=1)
import tensorflow_docs.vis.embed as embed
embed.embed_file(image_file)

gif

Próximos pasos

Este tutorial demostró cómo implementar el método actor-crítico usando Tensorflow.

Como siguiente paso, puede intentar entrenar a un modelo en un entorno diferente en OpenAI Gym.

Para obtener información adicional sobre los métodos de actor crítico y el problema Cartpole-v0, puede consultar los siguientes recursos:

Para obtener más ejemplos de aprendizaje por refuerzo en TensorFlow, puede consultar los siguientes recursos: