这篇文章主要介绍 DQN 的三种改进：Nature DQN、Double DQN、Dueling DQN 在运筹学中的应用，并给出三者的对比，同时也会给出不同增量学习下的效果。

这三种具体的改进方式可以从之前的文章中学习 《【强化学习】DQN 的各种改进》

背景

（搬运下背景）

假设有一个客服排班的任务，我们需要为 100 个人安排一个星期的排班问题，并且有以下约束条件：

• 一天被划分为 24 个时间段，即每个时间段为 1 个小时；

• 每个客服一个星期需要上七天班，每次上班八小时；

• 每个客服两次上班时间需要间隔 12 小时；

• 客服值班时，一个星期最早是 0，最晚 24*7 - 1。

评判标准：

• 现在有每个时间段所需客服人数，我们希望每个时段排班后的人数与实际人数尽量相近。

最优化问题可以使用启发式算法来做，上次用 DQN，这次用深度强化学习。

Nature DQN

之前给过 DQN 的代码，但是由于没有用批处理，所以速度非常慢，这里为了方便大家查看，给出完整版的 Nature DQN 代码，但是 Double DQN 和 Dueling DQN 的代码只会放上在前者需要修改的部分。

所以，Double DQN 的改进会加上 Nature DQN 的改进的部分，同理 Dueling DQN 实际上是 Nature DQN、Double DQN、Dueling DQN 三者的缝合怪。

import random
import numpy as np
import tensorflow as tf
from tensorflow.keras import layers
from copy import deepcopy
from collections import defaultdict, dequerandom.seed(2020)gpus = tf.config.experimental.list_physical_devices(device_type='GPU')
# GPU 随使用量增长
tf.config.experimental.set_memory_growth(gpus[0], True)
# 设定最大显存
tf.config.experimental.set_virtual_device_configuration(gpus[0],[tf.config.experimental.VirtualDeviceConfiguration(memory_limit=1024*16)]
)person_n = 10# 随机的一个排班需求
act_list = [5, 8, 8, 8, 5, 7, 9, 7, 5, 9, 7, 10, 10, 10, 7, 5, 10, 6, 7, 10, 7, 5, 6, 6, 10, 5, 9, 8, 8, 9, 9, 7, 6, 9, 7, 5, 9, 8, 7, 9, 10, 6, 7, 6, 6, 5, 8, 8, 9, 7, 8, 9, 8, 7, 7, 8, 9, 8, 7, 8, 9, 7, 10, 7, 5, 10, 10, 10, 7, 5, 6, 5, 9, 7, 5, 8, 7, 5, 5, 5, 7, 9, 9, 7, 9, 6, 9, 9, 9, 8, 9, 10, 5, 6, 6, 8, 7, 6, 5, 5, 9, 6, 7, 8, 6, 8, 9, 8, 5, 5, 8, 8, 6, 7, 9, 9, 10, 7, 8, 6, 6, 9, 6, 5, 6, 7, 5, 5, 8, 6, 5, 10, 10, 8, 10, 10, 6, 9, 8, 6, 5, 8, 6, 9, 8, 9, 6, 7, 6, 5, 9, 7, 7, 9, 6, 10, 7, 9, 5, 9, 9, 8, 7, 9, 9, 8, 8, 5]

class Env():def __init__(self):# 10 个人, 7 天，每个 bar 都可以向左向右移动，也可以不移动 '-1'self.actions_space = ['{}{}L'.format(i, j) for i in range(person_n) for j in range(7)] + \['{}{}R'.format(i, j) for i in range(person_n) for j in range(7)] + ['-1']self.n_actions = len(self.actions_space)self.act_list = act_listself.w_list = [i / sum(self.act_list) for i in self.act_list]self.state = [[i*24 for i in range(7)] for i in range(person_n)]self.n_state = person_n * 7 * 24self.punish = -1print(self.act_list)def list_2_str(self, l):# 拼接完整的 liststate_list = [[0 for i in range(24*7)] for j in range(person_n)]for person in range(person_n):for i in l[person]:for j in range(8):state_list[person][i+j] = 1return [i for state in state_list for i in state]def reset(self):self.state = [[i*24 for i in range(7)] for i in range(person_n)]return self.list_2_str(self.state)# 给当前排班打分，考虑权重def reward(self, tmp_state):# 判断每个人的排班要间隔 8+12 小时，否则 socre = -1000for i in range(person_n):# 星期天和星期一的排班间隔 8+12 小时if (tmp_state[i][0] + (24*7-1) - tmp_state[i][6]) < 20:return self.punishfor j in range(6):if (tmp_state[i][j+1] - tmp_state[i][j]) < 20:return self.punish# 拼接完整的 liststate_list = [[0] * 24 * 7] * person_nfor person in range(person_n):for i in tmp_state[person]:for j in range(8):state_list[person][i+j] = 1plan_list = np.sum(state_list, axis=0).tolist()s_list = [abs(plan_list[i] - self.act_list[i])/self.act_list[i] for i in range(len(plan_list))]# 奖励越大越好，所以加个负号score = 1-np.sum([s_list[i]*self.w_list[i] for i in range(len(s_list))])return scoredef step(self, action):actions_str = self.actions_space[action]if actions_str == '-1':return self.list_2_str(self.state), self.reward(self.state)else:num = int(actions_str[0])day = int(actions_str[1])move = actions_str[2]tmp_state = deepcopy(self.state)if move == 'R':if tmp_state[num][day] == (24*7-8-1):tmp_state[num][day] = tmp_state[num][day] + 1return self.list_2_str(tmp_state), self.punishtmp_state[num][day] = tmp_state[num][day] + 1if move == 'L':if tmp_state[num][day] == 0:tmp_state[num][day] = tmp_state[num][day] - 1return self.list_2_str(tmp_state), self.punishtmp_state[num][day] = tmp_state[num][day] - 1reward = self.reward(tmp_state)if reward == self.punish:return self.list_2_str(tmp_state), self.punishself.state = tmp_statereturn self.list_2_str(self.state), self.reward(self.state)

class DQNAgent:def __init__(self, state_size, action_size):self.state_size = state_sizeself.action_size = action_sizeself.memory = deque(maxlen=2000)self.discount_factor = 0.9self.epsilon = 1.0  # exploration rateself.epsilon_min = 0.1self.epsilon_decay = 0.999self.learning_rate = 0.01# Nature DQN 就是创建了两个 DQN，防止又踢球又当裁判self.model = self._build_model()  # 用于选择动作、更新参数self.model_Q = self._build_model()  # 用于计算 Q 值，定期从 model 中拷贝数据。def _build_model(self):model = tf.keras.Sequential()model.add(layers.Dense((512), input_shape=(self.state_size, ), activation='relu'))model.add(layers.BatchNormalization())model.add(layers.Dense((512), activation='relu'))model.add(layers.BatchNormalization())model.add(layers.Dense((256), activation='relu'))model.add(layers.BatchNormalization())model.add(layers.Dense((self.action_size), activation='sigmoid'))model.compile(loss='mse', optimizer=tf.keras.optimizers.Adam(lr=self.learning_rate))return modeldef update_model_Q(self):self.model_Q.set_weights(self.model.get_weights())def memorize(self, state, action, reward, next_state):self.memory.append((state, action, reward, next_state))def get_action(self, state):if np.random.rand() <= self.epsilon:return random.randrange(self.action_size)act_values = self.model.predict(state)act_values = act_values[0]max_action = np.random.choice(np.where(act_values == np.max(act_values))[0])return max_action  # returns actiondef replay(self, batch_size):minibatch = random.sample(self.memory, batch_size)state_batch = [data[0] for data in minibatch]action_batch = [data[1] for data in minibatch]reward_batch = [data[2] for data in minibatch]next_state_batch = [data[3] for data in minibatch]next_state_batch = np.array(next_state_batch)next_state_batch = next_state_batch.reshape(batch_size, self.state_size)next_state_Q_batch = self.model_Q.predict(next_state_batch)state_batch = np.array(state_batch)state_batch = state_batch.reshape(batch_size, self.state_size)state_Q_batch = self.model_Q.predict(state_batch)y_batch = []for i in range(batch_size):target = reward_batch[i] + self.discount_factor * np.amax(next_state_Q_batch[i]) target_f = state_Q_batch[i]target_f[action] = targety_batch.append(target_f)y_batch = np.array(y_batch)y_batch = y_batch.reshape(batch_size, self.action_size)self.model.fit(state_batch, y_batch, epochs=5, verbose=0)if self.epsilon > self.epsilon_min:self.epsilon *= self.epsilon_decaydef load(self, name):self.model.load_weights(name)def save(self, name):self.model.save_weights(name)

env = Env()
bst_state = env.state
agent = DQNAgent(env.n_state, env.n_actions)episodes = 1
update_model_Q_freq = 50
batch_size = 32
bst_reward = -500for e in range(episodes):state = env.reset()print('---------- ', e, ' ------------')for i in range(20000):state = np.reshape(state, [1, env.n_state])action = agent.get_action(state)next_state, reward = env.step(action)next_state = np.reshape(next_state, [1, env.n_state])if i % update_model_Q_freq == 0:agent.update_model_Q()if reward != env.punish:state = deepcopy(next_state)agent.memorize(state, action, reward, next_state)if len(agent.memory) > batch_size:agent.replay(batch_size)if bst_reward < reward:bst_reward = rewardbst_state = deepcopy(env.state)print('episode: {}/{}, i:{}, reward: {}, e: {:.2}'.format(e, episodes, i, bst_reward, agent.epsilon))

Double DQN

Double DQN 使用当前 Q 网络计算每一个 action 对应的 q 值，然后记录最大 q 值对应的 max action，然后用目标网络和 max action 计算目标 q 值。

em... 这块看公式比较清楚。

class DQNAgent:... ...def replay(self, batch_size):minibatch = random.sample(self.memory, batch_size)state_batch = [data[0] for data in minibatch]action_batch = [data[1] for data in minibatch]reward_batch = [data[2] for data in minibatch]next_state_batch = [data[3] for data in minibatch]# 修改了这里next_state_batch = np.array(next_state_batch).reshape(batch_size, self.state_size)cur_state_Q_batch = self.model.predict(next_state_batch)max_action_next = np.argmax(cur_state_Q_batch, axis=1)next_state_Q_batch = self.model_Q.predict(next_state_batch)state_batch = np.array(state_batch).reshape(batch_size, self.state_size)state_Q_batch = self.model_Q.predict(state_batch)y_batch = []for i in range(batch_size):target = reward_batch[i] + self.discount_factor * next_state_Q_batch[i, max_action_next[i]]target_f = state_Q_batch[i]target_f[action] = targety_batch.append(target_f)y_batch = np.array(y_batch)y_batch = y_batch.reshape(batch_size, self.action_size)self.model.fit(state_batch, y_batch, epochs=5, verbose=0)if self.epsilon > self.epsilon_min:self.epsilon *= self.epsilon_decay

Dueling DQN

Dueling Network：将 Q 网络分成两个通道，一个输出 V，一个输出 A，最后再合起来得到 Q。

class DQNAgent:...def _build_model(self):inputs = tf.keras.Input(shape=(self.state_size, ))x = layers.Dense(512, activation='relu')(inputs)x = layers.BatchNormalization()(x)x = layers.Dense(512, activation='relu')(x)x = layers.BatchNormalization()(x)x = layers.Dense(256, activation='relu')(x)x = layers.BatchNormalization()(x)v = layers.Dense(1, activation='sigmoid')(x)a = layers.Dense(self.action_size, activation='sigmoid')(x)a = a - tf.reduce_mean(a)outputs = a + vmodel = tf.keras.Model(inputs=inputs, outputs=outputs)model.compile(loss='mse', optimizer=tf.keras.optimizers.Adam(lr=self.learning_rate))return model;

三者对比

放一下三者的对比：

PS：只实验了一次，结果仅供参考。

往期精彩回顾适合初学者入门人工智能的路线及资料下载机器学习及深度学习笔记等资料打印机器学习在线手册深度学习笔记专辑《统计学习方法》的代码复现专辑
AI基础下载机器学习的数学基础专辑
本站知识星球“黄博的机器学习圈子”（92416895）
本站qq群704220115。
加入微信群请扫码：