算法代码如下
import numpy
import tensorflow as tf
from tensorflow import keras
import copy
import numpy as np
import pandas as pd
from matplotlib import pyplot as plt
import gym
def DQN_run(DQN_agent=None, episode=200):
DQN_agent = DQN_agent.DQN(n_actions=2, n_features=4, n_experience_pool=300)
DQN_agent.net_init()
score = []
env = gym.make('CartPole-v1')
for i_episode in range(episode):
# 初始化，
observation = env.reset()
for t in range(1000):
# 刷新环境
env.render()
# print(observation)
# 从动作空间采样，得到动作
action = DQN_agent.choose_action(observation)
# print(action)
# 将动作输入到环境中，得到状态，回报 observation_ ----> s'
observation_, reward, done, info = env.step(action)
x, x_dot, theta, theta_dot = observation
# print(f"x = {x},theta = {theta}")
r2 = - abs(theta) * 2
# DQN的性能受到奖励函数的影响，合理的设计奖励函数，能够DQN的性能变的更好
DQN_agent.experience_store(s=observation, a=action, r=reward + r2, s_=observation_, done=done)
# print(t)
DQN_agent.learn()
observation = observation_
if done:
print("Episode finished after {} time steps".format(t + 1))
if DQN_agent.experience_pool_is_full:
score.append(t + 1)
break
plt.plot(score, color='red')
plt.show()
print()
class DQN:
def __init__(self,
n_actions,
n_features,
epsilon=0.1,
batch_size=64,
learning_rate=0.001,
gamma=0.9,
replace_time=300,
n_experience_pool=300):
self.n_actions = n_actions
self.n_features = n_features
self.batch_size = batch_size
# 学习率
self.learning_rate = learning_rate
self.gamma = gamma
# epsilon-贪心算法
self.epsilon = epsilon
# 经验池大小
self.n_experience_pool = n_experience_pool
# 建立经验池 建立一个n_experience_pool行，n_features * 2 + 1 + 1 列的矩阵 s a r s_
self.experience_pool = pd.DataFrame(np.zeros([self.n_experience_pool, self.n_features * 2 + 1 + 1 + 1]))
self.experience_pool_index = 0
self.experience_pool_is_full = False
# 两个神经网络定义
self.q_pred = None
self.q_target = None
# 优化器定义
self.opt = tf.keras.optimizers.Adam(self.learning_rate)
# 间隔C次，进行参数更新
self.replace_time = replace_time
self.now_learn_time = 0
def loss_f(self, y_true, y_pred):
return keras.losses.mse(y_true, y_pred)
def choose_action(self, s):
s = s.reshape(1, 4)
rand = np.random.rand(1)
if rand < self.epsilon:
self.epsilon = self.epsilon * 0.999999
return np.random.randint(0, self.n_actions)
else:
# 采用numpy中argmax来得到最大值所对应的元素下标
# 这里尤其要注意，不能直接输入s，这里必须输入二维的数组，而不是单独的s
action_value = self.q_pred.predict(np.array(s))
return np.argmax(action_value)
def net_init(self):
# 神经网络的输入维数就是特征的维数，在强化学习中，就是状态
# shape=(self.n_features,) shape=（32，）`表示预期的输入将是一批32维向量 32列 n行
input_features = tf.keras.Input(shape=(self.n_features,), name='input_features')
# Dense（全连接层）的第一个参数32是他的输出维度,将输入输入层加入到全连接层上
dense_0 = tf.keras.layers.Dense(32, activation='relu')(input_features)
dense_1 = tf.keras.layers.Dense(64, activation='relu')(dense_0)
dense_2 = tf.keras.layers.Dense(32, activation='relu')(dense_1)
out_put = tf.keras.layers.Dense(self.n_actions, name='prediction_q_pred')(dense_2)
self.q_pred = tf.keras.Model(inputs=input_features, outputs=out_put)
# q_target
input_features_target = tf.keras.Input(shape=(self.n_features,), name='input_features')
dense_0_target = tf.keras.layers.Dense(32, activation='relu')(input_features_target)
dense_1_target = tf.keras.layers.Dense(64, activation='relu')(dense_0_target)
dense_2_target = tf.keras.layers.Dense(32, activation='relu')(dense_1_target)
out_put_target = tf.keras.layers.Dense(self.n_actions, name='prediction_q_target')(dense_2_target)
self.q_target = tf.keras.Model(inputs=input_features_target, outputs=out_put_target)
# 这里使用copy.deepcopy()进行神经网络的复制是不可取的，导致无法进行预测输出，原因目前还不知道
# self.q_target = copy.deepcopy(self.q_target)
self.q_target.set_weights(self.q_pred.get_weights())
# print(self.q_target.summary())
def experience_store(self, s, a, r, s_, done):
experience = []
for i in range(self.n_features * 2 + 2 + 1):
if i < self.n_features:
experience.append(s[i])
elif self.n_features <= i < self.n_features + 1:
experience.append(a)
elif self.n_features + 1 <= i < self.n_features + 2:
experience.append(r)
elif self.n_features + 2 <= i < self.n_features * 2 + 2:
experience.append(s_[i - self.n_features - 2])
else:
experience.append(done)
self.experience_pool.loc[self.experience_pool_index] = copy.deepcopy(experience)
self.experience_pool_index += 1
# print(self.experience_pool_index)
if self.experience_pool_index == self.n_experience_pool:
self.experience_pool_is_full = True
self.experience_pool_index = 0
def learn(self):
if not self.experience_pool_is_full:
return
# 注意这里，如果要自己建立数据集的话，最好使用pd中的dataframe，其自带了sample函数，可以进行取样
# 其他情况可以使用tensorflow的tf.data.Dataset.from_tensor_slices() 来进行数据集的建立，在使用shuffle进行打乱训练。
data_pool = self.experience_pool.sample(self.batch_size)
# 这里应该注意把DataFrame格式的转换为ndarray
s = np.array(data_pool.loc[:, 0:self.n_features - 1])
a = np.array(data_pool.loc[:, self.n_features], dtype=np.int32)
r = np.array(data_pool.loc[:, self.n_features + 1])
s_ = np.array(data_pool.loc[:, self.n_features + 2:self.n_features * 2 + 1])
done = np.array(data_pool.loc[:, self.n_features * 2 + 2])
with tf.GradientTape() as Tape:
y_pred = self.q_pred(s)
y_target = y_pred.numpy()
q_target = self.q_target(s_)
q_target = q_target.numpy()
index = np.arange(self.batch_size, dtype=np.int32)
# 注意这个地方的语法使用
y_target[index, a] = r + (1 - done) * self.gamma * np.max(q_target, axis=1)
loss_val = tf.keras.losses.mse(y_target, y_pred)
gradients = Tape.gradient(loss_val, self.q_pred.trainable_variables)
self.opt.apply_gradients(zip(gradients, self.q_pred.trainable_variables))
# 首先选出Q-target网络的最大值
self.now_learn_time += 1
if self.now_learn_time == self.replace_time:
self.replace_param()
self.now_learn_time = 0
def replace_param(self):
print("replace the param")
self.q_target.set_weights(self.q_pred.get_weights())
主函数调用：
from RL_algorithm_package import DQN
if __name__ == '__main__':
# 运行DQN
DQN.DQN_run(DQN_agent=DQN)
①设计经验池，可以用pandas中的DataFrame来实现，经验池要存储s,a,r,done,s_五个元素，根据元素的维数设计经验池的大小。
self.experience_pool = pd.DataFrame(np.zeros([self.n_experience_pool, self.n_features * 2 + 1 + 1 + 1]))
同时设计经验池的存储函数，从上往下依次存储，等完全存储完成之后，再从头开始存储
def experience_store(self, s, a, r, s_, done):
experience = []
for i in range(self.n_features * 2 + 2 + 1):
if i < self.n_features:
experience.append(s[i])
elif self.n_features <= i < self.n_features + 1:
experience.append(a)
elif self.n_features + 1 <= i < self.n_features + 2:
experience.append(r)
elif self.n_features + 2 <= i < self.n_features * 2 + 2:
experience.append(s_[i - self.n_features - 2])
else:
experience.append(done)
self.experience_pool.loc[self.experience_pool_index] = copy.deepcopy(experience)
self.experience_pool_index += 1
# print(self.experience_pool_index)
if self.experience_pool_index == self.n_experience_pool:
self.experience_pool_is_full = True
self.experience_pool_index = 0
②深度神经网络建立
这里采用的tensorflow中的Keras函数式编程算法实现。
def net_init(self):
# 神经网络的输入维数就是特征的维数，在强化学习中，就是状态
# shape=(self.n_features,) shape=（32，）`表示预期的输入将是一批32维向量 32列 n行
input_features = tf.keras.Input(shape=(self.n_features,), name='input_features')
# Dense（全连接层）的第一个参数32是他的输出维度,将输入输入层加入到全连接层上
dense_0 = tf.keras.layers.Dense(32, activation='relu')(input_features)
dense_1 = tf.keras.layers.Dense(64, activation='relu')(dense_0)
dense_2 = tf.keras.layers.Dense(32, activation='relu')(dense_1)
out_put = tf.keras.layers.Dense(self.n_actions, name='prediction_q_pred')(dense_2)
self.q_pred = tf.keras.Model(inputs=input_features, outputs=out_put)
# q_target
input_features_target = tf.keras.Input(shape=(self.n_features,), name='input_features')
dense_0_target = tf.keras.layers.Dense(32, activation='relu')(input_features_target)
dense_1_target = tf.keras.layers.Dense(64, activation='relu')(dense_0_target)
dense_2_target = tf.keras.layers.Dense(32, activation='relu')(dense_1_target)
out_put_target = tf.keras.layers.Dense(self.n_actions, name='prediction_q_target')(dense_2_target)
self.q_target = tf.keras.Model(inputs=input_features_target, outputs=out_put_target)
# 这里使用copy.deepcopy()进行神经网络的复制是不可取的，导致无法进行预测输出，原因目前还不知道
# self.q_target = copy.deepcopy(self.q_target)
self.q_target.set_weights(self.q_pred.get_weights())
# print(self.q_target.summary())
这里要注意的就是参数提取以及设置的方式
self.q_target.set_weights(self.q_pred.get_weights())
③DQN参数更新
之前采用DataFrame的格式来存储经验，这里就显示出了其的好处，可以直接调用DataFrame中的sample()函数，来显示随机采样。以实现打散数据间关联性
data_pool = self.experience_pool.sample(self.batch_size)
随后再将这个采样后的经验池中的经验提取出来，注意，这里要转换成ndarray格式。
s = np.array(data_pool.loc[:, 0:self.n_features - 1])
a = np.array(data_pool.loc[:, self.n_features], dtype=np.int32)
r = np.array(data_pool.loc[:, self.n_features + 1])
s_ = np.array(data_pool.loc[:, self.n_features + 2:self.n_features * 2 + 1])
done = np.array(data_pool.loc[:, self.n_features * 2 + 2])
采用的是DataFrame的loc函数，该函数可以返回某行某列的元素内容，[:, 0:self.n_features - 1]意思是从所有的行中，返回0到self.n_features - 1列的内容，这里是包括后一个元素的。这里a要注意转成int格式，方便后续的操作。
梯度下降算法是固定的写法
with tf.GradientTape() as Tape:
# 首先得到预测值，将其转换成numpy
y_pred = self.q_pred(s)
y_target = y_pred.numpy()
q_target = self.q_target(s_)
q_target = q_target.numpy()
index = np.arange(self.batch_size, dtype=np.int32)
# 注意这个地方的语法使用
y_target[index, a] = r + (1 - done) * self.gamma * np.max(q_target, axis=1)
loss_val = tf.keras.losses.mse(y_target, y_pred)
gradients = Tape.gradient(loss_val, self.q_pred.trainable_variables)
self.opt.apply_gradients(zip(gradients, self.q_pred.trainable_variables))
这里尤其要注意，只能对Q-target进行操作，不要对y_pred进行操作，否则会出现梯度为None的错误
y_target[index, a] = r + (1 - done) * self.gamma * np.max(q_target, axis=1)
就是对所有index的a列元素进行操作，这是一个批操作，可以简化代码量
值得注意的是，这里包含了DQN的更新公式，要对Q-pred网络的a（就是经验中存储的a）所在的位置更新，而对其他位置不更新。简单来说就是要进行如下步骤
1.将y-target复制Q-pred的输出y-pred
2.将y-target中a位置的元素，更换为更新公式计算出来的值，其余值不要动，仍与y-pred相同
3.进行mse求loss
④动作选择
将s输入到Q-pred网络中，进行输出，这里s要是tensor类型
⑤环境迭代
1.初始化环境，得到初始状态s
2.将s带入神经网络中的动作选择函数，得到动作a
3.将a带入到环境中，更新，得到s_,r,done（r也可以自己设计，不必使用环境得到的）
4.将s,a,r,done,s_存储到经验池中
5.若经验池满了，进行DQN学习（也可以隔多少次学习一次）
6.s = s_
7.执行2
效果图没保存，训练比较慢，并且不是很稳定。
类似于Q-learning，将表格更改成了神经网络。其余的算法原理网上内容很多。
由于DQN存在Q值高估的问题，所以可以进行改进，将其改进成DDQN
只需要将算法中的
y_target[index, a] = r + (1 - done) * self.gamma * np.max(q_target, axis=1)
np.max(q_target, axis=1)改成不是来自于q_target，而是先从将s_带入到q_pred网络中，得到最大的a，再将s_带入到q_target网络中，选择a所对应的值，进行更新。
with tf.GradientTape() as Tape:
# 这里是DQN与DDQN的区别之处
y_pred = self.q_pred(s)
y_target = y_pred.numpy()
q = self.q_pred(s_)
q = q.numpy()
arg_max_a = np.argmax(q,axis=1)
q_target = self.q_target(s_)
q_target = q_target.numpy()
index = np.arange(self.batch_size, dtype=np.int32)
y_target[index, a] = r + (1 - done) * self.gamma * q_target[index, arg_max_a]
loss_val = tf.keras.losses.mse(y_target, y_pred)
gradients = Tape.gradient(loss_val, self.q_pred.trainable_variables)
self.opt.apply_gradients(zip(gradients, self.q_pred.trainable_variables))
其余内容均与DQN相同，测试了一下，效果比DQN要好
竞争DQN网络，这个网络是把原本的Q网络进行分割，分割成价值网络和优势网络。
相比与DQN，只需要在网络架构上进行更改就行，其余内容不变
def net_init(self):
# 神经网络的输入维数就是特征的维数，在强化学习中，就是状态
# shape=(self.n_features,) shape=（32，）`表示预期的输入将是一批32维向量 32列 n行
input_features = tf.keras.Input(shape=(self.n_features,), name='input_features')
# Dense（全连接层）的第一个参数32是他的输出维度,将输入输入层加入到全连接层上
dense_0 = tf.keras.layers.Dense(32, activation='relu')(input_features)
dense_1 = tf.keras.layers.Dense(64, activation='relu')(dense_0)
dense_2 = tf.keras.layers.Dense(32, activation='relu')(dense_1)
# Dueling DQN 的特别之处，学习这种修改网络中间参数的方法
# 价值网络Vs输出
s_value = tf.keras.layers.Dense(1, name='prediction_s_value')(dense_2)
# 优势网络输出
A_s_a_ori = tf.keras.layers.Dense(self.n_actions, name='prediction_A_s_a')(dense_2)
# 对优势网络求平均，再相减,再加上价值网络，得到最后的输出，keepdims=True 要有
out_put = s_value + A_s_a_ori - tf.reduce_mean(A_s_a_ori, axis=1, keepdims=True)
self.q_pred = tf.keras.Model(inputs=input_features, outputs=out_put)
# q_target
self.q_target = tf.keras.Model(inputs=input_features, outputs=out_put)
# 这里使用copy.deepcopy()进行神经网络的复制是不可取的，导致无法进行预测输出，原因目前还不知道
# self.q_target = copy.deepcopy(self.q_target)
self.q_target.set_weights(self.q_pred.get_weights())
# print(self.q_target.summary())
效果图
