משחק CartPole בשיטת השחקן-מבקר

הצג באתר TensorFlow.org

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

צפה במקור ב-GitHub

הורד מחברת

מדריך זה מדגים כיצד ליישם את שיטת ה- Actor-Critic באמצעות TensorFlow כדי להכשיר סוכן בסביבת Open AI Gym CartPole-V0. ההנחה היא שלקורא יש היכרות מסוימת עם שיטות שיפוע מדיניות של למידת חיזוק.

שיטות שחקן-מבקר

שיטות השחקן-מבקר הן שיטות למידה של הבדל זמני (TD) המייצגות את פונקציית המדיניות ללא תלות בפונקציית הערך.

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

בשיטת ה-Actor-Critic, המדיניות מכונה השחקן המציע סט של פעולות אפשריות בהינתן מצב, ופונקציית הערך המוערך מכונה המבקר , אשר מעריכה את הפעולות שננקטו על ידי השחקן בהתבסס על המדיניות הנתונה. .

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

CartPole-v0

בסביבת CartPole-v0 , מוט מחובר לעגלה הנעה לאורך מסילה ללא חיכוך. המוט מתחיל זקוף ומטרת הסוכן היא למנוע ממנו ליפול על ידי הפעלת כוח של -1 או +1 על העגלה. פרס של 1+ ניתן על כל צעד שהמוט נשאר זקוף. פרק מסתיים כאשר (1) המוט נמצא במרחק של יותר מ-15 מעלות ממאונך או (2) העגלה זזה יותר מ-2.4 יחידות מהמרכז.

הבעיה נחשבת ל"פתורה" כאשר התגמול הכולל הממוצע עבור הפרק מגיע ל-195 על פני 100 ניסויים רצופים.

להכין

ייבא חבילות נחוצות והגדר הגדרות גלובליות.

pip install gym
pip install pyglet

# Install additional packages for visualization
sudo apt-get install -y xvfb python-opengl > /dev/null 2>&1
pip install pyvirtualdisplay > /dev/null 2>&1
pip install git+https://github.com/tensorflow/docs > /dev/null 2>&1

import collections
import gym
import numpy as np
import statistics
import tensorflow as tf
import tqdm

from matplotlib import pyplot as plt
from tensorflow.keras import layers
from typing import Any, List, Sequence, Tuple


# Create the environment
env = gym.make("CartPole-v0")

# Set seed for experiment reproducibility
seed = 42
env.seed(seed)
tf.random.set_seed(seed)
np.random.seed(seed)

# Small epsilon value for stabilizing division operations
eps = np.finfo(np.float32).eps.item()

דֶגֶם

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

במהלך המעבר קדימה, המודל ייקח את המצב כקלט ויוציא הן הסתברויות פעולה והן את ערך הביקורת \(V\), אשר מדגמן את פונקציית הערך התלויה במצב . המטרה היא להכשיר מודל שבוחר פעולות על סמך מדיניות \(\pi\) שממקסמת את התשואה הצפויה .

עבור Cartpole-v0, ישנם ארבעה ערכים המייצגים את המצב: מיקום העגלה, מהירות העגלה, זווית הקוטב ומהירות הקוטב בהתאמה. הסוכן יכול לבצע שתי פעולות כדי לדחוף את העגלה שמאלה (0) וימינה (1) בהתאמה.

עיין בדף הוויקי CartPole-v0 של OpenAI Gym למידע נוסף.

class ActorCritic(tf.keras.Model):
  """Combined actor-critic network."""

  def __init__(
      self, 
      num_actions: int, 
      num_hidden_units: int):
    """Initialize."""
    super().__init__()

    self.common = layers.Dense(num_hidden_units, activation="relu")
    self.actor = layers.Dense(num_actions)
    self.critic = layers.Dense(1)

  def call(self, inputs: tf.Tensor) -> Tuple[tf.Tensor, tf.Tensor]:
    x = self.common(inputs)
    return self.actor(x), self.critic(x)

num_actions = env.action_space.n  # 2
num_hidden_units = 128

model = ActorCritic(num_actions, num_hidden_units)

הַדְרָכָה

כדי להכשיר את הסוכן, תבצע את השלבים הבאים:

1. הפעל את הסוכן על הסביבה כדי לאסוף נתוני אימון לכל פרק.
2. חישוב התשואה הצפויה בכל שלב בזמן.
3. חשב את ההפסד עבור המודל המשולב של שחקן-מבקר.
4. חישוב מעברי צבע ועדכן פרמטרים של רשת.
5. חזור על 1-4 עד שהושג קריטריון הצלחה או מקסימום פרקים.

1. איסוף נתוני הכשרה

כמו בלמידה מפוקחת, כדי להכשיר את מודל השחקן-מבקר, אתה צריך לקבל נתוני אימון. עם זאת, על מנת לאסוף נתונים כאלה, המודל יצטרך להיות "להפעיל" בסביבה.

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

הפעולה הבאה תידגם מהסתברויות הפעולה שנוצרו על ידי המודל, שיושמו לאחר מכן על הסביבה, מה שיגרום ליצירת המצב והתגמול הבאים.

תהליך זה מיושם בפונקציית run_episode , המשתמשת בפעולות TensorFlow כך שניתן יהיה להרכיב אותו מאוחר יותר לתוך גרף TensorFlow לאימון מהיר יותר. שים לב ש- tf.TensorArray s שימשו לתמיכה באיטרציה של Tensor על מערכים באורך משתנה.

# Wrap OpenAI Gym's `env.step` call as an operation in a TensorFlow function.
# This would allow it to be included in a callable TensorFlow graph.

def env_step(action: np.ndarray) -> Tuple[np.ndarray, np.ndarray, np.ndarray]:
  """Returns state, reward and done flag given an action."""

  state, reward, done, _ = env.step(action)
  return (state.astype(np.float32), 
          np.array(reward, np.int32), 
          np.array(done, np.int32))


def tf_env_step(action: tf.Tensor) -> List[tf.Tensor]:
  return tf.numpy_function(env_step, [action], 
                           [tf.float32, tf.int32, tf.int32])

def run_episode(
    initial_state: tf.Tensor,  
    model: tf.keras.Model, 
    max_steps: int) -> Tuple[tf.Tensor, tf.Tensor, tf.Tensor]:
  """Runs a single episode to collect training data."""

  action_probs = tf.TensorArray(dtype=tf.float32, size=0, dynamic_size=True)
  values = tf.TensorArray(dtype=tf.float32, size=0, dynamic_size=True)
  rewards = tf.TensorArray(dtype=tf.int32, size=0, dynamic_size=True)

  initial_state_shape = initial_state.shape
  state = initial_state

  for t in tf.range(max_steps):
    # Convert state into a batched tensor (batch size = 1)
    state = tf.expand_dims(state, 0)

    # Run the model and to get action probabilities and critic value
    action_logits_t, value = model(state)

    # Sample next action from the action probability distribution
    action = tf.random.categorical(action_logits_t, 1)[0, 0]
    action_probs_t = tf.nn.softmax(action_logits_t)

    # Store critic values
    values = values.write(t, tf.squeeze(value))

    # Store log probability of the action chosen
    action_probs = action_probs.write(t, action_probs_t[0, action])

    # Apply action to the environment to get next state and reward
    state, reward, done = tf_env_step(action)
    state.set_shape(initial_state_shape)

    # Store reward
    rewards = rewards.write(t, reward)

    if tf.cast(done, tf.bool):
      break

  action_probs = action_probs.stack()
  values = values.stack()
  rewards = rewards.stack()

  return action_probs, values, rewards

2. חישוב התשואות הצפויות

רצף התגמולים עבור כל שלב זמן \(t\), \(\{r_{t}\}^{T}_{t=1}\) שנאסף במהלך פרק אחד מומר לרצף של החזרות צפויות \(\{G_{t}\}^{T}_{t=1}\) שבו סכום התגמולים נלקח משלב הזמן הנוכחי \(t\) ל- \(T\) וכל התגמול מוכפל עם גורם הנחה מתפוגג באופן אקספוננציאלי \(\gamma\):

\[G_{t} = \sum^{T}_{t'=t} \gamma^{t'-t}r_{t'}\]

מאז \(\gamma\in(0,1)\), לתגמולים רחוקים משלב הזמן הנוכחי ניתן משקל נמוך יותר.

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

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

def get_expected_return(
    rewards: tf.Tensor, 
    gamma: float, 
    standardize: bool = True) -> tf.Tensor:
  """Compute expected returns per timestep."""

  n = tf.shape(rewards)[0]
  returns = tf.TensorArray(dtype=tf.float32, size=n)

  # Start from the end of `rewards` and accumulate reward sums
  # into the `returns` array
  rewards = tf.cast(rewards[::-1], dtype=tf.float32)
  discounted_sum = tf.constant(0.0)
  discounted_sum_shape = discounted_sum.shape
  for i in tf.range(n):
    reward = rewards[i]
    discounted_sum = reward + gamma * discounted_sum
    discounted_sum.set_shape(discounted_sum_shape)
    returns = returns.write(i, discounted_sum)
  returns = returns.stack()[::-1]

  if standardize:
    returns = ((returns - tf.math.reduce_mean(returns)) / 
               (tf.math.reduce_std(returns) + eps))

  return returns

3. הפסד השחקן-מבקר

מכיוון שמשתמשים במודל היברידי של שחקן-מבקר, פונקציית ההפסד שנבחרה היא שילוב של הפסדי שחקן ומבקר לאימון, כפי שמוצג להלן:

\[L = L_{actor} + L_{critic}\]

אובדן שחקן

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

\[L_{actor} = -\sum^{T}_{t=1} \log\pi_{\theta}(a_{t} | s_{t})[G(s_{t}, a_{t}) - V^{\pi}_{\theta}(s_{t})]\]

איפה:

• \(T\): מספר שלבי הזמן לכל פרק, שיכול להשתנות לפי פרק
• \(s_{t}\): המדינה בשלב הזמן \(t\)
• \(a_{t}\): הפעולה שנבחרה בשלב הזמן \(t\) נתון למצב \(s\)
• \(\pi_{\theta}\): האם המדיניות (השחקן) מוגדרת על ידי \(\theta\)
• \(V^{\pi}_{\theta}\): האם פונקציית הערך (critic) מוגדרת גם על ידי \(\theta\)
• \(G = G_{t}\): התשואה הצפויה עבור מצב נתון, צמד פעולות בשלב הזמן \(t\)

מונח שלילי מתווסף לסכום מכיוון שהרעיון הוא למקסם את ההסתברויות לפעולות שיניבו תגמול גבוה יותר על ידי מזעור ההפסד המשולב.

יתרון

המונח \(L_{actor}\) בניסוח l10n- \(G - V\) שלנו נקרא היתרון , המציין עד כמה ניתנת לפעולה במצב מסוים על פני פעולה אקראית שנבחרה בהתאם למדיניות \(\pi\) עבור מצב זה.

אמנם ניתן לא לכלול קו בסיס, אך הדבר עלול לגרום לשונות גבוהה במהלך האימון. והדבר היפה בבחירת המבקר \(V\) כבסיס הוא שהוא הוכשר להיות קרוב ככל האפשר ל- \(G\), מה שמוביל לשונות נמוכה יותר.

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

לדוגמה, נניח ששתי פעולות עבור מדינה נתונה יניבו תשואה צפויה זהה. ללא המבקר, האלגוריתם ינסה להעלות את ההסתברות לפעולות אלו בהתבסס על האובייקטיבי \(J\). עם המבקר, עשוי להתברר שאין יתרון (\(G - V = 0\)) ולכן לא תועלת בהגדלת ההסתברויות של הפעולות והאלגוריתם יקבע את ההדרגות לאפס.

הפסד של מבקר

אימון \(V\) להיות קרוב ככל האפשר ל- \(G\) יכול להיות מוגדר כבעיית רגרסיה עם פונקציית האובדן הבאה:

\[L_{critic} = L_{\delta}(G, V^{\pi}_{\theta})\]

כאשר \(L_{\delta}\) הוא אובדן Huber , שהוא פחות רגיש לחריגים בנתונים מאשר אובדן שגיאה בריבוע.

huber_loss = tf.keras.losses.Huber(reduction=tf.keras.losses.Reduction.SUM)

def compute_loss(
    action_probs: tf.Tensor,  
    values: tf.Tensor,  
    returns: tf.Tensor) -> tf.Tensor:
  """Computes the combined actor-critic loss."""

  advantage = returns - values

  action_log_probs = tf.math.log(action_probs)
  actor_loss = -tf.math.reduce_sum(action_log_probs * advantage)

  critic_loss = huber_loss(values, returns)

  return actor_loss + critic_loss

4. הגדרת שלב האימון לעדכון פרמטרים

כל השלבים לעיל משולבים לשלב אימון המופעל בכל פרק. כל השלבים המובילים לפונקציית ההפסד מבוצעים עם ההקשר tf.GradientTape כדי לאפשר בידול אוטומטי.

מדריך זה משתמש בכלי האופטימיזציה של Adam כדי להחיל את ההדרגות על פרמטרי המודל.

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

ההקשר tf.function מיושם על הפונקציה train_step כך שניתן להרכיב אותה לתוך גרף TensorFlow הניתן להתקשרות, מה שיכול להוביל להאצה פי 10 באימון.

optimizer = tf.keras.optimizers.Adam(learning_rate=0.01)


@tf.function
def train_step(
    initial_state: tf.Tensor, 
    model: tf.keras.Model, 
    optimizer: tf.keras.optimizers.Optimizer, 
    gamma: float, 
    max_steps_per_episode: int) -> tf.Tensor:
  """Runs a model training step."""

  with tf.GradientTape() as tape:

    # Run the model for one episode to collect training data
    action_probs, values, rewards = run_episode(
        initial_state, model, max_steps_per_episode) 

    # Calculate expected returns
    returns = get_expected_return(rewards, gamma)

    # Convert training data to appropriate TF tensor shapes
    action_probs, values, returns = [
        tf.expand_dims(x, 1) for x in [action_probs, values, returns]] 

    # Calculating loss values to update our network
    loss = compute_loss(action_probs, values, returns)

  # Compute the gradients from the loss
  grads = tape.gradient(loss, model.trainable_variables)

  # Apply the gradients to the model's parameters
  optimizer.apply_gradients(zip(grads, model.trainable_variables))

  episode_reward = tf.math.reduce_sum(rewards)

  return episode_reward

5. הפעל את לולאת האימון

האימון מבוצע על ידי הפעלת שלב האימון עד שמגיעים לקריטריון ההצלחה או למספר המרבי של פרקים.

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

בהתאם לזמן הריצה שלך, האימון יכול להסתיים תוך פחות מדקה.

%%time

min_episodes_criterion = 100
max_episodes = 10000
max_steps_per_episode = 1000

# Cartpole-v0 is considered solved if average reward is >= 195 over 100 
# consecutive trials
reward_threshold = 195
running_reward = 0

# Discount factor for future rewards
gamma = 0.99

# Keep last episodes reward
episodes_reward: collections.deque = collections.deque(maxlen=min_episodes_criterion)

with tqdm.trange(max_episodes) as t:
  for i in t:
    initial_state = tf.constant(env.reset(), dtype=tf.float32)
    episode_reward = int(train_step(
        initial_state, model, optimizer, gamma, max_steps_per_episode))

    episodes_reward.append(episode_reward)
    running_reward = statistics.mean(episodes_reward)

    t.set_description(f'Episode {i}')
    t.set_postfix(
        episode_reward=episode_reward, running_reward=running_reward)

    # Show average episode reward every 10 episodes
    if i % 10 == 0:
      pass # print(f'Episode {i}: average reward: {avg_reward}')

    if running_reward > reward_threshold and i >= min_episodes_criterion:  
        break

print(f'\nSolved at episode {i}: average reward: {running_reward:.2f}!')

Episode 361:   4%|▎         | 361/10000 [01:13<32:33,  4.93it/s, episode_reward=182, running_reward=195]
Solved at episode 361: average reward: 195.14!
CPU times: user 2min 46s, sys: 35.4 s, total: 3min 21s
Wall time: 1min 13s

רְאִיָה

לאחר האימון, יהיה טוב לדמיין כיצד המודל מתפקד בסביבה. אתה יכול להפעיל את התאים למטה כדי ליצור אנימציית GIF של ריצת פרק אחד של המודל. שימו לב שצריך להתקין חבילות נוספות כדי ש-OpenAI Gym יציג את תמונות הסביבה בצורה נכונה ב-Colab.

# Render an episode and save as a GIF file

from IPython import display as ipythondisplay
from PIL import Image
from pyvirtualdisplay import Display


display = Display(visible=0, size=(400, 300))
display.start()


def render_episode(env: gym.Env, model: tf.keras.Model, max_steps: int): 
  screen = env.render(mode='rgb_array')
  im = Image.fromarray(screen)

  images = [im]

  state = tf.constant(env.reset(), dtype=tf.float32)
  for i in range(1, max_steps + 1):
    state = tf.expand_dims(state, 0)
    action_probs, _ = model(state)
    action = np.argmax(np.squeeze(action_probs))

    state, _, done, _ = env.step(action)
    state = tf.constant(state, dtype=tf.float32)

    # Render screen every 10 steps
    if i % 10 == 0:
      screen = env.render(mode='rgb_array')
      images.append(Image.fromarray(screen))

    if done:
      break

  return images


# Save GIF image
images = render_episode(env, model, max_steps_per_episode)
image_file = 'cartpole-v0.gif'
# loop=0: loop forever, duration=1: play each frame for 1ms
images[0].save(
    image_file, save_all=True, append_images=images[1:], loop=0, duration=1)

import tensorflow_docs.vis.embed as embed
embed.embed_file(image_file)

הצעדים הבאים

מדריך זה הדגים כיצד ליישם את שיטת השחקן-מבקר באמצעות Tensorflow.

כשלב הבא, תוכל לנסות לאמן מודל בסביבה אחרת ב-OpenAI Gym.

למידע נוסף לגבי שיטות שחקן-מבקר ובעיית Cartpole-v0, אתה יכול לעיין במשאבים הבאים:

• שיטת מבקר השחקן
• הרצאת ביקורת שחקנים (CAL)
• בעיית בקרת למידה של קרטון [Barto, et al. 1983]

לדוגמאות נוספות ללימוד חיזוק ב-TensorFlow, תוכל לבדוק את המשאבים הבאים:

• דוגמאות לקוד למידת חיזוק (keras.io)
• ספריית למידה לחיזוק TF-Agents