✱ 핸즈온 머신러닝 18장 강화 학습 ✱

18.1 보상을 최적화하기 위한 학습

강화 학습에서 소프트웨어 에이전트는 관측을 하고 주어진 환경에서 행동을 한다. 

그리고 결과에 따라 환경으로부터 보상을 받는다. 

에이전트의 목적은 보상의 장기간 기대치를 최대로 만드는 행동을 학습하는 것이다.

양의 보상은 기쁨으로, 음의 보상은 아픔으로 생각할 수 있다.

---

18.2 정책 탐색

정책 : 소프트웨어 에이전트가 행동을 결정하기 위해 사용하는 알고리즘이다. 

예) 관측을 입력으로 받고 수행할 행동을 출력하는 신경망

정책은 모든 알고리즘이 될 수 있으며 결정적일 필요는 없다. 어떤 경우에는 환경을 관측할 필요도 없다.

무작위성이 포함되어 있는 정책을 확률적 정책이라고 한다. 

 

예) 30분 동안 수집한 먼지의 양을 보상으로 받는 로봇 진공청소기가 있다.

여기에는 변경이 가능한 두 개의 정책 파라미터인 확률 p와 각도의 범위 r이 있다. 

이 파라미터에 많은 다른 값을 시도해보고 가장 성능이 좋은 조합을 고르는 학습 알고리즘을 사용할 수 있다.

하지만 정책 공간이 매우 큰 경우에 좋은 파라미터 조합을 찾는 데에 시간이 오래 걸린다. 

 

다른 방법으로 유전 알고리즘이 있다. 

예) 1세대 정책 100개를 랜덤하게 생성해서 시도해본 뒤, 성능이 낮은 정책 80개는 버리고 20개를 살려 각각 자식 정책 4개를 생산한다.

이 자식 정책은 부모를 복사한 것이 약간의 무작위성을 더한 것이다.

살아남은 정책과 그 자식은 2세대를 구성하며, 이런 식으로 좋은 정책을 찾을 때까지 여러 세대에 걸쳐 반복한다.

또 다른 방법으로 정책 파라미터에 대한 보상의 그레이디언트를 평가해서 높은 보상의 방향을 따르는 그레이디언트로 파라미터를 수정하는 최적화 기법이 있다. (정책 그레이디언트(PG))

예) 로봇청소기의 예에서 p를 조금 증가시켜서 30분 동안 로봇이 수집한 먼지 양이 증가했는지 평가할 수 있다.

먼지 양이 많아졌다면 P를 조금 증가시키고 그렇지 않다면 p를 감소시킨다.

PG 알고리즘은 텐서플로를 사용해 구현하지만 그전에 에이전트가 활동할 환경을 만들어야 한다.

---

18.3 OpenAI Gym

강화 학습에서 어려운 점은 에이전트를 훈련하기 위해 먼저 작업 환경을 마련해야 한다는 것이다.

실제 세상에서 훈련하는 것은 어렵고 느리기에 훈련을 위한 최소한의 시뮬레이션 환경이 필요하다.

 

OpenAI Gym은 다양한 종류의 시뮬레이션 환경(아타리 게임, 보드 게임, 2D와 3D 물리 시뮬레이션 등)을 제공하는 툴깃으로, 이를 사용하여 에이전트를 훈련하고 이들을 비교하거나 새로운 RL 알고리즘을 개발할 수 있다.

 

%pip install -q -U gymnasium
%pip install swig
%pip install -q -U gymnasium[classic_control, box2d, atari, accept-rom-license]

첫 번째 %pip 명령은 Gym을 최신 버전으로 업그레이드한다.

-q 옵션은 조용히 설치하는 것을 의미하며 장황한 출력을 만들지 않고, -U 옵션은 업그레이드를 의미한다.

두 번째 %pip 명령은 다양한 종류의 환경을 실행하는 데 필요한 라이브러리를 설치한다. 카트에서 막대의 균형을 잡은 것과 같은 고전적인 제어 이론의 환경이 포함된다.

import gymnasium as gym
 
env = gym.make("CartPole-v1", render_mode="rgb_array")

Gym을 임포트하고 환경을 만든다.

CartPole 환경을 만든다. 카트 위에 놓인 막대가 넘어지지 않도록 왼쪽이나 오른쪽으로 가속할 수 있는 2D 시뮬레이션이다.

obs, info = env.reset(seed=42)
obs
# array([ 0.0273956, -0.00611216, 0.03585979, 0.0197368 ], dtype=float32)
info
# {}

환경을 만든 후 reset() 메서드로 초기화해야 한다. 이 메서드는 첫 번째 관측을 반환한다. 관측은 환경의 종류에 따라 다르다.

CartPole 환경의 경우 각 관측은 네 개의 실수를 담은 1D 넘파이 배열이다.

이 실수는 카트의 수평 위치(0.0 = 중앙), 카트의 속도(양수는 우측 방향을 의미), 막대의 각도(0.0 = 수직), 막대의 각속도(양수는 시계 방향을 의미)를 나타낸다.

reset() 메서드는 추가적으로 환경에 관련된 정보를 담은 딕셔너리도 반환한다.

CartPole 환경

img = env.render()
img.shape # 높이, 너비, 채널
# (400, 600, 3)

render() 메서드를 호출해 이 환경을 이미지로 렌더링한다.

이 환경을 만들 떄 render_mode="rgb_array"로 지정했기 때문에 이미지는 넘파이 배열로 반환된다.

env.action_space
# Discrete(2)

맷플롯립의 imshow() 함수를 사용해 이미지를 화면에 그릴 수 있다.

Discrete(2)는 가능한 행동이 정수 0과 1이라는 것을 의미한다. (왼쪽 가속, 오른쪽 가속)

다른 환경은 더 많은 개별적인 행동을 가지거나 다른 종류의 행동(예. 연속적인)을 가질 수 있다.

action = 1 # 오른쪽으로 가속
obs, reward, done, truncated, info = env.step(action)
obs
# array([ 0.02727336, 0.18847767, 0.03625453, -0.26141977], dtype=float32)
reward
# 1.0
done
# False
truncated
# False
info
# {}

막대가 오른쪽으로 기울어져 있기 때문에 카트를 오른쪽으로 가속할 수 있다.

step() 메서드는 주어진 행동을 실행하고 네 가지 값을 반환한다.

 

obs : 새로운 관측값이다. 이제 카트가 오른쪽 방향으로 움직인다. (obs[1] > 0). 막대가 여전히 오른쪽 방향으로 기울어져 있지만 (obs[2] > 0) 각속도가 음수가 되었으므로 (obs[3] < 0) 다음 스텝 후에는 왼쪽으로 기울어질 가능성이 크다.

reward : 이 환경에서는 어떤 행동을 실행해도 매 스텝마다 1.0의 보상을 받는다. 그러므로 시스템의 목적은 가능한 한 오랫동안 실행하는 것이다.

doen : 이 값이 True이면 이 에피소드가 끝난 것이다. 막대가 너무 기울어지거나 화면 밖으로 나가거나 200 스텝을 넘기면 에피소드가 끝난다. (마지막의 경우 에이전트가 이긴 것이다.). 에피소드가 끝나면 환경을 다시 사용하기 전에 꼭 초기화해야 한다.

trundated : 이 값은 에피소드가 조기에 중단되는 경우 True가 된다. 일부 RL 알고리즘은 중단된 에피소드를 정상적으로 완료된 에피소드와 다르게 처리하지만 이 장에서는 동일하게 처리한다.

info : reset() 메서드가 반환하는 값처럼 환경에 관련된 추가 정보를 담은 딕셔너리이다. 

 

def basic_policy(obs)::
    angle = obs[2]
    return 0 if angle < 0 else 1
    
totals = []
for episode in range(500):
    episode_rewards = 0
    obs, info = env.reset(seed=episode)
    for step in range(200):
        action = basic_policy(obs)
        obs, reward, done, truncated, info = env.step(action)
        episode_rewards += reward
        if done or truncated:
            break
    
    totals.append(episode_rewards)

이 정책은 막대가 왼쪽으로 기울어지면 카트를 왼쪽으로 가속하고 오른쪽으로 기울어지면 오른쪽으로 가속한다.

import numpy as np
np.mean(totals), np.std(totals), min(totals), max(totals)
# (41.698, 8.389445512070509, 24.0, 63.0)

---

18.4 신경망 정책

신경망은 관측을 입력으로 받고 실행할 행동을 출력한다.

각 행동에 대한 확률을 추정하고 추정된 확률에 따라 랜덤하게 행동을 선택한다.

랜덤하게 행동을 선택하면 에이전트가 새로운 행동을 탐험하는 것과 잘 할 수 있는 행동을 활용하는 것 사이에 균형을 맞추게 된다. 

이런 특별한 환경에서는 과거의 행동과 관측을 무시할 수 있다. 

각 관측이 환경에 대한 완전한 상태를 담고 있기 때문이다. 

만약 어떤 상태가 숨겨져 있을 때 과거의 행동과 관측도 고려해야 한다. 

관측에 잡음이 있을 때는 가장 가능성 있는 현재의 상태를 추정하기 위해 지난 관측 몇 개를 사용하는 것이 좋다.

 

import tensorflow as tf

model = tf.keras.Sequential([
    tf.keras.layers.Dense(5, activation="relu"),
    tf.keras.layers.Dense(1, activation="sigmoid"),
])

Sequential 모델을 사용해 정책 네트워크를 저의한다.

입력의 개수는 관측 공간의 크기이다. 이 경우는 4이며, 간단한 문제이므로 은닉 유닛 5개를 사용한다.

마지막으로 하나의 확률(왼쪽 방향일 확률)이 필요하므로 시그모이드 활성화 함수를 사용한 하나의 출력 뉴런을 둔다.

만약 가능한 행동이 두 개보다 많으면 행동마다 하나의 출력 뉴런을 두고 소프트맥스 활성화 함수를 사용해야 한다.

---

18.5 행동 평가: 신용 할당 문제

각 스텝에서 가장 좋은 행동이 무엇인지 알고 있다면 평소처럼 추정된 확률과 타깃 확률 사이의 크로스 엔트로피를 최소화하도록 신경망을 훈련할 수 있다. 

일반적인 지도 학습과 같으나 강화 학습에서 에이전트가 얻을 수 있는 가이드는 보상뿐이다.

보상은 일반적으로 드물고 지연되어 나타난다. 

예) 에이전트가 100 스텝 동안 막대의 균형을 유지했을 때 우리가 아는 것은 마지막 행동 뒤에 막대가 쓰려졌다는 것뿐이다. 

이 100번의 행동 중 어느 것이 좋고, 어느 것이 나쁜지 알 수 없다.

신용 할당 문제 : 하지만 모든 책임이 마지막 행동에 있는 것은 아니다. 

에이전트가 보상을 받았을 때 어떤 행동 덕분인지 알기 어렵다.

 

→ 행동이 일어난 후 각 단계마다 할인 계수 γ를 적용한 보상을 모두 합하여 행동을 평가하는 방법을 사용할 수 있다.

할인된 보상의 합을 행동의 대가라고 한다. 

할인 계수가 0에 가까우면 미래의 보상은 현재의 보상만큼 중요하게 취급되지 않을 것이다.

반대로 할인 계수가 1에 가까우면 먼 미래의 보상이 현재의 보상만큼 중요하게 고려될 것이다.

전형적인 할인 계수의 값은 0.9에서 0.99사이이다.

행동 이익 : 우리는 평균적으로 다른 가능한 행동과 비교해서 각 행동이 얼마나 좋은지 나쁜지를 추정해야 한다.

많은 에피소드를 실행하고 모든 행동의 대가를 정규화해야 한다.

행동 이익이 음수인 행동은 나쁘고, 양수인 행동은 좋다고 가정할 수 있다.

---

18.6 정책 그레이디언트

높은 보상을 얻는 방향의 그레이디언트를 따르도록 정책의 파라미터를 최적화하는 알고리즘이다.

 

REINFORCEMENT 알고리즘이 인기 있다.

1. 신경망 정책이 여러 번에 걸쳐 게임을 플레이하고 매 스텝마다 선택된 행동이 더 높은 가능성을 가지도록 만드는 그레이디언트를 계산한다. 하지만 아직 이 그레이디언트르 적용하지는 않는다.

2. 에피소드를 몇 번 실행한 다음, 각 행동의 이익을 게산한다.

3. 한 행동의 이익이 양수이면 이 행동이 좋은 것임을 의미하므로 미래에 선택될 가능성이 높도록 앞서 계산한 그레이디언트를 적용한다. 그러나 행동 이익이 음수이면 이 행동이 나쁜 것임을 의미하므로 미래에 이 행동이 덜 선택되도록 반대의 그레이디언트를 적용한다. 이는 각 그레이디언트 벡터와 그에 상응하는 행동의 이익을 곱하면 된다.

4. 마지막으로 모든 결과 그레이디언트 벡터를 평균 내어 경사 하강법 스텝을 수행한다. 

 

def play_one_step(env, obs, model, loss_fn):
    with tf.GradientTape() as tape:
        left_prova = model(obs[np.newaxis])
        action = (tf.random.uniform([1, 1]) > left_proba)
        y_target = tf.constant([[1.]]) - tf.cast(action, tf.float32)
        loss = tf.reduce_mean(loss_fn(y_target, left_proba))
        
    grads = tape.gradient(loss, model.trainable_variables)
    obs, reward, done, truncated, info = env.step(int(action))
    return obs, reward, done, truncated, grads

먼저 한 스텝을 진행할 함수가 필요하다.

어떤 행동을 선택하더라도 손실과 그레이디언트를 계산하기 위해 옳은 선택이라고 가정한다. 

(이 그레이디언트를 잠시 동안 일단 저장했다가 이 행동이 좋은지 나쁜지 판명된 후에 조정한다.)

 

GradientTape 블록 안에서 하나의 관측과 함께 모델을 출력한다. 모델은 배치를 기대하므로 하나의 샘플이 들어 있는 배치가 되도록 관측의 크기를 바꾼다. 이 모델은 왼쪽으로 이동할 확률을 출력한다.

0에서 1 사이의 랜덤한 실수를 샘플링하고 이 값이 left_proba보다 큰지 확인한다. action은 left_proba 확률로 False가 되고 1 - left_proba 확률로 True가 된다. 이 불리언 값을 정수로 변환하면 action은 출력된 확률에 맞게 0(왼쪽) 또는 1(오른쪽)이 된다.

왼쪽으로 이동할 타깃 확률을 정의한다. 이 값은 1 - (실수로 변환된) 행동이다. 행동이 0(왼쪽)이면 왼쪽으로 이동할 타깃 확률은 1이 되며, 행동이 1(오른쪽)이면 타깃 확률이 0이 된다.

주어진 손실 함수를 사용해 손실을 계산하고 테이프를 사용해 모델의 훈련 가능한 변수에 대한 손실의 그레이디언트를 계산한다. 이 그레이디언트도 나중에 적용하기 전에 이 행동이 좋은지 나쁜지에 따라 조정될 것이다.

선택한 행동을 플레이하고 새로운 관측, 보상, 에피소드 종료 여부, 에피소드 중단 여부, 계산한 그레이디언트를 반환한다.

 

def play_multiple_episodes(env, n_episodes, n_max_steps, model, loss_fn):
    all_rewards = []
    all_grads = []
    for episode in range(n_episodes):
        current_rewards = []
        current_grads = []
        obs, info = env.reset()
        for step in range(n_max_steps):
            obs, reward, done, truncated, grads = play_one_step(
                env, obs, model, loss_fn)
            current_rewards.append(reward)
            current_grads.append(grads)
            if done or truncated:
                break
        
        all_rewards.append(current_rewards)
        all_grads.append(current_grads)
    
    return all_rewards, all_grad

play_one_step() 함수를 사용해 여러 에피소드를 플레이하고, 전체 보상 및 각 에피소드와 스텝의 그레이디언트를 반환하는 함수이다. 

보상 리스트의 리스트(에피소드마다 보상 리스트 하나, 이 리스트는 스텝마다 보상 하나를 포함한다.)와 그레이디언트 리스트의 리스트(에피소드마다 그레이디언트 리스트 하나, 이 리스트는 스텝마다 그레이디언트 튜플을 하나 포함하고 각 튜플은 훈련 가능한 변수마다 그레이디언트 텐서 하나를 포함한다.)를 반환한다.

def discount_rewards(rewards, discount_factor):
    discounted = np.array(rewards)
    for step in range(len(rewards) -2, -1, -1):
        discounted[step] += discounted[step + 1] * discount_factor
    return discounted
    
def discount_and_normalize_rewards(all_rewards, discount_factor):
    all_discounted_rewards = [discount_rewards(rewards, discount_factor)
                              for rewards in all_rewards]
    flat_rewards = np.concatenate(all_discounted_rewards)
    reward_mean = flat_rewards.mean()
    reward_std = flat_rewards.std()
    return [(discounted_rewards - reward_nean) / reward_std
            for discounted_rewards in all_discounted_rewards]

이 알고리즘은 play_multiple_episodes() 함수를 사용하여 여러 번 게임을 플레이한다.

그다음 처음부터 모든 보상을 살펴서 각 보상을 할인하고 정규화한다.

첫 번째 함수는 각 스텝에서 할인된 미래 보상의 합을 계산한다.

두 번째 함수는 여러 에피소드에 걸쳐 계산된 할인된 모든 보상(대가)에서 평균을 빼고 표준 편차로 나누어 졍규화한다.

n_iterations = 150
n_episodes_per_update = 10
n_max_steps = 200
discount_factor = 0.95

하이퍼파라미터를 정의한다.

optimizer = tf.keras.optimizers.Nadam(learning_rate=0.01)
loss_fn = tf.keras.losses.binary_crossentropy

옵티마이저와 손실 함수도 필요하다.

for iteration in range(n_iterations):
    all_rewards, all_grads = play_multiple_episodes(
        env, n_episodes_per_update, n_max_steps, model, loss_fn)
    all_final_rewards = discount_and_normalize_rewards(all_rewards, discount_factor)
    all_mean_grads = []
    for var_index in range(len(model, trainable_variables)):
        mean_grads = tf.reduce_mean(
            [final_reward * all_grads[episode_index][step][var_index]
            for episode_index, final_rewards in enumerate(all_final_rewards)
                for step, final_reward in enumerate(final_rewards)], axis=0)
        all_mean_grads.append(mean_grads)
    optimizer.apply_gradients(zip(all_mean_grads, model.trainable_variables))

훈련 반복을 만들어 실행할 준비를 한다. 

 

각 훈련 반복에서 play_multiple_episodes() 함수를 호출한다. 

discount_and_normalize_rewards() 함수를 호출하여 각 행동의 정규화된 이익(final_reward)를 계산한다. 이 값은 각 행동이 실제로 얼마나 좋은지 나쁜지를 알려준다.

훈련 가능한 변수를 순회하면서 모든 에피소드와 모든 스텝에 대한 각 변수의 그레이디언트를 final_reward로 가중치를 두어 평균한다.

평균 그레이디언트를 옵티마이저에 적용한다. 모델의 훈련 가능한 변수가 변경되고 아마 정책이 조금 더 나아질 것이다. 

 

크고 복잡한 문제에는 잘 적용하지 못한다.

아주 긴 시간 동안 게임을 플레이해야 정책을 많이 개선할 수 있으므로 샘플 효율성이 매우 좋지 못하다. 

각 행동의 이익을 추정하기 위해 많은 에피소드를 실행해야 한다.

액터-크리틱 알고리즘 같은 더 강력한 알고리즘도 있다. 

---

18.7 마르코프 결정 과정

마르코프 연쇄 : 정해진 개수의 상태를 가지고 있으며, 각 스텝마다 한 상태에서 다른 상태로 랜덤하게 전이된다.

상태 s에서 상태 s'로 전이하기 위한 확률은 고정되어 있으며, 시스템에 메모리가 없으므로 과거 상태에 상관없이 (s, s') 쌍에만 의존한다.

나오는 길이 없어서 영원히 그 상태에 남게 되는 상태를 종료 상태라고 한다. 

 

마르코프 결정 과정 : 마르코프 연쇄와 비슷하지만 약간 다르다.

각 스텝에서 에이전트는 여러 가능한 행동 중 하나를 선택할 수 있고, 전이 확률은 선택된 행동에 따라 달라진다.

또한 어떤 상태 전이는 보상(음수 혹은 양수)을 반환한다. 

에이전트의 목적은 시간이 지남에 따라 보상을 최대화하기 위한 정책을 찾는 것이다.

 

최적의 상태 가치 V*(s) : 에이전트가 상태 s에 도달한 후 최적으로 행동한다고 가정하고 평균적으로 기대할 수 있는 할인된 미래 보상의 합이다.

에이전트가 최적으로 행동하면 벨만 최적 방정식이 적용된다는 것을 입증했다. 

이 재귀 식은 에이전트가 최적으로 행동하면 현재 상태의 최적 가치는 하나의 최적 행동으로 인해 평균적으로 받게 될 보상과 이 행동이 유발할 수 있는 가능한 모든 다음 상태의 최적 가치의 기대치를 합한 것과 같다는 것을 의미한다.

알고리즘이 가능한 모든 상태에 대한 최적의 상태 가치를 정확히 추정할 수 있도록 도와준다. 

먼저 모든 상태 가치를 0으로 초기화한다. 그런 다음 가치 반복 알고리즘을 사용하여 반복적으로 업데이트한다.

[가치 반복 알고리즘]

V_k(s)는 알고리즘의 k번째 반복에서 상태 s의 추정 가치이다.

충분한 시간이 주어지면 이 추정값이 최적의 정책에 대응하는 최적의 상태 가치에 수렴하는 것이 보장된다.

 

최적의 상태 가치를 아는 것은 특히 정책을 평가할 때 유용하다. 하지만 에이전트를 위한 최적의 정책을 알려주지는 않는다. 

Q-가치라고 부르는 최적의 상태-행동 가치를 추정할 수 있는 매우 비슷한 알고리즘이 발견되었다. 

상태-행동 (s, a) 쌍에 대한 최적의 Q-가치인 Q'(s, a)는 에이전트가 상태 s에 도달해서 행동 a를 선택한 후 이 행동의 결과를 얻기 전에 평균적으로 기대할 수 있는 할인된 미래 보상의 합이다. 

에이전트는 여기서 이 행동 이후에 최적으로 행동할 것이라고 가정한다.

 

Q-가치의 추정을 모두 0으로 초기화한다.

Q-가치 반복 알고리즘을 사용해 업데이트한다.

[Q-가치 반복 알고리즘]

 

최적의 Q-가치를 구하고 나면 최적의 정책인 π*(s)를 정의하는 것은 간단하다.

에이전트가 상태 s에 도달했을 때 가장 높은 Q-가치를 가진 행동을 선택하면 된다. 

 

transition_probabilities = [ # 크기는 [s, a, s']
    [[0.7, 0.3, 0.0], [1.0, 0.0, 0.01, [0.8, 0.2, 0.0]],
    [[0.0, 1.0, 0.0], None, [0.0, 0.0, 1.0]],
    [None, [0.8, 0.1, 0.1], None]
]
rewards = [ # 크기는 [s, a, s']
    [[+10, 0, 0], [O, 0, 0], [0, 0, 0]],
    [[O, 0, 0], [0, 0, 0], [0, 0, -50]],
    [[O, 0, 01, [+40, 0, 0], [0, 0, 0]]
]
possible_actions = [[O, 1, 2], [0, 2], [1]]

예) 행동 a_1을 플레이한 후 s_2에서 s_0으로 전이할 확률을 알기 위해서는 transition_probabilities[2][1][0]을 참조한다.

이해 해당하는 보상을 얻으려면 rewards[2][1][0]을 참조한다.

s_2에서 가능한 행동의 리스트를 얻을면 possible_actions[2]를 참조한다.

Q_values = np.full((3, 3), -np.inf) # 불가능한 행동에 대해서는 -np.inf
for state, actions in enumerate(possible_actions):
    Q_values[state, actions] = 0.0 # 모든 가능한 행동에 대해서

그다음 모든 Q-가치를 0으로 초기화해야 한다.

gamma = 0.90 # 할인 계수

for iteration in range(50):
    Q_prev = Q_values.copy()
    for s in range(3):
        for a in possible_actions[s]:
            Q_values[s, a] = np.sum([
                transition_probabilities[s][a][sp]
                * (rewards[s][a][sp] + gamma * Q_prev[sp].max())
            for sp in range(3)])

모든 상태와 모든 가능한 행동에 대해 모든 Q-가치에 Q-가치 반복 알고리즘을 반복적으로 적용한다. 

Q_values
# array ([[18.91891892, 17.02702702, 13.62162162],
#         [ 0.        ,        -inf, -4.87971488],
#         [ -inf      , 50.13365013,        -inf]])

예) 에이전트가 상태 s_0에 있고 행동 a_1을 선택했을 때 할인된 미래 보상의 기대 합은 약 17.0이다.

np.argmax(Q_values, axis=1) # 각 상태에 대해 최적의 행동
# array([0, 0, 1])

각 상태에 대해 가장 높은 Q-가치를 갖는 행동이다.

---

18.8 시간차 학습

독립적인 행동으로 이루어진 강화 학습 문제는 보통 마르코프 결정 과정으로 모델링될 수 있지만 초기에 에이전트는 전이 확률에 대해 알지 못하며(T(s, a, s')를 모른다.), 보상이 얼마나 되는지도 알지 못한다.(R(s, a, s')를 모른다.)

보상에 대해 알기 위해서는 적어도 한 번은 각 상태와 전이를 경험해야 한다.

그리고 전이 확률에 대해 신뢰할 만한 추정을 얻으려면 여러 번 경험을 해야 한다.

 

시간차 학습(TD 학습) : Q-가치 반복 알고리즘과 매우 비슷하지만 에이전트가 MDP에 대해 일부 정보만 알고 있을 때를 다룰 수 있도록 변형한 것이다.

일반적으로 에이전트가 초기에 가능한 상태와 행동만 알고 다른 것은 모른다고 가정한다.

에이전트는 탐험 정책을 사용해 MDP를 탐험한다.

탐험이 진행될수록 TD 학습 알고리즘이 실제로 관측된 전이와 보상에 근거하여 상태 가치의 추정값을 업데이트한다.

[TD 학습 알고리즘]

이 식의 첫 번째 형태를 더 간단히 쓰는 방법은

표기법을 사용하는 것이다.

를 의미한다.

↓

각 상태 s에서 이 알고리즘은 에이전트가 해당 상태를 떠났을 때 얻을 수 있는 당장의 보상과(최적으로 행동한다고 가정하여) 나중에 기대할 수 있는 보상을 더한 이동 평균을 저장한다.

---

18.9 Q-러닝

Q-러닝 알고리즘은 전이 확률과 보상을 초기에 알지 못한 상황에서 Q-가치 반복 알고리즘을 적용한 것이다.

Q-러닝은 에이전트가 플레이하는 것을 보고 점진적으로 Q-가치 추정을 향상하는 방식으로 작동한다.

정확한(충분히 근접한) Q-가치 추정을 얻게 되면 최적의 정책은 가장 높은 Q-가치를 가지는 행동을 선택한다. (탐욕적 정책)

 

[Q-러닝 알고리즘]

각 상태-행동 (s, a) 쌍마다 이 알고리즘이 행동 a를 선택해 상태 s를 떠났을 때 에이전트가 받을 수 있는 보상 r과 기대할 수 있는 할인된 미래 보상의 합을 더한 이동 평균을 저장한다.

미래 보상의 합을 추정하기 위해서는 타깃 정책이 이후로 최적으로 행동한다고 가정하고 다음 상태 s'에 대한 Q-가치 추정의 최댓값을 선택한다. 

 

def step(state, action): 
    probas = transition_probabilities[state][action]
    next_state = np.random.choice([0, 1, 2], p=probas)
    reward = rewards[state][action][next_state]
    return next_state, reward

먼저 에이전트가 환경을 탐색하게 만들어야 한다.

이를 위해 에이전트가 한 행동을 실행하고 결과 상태와 보상을 받을 수 있는 스텝 함수가 필요하다.

def exploration_policy(state):
    return np.random.choice(possible_actions[state])

이 상태 공간은 매우 작기 때문에 단순한 랜덤 정책으로 충분하다.

알고리즘을 충분히 오랫동안 실행하면 에이전트가 각 상태를 여러 번 방문하고 가능한 모든 행동을 여러 번 실행할 수 있다.

alpha0 = 0.05 # 초기 학습률
decay = 0.005 # 학습률 감쇠
gamma = 0.90  # 할인 계수
state = 0     # 초기 상태

for iteration in range(10_000):
    action = exploration_policy(state)
    next_state, reward = step(state, action)
    next_value = Q_values[next_state].max() # 다음 스텝에서 탐욕적 정책
    alpha = alpha0 / (1 + iteration * decay)
    Q_values[state, action] *= 1 - alpha
    Q_values[state, action] += alpha * (reward + gamma * next_value)
    state = next_state

Q-가치를 초기화한 후 학습률 감쇠(거듭제곱 기반 스케줄링)를 사용해 Q-러닝 알고리즘을 실행한다.

최적의 Q-가치에 수렴하겠지만 많은 반복과 하이퍼파라미터 튜닝이 필요하다.

Q-가치 반복 알고리즘(왼쪽)과 Q-러닝 알고리즘(오른쪽)

훈련된 정책을 훈련 중에 반드시 사용하는 것은 아니다. (오프-폴리시 알고리즘)

예) 앞선 코드에서 실행된 정책(탐험 정책)은 완전히 랜덤한 정책이며 훈련된 정책은 전혀 사용되지 않는다.)

최적의 정책은 항상 가장 높은 Q-가치를 가진 행동을 선택하는 것이다.

반대로 정책 그레이디언트 알고리즘은 온-폴리시 알고리즘이다.

훈련된 정책을 사용해 환경을 탐험한다. 

18.9.1 탐험 정책

Q-러닝은 탐험 정책이 MDP를 충분히 탐험해야 작동한다.

완전한 랜덤 정책이 결국에는 모든 상태와 전이를 여러 번 경험하도록 보장하지만 이렇게 하려면 극단적으로 오랜 시간이 걸릴 수 있다.

더 나은 방법은 ε-그리디 정책을 사용하는 것이다.

각 스텝에서 ε 확률로 랜덤하게 행동하거나 1 - ε 확률로 그 순간 가장 최선인 것으로(가장 높은 Q-가치를 선택하여) 행동한다. 

완전한 랜덤 정책에 비해 Q-가정 추정이 점점 더 향상되기 때문에 환경에서 관심 있는 부분을 살피는 데 점점 더 많은 시간을 사용할 수 있다.

그럼에도 여전히 MDP의 알려지지 않은 지역을 방문하는 데 일정 시간을 사용할 것이다. 

 

이 외에도 탐험의 가능성에 의존하는 대신 이전에 많이 하지 않았던 행동을 시도하도록 탐험 정책을 강조하는 방법이 있다.

이런 방식은 Q-가치 추정에 보너스를 추가하는 방식으로 구현된다.

[탐험 함수를 사용한 Q-러닝

18.9.2 근사 Q-러닝과 심층 Q-러닝

Q-러닝의 주요 문제는 많은 상태와 행동을 가진 대규모의 MDP에 적용하기 어렵다는 것이다.

→ 어떤 상태-행동 (s, a) 쌍의 Q-가치를 근사하는 함수 Q_θ(s, a)를 적절한 개수의 파라미터를 사용하여 찾는 것이다. (근사 Q-러닝)

심층 Q-네트워크(DQN) : Q-가치를 추정하기 위해 사용하는 DNN

심층 Q-러닝 : 근사 Q-러닝을 위해 DQN을 사용하는 것

 

주어진 상태-행동 쌍 (s, a)에 대해 DQN이 계산한 근사 Q-가치를 생각해보겠다.

벨만 식 덕분이 이 근사 Q-가치는 상태 s에서 행동 a를 실행했을 때 관측된 보상 r과 그 이후에 최적으로 행동해서 얻은 할인된 가치를 더한 값에 가능한 한 가까워야 한다. 

이 미래의 할인된 가치를 추정하기 위해서는 간단하게 다음 상태 s'와 모든 가능한 행동 a'에 대해 DQN을 실행하면 된다.

그럼 모든 가능한 행동에 대한 미래의 근사 Q-가치를 얻을 수 있다. 

그다음 근사 Q-가치가 가장 높은 것을 고르고 할인을 적용하면 할인된 미래 보상의 추정을 얻을 수 있다.

보상 r과 미래의 할인된 가치 추정을 더하면 상태-행동 쌍 (s, a)에 대한 타깃 Q-가치 y(s, a)를 얻게 된다.

이 타깃 Q-가치로 경사 하강법을 사용해 훈련 단계를 수행할 수 있다.

추정된 Q-가치 Q(s, a)와 타깃 Q-가치 y(s, a) 사이의 제곱 오차를 최소화한다.

또는 알고리즘이 큰 오차에 민감하지 않도록 후버 손실을 사용한다.

---

18.10 심층 Q-러닝 구현

첫 번째로 필요한 것은 심층 Q-네트워크이다.

이론적으로는 상태-행동 쌍을 입력으로 받고 근사 Q-가치를 출력하는 신경망이 필요하다.

하지만 실전에서는 상태만 입력으로 받고 가능한 모든 행동에 대한 근사 Q-가치를 각각 출력하는 것이 훨씬 효율적이다.

input_shape = [4] # == env.observation_space.shape
n_outputs = 2 # == env.action_space.n

model = tf.keras.Sequential([
    tf.keras.layers.Dense(32, activation="elu", input_shape=input_shape),
    tf.keras.layers.dense(32, activation="elu")
    tf.keras.layers.Dense(n_outputs)
])

def epsilon_greedy_policy(state, epsilon=0):
    if np.random.rand() < epsilon:
        return np.random.randint(n_outputs) # 랜덤 행동
    else:
        Q_values = model.predict(state[np.newaxis], verbose=0)[0]
        return Q_values.argmax() # DQN에 따른 최적의 행동

이 DQN으로 행동을 선택하려면 예측 Q-가치가 가장 큰 행동을 선택한다.

에이전트가 환경을 탐험하도록 만들기 위해 ε-그리디 정책을 사용한다. (즉, 확률 ε만큼 랜덤한 행동을 선택한다.)

from collections import deque

replay_buffer = deque(maxlen=2000)

최근의 경험에만 의지하여 DQN을 훈련하는 대신 재생 버퍼(재생 메모리)에 모든 경험을 저장하고 훈련 반복마다 여기에서 랜덤한 훈련 배치를 샘플링할 수 있다. 경험과 훈련 배치 사이의 상관관계가 줄어들어 훈련에 도움이 된다.

이를 위해 덱을 사용한다.

def sample_experiences(batch_size):
    indices = np.random.randint(len(replay_buffer), size=batch_size)
    batch = [replay_buffer[index] for index in indices]
    return [
        np.array([experience[field_index] for experience in batch])
        for field_index in range(6)
    ] # [states, actions, rewards, next_states, dones, truncateds]

각 경험은 원소 6개로 구성된다.

상태 s, 에이전트가 선택한 행동 a, 결과 보상 r, 도달한 다음 상태 s', 에피소드가 이때 종료되었는지 여부를 나타내는 불리언 값(done), 마지막으로 에피소드가 중단되었는지 여부를 나타내는 불리언 값(truncated)이다.

재생 버퍼에서 경험을 랜덤하게 샘프링하기 위해 작은 함수를 만든다.

이 함수는 경험 원소 6개에 상응하는 넘파이 배열 6개를 반환한다.

def play_one_step(env, state, epsilon):
    action = epsilon_greedy_policy(state, epsilon)
    next_state, reward, done, truncated, info = env.step(action)
    replay_buffer.append((state, action, reward, next_state, done, truncated))
    return next_state, reward, done, truncated, info

ε-그리디 정책을 사용해 하나의 스텝을 플레이하고 반환된 경험을 재생 버퍼에 저장한다. 

batch_size = 32
discount_factor = 0.95
optimizer = tf.keras.optimizers.Nadam(learning_rate=1e-2)
loss_fn = tf.keras.losses.mean_squared_error

def training_step(batch_size):
    experiences = sample_experiences(batch_size)
    states, actions, rewards, next_states, dones, truncateds = experiences
    next_Q_values = next_Q_values.max(axis=0)
    max_next_Q_values = next_Q_values.max(axis=1)
    runs = 1.0 - (dones | truncateds) # 에피소드가 중지되거나 종료되지 않음
    target_Q_values = rewards + runs * discount_factor * max_next_Q_values
    target_Q_values = target_Q_values.reshape(-1, 1)
    nask = tf.one_hot(actions, n_outputs)
    with tf.GradientTape() as tape:
        all_Q_values = model(states)
        Q_values = tf.reduce_sum(all_Q_values * mask, axis=1, keepdims=True)
        loss = tf.reduce_mean(loss_fn(target_Q_values, Q_values))
        
    grads = tape.gradient(loss, model.trainable_variables)
    optimizer.apply_gradients(zip(grads, model.trainable_variables))

재생 버퍼에서 경험 배치를 샘플링하고 이 배치에서 경사 하강법 한 스텝을 수행하여 DQN을 훈련하는 함수이다. 

 

하이퍼파라미터 몇 개를 정의하고 옵티마이저와 손실 함수를 만든다.

training_step() 함수를 만든다. 이 함수는 경험 배치를 샘플링한 다음 DQN을 사용하여 각 경험의 다음 상태에서 가능한 모든 행동에 대한 Q-가치를 예측한다. 에이전트가 최적으로 플레이한다고 가정하므로 다음 상태에 대한 최대 Q-가치만 저장한다. 그다음 타깃 Q-가치 식을 사용해 각 경험의 상태-행동 쌍에 대한 타깃 Q-가치를 계산한다.

DQN이 경험한 각 상태-행동 쌍의 Q-가치를 계산하길 원한다. 하지만 이 DQN은 에이전트가 실제로 선택한 행동뿐만 아니라 다른 가능한 행동에 대한 Q-가치도 출력할 것이다. 따라서 필요하지 않은 모든 Q-가치를 마스크 처리해야 한다. tf.one_hot() 함수는 행동 인덱스의 배열을 마스크로 변환해준다. 예) 처음 3개의 경험이 행동 1, 1, 0을 각각 담고 있다면 마스크는 [[0, 1], [0, 1], [1, 0], ...]과 같다. 이 마스크를 DQN의 출력과 곱하여 원하지 않은 Q-가치를 0으로 만들 수 있다. 그다음 0을 없애기 위해 열(axis=1) 방향으로 덧셈하여 경험된 상태-행동 쌍의 Q-가치만 남긴다. 결국 배치에 있는 각 경험에 대해 예측된 Q-가치 하나를 담은 텐서인 Q-values를 얻는다.

손실을 계산한다. 손실은 경험된 상태-행동 쌍에 대한 타깃과 예측된 Q-가치 사이의 평균 제곱 오차이다.

모델의 훈련 가능한 변수에 관한 손실을 최소화하기 위해 경사 하강법을 수행한다.

for episode in range(600):
    obs, info = env.reset)_
    
for step in range(200):
    epsilon = max(1 - episode / 500, 0.01)
    obs, reward, done, truncated, info = play_one_step(env, obs, epsilon)
    if done or truncated:
        break
        
if episode > 50:
    training_step(batch_size)

최대 스텝 200번으로 이루어진 에피소드 600개를 실행한다.

각 스텝에서 먼저 ε-그리디 정책에 대한 epsilon 값을 계산한다.

이 값은 500 에피소드 직전까지 1에서 0.01로 선형적으로 줄어든다.

그다음 play_one_step() 함수를 호출한다. 이 함수는 ε-그리디 정책을 사용해 행동을 선택하여 실행하고 그 경험을 재생 버퍼에 기록한다.

에피소드가 종료되거나 중단되면 반복을 끝낸다.

마지막으로 50 에피소드 이후에는 training_step() 함수를 호출해 재생 버퍼에서 샘플링한 배치로 모델을 훈련한다.

훈련 없이 에피소드를 50번 플레이하는 이유는 재생 버퍼가 채워질 시간을 주기 위해서이다.

최대 보상 근처에서 안정된 것처럼 보였으나 점수가 급격히 떨어지는 걸 최악의 망각이라고 부른다. 

에이전트가 환경을 탐색하면 정책을 업데이트한다. 하지만 환경의 한 부분에서 학습한 것이 앞서 다른 부분에서 학습한 것을 망가뜨릴 수 있다. 경험은 크게 연관되며 학습 환경은 계속 바뀐다. 

---

18.11 심층 Q-러닝의 변형

18.11.1 고정 Q-가치 타깃

기본 심층 Q-러닝 알고리즘에서 모델은 예측을 만들고 타깃을 설정하는 데 모두 사용된다.

이런 피드백 순환 과정은 네트워크를 불안정하게 만들어 발산, 진동, 동결 등의 문제가 생긴다.

→ 한 개가 아닌 두 개의 DQN을 사용한다.

첫 번째 DQN은 각 스텝에서 학습하고 에이전트를 움직이는 데 사용하는 온라인 모델이다. 

두 번째 DQN은 타깃을 정의하기 위해서만 사용하는 타깃 모델이며, 이는 온라인 모델의 단순한 복사본이다. 

target = tf.keras.models.clone_model(model) # 모델 구조 복사
target.set_weights(model.get_weights()) # 가중치 복사

next_Q_values = target.predict(next_states, verbose=0)

그다음 training_step() 함수에서 다음 상태의 Q-가치를 계산할 때 온라인 모델 대신 타깃 모델을 사용하도록 한 줄을 바꾸어야 한다.

if episode % 50 == 0:
    target.set_weights(model.get_weights())

마지막으로 훈련 반복에서 일정한 간격으로 온라인 모델의 가중치를 타깃 모델로 복사해야 한다.

 

타깃 모델은 온라인 모델보다 자주 업데이트되지 않으므로 Q-가치 타깃이 더 안정적이며 피드백 반복을 완화하고 이에 대한 영향이 감소된다. 

18.11.2 더블 DQN

타깃 네트워크가 Q-가치를 과대평가하기 쉽다는 관측을 기반으로 한다.

모든 행동이 동일하게 좋다고 가정한다.

타깃 모델이 추정한 Q-가치가 동일해야 하지만 근삿값이기 때문에 우연히 다른 것보다 조금 높은 값이 있을 것이다.

타깃 모델은 항상 가장 큰 Q-가치를 선택하므로 평균 Q-가치보다 조금 더 커지고 실제 Q-가치를 과대평가할 가능성이 높다.

→ 다음 상태에서 최선의 행동을 선택할 때 타깃 모델 대신 온라인 모델을 사용하도록 제안되었다.

타깃 모델은 최선의 행동에 대한 Q-가치를 추정할 때만 사용한다.

def training_step(batch_size):
    experiences = sample_experiences(batch_size)
    states, actions, rewards, next_states, dones, truncateds = experiences
    next_Q_value = model.predict(next_states, vervose=0)
    best_next_actions = next_Q_values.argmax(axis=1)
    next_mask = tf.one_hot(best_next_actions, n_outputs).numpy()
    max_next_Q_values = (target.predict(next_states, verbose=0) * next_mask).sum(axis=1)
    [...]

18.11.3 우선 순위 기반 경험 재생

중요도 샘플링(IS), 우선 순위 기반 경험 재생(PER) : 재생 버퍼에서 경험을 균일하게 샘플링하는 것이 아니라 중요한 경험을 더 자주 샘플링한다. 

18.11.4 듀얼링 DQN