Tout d'abord, comme valeur initiale, considérons le cas où il y a plusieurs options et l'une d'elles est sélectionnée au hasard avec la même probabilité.

def normalize_pi(theta):
    [m, n] = theta.shape
    pi = np.zeros((m, n))
    for i in range(0, m):
        pi[i, :] = theta[i, :] / np.nansum(theta[i, :])
        
    return np.nan_to_num(pi)

De cette façon, la valeur initiale de pi`` pi_zero est obtenue.

pi_zero = normalize_pi(theta_zero)

Visualisons «pi_zero».

visualize_matrix(pd.DataFrame(pi_zero, columns=location[:, 0]))

Notez que «pi_zero» n'est pas une matrice symétrique. Par exemple, il y a un côté de Sapporo à Aomori et un côté d'Aomori à Sapporo, mais les poids ne sont pas les mêmes. En effet, dans ce graphique, il n'y a qu'une seule option de Sapporo à la ville voisine, mais il existe plusieurs options d'Aomori à la ville suivante.

À propos du "comportement" et de "l'état"

Dans l'apprentissage par renforcement, «comportement» et «état» sont traités séparément. À la suite de la sélection d'une certaine action dans un certain état, il passe à un état différent.

Les exemples traités dans cet article sont des cas particuliers où il n'est pas nécessaire de faire la distinction entre «comportement» et «état». En d'autres termes, «état» se réfère à la «ville à ce moment-là», «action» se réfère à la «ville suivante», et c'est le prochain «état». Le code est plus simple car vous n'avez pas à faire la distinction entre les actions et les états.

Marche aléatoire

Tout d'abord, essayons un voyage à travers le Japon avec cette politique pi encore pi_zero. Une fois que vous atteignez une ville, c'est une promenade aléatoire qui décide au hasard dans quelle ville aller.

Choisissez probablement l'action suivante

Créez une fonction qui sélectionne la ville suivante en fonction de la probabilité indiquée par la politique «pi».

def get_next(town, pi):
    return np.random.choice(range(len(pi)), p=pi[town, :])

Par exemple, selon «pi_zero», la prochaine ville de Tokyo (n ° 12) est calculée comme suit.

get_next(12, pi_zero)

13

Bien sûr, le résultat changera à chaque exécution car il est décidé au hasard.

Recherche par marche aléatoire

Maintenant, recherchons par marche aléatoire.

def explore(pi, start=0, goal=46): #Le départ est 0 (Sapporo), l'objectif est 46 (Naha)
    route = [start] #Liste pour mettre l'histoire de voyage
    town = start
    while True:
        next_town = get_next(town, pi) #Déterminez la prochaine ville
        if next_town in route: #Si vous visitez la même ville deux fois
            break #Jeu terminé

        route.append(next_town) #Ajouter la prochaine ville à l'histoire

        if next_town == goal: #Arriver au but
            break #Toutes nos félicitations
        else: #Pas encore un objectif
            town = next_town #Définir "ville suivante" comme "position actuelle"

    return route

Commencez une marche aléatoire selon pi_zero

route = explore(pi_zero)
route

[0, 1, 2, 3, 4, 5]

C'est aléatoire, donc bien sûr, le résultat changera à chaque exécution. S'il est difficile d'imaginer uniquement le numéro de la ville, il sera plus facile d'imaginer si vous le convertissez en nom de ville comme suit.

[location[i, 0] for i in route]

['Sapporo', 'Aomori', 'Morioka', 'Sendai', 'Akita', 'Yamagata']

Dans cet exemple, le jeu est terminé car j'ai marché de Sapporo à Yamagata et j'ai sélectionné par erreur une ville visitée à la destination suivante.

Au fait, pouvons-nous vraiment atteindre de Sapporo à Naha avec une promenade aussi aléatoire?

Index d'évaluation des résultats de recherche

Créons un moyen d'évaluer la qualité des résultats de recherche. Calculez la distance directe entre votre position actuelle et votre destination (Naha) et la distance totale parcourue entre votre emplacement de départ (Sapporo) et votre position actuelle.

def evaluate(route, goal=46):
    dist_goal = dist_mat[route[-1], goal]
    len_route = sum([dist_mat[route[i], route[i + 1]] for i in range(len(route) - 1)])
    return [dist_goal, len_route]

dist_goal est la" distance directe de votre position actuelle à votre destination (Naha) ", et len_route est la" distance totale de votre point de départ (Sapporo) à votre position actuelle ".

Ensuite, c'est un exemple d'exécution.

route = explore(pi_zero)
[location[i, 0] for i in route], evaluate(route)

(['Sapporo', 'Aomori'], [19.597025248636594, 2.3205099949149073])

Dans cet exemple, le jeu est terminé car j'ai choisi le chemin de Sapporo à Aomori et de retour d'Aomori à Sapporo. La distance entre l'emplacement actuel (Aomori) et la destination (Naha) est «19,597025248636594», et la distance entre le point de départ (Sapporo) et l'emplacement actuel (Aomori) est «2,3205099949149073».

Le but d'un voyage à travers le Japon est de trouver un itinéraire où ce dist_goal est nul, puis de trouver celui avec le plus petit len_route. Un len_route plus court n'est pas une bonne chose.

À partir de maintenant, je vais essayer le jeu de passer de Sapporo à Naha plusieurs fois, mais lorsque j'obtiens un certain résultat, créons une fonction pour juger si c'est la meilleure à ce jour.

best_dist_goal = 1000000 #Initialiser une valeur incroyablement élevée
best_len_route = 1000000 #Initialiser une valeur incroyablement élevée

def is_best_ever():
    if best_dist_goal >= dist_goal: #Meilleur dist_Si égal ou inférieur à la valeur de l'objectif
        if best_dist_goal > dist_goal: #Meilleur dist du passé_Si inférieur à la valeur de l'objectif
            return True #Le meilleur
        elif best_len_route > len_route: #Meilleur len du passé_S'il est inférieur à la valeur de l'itinéraire
            return True #Le meilleur
        else:
            return False #Pas le meilleur
    else:
        return False #Pas le meilleur

Exécution de la recherche

Maintenant que nous sommes prêts, nous allons répéter la recherche de marche aléatoire 50 000 fois.

%%time #Mesurer le temps d'exécution
pi = pi_zero.copy() #Initialisation de pi
theta = theta_zero.copy() #initialisation de theta

best_dist_goal = 1000000 #Initialiser une valeur incroyablement élevée
best_len_route = 1000000 #Initialiser une valeur incroyablement élevée

dist_goal_history0 = [] # dist_histoire du but
len_route_history0 = [] # len_historique de l'itinéraire
best_route0 = [] #Meilleur itinéraire

for itera in range(50000): #Répétez 50000 fois
    route = explore(pi) #chercher
    dist_goal, len_route = evaluate(route) #Évaluation
    dist_goal_history0.append(dist_goal) #Record
    len_route_history0.append(len_route) #Record
    
    if is_best_ever(): #Si le meilleur du passé
        best_dist_goal = dist_goal #Meilleur dist_goal
        best_len_route = len_route #Meilleur len_route
        best_route0 = route #Meilleur itinéraire

CPU times: user 10 s, sys: 118 ms, total: 10.2 s
Wall time: 11.8 s

Résultats de recherche

Créez une fonction pour visualiser le résultat. Tout d'abord, une fonction qui visualise la distribution du dist_goal`` len_route obtenu

def draw_histgrams(dist_goal_history, len_route_history):
    plt.figure(figsize=(max(len_route_history) / 4, max(dist_goal_history) / 4))
    x_max = max(dist_goal_history + len_route_history)
    ax = plt.subplot(2,1,1)
    ax.hist(dist_goal_history, label='dist_goal', bins=20)
    ax.set_xlim([0, x_max])
    ax.grid()
    ax.legend()
    ax = plt.subplot(2,1,2)
    ax.hist(len_route_history, label='len_route', bins=20)
    ax.set_xlim([0, x_max])
    ax.grid()
    ax.legend()

draw_histgrams(dist_goal_history0, len_route_history0)

En partant de Sapporo, il est difficile d'atteindre Naha. Si vous avez de la chance, vous pouvez y arriver. Vous pouvez voir qu'il existe de nombreux cas où le jeu est terminé au début de la recherche.

Ensuite, regardons la relation de dist_goal`` len_route.

def draw_scatter(dist_goal_history, len_route_history):
    plt.figure(figsize=(max(len_route_history) / 4, max(dist_goal_history) / 4))
    plt.scatter(len_route_history, dist_goal_history, alpha=0.5)
    plt.ylabel('dist_goal')
    plt.xlabel('len_route')
    plt.ylim([0, max(dist_goal_history) + 1])
    plt.xlim([0, max(len_route_history) + 1])
    plt.grid()
    plt.show()

draw_scatter(dist_goal_history0, len_route_history0)

Vous pouvez voir que nous n'avons pas atteint Naha, et même si le jeu est terminé au même endroit, il existe un large éventail de distances de voyage (différents itinéraires sont sélectionnés).

Ensuite, voyons comment dist_goal`` len_route a changé.

def visualize_history(history, interval=50, window=50):
    plt.plot(range(0, len(history), interval),  #Maximum global
             [np.array(history)[:i].max() for i in range(len(history)) if (i%interval) == 1], label='max')
    plt.plot(range(window, len(history)+window, interval), 
             [np.array(history)[i:i + window].mean() for i in range(len(history)) if (i%interval) == 1], 
             label='mean(recent)') #Moyenne des intervalles récents
    plt.plot(range(0, len(history), interval), 
             [np.array(history)[:i].mean() for i in range(len(history)) if (i%interval) == 1], label='mean') #Valeur moyenne globale
    plt.plot(range(0, len(history), interval), 
             [np.array(history)[:i].min() for i in range(len(history)) if (i%interval) == 1], label='min') #Minimum global
    plt.legend()
    plt.grid()
    plt.show()

visualize_history(dist_goal_history0)

dist_goal met parfois à jour la valeur minimale, mais elle n'atteint pas Naha. Et quel que soit le nombre de fois que vous répétez la recherche, la valeur moyenne ne s'améliorera pas. C'est naturel parce que je n'ai rien appris.

visualize_history(len_route_history0)

De même, len_route met parfois à jour la valeur maximale, mais elle n'atteint pas Naha, et la valeur moyenne ne s'améliore pas quel que soit le nombre de fois que la recherche est répétée. C'est naturel parce que je n'ai rien appris.

Enfin, affichons le meilleur itinéraire trouvé dans cette promenade aléatoire.

def draw_route(route):
    plt.figure(figsize=(11, 9))
    for i in route:
        plt.text(location[i, 2], location[i, 1], location[i, 0], alpha=0.4, size=12)
    plt.grid()
    plt.plot([location[i][2] for i in route], [location[i][1] for i in route])
    plt.ylabel('latitude')
    plt.xlabel('longitude')
    plt.show()

draw_route(best_route0)

J'ai essayé très fort, mais je n'ai pu atteindre que Saga. Si vous avez de la chance, vous pouvez atteindre Naha. Mais tu ne l'apprends pas.

Méthode du gradient de politique

Il est maintenant temps de commencer à renforcer l'apprentissage </ b>. La méthode d'apprentissage par renforcement semble être grossièrement divisée en Méthode de gradient de politique </ b> et Méthode d'itération de valeur </ b>. Commençons par la méthode du gradient de politique.

def update_pi(pi, route, goal=46, eta=0.1): #Eta (η),eta) est le taux d'apprentissage
    new_pi = pi.copy() #Copier le tableau numpy
    for i in range(len(route) - 1): #Pour chaque côté de l'itinéraire
        town1 = route[i] #Ville d'origine du côté i
        town2 = route[i + 1] #Ville au bout du côté i
        new_pi[town1, town2] +=  eta / (dist_mat[route[-1], goal] + 1)
        #Plus le point final de l'itinéraire est proche de l'objectif, plus le score est élevé
    
    return normalize_pi(new_pi) #Mettre à jour pi

%%time
pi = pi_zero.copy()

best_dist_goal = 1000000
best_len_route = 1000000

dist_goal_history1 = []
len_route_history1 = []
best_route1 = []
for itera in range(50000):
    route = explore(pi)
    dist_goal, len_route = evaluate(route)
    dist_goal_history1.append(dist_goal)
    len_route_history1.append(len_route)
        
    pi = update_pi(pi, route)
    
    if is_best_ever():
        best_dist_goal = dist_goal
        best_len_route = len_route
        best_route1 = route

CPU times: user 59.9 s, sys: 340 ms, total: 1min
Wall time: 1min 1s

draw_histgrams(dist_goal_history1, len_route_history1)

draw_scatter(dist_goal_history1, len_route_history1)

visualize_history(dist_goal_history1)

visualize_history(dist_goal_history1[:5000])

visualize_history(len_route_history1)

visualize_history(len_route_history1[:5000])

draw_route(best_route1)

pi_pg = pi.copy()

visualize_matrix(pd.DataFrame(pi_pg, columns=location[:, 0]))

visualize_matrix(pd.DataFrame(pi_pg - pi_zero, columns=location[:, 0]))

def generate_Q(theta): #Générez des probabilités aléatoires inégales à partir des choix que vous pouvez faire
    [m, n] = theta.shape
    Q = np.zeros((m, n))
    rand = np.random.rand(m, n)
    for i in range(m):
        for j in range(n):
            if theta[i, j] == 1:
                Q[i, j] = rand[i, j]
        
    return normalize_pi(Q)

Q_zero = generate_Q(theta_zero)

visualize_matrix(pd.DataFrame(Q_zero, columns=location[:, 0]))

visualize_matrix(pd.DataFrame(Q_zero - pi_zero, columns=location[:, 0]))

def get_next(town, Q, epsilon=0.1):
    if np.random.rand() < epsilon:
        return np.random.choice(range(len(Q)), p=Q[town, :])
    else:
        return np.nanargmax(Q[town, :])

def sarsa(town, Q, prev_t, next_t, reward, eta=0.1, gamma=0.9, goal=46):
    if reward == 1: #dist_mat[town, goal] == 0:
        Q[prev_t, town] = Q[prev_t, town] + eta * (reward - Q[prev_t, town])
    elif reward >= 0:
        Q[prev_t, town] = Q[prev_t, town] + eta * (reward - Q[prev_t, town] + gamma * Q[town, next_t])
    else:
        Q[prev_t, town] = Q[prev_t, town] - eta * Q[prev_t, town]
        
    return normalize_pi(Q)

def explore_epsilon_greedy(Q, epsilon=0.1, eta=0.1, gamma=0.9, start=0, goal=46, min_epsilon=0.01):
    epsilon = max(epsilon, min_epsilon)
    route = [start]
    town = get_next(start, Q, epsilon)
    prev_t = start
    while True:
        if town in route:
            reward = -1
            Q = sarsa(town, Q, prev_t, next_t, reward, eta, gamma)
            break
        elif town == goal:
            reward = 1
            route.append(town)
            Q = sarsa(town, Q, prev_t, next_t, reward, eta, gamma)
            break
        else:
            reward = 0
            route.append(town)
            next_t = get_next(town, Q, epsilon)
            Q = sarsa(town, Q, prev_t, next_t, reward, eta, gamma)
            prev_t = town
            town = next_t
    
    return [route, Q]

%%time
eta = 0.1 #Taux d'apprentissage
gamma = 0.9 #Taux d'actualisation du temps
epsilon = 0.5
Q = Q_zero.copy()

best_dist_goal = 1000000
best_len_route = 1000000

best_route2 = []
dist_goal_history2 = []
len_route_history2 = []
for itera in range(50000):
    epsilon = epsilon * 0.99
    route, Q = explore_epsilon_greedy(Q, epsilon, eta, gamma)
    dist_goal, len_route = evaluate(route)
    dist_goal_history2.append(dist_goal)
    len_route_history2.append(len_route)

    if is_best_ever():
        best_dist_goal = dist_goal
        best_len_route = len_route
        best_route2 = route

CPU times: user 7min 50s, sys: 948 ms, total: 7min 50s
Wall time: 7min 53s

draw_histgrams(dist_goal_history2, len_route_history2)

draw_scatter(dist_goal_history2, len_route_history2) 

visualize_history(dist_goal_history2)

visualize_history(dist_goal_history2[:5000])

visualize_history(len_route_history2)

visualize_history(len_route_history2[:5000])

draw_route(best_route2)

Q_eg = Q.copy()

visualize_matrix(pd.DataFrame(Q_eg, columns=location[:, 0]))

def Q_learning(town, Q, prev_t, reward, eta=0.1, gamma=0.9, goal=46):
    if reward == 1: #dist_mat[town, goal] == 0:
        Q[prev_t, town] = Q[prev_t, town] + eta * (reward - Q[prev_t, town])
    elif reward >= 0:
        Q[prev_t, town] = Q[prev_t, town] + eta * (reward - Q[prev_t, town] + gamma * np.nanmax(Q[town, :]))
    else:
        Q[prev_t, town] = Q[prev_t, town] - eta * Q[prev_t, town]
        
    return normalize_pi(Q)

def explore_Q_learning(Q, epsilon=0.1, eta=0.1, gamma=0.9, start=0, goal=46):
    prev_t = start
    route = [start]
    town = get_next(start, Q, epsilon)
    while True:
        if town in route:
            reward = -1
            Q = Q_learning(town, Q, prev_t, reward, eta, gamma)
            break
        elif town == goal:
            reward = 1 
            route.append(town)
            Q = Q_learning(town, Q, prev_t, reward, eta, gamma)
            break
        else:
            reward = 0
            dist_goal, len_route = evaluate(route)
            if best_dist_goal > dist_goal:
                reward = 1
            route.append(town)
            next_t = get_next(town, Q, epsilon)
            Q = Q_learning(town, Q, prev_t, reward, eta, gamma)
            prev_t = town
            town = next_t
    
    return [route, Q]

%%time
eta = 0.1 #Taux d'apprentissage
gamma = 0.9 #Taux d'actualisation du temps
epsilon = 0.5
Q = Q_zero.copy()

best_dist_goal = 1000000
best_len_route = 1000000

best_route3 = []
dist_goal_history3 = []
len_route_history3 = []
for itera in range(50000):
    epsilon = epsilon * 0.99
    route, Q = explore_Q_learning(Q, epsilon, eta, gamma)
    dist_goal, len_route = evaluate(route)
    dist_goal_history3.append(dist_goal)
    len_route_history3.append(len_route)

    if is_best_ever():
        best_dist_goal = dist_goal
        best_len_route = len_route
        best_route3 = route

CPU times: user 9min 50s, sys: 1.41 s, total: 9min 52s
Wall time: 9min 54s

draw_histgrams(dist_goal_history3, len_route_history3)

draw_scatter(dist_goal_history3, len_route_history3) 

visualize_history(dist_goal_history3)

visualize_history(dist_goal_history3[:5000])

visualize_history(len_route_history3)

visualize_history(len_route_history3[:5000])

draw_route(best_route3)

Q_qlearn = Q.copy()

visualize_matrix(pd.DataFrame(Q_qlearn, columns=location[:, 0]))