pengantar
Suatu ketika, saat membaca buku "Reinforcement Learning: An Introduction", saya berpikir untuk melengkapi pengetahuan teoretis saya dengan pengetahuan praktis, tetapi tidak ada keinginan untuk memecahkan masalah berikutnya yaitu menyeimbangkan batang, mengajari agen bermain catur, atau menciptakan sepeda lain.
Pada saat yang sama, buku tersebut memuat satu contoh menarik tentang optimalisasi antrian pelanggan yang di satu sisi tidak terlalu sulit dalam hal implementasi / pemahaman prosesnya, dan di sisi lain cukup menarik dan dapat diimplementasikan dengan sukses di kehidupan nyata.
Setelah sedikit mengubah contoh ini, saya mendapatkan ide, yang akan dibahas lebih lanjut.
Rumusan masalah
Jadi, bayangkan gambar berikut:
Kami memiliki sebuah toko roti yang memproduksi 6 (secara bersyarat) ton pai raspberry setiap hari dan mendistribusikan produk ini ke tiga toko setiap hari.
Namun, cara terbaik untuk melakukan ini sehingga hanya ada sesedikit mungkin produk yang kadaluwarsa (asalkan umur simpan pai adalah tiga hari), jika kita hanya memiliki tiga truk dengan kapasitas masing-masing 1, 2 dan 3 ton, di setiap titik penjualan akan paling menguntungkan mengirim hanya satu truk (karena lokasinya cukup jauh dari satu sama lain) dan, terlebih lagi, hanya sekali sehari setelah memanggang pai, dan selain itu, kami tidak mengetahui daya beli di toko kami (karena bisnis baru saja dimulai)?
Mari kita sepakati bahwa strategi tata letak FIFO bekerja dengan sempurna di toko, di mana pelanggan hanya mengambil barang yang diproduksi lebih lambat dari yang lain, tetapi jika pai raspberry tidak dibeli dalam waktu tiga hari, staf toko akan menyingkirkannya.
Kami (secara kondisional) tidak tahu berapa permintaan pai pada hari tertentu di toko tertentu, namun, dalam simulasi kami, kami menetapkannya sebagai berikut untuk masing-masing dari tiga toko: 3 ± 0,1, 1 ± 0,1, 2 ± 0,1.
Jelas sekali, opsi yang paling menguntungkan bagi kami adalah mengirim tiga ton ke toko pertama, satu ke toko kedua, dan dua ton pai ke toko ketiga.
Untuk mengatasi masalah ini, kami menggunakan lingkungan gym kustom, serta Deep Q Learning (implementasi Keras).
Lingkungan khusus
Kami akan menjelaskan keadaan lingkungan dengan tiga angka positif nyata - sisa produk untuk hari ini di masing-masing dari tiga toko. Tindakan agen adalah angka dari 0 sampai 5 inklusif, menunjukkan indeks permutasi bilangan bulat 1, 2 dan 3. Jelas bahwa tindakan yang paling menguntungkan akan berada di bawah indeks ke-4 (3, 1, 2). Kami menganggap tugas sebagai episodik, dalam satu episode 30 hari.
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('')
Impor utama
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 #
Mendefinisikan agen
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
Latih agennya
# 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
Menguji agen
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
Dengan demikian, agen dengan cepat memahami bagaimana bertindak yang paling menguntungkan.
Secara umum, masih ada banyak ruang untuk eksperimen: Anda dapat menambah jumlah penyimpanan, mendiversifikasi tindakan, atau bahkan hanya mengubah hyperparameter model pelatihan - dan ini hanyalah permulaan dari daftar.