環境

Copyright 2023 TF-Agents 作者群。

在 TensorFlow.org 上檢視

在 Google Colab 中執行

在 GitHub 上檢視原始碼

下載筆記本

簡介

強化學習 (RL) 的目標是設計能透過與環境互動來學習的代理程式。在標準強化學習設定中，代理程式會在每個時間步收到觀察結果並選擇一個動作。該動作會套用至環境，而環境會傳回獎勵和新的觀察結果。代理程式會訓練策略來選擇動作，以最大化獎勵總和 (也稱為回報)。

在 TF-Agents 中，環境可以使用 Python 或 TensorFlow 實作。Python 環境通常較容易實作、理解和偵錯，但 TensorFlow 環境效率更高，並允許自然平行化。最常見的工作流程是在 Python 中實作環境，並使用我們的其中一個包裝函式自動將其轉換為 TensorFlow。

我們先來看看 Python 環境。TensorFlow 環境遵循非常相似的 API。

設定

如果您尚未安裝 tf-agents 或 gym，請執行

pip install tf-agents[reverb]
pip install tf-keras

import os
# Keep using keras-2 (tf-keras) rather than keras-3 (keras).
os.environ['TF_USE_LEGACY_KERAS'] = '1'

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

2023-12-22 12:20:01.730535: E external/local_xla/xla/stream_executor/cuda/cuda_dnn.cc:9261] Unable to register cuDNN factory: Attempting to register factory for plugin cuDNN when one has already been registered
2023-12-22 12:20:01.730578: E external/local_xla/xla/stream_executor/cuda/cuda_fft.cc:607] Unable to register cuFFT factory: Attempting to register factory for plugin cuFFT when one has already been registered
2023-12-22 12:20:01.732199: E external/local_xla/xla/stream_executor/cuda/cuda_blas.cc:1515] Unable to register cuBLAS factory: Attempting to register factory for plugin cuBLAS when one has already been registered

Python 環境

Python 環境具有 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) 中。

所有 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() 時自動重設。

請注意，子類別不會直接實作 step() 或 reset()。它們會改為覆寫 _step() 和 _reset() 方法。從這些方法傳回的時間步將會快取，並透過 current_time_step() 公開。

observation_spec 和 action_spec 方法會傳回 (Bounded)ArraySpecs 的巢狀結構，這些結構分別描述觀察和動作的名稱、形狀、資料類型和範圍。

在 TF-Agents 中，我們會重複提及巢狀結構，其定義為由清單、Tuple、具名 Tuple 或字典組成的任何樹狀結構。這些可以任意組合，以維持觀察和動作的結構。我們發現這對於更複雜的環境非常有用，在這些環境中，您有許多觀察和動作。

使用標準環境

TF Agents 針對許多標準環境 (例如 OpenAI Gym、DeepMind-control 和 Atari) 具有內建包裝函式，以便它們遵循我們的 py_environment.PyEnvironment 介面。這些包裝的環境可以使用我們的環境套件輕鬆載入。讓我們從 OpenAI gym 載入 CartPole 環境，並查看動作和 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，其中觀察結果是長度為 4 的 float32 向量，而折扣因子是 [0.0, 1.0] 中的 float32。現在，讓我們嘗試對整個回合採取固定的動作 (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.02944653,  0.04422915,  0.03086922, -0.04273267], dtype=float32)})
TimeStep(
{'step_type': array(1, dtype=int32),
 'reward': array(1., dtype=float32),
 'discount': array(1., dtype=float32),
 'observation': array([-0.02856195,  0.23889515,  0.03001456, -0.32551846], dtype=float32)})
TimeStep(
{'step_type': array(1, dtype=int32),
 'reward': array(1., dtype=float32),
 'discount': array(1., dtype=float32),
 'observation': array([-0.02378405,  0.43357718,  0.02350419, -0.608587  ], dtype=float32)})
TimeStep(
{'step_type': array(1, dtype=int32),
 'reward': array(1., dtype=float32),
 'discount': array(1., dtype=float32),
 'observation': array([-0.01511251,  0.6283628 ,  0.01133245, -0.89377517], dtype=float32)})
TimeStep(
{'step_type': array(1, dtype=int32),
 'reward': array(1., dtype=float32),
 'discount': array(1., dtype=float32),
 'observation': array([-0.00254525,  0.8233292 , -0.00654305, -1.1828743 ], dtype=float32)})
TimeStep(
{'step_type': array(1, dtype=int32),
 'reward': array(1., dtype=float32),
 'discount': array(1., dtype=float32),
 'observation': array([ 0.01392134,  1.0185355 , -0.03020054, -1.477601  ], dtype=float32)})
TimeStep(
{'step_type': array(1, dtype=int32),
 'reward': array(1., dtype=float32),
 'discount': array(1., dtype=float32),
 'observation': array([ 0.03429205,  1.214013  , -0.05975256, -1.7795612 ], dtype=float32)})
TimeStep(
{'step_type': array(1, dtype=int32),
 'reward': array(1., dtype=float32),
 'discount': array(1., dtype=float32),
 'observation': array([ 0.0585723 ,  1.4097542 , -0.09534378, -2.090206  ], dtype=float32)})
TimeStep(
{'step_type': array(1, dtype=int32),
 'reward': array(1., dtype=float32),
 'discount': array(1., dtype=float32),
 'observation': array([ 0.08676739,  1.6056995 , -0.1371479 , -2.4107776 ], dtype=float32)})
TimeStep(
{'step_type': array(1, dtype=int32),
 'reward': array(1., dtype=float32),
 'discount': array(1., dtype=float32),
 'observation': array([ 0.11888137,  1.8017205 , -0.18536346, -2.7422433 ], dtype=float32)})
TimeStep(
{'step_type': array(2, dtype=int32),
 'reward': array(1., dtype=float32),
 'discount': array(0., dtype=float32),
 'observation': array([ 0.1549158 ,  1.9976014 , -0.24020831, -3.0852165 ], dtype=float32)})

建立您自己的 Python 環境

對於許多客戶而言，常見的使用案例是將 TF-Agents 中的標準代理程式 (請參閱 agents/) 套用至他們的問題。若要執行此操作，他們必須將自己的問題構建為環境。因此，讓我們看看如何在 Python 中實作環境。

假設我們想要訓練代理程式玩以下 (Black Jack 啟發) 紙牌遊戲

1. 此遊戲使用編號為 1...10 的無限牌組進行。
2. 在每個回合中，代理程式可以執行 2 件事：取得新的隨機紙牌，或停止目前的回合。
3. 目標是在回合結束時讓您的紙牌總和盡可能接近 21，但不要超過。

代表此遊戲的環境可能如下所示

1. 動作：我們有 2 個動作。動作 0：取得新紙牌，以及動作 1：終止目前的回合。
2. 觀察結果：目前回合中紙牌的總和。
3. 獎勵：目標是盡可能接近 21 但不要超過，因此我們可以在回合結束時使用以下獎勵來達成此目標：sum_of_cards - 21 (如果 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([1], 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([4], dtype=int32)})
TimeStep(
{'step_type': array(2, dtype=int32),
 'reward': array(-17., dtype=float32),
 'discount': array(0., dtype=float32),
 'observation': array([4], dtype=int32)})
Final Reward =  -17.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 的執行個體，可以像一般 Python 環境一樣處理。

TensorFlow 環境

TF 環境的介面在 environments/tf_environment.TFEnvironment 中定義，看起來與 Python 環境非常相似。TF 環境與 Python 環境在幾個方面有所不同

• 它們產生張量物件而不是陣列
• 與規格相比，TF 環境會在產生的張量中新增批次維度。

將 Python 環境轉換為 TFEnv 可讓 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() 方法會傳回目前的時間步，並在需要時初始化環境。

reset() 方法會強制環境重設，並傳回 current_step。

如果 action 不依賴先前的 time_step，則在 Graph 模式中需要 tf.control_dependency。

現在，讓我們看看如何建立 TFEnvironments。

建立您自己的 TensorFlow 環境

這比在 Python 中建立環境更複雜，因此我們不會在本 Colab 中涵蓋它。範例位於此處。更常見的使用案例是在 Python 中實作您的環境，並使用我們的 TFPyEnvironment 包裝函式在 TensorFlow 中包裝它 (請參閱下文)。

在 TensorFlow 中包裝 Python 環境

我們可以使用 TFPyEnvironment 包裝函式輕鬆地將任何 Python 環境包裝到 TensorFlow 環境中。

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。

使用範例

簡單範例

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.00848238, -0.0419628 ,  0.02369678, -0.03962697]],
      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.00932164, -0.2374164 ,  0.02290425,  0.26043734]],
      dtype=float32)})]
[TimeStep(
{'step_type': array([1], dtype=int32),
 'reward': array([1.], dtype=float32),
 'discount': array([1.], dtype=float32),
 'observation': array([[-0.00932164, -0.2374164 ,  0.02290425,  0.26043734]],
      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.01406997, -0.04262878,  0.02811299, -0.02493422]],
      dtype=float32)})]
[TimeStep(
{'step_type': array([1], dtype=int32),
 'reward': array([1.], dtype=float32),
 'discount': array([1.], dtype=float32),
 'observation': array([[-0.01406997, -0.04262878,  0.02811299, -0.02493422]],
      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.01492254, -0.23814237,  0.02761431,  0.27648443]],
      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: 124
avg_length 24.8 avg_reward: 24.8