Ta strona została przetłumaczona przez Cloud Translation API.

Środowiska

Copyright 2021 Autorzy TF-Agents.

Zobacz na TensorFlow.org

Uruchom w Google Colab

Wyświetl źródło na GitHub

Pobierz notatnik

Wstęp

Celem uczenia się przez wzmacnianie (RL) jest zaprojektowanie agentów, które uczą się poprzez interakcję ze środowiskiem. W standardowym ustawieniu RL agent otrzymuje obserwację w każdym kroku czasowym i wybiera akcję. Akcja jest stosowana do otoczenia, a środowisko zwraca nagrodę i nową obserwację. Agent szkoli politykę wyboru działań, aby zmaksymalizować sumę nagród, nazywaną również zwrotem.

W TF-Agents środowiska mogą być implementowane w Pythonie lub TensorFlow. Środowiska Python są zwykle łatwiejsze do zaimplementowania, zrozumienia i debugowania, ale środowiska TensorFlow są bardziej wydajne i umożliwiają naturalną równoległość. Najpopularniejszym przepływem pracy jest implementacja środowiska w Pythonie i użycie jednego z naszych wrapperów, aby automatycznie przekonwertować je na TensorFlow.

Przyjrzyjmy się najpierw środowiskom Pythona. Środowiska TensorFlow korzystają z bardzo podobnego API.

Ustawiać

Jeśli nie zainstalowałeś jeszcze agentów tf ani siłowni, uruchom:

pip install "gym>=0.21.0"
pip install 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

Środowiska Pythona

Środowiskach Python posiada step(action) -> next_time_step metodę, która dotyczy działania na środowisko i zwraca następujące informacje dotyczące następnego kroku:

observation : To jest częścią stanu środowiska, że agent może obserwować wybrać swoje działania w następnym kroku.
reward : Agent uczy się maksymalizować sumę tych nagród na wielu etapach.
step_type : Interakcje z otoczeniem są zwykle częścią sekwencji / epizodu. np. wielokrotne ruchy w grze w szachy. step_type może być FIRST , MID lub LAST aby wskazać, czy czas kroku jest pierwszym pośrednim lub ostatni etap w sekwencji.
discount : To jest pływak reprezentujący ile waga nagrodę w następnym kroku czasowym w stosunku do wynagrodzenia w bieżącym kroku czasowym.

Są one zgrupowane w nazwie krotki TimeStep(step_type, reward, discount, observation) .

Interfejs że wszystkie środowiska Python musi wdrożyć w environments/py_environment.PyEnvironment . Główne metody to:

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."""

Ponadto na step() sposobu środowiska stanowią również reset() sposobu, który uruchamia nowe sekwencje i zapewnia wstępny TimeStep . Nie jest konieczne, aby zadzwonić do reset sposób jawny. Zakładamy, że środowiska resetują się automatycznie, albo po dojściu do końca odcinka, albo po pierwszym wywołaniu funkcji step().

Należy zauważyć, że podklasy nie realizować step() lub reset() bezpośrednio. Oni zamiast zastąpić _step() i _reset() metod. Kroki czas wrócili z tych metod będą buforowane i wystawiony przez current_time_step() .

observation_spec i action_spec metody powrotu gniazdo (Bounded)ArraySpecs opisujących nazwę, kształt, typ danych i zakresy obserwacji i działań odpowiednio.

W TF-Agents wielokrotnie odwołujemy się do gniazd, które definiuje się jako dowolne drzewo, które składa się z list, krotek, krotek nazwanych lub słowników. Można je dowolnie komponować, aby zachować strukturę obserwacji i działań. Odkryliśmy, że jest to bardzo przydatne w bardziej złożonych środowiskach, w których masz wiele obserwacji i działań.

Korzystanie ze standardowych środowisk

TF Agencje ma wbudowane opakowania dla wielu standardowych środowiskach, takich jak OpenAI Siłownia, DeepMind-kontrola i Atari, dzięki czemu są zgodne z naszymi py_environment.PyEnvironment interfejs. Te opakowane środowiska można łatwo załadować za pomocą naszych pakietów środowiskowych. Załadujmy środowisko CartPole z siłowni OpenAI i spójrzmy na akcję i 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')

Widzimy więc, że środowisko oczekuje działań typu int64 w [0, 1] i zwraca TimeSteps gdzie obserwacje float32 wektor o długości 4 i współczynnika dyskonta jest float32 w [0.0, 1.0]. Teraz spróbujmy podjąć stałą akcję (1,) na całym odcinku.

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(
{'discount': array(1., dtype=float32),
 'observation': array([ 0.0138565 , -0.03582913,  0.04861612, -0.03755046], dtype=float32),
 'reward': array(0., dtype=float32),
 'step_type': array(0, dtype=int32)})
TimeStep(
{'discount': array(1., dtype=float32),
 'observation': array([ 0.01313992,  0.15856317,  0.0478651 , -0.3145069 ], dtype=float32),
 'reward': array(1., dtype=float32),
 'step_type': array(1, dtype=int32)})
TimeStep(
{'discount': array(1., dtype=float32),
 'observation': array([ 0.01631118,  0.35297176,  0.04157497, -0.5917188 ], dtype=float32),
 'reward': array(1., dtype=float32),
 'step_type': array(1, dtype=int32)})
TimeStep(
{'discount': array(1., dtype=float32),
 'observation': array([ 0.02337062,  0.54748774,  0.02974059, -0.87102115], dtype=float32),
 'reward': array(1., dtype=float32),
 'step_type': array(1, dtype=int32)})
TimeStep(
{'discount': array(1., dtype=float32),
 'observation': array([ 0.03432037,  0.74219286,  0.01232017, -1.1542072 ], dtype=float32),
 'reward': array(1., dtype=float32),
 'step_type': array(1, dtype=int32)})
TimeStep(
{'discount': array(1., dtype=float32),
 'observation': array([ 0.04916423,  0.93715197, -0.01076398, -1.4430016 ], dtype=float32),
 'reward': array(1., dtype=float32),
 'step_type': array(1, dtype=int32)})
TimeStep(
{'discount': array(1., dtype=float32),
 'observation': array([ 0.06790727,  1.1324048 , -0.03962401, -1.7390285 ], dtype=float32),
 'reward': array(1., dtype=float32),
 'step_type': array(1, dtype=int32)})
TimeStep(
{'discount': array(1., dtype=float32),
 'observation': array([ 0.09055536,  1.327955  , -0.07440457, -2.04377   ], dtype=float32),
 'reward': array(1., dtype=float32),
 'step_type': array(1, dtype=int32)})
TimeStep(
{'discount': array(1., dtype=float32),
 'observation': array([ 0.11711447,  1.523758  , -0.11527998, -2.3585167 ], dtype=float32),
 'reward': array(1., dtype=float32),
 'step_type': array(1, dtype=int32)})
TimeStep(
{'discount': array(1., dtype=float32),
 'observation': array([ 0.14758962,  1.7197047 , -0.16245031, -2.6843033 ], dtype=float32),
 'reward': array(1., dtype=float32),
 'step_type': array(1, dtype=int32)})
TimeStep(
{'discount': array(0., dtype=float32),
 'observation': array([ 0.18198372,  1.9156038 , -0.21613638, -3.0218334 ], dtype=float32),
 'reward': array(1., dtype=float32),
 'step_type': array(2, dtype=int32)})

Tworzenie własnego środowiska Pythona

Dla wielu klientów częstym przypadkiem użycia jest zastosowanie jednego ze standardowych agentów (patrz agenty/) w TF-Agents do ich problemu. Aby to zrobić, muszą przedstawić swój problem jako środowisko. Przyjrzyjmy się więc, jak zaimplementować środowisko w Pythonie.

Załóżmy, że chcemy wyszkolić agenta do gry w następującą (inspirowaną Black Jackiem) grę karcianą:

Gra toczy się za pomocą nieskończonej talii kart o numerach 1...10.
W każdej turze agent może zrobić 2 rzeczy: zdobyć nową losową kartę lub zatrzymać obecną rundę.
Celem jest, aby suma twoich kart była jak najbardziej zbliżona do 21 na koniec rundy, bez przekraczania.

Środowisko reprezentujące grę może wyglądać tak:

Akcje: Mamy 2 akcje. Akcja 0: zdobądź nową kartę i Akcja 1: zakończ bieżącą rundę.
Obserwacje: Suma kart w bieżącej rundzie.
Nagroda: Celem jest zbliżyć się do 21, jak to możliwe bez przekraczania, więc możemy to osiągnąć, korzystając z następującej nagrody na koniec rundy: sum_of_cards - 21, jeśli sum_of_cards <= 21, w przeciwnym razie -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)

Upewnijmy się, że zrobiliśmy wszystko poprawnie definiując powyższe środowisko. Tworząc własne środowisko, musisz upewnić się, że generowane obserwacje i time_steps są zgodne z prawidłowymi kształtami i typami określonymi w specyfikacji. Są one używane do generowania wykresu TensorFlow i jako takie mogą powodować trudne do debugowania problemy, jeśli źle je zrobimy.

Aby zweryfikować nasze środowisko, użyjemy losowych zasad do generowania działań i będziemy iterować przez 5 odcinków, aby upewnić się, że wszystko działa zgodnie z przeznaczeniem. Jeśli otrzymamy time_step, który nie jest zgodny ze specyfikacją środowiska, zgłaszany jest błąd.

environment = CardGameEnv()
utils.validate_py_environment(environment, episodes=5)

Teraz, gdy wiemy, że środowisko działa zgodnie z przeznaczeniem, uruchommy to środowisko przy użyciu ustalonej zasady: poproś o 3 karty, a następnie zakończ rundę.

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(
{'discount': array(1., dtype=float32),
 'observation': array([0], dtype=int32),
 'reward': array(0., dtype=float32),
 'step_type': array(0, dtype=int32)})
TimeStep(
{'discount': array(1., dtype=float32),
 'observation': array([9], dtype=int32),
 'reward': array(0., dtype=float32),
 'step_type': array(1, dtype=int32)})
TimeStep(
{'discount': array(1., dtype=float32),
 'observation': array([12], dtype=int32),
 'reward': array(0., dtype=float32),
 'step_type': array(1, dtype=int32)})
TimeStep(
{'discount': array(0., dtype=float32),
 'observation': array([21], dtype=int32),
 'reward': array(0., dtype=float32),
 'step_type': array(2, dtype=int32)})
TimeStep(
{'discount': array(0., dtype=float32),
 'observation': array([21], dtype=int32),
 'reward': array(0., dtype=float32),
 'step_type': array(2, dtype=int32)})
Final Reward =  0.0

Owijarki środowiskowe

Opakowanie środowiska przyjmuje środowisko Pythona i zwraca zmodyfikowaną wersję środowiska. Zarówno pierwotny środowiska i zmodyfikowany środowisko są przypadki py_environment.PyEnvironment oraz wiele banderoli może być połączony ze sobą.

Niektóre wspólne owijarki można znaleźć w environments/wrappers.py . Na przykład:

ActionDiscretizeWrapper : Konwersja ciągłą przestrzeń działania, oddzielnego miejsca działania.
RunStats : Przechwytuje uruchomić statystyki środowiska, takich jak liczba kroków podjętych, liczba epizodów zakończone etc.
TimeLimit : Kończy epizod po ustalonej liczbie kroków.

Przykład 1: Akcja Dyskretyzuj opakowanie

InvertedPendulum jest środowiskiem PyBullet przyjmująca ciągłego działania w zakresie [-2, 2] . Jeśli chcemy wyszkolić dyskretnego agenta akcji, takiego jak DQN, w tym środowisku, musimy zdyskretyzować (skwantować) przestrzeń akcji. To jest dokładnie to, co ActionDiscretizeWrapper robi. Porównaj action_spec przed i po opakowaniu:

env = suite_gym.load('Pendulum-v1')
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)

Owinięty discrete_action_env jest wystąpienie py_environment.PyEnvironment i może być traktowany jak zwykły środowisku Pythona.

Środowiska TensorFlow

Interfejs dla środowisk TF jest zdefiniowana w environments/tf_environment.TFEnvironment i wygląda bardzo podobnie do środowisk Pythona. Środowiska TF różnią się od środowisk środowiska Python na kilka sposobów:

Generują obiekty tensorowe zamiast tablic
Środowiska TF dodają wymiar wsadowy do tensorów generowanych w porównaniu ze specyfikacjami.

Konwersja środowisk Pythona na TFEnvs pozwala tensorflow na zrównoleglenie operacji. Na przykład, można zdefiniować collect_experience_op który zbiera dane z otoczenia i dodaje do replay_buffer , i train_op który odczytuje z replay_buffer i trenuje środka i uruchomić je równolegle naturalnie w 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() metoda zwraca bieżącą time_step i inicjuje środowiska w razie potrzeby.

W reset() siły metoda resetu w środowisku i zwraca CURRENT_STEP.

Jeśli action nie zależy od poprzedniego time_step tf.control_dependency jest potrzebne w Graph trybie.

Teraz przyjrzyjmy się, jak TFEnvironments są tworzone.

Tworzenie własnego środowiska TensorFlow

Jest to bardziej skomplikowane niż tworzenie środowisk w Pythonie, więc nie będziemy tego omawiać w tej kolaboracji. Przykładem jest dostępny tutaj . Im bardziej powszechny przypadek użycia jest wdrożenie środowiska w Pythonie i zawinąć go w TensorFlow chętnie TFPyEnvironment opakowanie (patrz niżej).

Zawijanie środowiska Pythona w TensorFlow

Możemy łatwo owinąć każdą środowisko Python w środowisku TensorFlow używając TFPyEnvironment opakowanie.

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(
{'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)),
 'reward': TensorSpec(shape=(), dtype=tf.float32, name='reward'),
 'step_type': TensorSpec(shape=(), dtype=tf.int32, name='step_type')})
Action Specs: BoundedTensorSpec(shape=(), dtype=tf.int64, name='action', minimum=array(0), maximum=array(1))

Uwaga specyfikacje są teraz typu: (Bounded)TensorSpec .

Przykłady użycia

Prosty przykład

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(
{'discount': array([1.], dtype=float32),
 'observation': array([[-0.0078796 , -0.04736348, -0.04966116,  0.04563603]],
      dtype=float32),
 'reward': array([0.], dtype=float32),
 'step_type': array([0], dtype=int32)}), array([0], dtype=int32), TimeStep(
{'discount': array([1.], dtype=float32),
 'observation': array([[-0.00882687, -0.24173944, -0.04874843,  0.32224613]],
      dtype=float32),
 'reward': array([1.], dtype=float32),
 'step_type': array([1], dtype=int32)})]
[TimeStep(
{'discount': array([1.], dtype=float32),
 'observation': array([[-0.00882687, -0.24173944, -0.04874843,  0.32224613]],
      dtype=float32),
 'reward': array([1.], dtype=float32),
 'step_type': array([1], dtype=int32)}), array([1], dtype=int32), TimeStep(
{'discount': array([1.], dtype=float32),
 'observation': array([[-0.01366166, -0.04595843, -0.04230351,  0.01459712]],
      dtype=float32),
 'reward': array([1.], dtype=float32),
 'step_type': array([1], dtype=int32)})]
[TimeStep(
{'discount': array([1.], dtype=float32),
 'observation': array([[-0.01366166, -0.04595843, -0.04230351,  0.01459712]],
      dtype=float32),
 'reward': array([1.], dtype=float32),
 'step_type': array([1], dtype=int32)}), array([0], dtype=int32), TimeStep(
{'discount': array([1.], dtype=float32),
 'observation': array([[-0.01458083, -0.24044897, -0.04201157,  0.2936384 ]],
      dtype=float32),
 'reward': array([1.], dtype=float32),
 'step_type': array([1], dtype=int32)})]
Total reward: [3.]

Całe odcinki

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: 131
avg_length 26.2 avg_reward: 26.2