このページは Cloud Translation API によって翻訳されました。
Switch to English

俳優批評家の方法でカートポールを再生する

TensorFlow.orgで表示 GoogleColabで実行 GitHubでソースを表示ノートブックをダウンロード

このチュートリアルでは、TensorFlowを使用してActor-Criticメソッドを実装し、 Open AI GymCartPole-V0環境でエージェントをトレーニングする方法を示します。読者は、強化学習のポリシー勾配法にある程度精通していることを前提としています。

アクター-批評家の方法

アクタークリティカルメソッドは、値関数から独立したポリシー関数を表す時間差(TD)学習メソッドです。

ポリシー関数(またはポリシー)は、指定された状態に基づいてエージェントが実行できるアクションの確率分布を返します。値関数は、特定の状態で開始し、その後ずっと特定のポリシーに従って動作するエージェントの期待収益を決定します。

Actor-Criticメソッドでは、ポリシーは、状態が与えられた一連の可能なアクションを提案するアクターと呼ばれ、推定値関数は、与えられたポリシーに基づいてアクターが実行したアクションを評価する批評家と呼ばれます。 。

このチュートリアルでは、アクター評論家の両方が、2つの出力を持つ1つのニューラルネットワークを使用して表されます。

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.

# 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()

モデル

アクター評論家は、それぞれアクション確率と評論家の価値を生成する1つのニューラルネットワークを使用してモデル化されます。モデルのサブクラス化を使用してモデルを定義します。

フォワードパスの間、モデルは状態を入力として受け取り、アクション確率と、状態に依存する値関数をモデル化する評論家の値$ V $の両方を出力します。目標は、期待収益を最大化するポリシー$ \ pi $に基づいてアクションを選択するモデルをトレーニングすることです。

Cartpole-v0の場合、状態を表す4つの値があります。それぞれ、カートの位置、カートの速度、極の角度、極の速度です。エージェントは、カートをそれぞれ左(0)と右(1)に押す2つのアクションを実行できます。

詳細については、 OpenAIGymのCartPole-v0wikiページ参照してください。

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.トレーニングデータの収集

教師あり学習と同様に、アクター批評モデルをトレーニングするには、トレーニングデータが必要です。ただし、このようなデータを収集するには、モデルを環境内で「実行」する必要があります。

エピソードごとにトレーニングデータを収集します。次に、各タイムステップで、モデルのフォワードパスが環境の状態で実行され、モデルの重みによってパラメーター化された現在のポリシーに基づいてアクション確率と批評家の値が生成されます。

次のアクションは、モデルによって生成されたアクション確率からサンプリングされ、環境に適用されて、次の状態と報酬が生成されます。

このプロセスはrun_episode関数に実装されてrun_episodeます。この関数は、TensorFlow操作を使用して、後でトレーニングを高速化するためにTensorFlowグラフにコンパイルできるようにします。 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.期待収益の計算

1つのエピソード中に収集された各タイムステップ$ 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)$なので、現在のタイムステップからさらに離れた報酬には、より少ない重みが与えられます。

直感的には、期待収益は単に、現在の報酬が後の報酬よりも優れていることを意味します。数学的な意味では、報酬の合計が収束することを保証することです。

トレーニングを安定させるために、結果のリターンのシーケンスも標準化します(つまり、平均と単位標準偏差をゼロにします)。

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 $にできるだけ近くなるようにトレーニングされ、分散が低くなることです。

さらに、批評家がいなければ、アルゴリズムは期待収益に基づいて特定の状態で実行されるアクションの確率を上げようとします。これは、アクション間の相対確率が同じままである場合、大きな違いはない可能性があります。

たとえば、特定の状態に対する2つのアクションで、同じ期待収益が得られるとします。批評家がいなければ、アルゴリズムは目的$ J $に基づいてこれらのアクションの確率を上げようとします。批評家の場合、利点がない($ G-V = 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オプティマイザーを使用して、モデルパラメーターに勾配を適用します。

また、割引されていない報酬の合計、 episode_rewardをこのステップで計算します。これは、成功基準を満たしているかどうかを評価するために後で使用されます。

tf.functionコンテキストをtrain_step関数に適用して、呼び出し可能なTensorFlowグラフにコンパイルできるようにします。これにより、トレーニングが10倍高速化されます。

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



視覚化

トレーニング後、モデルが環境内でどのように機能するかを視覚化するとよいでしょう。以下のセルを実行して、モデルの1つのエピソードの実行のGIFアニメーションを生成できます。 Colabで環境の画像を正しくレンダリングするには、OpenAIGymに追加のパッケージをインストールする必要があることに注意してください。

# 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を使用してアクタークリティカルメソッドを実装する方法を示しました。

次のステップとして、OpenAIGymの別の環境でモデルをトレーニングしてみることができます。

アクタークリティカルなメソッドとCartpole-v0の問題に関する追加情報については、次のリソースを参照してください。

TensorFlowの強化学習の例については、次のリソースを確認してください。