Как сделать входные данные и модель одинаковой формы (обучение с подкреплением RLlib Ray Sagemaker)

Возможная причина:

Сообщение об ошибке:

ValueError: Input 0 of layer default/fc1 is incompatible with the layer: : expected min_ndim=2, found ndim=1. Full shape received: [None]

Ваше исходное пространство для наблюдения за окружающей средой - self.observation_space = Box(np.array(0.0),np.array(1000)).

Отображение формы вашего окружения в пространстве дает:

print(Box(np.array(0.0), np.array(1000), dtype=np.float32).shape) = ()

На это может указывать Full shape received: [None] в сообщении об ошибке.

Если вы передадите фигуру (1,1) в np.zeros, вы получите ожидаемый min_ndim=2:

x = np.zeros((1, 1)) print(x) [[0.]] print(x.ndim) 2

Предлагаемое решение:

Я предполагаю, что вы хотите, чтобы пространство obs вашего окружения находилось в диапазоне от 0,0 до 1000,0, как указано self.price = np.random.rand() в вашей reset функции.

Попробуйте использовать следующее для вашего окружения obs space:

self.observation_space = Box(0.0, 1000.0, shape=(1,1), dtype=np.float32)

Я надеюсь, что установка Box с явным shape поможет.

ИЗМЕНИТЬ (20190903):

# new imports
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function

import numpy as np
from ray.rllib.models import ModelCatalog
from ray.rllib.models.tf.tf_modelv2 import TFModelV2
from ray.rllib.models.tf.fcnet_v2 import FullyConnectedNetwork

from ray.rllib.utils import try_import_tf
from ray.tune import grid_search

tf = try_import_tf()
# end new imports

# Custom model class (fcnet)
class CustomModel(TFModelV2):
    """Example of a custom model that just delegates to a fc-net."""

    def __init__(self, obs_space, action_space, num_outputs, model_config,
                 name):
        super(CustomModel, self).__init__(obs_space, action_space, num_outputs,
                                          model_config, name)
        self.model = FullyConnectedNetwork(obs_space, action_space,
                                           num_outputs, model_config, name)
        self.register_variables(self.model.variables())

    def forward(self, input_dict, state, seq_lens):
        return self.model.forward(input_dict, state, seq_lens)

    def value_function(self):
        return self.model.value_function()

    def get_experiment_config(self):


        # Register custom model
        ModelCatalog.register_custom_model("my_model", CustomModel)


        return {
          "training": {
            "env": "ArrivalSim-v0",
            "run": "PPO",
            "stop": {
              "episode_reward_mean": 5000,
            },
            "config": {


              "model": {"custom_model": "my_model"}, # Add registered custom model


              "gamma": 0.995,
              "kl_coeff": 1.0,
              "num_sgd_iter": 10,
              "lr": 0.0001,
              "sgd_minibatch_size": 32768,
              "train_batch_size": 320000,
              "monitor": False,  # Record videos.
              "model": {
                "free_log_std": False
              },
              "use_gae": False,
              "num_workers": (self.num_cpus-1),
              "num_gpus": self.num_gpus,
              "batch_mode": "complete_episodes"
            }
          }
        }

РЕДАКТИРОВАТЬ 2 (20190910):

Чтобы показать, что это работает, усеченный вывод Sagemaker (экземпляр записной книжки Jupyter):

.
.
.
algo-1-y2ayw_1  | price b = 0.439261780930142
algo-1-y2ayw_1  | price a = 0.439261780930142
algo-1-y2ayw_1  | (self.price).shape = (1,)
algo-1-y2ayw_1  | [0.43926178] 10.103020961393266 False {}
algo-1-y2ayw_1  | price b = 0.439261780930142
algo-1-y2ayw_1  | price a = 0.439261780930142
algo-1-y2ayw_1  | (self.price).shape = (1,)
algo-1-y2ayw_1  | [0.43926178] 9.663759180463124 False {}
algo-1-y2ayw_1  | price b = 0.439261780930142
algo-1-y2ayw_1  | price a = 0.189261780930142
algo-1-y2ayw_1  | (self.price).shape = (1,)
algo-1-y2ayw_1  | [0.18926178] 5.67785342790426 False {}
algo-1-y2ayw_1  | price b = 0.189261780930142
algo-1-y2ayw_1  | price a = -0.06073821906985799
algo-1-y2ayw_1  | (self.price).shape = (1,)
algo-1-y2ayw_1  | [-0.06073822] 0 True {}
algo-1-y2ayw_1  | Result for PPO_ArrivalSim-v0_0:
algo-1-y2ayw_1  |   date: 2019-09-10_11-51-13
algo-1-y2ayw_1  |   done: true
algo-1-y2ayw_1  |   episode_len_mean: 126.72727272727273
algo-1-y2ayw_1  |   episode_reward_max: 15772.677709596366
algo-1-y2ayw_1  |   episode_reward_mean: 2964.4609668691965
algo-1-y2ayw_1  |   episode_reward_min: 0.0
algo-1-y2ayw_1  |   episodes: 5
algo-1-y2ayw_1  |   experiment_id: 5d3b9f2988854a0db164a2e5e9a7550f
algo-1-y2ayw_1  |   hostname: 2dae585dcc65
algo-1-y2ayw_1  |   info:
algo-1-y2ayw_1  |     cur_lr: 4.999999873689376e-05
algo-1-y2ayw_1  |     entropy: 1.0670874118804932
algo-1-y2ayw_1  |     grad_time_ms: 1195.066
algo-1-y2ayw_1  |     kl: 3.391784191131592
algo-1-y2ayw_1  |     load_time_ms: 44.725
algo-1-y2ayw_1  |     num_steps_sampled: 463
algo-1-y2ayw_1  |     num_steps_trained: 463
algo-1-y2ayw_1  |     policy_loss: -0.05383850634098053
algo-1-y2ayw_1  |     sample_time_ms: 621.282
algo-1-y2ayw_1  |     total_loss: 2194493.5
algo-1-y2ayw_1  |     update_time_ms: 145.352
algo-1-y2ayw_1  |     vf_explained_var: -5.519390106201172e-05
algo-1-y2ayw_1  |     vf_loss: 2194492.5
algo-1-y2ayw_1  |   iterations_since_restore: 2
algo-1-y2ayw_1  |   node_ip: 172.18.0.2
algo-1-y2ayw_1  |   pid: 77
algo-1-y2ayw_1  |   policy_reward_mean: {}
algo-1-y2ayw_1  |   time_since_restore: 4.55129861831665
algo-1-y2ayw_1  |   time_this_iter_s: 1.3484764099121094
algo-1-y2ayw_1  |   time_total_s: 4.55129861831665
algo-1-y2ayw_1  |   timestamp: 1568116273
algo-1-y2ayw_1  |   timesteps_since_restore: 463
algo-1-y2ayw_1  |   timesteps_this_iter: 234
algo-1-y2ayw_1  |   timesteps_total: 463
algo-1-y2ayw_1  |   training_iteration: 2
algo-1-y2ayw_1  |
algo-1-y2ayw_1  | A worker died or was killed while executing task 00000000781a7b5b94a203683f8f789e593abbb1.
algo-1-y2ayw_1  | A worker died or was killed while executing task 00000000d3507bc6b41ee1c9fc36292eeae69557.
algo-1-y2ayw_1  | == Status ==
algo-1-y2ayw_1  | Using FIFO scheduling algorithm.
algo-1-y2ayw_1  | Resources requested: 0/3 CPUs, 0/0 GPUs
algo-1-y2ayw_1  | Result logdir: /opt/ml/output/intermediate/training
algo-1-y2ayw_1  | TERMINATED trials:
algo-1-y2ayw_1  |  - PPO_ArrivalSim-v0_0:   TERMINATED [pid=77], 4 s, 2 iter, 463 ts, 2.96e+03 rew
algo-1-y2ayw_1  |
algo-1-y2ayw_1  | Saved model configuration.
algo-1-y2ayw_1  | Saved the checkpoint file /opt/ml/output/intermediate/training/PPO_ArrivalSim-v0_0_2019-09-10_11-50-53vd32vlux/checkpoint-2.extra_data as /opt/ml/model/checkpoint.extra_data
algo-1-y2ayw_1  | Saved the checkpoint file /opt/ml/output/intermediate/training/PPO_ArrivalSim-v0_0_2019-09-10_11-50-53vd32vlux/checkpoint-2.tune_metadata as /opt/ml/model/checkpoint.tune_metadata
algo-1-y2ayw_1  | Created LogSyncer for /root/ray_results/PPO_ArrivalSim-v0_2019-09-10_11-51-13xdn_5i34 -> None
algo-1-y2ayw_1  | 2019-09-10 11:51:13.941718: I tensorflow/core/common_runtime/process_util.cc:71] Creating new thread pool with default inter op setting: 2. Tune using inter_op_parallelism_threads for best performance.
algo-1-y2ayw_1  | reset -> (self.price).shape =  (1,)
algo-1-y2ayw_1  | LocalMultiGPUOptimizer devices ['/cpu:0']
algo-1-y2ayw_1  | reset -> (self.price).shape =  (1,)
algo-1-y2ayw_1  | INFO:tensorflow:No assets to save.
algo-1-y2ayw_1  | No assets to save.
algo-1-y2ayw_1  | INFO:tensorflow:No assets to write.
algo-1-y2ayw_1  | No assets to write.
algo-1-y2ayw_1  | INFO:tensorflow:SavedModel written to: /opt/ml/model/1/saved_model.pb
algo-1-y2ayw_1  | SavedModel written to: /opt/ml/model/1/saved_model.pb
algo-1-y2ayw_1  | Saved TensorFlow serving model!
algo-1-y2ayw_1  | A worker died or was killed while executing task 00000000f352d985b807ca399460941fe2264899.

algo-1-y2ayw_1  | 2019-09-10 11:51:20,075 sagemaker-containers INFO

 Reporting training SUCCESS

tmpwwb4b358_algo-1-y2ayw_1 exited with code 0

Aborting on container exit...
Failed to delete: /tmp/tmpwwb4b358/algo-1-y2ayw Please remove it manually.

===== Job Complete =====

На этот раз я редактирую все 3 файла. Ваша среда, сценарий обучения и записная книжка Jupyter, но оказывается, что нет необходимости определять пользовательские модели для вашей пользовательской среды. Однако это остается жизнеспособным. И вы правы, основная причина проблемы все еще находится в космосе.

Я установил self.price как одномерный массив numpy, чтобы он лучше взаимодействовал с Ray RLlib. Создание настраиваемой среды в обучающем сценарии было выполнено более простым способом, как показано ниже. Что касается ноутбука, я использовал версию 0.5.3 вместо 0.6.5 для toolkit_version, и обучение выполняется в локальном режиме (в контейнере докеров на экземпляре ноутбука Sagemaker Jupyter, все еще на AWS) только с ЦП. Однако он также будет работать с любым экземпляром ML (например, ml.m4.xlarge) с графическим процессором.

Весь пакет вместе со всеми зависимостями находится здесь.

Отредактированный env:

# new
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
# end new


from enum import Enum
import math

import gym
from gym import error, spaces, utils, wrappers
from gym.utils import seeding
from gym.envs.registration import register
from gym.spaces import Discrete, Box

import numpy as np


def sigmoid_price_fun(x, maxcust, gamma):
    return maxcust / (1 + math.exp(gamma * max(0, x)))


class Actions(Enum):
    DECREASE_PRICE = 0
    INCREASE_PRICE = 1
    HOLD = 2


PRICE_ADJUSTMENT = {
    Actions.DECREASE_PRICE: -0.25,
    Actions.INCREASE_PRICE: 0.25,
    Actions.HOLD: 0
}


class ArrivalSim(gym.Env):
    """ Simple environment for price optimising RL learner. """

    def __init__(self, price):
        """
        Parameters
        ----------
        price : float
            The initial price to use.
        """
        super().__init__()

        self.price = price
        self.revenue = 0
        self.action_space = Discrete(3)  # [0, 1, 2]  #increase or decrease
        # original obs space:
        #self.observation_space = Box(0.0, 1000.0, shape=(1,1), dtype=np.float32)
        # obs space initially suggested:
        #self.observation_space = Box(0.0, 1000.0, shape=(1,1), dtype=np.float32)
        # obs space suggested in this edit:
        self.observation_space = spaces.Box(np.array([0.0]), np.array([1000.0]), dtype=np.float32)

    def step(self, action):
        """ Enacts the specified action in the environment.

        Returns the new price, reward, whether we're finished and an empty dict for compatibility with Gym's
        interface. """

        self._take_action(Actions(action))

        next_state = self.price
        print('(self.price).shape =', (self.price).shape)
        #next_state = self.observation_space.sample()

        reward = self._get_reward()
        done = False

        if next_state < 0 or reward == 0:
            done = True

        print(next_state, reward, done, {})

        return np.array(next_state), reward, done, {}

    def reset(self):
        """ Resets the environment, selecting a random initial price. Returns the price. """
        #self.observation_space.value = np.random.rand()
        #return self.observation_space.sample()

        self.price = np.random.rand(1)

        print('reset -> (self.price).shape = ', (self.price).shape)

        return self.price

    def _take_action(self, action):
#         self.observation_space.value += PRICE_ADJUSTMENT[action]
        #print('price b =', self.price)
        print('price b =', self.price[0])
        #print('price b =', self.price[[0]])
        #self.price += PRICE_ADJUSTMENT[action]
        self.price[0] += PRICE_ADJUSTMENT[action]
        #self.price[[0]] += PRICE_ADJUSTMENT[action]
        #print('price a =', self.price)
        print('price a =', self.price[0])
        #print('price a =', self.price[[0]])

    #def _get_reward(self, price):
    def _get_reward(self):
#         price = self.observation_space.value
#         return max(np.random.poisson(sigmoid_price_fun(price, 50, 0.5)) * price, 0)
        #self.revenue = max(np.random.poisson(sigmoid_price_fun(self.price, 50, 0.5)) * self.price, 0)
        #return max(np.random.poisson(sigmoid_price_fun(self.price, 50, 0.5)) * self.price, 0)
        self.revenue = max(np.random.poisson(sigmoid_price_fun(self.price[0], 50, 0.5)) * self.price[0], 0)
        return max(np.random.poisson(sigmoid_price_fun(self.price[0], 50, 0.5)) * self.price[0], 0)

#     def render(self, mode='human'):
#         super().render(mode)

def testEnv():
    """
    register(
        id='ArrivalSim-v0',
        entry_point='env:ArrivalSim',
        kwargs= {'price' : 40.0}
    )
    env = gym.make('ArrivalSim-v0')
    """
    env = ArrivalSim(30.0)

    val = env.reset()
    print('val.shape = ', val.shape)

    for _ in range(5):
        print('env.observation_space =', env.observation_space)
        act = env.action_space.sample()
        print('\nact =', act)
        next_state, reward, done, _ = env.step(act)  # take a random action
        print('next_state = ', next_state)
    env.close()



if __name__ =='__main__':

    testEnv()

Отредактированный сценарий обучения:

import json
import os

import gym
import ray
from ray.tune import run_experiments
import ray.rllib.agents.a3c as a3c
import ray.rllib.agents.ppo as ppo
from ray.tune.registry import register_env
from mod_op_env import ArrivalSim

from sagemaker_rl.ray_launcher import SageMakerRayLauncher

"""
def create_environment(env_config):
    import gym
#     from gym.spaces import Space
    from gym.envs.registration import register

    # This import must happen inside the method so that worker processes import this code
    register(
        id='ArrivalSim-v0',
        entry_point='env:ArrivalSim',
        kwargs= {'price' : 40}
    )
    return gym.make('ArrivalSim-v0')
"""
def create_environment(env_config):
    price = 30.0
    # This import must happen inside the method so that worker processes import this code
    from mod_op_env import ArrivalSim
    return ArrivalSim(price)


class MyLauncher(SageMakerRayLauncher):
    def __init__(self):        
        super(MyLauncher, self).__init__()
        self.num_gpus = int(os.environ.get("SM_NUM_GPUS", 0))
        self.hosts_info = json.loads(os.environ.get("SM_RESOURCE_CONFIG"))["hosts"]
        self.num_total_gpus = self.num_gpus * len(self.hosts_info)

    def register_env_creator(self):
        register_env("ArrivalSim-v0", create_environment)

    def get_experiment_config(self):
        return {
          "training": {
            "env": "ArrivalSim-v0",
            "run": "PPO",
            "stop": {
              "training_iteration": 3,
            },

            "local_dir": "/opt/ml/model/",
            "checkpoint_freq" : 3,

            "config": {                                
              #"num_workers": max(self.num_total_gpus-1, 1),
              "num_workers": max(self.num_cpus-1, 1),
              #"use_gpu_for_workers": False,
              "train_batch_size": 128, #5,
              "sample_batch_size": 32, #1,
              "gpu_fraction": 0.3,
              "optimizer": {
                "grads_per_step": 10
              },
            },
            #"trial_resources": {"cpu": 1, "gpu": 0, "extra_gpu": max(self.num_total_gpus-1, 1), "extra_cpu": 0},
            #"trial_resources": {"cpu": 1, "gpu": 0, "extra_gpu": max(self.num_total_gpus-1, 0),
            #                    "extra_cpu": max(self.num_cpus-1, 1)},
            "trial_resources": {"cpu": 1,
                                "extra_cpu": max(self.num_cpus-1, 1)},              
          }
        }

if __name__ == "__main__":
    os.environ["LC_ALL"] = "C.UTF-8"
    os.environ["LANG"] = "C.UTF-8"
    os.environ["RAY_USE_XRAY"] = "1"
    print(ppo.DEFAULT_CONFIG)
    MyLauncher().train_main()

Код записной книжки:

!/bin/bash ./setup.sh

from time import gmtime, strftime
import sagemaker 
role = sagemaker.get_execution_role()

sage_session = sagemaker.session.Session()
s3_bucket = sage_session.default_bucket()  
s3_output_path = 's3://{}/'.format(s3_bucket)
print("S3 bucket path: {}".format(s3_output_path))

job_name_prefix = 'ArrivalSim'

from sagemaker.rl import RLEstimator, RLToolkit, RLFramework

estimator = RLEstimator(entry_point="mod_op_train.py", # Our launcher code
                        source_dir='src', # Directory where the supporting files are at. All of this will be
                                          # copied into the container.
                        dependencies=["common/sagemaker_rl"], # some other utils files.
                        toolkit=RLToolkit.RAY, # We want to run using the Ray toolkit against the ray container image.
                        framework=RLFramework.TENSORFLOW, # The code is in tensorflow backend.
                        toolkit_version='0.5.3', # Toolkit version. This will also choose an apporpriate tf version.                                               
                        #toolkit_version='0.6.5', # Toolkit version. This will also choose an apporpriate tf version.                        
                        role=role, # The IAM role that we created at the begining.
                        #train_instance_type="ml.m4.xlarge", # Since we want to run fast, lets run on GPUs.
                        train_instance_type="local", # Since we want to run fast, lets run on GPUs.
                        train_instance_count=1, # Single instance will also work, but running distributed makes things 
                                                # fast, particularly in the case of multiple rollout training.
                        output_path=s3_output_path, # The path where we can expect our trained model.
                        base_job_name=job_name_prefix, # This is the name we setup above to be to track our job.
                        hyperparameters = {      # Some hyperparameters for Ray toolkit to operate.
                          "s3_bucket": s3_bucket,
                          "rl.training.stop.training_iteration": 2, # Number of iterations.
                          "rl.training.checkpoint_freq": 2,
                        },
                        #metric_definitions=metric_definitions, # This will bring all the logs out into the notebook.
                    )

estimator.fit()