סביבות

הצג באתר TensorFlow.org הפעל בגוגל קולאב צפה במקור ב-GitHub הורד מחברת

מבוא

המטרה של למידת חיזוק (RL) היא לעצב סוכנים הלומדים על ידי אינטראקציה עם סביבה. בהגדרת RL הסטנדרטית, הסוכן מקבל תצפית בכל שלב בזמן ובוחר פעולה. הפעולה מיושמת על הסביבה והסביבה מחזירה פרס והתבוננות חדשה. הסוכן מאמן מדיניות לבחירת פעולות כדי למקסם את סכום התגמולים, המכונה גם החזר.

ב-TF-Agents, ניתן ליישם סביבות ב- Python או TensorFlow. סביבות Python בדרך כלל קלות יותר ליישום, הבנה וניפוי באגים, אך סביבות TensorFlow יעילות יותר ומאפשרות הקבלה טבעית. זרימת העבודה הנפוצה ביותר היא ליישם סביבה ב-Python ולהשתמש באחד מהעטיפות שלנו כדי להמיר אותה אוטומטית ל-TensorFlow.

תחילה נסתכל על סביבות Python. סביבות TensorFlow עוקבות אחר API דומה מאוד.

להכין

אם עדיין לא התקנת tf-agents או חדר כושר, הפעל:

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

סביבות פייתון

יש סביבות פייתון step(action) -> next_time_step שיטה כי חל פעולה לסביבה, ומחזיר את המידע הבא על הצעד הבא:

  1. observation : זהו חלק של מדינת הסביבה כי הסוכן יכול לצפות לבחור מעשייה על הצעד הבא.
  2. reward : הסוכן לומד על מנת למקסם את סכום התגמולים אלה בשלבים מרובים.
  3. step_type : אינטראקציות עם הסביבה הן בדרך כלל חלק מרצף / פארק. למשל מספר מהלכים במשחק שח. step_type יכול להיות FIRST , MID או LAST כדי לציין אם צעד פעם הוא הצעד הראשון, ביניים או אחרון ברצף.
  4. discount : זהו לצוף המייצג כמה לשקלל את הפרס על יחסי הצעד בפעם הבאה אל הפרס על הצעד השעה הנוכחית.

אלה מקובצים tuple בשם TimeStep(step_type, reward, discount, observation) .

הממשק שבו כל סביבות פייתון חייב ליישם נמצא 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() השיטה, סביבות גם לספק reset() שיטה שמתחילה רצף חדש ומספקת ראשונית TimeStep . אין צורך להתקשר reset מפורש לשיטה. אנו מניחים שסביבות מתאפסות באופן אוטומטי, או כשהן מגיעות לסוף פרק או כאשר step() נקרא בפעם הראשונה.

הערה כי subclasses לא ליישם step() או reset() ישירות. הם במקום לעקוף את _step() ו _reset() שיטות. שלבי הזמן חזרו משיטות אלה יהיו במטמון וחשוף באמצעות current_time_step() .

observation_spec ואת action_spec שיטות לחזור קן של (Bounded)ArraySpecs המתארות את שמו, צורה, סוג הנתונים ואת טווחי התצפיות ופעולות בהתאמה.

ב-TF-Agents אנו מתייחסים שוב ושוב לקנים המוגדרים ככל מבנה כמו עץ ​​המורכב מרשימות, tuples, name-tuples או מילונים. אלה יכולים להיות מורכבים באופן שרירותי כדי לשמור על מבנה של תצפיות ופעולות. מצאנו שזה מאוד שימושי עבור סביבות מורכבות יותר שבהן יש לך תצפיות ופעולות רבות.

שימוש בסביבות סטנדרטיות

סוכני TF מוכלל עטיפות עבור סביבות סטנדרטיות רבות כמו כושר OpenAI, DeepMind-מלא עטרי, כך שהן תהיינה בהתאם שלנו py_environment.PyEnvironment הממשק. ניתן לטעון את הסביבה העטופה הללו בקלות באמצעות חבילות הסביבה שלנו. בואו נטען את סביבת CartPole מחדר הכושר של OpenAI ונסתכל על הפעולה ו-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')

אז אנחנו רואים שהסביבה מצפה פעולות מסוג int64 ב [0, 1] וחוזר TimeSteps שבו התצפיות הן float32 וקטור באורך 4 ו מקדם היוון הוא float32 ב [0.0, 1.0]. עכשיו, בואו ננסה לקחת פעולה קבוע (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(
{'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)})

יצירת סביבת Python משלך

עבור לקוחות רבים, מקרה שימוש נפוץ הוא להחיל את אחד מהסוכנים הסטנדרטיים (ראה סוכנים/) ב-TF-Agents על הבעיה שלהם. כדי לעשות זאת, הם צריכים למסגר את הבעיה שלהם כסביבה. אז הבה נבחן כיצד ליישם סביבה ב- Python.

נניח שאנו רוצים להכשיר סוכן לשחק את משחק הקלפים הבא (בהשראת בלק ג'ק):

  1. המשחק משוחק באמצעות חפיסת קלפים אינסופית שמספרה 1...10.
  2. בכל תור הסוכן יכול לעשות 2 דברים: לקבל קלף אקראי חדש, או לעצור את הסיבוב הנוכחי.
  3. המטרה היא להגיע לסכום הקלפים שלך כמה שיותר קרוב ל-21 בסוף הסיבוב, מבלי לעבור.

סביבה שמייצגת את המשחק יכולה להיראות כך:

  1. פעולות: יש לנו 2 פעולות. פעולה 0: קבל קלף חדש, ופעולה 1: סיום הסיבוב הנוכחי.
  2. תצפיות: סכום הקלפים בסיבוב הנוכחי.
  3. תגמול: המטרה היא להגיע כמה שיותר קרוב ל-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)

בואו נוודא שעשינו הכל בצורה נכונה בהגדרת הסביבה הנ"ל. בעת יצירת סביבה משלך, עליך לוודא שהתצפיות ושלבי הזמן שנוצרו עוקבים אחר הצורות והסוגים הנכונים כפי שהוגדרו במפרטים שלך. אלה משמשים ליצירת גרף 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(
{'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

עטיפות סביבה

מעטפת סביבה לוקחת סביבת Python ומחזירה גרסה שונה של הסביבה. שניהם הסביבה המקורית ואיכות הסביבה השונה מקרים של py_environment.PyEnvironment , ועטיפות מרובות יכולות להיות כבול יחד.

ניתן למצוא כמה עטיפות נפוצות environments/wrappers.py . לדוגמה:

  1. ActionDiscretizeWrapper : המרת מרחב פעולה רציף למרחב פעולה דיסקרטי.
  2. RunStats : לוכד לרוץ סטטיסטיקה של הסביבה כגון מספר הצעדים שננקט, מספר פרק הושלם וכו '
  3. TimeLimit : מסיים את פרק לאחר מספר קבוע של צעדים.

דוגמה 1: פעולה דיסקרטיז עטיפה

InvertedPendulum היא סביבת PyBullet שמקבלת פעולות רציפות בטווח [-2, 2] . אם אנחנו רוצים להכשיר סוכן פעולה בדיד כמו DQN על סביבה זו, עלינו לבצע דיסקרטיות (לכמת) את מרחב הפעולה. זה בדיוק מה ActionDiscretizeWrapper עושה. השווה את action_spec לפני ואחרי גלישה:

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)

עטוף discrete_action_env הוא מופע של py_environment.PyEnvironment ויכול שיתייחס אל כמו סביבת פיתון רגילה.

סביבות TensorFlow

הממשק עבור סביבות TF מוגדר environments/tf_environment.TFEnvironment נראה מאוד דומה סביבות פייתון. סביבות TF שונות מ- Python env בכמה דרכים:

  • הם יוצרים אובייקטים טנסוריים במקום מערכים
  • סביבות TF מוסיפות מימד אצווה לטנזורים שנוצרו בהשוואה למפרט.

המרת סביבות Python ל-TFEnvs מאפשרת ל-tensorflow לבצע במקביל פעולות. לדוגמה, אפשר להגדיר collect_experience_op כי נתוני גובה הסביבה ומוסיפה על replay_buffer , וכן train_op שקוראת מן replay_buffer ומכשירה הסוכן, ולהפעיל אותם במקביל באופן טבעי 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.

אם action אינה תלויה הקודם time_step tf.control_dependency שדרוש Graph מצב.

לעת עתה, הבה נתבונן כיצד TFEnvironments נוצרים.

יצירת סביבת TensorFlow משלך

זה יותר מסובך מיצירת סביבות ב-Python, אז לא נעסוק בזה בקולאב הזה. דוגמה זמינה כאן . השימוש הנפוץ יותר הוא ליישם את הסביבה שלך Python ולעטוף אותו TensorFlow באמצעות שלנו TFPyEnvironment מעטפת (ראה להלן).

עטיפת סביבת פייתון ב-TensorFlow

אנחנו יכולים לעטוף בכל סביבה Python לסביבה TensorFlow באמצעות בקלות TFPyEnvironment מעטפת.

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

הערה המשקפיים נמצאים כעת מסוג: (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(
{'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.]

פרקים שלמים

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