Today I started to learn a new RL method: Actor-Critic. The basic idea is quite similar to GAN network. Unlike Policy Gradient in which a single neural network is trained, Actor-Critic uses two neural networks: an "actor" for performing actions and a "critic" for evaluating action of the actor. I refer to this page for details of this method.
Model presentation
Given a state s_t at time t, the network calculates the action probabilities and chooses an action a_t to apply on env:
After the application of action a_t, then env updated accordingly, yields reward r_t, we observe at next time step t+1 a new state s_t+1 which results in a new sampled action a_t+1:
Steps of the basic Actor-Critic method
Unlike the Policy Gradient which needs to gather all data of an episode to calculate gradients, Actor-Critic method performs model update in every agent-env interaction loop :
Implementation with CartPole game
I followed this youtube video for my first Actor-Critic exercise. It was initially based on Keras thus I have modified a little bit of his code for using tensorflow. I also disabled eager mode because of a runtime error (I am using Tensorflow v2 on my laptop).
import tensorflow as tf tf.compat.v1.disable_eager_execution() from tensorflow.keras import backend as K from tensorflow.keras.layers import Activation, Dense, Input from tensorflow.keras.models import Model from tensorflow.keras.optimizers import Adam import numpy as np import gym class Agent(object): def __init__(self, alpha, beta, gamma=0.99, n_actions=2, \ layer1_size=1024, layer2_size=512, input_dims=4): self.gamma = gamma self.alpha = alpha self.beta = beta self.input_dims = input_dims self.fc1_dims = layer1_size self.fc2_dims = layer2_size self.n_actions = n_actions self.actor, self.critic, self.policy = self.build_network() self.action_space = [i for i in range(self.n_actions)] def build_network(self): input = Input(shape=(self.input_dims, )) delta = Input(shape=[1]) dense1 = Dense(self.fc1_dims, activation="relu")(input) dense2 = Dense(self.fc2_dims, activation="relu")(dense1) probs = Dense(self.n_actions, activation="softmax")(dense2) values = Dense(1, activation="linear")(dense2) def custom_loss(y_true, y_pred): out = K.clip(y_pred, 1e-8, 1-1e-8) log_lik = y_true * K.log(out) return K.sum(-log_lik*delta) actor = Model(inputs=[input, delta], outputs=[probs]) actor.compile(optimizer=Adam(lr=self.alpha), loss=custom_loss) critic = Model(inputs=[input], outputs=[values]) critic.compile(optimizer=Adam(lr=self.beta), loss="mse") policy = Model(inputs=[input], outputs=[probs]) return actor, critic, policy def choose_action(self, observation): state = observation[np.newaxis, :] probs = self.policy.predict(state)[0] action = np.random.choice(self.action_space, p=probs) return action def learn(self, state, action, reward, state_, done): state = state[np.newaxis, :] state_ = state_[np.newaxis, :] critic_value_ = self.critic.predict(state_) critic_value = self.critic.predict(state) target = reward + self.gamma*critic_value_*(1-int(done)) delta = target - critic_value actions = np.zeros([1, self.n_actions]) actions[np.arange(1), action] = 1.0 self.actor.fit([state, delta], actions, verbose=0) self.critic.fit(state, target, verbose=0) if __name__ == "__main__": agent=Agent(alpha=0.0001, beta=0.0005) env = gym.make("CartPole-v0") score_history = [] num_episodes = 2000 for i in range(num_episodes): done = False score = 0 observation = env.reset() while not done: env.render() action = agent.choose_action(observation) observation_, reward, done, info= env.step(action) agent.learn(observation, action, reward, observation_, done) observation = observation_ score+=reward score_history.append(score) avg_score = np.mean(score_history[-100:]) print("episode ", i, "score %.2f average score %.2f" % (score, avg_score)) A newer version
I have updated the program shown above to allow eager mode. The main difference is the introduction of the custom loss function for actor. It works well, however, the running speed is much slower after my modification (I am looking forward to Refactoring it later).
import os os.environ['TF_CPP_MIN_LOG_LEVEL'] = "2" import tensorflow as tf if tf.__version__.startswith("1."): raise RuntimeError("Error!! You are using tensorflow-v1") from tensorflow.keras import backend as K from tensorflow.keras.layers import Activation, Dense, Input from tensorflow.keras.models import Model from tensorflow.keras.optimizers import Adam import numpy as np import gym class Agent(object): def __init__(self, alpha, beta, gamma=0.99, n_actions=2, \ layer1_size=1024, layer2_size=512, input_dims=4): self.gamma = gamma self.alpha = alpha self.beta = beta self.input_dims = input_dims self.fc1_dims = layer1_size self.fc2_dims = layer2_size self.n_actions = n_actions self.actor, self.critic, self.policy = self.build_network() self.action_space = [i for i in range(self.n_actions)] self.optimizer = tf.keras.optimizers.Adam(learning_rate=self.alpha) def build_network(self): input = Input(shape=(self.input_dims, )) delta = Input(shape=[1]) dense1 = Dense(self.fc1_dims, activation="relu")(input) dense2 = Dense(self.fc2_dims, activation="relu")(dense1) probs = Dense(self.n_actions, activation="softmax")(dense2) values = Dense(1, activation="linear")(dense2) actor = Model(inputs=[input, delta], outputs=[probs]) critic = Model(inputs=[input], outputs=[values]) critic.compile(optimizer=Adam(lr=self.beta), loss="mse") policy = Model(inputs=[input], outputs=[probs]) return actor, critic, policy def choose_action(self, observation): state = observation[np.newaxis, :] probs = self.policy.predict(state)[0] action = np.random.choice(self.action_space, p=probs) return action def learn(self, state, action, reward, state_, done): state = state[np.newaxis, :] state_ = state_[np.newaxis, :] critic_value_ = self.critic.predict(state_) critic_value = self.critic.predict(state) target = reward + self.gamma*critic_value_*(1-int(done)) delta = target - critic_value actions = np.zeros([1, self.n_actions]) actions[np.arange(1), action] = 1.0 self.critic.fit(state, target, verbose=0) with tf.GradientTape() as tape: y_pred = self.actor(state) out = K.clip(y_pred, 1e-8, 1-1e-8) log_lik = actions * K.log(out) myloss = K.sum(-log_lik*delta) grads = tape.gradient(myloss, self.actor.trainable_variables) self.optimizer.apply_gradients(zip(grads, self.actor.trainable_variables)) if __name__ == "__main__": agent=Agent(alpha=0.0001, beta=0.0005) ENV_SEED = 1024 ## Reproducibility of the game NP_SEED = 1024 ## Reproducibility of numpy random env = gym.make('CartPole-v0') env = env.unwrapped # use unwrapped version, otherwise episodes will terminate after 200 steps env.seed(ENV_SEED) np.random.seed(NP_SEED) ### The Discrete space allows a fixed range of non-negative numbers, so in this case valid actions are either 0 or 1. print(env.action_space) ### The Box space represents an n-dimensional box, so valid observations will be an array of 4 numbers. print(env.observation_space) ### We can also check the Box’s bounds: print(env.observation_space.high) print(env.observation_space.low) score_history = [] num_episodes = 2000 for i in range(num_episodes): done = False score = 0 observation = env.reset() while not done: env.render() action = agent.choose_action(observation) observation_, reward, done, info= env.step(action) agent.learn(observation, action, reward, observation_, done) observation = observation_ score+=reward score_history.append(score) avg_score = np.mean(score_history[-100:]) print("episode ", i, "score %.2f average score %.2f" % (score, avg_score)) The training converges ideally. Below is a snapshot of the execution output:
References:
*https://github.com/wangshusen/DRL
*https://www.youtube.com/watch?v=2vJtbAha3To
Top comments (1)
Hi! I have the same implementation however it takes forever to converge. It does converge but it takes more than 24 hours often. Do you have the same? I have the same code als your first example here :) .
Thanks in advance!