環境

Copyright 2021 TF-AgentsAuthors。

TensorFlow.orgで表示

GoogleColabで実行

GitHubでソースを表示

ノートブックをダウンロード

前書き

強化学習（RL）の目標は、環境と相互作用することによって学習するエージェントを設計することです。標準のRL設定では、エージェントはすべてのタイムステップでオブザベーションを受け取り、アクションを選択します。アクションは環境に適用され、環境は報酬と新しい観察を返します。エージェントは、リターンとも呼ばれる報酬の合計を最大化するアクションを選択するポリシーをトレーニングします。

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

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

セットアップ

tf-agentsまたはジムをまだインストールしていない場合は、次を実行します。

pip install -q 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 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 ：環境との相互作用は通常、シーケンス/エピソードの一部です。たとえば、チェスのゲームでの複数の動き。 step_typeは、 FIRST 、 MIDまたはLASTいずれかで、このタイムステップがシーケンスの最初、中間、または最後のステップであるかどうかを示します。
discount ：これは、現在のタイムステップでの報酬と比較して、次のタイムステップでの報酬をどの程度重み付けするかを表すフロートです。

これらは、名前付きタプル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."""

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

サブクラスはstep()またはreset()直接実装しないことに注意してください。彼らは代わりに上書き_step()と_reset()メソッドを。これらのメソッドから返されるタイムステップはキャッシュされ、 current_time_step()介して公開されます。

observation_specとaction_spec方法はの巣返す(Bounded)ArraySpecsそれぞれ観察およびアクションの名前、形状、データ型及び範囲を記載しています。

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

標準環境の使用

TF Agentsには、OpenAI Gym、DeepMind-control、Atariなどの多くの標準環境用の組み込みラッパーがあり、 py_environment.PyEnvironmentインターフェイスに従います。これらのラップされた環境は、環境スイートを使用して簡単にロードできます。 OpenAIジムからCartPole環境をロードし、アクションと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)

action_spec: BoundedArraySpec(shape=(), dtype=dtype('int64'), name='action', minimum=0, maximum=1)
time_step_spec.observation: BoundedArraySpec(shape=(4,), dtype=dtype('float32'), name='observation', minimum=[-4.8000002e+00 -3.4028235e+38 -4.1887903e-01 -3.4028235e+38], maximum=[4.8000002e+00 3.4028235e+38 4.1887903e-01 3.4028235e+38])
time_step_spec.step_type: ArraySpec(shape=(), dtype=dtype('int32'), name='step_type')
time_step_spec.discount: BoundedArraySpec(shape=(), dtype=dtype('float32'), name='discount', minimum=0.0, maximum=1.0)
time_step_spec.reward: ArraySpec(shape=(), dtype=dtype('float32'), name='reward')

したがって、環境は[0、1]でint64型のアクションを予期し、 int64を返しますTimeStepsここで、観測値は長さ4のfloat32ベクトルであり、割引係数は[ float32 ]のfloat32です。それでは、エピソード全体に対して固定アクション(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)

TimeStep(step_type=array(0, dtype=int32), reward=array(0., dtype=float32), discount=array(1., dtype=float32), observation=array([ 0.01285449,  0.04769544,  0.01983412, -0.00245379], dtype=float32))
TimeStep(step_type=array(1, dtype=int32), reward=array(1., dtype=float32), discount=array(1., dtype=float32), observation=array([ 0.0138084 ,  0.24252741,  0.01978504, -0.2888134 ], dtype=float32))
TimeStep(step_type=array(1, dtype=int32), reward=array(1., dtype=float32), discount=array(1., dtype=float32), observation=array([ 0.01865895,  0.43736172,  0.01400878, -0.57519126], dtype=float32))
TimeStep(step_type=array(1, dtype=int32), reward=array(1., dtype=float32), discount=array(1., dtype=float32), observation=array([ 0.02740618,  0.6322845 ,  0.00250495, -0.8634283 ], dtype=float32))
TimeStep(step_type=array(1, dtype=int32), reward=array(1., dtype=float32), discount=array(1., dtype=float32), observation=array([ 0.04005187,  0.82737225, -0.01476362, -1.1553226 ], dtype=float32))
TimeStep(step_type=array(1, dtype=int32), reward=array(1., dtype=float32), discount=array(1., dtype=float32), observation=array([ 0.05659932,  1.0226836 , -0.03787007, -1.452598  ], dtype=float32))
TimeStep(step_type=array(1, dtype=int32), reward=array(1., dtype=float32), discount=array(1., dtype=float32), observation=array([ 0.07705299,  1.2182497 , -0.06692202, -1.7568679 ], dtype=float32))
TimeStep(step_type=array(1, dtype=int32), reward=array(1., dtype=float32), discount=array(1., dtype=float32), observation=array([ 0.10141798,  1.4140631 , -0.10205939, -2.069591  ], dtype=float32))
TimeStep(step_type=array(1, dtype=int32), reward=array(1., dtype=float32), discount=array(1., dtype=float32), observation=array([ 0.12969925,  1.6100639 , -0.1434512 , -2.3920157 ], dtype=float32))
TimeStep(step_type=array(1, dtype=int32), reward=array(1., dtype=float32), discount=array(1., dtype=float32), observation=array([ 0.16190052,  1.8061239 , -0.19129153, -2.725115  ], dtype=float32))
TimeStep(step_type=array(2, dtype=int32), reward=array(1., dtype=float32), discount=array(0., dtype=float32), observation=array([ 0.198023  ,  2.002027  , -0.24579382, -3.0695074 ], dtype=float32))

独自のPython環境の作成

多くのクライアントの一般的な使用例は、TF-Agentsの標準エージェント（agents /を参照）の1つを問題に適用することです。これを行うには、問題を環境として組み立てる必要があります。それでは、Pythonで環境を実装する方法を見てみましょう。

次の（ブラックジャックに触発された）カードゲームをプレイするようにエージェントをトレーニングしたいとします。

ゲームは、1 ... 10の番号が付けられたカードの無限のデッキを使用してプレイされます。
毎ターン、エージェントは2つのことを行うことができます。新しいランダムカードを入手するか、現在のラウンドを停止します。
目標は、ラウンドの終わりに、カードの合計を21にできるだけ近づけることです。

ゲームを表す環境は次のようになります。

アクション：2つのアクションがあります。アクション0：新しいカードを取得し、アクション1：現在のラウンドを終了します。
観察：現在のラウンドのカードの合計。
報酬：目的は、行き過ぎずに可能な限り21に近づくことです。そのため、ラウンドの最後に次の報酬を使用してこれを達成できます。sum_of_cards-21if 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)

上記の環境を正しく定義するためにすべてを行ったことを確認しましょう。独自の環境を作成するときは、生成されたobservationsとtime_stepsが、仕様で定義されている正しい形状とタイプに従っていることを確認する必要があります。これらはTensorFlowグラフを生成するために使用されるため、間違えるとデバッグが困難になる可能性があります。

環境を検証するために、ランダムポリシーを使用してアクションを生成し、5つのエピソードを繰り返して、意図したとおりに機能していることを確認します。環境仕様に従わない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)

TimeStep(step_type=array(0, dtype=int32), reward=array(0., dtype=float32), discount=array(1., dtype=float32), observation=array([0], dtype=int32))
TimeStep(step_type=array(1, dtype=int32), reward=array(0., dtype=float32), discount=array(1., dtype=float32), observation=array([2], dtype=int32))
TimeStep(step_type=array(1, dtype=int32), reward=array(0., dtype=float32), discount=array(1., dtype=float32), observation=array([7], dtype=int32))
TimeStep(step_type=array(1, dtype=int32), reward=array(0., dtype=float32), discount=array(1., dtype=float32), observation=array([8], dtype=int32))
TimeStep(step_type=array(2, dtype=int32), reward=array(-13., dtype=float32), discount=array(0., dtype=float32), observation=array([8], dtype=int32))
Final Reward =  -13.0

環境ラッパー

環境ラッパーはPython環境を取得し、環境の変更バージョンを返します。元の環境と変更された環境はどちらもpy_environment.PyEnvironmentインスタンスでpy_environment.PyEnvironment 、複数のラッパーをチェーン化できます。

いくつかの一般的なラッパーはenvironments/wrappers.pyます。例えば：

ActionDiscretizeWrapper ：連続アクションスペースを離散アクションスペースに変換します。
RunStats ：実行されたステップ数、完了したエピソード数など、環境の実行統計をキャプチャします。
TimeLimit ：一定のステップ数の後にエピソードを終了します。

例1：アクション離散化ラッパー

InvertedPendulumは、 [-2, 2]範囲の連続アクションを受け入れるPyBullet環境です。この環境でDQNなどの個別のアクションエージェントをトレーニングする場合は、アクションスペースを離散化（量子化）する必要があります。これはまさにActionDiscretizeWrapperが行うことです。ラッピングの前後のaction_spec比較し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())

Action Spec: BoundedArraySpec(shape=(1,), dtype=dtype('float32'), name='action', minimum=-2.0, maximum=2.0)
Discretized Action Spec: BoundedArraySpec(shape=(), dtype=dtype('int32'), name='action', minimum=0, maximum=4)

ラップされたdiscrete_action_envはpy_environment.PyEnvironmentインスタンスであり、通常のPython環境のように扱うことができます。

TensorFlow環境

TF環境のインターフェースはenvironments/tf_environment.TFEnvironment定義されており、Python環境と非常によく似ています。 TF環境は、いくつかの点でPython環境とは異なります。

配列の代わりにテンソルオブジェクトを生成します
TF環境は、仕様と比較したときに生成されたテンソルにバッチ次元を追加します。

Python環境をTFEnvsに変換すると、tensorflowで操作を並列化できます。たとえば、環境からデータを収集してreplay_bufferに追加するcollect_experience_opと、 train_opから読み取り、エージェントをトレーニングするreplay_bufferして、 replay_bufferで自然に並行して実行できます。

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を返します。

actionが前のtime_step依存しない場合、 Graphモードでtf.control_dependencyが必要です。

TFEnvironmentsがどのように作成されるかを見てみましょう。

独自のTensorFlow環境を作成する

これはPythonで環境を作成するよりも複雑なので、このコラボでは取り上げません。例はここにあります。より一般的な使用例は、Pythonで環境を実装し、 TFPyEnvironmentラッパーを使用してTFPyEnvironmentラップすることです（以下を参照）。

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

True
TimeStep Specs: TimeStep(step_type=TensorSpec(shape=(), dtype=tf.int32, name='step_type'), reward=TensorSpec(shape=(), dtype=tf.float32, name='reward'), discount=BoundedTensorSpec(shape=(), dtype=tf.float32, name='discount', minimum=array(0., dtype=float32), maximum=array(1., dtype=float32)), observation=BoundedTensorSpec(shape=(4,), dtype=tf.float32, name='observation', minimum=array([-4.8000002e+00, -3.4028235e+38, -4.1887903e-01, -3.4028235e+38],
      dtype=float32), maximum=array([4.8000002e+00, 3.4028235e+38, 4.1887903e-01, 3.4028235e+38],
      dtype=float32)))
Action Specs: BoundedTensorSpec(shape=(), dtype=tf.int64, name='action', minimum=array(0), maximum=array(1))

仕様のタイプが(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())

[TimeStep(step_type=array([0], dtype=int32), reward=array([0.], dtype=float32), discount=array([1.], dtype=float32), observation=array([[-0.03501577, -0.04957427,  0.00623939,  0.03762257]],
      dtype=float32)), array([0], dtype=int32), TimeStep(step_type=array([1], dtype=int32), reward=array([1.], dtype=float32), discount=array([1.], dtype=float32), observation=array([[-0.03600726, -0.24478514,  0.00699184,  0.33226755]],
      dtype=float32))]
[TimeStep(step_type=array([1], dtype=int32), reward=array([1.], dtype=float32), discount=array([1.], dtype=float32), observation=array([[-0.03600726, -0.24478514,  0.00699184,  0.33226755]],
      dtype=float32)), array([1], dtype=int32), TimeStep(step_type=array([1], dtype=int32), reward=array([1.], dtype=float32), discount=array([1.], dtype=float32), observation=array([[-0.04090296, -0.0497634 ,  0.01363719,  0.04179767]],
      dtype=float32))]
[TimeStep(step_type=array([1], dtype=int32), reward=array([1.], dtype=float32), discount=array([1.], dtype=float32), observation=array([[-0.04090296, -0.0497634 ,  0.01363719,  0.04179767]],
      dtype=float32)), array([0], dtype=int32), TimeStep(step_type=array([1], dtype=int32), reward=array([1.], dtype=float32), discount=array([1.], dtype=float32), observation=array([[-0.04189822, -0.24507822,  0.01447314,  0.33875188]],
      dtype=float32))]
Total reward: [3.]

全エピソード

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)

num_episodes: 5 num_steps: 138
avg_length 27.6 avg_reward: 27.6