1. はじめに
本記事は、エンジニア向けにChatGPTの活用例を紹介する 「ChatGPTと始めるシリーズ」第5弾です。今回は強化学習に挑戦したいと思います！
今回の題材はこちらです。
OpenAI Gym のシミュレーション環境 LunarLander-v2 を使用して、宇宙船を着陸させるエージェントを強化学習を用いて作成します。
（過去のシリーズ記事はこちらです）
2. 作業内容と問題
2.1. 作業内容について提案してもらう
少し複雑な作業になるので、まずは作業内容を ChatGPT と相談したいと思います。
ChatGPT からの回答がこちらです。
本記事では、上記の作業内容に沿って、強化学習の基本である Q-Learning、 Deep Q-Learning の順に試していきます。
以降は上記作業内容の順に、手順を聞いていく流れになります。プロンプトの多くは、過去のシリーズ記事と同じなので、今回は結果を中心に紹介します。
2.2. 問題設定
今回使用する環境は冒頭で紹介したように、下記のリンクの LunarLander-v2 になります。
問題の概要はこちらです。
宇宙船を墜落させることなく着陸できるように、エンジン出力をコントロールします。
3. Q-Learning
3.1. Q-Learning の概要
Q-Learningについて、ChatGTP に簡単に概要を紹介してもらいます。
実際に作業していたときは、「Exploration vs Exploitation についてもっと詳しく」など理解が足りないところについて色々聞いていたのですが、すごく長いのでカットします。
ChatGPT は何を聞いても根気よく教えてくれるので、不明点はぜひお手元のGPTに相談してみてください！
3.2. 使用環境
今回使用した環境はこちらです。
- Windows 11
- python 3.8.10
- 使用したパッケージ (gymは最新バージョンがうまくインストールできなかったのでバージョンを落としています)
requirements.txt
wheel
gym==0.21.0
numpy
matplotlib
Box2D
pyglet==1.5.27
3.3. Q-Learningのコード
ChatGPTにより生成してもらった Q-Learning のコードを以下に示します。
長いので折りたたんでいます
lunar_lander_qlearning.py
"""
強化学習（Q学習）によるLunarLanderの制御コードです。
"""
import gym
import numpy as np
import matplotlib.pyplot as plt
from gym import wrappers
import datetime
def create_env():
"""
環境の作成と初期化を行います。
Returns
-------
env : gym.Env
'LunarLander-v2'環境
initial_state : array
初期状態
"""
# 環境の初期化
env = gym.make('LunarLander-v2')
initial_state = env.reset()
return env, initial_state
def initialize_q_table(env, binning_specs):
"""
Qテーブルを初期化します。
Parameters
----------
env : gym.Env
学習対象の環境
binning_specs : list of array-like
各状態変数を離散化する際のビンの境界値リスト
Returns
-------
q_table : ndarray
初期化されたQテーブル
"""
# Q-Tableの初期化
# binning_specs の各リストの長さを取得してタプルを作る
state_space_size = tuple(len(specs) + 1 for specs in binning_specs)
action_space_size = env.action_space.n
# Q-Tableをゼロで初期化
q_table = np.zeros(state_space_size + (action_space_size,))
return q_table
def choose_action(state, q_table, action_space, epsilon):
"""
ε-greedy法による行動選択を行います。
Parameters
----------
state : tuple
離散化された現在の状態
q_table : ndarray
Qテーブル
action_space : gym.Space
行動空間
epsilon : float
ランダムな行動を選択する確率
Returns
-------
action : int
選択された行動
"""
# ε-greedy法による行動選択
if np.random.uniform(0, 1) < epsilon:
# 探索：ランダムな行動を選択
action = action_space.sample()
else:
# 活用：Q値が最大の行動を選択
action = np.argmax(q_table[state])
return action
def update_q_table(q_table, state, action, reward, next_state, alpha, gamma):
"""
Q値の更新を行います。
Parameters
----------
q_table : ndarray
Qテーブル
state : tuple
離散化された現在の状態
action : int
行動
reward : float
得られた報酬
next_state : tuple
離散化された次の状態
alpha : float
学習率
gamma : float
割引率
Returns
-------
q_table : ndarray
更新されたQテーブル
"""
# Q値の更新
# 古いQ値
old_value = q_table[state][action]
# 次の状態における最高のQ値
next_max = np.max(q_table[next_state])
# 新しいQ値の計算（Bellman方程式）
new_value = (1 - alpha) * old_value + alpha * (reward + gamma * next_max)
# Q-Tableの更新
q_table[state][action] = new_value
return q_table
def discretize_state(state, binning_specs):
"""
離散化を行う関数です。
Parameters
----------
state: list
離散化する前の状態変数のリスト
binning_specs: list of tuples
各状態変数に対するビンの境界値を定めるタプルのリスト
Returns
-------
discretized_state: tuple
離散化された状態変数
"""
discretized_state = []
for value, bins in zip(state, binning_specs):
discretized_state.append(np.digitize(value, bins) - 1)
return tuple(discretized_state)
def episode_step(env, q_table, max_steps, epsilon, alpha, gamma, binning_specs, is_recording):
"""
一エピソードを実行し、報酬と終了フラグを返します。
Parameters
----------
env : gym.Env
学習対象の環境
q_table : ndarray
Qテーブル
max_steps : int
1エピソードの最大ステップ数
epsilon : float
ランダムな行動を選択する確率
alpha : float
学習率
gamma : float
割引率
binning_specs : list of array-like
各状態変数を離散化する際のビンの境界値リスト
is_recording : bool
ビデオを録画するかどうか
Returns
-------
episode_reward : float
1エピソードの総報酬
done : bool
エピソードが終了したかどうか
"""
# 状態の初期化
state = env.reset()
state = discretize_state(state, binning_specs)
# 1エピソードの報酬
episode_reward = 0
done = False
# for step in range(max_steps):
while not done:
# 行動の選択
action = choose_action(state, q_table, env.action_space, epsilon)
# 行動の実行とフィードバックの取得
next_state, reward, done, _ = env.step(action)
next_state = discretize_state(next_state, binning_specs)
# Q-Tableの更新
q_table = update_q_table(q_table, state, action, reward, next_state, alpha, gamma)
# 状態の更新
state = next_state
# 報酬の加算
episode_reward += reward
# エピソード終了時
if done:
break
if is_recording:
env.close()
return episode_reward, done
def plot_rewards(rewards):
"""
報酬の移動平均を計算し、グラフを表示します。
Parameters
----------
rewards : list of float
各エピソードの報酬のリスト
"""
# 報酬の移動平均を計算
moving_avg = np.convolve(rewards, np.ones((100,))/100, mode='valid')
# グラフをプロット
plt.plot(np.arange(len(moving_avg)), moving_avg)
plt.title('Training Rewards')
plt.xlabel('Episode')
plt.ylabel('100 Episode Average')
plt.savefig("result.png")
plt.show()
def main():
"""
メインの関数。学習パラメータの設定、学習環境の準備、エピソードの実行、結果の表示などを行います。
Parameters
----------
num_episodes : int
学習を行うエピソード数
max_steps : int
1エピソード当たりの最大ステップ数
alpha : float
学習率
gamma : float
割引率
epsilon : float
ε-greedy法における探索の確率
"""
# パラメータの設定
num_episodes = 50000
max_steps = 300
alpha = 0.5
gamma = 0.95
epsilon = 0.1
# ビデオを保存するエピソードの間隔
video_interval = 1000
binning_specs = [
np.linspace(-1, 1, 10), # x座標
np.linspace(0, 1, 10), # y座標
np.linspace(-1, 1, 10), # x速度
np.linspace(-1, 1, 10), # y速度
np.linspace(-1, 1, 10), # ランダーの角度
np.linspace(-1, 1, 10), # ランダーの角速度
np.linspace(0, 1, 10), # 左脚が地面に接触しているかどうか
np.linspace(0, 1, 10) # 右脚が地面に接触しているかどうか
]
# 環境の作成
env = gym.make('LunarLander-v2')
# 環境の初期化とQテーブルの初期化
q_table = initialize_q_table(env, binning_specs)
rewards = []
# Q-Tableの初期化
q_table = initialize_q_table(env, binning_specs)
now = datetime.datetime.now()
# エピソードごとの報酬を格納するリスト
rewards = []
# 学習のループ
for episode in range(num_episodes):
if episode % video_interval == 0:
# Gymのラッパーを使用して環境をラップ
env_wrapped = wrappers.Monitor(env, f"./videos_{now.strftime('%Y%m%d_%H%M%S')}/{episode}", force=True)
episode_reward, done= episode_step(env_wrapped, q_table, max_steps, epsilon, alpha, gamma, binning_specs, True)
print(f"episode {episode}: {episode_reward}")
else:
# ビデオを保存しないエピソード
episode_reward, done= episode_step(env, q_table, max_steps, epsilon, alpha, gamma, binning_specs, False)
rewards.append(episode_reward)
# if episode % video_interval == 0:
# print(f"episode {episode}: {episode_reward}")
# 学習の結果を表示
print("Training finished.\n")
# 報酬の変動をグラフで表示
plot_rewards(rewards)
if __name__ == "__main__":
main()
補足
ChatGPTがそのまま出してくれたコードでは少し不具合があったので、以下の修正を行っています。
- 対象の状態空間を連続のまま取り扱うとメモリが足りなかったため、離散化を実施
- 一定間隔ごとにmp4で動画を保存するように設定
- 完成したコードに対して、コメントの追加を依頼
3.4. 実行結果
上記のコードを実行して、100エピソードごとの報酬の平均値を出したものはこちらです。50000回学習を行うことで、スコアはだんだんよくなりますが、0 程度で飽和し、安定的に着陸したことを示すスコア200には届いていません。
シミュレーション動画: 1エピソード
シミュレーション動画: 50000エピソード付近
50000回学習しても着陸を学ぶという段階には進んでおらず、機体が反転することがないように一応はコントロールしながらも、ただフラフラと落ちているだけのように見えます。
ChatGPT にもさんざん言われたのですが、Q-Learning で解くには問題自体がちょっと複雑なため、Q-Learning でこれ以上のパラメータ調整を頑張らずに、ニューラルネットワークを使った Deep Q-Learning に進みたいと思います。
4. Deep Q-Learning
4.1. Deep Q-Learning の概要
Deep Q-Learning の概要説明はこちらです。
4.2. 事前準備
ここからは、Pytorch のインストールをお願いします（バージョン等については Pytorchの公式サイトを参考に使用環境に応じて変更してください）。
pip install torch==1.13.1+cpu torchvision==0.14.1+cpu torchaudio==0.13.1 --extra-index-url https://download.pytorch.org/whl/cpu
4.3. Deep Q-Learning のコード
Deep Q-Learning について、同様に ChatGPT にコードを出力してもらいました。
長いので折りたたみます
lunar_lander_dqn.py
import torch
import torch.nn as nn
import torch.optim as optim
from collections import deque
import random
import torch.nn.functional as F
import numpy as np
import gym
import matplotlib.pyplot as plt
import datetime
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
class DQNetwork(nn.Module):
"""DQN (Deep Q-Network) モデルクラス.
Parameters
----------
state_size : int
状態の次元数
action_size : int
行動の数
hidden_size : int, default=64
隠れ層のノード数
hidden_size2 : int, default=16
2つ目の隠れ層のノード数
"""
def __init__(self, state_size, action_size, hidden_size=64, hidden_size2=16):
super(DQNetwork, self).__init__()
# 全結合層
self.fc1 = nn.Linear(state_size, hidden_size)
self.fc2 = nn.Linear(hidden_size, hidden_size2)
self.fc3 = nn.Linear(hidden_size2, action_size)
# Heの初期化を使って重みを初期化
nn.init.kaiming_normal_(self.fc1.weight, nonlinearity='relu')
nn.init.kaiming_normal_(self.fc2.weight, nonlinearity='relu')
nn.init.kaiming_normal_(self.fc3.weight, nonlinearity='relu')
def forward(self, state):
"""順伝播を定義します.
Parameters
----------
state : torch.Tensor
状態
Returns
-------
action_values : torch.Tensor
各行動の価値
"""
x = torch.relu(self.fc1(state))
x = torch.relu(self.fc2(x))
action_values = self.fc3(x)
return action_values
class ReplayBuffer:
"""経験再生 (Experience Replay) メモリクラス.
Parameters
----------
buffer_size : int
バッファの大きさ
"""
def __init__(self, buffer_size):
self.buffer_size = buffer_size
self.buffer = deque(maxlen=buffer_size)
def add(self, experience):
"""経験をメモリに追加します.
Parameters
----------
experience : tuple
経験 (状態, 行動, 報酬, 次の状態, 終了フラグ)
"""
self.buffer.append(experience)
def sample(self, batch_size):
"""メモリからランダムに経験をサンプリングします.
Parameters
----------
batch_size : int
サンプリングする経験の数
Returns
-------
list
バッチサイズ分の経験
"""
return random.sample(self.buffer, batch_size)
def __len__(self):
"""メモリの現在の長さを返します.
Returns
-------
int
メモリの長さ
"""
return len(self.buffer)
class DQNAgent:
"""DQNエージェントクラス.
Parameters
----------
state_size : int
状態の次元数
action_size : int
行動の数
hidden_size : int, default=64
隠れ層のノード数
buffer_size : int, default=10000
経験再生メモリの大きさ
batch_size : int, default=64
バッチサイズ
gamma : float, default=0.95
割引率
lr : float, default=0.01
学習率
update_every : int, default=5
何ステップごとにネットワークを更新するか
target_update_every : int, default=10
何ステップごとにターゲットネットワークを更新するか
"""
def __init__(self, state_size, action_size, hidden_size=64, buffer_size=10000,
batch_size=64, gamma=0.95, lr=0.01, update_every=5, target_update_every=10):
self.state_size = state_size
self.action_size = action_size
self.hidden_size = hidden_size
self.buffer_size = buffer_size
self.batch_size = batch_size
self.gamma = gamma
self.lr = lr
self.update_every = update_every
self.target_update_every = target_update_every
# Qネットワークとターゲットネットワークの初期化
self.qnetwork = DQNetwork(state_size, action_size, hidden_size).to(device)
self.target_network = DQNetwork(state_size, action_size, hidden_size).to(device)
self.optimizer = optim.Adam(self.qnetwork.parameters(), lr=lr)
# メモリの初期化
self.memory = ReplayBuffer(buffer_size)
self.t_step = 0
self.tq_step = 0
def step(self, state, action, reward, next_state, done):
"""エージェントのステップを進めます.
Parameters
----------
state : array_like
現在の状態
action : int
行動
reward : float
報酬
next_state : array_like
次の状態
done : bool
エピソードが終了したかどうか
"""
# 経験を再生メモリに保存
self.memory.add((state, action, reward, next_state, done))
# `update_every` ステップごとに学習
self.t_step = (self.t_step + 1) % self.update_every
if self.t_step == 0:
# メモリに十分なサンプルがある場合、ランダムにサブセットを取得して学習
if len(self.memory) > self.batch_size:
experiences = self.memory.sample(self.batch_size)
self.learn(experiences, self.gamma)
# DQNの全ての重みとバイアスをターゲットネットワークに更新する
# self.tq_step = (self.tq_step + 1) % self.target_update_every
# if self.tq_step == 0:
# self.update_target_network()
def update_target_network(self):
"""ターゲットネットワークを更新します.
"""
self.target_network.load_state_dict(self.qnetwork.state_dict())
def act(self, state, eps=0.):
"""行動を選択します.
Parameters
----------
state : array_like
現在の状態
eps : float, default=0.
イプシロン、イプシロン貪欲な行動選択に使用
Returns
-------
int
選択された行動
"""
# 行動を選択
state = torch.from_numpy(state).float().unsqueeze(0).to(device)
self.qnetwork.eval()
with torch.no_grad():
action_values = self.qnetwork(state)
self.qnetwork.train()
# イプシロン貪欲な行動選択
if random.random() > eps:
return np.argmax(action_values.cpu().data.numpy())
else:
return random.choice(np.arange(self.action_size))
def learn(self, experiences, gamma):
"""経験を元にネットワークを学習させます.
Parameters
----------
experiences : tuple of (array_like, array_like, array_like, array_like, array_like)
経験のタプル, (states, actions, rewards, next_states, dones)
gamma : float
割引率
"""
# 状態、行動、報酬、次の状態、終了情報を抽出
states, actions, rewards, next_states, dones = zip(*experiences)
states = torch.from_numpy(np.vstack(states)).float().to(device)
actions = torch.from_numpy(np.vstack(actions)).long().to(device)
rewards = torch.from_numpy(np.vstack(rewards)).float().to(device)
next_states = torch.from_numpy(np.vstack(next_states)).float().to(device)
dones = torch.from_numpy(np.vstack(dones)).float().to(device)
# qnetworkからQ値を取得
Q_expected = self.qnetwork(states).gather(1, actions)
# ターゲットネットワークから次の状態の最大予測Q値を取得
Q_targets_next = self.target_network(next_states).detach().max(1)[0].unsqueeze(1)
# 現在の状態のQターゲットを計算
Q_targets = rewards + (gamma * Q_targets_next * (1 - dones))
# 損失を計算
loss = F.mse_loss(Q_expected, Q_targets)
# 損失を最小化
self.optimizer.zero_grad()
loss.backward()
self.optimizer.step()
def dqn(env, agent, now, n_episodes=2000, max_t=1000, eps_start=1.0, eps_end=0.01, eps_decay=0.995):
"""深層Q学習.
パラメータ
----------
env : gym.Env
環境
agent : DQNAgent
エージェント
n_episodes : int, default=2000
エピソードの最大数
max_t : int, default=1000
エピソード内の最大ステップ数
eps_start : float, default=1.0
イプシロンの初期値
eps_end : float, default=0.01
イプシロンの最小値
eps_decay : float, default=0.995
イプシロンの減衰率
戻り値
-------
list
各エピソードのスコア
"""
scores = [] # 各エピソードからのスコアを格納するリスト
scores_window = deque(maxlen=100) # 最新の100のスコア
eps = eps_start # イプシロンを初期化
video_interval = 100
for i_episode in range(1, n_episodes+1):
state = env.reset()
score = 0
done = False
# 100エピソードごとに録画を開始
if i_episode % video_interval == 0:
env = gym.wrappers.Monitor(env, f"./dvideos_{now.strftime('%Y%m%d_%H%M%S')}/{i_episode}", force=True)
# env.seed(0)
env._max_episode_steps = 500
env.reset()
while not done:
action = agent.act(state, eps)
next_state, reward, done, _ = env.step(action)
agent.step(state, action, reward, next_state, done)
state = next_state
score += reward
if done:
agent.update_target_network()
break
# 100エピソードの終了時に録画を終了
if i_episode % video_interval == 0:
env.close()
env = gym.make('LunarLander-v2')
env._max_episode_steps = 500
# env.seed(0)
scores_window.append(score) # 最新のスコアを保存
scores.append(score) # 最新のスコアを保存
eps = max(eps_end, eps_decay*eps) # イプシロンを減少させる
print('\rEpisode {}\tAverage Score: {:.2f}'.format(i_episode, np.mean(scores_window)), end="")
if i_episode % 100 == 0:
print('\rEpisode {}\tAverage Score: {:.2f}'.format(i_episode, np.mean(scores_window)))
print(f'eps: {eps}')
# if np.mean(scores_window)>=200.0:
# print('\nEnvironment solved in {:d} episodes!\tAverage Score: {:.2f}'.format(i_episode-100, np.mean(scores_window)))
# torch.save(agent.qnetwork.state_dict(), 'checkpoint.pth')
# break
return scores
def plot_rewards(rewards):
"""
報酬の移動平均を計算し、グラフを表示します。
パラメータ
----------
rewards : list of float
各エピソードの報酬のリスト
"""
# 報酬の移動平均を計算
moving_avg = np.convolve(rewards, np.ones((100,))/100, mode='valid')
# グラフをプロット
plt.plot(np.arange(len(moving_avg)), moving_avg)
plt.title('Training Rewards')
plt.xlabel('Episode')
plt.ylabel('100 Episode Average')
plt.savefig("result.png")
plt.show()
def main():
buffer_size = int(1e3)
hidden_size = 32
batch_size = 64
gamma = 0.99
n_episodes=5000
update_every=5
lr = 0.00047
eps_decay = 0.998
# Gym環境を初期化
env = gym.make('LunarLander-v2')
# env.seed(0)
env._max_episode_steps = 500
now = datetime.datetime.now()
# 環境から状態サイズとアクションサイズを取得
state_size = env.observation_space.shape[0]
action_size = env.action_space.n
print('状態の形状: ', state_size)
print('アクションの数: ', action_size)
# DQNエージェントを初期化
agent = DQNAgent(state_size=state_size, action_size=action_size, buffer_size=buffer_size, batch_size=batch_size, gamma=gamma, hidden_size=hidden_size, update_every=update_every, lr=lr)
# エージェントを訓練
scores = dqn(env, agent, now, n_episodes, eps_decay=eps_decay)
# スコアをプロット
plot_rewards(scores)
if __name__ == '__main__':
main()
補足
上記コードは完成版ですが、この完成版のコードを出してもらうまでに、色々と問題が発生しました...
-
機械学習ライブラリの変更
一番初めは機械学習ライブラリとして、PytorchではなくKerasとtensorflowを使用するコードを出力してくれていたのですが、私のPCでメモリリークが発生し、学習が中断する問題が発生しました。色々探ってみたのですが原因がはっきりしないため、使用する機械学習ライブラリを Pytorch に変更するようお願いしました。
-
コードのバグ
ChatGPT が出してくれたコード上で、ターゲットネットワークが正しく更新されていませんでした。
特に2.についてパラメータがの調整不足が原因かと思い2日くらい試行錯誤していたのですが、単なるコードのバグでした (別のスレッドのChatGPTに「コードに変なところはない？」と聞いてみて発覚しました)
ChatGPTのことを信用しすぎず、ちゃんとコードはチェックした方がいいですね...
4.4. 学習結果
Q-Learningと同様に、100エピソードごとの報酬の平均値を出力したのが下記になります。
無事200点以上を達成し、着陸できるようになりました！
5000エピソードまで学習を行いましたが、3000くらいで切るのが良さそうですね。
4.5. 学習中のシミュレーション結果
学習初期の100エピソード目はこちら。$\epsilon$（強化学習における探索と利用のバランスを管理するパラメータ）が大きいので、とにかくランダムに色々な方法を試そうとします。
1000エピソードあたりで、ホバリングで点を稼ぐ（墜落ペナルティをもらわないようにする）ようになりました。パラメータの設定が悪いと、これ以上改善されません。
2000エピソード付近で、きれいに着陸することができるようになりました。
2000エピソードでは無駄な燃料消費があったり、ときどき墜落することがありましたが、3000エピソードあたりで安定して高速に着陸しています。
4.6. パラメータ調整について
ChatGPTにパラメータについて聞くと、概要やどのように調整した方がよいかはある程度指針は教えてくれますが、今回は最終的な微調整は自分で行う必要がありました。
-
ニューラルネットワークの構成
使用するニューラルネットワークについて、ChatGPTが提案してくれたのは隠れ層は1層だったのですが、少ないかなと思ったので2層に増やしました（あまり試行錯誤はできていないです）。
-
$\epsilon$ について
$\epsilon$ をあまりに早く下げると、動画も貼りましたが、着陸動作を学習せず、ホバリングし続ける（墜落によるマイナスペナルティを避ける）ような収束をしました。
eps_decay=0.995とすると $\epsilon$ の減少が早すぎてダメで、最終的に収束の速さとの兼ね合いから 0.998 としました。
色々と苦労はしましたが、初歩的な DQN でもちゃんと学習させることができました！
5. おわりに
今回、色々と苦労して強化学習は難しいなと改めて思いましたが、本を読むだけだとちゃんと理解できていないことが、手を動かすとわかってくるのでいいですね。
みんなも遊んでみてね。
関連記事