[카테고리:] 생활코딩

강아지 책에 나온 예제를 실행했다. 책 실행 환경과 내 그것이 달라 실행할 수 없었다. 나는 docker로 tensorflow를 사용한다. 그것도 cpu가 avx 등 을 지원하지 않아 직접 컴파일했다. openAi gym은 GUI 환경에서 실행되어 docker로 구동하기 힘들다. 데스크탑을 거실에 설치하여 다른 노트북에서 ssh로 접속하여 사용한다. 이런 모든 문제를 jypyter notebook로 해결했다. 전 글에 설명한대로 docker를 아래 명령으로 구동한다.

docker run -it -v /run/user/1000:/run/user/1000 -v /dev:/dev -v /tmp/.X11-unix:/tmp/.X11-unix:ro --privileged --ipc=host --shm-size=256m --net=host -e DISPLAY=$DISPLAY -e XDG_RUNTIME_DIR=/run/user/1000 --runtime=nvidia -e LC_ALL=C.UTF-8 -v /home/now0930/tensorflow/:/home/mnt tensorflow/tensorflow:1.12.0-rc2-gpu-py3-keras /bin/bash

jupyter notebook을 실행한다. uid 1000으로 실행하는 방법을 모르겠다.

docker exec -w /home/mnt/gym vibrant_banach /usr/bin/xvfb-run -s "-screen 0 1400x600x24" jupyter notebook --allow-root

책 코드를 실행하기 위해 아래와 같이 수정했다.

import sys
import gym
import pylab
import random
import numpy as np
from collections import deque
from keras.layers import Dense
from keras.optimizers import Adam
from keras.models import Sequential
from gym import wrappers

EPISODES = 300


# 카트폴 예제에서의 DQN 에이전트
class DQNAgent:
    def __init__(self, state_size, action_size):
        self.render = False
        self.load_model = False

        # 상태와 행동의 크기 정의
        self.state_size = state_size
        self.action_size = action_size
        print("self.state_size는", self.state_size)
        print("self.action_size는", self.action_size)
        # DQN 하이퍼파라미터
        self.discount_factor = 0.99
        self.learning_rate = 0.001
        self.epsilon = 1.0
        self.epsilon_decay = 0.999
        self.epsilon_min = 0.01
        self.batch_size = 64
        #self.batch_size = 1
        self.train_start = 1000

        # 리플레이 메모리, 최대 크기 2000
        self.memory = deque(maxlen=2000)

        # 모델과 타깃 모델 생성
        self.model = self.build_model()
        self.target_model = self.build_model()

        # 타깃 모델 초기화
        self.update_target_model()

        if self.load_model:
            self.model.load_weights("./save_model/cartpole_dqn_trained.h5")

    # 상태가 입력, 큐함수가 출력인 인공신경망 생성
    def build_model(self):
        model = Sequential()
        model.add(Dense(24, input_dim=self.state_size, activation='relu',
                        kernel_initializer='he_uniform'))
        model.add(Dense(24, activation='relu',
                        kernel_initializer='he_uniform'))
        model.add(Dense(self.action_size, activation='linear',
                        kernel_initializer='he_uniform'))
        model.summary()
        model.compile(loss='mse', optimizer=Adam(lr=self.learning_rate))
        return model

    # 타깃 모델을 모델의 가중치로 업데이트
    def update_target_model(self):
        self.target_model.set_weights(self.model.get_weights())

    # 입실론 탐욕 정책으로 행동 선택
    def get_action(self, state):
        if np.random.rand() <= self.epsilon:
            return random.randrange(self.action_size)
        else:
            q_value = self.model.predict(state)
            #print("q_value는", q_value)
            #print("argamx(q_value)는", np.argmax(q_value[0]))
            #print("State는", state)
            return np.argmax(q_value[0])

    # 샘플 <s, a, r, s'>을 리플레이 메모리에 저장
    def append_sample(self, state, action, reward, next_state, done):
        self.memory.append((state, action, reward, next_state, done))

    # 리플레이 메모리에서 무작위로 추출한 배치로 모델 학습
    def train_model(self):
        if self.epsilon > self.epsilon_min:
            self.epsilon *= self.epsilon_decay

        # 메모리에서 배치 크기만큼 무작위로 샘플 추출
        mini_batch = random.sample(self.memory, self.batch_size)
        states = np.zeros((self.batch_size, self.state_size))
        next_states = np.zeros((self.batch_size, self.state_size))
        actions, rewards, dones = [], [], []

        for i in range(self.batch_size):
            states[i] = mini_batch[i][0]
            actions.append(mini_batch[i][1])
            rewards.append(mini_batch[i][2])
            next_states[i] = mini_batch[i][3]
            dones.append(mini_batch[i][4])

        # 현재 상태에 대한 모델의 큐함수
        # 다음 상태에 대한 타깃 모델의 큐함수
        target = self.model.predict(states)
        target_val = self.target_model.predict(next_states)

        # 벨만 최적 방정식을 이용한 업데이트 타깃
        for i in range(self.batch_size):
            #print("target은", target[i])
            if dones[i]:
                target[i][actions[i]] = rewards[i]
            else:
                target[i][actions[i]] = rewards[i] + self.discount_factor * (
                    np.amax(target_val[i]))

        self.model.fit(states, target, batch_size=self.batch_size,
                       epochs=10, verbose=0)


if __name__ == "__main__":
    # CartPole-v1 환경, 최대 타임스텝 수가 500
    env = gym.make('CartPole-v1')
    env = wrappers.Monitor(env, "./gym-results-Cart", force=True, video_callable=lambda episode_id: episode_id%20==0)

    state_size = env.observation_space.shape[0]
    action_size = env.action_space.n

    # DQN 에이전트 생성
    agent = DQNAgent(state_size, action_size)

    scores, episodes = [], []

    for e in range(EPISODES):
        done = False
        score = 0
        # env 초기화
        #env = wrappers.Monitor(env, "./gym-results-Cart", force=True)
        state = env.reset()
        state = np.reshape(state, [1, state_size])
        #print("state reshape는",state)

        while not done:
            
            if agent.render:
                env.render()

            # 현재 상태로 행동을 선택
            action = agent.get_action(state)
            # 선택한 행동으로 환경에서 한 타임스텝 진행
            next_state, reward, done, info = env.step(action)
            next_state = np.reshape(next_state, [1, state_size])
            # 에피소드가 중간에 끝나면 -100 보상
            reward = reward if not done or score == 499 else -100

            # 리플레이 메모리에 샘플 <s, a, r, s'> 저장
            agent.append_sample(state, action, reward, next_state, done)
            # 매 타임스텝마다 학습
            if len(agent.memory) >= agent.train_start:
                agent.train_model()

            score += reward
            state = next_state

            if done:
                # 각 에피소드마다 타깃 모델을 모델의 가중치로 업데이트
                agent.update_target_model()

                score = score if score == 500 else score + 100
                # 에피소드마다 학습 결과 출력
                scores.append(score)
                episodes.append(e)
                #pylab.plot(episodes, scores, 'b')
                #pylab.savefig("./save_graph/cartpole_dqn.png")
                print("episode:", e, "  score:", score, "  memory length:",
                      len(agent.memory), "  epsilon:", agent.epsilon)

                # 이전 10개 에피소드의 점수 평균이 490보다 크면 학습 중단
                if np.mean(scores[-min(10, len(scores)):]) > 490:
                    agent.model.save_weights("./save_model/cartpole_dqn.h5")
                    sys.exit()

다음 내용을 수정했다.

• 10 행: from gym import wrappers: wrappers 사용.
• 120 행: env = wrappers.Monitor(env, “./gym-results-Cart”, force=True, video_callable=lambda episode_id: episode_id%20==0): 20번마다 동영상으로 저장. 여기를 설정하지 않으면 64번째 에피소드 동영상을 저장하고, 1,000번째 에피소드로 넘어간다. 학습 과정을 알 수 없다.

cartPole-v1은 10초 동안 막대기가 넘어지지 않으면 끝나는 모델이다. 10초 전 막대기가 12도 넘게 기울어지면 끝난다. Q-learning은 아래 식으로 정의된다. 는 learning rate, 는 discount factor.

위 식에서 와 로 오차를 줄여 나간다. 오차는 정답 – 예측 로 정의한다. keras.fit에서 state에 따른 예측과 정답인 target[i]로 학습한다. keras.fit을 부르기 전에 miniBatch 크기 모든 target[size]를 업데이트한다. print로 taget[i]값을 확인할 수 있다.

episode: 73   score: 64.0   memory length: 2000   epsilon: 0.2699131774597243
update전 target[20] [41.376812 37.715374]
update후 target[20] [41.376812 40.977367]
#action에 따른 정확한 Q 값을 수정.
##여기에서 무작위로 다시 뽑음.
update전 target[20] [46.597427 39.114082]
update후 target[20] [46.597427 42.882465]
...
episode: 132   score: 419.0   memory length: 2000   epsilon: 0.009998671593271896
update전 target[20] [113.11228 112.72044]
update후 target[20] [113.11228 113.5387 ]
update전 target[20] [109.97069 109.24642]
update후 target[20] [109.97069 109.48601]
update전 target[20] [112.96742 112.18164]
update후 target[20] [112.92011 112.18164]
# 132번 episode가 73번 episode보다 정답에 더 근접하여 업데이트 량이 줄었다.

이렇게 10개 평균 score가 490 이상이면 학습을 중단한다. 적당한 Q 값을 찾는 문제라 value based Reinforce Learning이라 한다. 모델을 정의할 때 Q 값을 제한하지 않으려 activation을 linear로 설정했다.

정책을 직접 근사하는 방법은 policy based Reinforce Learning인데, 마지막에 softmax 로 확율을 나오도록 해야 한다.

Sutton & Barto 가 쓴 책을 보면 잘 이해가지 않는데, 직접 돌려 확인하면 이해했다 착각한다.