Actor-CriticメソッドでCartPoleを再生する

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ユニット以上移動したときに終了します。

エピソードの平均合計報酬が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\)両方を出力します。目標は、期待収益を最大化するポリシー \(\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関数に実装されています。この関数は、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) -> 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\) placeholder7に取得されます。報酬は、指数関数的に減衰する割引係数 \(\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}\)：l10n- \(\theta\)によってパラメーター化されたポリシー（アクター）です
• \(V^{\pi}_{\theta}\)：l10n- \(\theta\)によってもパラメーター化された値関数（クリティカル）です
• \(G = G_{t}\)：タイムステップ \(t\)での特定の状態、アクションペアの期待収益

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

アドバンテージ

\(L_{actor}\) 定式化における \(G - V\) の用語は、アドバンテージと呼ばれます。これは、特定の状態のポリシー \(\pi\) に従って選択されたランダムなアクションよりも、その状態にアクションがどれだけ優れているかを示します。

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

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

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

批評家の喪失

l10n-placeholder32を \(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

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:13<32:33,  4.93it/s, episode_reward=182, running_reward=195]
Solved at episode 361: average reward: 195.14!
CPU times: user 2min 46s, sys: 35.4 s, total: 3min 21s
Wall time: 1min 13s

視覚化

トレーニング後、モデルが環境内でどのように機能するかを視覚化するとよいでしょう。以下のセルを実行して、モデルの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)

次のステップ

このチュートリアルでは、Tensorflowを使用してアクタークリティカルメソッドを実装する方法を示しました。

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

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

• 俳優評論家の方法
• 俳優評論家レクチャー（CAL）
• カートポール学習制御問題[Barto、etal。 1983]

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

• 強化学習コードの例（keras.io）
• TF-エージェント強化学習ライブラリ