Split View: [심층 강화학습] 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

        # 두 개의 Q 네트워크 (쌍둥이 Critic)
        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

        # === Critic 업데이트 ===
        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()

        # === Actor 업데이트 ===
        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)
        mu = self.mu(features)
        log_std = self.log_std(features).clamp(-20, 2)
        return mu, log_std

    def sample(self, obs):
        mu, log_std = self.forward(obs)
        std = log_std.exp()
        dist = torch.distributions.Normal(mu, std)
        # Reparameterization trick
        z = dist.rsample()
        action = torch.tanh(z)
        # 로그 확률 보정 (tanh squashing)
        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):
        # 기존 Q 학습 손실
        td_loss = self.compute_td_loss(q_network, states, actions)

        # CQL 정규화: OOD 행동의 Q값을 낮춤
        # 데이터에 없는 행동의 가치를 과대평가하지 않도록
        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: 단일 파일 구현, 교육용으로 최고
# 각 알고리즘이 하나의 파일에 완전히 구현됨

# 3. RLlib (Ray): 분산 학습, 프로덕션 레벨
# pip install ray[rllib]

# 4. Tianshou: PyTorch 기반, 유연한 구조
# pip install tianshou