ホーム » 「TF-Agents 0.4」タグがついた投稿

タグアーカイブ: TF-Agents 0.4

TF-Agents 0.4 : Tutorials : DQN C51/Rainbow

TF-Agents 0.4 Tutorials : DQN C51/Rainbow (翻訳/解説)

翻訳 : (株)クラスキャット セールスインフォメーション
作成日時 : 04/23/2020 (0.4)

* 本ページは、TF Agents の以下のドキュメントを翻訳した上で適宜、補足説明したものです:

* サンプルコードの動作確認はしておりますが、必要な場合には適宜、追加改変しています。
* ご自由にリンクを張って頂いてかまいませんが、sales-info@classcat.com までご一報いただけると嬉しいです。

 

DQN C51/Rainbow

イントロダクション

このサンプルは TF-Agents ライブラリでカートポール環境上カテゴリカル DQN (C51) エージェントをどのように訓練するかを示します。

必要条件として DQN チュートリアル に目を通していることを確実にしてください。このチュートリアルは DQN チュートリアルに精通していることを仮定しています ; それは DQN と C51 の間の違いに主として焦点を当てます。

 

セットアップ

tf-agents をまだインストールしていないのであれば、以下を実行します :

!sudo apt-get install -y xvfb ffmpeg
!pip install 'gym==0.10.11'
!pip install 'imageio==2.4.0'
!pip install PILLOW
!pip install 'pyglet==1.3.2'
!pip install pyvirtualdisplay
!pip install --upgrade tensorflow-probability
!pip install tf-agents
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function

import base64
import imageio
import IPython
import matplotlib
import matplotlib.pyplot as plt
import PIL.Image
import pyvirtualdisplay

import tensorflow as tf

from tf_agents.agents.categorical_dqn import categorical_dqn_agent
from tf_agents.drivers import dynamic_step_driver
from tf_agents.environments import suite_gym
from tf_agents.environments import tf_py_environment
from tf_agents.eval import metric_utils
from tf_agents.metrics import tf_metrics
from tf_agents.networks import categorical_q_network
from tf_agents.policies import random_tf_policy
from tf_agents.replay_buffers import tf_uniform_replay_buffer
from tf_agents.trajectories import trajectory
from tf_agents.utils import common

tf.compat.v1.enable_v2_behavior()


# Set up a virtual display for rendering OpenAI gym environments.
display = pyvirtualdisplay.Display(visible=0, size=(1400, 900)).start()

 

ハイパーパラメータ

env_name = "CartPole-v1" # @param {type:"string"}
num_iterations = 15000 # @param {type:"integer"}

initial_collect_steps = 1000  # @param {type:"integer"} 
collect_steps_per_iteration = 1  # @param {type:"integer"}
replay_buffer_capacity = 100000  # @param {type:"integer"}

fc_layer_params = (100,)

batch_size = 64  # @param {type:"integer"}
learning_rate = 1e-3  # @param {type:"number"}
gamma = 0.99
log_interval = 200  # @param {type:"integer"}

num_atoms = 51  # @param {type:"integer"}
min_q_value = -20  # @param {type:"integer"}
max_q_value = 20  # @param {type:"integer"}
n_step_update = 2  # @param {type:"integer"}

num_eval_episodes = 10  # @param {type:"integer"}
eval_interval = 1000  # @param {type:"integer"}

 

環境

前のように環境をロードします、一つは訓練のためそして一つは評価のためです。ここでは CartPole-v1 (vs. CartPole-v0 in DQN チュートリアル) を使用します、これは 200 よりも大きい 500 の大きな最大報酬を持ちます。

train_py_env = suite_gym.load(env_name)
eval_py_env = suite_gym.load(env_name)

train_env = tf_py_environment.TFPyEnvironment(train_py_env)
eval_env = tf_py_environment.TFPyEnvironment(eval_py_env)

 

エージェント

C51 は DQN に基づく Q-学習アルゴリズムです。DQN のように、それは離散アクション空間を持つ任意の環境上で利用できます。

C51 と DQN の間の主要な違いは各状態-アクションペアのための Q-値を単純に予測するのではなく、C51 は Q-値の確率分布のためのヒストグラム・モデルを予測します :

単純な期待値よりも分布を学習することにより、アルゴリズムは訓練の間よりステーブルのままでいることが可能で、改良された最終的なパフォーマンスに繋がります。これは特に二峰性分布の状況あるいは多峰性値分布でさえ特に真で、そこでは単一平均は正確な描写を提供しません。

値の上ではなく確率分布上で訓練するため、C51 はその損失関数を計算するためにある複雑な分布計算を遂行しなければなりません。しかし心配しないでください、これの総ては貴方のために TF-Agents でケアされます!

C51 エージェントを作成するため、最初に CategoricalQNetwork を作成する必要があります。CategoricalQNetwork の API は QNetwork のそれと同じです、追加引数 num_atoms があることを除いて。これは確率分布推定のサポート点の数を表します。(上の画像は 10 サポート点を含みます、各々は垂直の青色棒により表されます。) 名前から分かるように、アトムのデフォルト数は 51 です。

categorical_q_net = categorical_q_network.CategoricalQNetwork(
    train_env.observation_spec(),
    train_env.action_spec(),
    num_atoms=num_atoms,
    fc_layer_params=fc_layer_params)

ちょうど作成したネットワークを訓練するために optimizer、そしてネットワークが何回更新されたかを追跡するために train_step_counter 変数も必要です。

vanilla DqnAgent との一つの他の重要な違いは今は引数として min_q_value と max_q_value を指定する必要があることです。これらはサポートの最も極値を指定します (換言すれば、両側の 51 アトムの最も極値)。貴方の特定の環境のためにこれらを適切に選択することを確実にしてください。ここでは -20 と 20 を使用します。

optimizer = tf.compat.v1.train.AdamOptimizer(learning_rate=learning_rate)

train_step_counter = tf.compat.v2.Variable(0)

agent = categorical_dqn_agent.CategoricalDqnAgent(
    train_env.time_step_spec(),
    train_env.action_spec(),
    categorical_q_network=categorical_q_net,
    optimizer=optimizer,
    min_q_value=min_q_value,
    max_q_value=max_q_value,
    n_step_update=n_step_update,
    td_errors_loss_fn=common.element_wise_squared_loss,
    gamma=gamma,
    train_step_counter=train_step_counter)
agent.initialize()

注意すべき一つの最後のことは $n$ = 2 で n-step 更新を使用するために引数も追加したことです。私達は (bellman 最適方程式に基づいて) 単一ステップのリターンを使用して現在の時間ステップと次の時間ステップの Q-値の間の誤差を計算するだけです。単一ステップのリターンは次のように定義されます :

$G_t = R_{t + 1} + \gamma V(s_{t + 1})$

ここで $V(s) = \max_a{Q(s, a)}$ を定義します。

N-ステップ更新は標準的な単一ステップの報酬関数を $n$ 回展開を伴います :

$G_t^n = R_{t + 1} + \gamma R_{t + 2} + \gamma^2 R_{t + 3} + \dots + \gamma^n V(s_{t + n})$

N-ステップ更新は更なる将来からブートストラップすることを可能にし、そして $n$ の正しい値で、これはしばしばより高速な学習に繋がります。

C51 と n-ステップ更新はしばしば Rainbow エージェントのコアを形成するために優先付けられた再生と結合されますが、私達は優先付けられた再生の実装から測定可能な改良は見ませんでした。その上、C51 エージェントを n-ステップ更新だけと結合するとき、エージェントはテストしたAtari 環境のサンプル上他の Rainbow エージェントと同じように上手く遂行します。

 

メトリクスと評価

ポリシーを評価するために使用される最も一般的なメトリックは平均リターンです。リターンはエピソードのために環境でポリシーを実行する間に得られた報酬の総計です、そして通常はこれを 2, 3 のエピソードに渡り平均します。平均リターンメトリックを次のように計算できます。

def compute_avg_return(environment, policy, num_episodes=10):

  total_return = 0.0
  for _ in range(num_episodes):

    time_step = environment.reset()
    episode_return = 0.0

    while not time_step.is_last():
      action_step = policy.action(time_step)
      time_step = environment.step(action_step.action)
      episode_return += time_step.reward
    total_return += episode_return

  avg_return = total_return / num_episodes
  return avg_return.numpy()[0]


random_policy = random_tf_policy.RandomTFPolicy(train_env.time_step_spec(),
                                                train_env.action_spec())

compute_avg_return(eval_env, random_policy, num_eval_episodes)

# Please also see the metrics module for standard implementations of different

 

データ収集

DQN チュートリアル内のように、再生バッファとランダムポリシーで初期データ収集をセットアップします。

#@test {"skip": true}
replay_buffer = tf_uniform_replay_buffer.TFUniformReplayBuffer(
    data_spec=agent.collect_data_spec,
    batch_size=train_env.batch_size,
    max_length=replay_buffer_capacity)

def collect_step(environment, policy):
  time_step = environment.current_time_step()
  action_step = policy.action(time_step)
  next_time_step = environment.step(action_step.action)
  traj = trajectory.from_transition(time_step, action_step, next_time_step)

  # Add trajectory to the replay buffer
  replay_buffer.add_batch(traj)

for _ in range(initial_collect_steps):
  collect_step(train_env, random_policy)

# This loop is so common in RL, that we provide standard implementations of
# these. For more details see the drivers module.

# Dataset generates trajectories with shape [BxTx...] where
# T = n_step_update + 1.
dataset = replay_buffer.as_dataset(
    num_parallel_calls=3, sample_batch_size=batch_size,
    num_steps=n_step_update + 1).prefetch(3)

iterator = iter(dataset)

 

エージェントを訓練する

訓練ループは環境からデータを集めることとエージェントのネットワークを最適化することの両者を含みます。途中で、上手くやっているかを見るためにエージェントのポリシーを時々評価します。

以下は実行に ~7 分かかります。

#@test {"skip": true}
try:
  %%time
except:
  pass

# (Optional) Optimize by wrapping some of the code in a graph using TF function.
agent.train = common.function(agent.train)

# Reset the train step
agent.train_step_counter.assign(0)

# Evaluate the agent's policy once before training.
avg_return = compute_avg_return(eval_env, agent.policy, num_eval_episodes)
returns = [avg_return]

for _ in range(num_iterations):

  # Collect a few steps using collect_policy and save to the replay buffer.
  for _ in range(collect_steps_per_iteration):
    collect_step(train_env, agent.collect_policy)

  # Sample a batch of data from the buffer and update the agent's network.
  experience, unused_info = next(iterator)
  train_loss = agent.train(experience)

  step = agent.train_step_counter.numpy()

  if step % log_interval == 0:
    print('step = {0}: loss = {1}'.format(step, train_loss.loss))

  if step % eval_interval == 0:
    avg_return = compute_avg_return(eval_env, agent.policy, num_eval_episodes)
    print('step = {0}: Average Return = {1:.2f}'.format(step, avg_return))
    returns.append(avg_return)

 

可視化

プロット

エージェントのパフォーマンスを見るためにリターン vs グローバルステップをプロットすることができます。Cartpole-v1 では、ポールが直立し続ける総ての時間ステップのために環境は +1 の報酬を与え、そしてステップのあ最大数は 500 ですから、最大の可能なリターンもまた 500 です。

#@test {"skip": true}

steps = range(0, num_iterations + 1, eval_interval)
plt.plot(steps, returns)
plt.ylabel('Average Return')
plt.xlabel('Step')
plt.ylim(top=550)

 

ビデオ

各ステップで環境をレンダリングすることによりエージェントのパフォーマンスを可視化することは有用です。それを行なう前に、最初にビデオをこの colab に埋め込む関数を作成しましょう。

def embed_mp4(filename):
  """Embeds an mp4 file in the notebook."""
  video = open(filename,'rb').read()
  b64 = base64.b64encode(video)
  tag = '''
  '''.format(b64.decode())

  return IPython.display.HTML(tag)

以下のコードは 2,3 のエピソードのためにエージェントのポリシーを可視化します :

num_episodes = 3
video_filename = 'imageio.mp4'
with imageio.get_writer(video_filename, fps=60) as video:
  for _ in range(num_episodes):
    time_step = eval_env.reset()
    video.append_data(eval_py_env.render())
    while not time_step.is_last():
      action_step = agent.policy.action(time_step)
      time_step = eval_env.step(action_step.action)
      video.append_data(eval_py_env.render())

embed_mp4(video_filename)

C51 は CartPole-v1 上 DQN よりも僅かに良く行なう傾向がありますが、2 つのエージェントの間の違いはますます複雑な環境で更に更に重要になります。例えば、full Atari 2600 ベンチマーク上、C51 はランダムエージェントに関して正規化後 DQN を越えて 126 % の平均スコア改善を実演します。n-ステップ更新を含むことにより追加の改善が獲得できます。

For a deeper dive into the C51 algorithm, see A Distributional Perspective on Reinforcement Learning (2017).

 

以上






TF-Agents 0.4 : Tutorials : Checkpointer と PolicySaver

TF-Agents 0.4 Tutorials : Checkpointer と PolicySaver (翻訳/解説)

翻訳 : (株)クラスキャット セールスインフォメーション
作成日時 : 04/23/2020 (0.4)

* 本ページは、TF Agents の以下のドキュメントを翻訳した上で適宜、補足説明したものです:

* サンプルコードの動作確認はしておりますが、必要な場合には適宜、追加改変しています。
* ご自由にリンクを張って頂いてかまいませんが、sales-info@classcat.com までご一報いただけると嬉しいです。

 

Checkpointer と PolicySaver

イントロダクション

tf_agents.utils.common.Checkpointer は訓練状態、ポリシー状態と reply_buffer 状態をローカルストレージに/からセーブ/ロードするユティリティです。

tf_agents.policies.policy_saver.PolicySaver はポリシーだけをセーブ/ロードするツールで Checkpointer よりも軽いです。ポリシーを作成したコードのどのような知識なしでもモデルを配備するために PolicySaver を利用できます。

このチュートリアルでは、モデルを訓練するために DQN を使用てから、状態とモデルを対話的にどのようにストアしてロードできるかを示すために Checkpointer と PolicySaver を使用します。PolicySaver のために TF 2.0 の新しい saved_model ツールと形式を使用することに注意してください。

 

セットアップ

以下の依存性をインストールしていないのであれば、以下を実行します :

#@test {"skip": true}
!sudo apt-get install -y xvfb ffmpeg
!pip install 'gym==0.10.11'
!pip install 'imageio==2.4.0'
!pip install 'pyglet==1.3.2'
!pip install 'xvfbwrapper==0.2.9'
!pip install --upgrade tensorflow-probability
!pip install tf-agents
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function

import base64
import imageio
import io
import matplotlib
import matplotlib.pyplot as plt
import os
import shutil
import tempfile
import tensorflow as tf
import zipfile
import IPython

try:
  from google.colab import files
except ImportError:
  files = None
from tf_agents.agents.dqn import dqn_agent
from tf_agents.drivers import dynamic_step_driver
from tf_agents.environments import suite_gym
from tf_agents.environments import tf_py_environment
from tf_agents.eval import metric_utils
from tf_agents.metrics import tf_metrics
from tf_agents.networks import q_network
from tf_agents.policies import policy_saver
from tf_agents.policies import py_tf_eager_policy
from tf_agents.policies import random_tf_policy
from tf_agents.replay_buffers import tf_uniform_replay_buffer
from tf_agents.trajectories import trajectory
from tf_agents.utils import common

tf.compat.v1.enable_v2_behavior()

tempdir = os.getenv("TEST_TMPDIR", tempfile.gettempdir())
#@test {"skip": true}
# Set up a virtual display for rendering OpenAI gym environments.
import xvfbwrapper
xvfbwrapper.Xvfb(1400, 900, 24).start()

 

DQN エージェント

ちょうど前の colab のように、DQN エージェントをセットアップしていきます。詳細はデフォルトでは隠されます、何故ならばこの colab の中核パートではないからです、しかし詳細を見るために ‘SHOW CODE’ 上をクリックできます。

 

ハイパーパラメータ

env_name = "CartPole-v1"

collect_steps_per_iteration = 100
replay_buffer_capacity = 100000

fc_layer_params = (100,)

batch_size = 64
learning_rate = 1e-3
log_interval = 5

num_eval_episodes = 10
eval_interval = 1000

 

環境

train_py_env = suite_gym.load(env_name)
eval_py_env = suite_gym.load(env_name)

train_env = tf_py_environment.TFPyEnvironment(train_py_env)
eval_env = tf_py_environment.TFPyEnvironment(eval_py_env)

 

エージェント

#@title
q_net = q_network.QNetwork(
    train_env.observation_spec(),
    train_env.action_spec(),
    fc_layer_params=fc_layer_params)

optimizer = tf.compat.v1.train.AdamOptimizer(learning_rate=learning_rate)

global_step = tf.compat.v1.train.get_or_create_global_step()

agent = dqn_agent.DqnAgent(
    train_env.time_step_spec(),
    train_env.action_spec(),
    q_network=q_net,
    optimizer=optimizer,
    td_errors_loss_fn=common.element_wise_squared_loss,
    train_step_counter=global_step)
agent.initialize()

 

データ収集

#@title
replay_buffer = tf_uniform_replay_buffer.TFUniformReplayBuffer(
    data_spec=agent.collect_data_spec,
    batch_size=train_env.batch_size,
    max_length=replay_buffer_capacity)

collect_driver = dynamic_step_driver.DynamicStepDriver(
    train_env,
    agent.collect_policy,
    observers=[replay_buffer.add_batch],
    num_steps=collect_steps_per_iteration)

# Initial data collection
collect_driver.run()

# Dataset generates trajectories with shape [BxTx...] where
# T = n_step_update + 1.
dataset = replay_buffer.as_dataset(
    num_parallel_calls=3, sample_batch_size=batch_size,
    num_steps=2).prefetch(3)

iterator = iter(dataset)

 

エージェントを訓練する

#@title
# (Optional) Optimize by wrapping some of the code in a graph using TF function.
agent.train = common.function(agent.train)

def train_one_iteration():

  # Collect a few steps using collect_policy and save to the replay buffer.
  for _ in range(collect_steps_per_iteration):
    collect_driver.run()

  # Sample a batch of data from the buffer and update the agent's network.
  experience, unused_info = next(iterator)
  train_loss = agent.train(experience)

  iteration = agent.train_step_counter.numpy()
  print ('iteration: {0} loss: {1}'.format(iteration, train_loss.loss))

 

ビデオ生成

#@title
def embed_gif(gif_buffer):
  """Embeds a gif file in the notebook."""
  tag = ''.format(base64.b64encode(gif_buffer).decode())
  return IPython.display.HTML(tag)

def run_episodes_and_create_video(policy, eval_tf_env, eval_py_env):
  num_episodes = 3
  frames = []
  for _ in range(num_episodes):
    time_step = eval_tf_env.reset()
    frames.append(eval_py_env.render())
    while not time_step.is_last():
      action_step = policy.action(time_step)
      time_step = eval_tf_env.step(action_step.action)
      frames.append(eval_py_env.render())
  gif_file = io.BytesIO()
  imageio.mimsave(gif_file, frames, format='gif', fps=60)
  IPython.display.display(embed_gif(gif_file.getvalue()))

 

ビデオを生成する

ビデオを生成することによりポリシーのパフォーマンスを確認する。

print ('global_step:')
print (global_step)
run_episodes_and_create_video(agent.policy, eval_env, eval_py_env)

 

Checkpointer と PolicySaver をセットアップする

今は Checkpointer と PolicySaver を使用する準備ができました。

 

Checkpointer

checkpoint_dir = os.path.join(tempdir, 'checkpoint')
train_checkpointer = common.Checkpointer(
    ckpt_dir=checkpoint_dir,
    max_to_keep=1,
    agent=agent,
    policy=agent.policy,
    replay_buffer=replay_buffer,
    global_step=global_step
)

 

Policy Saver

policy_dir = os.path.join(tempdir, 'policy')
tf_policy_saver = policy_saver.PolicySaver(agent.policy)

 

1 反復訓練する

#@test {"skip": true}
print('Training one iteration....')
train_one_iteration()

 

チェックポイントにセーブする

train_checkpointer.save(global_step)

 

チェックポイントをリストアする

これが動作するするためには、オブジェクトのセット全体がチェックポイントが作成されたときと同じ方法で再作成されるべきです。

train_checkpointer.initialize_or_restore()
global_step = tf.compat.v1.train.get_global_step()

ポリシーもセーブして位置にエクスポートします。

tf_policy_saver.save(policy_dir)

ポリシーは、それを作成するために何のエージェントやネットワークが使用されたかのどのような知識も持つことなくロードできます。これはポリシーの配備を遥かに容易にします。

セーブされたポリシーをロードしてそれがどのように遂行するか確認します。

saved_policy = tf.compat.v2.saved_model.load(policy_dir)
run_episodes_and_create_video(saved_policy, eval_env, eval_py_env)

 

Export と import

colab の残りは、後の時点で訓練をで続けてそして再度訓練しなければならないことなくモデルを配備できるように checkpointer とポリシー・ディレクトリを export / import するに役立つでしょう。

今は「1 反復を訓練する」に戻って後で差異を理解できるように更に 2, 3 回訓練することができます。僅かばかりより良い結果をひとたび見始めるのであれば、以下を続けてください。

#@title Create zip file and upload zip file (double-click to see the code)
def create_zip_file(dirname, base_filename):
  return shutil.make_archive(base_filename, 'zip', dirname)

def upload_and_unzip_file_to(dirname):
  if files is None:
    return
  uploaded = files.upload()
  for fn in uploaded.keys():
    print('User uploaded file "{name}" with length {length} bytes'.format(
        name=fn, length=len(uploaded[fn])))
    shutil.rmtree(dirname)
    zip_files = zipfile.ZipFile(io.BytesIO(uploaded[fn]), 'r')
    zip_files.extractall(dirname)
    zip_files.close()

チェックポイント・ディレクトリから zip されたファイルを作成します。

train_checkpointer.save(global_step)
checkpoint_zip_filename = create_zip_file(checkpoint_dir, os.path.join(tempdir, 'exported_cp'))

zip ファイルをダウンロードします。

#@test {"skip": true}
if files is not None:
  files.download(checkpoint_zip_filename) # try again if this fails: https://github.com/googlecolab/colabtools/issues/469

ある程度の時間 (10-15 回) の間訓練した後、チェックポイント zip ファイルをダウンロードして、そして訓練をリセットするために “Runtime > Restart and run all” に行き、そしてこのセルに戻ります。今はダウンロードされた zip ファイルをアップロードして訓練を続けることができます。

#@test {"skip": true}
upload_and_unzip_file_to(checkpoint_dir)
train_checkpointer.initialize_or_restore()
global_step = tf.compat.v1.train.get_global_step()

ひとたびチェックポイント・ディレクトリをアップロードしたら、訓練を続けるために「1 反復を訓練する」に戻るかロードされたポリシーのパフォーマンスを確認するために「ビデオを生成する」に戻ります。

代わりに、ポリシー (モデル) をセーブしてそれをリストアできます。checkpointer とは違い、訓練を続けることはできませんが、依然としてモデルを配備できます。ダウンロードされたファイルは checkpointer のそれよりも遥かに小さいことに注意してください。

tf_policy_saver.save(policy_dir)
policy_zip_filename = create_zip_file(policy_dir, os.path.join(tempdir, 'exported_policy'))
#@test {"skip": true}
if files is not None:
  files.download(policy_zip_filename) # try again if this fails: https://github.com/googlecolab/colabtools/issues/469

ダウンロードされたポリシー・ディレクトリ (exported_policy.zip) をアップロードしてセーブされたポリシーがどのように遂行するかを確認します。

#@test {"skip": true}
upload_and_unzip_file_to(policy_dir)
saved_policy = tf.compat.v2.saved_model.load(policy_dir)
run_episodes_and_create_video(saved_policy, eval_env, eval_py_env)

 

SavedModelPyTFEagerPolicy

TF ポリシーを使用することを望まない場合、py_tf_eager_policy.SavedModelPyTFEagerPolicy の利用を通して Python env で saved_model を直接使用することもできます。

これは eager mode が有効であるときに限り動作するだけであることに注意してください。

eager_py_policy = py_tf_eager_policy.SavedModelPyTFEagerPolicy(
    policy_dir, eval_py_env.time_step_spec(), eval_py_env.action_spec())

# Note that we're passing eval_py_env not eval_env.
run_episodes_and_create_video(eager_py_policy, eval_py_env, eval_py_env)
 

以上






TF-Agents 0.4 : Tutorials : ネットワーク

TF-Agents 0.4 Tutorials : ネットワーク (翻訳/解説)

翻訳 : (株)クラスキャット セールスインフォメーション
作成日時 : 04/22/2020 (0.4)

* 本ページは、TF Agents の以下のドキュメントを翻訳した上で適宜、補足説明したものです:

* サンプルコードの動作確認はしておりますが、必要な場合には適宜、追加改変しています。
* ご自由にリンクを張って頂いてかまいませんが、sales-info@classcat.com までご一報いただけると嬉しいです。

 

ネットワーク

イントロダクション

この colab では貴方のエージェントのためにどのようにカスタム・ネットワークを定義するかをカバーします。ネットワークはエージェントにより訓練されるモデルを定義することを手助けします。TF-Agents では幾つかの異なるタイプのネットワークを見つけるでしょう、これらはエージェントに渡り有用です :

 
主要ネットワーク

  • QNetwork: 離散アクションを持つ環境のために Qlearning で使用されます、このネットワークは観測を各々の可能なアクションのための値推定にマップします。
  • CriticNetworks: 文献では ValueNetworks としても参照されます、ある状態をポリシーの期待リターンのための推定にマップする価値 (Value) 関数のあるバージョンを推定するために学習します。これらのネットワークはエージェントが現在在る状態がどれほど良いかを推定します。
  • ActorNetworks: 観測からアクションへのマッピングを学習します。これらのネットワークは通常はアクションを生成するためにポリシーにより使用されます。
  • ActorDistributionNetworks: ActorNetworks に類似していますが、これらはポリシーがアクションを生成するためにサンプリングできる分布を生成します。

 
Helper ネットワーク

  • EncodingNetwork: ネットワークの入力に適用する前処理層のマッピングを容易に定義することをユーザに可能にします。
  • DynamicUnrollLayer:それが時間シークエンスに渡り適用されるときエピソード境界上でネットワークの状態を自動的にリセットします。
  • ProjectionNetwork: CategoricalProjectionNetwork or NormalProjectionNetwork のようなネットワークは入力を取りカテゴリカル or 正規分布を生成するために必要なパラメータを生成します。

TF-Agents の総てのサンプルは事前構成されたネットワークを装備しています。けれどもこれらのネットワークは複雑な観測を処理するためにセットアップされていません。

一つの観測/アクションより多くを公開する環境を持ちネットワークをカスタマイズする必要がある場合には、このチュートリアルは貴方のためです!

 

セットアップ

tf-agents をまだインストールしてないのであれば、以下を実行します :

!pip install --upgrade tensorflow_probability
!pip install tf-agents
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function

import abc
import tensorflow as tf
import numpy as np

from tf_agents.environments import random_py_environment
from tf_agents.environments import tf_py_environment
from tf_agents.networks import encoding_network
from tf_agents.networks import network
from tf_agents.networks import utils
from tf_agents.specs import array_spec
from tf_agents.utils import common as common_utils
from tf_agents.utils import nest_utils

tf.compat.v1.enable_v2_behavior()

 

ネットワークを定義する

Network API

TF-Agents では Keras ネットワーク からサブクラス化します。それで以下が可能です :

  • ターゲットネットワークを作成するとき必要な copy 演算を単純化する。
  • network.variables() を呼び出すとき自動変数作成を遂行する。
  • ネットワーク input_specs に基づいて入力を検証する。

 

EncodingNetwork

上で言及したように EncodingNetwork はあるエンコーディングを生成するためにネットワークの入力に適用する前処理層のマッピングを容易に定義することを可能にします。

EncodingNetwork は以下の殆どオプションの層から成ります :

  • 前処理層
  • 前処理 combiner
  • Conv2D
  • Flatten
  • Dense

encoding ネットワークの特別なことは入力前処理が適用されることです。入力前処理は preprocessing_layers と preprocessing_combiner 層を通して可能です。これらの各々はネストされた構造として指定できます。もし preprocessing_layers ネストが input_tensor_spec よりも浅い場合、層はサブネストを得ます。例えば、

if :
    input_tensor_spec = ([TensorSpec(3)] * 2, [TensorSpec(3)] * 5)
    preprocessing_layers = (Layer1(), Layer2())

then 前処理は次を呼び出します :
    preprocessed = [preprocessing_layers[0](observations[0]),
                preprocessing_layers[1](obsrevations[1])]

けれども if
    preprocessing_layers = ([Layer1() for _ in range(2)],
                        [Layer2() for _ in range(5)])

then 前処理は次を呼び出します :
    preprocessed = [
      layer(obs) for layer, obs in zip(flatten(preprocessing_layers),
                                    flatten(observations))
    ]

 

カスタムネットワーク

貴方自身のネットワークを作成するためには __init__ と call メソッドを override しなければならないだけです。画像とベクトルを含む観測を取る ActorNetwork を作成するため EncodingNetworks について学習したことを使用してカスタムネットワークを作成しましょう。

class ActorNetwork(network.Network):

  def __init__(self,
               observation_spec,
               action_spec,
               preprocessing_layers=None,
               preprocessing_combiner=None,
               conv_layer_params=None,
               fc_layer_params=(75, 40),
               dropout_layer_params=None,
               activation_fn=tf.keras.activations.relu,
               enable_last_layer_zero_initializer=False,
               name='ActorNetwork'):
    super(ActorNetwork, self).__init__(
        input_tensor_spec=observation_spec, state_spec=(), name=name)

    # For simplicity we will only support a single action float output.
    self._action_spec = action_spec
    flat_action_spec = tf.nest.flatten(action_spec)
    if len(flat_action_spec) > 1:
      raise ValueError('Only a single action is supported by this network')
    self._single_action_spec = flat_action_spec[0]
    if self._single_action_spec.dtype not in [tf.float32, tf.float64]:
      raise ValueError('Only float actions are supported by this network.')

    kernel_initializer = tf.keras.initializers.VarianceScaling(
        scale=1. / 3., mode='fan_in', distribution='uniform')
    self._encoder = encoding_network.EncodingNetwork(
        observation_spec,
        preprocessing_layers=preprocessing_layers,
        preprocessing_combiner=preprocessing_combiner,
        conv_layer_params=conv_layer_params,
        fc_layer_params=fc_layer_params,
        dropout_layer_params=dropout_layer_params,
        activation_fn=activation_fn,
        kernel_initializer=kernel_initializer,
        batch_squash=False)

    initializer = tf.keras.initializers.RandomUniform(
        minval=-0.003, maxval=0.003)

    self._action_projection_layer = tf.keras.layers.Dense(
        flat_action_spec[0].shape.num_elements(),
        activation=tf.keras.activations.tanh,
        kernel_initializer=initializer,
        name='action')

  def call(self, observations, step_type=(), network_state=()):
    outer_rank = nest_utils.get_outer_rank(observations, self.input_tensor_spec)
    # We use batch_squash here in case the observations have a time sequence
    # compoment.
    batch_squash = utils.BatchSquash(outer_rank)
    observations = tf.nest.map_structure(batch_squash.flatten, observations)

    state, network_state = self._encoder(
        observations, step_type=step_type, network_state=network_state)
    actions = self._action_projection_layer(state)
    actions = common_utils.scale_to_spec(actions, self._single_action_spec)
    actions = batch_squash.unflatten(actions)
    return tf.nest.pack_sequence_as(self._action_spec, [actions]), network_state

構造化観測を生成して実装を確認するために RandomPyEnvironment を作成しましょう。

action_spec = array_spec.BoundedArraySpec((3,), np.float32, minimum=0, maximum=10)
observation_spec =  {
    'image': array_spec.BoundedArraySpec((16, 16, 3), np.float32, minimum=0,
                                        maximum=255),
    'vector': array_spec.BoundedArraySpec((5,), np.float32, minimum=-100,
                                          maximum=100)}

random_env = random_py_environment.RandomPyEnvironment(observation_spec, action_spec=action_spec)

# Convert the environment to a TFEnv to generate tensors.
tf_env = tf_py_environment.TFPyEnvironment(random_env)

観測を辞書として定義しましたので、これらを処理するために前処理層を作成する必要があります。

preprocessing_layers = {
    'image': tf.keras.models.Sequential([tf.keras.layers.Conv2D(8, 4),
                                        tf.keras.layers.Flatten()]),
    'vector': tf.keras.layers.Dense(5)
    }
preprocessing_combiner = tf.keras.layers.Concatenate(axis=-1)
actor = ActorNetwork(tf_env.observation_spec(), 
                     tf_env.action_spec(),
                     preprocessing_layers=preprocessing_layers,
                     preprocessing_combiner=preprocessing_combiner)

actor ネットワークを持つ今、環境からの観測を処理できます。

time_step = tf_env.reset()
actor(time_step.observation, time_step.step_type)

エージェントに使用される任意の主要ネットワークをカスタマイズするために同じストラテジーが使用できます。前処理が何であろうと定義してそれをネットワークの残りに接続できます。貴方自身のカスタムを定義するときネットワークの出力層定義が適合することを確実にしてください。

 

以上






TF-Agents 0.4 : Tutorials : SAC minitaur

TF-Agents 0.4 Tutorials : SAC minitaur (翻訳/解説)

翻訳 : (株)クラスキャット セールスインフォメーション
作成日時 : 04/21/2020 (0.4)

* 本ページは、TF Agents の以下のドキュメントを翻訳した上で適宜、補足説明したものです:

* サンプルコードの動作確認はしておりますが、必要な場合には適宜、追加改変しています。
* ご自由にリンクを張って頂いてかまいませんが、sales-info@classcat.com までご一報いただけると嬉しいです。

 

SAC minitaur

イントロダクション

このサンプルは TF-Agents ライブラリを使用して Minitaur 環境上 Soft Actor Critic エージェントをどのように訓練するかを示します。

DQN Colab を通して作業したのであれば、これは非常に馴染みやすいはずです。顕著な変更は以下を含みます :

  • エージェントを DQN から SAC に変更する。
  • Minitaur 上で訓練します、これはカートポールよりも遥かに複雑な環境です。Minitaur 環境は四足ロボットを前に進むように訓練することを目的とします。
  • 初期データ収集を遂行するためにランダムポリシーは使用しません。

以下の依存性をまだインストールしてないのであれば、以下を実行します :

!sudo apt-get install -y xvfb ffmpeg
!pip install 'gym==0.10.11'
!pip install 'imageio==2.4.0'
!pip install matplotlib
!pip install PILLOW
!pip install --upgrade tensorflow-probability
!pip install tf-agents
!pip install 'pybullet==2.4.2'

 

セットアップ

最初に必要な各種ツールをインポートして TF-V2 動作を有効にします、何故ならば colab を通して Eager モードで反復することは容易だからです。

from __future__ import absolute_import
from __future__ import division
from __future__ import print_function

import base64
import imageio
import IPython
import matplotlib
import matplotlib.pyplot as plt
import PIL.Image

import tensorflow as tf
tf.compat.v1.enable_v2_behavior()

from tf_agents.agents.ddpg import critic_network
from tf_agents.agents.sac import sac_agent
from tf_agents.drivers import dynamic_step_driver
from tf_agents.environments import suite_pybullet
from tf_agents.environments import tf_py_environment
from tf_agents.eval import metric_utils
from tf_agents.metrics import tf_metrics
from tf_agents.networks import actor_distribution_network
from tf_agents.networks import normal_projection_network
from tf_agents.policies import greedy_policy
from tf_agents.policies import random_tf_policy
from tf_agents.replay_buffers import tf_uniform_replay_buffer
from tf_agents.trajectories import trajectory
from tf_agents.utils import common

 

ハイパーパラメータ

env_name = "MinitaurBulletEnv-v0" # @param {type:"string"}

# use "num_iterations = 1e6" for better results,
# 1e5 is just so this doesn't take too long. 
num_iterations = 100000 # @param {type:"integer"}

initial_collect_steps = 10000 # @param {type:"integer"} 
collect_steps_per_iteration = 1 # @param {type:"integer"}
replay_buffer_capacity = 1000000 # @param {type:"integer"}

batch_size = 256 # @param {type:"integer"}

critic_learning_rate = 3e-4 # @param {type:"number"}
actor_learning_rate = 3e-4 # @param {type:"number"}
alpha_learning_rate = 3e-4 # @param {type:"number"}
target_update_tau = 0.005 # @param {type:"number"}
target_update_period = 1 # @param {type:"number"}
gamma = 0.99 # @param {type:"number"}
reward_scale_factor = 1.0 # @param {type:"number"}
gradient_clipping = None # @param

actor_fc_layer_params = (256, 256)
critic_joint_fc_layer_params = (256, 256)

log_interval = 5000 # @param {type:"integer"}

num_eval_episodes = 30 # @param {type:"integer"}
eval_interval = 10000 # @param {type:"integer"}

 

環境

RL の環境は解こうとしているタスクや問題を表します。標準的な環境はスーツを使用して TF-Agents で容易に作成できます。文字列環境名が与えられたとき、OpenAI Gym, Atari, DM Control 等のようなソースから環境をロードするために様々なスーツを持ちます。

今は Pybullet スーツから Minituar 環境をロードしましょう。

env = suite_pybullet.load(env_name)
env.reset()
PIL.Image.fromarray(env.render())

この環境では目標はエージェントが Minitaur ロボットを制御するポリシーを訓練してできる限り速く前に進ませることです。エピソードは 1000 ステップ続いてリターンはエピソードを通した報酬の総計です。環境が観測として提供する情報を見ましょう、これをポリシーはアクションを生成するために利用します。

print('Observation Spec:')
print(env.time_step_spec().observation)
print('Action Spec:')
print(env.action_spec())

見れるように観測は非常に複雑です。総てのモータのための角度、速度とトルクを表す 28 値を受け取ります。返しでは環境は [-1, 1] の間のアクションのための 8 値を想定します。これらは望まれるモーター角度です。

通常は 2 つの環境を作成します : 一つは訓練のためそして 1 つは評価のためです。殆どの環境は純粋は python で書かれますが、それらは TFPyEnvironment ラッパーを使用して容易に TensorFlow に変換できます。元の環境の API は numpy 配列を使用し、TFPyEnvironment はこれらを TensorFlow ポリシーとエージェントとより容易に相互作用するために貴方のために Tensor に/から変換します。

train_py_env = suite_pybullet.load(env_name)
eval_py_env = suite_pybullet.load(env_name)

train_env = tf_py_environment.TFPyEnvironment(train_py_env)
eval_env = tf_py_environment.TFPyEnvironment(eval_py_env)

 

エージェント

SAC エージェントを作成するため、それが訓練するネットワークを最初に作成する必要があります。SAC は actor-critic エージェントですので、2 つのネットワークを必要とします。

critic (批評家) は Q(s, a) のための値推定を与えます。つまり、それは入力として観測とアクションを受け取り、そしてそれは与えられた状態のためにアクションがどのくらい良いかの推定を与えます。

observation_spec = train_env.observation_spec()
action_spec = train_env.action_spec()
critic_net = critic_network.CriticNetwork(
    (observation_spec, action_spec),
    observation_fc_layer_params=None,
    action_fc_layer_params=None,
    joint_fc_layer_params=critic_joint_fc_layer_params)

観測が与えられたときアクションを生成することを可能にする actor ネットワークを訓練するためにこの critic を利用します。

ActorNetwork は正規分布のためのパラメータを予測します。それからアクションを生成する必要があるとき、この分布はサンプリングされ、現在の観測で条件付けられます。

def normal_projection_net(action_spec,init_means_output_factor=0.1):
  return normal_projection_network.NormalProjectionNetwork(
      action_spec,
      mean_transform=None,
      state_dependent_std=True,
      init_means_output_factor=init_means_output_factor,
      std_transform=sac_agent.std_clip_transform,
      scale_distribution=True)


actor_net = actor_distribution_network.ActorDistributionNetwork(
    observation_spec,
    action_spec,
    fc_layer_params=actor_fc_layer_params,
    continuous_projection_net=normal_projection_net)

手元のこれらのネットワークで今ではエージェントをインスタンス化できます。

global_step = tf.compat.v1.train.get_or_create_global_step()
tf_agent = sac_agent.SacAgent(
    train_env.time_step_spec(),
    action_spec,
    actor_network=actor_net,
    critic_network=critic_net,
    actor_optimizer=tf.compat.v1.train.AdamOptimizer(
        learning_rate=actor_learning_rate),
    critic_optimizer=tf.compat.v1.train.AdamOptimizer(
        learning_rate=critic_learning_rate),
    alpha_optimizer=tf.compat.v1.train.AdamOptimizer(
        learning_rate=alpha_learning_rate),
    target_update_tau=target_update_tau,
    target_update_period=target_update_period,
    td_errors_loss_fn=tf.compat.v1.losses.mean_squared_error,
    gamma=gamma,
    reward_scale_factor=reward_scale_factor,
    gradient_clipping=gradient_clipping,
    train_step_counter=global_step)
tf_agent.initialize()

 

ポリシー

TF-Agents では、ポリシーは RL のポリシーの標準的な概念を表します : time_step が与えられたときアクションかアクションに渡る分布を生成します。主要メソッドは policy_step = policy.step(time_step) です、そこでは policy_step は名前付きタプル PolicyStep(action, state, info) です。policy_step.action は環境に適用されるべきアクションで、state はステートフル (RNN) ポリシーのための状態を表して info はアクションの対数確率のような補助的情報を含むかもしれません。

エージェントは 2 つのポリシーを含みます : 主要ポリシー (agent.policy) とデータ収集のために使用される行動 (= behavioral) ポリシー (agent.collect_policy) です。評価/配備については、主要ポリシーを GreedyPolicy() でラップすることにより平均 (= mean) アクションを取ります。

eval_policy = greedy_policy.GreedyPolicy(tf_agent.policy)
collect_policy = tf_agent.collect_policy

 

メトリクスと評価

ポリシーを評価するために使用される最も一般的なメトリックは平均リターンです。リターンはエピソードのための環境でポリシーを実行する間に得られた報酬の総計で、通常はこれを幾つかのエピソードに渡り平均します。次のように平均リターン・メトリックを計算できます。

def compute_avg_return(environment, policy, num_episodes=5):

  total_return = 0.0
  for _ in range(num_episodes):

    time_step = environment.reset()
    episode_return = 0.0

    while not time_step.is_last():
      action_step = policy.action(time_step)
      time_step = environment.step(action_step.action)
      episode_return += time_step.reward
    total_return += episode_return

  avg_return = total_return / num_episodes
  return avg_return.numpy()[0]


compute_avg_return(eval_env, eval_policy, num_eval_episodes)

# Please also see the metrics module for standard implementations of different
# metrics.

 

再生バッファ

環境から収集されたデータを追跡するために、TFUniformReplayBuffer を利用します。この再生バッファはストアされる tensor を記述する specs を使用して構築されます、これは tf_agent.collect_data_spec を使用してエージェントから得られます。

replay_buffer = tf_uniform_replay_buffer.TFUniformReplayBuffer(
    data_spec=tf_agent.collect_data_spec,
    batch_size=train_env.batch_size,
    max_length=replay_buffer_capacity)

殆どのエージェントについて、collect_data_spec は Trajectory と名前付けられたタプルで観測、アクション、報酬等を含みます。

 

データ収集

今はそれで再生バッファを seed するための経験を収集するためにドライバーを作成します。ドライバーは指定されたポリシーを使用して環境上の n ステップかエピソードを収集する単純な方法を提供します。

initial_collect_driver = dynamic_step_driver.DynamicStepDriver(
        train_env,
        collect_policy,
        observers=[replay_buffer.add_batch],
        num_steps=initial_collect_steps)
initial_collect_driver.run()

再生バッファからデータをサンプリングするため、tf.data パイプラインを作成します、これを後で訓練のためにエージェントに供給できます。再生バッファからサンプリングされた項目の数を設定するために sample_batch_size を指定できます。並列呼び出しと先取りを使用してデータパイプラインを最適化することもできます。

スペースを節約するため、再生バッファの各行で現在の観測をストアするだけです。しかし SAC エージェントは損失を計算するために現在と次の観測の両者を必要としますので、num_steps=2 を設定することによりバッチの各項目に対して 2 つの隣接する行を常にサンプリングします。

# Dataset generates trajectories with shape [Bx2x...]
dataset = replay_buffer.as_dataset(
    num_parallel_calls=3, sample_batch_size=batch_size, num_steps=2).prefetch(3)

iterator = iter(dataset)

 

エージェントを訓練する

訓練ループは環境からのデータ収集とエージェントのネットワークの最適化の両者を含みます。道中、どのようにやっているかを見るためにエージェントのポリシーを時々評価します。

collect_driver = dynamic_step_driver.DynamicStepDriver(
    train_env,
    collect_policy,
    observers=[replay_buffer.add_batch],
    num_steps=collect_steps_per_iteration)
#@test {"skip": true}
try:
  %%time
except:
  pass

# (Optional) Optimize by wrapping some of the code in a graph using TF function.
tf_agent.train = common.function(tf_agent.train)
collect_driver.run = common.function(collect_driver.run)

# Reset the train step
tf_agent.train_step_counter.assign(0)

# Evaluate the agent's policy once before training.
avg_return = compute_avg_return(eval_env, eval_policy, num_eval_episodes)
returns = [avg_return]

for _ in range(num_iterations):

  # Collect a few steps using collect_policy and save to the replay buffer.
  for _ in range(collect_steps_per_iteration):
    collect_driver.run()

  # Sample a batch of data from the buffer and update the agent's network.
  experience, unused_info = next(iterator)
  train_loss = tf_agent.train(experience)

  step = tf_agent.train_step_counter.numpy()

  if step % log_interval == 0:
    print('step = {0}: loss = {1}'.format(step, train_loss.loss))

  if step % eval_interval == 0:
    avg_return = compute_avg_return(eval_env, eval_policy, num_eval_episodes)
    print('step = {0}: Average Return = {1}'.format(step, avg_return))
    returns.append(avg_return)

 

可視化

プロット

エージェントのパフォーマンスを見るために平均リターン vs グルーバルステップをプロットできます。Minitaur では、報酬関数は minitaur が 1000 ステップでどこまで遠くまで歩くかに基づきそしてエネルギー消費にペナルティを与えるます。

#@test {"skip": true}

steps = range(0, num_iterations + 1, eval_interval)
plt.plot(steps, returns)
plt.ylabel('Average Return')
plt.xlabel('Step')
plt.ylim()

 

ビデオ

各ステップで環境をレンダリングすることによりエージェントのパフォーマンスを可視化することは役立ちます。それを行なう前に、ビデオをこの colab に埋め込む関数を最初に作成しましょう。

def embed_mp4(filename):
  """Embeds an mp4 file in the notebook."""
  video = open(filename,'rb').read()
  b64 = base64.b64encode(video)
  tag = '''
  '''.format(b64.decode())

  return IPython.display.HTML(tag)

以下のコードは幾つかのエピソードのためにエージェントのポリシーを可視化します :

num_episodes = 3
video_filename = 'sac_minitaur.mp4'
with imageio.get_writer(video_filename, fps=60) as video:
  for _ in range(num_episodes):
    time_step = eval_env.reset()
    video.append_data(eval_py_env.render())
    while not time_step.is_last():
      action_step = tf_agent.policy.action(time_step)
      time_step = eval_env.step(action_step.action)
      video.append_data(eval_py_env.render())

embed_mp4(video_filename)
 

以上






TF-Agents 0.4 : Tutorials : REINFORCE エージェント

TF-Agents 0.4 Tutorials : REINFORCE エージェント (翻訳/解説)

翻訳 : (株)クラスキャット セールスインフォメーション
作成日時 : 04/21/2020 (0.4)

* 本ページは、TF Agents の以下のドキュメントを翻訳した上で適宜、補足説明したものです:

* サンプルコードの動作確認はしておりますが、必要な場合には適宜、追加改変しています。
* ご自由にリンクを張って頂いてかまいませんが、sales-info@classcat.com までご一報いただけると嬉しいです。

 

REINFORCE エージェント

イントロダクション

このサンプルは TF-Agents ライブラリを使用してカートポール環境上 REINFORCE エージェントをどのように訓練するかを示します、DQN チュートリアル に似ています。

訓練、評価とデータ収集のための強化学習 (RL) パイプラインの総てのコンポーネントを通して貴方をガイドします。

 

セットアップ

以下の依存性をインストールしていないのであれば、次を実行します :

!sudo apt-get install -y xvfb ffmpeg
!pip install 'gym==0.10.11'
!pip install 'imageio==2.4.0'
!pip install PILLOW
!pip install 'pyglet==1.3.2'
!pip install pyvirtualdisplay
!pip install --upgrade tensorflow-probability
!pip install tf-agents
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function

import base64
import imageio
import IPython
import matplotlib
import matplotlib.pyplot as plt
import numpy as np
import PIL.Image
import pyvirtualdisplay

import tensorflow as tf

from tf_agents.agents.reinforce import reinforce_agent
from tf_agents.drivers import dynamic_step_driver
from tf_agents.environments import suite_gym
from tf_agents.environments import tf_py_environment
from tf_agents.eval import metric_utils
from tf_agents.metrics import tf_metrics
from tf_agents.networks import actor_distribution_network
from tf_agents.replay_buffers import tf_uniform_replay_buffer
from tf_agents.trajectories import trajectory
from tf_agents.utils import common

tf.compat.v1.enable_v2_behavior()


# Set up a virtual display for rendering OpenAI gym environments.
display = pyvirtualdisplay.Display(visible=0, size=(1400, 900)).start()

 

ハイパーパラメータ

env_name = "CartPole-v0" # @param {type:"string"}
num_iterations = 250 # @param {type:"integer"}
collect_episodes_per_iteration = 2 # @param {type:"integer"}
replay_buffer_capacity = 2000 # @param {type:"integer"}

fc_layer_params = (100,)

learning_rate = 1e-3 # @param {type:"number"}
log_interval = 25 # @param {type:"integer"}
num_eval_episodes = 10 # @param {type:"integer"}
eval_interval = 50 # @param {type:"integer"}

 

環境

RL の環境は解こうとしているタスクや問題を表します。標準的な環境はスーツを使用して TF-Agents で容易に作成できます。文字列環境名が与えられたとき、OpenAI Gym, Atari, DM Control 等のようなソースから環境をロードするために様々なスーツを持ちます。

今は OpenAI Gym スーツからカートポール環境をロードしましょう。

env = suite_gym.load(env_name)

この環境をそれがどのようなものか見るためにレンダリングできます。free-swinging ポールはカートに装着されています。ゴールはポールを直立に維持するためにカートを右か左に動かすことです。

#@test {"skip": true}
env.reset()
PIL.Image.fromarray(env.render())

time_step = environment.step(action) ステートメントは環境でアクションを取ります。返される TimeStep タプルはそのアクションのための環境の次の観測と報酬を含みます。環境の time_step_spec() と action_spec() メソッドは time_step とアクションの仕様 (types, shapes, bounds) をそれぞれ返します。

print('Observation Spec:')
print(env.time_step_spec().observation)
print('Action Spec:')
print(env.action_spec())

そして、観測は 4 floats の配列であることを見ます : カートの位置と速度、そしてポールの角度位置と角速度です。2 つのアクション (左に動かす or 右に動かす) だけが可能ですので、action_spec はスカラーでそこでは 0 は「左に動かす」を意味して 1 は「右に動かす」を意味します。

time_step = env.reset()
print('Time step:')
print(time_step)

action = np.array(1, dtype=np.int32)

next_time_step = env.step(action)
print('Next time step:')
print(next_time_step)

通常は 2 つの環境を作成します : 訓練のために一つと評価のために一つです。殆どの環境は純粋な python で書かれていますが、それらは TFPyEnvironment ラッパーを使用して容易に TensorFlow に変換できます。元の環境の API は numpy 配列を使用し、TFPyEnvironment はこれらを TensorFlow ポリシーとエージェントとより容易に相互作用するために貴方のために Tensor に/から変換します。

train_py_env = suite_gym.load(env_name)
eval_py_env = suite_gym.load(env_name)

train_env = tf_py_environment.TFPyEnvironment(train_py_env)
eval_env = tf_py_environment.TFPyEnvironment(eval_py_env)

 

エージェント

RL 問題を解くために使用するアルゴリズムはエージェントとして表されます。REINFORCE エージェントに加えて、TF-Agents は DQN, DDPG, TD3, PPOSAC のような様々なエージェントの標準的な実装を提供します。

REINFORCE エージェントを作成するため、最初に Actor ネットワークを必要とします、これは環境からの観測が与えられたときアクションを予測することを学習できます。

観測とアクションの specs を使用して容易に Actor ネットワークを作成できます。ネットワークの層を指定できます、これは例えば各隠れ層のサイズを表す ints のタプルに設定される fc_layer_params 引数です (上のハイパーパラメータ・セクション参照)。

actor_net = actor_distribution_network.ActorDistributionNetwork(
    train_env.observation_spec(),
    train_env.action_spec(),
    fc_layer_params=fc_layer_params)

作成したばかりのネットワークを訓練するために optimizer と,ネットワークが何回更新されたかを追跡するための train_step_counter 変数もまた必要です。

optimizer = tf.compat.v1.train.AdamOptimizer(learning_rate=learning_rate)

train_step_counter = tf.compat.v2.Variable(0)

tf_agent = reinforce_agent.ReinforceAgent(
    train_env.time_step_spec(),
    train_env.action_spec(),
    actor_network=actor_net,
    optimizer=optimizer,
    normalize_returns=True,
    train_step_counter=train_step_counter)
tf_agent.initialize()

 

ポリシー

TF-Agents では、ポリシーは RL のポリシーの標準的な概念を表します : time_step が与えられたときアクションかアクションに渡る分布を生成します。主要メソッドは policy_step = policy.step(time_step) です、そこでは policy_step は名前付きタプル PolicyStep(action, state, info) です。policy_step.action は環境に適用されるべきアクションで、state はステートフル (RNN) ポリシーのための状態を表して info はアクションの対数確率のような補助的情報を含むかもしれません。

エージェントは 2 つのポリシーを含みます : 主要ポリシーは評価/配備のため (agent.policy) に使用されてもう一つのポリシーはデータ収集のため (agent.collect_policy) に使用されます。

eval_policy = tf_agent.policy
collect_policy = tf_agent.collect_policy

 

メトリクスと評価

ポリシーを評価するために使用される最も一般的なメトリックは平均リターンです。リターンはエピソードのための環境でポリシーを実行する間に得られた報酬の総計で、通常はこれを幾つかのエピソードに渡り平均します。次のように平均リターン・メトリックを計算できます。

def compute_avg_return(environment, policy, num_episodes=10):

  total_return = 0.0
  for _ in range(num_episodes):

    time_step = environment.reset()
    episode_return = 0.0

    while not time_step.is_last():
      action_step = policy.action(time_step)
      time_step = environment.step(action_step.action)
      episode_return += time_step.reward
    total_return += episode_return

  avg_return = total_return / num_episodes
  return avg_return.numpy()[0]


# Please also see the metrics module for standard implementations of different
# metrics.

 

再生バッファ

環境から収集されたデータを追跡するために、TFUniformReplayBuffer を利用します。この再生バッファはストアされる tensor を記述する specs を使用して構築されます、これは tf_agent.collect_data_spec を使用してエージェントから得られます。

replay_buffer = tf_uniform_replay_buffer.TFUniformReplayBuffer(
    data_spec=tf_agent.collect_data_spec,
    batch_size=train_env.batch_size,
    max_length=replay_buffer_capacity)

殆どのエージェントについて、collect_data_spec は Trajectory と名前付けられたタプルで観測、アクション、報酬等を含みます。

 

データ収集

REINFORCE はエピソード全体から学習しますので、与えられたデータ収集ポリシーを使用してエピソードを集める関数を定義します、そしてデータ (観測、アクション、報酬等) を再生バッファの trajectories としてセーブします。

#@test {"skip": true}

def collect_episode(environment, policy, num_episodes):

  episode_counter = 0
  environment.reset()

  while episode_counter < num_episodes:
    time_step = environment.current_time_step()
    action_step = policy.action(time_step)
    next_time_step = environment.step(action_step.action)
    traj = trajectory.from_transition(time_step, action_step, next_time_step)

    # Add trajectory to the replay buffer
    replay_buffer.add_batch(traj)

    if traj.is_boundary():
      episode_counter += 1


# This loop is so common in RL, that we provide standard implementations of
# these. For more details see the drivers module.

 

エージェントを訓練する

訓練ループは環境からのデータ収集とエージェントのネットワークの最適化の両者を含みます。道中、どのようにやっているかを見るためにエージェントのポリシーを時々評価します。

以下は実行するために ~3 分かかります。

#@test {"skip": true}
try:
  %%time
except:
  pass

# (Optional) Optimize by wrapping some of the code in a graph using TF function.
tf_agent.train = common.function(tf_agent.train)

# Reset the train step
tf_agent.train_step_counter.assign(0)

# Evaluate the agent's policy once before training.
avg_return = compute_avg_return(eval_env, tf_agent.policy, num_eval_episodes)
returns = [avg_return]

for _ in range(num_iterations):

  # Collect a few episodes using collect_policy and save to the replay buffer.
  collect_episode(
      train_env, tf_agent.collect_policy, collect_episodes_per_iteration)

  # Use data from the buffer and update the agent's network.
  experience = replay_buffer.gather_all()
  train_loss = tf_agent.train(experience)
  replay_buffer.clear()

  step = tf_agent.train_step_counter.numpy()

  if step % log_interval == 0:
    print('step = {0}: loss = {1}'.format(step, train_loss.loss))

  if step % eval_interval == 0:
    avg_return = compute_avg_return(eval_env, tf_agent.policy, num_eval_episodes)
    print('step = {0}: Average Return = {1}'.format(step, avg_return))
    returns.append(avg_return)

 

可視化

プロット

エージェントのパフォーマンスを見るためにリターン vs グルーバルステップをプロットできます。Cartpole-v0 では、ポールが直立している総ての時間ステップについて環境は +1 の報酬を与えます、そしてステップの最大数は 200 ですから、最大の可能なリターンもまた 200 です。

#@test {"skip": true}

steps = range(0, num_iterations + 1, eval_interval)
plt.plot(steps, returns)
plt.ylabel('Average Return')
plt.xlabel('Step')
plt.ylim(top=250)

 

ビデオ

各ステップで環境をレンダリングすることによりエージェントのパフォーマンスを可視化することは役立ちます。それを行なう前に、ビデオをこの colab に埋め込む関数を最初に作成しましょう。

def embed_mp4(filename):
  """Embeds an mp4 file in the notebook."""
  video = open(filename,'rb').read()
  b64 = base64.b64encode(video)
  tag = '''
  '''.format(b64.decode())

  return IPython.display.HTML(tag)

以下のコードは幾つかのエピソードのためにエージェントのポリシーを可視化します :

num_episodes = 3
video_filename = 'imageio.mp4'
with imageio.get_writer(video_filename, fps=60) as video:
  for _ in range(num_episodes):
    time_step = eval_env.reset()
    video.append_data(eval_py_env.render())
    while not time_step.is_last():
      action_step = tf_agent.policy.action(time_step)
      time_step = eval_env.step(action_step.action)
      video.append_data(eval_py_env.render())

embed_mp4(video_filename)
 

以上






TF-Agents 0.4 : Tutorials : 再生バッファ

TF-Agents 0.4 Tutorials : 再生バッファ (翻訳/解説)

翻訳 : (株)クラスキャット セールスインフォメーション
作成日時 : 04/21/2020 (0.4)

* 本ページは、TF Agents の以下のドキュメントを翻訳した上で適宜、補足説明したものです:

* サンプルコードの動作確認はしておりますが、必要な場合には適宜、追加改変しています。
* ご自由にリンクを張って頂いてかまいませんが、sales-info@classcat.com までご一報いただけると嬉しいです。

 

再生バッファ

イントロダクション

強化学習アルゴリズムは環境でポリシーを実行するとき経験の軌道 (= trajectories) をストアするために再生バッファを利用します。訓練の間、エージェントの経験を「再生する」ために軌道のサブセット (シークエンシャルなサブセットかサンプル) のために再生バッファが問い合わせがされます。

この colab では、2 つのタイプの再生バッファを調べます : python-支援と tensorflow-支援で、共通 API を共有します。次のセクションで、API、各々のバッファ実装そしてそれらをデータ収集訓練の間にどのように利用するかを説明します。

 

セットアップ

tf-agents をまだであればインストールします。

!pip install --upgrade tensorflow-probability
!pip install tf-agents
!pip install gym
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function

import tensorflow as tf
import numpy as np

from tf_agents import specs
from tf_agents.agents.dqn import dqn_agent
from tf_agents.drivers import dynamic_step_driver
from tf_agents.environments import suite_gym
from tf_agents.environments import tf_py_environment
from tf_agents.networks import q_network
from tf_agents.replay_buffers import py_uniform_replay_buffer
from tf_agents.replay_buffers import tf_uniform_replay_buffer
from tf_agents.specs import tensor_spec
from tf_agents.trajectories import time_step

tf.compat.v1.enable_v2_behavior()

 

再生バッファ API

再生バッファ・クラスは次の定義とメソッドを持ちます :

class ReplayBuffer(tf.Module):
  """Abstract base class for TF-Agents replay buffer."""

  def __init__(self, data_spec, capacity):
    """Initializes the replay buffer.

    Args:
      data_spec: A spec or a list/tuple/nest of specs describing
        a single item that can be stored in this buffer
      capacity: number of elements that the replay buffer can hold.
    """

  @property
  def data_spec(self):
    """Returns the spec for items in the replay buffer."""

  @property
  def capacity(self):
    """Returns the capacity of the replay buffer."""

  def add_batch(self, items):
    """Adds a batch of items to the replay buffer."""

  def get_next(self,
               sample_batch_size=None,
               num_steps=None,
               time_stacked=True):
    """Returns an item or batch of items from the buffer."""

  def as_dataset(self,
                 sample_batch_size=None,
                 num_steps=None,
                 num_parallel_calls=None):
    """Creates and returns a dataset that returns entries from the buffer."""


  def gather_all(self):
    """Returns all the items in buffer."""
    return self._gather_all()

  def clear(self):
    """Resets the contents of replay buffer"""

再生バッファが初期化されるとき、それがストアする要素の data_spec を要求することに注意してください。この spec はバッファに追加される trajectory 要素の TensorSpec に相当します。この spec は通常はエージェントの agent.collect_data_spec を見ることにより獲得されます、これは訓練の時 (more on that later) エージェントにより想定される shape, 型と構造を定義します。

 

TFUniformReplayBuffer

TFUniformReplayBuffer は TF-Agents で最も一般的に使用される再生バッファで、そのためここでチュートリアル内でそれを使用します。TFUniformReplayBuffer では支援 (= backing) バッファ・ストレージは tensorflow 変数により成されるため計算グラフの一部です。

バッファは要素のバッチをストアしてバッチセグメント毎に最大容量 max_length 要素を持ちます。このため、総バッファ容量は batch_size x max_length 要素です。バッファにストアされる要素総ては適合するデータ spec を持たなければなりません。再生バッファがデータ収集のために使用されるとき、spec はエージェントの collect data spec です。

 

バッファを作成する

TFUniformReplayBuffer を作成するために以下を渡します :

  1. バッファがストアするデータ要素の spec
  2. バッファのバッチサイズに対応するバッチサイズ
  3. バッチセグメント毎の要素の max_length 数

ここにサンプル data specs, batch_size 32 と max_length 1000 を持つ TFUniformReplayBuffer を作成するサンプルがあります。

data_spec =  (
        tf.TensorSpec([3], tf.float32, 'action'),
        (
            tf.TensorSpec([5], tf.float32, 'lidar'),
            tf.TensorSpec([3, 2], tf.float32, 'camera')
        )
)

batch_size = 32
max_length = 1000

replay_buffer = tf_uniform_replay_buffer.TFUniformReplayBuffer(
    data_spec,
    batch_size=batch_size,
    max_length=max_length)

 

バッファに書く

要素を再生バッファに追加するため、add_batch(items) メソッドを使用します、そこでは項目はバッファに追加されるべき項目のバッチを表す tensor のリスト/タプル/ネストです。項目の各要素は batch_size に等しい外側の次元を持たなければならずそして残りの次元は (再生バッファに渡された data specs と同じ) 項目の data spec を忠実に守らなければなりません。

ここに項目のバッチを追加するサンプルがあります :

action = tf.constant(1 * np.ones(
    data_spec[0].shape.as_list(), dtype=np.float32))
lidar = tf.constant(
    2 * np.ones(data_spec[1][0].shape.as_list(), dtype=np.float32))
camera = tf.constant(
    3 * np.ones(data_spec[1][1].shape.as_list(), dtype=np.float32))
  
values = (action, (lidar, camera))
values_batched = tf.nest.map_structure(lambda t: tf.stack([t] * batch_size),
                                       values)
  
replay_buffer.add_batch(values_batched)

 

バッファから読む

TFUniformReplayBuffer からデータを読む 3 つの方法があります :

  1. get_next() – バッファから 1 つのサンプルを返します。返されるサンプル・バッチサイズと時間ステップの数はこのメソッドへの引数を通して指定できます。
  2. as_dataset() – 再生バッファを tf.data.Dataset として返します。それからデータセット iterator を作成できてバッファの項目のサンプルを通して反復できます。
  3. gather_all() – バッファの総ての項目を shape [batch, time, data_spec] を持つ Tensor として返します。

下はこれらのメソッドの各々を使用して再生バッファからどのように読むかのサンプルです :

# add more items to the buffer before reading
for _ in range(5):
  replay_buffer.add_batch(values_batched)

# Get one sample from the replay buffer with batch size 10 and 1 timestep:

sample = replay_buffer.get_next(sample_batch_size=10, num_steps=1)

# Convert the replay buffer to a tf.data.Dataset and iterate through it
dataset = replay_buffer.as_dataset(
    sample_batch_size=4,
    num_steps=2)

iterator = iter(dataset)
print("Iterator trajectories:")
trajectories = []
for _ in range(3):
  t, _ = next(iterator)
  trajectories.append(t)
  
print(tf.nest.map_structure(lambda t: t.shape, trajectories))

# Read all elements in the replay buffer:
trajectories = replay_buffer.gather_all()

print("Trajectories from gather all:")
print(tf.nest.map_structure(lambda t: t.shape, trajectories))

 

PyUniformReplayBuffer

PyUniformReplayBuffer は TFUniformReplayBuffer と同じ機能を持ちますが、tf 変数の代わりに、そのデータは numpy 配列でストアされます。このバッファは out-of-graph データ収集のために使用できます。TensorFlow 変数を使用することなく numpy で支援ストレージを持つことは幾つかのアプリケーションにとっては (優先度を更新するためのインデキシングのような) データ操作を行なうことを容易にします。けれども、この実装は TensorFlow によるグラフ最適化の恩恵を持ちません。

下はエージェントのポリシー trajectory specs から PyUniformReplayBuffer をインスタンス化するサンプルです :

replay_buffer_capacity = 1000*32 # same capacity as the TFUniformReplayBuffer

py_replay_buffer = py_uniform_replay_buffer.PyUniformReplayBuffer(
    capacity=replay_buffer_capacity,
    data_spec=tensor_spec.to_nest_array_spec(data_spec))

 

訓練の間に再生バッファを使用する

再生バッファをどのように作成し、項目をそれに書いてそれから読むかを知った今、エージェントの訓練の間に trajectories をストアするためにそれを利用できます。

 

データ・コレクション

最初に、データコレクションの間に再生バッファをどのように使用するかを見ましょう。

TF-Agents では環境での経験を収集するために Driver (より多くの詳細については Driver チュートリアル 参照) を使用します。Driver を使用するために、Observer を指定します、これは trajectory を受け取るとき Driver が実行するための関数です。

こうして、trajectory 要素を再生バッファに追加するため、再生バッファに項目 (のバッチ) を追加するために add_batch(items) を呼び出す観測者を追加します。

下は TFUniformReplayBuffer を伴うこれのサンプルです。最初に環境、ネットワークとエージェントを作成します。それからTFUniformReplayBuffer を作成します。再生バッファの trajectory 要素の specs はエージェントの collect data spec に等しいことに注意してください。そしてその add_batch メソッドをドライバーのための観測者として設定します、これは訓練の間にデータ収集を行ないます :

env = suite_gym.load('CartPole-v0')
tf_env = tf_py_environment.TFPyEnvironment(env)

q_net = q_network.QNetwork(
    tf_env.time_step_spec().observation,
    tf_env.action_spec(),
    fc_layer_params=(100,))

agent = dqn_agent.DqnAgent(
    tf_env.time_step_spec(),
    tf_env.action_spec(),
    q_network=q_net,
    optimizer=tf.compat.v1.train.AdamOptimizer(0.001))

replay_buffer_capacity = 1000

replay_buffer = tf_uniform_replay_buffer.TFUniformReplayBuffer(
    agent.collect_data_spec,
    batch_size=tf_env.batch_size,
    max_length=replay_buffer_capacity)

# Add an observer that adds to the replay buffer:
replay_observer = [replay_buffer.add_batch]

collect_steps_per_iteration = 10
collect_op = dynamic_step_driver.DynamicStepDriver(
  tf_env,
  agent.collect_policy,
  observers=replay_observer,
  num_steps=collect_steps_per_iteration).run()

 

訓練ステップのためにデータを読む

再生バッファに trajectory 要素を追加した後、訓練ステップのための入力データとして使用するために再生バッファから trajectory のバッチを読むことができます。

ここに訓練ループで再生バッファからの trajectory 上でどのように訓練するかのサンプルがあります :

# Read the replay buffer as a Dataset,
# read batches of 4 elements, each with 2 timesteps:
dataset = replay_buffer.as_dataset(
    sample_batch_size=4,
    num_steps=2)

iterator = iter(dataset)

num_train_steps = 10

for _ in range(num_train_steps):
  trajectories, _ = next(iterator)
  loss = agent.train(experience=trajectories)
 

以上






TF-Agents 0.4 : Tutorials : ドライバー

TF-Agents 0.4 Tutorials : ドライバー (翻訳/解説)

翻訳 : (株)クラスキャット セールスインフォメーション
作成日時 : 04/20/2020 (0.4)

* 本ページは、TF Agents の以下のドキュメントを翻訳した上で適宜、補足説明したものです:

* サンプルコードの動作確認はしておりますが、必要な場合には適宜、追加改変しています。
* ご自由にリンクを張って頂いてかまいませんが、sales-info@classcat.com までご一報いただけると嬉しいです。

 

ドライバー

イントロダクション

強化学習の一般的なパターンはステップかエピソードの指定数のために環境でポリシーを実行します。これは例えば、データ収集、評価そしてエージェントのビデオ生成する間に発生します。

これは Python で書くことは比較的簡単である一方、TensorFlow で書いてデバッグすることは遥かにより複雑です、何故ならばそれは tf.while loops, tf.cond と tf.control_dependencies を伴うからです。そのためこの実行ループの概念をドライバーと呼ばれるクラスに抽象して Python と TensorFlow の両者で良くテストされた実装を提供します。

追加で、各ステップでドライバーに遭遇したデータは Trajectory と呼ばれる名前付きタプルにセーブされて再生バッファとメトリクスのような観測者のセットにブロードキャストされます。このデータは環境からの観測、ポリシーにより勧められるアクション、得られた報酬、現在と次のステップのタイプ、等。

 

セットアップ

tf-agents か gym をまだインストールしていないのであれば、以下を実行します :

!pip install --upgrade tensorflow-probability
!pip install tf-agents
!pip install gym
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function

import tensorflow as tf


from tf_agents.environments import suite_gym
from tf_agents.environments import tf_py_environment
from tf_agents.policies import random_py_policy
from tf_agents.policies import random_tf_policy
from tf_agents.metrics import py_metrics
from tf_agents.metrics import tf_metrics
from tf_agents.drivers import py_driver
from tf_agents.drivers import dynamic_episode_driver

tf.compat.v1.enable_v2_behavior()

 

Python ドライバー

PyDriver クラスは各ステップで更新するために python 環境、python ポリシーそして観測者のリストを取ります。主要なメソッドは run() で、これはポリシーからのアクションを使用して少なくとも以下の停止基準の一つに遭遇するまで環境に踏み入ります : ステップ数が max_steps に到達するかエピソード数が max_episodes に達する。

実装はおおよそ次のようなものです :

class PyDriver(object):

  def __init__(self, env, policy, observers, max_steps=1, max_episodes=1):
    self._env = env
    self._policy = policy
    self._observers = observers or []
    self._max_steps = max_steps or np.inf
    self._max_episodes = max_episodes or np.inf

  def run(self, time_step, policy_state=()):
    num_steps = 0
    num_episodes = 0
    while num_steps < self._max_steps and num_episodes < self._max_episodes:

      # Compute an action using the policy for the given time_step
      action_step = self._policy.action(time_step, policy_state)

      # Apply the action to the environment and get the next step
      next_time_step = self._env.step(action_step.action)

      # Package information into a trajectory
      traj = trajectory.Trajectory(
         time_step.step_type,
         time_step.observation,
         action_step.action,
         action_step.info,
         next_time_step.step_type,
         next_time_step.reward,
         next_time_step.discount)

      for observer in self._observers:
        observer(traj)

      # Update statistics to check termination
      num_episodes += np.sum(traj.is_last())
      num_steps += np.sum(~traj.is_boundary())

      time_step = next_time_step
      policy_state = action_step.state

    return time_step, policy_state

今は、カートポール環境でランダムポリシーを実行するサンプルを通して実行しましょう、結果を再生バッファにセーブして幾つかのメトリクスを計算します。

env = suite_gym.load('CartPole-v0')
policy = random_py_policy.RandomPyPolicy(time_step_spec=env.time_step_spec(), 
                                         action_spec=env.action_spec())
replay_buffer = []
metric = py_metrics.AverageReturnMetric()
observers = [replay_buffer.append, metric]
driver = py_driver.PyDriver(
    env, policy, observers, max_steps=20, max_episodes=1)

initial_time_step = env.reset()
final_time_step, _ = driver.run(initial_time_step)

print('Replay Buffer:')
for traj in replay_buffer:
  print(traj)

print('Average Return: ', metric.result())

 

TensorFlow ドライバー

私達はまた TensorFlow のドライバーも持ちます、これは機能的には Python ドライバーに類似していますが、TF 環境, TF ポリシー, TF 観測者等を利用します。現在は 2 つの TensorFlow ドライバーを持ちます : DynamicStepDriver, これは (正当な) 環境ステップの与えられた数の後停止します、そして DynamicEpisodeDriver, これはエピソードの与えられた数の後停止します。アクションの DynamicEpisode のサンプルを見ましょう。

env = suite_gym.load('CartPole-v0')
tf_env = tf_py_environment.TFPyEnvironment(env)

tf_policy = random_tf_policy.RandomTFPolicy(action_spec=tf_env.action_spec(),
                                            time_step_spec=tf_env.time_step_spec())


num_episodes = tf_metrics.NumberOfEpisodes()
env_steps = tf_metrics.EnvironmentSteps()
observers = [num_episodes, env_steps]
driver = dynamic_episode_driver.DynamicEpisodeDriver(
    tf_env, tf_policy, observers, num_episodes=2)

# Initial driver.run will reset the environment and initialize the policy.
final_time_step, policy_state = driver.run()

print('final_time_step', final_time_step)
print('Number of Steps: ', env_steps.result().numpy())
print('Number of Episodes: ', num_episodes.result().numpy())
# Continue running from previous state
final_time_step, _ = driver.run(final_time_step, policy_state)

print('final_time_step', final_time_step)
print('Number of Steps: ', env_steps.result().numpy())
print('Number of Episodes: ', num_episodes.result().numpy())
 

以上






TF-Agents 0.4 : Tutorials : ポリシー

TF-Agents 0.4 Tutorials : ポリシー (翻訳/解説)

翻訳 : (株)クラスキャット セールスインフォメーション
作成日時 : 04/19/2020 (0.4)

* 本ページは、TF Agents の以下のドキュメントを翻訳した上で適宜、補足説明したものです:

* サンプルコードの動作確認はしておりますが、必要な場合には適宜、追加改変しています。
* ご自由にリンクを張って頂いてかまいませんが、sales-info@classcat.com までご一報いただけると嬉しいです。

 

ポリシー

イントロダクション

強化学習の用語では、ポリシーは環境からの観測をアクションかアクションに渡る分布にマップします。TF-Agents では、環境からの観測は名前付きタプル TimeStep(‘step_type’, ‘discount’, ‘reward’, ‘observation’) に含まれて、そしてポリシーは時間ステップをアクションかアクションに渡る分布にマップします。殆どのポリシーは timestep.observation を利用し、幾つかのポリシーは timestep.step_type を利用します (e.g. ステートフル・ポリシーでエピソードの始まりで状態をリセットするため) が、timestep.discount と timestep.reward は通常は無視されます。

ポリシーは次のような方法で TF-Agents で他のコンポーネントに関係します。殆どのポリシーはアクション and/or TimeSteps からのアクションに渡る分布を計算するためにニューラルネットワークを持ちます。エージェントは異なる目的のために 1 つまたはそれ以上のポリシーを含むことができます、e.g. 配備のために訓練される主要ポリシー、そしてデータ収集のための noisy ポリシー。ポリシーはセーブ/リストアできて、データコレクション、評価等のためにエージェントとは無関係に使用できます。

幾つかのポリシーは TensorFlow で書くことがより容易です (e.g. ニューラルネットワークを持つもの)、他は Python で書くことが容易であることに反して (e.g. アクションのスクリプトに従っている)。そこで TF agents では、Python と TensorFlow ポリシーの両者を許容します。更に、TensorFlow で書かれたポリシーは Python 環境で使用されなければならないかもしれません、or vice versa、e.g. TensorFlow ポリシーは訓練のために使用されますが後で製品 python 環境で配備されます。これを容易にするために、python と TensorFlow ポリシーの間で変換するためのラッパーを提供します。

ポリシーのもう一つの興味深いクラスはポリシー・ラッパーです、これは与えられたポリシーをある方法で変更します、e.g. 特定のタイプのノイズを追加する、確率的ポリシーの greedy or epsilon-greedy 版を作成する、ランダムにマルチポリシーをミックスする等。

 

セットアップ

まだ tf-agents をインストールしていないのであれば、以下を実行します :

!pip install --upgrade tensorflow-probability
!pip install tf-agents
from __future__ import division
from __future__ import print_function

import abc
import tensorflow as tf
import tensorflow_probability as tfp
import numpy as np

from tf_agents.specs import array_spec
from tf_agents.specs import tensor_spec
from tf_agents.networks import network

from tf_agents.policies import py_policy
from tf_agents.policies import random_py_policy
from tf_agents.policies import scripted_py_policy

from tf_agents.policies import tf_policy
from tf_agents.policies import random_tf_policy
from tf_agents.policies import actor_policy
from tf_agents.policies import q_policy
from tf_agents.policies import greedy_policy

from tf_agents.trajectories import time_step as ts

tf.compat.v1.enable_v2_behavior()

 

Python ポリシー

Python ポリシーのためのインターフェイスは policies/py_policy.Base で定義されます。主要なメソッドは :

class Base(object):

  @abc.abstractmethod
  def __init__(self, time_step_spec, action_spec, policy_state_spec=()):
    self._time_step_spec = time_step_spec
    self._action_spec = action_spec
    self._policy_state_spec = policy_state_spec

  @abc.abstractmethod
  def reset(self, policy_state=()):
    # return initial_policy_state.
    pass

  @abc.abstractmethod
  def action(self, time_step, policy_state=()):
    # return a PolicyStep(action, state, info) named tuple.
    pass

  @abc.abstractmethod
  def distribution(self, time_step, policy_state=()):
    # Not implemented in python, only for TF policies.
    pass

  @abc.abstractmethod
  def update(self, policy):
    # update self to be similar to the input `policy`.
    pass

  @abc.abstractmethod
  def copy(self):
    # return a copy of self.
    pass

  @property
  def time_step_spec(self):
    return self._time_step_spec

  @property
  def action_spec(self):
    return self._action_spec

  @property
  def policy_state_spec(self):
    return self._policy_state_spec

最も重要なメソッドは action(time_step) です、これは環境からの観測を含む time_step を以下の属性を含む PolicyStep 名前付きタプルにマップします :

  • action: 環境に適用されるアクション。
  • state: アクションへの次の呼び出しで供給されるポリシーの状態 (e.g. RNN 状態)。
  • info: アクション対数確率のようなオプションの副次情報。

time_step_spec と action_spec は入力時間ステップと出力アクションのための仕様です。ポリシーはまたリセット関数を持ちます、これは典型的にはステートフル・ポリシーの状態をリセットするために使用されます。copy 関数は自身のコピーを返しそして update(new_policy) 関数は自身を new_policy に向けて更新します。

今は、python ポリシーの 2, 3 のサンプルを見ましょう。

 

サンプル 1: ランダム Python ポリシー

PyPolicy の単純なサンプルは RandomPyPolicy で、これは離散/連続な与えられた action_spec のためのランダムアクションを生成します。入力 time_step は無視されます。

action_spec = array_spec.BoundedArraySpec((2,), np.int32, -10, 10)
my_random_py_policy = random_py_policy.RandomPyPolicy(time_step_spec=None,
    action_spec=action_spec)
time_step = None
action_step = my_random_py_policy.action(time_step)
print(action_step)
action_step = my_random_py_policy.action(time_step)
print(action_step)

 

サンプル 2: Scripted Python ポリシー

script ポリシーは (num_repeats, action) タプルのリストとして表されたアクションのスクリプトをプレイバックします。action 関数が呼び出されるたびに、それは反復の指定回数が成されるまでリストから次のアクションを返し、それからリストの次のアクションに進みます。reset メソッドはリストの最初から実行を開始するために呼び出すことができます。

action_spec = array_spec.BoundedArraySpec((2,), np.int32, -10, 10)
action_script = [(1, np.array([5, 2], dtype=np.int32)), 
                 (0, np.array([0, 0], dtype=np.int32)), # Setting `num_repeates` to 0 will skip this action.
                 (2, np.array([1, 2], dtype=np.int32)), 
                 (1, np.array([3, 4], dtype=np.int32))]

my_scripted_py_policy = scripted_py_policy.ScriptedPyPolicy(
    time_step_spec=None, action_spec=action_spec, action_script=action_script)

policy_state = my_scripted_py_policy.get_initial_state()
time_step = None
print('Executing scripted policy...')
action_step = my_scripted_py_policy.action(time_step, policy_state)
print(action_step)
action_step= my_scripted_py_policy.action(time_step, action_step.state)
print(action_step)
action_step = my_scripted_py_policy.action(time_step, action_step.state)
print(action_step)

print('Resetting my_scripted_py_policy...')
policy_state = my_scripted_py_policy.get_initial_state()
action_step = my_scripted_py_policy.action(time_step, policy_state)
print(action_step)

 

TensorFlow ポリシー

TensorFlow ポリシーは Python ポリシーと同じインターフェイスに従います。2, 3 のサンプルを見ましょう。

 

Example 1: Random TF ポリシー

RandomTFPolicy は与えられた離散/連続な action_spec に従ってランダムアクションを生成するために使用できます。入力 time_step は無視されます。

action_spec = tensor_spec.BoundedTensorSpec(
    (2,), tf.float32, minimum=-1, maximum=3)
input_tensor_spec = tensor_spec.TensorSpec((2,), tf.float32)
time_step_spec = ts.time_step_spec(input_tensor_spec)

my_random_tf_policy = random_tf_policy.RandomTFPolicy(
    action_spec=action_spec, time_step_spec=time_step_spec)
observation = tf.ones(time_step_spec.observation.shape)
time_step = ts.restart(observation)
action_step = my_random_tf_policy.action(time_step)

print('Action:')
print(action_step.action)

 

サンプル 2: Actor ポリシー

actor ポリシーは time_steps をアクションにマップするネットワークか、time_steps をアクションに渡る分散にマップするネットワークを使用して作成できます。

 
アクションネットワークを使用する

ネットワークを次のように定義しましょう :


class ActionNet(network.Network):

  def __init__(self, input_tensor_spec, output_tensor_spec):
    super(ActionNet, self).__init__(
        input_tensor_spec=input_tensor_spec,
        state_spec=(),
        name='ActionNet')
    self._output_tensor_spec = output_tensor_spec
    self._sub_layers = [
        tf.keras.layers.Dense(
            action_spec.shape.num_elements(), activation=tf.nn.tanh),
    ]

  def call(self, observations, step_type, network_state):
    del step_type

    output = tf.cast(observations, dtype=tf.float32)
    for layer in self._sub_layers:
      output = layer(output)
    actions = tf.reshape(output, [-1] + self._output_tensor_spec.shape.as_list())

    # Scale and shift actions to the correct range if necessary.
    return actions, network_state

TensorFlow では殆どのネットワーク層はバッチ演算のために設計されていますので、入力 time_steps はバッチ処理されてネットワーク出力もバッチ処理されることを想定します。ネットワークはまた与えられた action_spec の正しい範囲でアクションを生成する責任も負います。これは慣習的に e.g. [-1, 1] のアクションを生成するために最終層のための tanh 活性を使用してからこれを入力 action_spec として正しい範囲にスケールしてシフトすることにより成されます (e.g. tf_agents/agents/ddpg/networks.actor_network() 参照)。

今は、上のネットワークを使用して actor ポリシーを作成できます。

input_tensor_spec = tensor_spec.TensorSpec((4,), tf.float32)
time_step_spec = ts.time_step_spec(input_tensor_spec)
action_spec = tensor_spec.BoundedTensorSpec((3,),
                                            tf.float32,
                                            minimum=-1,
                                            maximum=1)

action_net = ActionNet(input_tensor_spec, action_spec)

my_actor_policy = actor_policy.ActorPolicy(
    time_step_spec=time_step_spec,
    action_spec=action_spec,
    actor_network=action_net)

それを time_step_spec に従う time_steps の任意のバッチに適用できます :

batch_size = 2
observations = tf.ones([2] + time_step_spec.observation.shape.as_list())

time_step = ts.restart(observations, batch_size)

action_step = my_actor_policy.action(time_step)
print('Action:')
print(action_step.action)

distribution_step = my_actor_policy.distribution(time_step)
print('Action distribution:')
print(distribution_step.action)

上のサンプルで、アクション tensor を生成するアクション・ネットワークを使用してポリシーを作成しました。この場合、policy.distribution(time_step) は policy.action(time_step) の出力回りの決定論的 (delta) 分布です。確率的ポリシーを生成する一つの方法は actor ポリシーをアクションに noise を追加するポリシー・ラッパーでラップすることです。もう一つの方法は下で示されるようにアクションネットワークの代わりにアクション分布ネットワークを使用して actor ポリシーを作成することです。

 
アクション分布ネットワークを使用する

class ActionDistributionNet(ActionNet):

  def call(self, observations, step_type, network_state):
    action_means, network_state = super(ActionDistributionNet, self).call(
        observations, step_type, network_state)

    action_std = tf.ones_like(action_means)
    return tfp.distributions.Normal(action_means, action_std), network_state


action_distribution_net = ActionDistributionNet(input_tensor_spec, action_spec)

my_actor_policy = actor_policy.ActorPolicy(
    time_step_spec=time_step_spec,
    action_spec=action_spec,
    actor_network=action_distribution_net)

action_step = my_actor_policy.action(time_step)
print('Action:')
print(action_step.action)
distribution_step = my_actor_policy.distribution(time_step)
print('Action distribution:')
print(distribution_step.action)

上で、アクションは与えられたアクション spec [-1, 1] の範囲にクリップされることに注意してください。これはデフォルトで ActorPolicy のコンストラクタ引数 clip=True であるためです。これを false に設定することはネットワークにより生成されるクリップされないアクションを返します。

確率的ポリシーは例えば、GreedyPolicy ラッパーを使用して決定論的ポリシーに変換できます、これはそのアクションとして stochastic_policy.distribution().mode() を、そしてその distribution() としてこの greedy アクション回りの決定論的/delta 分布を選択します。

 

サンプル 3: Q ポリシー

Q ポリシーは DQN のようなエージェントで使用されて、各離散アクションのための Q 値を予測する Q ネットワークに基づきます。与えられた時間ステップについて、Q ポリシーのアクション分布はロジットとして q 値を使用して作成される categorical 分布です。

input_tensor_spec = tensor_spec.TensorSpec((4,), tf.float32)
time_step_spec = ts.time_step_spec(input_tensor_spec)
action_spec = tensor_spec.BoundedTensorSpec((1,),
                                            tf.int32,
                                            minimum=-1,
                                            maximum=1)
num_actions = action_spec.maximum - action_spec.minimum + 1


class QNetwork(network.Network):

  def __init__(self, input_tensor_spec, action_spec, num_actions=num_actions, name=None):
    super(QNetwork, self).__init__(
        input_tensor_spec=input_tensor_spec,
        state_spec=(),
        name=name)
    self._sub_layers = [
        tf.keras.layers.Dense(num_actions),
    ]

  def call(self, inputs, step_type=None, network_state=()):
    del step_type
    inputs = tf.cast(inputs, tf.float32)
    for layer in self._sub_layers:
      inputs = layer(inputs)
    return inputs, network_state


batch_size = 2
observation = tf.ones([batch_size] + time_step_spec.observation.shape.as_list())
time_steps = ts.restart(observation, batch_size=batch_size)

my_q_network = QNetwork(
    input_tensor_spec=input_tensor_spec,
    action_spec=action_spec)
my_q_policy = q_policy.QPolicy(
    time_step_spec, action_spec, q_network=my_q_network)
action_step = my_q_policy.action(time_steps)
distribution_step = my_q_policy.distribution(time_steps)

print('Action:')
print(action_step.action)

print('Action distribution:')
print(distribution_step.action)

 

ポリシー・ラッパー

ポリシー・ラッパーは与えられたポリシー, e.g. add noise をラップして変更するために使用できます。ポリシー・ラッパーは Policy (Python/TensorFlow) のサブクラスで従ってちょうど任意の他のポリシーのように使用できます。

 

サンプル: Greedy ポリシー

greedy ラッパーは distribution() を実装する任意の TensorFlow ポリシーをラップするために使用できます。GreedyPolicy.action() は wrapped_policy.distribution().mode() を返してそして GreedyPolicy.distribution() は GreedyPolicy.action() 回りの決定論的/delta 分布です :


my_greedy_policy = greedy_policy.GreedyPolicy(my_q_policy)

action_step = my_greedy_policy.action(time_steps)
print('Action:')
print(action_step.action)

distribution_step = my_greedy_policy.distribution(time_steps)
print('Action distribution:')
print(distribution_step.action)
 

以上






TF-Agents 0.4 : Tutorials : 環境

TF-Agents 0.4 Tutorials : 環境 (翻訳/解説)

翻訳 : (株)クラスキャット セールスインフォメーション
作成日時 : 04/19/2020 (0.4)

* 本ページは、TF Agents の以下のドキュメントを翻訳した上で適宜、補足説明したものです:

* サンプルコードの動作確認はしておりますが、必要な場合には適宜、追加改変しています。
* ご自由にリンクを張って頂いてかまいませんが、sales-info@classcat.com までご一報いただけると嬉しいです。

 

環境

イントロダクション

強化学習 (RL) の目標は環境と相互作用することにより学習するエージェントを設計することです。標準的な RL 設定では、エージェントは総ての時間ステップで観測を受け取りそしてアクションを選択します。アクションは環境に適用されて環境は報酬と新しい観測を返します。エージェントはリターンとしても知られる、報酬の総計を最大化するためのアクションを選択するポリシーを訓練します。

TF-Agents では、環境は Python か TensorFlow で実装できます。Python 環境は実装し、理解し、そしてデバッグするために通常はより容易ですが、TensorFlow 環境はより効率的で自然な並列性が可能です。最も一般的なワークフローは Python で環境を実装してそしてそれを自動的に TensorFlow に変換するために私達のラッパーの一つを使用することです。

最初に Python 環境を見ましょう。TensorFlow 環境は非常に類似した API に従います。

 

セットアップ

tf-agents か gym をまだインストールしてないのであれば、以下を実行してください :

!pip install tf-agents
!pip install 'gym==0.10.11'
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function

import abc
import tensorflow as tf
import numpy as np

from tf_agents.environments import py_environment
from tf_agents.environments import tf_environment
from tf_agents.environments import tf_py_environment
from tf_agents.environments import utils
from tf_agents.specs import array_spec
from tf_agents.environments import wrappers
from tf_agents.environments import suite_gym
from tf_agents.trajectories import time_step as ts

tf.compat.v1.enable_v2_behavior()

 

Python 環境

Python 環境は step(action) -> next_time_step メソッドを持ちます、これは環境にアクションを適用して、次のステップについて以下の情報を返します :

  1. observation (観測): これはエージェントが次のステップでそのアクションを選択するために観測できる環境状態の一部です。
  2. reward (報酬): エージェントはマルチステップに渡りこれらの報酬の総計を最大化することを学習しています。
  3. step_type: 環境との相互作用は通常はシークエンス/エピソードの一部です、e.g. チェスゲームの複数の移動。step_type は FIRST, MID or LAST のいずれかであり得て、この時間ステップが最初、中間 or 最後のステップであるかを示します。
  4. discount (割引き): これは、現在の時間ステップにおける報酬に比較して次の時間ステップで報酬をどのくらい重み付けるかを表す float です。

これらは TimeStep(step_type, reward, discount, observation) にグループ分けされます。

総ての Python 環境が実装しなければならないインターフェイスは environments/py_environment.PyEnvironment にあります。主要なメソッドは :

class PyEnvironment(object):

  def reset(self):
    """Return initial_time_step."""
    self._current_time_step = self._reset()
    return self._current_time_step

  def step(self, action):
    """Apply action and return new time_step."""
    if self._current_time_step is None:
        return self.reset()
    self._current_time_step = self._step(action)
    return self._current_time_step

  def current_time_step(self):
    return self._current_time_step

  def time_step_spec(self):
    """Return time_step_spec."""

  @abc.abstractmethod
  def observation_spec(self):
    """Return observation_spec."""

  @abc.abstractmethod
  def action_spec(self):
    """Return action_spec."""

  @abc.abstractmethod
  def _reset(self):
    """Return initial_time_step."""

  @abc.abstractmethod
  def _step(self, action):
    """Apply action and return new time_step."""
    self._current_time_step = self._step(action)
    return self._current_time_step

step() メソッドに加えて、環境はまた新しいシークエンスを始めて初期 TimeStep を提供する reset() メソッドも提供します。reset メソッドを明示的に呼び出す必要はありません。エピソードの最後に達するときか step() が最初に呼び出されるとき、環境は自動的にリセットすることを仮定しています。

サブクラスは step() や reset() を直接実装しないことに注意してください。それらは代わりに _step() と _reset() メソッドをオーバーライドします。これらのメソッドから返される時間ステップは current_time_step() を通してキャッシュされて開示されます。

observation_spec と action_spec メソッドは (Bounded)ArraySpecs のネストを返します、これはそれぞれ観測とアクションの名前、shape、データ型と範囲を記述します。

TF-Agents では繰り返しネストを参照します、これはリスト、タプル、名前付きタプルや辞書から成る構造のような任意のツリーとして定義されます。これらは観測とアクションの構造を保持するために任意に構成できます。私達は多くの観測とアクションを持つようなより複雑な環境のためにこれが非常に有用であることを見出しました。

 

標準的な環境を使用する

TF Agents は OpenAI Gym, DeepMind-control と Atari のような多くの標準的な環境のための組込みラッパーを持ちます、その結果それらは私達の py_environment.PyEnvironment インターフェイスに従います。これらのラップされた環境は環境スーツを使用して容易にロードできます。OpenAI gym からカートポール環境をロードしてアクションと time_step_spec を見ましょう。

environment = suite_gym.load('CartPole-v0')
print('action_spec:', environment.action_spec())
print('time_step_spec.observation:', environment.time_step_spec().observation)
print('time_step_spec.step_type:', environment.time_step_spec().step_type)
print('time_step_spec.discount:', environment.time_step_spec().discount)
print('time_step_spec.reward:', environment.time_step_spec().reward)

そして環境は [0, 1] の型 int64 のアクションを想定して観測が長さ 4 の float32 ベクトルで割引き因子が [0.0, 1.0] の float32 であるような TimeSteps を返すことを見ます。今は、総てエピソードのために固定されたアクション (1,) を取ってみましょう。

action = np.array(1, dtype=np.int32)
time_step = environment.reset()
print(time_step)
while not time_step.is_last():
  time_step = environment.step(action)
  print(time_step)

 

貴方自身の Python 環境を作成する

多くのクライアントについて、一般的なユースケースは彼らの問題に対して TF-Agents で標準的なエージェント (agents/ 参照) の一つを適用することです。これを行なうため、彼らの問題を環境に組み入れなければなりません。そこで環境を Python でどのように実装するかを見ましょう。

次の (ブラックジャックにインスパイアされた) カードゲームをプレイするためにエージェントを訓練することを望むとしましょう :

  1. ゲームは 1…10 と番号付けられたカードの無限のデッキを使用してプレイされます。
  2. 総てのターンでエージェントは 2 つのことを行えます : 新しいランダムカードを得るか、現在のラウンドを停止するかです。
  3. 目標はラウンドの最後で、越えることなく、カードの総計をできる限り 21 に近づけることです。

ゲームを表す環境はこのように見えるでしょう :

  1. アクション: 2 つのアクションを持ちます。Action 0: 新しいカードを得る、そして Action 1: 現在のラウンドを停止する。
  2. 観測: 現在のラウンドのカードの総計。
  3. 報酬: 目的は越えることなく出来る限り 21 に近づけて得ることですから、ラウンドの最後の次の報酬を使用してこれを達成できます : sum_of_cards – 21 if sum_of_cards <= 21, else -21
class CardGameEnv(py_environment.PyEnvironment):

  def __init__(self):
    self._action_spec = array_spec.BoundedArraySpec(
        shape=(), dtype=np.int32, minimum=0, maximum=1, name='action')
    self._observation_spec = array_spec.BoundedArraySpec(
        shape=(1,), dtype=np.int32, minimum=0, name='observation')
    self._state = 0
    self._episode_ended = False

  def action_spec(self):
    return self._action_spec

  def observation_spec(self):
    return self._observation_spec

  def _reset(self):
    self._state = 0
    self._episode_ended = False
    return ts.restart(np.array([self._state], dtype=np.int32))

  def _step(self, action):

    if self._episode_ended:
      # The last action ended the episode. Ignore the current action and start
      # a new episode.
      return self.reset()

    # Make sure episodes don't go on forever.
    if action == 1:
      self._episode_ended = True
    elif action == 0:
      new_card = np.random.randint(1, 11)
      self._state += new_card
    else:
      raise ValueError('`action` should be 0 or 1.')

    if self._episode_ended or self._state >= 21:
      reward = self._state - 21 if self._state <= 21 else -21
      return ts.termination(np.array([self._state], dtype=np.int32), reward)
    else:
      return ts.transition(
          np.array([self._state], dtype=np.int32), reward=0.0, discount=1.0)

上の環境を定義して総てを正しく行なったことを確かなものにしましょう。貴方自身の環境を作成するとき生成された観測と time_steps が貴方の spes で定義されたように正しい shape と型に従うことを確実にしなければなりません。これらは TensorFlow グラフを生成するために使用されてそれらを間違える場合には問題をデバッグすることを困難にするようなものとして作成するかもしれません。

環境を検証するため、アクションを生成するためにランダムポリシーを使用して物事が意図したように動作していることを確認するために 5 エピソードに渡り反復します。環境 specs に従わない time_step を受け取る場合エラーが上げられます。

environment = CardGameEnv()
utils.validate_py_environment(environment, episodes=5)

環境が意図通りに動作していることを知った今、固定ポリシーを使用してこの環境を実行しましょう : 3 カードを求めてからラウンドを終了します。

get_new_card_action = np.array(0, dtype=np.int32)
end_round_action = np.array(1, dtype=np.int32)

environment = CardGameEnv()
time_step = environment.reset()
print(time_step)
cumulative_reward = time_step.reward

for _ in range(3):
  time_step = environment.step(get_new_card_action)
  print(time_step)
  cumulative_reward += time_step.reward

time_step = environment.step(end_round_action)
print(time_step)
cumulative_reward += time_step.reward
print('Final Reward = ', cumulative_reward)

 

環境ラッパー

環境ラッパーは python 環境を取り環境の変更されたバージョンを返します。元の環境と変更された環境の両者は py_environment.PyEnvironment のインスタンスで、マルチラッパーが一緒に連鎖できます。

幾つかの一般的なラッパーは environments/wrappers.py で見つかります。例えば :

  1. ActionDiscretizeWrapper: 連続アクション空間を離散アクション空間に変換します。
  2. RunStats: 取られるステップ数、完了したエピソードの数等のような環境の実行統計を捕捉します。
  3. TimeLimit: ステップの固定数の後エピソードを終わらせる。

 
サンプル 1: アクション離散化 (= Discretize) ラッパー

InvertedPendulum は範囲 [-1, 1] の連続的アクションを受け取る PyBullet 環境です。この環境上の DQN のような離散アクションエージェントを訓練することを望む場合、アクション空間を離散化 (量子化) しなければなりません。これは正確に ActionDiscretizeWrapper が行なうことです。ラッピング前後で action_spec を比較します :

env = suite_gym.load('Pendulum-v0')
print('Action Spec:', env.action_spec())

discrete_action_env = wrappers.ActionDiscretizeWrapper(env, num_actions=5)
print('Discretized Action Spec:', discrete_action_env.action_spec())

ラップされた discrete_action_env は py_environment.PyEnvironment のインスタンスで通常の python 環境のように扱えます。

 

TensorFlow 環境

TF 環境のためのインターフェイスは environments/tf_environment.TFEnvironment で定義されて Python 環境に非常に類似して見えます。TF 環境は 2, 3 の方法で python envs と異なります :

  • それらは配列の代わりに tensor オブジェクトを生成します。
  • TF 環境は specs と比較するとき生成された tensor にバッチ次元を追加します。

python 環境の TFEnvs への変換は tensorflow に演算を並列化することを可能にします。例えば、環境からデータを収集して replay_buffer に追加する collect_experience_op そして replay_buffer から読みエージェントを訓練する train_op を定義し、それらを TensorFlow で自然に並列に実行できるでしょう。

class TFEnvironment(object):

  def time_step_spec(self):
    """Describes the `TimeStep` tensors returned by `step()`."""

  def observation_spec(self):
    """Defines the `TensorSpec` of observations provided by the environment."""

  def action_spec(self):
    """Describes the TensorSpecs of the action expected by `step(action)`."""

  def reset(self):
    """Returns the current `TimeStep` after resetting the Environment."""
    return self._reset()

  def current_time_step(self):
    """Returns the current `TimeStep`."""
    return self._current_time_step()

  def step(self, action):
    """Applies the action and returns the new `TimeStep`."""
    return self._step(action)

  @abc.abstractmethod
  def _reset(self):
    """Returns the current `TimeStep` after resetting the Environment."""

  @abc.abstractmethod
  def _current_time_step(self):
    """Returns the current `TimeStep`."""

  @abc.abstractmethod
  def _step(self, action):
    """Applies the action and returns the new `TimeStep`."""

current_time_step() メソッドは現在の time_step を返して必要であれば環境を初期化します。

reset() メソッドは環境でリセットを強要して current_step を返します。

アクションが前の time_step に依拠しないのであれば、tf.control_dependency が Graph モードで必要です。

今は、TFEnvironments がどのように作成されるかを見ましょう。

 

貴方自身の TensorFlow 環境を作成する

これは Python で環境を作成するよりも複雑ですので、それをこの colab ではカバーしません。サンプルは ここ で利用可能です。より一般的なケースは貴方の環境を Python で実装してそれを TFPyEnvironment ラッパーを使用して TensorFlow でラップすることです (下を見てください)。

 

Python 環境を TensorFlow にラップする

TFPyEnvironment ラッパーを使用して任意の Python 環境を TensorFlow 環境に容易にラップできます。

env = suite_gym.load('CartPole-v0')
tf_env = tf_py_environment.TFPyEnvironment(env)

print(isinstance(tf_env, tf_environment.TFEnvironment))
print("TimeStep Specs:", tf_env.time_step_spec())
print("Action Specs:", tf_env.action_spec())

specs は今では次の型: (Bounded)TensorSpec であることに注意してください。

 

使用方法サンプル

単純なサンプル

env = suite_gym.load('CartPole-v0')

tf_env = tf_py_environment.TFPyEnvironment(env)
# reset() creates the initial time_step after resetting the environment.
time_step = tf_env.reset()
num_steps = 3
transitions = []
reward = 0
for i in range(num_steps):
  action = tf.constant([i % 2])
  # applies the action and returns the new TimeStep.
  next_time_step = tf_env.step(action)
  transitions.append([time_step, action, next_time_step])
  reward += next_time_step.reward
  time_step = next_time_step

np_transitions = tf.nest.map_structure(lambda x: x.numpy(), transitions)
print('\n'.join(map(str, np_transitions)))
print('Total reward:', reward.numpy())

 
エピソード全体

env = suite_gym.load('CartPole-v0')
tf_env = tf_py_environment.TFPyEnvironment(env)

time_step = tf_env.reset()
rewards = []
steps = []
num_episodes = 5

for _ in range(num_episodes):
  episode_reward = 0
  episode_steps = 0
  while not time_step.is_last():
    action = tf.random.uniform([1], 0, 2, dtype=tf.int32)
    time_step = tf_env.step(action)
    episode_steps += 1
    episode_reward += time_step.reward.numpy()
  rewards.append(episode_reward)
  steps.append(episode_steps)
  time_step = tf_env.reset()

num_steps = np.sum(steps)
avg_length = np.mean(steps)
avg_reward = np.mean(rewards)

print('num_episodes:', num_episodes, 'num_steps:', num_steps)
print('avg_length', avg_length, 'avg_reward:', avg_reward)
 

以上






TF-Agents 0.4 : Tutorials : 深層 Q ネットワークを TF-Agents で訓練する

TF-Agents 0.4 Tutorials : 深層 Q ネットワークを TF-Agents で訓練する (翻訳/解説)

翻訳 : (株)クラスキャット セールスインフォメーション
作成日時 : 04/18/2020 (0.4)

* 本ページは、TF Agents の以下のドキュメントを翻訳した上で適宜、補足説明したものです:

* サンプルコードの動作確認はしておりますが、必要な場合には適宜、追加改変しています。
* ご自由にリンクを張って頂いてかまいませんが、sales-info@classcat.com までご一報いただけると嬉しいです。

 

深層 Q ネットワークを TF-Agents で訓練する

イントロダクション

このサンプルは TF-Agents ライブラリを使用してカートポール環境でどのように DQN (深層 Q ネットワーク) エージェントを訓練するかを示します。

それは訓練、評価とデータ収集のための強化学習 (RL) パイプラインの総てのコンポーネントを段階的に貴方に説明します。

 

セットアップ

以下の依存性をインストールしていないのであれば、次を実行してください :

!sudo apt-get install -y xvfb ffmpeg
!pip install 'gym==0.10.11'
!pip install 'imageio==2.4.0'
!pip install PILLOW
!pip install 'pyglet==1.3.2'
!pip install pyvirtualdisplay
!pip install --upgrade tensorflow-probability
!pip install tf-agents
from __future__ import absolute_import, division, print_function

import base64
import imageio
import IPython
import matplotlib
import matplotlib.pyplot as plt
import numpy as np
import PIL.Image
import pyvirtualdisplay

import tensorflow as tf

from tf_agents.agents.dqn import dqn_agent
from tf_agents.drivers import dynamic_step_driver
from tf_agents.environments import suite_gym
from tf_agents.environments import tf_py_environment
from tf_agents.eval import metric_utils
from tf_agents.metrics import tf_metrics
from tf_agents.networks import q_network
from tf_agents.policies import random_tf_policy
from tf_agents.replay_buffers import tf_uniform_replay_buffer
from tf_agents.trajectories import trajectory
from tf_agents.utils import common
tf.compat.v1.enable_v2_behavior()

# Set up a virtual display for rendering OpenAI gym environments.
display = pyvirtualdisplay.Display(visible=0, size=(1400, 900)).start()
tf.version.VERSION

 

ハイパーパラメータ

num_iterations = 20000 # @param {type:"integer"}

initial_collect_steps = 1000  # @param {type:"integer"} 
collect_steps_per_iteration = 1  # @param {type:"integer"}
replay_buffer_max_length = 100000  # @param {type:"integer"}

batch_size = 64  # @param {type:"integer"}
learning_rate = 1e-3  # @param {type:"number"}
log_interval = 200  # @param {type:"integer"}

num_eval_episodes = 10  # @param {type:"integer"}
eval_interval = 1000  # @param {type:"integer"}

 

環境

強化学習では、環境は解かれるべきタスクあるいは問題を表します。TF-Agents では標準的な環境は tf_agents.environments スーツを使用して作成できます。TF-Agents は OpenAI Gym, Atari と DM Control のようなソースから環境をロードするためにスーツを持ちます。

OpenAI Gym スーツからカートポール環境をロードします。

env_name = 'CartPole-v0'
env = suite_gym.load(env_name)

この環境をそれがどのようなものかを見るためにレンダリングできます。free-swinging ポールはカートに装着されています。目標はポールを直立するように保持するためにカートを右または左に動かすことです。

#@test {"skip": true}
env.reset()
PIL.Image.fromarray(env.render())

environment.step メソッドは環境でアクションを取り TimeStep タプルを返します、これは環境の次の観測とアクションに対する報酬を含みます。

time_step_spec() メソッドは TimeStep タプルのための仕様を返します。その observation 属性は観測の shape, データ型そして許される値の範囲を示します。reward 属性は報酬のための同じ属性を示します。

print('Observation Spec:')
print(env.time_step_spec().observation)
print('Reward Spec:')
print(env.time_step_spec().reward)

action_spec() メソッドは shape, データ型と正当なアクションの許される値を返します。

print('Action Spec:')
print(env.action_spec())

カートポール環境では :

  • 観測は 4 floats の配列です :
    • カートの位置と速度
    • ポールの角度位置と (角) 速度
  • 報酬はスカラー float 阿智
  • アクションは 2 つだけの可能な値を持つスカラー整数です :
    • 0 — “move left”
    • 1 — “move right”
time_step = env.reset()
print('Time step:')
print(time_step)

action = np.array(1, dtype=np.int32)

next_time_step = env.step(action)
print('Next time step:')
print(next_time_step)

通常は 2 つの環境がインスタンス化されます : 一つは訓練のためそして一つは評価のためです。

train_py_env = suite_gym.load(env_name)
eval_py_env = suite_gym.load(env_name)

殆どの環境のように、カートポール環境は純粋な Python で書かれています。これは TFPyEnvironment ラッパーを使用して TensorFlow に変換されます。

元の環境の API は Numpy 配列を使用しています。TFPyEnvironment はそれを TensorFlow エージェントとポリシーと互換にするためにこれらを Tensor に変換します。

train_env = tf_py_environment.TFPyEnvironment(train_py_env)
eval_env = tf_py_environment.TFPyEnvironment(eval_py_env)

 

エージェント

RL 問題を解くために使用される有りごリズムはエージェントにより表されます。TF-Agents は以下を含む、様々なエージェントの標準的な実装を提供します :

DQN エージェントは離散アクション空間を持つ任意の環境で利用できます。

DQN エージェントの心臓部は QNetwork です、これは環境からの観測が与えられたとき総てのアクションに対して QValue (期待リターン) を予測することを学習できるニューラルネットワーク・モデルです。

QNetwork を作成するために tf_agents.networks.q_network を使用し、observation_spec, action_spec とモデルの隠れ層の数とサイズを記述するタプルを渡します。

fc_layer_params = (100,)

q_net = q_network.QNetwork(
    train_env.observation_spec(),
    train_env.action_spec(),
    fc_layer_params=fc_layer_params)

今は DqnAgent をインスタンス化するために tf_agents.agents.dqn.dqn_agent を使用します。time_step_spec, action_spec と QNetwork に加えて、エージェント・コンストラクタは optimizer (この場合、AdamOptimizer), 損失関数そして整数ステップ・カウンターも必要とします。

optimizer = tf.compat.v1.train.AdamOptimizer(learning_rate=learning_rate)

train_step_counter = tf.Variable(0)

agent = dqn_agent.DqnAgent(
    train_env.time_step_spec(),
    train_env.action_spec(),
    q_network=q_net,
    optimizer=optimizer,
    td_errors_loss_fn=common.element_wise_squared_loss,
    train_step_counter=train_step_counter)

agent.initialize()

 

ポリシー

ポリシーはエージェントが環境で動作する方法を定義します。典型的には、強化学習の目標は基礎的なモデルをポリシーが望まれる結果を生成するまで訓練することです。

このチュートリアルでは :

  • 望まれる結果はポールがカートを越えて直立にバランスが取られ続けることです。
  • ポリシーは各 time_step 観測に対してアクション (左 or 右) を返します。

エージェントは 2 つのポリシーを含みます :

  • agent.policy — 評価と配備のために使用される主要ポリシー。
  • agent.collect_policy — データ収集のために使用される 2 番目のポリシー。
eval_policy = agent.policy
collect_policy = agent.collect_policy

ポリシーはエージェントとは無関係に作成できます。例えば、ポリシーを作成するために tf_agents.policies.random_tf_policy を使用できます、これは各 time_step のためにランダムにアクションを選択します。

random_policy = random_tf_policy.RandomTFPolicy(train_env.time_step_spec(),
                                                train_env.action_spec())

ポリシーからアクションを得るには、policy.action(time_step) メソッドを呼び出します。time_step は環境からの観測を含みます。このメソッドは PolicyStep を返します、これは 3 つのコンポーネントを持つ名前付きタプルです :

  • action — 取られるアクション (この場合、0 or 1)
  • state — ステートフル (つまり、RNN-ベースの) ポリシーのために使用されます
  • info — アクションの log 確率のような、補助的データ
example_environment = tf_py_environment.TFPyEnvironment(
    suite_gym.load('CartPole-v0'))
time_step = example_environment.reset()
random_policy.action(time_step)

 

メトリクスと評価

ポリシーを評価するために使用される最も一般的なメトリックは平均リターンです。リターンはエピソードのための環境でポリシーを実行している間に得られた報酬の総計です。幾つかのエピソードが実行されて、平均リターンを作成します。

次の関数は、ポリシー、環境とエピソードの数が与えられたとき、ポリシーの平均リターンを計算します。

#@test {"skip": true}
def compute_avg_return(environment, policy, num_episodes=10):

  total_return = 0.0
  for _ in range(num_episodes):

    time_step = environment.reset()
    episode_return = 0.0

    while not time_step.is_last():
      action_step = policy.action(time_step)
      time_step = environment.step(action_step.action)
      episode_return += time_step.reward
    total_return += episode_return

  avg_return = total_return / num_episodes
  return avg_return.numpy()[0]


# See also the metrics module for standard implementations of different metrics.
# https://github.com/tensorflow/agents/tree/master/tf_agents/metrics

random_policy 上でのこの計算の実行は環境でのベースライン・パフォーマンスを示します。

compute_avg_return(eval_env, random_policy, num_eval_episodes)

 

再生バッファ

再生バッファは環境から収集されたデータを追跡します。このチュートリアルは tf_agents.replay_buffers.tf_uniform_replay_buffer.TFUniformReplayBuffer を使用します、それが最も一般的であるからです。

コンストラクタはそれが収集するデータのための仕様を必要とします。これは collect_data_spec メソッドを使用してエージェントから利用可能です。バッチサイズと最大バッファ長もまた必要です。

replay_buffer = tf_uniform_replay_buffer.TFUniformReplayBuffer(
    data_spec=agent.collect_data_spec,
    batch_size=train_env.batch_size,
    max_length=replay_buffer_max_length)

殆どのエージェントについて、collect_data_spec は Trajectory と呼ばれる名前付きタプルで、観測、アクション、報酬と他の項目のための仕様を含みます。

agent.collect_data_spec
agent.collect_data_spec._fields

 

データ収集

今は数ステップのために環境でランダム・ポリシーを実行します、データを再生バッファで記録します。

#@test {"skip": true}
def collect_step(environment, policy, buffer):
  time_step = environment.current_time_step()
  action_step = policy.action(time_step)
  next_time_step = environment.step(action_step.action)
  traj = trajectory.from_transition(time_step, action_step, next_time_step)

  # Add trajectory to the replay buffer
  buffer.add_batch(traj)

def collect_data(env, policy, buffer, steps):
  for _ in range(steps):
    collect_step(env, policy, buffer)

collect_data(train_env, random_policy, replay_buffer, steps=100)

# This loop is so common in RL, that we provide standard implementations. 
# For more details see the drivers module.
# https://github.com/tensorflow/agents/blob/master/tf_agents/docs/python/tf_agents/drivers.md

再生バッファは今では Trajectories のコレクションです。

# For the curious:
# Uncomment to peel one of these off and inspect it.
# iter(replay_buffer.as_dataset()).next()

エージェントは再生バッファへのアクセスを必要とします。これは iterable tf.data.Dataset パイプラインを作成することにより提供されます、これはデータをエージェントに供給します。

再生バッファの各行は単一の観測ステップをストアするだけです。しかし DQN エージェントは損失を計算するために現在と次の両者の観測を必要としますので、データセット・パイプラインはバッチ (num_steps=2) の各項目のために 2 つの隣接行をサンプリングします。

データセットはまた並列呼び出しの実行とデータの先取りにより最適化もされます。

# Dataset generates trajectories with shape [Bx2x...]
dataset = replay_buffer.as_dataset(
    num_parallel_calls=3, 
    sample_batch_size=batch_size, 
    num_steps=2).prefetch(3)


dataset
iterator = iter(dataset)

print(iterator)
# For the curious:
# Uncomment to see what the dataset iterator is feeding to the agent.
# Compare this representation of replay data 
# to the collection of individual trajectories shown earlier.

# iterator.next()

 

エージェントを訓練する

訓練ループの間に 2 つのことが起きなければなりません :

  • 環境からデータを収集する
  • エージェントのニューラルネットワークを訓練するためにデータを使用する

このサンプルはまた定期的にポリシーを評価して現在のスコアをプリントします。

以下は実行に ~5 分かかります。

#@test {"skip": true}
try:
  %%time
except:
  pass

# (Optional) Optimize by wrapping some of the code in a graph using TF function.
agent.train = common.function(agent.train)

# Reset the train step
agent.train_step_counter.assign(0)

# Evaluate the agent's policy once before training.
avg_return = compute_avg_return(eval_env, agent.policy, num_eval_episodes)
returns = [avg_return]

for _ in range(num_iterations):

  # Collect a few steps using collect_policy and save to the replay buffer.
  for _ in range(collect_steps_per_iteration):
    collect_step(train_env, agent.collect_policy, replay_buffer)

  # Sample a batch of data from the buffer and update the agent's network.
  experience, unused_info = next(iterator)
  train_loss = agent.train(experience).loss

  step = agent.train_step_counter.numpy()

  if step % log_interval == 0:
    print('step = {0}: loss = {1}'.format(step, train_loss))

  if step % eval_interval == 0:
    avg_return = compute_avg_return(eval_env, agent.policy, num_eval_episodes)
    print('step = {0}: Average Return = {1}'.format(step, avg_return))
    returns.append(avg_return)

 

可視化

プロット

訓練の間にポリシーがどのように改良されるを図表にするために matplotlib.pyplot を使用します。

Cartpole-v0 の 1 反復は 200 時間ステップから成ります。環境はポールが直立し続ける各ステップのために +1 の報酬を与えますので、1 エピソードのための最大リターンは 200 です。チャートはリターンが訓練の間に評価されるたびにその最大値に向かって増えているうことを示します。(それは少し不安定で毎回単調には増えないかもしれません。)

#@test {"skip": true}

iterations = range(0, num_iterations + 1, eval_interval)
plt.plot(iterations, returns)
plt.ylabel('Average Return')
plt.xlabel('Iterations')
plt.ylim(top=250)

 

ビデオ

チャートは良いです。しかしエージェントが実際に環境でタスクを遂行するのを見ることはより刺激的です。

最初に、ビデオをノートブックに埋め込む関数を作成します。

def embed_mp4(filename):
  """Embeds an mp4 file in the notebook."""
  video = open(filename,'rb').read()
  b64 = base64.b64encode(video)
  tag = '''
  '''.format(b64.decode())

  return IPython.display.HTML(tag)

今はエージェントでカートボール・ゲームの 2, 3 のエピソードを通して反復します。基礎的な Python 環境 (TensorFlow 環境ラッパーの「内側」の一つ) が render() メソッドを提供し、これは環境状態の画像を出力します。これらはビデオに収集できます。

def create_policy_eval_video(policy, filename, num_episodes=5, fps=30):
  filename = filename + ".mp4"
  with imageio.get_writer(filename, fps=fps) as video:
    for _ in range(num_episodes):
      time_step = eval_env.reset()
      video.append_data(eval_py_env.render())
      while not time_step.is_last():
        action_step = policy.action(time_step)
        time_step = eval_env.step(action_step.action)
        video.append_data(eval_py_env.render())
  return embed_mp4(filename)




create_policy_eval_video(agent.policy, "trained-agent")

For fun, compare the trained agent (above) to an agent moving randomly. (It does not do as well.)

create_policy_eval_video(random_policy, "random-agent")
 

以上






AI導入支援 #2 ウェビナー

スモールスタートを可能としたAI導入支援   Vol.2
[無料 WEB セミナー] [詳細]
「画像認識 AI PoC スターターパック」の紹介
既に AI 技術を実ビジネスで活用し、成果を上げている日本企業も多く存在しており、競争優位なビジネスを展開しております。
しかしながら AI を導入したくとも PoC (概念実証) だけでも高額な費用がかかり取組めていない企業も少なくないようです。A I導入時には欠かせない PoC を手軽にしかも短期間で認知度を確認可能とするサービの紹介と共に、AI 技術の特性と具体的な導入プロセスに加え運用時のポイントについても解説いたします。
日時:2021年10月13日(水)
会場:WEBセミナー
共催:クラスキャット、日本FLOW(株)
後援:働き方改革推進コンソーシアム
参加費: 無料 (事前登録制)
人工知能開発支援
◆ クラスキャットは 人工知能研究開発支援 サービスを提供しています :
  • テクニカルコンサルティングサービス
  • 実証実験 (プロトタイプ構築)
  • アプリケーションへの実装
  • 人工知能研修サービス
◆ お問合せ先 ◆
(株)クラスキャット
セールス・インフォメーション
E-Mail:sales-info@classcat.com