이 페이지는 Cloud Translation API를 통해 번역되었습니다.
Switch to English

배우 비평 방법으로 CartPole 플레이

TensorFlow.org에서보기 Google Colab에서 실행 GitHub에서 소스보기 노트북 다운로드

이 자습서에서는 TensorFlow를 사용하여 Actor-Critic 메서드를 구현하여 Open AI Gym CartPole-V0 환경에서 에이전트를 교육하는 방법을 보여줍니다. 독자는 강화 학습의 정책 그라데이션 방법 에 어느 정도 익숙하다고 가정합니다.

배우 비평 방법

Actor-Critic 방법은 가치 함수와 독립적 인 정책 함수를 나타내는 TD (temporal Difference) 학습 방법입니다.

정책 함수 (또는 정책)는 에이전트가 주어진 상태에 따라 취할 수있는 작업에 대한 확률 분포를 반환합니다. 가치 함수는 주어진 상태에서 시작하여 영원히 특정 정책에 따라 행동하는 에이전트의 예상 수익을 결정합니다.

Actor-Critic 방법에서 정책은 주에 대해 가능한 일련의 가능한 조치를 제안하는 행위자 라고하며, 추정값 함수는 주어진 정책에 따라 행위자가 취한 조치를 평가하는 비평가 라고합니다. .

이 튜토리얼에서 ActorCritic 은 두 개의 출력이있는 하나의 신경망을 사용하여 표현됩니다.

CartPole-v0

CartPole-v0 환경 에서는 마찰이없는 트랙을 따라 이동하는 카트에 기둥이 부착됩니다. 장대가 똑바로 시작되고 에이전트의 목표는 카트에 -1 또는 +1의 힘을 가하여 넘어지는 것을 방지하는 것입니다. 기둥이 똑바로 유지 될 때마다 +1의 보상이 주어집니다. 에피소드는 (1) 기둥이 수직에서 15도 이상 떨어져 있거나 (2) 수레가 중앙에서 2.4 단위 이상 이동하면 끝납니다.

Cartpole-v0 환경에서 훈련 된 행위자 비판 모델

이 문제는 에피소드에 대한 평균 총 보상이 100 회 연속 시도에서 195에 도달하면 "해결 된"것으로 간주됩니다.

설정

필요한 패키지를 가져오고 전역 설정을 구성합니다.

pip install -q gym
WARNING: You are using pip version 20.2.2; however, version 20.2.3 is available.
You should consider upgrading via the '/tmpfs/src/tf_docs_env/bin/python -m pip install --upgrade pip' command.

%%bash
# Install additional packages for visualization
sudo apt-get install -y xvfb python-opengl > /dev/null 2>&1
pip install -q pyvirtualdisplay > /dev/null 2>&1
pip install -q git+https://github.com/tensorflow/docs > /dev/null 2>&1
import collections
import gym
import numpy as np
import tensorflow as tf
import tqdm

from matplotlib import pyplot as plt
from tensorflow.keras import layers
from typing import Any, List, Sequence, Tuple


# Create the environment
env = gym.make("CartPole-v0")

# Set seed for experiment reproducibility
seed = 42
env.seed(seed)
tf.random.set_seed(seed)
np.random.seed(seed)

# Small epsilon value for stabilizing division operations
eps = np.finfo(np.float32).eps.item()

모델

ActorCritic 은 각각 행동 확률과 비평 값을 생성하는 하나의 신경망을 사용하여 모델링됩니다. 모델 서브 클래 싱을 사용하여 모델을 정의합니다.

순방향 패스 동안 모델은 상태를 입력으로 취하고 상태 종속 값 함수 를 모델링하는 행동 확률과 비판 값 $ V $를 모두 출력 합니다 . 목표는 기대 수익 을 극대화하는 $ \ pi $ 정책을 기반으로 작업을 선택하는 모델을 학습시키는 것입니다.

Cartpole-v0의 경우 상태를 나타내는 네 가지 값이 있습니다. 각각 카트 위치, 카트 속도, 폴 각도 및 폴 속도입니다. 상담원은 카트를 각각 왼쪽 (0)과 오른쪽 (1)으로 밀기 위해 두 가지 작업을 수행 할 수 있습니다.

자세한 내용은 OpenAI Gym의 CartPole-v0 위키 페이지 를 참조하십시오.

class ActorCritic(tf.keras.Model):
  """Combined actor-critic network."""

  def __init__(
      self, 
      num_actions: int, 
      num_hidden_units: int):
    """Initialize."""
    super().__init__()

    self.common = layers.Dense(num_hidden_units, activation="relu")
    self.actor = layers.Dense(num_actions)
    self.critic = layers.Dense(1)

  def call(self, inputs: tf.Tensor) -> Tuple[tf.Tensor, tf.Tensor]:
    x = self.common(inputs)
    return self.actor(x), self.critic(x)
num_actions = env.action_space.n  # 2
num_hidden_units = 128

model = ActorCritic(num_actions, num_hidden_units)

훈련

에이전트를 교육하려면 다음 단계를 따르십시오.

  1. 환경에서 에이전트를 실행하여 에피소드 당 학습 데이터를 수집합니다.
  2. 각 시간 단계에서 기대 수익을 계산합니다.
  3. 결합 된 행위자 비평 모델의 손실을 계산합니다.
  4. 기울기를 계산하고 네트워크 매개 변수를 업데이트합니다.
  5. 성공 기준 또는 최대 에피소드에 도달 할 때까지 1-4를 반복합니다.

1. 훈련 데이터 수집

지도 학습에서와 같이 행위자 비평 모델을 훈련하려면 훈련 데이터가 필요합니다. 그러나 이러한 데이터를 수집하려면 모델이 환경에서 "실행"되어야합니다.

우리는 각 에피소드에 대한 훈련 데이터를 수집합니다. 그런 다음 각 시간 단계에서 모델의 가중치에 의해 매개 변수화 된 현재 정책을 기반으로 행동 확률과 비평 값을 생성하기 위해 모델의 순방향 패스가 환경의 상태에서 실행됩니다.

다음 액션은 모델에 의해 생성 된 액션 확률에서 샘플링되며, 그런 다음 환경에 적용되어 다음 상태와 보상이 생성됩니다.

이 프로세스는 더 빠른 학습을 위해 나중에 TensorFlow 그래프로 컴파일 할 수 있도록 TensorFlow 작업을 사용하는 run_episode 함수에서 구현됩니다. tf.TensorArray 는 가변 길이 배열에서 Tensor 반복을 지원하는 데 사용되었습니다.

# Wrap OpenAI Gym's `env.step` call as an operation in a TensorFlow function.
# This would allow it to be included in a callable TensorFlow graph.

def env_step(action: np.ndarray) -> Tuple[np.ndarray, np.ndarray, np.ndarray]:
  """Returns state, reward and done flag given an action."""

  state, reward, done, _ = env.step(action)
  return (state.astype(np.float32), 
          np.array(reward, np.int32), 
          np.array(done, np.int32))


def tf_env_step(action: tf.Tensor) -> List[tf.Tensor]:
  return tf.numpy_function(env_step, [action], 
                           [tf.float32, tf.int32, tf.int32])
def run_episode(
    initial_state: tf.Tensor,  
    model: tf.keras.Model, 
    max_steps: int) -> List[tf.Tensor]:
  """Runs a single episode to collect training data."""

  action_probs = tf.TensorArray(dtype=tf.float32, size=0, dynamic_size=True)
  values = tf.TensorArray(dtype=tf.float32, size=0, dynamic_size=True)
  rewards = tf.TensorArray(dtype=tf.int32, size=0, dynamic_size=True)

  initial_state_shape = initial_state.shape
  state = initial_state

  for t in tf.range(max_steps):
    # Convert state into a batched tensor (batch size = 1)
    state = tf.expand_dims(state, 0)
  
    # Run the model and to get action probabilities and critic value
    action_logits_t, value = model(state)
  
    # Sample next action from the action probability distribution
    action = tf.random.categorical(action_logits_t, 1)[0, 0]
    action_probs_t = tf.nn.softmax(action_logits_t)

    # Store critic values
    values = values.write(t, tf.squeeze(value))

    # Store log probability of the action chosen
    action_probs = action_probs.write(t, action_probs_t[0, action])
  
    # Apply action to the environment to get next state and reward
    state, reward, done = tf_env_step(action)
    state.set_shape(initial_state_shape)
  
    # Store reward
    rewards = rewards.write(t, reward)

    if tf.cast(done, tf.bool):
      break

  action_probs = action_probs.stack()
  values = values.stack()
  rewards = rewards.stack()
  
  return action_probs, values, rewards

2. 기대 수익 계산

한 에피소드 동안 수집 된 각 타임 스텝 $ t $, $ {r_ {t}} ^ {T} _ {t = 1} $의 보상 순서를 예상 수익률 순서 $ {G_ {t}} ^ {로 변환합니다. T} _ {t = 1} $ : 현재 시간 단계 $ t $에서 $ T $까지의 보상 합계를 가져오고 각 보상에 기하 급수적으로 감소하는 할인 계수 $ \ gamma $를 곱합니다.

$$G_{t} = \sum^{T}_{t'=t} \gamma^{t'-t}r_{t'}$$

$ \ gamma \ in (0,1) $ 이후 현재 시간 단계에서 더 멀리 떨어진 보상에는 더 적은 가중치가 부여됩니다.

직관적으로, 기대 수익은 단순히 지금 보상이 나중에 보상보다 낫다는 것을 의미합니다. 수학적 의미에서 보상의 합이 수렴되도록하는 것입니다.

훈련을 안정화하기 위해 우리는 또한 결과의 반환 순서를 표준화합니다 (즉, 평균이 0이고 단위 표준 편차가 있음).

def get_expected_return(
    rewards: tf.Tensor, 
    gamma: float, 
    standardize: bool = True) -> tf.Tensor:
  """Compute expected returns per timestep."""

  n = tf.shape(rewards)[0]
  returns = tf.TensorArray(dtype=tf.float32, size=n)

  # Start from the end of `rewards` and accumulate reward sums
  # into the `returns` array
  rewards = tf.cast(rewards[::-1], dtype=tf.float32)
  discounted_sum = tf.constant(0.0)
  discounted_sum_shape = discounted_sum.shape
  for i in tf.range(n):
    reward = rewards[i]
    discounted_sum = reward + gamma * discounted_sum
    discounted_sum.set_shape(discounted_sum_shape)
    returns = returns.write(i, discounted_sum)
  returns = returns.stack()[::-1]

  if standardize:
    returns = ((returns - tf.math.reduce_mean(returns)) / 
               (tf.math.reduce_std(returns) + eps))

  return returns

3. 행위자 비판적 손실

우리는 하이브리드 행위자 비평 모델을 사용하고 있기 때문에 아래와 같이 훈련을 위해 행위자와 비평가 손실의 조합 인 손실 함수를 사용합니다.

$$L = L_{actor} + L_{critic}$$

배우 손실

우리 는 비평가를 상태 종속 기준으로 정책 기울기를 기반으로 행위자 손실을 공식화하고 단일 샘플 (에피소드 당) 추정치를 계산합니다.

$$L_{actor} = -\sum^{T}_{t=1} log\pi_{\theta}(a_{t} | s_{t})[G(s_{t}, a_{t}) - V^{\pi}_{\theta}(s_{t})]$$

어디:

  • $ T $ : 에피소드 당 타임 스텝 수, 에피소드마다 다를 수 있음
  • $ s_ {t} $ : $ t $ 타임 스텝의 주
  • $ a_ {t} $ : $ s $ 주에서 $ t $ 시간 단계에서 선택한 작업
  • $ \ pi _ {\ theta} $ : $ \ theta $에 의해 매개 변수화 된 정책 (액터)입니다.
  • $ V ^ {\ pi} _ {\ theta} $ : $ \ theta $로 매개 변수화되는 값 함수 (비평)입니다.
  • $ G = G_ {t} $ : 주어진 상태에 대한 예상 수익률, 시간 단계 $ t $에서의 작업 쌍

결합 손실을 최소화하여 더 높은 보상을 산출하는 행동의 확률을 최대화하기 위해 합계에 음의 항을 추가합니다.


이점

$ L_ {actor} $ 공식에서 $ G-V $ 용어는 이점 이라고하며, 이는 해당 주에 대한 $ \ pi $ 정책에 따라 선택된 임의의 행동보다 특정 주에 얼마나 더 나은 행동이 제공되는지를 나타냅니다.

기준선을 제외 할 수 있지만 이로 인해 훈련 ​​중에 큰 차이가 발생할 수 있습니다. 그리고 비평가 $ V $를 기준선으로 선택하는 것의 좋은 점은 가능한 한 $ G $에 가깝게 훈련되어 더 낮은 분산으로 이어진다는 것입니다.

또한 비평가없이 알고리즘은 예상 수익률을 기반으로 특정 상태에서 취해진 조치에 대한 확률을 높이려고 시도합니다. 이는 조치 간의 상대 확률이 동일하게 유지되는 경우 큰 차이를 만들지 않을 수 있습니다.

예를 들어, 주어진 상태에 대한 두 가지 작업이 동일한 예상 수익을 산출한다고 가정합니다. 비평가가 없으면 알고리즘은 목표 $ J $에 따라 이러한 행동의 확률을 높이려고합니다. 비평가의 경우 이점 ($ G-V = 0 $)이 없으므로 행동의 확률을 높이는 데 따른 이점이 없으며 알고리즘이 기울기를 0으로 설정합니다.


비판적 손실

$ V $를 $ G $에 최대한 가깝게 훈련하는 것은 다음 손실 함수를 사용하여 회귀 문제로 설정할 수 있습니다.

$$L_{critic} = L_{\delta}(G, V^{\pi}_{\theta})$$

여기서 $ L _ {\ delta} $는 Huber 손실 이며 제곱 오차 손실보다 데이터의 이상 값에 덜 민감합니다.

huber_loss = tf.keras.losses.Huber(reduction=tf.keras.losses.Reduction.SUM)

def compute_loss(
    action_probs: tf.Tensor,  
    values: tf.Tensor,  
    returns: tf.Tensor) -> tf.Tensor:
  """Computes the combined actor-critic loss."""

  advantage = returns - values

  action_log_probs = tf.math.log(action_probs)
  actor_loss = -tf.math.reduce_sum(action_log_probs * advantage)

  critic_loss = huber_loss(values, returns)

  return actor_loss + critic_loss

4. 매개 변수 업데이트를위한 훈련 단계 정의

위의 모든 단계를 모든 에피소드에서 실행되는 교육 단계로 결합합니다. 손실 함수로 이어지는 모든 단계는 tf.GradientTape 컨텍스트로 실행되어 자동 차별화를 활성화합니다.

Adam Optimizer를 사용하여 모델 매개 변수에 기울기를 적용합니다.

또한이 단계에서 성공 기준을 충족했는지 평가하는 데 사용되는 할인되지 않은 보상의 합계 인 episode_reward 를 계산합니다.

tf.function 컨텍스트를 train_step 함수에 적용하여 호출 가능한 TensorFlow 그래프로 컴파일 할 수 있습니다. 그러면 학습 속도가 10 배 train_step .

optimizer = tf.keras.optimizers.Adam(learning_rate=0.01)


@tf.function
def train_step(
    initial_state: tf.Tensor, 
    model: tf.keras.Model, 
    optimizer: tf.keras.optimizers.Optimizer, 
    gamma: float, 
    max_steps_per_episode: int) -> tf.Tensor:
  """Runs a model training step."""

  with tf.GradientTape() as tape:

    # Run the model for one episode to collect training data
    action_probs, values, rewards = run_episode(
        initial_state, model, max_steps_per_episode) 

    # Calculate expected returns
    returns = get_expected_return(rewards, gamma)

    # Convert training data to appropriate TF tensor shapes
    action_probs, values, returns = [
        tf.expand_dims(x, 1) for x in [action_probs, values, returns]] 

    # Calculating loss values to update our network
    loss = compute_loss(action_probs, values, returns)

  # Compute the gradients from the loss
  grads = tape.gradient(loss, model.trainable_variables)

  # Apply the gradients to the model's parameters
  optimizer.apply_gradients(zip(grads, model.trainable_variables))

  episode_reward = tf.math.reduce_sum(rewards)

  return episode_reward

5. 훈련 루프 실행

성공 기준 또는 최대 에피소드 수에 도달 할 때까지 훈련 단계를 실행하여 훈련을 실행합니다.

우리는 대기열을 사용하여 에피소드 보상의 실행 기록을 유지합니다. 100 회 시도에 도달하면 가장 오래된 보상이 대기열의 왼쪽 (꼬리) 끝에서 제거되고 최신 보상이 머리 (오른쪽)에 추가됩니다. 계산 효율성을 위해 보상의 누적 합계도 유지됩니다.

런타임에 따라 학습은 1 분 이내에 완료 될 수 있습니다.

%%time

max_episodes = 10000
max_steps_per_episode = 1000

# Cartpole-v0 is considered solved if average reward is >= 195 over 100 
# consecutive trials
reward_threshold = 195
running_reward = 0

# Discount factor for future rewards
gamma = 0.99

with tqdm.trange(max_episodes) as t:
  for i in t:
    initial_state = tf.constant(env.reset(), dtype=tf.float32)
    episode_reward = int(train_step(
        initial_state, model, optimizer, gamma, max_steps_per_episode))

    running_reward = episode_reward*0.01 + running_reward*.99
  
    t.set_description(f'Episode {i}')
    t.set_postfix(
        episode_reward=episode_reward, running_reward=running_reward)
  
    # Show average episode reward every 10 episodes
    if i % 10 == 0:
      pass # print(f'Episode {i}: average reward: {avg_reward}')
  
    if running_reward > reward_threshold:  
        break

print(f'\nSolved at episode {i}: average reward: {running_reward:.2f}!')
Episode 1524:  15%|█▌        | 1524/10000 [08:16<46:00,  3.07it/s, episode_reward=200, running_reward=195]

Solved at episode 1524: average reward: 195.03!
CPU times: user 20min 43s, sys: 4min 52s, total: 25min 35s
Wall time: 8min 16s



심상

훈련 후에는 모델이 환경에서 어떻게 작동하는지 시각화하는 것이 좋습니다. 아래 셀을 실행하여 모델의 한 에피소드 실행에 대한 GIF 애니메이션을 생성 할 수 있습니다. Colab에서 환경의 이미지를 올바르게 렌더링하려면 OpenAI Gym에 대한 추가 패키지를 설치해야합니다.

# Render an episode and save as a GIF file

from IPython import display as ipythondisplay
from PIL import Image
from pyvirtualdisplay import Display


display = Display(visible=0, size=(400, 300))
display.start()


def render_episode(env: gym.Env, model: tf.keras.Model, max_steps: int): 
  screen = env.render(mode='rgb_array')
  im = Image.fromarray(screen)

  images = [im]
  
  state = tf.constant(env.reset(), dtype=tf.float32)
  for i in range(1, max_steps + 1):
    state = tf.expand_dims(state, 0)
    action_probs, _ = model(state)
    action = np.argmax(np.squeeze(action_probs))

    state, _, done, _ = env.step(action)
    state = tf.constant(state, dtype=tf.float32)

    # Render screen every 10 steps
    if i % 10 == 0:
      screen = env.render(mode='rgb_array')
      images.append(Image.fromarray(screen))
  
    if done:
      break
  
  return images


# Save GIF image
images = render_episode(env, model, max_steps_per_episode)
image_file = 'cartpole-v0.gif'
# loop=0: loop forever, duration=1: play each frame for 1ms
images[0].save(
    image_file, save_all=True, append_images=images[1:], loop=0, duration=1)
import tensorflow_docs.vis.embed as embed
embed.embed_file(image_file)

gif

다음 단계

이 가이드에서는 Tensorflow를 사용하여 행위자 비판 방법을 구현하는 방법을 보여주었습니다.

다음 단계로 OpenAI Gym의 다른 환경에서 모델 훈련을 시도 할 수 있습니다.

행위자 비판 방법 및 Cartpole-v0 문제에 대한 추가 정보는 다음 리소스를 참조 할 수 있습니다.

TensorFlow의 더 많은 강화 학습 예제는 다음 리소스를 확인할 수 있습니다.