[강화학습] gym으로 강화학습 custom 환경 생성부터 Dueling DDQN 학습까지

인터넷을 다 뒤져봤는데 강화학습을 gym에서 제공하는 게임 agent 사용해서 하는 예제는 육천만 개고 커스텀해서 학습을 하는 예제는 단 한 개 있었다. 이제 막 공부를 시작하는 사람들에게 도움이 되었으면 하는 마음으로 간단하게 써보고자 한다.

1. Gym의 Env 구조 살펴보기

꼭 그래야하는 것은 아니지만(밑바닥부터 구현하는 방법도 있긴 하다) 어쨌든 gym 라이브러리의 environment 구조를 기반으로 해서 구현해볼 것이다.

!pip install gym

gym 라이브러리의 env 구조는 대충 아래와 같다. site-packages/gym/core.py 에서 직접 볼 수 있다.

class Env(Generic[ObsType, ActType]):m.Generator] = None
		"""
		The main API methods that users of this class need to know are:

    - :meth:`step` - Takes a step in the environment using an action returning the next observation, reward,
      if the environment terminated and observation information.
    - :meth:`reset` - Resets the environment to an initial state, returning the initial observation and observation information.
    - :meth:`render` - Renders the environment observation with modes depending on the output
    - :meth:`close` - Closes the environment, important for rendering where pygame is imported

    And set the following attributes:

    - :attr:`action_space` - The Space object corresponding to valid actions
    - :attr:`observation_space` - The Space object corresponding to valid observations
    - :attr:`reward_range` - A tuple corresponding to the minimum and maximum possible rewards
    - :attr:`spec` - An environment spec that contains the information used to initialise the environment from `gym.make`
    - :attr:`metadata` - The metadata of the environment, i.e. render modes
    - :attr:`np_random` - The random number generator for the environment
    """

		def step(self, action: ActType) -> Tuple[ObsType, float, bool, bool, dict]:
        raise NotImplementedError

    def reset(
        self,
        *,
        seed: Optional[int] = None,
        options: Optional[dict] = None,
    ) -> Tuple[ObsType, dict]:

        # Initialize the RNG if the seed is manually passed
        if seed is not None:
            self._np_random, seed = seeding.np_random(seed)

    def render(self) -> Optional[Union[RenderFrame, List[RenderFrame]]]:
        raise NotImplementedError

    def close(self):
        pass

    @property
    def unwrapped(self) -> "Env":
        return self

    def __str__(self):
        """Returns a string of the environment with the spec id if specified."""
        if self.spec is None:
            return f"<{type(self).__name__} instance>"
        else:
            return f"<{type(self).__name__}<{self.spec.id}>>"

    def __enter__(self):
        """Support with-statement for the environment."""
        return self

    def __exit__(self, *args):
        """Support with-statement for the environment."""
        self.close()
        # propagate exception
        return False

• 공식 문서 설명

Core - Gym Documentation

중간 주석을 죄다 지워서 짧아보인다. 아무튼 중요한 것은 아래와 같다.

• Method
  • step 함수는 어떤 action 을 취했을 때, state의 변화와 그에 따른 보상을 책정하는 부분이다. 온도로 예를 들면 현재 state 가 20도이고, 목표 온도가 30도라면 어떤 action 을 취하고 난 뒤에 state의 변화가 긍정적인지, 부정적인지에 따라서 reward를 주는 부분이 이 부분이다.
    • return → observation, reward, terminated, truncated, info, done
  • reset 함수는 한 episode(혹은 게임, 일정 시간 등 학습 주기라고 생각하면 된다)를 시작할 때 attributes를 초기화하는 함수이다.
    • return → observation, info
• Attributes
  • action_space 는 이 모델이 취할 수 있는 행동들의 집합이다. 정수 집합일 수도 있고, DDPG 모델과 같은 경우에는 연속적인 공간으로 설정할 수도 있다. 이를테면 불을 꺼라, 켜라, 밝기를 반으로 낮춰라와 같은 행동명령의 리스트에 가깝다고 보면 된다.
  • observation_space 는 반환할 state의 형식과 범위의 정의이다.
  • state 는 환경의 상태이다.
  • reward 는 state에 대해 action을 취했을 때 반환해줄 보상이다. reward function은 누가 정해줄 수 있는 것은 아니고, 환경에 맞춰서 적절히 만들어줘야한다. 이 부분도 예제에 넣긴 했으니 걱정말자.

그럼 거두절미하고 곧장 한 번 짜보도록 하자. 

* 이 예제의 목표는 랜덤하게 설정된 숫자를 원하는 범위 안으로 들어오도록 하는 것이다.

 

2. Gym Custom Environment 작성하기

gym library의 Env 를 가져와서 상속받을 것이니 우선 import 한다.

from gym import Env
from gym.spaces import Box # observation space 용

__init__ 함수 아래에 action space, observation space, state, 그리고 episode length 를 선언해주었다.

a. Attributes 설정

class ENV(Env):
		def __init__(self):
        self.action_space = [i for i in range(-2, 3)]
        self.observation_space = Box(low=np.array([0]), high=np.array([100]), dtype=np.int8)
        self.state = np.random.choice([-20, 0, 20, 40, 60])
				self.prev_state = self.state
        self.episode_length = 100

		def step(self, action):
				pass
		
		def reset(self):
				pass

하나씩 보자.

• action space → -2에서 2 사이의 정수들로 이루어진 길이 5짜리 리스트이다. gym.spaces.Discrete 를 사용해서 정의하는 방법도 있는데, 이번 예제에서 사용할 예제 모델인 Dueling DQN은 action 추론을 오직 양수로만 한다. 따라서 action space의 index를 action으로 학습하기 위해서 이렇게 선언을 해줬다.
• observation space → gym.spaces.Box 를 사용했다. low와 high에 아주 큰 의미가 있지는 않다.
• state → 현재 우리는 실제 시험 환경이 없으므로 적당히 랜덤하게 설정하도록 했다.
• episode_length → 하나의 episode에 몇 개의 action을 실행할 것인지를 설정할 수 있다. 이 녀석이 0이 되면 step 함수의 return 중 하나인 done 을 True로 만들면 된다.

b. reset 함수 작성

이제 reset 함수를 작성한다. 초기화라 만만하다.

def reset(self):
    self.state = np.random.choice([-20, 0, 20, 40, 60])
    self.episode_length = 100 
    return self.get_obs()

def get_obs(self):
    return np.array([self.state], dtype=int)

get_obs 라는 녀석이 등장했다. 보시면 아시겠지만 Dueling Double DQN에 포함된 neural network의 input으로 사용될 state를 미리 numpy array의 (1, ) shape로 만들어서 신경망 통과 과정에서 문제가 생기지 않도록 해준 것인데, 이건 사용하는 모델에 맞춰서 조정해주어야한다.

c. step 함수 작성

조금 어려운 step 함수를 작성해보자.

def step(self, action):
    self.state += self.action_space[action]
    self.episode_length -= 1 
    
	# Reward 책정
    if self.state >= 20 and self.state <= 25:
        reward = +100
    else:
        reward = -100

		prev_diff = min(abs(self.prev_state - 20), abs(self.prev_state - 25))
    curr_diff = min(abs(self.state - 20), abs(self.state - 25))

    if curr_diff <= prev_diff: 
        if reward != 100: reward = reward + 50 
        else: reward = 100
    if curr_diff > prev_diff: reward -= 50

		self.prev_state = self.state
	
	# Episode 끝났는지 확인
    if self.episode_length <= 0: 
        done = True
    else:
        done = False
    
    info = {}
    
    return self.get_obs(), reward, done, info

다소 조잡해보이는 reward function은 이렇게 생성해주었다.

1. 범위 안에 state가 있으면 reward = 100, 없으면 reward = -100
2. 이전보다 범위에 가까워졌으면 +50, 멀어졌으면 -50
   • 단, 이미 범위 안에 있는 경우 100으로 고정

그리고 episode length에 따라 episode가 끝났는지 판별해주고, return하면 된다.

유의해야할 것은, 지금 이 보상함수는 잘 만든 보상함수가 아니다!!!!!!!!!!!!!! 더 많은 상황을 고려해서 적절한 보상함수를 만들어나가는 것이 좋다. 우리 모델은 궁극적으로 0과 1로 이루어져 있으므로 기면 기고 아니면 아니라는 것을 염두에 두자. 단순한 녀석에게 복잡한 것을 가르치려면 선생님이 연구를 좀 해야한다. (필자도 아직 잘 모른다ㅎ)

이외에도 render 함수나 close 함수 등 gym environment class에서 추가적으로 작성해줄 수 있는 함수들이 있지만 나는 시각화를 한다던가 하는 욕심까지는 없으므로 지금은 생략하도록 하겠다. 

 

3. 모델 적용

그럼 이제 모델과 함께 사용해보자. 코드는 미리 만들어둔 아래 repository 코드를 사용할 것이다. 아래 레포지토리 하위의 DDDQN 폴더의 코드를 보자.

GitHub - melli0505/Deep-RL

원본 코드는 여기에서 확인할 수 있다. ReinforcementLearning/DeepQLearning 아래에 있다.

GitHub - philtabor/Youtube-Code-Repository: Repository for most of the code from my YouTube channel

DDDQN 모델 구조

DDDQN(Dueling Double Deep Q-Network) 모델은 DDQN(Double Deep Q-Network) 모델과 Dueling DQN 모델의 장점을 섞어 만든 모델이다. 여기서는 자세한 모델 설명은 하지 않을 것이므로 추후에 Dueling DQN / Double DQN모델에 대한 포스팅을 하게되면 추가하도록 하겠다. 모델 구조는 우리에게 중요하

긴 하지만 아주 중요하진 않다. 코드만 잘 보면 된다.

a. DDDQN 톺아보기

env = ENV()
agent = Agent(env=env, lr=1e-3, gamma=0.99, n_actions=5, epsilon=1.0,
              batch_size=64, input_dims=[1])

우리가 위에 만들어둔 ENV를 env로 정의한 뒤에 DDDQN agent를 선언해준다. agent를 간단히 보도록 하자.

class Agent():
    def __init__(self, input_dims, env, epsilon=1, lr=1e-3, gamma=0.99, n_actions=2, batch_size=64,
                 epsilon_dec=1e-3, eps_end=0.01, 
                 mem_size=100000, fc1_dims=128,
                 fc2_dims=128, replace=100):
                 
        self.env = env
        self.gamma = gamma
        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 = 1
        self.memory = ReplayBuffer(mem_size, input_dims)
        self.q_eval = DuelingDeepQNetwork(n_actions, fc1_dims, fc2_dims)
        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')
        self.action_space = [i for i in range(n_actions)]

아아아주 간단하게 기본적인 모델에 대한 설명을 하자면, 이 모델은 두 개의 네트워크를 사용해서 하나는 action-reward에 따라 바로바로 업데이트를 하고, 하나는 일정기간 업데이트를 하지 않다가 업데이트를 하는 지연된 target 네트워크를 가진다.

또한 state-action-reward에 대한 정보를 학습할 때는 바로 실시간으로 들어오는 정보들을 활용하지 않고, Replay Memory라는 일종의 저장소에 저장해두었다가 랜덤하게 꺼낸 데이터를 가지고 학습하여 의존도라던가 뭐 기타등등 성능향상을 도모했다. 

더불어 학습 초반에 epsilon에 의해 랜덤하게 몇 번 action을 취하도록 해서 과적합(...맞는 표현인지 모르겠다.) 문제를 해결하려고 했다고 한다. 

gamma의 경우에는 값이 1에 가까울 수록 미래지향적으로, 현재의 값이 미래에 미칠 영향을 더 많이 고려한다는 뜻이다. Gradient Vanishing 문제에서 네트워크가 진행될수록 앞선 값이 잊히는 것과 비슷한 느낌이라고 생각할 수 있겠는데, gamma가 1이면 나중에 현재 네트워크의 예측 결과가 미래까지 영향을 온전히, 많이 미치는 것이다. 

n_action은 action space의 길이(action 개수)이다. self.action space는 environment 구성할 때 이야기한 것처럼 이번 예제에서 사용할 예제 모델인 Dueling DQN은 action 추론을 오직 양수로만 한다. 따라서 action space의 index를 environment step의 action으로 학습하기 위해 0~n_action 사이로 설정해주었다.

DDDQN의 학습 과정은 다음과 같다.

1. 네트워크를 기반으로 action 고르기 
2. env.step으로 state와 reward 받기
3. 받은 reward와 state, action 등을 memory에 저장하기
4. memory에 저장된 데이터를 랜덤하게 뽑아서 학습하고 네트워크를 개선하기
5. 1~4 반복

그리고 하나 유의해야할 것이 있다면 Readme 파일에도 작성해둔 것처럼 학습 과정에서 reward function을 상당히 많이 믿고!!! 영향을 많이 받아서 업데이트를 시키기 위해서 network target 값에 reward의 반영율 10배 이벤트를 설정했다. 원래대로라면 아래의 * 10이 없어야한다.

# dddqn.py, line 59-62

for idx, terminal in enumerate(dones):
    q_target[idx, actions[idx]] = rewards[idx] * 10 + \
            self.gamma*q_next[idx, max_actions[idx]]*(1-int(dones[idx]))
self.q_eval.train_on_batch(states, q_target)

그러니 뭔가 잘못되고 있는 것 같다면 이 녀석을 의심해보시길 바란다. 

 

4. 실행

python DDDQN_custom/main.py

이제 실행을 해주고, load checkpoint나 training resume에 다 False를 처리해주면 처음부터 학습이 시작된다. 

이런 식으로 학습이 되는데, 만약에 뭔가...학습이...이상하다ㅏㅏ? 하면 choose_action을 아래와 같이 수정해주자. 단 evaluate를 하고 싶을 때는 다시 돌려줘야한다 ㅎ.. 해결되면 깃헙에 업데이트 하도록 하겠다.

    def choose_action(self, observation, evaluate=False):
        if np.random.random() < self.epsilon:# and evaluate is False:
            action = np.random.choice(self.action_space)
        else:
            state = np.array([observation])
            actions = self.q_eval.advantage(state)
            action = tf.math.argmax(actions, axis=1).numpy()[0]

        return action

 

이렇게 하고 결과를 보면

initial state:  [80]    - action: 1 | state: [79] | reward: -50
- action: 1 | state: [78] | reward: -50
- action: 1 | state: [77] | reward: -50
- action: 1 | state: [76] | reward: -50
- action: 1 | state: [75] | reward: -50
- action: 1 | state: [74] | reward: -50
- action: 1 | state: [73] | reward: -50
- action: 1 | state: [72] | reward: -50
- action: 1 | state: [71] | reward: -50
- action: 1 | state: [70] | reward: -50
- action: 1 | state: [69] | reward: -50
- action: 1 | state: [68] | reward: -50
- action: 1 | state: [67] | reward: -50
- action: 1 | state: [66] | reward: -50
- action: 0 | state: [64] | reward: -50
- action: 0 | state: [62] | reward: -50
- action: 0 | state: [60] | reward: -50
- action: 0 | state: [58] | reward: -50

...

- action: 2 | state: [24] | reward: 100
- action: 2 | state: [24] | reward: 100
- action: 2 | state: [24] | reward: 100
- action: 2 | state: [24] | reward: 100
- action: 2 | state: [24] | reward: 100
- action: 2 | state: [24] | reward: 100
- action: 2 | state: [24] | reward: 100
- action: 2 | state: [24] | reward: 100
- action: 2 | state: [24] | reward: 100
- action: 2 | state: [24] | reward: 100
- action: 2 | state: [24] | reward: 100
- action: 2 | state: [24] | reward: 100
- action: 2 | state: [24] | reward: 100
- action: 2 | state: [24] | reward: 100
- action: 2 | state: [24] | reward: 100
- action: 2 | state: [24] | reward: 100
- action: 2 | state: [24] | reward: 100
- action: 2 | state: [24] | reward: 100
| episode:  0   | score: 4900.00        | average score 4900.00
 - last state:  [24]    | reward:  100  | action:  2    | epsilon:  0.999

요로코롬 잘 끝나는 것을 볼 수 있었다. 

끝!!!!!!!!!!!! 하다가 문제가 발생했거나 설명 중 틀린 부분이 있다면 얼마든지 댓글 부탁드립니다(_ _)