Tarihi kaydet! Google I / O 18-20 Mayıs'ta geri dönüyor Şimdi kaydolun
Bu sayfa, Cloud Translation API ile çevrilmiştir.
Switch to English

Aktör-Eleştirmen Yöntemi ile CartPole Oynamak

TensorFlow.org'da görüntüleyin Google Colab'da çalıştırın Kaynağı GitHub'da görüntüleyin Defteri indirin

Bu eğitici, Actor-Critic yönteminin bir temsilciyi Open AI Gym CartPole-V0 ortamında eğitmek için TensorFlow kullanarak nasıl uygulanacağını gösterir. Okuyucunun, pekiştirmeli öğrenmenin politika gradyan yöntemlerine aşina olduğu varsayılır.

Aktör-Eleştirmen yöntemleri

Aktör-Eleştirmen yöntemleri, değer işlevinden bağımsız politika işlevini temsil eden zamansal fark (TD) öğrenme yöntemleridir.

Bir ilke işlevi (veya ilke), temsilcinin verilen duruma göre gerçekleştirebileceği eylemler üzerinden bir olasılık dağılımı döndürür. Bir değer işlevi, belirli bir durumda başlayan ve sonsuza kadar belirli bir politikaya göre hareket eden bir temsilci için beklenen getiriyi belirler.

Aktör-Eleştirmen yönteminde, politika, bir durum verilen bir dizi olası eylemi öneren aktör olarak adlandırılır ve tahmini değer işlevi, aktör tarafından verilen politikaya göre alınan eylemleri değerlendiren eleştirmen olarak adlandırılır. .

Bu eğitimde, hem Aktör hem de Eleştirmen , iki çıkışlı bir sinir ağı kullanılarak temsil edilecek.

Sepet Kutbu-v0

CartPole-v0 ortamında , sürtünmesiz bir yol boyunca hareket eden bir arabaya bir direk takılır. Direk dik olarak başlar ve temsilcinin amacı, arabaya -1 veya +1 kuvvet uygulayarak düşmesini önlemektir. Direğin dik kaldığı her adım için +1 ödül verilir. Bir bölüm, (1) direk dikey olarak 15 dereceden fazla olduğunda veya (2) araba merkezden 2.4 birimden fazla hareket ettiğinde sona erer.

Cartpole-v0 ortamında eğitimli oyuncu-eleştirmen modeli

Bölüm için ortalama toplam ödül art arda 100 denemede 195'e ulaştığında sorun "çözülmüş" olarak kabul edilir.

Kurulum

Gerekli paketleri içe aktarın ve genel ayarları yapılandırın.

pip install -q gym
# Install additional packages for visualization
sudo apt-get install -y xvfb python-opengl > /dev/null 2>&1
pip install -q pyvirtualdisplay > /dev/null 2>&1
pip install -q 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()

Modeli

Aktör ve Eleştirmen , sırasıyla eylem olasılıklarını ve kritik değeri üreten bir sinir ağı kullanılarak modellenecek. Bu eğitici, modeli tanımlamak için model alt sınıflarını kullanır.

İleri geçiş sırasında, model durumu girdi olarak alacak ve duruma bağlı değer işlevini modelleyen hem eylem olasılıklarını hem de kritik değeri $ V $ çıkaracaktır. Amaç, beklenen getiriyi en üst düzeye çıkaran $ \ pi $ politikasına göre eylemler seçen bir model eğitmektir.

Cartpole-v0 için durumu temsil eden dört değer vardır: sırasıyla araba konumu, araba hızı, kutup açısı ve kutup hızı. Temsilci, sepeti sırasıyla sola (0) ve sağa (1) itmek için iki eylem gerçekleştirebilir.

Daha fazla bilgi için OpenAI Gym'in CartPole-v0 wiki sayfasına bakın.

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)

Eğitim

Temsilciyi eğitmek için şu adımları takip edeceksiniz:

  1. Bölüm başına eğitim verilerini toplamak için aracıyı ortamda çalıştırın.
  2. Her adımda beklenen getiriyi hesaplayın.
  3. Birleşik aktör-eleştirmen modeli için kaybı hesaplayın.
  4. Degradeleri hesaplayın ve ağ parametrelerini güncelleyin.
  5. Başarı kriterine veya maksimum bölüme ulaşılıncaya kadar 1-4'ü tekrarlayın.

1. Eğitim verilerinin toplanması

Denetimli öğrenmede olduğu gibi, aktör-eleştirmen modelini eğitmek için eğitim verilerine sahip olmanız gerekir. Ancak, bu tür verileri toplamak için, modelin ortamda "çalıştırılması" gerekir.

Her bölüm için eğitim verileri toplanır. Daha sonra her zaman adımında, modelin ağırlıkları ile parametrelendirilen mevcut politikaya dayalı olarak eylem olasılıklarını ve kritik değeri oluşturmak için modelin ileri geçişi ortamın durumunda çalıştırılacaktır.

Bir sonraki eylem, model tarafından oluşturulan eylem olasılıklarından örneklenecek ve daha sonra ortama uygulanacak ve bir sonraki durum ve ödülün üretilmesine neden olacak.

Bu işlem, daha sonra daha hızlı eğitim için bir TensorFlow grafiğinde derlenebilmesi için TensorFlow işlemlerini kullanan run_episode işlevinde uygulanır. tf.TensorArray s değişken uzunluklu dizilerde Tensor yinelemesini desteklemek için kullanıldı.

# 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. Beklenen getirilerin hesaplanması

Bir bölüm boyunca toplanan her zaman adımı $ t $, $ {r_ {t}} ^ {T} _ {t = 1} $ için ödül dizisi beklenen getiri dizisine dönüştürülür $ {G_ {t}} ^ { T} _ {t = 1} $, ödüllerin toplamının geçerli zaman adımı $ t $ ile $ T $ arasında alındığı ve her ödülün katlanarak azalan indirim faktörü $ \ gamma $ ile çarpıldığı:

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

$ \ Gamma \ in (0,1) $ olduğundan, geçerli zaman adımından daha uzaktaki ödüllere daha az ağırlık verilir.

Sezgisel olarak, beklenen getiri basitçe şu andaki ödüllerin sonraki ödüllerden daha iyi olduğu anlamına gelir. Matematiksel anlamda, ödüllerin toplamının yakınsamasını sağlamaktır.

Eğitimi stabilize etmek için, ortaya çıkan geri dönüş sırası da standartlaştırılır (yani sıfır ortalamaya ve birim standart sapmaya sahip olmak).

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. Oyuncu-eleştirmen kaybı

Karma bir aktör-eleştirmen modeli kullanıldığından, seçilen kayıp işlevi, aşağıda gösterildiği gibi eğitim için aktör ve kritik kayıpların bir kombinasyonudur:

$$L = L_{actor} + L_{critic}$$

Oyuncu kaybı

Aktör kaybı, duruma bağlı bir taban çizgisi olarak eleştirmenle politika gradyanlarına dayanır ve tek örnekli (bölüm başına) tahminlerle hesaplanır.

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

nerede:

  • $ T $: bölüm başına değişebilen bölüm başına zaman adımı sayısı
  • $ s_ {t} $: $ t $ zaman adımındaki durum
  • $ a_ {t} $: $ t $ zaman adımında seçilen eylem $ s $ durumunda
  • $ \ pi _ {\ theta} $: $ \ theta $ ile parametrelendirilen politikadır (aktör)
  • $ V ^ {\ pi} _ {\ theta} $: ayrıca $ \ theta $ tarafından parametrelendirilen değer işlevi (kritik)
  • $ G = G_ {t} $: belirli bir durum için beklenen getiri, $ t $ zaman adımındaki eylem çifti

Ana fikir, birleşik kaybı en aza indirerek daha yüksek ödüller sağlayan eylemlerin olasılıklarını en üst düzeye çıkarmak olduğundan, toplama negatif bir terim eklenir.


Avantajı

$ L_ {actor} $ formülasyonumuzdaki $ G - V $ terimine avantaj denir ve bu durum, bir eyleme belirli bir durumun o durum için $ \ pi $ politikasına göre seçilen rastgele bir eyleme göre ne kadar iyi verildiğini gösterir.

Bir temeli hariç tutmak mümkün olsa da, bu eğitim sırasında yüksek varyansa neden olabilir. Ve temel olarak $ V $ eleştirmenini seçmenin güzel yanı, $ G $ 'a olabildiğince yakın olacak şekilde eğitilmesi ve bu da daha düşük bir varyansa yol açmasıdır.

Ek olarak, eleştirmen olmadan, algoritma beklenen getiriye dayalı olarak belirli bir durumda gerçekleştirilen eylemlerin olasılıklarını artırmaya çalışacaktır; bu, eylemler arasındaki göreli olasılıklar aynı kalırsa pek bir fark yaratmayabilir.

Örneğin, belirli bir durum için iki eylemin aynı beklenen getiriyi sağlayacağını varsayalım. Eleştirmen olmasaydı, algoritma bu eylemlerin olasılığını $ J $ hedefine göre artırmaya çalışırdı. Eleştirmenle, herhangi bir avantajın olmadığı ($ G - V = 0 $) ve dolayısıyla eylemlerin olasılıklarının artırılmasında hiçbir fayda sağlanamadığı ve algoritmanın gradyanları sıfıra ayarlayacağı ortaya çıkabilir.


Kritik kayıp

$ V $ 'ın $ G $' a olabildiğince yakın olacak şekilde eğitilmesi, aşağıdaki kayıp fonksiyonu ile bir gerileme problemi olarak ayarlanabilir:

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

$ L _ {\ delta} $, verilerdeki aykırı değerlere karesel hata kaybından daha az duyarlı olan Huber kaybıdır.

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. Parametreleri güncellemek için eğitim adımını tanımlama

Yukarıdaki adımların tümü, her bölümde çalıştırılan bir eğitim adımında birleştirilir. Otomatik farklılaştırmayı etkinleştirmek için kayıp işlevine giden tüm adımlartf.GradientTape bağlamıyla yürütülür.

Bu eğitici, degradeleri model parametrelerine uygulamak için Adam iyileştiricisini kullanır.

episode_reward ödüllerin toplamı olan episode_reward da bu adımda hesaplanır. Bu değer daha sonra başarı kriterinin karşılanıp karşılanmadığını değerlendirmek için kullanılacaktır.

tf.function bağlamı train_step işlevine uygulanır, böylece çağrılabilir bir TensorFlow grafiğinde derlenebilir ve bu da eğitimde 10 kat hızlanmaya yol açabilir.

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. Eğitim döngüsünü çalıştırın

Eğitim, başarı kriterine veya maksimum bölüm sayısına ulaşılana kadar eğitim adımı çalıştırılarak yürütülür.

Bölüm ödüllerinin sürekli kaydı bir kuyrukta tutulur. 100 denemeye ulaşıldığında, en eski ödül kuyruğun sol (kuyruk) ucundan kaldırılır ve en yenisi başa (sağ) eklenir. Hesaplama verimliliği için ödüllerin sürekli bir toplamı da korunur.

Çalışma sürenize bağlı olarak eğitim bir dakikadan daha kısa sürede tamamlanabilir.

%%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 722:   7%|▋         | 722/10000 [03:41<47:21,  3.26it/s, episode_reward=200, running_reward=195]
Solved at episode 722: average reward: 195.14!
CPU times: user 11min 21s, sys: 3min 8s, total: 14min 30s
Wall time: 3min 41s

Görselleştirme

Eğitimden sonra, modelin ortamda nasıl performans gösterdiğini görselleştirmek iyi olacaktır. Modelin bir bölüm çalışmasının bir GIF animasyonunu oluşturmak için aşağıdaki hücreleri çalıştırabilirsiniz. Ortamın görüntülerini Colab'de doğru şekilde işlemek için OpenAI Gym için ek paketlerin yüklenmesi gerektiğini unutmayın.

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

gif

Sonraki adımlar

Bu eğitim, aktör-eleştirmen yönteminin Tensorflow kullanılarak nasıl uygulanacağını gösterdi.

Bir sonraki adım olarak, bir modeli OpenAI Gym'de farklı bir ortamda eğitmeyi deneyebilirsiniz.

Aktör-eleştirmen yöntemleri ve Cartpole-v0 sorunu hakkında ek bilgi için aşağıdaki kaynaklara başvurabilirsiniz:

TensorFlow'da daha fazla pekiştirici öğrenme örnekleri için aşağıdaki kaynakları kontrol edebilirsiniz: