PPO (Proximal Policy Optimization)

PPO는 OpenAI가 2017년에 발표한 Policy Gradient 알고리즘으로, 구현이 간단하면서도 높은 성능을 보여 현재 가장 널리 사용되는 강화학습 알고리즘 중 하나입니다. ChatGPT의 RLHF 학습에도 PPO가 사용되었습니다.

---

1. PPO가 해결하는 문제

Policy Gradient의 문제점

기본 Policy Gradient (REINFORCE, A2C)는 학습이 불안정합니다:

• 스텝 크기가 너무 크면: 정책이 급격히 변해서 성능이 붕괴
• 스텝 크기가 너무 작으면: 학습이 너무 느림

정책 변화가 너무 크면:
π_old: "왼쪽 70%, 오른쪽 30%"
    ↓ (한 번의 업데이트)
π_new: "왼쪽 10%, 오른쪽 90%"
    → 갑자기 완전히 다른 행동 → 성능 붕괴 가능

TRPO의 접근

TRPO (Trust Region Policy Optimization)는 KL divergence 제약을 두어 정책 변화를 제한합니다:

$\max_\theta \mathbb{E}[r(\theta) \hat{A}] \quad \text{subject to} \quad D_{KL}(\pi_{old} || \pi_{new}) \leq \delta$

하지만 TRPO는 제약 조건 최적화가 복잡합니다.

PPO의 해결책

PPO는 클리핑을 사용하여 간단하게 정책 변화를 제한합니다.

---

2. PPO 핵심 개념

Probability Ratio

$r(\theta) = \frac{\pi_\theta(a|s)}{\pi_{\theta_{old}}(a|s)}$

• r = 1: 새 정책과 이전 정책이 같음
• r > 1: 새 정책이 해당 행동을 더 많이 선택
• r < 1: 새 정책이 해당 행동을 덜 선택

PPO-Clip 목적 함수

$L^{CLIP}(\theta) = \mathbb{E}\left[ \min\left( r(\theta) \hat{A}, \text{clip}(r(\theta), 1-\epsilon, 1+\epsilon) \hat{A} \right) \right]$

• ε: 클리핑 범위 (보통 0.2)
• Â: Advantage 추정값

클리핑의 효과

r(θ)가 [1-ε, 1+ε] 범위를 벗어나면 그래디언트가 0이 됨
→ 정책이 너무 많이 변하는 것을 방지

---

3. PPO 알고리즘

1. 현재 정책 π_θ_old로 경험 수집 (여러 에피소드 또는 고정 스텝)
2. Advantage 계산 (GAE 사용)
3. 수집된 데이터로 여러 epoch 학습:
   a. 미니배치 샘플링
   b. r(θ) = π_θ(a|s) / π_θ_old(a|s) 계산
   c. Clipped 손실 계산
   d. 네트워크 업데이트
4. θ_old ← θ
5. 반복

핵심 특징: 같은 데이터로 여러 번 업데이트 가능 (샘플 효율성)

---

4. PyTorch 구현

네트워크 정의

import torch
import torch.nn as nn
import torch.optim as optim
import torch.nn.functional as F
from torch.distributions import Categorical
import gymnasium as gym
import numpy as np

class PPONetwork(nn.Module):
    """Actor-Critic 네트워크"""

    def __init__(self, state_dim, action_dim):
        super().__init__()

        # 공유 레이어
        self.shared = nn.Sequential(
            nn.Linear(state_dim, 64),
            nn.Tanh(),
            nn.Linear(64, 64),
            nn.Tanh()
        )

        # Actor (정책)
        self.actor = nn.Linear(64, action_dim)

        # Critic (가치)
        self.critic = nn.Linear(64, 1)

    def forward(self, x):
        shared = self.shared(x)
        return self.actor(shared), self.critic(shared)

    def get_action(self, state):
        logits, value = self(state)
        probs = F.softmax(logits, dim=-1)
        dist = Categorical(probs)
        action = dist.sample()
        log_prob = dist.log_prob(action)
        return action, log_prob, value.squeeze()

    def evaluate(self, states, actions):
        logits, values = self(states)
        probs = F.softmax(logits, dim=-1)
        dist = Categorical(probs)
        log_probs = dist.log_prob(actions)
        entropy = dist.entropy()
        return log_probs, values.squeeze(), entropy

경험 버퍼

class RolloutBuffer:
    def __init__(self):
        self.states = []
        self.actions = []
        self.rewards = []
        self.dones = []
        self.log_probs = []
        self.values = []

    def store(self, state, action, reward, done, log_prob, value):
        self.states.append(state)
        self.actions.append(action)
        self.rewards.append(reward)
        self.dones.append(done)
        self.log_probs.append(log_prob)
        self.values.append(value)

    def clear(self):
        self.states = []
        self.actions = []
        self.rewards = []
        self.dones = []
        self.log_probs = []
        self.values = []

    def get(self):
        return (
            torch.FloatTensor(np.array(self.states)),
            torch.LongTensor(self.actions),
            torch.FloatTensor(self.rewards),
            torch.FloatTensor(self.dones),
            torch.stack(self.log_probs),
            torch.stack(self.values)
        )

PPO 에이전트

class PPOAgent:
    def __init__(self, state_dim, action_dim,
                 lr=3e-4, gamma=0.99, lam=0.95,
                 clip_epsilon=0.2, epochs=10, batch_size=64,
                 value_coef=0.5, entropy_coef=0.01):

        self.gamma = gamma
        self.lam = lam
        self.clip_epsilon = clip_epsilon
        self.epochs = epochs
        self.batch_size = batch_size
        self.value_coef = value_coef
        self.entropy_coef = entropy_coef

        self.network = PPONetwork(state_dim, action_dim)
        self.optimizer = optim.Adam(self.network.parameters(), lr=lr)
        self.buffer = RolloutBuffer()

    def select_action(self, state):
        state_tensor = torch.FloatTensor(state).unsqueeze(0)
        with torch.no_grad():
            action, log_prob, value = self.network.get_action(state_tensor)
        return action.item(), log_prob, value

    def compute_gae(self, rewards, values, dones, next_value):
        """Generalized Advantage Estimation"""
        advantages = []
        gae = 0

        values_list = values.tolist() + [next_value]

        for t in reversed(range(len(rewards))):
            delta = rewards[t] + self.gamma * values_list[t + 1] * (1 - dones[t]) - values_list[t]
            gae = delta + self.gamma * self.lam * (1 - dones[t]) * gae
            advantages.insert(0, gae)

        advantages = torch.FloatTensor(advantages)
        returns = advantages + values

        return advantages, returns

    def update(self, next_value):
        states, actions, rewards, dones, old_log_probs, old_values = self.buffer.get()

        # GAE 계산
        advantages, returns = self.compute_gae(rewards, old_values, dones, next_value)

        # Advantage 정규화
        advantages = (advantages - advantages.mean()) / (advantages.std() + 1e-8)

        # 여러 epoch 학습
        dataset_size = len(states)
        indices = np.arange(dataset_size)

        for _ in range(self.epochs):
            np.random.shuffle(indices)

            for start in range(0, dataset_size, self.batch_size):
                end = start + self.batch_size
                batch_indices = indices[start:end]

                batch_states = states[batch_indices]
                batch_actions = actions[batch_indices]
                batch_old_log_probs = old_log_probs[batch_indices]
                batch_advantages = advantages[batch_indices]
                batch_returns = returns[batch_indices]

                # 현재 정책으로 평가
                new_log_probs, new_values, entropy = self.network.evaluate(
                    batch_states, batch_actions
                )

                # Probability ratio
                ratio = torch.exp(new_log_probs - batch_old_log_probs.detach())

                # Clipped surrogate loss
                surr1 = ratio * batch_advantages
                surr2 = torch.clamp(ratio, 1 - self.clip_epsilon, 1 + self.clip_epsilon) * batch_advantages
                actor_loss = -torch.min(surr1, surr2).mean()

                # Value loss
                value_loss = F.mse_loss(new_values, batch_returns.detach())

                # Entropy bonus
                entropy_loss = -entropy.mean()

                # Total loss
                loss = actor_loss + self.value_coef * value_loss + self.entropy_coef * entropy_loss

                self.optimizer.zero_grad()
                loss.backward()
                torch.nn.utils.clip_grad_norm_(self.network.parameters(), 0.5)
                self.optimizer.step()

        self.buffer.clear()

        return actor_loss.item(), value_loss.item()

학습 함수

def train_ppo(env_name='CartPole-v1', total_timesteps=100000, rollout_steps=2048):
    env = gym.make(env_name)
    state_dim = env.observation_space.shape[0]
    action_dim = env.action_space.n

    agent = PPOAgent(state_dim, action_dim)
    episode_rewards = []
    current_episode_reward = 0

    state, _ = env.reset()
    timestep = 0

    while timestep < total_timesteps:
        # Rollout 수집
        for _ in range(rollout_steps):
            action, log_prob, value = agent.select_action(state)
            next_state, reward, terminated, truncated, _ = env.step(action)
            done = terminated or truncated

            agent.buffer.store(state, action, reward, float(done), log_prob, value)

            current_episode_reward += reward
            timestep += 1

            if done:
                episode_rewards.append(current_episode_reward)
                current_episode_reward = 0
                state, _ = env.reset()
            else:
                state = next_state

        # 마지막 상태의 가치 추정
        with torch.no_grad():
            state_tensor = torch.FloatTensor(state).unsqueeze(0)
            _, _, next_value = agent.network.get_action(state_tensor)
            next_value = next_value.item() if not done else 0

        # PPO 업데이트
        actor_loss, value_loss = agent.update(next_value)

        # 로깅
        if len(episode_rewards) > 0 and len(episode_rewards) % 10 == 0:
            avg_reward = np.mean(episode_rewards[-100:])
            print(f"Timestep {timestep}, Episodes: {len(episode_rewards)}, "
                  f"Avg Reward: {avg_reward:.1f}")

    env.close()
    return episode_rewards


if __name__ == "__main__":
    rewards = train_ppo()

---

5. 연속 행동 공간

로봇 제어 등 연속 행동에서는 가우시안 정책을 사용합니다.

class PPOContinuous(nn.Module):
    def __init__(self, state_dim, action_dim):
        super().__init__()

        self.shared = nn.Sequential(
            nn.Linear(state_dim, 64),
            nn.Tanh(),
            nn.Linear(64, 64),
            nn.Tanh()
        )

        # 평균 출력
        self.mean = nn.Linear(64, action_dim)
        # 로그 표준편차 (학습 가능 파라미터)
        self.log_std = nn.Parameter(torch.zeros(action_dim))

        # Critic
        self.critic = nn.Linear(64, 1)

    def forward(self, x):
        shared = self.shared(x)
        mean = self.mean(shared)
        std = self.log_std.exp()
        value = self.critic(shared)
        return mean, std, value

    def get_action(self, state):
        mean, std, value = self(state)
        dist = torch.distributions.Normal(mean, std)
        action = dist.sample()
        action = torch.tanh(action)  # 범위 제한

        log_prob = dist.log_prob(action).sum(dim=-1)
        return action, log_prob, value.squeeze()

    def evaluate(self, states, actions):
        mean, std, values = self(states)
        dist = torch.distributions.Normal(mean, std)

        log_probs = dist.log_prob(actions).sum(dim=-1)
        entropy = dist.entropy().sum(dim=-1)

        return log_probs, values.squeeze(), entropy

---

6. 주요 하이퍼파라미터

파라미터 튜닝 팁

• 학습이 불안정: clip_epsilon 줄이기, epochs 줄이기
• 학습이 느림: clip_epsilon 늘리기, lr 늘리기
• 조기 수렴: entropy_coef 늘리기

---

7. PPO의 장단점

장점

1. 구현 간단: TRPO 대비 매우 간단
2. 안정적 학습: 클리핑으로 급격한 정책 변화 방지
3. 샘플 효율적: 같은 데이터로 여러 번 학습
4. 범용성: 이산/연속 행동 모두 지원
5. 확장성: 병렬 환경으로 쉽게 확장

단점

1. On-policy: 오래된 데이터 재사용 불가
2. 하이퍼파라미터 민감: 환경마다 조정 필요
3. 샘플 효율성 한계: Off-policy 방법 대비

---

8. PPO vs 다른 알고리즘

---

9. 실전 사용 예시

Stable-Baselines3 사용

from stable_baselines3 import PPO
from stable_baselines3.common.env_util import make_vec_env

# 병렬 환경 생성
env = make_vec_env("CartPole-v1", n_envs=4)

# PPO 에이전트 생성
model = PPO(
    "MlpPolicy",
    env,
    learning_rate=3e-4,
    n_steps=2048,
    batch_size=64,
    n_epochs=10,
    gamma=0.99,
    gae_lambda=0.95,
    clip_range=0.2,
    verbose=1
)

# 학습
model.learn(total_timesteps=100000)

# 저장
model.save("ppo_cartpole")

# 테스트
model = PPO.load("ppo_cartpole")
obs = env.reset()
for _ in range(1000):
    action, _ = model.predict(obs)
    obs, reward, done, info = env.step(action)

---

10. PPO 변형들

PPO-Penalty

클리핑 대신 KL divergence 페널티 사용:

$L(\theta) = \mathbb{E}[r(\theta)\hat{A} - \beta D_{KL}(\pi_{old} || \pi_\theta)]$

APPO (Asynchronous PPO)

비동기 병렬 학습으로 처리량 증가

2025년 트렌드: GRPO

DeepSeek-R1 등에서 사용되는 GRPO (Group Relative Policy Optimization)는 PPO의 변형으로, 보상 모델 없이 그룹 내 상대 비교를 통해 학습합니다.

---