Deep Q-Network (DQN) とは？

!pip install gymnasium torch matplotlib
!pip install stable-baselines3[extra]

import gymnasium as gym
import torch
import torch.nn as nn
import torch.optim as optim
import random
from collections import deque
import numpy as np

# GPU（CUDA）が利用可能であればそれを使用し、不可であればCPUを使用する設定を行います
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")

# QNetwork: 状態を入力とし、各行動のQ値を出力するニューラルネットワークを定義します
# 記事内の Q(s, a; θ) に対応するメインネットワーク（Policy Network）と、
# Q(s, a; θ^-) に対応するターゲットネットワーク（Target Network）の両方でこのクラスがインスタンス化されます
class QNetwork(nn.Module):
    def __init__(self, state_dim, action_dim):
        super(QNetwork, self).__init__()
        # 入力層から隠れ層1への全結合（状態の次元数から128次元へ変換）
        self.fc1 = nn.Linear(state_dim, 128)
        # 隠れ層1から隠れ層2への全結合
        self.fc2 = nn.Linear(128, 128)
        # 隠れ層2から出力層への全結合（行動の数だけの次元を出力し、それぞれのQ値を算出）
        self.fc3 = nn.Linear(128, action_dim)

    def forward(self, x):
        # 活性化関数ReLUを適用して非線形性を導入します
        x = torch.relu(self.fc1(x))
        x = torch.relu(self.fc2(x))
        # 出力層では活性化関数を通さず、生のQ値をそのまま返します
        return self.fc3(x)

# ReplayBuffer: エージェントの過去の経験を保存し、学習時にランダムにサンプリングしてデータの時間的相関を解消するクラス
class ReplayBuffer:
    def __init__(self, capacity):
        # Pythonの標準ライブラリdequeを使用し、最大容量（capacity）に達すると自動的に最も古いデータが破棄されるリングバッファを作成します
        self.buffer = deque(maxlen=capacity)

    def push(self, state, action, reward, next_state, done):
        # 経験（現在の状態、選択した行動、得られた報酬、遷移先の次の状態、エピソード終了フラグ）をタプルとしてバッファに追加します
        self.buffer.append((state, action, reward, next_state, done))

    def sample(self, batch_size):
        # バッファの中から重複なしでbatch_size分のデータをランダムに抽出し、リストとして返します
        return random.sample(self.buffer, batch_size)

    def __len__(self):
        # 現在バッファに保存されているデータの個数を返します
        return len(self.buffer)

# DQNAgent: DQN（Deep Q-Network）の行動選択、経験の蓄積、および重みの更新処理を取りまとめるエージェントクラス
class DQNAgent:
    def __init__(self, state_dim, action_dim, replay_buffer_capacity=10000, batch_size=128, gamma=0.99, lr=1e-4):
        # 状態空間の次元数（CartPoleでは4次元）
        self.state_dim = state_dim
        # 行動空間の数（CartPoleでは2通り: 左・右）
        self.action_dim = action_dim
        # 1回の学習で使用するミニバッチのサイズ
        self.batch_size = batch_size
        # 将来の報酬を割り引く割合（割引率 γ）
        self.gamma = gamma

        # メインネットワーク（Policy Network θ）: 行動選択と頻繁な重み更新を行います
        self.policy_net = QNetwork(state_dim, action_dim).to(device)
        # ターゲットネットワーク（Target Network θ^-）: TDターゲットの目標値算出のみに使用し、重みは一定期間固定します
        self.target_net = QNetwork(state_dim, action_dim).to(device)
        # 初期状態ではメインネットワークの重みをターゲットネットワークに完全に同期させます
        self.target_net.load_state_dict(self.policy_net.state_dict())
        # ターゲットネットワークは勾配計算やDropout、BatchNormの挙動を防ぐために評価モード（eval）にします
        self.target_net.eval()

        # メインネットワークを最適化するためのAdamオプティマイザを設定します
        self.optimizer = optim.Adam(self.policy_net.parameters(), lr=lr)
        # 経験データを一時保存するためのリプレイバッファを生成します
        self.buffer = ReplayBuffer(replay_buffer_capacity)
        # 誤差逆伝播の基準となる損失関数として、平均二乗誤差（MSE Loss）を定義します
        self.loss_fn = nn.MSELoss()

    def select_action(self, state, epsilon):
        # ε-greedy法に基づき、探索と活用のバランスをとって行動を決定します
        if random.random() < epsilon:
            # 確率εでランダムに行動を選択します（探索: Exploration）
            return random.randrange(self.action_dim)
        with torch.no_grad():
            # 確率 (1 - ε) で現在のQ値が最も高い行動を貪欲（Greedy）に選択します（活用: Exploitation）
            # 入力状態をテンソルに変換し、バッチサイズ1の次元（unsqueeze(0)）を追加してデバイスに転送します
            state = torch.FloatTensor(state).unsqueeze(0).to(device)
            # メインネットワークに状態を通し、各行動の予想Q値を出力します
            q_values = self.policy_net(state)
            # 最も高いQ値を持つ行動のインデックスを取得します
            return q_values.max(1)[1].item()

    def learn(self):
        # バッファに十分なデータ（batch_size以上）が貯まるまでは学習を行いません
        if len(self.buffer) < self.batch_size:
            return

        # リプレイバッファからランダムにbatch_size分の遷移データをサンプリングします
        transitions = self.buffer.sample(self.batch_size)
        # サンプリングされたデータの構造を[(s, a, r, s', done), ...]から、各要素ごとのリストにアンパックして再構成します
        batch = list(zip(*transitions))

        # PyTorchの計算効率を最大化するため、各データをテンソルに変換しGPU/CPUデバイスに送ります
        state_batch = torch.FloatTensor(np.array(batch[0])).to(device)
        action_batch = torch.LongTensor(batch[1]).unsqueeze(1).to(device)
        reward_batch = torch.FloatTensor(batch[2]).to(device)
        next_state_batch = torch.FloatTensor(np.array(batch[3])).to(device)
        done_batch = torch.FloatTensor(batch[4]).to(device)
        
        # 1. 現在のQ値 Q(s, a; θ) の算出
        # メインネットワークに状態バッチを入力し、gatherを用いて実際に選択された行動(action_batch)に対応するQ値のみを抽出します
        current_q_values = self.policy_net(state_batch).gather(1, action_batch)

        # 2. ターゲットとなるQ値（TDターゲット）の算出
        # ターゲットネットワークを用いた目標値計算では、メインネットワークの頻繁な更新による影響を防ぐため勾配計算を無効化します
        with torch.no_grad():
            # 次の状態における各行動の最大Q値 max_a' Q(s', a'; θ^-) をターゲットネットワークで計算します
            next_q_values = self.target_net(next_state_batch).max(1)[0]
            # ベルマン更新目標を算出: r + γ * (1 - done) * max_a' Q(s', a'; θ^-)
            # エピソードが終了（doneがTrue＝1）している場合は、遷移先が存在しないため将来価値を0とします
            target_q_values = reward_batch + (1 - done_batch) * self.gamma * next_q_values

        # 3. 損失 L(θ) の計算
        # 予測Q値(current_q_values)と、目標Q値(target_q_values)の間の平均二乗誤差を算出します
        loss = self.loss_fn(current_q_values.squeeze(), target_q_values)
        
        # 4. バックプロパゲーションとパラメータ更新
        # 勾配バッファをゼロクリアし、誤差逆伝播で勾配を計算後、メインネットワークのパラメータ θ を更新します
        self.optimizer.zero_grad()
        loss.backward()
        self.optimizer.step()

    def update_target_net(self):
        # 定期的なタイミングで、メインネットワークの重み（θ）をターゲットネットワークの重み（θ^-）に完全にコピー（同期）します
        self.target_net.load_state_dict(self.policy_net.state_dict())

# --- ハイパーパラメータの設定 ---
EPISODES = 500                    # 学習を実行する最大エピソード数
TARGET_UPDATE_FREQUENCY = 100     # メインネットワークからターゲットネットワークへ重みを同期する間隔（総ステップ数単位）
EPSILON_START = 1.0               # 学習初期における探索率 ε の初期値（100%ランダムに探索）
EPSILON_END = 0.01                # 探索率 ε の最小値（最低でも1%は探索を残す）
EPSILON_DECAY = 0.995             # エピソードごとに ε を減衰させる割合（徐々に活用を優先にするため）

# --- 環境とエージェントの初期化 ---
# GymnasiumよりCartPole-v1環境を作成します
env = gym.make('CartPole-v1')
# 環境の観測空間（状態）の次元数を取得（CartPoleは4次元）
state_dim = env.observation_space.shape[0]
# 環境の行動空間（エージェントが取れる行動の種類）の数を取得（CartPoleは2種類）
action_dim = env.action_space.n
# 事前に定義したDQNAgentインスタンスを作成
agent = DQNAgent(state_dim, action_dim)

# --- 訓練（学習）メインループ ---
scores = []                       # 各エピソードの獲得スコアを記録するリスト
total_steps = 0                   # 累積された全体の総ステップ数
epsilon = EPSILON_START           # 現在の探索率を初期値にセット

print("--- DQN (スクラッチ) 学習開始 ---")
for episode in range(EPISODES):
    # エピソードの開始時に環境をリセットし、初期状態を取得します
    state, _ = env.reset()
    episode_score = 0             # このエピソードにおける累積報酬（スコア）
    done = False                  # エピソードが終了したかどうかのフラグ
    
    while not done:
        # 現在の状態と現在の探索率に基づいて行動を選択（ε-greedy法）
        action = agent.select_action(state, epsilon)
        # 選択した行動を実行し、次の状態、報酬、終了フラグ（terminated/truncated）などを取得
        next_state, reward, terminated, truncated, _ = env.step(action)
        # 通常終了、または時間切れ（最大ステップ到達など）のいずれかでエピソード終了と判定
        done = terminated or truncated
        
        # 経験（遷移データ）をエージェントのリプレイバッファに保存します
        agent.buffer.push(state, action, reward, next_state, done)
        # リプレイバッファ内のデータを用いてネットワークを1ステップ更新します
        agent.learn()
        
        # 状態を次のステップへ移行
        state = next_state
        # 報酬をスコアに加算（CartPoleでは1ステップ生存するごとに報酬 +1）
        episode_score += reward
        # 総ステップ数をカウントアップ
        total_steps += 1
        
        # 設定された更新ステップ間隔ごとに、ターゲットネットワークの重みを同期します
        if total_steps % TARGET_UPDATE_FREQUENCY == 0:
            agent.update_target_net()

    # エピソード終了後、探索率 ε を少しずつ減衰させます（最小値以下には下げない）
    epsilon = max(EPSILON_END, epsilon * EPSILON_DECAY)
    
    # このエピソードのスコアを記録
    scores.append(episode_score)
    # 50エピソードごとに直近の平均スコアを出力し、学習の進捗を確認します
    if (episode + 1) % 50 == 0:
        print(f"エピソード {episode + 1}/{EPISODES}: 平均スコア (直近50) {np.mean(scores[-50:])}")

# スクラッチ実装による学習履歴を保存
scratch_dqn_scores = scores
print("--- DQN (スクラッチ) 学習完了 ---")

--- DQN (スクラッチ) 学習開始 ---
エピソード 50/500: 平均スコア (直近50) 20.48
エピソード 100/500: 平均スコア (直近50) 47.38
エピソード 150/500: 平均スコア (直近50) 117.38
エピソード 200/500: 平均スコア (直近50) 188.26
エピソード 250/500: 平均スコア (直近50) 257.06
エピソード 300/500: 平均スコア (直近50) 265.8
エピソード 350/500: 平均スコア (直近50) 338.42
エピソード 400/500: 平均スコア (直近50) 454.02
エピソード 450/500: 平均スコア (直近50) 401.18
エピソード 500/500: 平均スコア (直近50) 434.04
--- DQN (スクラッチ) 学習完了 ---

# ==========================================
# ランダム行動エージェントのスコアを収集
# ==========================================
random_scores = []
env_random = gym.make('CartPole-v1')
print("--- ランダムエージェント 評価開始 ---")
for episode in range(EPISODES):
    env_random.reset()
    done = False
    episode_score = 0
    while not done:
        action = env_random.action_space.sample()
        _, reward, terminated, truncated, _ = env_random.step(action)
        done = terminated or truncated
        episode_score += reward
    random_scores.append(episode_score)
print("--- ランダムエージェント 評価完了 ---")

# ==========================================
# グラフ描画（学習曲線の比較）
# ==========================================
def moving_average(data, window_size):
    if len(data) < window_size:
        return np.mean(data)
    return np.convolve(data, np.ones(window_size), 'valid') / window_size

plt.figure(figsize=(12, 7))
window = 50

# スクラッチ実装（学習中）
plt.plot(moving_average(scratch_dqn_scores, window), label=f'DQN (Scratch) - MA({window})', color='blue')

# ランダム行動
plt.plot(moving_average(random_scores, window), label=f'Random - MA({window})', color='orange', linestyle='--')

plt.title('Training Learning Curve: Scratch vs Random')
plt.xlabel(f'Episodes (Moving Average Window = {window})')
plt.ylabel('Score')
plt.legend()
plt.grid(True)
plt.show()

# Gymnasium（環境）とStable Baselines3（DQN）のインポート
import gymnasium as gym
from stable_baselines3 import DQN
from stable_baselines3.common.evaluation import evaluate_policy
from stable_baselines3.common.monitor import Monitor
import matplotlib.pyplot as plt

# CartPole環境を初期化
env_sb3 = gym.make('CartPole-v1')
# Monitorラッパーを使ってエピソードごとの報酬などを自動記録
env_sb3 = Monitor(env_sb3)

# DQNモデルの定義
# MlpPolicy: 多層パーセプトロン（全結合NN）を使用
model_sb3_dqn = DQN(
    'MlpPolicy',
    env_sb3,
    learning_rate=1e-3,          # 学習率
    buffer_size=20000,           # Experience Replayのバッファサイズ
    learning_starts=500,         # 学習を開始するまでのランダム行動ステップ数
    batch_size=64,               # 1回の学習に使うミニバッチサイズ
    gamma=0.99,                  # 割引率
    train_freq=1,                # ネットワークの更新頻度（1ステップごと）
    gradient_steps=1,            # 1回の更新で行う勾配ステップ数
    target_update_interval=500,  # Target Networkを更新する間隔
    exploration_fraction=0.5,    # 探索率(ε)を減衰させる期間の割合
    exploration_final_eps=0.01,  # 探索率(ε)の最小値
    verbose=0                    # ログの出力レベル（0は出力なし）
)

print("--- DQN (Stable Baselines3) 学習開始 ---")
# 指定した総ステップ数だけ学習を実行
model_sb3_dqn.learn(total_timesteps=150_000)
print("--- DQN (Stable Baselines3) 学習完了 ---")

# 学習中のエピソードごとのスコア推移をプロット
plt.plot(env_sb3.get_episode_rewards())
plt.title('DQN (Stable Baselines3) Learning Curve')
plt.xlabel('Episodes')
plt.ylabel('Score')
plt.show()

--- DQN (Stable Baselines3) 学習開始 ---
--- DQN (Stable Baselines3) 学習完了 ---