T TP 5 : Advantage Actor Critic¶
Ce TP est disponible dans google colab ici ou téléchargeable ici
Actor-Critic¶
Advantage Actor Critic est une méthode optimisation de la politique populaire en RL. Cette approche consiste à maximiser le retour moyen en favorisant les probailités liées aux actions donnant le plus de récompenses. Si \(\pi_\theta\) consiste en une politique paramètrée par \(\theta\), le gradient de notre fonction objectif \(\mathcal{J}(\pi_\theta)\) s’écrit
avec \(\tau\) la trajectoire et \(A^{\pi_\theta}\) la fonction avantage. L’approche policy gradient consiste ) mettre à jour les paramètre de \(\pi\) par montée de gradient :
avec \(\alpha\) le learning rate. L’agent est entraîné de façon on-policy c’est à dire que les paramètres sont mis à jour par la politique courrante.
Advantage Function¶
La fonction avantage \(A\) permet de réduire la variance de la fonction de valeur et est défini comme suit
On peut remplacer \(Q\) avec \(r+\gamma V(s')\) et estimer \(A\) de la façon suivante:
Maximization Entropy¶
L’entropie est une mesure d’incertitude. À des actions avec une probabilité uniforme correspond une entropie maximale tandis que l’entropie est nulle lorsque toute la masse est concentrée sur une action. L’entropie \(H\) par rapport à la probabilité \(p\) sur les action est
import module and seed¶
import random
from typing import List, Tuple
import gym
import matplotlib.pyplot as plt
import numpy as np
import torch
import torch.nn as nn
import torch.nn.functional as F
import torch.optim as optim
from IPython.display import clear_output
from torch.distributions import Normal
if torch.backends.cudnn.enabled:
torch.backends.cudnn.benchmark = False
torch.backends.cudnn.deterministic = True
seed = 777
torch.manual_seed(seed)
np.random.seed(seed)
random.seed(seed)
Modèle¶
Nous allons utiliser deux modèles séparés pour l’acteur et le critique.
Acteur :
Une couche linéaire cachée de taille in_dimx128
ReLU
Deux sortie correspondant à la moyenne de la densité et à son écart type correspondant chacune à une couche linéaire de taille 128xout_dim
Activation pour la moyenne : tanh
Activation pour écart type : softplus puis exponentielle
Critique :
Une couche linéaire cachée de taille in_dimx128
Une couche linéaire de taille 128x1 donnant la valeur
Coding Task :¶
Compléter les modèles suivant pour l’acteur et le critique
def initialize_uniformly(layer: nn.Linear, init_w: float = 3e-3):
"""Initialize the weights and bias in [-init_w, init_w]."""
layer.weight.data.uniform_(-init_w, init_w)
layer.bias.data.uniform_(-init_w, init_w)
class Actor(nn.Module):
def __init__(self, in_dim: int, out_dim: int):
"""Initialize."""
super(Actor, self).__init__()
self.hidden1 = ...
self.mu_layer = ...
self.log_std_layer = ...
initialize_uniformly(self.mu_layer)
initialize_uniformly(self.log_std_layer)
def forward(self, state: torch.Tensor) -> torch.Tensor:
"""Forward method implementation."""
x = ..
mu = ...
log_std = ...
std = ...
dist = Normal(mu, std)
action = dist.sample()
return action, dist
class Critic(nn.Module):
def __init__(self, in_dim: int):
"""Initialize."""
super(Critic, self).__init__()
self.hidden1 = ...
self.out = ...
initialize_uniformly(self.out)
def forward(self, state: torch.Tensor) -> torch.Tensor:
"""Forward method implementation."""
x = ...
value = ...
return value
A2C Agent¶
La classe A2CAgent permet de sélectioner une action, avancer dans l’environnement, mettre à jour le modèle, l’entrâiner et le tester.
Method |
Note |
---|---|
select_action |
select an action from the input state. |
step |
take an action and return the response of the env. |
update_model |
update the model by gradient descent. |
train |
train the agent during num_frames. |
test |
test the agent (1 episode). |
plot |
plot the training progresses. |
Coding Task¶
Compléter la méthode update model
:
Estimer la valeur grâce au critique
Calculer la target du crtique avec une approche TD
Minimiser la loss du critique et mettre à jour la critique
Caculer l’avantage
- caculer la loss pour le modèle gérant la politique.NB: Ne pas oublier la régularisation donnée par l’entropie
mettre à jour la politique
class A2CAgent:
"""A2CAgent interacting with environment.
Atribute:
env (gym.Env): openAI Gym environment
gamma (float): discount factor
entropy_weight (float): rate of weighting entropy into the loss function
device (torch.device): cpu / gpu
actor (nn.Module): target actor model to select actions
critic (nn.Module): critic model to predict state values
actor_optimizer (optim.Optimizer) : optimizer of actor
critic_optimizer (optim.Optimizer) : optimizer of critic
transition (list): temporory storage for the recent transition
total_step (int): total step numbers
is_test (bool): flag to show the current mode (train / test)
"""
def __init__(self, env: gym.Env, gamma: float, entropy_weight: float):
"""Initialize."""
self.env = env
self.gamma = gamma
self.entropy_weight = entropy_weight
# device: cpu / gpu
self.device = torch.device(
"cuda" if torch.cuda.is_available() else "cpu"
)
print(self.device)
# networks
obs_dim = env.observation_space.shape[0]
action_dim = env.action_space.shape[0]
self.actor = Actor(obs_dim, action_dim).to(self.device)
self.critic = Critic(obs_dim).to(self.device)
# optimizer
self.actor_optimizer = optim.Adam(self.actor.parameters(), lr=1e-4)
self.critic_optimizer = optim.Adam(self.critic.parameters(), lr=1e-3)
# transition (state, log_prob, next_state, reward, done)
self.transition: list = list()
# total steps count
self.total_step = 0
# mode: train / test
self.is_test = False
def select_action(self, state: np.ndarray) -> np.ndarray:
"""Select an action from the input state."""
state = torch.FloatTensor(state).to(self.device)
action, dist = self.actor(state)
selected_action = dist.mean if self.is_test else action
if not self.is_test:
log_prob = dist.log_prob(selected_action).sum(dim=-1)
self.transition = [state, log_prob]
return selected_action.clamp(-2.0, 2.0).cpu().detach().numpy()
def step(self, action: np.ndarray) -> Tuple[np.ndarray, np.float64, bool]:
"""Take an action and return the response of the env."""
next_state, reward, done, _, _ = self.env.step(action)
if not self.is_test:
self.transition.extend([next_state, reward, done])
return next_state, reward, done
def update_model(self) -> Tuple[torch.Tensor, torch.Tensor]:
"""Update the model by gradient descent."""
state, log_prob, next_state, reward, done = self.transition
############ Votre code ici #####################
#### Critique : estimation et mise à jour ######
# Q_t = r + gamma * V(s_{t+1}) if state != Terminal
# = r otherwise
# update value
# advantage = Q_t - V(s_t)
# Policy loss : sum(-advantage * log(politique) )
# update policy
#########################################################
return policy_loss.item(), value_loss.item()
def train(self, num_frames: int, plotting_interval: int = 200):
"""Train the agent."""
self.is_test = False
actor_losses, critic_losses, scores = [], [], []
state = self.env.reset()[0]
score = 0
for self.total_step in range(1, num_frames + 1):
action = self.select_action(state)
next_state, reward, done = self.step(action)
actor_loss, critic_loss = self.update_model()
actor_losses.append(actor_loss)
critic_losses.append(critic_loss)
state = next_state
score += reward
# if episode ends
if done:
state = env.reset()[0]
scores.append(score)
score = 0
# plot
if self.total_step % plotting_interval == 0:
self._plot(self.total_step, scores, actor_losses, critic_losses)
self.env.close()
def test(self):
"""Test the agent."""
self.is_test = True
state = self.env.reset()[0]
done = False
score = 0
frames = []
while not done:
frames.append(self.env.render(mode="rgb_array"))
action = self.select_action(state)
next_state, reward, done = self.step(action)
state = next_state
score += reward
print("score: ", score)
self.env.close()
return frames
def _plot(
self,
frame_idx: int,
scores: List[float],
actor_losses: List[float],
critic_losses: List[float],
):
"""Plot the training progresses."""
def subplot(loc: int, title: str, values: List[float]):
plt.subplot(loc)
plt.title(title)
plt.plot(values)
subplot_params = [
(131, f"frame {frame_idx}. score: {np.mean(scores[-10:])}", scores),
(132, "actor_loss", actor_losses),
(133, "critic_loss", critic_losses),
]
clear_output(True)
plt.figure(figsize=(30, 5))
for loc, title, values in subplot_params:
subplot(loc, title, values)
plt.show()
Environment¶
L’environnement ici est continue est consiste en un pendule inversé.
env_id = "Pendulum-v1"
env = gym.make(env_id,render_mode="rgb_array")
Pour initialiser l’agent nous pouvons utiliser le code suivant
num_frames = 100000
gamma = 0.9
entropy_weight = 1e-2
agent = A2CAgent(env, gamma, entropy_weight)
cuda
Enfin l’entrainement se fait avec la méthode train
agent.train(num_frames)
Test¶
from gym.wrappers import RecordVideo
env = RecordVideo(agent.env,video_folder = './video')
frames = agent.test()
score: -135.17415452820998
Render¶
! pip install JSAnimation
from matplotlib import animation
from JSAnimation.IPython_display import display_animation
from IPython.display import display
def display_frames_as_gif(frames):
"""Displays a list of frames as a gif, with controls."""
patch = plt.imshow(frames[0])
plt.axis('off')
def animate(i):
patch.set_data(frames[i])
anim = animation.FuncAnimation(
plt.gcf(), animate, frames = len(frames), interval=50
)
display(display_animation(anim, default_mode='loop'))
# display
display_frames_as_gif(frames)