LabHub

블로그

[심층 강화학습] 04. Cross-Entropy 방법으로 CartPole 풀기

한국어English日本語

강화학습 방법론의 분류

강화학습 알고리즘은 다양한 기준으로 분류할 수 있습니다. 전체 그림을 이해하면 각 알고리즘의 위치를 파악하기 쉽습니다.

모델 기반 vs 모델 프리

가치 기반 vs 정책 기반

온폴리시 vs 오프폴리시

분류알고리즘 예시
가치 기반, 오프폴리시DQN, Double DQN
정책 기반, 온폴리시REINFORCE, PPO
액터-크리틱, 온폴리시A2C, A3C
액터-크리틱, 오프폴리시SAC, DDPG, TD3

Cross-Entropy 방법의 핵심 아이디어

Cross-Entropy 방법은 강화학습에서 가장 단순한 알고리즘 중 하나입니다. 핵심 아이디어는 다음과 같습니다.

  1. 현재 정책으로 여러 에피소드를 실행합니다
  2. 전체 에피소드 중 보상이 높은 상위 에피소드들만 선별합니다 (엘리트 에피소드)
  3. 엘리트 에피소드에서의 상태-행동 쌍을 이용해 정책을 업데이트합니다
  4. 위 과정을 반복합니다

이 방법은 진화 전략의 일종으로, 좋은 에피소드를 모방하는 방식으로 정책을 개선합니다.

알고리즘 의사 코드

1. 정책 네트워크 초기화
2. 반복:
   a. N개의 에피소드를 현재 정책으로 수집
   b.  에피소드의 총 보상을 계산
   c. 상위 p%의 엘리트 에피소드를 선별 (보상 기준 상위 임계값)
   d. 엘리트 에피소드의 (상태, 행동) 쌍으로 정책 네트워크 학습
      - 손실 함수: Cross-Entropy Loss (분류 문제로 취급)
   e. 평균 보상이 충분히 높으면 종료

CartPole 구현

정책 네트워크 정의

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

# 에피소드 데이터 구조
Episode = namedtuple('Episode', ['reward', 'steps'])
EpisodeStep = namedtuple('EpisodeStep', ['observation', 'action'])

class PolicyNetwork(nn.Module):
    """Cross-Entropy 방법을 위한 정책 네트워크"""
    def __init__(self, obs_size, hidden_size, n_actions):
        super().__init__()
        self.net = nn.Sequential(
            nn.Linear(obs_size, hidden_size),
            nn.ReLU(),
            nn.Linear(hidden_size, n_actions),
        )

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

에피소드 수집

def generate_batch(env, net, batch_size, device="cpu"):
    """현재 정책으로 에피소드 배치를 생성"""
    batch = []
    episode_reward = 0.0
    episode_steps = []
    obs, _ = env.reset()
    sm = nn.Softmax(dim=1)

    while True:
        obs_tensor = torch.tensor(np.array([obs]), dtype=torch.float32).to(device)
        action_probs = sm(net(obs_tensor))
        action_probs_np = action_probs.data.cpu().numpy()[0]

        # 확률에 따라 행동 선택
        action = np.random.choice(len(action_probs_np), p=action_probs_np)

        next_obs, reward, terminated, truncated, _ = env.step(action)
        episode_reward += reward
        episode_steps.append(EpisodeStep(observation=obs, action=action))

        if terminated or truncated:
            batch.append(Episode(reward=episode_reward, steps=episode_steps))
            episode_reward = 0.0
            episode_steps = []
            next_obs, _ = env.reset()

            if len(batch) >= batch_size:
                break

        obs = next_obs

    return batch

엘리트 에피소드 필터링

def filter_elite_episodes(batch, percentile):
    """상위 percentile%의 엘리트 에피소드 선별"""
    rewards = [episode.reward for episode in batch]
    reward_threshold = np.percentile(rewards, percentile)

    elite_observations = []
    elite_actions = []

    for episode in batch:
        if episode.reward >= reward_threshold:
            for step in episode.steps:
                elite_observations.append(step.observation)
                elite_actions.append(step.action)

    return (
        torch.tensor(np.array(elite_observations), dtype=torch.float32),
        torch.tensor(np.array(elite_actions), dtype=torch.long),
        reward_threshold,
        np.mean(rewards),
    )

학습 루프

def train_cross_entropy_cartpole():
    """Cross-Entropy 방법으로 CartPole 학습"""
    # 하이퍼파라미터
    HIDDEN_SIZE = 128
    BATCH_SIZE = 16
    PERCENTILE = 70
    LEARNING_RATE = 0.01
    GOAL_REWARD = 475  # CartPole-v1의 목표 보상

    env = gym.make("CartPole-v1")
    obs_size = env.observation_space.shape[0]
    n_actions = env.action_space.n

    net = PolicyNetwork(obs_size, HIDDEN_SIZE, n_actions)
    optimizer = optim.Adam(net.parameters(), lr=LEARNING_RATE)
    loss_fn = nn.CrossEntropyLoss()

    for iteration in range(100):
        # 1. 에피소드 배치 생성
        batch = generate_batch(env, net, BATCH_SIZE)

        # 2. 엘리트 에피소드 선별
        elite_obs, elite_actions, reward_threshold, mean_reward = \
            filter_elite_episodes(batch, PERCENTILE)

        # 3. 정책 네트워크 학습
        optimizer.zero_grad()
        action_logits = net(elite_obs)
        loss = loss_fn(action_logits, elite_actions)
        loss.backward()
        optimizer.step()

        print(
            f"반복 {iteration:3d}: "
            f"평균 보상={mean_reward:6.1f}, "
            f"임계값={reward_threshold:6.1f}, "
            f"손실={loss.item():.4f}"
        )

        if mean_reward >= GOAL_REWARD:
            print(f"목표 달성! 평균 보상: {mean_reward:.1f}")
            break

    env.close()
    return net

# 학습 실행
trained_net = train_cross_entropy_cartpole()

예상 출력

반복   0: 평균 보상=  21.3, 임계값=  23.0, 손실=0.6897
반복   1: 평균 보상=  25.1, 임계값=  27.0, 손실=0.6753
...
반복  20: 평균 보상= 152.4, 임계값= 185.0, 손실=0.5832
...
반복  40: 평균 보상= 421.8, 임계값= 500.0, 손실=0.5106
반복  45: 평균 보상= 487.2, 임계값= 500.0, 손실=0.4923
목표 달성! 평균 보상: 487.2

학습된 에이전트 평가

def evaluate_agent(env_name, net, n_episodes=100, render=False):
    """학습된 에이전트의 성능을 평가"""
    render_mode = "human" if render else None
    env = gym.make(env_name, render_mode=render_mode)
    rewards = []

    for _ in range(n_episodes):
        obs, _ = env.reset()
        total_reward = 0

        while True:
            obs_tensor = torch.tensor(np.array([obs]), dtype=torch.float32)
            with torch.no_grad():
                action_logits = net(obs_tensor)
            action = action_logits.argmax(dim=1).item()

            obs, reward, terminated, truncated, _ = env.step(action)
            total_reward += reward

            if terminated or truncated:
                break

        rewards.append(total_reward)

    env.close()
    print(f"평가 결과 ({n_episodes}회):")
    print(f"  평균 보상: {np.mean(rewards):.1f} +/- {np.std(rewards):.1f}")
    print(f"  최소/최대: {np.min(rewards):.1f} / {np.max(rewards):.1f}")

# evaluate_agent("CartPole-v1", trained_net)

FrozenLake 구현

FrozenLake는 4x4 얼음판 위에서 시작점에서 목표점까지 도달하는 문제입니다. 미끄러운 얼음판 때문에 의도한 방향으로 움직이지 않을 수 있습니다.

FrozenLake의 특성

원핫 인코딩으로 상태 표현

FrozenLake의 관찰은 정수 하나(현재 위치)이므로, 신경망에 입력하기 위해 원핫 인코딩을 사용합니다.

class DiscreteOneHotWrapper(gym.ObservationWrapper):
    """이산 관찰을 원핫 벡터로 변환하는 래퍼"""
    def __init__(self, env):
        super().__init__(env)
        n = env.observation_space.n
        self.observation_space = gym.spaces.Box(
            low=0.0, high=1.0, shape=(n,), dtype=np.float32
        )

    def observation(self, observation):
        result = np.zeros(self.observation_space.shape[0], dtype=np.float32)
        result[observation] = 1.0
        return result

FrozenLake에 Cross-Entropy 적용

FrozenLake에서는 몇 가지 수정이 필요합니다. 보상이 매우 희소(목표 도달 시에만 1)하기 때문입니다.

def train_cross_entropy_frozenlake():
    """Cross-Entropy 방법으로 FrozenLake 학습"""
    HIDDEN_SIZE = 128
    BATCH_SIZE = 100  # 더 많은 에피소드 필요
    PERCENTILE = 30   # 더 낮은 퍼센타일 (성공 에피소드가 적으므로)
    LEARNING_RATE = 0.001

    env = DiscreteOneHotWrapper(gym.make("FrozenLake-v1"))
    obs_size = env.observation_space.shape[0]
    n_actions = env.action_space.n

    net = PolicyNetwork(obs_size, HIDDEN_SIZE, n_actions)
    optimizer = optim.Adam(net.parameters(), lr=LEARNING_RATE)
    loss_fn = nn.CrossEntropyLoss()

    for iteration in range(200):
        batch = generate_batch(env, net, BATCH_SIZE)
        elite_obs, elite_actions, reward_threshold, mean_reward = \
            filter_elite_episodes(batch, PERCENTILE)

        # 엘리트 에피소드가 없는 경우 건너뛰기
        if len(elite_obs) == 0:
            continue

        optimizer.zero_grad()
        action_logits = net(elite_obs)
        loss = loss_fn(action_logits, elite_actions)
        loss.backward()
        optimizer.step()

        if iteration % 10 == 0:
            print(
                f"반복 {iteration:3d}: "
                f"평균 보상={mean_reward:.3f}, "
                f"임계값={reward_threshold:.3f}, "
                f"손실={loss.item():.4f}"
            )

        if mean_reward > 0.8:
            print(f"목표 달성! 평균 보상: {mean_reward:.3f}")
            break

    env.close()
    return net

# trained_net_fl = train_cross_entropy_frozenlake()

Cross-Entropy 방법의 이론적 배경

왜 "Cross-Entropy"인가

이 방법의 이름은 손실 함수에서 비롯됩니다. 엘리트 에피소드의 행동을 지도학습의 정답 레이블로 취급하고, 정책 네트워크의 출력과 비교할 때 Cross-Entropy 손실을 사용합니다.

Cross-Entropy 손실의 수학적 의미는 두 확률 분포 사이의 거리를 측정하는 것입니다.

H(p, q) = -sum( p(x) * log(q(x)) )

여기서 p는 엘리트 에피소드의 행동 분포(정답)이고, q는 정책 네트워크가 출력하는 행동 확률입니다.

방법의 장단점

장점

단점


하이퍼파라미터의 영향

def experiment_hyperparameters():
    """다양한 하이퍼파라미터 조합 실험"""
    configs = [
        {"batch_size": 8, "percentile": 70, "hidden": 64},
        {"batch_size": 16, "percentile": 70, "hidden": 128},
        {"batch_size": 32, "percentile": 80, "hidden": 128},
        {"batch_size": 16, "percentile": 50, "hidden": 256},
    ]

    for i, config in enumerate(configs):
        print(f"\n=== 설정 {i+1}: batch={config['batch_size']}, "
              f"percentile={config['percentile']}, hidden={config['hidden']} ===")

        env = gym.make("CartPole-v1")
        obs_size = env.observation_space.shape[0]
        n_actions = env.action_space.n

        net = PolicyNetwork(obs_size, config["hidden"], n_actions)
        optimizer = optim.Adam(net.parameters(), lr=0.01)
        loss_fn = nn.CrossEntropyLoss()

        for iteration in range(50):
            batch = generate_batch(env, net, config["batch_size"])
            elite_obs, elite_actions, threshold, mean_reward = \
                filter_elite_episodes(batch, config["percentile"])

            if len(elite_obs) == 0:
                continue

            optimizer.zero_grad()
            loss = loss_fn(net(elite_obs), elite_actions)
            loss.backward()
            optimizer.step()

            if mean_reward >= 475:
                print(f"  반복 {iteration}에서 목표 달성!")
                break

        env.close()

# experiment_hyperparameters()

정리

  1. 강화학습 분류: 모델 기반/프리, 가치/정책 기반, 온/오프 폴리시로 분류
  2. Cross-Entropy 방법: 상위 에피소드를 선별하여 정책을 지도학습 방식으로 업데이트
  3. CartPole: 약 40~50회 반복으로 최대 보상에 도달 가능
  4. FrozenLake: 희소 보상 환경에서는 더 많은 에피소드와 낮은 퍼센타일 필요
  5. 이론: Cross-Entropy 손실로 엘리트 에피소드의 행동 분포를 모방

Cross-Entropy 방법은 간단하지만, 더 복잡한 환경에서는 한계가 있습니다. 다음 글에서는 벨만 방정식과 가치 반복법을 통해 더 강력한 알고리즘의 기초를 쌓겠습니다.

댓글

아직 댓글이 없습니다.

로그인하면 댓글을 쓸 수 있습니다