סביבות

זכויות יוצרים 2021 מחברי TF-Agents.

צפה ב- TensorFlow.org

הפעל בגוגל קולאב

צפה במקור ב- GitHub

הורד מחברת

מבוא

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

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

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

להכין

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

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

סביבות פיתון

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

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

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

שים לב כי מחלקות המשנה אינן מיישמות את step() או reset() ישירות. במקום זאת הם עוקפים את _step() ו- _reset() . שלבי הזמן שיוחזרו משיטות אלה יישמרו במטמון current_time_step() דרך current_time_step() .

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

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

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

ל- TF Agents יש מעטפות מובנות לסביבות סטנדרטיות רבות כמו OpenAI Gym, DeepMind-control ו- Atari, כך שהן עוקבות אחר ממשק 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(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))

יצירת סביבת פייתון משלך

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

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

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

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

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

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

עטיפות סביבתיות

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

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

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

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

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

סביבות TensorFlow

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

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

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

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

לעת עתה, הבה נבחן כיצד TFEnvironments .

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

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

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

אנו יכולים לעטוף בקלות כל סביבת פיתון לסביבת 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(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 עכשיו הוא מהסוג: (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