Apprenez Pendulum-v0 avec DDPG

Aperçu

Dans des méthodes telles que DQN, les Q (s, a) de chaque état étaient calculés par une politique, et l'action qui maximisait la valeur Q était sélectionnée et agie, mais cela ne pouvait gérer que des actions discrètes. D'autre part, DDPG a répondu en paramétrant la politique et en sortant l'action directement, au lieu de rechercher l'action qui maximise la valeur Q afin de correspondre à l'espace d'action continue. C'est donc une mesure décisive.

la mise en oeuvre

Tampon de relecture

C'est un tampon de relecture familier dans l'apprentissage par renforcement profond. L'état actuel, l'action à ce moment-là, l'état suivant, la récompense immédiate et l'état terminal sont enregistrés sous la forme d'un taple.

from collections import deque, namedtuple
import random

Transition = namedtuple('Transition', ('state', 'action', 'next_state', 'reward', 'done'))

class ReplayBuffer(object):
    def __init__(self, capacity=1e6):
        self.capacity = capacity
        self.memory = deque([], maxlen=int(capacity))

    def append(self, *args):
        transition = Transition(*args)
        self.memory.append(transition)

    def sample(self, batch_size):
        return random.sample(self.memory, batch_size)

    def reset(self):
        self.memory.clear()

    def length(self):
        return len(self.memory)

    def __len__(self):
        return len(self.memory)

modèle

Dans DDPG, il y a Actor $ \ mu (s) $ qui sort l'action en continu à partir de l'état actuel et Critic $ Q (s, a) $ qui sort la valeur Q de l'état et de l'action actuels. L'initialisation des poids de chaque couche est conforme au papier d'origine, veuillez donc vérifier ici pour plus de détails (lien ci-dessous). Ce qui est caractéristique, c'est qu'il y a un tanh dans la couche finale de l'acteur, et lors de la réception d'une action dans Critic, il est reçu dans la deuxième couche. Si vous expérimentez avec Pendulum, vous pourrez peut-être écrire 2 dans la sortie car la plage d'action est [-2, 2].

import torch
import torch.nn as nn
import torch.nn.functional as F
import numpy as np

def init_weight(size):
    f = size[0]
    v = 1. / np.sqrt(f)
    return torch.tensor(np.random.uniform(low=-v, high=v, size=size), dtype=torch.float)

class ActorNetwork(nn.Module):
    def __init__(self, num_state, num_action, hidden1_size=400, hidden2_size=300, init_w=3e-3):
        super(ActorNetwork, self).__init__()
        self.fc1 = nn.Linear(num_state[0], hidden1_size)
        self.fc2 = nn.Linear(hidden1_size, hidden2_size)
        self.fc3 = nn.Linear(hidden2_size, num_action[0])

        self.num_state = num_state
        self.num_action = num_action

        self.fc1.weight.data = init_weight(self.fc1.weight.data.size())
        self.fc2.weight.data = init_weight(self.fc2.weight.data.size())
        self.fc3.weight.data.uniform_(-init_w, init_w)

    def forward(self, x):
        h = F.relu(self.fc1(x))
        h = F.relu(self.fc2(h))
        y = torch.tanh(self.fc3(h))  #Puis-je multiplier par 2?
        return y

class CriticNetwork(nn.Module):
    def __init__(self, num_state, num_action, hidden1_size=400, hidden2_size=300, init_w=3e-4):
        super(CriticNetwork, self).__init__()
        self.fc1 = nn.Linear(num_state[0], hidden1_size)
        self.fc2 = nn.Linear(hidden1_size+num_action[0], hidden2_size)
        self.fc3 = nn.Linear(hidden2_size, 1)
        
        self.num_state = num_state
        self.num_action = num_action

        self.fc1.weight.data = init_weight(self.fc1.weight.data.size())
        self.fc2.weight.data = init_weight(self.fc2.weight.data.size())
        self.fc3.weight.data.uniform_(-init_w, init_w)

    def forward(self, x, action):
        h = F.relu(self.fc1(x))
        h = F.relu(self.fc2(torch.cat([h, action], dim=1)))
        y = self.fc3(h)
        return y

Agent

Dans l'agent, lors de la sélection d'une action, l'action devient décisive telle quelle, alors ajoutez du bruit $ \ mathcal {N} $. Le bruit à ce moment est [processus Ornstein-Woolenbeck](https://ja.wikipedia.org/wiki/%E3%82%AA%E3%83%AB%E3%83%B3%E3%82%B7% E3% 83% A5% E3% 82% BF% E3% 82% A4% E3% 83% B3% EF% BC% 9D% E3% 82% A6% E3% 83% BC% E3% 83% AC% E3% 83% B3% E3% 83% 99% E3% 83% 83% E3% 82% AF% E9% 81% 8E% E7% A8% 8B). Je ne connais pas les détails. Considérez-le comme un bruit qui s'approche de la moyenne au fil du temps. Je ne sais pas.

a = \mu(s) + \mathcal{N}

Pour l'apprentissage de chaque modèle, Critic met à jour le modèle en trouvant le gradient afin de minimiser l'erreur TD, similaire à DQN. La fonction de perte est la suivante. N est la taille du lot.

L = \frac{1}{N} \sum_{i=1}^N (r_i + \gamma Q^{\prime}(s_{i+1}, \mu^{\prime}(s_{i+1})) - Q(s_i, a_i))^2

L'acteur met à jour le modèle pour maximiser la valeur Q. Notez que la perte sera négative car elle sera maximisée à ce moment. La fonction objectif est la suivante.

J = \frac{1}{N}\sum_{i=1}^N Q(s_{i}, \mu{s_i})

Dans la fonction objectif ci-dessus, celui avec un tiret est le réseau cible. Ceci est souvent utilisé pour stabiliser l'apprentissage. Dans DQN etc., ce réseau cible est mis à jour toutes les quelques époques, alors que dans DDPG, l'hyper paramètre $ \ tau (\ ll 1) $ est utilisé.

\theta \gets \tau \theta + (1 - \tau) \theta^{\prime}

Il sera mis à jour lentement, comme. Cela stabilise l'apprentissage, mais il semble que le temps d'apprentissage sera légèrement plus long.

import torch
import torch.nn.functional as F
import numpy as np
import copy

class OrnsteinUhlenbeckProcess:
    def __init__(self, theta=0.15, mu=0.0, sigma=0.2, dt=1e-2, x0=None, size=1, sigma_min=None, n_steps_annealing=1000):
        self.theta = theta
        self.mu = mu
        self.sigma = sigma
        self.dt = dt
        self.x0 = x0
        self.size = size
        self.num_steps = 0

        self.x_prev = self.x0 if self.x0 is not None else np.zeros(self.size)

        if sigma_min is not None:
            self.m = -float(sigma - sigma_min) / float(n_steps_annealing)
            self.c = sigma
            self.sigma_min = sigma_min
        else:
            self.m = 0
            self.c = sigma
            self.sigma_min = sigma

    def current_sigma(self):
        sigma = max(self.sigma_min, self.m * float(self.num_steps) + self.c)
        return sigma

    def sample(self):
        x = self.x_prev + self.theta * (self.mu - self.x_prev) * self.dt + self.current_sigma() * np.sqrt(self.dt) * np.random.normal(size=self.size)
        self.x_prev = x
        self.num_steps += 1
        return x


class DDPG:
    def __init__(self, actor, critic, optimizer_actor, optimizer_critic, replay_buffer, device, gamma=0.99, tau=1e-3, epsilon=1.0, batch_size=64):
        self.actor = actor
        self.critic = critic
        self.actor_target = copy.deepcopy(self.actor)
        self.critic_target = copy.deepcopy(self.critic)
        self.optimizer_actor = optimizer_actor
        self.optimizer_critic = optimizer_critic
        self.replay_buffer = replay_buffer
        self.device = device
        self.gamma = gamma
        self.tau = tau
        self.epsilon = epsilon
        self.batch_size = batch_size
        self.random_process = OrnsteinUhlenbeckProcess(size=actor.num_action[0])

        self.num_state = actor.num_state
        self.num_action = actor.num_action

    def add_memory(self, *args):
        self.replay_buffer.append(*args)
    
    def reset_memory(self):
        self.replay_buffer.reset()

    def get_action(self, state, greedy=False):
        state_tensor = torch.tensor(state, dtype=torch.float, device=self.device).view(-1, *self.num_state)
        action = self.actor(state_tensor)
        if not greedy:
            action += self.epsilon*torch.tensor(self.random_process.sample(), dtype=torch.float, device=self.device)

        return action.squeeze(0).detach().cpu().numpy()

    def train(self):
        if len(self.replay_buffer) < self.batch_size:
            return None
        transitions = self.replay_buffer.sample(self.batch_size)
        batch = Transition(*zip(*transitions))

        state_batch = torch.tensor(batch.state, device=self.device, dtype=torch.float)
        action_batch = torch.tensor(batch.action, device=self.device, dtype=torch.float)
        next_state_batch = torch.tensor(batch.next_state, device=self.device, dtype=torch.float)
        reward_batch = torch.tensor(batch.reward, device=self.device, dtype=torch.float).unsqueeze(1)
        not_done = np.array([(not done) for done in batch.done])
        not_done_batch = torch.tensor(not_done, device=self.device, dtype=torch.float).unsqueeze(1)

        # need to change
        qvalue = self.critic(state_batch, action_batch)
        next_qvalue = self.critic_target(next_state_batch, self.actor_target(next_state_batch))
        target_qvalue = reward_batch + (self.gamma * next_qvalue * not_done_batch) 

        critic_loss = F.mse_loss(qvalue, target_qvalue)
        self.optimizer_critic.zero_grad()
        critic_loss.backward()
        self.optimizer_critic.step()

        actor_loss = -self.critic(state_batch, self.actor(state_batch)).mean()
        self.optimizer_actor.zero_grad()
        actor_loss.backward()
        self.optimizer_actor.step()

        # soft parameter update
        for target_param, param in zip(self.actor_target.parameters(), self.actor.parameters()):
            target_param.data.copy_(target_param.data * (1.0 - self.tau) + param.data * self.tau)
        for target_param, param in zip(self.critic_target.parameters(), self.critic.parameters()):
            target_param.data.copy_(target_param.data * (1.0 - self.tau) + param.data * self.tau)

Apprentissage

Il n'y a rien de nouveau à ce sujet. Comme d'autres algorithmes d'apprentissage par renforcement, il reçoit l'état de l'environnement et vous donne l'impression d'agir et d'apprendre. Chaque hyper paramètre est conforme au papier d'origine. (Peut-être)

import torch
import torch.optim as optim
import gym

max_episodes = 300
memory_capacity = 1e6  #Capacité tampon
gamma = 0.99  #Taux de remise
tau = 1e-3  #Taux de mise à jour cible
epsilon = 1.0  #Si vous voulez gâcher la quantité de bruit, vous n'en avez probablement pas besoin
batch_size = 64
lr_actor = 1e-4
lr_critic = 1e-3
logger_interval = 10
weight_decay = 1e-2

env = gym.make('Pendulum-v0')
num_state = env.observation_space.shape
num_action = env.action_space.shape
max_steps = env.spec.max_episode_steps

device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')

actorNet = ActorNetwork(num_state, num_action).to(device)
criticNet = CriticNetwork(num_state, num_action).to(device)
optimizer_actor = optim.Adam(actorNet.parameters(), lr=lr_actor)
optimizer_critic = optim.Adam(criticNet.parameters(), lr=lr_critic, weight_decay=weight_decay)
replay_buffer = ReplayBuffer(capacity=memory_capacity)
agent = DDPG(actorNet, criticNet, optimizer_actor, optimizer_critic, replay_buffer, device, gamma, tau, epsilon, batch_size)

for episode in range(max_episodes):
    observation = env.reset()
    total_reward = 0

    for step in range(max_steps):
        action = agent.get_action(observation)
        next_observation, reward, done, _ = env.step(action)
        total_reward += reward
        agent.add_memory(observation, action, next_observation, reward, done)

        agent.train()

        observation = next_observation

        if done:
            break

    if episode % logger_interval == 0:
        print("episode:{} total reward:{}".format(episode, total_reward))

for episode in range(3):
    observation = env.reset()
    env.render()
    for step in range(max_steps):
        action = agent.get_action(observation, greedy=True)
        next_observation, reward, done, _ = env.step(action)
        observation = next_observation
        env.render()

        if done:
            break

env.close()

Si vous modifiez l'environnement de gym.make, vous devriez pouvoir étudier dans d'autres environnements.

résultat

C'est un graphique des récompenses cumulées et des épisodes d'apprentissage. Je pense que vous apprenez d'une bonne manière. Figure_1-1.png Il a été construit correctement. Génial. (Gif n'est pas posté car je ne sais pas comment le faire)

en conclusion

En gros, si vous exécutez le code ci-dessus tel quel dans un fichier, vous pouvez vérifier l'opération. Comme l'implémentation était complète, certaines variables supplémentaires, etc. subsistent. Avec Pendulum-v0, vous ne pouvez apprendre qu'avec le processeur, même si vous n'êtes pas dans un environnement où le GPU est disponible. Cependant, l'apprentissage peut devenir un peu instable, veuillez donc réessayer. S'il y a une opportunité, nous mettrons en œuvre d'autres méthodes.

Les références

Continuous Control with Deep Reinforcement Learning

Recommended Posts

Apprenez Pendulum-v0 avec DDPG
Apprenez Python avec ChemTHEATER
Apprenez Zundokokiyoshi en utilisant LSTM
Pandas apprenant avec la chimioinfomatique
Apprentissage Scikit-Learn avec la chimioinfomatique
Apprenez avec Chemo Informatics Matplotlib
Apprenez avec Chemo Informatics NumPy
DCGAN avec TF Learn
Apprenez librosa avec un tutoriel 1
Apprenez les orbites elliptiques avec Chainer
Apprenez de nouvelles données avec PaintsChainer
Apprendre avec Causal ML Package Meta-Learner
Apprenez avec FizzBuzz Iterator, Generator, Decorator
Apprenez avec les réseaux convolutifs PyTorch Graph
[TensorFlow 2] Apprendre RNN avec perte CTC
Apprendre la catégorisation de documents avec la CLI spaCy
Premiers pas avec python3 # 1 Apprenez les connaissances de base
Apprenez à coloriser les images monochromes avec Chainer
Apprenez les données distribuées avec TensorFlow Y = 2X
Apprendre Python! Comparaison avec Java (fonction de base)
Apprenez avec Supra Nervous Weakness! La théorie des graphes
Se préparer à apprendre les indicateurs techniques avec TFlearn
Apprenez le modèle de conception "Façade" avec Python