# # Unity ML Agents # ## ML-Agent Learning (PPO) # Contains an implementation of PPO as described (https://arxiv.org/abs/1707.06347). import logging import os import numpy as np import tensorflow as tf from unityagents import AllBrainInfo from unitytrainers.buffer import Buffer from unitytrainers.ppo.models import PPOModel from unitytrainers.trainer import UnityTrainerException, Trainer logger = logging.getLogger("unityagents") class PPOTrainer(Trainer): """The PPOTrainer is an implementation of the PPO algorithm.""" def __init__(self, sess, env, brain_name, trainer_parameters, training, seed): """ Responsible for collecting experiences and training PPO model. :param sess: Tensorflow session. :param env: The UnityEnvironment. :param trainer_parameters: The parameters for the trainer (dictionary). :param training: Whether the trainer is set for training. """ self.param_keys = ['batch_size', 'beta', 'buffer_size', 'epsilon', 'gamma', 'hidden_units', 'lambd', 'learning_rate', 'max_steps', 'normalize', 'num_epoch', 'num_layers', 'time_horizon', 'sequence_length', 'summary_freq', 'use_recurrent', 'graph_scope', 'summary_path', 'memory_size', 'use_curiosity', 'curiosity_strength', 'curiosity_enc_size'] for k in self.param_keys: if k not in trainer_parameters: raise UnityTrainerException("The hyperparameter {0} could not be found for the PPO trainer of " "brain {1}.".format(k, brain_name)) super(PPOTrainer, self).__init__(sess, env, brain_name, trainer_parameters, training) self.use_recurrent = trainer_parameters["use_recurrent"] self.use_curiosity = bool(trainer_parameters['use_curiosity']) self.sequence_length = 1 self.step = 0 self.has_updated = False self.m_size = None if self.use_recurrent: self.m_size = trainer_parameters["memory_size"] self.sequence_length = trainer_parameters["sequence_length"] if self.m_size == 0: raise UnityTrainerException("The memory size for brain {0} is 0 even though the trainer uses recurrent." .format(brain_name)) elif self.m_size % 4 != 0: raise UnityTrainerException("The memory size for brain {0} is {1} but it must be divisible by 4." .format(brain_name, self.m_size)) self.variable_scope = trainer_parameters['graph_scope'] with tf.variable_scope(self.variable_scope): tf.set_random_seed(seed) self.model = PPOModel(env.brains[brain_name], lr=float(trainer_parameters['learning_rate']), h_size=int(trainer_parameters['hidden_units']), epsilon=float(trainer_parameters['epsilon']), beta=float(trainer_parameters['beta']), max_step=float(trainer_parameters['max_steps']), normalize=trainer_parameters['normalize'], use_recurrent=trainer_parameters['use_recurrent'], num_layers=int(trainer_parameters['num_layers']), m_size=self.m_size, use_curiosity=bool(trainer_parameters['use_curiosity']), curiosity_strength=float(trainer_parameters['curiosity_strength']), curiosity_enc_size=float(trainer_parameters['curiosity_enc_size'])) stats = {'cumulative_reward': [], 'episode_length': [], 'value_estimate': [], 'entropy': [], 'value_loss': [], 'policy_loss': [], 'learning_rate': []} if self.use_curiosity: stats['forward_loss'] = [] stats['inverse_loss'] = [] stats['intrinsic_reward'] = [] self.intrinsic_rewards = {} self.stats = stats self.training_buffer = Buffer() self.cumulative_rewards = {} self.episode_steps = {} self.is_continuous_action = (env.brains[brain_name].vector_action_space_type == "continuous") self.is_continuous_observation = (env.brains[brain_name].vector_observation_space_type == "continuous") self.use_visual_obs = (env.brains[brain_name].number_visual_observations > 0) self.use_vector_obs = (env.brains[brain_name].vector_observation_space_size > 0) self.summary_path = trainer_parameters['summary_path'] if not os.path.exists(self.summary_path): os.makedirs(self.summary_path) self.summary_writer = tf.summary.FileWriter(self.summary_path) self.inference_run_list = [self.model.output, self.model.all_probs, self.model.value, self.model.entropy, self.model.learning_rate] if self.is_continuous_action: self.inference_run_list.append(self.model.output_pre) if self.use_recurrent: self.inference_run_list.extend([self.model.memory_out]) if (self.is_training and self.is_continuous_observation and self.use_vector_obs and self.trainer_parameters['normalize']): self.inference_run_list.extend([self.model.update_mean, self.model.update_variance]) def __str__(self): return '''Hyperparameters for the PPO Trainer of brain {0}: \n{1}'''.format( self.brain_name, '\n'.join(['\t{0}:\t{1}'.format(x, self.trainer_parameters[x]) for x in self.param_keys])) @property def parameters(self): """ Returns the trainer parameters of the trainer. """ return self.trainer_parameters @property def graph_scope(self): """ Returns the graph scope of the trainer. """ return self.variable_scope @property def get_max_steps(self): """ Returns the maximum number of steps. Is used to know when the trainer should be stopped. :return: The maximum number of steps of the trainer """ return float(self.trainer_parameters['max_steps']) @property def get_step(self): """ Returns the number of steps the trainer has performed :return: the step count of the trainer """ return self.step @property def get_last_reward(self): """ Returns the last reward the trainer has had :return: the new last reward """ return self.sess.run(self.model.last_reward) def increment_step_and_update_last_reward(self): """ Increment the step count of the trainer and Updates the last reward """ if len(self.stats['cumulative_reward']) > 0: mean_reward = np.mean(self.stats['cumulative_reward']) self.sess.run([self.model.update_reward, self.model.increment_step], feed_dict={self.model.new_reward: mean_reward}) else: self.sess.run(self.model.increment_step) self.step = self.sess.run(self.model.global_step) def take_action(self, all_brain_info: AllBrainInfo): """ Decides actions given observations information, and takes them in environment. :param all_brain_info: A dictionary of brain names and BrainInfo from environment. :return: a tuple containing action, memories, values and an object to be passed to add experiences """ curr_brain_info = all_brain_info[self.brain_name] if len(curr_brain_info.agents) == 0: return [], [], [], None feed_dict = {self.model.batch_size: len(curr_brain_info.vector_observations), self.model.sequence_length: 1} if self.use_recurrent: if not self.is_continuous_action: feed_dict[self.model.prev_action] = curr_brain_info.previous_vector_actions.flatten() if curr_brain_info.memories.shape[1] == 0: curr_brain_info.memories = np.zeros((len(curr_brain_info.agents), self.m_size)) feed_dict[self.model.memory_in] = curr_brain_info.memories if self.use_visual_obs: for i, _ in enumerate(curr_brain_info.visual_observations): feed_dict[self.model.visual_in[i]] = curr_brain_info.visual_observations[i] if self.use_vector_obs: feed_dict[self.model.vector_in] = curr_brain_info.vector_observations values = self.sess.run(self.inference_run_list, feed_dict=feed_dict) run_out = dict(zip(self.inference_run_list, values)) self.stats['value_estimate'].append(run_out[self.model.value].mean()) self.stats['entropy'].append(run_out[self.model.entropy].mean()) self.stats['learning_rate'].append(run_out[self.model.learning_rate]) if self.use_recurrent: return run_out[self.model.output], run_out[self.model.memory_out], None, run_out else: return run_out[self.model.output], None, None, run_out def generate_intrinsic_rewards(self, curr_info, next_info): """ Generates intrinsic reward used for Curiosity-based training. :param curr_info: Current BrainInfo. :param next_info: Next BrainInfo. :return: Intrinsic rewards for all agents. """ if self.use_curiosity: if curr_info.agents != next_info.agents: raise UnityTrainerException("Training with Curiosity-driven exploration" " and On-Demand Decision making is currently not supported.") feed_dict = {self.model.batch_size: len(curr_info.vector_observations), self.model.sequence_length: 1} if self.is_continuous_action: feed_dict[self.model.output] = next_info.previous_vector_actions else: feed_dict[self.model.action_holder] = next_info.previous_vector_actions.flatten() if self.use_visual_obs: for i in range(len(curr_info.visual_observations)): feed_dict[self.model.visual_in[i]] = curr_info.visual_observations[i] feed_dict[self.model.next_visual_in[i]] = next_info.visual_observations[i] if self.use_vector_obs: feed_dict[self.model.vector_in] = curr_info.vector_observations feed_dict[self.model.next_vector_in] = next_info.vector_observations if self.use_recurrent: if curr_info.memories.shape[1] == 0: curr_info.memories = np.zeros((len(curr_info.agents), self.m_size)) feed_dict[self.model.memory_in] = curr_info.memories intrinsic_rewards = self.sess.run(self.model.intrinsic_reward, feed_dict=feed_dict) * float(self.has_updated) return intrinsic_rewards else: return None def generate_value_estimate(self, brain_info, idx): """ Generates value estimates for bootstrapping. :param brain_info: BrainInfo to be used for bootstrapping. :param idx: Index in BrainInfo of agent. :return: Value estimate. """ feed_dict = {self.model.batch_size: 1, self.model.sequence_length: 1} if self.use_visual_obs: for i in range(len(brain_info.visual_observations)): feed_dict[self.model.visual_in[i]] = [brain_info.visual_observations[i][idx]] if self.use_vector_obs: feed_dict[self.model.vector_in] = [brain_info.vector_observations[idx]] if self.use_recurrent: if brain_info.memories.shape[1] == 0: brain_info.memories = np.zeros( (len(brain_info.vector_observations), self.m_size)) feed_dict[self.model.memory_in] = [brain_info.memories[idx]] if not self.is_continuous_action and self.use_recurrent: feed_dict[self.model.prev_action] = brain_info.previous_vector_actions[idx].flatten() value_estimate = self.sess.run(self.model.value, feed_dict) return value_estimate def add_experiences(self, curr_all_info: AllBrainInfo, next_all_info: AllBrainInfo, take_action_outputs): """ Adds experiences to each agent's experience history. :param curr_all_info: Dictionary of all current brains and corresponding BrainInfo. :param next_all_info: Dictionary of all current brains and corresponding BrainInfo. :param take_action_outputs: The outputs of the take action method. """ curr_info = curr_all_info[self.brain_name] next_info = next_all_info[self.brain_name] intrinsic_rewards = self.generate_intrinsic_rewards(curr_info, next_info) for agent_id in curr_info.agents: self.training_buffer[agent_id].last_brain_info = curr_info self.training_buffer[agent_id].last_take_action_outputs = take_action_outputs for agent_id in next_info.agents: stored_info = self.training_buffer[agent_id].last_brain_info stored_take_action_outputs = self.training_buffer[agent_id].last_take_action_outputs if stored_info is not None: idx = stored_info.agents.index(agent_id) next_idx = next_info.agents.index(agent_id) if not stored_info.local_done[idx]: if self.use_visual_obs: for i, _ in enumerate(stored_info.visual_observations): self.training_buffer[agent_id]['visual_obs%d' % i].append( stored_info.visual_observations[i][idx]) self.training_buffer[agent_id]['next_visual_obs%d' % i].append( next_info.visual_observations[i][idx]) if self.use_vector_obs: self.training_buffer[agent_id]['vector_obs'].append(stored_info.vector_observations[idx]) self.training_buffer[agent_id]['next_vector_in'].append( next_info.vector_observations[next_idx]) if self.use_recurrent: if stored_info.memories.shape[1] == 0: stored_info.memories = np.zeros((len(stored_info.agents), self.m_size)) self.training_buffer[agent_id]['memory'].append(stored_info.memories[idx]) actions = stored_take_action_outputs[self.model.output] if self.is_continuous_action: actions_pre = stored_take_action_outputs[self.model.output_pre] self.training_buffer[agent_id]['actions_pre'].append(actions_pre[idx]) a_dist = stored_take_action_outputs[self.model.all_probs] value = stored_take_action_outputs[self.model.value] self.training_buffer[agent_id]['actions'].append(actions[idx]) self.training_buffer[agent_id]['prev_action'].append(stored_info.previous_vector_actions[idx]) self.training_buffer[agent_id]['masks'].append(1.0) if self.use_curiosity: self.training_buffer[agent_id]['rewards'].append(next_info.rewards[next_idx] + intrinsic_rewards[next_idx]) else: self.training_buffer[agent_id]['rewards'].append(next_info.rewards[next_idx]) self.training_buffer[agent_id]['action_probs'].append(a_dist[idx]) self.training_buffer[agent_id]['value_estimates'].append(value[idx][0]) if agent_id not in self.cumulative_rewards: self.cumulative_rewards[agent_id] = 0 self.cumulative_rewards[agent_id] += next_info.rewards[next_idx] if self.use_curiosity: if agent_id not in self.intrinsic_rewards: self.intrinsic_rewards[agent_id] = 0 self.intrinsic_rewards[agent_id] += intrinsic_rewards[next_idx] if not next_info.local_done[next_idx]: if agent_id not in self.episode_steps: self.episode_steps[agent_id] = 0 self.episode_steps[agent_id] += 1 def process_experiences(self, current_info: AllBrainInfo, new_info: AllBrainInfo): """ Checks agent histories for processing condition, and processes them as necessary. Processing involves calculating value and advantage targets for model updating step. :param current_info: Dictionary of all current brains and corresponding BrainInfo. :param new_info: Dictionary of all next brains and corresponding BrainInfo. """ info = new_info[self.brain_name] for l in range(len(info.agents)): agent_actions = self.training_buffer[info.agents[l]]['actions'] if ((info.local_done[l] or len(agent_actions) > self.trainer_parameters['time_horizon']) and len(agent_actions) > 0): agent_id = info.agents[l] if info.local_done[l] and not info.max_reached[l]: value_next = 0.0 else: if info.max_reached[l]: bootstrapping_info = self.training_buffer[agent_id].last_brain_info idx = bootstrapping_info.agents.index(agent_id) else: bootstrapping_info = info idx = l value_next = self.generate_value_estimate(bootstrapping_info, idx) self.training_buffer[agent_id]['advantages'].set( get_gae( rewards=self.training_buffer[agent_id]['rewards'].get_batch(), value_estimates=self.training_buffer[agent_id]['value_estimates'].get_batch(), value_next=value_next, gamma=self.trainer_parameters['gamma'], lambd=self.trainer_parameters['lambd'])) self.training_buffer[agent_id]['discounted_returns'].set( self.training_buffer[agent_id]['advantages'].get_batch() + self.training_buffer[agent_id]['value_estimates'].get_batch()) self.training_buffer.append_update_buffer(agent_id, batch_size=None, training_length=self.sequence_length) self.training_buffer[agent_id].reset_agent() if info.local_done[l]: self.stats['cumulative_reward'].append( self.cumulative_rewards.get(agent_id, 0)) self.stats['episode_length'].append( self.episode_steps.get(agent_id, 0)) self.cumulative_rewards[agent_id] = 0 self.episode_steps[agent_id] = 0 if self.use_curiosity: self.stats['intrinsic_reward'].append( self.intrinsic_rewards.get(agent_id, 0)) self.intrinsic_rewards[agent_id] = 0 def end_episode(self): """ A signal that the Episode has ended. The buffer must be reset. Get only called when the academy resets. """ self.training_buffer.reset_all() for agent_id in self.cumulative_rewards: self.cumulative_rewards[agent_id] = 0 for agent_id in self.episode_steps: self.episode_steps[agent_id] = 0 if self.use_curiosity: for agent_id in self.intrinsic_rewards: self.intrinsic_rewards[agent_id] = 0 def is_ready_update(self): """ Returns whether or not the trainer has enough elements to run update model :return: A boolean corresponding to whether or not update_model() can be run """ size_of_buffer = len(self.training_buffer.update_buffer['actions']) return size_of_buffer > max(int(self.trainer_parameters['buffer_size'] / self.sequence_length), 1) def update_model(self): """ Uses training_buffer to update model. """ n_sequences = max(int(self.trainer_parameters['batch_size'] / self.sequence_length), 1) value_total, policy_total, forward_total, inverse_total = [], [], [], [] advantages = self.training_buffer.update_buffer['advantages'].get_batch() self.training_buffer.update_buffer['advantages'].set( (advantages - advantages.mean()) / (advantages.std() + 1e-10)) num_epoch = self.trainer_parameters['num_epoch'] for k in range(num_epoch): self.training_buffer.update_buffer.shuffle() buffer = self.training_buffer.update_buffer for l in range(len(self.training_buffer.update_buffer['actions']) // n_sequences): start = l * n_sequences end = (l + 1) * n_sequences feed_dict = {self.model.batch_size: n_sequences, self.model.sequence_length: self.sequence_length, self.model.mask_input: np.array(buffer['masks'][start:end]).flatten(), self.model.returns_holder: np.array(buffer['discounted_returns'][start:end]).flatten(), self.model.old_value: np.array(buffer['value_estimates'][start:end]).flatten(), self.model.advantage: np.array(buffer['advantages'][start:end]).reshape([-1, 1]), self.model.all_old_probs: np.array(buffer['action_probs'][start:end]).reshape( [-1, self.brain.vector_action_space_size])} if self.is_continuous_action: feed_dict[self.model.output_pre] = np.array(buffer['actions_pre'][start:end]).reshape( [-1, self.brain.vector_action_space_size]) else: feed_dict[self.model.action_holder] = np.array(buffer['actions'][start:end]).flatten() if self.use_recurrent: feed_dict[self.model.prev_action] = np.array(buffer['prev_action'][start:end]).flatten() if self.use_vector_obs: if self.is_continuous_observation: total_observation_length = self.brain.vector_observation_space_size * \ self.brain.num_stacked_vector_observations feed_dict[self.model.vector_in] = np.array(buffer['vector_obs'][start:end]).reshape( [-1, total_observation_length]) if self.use_curiosity: feed_dict[self.model.next_vector_in] = np.array(buffer['next_vector_in'][start:end]) \ .reshape([-1, total_observation_length]) else: feed_dict[self.model.vector_in] = np.array(buffer['vector_obs'][start:end]).reshape( [-1, self.brain.num_stacked_vector_observations]) if self.use_curiosity: feed_dict[self.model.next_vector_in] = np.array(buffer['next_vector_in'][start:end]) \ .reshape([-1, self.brain.num_stacked_vector_observations]) if self.use_visual_obs: for i, _ in enumerate(self.model.visual_in): _obs = np.array(buffer['visual_obs%d' % i][start:end]) if self.sequence_length > 1 and self.use_recurrent: (_batch, _seq, _w, _h, _c) = _obs.shape feed_dict[self.model.visual_in[i]] = _obs.reshape([-1, _w, _h, _c]) else: feed_dict[self.model.visual_in[i]] = _obs if self.use_curiosity: for i, _ in enumerate(self.model.visual_in): _obs = np.array(buffer['next_visual_obs%d' % i][start:end]) if self.sequence_length > 1 and self.use_recurrent: (_batch, _seq, _w, _h, _c) = _obs.shape feed_dict[self.model.next_visual_in[i]] = _obs.reshape([-1, _w, _h, _c]) else: feed_dict[self.model.next_visual_in[i]] = _obs if self.use_recurrent: mem_in = np.array(buffer['memory'][start:end])[:, 0, :] feed_dict[self.model.memory_in] = mem_in run_list = [self.model.value_loss, self.model.policy_loss, self.model.update_batch] if self.use_curiosity: run_list.extend([self.model.forward_loss, self.model.inverse_loss]) values = self.sess.run(run_list, feed_dict=feed_dict) self.has_updated = True run_out = dict(zip(run_list, values)) value_total.append(run_out[self.model.value_loss]) policy_total.append(np.abs(run_out[self.model.policy_loss])) if self.use_curiosity: inverse_total.append(run_out[self.model.inverse_loss]) forward_total.append(run_out[self.model.forward_loss]) self.stats['value_loss'].append(np.mean(value_total)) self.stats['policy_loss'].append(np.mean(policy_total)) if self.use_curiosity: self.stats['forward_loss'].append(np.mean(forward_total)) self.stats['inverse_loss'].append(np.mean(inverse_total)) self.training_buffer.reset_update_buffer() def discount_rewards(r, gamma=0.99, value_next=0.0): """ Computes discounted sum of future rewards for use in updating value estimate. :param r: List of rewards. :param gamma: Discount factor. :param value_next: T+1 value estimate for returns calculation. :return: discounted sum of future rewards as list. """ discounted_r = np.zeros_like(r) running_add = value_next for t in reversed(range(0, r.size)): running_add = running_add * gamma + r[t] discounted_r[t] = running_add return discounted_r def get_gae(rewards, value_estimates, value_next=0.0, gamma=0.99, lambd=0.95): """ Computes generalized advantage estimate for use in updating policy. :param rewards: list of rewards for time-steps t to T. :param value_next: Value estimate for time-step T+1. :param value_estimates: list of value estimates for time-steps t to T. :param gamma: Discount factor. :param lambd: GAE weighing factor. :return: list of advantage estimates for time-steps t to T. """ value_estimates = np.asarray(value_estimates.tolist() + [value_next]) delta_t = rewards + gamma * value_estimates[1:] - value_estimates[:-1] advantage = discount_rewards(r=delta_t, gamma=gamma * lambd) return advantage