Frozenlake 基準測試¶
在這篇文章中，我們將比較在強化學習 Gymnasium 套件中使用 Q-learning 演算法，於 FrozenLake 環境中不同地圖尺寸的效能。
依賴套件¶
首先，讓我們導入一些我們需要的依賴套件。
# Author: Andrea Pierré
# License: MIT License
from pathlib import Path
from typing import NamedTuple
import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
import seaborn as sns
from tqdm import tqdm
import gymnasium as gym
from gymnasium.envs.toy_text.frozen_lake import generate_random_map
sns.set_theme()
# %load_ext lab_black
我們將使用的參數¶
class Params(NamedTuple):
total_episodes: int # Total episodes
learning_rate: float # Learning rate
gamma: float # Discounting rate
epsilon: float # Exploration probability
map_size: int # Number of tiles of one side of the squared environment
seed: int # Define a seed so that we get reproducible results
is_slippery: bool # If true the player will move in intended direction with probability of 1/3 else will move in either perpendicular direction with equal probability of 1/3 in both directions
n_runs: int # Number of runs
action_size: int # Number of possible actions
state_size: int # Number of possible states
proba_frozen: float # Probability that a tile is frozen
savefig_folder: Path # Root folder where plots are saved
params = Params(
total_episodes=2000,
learning_rate=0.8,
gamma=0.95,
epsilon=0.1,
map_size=5,
seed=123,
is_slippery=False,
n_runs=20,
action_size=None,
state_size=None,
proba_frozen=0.9,
savefig_folder=Path("../../_static/img/tutorials/"),
)
params
# Set the seed
rng = np.random.default_rng(params.seed)
# Create the figure folder if it doesn't exist
params.savefig_folder.mkdir(parents=True, exist_ok=True)
FrozenLake 環境¶
env = gym.make(
"FrozenLake-v1",
is_slippery=params.is_slippery,
render_mode="rgb_array",
desc=generate_random_map(
size=params.map_size, p=params.proba_frozen, seed=params.seed
),
)
建立 Q 表格¶
在本教學中，我們將使用 Q-learning 作為我們的學習演算法，並使用 \(\epsilon\)-greedy 來決定在每個步驟選擇哪個動作。您可以查看參考文獻章節以複習一些理論。現在，讓我們建立我們的 Q 表格，並將其初始化為零，其中狀態數量作為行，動作數量作為列。
params = params._replace(action_size=env.action_space.n)
params = params._replace(state_size=env.observation_space.n)
print(f"Action size: {params.action_size}")
print(f"State size: {params.state_size}")
class Qlearning:
def __init__(self, learning_rate, gamma, state_size, action_size):
self.state_size = state_size
self.action_size = action_size
self.learning_rate = learning_rate
self.gamma = gamma
self.reset_qtable()
def update(self, state, action, reward, new_state):
"""Update Q(s,a):= Q(s,a) + lr [R(s,a) + gamma * max Q(s',a') - Q(s,a)]"""
delta = (
reward
+ self.gamma * np.max(self.qtable[new_state, :])
- self.qtable[state, action]
)
q_update = self.qtable[state, action] + self.learning_rate * delta
return q_update
def reset_qtable(self):
"""Reset the Q-table."""
self.qtable = np.zeros((self.state_size, self.action_size))
class EpsilonGreedy:
def __init__(self, epsilon):
self.epsilon = epsilon
def choose_action(self, action_space, state, qtable):
"""Choose an action `a` in the current world state (s)."""
# First we randomize a number
explor_exploit_tradeoff = rng.uniform(0, 1)
# Exploration
if explor_exploit_tradeoff < self.epsilon:
action = action_space.sample()
# Exploitation (taking the biggest Q-value for this state)
else:
# Break ties randomly
# Find the indices where the Q-value equals the maximum value
# Choose a random action from the indices where the Q-value is maximum
max_ids = np.where(qtable[state, :] == max(qtable[state, :]))[0]
action = rng.choice(max_ids)
return action
執行環境¶
讓我們實例化學習器和探索器。
learner = Qlearning(
learning_rate=params.learning_rate,
gamma=params.gamma,
state_size=params.state_size,
action_size=params.action_size,
)
explorer = EpsilonGreedy(
epsilon=params.epsilon,
)
這將是我們的主要函數，用於執行我們的環境，直到達到最大 episode 數量 params.total_episodes。為了考慮隨機性，我們還將多次執行我們的環境。
def run_env():
rewards = np.zeros((params.total_episodes, params.n_runs))
steps = np.zeros((params.total_episodes, params.n_runs))
episodes = np.arange(params.total_episodes)
qtables = np.zeros((params.n_runs, params.state_size, params.action_size))
all_states = []
all_actions = []
for run in range(params.n_runs): # Run several times to account for stochasticity
learner.reset_qtable() # Reset the Q-table between runs
for episode in tqdm(
episodes, desc=f"Run {run}/{params.n_runs} - Episodes", leave=False
):
state = env.reset(seed=params.seed)[0] # Reset the environment
step = 0
done = False
total_rewards = 0
while not done:
action = explorer.choose_action(
action_space=env.action_space, state=state, qtable=learner.qtable
)
# Log all states and actions
all_states.append(state)
all_actions.append(action)
# Take the action (a) and observe the outcome state(s') and reward (r)
new_state, reward, terminated, truncated, info = env.step(action)
done = terminated or truncated
learner.qtable[state, action] = learner.update(
state, action, reward, new_state
)
total_rewards += reward
step += 1
# Our new state is state
state = new_state
# Log all rewards and steps
rewards[episode, run] = total_rewards
steps[episode, run] = step
qtables[run, :, :] = learner.qtable
return rewards, steps, episodes, qtables, all_states, all_actions
視覺化¶
為了方便使用 Seaborn 繪製結果，我們將把模擬的主要結果保存在 Pandas dataframes 中。
def postprocess(episodes, params, rewards, steps, map_size):
"""Convert the results of the simulation in dataframes."""
res = pd.DataFrame(
data={
"Episodes": np.tile(episodes, reps=params.n_runs),
"Rewards": rewards.flatten(order="F"),
"Steps": steps.flatten(order="F"),
}
)
res["cum_rewards"] = rewards.cumsum(axis=0).flatten(order="F")
res["map_size"] = np.repeat(f"{map_size}x{map_size}", res.shape[0])
st = pd.DataFrame(data={"Episodes": episodes, "Steps": steps.mean(axis=1)})
st["map_size"] = np.repeat(f"{map_size}x{map_size}", st.shape[0])
return res, st
我們想要繪製代理程式最終學習到的策略。為此，我們將：1. 從每個狀態的 Q 表格中提取最佳 Q 值，2. 取得這些 Q 值對應的最佳動作，3. 將每個動作映射到一個箭頭，以便我們可以將其視覺化。
def qtable_directions_map(qtable, map_size):
"""Get the best learned action & map it to arrows."""
qtable_val_max = qtable.max(axis=1).reshape(map_size, map_size)
qtable_best_action = np.argmax(qtable, axis=1).reshape(map_size, map_size)
directions = {0: "←", 1: "↓", 2: "→", 3: "↑"}
qtable_directions = np.empty(qtable_best_action.flatten().shape, dtype=str)
eps = np.finfo(float).eps # Minimum float number on the machine
for idx, val in enumerate(qtable_best_action.flatten()):
if qtable_val_max.flatten()[idx] > eps:
# Assign an arrow only if a minimal Q-value has been learned as best action
# otherwise since 0 is a direction, it also gets mapped on the tiles where
# it didn't actually learn anything
qtable_directions[idx] = directions[val]
qtable_directions = qtable_directions.reshape(map_size, map_size)
return qtable_val_max, qtable_directions
使用以下函數，我們將在左側繪製模擬的最後一幀。如果代理程式學習到解決任務的良好策略，我們期望在影片最後一幀的寶藏方塊上看到它。在右側，我們將繪製代理程式學習到的策略。每個箭頭將代表每個方塊/狀態要選擇的最佳動作。
def plot_q_values_map(qtable, env, map_size):
"""Plot the last frame of the simulation and the policy learned."""
qtable_val_max, qtable_directions = qtable_directions_map(qtable, map_size)
# Plot the last frame
fig, ax = plt.subplots(nrows=1, ncols=2, figsize=(15, 5))
ax[0].imshow(env.render())
ax[0].axis("off")
ax[0].set_title("Last frame")
# Plot the policy
sns.heatmap(
qtable_val_max,
annot=qtable_directions,
fmt="",
ax=ax[1],
cmap=sns.color_palette("Blues", as_cmap=True),
linewidths=0.7,
linecolor="black",
xticklabels=[],
yticklabels=[],
annot_kws={"fontsize": "xx-large"},
).set(title="Learned Q-values\nArrows represent best action")
for _, spine in ax[1].spines.items():
spine.set_visible(True)
spine.set_linewidth(0.7)
spine.set_color("black")
img_title = f"frozenlake_q_values_{map_size}x{map_size}.png"
fig.savefig(params.savefig_folder / img_title, bbox_inches="tight")
plt.show()
作為健全性檢查，我們將使用以下函數繪製狀態和動作的分佈
def plot_states_actions_distribution(states, actions, map_size):
"""Plot the distributions of states and actions."""
labels = {"LEFT": 0, "DOWN": 1, "RIGHT": 2, "UP": 3}
fig, ax = plt.subplots(nrows=1, ncols=2, figsize=(15, 5))
sns.histplot(data=states, ax=ax[0], kde=True)
ax[0].set_title("States")
sns.histplot(data=actions, ax=ax[1])
ax[1].set_xticks(list(labels.values()), labels=labels.keys())
ax[1].set_title("Actions")
fig.tight_layout()
img_title = f"frozenlake_states_actions_distrib_{map_size}x{map_size}.png"
fig.savefig(params.savefig_folder / img_title, bbox_inches="tight")
plt.show()
現在我們將在幾個增加的地圖尺寸上執行我們的代理程式：- \(4 \times 4\), - \(7 \times 7\), - \(9 \times 9\), - \(11 \times 11\)。
整合在一起
map_sizes = [4, 7, 9, 11]
res_all = pd.DataFrame()
st_all = pd.DataFrame()
for map_size in map_sizes:
env = gym.make(
"FrozenLake-v1",
is_slippery=params.is_slippery,
render_mode="rgb_array",
desc=generate_random_map(
size=map_size, p=params.proba_frozen, seed=params.seed
),
)
params = params._replace(action_size=env.action_space.n)
params = params._replace(state_size=env.observation_space.n)
env.action_space.seed(
params.seed
) # Set the seed to get reproducible results when sampling the action space
learner = Qlearning(
learning_rate=params.learning_rate,
gamma=params.gamma,
state_size=params.state_size,
action_size=params.action_size,
)
explorer = EpsilonGreedy(
epsilon=params.epsilon,
)
print(f"Map size: {map_size}x{map_size}")
rewards, steps, episodes, qtables, all_states, all_actions = run_env()
# Save the results in dataframes
res, st = postprocess(episodes, params, rewards, steps, map_size)
res_all = pd.concat([res_all, res])
st_all = pd.concat([st_all, st])
qtable = qtables.mean(axis=0) # Average the Q-table between runs
plot_states_actions_distribution(
states=all_states, actions=all_actions, map_size=map_size
) # Sanity check
plot_q_values_map(qtable, env, map_size)
env.close()
地圖尺寸：\(4 \times 4\)¶
地圖尺寸：\(7 \times 7\)¶
地圖尺寸：\(9 \times 9\)¶
地圖尺寸：\(11 \times 11\)¶
DOWN 和 RIGHT 動作被選擇的次數更多，這是合理的，因為代理程式從地圖的左上角開始，需要找到通往右下角的路。而且地圖越大，離起始狀態越遠的狀態/方塊被訪問的次數就越少。
為了檢查我們的代理程式是否正在學習，我們想要繪製獎勵的累積總和，以及達到 episode 結束所需的步驟數。如果我們的代理程式正在學習，我們期望看到獎勵的累積總和增加，並且解決任務的步驟數減少。
def plot_steps_and_rewards(rewards_df, steps_df):
"""Plot the steps and rewards from dataframes."""
fig, ax = plt.subplots(nrows=1, ncols=2, figsize=(15, 5))
sns.lineplot(
data=rewards_df, x="Episodes", y="cum_rewards", hue="map_size", ax=ax[0]
)
ax[0].set(ylabel="Cumulated rewards")
sns.lineplot(data=steps_df, x="Episodes", y="Steps", hue="map_size", ax=ax[1])
ax[1].set(ylabel="Averaged steps number")
for axi in ax:
axi.legend(title="map size")
fig.tight_layout()
img_title = "frozenlake_steps_and_rewards.png"
fig.savefig(params.savefig_folder / img_title, bbox_inches="tight")
plt.show()
plot_steps_and_rewards(res_all, st_all)
在 \(4 \times 4\) 地圖上，學習收斂得非常快，而在 \(7 \times 7\) 地圖上，代理程式需要 \(\sim 300\) 個 episodes，在 \(9 \times 9\) 地圖上，它需要 \(\sim 800\) 個 episodes，而在 \(11 \times 11\) 地圖上，它需要 \(\sim 1800\) 個 episodes 才能收斂。有趣的是，代理程式在 \(9 \times 9\) 地圖上似乎獲得了比 \(7 \times 7\) 地圖更多的獎勵，這可能意味著它在 \(7 \times 7\) 地圖上沒有達到最佳策略。
最後，如果代理程式沒有獲得任何獎勵，獎勵就不會在 Q 值中傳播，並且代理程式不會學習任何東西。根據我在這個環境中使用 \(\epsilon\)-greedy 以及這些超參數和環境設定的經驗，超過 \(11 \times 11\) 個方塊的地圖開始變得難以解決。也許使用不同的探索演算法可以克服這個問題。另一個有很大影響的參數是 proba_frozen，即方塊被凍結的機率。如果洞太多，即 \(p<0.9\)，Q-learning 很難不掉入洞中並獲得獎勵訊號。
參考文獻¶
程式碼靈感來自 Thomas Simonini 的 Deep Reinforcement Learning Course (http://simoninithomas.com/)
David Silver 的課程，特別是第 4 課和第 5 課
Q-Learning: Off-Policy TD Control，出自 Reinforcement Learning: An Introduction, by Richard S. Sutton and Andrew G. Barto
Tim Miller（墨爾本大學）的 Introduction to Reinforcement Learning