import tensorflow as tf
import numpy as np
import gym
import time
import random
from collections import deque#####################  hyper parameters  ##################### Hyper Parameters for DQN
GAMMA = 0.9 # discount factor for target Q
INITIAL_EPSILON = 0.5 # starting value of epsilon
FINAL_EPSILON = 0.01 # final value of epsilon
REPLAY_SIZE = 10000 # experience replay buffer size
BATCH_SIZE = 32 # size of minibatch###############################  DQN  ####################################class DQN():# DQN Agentdef __init__(self, env):# init experience replayself.replay_buffer = deque()# init some parametersself.time_step = 0self.epsilon = INITIAL_EPSILONself.state_dim = env.observation_space.shape[0]self.action_dim = env.action_space.nself.create_Q_network()self.create_training_method()# Init sessionself.session = tf.InteractiveSession()self.session.run(tf.global_variables_initializer())def create_Q_network(self):# network weightsW1 = self.weight_variable([self.state_dim,20])b1 = self.bias_variable([20])W2 = self.weight_variable([20,self.action_dim])b2 = self.bias_variable([self.action_dim])# input layerself.state_input = tf.placeholder("float",[None,self.state_dim])# hidden layersh_layer = tf.nn.relu(tf.matmul(self.state_input,W1) + b1)# Q Value layerself.Q_value = tf.matmul(h_layer,W2) + b2def create_training_method(self):self.action_input = tf.placeholder("float",[None,self.action_dim]) # one hot presentationself.y_input = tf.placeholder("float",[None])Q_action = tf.reduce_sum(tf.multiply(self.Q_value,self.action_input),reduction_indices = 1)self.cost = tf.reduce_mean(tf.square(self.y_input - Q_action))self.optimizer = tf.train.AdamOptimizer(0.0001).minimize(self.cost)def perceive(self,state,action,reward,next_state,done):one_hot_action = np.zeros(self.action_dim)one_hot_action[action] = 1self.replay_buffer.append((state,one_hot_action,reward,next_state,done))if len(self.replay_buffer) > REPLAY_SIZE:self.replay_buffer.popleft()if len(self.replay_buffer) > BATCH_SIZE:self.train_Q_network()def train_Q_network(self):self.time_step += 1# Step 1: obtain random minibatch from replay memoryminibatch = random.sample(self.replay_buffer,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]# Step 2: calculate yy_batch = []Q_value_batch = self.Q_value.eval(feed_dict={self.state_input:next_state_batch})for i in range(0,BATCH_SIZE):done = minibatch[i][4]if done:y_batch.append(reward_batch[i])else :y_batch.append(reward_batch[i] + GAMMA * np.max(Q_value_batch[i]))self.optimizer.run(feed_dict={self.y_input:y_batch,self.action_input:action_batch,self.state_input:state_batch})def egreedy_action(self,state):Q_value = self.Q_value.eval(feed_dict = {self.state_input:[state]})[0]if random.random() <= self.epsilon:self.epsilon -= (INITIAL_EPSILON - FINAL_EPSILON) / 10000return random.randint(0,self.action_dim - 1)else:self.epsilon -= (INITIAL_EPSILON - FINAL_EPSILON) / 10000return np.argmax(Q_value)def action(self,state):return np.argmax(self.Q_value.eval(feed_dict = {self.state_input:[state]})[0])def weight_variable(self,shape):initial = tf.truncated_normal(shape)return tf.Variable(initial)def bias_variable(self,shape):initial = tf.constant(0.01, shape = shape)return tf.Variable(initial)

from DQN import DQN
import gym
import numpy as np
import timeENV_NAME = 'CartPole-v1'
EPISODE = 3000 # Episode limitation
STEP = 300 # Step limitation in an episode
TEST = 10 # The number of experiment test every 100 episodedef main():# initialize OpenAI Gym env and dqn agentenv = gym.make(ENV_NAME)agent = DQN(env)for episode in range(EPISODE):# initialize taskstate = env.reset()# Trainep_reward = 0for step in range(STEP):action = agent.egreedy_action(state) # e-greedy action for trainnext_state,reward,done,_ = env.step(action)# Define reward for agentreward = -10 if done else 1ep_reward += rewardagent.perceive(state,action,reward,next_state,done)state = next_stateif done:#print('episode complete, reward: ', ep_reward)break# Test every 100 episodesif episode % 100 == 0:total_reward = 0for i in range(TEST):state = env.reset()for j in range(STEP):#env.render()action = agent.action(state) # direct action for teststate,reward,done,_ = env.step(action)total_reward += rewardif done:breakave_reward = total_reward/TESTprint ('episode: ',episode,'Evaluation Average Reward:',ave_reward)if __name__ == '__main__':main()