개요
이전 글에서는 DQN(Deep Q-Network)의 원리에 대해 자세히 알아보았는데, 이젠 이를 실제로 활용하여 학습을 진행한다
테트리스 게임을 환경으로 선택하여 학습을 진행했으며, 이를 통해 깨달은 내용들에 대해 끄적여본다
- Kaggle
Gymnasium- 학습 코드
- 마무리(하소연?)
Kaggle
무료로 P100을 최대 연속 9시간 사용할 수 있는 서비스가 있다?
이전까지 Colab을 사용하면서, 브라우저로 작업을 돌면서 항상 세션이 종료되지 않을까 걱정했었는데 찾아보니 Kaggle을 이용하면 이런 불편함을 해소할 수 있었다
Kaggle에서 Job은 코드 셀 실행 시 자동으로 시작되며, 사용자 개입 없이 백그라운드에서 작업을 처리한다
즉, 기존에 노트북파일을 직접 연결해서 수행하던 걸 그냥 해당 파일만 전달하면 알아서 실행하게 하는거라 이해하면 된다
덕분에 그냥 노트북파일이 오류없이 돌아가는지 확인 후, 우측 상단에 Save Version만 눌러서 백그라운드로 실행시키면 된다
아래 사진과 같이 동시에 최대 2개의 GPU Session을 사용할 수 있다
Gymnasium
Gymnasium은 강화학습 연구와 교육에 자주 사용되는 오픈 소스 라이브러리로, 강화학습을 위한 환경을 제공하는 것이 주된 목적이다
Gymnasium의 환경은 모두 통일된 API로 제공되어, 강화학습 알고리즘을 테스트할 때 환경을 쉽게 교체하거나 확장할 수 있다
원래는 나의 게임 환경을 해당 프레임워크를 통해 배포하는 게 목표로 했었는데, 여러가지 공부를 하며 생각보다 오래 걸렸어서 이 인터페이스에 맞게 환경을 구현하지는 못했다…
학습 코드
위의 깃헙레포는 게임환경과 상호작용하며 학습하는 프로젝트 코드를 작성하는데 참고한 코드다
train.py를 보면, 상태를 제한하여 Deep Q-learning 만으로 학습을 수행한다
전체 학습 코드를 간단하게 이해하기 앞서, 우선 환경과 학습 알고리즘 부분으로 나눠서 정리한다
환경
게임 환경 객체는 특정 규칙에 따라 명확하게 정의되어 있으며, 몇 가지 공통적인 메서드만 구현하면 된다
특히 다음과 같은 메서드들을 공통적으로 포함해야 한다
- reset() - 환경을 초기화
- step(action) - 주어진 action을 환경에서 수행한 뒤, 환경의 상태와 관련된 여러 정보를 반환
- render() - 현재 환경을 시각적으로 렌더링
여기서 action은 환경에서 수행할 수 있는 특정 동작을 의미한다
Gymnasium 기반의 게임 환경과 사용자 입력을 통해 이루어지는 상호작용을 간단히 구현한 예시는 다음과 같다
import gymnasium as gym
agent = SomeAgent()
env = gym.make("some_env")
observation, _ = env.reset(seed=42)
terminated = False
while not terminated:
action = agent.get_action(observation)
observation, reward, terminated, _, _ = env.step(action)
print("Game Over!")- agent - 환경에 대한 관측(observation)을 바탕으로 action을 얻을 수 있는 객체
- env - Gymnasium을 기반으로 만들어진 환경 객체
즉, observation을 바탕으로 action을 선택할 수 있는 SomeAgent를 구현하면, Gymnasium을 이용한 env 객체와 위처럼 상호작용한다
우선 환경을 초기화하고, 이 초기 관측을 agent로 넘겨 action을 얻으면 이를 바탕으로 다음 상태를 얻는 방식이다
학습 알고리즘
위에서 언급한 레포의 train.py에서 Target Network 없이 학습을 하는 과정은 다음과 같다
- 환경 초기화
- 학습을 시작하기 전, 환경을 초기 상태로 설정
- 환경과 상호작용 반복
- 현재 상태를 관측하여 행동(action)을 선택
- 경험버퍼에 (현재 상태, 보상, 다음 상태, 종료 여부) 추가
- 종료라면 학습 시작
- 학습 시작
- 경험 버퍼에서 데이터 샘플링(미니배치)
- 현재 상태와 다음 상태의 Q-value 계산
- 타겟 Q-value 계산(벨만 최적 방정식) ← reward + gamma * max(next_state_q_value)
- 예측 Q-value와 타겟 Q-value의 차이를 최소화하도록 모델 업데이트
- 종료 조건 확인
- 설정한 에포크(epoch)에 도달할 때까지 위 단계를 반복
아래 코드에서 경험 버퍼는 일정 수 이상의 데이터가 쌓인 뒤 샘플링을 통해 학습을 시작한다
이로 인해 에피소드가 여러 번 진행된 후에야 첫 에포크가 시작될 수 있어서 에피소드와 에포크는 항상 1:1로 대응되지 않는다
에포크는 배치 단위로 진행되며, 배치 안의 데이터를 학습하는 과정에서 여러 스텝이 발생한다
즉 이 키워드들을 정리해보면 다음과 같다
- Episode - 환경 내에서 하나의 시뮬레이션 또는 게임 플레이.
- Batch - 학습을 위해 경험 버퍼에서 샘플링한 데이터 묶음.
- Epoch - 배치를 기반으로 모델을 한 번 업데이트하는 과정.
- Step - 배치 단위로 모델을 한 번 업데이트하는 과정.
def train(opt):
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
# Model, Optimizer, LR scheduler
model = DQN().to(device)
optimizer = torch.optim.Adam(model.parameters(), lr=opt.lr)
scheduler = CosineAnnealingLR(
optimizer, T_max=opt.replay_memory_size//opt.batch_size+1, eta_min=1e-5)
# Environment
env = Tetris(width=opt.width, height=opt.height, block_size=opt.block_size)
# Replay memory
replay_memory = deque(maxlen=opt.replay_memory_size)
minimum_replay_memory_size = opt.replay_memory_size // 10
max_cleared_lines = 0
# Epoch = 학습이 진행된 Episode
epoch = 0
while epoch < opt.num_epochs:
# epsilon with linear decay
epsilon = opt.final_epsilon + (max(opt.num_decay_epochs - epoch, 0) * (
opt.initial_epsilon - opt.final_epsilon) / opt.num_decay_epochs)
# 환경 초기화
state = env.reset().to(device)
done = False
while not done:
# 현재 상태에서 가능한 모든 행동들과 다음 상태들을 가져옴
next_steps = env.get_next_states()
next_actions, next_states = zip(*next_steps.items())
# 다음 상태들에 대한 q-value를 계산하고, 이를 바탕으로 현재 상태에서 최적의 행동을 선택(또는 랜덤하게 행동을 선택)
next_states = torch.stack(next_states).to(device)
with torch.no_grad():
predictions = model(next_states)[:, 0]
# 행동 선택은 epsilon-greedy policy을 따름
index = epsilon_greedy_policy(
predictions, len(next_steps), epsilon)
next_state = next_states[index, :]
action = next_actions[index]
# 환경과 상호작용
reward, done = env.step(action, render=False)
# Replay memory에 저장
replay_memory.append([state, reward, next_state, done])
# 상태 업데이트
if not done:
state = next_state.to(device)
# Replay memory가 충분히 쌓여야 학습 시작
if len(replay_memory) < max(opt.batch_size, minimum_replay_memory_size):
continue
# 학습 시작
epoch += 1
# Batch sampling
batch = sample(replay_memory, opt.batch_size)
state_batch, reward_batch, next_state_batch, done_batch = zip(
*batch)
state_batch = torch.stack(
tuple(state for state in state_batch)).to(device)
reward_batch = torch.from_numpy(
np.array(reward_batch, dtype=np.int32)[:, None]).to(device)
next_state_batch = torch.stack(
tuple(state for state in next_state_batch)).to(device)
# 현재 상태에서의 행동에 대한 q-value
q_values = model(state_batch)
# 다음 상태에서의 q-value를 계산하고, target q-value를 계산
with torch.no_grad():
next_q_values = model(next_state_batch)
done_batch = torch.tensor(done_batch, dtype=torch.float32)[
:, None].to(device)
target_q_values = reward_batch + opt.gamma * \\
next_q_values * (1 - done_batch)
# 최적화
optimizer.zero_grad()
F.mse_loss(q_values, target_q_values).backward()
optimizer.step()
scheduler.step()마무리
이번 프로젝트에서는 DQN 기반 강화학습을 실제로 구현하면서 다양한 시행착오를 겪고, 이를 통해 나름의 인사이트를 얻을 수 있었다
처음엔 타겟 네트워크를 도입하여 안정적인 학습을 목표로 했지만, 의도대로 학습이 이뤄지지 않아 단일 네트워크로 학습을 진행했으며, 오히려 더 나은 결과를 얻었다
그러나 이러한 결과의 근본적인 이유를 명확히 파악하지 못한 점은 다음 연구의 과제로 남는다…
강화학습에서 중요한 지표인 학습 곡선이라 불리는 에피소드 길이 (Episode Length), 보상(Total Reward)과 손실(Loss)을 지속적으로 모니터링하는 것은 학습 상태를 파악하고 문제를 진단하는데 필수적이다
지도학습과 달리 강화학습에서는 정답 레이블이 아닌 환경과의 상호작용을 통해 보상을 최대화하려는 점에서 학습이 올바르게 진행되고 있는지를 단순히 손실값만으로 판단하기 어렵다
따라서 단순히 손실 값만 관찰하기보단 이들을 함께 관찰하는 것이 학습 진행 상황을 평가하는 더 신뢰할 수 있는 방법이라는 것을 한번 더 기억에 각인시켰다
또한 학습을 하면서 손실과 보상이 주기적으로 큰 폭으로 변동하며 학습이 불안정한 현상을 관찰했다
이 문제를 해결하기 위해 여러 방안을 모색했는데, 학습률 스케줄링이 나름의 개선을 이끌었다
이러한 하이퍼파라미터 조정의 중요성을 실감하게 한 유익한 경험이었는데, 그래서 그런지 향후 이런 프로젝트를 또 하게 된다면 자동화된 하이퍼파라미터 최적화 도구를 활용해 실험 설계를 체계화하고, 최적의 조합을 더 효율적으로 탐색할 것이다
모델 테스트
참고로 해당 게임 환경에서 행동은 현재 블록의 (x좌표, n회전횟수)으로, 만약 보드의 중앙 상단에 생성되는 블록이 행동을 취하면 x좌표에 n번 회전된 상태로 hard drop된다
