Блог

Правка: Проблема Решена. Решение приведено ниже.

Попытка построить модель RL для решения задачи. Есть два входа: x и y, оба измеряются по шкале int от 1 до 100. На основе этих двух входных данных должен быть вывод (действие, которое необходимо предпринять, дискретное(5)) и уверенность.

Кроме того, я очень новичок на этой территории. Пожалуйста, не стесняйтесь спрашивать меня о чем угодно или поправлять меня в чем-то, что кажется совершенно глупым/неправильным.

Вот моя программа (импорт не был очищен….):

 from abc import ABC
import gym
from tensorflow import keras
from gym import Env
from gym.spaces import Discrete, Box
import random
import numpy as np
from tensorflow.keras.models import Sequential
from tensorflow.keras import layers, losses, metrics
from tensorflow.keras.layers import Dense, Flatten, Input
from tensorflow.keras.optimizers import Adam
import os
from rl.agents import DQNAgent
from rl.policy import BoltzmannQPolicy
from rl.memory import SequentialMemory
os.environ['TF_CPP_MIN_LOG_LEVEL'] = '2'

steps = 10000
episodes = 100
score_requirement = 1000

class PlantEnv(Env, ABC):
    def __init__(self):
        # Actions = water: 0=(none), 1=(3 seconds), 2=(4 seconds), 3=(5 seconds), 4=(6 seconds)
        self.action_space = Discrete(5)

        # Starting Moisture
        moisture = 20   random.randint(-10, 10)
        # Starting Chance of Rain
        chance_of_rain = 50   random.randint(-50, 50)

        # Observations
        self.observation_space = Box(low=np.array([0, 0]), high=np.array([100, 100]), dtype=np.int)
        self.state = moisture, chance_of_rain

        # Number of water steps left
        self.water_length = steps

    def step(self, action):
        # Action section
        water = 0

        if action == 1:
            water = 2
        elif action == 2:
            water = 3
        elif action == 3:
            water = 4
        elif action == 4:
            water = 5

        moisture, chance_of_rain = self.state

        moisture  = (water * 5)
        self.water_length -= 1

        # Reward Section
        reward = 0
        if 40 <= moisture <= 60:
            reward = 2
        # If moisture is dry or wet
        elif 60 < moisture <= 80 or 20 <= moisture < 40:
            reward = 0.5
        # If moisture is really dry or really wet
        elif 80 < moisture <= 100 or 0 <= moisture < 20:
            reward = -1
        # If moisture is really dry or really wet
        elif 100 < moisture or moisture < 0:
            reward = -2

        # Check if shower is done
        if self.water_length <= 0:
            done = True
        else:
            done = False

        moistureLoss = random.randint(15, 25)
        moisture -= moistureLoss
        chance_of_rain = 50   random.randint(-50, 50)
        xfactor = chance_of_rain   random.randint(-50, 50)
        if xfactor > 100:
            moisture  = (10   random.randint(0, 15))

        # Set placeholder for info
        info = {}

        # Save current state
        self.state = moisture, chance_of_rain

        # Return step information
        return self.state, reward, done, info

    def reset(self):
        # Reset test environment
        # Set starting moisture
        moisture = 50   random.randint(-10, 10)
        # Set starting chance of rain array
        chance_of_rain = 50   random.randint(-50, 50)
        self.state = moisture, chance_of_rain
        # Reset Test time
        self.water_length = steps
        return self.state


def build_model():
    model = Sequential()
    model.add(Flatten(input_shape=(1, 4)))
    model.add(Dense(24, activation='relu'))
    model.add(Dense(24, activation='relu'))
    model.add(Dense(2, activation='linear'))
    return model


def build_agent(model):
    policy = BoltzmannQPolicy()
    memory = SequentialMemory(limit=50000, window_length=1)
    dqn = DQNAgent(model=model, memory=memory, policy=policy, nb_actions=2,
                   nb_steps_warmup=10, target_model_update=1e-2)
    return dqn


# Create environment
env = PlantEnv()

accepted_scores = []
training_data = []
scores = []
good_episodes = 0

# Create episodes and initiate simulation
for episode in range(1, episodes   1):
    observation = env.reset()
    done = False
    score = 0
    history = []
    prev_observation = []

    while not done:
        action = env.action_space.sample()
        if observation[0] > 100:
            action = 0
        elif observation[0] < 0:
            action = 4
        observation, reward, done, info = env.step(action)
        score  = reward
        if len(prev_observation) > 0:
            history.append([prev_observation, action])
        prev_observation = observation

    if score >= score_requirement:
        good_episodes  = 1
        accepted_scores.append(score)
        for data in history:
            if data[1] == 1:
                output = [1]
            else:
                output = [0]

            training_data.append([data[0], output])

    scores.append(score)

if len(accepted_scores) > 0:
    print("Average accepted score: ", np.mean(accepted_scores))
    print("Median accepted score : ", np.median(accepted_scores))
print("Episodes above accepted score of {}: {}/{}n".format(score_requirement, good_episodes, episodes))

model = build_model()
model.summary()

dqn = build_agent(model)
dqn.compile(Adam(learning_rate=1e-3), metrics=['mae'])
dqn.fit(env, nb_steps=50000, visualize=False, verbose=1)
 

Первая модель выдает эту ошибку при попытке dqn.fit:
Ошибка при проверке ввода: ожидалось, что dense_input будет иметь 2 измерения, но получил массив с формой (1, 1, 2)

Вторая модель выдает эту ошибку при попытке построить_агент: ошибка атрибута: объект «список» не имеет атрибута «форма»

Любые идеи о том, что я делаю неправильно или как это исправить, окажут огромную помощь. Я на 95% уверен, что правильно настроил свою среду.

Сначала я выбрал первую модель просто для того, чтобы посмотреть, смогу ли я заставить программу компилироваться и работать. Затем, после дальнейших исследований, я построил вторую модель, потому что понял, что она способна дать мне действие с рейтингом доверия. Получение ошибок на обоих поворотах.

Ваша модель ожидает 2D-ввода, но вы определили его как 1D. Вот рабочий пример:

 from abc import ABC
import gym
from tensorflow import keras
from gym import Env
from gym.spaces import Discrete, Box
import random
import numpy as np
from tensorflow.keras.models import Sequential
from tensorflow.keras import layers, losses, metrics
from tensorflow.keras.layers import Dense, Flatten, Input
from tensorflow.keras.optimizers import Adam
import os
from rl.agents import DQNAgent
from rl.policy import BoltzmannQPolicy
from rl.memory import SequentialMemory
os.environ['TF_CPP_MIN_LOG_LEVEL'] = '2'

steps = 10000
episodes = 100
score_requirement = 1000

class PlantEnv(Env, ABC):
    def __init__(self):
        # Actions = water: 0=(none), 1=(3 seconds), 2=(4 seconds), 3=(5 seconds), 4=(6 seconds)
        self.action_space = Discrete(5)

        # Starting Moisture
        moisture = 20   random.randint(-10, 10)
        # Starting Chance of Rain
        chance_of_rain = 50   random.randint(-50, 50)

        # Observations
        self.observation_space = Box(low=np.array([0, 0]), high=np.array([100, 100]), dtype=np.int)
        self.state = moisture, chance_of_rain

        # Number of water steps left
        self.water_length = steps

    def step(self, action):
        # Action section
        water = 0

        if action == 1:
            water = 2
        elif action == 2:
            water = 3
        elif action == 3:
            water = 4
        elif action == 4:
            water = 5

        moisture, chance_of_rain = self.state

        moisture  = (water * 5)
        self.water_length -= 1

        # Reward Section
        reward = 0
        if 40 <= moisture <= 60:
            reward = 2
        # If moisture is dry or wet
        elif 60 < moisture <= 80 or 20 <= moisture < 40:
            reward = 0.5
        # If moisture is really dry or really wet
        elif 80 < moisture <= 100 or 0 <= moisture < 20:
            reward = -1
        # If moisture is really dry or really wet
        elif 100 < moisture or moisture < 0:
            reward = -2

        # Check if shower is done
        if self.water_length <= 0:
            done = True
        else:
            done = False

        moistureLoss = random.randint(15, 25)
        moisture -= moistureLoss
        chance_of_rain = 50   random.randint(-50, 50)
        xfactor = chance_of_rain   random.randint(-50, 50)
        if xfactor > 100:
            moisture  = (10   random.randint(0, 15))

        # Set placeholder for info
        info = {}

        # Save current state
        self.state = moisture, chance_of_rain

        # Return step information
        return self.state, reward, done, info

    def reset(self):
        # Reset test environment
        # Set starting moisture
        moisture = 50   random.randint(-10, 10)
        # Set starting chance of rain array
        chance_of_rain = 50   random.randint(-50, 50)
        self.state = moisture, chance_of_rain
        # Reset Test time
        self.water_length = steps
        return self.state


def build_model():
    model = Sequential()
    model.add(Flatten(input_shape=(1, 2)))
    model.add(Dense(24, activation='relu'))
    model.add(Dense(24, activation='relu'))
    model.add(Dense(5, activation='linear'))
    return model


def build_agent(model):
    policy = BoltzmannQPolicy()
    memory = SequentialMemory(limit=50000, window_length=1)
    dqn = DQNAgent(model=model, memory=memory, policy=policy, nb_actions=5,
                   nb_steps_warmup=10, target_model_update=1e-2)
    return dqn


# Create environment
env = PlantEnv()

accepted_scores = []
training_data = []
scores = []
good_episodes = 0

# Create episodes and initiate simulation
for episode in range(1, episodes   1):
    observation = env.reset()
    done = False
    score = 0
    history = []
    prev_observation = []

    while not done:
        action = env.action_space.sample()
        if observation[0] > 100:
            action = 0
        elif observation[0] < 0:
            action = 4
        observation, reward, done, info = env.step(action)
        score  = reward
        if len(prev_observation) > 0:
            history.append([prev_observation, action])
        prev_observation = observation

    if score >= score_requirement:
        good_episodes  = 1
        accepted_scores.append(score)
        for data in history:
            if data[1] == 1:
                output = [1]
            else:
                output = [0]

            training_data.append([data[0], output])

    scores.append(score)

if len(accepted_scores) > 0:
    print("Average accepted score: ", np.mean(accepted_scores))
    print("Median accepted score : ", np.median(accepted_scores))
print("Episodes above accepted score of {}: {}/{}n".format(score_requirement, good_episodes, episodes))

model = build_model()
model.summary()

dqn = build_agent(model)
dqn.compile(Adam(learning_rate=1e-3), metrics=['mae'])
dqn.fit(env, nb_steps=50000, visualize=False, verbose=1)
 

Обратите внимание, что input_shape 4 равно числу states , а выходные узлы (2) равны числу actions . Вам придется изменить эти параметры в соответствии с вашим набором данных.