ClassCat® AI Research

Acme : サンプル : チュートリアル

Posted on by Sales Information

Acme : サンプル : チュートリアル (翻訳/解説)

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

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

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

 

サンプル : チュートリアル

この colab は強化学習エージェントを作成するために Acme のモジュールがどのように一緒にスタックされるかの概要を提供します。それはどのようにネットワークを環境仕様に適合させるか、アクター、leaner、再生バッファ、データセット、adder そして完全なエージェントをどのように作成するかを示します。それはまた貴方自身の Acme ベースのエージェントを作成するためにどこで、あるモジュールをスワップアウトできるかも明らかにします。

 

インストール

最初の 2, 3 セルで総ての必要な依存性 (そして 2, 3 のオプションのもの) をインストールすることから始めます。

#@title Install necessary dependencies.

!sudo apt-get install -y xvfb ffmpeg
!pip install 'gym==0.10.11'
!pip install imageio
!pip install PILLOW
!pip install 'pyglet==1.3.2'
!pip install pyvirtualdisplay

!pip install dm-acme
!pip install dm-acme[reverb]
!pip install dm-acme[tf]
!pip install dm-acme[envs]

from IPython.display import clear_output
clear_output()

 

dm_control をインストールする

次のセルは貴方が institutional MuJoCo ライセンスを持つ場合 dm_control により提供される環境をインストールします。これは必要ではありませんが、これなしでは下の dm_cartpole 環境を利用することができません、そして代わりに gym 環境を利用してこの colab に従うことができます。それを行なうには次のセルを単純に展開し、貴方のライセンスファイルを貼り付けて、そしてセルを実行します。

代わりに、Colab は貴方のローカルマシン上の Jupyter カーネルの使用もサポートします、これはここのガイドラインに従うことにより成されます : https://research.google.com/colaboratory/local-runtimes.html 。これは https://github.com/deepmind/dm_control の手順に従い personal MuJoCo ライセンスを使用することにより dm_control をインストールすることを可能にします。

#@title Add your License
#@test {"skip": true}
mjkey = """
""".strip()

mujoco_dir = "$HOME/.mujoco"

# Install OpenGL dependencies
!apt-get update && apt-get install -y --no-install-recommends \
  libgl1-mesa-glx libosmesa6 libglew2.0

# Get MuJoCo binaries
!wget -q https://www.roboti.us/download/mujoco200_linux.zip -O mujoco.zip
!unzip -o -q mujoco.zip -d "$mujoco_dir"

# Copy over MuJoCo license
!echo "$mjkey" > "$mujoco_dir/mjkey.txt"

# Install dm_control
!pip install dm_control

# Configure dm_control to use the OSMesa rendering backend
%env MUJOCO_GL=osmesa

# Check that the installation succeeded
try:
  from dm_control import suite
  env = suite.load('cartpole', 'swingup')
  pixels = env.physics.render()
except Exception as e:
  raise e from RuntimeError(
      'Something went wrong during installation. Check the shell output above '
      'for more information.')
else:
  from IPython.display import clear_output
  clear_output()
  del suite, env, pixels

 

モジュールをインポートする

今は総ての関連モジュールをインポートできます。

#@title Import modules.
#python3

%%capture
import copy
import pyvirtualdisplay
import imageio 
import base64
import IPython


from acme import environment_loop
from acme.tf import networks
from acme.adders import reverb as adders
from acme.agents.tf import actors as actors
from acme.datasets import reverb as datasets
from acme.wrappers import gym_wrapper
from acme import specs
from acme import wrappers
from acme.agents.tf import d4pg
from acme.agents import agent
from acme.tf import utils as tf2_utils
from acme.utils import loggers

import gym 
import dm_env
import matplotlib.pyplot as plt
import numpy as np
import reverb
import sonnet as snt
import tensorflow as tf

# Import dm_control if it exists.
try:
  from dm_control import suite
except (ModuleNotFoundError, OSError):
  pass

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

 

環境をロードする

今は環境をロードできます。以後、環境から単一状態を生成して可視化するために環境を作成します。利用したい環境を単に選択してセルを実行します。

environment_name = 'dm_cartpole'  # @param ['dm_cartpole', 'gym_mountaincar']
# task_name = 'balance'  # @param ['swingup', 'balance']

def make_environment(domain_name='cartpole', task='balance'):
  env = suite.load(domain_name, task)
  env = wrappers.SinglePrecisionWrapper(env)
  return env

if 'dm_cartpole' in environment_name:
  environment = make_environment('cartpole')
  def render(env):
    return env._physics.render(camera_id=0)  #pylint: disable=protected-access

elif 'gym_mountaincar' in environment_name:
  environment = gym_wrapper.GymWrapper(gym.make('MountainCarContinuous-v0'))
  environment = wrappers.SinglePrecisionWrapper(environment)
  def render(env):
    return env.environment.render(mode='rgb_array')
else:
  raise ValueError('Unknown environment: {}.'.format(environment_name))

# Show the frame.
frame = render(environment)
plt.imshow(frame)
plt.axis('off')

 

環境仕様 (= Spec)

後で次の図に対応するループで環境と相互作用します :

しかしこの環境と相互作用するエージェントの構築を始める前に、最初に環境が返す (e.g. 観測) あるいは消費する (e.g. アクション) オブジェクトのタイプを見ましょう。environment_spec は環境が公開する観測、報酬と割引きの形式と取られるアクションの形式を貴方に示します、

environment_spec = specs.make_environment_spec(environment)

print('actions:\n', environment_spec.actions, '\n')
print('observations:\n', environment_spec.observations, '\n')
print('rewards:\n', environment_spec.rewards, '\n')
print('discounts:\n', environment_spec.discounts, '\n')

 

観測をアクションにマップするポリシー・ネットワークを構築する

潜在的に強化学習アルゴリズムの最も重要なパートは環境観測をアクションにマップするポリシーです。ポリシーを作成するために単純なニューラルネットワークを利用できます、この場合には層 norm を持つ単純な順伝播 MLP です。私達の TensorFlow エージェントについてはネットワークやもジュールを指定するために sonnet ライブラリを利用します ; それとともに作業するネットワークの総ては初期バッチ次元を持ちます、これはバッチ化された推論/学習を可能にします。

環境から返される観測は何らかの方法でネストされることが可能です : e.g. dm_control スーツからの環境はしばしば位置と速度エントリを含む辞書として返されます。私達のネットワークはこの辞書をアクションを生成するために恣意的にマップすることが許されます、しかしこの場合それを MLP を通して供給する前にこれらの観測を単純に結合します。ネストされた観測を各バッチのための単一次元に平坦化する Acme の batch_concat ユティリティを使用してそれを行なうことができます。観測が既にフラットであるならばこれは no-op となるでしょう。

同様に、MLP の出力はアクション spec が規定するものとは異なる値の範囲を持つかもしれません。このためには、アクションを spec に適合するように再スケールするため Acme の TanhToSpec を利用できます。

# Calculate how big the last layer should be based on total # of actions.
action_spec = environment_spec.actions
action_size = np.prod(action_spec.shape, dtype=int)
exploration_sigma = 0.3

# In order the following modules:
# 1. Flatten the observations to be [B, ...] where B is the batch dimension.
# 2. Define a simple MLP which is the guts of this policy.
# 3. Make sure the output action matches the spec of the actions.
policy_modules = [
    tf2_utils.batch_concat,
    networks.LayerNormMLP(layer_sizes=(300, 200, action_size)),
    networks.TanhToSpec(spec=environment_spec.actions)]

policy_network = snt.Sequential(policy_modules)

# We will also create a version of this policy that uses exploratory noise.
behavior_network = snt.Sequential(
    policy_modules + [networks.ClippedGaussian(exploration_sigma),
                      networks.ClipToSpec(action_spec)])

 

アクターを作成する

アクターは私達のフレームワークの一部でアクションを生成することにより環境と直接相互作用します。より詳細には先の図はこの相互作用がどのように起きるかを正確に示すために展開できます :

貴方自身のアクターを常に書ける一方で、Acme では幾つかの有用な事前作成されたバージョンも提供します。上で指定したネットワークについては FeedForwardActor を利用します、これは 単一の順伝播ネットワークをラップしてそして任意のバッチ次元を処理したり観測された遷移を記録するようなことをどのように行なうかを知っています。

actor = actors.FeedForwardActor(policy_network)

総てのアクターは次の公開メソッドと属性を持ちます :

[method_or_attr for method_or_attr in dir(actor)  # pylint: disable=expression-not-assigned
 if not method_or_attr.startswith('_')]

 

ランダムアクターのポリシーを評価する

ポリシーを持つアクターをインスタンス化しましたが、ポリシーはどのようなタスク報酬を獲得することもまだ学習していませんので、本質的には単にランダムに動作するだけです。けれどもこれはアクターと環境がどのように相互作用するかを見るために完全な機会です。下でフレームが与えられたときに動画を表示する単純なヘルパー関数を定義します、そして世界でアクションを取るアクターの 500 ステップを示します。

def display_video(frames, filename='temp.mp4'):
  """Save and display video."""
  # Write video
  with imageio.get_writer(filename, fps=60) as video:
    for frame in frames:
      video.append_data(frame)
  # Read video and display the video
  video = open(filename, 'rb').read()
  b64_video = base64.b64encode(video)
  video_tag = ('
# Run the actor in the environment for desired number of steps.
frames = []
num_steps = 500
timestep = environment.reset()

for _ in range(num_steps):
  frames.append(render(environment))
  action = actor.select_action(timestep.observation)
  timestep = environment.step(action)

# Save video of the behaviour.
display_video(np.array(frames))

 

再生バッファにアクター経験をストアする

多くの RL エージェントはアクターにより取られたアクションと一緒に環境からのデータ (e.g. 観測) をストアするために再生バッファのようなデータ構造を利用します。このデータはポリシーを更新するために後で学習プロセスに供給されます。このステップを含むために再度前の図を展開することができます :

これを可能にするため、Acme は Reverb を活用します、これは機械学習研究のために設計された効率的で利用が容易なデータストレージと輸送系です。下でそれと相互作用する前に再生バッファを作成します。

# Create a table with the following attributes:
# 1. when replay is full we remove the oldest entries first.
# 2. to sample from replay we will do so uniformly at random.
# 3. before allowing sampling to proceed we make sure there is at least
#    one sample in the replay table.
# 4. we use a default table name so we don't have to repeat it many times below;
#    if we left this off we'd need to feed it into adders/actors/etc. below.
replay_buffer = reverb.Table(
    name=adders.DEFAULT_PRIORITY_TABLE,
    max_size=1000000,
    remover=reverb.selectors.Fifo(),
    sampler=reverb.selectors.Uniform(),
    rate_limiter=reverb.rate_limiters.MinSize(min_size_to_sample=1))

# Get the server and address so we can give it to the modules such as our actor
# that will interact with the replay buffer.
replay_server = reverb.Server([replay_buffer], port=None)
replay_server_address = 'localhost:%d' % replay_server.port

データを再生 (バッファ) に追加するために Reverb と直接相互作用できるでしょう。けれども Acme ではこのデータストレージの上に追加の層を持ちます、これは挿入しているデータがどのような種類であろうとも同じインターフェイスを使用することを可能にします。

Acme のこの層は経験をデータテーブルに追加する Adder に相当します。私達はテーブルにストアされることが望まれる情報のタイプに基づいて異なる幾つかの adder を提供します、けれどもこの場合 NStepTransitionAdder を利用します、これは単純な遷移をストアするか (if N=1) 塊状の遷移を形成する N-ステップを蓄積します。

# Create a 5-step transition adder where in between those steps a discount of
# 0.99 is used (which should be the same discount used for learning).
adder = adders.NStepTransitionAdder(
    client=reverb.Client(replay_server_address),
    n_step=5,
    discount=0.99)

以下のように add() と add_first() メソッドを使用して遷移を再生バッファに直接追加するために adder を直接利用することもできます :

num_episodes = 2  #@param

for episode in range(num_episodes):
  timestep = environment.reset()
  adder.add_first(timestep)

  while not timestep.last():
    action = actor.select_action(timestep.observation)
    timestep = environment.step(action)
    adder.add(action=action, next_timestep=timestep)

これはデータを観測するために共通の十分な方法ですので、Acme のアクターは一般的に Adder インスタンスを取ります、これは observation メソッドを定義するために利用します。先に総てのアクターのように FeedForwardActor が observe と observe_first メソッドを定義することを見ました。アクターに Adder インスタンスを初期化時に与えれば、それは観測を作成するためにこの adder を利用します。

actor = actors.FeedForwardActor(policy_network=behavior_network, adder=adder)

下で同じプロセスを繰り返しますが、アクターとその observe メソッドを利用します。下でこられの微妙な変更に注意します。

num_episodes = 2  #@param

for episode in range(num_episodes):
  timestep = environment.reset()
  actor.observe_first(timestep)  # Note: observe_first.

  while not timestep.last():
    action = actor.select_action(timestep.observation)
    timestep = environment.step(action)
    actor.observe(action=action, next_timestep=timestep)  # Note: observe.

 

再生バッファの経験から学習する

Acme は複数の学習アルゴリズム/エージェントを提供します。ここではアクターにより収集されたデータから学習するために D4PG 学習アルゴリズムを使用します。そのため、make_dataset 関数を使用して Reverb テーブルから TensorFlow データセットを最初に作成します。

# This connects to the created reverb server; also note that we use a transition
# adder above so we'll tell the dataset function that so that it knows the type
# of data that's coming out.
dataset = datasets.make_dataset(
    client=reverb.TFClient(server_address=replay_server_address),
    batch_size=256,
    environment_spec=environment_spec,
    transition_adder=True)

以後、D4PG を利用します、actor-critic 学習アルゴリズムです。D4PG は幾分複雑なアルゴリズムですので、このメソッドの完全な説明は添付のペーパーに委ねます (ドキュメント参照)。

けれども、D4PG は actor-critic アルゴリズムですので、そのために critic (値関数) を指定しなければなりません。この場合 D4PG は更に分布 critic も使用します。D4PG はまたオンラインとターゲットネットワークも利用しますので、(前からの) policy_network と作成しようとしている新しい critic ネットワークの両者のコピーを作成する必要があります。

critic ネットワークを構築するため、multiplexer を利用します、これは単純にニューラルネットワーク・モジュールで複数の入力を取りそれらを結合して更に処理する前にそれらを異なる方法で処理します。Acme の CriticMultiplexer の場合、入力は観測とアクションで、それぞれ自身のネットワーク・トルソーを伴います。それから critic ネットワーク・もジュールがあります、これは観測ネットワークとアクションネットワークの出力を処理して tensor を出力します。

最後に、これらのネットワークを最適化するために learner は作成された変数を持つネットワークを受け取らなければなりません。Acme でこれを正確に扱うユティリティを持ちます、そして次のコードブロックの最後の行でそれを行ないます。

critic_network = snt.Sequential([
    networks.CriticMultiplexer(
        observation_network=tf2_utils.batch_concat,
        action_network=tf.identity,
        critic_network=networks.LayerNormMLP(
            layer_sizes=(400, 300),
            activate_final=True)),
    # Value-head gives a 51-atomed delta distribution over state-action values.
    networks.DiscreteValuedHead(vmin=-150., vmax=150., num_atoms=51)])

# Create the target networks
target_policy_network = copy.deepcopy(policy_network)
target_critic_network = copy.deepcopy(critic_network)

# We must create the variables in the networks before passing them to learner.
tf2_utils.create_variables(network=policy_network,
                           input_spec=[environment_spec.observations])
tf2_utils.create_variables(network=critic_network,
                           input_spec=[environment_spec.observations,
                                       environment_spec.actions])
tf2_utils.create_variables(network=target_policy_network,
                           input_spec=[environment_spec.observations])
tf2_utils.create_variables(network=target_critic_network,
                           input_spec=[environment_spec.observations,
                                       environment_spec.actions])

今はこれらのネットワークを使用する learner を作成できます。ここでは遷移 adder で使用されたのと同じ割引き因子を使用していることに注意してください。残りのパラメータは合理的なデフォルトです。

けれども通常の間隔で出力を端末にロギングしていることに注意してください。ネットワーク重みのチェックポイント (i.e. それらのセーブ) も無効にしています。これはデフォルトでは通常は使用されますが、対話的な colab セッションでは問題を引き起こす可能性がありおます。

learner = d4pg.D4PGLearner(policy_network=policy_network,
                           critic_network=critic_network,
                           target_policy_network=target_policy_network,
                           target_critic_network=target_critic_network,
                           dataset=dataset,
                           discount=0.99,
                           target_update_period=100,
                           policy_optimizer=snt.optimizers.Adam(1e-4),
                           critic_optimizer=snt.optimizers.Adam(1e-4),
                           # Log learner updates to console every 10 seconds.
                           logger=loggers.TerminalLogger(time_delta=10.),
                           checkpoint=False)

learner の公開メソッドを調べるとそれは主として変数を公開してそれらを更新するために存在することを見ます。つまりこれは教師あり学習に非常に類似しています。

[method_or_attr for method_or_attr in dir(learner)  # pylint: disable=expression-not-assigned
 if not method_or_attr.startswith('_')]

learner の step() メソッドはそれに与えられた再生データセットからデータのバッチをサンプリングして、そして optimizer を使用し、そこまでの損失メトリクスをログ記録しながら最適化を遂行します。Note: 再生データセットからサンプリングするためには、再生バッファに少なくとも 1000 要素がなければなりません (それはアクターの追加された経験から既に持つはずです)。

learner.step()

 

訓練ループ

最後に、ピースの総てをひとつにまとめて環境で幾つかの訓練ステップを実行できます、アクターの経験収集と learner の学習を交互に行ないます。

これは num_training_episodes エピソードの間実行する単純な訓練ループです、そこではアクターと learner は交代でそれぞれ (経験を) 生成して経験から学習します :

  • アクターは環境で動作し & min_actor_steps_per_iteration ステップの間経験を再生に追加します。
  • learner は再生データからサンプリングしてそれから num_learner_steps_per_iteration ステップの間学習します。

Note: learner とアクターはポリシーネットワークを共有しますので、learner 上で成された任意の学習は、自動的にアクターのポリシーに転送されます。

num_training_episodes =  10 # @param {type: "integer"}
min_actor_steps_before_learning = 1000  # @param {type: "integer"}
num_actor_steps_per_iteration =   100 # @param {type: "integer"}
num_learner_steps_per_iteration = 1  # @param {type: "integer"}

learner_steps_taken = 0
actor_steps_taken = 0
for episode in range(num_training_episodes):
  
  timestep = environment.reset()
  actor.observe_first(timestep)
  episode_return = 0

  while not timestep.last():
    # Get an action from the agent and step in the environment.
    action = actor.select_action(timestep.observation)
    next_timestep = environment.step(action)

    # Record the transition.
    actor.observe(action=action, next_timestep=next_timestep)

    # Book-keeping.
    episode_return += next_timestep.reward
    actor_steps_taken += 1
    timestep = next_timestep

    # See if we have some learning to do.
    if (actor_steps_taken >= min_actor_steps_before_learning and
        actor_steps_taken % num_actor_steps_per_iteration == 0):
      # Learn.
      for learner_step in range(num_learner_steps_per_iteration):
        learner.step()
      learner_steps_taken += num_learner_steps_per_iteration

  # Log quantities.
  print('Episode: %d | Return: %f | Learner steps: %d | Actor steps: %d'%(
      episode, episode_return, learner_steps_taken, actor_steps_taken))

 

ひとつにまとめる: Acme エージェント

ピースの総てを使用してそれらがどのように相互作用できるかを見た今、それらを一つにまとめられるもう一つの方法があります。Acme 設計スキームでは、エージェントは learner とアクター・コンポーネントの両者を持つ実在で、それらの相互作用を内部的にまとめます。エージェントは経験を再生バッファに追加するアクターと、それからサンプリングして学習し、交代で、重みをアクターと共有し戻す leaner の間のやり取りを処理します。

上のカスタム訓練ループで num_actor_steps_per_iteration と num_learner_steps_per_iteration パラメータをどのように使用したかに類似して、エージェント・パラメータ min_observations と observations_per_step はエージェントの訓練ループの構造を指定します。

  • min_observations は学習を開始するために幾つのアクター・ステップが起きる必要があるかを指定します。
  • observations_per_step は各 learner ステップの間内に幾つのアクター・ステップが発生するかを指定します。 
d4pg_agent = agent.Agent(actor=actor,
                         learner=learner,
                         min_observations=1000,
                         observations_per_step=8.)

もちろん agents.D4PG エージェントを単に直接使用することもできました、これはこの総てを私達のために設定します。ちょうど作成したこのエージェントで続けますが、このチュートリアルで概説されるスキップの殆どは単に事前構築されたエージェントと環境ループを利用することによりスキップできます。

 

完全な (= full) エージェントを訓練する

経験を収集してストアすることを単純化するため、Acme の EnvironmentLoop を直接使用することもできます、これは指定された数のエピソードのために環境ループを実行します。各エピソードはそれ自身がループで、観測を得てからその観測を (アクションを取得するため) エージェントに与えるように環境と最初に相互作用します。エピソードの終了時、新しいエピソードが開始されます。エピソードの数が与えられない場合にはこれは環境と無限に相互作用します。

# This may be necessary if any of the episodes were cancelled above.
adder.reset()

# We also want to make sure the logger doesn't write to disk because that can
# cause issues in colab on occasion.
logger = loggers.TerminalLogger(time_delta=10.)

loop = environment_loop.EnvironmentLoop(environment, d4pg_agent, logger=logger)
loop.run(num_episodes=50)

 

D4PG エージェントを評価する

今はエージェントを評価できます。これは noisy behavior ポリシーを使用しますので、最適ではないことに注意してください。絶対的に精密であることを望んだ場合にはこれを noise-free ポリシーと容易に置き換えられたでしょう。最適なポリシーはこの環境でおよそ 1000 報酬を得られることに注意してください。D4PG は一般には 50-100 leaner ステップ内でそれに到達するはずです。このチュートリアルを単に単純化するためにそれを 50 でカットして behavior noise を破棄しません。

# Run the actor in the environment for desired number of steps.
frames = []
num_steps = 500
timestep = environment.reset()

for _ in range(num_steps):
  frames.append(render(environment))
  action = d4pg_agent.select_action(timestep.observation)
  timestep = environment.step(action)

# Save video of the behaviour.
display_video(np.array(frames))
 

以上