[강화학습] Dueling Double Deep Q Learning(DDDQN / Dueling DQN / D3QN)

Dueling Double DQN

https://arxiv.org/pdf/1509.06461.pdf

https://arxiv.org/pdf/1511.06581.pdf

Double DQN

• DQN에서 reward를 과대 평가하는 문제가 있음.
• Q Value가 agent가 실제보다 높은 리턴을 받을 것이라고 생각하는 경향
• ⇒ Q learning update 방정식에 다음 상태(state)에 대한 Q value 최대값이 존재하기 때문
• Q 값에 대한 max 연산은 편향을 최대화한다.
  • 환경의 최대 true value가 0인데 agent가 추정하는 최대 true value가 양수인 경우에 성능 저하

해결을 위해 두 개의 network 사용.

• Q Next : action selection → 다음 액션으로 가장 좋은 것 선택
• Q Eval : action evaluation → 선택한 액션이 좋은 액션이었는지 평가

Dueling DQN

두 개의 network 사용

• V : Value function → 상태(state)로부터 받을 reward 양에 따른 상태 가치
• A : Advantage function → 부분적인 상태(state)에서 어떤 action이 다른 action에 비한 가치
• Q function은 V + A

1. DQN

import tensorflow as tf
import tensorflow.keras as keras


class DuelingDeepQNetwork(keras.Model):
    def __init__(self, n_actions, fc1_dims, fc2_dims):
        super(DuelingDeepQNetwork, self).__init__()
        self.dense1 = keras.layers.Dense(fc1_dims, activation='relu')
        self.dense2 = keras.layers.Dense(fc2_dims, activation='relu')
        self.V = keras.layers.Dense(1, activation=None)
        self.A = keras.layers.Dense(n_actions, activation=None)

    def call(self, state):
        x = self.dense1(state)
        x = self.dense2(x)
				# Value function -> 현재 상태 가치
        V = self.V(x)
				# Advantage function -> action 가치
        A = self.A(x)

        Q = (V + (A - tf.math.reduce_mean(A, axis=1, keepdims=True)))

        return Q
    
    def advantage(self, state):
        x = self.dense1(state)
        x = self.dense2(x)
        A = self.A(x)

        return A

 

2. Replay Buffer

• Replay Buffer를 사용하면 아래와 같은 이점 존재
  • 신경망 업데이트 시에 example 간의 의존도 감소
  • mini batch 사용으로 학습속도 가속
  • 과거의 transition을 다시 사용해서 망각을 방지

import numpy as np

class ReplayBuffer():
    def __init__(self, max_size, input_shape):
        self.mem_size = max_size
        self.mem_cntr = 0

        self.state_memory = np.zeros((self.mem_size, *input_shape),
                                        dtype=np.float32)
        self.new_state_memory = np.zeros((self.mem_size, *input_shape),
                                        dtype=np.float32)
        self.action_memory = np.zeros(self.mem_size, dtype=np.int32)
        self.reward_memory = np.zeros(self.mem_size, dtype=np.float32)
        self.terminal_memory = np.zeros(self.mem_size, dtype=np.bool)

    def store_transition(self, state, action, reward, state_, done):
        index = self.mem_cntr % self.mem_size
        self.state_memory[index] = state
        self.new_state_memory[index] = state_
        self.action_memory[index] = action
        self.reward_memory[index] = reward
        self.terminal_memory[index] = done

        self.mem_cntr += 1

    def sample_buffer(self, batch_size):
        max_mem = min(self.mem_cntr, self.mem_size)
        batch = np.random.choice(max_mem, batch_size, replace=False)

        states = self.state_memory[batch]
        new_states = self.new_state_memory[batch]
        actions = self.action_memory[batch]
        rewards = self.reward_memory[batch]
        dones = self.terminal_memory[batch]

        return states, actions, rewards, new_states, dones

 

3. Agent

from importlib import import_module
import tensorflow as tf
from dueling_ddqn_lunar import DuelingDeepQNetwork
from dueling_ddqn_replay_buffer_lunar import ReplayBuffer
from tensorflow.keras.optimizers import Adam
import numpy as np


class Agent():
    def __init__(self, lr, gamma, n_actions, epsilon, batch_size,
                 input_dims, epsilon_dec=1e-3, eps_end=0.01, 
                 mem_size=100000, fc1_dims=128,
                 fc2_dims=128, replace=100):
        self.action_space = [i for i in range(n_actions)]

# gamma = discount factor
        self.gamma = gamma
# epsilon = agent가 다른 action을 선택하도록 하는 random 지수
        self.epsilon = epsilon
        self.eps_dec = epsilon_dec
        self.eps_min = eps_end
        self.replace = replace
        self.batch_size = batch_size

        self.learn_step_counter = 0
        self.memory = ReplayBuffer(mem_size, input_dims)

# 이 action을 평가하는 network
        self.q_eval = DuelingDeepQNetwork(n_actions, fc1_dims, fc2_dims)
# 현재 상황에 따른 action을 선택하는 network
        self.q_next = DuelingDeepQNetwork(n_actions, fc1_dims, fc2_dims)

        self.q_eval.compile(optimizer=Adam(learning_rate=lr),
                            loss='mean_squared_error')
                            
        self.q_next.compile(optimizer=Adam(learning_rate=lr),
                            loss='mean_squared_error')

    def store_transition(self, state, action, reward, new_state, done):
        self.memory.store_transition(state, action, reward, new_state, done)

    def choose_action(self, observation):
        if np.random.random() < self.epsilon:
            action = np.random.choice(self.action_space)
        else:
            state = np.array([observation])
# action 평가
            actions = self.q_eval.advantage(state)
# 평가 값 중 최고
            action = tf.math.argmax(actions, axis=1).numpy()[0]

        return action

    def learn(self):
        if self.memory.mem_cntr < self.batch_size:
            return

        if self.learn_step_counter % self.replace == 0:
            self.q_next.set_weights(self.q_eval.get_weights())

        states, actions, rewards, states_, dones = \
                                    self.memory.sample_buffer(self.batch_size)

        q_pred = self.q_eval(states)
        q_target = q_pred.numpy()
        max_actions = tf.math.argmax(self.q_eval(states_), axis=1)
        
        # improve on my solution!
        for idx, terminal in enumerate(dones):
            #if terminal:
                #q_next[idx] = 0.0
            q_target[idx, actions[idx]] = rewards[idx] + \
                    self.gamma*q_next[idx, max_actions[idx]]*(1-int(dones[idx]))
        self.q_eval.train_on_batch(states, q_target)

        self.epsilon = self.epsilon - self.eps_dec if self.epsilon > \
                        self.eps_min else self.eps_min

        self.learn_step_counter += 1