นโยบาย

ลิขสิทธิ์ 2021 The TF-Agents Authors.

บทนำ

ในคำศัพท์เกี่ยวกับการเรียนรู้การเสริมแรง นโยบายจะจับคู่การสังเกตจากสภาพแวดล้อมไปยังการกระทำหรือการแจกแจงการกระทำ ในตัวแทน TF สังเกตจากสภาพแวดล้อมที่มีอยู่ในชื่อ tuple TimeStep('step_type', 'discount', 'reward', 'observation') และนโยบาย map timesteps การกระทำหรือการกระจายการกระทำ นโยบายส่วนใหญ่ใช้ timestep.observation นโยบายบางคนใช้ timestep.step_type (เช่นการรีเซ็ตรัฐที่จุดเริ่มต้นของเรื่องราวในนโยบาย stateful นั้น) แต่ timestep.discount และ timestep.reward มักจะถูกละเลย

นโยบายเกี่ยวข้องกับส่วนประกอบอื่นๆ ใน TF-Agents ดังต่อไปนี้ นโยบายส่วนใหญ่มีโครงข่ายประสาทเทียมในการคำนวณการกระทำและ/หรือการกระจายการกระทำจาก TimeSteps เอเจนต์สามารถมีนโยบายตั้งแต่หนึ่งรายการขึ้นไปเพื่อวัตถุประสงค์ที่แตกต่างกัน เช่น นโยบายหลักที่ได้รับการฝึกอบรมสำหรับการปรับใช้ และนโยบายที่ส่งเสียงดังสำหรับการรวบรวมข้อมูล สามารถบันทึก/กู้คืนนโยบายได้ และสามารถใช้ตัวแทนในการรวบรวมข้อมูล ประเมินผล ฯลฯ ได้โดยอิสระ

นโยบายบางอย่างเขียนได้ง่ายกว่าใน Tensorflow (เช่น นโยบายที่มีโครงข่ายประสาทเทียม) ในขณะที่บางนโยบายจะเขียนใน Python ได้ง่ายกว่า (เช่น ทำตามสคริปต์ของการกระทำ) ดังนั้นในตัวแทน TF เราอนุญาตทั้งนโยบาย Python และ Tensorflow ยิ่งไปกว่านั้น นโยบายที่เขียนด้วย TensorFlow อาจต้องใช้ในสภาพแวดล้อม Python หรือในทางกลับกัน เช่น นโยบาย TensorFlow ใช้สำหรับการฝึกอบรม แต่จะปรับใช้ในภายหลังในสภาพแวดล้อม Python ที่ใช้งานจริง เพื่อให้ง่ายขึ้น เราได้จัดเตรียม wrapper สำหรับการแปลงระหว่างนโยบาย Python และ TensorFlow

นโยบายที่น่าสนใจอีกประเภทหนึ่ง ได้แก่ ตัวห่อหุ้มนโยบาย ซึ่งปรับเปลี่ยนนโยบายที่กำหนดในลักษณะใดรูปแบบหนึ่ง เช่น เพิ่มประเภทของสัญญาณรบกวน สร้างนโยบายสุ่มรุ่นที่มีความโลภหรือมีความโลภ สุ่มผสมหลายนโยบาย เป็นต้น

ติดตั้ง

หากคุณยังไม่ได้ติดตั้ง tf-agents ให้รัน:

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 tensorflow_probability as tfp
import numpy as np

from tf_agents.specs import array_spec
from tf_agents.specs import tensor_spec
from tf_agents.networks import network

from tf_agents.policies import py_policy
from tf_agents.policies import random_py_policy
from tf_agents.policies import scripted_py_policy

from tf_agents.policies import tf_policy
from tf_agents.policies import random_tf_policy
from tf_agents.policies import actor_policy
from tf_agents.policies import q_policy
from tf_agents.policies import greedy_policy

from tf_agents.trajectories import time_step as ts

นโยบายหลาม

อินเตอร์เฟซสำหรับนโยบายงูใหญ่ถูกกำหนดไว้ใน policies/py_policy.PyPolicy วิธีการหลักคือ:

class Base(object):

  @abc.abstractmethod
  def __init__(self, time_step_spec, action_spec, policy_state_spec=()):
    self._time_step_spec = time_step_spec
    self._action_spec = action_spec
    self._policy_state_spec = policy_state_spec

  @abc.abstractmethod
  def reset(self, policy_state=()):
    # return initial_policy_state.
    pass

  @abc.abstractmethod
  def action(self, time_step, policy_state=()):
    # return a PolicyStep(action, state, info) named tuple.
    pass

  @abc.abstractmethod
  def distribution(self, time_step, policy_state=()):
    # Not implemented in python, only for TF policies.
    pass

  @abc.abstractmethod
  def update(self, policy):
    # update self to be similar to the input `policy`.
    pass

  @property
  def time_step_spec(self):
    return self._time_step_spec

  @property
  def action_spec(self):
    return self._action_spec

  @property
  def policy_state_spec(self):
    return self._policy_state_spec

วิธีการที่สำคัญที่สุดคือ action(time_step) ซึ่งแม time_step มีข้อสังเกตจากสภาพแวดล้อมที่จะ PolicyStep ชื่อ tuple ที่มีคุณสมบัติดังต่อไปนี้:

• action : การกระทำที่จะนำไปใช้กับสภาพแวดล้อม
• state : รัฐของนโยบาย (เช่น RNN รัฐ) ที่จะป้อนเข้าโทรไปเพื่อดำเนินการ
• info : ข้อมูลด้านตัวเลือกเช่นความน่าจะเป็นบันทึกการกระทำ

time_step_spec และ action_spec มีข้อกำหนดสำหรับขั้นตอนใส่เวลาและการกระทำที่ส่งออก นโยบายยังมี reset ฟังก์ชั่นซึ่งมักจะถูกนำมาใช้ในการรีเซ็ตรัฐในนโยบาย stateful update(new_policy) ฟังก์ชั่นการปรับปรุง self ต่อ new_policy

ตอนนี้ ให้เราดูตัวอย่างสองสามนโยบายของ Python

ตัวอย่างที่ 1: นโยบายหลามสุ่ม

ตัวอย่างง่ายๆของ PyPolicy เป็น RandomPyPolicy ซึ่งจะสร้างการกระทำที่สุ่มสำหรับที่ไม่ต่อเนื่อง / ต่อเนื่อง action_spec รับ อินพุต time_step จะถูกละเว้น

action_spec = array_spec.BoundedArraySpec((2,), np.int32, -10, 10)
my_random_py_policy = random_py_policy.RandomPyPolicy(time_step_spec=None,
    action_spec=action_spec)
time_step = None
action_step = my_random_py_policy.action(time_step)
print(action_step)
action_step = my_random_py_policy.action(time_step)
print(action_step)

PolicyStep(action=array([10, -4], dtype=int32), state=(), info=())
PolicyStep(action=array([7, 6], dtype=int32), state=(), info=())

ตัวอย่างที่ 2: สคริปต์ Python Policy

เล่นนโยบายสคริปต์สำรองสคริปต์ของการกระทำแสดงเป็นรายการ (num_repeats, action) tuples เวลาทุก action ฟังก์ชั่นที่เรียกว่ามันส่งกลับดำเนินการต่อไปจากรายการจนถึงจำนวนที่ระบุจะทำซ้ำและจากนั้นย้ายไปดำเนินการต่อไปในรายการ reset วิธีสามารถเรียกได้ว่าการดำเนินการที่จะเริ่มต้นจากจุดเริ่มต้นของรายการ

action_spec = array_spec.BoundedArraySpec((2,), np.int32, -10, 10)
action_script = [(1, np.array([5, 2], dtype=np.int32)), 
                 (0, np.array([0, 0], dtype=np.int32)), # Setting `num_repeats` to 0 will skip this action.
                 (2, np.array([1, 2], dtype=np.int32)), 
                 (1, np.array([3, 4], dtype=np.int32))]

my_scripted_py_policy = scripted_py_policy.ScriptedPyPolicy(
    time_step_spec=None, action_spec=action_spec, action_script=action_script)

policy_state = my_scripted_py_policy.get_initial_state()
time_step = None
print('Executing scripted policy...')
action_step = my_scripted_py_policy.action(time_step, policy_state)
print(action_step)
action_step= my_scripted_py_policy.action(time_step, action_step.state)
print(action_step)
action_step = my_scripted_py_policy.action(time_step, action_step.state)
print(action_step)

print('Resetting my_scripted_py_policy...')
policy_state = my_scripted_py_policy.get_initial_state()
action_step = my_scripted_py_policy.action(time_step, policy_state)
print(action_step)

Executing scripted policy...
PolicyStep(action=array([5, 2], dtype=int32), state=[0, 1], info=())
PolicyStep(action=array([1, 2], dtype=int32), state=[2, 1], info=())
PolicyStep(action=array([1, 2], dtype=int32), state=[2, 2], info=())
Resetting my_scripted_py_policy...
PolicyStep(action=array([5, 2], dtype=int32), state=[0, 1], info=())

นโยบาย TensorFlow

นโยบาย TensorFlow ใช้อินเทอร์เฟซเดียวกับนโยบาย Python ลองมาดูตัวอย่างกัน

ตัวอย่างที่ 1: นโยบาย TF แบบสุ่ม

RandomTFPolicy สามารถนำมาใช้ในการสร้างการกระทำที่สุ่มตามที่ได้รับไม่ต่อเนื่อง / ต่อเนื่อง action_spec อินพุต time_step จะถูกละเว้น

action_spec = tensor_spec.BoundedTensorSpec(
    (2,), tf.float32, minimum=-1, maximum=3)
input_tensor_spec = tensor_spec.TensorSpec((2,), tf.float32)
time_step_spec = ts.time_step_spec(input_tensor_spec)

my_random_tf_policy = random_tf_policy.RandomTFPolicy(
    action_spec=action_spec, time_step_spec=time_step_spec)
observation = tf.ones(time_step_spec.observation.shape)
time_step = ts.restart(observation)
action_step = my_random_tf_policy.action(time_step)

print('Action:')
print(action_step.action)

Action:
tf.Tensor([-0.9448042  1.9039011], shape=(2,), dtype=float32)

ตัวอย่างที่ 2: นโยบายนักแสดง

นโยบายนักแสดงที่สามารถสร้างขึ้นโดยใช้เครือข่ายที่แม time_steps การกระทำหรือเครือข่ายที่แมป time_steps เพื่อกระจายการกระทำ

การใช้เครือข่ายการดำเนินการ

ให้เรากำหนดเครือข่ายดังนี้:

class ActionNet(network.Network):

  def __init__(self, input_tensor_spec, output_tensor_spec):
    super(ActionNet, self).__init__(
        input_tensor_spec=input_tensor_spec,
        state_spec=(),
        name='ActionNet')
    self._output_tensor_spec = output_tensor_spec
    self._sub_layers = [
        tf.keras.layers.Dense(
            action_spec.shape.num_elements(), activation=tf.nn.tanh),
    ]

  def call(self, observations, step_type, network_state):
    del step_type

    output = tf.cast(observations, dtype=tf.float32)
    for layer in self._sub_layers:
      output = layer(output)
    actions = tf.reshape(output, [-1] + self._output_tensor_spec.shape.as_list())

    # Scale and shift actions to the correct range if necessary.
    return actions, network_state

ใน TensorFlow เลเยอร์เครือข่ายส่วนใหญ่ได้รับการออกแบบสำหรับการดำเนินการแบบแบตช์ ดังนั้นเราจึงคาดว่าอินพุต time_steps จะถูกจัดแบทช์ และเอาต์พุตของเครือข่ายก็จะถูกแบทช์ด้วยเช่นกัน นอกจากนี้ เครือข่ายมีหน้าที่รับผิดชอบในการสร้างการดำเนินการในช่วงที่ถูกต้องของ action_spec ที่กำหนด นี้จะกระทำโดยใช้อัตภาพเช่น tanh เปิดใช้งานสำหรับชั้นสุดท้ายที่จะดำเนินการผลิตใน [-1, 1] และจากนั้นปรับและขยับนี้ในช่วงที่ถูกต้องตาม action_spec การป้อนข้อมูล (เช่นดู tf_agents/agents/ddpg/networks.actor_network() )

ตอนนี้ เราสามารถสร้างนโยบายนักแสดงโดยใช้เครือข่ายข้างต้นได้

input_tensor_spec = tensor_spec.TensorSpec((4,), tf.float32)
time_step_spec = ts.time_step_spec(input_tensor_spec)
action_spec = tensor_spec.BoundedTensorSpec((3,),
                                            tf.float32,
                                            minimum=-1,
                                            maximum=1)

action_net = ActionNet(input_tensor_spec, action_spec)

my_actor_policy = actor_policy.ActorPolicy(
    time_step_spec=time_step_spec,
    action_spec=action_spec,
    actor_network=action_net)

เราสามารถนำไปใช้กับชุดของ time_steps ใดๆ ที่เป็นไปตาม time_step_spec:

batch_size = 2
observations = tf.ones([2] + time_step_spec.observation.shape.as_list())

time_step = ts.restart(observations, batch_size)

action_step = my_actor_policy.action(time_step)
print('Action:')
print(action_step.action)

distribution_step = my_actor_policy.distribution(time_step)
print('Action distribution:')
print(distribution_step.action)

Action:
tf.Tensor(
[[0.9318627 0.7770741 0.8645338]
 [0.9318627 0.7770741 0.8645338]], shape=(2, 3), dtype=float32)
Action distribution:
tfp.distributions.Deterministic("Deterministic", batch_shape=[2, 3], event_shape=[], dtype=float32)

ในตัวอย่างข้างต้น เราสร้างนโยบายโดยใช้เครือข่ายการดำเนินการที่สร้างเมตริกซ์การดำเนินการ ในกรณีนี้ policy.distribution(time_step) เป็นกำหนด (Delta) กระจายรอบการส่งออกของ policy.action(time_step) วิธีหนึ่งในการสร้างนโยบายสุ่มคือการห่อนโยบายของนักแสดงไว้ในเสื้อคลุมนโยบายที่เพิ่มเสียงรบกวนให้กับการกระทำ อีกวิธีหนึ่งคือการสร้างนโยบายนักแสดงโดยใช้เครือข่ายการกระจายการดำเนินการแทนเครือข่ายการดำเนินการดังที่แสดงด้านล่าง

การใช้เครือข่ายการกระจายการกระทำ

class ActionDistributionNet(ActionNet):

  def call(self, observations, step_type, network_state):
    action_means, network_state = super(ActionDistributionNet, self).call(
        observations, step_type, network_state)

    action_std = tf.ones_like(action_means)
    return tfp.distributions.MultivariateNormalDiag(action_means, action_std), network_state


action_distribution_net = ActionDistributionNet(input_tensor_spec, action_spec)

my_actor_policy = actor_policy.ActorPolicy(
    time_step_spec=time_step_spec,
    action_spec=action_spec,
    actor_network=action_distribution_net)

action_step = my_actor_policy.action(time_step)
print('Action:')
print(action_step.action)
distribution_step = my_actor_policy.distribution(time_step)
print('Action distribution:')
print(distribution_step.action)

Action:
tf.Tensor(
[[ 0.96731853  1.          1.        ]
 [ 0.94488937 -0.29294527  1.        ]], shape=(2, 3), dtype=float32)
Action distribution:
tfp.distributions.MultivariateNormalDiag("ActionNet_MultivariateNormalDiag", batch_shape=[2], event_shape=[3], dtype=float32)

โปรดทราบว่าในข้างต้น การดำเนินการจะถูกตัดไปยังช่วงของข้อกำหนดการดำเนินการที่กำหนด [-1, 1] นี่เป็นเพราะอาร์กิวเมนต์คอนสตรัคเตอร์ของ ActorPolicy clip=True โดยค่าเริ่มต้น การตั้งค่านี้เป็น "เท็จ" จะคืนค่าการดำเนินการที่ไม่ได้ตัดออกซึ่งสร้างโดยเครือข่าย

นโยบาย Stochastic สามารถแปลงเป็นนโยบายที่กำหนดโดยใช้ตัวอย่างเช่นเสื้อคลุม GreedyPolicy ซึ่งเลือก stochastic_policy.distribution().mode() ว่าการกระทำของตนและการกระจาย / เดลต้ากำหนดรอบการกระทำโลภนี้เป็นของ distribution()

ตัวอย่างที่ 3: นโยบาย Q

นโยบาย AQ ใช้ในเอเจนต์เช่น DQN และอิงตามเครือข่าย Q ที่คาดการณ์ค่า Q สำหรับแต่ละการดำเนินการที่ไม่ต่อเนื่อง สำหรับขั้นตอนเวลาที่กำหนด การกระจายการดำเนินการในนโยบาย Q เป็นการแจกแจงตามหมวดหมู่ที่สร้างขึ้นโดยใช้ค่า q เป็นบันทึก

input_tensor_spec = tensor_spec.TensorSpec((4,), tf.float32)
time_step_spec = ts.time_step_spec(input_tensor_spec)
action_spec = tensor_spec.BoundedTensorSpec((),
                                            tf.int32,
                                            minimum=0,
                                            maximum=2)
num_actions = action_spec.maximum - action_spec.minimum + 1


class QNetwork(network.Network):

  def __init__(self, input_tensor_spec, action_spec, num_actions=num_actions, name=None):
    super(QNetwork, self).__init__(
        input_tensor_spec=input_tensor_spec,
        state_spec=(),
        name=name)
    self._sub_layers = [
        tf.keras.layers.Dense(num_actions),
    ]

  def call(self, inputs, step_type=None, network_state=()):
    del step_type
    inputs = tf.cast(inputs, tf.float32)
    for layer in self._sub_layers:
      inputs = layer(inputs)
    return inputs, network_state


batch_size = 2
observation = tf.ones([batch_size] + time_step_spec.observation.shape.as_list())
time_steps = ts.restart(observation, batch_size=batch_size)

my_q_network = QNetwork(
    input_tensor_spec=input_tensor_spec,
    action_spec=action_spec)
my_q_policy = q_policy.QPolicy(
    time_step_spec, action_spec, q_network=my_q_network)
action_step = my_q_policy.action(time_steps)
distribution_step = my_q_policy.distribution(time_steps)

print('Action:')
print(action_step.action)

print('Action distribution:')
print(distribution_step.action)

Action:
tf.Tensor([2 2], shape=(2,), dtype=int32)
Action distribution:
tfp.distributions.Categorical("Categorical", batch_shape=[2], event_shape=[], dtype=int32)

กระดาษห่อนโยบาย

กระดาษห่อนโยบายสามารถใช้ห่อและแก้ไขนโยบายที่กำหนดได้ เช่น เพิ่มสัญญาณรบกวน Wrapper ของนโยบายเป็นคลาสย่อยของ Policy (Python/TensorFlow) ดังนั้นจึงสามารถใช้ได้เหมือนกับนโยบายอื่นๆ

ตัวอย่าง: Greed Policy

เสื้อคลุมโลภสามารถนำมาใช้ในการห่อนโยบาย TensorFlow ใด ๆ ที่ดำเนิน distribution() GreedyPolicy.action() จะกลับ wrapped_policy.distribution().mode() และ GreedyPolicy.distribution() คือการกระจาย / เดลต้ากำหนดรอบ GreedyPolicy.action() :

my_greedy_policy = greedy_policy.GreedyPolicy(my_q_policy)

action_step = my_greedy_policy.action(time_steps)
print('Action:')
print(action_step.action)

distribution_step = my_greedy_policy.distribution(time_steps)
print('Action distribution:')
print(distribution_step.action)

Action:
tf.Tensor([1 1], shape=(2,), dtype=int32)
Action distribution:
tfp.distributions.DeterministicWithLogProb("Deterministic", batch_shape=[2], event_shape=[], dtype=int32)