Breakout DQN : L'intérieur du Deep Q Network

Suite du sujet Breakout DQN : Agent avec Double DQN :

Pour créer notre réseau de neurones, on se basera sur l’architecture et les hyperparamètres utilisés dans cet article à la page 9.

La classe BreakoutDQN

Le cœur du Deep Q Network (DQN)

Voici le code de notre réseau de neurones:

import keras
from keras import backend as K
import numpy as np

class BreakOutDQN:
    def __init__(self):
        nbr_actions=4
        #Entrées
        states = keras.layers.Input((84,84,4), name='frames')# 4 frames de taille 84*84
        normalisation = keras.layers.Lambda(lambda x: x / 255.0)(states)# [0;255]->[0;1]
        actions = keras.layers.Input((nbr_actions,), name='actions_oh')
        #les 4 frames passent dans 3 couches de CNN
        #1er filtre : 32 filtres de taille 8*8 avec un pas de 4*4
        conv_1 = keras.layers.convolutional.Conv2D(32, (8, 8), activation='relu', strides=(4, 4), kernel_initializer='he_normal')(normalisation)
        conv_2 = keras.layers.convolutional.Conv2D(64, (4, 4), activation='relu', strides=(2, 2), kernel_initializer='he_normal')(conv_1)
        conv_3 = keras.layers.convolutional.Conv2D(64, (3, 3), activation='relu', strides=(1, 1), kernel_initializer='he_normal')(conv_2)
        # On "aplati" la sortie
        conv_flattened = keras.layers.core.Flatten()(conv_3)
        #couche cachée de 512 neurones 
        q_actions = keras.layers.Dense(512, activation='relu', kernel_initializer='he_normal')(conv_flattened)
        #Sortie filtré avec le masque (l'action choisi en format one-hot)
        q_action = keras.layers.Multiply()([q_actions, actions])
        #on crée le modèle et on choisi l'optimiseur
        self.model = keras.models.Model(inputs=[states, actions], outputs=q_action)
        optimizer =keras.optimizers.RMSprop(lr=0.00025, rho=0.95, epsilon=0.01)
        self.model.compile(optimizer, loss=huber_loss)

La normalisation des données se fait dans le DQN

Normaliser les valeurs des frames en dehors de cette classe rendra l’apprentissage 2 fois plus lent pour 2 raisons. La première est que seul le code du DQN sera exécuté sur le GPU, on ne bénéficiera donc pas de l’avantage du GPU pour le calcul parallèle. La seconde est que le calcul sur le CPU transforme l’encodage des données de 8 bit en 64 bit: en procédant ainsi, le transfert des données de la RAM au GPU sera plus long que le temps de calcul du GPU, on est en présence d’un goulot d’étranglement.

La fonction de perte de Huber

Pour éviter que nos valeurs Q(s,a) ne deviennent trop grandes on va utiliser une fonction de perte (loss function) qui a une propriété intéressante: la perte sera quadratique pour les petites erreurs et linéaire pour les grosses. Cette fonction s’appelle Hubert Loss et son équation est donnée ci dessous:

f(x) = \left\{ \begin{array}{} \frac{x^2}{2} & \mbox{si } x<\delta \\ \delta(|x|-\frac{\delta}{2}) & \mbox{sinon.} \end{array} \right.

\delta représente la limite entre les petites et grosses erreurs. Dans notre cas on prendra \delta=1 car les récompenses de notre agent sont soit 0 soit 1.
La dérivée 1ère de la fonction est continue en \delta. Le graphe ci dessous permet de comparer la perte de Hubert (en vert) et la perte Mean Squared Error (en bleu).
Huber_loss

Implémentation personnalisée de Huber Loss avec Keras

La sortie de notre réseau de neurone a une particularité: on a toujours 3 sorties sur 4 qui sont nulles. On devra donc écrire notre propre fonction de perte qui prendra en compte cette subtilité.

Voici le code de notre fonction de perte personnalisée:

# rappel K=keras.backend
def huber_loss(y_true,y_predict):
    delta=1.0
    error=y_true-y_predict
    error=K.tf.reduce_sum(error,axis=1)
    cond=K.abs(error)<delta
    squared_loss=0.5*K.square(error)
    linear_loss=delta*(K.abs(error)-0.5*delta)
    return K.tf.where(cond,squared_loss, linear_loss)

y_true et y_predict ont la même dimension (batch_size,4), error possède donc aussi la dimension (batch_size,4). On élimine les 0 inutiles en sommant les valeurs q des 4 actions, error est maintenant de dimension batch_size. Le reste du code est simplement la définition de la perte de Hubert écrit avec le backend de keras. Il est obligatoire d’utiliser ce dernier pour que keras puisse créer un graphe de notre fonction et ensuite la dériver durant la rétropropagation du gradient.
Ce code se place avant notre classe.

Prédictions des actions par le DQN

def GetAction(self,state):
        state4predict=np.expand_dims(state,axis=0)
        action_oh=np.ones((1,4))
        q_actions=self.model.predict([state4predict, action_oh],batch_size=1)
        q_actions=np.squeeze(q_actions, axis=0)
        action=np.argmax(q_actions)
        return action

La variable state possède 3 dimensions (84*84*4) hors le DQN s’attend à recevoir des données à 4 dimensions: une pour le nombre de données envoyées et les trois dimensions de state. Pour cette raison on rajoute la dimension manquante avec expand_dims, on envoie l’état au DQN et on enlève ensuite de notre résultat la dimension superflue avec squeeze.

La méthode où notre DQN doit prédire la meilleure action de plusieurs états:

def GetActions(self,states,nbr_xp):      
        action_oh=np.ones((nbr_xp,4))
        actions=np.zeros(nbr_xp)
        q_actions=self.model.predict([states, action_oh],batch_size=nbr_xp)
        for i in range(0,nbr_xp):
            actions[i]=np.argmax(q_actions[i])
        return actions

Sauvegarde du DQN

On a 2 options : soit sauvegarder uniquement les poids du réseau de neurones soit sauvegarder les poids mais aussi l’architecture du réseau de neurones. J’ai choisi de sauvegarder uniquement les poids:

def SaveWeights(self,file_name):
    self.model.save_weights(file_name+".h5")

def LoadWeights(self,file_name):
    self.model.load_weights(file_name+".h5")

L’extension de fichier .h5 indique au programme de stocker les poids sous le format HDF5.

Mise à jour des poids du DQN

def GetQValues(self, states_prime, game_over,n):
    actions=np.ones((nbr_xp,4))
    QValues_prime= self.model.predict([states_prime, actions], batch_size=n)
    for i in range(0,nbr_xp):
        if game_over[i]:
            #si on perds la qualité de l'action est nulle
            QValues_prime[i][:]=0.0
    return QValues_prime

Cette méthode prédit la qualité des actions de l’état suivant, si l’état suivant correspond à une défaite on fixe à 0 ses valeurs q sinon les valeurs restent inchangées.

def SetWeights(self, states, actions, QValues_true, nbr_xp):
        self.model.fit([states, actions], QValue_true, batch_size=nbr_xp, epochs=1, verbose=0)

On est dans le cadre d’un apprentissage supervisé:
-le DQN prend des états et des actions en entrée et produits des QValues.
-QValues_true correspond aux “vraies” valeurs que le DQN doit estimer et qu’il optimise via l’algorithme de descente de gradient.

Amélioration: Le Dueling DQN

En cours.