DQN

• 2013년 발표 버전
  • Target Q-Network를 따로 두고 Main Q-Network를 학습시킨다
  • 몇 단계마다 한번씩 Target Q-Network를 수정해서 Main Q-Network와 일치시킨다
• 2015년 네이처 발표 버전
  $Q_m(s_t,a_t)=Q_m(s_t,a_t)+\eta*(R_{t+1}+\gamma\max_aQ_t(s_{t+1},a)-Q_m(s_t,a_t))$
  • $Q_m$ : Main Q-network
  • $Q_t$ : Target Q-network
  • 다음 상태 $s_{t+1}$에서 Q값이 최대가 되는 행동 $a$와 그때의 Q값을 Target Q-network에서 계산

DDQN

• 안정화된 수정식
  $a_m=\arg\max_aQ_m(s_{t+1},a)\\Q_m(s_t,a_t)=Q_m(s_t,a_t)+\eta*(R_{t+1}+\gamma Q_t(s_{t+1},a_m)-Q_m(s_t,a_t))$
• 다음 상태 $s_{t+1}$에서 Q값이 최대가 되는 행동 $a_m$은 Main Q-Network에서 구하고
• 그때의 Q값은 Target Q-Network에서 구하는 것

DDQN 구현

• 신경망을 2개 사용한다

# 신경망 구성
import torch.nn as nn
import torch.nn.functional as F

class Net(nn.Module):
    def __init__(self, n_in, n_mid, n_out):
        super(Net, self).__init__()
        self.fc1 = nn.Linear(n_in, n_mid)
        self.fc2 = nn.Linear(n_mid, n_mid)
        self.fc3 = nn.Linear(n_mid, n_out)

    def forward(self, x):
        h1 = F.relu(self.fc1(x))
        h2 = F.relu(self.fc2(h1))
        output = self.fc3(h2)
        return output

Brain 클래스

# 에이전트의 두뇌 역할을 하는 클래스, DDQN을 실제 수행
# Q함수를 딥러닝 신경망 형태로 정의

import numpy as np
import random
import torch
from torch import nn
from torch import optim
import torch.nn.functional as F

BATCH_SIZE = 32
CAPACITY = 10000

class Brain:
    def __init__(self, num_states, num_actions):
        self.num_actions = num_actions # 행동 가짓수(2)를 구함

        # transition을 기억하기 위한 메모리 객체 생성
        self.memory = ReplayMemory(CAPACITY)

        # 신경망 구성
        n_in, n_mid, n_out = num_states, 32, num_actions
        self.main_q_network = Net(n_in, n_mid, n_out) # Net 클래스 사용
        self.target_q_network = Net(n_in, n_mid, n_out)
        print(self.main_q_network) # 신경망 구조 출력

        # 최적화 기법 선택
        self.optimizer = optim.Adam(self.main_q_network.parameters(), lr=0.0001)

    def replay(self):
        '''Experience Replay로 신경망 결합 가중치 학습'''

        # 1. 저장된 transition 수 확인
        if len(self.memory) < BATCH_SIZE :
            return
        
        # 2. 미니배치 생성
        self.batch, self.state_batch, self.action_batch, self.reward_batch, self.non_final_next_states \
        = self.make_minibatch()

        # 3. 정답신호로 사용할 Q(s_t, a_t)를 계산
        self.expected_state_action_values = self.get_expected_state_action_values()

        # 4. 결합 가중치 수정
        self.update_main_q_network()

    def decide_action(self, state, episode):
        '''현재 상태로부터 행동을 결정함'''
        # Epsilon-greedy 알고리즘에서 서서히 최적행동의 비중을 늘림
        epsilon = 0.5 * (1 / (episode + 1))

        if epsilon <= np.random.uniform(0, 1):
            self.main_q_network.eval() # 신경망을 추론 모드로 전환
            with torch.no_grad():
                action = self.main_q_network(state).max(1)[1].view(1, 1)
            # 신경망 출력의 최댓값에 대한 인덱스 = max(1)[1]
            # .view(1, 1)은 [torch.LongTensor of size 1]을 size 1*1로 변환

        else :
            # 행동을 무작위로 반환(0 or 1)
            action = torch.LongTensor(
                [[random.randrange(self.num_actions)]]
            )
            # action은 [torch.LongTensor of size 1*1]

        return action
    
    def make_minibatch(self):
        '''2. 미니배치 생성'''
        # 2.1 메모리 객체에서 미니배치를 추출
        transitions = self.memory.sample(BATCH_SIZE)

        # 2.2 각 변수를 미니배치에 맞는 형태로 변형
        # transitions는 각 단계별로 (state, action, state_next, reward) 형태로 BATCH_SIZE 만큼 저장
        batch = Transition(*zip(*transitions))
        
        # 2.3 각 변수의 요소를 미니배치에 맞게 변형하고 Variable로 만듬
        state_batch = torch.cat(batch.state)
        action_batch = torch.cat(batch.action)
        reward_batch = torch.cat(batch.reward)
        non_final_next_states = torch.cat([s for s in batch.next_state
                                           if s is not None])
        
        return batch, state_batch, action_batch, reward_batch, non_final_next_states
    
    def get_expected_state_action_values(self):
        '''정답신호로 사용할 Q(s_t, a_t)를 계산'''

        # 3.1 신경망을 추론 모드로 전환
        self.main_q_network.eval()
        self.target_q_network.eval()

        # 3.2 신경망으로 Q(s_t, a_t)를 계산
        # self.model(state_batch)는 왼쪽, 오른쪽에 대한 Q값을 출력
        # [torch.FloatTensor of size BATCH_SIZE * 2] 형태
        # 여기서부터 실행한 행동 a_t 에 대한 Q값을 계산하므로 action_batch에서 취한 행동
        # a_t가 왼쪽이냐 오른쪽이냐에 대한 인덱스를 구하고, 이에 대한 Q값을 gather 메서드로 모아온다
        self.state_action_values = self.main_q_network(
            self.state_batch
        ).gather(1, self.action_batch)

        # 3.3 max{Q(s_t+1, a)}값을 계산. 다음 상태가 존재하는지 주의
        # cartpole이 done이 아니고, next_state가 존재하는지 확인
        non_final_mask = torch.ByteTensor(tuple(map(lambda s: s is not None,
                                                    self.batch.next_state)))
        
        # 먼저 전체를 0으로 초기화
        next_state_values = torch.zeros(BATCH_SIZE)
        a_m = torch.zeros(BATCH_SIZE).type(torch.LongTensor)

        # 다음 상태에서 Q값이 최대가 되는 행동 a_m을 Main Q-Network로 계산
        # 마지막에 붙은 [1]로 행동에 해당하는 인덱스를 구함
        a_m[non_final_mask] = self.main_q_network(
            self.non_final_next_states
        ).detach().max(1)[1]

        # 다음 상태가 있는 것만을 걸러내고, size 32를 32*1로 변환
        a_m_non_final_next_states = a_m[non_final_mask].view(-1, 1)

        # 다음 상태가 있는 인덱스에 대해 행동 a_m의 Q값을 target Q-network로 계산
        # detach() 메서드로 값을 꺼내옴
        # squeeze() 메서드로 size[minibatch*1]을 [minibatch]로 변환
        next_state_values[non_final_mask] = self.target_q_network(
            self.non_final_next_states
        ).gather(1, a_m_non_final_next_states).detach().squeeze()

        # 3.4 정답신호로 사용할 Q(s_t, a_t) 값을 Q러닝 식으로 계산
        expected_state_action_values = self.reward_batch + GAMMA * next_state_values

        return expected_state_action_values
    
    def update_main_q_network(self):
        '''4. 결합 가중치 수정'''

        # 4.1 신경망을 학습 모드로 전환
        self.main_q_network.train()

        # 4.2 손실함수를 계산(smooth_l1_loss는 Huber 함수)
        # expected_state_action_values은
        # size가 [minibatch]이므로 unsqueeze해서 [minibatch*1]로 만듬
        loss = F.smooth_l1_loss(self.state_action_values,
                                self.expected_state_action_values.unsqueeze(1))
        
        # 4.3 결합 가중치를 수정
        self.optimizer.zero_grad() # 경사를 초기화
        loss.backward() # 역전파 계산
        self.optimizer.step() # 결합 가중치 수정

    def update_target_q_network(self): # DDQN에서 추가됨
        '''Target Q-network을 Main Q-Network와 맞춤'''
        self.target_q_network.load_state_dict(self.main_q_network.state_dict())

• 최적화 기법 설정시 self.main_q_network.parameters()를 통해 인자로 주어 학습 대상이 Main Q-Network가 되도록 한다
• get_expected_state_action_values 메서드를 DDQN용으로 수정한다
• update_target_q_network를 정의해 Target Q-Network 결합 가중치를 Main Q-Network와 맞춘다

Agent 클래스

• 에피소드 종료시 update_target_q_function 호출

# CartPole 태스크의 에이전트 클래스 : 봉 달린 수레

class Agent :
    def __init__(self, num_states, num_actions):
        '''태스크의 상태 및 행동의 가짓수를 설정'''
        self.brain = Brain(num_states, num_actions) # 에이전트의 행동을 결정할 두뇌

    def update_q_function(self):
        '''Q함수를 수정'''
        self.brain.replay()

    def get_action(self, state, episode):
        '''행동을 결정'''
        action = self.brain.decide_action(state, episode)
        return action
    
    def memorize(self, state, action, state_next, reward):
        '''memory 객체에 state, action, state_next, reward 저장'''
        self.brain.memory.push(state, action, state_next, reward)

    def update_target_q_function(self):
        '''Target Q-Network를 Main Q-Network와 맞춤'''
        self.brain.update_target_q_network()

Environment 클래스

# CartPole을 실행하는 환경 역할을 하는 클래스
class Environment:
    def __init__(self):
        self.env = gym.make(ENV, render_mode = 'rgb_array')
        num_states = self.env.observation_space.shape[0] # 태스크의 상태 변수 수(4)를 받아옴
        num_actions = self.env.action_space.n # 태스크의 행동 가짓수(2)를 받아옴
        self.agent = Agent(num_states, num_actions) # 에이전트 역할을 할 객체를 생성

    def run(self):
        '''실행'''
        episode_10_list = np.zeros(10) # 최근 10에피소드 동안 버틴 단계 수 저장
        complete_episodes = 0 # 현재까지 195단계를 버틴 에피소드 수
        episode_final = False # 마지막 에피소드 여부
        frames = [] # 애니메이션 저장용 프레임

        for episode in range(NUM_EPISODES): # 최대 에피소드 수만큼 반복
            observation = self.env.reset() # 환경 초기화

            state = observation # 관측을 변환 없이 그대로 상태 s로 사용
            state = torch.from_numpy(state[0]).type(
                torch.FloatTensor # Numpy 변수를 파이토치 텐서로 변환
            )
            state = torch.unsqueeze(state, 0) # size 4를 size 1*4로 변환

            for step in range(MAX_STEPS):
                if episode_final is True : # 마지막에피소드에서는 frame에 저장
                    frames.append(self.env.render())
                action = self.agent.get_action(state, episode) # 다음 행동을 결정

                # 행동 a_t를 실행해 다음 상태 s_{t+1}과 done 플래그 값을 결정
                # action에 .item()을 호출해 행동 내용을 구함
                observation_next, _, done, _, _ = self.env.step(
                    action.item() # reward와 info는 사용하지 않으므로 _ 처리
                )

                # 보상을 부여하고 episode의 종료 판정 및 state_next를 설정
                if done : # 단계 수가 200을 넘었거나 봉이 일정 각도 이상 기울면 Done
                    state_next = None # 다음 상태가 없으므로 None

                    # 최근 10 에피소드에서 버틴 단계 수를 리스트에 저장
                    episode_10_list = np.hstack(
                        (episode_10_list[1:], step+1)
                    )

                    if step < 195:
                        reward = torch.FloatTensor(
                            [-1.0] # 도중에 봉이 쓰러졌다면 페널티로 보상 -1을 부여
                        )
                        complete_episodes = 0 # 연속 성공 에피소드 기록을 초기화
                    else :
                        reward = torch.FloatTensor([1.0]) # 봉이 살아있으면 보상 1
                        complete_episodes += 1 # 연속 성공 기록 갱신

                else :
                    reward = torch.FloatTensor([0.0]) # 그 외의 경우는 보상 0을 부여
                    state_next = observation_next # 관측 결과를 그대로 상태로 사용
                    state_next = torch.from_numpy(state_next).type(
                        torch.FloatTensor # numpy 변수를 파이토치 텐서로 변환
                    )
                    state_next = torch.unsqueeze(state_next, 0) # size 4를 size 1*4로 변환

                # 메모리에 경험을 저장
                self.agent.memorize(state, action, state_next, reward)

                # Experience Replay로 Q함수를 수정
                self.agent.update_q_function()

                # 관측 결과를 업데이트
                state = state_next

                # 에피소드 종료 처리
                if done:
                    print('%d Episode : Finished after %d steps : 최근 10 에피소드의 평균 단계 수 = %.1lf' % (episode, step + 1, episode_10_list.mean()))
                    
                    # DDQN 추가 2에피소드마다 한 번씩 Target Q-Network를 Main Q-Network와 맞춰줌
                    if (episode % 2 == 0) :
                        self.agent.update_target_q_function()
                    break

                if episode_final is True:
                    # 애니메이션 생성 및 저장
                    display_frames_as_gif(frames)
                    break

                # 10 에피소드 연속으로 195단계를 버티면 태스크 성공
                if complete_episodes >= 10:
                    print("10 연속 에피소드 성공")
                    episode_final = True # 다음 에피소드에서 애니메이션 생성

DQN 보다 훨씬 빨리 학습이 끝났다