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

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

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

Environments

* サンプルコードの動作確認はしておりますが、必要な場合には適宜、追加改変しています。
* ご自由にリンクを張って頂いてかまいませんが、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 メソッドを持ちます、これは環境にアクションを適用して、次のステップについて以下の情報を返します :

observation (観測): これはエージェントが次のステップでそのアクションを選択するために観測できる環境状態の一部です。
reward (報酬): エージェントはマルチステップに渡りこれらの報酬の総計を最大化することを学習しています。
step_type: 環境との相互作用は通常はシークエンス/エピソードの一部です、e.g. チェスゲームの複数の移動。step_type は FIRST, MID or LAST のいずれかであり得て、この時間ステップが最初、中間 or 最後のステップであるかを示します。
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…10 と番号付けられたカードの無限のデッキを使用してプレイされます。
総てのターンでエージェントは 2 つのことを行えます : 新しいランダムカードを得るか、現在のラウンドを停止するかです。
目標はラウンドの最後で、越えることなく、カードの総計をできる限り 21 に近づけることです。

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

アクション: 2 つのアクションを持ちます。Action 0: 新しいカードを得る、そして Action 1: 現在のラウンドを停止する。
観測: 現在のラウンドのカードの総計。
報酬: 目的は越えることなく出来る限り 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 で見つかります。例えば :

ActionDiscretizeWrapper: 連続アクション空間を離散アクション空間に変換します。
RunStats: 取られるステップ数、完了したエピソードの数等のような環境の実行統計を捕捉します。
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)

以上

月	火	水	木	金	土	日
		1	2	3	4	5
6	7	8	9	10	11	12
13	14	15	16	17	18	19
20	21	22	23	24	25	26
27	28	29	30