-------------初始化---------------
- 创建Gym环境实例
- 设置DQN参数
- 创建DQN实例- 载入DQN记忆（可选）- 载入DQN权重（可选）
- 创建tensorflow summary writer
- 设置训练参数
--------------训练----------------
- 循环N次
|     运行一局,得到该局内的reward和losses
|     记录数据到summary
|     显示数据到命令行
- 结束循环
--------------结束----------------- 保存DQN记忆- 保存DQN权重- 录制视频
- 关闭环境

from cart_pole import MyModel
from cart_pole import DQN
from cart_pole import play_game
from cart_pole import make_videoimport numpy as np
import tensorflow as tf
import gym
import os
import datetime
from statistics import mean
from gym import wrappersdef main():####################初始化#####################env = gym.make('CartPole-v0')gamma = 0.9num_states = len(env.observation_space.sample())num_actions = env.action_space.nhidden_units = [20, 20]max_experiences = 2000min_experiences = 1000batch_size = 32lr = 0.01e_greedy = 0.9e_greedy_increment = 1.001replace_target_iter = 50DQN_ = DQN(num_states, num_actions, hidden_units, gamma, max_experiences, min_experiences, batch_size, lr,e_greedy, replace_target_iter,e_greedy_increment)DQN_.loadModel()DQN_.loadMemory()current_time = datetime.datetime.now().strftime("%Y%m%d-%H%M%S")log_dir = 'logs/dqn/' + current_timesummary_writer = tf.summary.create_file_writer(log_dir)N = 10000 # 总训练轮次Dispaly_interval = 100total_rewards = np.empty(N)####################主循环#####################for n in range(N):total_reward, losses = play_game(env, DQN_)total_rewards[n] = total_rewardavg_rewards = total_rewards[max(0, n - Dispaly_interval):(n + 1)].mean()with summary_writer.as_default():tf.summary.scalar('episode reward', total_reward, step=n)tf.summary.scalar('running avg reward(100)', avg_rewards, step=n)tf.summary.scalar('average loss)', losses, step=n)if n % Dispaly_interval == 0:print("episode:", n, "eps:", DQN_.epsilon, "avg reward (last ",Dispaly_interval,"):", avg_rewards,"episode loss: ", losses)####################结束#####################make_video(env, DQN_)# DQN_.saveModel()# DQN_.saveMemory()env.close()

def play_game(env, DQN):rewards = 0done = Falseobservations = env.reset()losses = list()DQN.epsilon_incre()while not done:# env.render() #注释掉则不显示GUI，训练加速action = DQN.get_action(observations)prev_observations = observationsobservations, reward, done, _ = env.step(action)reward = My_reward(env,observation)rewards += rewardif done:reward = -10 #适用于cart_pole环境env.reset()exp = {'s': prev_observations, 'a': action, 'r': reward, 's2': observations, 'done': done}DQN.add_experience(exp)loss = DQN.train()if isinstance(loss, int): #当记忆库数量不足时loss=0losses.append(loss)else:losses.append(loss.numpy()) #类型转换return rewards, mean(losses)

def My_reward(env,observations):x, _, theta, _ = observations # observations = [x,x导数,theta,theta导数]rp = (env.x_threshold - abs(x))/env.x_threshold - 0.8 # reward positionra = (env.theta_threshold_radians - abs(theta))/env.theta_threshold_radians - 0.5 # reward anglereward = rp + ra return reward

def make_video(env, DQN):env = wrappers.Monitor(env, r"D:桌面RLvideos", force=True)rewards = 0steps = 0done = Falseobservation = env.reset()DQN.epsilon = 1 #epsilon = 1代表严格按照DQN输出行动，没有随机因素while not done:env.render() #调出并更新窗口action = DQN.get_action(observation)observation, reward, done, _ = env.step(action)reward = My_reward(env,observations)steps += 1rewards += rewardprint("Testing steps: {} rewards {}: ".format(steps, rewards))

# keras.model的自定义子类
class MyModel(tf.keras.Model):def __init__(self, num_states, hidden_units, num_actions):super(MyModel, self).__init__()self.input_layer = tf.keras.layers.InputLayer(input_shape=(num_states,))self.hidden_layers = []for i in hidden_units:self.hidden_layers.append(tf.keras.layers.Dense(i, activation='tanh', kernel_initializer='RandomNormal'))self.output_layer = tf.keras.layers.Dense(num_actions, activation='linear', kernel_initializer='RandomNormal')@tf.functiondef call(self, inputs):z = self.input_layer(inputs)for layer in self.hidden_layers:z = layer(z)output = self.output_layer(z)return outputdef predict(self, inputs):return self(np.atleast_2d(inputs.astype('float32')))

import numpy as np
import tensorflow as tf
import gym
import os
import datetime
from statistics import mean
from gym import wrappers
import pickleclass DQN:def __init__(self, num_states, num_actions, hidden_units, gamma, max_experiences, min_experiences,batch_size, lr,e_greedy, replace_target_iter = 50,e_greedy_increment=None):self.num_actions = num_actionsself.batch_size = batch_sizeself.optimizer = tf.optimizers.Adam(lr)self.gamma = gammaself.model = MyModel(num_states, hidden_units, num_actions)self.target = MyModel(num_states, hidden_units, num_actions)self.experience = {'s': [], 'a': [], 'r': [], 's2': [], 'done': []} #字典形式的记忆库self.max_experiences = max_experiencesself.min_experiences = min_experiencesself.learn_step_counter = 0self.replace_target_iter = replace_target_iterself.epsilon_increment = e_greedy_incrementself.epsilon_max = e_greedyself.epsilon = 0.1 if self.epsilon_increment is not None else self.epsilon_max# 以Target net作为参考训练def train(self):if self.learn_step_counter % self.replace_target_iter == 0:self.learn_step_counter = 0self.copy_weights()if len(self.experience['s']) < self.min_experiences:return 0 # 0 代表啥也没学到ids = np.random.randint(low=0, high=len(self.experience['s']), size=self.batch_size)# {'s': array([-0.02874029, -0.58186904,  0.00809103,  0.89257765]), # 'a': 1, 'r': 1.0, 's2': array([-0.04037767, -0.38685777,  0.02594258,  0.60244905]), 'done': False}# {'s': array([-0.13791506, -0.78732364,  0.19086538,  1.4356326 ]),#  'a': 0, 'r': -200, 's2': array([-0.15366153, -0.98421693,  0.21957803,  1.78138363]), 'done': True}#  state = x, x_dot, theta, theta_dotstates = np.asarray([self.experience['s'][i] for i in ids])actions = np.asarray([self.experience['a'][i] for i in ids])rewards = np.asarray([self.experience['r'][i] for i in ids])states_next = np.asarray([self.experience['s2'][i] for i in ids])dones = np.asarray([self.experience['done'][i] for i in ids])value_next = np.max(self.target.predict(states_next), axis=1)# np.where(condition, x, y)# 满足条件(condition)，输出x，不满足输出y。actual_values = np.where(dones, rewards, rewards+self.gamma*value_next)with tf.GradientTape() as tape:selected_action_values = tf.math.reduce_sum(self.model.predict(states) * tf.one_hot(actions, self.num_actions), axis=1)loss = tf.math.reduce_mean(tf.square(actual_values - selected_action_values))variables = self.model.trainable_variablesgradients = tape.gradient(loss, variables)self.optimizer.apply_gradients(zip(gradients, variables))self.learn_step_counter += 1   return loss # 选取行动def get_action(self, states):if np.random.random() > self.epsilon:return np.random.choice(self.num_actions)else:return np.argmax(self.model.predict(np.atleast_2d(states))[0])# 记忆库def add_experience(self, exp):if len(self.experience['s']) >= self.max_experiences:for key in self.experience.keys():self.experience[key].pop(0) # 满了就删除一个for key, value in exp.items():self.experience[key].append(value)# 更新fronzen/target netdef copy_weights(self):variables1 = self.target.trainable_variablesvariables2 = self.model.trainable_variablesfor v1, v2 in zip(variables1, variables2):v1.assign(v2.numpy())def epsilon_incre(self):self.epsilon = self.epsilon * self.epsilon_increment if self.epsilon < self.epsilon_max else self.epsilon_maxdef saveModel(self):self.model.save_weights(r'D:桌面RLDQNeval_net.h5')self.target.save_weights(r'D:桌面RLDQNtarget_net.h5')print('Weights Saved.')def loadModel(self):# 模型没有被调用过则没有权重，调用一下生成初始权重self.model.predict(np.array([1,1,1,1]))self.target.predict(np.array([1,1,1,1]))self.model.load_weights(r'D:桌面RLDQNeval_net.h5')self.target.load_weights(r'D:桌面RLDQNtarget_net.h5')print('Weights loaded.')def saveMemory(self):f = open(r'D:桌面RLMemory.data', 'wb')# 将变量存储到目标文件中区pickle.dump(self.experience, f)# 关闭文件f.close()print('Memory saved.')def loadMemory(self):fr = open(r'D:桌面RLMemory.data','rb')self.experience  = pickle.load(fr)print('Memory loaded.')

tensorboard --logdir=20201112-111458
#或者
#tensorboard --logdir=D:RLlogsdqn20201112-111458