前書き
かつて「強化学習：はじめに」という本を読みながら、理論的な知識を実践的な知識で補うことを考えましたが、バーのバランスをとるという次の問題を解決したり、エージェントにチェスをしたり、別の自転車を発明したりすることを望んでいませんでした。
同時に、この本には、顧客キューを最適化する興味深い例が1つ含まれています。これは、一方ではプロセスの実装/理解の点でそれほど複雑ではなく、他方では非常に興味深く、実際の生活である程度の成功を収めて実装できます。
この例を少し変更して、後で説明するアイデアにたどり着きました。
問題の定式化
したがって、次の図を想像してみてください。
毎日6（条件付き）トンのラズベリーパイを生産し、これらの製品を毎日3つの店舗に配布するパン屋があります。
ただし、期限切れの製品をできるだけ少なくするためにこれを行う最善の方法（パイの保管寿命が3日である場合）、各販売ポイントでそれぞれ1、2、および3トンの容量を持つトラックが3台しかない場合、最も収益性が高くなります。トラックを1台だけ送る（お互いに十分に離れているため）、さらに、パイを焼いた後、1日1回だけ送るだけでなく、店舗での購入力がわかりません（事業が始まったばかりなので）？
FIFOレイアウト戦略は、顧客が他の商品よりも遅れて生産された商品のみを受け取る店舗で完全に機能することに同意しましょう。ただし、ラズベリーパイが3日以内に購入されなかった場合、店舗スタッフはそれを取り除きます。
特定の店舗での特定の日のパイの需要が（条件付きで）わかりませんが、シミュレーションでは、3つの店舗のそれぞれについて次のように設定しました：3±0.1、1±0.1、2±0.1。
明らかに、私たちにとって最も有益なオプションは、最初の店舗に3トン、2番目の店舗に1トン、3番目の店舗に2トンのパイを送ることです。
この問題を解決するために、カスタムジム環境とDeep Q Learning（Keras実装）を使用します。
カスタム環境
環境の状態を3つの実際の正の数値で説明します。3つの店舗のそれぞれにおける当日の残りの製品です。エージェントのアクションは0から5までの数字で、整数1、2、3の順列のインデックスを示します。最も有益なアクションが4番目のインデックス（3、1、2）の下にあることは明らかです。1つのエピソードで30日間、タスクを一時的なものと見なします。
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('')
主な輸入品
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 #
エージェントの定義
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
エージェントをトレーニングする
# 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
エージェントのテスト
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()
合計
したがって、エージェントは最も有益に行動する方法をすぐに理解しました。
一般に、まだ実験の余地はたくさんあります。ストアの数を増やしたり、アクションを多様化したり、トレーニングモデルのハイパーパラメータを変更したりすることもできます。これはリストの始まりにすぎません。