MLコミュニティデーは11月9日です! TensorFlow、JAXからの更新のために私たちに参加し、より多くの詳細をご覧ください

俳優批評家の方法でCartPoleを再生する

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

このチュートリアルでは、実装する方法を示し俳優、評論家の上のエージェントを訓練するTensorFlowを用いた方法を開きAIジムCartPole-V0環境を。読者がある程度精通有するものとする政策勾配法強化学習のを。

アクター-批評家の方法

俳優、評論家の方法がある時間差(TD)学習値関数のポリシー機能の独立を表現する方法。

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

アクター評論家方法では、ポリシーは、状態、所与の可能なアクションのセットを提案している俳優と呼ばれ、推定値関数は、与えられたポリシーに基づいてアクターによって取られるアクションを評価評論家と呼ばれ。

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

CartPole-v0

CartPole-V0環境、ポールは、摩擦の軌道に沿って移動台車に取り付けられています。ポールは直立して始まり、エージェントの目標は、カートに-1または+1の力を加えることによってポールが倒れるのを防ぐことです。ポールが直立したままになるたびに、+ 1の報酬が与えられます。エピソードは、(1)ポールが垂直から15度以上離れているか、(2)カートが中心から2.4ユニット以上移動したときに終了します。

Cartpole-v0環境で訓練された俳優批評家モデル

エピソードの平均合計報酬が100回の連続試行で195に達すると、問題は「解決済み」と見なされます。

設定

必要なパッケージをインポートし、グローバル設定を構成します。

pip install gym
pip install pyglet
# Install additional packages for visualization
sudo apt-get install -y xvfb python-opengl > /dev/null 2>&1
pip install pyvirtualdisplay > /dev/null 2>&1
pip install git+https://github.com/tensorflow/docs > /dev/null 2>&1
import collections
import gym
import numpy as np
import statistics
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 $、どのモデル状態依存取る値関数を。目標は、選択したアクションが最大化が期待されることをポリシー$ \パイ$に基づいていることをモデル訓練することであるリターンを

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

参照してくださいOpenAIジムのCartPole-V0のwikiページの詳細については。

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 、後で高速トレーニング用TensorFlowグラフにコンパイルすることができるようにTensorFlow操作を使用する関数、。そのノートtf.TensorArray sは可変長アレイ上テンソル反復をサポートするために使用しました。

# 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) -> Tuple[tf.Tensor, tf.Tensor, 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 $での特定の状態、アクションペアの期待収益

合計損失を最小化することにより、より高い報酬を生み出すアクションの確率を最大化することが目的であるため、合計に負の項が追加されます。


アドバンテージ

$ G -私達の$ L_ {アクター} $製剤中のV $用語が呼び出される利点より良いアクションがその状態のポリシー$ \パイ$に応じて選択されたランダム行動上の特定の状態が与えられているどの程度を示し、。

ベースラインを除外することは可能ですが、これによりトレーニング中に大きな変動が生じる可能性があります。そして、批評家の$ V $をベースラインとして選択することの良い点は、それが$ G $にできるだけ近くなるようにトレーニングされ、分散が低くなることです。

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

たとえば、特定の状態に対する2つのアクションが、同じ期待収益をもたらすと仮定します。批評家がいなければ、アルゴリズムは目的$ J $に基づいてこれらのアクションの確率を上げようとします。批評家の場合、利点がない($ G-V = 0 $)ため、アクションの確率を上げることで得られる利点がなく、アルゴリズムが勾配をゼロに設定することが判明する場合があります。


批評家の喪失

$ V $を$ G $にできるだけ近づけるようにトレーニングすることは、次の損失関数を使用した回帰問題として設定できます。

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

$ L _ {\デルタ} $である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それは訓練で10倍高速化につながることができます呼び出し可能TensorFlowグラフにコンパイルすることができるように機能。

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

min_episodes_criterion = 100
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

# Keep last episodes reward
episodes_reward: collections.deque = collections.deque(maxlen=min_episodes_criterion)

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

    episodes_reward.append(episode_reward)
    running_reward = statistics.mean(episodes_reward)

    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 and i >= min_episodes_criterion:  
        break

print(f'\nSolved at episode {i}: average reward: {running_reward:.2f}!')
Episode 361:   4%|▎         | 361/10000 [01:16<34:10,  4.70it/s, episode_reward=182, running_reward=195]
Solved at episode 361: average reward: 195.14!
CPU times: user 2min 50s, sys: 40.3 s, total: 3min 30s
Wall time: 1min 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の強化学習の例については、次のリソースを確認してください。