DQN(Deep Q-Networks) (1)

DQN(Deep Q-Networks)

논문 Playing Atari with Deep Reinforcement Learning 으로 소개되었습니다.

 

Q-learning 을 기반으로 하며 연속된 상태 공간(continuous state space)을 가지는 환경의 문제를 해결하기 위한 아이디어입니다. $Q$ 테이블에 모든 상태의 행동 가치를 저장하는 대신 신경망을 이용해서 $Q_*$ 에 근사하는 것을 목적합니다. 

 

 

또한 안정적인 학습과 성능 향상을 위해 fixed Q-targets 과 experience replay 기법을 이용합니다.

 

 

손실 함수

벨만 최적 방정식을 기반으로 하여 각 상태에서 취할 수 있는 최적의 전략은 다음 상태 $s'$ 와 다음 행동 $a'$ 에 대한 $r + \gamma Q^*(s',a')$ 의 기댓값을 최대화하는 행동을 선택하는 것입니다.

 

$Q^*(s,a) = \mathbb{E}_{s'\thicksim \varepsilon} \left[ r + \gamma {\underset {a'} {\text{max}} } Q^*(s',a') | s,a \right]$

 

 

value iteration 과 같은 과정을 무한히 반복하여 계산하면 $Q_i$ 는 최적 $Q^*$ 로 수렴하는 것입니다.

 

$Q_{i+1}(s,a) = \mathbb{E}[ r + \gamma \text{max}_{a'} Q_i(s',a'|s,a ]$

 

$Q_i \rightarrow Q^* \text{ as } i \rightarrow \infty$ 

 

하지만 이는 선형(linear) 문제이며 신경망과 같은 비선형 function approximator 를 사용한다면 다음과 같은 손실 함수를 사용해야 합니다.

 

$L_i(\theta_i) = \mathbb{E}_{s,a \thicksim p(\cdot)} \left[ (y_i - Q(s,a;\theta_i))^2 \right]$

 

타겟과 예측의 오차를 계산하는 것으로 이를 최소화하는 방향으로 학습하는 것입니다.

 

 

타겟 $y_i$ 는 이전 파라미터 $\theta_{i-1}$ 에서 계산합니다.

 

$y_i = \mathbb{E}_{s,a \thicksim p(\cdot)} \left[ ( r + \gamma {\underset {a'} {\text{max}}}Q(s',a'; \theta_{i-1})|s,a \right]$ 

 

 

손실 함수를 미분하면 다음과 같은 그래디언트 식을 얻을 수 있습니다.

 

$\nabla_{\theta_i} L_i(\theta_i) = \mathbb{E}_{s,a \thicksim p(\cdot)} \left[ ( r + \gamma {\underset {a'} {\text{max}}}Q(s',a'; \theta_{i-1} - Q(s,a;\theta_i) ) \nabla_{\theta_i} Q(s,a;\theta_i) \right]$

 

 

타겟 네트워크(Target Network)

위 식 $L_i(\theta_i)$ 의 문제는 타겟 $y_i$ 가 고정되어 있지 않다는 것인데 같은 신경망으로부터 타겟과 예측을 계산한다면 매번 다른 값이 나올 수 있기 때문에 안정적이지 못합니다. 따라서 동일한 구조의 타겟 신경망을 두어 안정적인 학습을 가능하게 하는 것입니다.

 

일반 신경망은 매 타임스텝마다 업데이트되며 타겟 신경망은 일정 주기마다 일반 신경망의 파라미터를 타겟 신경망의 파라미터로 한 번에 복사하게 됩니다. 즉 타겟 신경망을 일정 주기만큼 고정하는 것이라고 할 수 있습니다.

 

이를 통해 손실 함수를 다시 정의하면 다음과 같습니다.

 

$L_i(\theta_i) = \mathbb{E}_{s,a \thicksim p(\cdot)} \left[ \left( r + \gamma {\underset {a'} {\text{max}}} Q(s',a'; \color {red} {\theta_i^-}) - Q(s,a;\theta_i) \right)^2 \right]$

 

 

경험 리플레이(Experience replay)

이전 경험들을 반복하는 기법을 경험 리플레이(experience replay)라고 합니다. 

 

에피소드를 진행하면서 얻은 샘플 데이터를 리플레이 메모리(replay memory)에 저장하고 일정 개수만큼 무작위로 추출하여 미니배치(mini-batch)를 구성하고 업데이트를 수행합니다. 이에 따라 데이터간에 상관관계(correlation)를 줄여준다는 이점이 있습니다.

 

이러한 경험 리플레이 기법은 DQN, DDPG 과 같은 off-policy 알고리즘에만 사용할 수 있습니다.

 

또한 메모리안의 샘플 데이터는 큐(queue)를 이용한 FIFO(First In First Out) 구조로 저장하여 오래된 데이터가 먼저 폐기됩니다.

 

 

 

DQN 에 사용된 기법들의 사용유무에 대한 실험 결과는 다음과 같습니다.

 

 

 

알고리즘

 

 

 

구현

알고리즘 테스트 환경은 OpenAI Gym 의 CartPole-v0 입니다.

 

 

CartPole 은 좌우로 카트를 이동하여 막대기가 쓰러지지 않도록 오래 유지하는 문제입니다. 

 

환경으로부터 받을 수 있는 정보를 출력합니다.

import gym

env = gym.make('CartPole-v0')

print('obs:', env.observation_space.shape[0])
print('act:', env.action_space.n)

 

상태 정보는 4차원의 벡터로 연속적인 공간을 갖으며 에이전트가 취할 수 있는 행동은 스칼라로 2개의 범위를 갖는 이산적인 공간입니다.

obs: 4
act: 2

 

신경망을 정의합니다.

class Network:
    def __init__(self, scope):
        with tf.variable_scope(scope):
            with tf.name_scope('inputs'):
                self.s = tf.placeholder(tf.float32, [None, N_S])
                self.q_target = tf.placeholder(tf.float32, [None, N_A])
            
            with tf.variable_scope('layer'):
                self.fc = tf.layers.dense(self.s, 16, tf.nn.relu)
            
            with tf.variable_scope('out'):
                self.q = tf.layers.dense(self.fc, N_A)
                
            with tf.name_scope('loss'):
                self.loss = tf.losses.mean_squared_error(self.q_target, self.q)
                
            with tf.name_scope('optimizer'):
                self.train_op = tf.train.AdamOptimizer(LR).minimize(self.loss)
        
            self.params = tf.get_collection(tf.GraphKeys.TRAINABLE_VARIABLES, scope)

 

상태와 $Q$ 타겟을 입력으로 받습니다.

with tf.name_scope('input'):
  self.s = tf.placeholder(tf.float32, [None, N_S])
  self.q_target = tf.placeholder(tf.float32, [None, N_A])

 

출력층으로 입력으로 받은 상태에 대한 $Q$ 를 출력합니다.

with tf.variable_scope('output'):
    self.q = tf.layers.dense(self.fc, N_A)

 

손실 함수를 정의합니다.

with tf.name_scope('loss'):
    self.loss = tf.losses.mean_squared_error(self.q_target, self.q)

 

타겟과 예측 사이의 평균 제곱 오차(mean squared error, MSE)로 $Q$ 타겟과 추정 $Q$ 간의 계산입니다.

 

손실값을 최소화하는 방향으로 학습하도록 설정합니다.

with tf.name_scope('optimizer'):
    self.train_op = tf.train.AdamOptimizer(LR).minimize(self.loss)

 

$\epsilon$-greedy 정책을 정의합니다.

def decay_epsilon(self):
    if self.epsilon > 0.1:
        self.epsilon -= self.decay_rate
    return self.epsilon

def get_action(self, s):
    if np.random.rand() < self.epsilon:
        action = np.random.randint(N_A)
    else:
        q = self.sess.run(self.main.q, {self.main.s: s[np.newaxis, :]})
        action = np.argmax(q)
    return action

...

for i in range(MAX_EPI + 1):
    agent.decay_epsilon()
    
    state = env.reset()
    done = False
    while not done:
        action = agent.get_action(state)
        ...

 

감쇠 비율은 전체 에피소드 수에 대한 현재 에피소드의 비율입니다. 따라서 매 에피소드마다 엡실론 $\epsilon$ 을 감쇠하여 랜덤한 행동과 최선의 행동 사이의 확률을 조정합니다. $\epsilon$ 의 확률로 랜덤한 행동을 선택하고 $1 - \epsilon$ 의 확률로 최선의 행동을 선택하게 됩니다.

 

 

에이전트와 환경의 상호작용입니다.

agent = DQN(epsilon=1, decay_rate=1/MAX_EPI)

for i in range(MAX_EPI + 1):
    epi_steps = 0

    agent.decay_epsilon()
    
    state = env.reset()
    done = False
    while not done:
        action = agent.get_action(state)
        
        next_state, reward, done, _ = env.step(action)
        
        if done:
        	reward = -1
        
        agent.replay_memory.append((state, action, reward, done, next_state))
        
        if len(agent.replay_memory) >= BATCH_SIZE:
            agent.train()
        
        if (agent.train_cnt + 1) % UPDATE == 0:
            agent.update_target()
            print('** update **')
        
        state = next_state
        epi_steps += 1

 

매 타임스텝마다 환경으로 받은 정보 $s, a, r, d, s'$ 를 리플레이 메모리에 저장합니다.

for i in range(MAX_EPI + 1):
    state = env.reset()
    done = False
    while not done:
        action = agent.get_action(state)   
        next_state, reward, done, _ = env.step(action)
        agent.replay_memory.append((state, action, reward, done, next_state))

 

메모리의 크기가 일정 크기 이상이면 매 타임스텝마다 일반 신경망 업데이트를 수행합니다.

if len(agent.replay_memory) > agent.replay_memory * 0.1:
    agent.train()

 

일반 신경망을 업데이트합니다.

def train(self):
    minibatch = random.sample(self.replay_memory, BATCH_SIZE)
    minibatch = np.array(minibatch)

    target_q = self.sess.run(self.target.q, {self.target.s: minibatch[:, 4].tolist()})
    update_value = minibatch[:, 2] + GAMMA * np.max(target_q, axis=1) * ~minibatch[:, 3].astype(np.bool)

    q_target = self.sess.run(self.main.q, {self.main.s: minibatch[:, 0].tolist()})
    q_target[range(len(minibatch)), minibatch[:, 1].tolist()] = update_value

    _, loss = self.sess.run([self.main.train_op, self.main.loss], {self.main.s: minibatch[:, 0].tolist(),
                                                                   self.main.q_target: q_target})

    self.train_cnt += 1

    return loss

 

리플레이 메모리에 저장된 샘플을 배치 사이즈만큼 무작위로 추출합니다.

minibatch = random.sample(self.replay_memory, BATCH_SIZE)

 

계산의 편의를 위해 리스트를 numpy.ndarray 타입으로 변환합니다.

minibatch = np.array(minibatch)

 

다음 상태의 시퀀스를 입력으로 타겟 신경망의 $Q(s';\theta^-)$ 를 구합니다. 

target_q = self.sess.run(self.target.q, {self.target.s: minibatch[:, 4].tolist()})

 

 

타겟인 $r + \gamma {\underset{a'} {\text{maxQ}}}(s', a'; \theta^-)$ 를 계산합니다.

update_value = minibatch[:, 2] + GAMMA * np.max(target_q, axis=1) * ~minibatch[:, 3].astype(np.bool)

 

여기서 minibatch[:, 2] 는 보상의 시퀀스이며 np.max(target_q, axis=1) 은 최대 행동 가치 $\underset{a'} {\text{max}} Q(s', a'; \theta^-)$ 시퀀스입니다. 또한 minibatch[:, 3] 은 에피소드 종료 여부를 나타내는 True/False 의 bool 자료형을 가지는 시퀀스이며 앞에 ~ 연산자는 numpy.ndarray 타입의 자료형에서 반전을 나타냅니다. 따라서 종료되지 않은 에피소드에 한해서만 $\gamma \underset{a'} {\text{max}} Q(s', a'; \theta^-)$ 을 계산하라는 것입니다.

 

 

 

일반 신경망의 $Q(s;\theta)$ 에 취한 행동에 대해 계산된 가치를 넣어 타겟의 형태를 완성합니다.

q_target = self.sess.run(self.main.q, {self.main.s: minibatch[:, 0].tolist()})
q_target[range(len(minibatch)), minibatch[:, 1].tolist()] = update_value

 

일반 신경망 업데이트를 수행합니다.

_, loss = self.sess.run([self.main.train_op, self.main.loss], {self.main.s: minibatch[:, 0].tolist(), self.main.q_target: q_target}

 

일반 신경망 업데이트 횟수가 타겟 네트워크 업데이트 주기에 걸리면 그 동안 학습된 일반 신경망의 파라미터 $\theta$ 를 타겟 신경망 파라미터 $\theta^-$ 로 복사합니다. 

self.update_op = [t_p.assign(m_p) for t_p, m_p in zip(self.target.params, self.main.params)]

...

def update_target(self):
    self.sess.run(self.update_op)

...

if (agent.train_cnt + 1) % UPDATE == 0:
    agent.update_target()

 

 

200번의 에피소드를 학습한 결과입니다.

 

 

저장한 모델을 불러와 테스트합니다.

INFO:tensorflow:Restoring parameters from ./tmp/dqn_cartpole/model/model
episode: 0 / step 200
episode: 1 / step 200
episode: 2 / step 200
episode: 3 / step 200
episode: 4 / step 200