Copyright 2021 Les auteurs TF-Agents.
 Voir sur TensorFlow.org |  Exécuter dans Google Colab |  Voir la source sur GitHub |  Télécharger le cahier |
introduction
Un modèle courant dans l'apprentissage par renforcement consiste à exécuter une politique dans un environnement pour un nombre spécifié d'étapes ou d'épisodes. Cela se produit, par exemple, lors de la collecte de données, de l'évaluation et de la génération d'une vidéo de l'agent.
Bien que ce soit relativement simple d'écrire en python, il est beaucoup plus complexe à écrire et déboguer dans tensorflow car elle implique tf.while boucles, tf.cond et tf.control_dependencies . Par conséquent , nous faisons abstraction de cette notion d'une boucle d'exécution dans une classe appelée driver et fournir des implémentations éprouvées à la fois en Python et tensorflow.
De plus, les données rencontrées par le pilote à chaque étape sont enregistrées dans un tuple nommé appelé Trajectoire et diffusées à un ensemble d'observateurs tels que des tampons de relecture et des métriques. Ces données comprennent l'observation de l'environnement, l'action recommandée par la politique, la récompense obtenue, le type de l'étape actuelle et la prochaine étape, etc.
Installer
Si vous n'avez pas encore installé tf-agents ou gym, exécutez :
pip install tf-agents
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
Pilotes Python
La PyDriver classe prend un environnement python, une politique de python et une liste des observateurs à mettre à jour à chaque étape. La principale méthode est run() , quelles étapes l'environnement en utilisant des actions de la politique jusqu'à ce qu'au moins l' un des critères de terminaison suivantes est remplie: le nombre d'étapes atteint max_steps ou le nombre d'épisodes atteint max_episodes .
La mise en œuvre est grosso modo la suivante :
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
Examinons maintenant l'exemple d'exécution d'une stratégie aléatoire sur l'environnement CartPole, en enregistrant les résultats dans un tampon de relecture et en calculant certaines métriques.
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())
Replay Buffer:
Trajectory(
{'action': array(1),
'discount': array(1., dtype=float32),
'next_step_type': array(1, dtype=int32),
'observation': array([-0.01483762, -0.0301547 , -0.02482025, 0.00477367], dtype=float32),
'policy_info': (),
'reward': array(1., dtype=float32),
'step_type': array(0, dtype=int32)})
Trajectory(
{'action': array(1),
'discount': array(1., dtype=float32),
'next_step_type': array(1, dtype=int32),
'observation': array([-0.01544072, 0.16531426, -0.02472478, -0.29563585], dtype=float32),
'policy_info': (),
'reward': array(1., dtype=float32),
'step_type': array(1, dtype=int32)})
Trajectory(
{'action': array(1),
'discount': array(1., dtype=float32),
'next_step_type': array(1, dtype=int32),
'observation': array([-0.01213443, 0.3607798 , -0.0306375 , -0.5960129 ], dtype=float32),
'policy_info': (),
'reward': array(1., dtype=float32),
'step_type': array(1, dtype=int32)})
Trajectory(
{'action': array(1),
'discount': array(1., dtype=float32),
'next_step_type': array(1, dtype=int32),
'observation': array([-0.00491884, 0.5563168 , -0.04255775, -0.8981868 ], dtype=float32),
'policy_info': (),
'reward': array(1., dtype=float32),
'step_type': array(1, dtype=int32)})
Trajectory(
{'action': array(0),
'discount': array(1., dtype=float32),
'next_step_type': array(1, dtype=int32),
'observation': array([ 0.0062075 , 0.75198895, -0.06052149, -1.2039375 ], dtype=float32),
'policy_info': (),
'reward': array(1., dtype=float32),
'step_type': array(1, dtype=int32)})
Trajectory(
{'action': array(0),
'discount': array(1., dtype=float32),
'next_step_type': array(1, dtype=int32),
'observation': array([ 0.02124728, 0.5576993 , -0.08460024, -0.9308191 ], dtype=float32),
'policy_info': (),
'reward': array(1., dtype=float32),
'step_type': array(1, dtype=int32)})
Trajectory(
{'action': array(0),
'discount': array(1., dtype=float32),
'next_step_type': array(1, dtype=int32),
'observation': array([ 0.03240127, 0.36381477, -0.10321662, -0.6658752 ], dtype=float32),
'policy_info': (),
'reward': array(1., dtype=float32),
'step_type': array(1, dtype=int32)})
Trajectory(
{'action': array(1),
'discount': array(1., dtype=float32),
'next_step_type': array(1, dtype=int32),
'observation': array([ 0.03967756, 0.17026839, -0.11653412, -0.40739253], dtype=float32),
'policy_info': (),
'reward': array(1., dtype=float32),
'step_type': array(1, dtype=int32)})
Trajectory(
{'action': array(0),
'discount': array(1., dtype=float32),
'next_step_type': array(1, dtype=int32),
'observation': array([ 0.04308293, 0.36683324, -0.12468197, -0.7344236 ], dtype=float32),
'policy_info': (),
'reward': array(1., dtype=float32),
'step_type': array(1, dtype=int32)})
Trajectory(
{'action': array(0),
'discount': array(1., dtype=float32),
'next_step_type': array(1, dtype=int32),
'observation': array([ 0.0504196 , 0.17363413, -0.13937044, -0.48343614], dtype=float32),
'policy_info': (),
'reward': array(1., dtype=float32),
'step_type': array(1, dtype=int32)})
Trajectory(
{'action': array(1),
'discount': array(1., dtype=float32),
'next_step_type': array(1, dtype=int32),
'observation': array([ 0.05389228, -0.0192741 , -0.14903916, -0.23772195], dtype=float32),
'policy_info': (),
'reward': array(1., dtype=float32),
'step_type': array(1, dtype=int32)})
Trajectory(
{'action': array(1),
'discount': array(1., dtype=float32),
'next_step_type': array(1, dtype=int32),
'observation': array([ 0.0535068 , 0.17762792, -0.1537936 , -0.5734562 ], dtype=float32),
'policy_info': (),
'reward': array(1., dtype=float32),
'step_type': array(1, dtype=int32)})
Trajectory(
{'action': array(0),
'discount': array(1., dtype=float32),
'next_step_type': array(1, dtype=int32),
'observation': array([ 0.05705936, 0.37453365, -0.16526273, -0.910366 ], dtype=float32),
'policy_info': (),
'reward': array(1., dtype=float32),
'step_type': array(1, dtype=int32)})
Trajectory(
{'action': array(0),
'discount': array(1., dtype=float32),
'next_step_type': array(1, dtype=int32),
'observation': array([ 0.06455003, 0.18198717, -0.18347006, -0.6738478 ], dtype=float32),
'policy_info': (),
'reward': array(1., dtype=float32),
'step_type': array(1, dtype=int32)})
Trajectory(
{'action': array(0),
'discount': array(1., dtype=float32),
'next_step_type': array(1, dtype=int32),
'observation': array([ 0.06818977, -0.01017502, -0.19694701, -0.44408032], dtype=float32),
'policy_info': (),
'reward': array(1., dtype=float32),
'step_type': array(1, dtype=int32)})
Trajectory(
{'action': array(0),
'discount': array(0., dtype=float32),
'next_step_type': array(2, dtype=int32),
'observation': array([ 0.06798627, -0.20204504, -0.20582862, -0.21936782], dtype=float32),
'policy_info': (),
'reward': array(1., dtype=float32),
'step_type': array(1, dtype=int32)})
Trajectory(
{'action': array(1),
'discount': array(1., dtype=float32),
'next_step_type': array(0, dtype=int32),
'observation': array([ 0.06394537, -0.39372152, -0.21021597, 0.00199082], dtype=float32),
'policy_info': (),
'reward': array(0., dtype=float32),
'step_type': array(2, dtype=int32)})
Average Return: 16.0
Pilotes TensorFlow
Nous avons aussi les conducteurs à tensorflow qui sont fonctionnellement similaires aux pilotes Python, mais utiliser des environnements de TF, les politiques de TF, les observateurs de TF etc. Nous avons actuellement 2 pilotes tensorflow: DynamicStepDriver , qui se termine après un nombre donné de (valide) étapes de l' environnement et DynamicEpisodeDriver , qui se termine après un nombre donné d'épisodes. Regardons un exemple de DynamicEpisode en action.
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())
final_time_step TimeStep(
{'discount': <tf.Tensor: shape=(1,), dtype=float32, numpy=array([1.], dtype=float32)>,
'observation': <tf.Tensor: shape=(1, 4), dtype=float32, numpy=array([[0.01182632, 0.01372784, 0.03056967, 0.04454206]], dtype=float32)>,
'reward': <tf.Tensor: shape=(1,), dtype=float32, numpy=array([0.], dtype=float32)>,
'step_type': <tf.Tensor: shape=(1,), dtype=int32, numpy=array([0], dtype=int32)>})
Number of Steps: 24
Number of Episodes: 2
# 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())
final_time_step TimeStep(
{'discount': <tf.Tensor: shape=(1,), dtype=float32, numpy=array([1.], dtype=float32)>,
'observation': <tf.Tensor: shape=(1, 4), dtype=float32, numpy=
array([[-0.02565088, 0.04813434, -0.04199163, 0.03810809]],
dtype=float32)>,
'reward': <tf.Tensor: shape=(1,), dtype=float32, numpy=array([0.], dtype=float32)>,
'step_type': <tf.Tensor: shape=(1,), dtype=int32, numpy=array([0], dtype=int32)>})
Number of Steps: 70
Number of Episodes: 4