강화학습 (reinforcement learning) 정책 학습과 가치 학습 비교 - Deep Q Network를 이용한 Atari Space Invaders 와 CartPole

정책 학습과 가치 학습

 강화 학습(reinforcement learning)의 두가지 범주에 속하는 정책 학습과 가치 학습을 비교하기 위해 이전 포스트에서 정책 경사(policy gradient)를 이용한 Atari 게임 Space Invaders 학습을 알아본 적이 있다.


 본 포스트에서는 OpenAI gym의 CartPole과 Atari 게임 Space Invaders 환경에서 Deep Q-network을 이용한 Agent를 만들고 학습이 잘 이루어지는지 테스트하였다. 이전 정책 경사(policy gradient) 에이전트에서 Space Invaders학습이 잘 이루어지지 않아 학습 방법 비교를 위해 간단한 게임인 CartPole도 테스트에 사용하였다.


DQN(Deep Q-Network)을 이용한 Atari 게임 Space Invaders 와 CartPole 학습

openai gym Atari 게임 Space Invaders / CartPole 환경


 Space Invaders는 128바이트 형태의 상태 데이터를 반환하는 'SpaceInvaders-ram-v0'를 사용하였다.

import gym

env = gym.make("SpaceInvaders-ram-v0")


 CartPole은 'CartPole-v0'을 사용하였으며, 테스트 목표에 맞게 _max_episode_steps도 설정 했다.


import gym

env = gym.make("CartPole-v0')
env._max_episode_steps=MAX_EPISODE_STEP+1



Q-learning

 Q-learning은 정책을 직접 학습하는 대신 상태(state)와 행동(action)의 가치 (Q-value)를 학습한다. 가치 Q-value는 어떤 상태(state)에서 어떤 행동(action)을 행한 후 후속되는 기대되는 미래의 보상을 의미한다. 


 위 그림의 예를 들면, 로봇이 행동 a를 한 후 상태 s에서, 가장 많은 보상을 받기 위해선, 기대되는 Q-value가 가장 높은 행동을 해야 한다. 이 상태(state)별 각 행동(action)의 Q-value를 가진 테이블을 Q table이라고 한다. 

 제한된 환경이 아닌 일반적인 환경에서는 상태(state)도 무수히 많고, 행할 수 있는 행동(action)도 많기 때문에 Q-table의 크기도 커지고, Q-value를 만드는 것에 어려움이 있다. 이를 해결하고자 인공 신경망을 사용한다. 이것이 Deep Q Network이다. 

 Deep Q Network Agent의 학습은 상태(state)에서 각 행동(action)별 Q-value를 생성하는 과정이다. 

학습의 목적 함수는 아래와 같다.


Deep Q-Network 분리

학습의 목적함수를 보면 현재의 Q-value에서 미래의 Q-value를 차이를 기반으로 손실 함수를 정의하고 있다. 이렇게 되면 학습과정에 파라메터들이 업데이트 될 때마다 기준도 변하게 되어 파라메터들이 진동하고 손실(loss)값이 발산하게 되는 문제가 발생한다. 

이 문제를 해결하기 위해 Q-Network를 Predict Q-Network와 Target Q-Network로 분리하여, 학습 시 Predict Q-Network의 파라메터만 업데이트 한 후 일정 회수의 학습이 끝난 후 Target Q-Network를 업데이트 하는 방식을 사용해야 한다. 


Experience Replay 

DQN Agent 학습 시 최근의 경험만을 학습한다면, 이 Agent는 전체 환경의 대표(아래 그림의 A)하는 것이 아니라 일부 지역만을 대표(아래 그림의 B)하게 되는 문제가 발생할 수 있다. 

이 문제를 피하기 위해 학습과정에서 과거의 경험까지 함께 학습해야 한다. 


episode 경험 저장 class 소스 코드

experience replay, discounted reward, 데이터 정규화(normalize) 등이 구형되어 있다. 학습은 정해진 에피소드가 끝난 후 주기를 가지고 시행하며, 그 동안의 경험 데이터를 저장한다. 각 에피소드가 끝난 후 보상(reward)데이터는 discount와 normalize를 거치게 된다. 


class experience:
    def __init__(self, max_size=50000):
        self.statet0 = []
        self.action = []
        self.reward = []
        self.statet1 = []
        self.done= []
        self.max_size = max_size
    
    def reset(self):
        self.statet0 = []
        self.action = []
        self.reward = []
        self.statet1 = []
        self.done= []
        
    def append(self, statet0, action, reward, statet1, done):
        self.statet0.append(statet0)
        self.action.append(action)
        self.reward.append(reward)
        self.statet1.append(statet1)
        self.done.append(done)
        self.prune()
    
    def add(self, statet0, action, reward, statet1, done):
        self.statet0 += statet0
        self.action += action
        self.reward += reward
        self.statet1 += statet1
        self.done += done
        self.prune()
        
    def prune(self):
        while len(self.statet0) > self.max_size:
            self.statet0.pop(0)
            self.action.pop(0)
            self.reward.pop(0)
            self.statet1.pop(0)
            self.done.pop(0)
            
    def discount(self, data, gamma = 0.95):
        discount_data = [0 for _ in data]
        cumulative = 0
        for t in reversed(range(len(data))):
            cumulative = cumulative * gamma + data[t]
            discount_data[t] = cumulative
            
        discount_data -= np.mean(discount_data)
        if np.std(discount_data) == 0:
            return discount_data
        discount_data /= np.std(discount_data)
        
        return discount_data
    
    def normalize(self, data):
        normal_data = [x for x in data]
        normal_data -= np.mean(normal_data) 
        if np.std(normal_data) == 0:
            return normal_data
        normal_data /= np.std(normal_data)

        return normal_data

    
    def sample_batch(self, batch_size):
        
        statet0,action,reward,statet1,done=[],[],[],[],[]
        
        rands = np.arange(len(self.statet0))
        np.random.shuffle(rands)
        
        if batch_size > len(self.statet0):
            batch_size = len(self.statet0)
            
        rands = rands[:batch_size]
        for i in rands:
            statet0.append(self.statet0[i])
            action.append(self.action[i])
            reward.append(self.reward[i])
            statet1.append(self.statet1[i])
            done.append(self.done[i])
            
        return statet0,action,reward,statet1,done



Deep Q-Network Agent 소스 코드

인공 신경망은 keras를 사용해 구현했다. DQN 모델 간의 파라메터 업데이트가 편해 keras를 사용했다.  코드가 tensorflow api 만을 사용할 때 보다 훨씬 간결해진 느낌이다.


import numpy as np
import random
from collections import deque
from keras.layers import Input, Dense, Dropout, regularizers
from keras.models import Model
from keras.optimizers import Adam

class DQNAgent:
    
    def __init__(self,state_size, num_actions, learning_rate = 0.01, gamma=0.9, hidden_layer_units=10, replay_buffer_size = 50000):
        
        self.gamma = gamma
        self.num_actions = num_actions
        self.state_size = state_size
        self.replay_buffer_size = replay_buffer_size

        self.mainDQN = self.build_model(in_size=state_size,out_sizes=num_actions,layer_units=hidden_layer_units,lr=learning_rate)
        self.targetDQN = self.build_model(in_size=state_size,out_sizes=num_actions,layer_units=hidden_layer_units,lr=learning_rate)
        #self.mainDQN.summary()
        #self.targetDQN.summary()
        self.update_targetDQN()
        self.replay_buffer = experience()
        
        
    def build_model(self,in_size, out_sizes, depth=5, layer_units = 10, lr=0.01):
        in_layer = Input(shape=(in_size,),name='input_layer')
        prev = in_layer
        
        for fc in range(depth):
            prev = Dense(units=layer_units, activation='tanh', 
            kernel_regularizer=regularizers.l2(l=0.01), name='dense%d'%fc)(prev)
        
        out_layer = Dense(out_sizes,activation='linear', 
                          kernel_regularizer=regularizers.l2(l=0.01), name='out_layers')(prev)
        model = Model(inputs=[in_layer],outputs=[out_layer])
        optimizer = Adam(lr=lr)
        model.compile(loss='mean_squared_error', optimizer=optimizer,metrics=['accuracy'])
        return model
            
    def store_experience(self, state_t0, action, reward, state_t1, done):
        normal_reward = [0 for _ in reward]
        normal_reward[:] = self.replay_buffer.discount(reward)[:]
        self.replay_buffer.add(state_t0, action, normal_reward, state_t1, done)
        
    def update_targetDQN(self):
        self.targetDQN.set_weights(self.mainDQN.get_weights())
        
    def learn(self, batch_size = 10):
        # batch samples      
        statet0_batch,action_batch,reward_batch,statet1_batch,done_batch = self.replay_buffer.sample_batch(batch_size=batch_size)
        state = np.empty(0).reshape(0,self.state_size)
        targetQ = np.empty(0).reshape(0,self.num_actions)
        
        for state_t0, action, reward, state_t1, done in zip(statet0_batch,action_batch,reward_batch,statet1_batch,done_batch):
            Q = self.mainDQN.predict(self.state_reshape(state_t0,self.state_size))
            
            if done:
                Q[0,action]= reward
            else:
                Q[0,action]= reward + (self.gamma*np.max(self.targetDQN.predict(self.state_reshape(state_t1,self.state_size))))
                
            targetQ = np.vstack([targetQ,Q])
            state = np.vstack([state,state_t0])
        
        loss =self.mainDQN.train_on_batch(state,targetQ)
        
        return loss[0]
     
    def state_reshape(self, state, size):
        return np.reshape(state, [1,size]);

    def predict_action(self,env_action_sample, state, epsilon):
        if np.random.rand(1) < epsilon:
            return env_action_sample
        action = np.argmax(self.mainDQN.predict(self.state_reshape(state,self.state_size)))
        return action



DQN(Deep Q-Network) Agent를 사용한 CartPole-v0

CartPole 환경의 학습 코드는 아래와 같다. 


import gym
import time
import numpy as np
from RyanDQNAgent import DQNAgent,experience
import matplotlib.pyplot as plt

MAX_EPISODES = 5000
MAX_EPISODES_STEP = 10000
TRAINING_PERIOD = 10
TRAINING_START_EP = 50

def training(env,agent):
    
    if env == None or agent == None:
        return
    
    training_done_count = 0
    loss_log = []
    steps_log = []

    for episode in range(MAX_EPISODES):
        
        exp = experience()
        state_t0 = env.reset()
        epsilon = 1.0/((episode/10)+1)
        done = False
        step = 0
         
        #
        while done == False:
            #env.render()
            action = agent.predict_action(env_action_sample=env.action_space.sample(),state=state_t0,epsilon= epsilon)
                
            state_t1, reward, done, info = env.step(action)            
            if done:
                reward -= 1
                
            exp.append(statet0=state_t0,action=action,reward=reward,statet1=state_t1,done=done)
            state_t0 = state_t1
            step += 1
        
        #
        print('episode ',episode,' - steps:', step)
        agent.store_experience(state_t0=exp.statet0,action=exp.action,reward=exp.reward,state_t1=exp.statet1,done=exp.done)
        steps_log.append(step)
        
        if episode >= TRAINING_START_EP and episode % TRAINING_PERIOD == 0 and step < MAX_EPISODES_STEP:
            loss = 0
            for _ in range(50):
                loss += agent.learn(batch_size=10)               
            loss /= 50
            loss_log.append(loss)
            print('loss ..... ',loss)
            
            agent.update_targetDQN()
        
        if step >= MAX_EPISODES_STEP:
            training_done_count += 1
        else:
            training_done_count = 0
            
        if training_done_count >= 10: 
            print('training_done')
            break
        
    #
    print('loss')
    plt.bar(range(len(loss_log)), loss_log, color="red")
    plt.show()
    print('steps')
    plt.bar(range(len(steps_log)), steps_log, color="green")
    plt.show()


if __name__ == "__main__":
env = gym.make('CartPole-v0')
    env._max_episode_steps = MAX_EPISODES_STEP+2

    agent= DQNAgent(state_size = env.observation_space.shape[0],num_actions=env.action_space.n,hidden_layer_units=10)
    
    training(env,agent)
    
    env.close()


위 코드를 사용하여 학습한 결과는 아래와 같다.



이전 정책 경사(policy gradient)를 이용한 학습보다 빠르게 학습되는 것을 볼 수 있다. 


Discounted Reward와 Normalize 

Agent 학습 시 discounted reward와 normalize 여부에 따라 학습 속도와 능력에 많은 차이를 가져온다. 아래처럼 보상 감쇠와 정규화를 사용하지 않으면 사용한 경우에 비해 학습 속도가 느리거나 아예 학습을 실패하는 사례가 많아진다.




DQN(Deep Q-Network) Agent를 사용한 Atari Space Invaders

Space Invaders 학습 코드는 아래와 같다. 

import gym
import time
import numpy as np
from RyanDQNAgent import experience,DQNAgent
import matplotlib.pyplot as plt

MAX_EPISODES = 5000
MAX_EPISODES_STEP = 10000
TRAINING_START_EP = 10
TRAINING_PERIOD = 5

def training(env,agent):
    
    if env == None or agent == None:
        return
    
    training_done_count = 0
    rewards_log = []
    steps_log = []
    loss_log = []

    for episode in range(1,MAX_EPISODES+1,1):
        
        exp = experience()
        episode_reward = 0
        ale_lives = 3
        state_t0 = env.reset()
        step = 0
        epsilon = 1.0/((episode/10.0)+1.0)

        while step<=MAX_EPISODES_STEP:
            #env.render()
            step += 1
            action = agent.predict_action(env_action_sample=env.action_space.sample(),state=state_t0,epsilon= epsilon)
            state_t1, reward, done, info = env.step(action)
            episode_reward += reward
            
            if info['ale.lives'] != ale_lives or done:
                ale_lives = info['ale.lives']
                reward -= 1
                exp.append(statet0=state_t0,action=action,reward=reward,statet1=state_t1,done=done)
                agent.store_experience(state_t0=exp.statet0,action=exp.action,reward=exp.reward,state_t1=exp.statet1,done=exp.done)
                exp.reset()
            else:
                exp.append(statet0=state_t0,action=action,reward=reward,statet1=state_t1,done=done)
                        
            state_t0 = state_t1 # update new state
            
            if done:
                break
        
        #
        print('episode ',episode,' - steps: ', step, ' episode_reward: ',episode_reward)
        steps_log.append(step)
        rewards_log.append(episode_reward)
        
        if episode >= TRAINING_START_EP and episode % TRAINING_PERIOD == 0 and step < MAX_EPISODES_STEP:
            loss = 0
            for _ in range(50):
                loss += agent.learn(batch_size=10)               
            loss /= 50
            loss_log.append(loss)
            print('loss ..... ',loss)
            
            agent.update_targetDQN()
        
        if step >= MAX_EPISODES_STEP:
            training_done_count += 1
        else:
            training_done_count = 0
            
        if training_done_count >= TRAINING_PERIOD: 
            print('training_done')
            break    
    #
    print('loss')
    plt.bar(range(len(loss_log)), loss_log, color="red")
    plt.show()
    print('steps')
    plt.bar(range(len(steps_log)), steps_log, color="green")
    plt.show()
    print('reward')
    plt.bar(range(len(rewards_log)), rewards_log, color="blue")
    plt.show()

if __name__ == "__main__":
    
    env = gym.make("SpaceInvaders-ram-v0")

    agent= DQNAgent(state_size = env.observation_space.shape[0],num_actions=env.action_space.n,hidden_layer_units=256)
   
    training(env,agent)
    
    env.close()


위 코드를 이용한 학습 결과는 아래와 같다. 


이전 정책 경사(policy gradient)와 마찬가지로 DQN에서도 학습이 이루어지지 않았다.

Cartpole에서 성공한 것과는 대조적이다. 학습 성공에 다른 테크닉이 필요해 보인다. 실패의 원인이라고 생각되는 것이 있지만, 확실하진 않다. 그래도 policy gradient 보다 학습 성능이 좋아 보인다. 추후 별도의 포스트에서 학습에 성공해 성공 사례를 올리도록 하겠다. 


[관련 포스트]
openai gym을 사용하여 강화학습(reinforcement learning) 환경 만들기
강화학습과 마르코프 결정 과정( MDP, Markov Decision process) 관계 요약
강화학습 (reinforcement learning) 정책 학습과 가치 학습 비교 - 정책 경사 (policy gradient)를 이용한 Atari Space Invaders

댓글

댓글 쓰기

이 블로그의 인기 게시물

간단한 cfar 알고리즘에 대해

이미지
Constant False Alarm Rate (CFAR)은 기본적으로 테스트하고자 하는 위치의 cell과 주변 cell의 관계를 보고 테스트 cell이 Target 인지 아닌지를 구분하는 알고리즘이다. 주변 cell과의 관계를 어떤 방식으로 비교하는가에 따라 여러가지 알고리즘이 존재한다. CA-CFAR, GO-CFAR, SO-CFAR, OS-CFAR, VI-CFAR, OSVI-CFAR등이 있다. 이 많은 알고리즘 중 간단하고 기본적인 CA-CFAR과 OS-CFAR에 대해 알아보자. CA-CFAR (Cell Average CFAR)   CA-CFAR의 원리는 위 그림과 같다. 좌우의 Reference Cell의 평균에 Scale Factor T를 곱하고, ADT (Average Decision Threshold)를 합하여 CUT가 타겟인지 아닌지를 구분한다. ADT는 신호의 white noise를 고려하기 위해 더해준다. Scale Factor T와 ADT는 아래와 같은 공식으로 사용할 수 있다. 사용하려는 시스템에 따라 공식은 달라질 수도 있다.   Pfa는 False Alarm 발생 확률을 나타내고, M과N는각 좌우의 Reference Cell의 개수를 나타낸다. OS-CFAR (Order Static CFAR)   OS-CFAR의 원리는 위 그림과 같다. 좌우의 reference cell중 K번째 큰 값의 cell을 Reference Cell의 대표 값으로 정해 CUT가 타겟인지 아닌지를 구분한다. 보통 전제 reference cell의 크기 순에서 (3*N)/4 번째 cell이 성능이 좋다고 알려져 있다. CFAR 사용 예 아래 사진은 위 두 cfar을 사용한 예다. 파란색은 신호이며, 붉은 색은 char을 사용해 만든 threshold이다. 이 threshold보다 큰 값의 신호가 있는 위치에 실제 Tar...
자세한 내용 보기

windows에서 간단하게 크롬캐스트(Chromecast)를 통해 윈도우 화면 미러링 방법

이미지
 Windows 10에서는 미라캐스트를 사용한 무선 디스플레이연결을 지원한다. 하지만, 만약 크롬캐스트(Chromecast)를 사용해 윈도우 화면을 미러링 하고자 할 때, 구글 크롬 브라우저를 사용하면 쉽고 편하게 할 수 있다. https://www.google.co.kr/intl/ko/chrome/ VLC에서도 비디오 화면을 미러링 할 수 있다고 하는데, 잘 동작하지 않고 윈도우 화면 전체를 미러링하지 않아 크롬 브라우저를 사용하고 있다. 크롬 캐스트 윈도우 화면 전송 방법 0. 크롬 캐스트 디바이스와 windows pc는 같은 네트웍에 연결되어 있어야 한다. (같은 공유기에 연결되어 있어야 한다.) 1. 크롬 브라우저의 우측 상단의 메뉴 > 전송을 클릭하면, 현재 사용가능한 크롬캐스트 디바이스 항목을 볼 수 있다. 2. 편의에 따라 탭 전송 아이콘에서 우측 버튼을 클릭하여 항상 표시되게 설정할 수도 있다. 3.  윈도우 화면 전체를 전송하고자 할 때는 탭 전송의 소스에서 '데스크톱 전송'을 선택하고, 바로 위의 크롬캐스트 디바이스를 클릭하면 된다. ** 소스는 미러링 중에는 변경되지 않는다. 먼저 소스를 선택하고 미러링해야 한다.  4. 화면 전송을 끝내고자 할 때는, 탭 전송에서 화면 전송중인 크롬캐스트 디바이스를 클릭하거나, 크롬 브라우저를 종료하면 된다.
자세한 내용 보기

base64 인코딩 디코딩 예제 c 소스

이미지
본 글은 base64 encoder/decoder예제 c 소스 코드를 싣고 있다. BASE64 BASE64는 일련의 데이터를 64종류" ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/ "의 문자만을 이용하여 표현하도록 하는 부호화 방식이다. Base64는 인터넷 이메일 프로토콜이나 http url등에서 사용된다.  부호화 과정은 아래와 같다. 문자 : Hello Hex 코드 : 0x48 0x65 0x6c 0x6c 0x6f 8bits bin 코드: 01001000 01100101 01101100 01101100 01101111 8비트 단위의 문자를 6비트 단위로 분리한다. 6bits bin 코드 : 010010  000110  010101  101100  011011  000110  1111  00  ‘=’ 6비트 2진 코드 값은 BASE64 코드 테이블의 인덱스이고 각 인덱스의 문자로 base64 부호화 한다. 마지막 모자라는 비트는 0으로 채우며, 4글자 단위로 부호화 하기 때문에 4글자가 안되면 ‘=’기호를 사용해 4글자를 맞춘다. Base64 코드 : aGVsbG8= Base64 Encoder source code Base64 인코더 소스는 아래와 같다. 위의 부호화 예대로 text데이터를 받아 6bit 단위로 나누어 base64 코드 테이블에 맞춰 부호화를 진행한다. static const char base64code_ascii[] = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/"; int text2base64_size(const char *text) {...
자세한 내용 보기

아두이노(arduino) 심박센서 (heart rate sensor) 심박수 측정 example code

이미지
하드 정리 중 예전에 만든 아두이노에서 심박센서를 사용하여 심박수 측정하는 소스 코드가 있어, 이를 블로그 글로 남긴다. 인터넷 쇼핑몰에서 쉽게 구할 수 있는 저렴함 센서가 사용 되었으며, 연결은 아래 그림과 같이 연결 되었다.      심박센서의 하트 부위에 손가락을 올려 놓고, 아날로그 A0값을 읽으면 심박의 움직임에 따라 A0값이 변하는 동작이 간단한 센서이다. 아래 그림의 왼쪽은 일정 시간 간격 마다 A0값을 읽어 시리얼 플로터로 본 심박에 따른 데이터이고, 오른쪽은 평균 필터를 사용해 노이즈를 제거한 후 시리얼 플로터로 본 데이터이다.   심박수는 수집된 데이터에서 threshold를 넘는 peak들의 시간 간격을 측정하여 계산한다.  peak위치를 찾는 방법은 여러 가지 방법이 있으나, 소스에 구현된 방법은 신호의 기울기가 +에서 0또는 -로 바뀌는 위치를 peak으로 선정하도록 하였다. 심장이 t시간동안 N회 뛰는 경우 심박수는 heart rate bpm = (N/t)*60 을 표현 할 수 있다.   threshold는 데이터를 보고 임의로 정했으며, 측정 데이터에 따라 자동으로 정하고 싶을 때는 cfar 같은 방법을 사용하면 된다.  소스는 아래와 같다. #define heartbit_adc_threshold      511 #define meanfilter_size             5 #define heartbit_tick_buffer_size   10 ...
자세한 내용 보기

python asyncio를 이용한 async socket server client example code

 본 포스트는 python asyncio 라이브러리를 사용한 async socket 서버와 클라이언트 예제 코드를 싣고 있다. 이전 non-blocking socket의 예제를 asyncio 라이브러리에 맞게 변형하였다.  python socket API 와 non-blocking socket server client example code 1. asyncio (Asynchronous I/O) asyncio는 async/await 구문을 사용해 multi-thread처럼 동시에 진행하는 코드를 작성하는 라이브러리다. 네트워크 및 웹 서버, DB 연결, 분산 작업 대기열을 제공하는 분산 프레임 워크의 기반으로 사용된다고 한다. (  https://docs.python.org/3.7/library/asyncio.html ) async socket server/client 구현에 필요한 asyncio 라이브러리 API는 예제 코드를 보면서, 설명하도록 하겠다. 2. Async socket client example code import asyncio async def pingpong_client():     try:         reader, writer = await asyncio.open_connection(host='localhost',port=8000)     except OSError:         print('connection fail')         return          for _ in range(10):         writer.write(b'ping')     ...
자세한 내용 보기