Ayuda a proteger la Gran Barrera de Coral con TensorFlow en Kaggle Únete Challenge

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

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

Este tutorial muestra cómo implementar el actor-crítico método que usa para entrenar TensorFlow un agente en el Abierto de AI Gimnasio ambiente CartPole-V0. El lector se supone que tiene cierta familiaridad con métodos de gradiente política de aprendizaje por refuerzo.

Métodos actor-crítico

Métodos Actor crítico son de aprendizaje temporales diferencia (TD) métodos 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 del actor-crítico, la política se conoce como el actor que propone un conjunto de acciones posibles dado un estado, y la función del valor estimado se conoce como el crítico, que evalúa las acciones tomadas por el actor sobre la base de la política determinada .

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

CartPole-v0

En el entorno de CartPole-V0 , un poste está unido a un carro en movimiento 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.

Configuración

Importe los paquetes necesarios y configure los ajustes globales.

pip install gym
pip install pyglet
# Install additional packages for visualization
sudo apt-get install -y xvfb python-opengl > /dev/null 2>&1
pip install pyvirtualdisplay > /dev/null 2>&1
pip install git+https://github.com/tensorflow/docs > /dev/null 2>&1
import collections
import gym
import numpy as np
import statistics
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 Crítico se modelaron usando una red neural que genera las probabilidades de acción y valor crítico respectivamente. Este tutorial usa subclases de modelos para definir el modelo.

Durante el pase hacia adelante, el modelo se llevará en el estado como la entrada y la voluntad de salida ambas probabilidades de acción y valor crítico \(V\), que modela el dependiente del estado de función de valor . El objetivo es formar un modelo que elige a las acciones basadas en una política \(\pi\) que maximiza la espera de retorno .

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

Capacitació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 que se alcance 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 entrenar el modelo actor-crítico, es necesario tener datos de entrenamiento. Sin embargo, para recopilar dichos datos, el modelo debería "ejecutarse" en el entorno.

Se recopilan 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 en función de 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 lleva a cabo en el run_episode función, que utiliza operaciones TensorFlow para que luego se puede compilar en un gráfico TensorFlow para el entrenamiento más rápido. Tenga en cuenta que tf.TensorArray s se utilizaron para apoyar iteración tensor de 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) -> Tuple[tf.Tensor, tf.Tensor, 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

La secuencia de recompensas para cada paso de tiempo \(t\), \(\{r_{t}\}^{T}_{t=1}\) recogida durante un episodio se convierte en una secuencia de rendimientos esperados \(\{G_{t}\}^{T}_{t=1}\) en el que se toma la suma de recompensas del paso de tiempo actual \(t\) a \(T\) y cada premio se multiplica por un factor de descuento que disminuye exponencialmente \(\gamma\):

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

Desde \(\gamma\in(0,1)\), premia más lejos del paso de tiempo actual se le da 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, la secuencia resultante de retornos también está estandarizada (es decir, para 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 se utiliza un modelo híbrido de actor-crítico, la función de pérdida elegida 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

La pérdida actor está basada en gradientes de política con la crítica como una línea de base depende del estado y se calcula con las 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 timesteps por episodio, que puede variar por episodio
  • \(s_{t}\): el estado en el paso de tiempo \(t\)
  • \(a_{t}\): acción elegida en timestep \(t\) dados estado \(s\)
  • \(\pi_{\theta}\): es la política (actor) parametrizado por \(\theta\)
  • \(V^{\pi}_{\theta}\): es la función de valor (crítico) también parametrizado por \(\theta\)
  • \(G = G_{t}\): el retorno esperado para un determinado estado, par la acción a paso de tiempo \(t\)

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


Ventaja

El \(G - V\) plazo en nuestra \(L_{actor}\) formulación se llama la ventaja , que indica cuánto mejor se le da una acción de un estado en particular a través de una acción aleatoria seleccionada de acuerdo con la política de \(\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 ser lo más cercano posible a \(G\), lo que lleva a una menor varianza.

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 el crítico, el algoritmo podría tratar de aumentar la probabilidad de que estas acciones en función del objetivo \(J\). Con el crítico, puede resultar que no hay ninguna ventaja (\(G - V = 0\)) y por lo tanto ningún beneficio obtenido en el aumento de las probabilidades de las acciones y el algoritmo podría establecer los gradientes a cero.


Pérdida crítica

El entrenamiento de \(V\) a estar lo más cerca posible \(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 de que la pérdida-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

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

Este tutorial utiliza el optimizador de Adam para aplicar los degradados a los parámetros del modelo.

La suma de las recompensas no descontados, episode_reward , se calcula también en este paso. Este valor se utilizará más adelante para evaluar si se cumple el criterio de éxito.

El tf.function contexto se aplica a la train_step función para que se puede compilar en un gráfico TensorFlow exigible, lo que puede llevar 10 veces más veloz en los entrenamientos.

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

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

Se mantiene en cola un registro actualizado de las recompensas de los episodios. 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

min_episodes_criterion = 100
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

# Keep last episodes reward
episodes_reward: collections.deque = collections.deque(maxlen=min_episodes_criterion)

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

    episodes_reward.append(episode_reward)
    running_reward = statistics.mean(episodes_reward)

    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 and i >= min_episodes_criterion:  
        break

print(f'\nSolved at episode {i}: average reward: {running_reward:.2f}!')
Episode 361:   4%|▎         | 361/10000 [01:16<34:10,  4.70it/s, episode_reward=182, running_reward=195]
Solved at episode 361: average reward: 195.14!
CPU times: user 2min 50s, sys: 40.3 s, total: 3min 30s
Wall time: 1min 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 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: