introduction
Une fois, en lisant le livre "Apprentissage par renforcement: une introduction", j'ai pensé à compléter mes connaissances théoriques par des connaissances pratiques, mais je n'avais aucune envie de résoudre le problème suivant: équilibrer une barre, apprendre à un agent à jouer aux échecs ou inventer un autre vélo.
Dans le même temps, le livre contenait un exemple intéressant d'optimisation de la file d'attente client, qui, d'une part, n'est pas trop compliqué en termes de mise en œuvre / compréhension du processus, et d'autre part, il est assez intéressant et peut être mis en œuvre avec un certain succès dans la vie réelle.
Ayant légèrement modifié cet exemple, j'en suis venu à l'idée, qui sera discutée plus loin.
Formulation du problème
Alors, imaginez l'image suivante:
Nous avons à notre disposition une boulangerie qui produit 6 tonnes (conditionnellement) de tartes aux framboises chaque jour et distribue ces produits dans trois magasins chaque jour.
Cependant, comment faire au mieux pour qu'il y ait le moins de produits périmés possible (à condition que la durée de conservation des tartes soit de trois jours), si nous n'avons que trois camions d'une capacité de 1, 2 et 3 tonnes, respectivement, à chaque point de vente, c'est le plus rentable d'envoyer un seul camion (car ils sont assez éloignés les uns des autres) et, de plus, une seule fois par jour après la cuisson des tartes, et d'ailleurs, nous ne connaissons pas le pouvoir d'achat dans nos magasins (puisque l'entreprise vient de démarrer)?
Admettons que la stratégie de mise en page FIFO fonctionne parfaitement dans les magasins, dans lesquels les clients ne prennent que les produits qui ont été produits plus tard que les autres, mais si la tarte aux framboises n'a pas été achetée dans les trois jours, le personnel du magasin s'en débarrasse.
Nous ne savons pas (conditionnellement) quelle sera la demande de tartes un jour particulier dans un magasin particulier, cependant, dans notre simulation, nous la définissons comme suit pour chacun des trois magasins: 3 ± 0,1, 1 ± 0,1, 2 ± 0,1.
Évidemment, l'option la plus rentable pour nous est d'envoyer respectivement trois tonnes au premier magasin, une au deuxième et deux tonnes de tartes au troisième.
Pour résoudre ce problème, nous utilisons un environnement de gym personnalisé, ainsi que Deep Q Learning (implémentation Keras).
Environnement personnalisé
L'état de l'environnement sera décrit par trois nombres réels positifs - le reste des produits pour la journée en cours dans chacun des trois magasins. Les actions de l'agent sont des nombres de 0 à 5 inclus, désignant les indices de permutation des entiers 1, 2 et 3. Il est clair que l'action la plus bénéfique se situera sous le 4ème indice (3, 1, 2). Nous considérons le problème comme épisodique, dans un épisode de 30 jours.
import gym
from gym import error, spaces, utils
from gym.utils import seeding
import itertools
import random
import time
class ShopsEnv(gym.Env):
metadata = {'render.modes': ['human']}
# ,
#
def __init__(self):
self.state = [0, 0, 0] #
self.next_state = [0, 0, 0] #
self.done = False #
self.actions = list(itertools.permutations([1, 2, 3])) #
self.reward = 0 #
self.time_tracker = 0 #
self.remembered_states = [] #
#
t = int( time.time() * 1000.0 )
random.seed( ((t & 0xff000000) >> 24) +
((t & 0x00ff0000) >> 8) +
((t & 0x0000ff00) << 8) +
((t & 0x000000ff) << 24) )
# ()
def step(self, action_num):
#
if self.done:
return [self.state, self.reward, self.done, self.next_state]
else:
#
self.state = self.next_state
#
self.remembered_states.append(self.state)
#
self.time_tracker += 1
#
action = self.actions[action_num]
# , ( )
self.next_state = [x + y for x, y in zip(action, self.state)]
#
self.next_state[0] -= (3 + random.uniform(-0.1, 0.1))
self.next_state[1] -= (1 + random.uniform(-0.1, 0.1))
self.next_state[2] -= (2 + random.uniform(-0.1, 0.1))
#
if any([x < 0 for x in self.next_state]):
self.reward = sum([x for x in self.next_state if x < 0])
else:
self.reward = 1
#
#
# ( ),
#
if self.time_tracker >= 3:
remembered_state = self.remembered_states.pop(0)
self.next_state = [max(x - y, 0) for x, y in zip(self.next_state, remembered_state)]
else:
self.next_state = [max(x, 0) for x in self.next_state]
# 30
self.done = self.time_tracker == 30
#
return [self.state, self.reward, self.done, self.next_state]
#
def reset(self):
#
self.state = [0, 0, 0]
self.next_state = [0, 0, 0]
self.done = False
self.reward = 0
self.time_tracker = 0
self.remembered_states = []
t = int( time.time() * 1000.0 )
random.seed( ((t & 0xff000000) >> 24) +
((t & 0x00ff0000) >> 8) +
((t & 0x0000ff00) << 8) +
((t & 0x000000ff) << 24) )
#
return self.state
# :
#
def render(self, mode='human', close=False):
print('-'*20)
print('First shop')
print('Pies:', self.state[0])
print('Second shop')
print('Pies:', self.state[1])
print('Third shop')
print('Pies:', self.state[2])
print('-'*20)
print('')
Importations majeures
import numpy as np #
import pandas as pd #
import gym #
import gym_shops #
from tqdm import tqdm #
#
import matplotlib.pyplot as plt
import seaborn as sns
from IPython.display import clear_output
sns.set_color_codes()
#
from collections import deque
from keras.models import Sequential
from keras.layers import Dense
from keras.optimizers import Adam
import random #
Définition d'un agent
class DQLAgent():
def __init__(self, env):
#
self.state_size = 3 #
self.action_size = 6 #
# replay()
self.gamma = 0.99
self.learning_rate = 0.01
# adaptiveEGreedy()
self.epsilon = 0.99
self.epsilon_decay = 0.99
self.epsilon_min = 0.0001
self.memory = deque(maxlen = 5000) # 5000 , -
# (NN)
self.model = self.build_model()
# Deep Q Learning
def build_model(self):
model = Sequential()
model.add(Dense(10, input_dim = self.state_size, activation = 'sigmoid')) #
model.add(Dense(50, activation = 'sigmoid')) #
model.add(Dense(10, activation = 'sigmoid')) #
model.add(Dense(self.action_size, activation = 'sigmoid')) #
model.compile(loss = 'mse', optimizer = Adam(lr = self.learning_rate))
return model
#
def remember(self, state, action, reward, next_state, done):
self.memory.append((state, action, reward, next_state, done))
#
def act(self, state):
# 0 1 epsilon
# (exploration)
if random.uniform(0,1) <= self.epsilon:
return random.choice(range(6))
else:
#
act_values = self.model.predict(state)
return np.argmax(act_values[0])
#
def replay(self, batch_size):
# ,
if len(self.memory) < batch_size:
return
minibatch = random.sample(self.memory, batch_size) # batch_size
#
for state, action, reward, next_state, done in minibatch:
if done: # -
target = reward
else:
#
target = reward + self.gamma * np.amax(self.model.predict(next_state)[0])
# target = R(s,a) + gamma * max Q`(s`,a`)
# target (max Q` value) , s`
train_target = self.model.predict(state) # s --> NN --> Q(s,a) = train_target
train_target[0][action] = target
self.model.fit(state, train_target, verbose = 0)
# exploration rate,
# epsilon
def adaptiveEGreedy(self):
if self.epsilon > self.epsilon_min:
self.epsilon *= self.epsilon_decay
Former l'agent
# gym
env = gym.make('shops-v0')
agent = DQLAgent(env)
#
batch_size = 100
episodes = 1000
#
progress_bar = tqdm(range(episodes), position=0, leave=True)
for e in progress_bar:
#
state = env.reset()
state = np.reshape(state, [1, 3])
# , id
time = 0
taken_actions = []
sum_rewards = 0
#
while True:
#
action = agent.act(state)
#
taken_actions.append(action)
#
next_state, reward, done, _ = env.step(action)
next_state = np.reshape(next_state, [1, 3])
#
sum_rewards += reward
#
agent.remember(state, action, reward, next_state, done)
#
state = next_state
# replay
agent.replay(batch_size)
# epsilon
agent.adaptiveEGreedy()
#
time += 1
#
progress_bar.set_postfix_str(s='mean reward: {}, time: {}, epsilon: {}'.format(round(sum_rewards/time, 3), time, round(agent.epsilon, 3)), refresh=True)
#
if done:
#
clear_output(wait=True)
sns.distplot(taken_actions, color="y")
plt.title('Episode: ' + str(e))
plt.xlabel('Action number')
plt.ylabel('Occurrence in %')
plt.show()
break
Tester l'agent
import time
trained_model = agent #
state = env.reset() #
state = np.reshape(state, [1,3])
#
time_t = 0
MAX_EPISOD_LENGTH = 1000 #
taken_actions = []
mean_reward = 0
#
progress_bar = tqdm(range(MAX_EPISOD_LENGTH), position=0, leave=True)
for time_t in progress_bar:
#
action = trained_model.act(state)
next_state, reward, done, _ = env.step(action)
next_state = np.reshape(next_state, [1,3])
state = next_state
taken_actions.append(action)
#
clear_output(wait=True)
env.render()
progress_bar.set_postfix_str(s='time: {}'.format(time_t), refresh=True)
print('Reward:', round(env.reward, 3))
time.sleep(0.5)
mean_reward += env.reward
if done:
break
#
sns.distplot(taken_actions, color='y')
plt.title('Test episode - mean reward: ' + str(round(mean_reward/(time_t+1), 3)))
plt.xlabel('Action number')
plt.ylabel('Occurrence in %')
plt.show()
Total
Ainsi, l'agent a rapidement compris comment agir de la manière la plus rentable.
En général, il y a encore beaucoup de place pour l'expérimentation: vous pouvez augmenter le nombre de magasins, diversifier les actions, ou même simplement changer les hyperparamètres du modèle de formation - et ce n'est que le début de la liste.