Ppo

mighty.mighty_agents.ppo

MightyPPOAgent

MightyPPOAgent(
    output_dir,
    env: MIGHTYENV,
    eval_env: Optional[MIGHTYENV] = None,
    seed: Optional[int] = None,
    learning_rate: float = 0.001,
    gamma: float = 0.99,
    batch_size: int = 64,
    learning_starts: int = 1,
    render_progress: bool = True,
    log_wandb: bool = False,
    wandb_kwargs: dict | None = None,
    log_infos: bool = False,
    rollout_buffer_class: Optional[
        str | DictConfig | Type[MightyRolloutBuffer]
    ] = MightyRolloutBuffer,
    rollout_buffer_kwargs: Optional[TypeKwargs] = {
        "buffer_size": 256
    },
    meta_methods: Optional[List[str | type]] = None,
    meta_kwargs: Optional[List[TypeKwargs]] = None,
    n_policy_units: int = 8,
    n_critic_units: int = 8,
    soft_update_weight: float = 0.01,
    policy_class: Optional[
        Union[
            str, DictConfig, Type[MightyExplorationPolicy]
        ]
    ] = None,
    policy_kwargs: Optional[Dict] = None,
    ppo_clip: float = 0.2,
    value_loss_coef: float = 0.5,
    entropy_coef: float = 0.01,
    max_grad_norm: float = 0.5,
    n_gradient_steps: int = 10,
    hidden_sizes: Optional[List[int]] = [64, 64],
    activation: Optional[str] = "tanh",
    n_epochs: int = 10,
    minibatch_size: int = 32,
    kl_target: float = 0.001,
    use_value_clip: bool = True,
    value_clip_eps: float = 0.2,
    total_timesteps: int = 1000000,
    normalize_obs: bool = False,
    normalize_reward: bool = False,
    rescale_action: bool = False,
    tanh_squash: bool = False,
)

Bases: MightyAgent

Creates all relevant class variables and calls the agent-specific init function.

:param env: Train environment :param eval_env: Evaluation environment :param seed: Seed for random number generators :param learning_rate: Learning rate for training :param gamma: Discount factor :param batch_size: Batch size for training :param learning_starts: Number of steps before learning starts :param render_progress: Whether to render progress :param log_tensorboard: Log to TensorBoard as well as to file :param log_wandb: Log to Weights and Biases :param wandb_kwargs: Arguments for Weights and Biases logging :param rollout_buffer_class: Rollout buffer class :param rollout_buffer_kwargs: Arguments for the rollout buffer :param meta_methods: Meta methods for the agent :param meta_kwargs: Arguments for meta methods :param n_policy_units: Number of units for the policy network :param n_critic_units: Number of units for the critic network :param soft_update_weight: Size of soft updates for the target network :param policy_class: Policy class :param policy_kwargs: Arguments for the policy :param ppo_clip: Clipping parameter for PPO :param value_loss_coef: Coefficient for the value loss :param entropy_coef: Coefficient for the entropy loss :param max_grad_norm: Maximum gradient norm :param n_gradient_steps: Number of gradient steps per update

Source code in mighty/mighty_agents/ppo.py

def __init__(
    self,
    output_dir,
    env: MIGHTYENV,  # type: ignore
    eval_env: Optional[MIGHTYENV] = None,  # type: ignore
    seed: Optional[int] = None,
    learning_rate: float = 0.001,
    gamma: float = 0.99,
    batch_size: int = 64,
    learning_starts: int = 1,
    render_progress: bool = True,
    log_wandb: bool = False,
    wandb_kwargs: dict | None = None,
    log_infos: bool = False,
    rollout_buffer_class: Optional[
        str | DictConfig | Type[MightyRolloutBuffer]
    ] = MightyRolloutBuffer,
    rollout_buffer_kwargs: Optional[TypeKwargs] = {
        "buffer_size": 256,
    },
    meta_methods: Optional[List[str | type]] = None,
    meta_kwargs: Optional[List[TypeKwargs]] = None,
    n_policy_units: int = 8,
    n_critic_units: int = 8,
    soft_update_weight: float = 0.01,
    policy_class: Optional[
        Union[str, DictConfig, Type[MightyExplorationPolicy]]
    ] = None,
    policy_kwargs: Optional[Dict] = None,
    ppo_clip: float = 0.2,
    value_loss_coef: float = 0.5,
    entropy_coef: float = 0.01,
    max_grad_norm: float = 0.5,
    n_gradient_steps: int = 10,
    hidden_sizes: Optional[List[int]] = [64, 64],
    activation: Optional[str] = "tanh",
    n_epochs: int = 10,
    minibatch_size: int = 32,
    kl_target: float = 0.001,
    use_value_clip: bool = True,
    value_clip_eps: float = 0.2,
    total_timesteps: int = 1_000_000,
    normalize_obs: bool = False,
    normalize_reward: bool = False,
    rescale_action: bool = False,
    tanh_squash: bool = False,
):
    """Initialize the PPO agent.

    Creates all relevant class variables and calls the agent-specific init function.

    :param env: Train environment
    :param eval_env: Evaluation environment
    :param seed: Seed for random number generators
    :param learning_rate: Learning rate for training
    :param gamma: Discount factor
    :param batch_size: Batch size for training
    :param learning_starts: Number of steps before learning starts
    :param render_progress: Whether to render progress
    :param log_tensorboard: Log to TensorBoard as well as to file
    :param log_wandb: Log to Weights and Biases
    :param wandb_kwargs: Arguments for Weights and Biases logging
    :param rollout_buffer_class: Rollout buffer class
    :param rollout_buffer_kwargs: Arguments for the rollout buffer
    :param meta_methods: Meta methods for the agent
    :param meta_kwargs: Arguments for meta methods
    :param n_policy_units: Number of units for the policy network
    :param n_critic_units: Number of units for the critic network
    :param soft_update_weight: Size of soft updates for the target network
    :param policy_class: Policy class
    :param policy_kwargs: Arguments for the policy
    :param ppo_clip: Clipping parameter for PPO
    :param value_loss_coef: Coefficient for the value loss
    :param entropy_coef: Coefficient for the entropy loss
    :param max_grad_norm: Maximum gradient norm
    :param n_gradient_steps: Number of gradient steps per update
    """

    self.total_timesteps = total_timesteps
    self.gamma = gamma
    self.n_policy_units = n_policy_units
    self.n_critic_units = n_critic_units
    self.soft_update_weight = soft_update_weight
    self.ppo_clip = ppo_clip
    self.value_loss_coef = value_loss_coef
    self.entropy_coef = entropy_coef
    self.max_grad_norm = max_grad_norm
    self.hidden_sizes = hidden_sizes
    self.activation = activation

    self.n_epochs = n_epochs
    self.minibatch_size = minibatch_size
    self.kl_target = kl_target
    self.use_value_clip = use_value_clip
    self.value_clip_eps = value_clip_eps
    self.tanh_squash = tanh_squash

    # Placeholder variables which are filled in self._initialize_agent
    self.model: PPOModel | None = None
    self.update_fn: PPOUpdate | None = None

    # Policy class
    policy_class = retrieve_class(cls=policy_class, default_cls=StochasticPolicy)  # type: ignore
    if policy_kwargs is None:
        policy_kwargs = {}
    self.policy_class = policy_class
    self.policy_kwargs = policy_kwargs

    super().__init__(
        env=env,
        output_dir=output_dir,
        seed=seed,
        eval_env=eval_env,
        learning_rate=learning_rate,
        batch_size=batch_size,
        learning_starts=learning_starts,
        n_gradient_steps=n_gradient_steps,
        render_progress=render_progress,
        log_wandb=log_wandb,
        wandb_kwargs=wandb_kwargs,
        replay_buffer_class=rollout_buffer_class,
        replay_buffer_kwargs=rollout_buffer_kwargs,
        meta_methods=meta_methods,
        meta_kwargs=meta_kwargs,
        normalize_obs=normalize_obs,
        normalize_reward=normalize_reward,
        rescale_action=rescale_action,
        log_infos=log_infos,
    )

    self.loss_buffer = {
        "Update/policy_loss": [],
        "Update/value_loss": [],
        "Update/entropy": [],
        "Update/approx_kl": [],
        "update_at_step": [],
    }

parameters property

parameters: List[Parameter]

Return all trainable parameters (policy + value) for PPO.

value_function property

value_function: Module

Return the value function model.

__del__

__del__() -> None

Close wandb upon deletion.

Source code in mighty/mighty_agents/base_agent.py

def __del__(self) -> None:
    """Close wandb upon deletion."""
    self.env.close()  # type: ignore
    if self.log_wandb:
        wandb.finish()

adapt_hps

adapt_hps(metrics: Dict) -> None

Set hyperparameters.

Source code in mighty/mighty_agents/base_agent.py

def adapt_hps(self, metrics: Dict) -> None:
    """Set hyperparameters."""
    old_hps = {
        "hps_after_n_steps": self.steps,
        "hp/lr": self.learning_rate,
        "hp/pi_epsilon": self._epsilon,
        "hp/batch_size": self._batch_size,
        "hp/learning_starts": self._learning_starts,
        "meta_modules": list(self.meta_modules.keys()),
    }
    self.learning_rate = metrics["hp/lr"]
    self._epsilon = metrics["hp/pi_epsilon"]
    self._batch_size = metrics["hp/batch_size"]
    self._learning_starts = metrics["hp/learning_starts"]

    updated_hps = {
        "hps_after_n_steps": self.steps,
        "hp/lr": self.learning_rate,
        "hp/pi_epsilon": self._epsilon,
        "hp/batch_size": self._batch_size,
        "hp/learning_starts": self._learning_starts,
        "meta_modules": list(self.meta_modules.keys()),
    }

    if any(old_hps[k] != updated_hps[k] for k in old_hps.keys()):
        self.hp_buffer = update_buffer(self.hp_buffer, updated_hps)

apply_config

apply_config(config: Dict) -> None

Apply config to agent.

Source code in mighty/mighty_agents/base_agent.py

def apply_config(self, config: Dict) -> None:
    """Apply config to agent."""
    for n in config:
        algo_name = n.split(".")[-1]
        if hasattr(self, algo_name):
            setattr(self, algo_name, config[n])
        elif hasattr(self, "_" + algo_name):
            setattr(self, "_" + algo_name, config[n])
        elif n in ["architecture", "n_units", "n_layers", "size"]:
            pass
        else:
            print(f"Trying to set hyperparameter {algo_name} which does not exist.")

evaluate

evaluate(
    eval_env: MIGHTYENV | None = None,
    log_infos: bool = False,
) -> Dict

Eval agent on an environment. (Full rollouts).

:param env: The environment to evaluate on :param episodes: The number of episodes to evaluate :return:

Source code in mighty/mighty_agents/base_agent.py

def evaluate(
    self, eval_env: MIGHTYENV | None = None, log_infos: bool = False
) -> Dict:  # type: ignore
    """Eval agent on an environment. (Full rollouts).

    :param env: The environment to evaluate on
    :param episodes: The number of episodes to evaluate
    :return:
    """

    terminated, truncated = False, False
    options: Dict = {}
    if eval_env is None:
        eval_env = self.eval_env

    if self.eval_state is None:
        self.eval_state, _ = eval_env.reset(options=options, seed=self.seed)  # type: ignore

    rewards = np.zeros(eval_env.num_envs)  # type: ignore
    steps = np.zeros(eval_env.num_envs)  # type: ignore
    mask = np.zeros(eval_env.num_envs)  # type: ignore
    if log_infos:
        infos = {}
        info_mask = np.zeros(eval_env.num_envs)
        last_info = None
    while not np.all(mask):
        action = self.policy(self.eval_state, evaluate=True)  # type: ignore
        self.eval_state, reward, terminated, truncated, info = eval_env.step(action)  # type: ignore
        rewards += reward * (1 - mask)
        steps += 1 * (1 - mask)
        dones = np.logical_or(terminated, truncated)
        if log_infos:
            if last_info is None:
                last_info = info
            info_mask_copy = np.where(dones, 1, info_mask)
            info_mask_copy = np.where(info_mask_copy == 3, 0, info_mask_copy)
            for k in info.keys():
                if not k.startswith("_") and "final" not in k:
                    if k not in infos.keys():
                        infos[k] = []
                    final_infos = last_info[k][info_mask_copy.astype(bool)]
                    if len(final_infos) > 0:
                        infos[k] += final_infos.tolist()
            info_mask = np.where(dones, 3, info_mask)
            last_info = info
        mask = np.where(dones, 1, mask)

    if hasattr(self.eval_env, "unwrapped") and (
        isinstance(self.eval_env.unwrapped, DACENV)
        or isinstance(self.eval_env.unwrapped, CARLENV)
    ):
        instances = eval_env.inst_ids  # type: ignore
    else:
        instances = "None"

    eval_metrics = {
        "eval_after_n_steps": self.steps,
        "seed": self.seed,
        "eval_episodes": np.array(rewards) / steps,
        "mean_eval_step_reward": np.mean(rewards) / steps,
        "mean_eval_reward": np.mean(rewards),
        "instances": instances,
        "eval_rewards": rewards,
    }

    if log_infos:
        eval_metrics.update(infos)
        for k in eval_metrics.keys():
            if not k.startswith("_") and k not in self.eval_buffer.keys():
                self.eval_buffer[k] = []
    self.eval_buffer = update_buffer(self.eval_buffer, eval_metrics)

    if self.log_wandb:
        log_to_wandb(eval_metrics)

    return eval_metrics

initialize_agent

initialize_agent() -> None

General initialization of tracer and buffer for all agents.

Algorithm specific initialization like policies etc. are done in _initialize_agent

Source code in mighty/mighty_agents/base_agent.py

def initialize_agent(self) -> None:
    """General initialization of tracer and buffer for all agents.

    Algorithm specific initialization like policies etc.
    are done in _initialize_agent
    """
    self._initialize_agent()

    if isinstance(self.buffer_class, type) and issubclass(
        self.buffer_class, PrioritizedReplay
    ):
        if isinstance(self.buffer_kwargs, DictConfig):
            self.buffer_kwargs = OmegaConf.to_container(
                self.buffer_kwargs, resolve=True
            )
        # 1) Get observation-space shape
        try:
            obs_space = self.env.single_observation_space
            obs_shape = tuple(obs_space.shape)
        except Exception:
            # Fallback: call env.reset() once and infer shape from returned numpy/torch array
            first_obs, _ = self.env.reset(seed=self.seed)
            obs_shape = tuple(np.array(first_obs).shape)

        # 2) Get action-space shape (if discrete, .n is number of actions)
        action_space = self.env.single_action_space
        if hasattr(action_space, "n"):
            # Discrete action space → action_shape = () (scalar), but Q-net will expect a single integer
            # We store it as a zero-length tuple, and treat it as int later.
            action_shape = ()
        else:
            # Continuous action space, e.g. Box(shape=(3,)), so we store that tuple
            action_shape = tuple(action_space.shape)

        # 3) Overwrite the YAML placeholders (null → actual)
        self.buffer_kwargs["obs_shape"] = obs_shape
        self.buffer_kwargs["action_shape"] = action_shape

    self.buffer = self.buffer_class(**self.buffer_kwargs)  # type: ignore

load

load(path: str) -> None

Load the internal state of the agent.

Source code in mighty/mighty_agents/ppo.py

def load(self, path: str) -> None:
    """Load the internal state of the agent."""
    base_path = Path(path)
    self.model.policy_head.load_state_dict(torch.load(base_path / "policy_head.pt"))  # type: ignore
    self.model.value_head.load_state_dict(torch.load(base_path / "value_head.pt"))  # type: ignore
    self.update_fn.optimizer.load_state_dict(  # type: ignore
        torch.load(base_path / "optimizer.pt")
    )

    if self.verbose:
        print(f"Loaded checkpoint at {path}")

make_checkpoint_dir

make_checkpoint_dir(t: int) -> None

Checkpoint model.

:param T: Current timestep :return:

Source code in mighty/mighty_agents/base_agent.py

def make_checkpoint_dir(self, t: int) -> None:
    """Checkpoint model.

    :param T: Current timestep
    :return:
    """
    self.upper_checkpoint_dir = Path(self.output_dir) / Path("checkpoints")
    if not self.upper_checkpoint_dir.exists():
        Path(self.upper_checkpoint_dir).mkdir()
    self.checkpoint_dir = self.upper_checkpoint_dir / f"{t}"
    if not self.checkpoint_dir.exists():
        Path(self.checkpoint_dir).mkdir()

run

run(
    n_steps: int,
    eval_every_n_steps: int = 1000,
    save_model_every_n_steps: int | None = 5000,
    env: MIGHTYENV = None,
) -> Dict

Run agent.

Source code in mighty/mighty_agents/base_agent.py

def run(  # noqa: PLR0915
    self,
    n_steps: int,
    eval_every_n_steps: int = 1_000,
    save_model_every_n_steps: int | None = 5000,
    env: MIGHTYENV = None,  # type: ignore
) -> Dict:
    """Run agent."""
    episodes = 0
    if env is not None:
        self.env = env

    logging_layout, progress, steps_task = self.make_logging_layout(n_steps)
    update_multiplier = 0

    with Live(logging_layout, refresh_per_second=10, vertical_overflow="visible"):
        steps_since_eval = 0
        steps_since_log = 0

        metrics = {
            "env": self.env,
            "vf": self.value_function,  # type: ignore
            "policy": self.policy,
            "env_step": self.steps,
            "hp/lr": self.learning_rate,
            "hp/pi_epsilon": self._epsilon,
            "hp/batch_size": self._batch_size,
            "hp/learning_starts": self._learning_starts,
        }

        # Reset env and initialize reward sum
        curr_s, info = self.env.reset(seed=self.seed)  # type: ignore
        if self.log_infos:
            for k in info.keys():
                if not k.startswith("_") and k not in self.result_buffer.keys():
                    self.result_buffer[k] = []
                if not k.startswith("_") and k not in self.eval_buffer.keys():
                    self.eval_buffer[k] = []
        if len(curr_s.squeeze().shape) == 0:
            episode_reward = [0]
        else:
            episode_reward = np.zeros(curr_s.squeeze().shape[0])  # type: ignore

        last_episode_reward = episode_reward
        if not torch.is_tensor(last_episode_reward):
            last_episode_reward = torch.tensor(last_episode_reward).float()

        recent_episode_reward = []
        recent_step_reward = []
        recent_actions = []
        evaluation_reward = []

        # Start logging
        eval_curve = [0]
        learning_curve = [0]
        curve_xs = [0]
        progress.update(steps_task, visible=True)
        logging_layout["lower"]["left"].update(
            self.get_plot(curve_xs, learning_curve, "Training Reward")
        )
        logging_layout["lower"]["right"].update(
            self.get_plot(curve_xs, eval_curve, "Evaluation Reward")
        )

        # Main loop: rollouts, training and evaluation
        while self.steps < n_steps:
            metrics["episode_reward"] = episode_reward

            action, log_prob = self.step(curr_s, metrics)
            # step the env as usual
            next_s, reward, terminated, truncated, infos = self.env.step(action)

            # decide which samples are true “done”
            replay_dones = terminated  # physics‐failure only
            dones = np.logical_or(terminated, truncated)

            # Overwrite next_s on truncation
            # Based on https://github.com/DLR-RM/stable-baselines3/issues/284
            real_next_s = next_s.copy()
            # infos["final_observation"] is a list/array of the last real obs
            for i, tr in enumerate(truncated):
                if tr:
                    real_next_s[i] = infos["final_observation"][i]
            episode_reward += reward

            # Log everything
            t = {
                "seed": self.seed,
                "env_step": self.steps,
                "reward": reward,
                "action": action,
                "state": curr_s,
                "next_state": real_next_s,
                "terminated": terminated.astype(int),
                "truncated": truncated.astype(int),
                "dones": replay_dones.astype(int),
                "mean_episode_reward": last_episode_reward.mean()
                .cpu()
                .numpy()
                .item(),
            }
            if isinstance(infos, list):
                if len(infos) > 0:
                    infos = {
                        k: infos[i][k]
                        for k in list(infos[0].keys())
                        for i in range(len(infos))
                    }
                else:
                    infos = {}

            t.update(infos)

            if hasattr(self.env, "unwrapped") and (
                isinstance(self.env.unwrapped, DACENV)
                or isinstance(self.env.unwrapped, CARLENV)
            ):
                t["instances"] = self.env.inst_ids

            metrics["log_prob"] = log_prob.detach().cpu().numpy()
            metrics["episode_reward"] = episode_reward
            metrics["transition"] = t

            recent_actions.append(np.mean(action))
            if len(recent_actions) > 100:
                recent_actions.pop(0)

            for k in self.meta_modules:
                self.meta_modules[k].post_step(metrics)

            transition_metrics = self.process_transition(
                metrics["transition"]["state"],
                metrics["transition"]["action"],
                metrics["transition"]["reward"],
                metrics["transition"]["next_state"],
                metrics["transition"]["dones"],
                metrics["log_prob"],
                metrics,
            )
            metrics.update(transition_metrics)
            self.result_buffer = update_buffer(self.result_buffer, t)

            if self.log_wandb:
                log_to_wandb(metrics)

            self.steps += len(action)
            metrics["env_step"] = self.steps
            steps_since_eval += len(action)
            steps_since_log += len(action)
            for _ in range(len(action)):
                progress.advance(steps_task)

            # Update agent
            if (
                len(self.buffer) >= self._batch_size  # type: ignore
                and self.steps >= self._learning_starts
            ):
                update_kwargs = {"next_s": next_s, "dones": dones}
                metrics = self.update(metrics, update_kwargs)

            # End step
            self.last_state = curr_s
            curr_s = next_s

            # Evaluate
            if eval_every_n_steps and steps_since_eval >= eval_every_n_steps:
                steps_since_eval = 0
                eval_metrics = self.evaluate(log_infos=self.log_infos)
                evaluation_reward = eval_metrics["eval_rewards"]

            # Log to command line via rich layout
            if self.steps >= 1000 * update_multiplier:
                metrics_table = self.make_logging_table(
                    self.steps,
                    recent_episode_reward,
                    recent_step_reward,
                    evaluation_reward,
                    recent_actions,
                )
                logging_layout["middle"]["left"].update(metrics_table)
                eval_curve.append(np.mean(evaluation_reward))
                learning_curve.append(np.mean(recent_episode_reward))
                curve_xs.append(self.steps)

                logging_layout["lower"]["left"].update(
                    self.get_plot(curve_xs, learning_curve, "Training Reward")
                )
                logging_layout["lower"]["right"].update(
                    self.get_plot(curve_xs, eval_curve, "Evaluation Reward")
                )
                update_multiplier += 1

            # Save model & metrics
            if (
                save_model_every_n_steps
                and steps_since_log >= save_model_every_n_steps
            ):
                steps_since_log = 0
                self.save(self.steps)
                log_to_file(
                    self.output_dir,
                    self.result_buffer,
                    self.hp_buffer,
                    self.eval_buffer,
                    self.loss_buffer,
                )

            # Perform resets as necessary
            if np.any(dones):
                last_episode_reward = np.where(  # type: ignore
                    dones, episode_reward, last_episode_reward
                )
                recent_episode_reward.append(np.mean(last_episode_reward))
                recent_step_reward.append(
                    np.mean(last_episode_reward) / len(last_episode_reward)
                )
                last_episode_reward = torch.tensor(last_episode_reward).float()
                if len(recent_episode_reward) > 10:
                    recent_episode_reward.pop(0)
                    recent_step_reward.pop(0)
                episode_reward = np.where(dones, 0, episode_reward)  # type: ignore
                # End episode
                episodes += 1
                for k in self.meta_modules:
                    self.meta_modules[k].post_episode(metrics)

                if "rollout_values" in metrics:
                    del metrics["rollout_values"]

                if "rollout_logits" in metrics:
                    del metrics["rollout_logits"]

                # Meta Module hooks
                for k in self.meta_modules:
                    self.meta_modules[k].pre_episode(metrics)

    # Final logging
    log_to_file(
        self.output_dir,
        self.result_buffer,
        self.hp_buffer,
        self.eval_buffer,
        self.loss_buffer,
    )
    return metrics

save

save(t: int) -> None

Save current agent state.

Source code in mighty/mighty_agents/ppo.py

def save(self, t: int) -> None:
    """Save current agent state."""
    super().make_checkpoint_dir(t)
    torch.save(
        self.model.policy_head.state_dict(),  # type: ignore
        self.checkpoint_dir / "policy_head.pt",
    )
    torch.save(
        self.model.value_head.state_dict(),  # type: ignore
        self.checkpoint_dir / "value_head.pt",
    )
    torch.save(
        self.update_fn.optimizer.state_dict(),  # type: ignore
        self.checkpoint_dir / "optimizer.pt",
    )

    if self.verbose:
        print(f"Saved checkpoint at {self.checkpoint_dir}")

update_agent

update_agent(
    transition_batch, batches_left, next_s, dones, **kwargs
) -> Dict

Update the agent using PPO.

:return: Dictionary containing the update metrics.

Source code in mighty/mighty_agents/ppo.py

def update_agent(
    self, transition_batch, batches_left, next_s, dones, **kwargs
) -> Dict:  # type: ignore
    """Update the agent using PPO.

    :return: Dictionary containing the update metrics.
    """
    metrics: Dict = {}
    metrics.update(self.update_fn.update(transition_batch))  # type: ignore

    for key, value in metrics.items():
        self.loss_buffer[key].append(value)
    self.loss_buffer["update_at_step"].append(self.steps)

    if batches_left == 0:
        self.buffer.reset()  # type: ignore

    return metrics