우선순위 경험 재생

Python으로 배우는 Deep Reinforcement Learning

Timothée Carayol

Principal Machine Learning Engineer, Komment

모든 경험이 동일하진 않다

경험 재생:
- 균일 샘플링은 중요한 기억을 놓칠 수 있음
우선순위 경험 재생:
- TD 오차 기반으로 각 경험에 우선순위를 부여
- 학습 잠재력이 큰 경험에 집중

학생들이 도서관에서 공부 중

우선순위 경험 재생 (PER)

for step = 1 to T do:
    # 가치 함수로 최적 행동 수행
    # 다음 상태와 보상 관측
    # 전이를 리플레이 버퍼에 추가

      # 가장 높은 우선순위(1) 부여

    # 과거 전이 배치 샘플

      # 우선순위 기반(2)

    # 배치에 대한 TD 오차 계산

    # 손실 계산 후 Q 네트워크 업데이트

      # 중요도 샘플링 가중치 사용 (4)

    # 샘플된 전이의 우선순위 갱신 (3)

# 중요도 샘플링을 시간에 따라 증가 (5)

(1) 새로운 전이는 최고 우선순위로 추가: $p_i = \max_k(p_k)$

(2) 전이 $i$를 다음 확률로 샘플링: $$P(i) = p_i^{\alpha} / \sum_k p_k^{\alpha}\ \ \ \ \ \ \ \ (0<\alpha<1)$$

(3) 샘플된 전이의 우선순위를 TD 오차로 갱신: $p_i = |\delta_i| + \varepsilon$

(4) 중요도 샘플링 가중치 사용: $$w_i = \left( \frac{1}{N} \cdot \frac{1}{P(i)} \right)^\beta\ \ \ \ \ \ \ \ (0<\beta<1)$$

(5) $\beta$를 1에 가깝게 점진적으로 증가

PER 구현

def __init__(self, capacity, alpha=0.6, beta=0.4, beta_increment=0.001, epsilon=0.001):
    # 메모리 버퍼 초기화
    self.memory = deque(maxlen=capacity)

    # 파라미터 저장 및 우선순위 초기화
    self.alpha, self.beta, self.beta_increment, self.epsilon = (alpha, beta, beta_increment, epsilon)
    self.priorities = deque(maxlen=capacity)


...

PER 구현

...

def push(self, state, action, reward, next_state, done):
    # 경험을 메모리 버퍼에 추가
    experience_tuple = (state, action, reward, next_state, done)
    self.memory.append(experience_tuple)

    # 새 전이의 우선순위를 최대값으로 설정
    max_priority = max(self.priorities) if self.memory else 1.0
    self.priorities.append(max_priority)


...

PER 구현

def sample(self, batch_size):
    priorities = np.array(self.priorities)
    # 샘플링 확률 계산
    probabilities = priorities**self.alpha / np.sum(priorities**self.alpha)

    # 무작위로 인덱스 선택
    indices = np.random.choice(len(self.memory), batch_size, p=probabilities)

    # 가중치 계산
    weights = (1 / (len(self.memory) * probabilities)) ** self.beta
    weights /= np.max(weights)
    states, actions, rewards, next_states, dones = zip(*[self.memory[idx] for idx in indices])
    weights = [weights[idx] for idx in indices]
    states, actions, rewards, next_states, dones = (zip(*[self.memory[idx] for idx in indices]))

    # 텐서 반환
    states = torch.tensor(states, dtype=torch.float32)
    ... # rewards, next_states, dones, weights 동일 처리
    actions = torch.tensor(actions, dtype=torch.long).unsqueeze(1)    
    return (states, actions, rewards, next_states, dones, indices, weights)

PER 구현

...

def update_priorities(self, indices, td_errors: torch.Tensor):
    # 샘플된 전이의 우선순위 갱신
    for idx, td_error in zip(indices, td_errors.abs()):
        self.priorities[idx] = abs(td_error.item()) + self.epsilon


def increase_beta(self):
    # beta를 1로 점진 증가
    self.beta = min(1.0, self.beta + self.beta_increment)

DQN 학습 루프에서의 PER

반복문 전 코드:

buffer = PrioritizedReplayBuffer(capacity)

각 에피소드 시작 시:
```
buffer.increase_beta()
```

매 스텝에서:

# 행동 선택 후
buffer.push(state, action, reward, 
            next_state, done)
...

# TD 오차 계산 전:
replay_buffer.sample(batch_size)
...

# TD 오차 계산 후
buffer.update_priorities(indices, td_errors)

loss = torch.sum(weights * (td_errors ** 2))

PER 실전: Cartpole

Cartpole 환경에서 100회 학습 실행:

우선순위 경험 재생 사용
균일 경험 재생 사용

PER가 균일 재생보다 학습이 더 빠르고 성능이 더 좋음

학습 곡선: PER가 더 빠르게 학습

100 에폭 후: Cartpole, 100 에폭 후 불안정

400 에폭 후: Cartpole, 400 에폭 후 안정적

PER 실전: Atari 환경

Atari 환경에서 PER로 성능이 크게 향상

막대그래프: 인간, DQN, DDQN, Dueling DDQN, Prioritized DDQN, Prioritized Dueling DQN 비교. 마지막이 Prioritized Experience Replay 도입으로 DDQN 성능 향상 표시.

¹ https://arxiv.org/abs/2303.11634

Ayo berlatih!

Python으로 배우는 Deep Reinforcement Learning