[深層強化学習] 20. 深層強化学習総まとめ：アルゴリズム比較と選択ガイド

# DQN系列アルゴリズムの核心的な違いを示す疑似コード

def dqn_target(reward, next_state, done, gamma, q_network, target_network):
    """基本DQN：ターゲットネットワークで最大Q値を計算"""
    with torch.no_grad():
        max_q = target_network(next_state).max(dim=-1)[0]
        target = reward + gamma * (1 - done) * max_q
    return target

def double_dqn_target(reward, next_state, done, gamma,
                       q_network, target_network):
    """Double DQN：行動選択と評価を分離"""
    with torch.no_grad():
        # メインネットワークで行動選択
        best_actions = q_network(next_state).argmax(dim=-1)
        # ターゲットネットワークで価値評価
        q_values = target_network(next_state)
        max_q = q_values.gather(1, best_actions.unsqueeze(1)).squeeze()
        target = reward + gamma * (1 - done) * max_q
    return target

def dueling_network_forward(features, advantage_stream, value_stream):
    """Dueling DQN：価値とアドバンテージを分離"""
    value = value_stream(features)        # V(s)
    advantage = advantage_stream(features) # A(s,a)
    # Q(s,a) = V(s) + A(s,a) - mean(A)
    q_values = value + advantage - advantage.mean(dim=-1, keepdim=True)
    return q_values

import torch

def reinforce_loss(log_probs, returns):
    """基本REINFORCE：高分散"""
    return -(log_probs * returns).mean()

def reinforce_with_baseline(log_probs, returns, values):
    """ベースラインREINFORCE：分散低減"""
    advantages = returns - values.detach()
    policy_loss = -(log_probs * advantages).mean()
    value_loss = (returns - values).pow(2).mean()
    return policy_loss + 0.5 * value_loss

def ppo_clipped_loss(log_probs, old_log_probs, advantages,
                     clip_epsilon=0.2):
    """PPO：安定した方策更新"""
    ratio = torch.exp(log_probs - old_log_probs)
    surr1 = ratio * advantages
    surr2 = torch.clamp(ratio, 1 - clip_epsilon,
                         1 + clip_epsilon) * advantages
    return -torch.min(surr1, surr2).mean()

import torch
import torch.nn as nn
import copy

class SACAgent:
    """SAC：最大エントロピー強化学習"""

    def __init__(self, obs_size, act_size, hidden=256,
                 lr=3e-4, gamma=0.99, tau=0.005):
        self.gamma = gamma
        self.tau = tau
        self.q1 = self._make_q(obs_size, act_size, hidden)
        self.q2 = self._make_q(obs_size, act_size, hidden)
        self.q1_target = copy.deepcopy(self.q1)
        self.q2_target = copy.deepcopy(self.q2)
        self.actor = self._make_actor(obs_size, act_size, hidden)
        self.log_alpha = torch.zeros(1, requires_grad=True)
        self.target_entropy = -act_size
        self.q1_opt = torch.optim.Adam(self.q1.parameters(), lr=lr)
        self.q2_opt = torch.optim.Adam(self.q2.parameters(), lr=lr)
        self.actor_opt = torch.optim.Adam(self.actor.parameters(), lr=lr)
        self.alpha_opt = torch.optim.Adam([self.log_alpha], lr=lr)

    def _make_q(self, obs_size, act_size, hidden):
        return nn.Sequential(
            nn.Linear(obs_size + act_size, hidden), nn.ReLU(),
            nn.Linear(hidden, hidden), nn.ReLU(),
            nn.Linear(hidden, 1),
        )

    def _make_actor(self, obs_size, act_size, hidden):
        return GaussianActor(obs_size, act_size, hidden)

    @property
    def alpha(self):
        return self.log_alpha.exp()

    def update(self, batch):
        states, actions, rewards, next_states, dones = batch

        with torch.no_grad():
            next_actions, next_log_probs = self.actor.sample(next_states)
            q1_next = self.q1_target(torch.cat([next_states, next_actions], -1))
            q2_next = self.q2_target(torch.cat([next_states, next_actions], -1))
            q_next = torch.min(q1_next, q2_next)
            target = rewards + self.gamma * (1 - dones) * (
                q_next - self.alpha * next_log_probs
            )

        q1_val = self.q1(torch.cat([states, actions], -1))
        q2_val = self.q2(torch.cat([states, actions], -1))
        q1_loss = (q1_val - target).pow(2).mean()
        q2_loss = (q2_val - target).pow(2).mean()

        self.q1_opt.zero_grad(); q1_loss.backward(); self.q1_opt.step()
        self.q2_opt.zero_grad(); q2_loss.backward(); self.q2_opt.step()

        new_actions, log_probs = self.actor.sample(states)
        q1_new = self.q1(torch.cat([states, new_actions], -1))
        q2_new = self.q2(torch.cat([states, new_actions], -1))
        q_new = torch.min(q1_new, q2_new)
        actor_loss = (self.alpha.detach() * log_probs - q_new).mean()

        self.actor_opt.zero_grad(); actor_loss.backward(); self.actor_opt.step()

        alpha_loss = -(self.log_alpha * (log_probs.detach() + self.target_entropy)).mean()
        self.alpha_opt.zero_grad(); alpha_loss.backward(); self.alpha_opt.step()

        self._soft_update(self.q1, self.q1_target)
        self._soft_update(self.q2, self.q2_target)

    def _soft_update(self, source, target):
        for s, t in zip(source.parameters(), target.parameters()):
            t.data.copy_(self.tau * s.data + (1 - self.tau) * t.data)

class GaussianActor(nn.Module):
    def __init__(self, obs_size, act_size, hidden):
        super().__init__()
        self.net = nn.Sequential(
            nn.Linear(obs_size, hidden), nn.ReLU(),
            nn.Linear(hidden, hidden), nn.ReLU(),
        )
        self.mu = nn.Linear(hidden, act_size)
        self.log_std = nn.Linear(hidden, act_size)

    def forward(self, obs):
        features = self.net(obs)
        return self.mu(features), self.log_std(features).clamp(-20, 2)

    def sample(self, obs):
        mu, log_std = self.forward(obs)
        std = log_std.exp()
        dist = torch.distributions.Normal(mu, std)
        z = dist.rsample()
        action = torch.tanh(z)
        log_prob = (dist.log_prob(z) - torch.log(1 - action.pow(2) + 1e-6)).sum(dim=-1, keepdim=True)
        return action, log_prob

# 各アルゴリズムの主要ハイパーパラメータと一般的な値

hyperparams = {
    'DQN': {
        'lr': 1e-4,
        'batch_size': 32,
        'buffer_size': 1000000,
        'target_update_freq': 1000,
        'epsilon_decay': 'linear to 0.01 over 1M steps',
        'sensitivity': 'medium',
    },
    'PPO': {
        'lr': 3e-4,
        'clip_epsilon': 0.2,
        'num_epochs': 10,
        'batch_size': 64,
        'gae_lambda': 0.95,
        'entropy_coef': 0.01,
        'sensitivity': 'low',
    },
    'SAC': {
        'lr': 3e-4,
        'batch_size': 256,
        'buffer_size': 1000000,
        'tau': 0.005,
        'auto_alpha': True,
        'sensitivity': 'low',
    },
    'DDPG': {
        'lr_actor': 1e-4,
        'lr_critic': 1e-3,
        'batch_size': 256,
        'tau': 0.005,
        'noise_type': 'OU or Gaussian',
        'sensitivity': 'high',
    },
}

class ConservativeQLearning:
    """CQL：保守的Q学習の核心アイデア"""

    def compute_cql_loss(self, q_network, states, actions, alpha=1.0):
        td_loss = self.compute_td_loss(q_network, states, actions)
        random_actions = torch.rand_like(actions) * 2 - 1
        random_q = q_network(states, random_actions)
        data_q = q_network(states, actions)
        cql_penalty = (
            torch.logsumexp(random_q, dim=0).mean() - data_q.mean()
        )
        return td_loss + alpha * cql_penalty

class SafeRLObjective:
    """制約ベースの安全強化学習の目的"""

    def compute_objective(self, policy, states):
        expected_reward = self.estimate_reward(policy, states)
        expected_cost = self.estimate_cost(policy, states)
        cost_limit = 25.0
        lagrangian = (expected_reward
                      - self.lambda_multiplier
                      * (expected_cost - cost_limit))
        return lagrangian

# 主要RLライブラリ

# 1. Stable-Baselines3: 検証済み実装、高速実験
# pip install stable-baselines3
from stable_baselines3 import PPO, SAC, DQN

model = PPO("MlpPolicy", "CartPole-v1", verbose=1)
model.learn(total_timesteps=100000)

# 2. CleanRL: 単一ファイル実装、教育用に最高
# 各アルゴリズムが1つのファイルに完全実装

# 3. RLlib (Ray): 分散学習、プロダクションレベル
# pip install ray[rllib]

# 4. Tianshou: PyTorchベース、柔軟な構造
# pip install tianshou