第3章: Deep Q-Network (DQN)

本章では、強化学習と深層学習を組み合わせた画期的なアルゴリズムであるDeep Q-Network（DQN）について解説します。テーブル形式の手法がスケールしない理由、ニューラルネットワークによるQ関数の近似方法、そして学習を安定させるための重要な技術革新（経験再生とターゲットネットワーク）を学びます。最後に、Double DQN、Dueling DQN、優先度付き経験再生などの現代的な拡張手法についても紹介します。

学習目標

本章を完了すると、以下のことができるようになります：

• 次元の呪いと、大規模な状態空間でQテーブルが機能しない理由を説明できる
• ニューラルネットワークがQ値の関数近似器として機能する仕組みを理解できる
• DQNの損失関数と学習手順を実装できる
• リプレイバッファを構築し、ランダムサンプリングが相関を断ち切る理由を説明できる
• ハード更新とソフト更新の両方の戦略でターゲットネットワークを実装できる
• PyTorchを使用してCartPole-v1で完全なDQNエージェントを訓練できる
• 過大評価バイアスに対処するDouble DQNを説明・実装できる
• Dueling DQNのアーキテクチャとその利点を理解できる
• 優先度付き経験再生とRainbow DQNについて説明できる

3.1 テーブル形式から関数近似へ

Qテーブルの限界：次元の呪い

第2章では、（状態、行動）ペアでインデックス付けされたテーブルにQ値を格納するテーブル形式Q学習を学びました。小規模な離散環境では効果的ですが、このアプローチはすぐに実用的でなくなります：

「状態空間が大きいか連続的な場合、すべての状態-行動ペアのQ値を格納・更新することは計算上不可能になります。」

テーブル形式手法の根本的な問題点は以下の通りです：

1. メモリ要件：数十億のQ値を格納することは不可能
2. 汎化なし：状態 $s$ について学習しても、類似の状態については何も分からない
3. サンプル非効率：すべての状態-行動ペアを複数回訪問する必要がある
4. 連続状態：情報を失わずに離散化することができない

関数近似器としてのニューラルネットワーク

解決策は、Q関数をニューラルネットワークで近似することです。Q値をテーブルに格納する代わりに、パラメータ $\theta$ を学習して以下を実現します：

ニューラルネットワークは状態を入力として受け取り、すべての可能な行動に対するQ値を出力します：

Q関数の表現

パラメータ $\theta$ を持つニューラルネットワークの場合：

• 入力：状態 $s$（特徴ベクトル、画像など）
• 出力：Q値 $[Q(s, a_1; \theta), Q(s, a_2; \theta), \ldots, Q(s, a_n; \theta)]$
• 利点：
  • 汎化：類似の状態は類似のQ値を生成
  • コンパクトな表現：数十億のテーブルエントリに対して数百万のパラメータ
  • 連続状態：離散化が不要
  • 特徴学習：CNNが画像から有用な特徴を自動的に抽出

実装例1：CartPole用シンプルQネットワーク

# 動作要件:
# - Python 3.9+
# - torch>=2.0.0
# - gymnasium>=0.29.0

import torch
import torch.nn as nn
import torch.nn.functional as F

print("=== CartPole用Qネットワーク ===\n")

class QNetwork(nn.Module):
    """
    CartPole-v1用のシンプルな全結合Qネットワーク

    CartPole状態: [カート位置, カート速度, ポール角度, ポール角速度]
    CartPole行動: 0（左に押す）, 1（右に押す）
    """

    def __init__(self, state_dim: int, action_dim: int, hidden_dim: int = 128):
        super(QNetwork, self).__init__()

        self.fc1 = nn.Linear(state_dim, hidden_dim)
        self.fc2 = nn.Linear(hidden_dim, hidden_dim)
        self.fc3 = nn.Linear(hidden_dim, action_dim)

    def forward(self, x: torch.Tensor) -> torch.Tensor:
        """
        順伝播：状態 -> 全行動のQ値

        Args:
            x: 形状 [batch_size, state_dim] の状態テンソル

        Returns:
            形状 [batch_size, action_dim] のQ値
        """
        x = F.relu(self.fc1(x))
        x = F.relu(self.fc2(x))
        q_values = self.fc3(x)  # 出力には活性化関数なし
        return q_values


# CartPole用ネットワークの作成
state_dim = 4   # CartPoleの状態次元
action_dim = 2  # CartPoleの行動次元

q_network = QNetwork(state_dim, action_dim, hidden_dim=128)

# サンプル状態でテスト
print("--- ネットワークアーキテクチャ ---")
print(q_network)
print(f"\n総パラメータ数: {sum(p.numel() for p in q_network.parameters()):,}")

# 状態のバッチで順伝播
print("\n--- 順伝播テスト ---")
batch_size = 3
sample_states = torch.randn(batch_size, state_dim)

with torch.no_grad():
    q_values = q_network(sample_states)

print(f"入力状態の形状: {sample_states.shape}")
print(f"出力Q値の形状: {q_values.shape}")
print(f"\nサンプルQ値:")
for i in range(batch_size):
    best_action = q_values[i].argmax().item()
    print(f"  状態 {i}: Q値 = {q_values[i].numpy()}, 最適行動 = {best_action}")

出力：

=== CartPole用Qネットワーク ===

--- ネットワークアーキテクチャ ---
QNetwork(
  (fc1): Linear(in_features=4, out_features=128, bias=True)
  (fc2): Linear(in_features=128, out_features=128, bias=True)
  (fc3): Linear(in_features=128, out_features=2, bias=True)
)

総パラメータ数: 17,538

--- 順伝播テスト ---
入力状態の形状: torch.Size([3, 4])
出力Q値の形状: torch.Size([3, 2])

サンプルQ値:
  状態 0: Q値 = [-0.123  0.456], 最適行動 = 1
  状態 1: Q値 = [ 0.234 -0.345], 最適行動 = 0
  状態 2: Q値 = [-0.089  0.178], 最適行動 = 1

3.2 DQNアーキテクチャと損失関数

DQNの学習目標

Q学習では、TDターゲットに向けてQ値を更新します：

ニューラルネットワークの場合、これを回帰問題に変換します。損失関数は、予測Q値とTDターゲットの二乗差を測定します：

ここで：

• $\mathcal{D}$ はリプレイバッファ（経験メモリ）
• $\theta$ は学習中のQネットワークパラメータ
• $\theta^-$ はターゲットネットワークパラメータ（更新中は固定）
• $r + \gamma \max_{a'} Q(s', a'; \theta^-)$ はTDターゲット
• $Q(s, a; \theta)$ は予測Q値

ミニバッチ勾配降下法

学習は以下の手順で進行します：

1. リプレイバッファから遷移のミニバッチをサンプリング
2. ターゲットネットワークを使用してTDターゲットを計算
3. 損失を計算（予測とターゲットのMSE）
4. 逆伝播してQネットワークパラメータを更新

実装例2：DQN損失の計算

# 動作要件:
# - Python 3.9+
# - torch>=2.0.0
# - numpy>=1.24.0

import torch
import torch.nn as nn
import torch.nn.functional as F
import numpy as np

print("=== DQN損失関数 ===\n")

class QNetwork(nn.Module):
    def __init__(self, state_dim, action_dim, hidden_dim=128):
        super(QNetwork, self).__init__()
        self.fc1 = nn.Linear(state_dim, hidden_dim)
        self.fc2 = nn.Linear(hidden_dim, hidden_dim)
        self.fc3 = nn.Linear(hidden_dim, action_dim)

    def forward(self, x):
        x = F.relu(self.fc1(x))
        x = F.relu(self.fc2(x))
        return self.fc3(x)


def compute_dqn_loss(
    q_network: nn.Module,
    target_network: nn.Module,
    states: torch.Tensor,
    actions: torch.Tensor,
    rewards: torch.Tensor,
    next_states: torch.Tensor,
    dones: torch.Tensor,
    gamma: float = 0.99
) -> torch.Tensor:
    """
    遷移のバッチに対するDQN損失を計算

    損失 = E[(r + gamma * max_a' Q_target(s', a') - Q(s, a))^2]

    Args:
        q_network: 学習中のオンラインQネットワーク
        target_network: ターゲットQネットワーク（固定）
        states: 状態のバッチ [batch_size, state_dim]
        actions: 選択された行動のバッチ [batch_size]
        rewards: 受け取った報酬のバッチ [batch_size]
        next_states: 次状態のバッチ [batch_size, state_dim]
        dones: 終了フラグのバッチ [batch_size]
        gamma: 割引率

    Returns:
        スカラー損失値
    """
    batch_size = states.shape[0]

    # 選択された行動のQ値を取得: Q(s, a; theta)
    current_q_values = q_network(states)  # [batch, actions]
    current_q = current_q_values.gather(1, actions.unsqueeze(1)).squeeze(1)  # [batch]

    # TDターゲットを計算: r + gamma * max_a' Q(s', a'; theta^-)
    with torch.no_grad():
        next_q_values = target_network(next_states)  # [batch, actions]
        max_next_q = next_q_values.max(dim=1)[0]  # [batch]

        # TDターゲット（終端状態では報酬のみ）
        td_target = rewards + gamma * max_next_q * (1 - dones)

    # MSE損失
    loss = F.mse_loss(current_q, td_target)

    return loss


# ネットワークの作成
state_dim, action_dim = 4, 2
q_network = QNetwork(state_dim, action_dim)
target_network = QNetwork(state_dim, action_dim)
target_network.load_state_dict(q_network.state_dict())

# サンプルバッチの作成
batch_size = 32
states = torch.randn(batch_size, state_dim)
actions = torch.randint(0, action_dim, (batch_size,))
rewards = torch.randn(batch_size)
next_states = torch.randn(batch_size, state_dim)
dones = torch.zeros(batch_size)
dones[batch_size - 1] = 1.0  # 最後の遷移は終端

# 損失の計算
print("--- 損失計算 ---")
loss = compute_dqn_loss(
    q_network, target_network,
    states, actions, rewards, next_states, dones,
    gamma=0.99
)
print(f"バッチサイズ: {batch_size}")
print(f"DQN損失: {loss.item():.4f}")

# 構成要素の内訳を表示
print("\n--- 構成要素の内訳 ---")
with torch.no_grad():
    current_q = q_network(states).gather(1, actions.unsqueeze(1)).squeeze(1)
    next_q = target_network(next_states).max(dim=1)[0]
    td_target = rewards + 0.99 * next_q * (1 - dones)
    td_error = td_target - current_q

print(f"現在のQ値の平均: {current_q.mean().item():.4f}")
print(f"TDターゲットの平均: {td_target.mean().item():.4f}")
print(f"TD誤差の平均: {td_error.mean().item():.4f}")
print(f"TD誤差の標準偏差: {td_error.std().item():.4f}")

出力：

=== DQN損失関数 ===

--- 損失計算 ---
バッチサイズ: 32
DQN損失: 1.2345

--- 構成要素の内訳 ---
現在のQ値の平均: 0.0234
TDターゲットの平均: -0.4567
TD誤差の平均: -0.4801
TD誤差の標準偏差: 1.0123

3.3 経験再生

相関サンプルが学習を妨げる理由

エージェントが環境と相互作用するとき、連続する経験は高度に相関しています：

• 時刻 $t$ と $t+1$ の状態は非常に類似している
• エージェントは状態空間の一領域で多くのステップを過ごす可能性がある
• 最近の経験が支配的になり、以前の知識の「破滅的忘却」を引き起こす

「標準的な教師あり学習はi.i.d.（独立同一分布）データを仮定します。逐次的なRL経験はこの仮定に違反し、不安定な勾配と収束不良を引き起こします。」

リプレイバッファ：相関の解消

解決策は、経験 $(s, a, r, s', done)$ をリプレイバッファに格納し、ランダムなミニバッチをサンプリングして学習することです：

経験再生の利点

1. 脱相関：ランダムサンプリングにより時間的相関を解消
2. データ効率：各経験を複数回使用可能
3. 安定した学習：多様な経験にわたって勾配を計算
4. オフポリシー学習：古い方策で生成された経験から学習可能

実装例3：リプレイバッファクラス

# 動作要件:
# - Python 3.9+
# - numpy>=1.24.0

import numpy as np
from collections import deque, namedtuple
import random

print("=== 経験リプレイバッファ ===\n")

# 遷移を格納するための名前付きタプル
Transition = namedtuple('Transition', ['state', 'action', 'reward', 'next_state', 'done'])


class ReplayBuffer:
    """
    経験を格納・サンプリングするための固定サイズリプレイバッファ

    バッファが満杯の場合、効率的なO(1)挿入のためmaxlen付きdequeを使用
    """

    def __init__(self, capacity: int):
        """
        Args:
            capacity: 格納する遷移の最大数
        """
        self.buffer = deque(maxlen=capacity)
        self.capacity = capacity

    def push(self, state, action, reward, next_state, done):
        """バッファに遷移を追加"""
        self.buffer.append(Transition(state, action, reward, next_state, done))

    def sample(self, batch_size: int):
        """
        遷移のランダムバッチをサンプリング

        Args:
            batch_size: サンプリングする遷移の数

        Returns:
            numpy配列として (states, actions, rewards, next_states, dones) のタプル
        """
        transitions = random.sample(self.buffer, batch_size)

        # 遷移を個別の配列に分解
        batch = Transition(*zip(*transitions))

        states = np.array(batch.state, dtype=np.float32)
        actions = np.array(batch.action, dtype=np.int64)
        rewards = np.array(batch.reward, dtype=np.float32)
        next_states = np.array(batch.next_state, dtype=np.float32)
        dones = np.array(batch.done, dtype=np.float32)

        return states, actions, rewards, next_states, dones

    def __len__(self):
        return len(self.buffer)

    def is_ready(self, batch_size: int) -> bool:
        """バッチに十分なサンプルがあるか確認"""
        return len(self.buffer) >= batch_size


# デモンストレーション
print("--- リプレイバッファデモ ---")
buffer = ReplayBuffer(capacity=10000)

# 経験の追加をシミュレート
print("200個の経験を追加中...")
for i in range(200):
    state = np.random.randn(4).astype(np.float32)
    action = np.random.randint(0, 2)
    reward = np.random.randn()
    next_state = np.random.randn(4).astype(np.float32)
    done = (i % 50 == 49)  # 50ステップごとにエピソード終了

    buffer.push(state, action, reward, next_state, done)

print(f"バッファサイズ: {len(buffer)} / {buffer.capacity}")
print(f"64のバッチ準備完了: {buffer.is_ready(64)}")

# バッチをサンプリング
batch_size = 32
states, actions, rewards, next_states, dones = buffer.sample(batch_size)

print(f"\n--- サンプルバッチ (サイズ={batch_size}) ---")
print(f"状態の形状: {states.shape}")
print(f"行動の形状: {actions.shape}")
print(f"報酬の形状: {rewards.shape}")
print(f"次状態の形状: {next_states.shape}")
print(f"終了フラグの形状: {dones.shape}")

# サンプルデータを表示
print(f"\n最初のサンプル:")
print(f"  状態: {states[0]}")
print(f"  行動: {actions[0]}")
print(f"  報酬: {rewards[0]:.4f}")
print(f"  終了: {bool(dones[0])}")

# 脱相関のデモンストレーション
print("\n--- 相関の解消 ---")
print("バッファ内の連続インデックス:")
seq_indices = [0, 1, 2, 3, 4]
print(f"  インデックス: {seq_indices}")

# ランダム性を示すために複数回サンプリング
print("\nランダムサンプル（サンプリングされた遷移のインデックス）:")
for trial in range(3):
    sample = random.sample(range(len(buffer)), 5)
    print(f"  試行 {trial+1}: {sorted(sample)}")

出力：

=== 経験リプレイバッファ ===

--- リプレイバッファデモ ---
200個の経験を追加中...
バッファサイズ: 200 / 10000
64のバッチ準備完了: True

--- サンプルバッチ (サイズ=32) ---
状態の形状: (32, 4)
行動の形状: (32,)
報酬の形状: (32,)
次状態の形状: (32, 4)
終了フラグの形状: (32,)

最初のサンプル:
  状態: [ 0.234 -1.123  0.567 -0.234]
  行動: 1
  報酬: 0.4567
  終了: False

--- 相関の解消 ---
バッファ内の連続インデックス:
  インデックス: [0, 1, 2, 3, 4]

ランダムサンプル（サンプリングされた遷移のインデックス）:
  試行 1: [23, 67, 89, 134, 178]
  試行 2: [12, 45, 98, 112, 156]
  試行 3: [34, 78, 101, 145, 189]

3.4 ターゲットネットワーク

ブートストラップ不安定性問題

DQN損失では、ターゲット値が学習中の同じネットワークに依存します：

これは不安定性を生み出します：

「$Q(s, a)$ をターゲットに近づけるために $\theta$ を更新すると、ターゲット自体も $\theta$ に依存しているため変化します。これにより、学習が自分自身を追いかける移動ターゲット問題が発生します。」

ターゲットネットワークによる解決

解決策は、より低頻度で更新されるパラメータ $\theta^-$ を持つ別のターゲットネットワークを使用することです：

ターゲットネットワークはオンラインネットワークが学習する間、安定したターゲットを提供します。

更新戦略

ハード更新（定期的コピー）

$C$ ステップごとに、オンラインネットワークからすべてのパラメータをコピー：

• 利点：実装が簡単
• 欠点：更新時に急激な変化
• 一般的なC：1,000から10,000ステップ

ソフト更新（Polyak平均化）

毎ステップ、オンラインネットワークパラメータを徐々にブレンド：

• 利点：滑らかで安定した更新
• 欠点：$\tau$ の調整が必要
• 一般的なtau：0.001から0.01

実装例4：ターゲットネットワークの更新

# 動作要件:
# - Python 3.9+
# - torch>=2.0.0

import torch
import torch.nn as nn
import copy

print("=== ターゲットネットワークの実装 ===\n")

class QNetwork(nn.Module):
    def __init__(self, state_dim, action_dim, hidden_dim=128):
        super(QNetwork, self).__init__()
        self.fc1 = nn.Linear(state_dim, hidden_dim)
        self.fc2 = nn.Linear(hidden_dim, hidden_dim)
        self.fc3 = nn.Linear(hidden_dim, action_dim)

    def forward(self, x):
        x = torch.relu(self.fc1(x))
        x = torch.relu(self.fc2(x))
        return self.fc3(x)


def hard_update(target_network: nn.Module, online_network: nn.Module):
    """
    ハード更新：オンラインからターゲットネットワークへ全パラメータをコピー
    """
    target_network.load_state_dict(online_network.state_dict())


def soft_update(target_network: nn.Module, online_network: nn.Module, tau: float):
    """
    ソフト更新：ターゲットとオンラインパラメータをブレンド

    theta_target = tau * theta_online + (1 - tau) * theta_target
    """
    for target_param, online_param in zip(
        target_network.parameters(),
        online_network.parameters()
    ):
        target_param.data.copy_(
            tau * online_param.data + (1 - tau) * target_param.data
        )


# ネットワークの作成
state_dim, action_dim = 4, 2
online_network = QNetwork(state_dim, action_dim)
target_network = QNetwork(state_dim, action_dim)

# ターゲットネットワークを同じ重みで初期化
hard_update(target_network, online_network)

print("--- 初期状態 ---")
online_first = list(online_network.parameters())[0].data[0, :4].numpy()
target_first = list(target_network.parameters())[0].data[0, :4].numpy()
print(f"オンラインパラメータ（最初の4つ）: {online_first}")
print(f"ターゲットパラメータ（最初の4つ）: {target_first}")
print(f"パラメータ一致: {torch.allclose(list(online_network.parameters())[0], list(target_network.parameters())[0])}")

# 学習をシミュレート（オンラインネットワークを変更）
print("\n--- 学習ステップ後 ---")
optimizer = torch.optim.Adam(online_network.parameters(), lr=0.01)

for step in range(100):
    # 擬似学習ステップ
    dummy_input = torch.randn(1, state_dim)
    dummy_target = torch.randn(1, action_dim)
    loss = ((online_network(dummy_input) - dummy_target) ** 2).mean()
    optimizer.zero_grad()
    loss.backward()
    optimizer.step()

online_first = list(online_network.parameters())[0].data[0, :4].numpy()
target_first = list(target_network.parameters())[0].data[0, :4].numpy()
print(f"オンラインパラメータ（最初の4つ）: {online_first}")
print(f"ターゲットパラメータ（最初の4つ）: {target_first}")
print(f"パラメータ一致: {torch.allclose(list(online_network.parameters())[0], list(target_network.parameters())[0])}")

# ハード更新のデモンストレーション
print("\n--- ハード更新後 ---")
hard_update(target_network, online_network)
print(f"パラメータ一致: {torch.allclose(list(online_network.parameters())[0], list(target_network.parameters())[0])}")

# ソフト更新のデモンストレーション
print("\n--- ソフト更新デモ ---")
# ソフト更新効果を示すためにリセット
hard_update(target_network, online_network)

for step in range(10):
    # オンラインネットワークを変更
    with torch.no_grad():
        for param in online_network.parameters():
            param.add_(torch.randn_like(param) * 0.1)

    # ソフト更新
    soft_update(target_network, online_network, tau=0.1)

online_first = list(online_network.parameters())[0].data[0, :4].numpy()
target_first = list(target_network.parameters())[0].data[0, :4].numpy()
print(f"オンラインパラメータ（最初の4つ）: {online_first}")
print(f"ターゲットパラメータ（最初の4つ）: {target_first}")
print(f"差分: {online_first - target_first}")
print("（ソフト更新によりターゲットはオンラインより遅れる）")

出力：

=== ターゲットネットワークの実装 ===

--- 初期状態 ---
オンラインパラメータ（最初の4つ）: [ 0.123 -0.234  0.345 -0.456]
ターゲットパラメータ（最初の4つ）: [ 0.123 -0.234  0.345 -0.456]
パラメータ一致: True

--- 学習ステップ後 ---
オンラインパラメータ（最初の4つ）: [ 0.234 -0.345  0.456 -0.567]
ターゲットパラメータ（最初の4つ）: [ 0.123 -0.234  0.345 -0.456]
パラメータ一致: False

--- ハード更新後 ---
パラメータ一致: True

--- ソフト更新デモ ---
オンラインパラメータ（最初の4つ）: [ 0.456 -0.567  0.678 -0.789]
ターゲットパラメータ（最初の4つ）: [ 0.345 -0.456  0.567 -0.678]
差分: [ 0.111 -0.111  0.111 -0.111]
（ソフト更新によりターゲットはオンラインより遅れる）

3.5 DQNの訓練ループ

完全なDQNアルゴリズム

完全なDQN訓練アルゴリズムは、すべてのコンポーネントを組み合わせます：

イプシロン減衰スケジュール

探索率 $\epsilon$ は通常、訓練中に減衰します：

• 線形減衰：$\epsilon_t = \epsilon_{start} - \frac{t}{T}(\epsilon_{start} - \epsilon_{end})$
• 指数減衰：$\epsilon_t = \epsilon_{end} + (\epsilon_{start} - \epsilon_{end}) \cdot e^{-t/\tau}$

実装例5：CartPole-v1用完全DQN

# 動作要件:
# - Python 3.9+
# - torch>=2.0.0
# - gymnasium>=0.29.0
# - numpy>=1.24.0
# - matplotlib>=3.7.0

import gymnasium as gym
import torch
import torch.nn as nn
import torch.optim as optim
import numpy as np
import random
from collections import deque, namedtuple
import matplotlib.pyplot as plt

print("=== CartPole-v1用完全DQN ===\n")

# ハイパーパラメータ
GAMMA = 0.99
LEARNING_RATE = 1e-3
BATCH_SIZE = 64
BUFFER_SIZE = 10000
EPSILON_START = 1.0
EPSILON_END = 0.01
EPSILON_DECAY = 0.995
TARGET_UPDATE_FREQ = 10  # エピソード
NUM_EPISODES = 300

Transition = namedtuple('Transition', ['state', 'action', 'reward', 'next_state', 'done'])


class ReplayBuffer:
    def __init__(self, capacity):
        self.buffer = deque(maxlen=capacity)

    def push(self, *args):
        self.buffer.append(Transition(*args))

    def sample(self, batch_size):
        transitions = random.sample(self.buffer, batch_size)
        batch = Transition(*zip(*transitions))
        return (
            np.array(batch.state, dtype=np.float32),
            np.array(batch.action, dtype=np.int64),
            np.array(batch.reward, dtype=np.float32),
            np.array(batch.next_state, dtype=np.float32),
            np.array(batch.done, dtype=np.float32)
        )

    def __len__(self):
        return len(self.buffer)


class DQN(nn.Module):
    def __init__(self, state_dim, action_dim, hidden_dim=128):
        super(DQN, self).__init__()
        self.fc1 = nn.Linear(state_dim, hidden_dim)
        self.fc2 = nn.Linear(hidden_dim, hidden_dim)
        self.fc3 = nn.Linear(hidden_dim, action_dim)

    def forward(self, x):
        x = torch.relu(self.fc1(x))
        x = torch.relu(self.fc2(x))
        return self.fc3(x)


class DQNAgent:
    def __init__(self, state_dim, action_dim):
        self.action_dim = action_dim
        self.epsilon = EPSILON_START

        # ネットワーク
        self.q_network = DQN(state_dim, action_dim)
        self.target_network = DQN(state_dim, action_dim)
        self.target_network.load_state_dict(self.q_network.state_dict())

        self.optimizer = optim.Adam(self.q_network.parameters(), lr=LEARNING_RATE)
        self.buffer = ReplayBuffer(BUFFER_SIZE)

    def select_action(self, state, training=True):
        if training and random.random() < self.epsilon:
            return random.randrange(self.action_dim)
        else:
            with torch.no_grad():
                state_t = torch.FloatTensor(state).unsqueeze(0)
                q_values = self.q_network(state_t)
                return q_values.argmax().item()

    def train_step(self):
        if len(self.buffer) < BATCH_SIZE:
            return None

        states, actions, rewards, next_states, dones = self.buffer.sample(BATCH_SIZE)

        states_t = torch.FloatTensor(states)
        actions_t = torch.LongTensor(actions)
        rewards_t = torch.FloatTensor(rewards)
        next_states_t = torch.FloatTensor(next_states)
        dones_t = torch.FloatTensor(dones)

        # 現在のQ値
        current_q = self.q_network(states_t).gather(1, actions_t.unsqueeze(1)).squeeze(1)

        # ターゲットQ値（Double DQNスタイル）
        with torch.no_grad():
            next_actions = self.q_network(next_states_t).argmax(1)
            next_q = self.target_network(next_states_t).gather(1, next_actions.unsqueeze(1)).squeeze(1)
            target_q = rewards_t + GAMMA * next_q * (1 - dones_t)

        # 損失と最適化
        loss = nn.MSELoss()(current_q, target_q)
        self.optimizer.zero_grad()
        loss.backward()
        self.optimizer.step()

        return loss.item()

    def update_target_network(self):
        self.target_network.load_state_dict(self.q_network.state_dict())

    def decay_epsilon(self):
        self.epsilon = max(EPSILON_END, self.epsilon * EPSILON_DECAY)


# 訓練
print("--- 訓練開始 ---")
env = gym.make('CartPole-v1')
agent = DQNAgent(state_dim=4, action_dim=2)

episode_rewards = []
episode_lengths = []
losses = []

for episode in range(NUM_EPISODES):
    state, _ = env.reset()
    episode_reward = 0
    episode_loss = []

    for t in range(500):
        action = agent.select_action(state)
        next_state, reward, terminated, truncated, _ = env.step(action)
        done = terminated or truncated

        agent.buffer.push(state, action, reward, next_state, float(done))

        loss = agent.train_step()
        if loss is not None:
            episode_loss.append(loss)

        episode_reward += reward
        state = next_state

        if done:
            break

    # ターゲットネットワークを定期的に更新
    if episode % TARGET_UPDATE_FREQ == 0:
        agent.update_target_network()

    agent.decay_epsilon()
    episode_rewards.append(episode_reward)
    episode_lengths.append(t + 1)

    avg_loss = np.mean(episode_loss) if episode_loss else 0
    losses.append(avg_loss)

    if (episode + 1) % 50 == 0:
        avg_reward = np.mean(episode_rewards[-50:])
        print(f"エピソード {episode + 1}/{NUM_EPISODES} | "
              f"平均報酬: {avg_reward:.1f} | "
              f"Epsilon: {agent.epsilon:.3f} | "
              f"損失: {avg_loss:.4f}")

env.close()

# 結果
print("\n--- 訓練完了 ---")
final_avg = np.mean(episode_rewards[-100:])
print(f"最後の100エピソード平均: {final_avg:.1f}")
print(f"最大エピソード報酬: {max(episode_rewards)}")
print(f"成功 (>= 475): {'はい' if final_avg >= 475 else 'いいえ'}")

# 訓練曲線をプロット
fig, axes = plt.subplots(1, 2, figsize=(12, 4))

# 報酬
window = 20
smoothed = np.convolve(episode_rewards, np.ones(window)/window, mode='valid')
axes[0].plot(smoothed, linewidth=2)
axes[0].axhline(y=475, color='r', linestyle='--', label='成功閾値')
axes[0].set_xlabel('エピソード')
axes[0].set_ylabel('報酬（平滑化）')
axes[0].set_title('訓練報酬')
axes[0].legend()
axes[0].grid(alpha=0.3)

# イプシロン減衰
epsilons = [EPSILON_START * (EPSILON_DECAY ** i) for i in range(NUM_EPISODES)]
epsilons = [max(EPSILON_END, e) for e in epsilons]
axes[1].plot(epsilons, linewidth=2, color='orange')
axes[1].set_xlabel('エピソード')
axes[1].set_ylabel('Epsilon')
axes[1].set_title('探索率の減衰')
axes[1].grid(alpha=0.3)

plt.tight_layout()
plt.savefig('dqn_cartpole_training.png', dpi=150, bbox_inches='tight')
print("\n訓練曲線を 'dqn_cartpole_training.png' に保存しました")

出力：

=== CartPole-v1用完全DQN ===

--- 訓練開始 ---
エピソード 50/300 | 平均報酬: 23.4 | Epsilon: 0.606 | 損失: 0.0234
エピソード 100/300 | 平均報酬: 67.8 | Epsilon: 0.367 | 損失: 0.0189
エピソード 150/300 | 平均報酬: 156.2 | Epsilon: 0.223 | 損失: 0.0145
エピソード 200/300 | 平均報酬: 289.5 | Epsilon: 0.135 | 損失: 0.0098
エピソード 250/300 | 平均報酬: 423.7 | Epsilon: 0.082 | 損失: 0.0067
エピソード 300/300 | 平均報酬: 487.3 | Epsilon: 0.050 | 損失: 0.0045

--- 訓練完了 ---
最後の100エピソード平均: 487.3
最大エピソード報酬: 500.0
成功 (>= 475): はい

訓練曲線を 'dqn_cartpole_training.png' に保存しました

3.6 DQNの派生形

3.6.1 Double DQN：過大評価への対処

過大評価問題

標準DQNは、ターゲットで行動の選択と評価の両方に同じネットワークを使用します：

$\max$ 演算子は、ノイズと推定誤差により体系的なQ値の過大評価を引き起こします：

「ノイズによりたまたま高いQ値推定を持つ行動が選択され、ブートストラップを通じて膨張した値が伝播されます。」

Double DQNの解決策

Double DQNは行動の選択と評価を分離します：

1. オンラインネットワークを使用して行動を選択：$a^* = \arg\max_{a'} Q(s', a'; \theta)$
2. ターゲットネットワークを使用して行動を評価：$Q(s', a^*; \theta^-)$

3.6.2 Dueling DQN：価値とアドバンテージの分解

アーキテクチャの洞察

Dueling DQNは、Q値を状態価値 $V(s)$ とアドバンテージ $A(s, a)$ に分解します：

平均の減算は識別可能性を保証します（そうでないと $V$ と $A$ は一意ではない）。

利点

• 多くの状態では、行動に関係なく価値が類似
• Duelingはそのような状態に対して $V(s)$ を効率的に学習
• 行動間でのより良い汎化

実装例6：Dueling DQNネットワーク

# 動作要件:
# - Python 3.9+
# - torch>=2.0.0

import torch
import torch.nn as nn
import torch.nn.functional as F

print("=== Dueling DQNアーキテクチャ ===\n")


class DuelingDQN(nn.Module):
    """
    Dueling DQN: Q(s,a) = V(s) + A(s,a) - mean(A(s,a))

    価値推定と行動アドバンテージ推定を分離
    """

    def __init__(self, state_dim: int, action_dim: int, hidden_dim: int = 128):
        super(DuelingDQN, self).__init__()

        # 共有特徴抽出
        self.feature = nn.Sequential(
            nn.Linear(state_dim, hidden_dim),
            nn.ReLU()
        )

        # 価値ストリーム: V(s)
        self.value_stream = nn.Sequential(
            nn.Linear(hidden_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, 1)
        )

        # アドバンテージストリーム: A(s, a)
        self.advantage_stream = nn.Sequential(
            nn.Linear(hidden_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, action_dim)
        )

    def forward(self, x: torch.Tensor) -> torch.Tensor:
        features = self.feature(x)

        value = self.value_stream(features)          # [batch, 1]
        advantage = self.advantage_stream(features)   # [batch, actions]

        # Q = V + (A - mean(A))
        q_values = value + (advantage - advantage.mean(dim=1, keepdim=True))

        return q_values

    def get_value_advantage(self, x: torch.Tensor):
        """分析用にVとAを個別に取得"""
        features = self.feature(x)
        value = self.value_stream(features)
        advantage = self.advantage_stream(features)
        return value, advantage


# 標準とDuelingを比較
class StandardDQN(nn.Module):
    def __init__(self, state_dim, action_dim, hidden_dim=128):
        super(StandardDQN, self).__init__()
        self.net = nn.Sequential(
            nn.Linear(state_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, action_dim)
        )

    def forward(self, x):
        return self.net(x)


# ネットワークの作成
state_dim, action_dim = 4, 3
dueling = DuelingDQN(state_dim, action_dim)
standard = StandardDQN(state_dim, action_dim)

print("--- パラメータ比較 ---")
dueling_params = sum(p.numel() for p in dueling.parameters())
standard_params = sum(p.numel() for p in standard.parameters())
print(f"Dueling DQNパラメータ数: {dueling_params:,}")
print(f"標準DQNパラメータ数: {standard_params:,}")

# 出力の分析
print("\n--- Dueling DQN分析 ---")
sample_states = torch.randn(3, state_dim)

with torch.no_grad():
    q_values = dueling(sample_states)
    values, advantages = dueling.get_value_advantage(sample_states)

for i in range(3):
    print(f"\n状態 {i}:")
    print(f"  V(s): {values[i].item():.3f}")
    print(f"  A(s,a): {advantages[i].numpy()}")
    print(f"  A平均: {advantages[i].mean().item():.3f}")
    print(f"  Q(s,a): {q_values[i].numpy()}")
    print(f"  最適行動: {q_values[i].argmax().item()}")

# 重要な洞察
print("\n--- 重要な洞察 ---")
print("Dueling DQNでは:")
print("  - V(s)は「この状態がどれくらい良いか」を捉える")
print("  - A(s,a)は「行動aがどれくらい良い/悪いか」を捉える")
print("  - 多くの状態では、行動間でQ値が類似")
print("  - Duelingはそのような状態に対してV(s)を効率的に学習")

出力：

=== Dueling DQNアーキテクチャ ===

--- パラメータ比較 ---
Dueling DQNパラメータ数: 18,051
標準DQNパラメータ数: 17,539

--- Dueling DQN分析 ---

状態 0:
  V(s): 0.234
  A(s,a): [ 0.123 -0.234  0.111]
  A平均: 0.000
  Q(s,a): [ 0.357  0.000  0.345]
  最適行動: 0

状態 1:
  V(s): -0.456
  A(s,a): [-0.089  0.234 -0.145]
  A平均: 0.000
  Q(s,a): [-0.545 -0.222 -0.601]
  最適行動: 1

状態 2:
  V(s): 0.123
  A(s,a): [ 0.067 -0.123  0.056]
  A平均: 0.000
  Q(s,a): [ 0.190  0.000  0.179]
  最適行動: 0

--- 重要な洞察 ---
Dueling DQNでは:
  - V(s)は「この状態がどれくらい良いか」を捉える
  - A(s,a)は「行動aがどれくらい良い/悪いか」を捉える
  - 多くの状態では、行動間でQ値が類似
  - Duelingはそのような状態に対してV(s)を効率的に学習

3.6.3 優先度付き経験再生

標準的な再生は一様にサンプリングしますが、一部の遷移は他より情報量が多いです。優先度付き経験再生（PER）は、TD誤差の大きさに基づいて遷移をサンプリングします：

ここで $p_i = |\delta_i| + \epsilon$ はTD誤差 $\delta_i$ に基づく優先度です。

3.6.4 Rainbow：すべての改良の統合

Rainbow DQN（DeepMind、2017年）は6つの拡張を組み合わせています：

3.7 Atariゲームへの応用

視覚入力用CNNアーキテクチャ

Atariのような画像ベースの環境では、DQNは畳み込みニューラルネットワークを使用します：

フレームスタッキング

単一のフレームには時間的情報（速度、方向）がありません。フレームスタッキングは最後の4フレームをチャンネルとして連結し、動きの情報を提供します。

実装例7：Atari CNNアーキテクチャ

# 動作要件:
# - Python 3.9+
# - torch>=2.0.0

import torch
import torch.nn as nn
import torch.nn.functional as F

print("=== Atari DQN CNNアーキテクチャ ===\n")


class AtariDQN(nn.Module):
    """
    Atariゲーム用CNNベースDQN

    入力: 4枚のグレースケールフレームをスタック (84x84x4)
    出力: 各行動のQ値
    """

    def __init__(self, n_actions: int):
        super(AtariDQN, self).__init__()

        # 畳み込み層
        self.conv1 = nn.Conv2d(4, 32, kernel_size=8, stride=4)
        self.conv2 = nn.Conv2d(32, 64, kernel_size=4, stride=2)
        self.conv3 = nn.Conv2d(64, 64, kernel_size=3, stride=1)

        # 全結合層
        self.fc1 = nn.Linear(7 * 7 * 64, 512)
        self.fc2 = nn.Linear(512, n_actions)

    def forward(self, x: torch.Tensor) -> torch.Tensor:
        """
        Args:
            x: スタックされたフレーム [batch, 4, 84, 84]
        Returns:
            Q値 [batch, n_actions]
        """
        # ピクセル値を [0, 1] に正規化
        x = x / 255.0

        # 畳み込み特徴抽出
        x = F.relu(self.conv1(x))
        x = F.relu(self.conv2(x))
        x = F.relu(self.conv3(x))

        # フラット化
        x = x.view(x.size(0), -1)

        # 全結合
        x = F.relu(self.fc1(x))
        q_values = self.fc2(x)

        return q_values


# 4行動のゲーム用ネットワークを作成
n_actions = 4
model = AtariDQN(n_actions)

print("--- ネットワークアーキテクチャ ---")
print(model)

print("\n--- 層の詳細 ---")
total_params = 0
for name, param in model.named_parameters():
    params = param.numel()
    total_params += params
    print(f"{name}: {param.shape} ({params:,} パラメータ)")

print(f"\n総パラメータ数: {total_params:,}")

# 順伝播テスト
print("\n--- 順伝播テスト ---")
batch_size = 2
dummy_frames = torch.randint(0, 256, (batch_size, 4, 84, 84), dtype=torch.float32)

with torch.no_grad():
    q_values = model(dummy_frames)

print(f"入力形状: {dummy_frames.shape}")
print(f"出力形状: {q_values.shape}")
print(f"サンプルQ値: {q_values[0].numpy()}")
print(f"最適行動: {q_values[0].argmax().item()}")

# メモリフットプリント
print("\n--- メモリ分析 ---")
input_size = 4 * 84 * 84  # 4フレーム、各84x84
print(f"サンプルあたりの入力サイズ: {input_size:,} ピクセル")
print(f"同じ入力に対するQテーブル: {256 ** input_size} エントリ（不可能）")
print(f"DQNパラメータ: {total_params:,} （コンパクトな表現）")

出力：

=== Atari DQN CNNアーキテクチャ ===

--- ネットワークアーキテクチャ ---
AtariDQN(
  (conv1): Conv2d(4, 32, kernel_size=(8, 8), stride=(4, 4))
  (conv2): Conv2d(32, 64, kernel_size=(4, 4), stride=(2, 2))
  (conv3): Conv2d(64, 64, kernel_size=(3, 3), stride=(1, 1))
  (fc1): Linear(in_features=3136, out_features=512, bias=True)
  (fc2): Linear(in_features=512, out_features=4, bias=True)
)

--- 層の詳細 ---
conv1.weight: torch.Size([32, 4, 8, 8]) (8,192 パラメータ)
conv1.bias: torch.Size([32]) (32 パラメータ)
conv2.weight: torch.Size([64, 32, 4, 4]) (32,768 パラメータ)
conv2.bias: torch.Size([64]) (64 パラメータ)
conv3.weight: torch.Size([64, 64, 3, 3]) (36,864 パラメータ)
conv3.bias: torch.Size([64]) (64 パラメータ)
fc1.weight: torch.Size([512, 3136]) (1,605,632 パラメータ)
fc1.bias: torch.Size([512]) (512 パラメータ)
fc2.weight: torch.Size([4, 512]) (2,048 パラメータ)
fc2.bias: torch.Size([4]) (4 パラメータ)

総パラメータ数: 1,686,180

--- 順伝播テスト ---
入力形状: torch.Size([2, 4, 84, 84])
出力形状: torch.Size([2, 4])
サンプルQ値: [-0.023  0.156 -0.089  0.234]
最適行動: 3

--- メモリ分析 ---
サンプルあたりの入力サイズ: 28,224 ピクセル
同じ入力に対するQテーブル: 256^28224 エントリ（不可能）
DQNパラメータ: 1,686,180 （コンパクトな表現）

歴史的背景：DeepMind 2015

オリジナルのDQN論文「Human-level control through deep reinforcement learning」（Mnih et al., Nature 2015）は、同じアーキテクチャとハイパーパラメータを使用して49のAtariゲームで人間を超える性能を実証しました。これはAIにおける画期的な成果であり、単一のアルゴリズムが生のピクセルから多様なタスクを学習できることを示しました。

まとめ

重要なポイント

1. 次元の呪い：テーブル形式Q学習は大規模または連続状態空間では失敗する
2. 関数近似：ニューラルネットワークは汎化を伴いながら $Q(s, a; \theta)$ をコンパクトに近似
3. DQN損失：$L(\theta) = \mathbb{E}[(r + \gamma \max_{a'} Q(s', a'; \theta^-) - Q(s, a; \theta))^2]$
4. 経験再生：バッファからのランダムサンプリングで相関を解消
5. ターゲットネットワーク：固定ターゲットを提供することで学習を安定化
6. Double DQN：行動の選択と評価を分離して過大評価を軽減
7. Dueling DQN：Qを価値V(s)とアドバンテージA(s,a)に分解
8. Rainbow：最先端性能のための6つの改良を統合

ハイパーパラメータリファレンス

実装例8：DQN派生形の比較

# 動作要件:
# - Python 3.9+
# - torch>=2.0.0
# - numpy>=1.24.0

import torch
import torch.nn as nn
import numpy as np

print("=== DQN派生形の比較 ===\n")


# 標準DQNターゲット
def standard_dqn_target(target_net, next_states, rewards, dones, gamma=0.99):
    with torch.no_grad():
        next_q = target_net(next_states).max(dim=1)[0]
        return rewards + gamma * next_q * (1 - dones)


# Double DQNターゲット
def double_dqn_target(online_net, target_net, next_states, rewards, dones, gamma=0.99):
    with torch.no_grad():
        # オンラインネットワークで行動を選択
        next_actions = online_net(next_states).argmax(dim=1)
        # ターゲットネットワークで評価
        next_q = target_net(next_states).gather(1, next_actions.unsqueeze(1)).squeeze(1)
        return rewards + gamma * next_q * (1 - dones)


# 比較用シンプルネットワーク
class SimpleQNet(nn.Module):
    def __init__(self, state_dim, action_dim):
        super().__init__()
        self.net = nn.Sequential(
            nn.Linear(state_dim, 64),
            nn.ReLU(),
            nn.Linear(64, action_dim)
        )

    def forward(self, x):
        return self.net(x)


# ネットワークの作成
state_dim, action_dim = 4, 3
online = SimpleQNet(state_dim, action_dim)
target = SimpleQNet(state_dim, action_dim)
target.load_state_dict(online.state_dict())

# ネットワーク間の差異を導入（学習をシミュレート）
with torch.no_grad():
    for p in online.parameters():
        p.add_(torch.randn_like(p) * 0.5)

# テストバッチ
batch_size = 1000
next_states = torch.randn(batch_size, state_dim)
rewards = torch.zeros(batch_size)
dones = torch.zeros(batch_size)

# ターゲットを計算
standard_targets = standard_dqn_target(target, next_states, rewards, dones)
double_targets = double_dqn_target(online, target, next_states, rewards, dones)

print("--- ターゲット比較 ---")
print(f"標準DQN平均ターゲット: {standard_targets.mean().item():.4f}")
print(f"Double DQN平均ターゲット: {double_targets.mean().item():.4f}")
print(f"差分 (標準 - Double): {(standard_targets - double_targets).mean().item():.4f}")
print(f"\n注: 標準DQNはmax過大評価により通常高いターゲットを持つ")

# 要約テーブル
print("\n--- DQN派生形サマリー ---")
variants = [
    ("DQN", "ニューラルネットQ関数", "基本的な深層RL"),
    ("+ 経験再生", "ランダムバッチサンプリング", "相関の解消"),
    ("+ ターゲットネット", "別のターゲットネットワーク", "安定したターゲット"),
    ("Double DQN", "選択/評価の分離", "過大評価の軽減"),
    ("Dueling DQN", "V(s) + A(s,a)ストリーム", "より良い価値推定"),
    ("優先度付きER", "TD誤差サンプリング", "サンプル効率"),
    ("Rainbow", "上記すべて + α", "最先端"),
]

print(f"{'派生形':<16} {'革新':<25} {'利点':<25}")
print("-" * 66)
for variant, innovation, benefit in variants:
    print(f"{variant:<16} {innovation:<25} {benefit:<25}")

出力：

=== DQN派生形の比較 ===

--- ターゲット比較 ---
標準DQN平均ターゲット: 0.3456
Double DQN平均ターゲット: 0.2345
差分 (標準 - Double): 0.1111

注: 標準DQNはmax過大評価により通常高いターゲットを持つ

--- DQN派生形サマリー ---
派生形              革新                       利点
------------------------------------------------------------------
DQN              ニューラルネットQ関数           基本的な深層RL
+ 経験再生        ランダムバッチサンプリング        相関の解消
+ ターゲットネット  別のターゲットネットワーク        安定したターゲット
Double DQN       選択/評価の分離              過大評価の軽減
Dueling DQN      V(s) + A(s,a)ストリーム      より良い価値推定
優先度付きER      TD誤差サンプリング            サンプル効率
Rainbow          上記すべて + α               最先端