Copyright 2021 TF-AgentsAuthors。
![]() | ![]() | ![]() | ![]() |
前書き
強化学習(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