Sac

mighty.mighty_agents.sac

MightySACAgent

MightySACAgent(
    output_dir: Path,
    env: MIGHTYENV,
    eval_env: Optional[MIGHTYENV] = None,
    seed: Optional[int] = None,
    n_policy_units: int = 64,
    n_critic_units: int = 64,
    soft_update_weight: float = 0.005,
    batch_size: int = 256,
    learning_starts: int = 10000,
    update_every: int = 50,
    n_gradient_steps: int = 1,
    policy_lr: float = 0.0003,
    q_lr: float = 0.0003,
    gamma: float = 0.99,
    alpha: float = 0.2,
    hidden_sizes: Optional[List[int]] = None,
    activation: str = "relu",
    log_std_min: float = -20,
    log_std_max: float = 2,
    render_progress: bool = True,
    log_wandb: bool = False,
    wandb_kwargs: Optional[Dict] = None,
    replay_buffer_class: Type[MightyReplay] = MightyReplay,
    replay_buffer_kwargs: Optional[TypeKwargs] = None,
    meta_methods: Optional[List[Union[str, type]]] = None,
    meta_kwargs: Optional[List[TypeKwargs]] = None,
    policy_class: Optional[
        Union[
            str, DictConfig, Type[MightyExplorationPolicy]
        ]
    ] = None,
    policy_kwargs: Optional[Dict] = None,
)

Bases: MightyAgent

Source code in mighty/mighty_agents/sac.py

def __init__(
    self,
    output_dir: Path,
    env: MIGHTYENV,
    eval_env: Optional[MIGHTYENV] = None,
    seed: Optional[int] = None,
    # --- PPO-style network sizes ---
    n_policy_units: int = 64,
    n_critic_units: int = 64,
    soft_update_weight: float = 0.005,
    # --- Replay & update scheduling ---
    batch_size: int = 256,
    learning_starts: int = 10000,
    update_every: int = 50,
    n_gradient_steps: int = 1,
    # --- Learning rates ---
    policy_lr: float = 3e-4,
    q_lr: float = 3e-4,
    # --- SAC hyperparameters ---
    gamma: float = 0.99,
    alpha: float = 0.2,
    # --- Network architecture (optional override) ---
    hidden_sizes: Optional[List[int]] = None,
    activation: str = "relu",
    log_std_min: float = -20,
    log_std_max: float = 2,
    # --- Logging & buffer ---
    render_progress: bool = True,
    log_wandb: bool = False,
    wandb_kwargs: Optional[Dict] = None,
    replay_buffer_class: Type[MightyReplay] = MightyReplay,
    replay_buffer_kwargs: Optional[TypeKwargs] = None,
    meta_methods: Optional[List[Union[str, type]]] = None,
    meta_kwargs: Optional[List[TypeKwargs]] = None,
    policy_class: Optional[
        Union[str, DictConfig, Type[MightyExplorationPolicy]]
    ] = None,
    policy_kwargs: Optional[Dict] = None,
):
    """Initialize SAC agent with tunable hyperparameters and backward-compatible names."""
    # Map PPO-style units to hidden_sizes if not provided
    if hidden_sizes is None:
        hidden_sizes = [n_policy_units, n_policy_units]
    tau = soft_update_weight

    # Save hyperparameters
    self.batch_size = batch_size
    self.learning_starts = learning_starts
    self.update_every = update_every
    self.n_gradient_steps = n_gradient_steps
    self.policy_lr = policy_lr
    self.q_lr = q_lr
    self.gamma = gamma
    self.tau = tau
    self.alpha = alpha
    self.hidden_sizes = hidden_sizes
    self.activation = activation
    self.log_std_min = log_std_min
    self.log_std_max = log_std_max

    # Placeholders for model and updater
    self.model: SACModel | None = None
    self.update_fn: SACUpdate | None = None

    # Exploration policy class
    policy_class = retrieve_class(cls=policy_class, default_cls=StochasticPolicy)
    self.policy_class = policy_class
    self.policy_kwargs = policy_kwargs or {}

    super().__init__(
        env=env,
        output_dir=output_dir,
        seed=seed,
        eval_env=eval_env,
        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=replay_buffer_class,
        replay_buffer_kwargs=replay_buffer_kwargs,
        meta_methods=meta_methods,
        meta_kwargs=meta_kwargs,
    )

    # Initialize loss buffer for logging
    self.loss_buffer = {
        "q_loss1": [],
        "q_loss2": [],
        "policy_loss": [],
        "td_error1": [],
        "td_error2": [],
        "step": [],
    }

value_function property

value_function: Module

Value function for compatibility: V(s) = min(Q1,Q2)(s, a_policy) - alpha * log_pi(a|s).

__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 = {
        "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,
        "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 = {
        "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,
        "meta_modules": list(self.meta_modules.keys()),
    }
    # FIXME: EWRL: Can we naively compare dictoinaries like this?
    if old_hps != updated_hps:
        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) -> 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) -> 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

    state, _ = eval_env.reset(options=options)  # 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
    while not np.all(mask):
        action = self.policy(state, evaluate=True)  # type: ignore
        state, reward, terminated, truncated, _ = eval_env.step(action)  # type: ignore
        rewards += reward * (1 - mask)
        steps += 1 * (1 - mask)
        dones = np.logical_or(terminated, truncated)
        mask = np.where(dones, 1, mask)

    eval_env.close()  # type: ignore

    if isinstance(self.eval_env, DACENV) or isinstance(self.env, CARLENV):
        instance = eval_env.instance  # type: ignore
    else:
        instance = "None"

    eval_metrics = {
        "step": 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),
        "instance": instance,
        "eval_rewards": rewards,
    }
    self.eval_buffer = update_buffer(self.eval_buffer, eval_metrics)

    if self.log_wandb:
        wandb.log(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()
    self.buffer = self.buffer_class(**self.buffer_kwargs)  # type: ignore

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 = Layout()
    logging_layout.split_column(
        Layout(name="upper"), Layout(name="middle"), Layout(name="lower")
    )
    logging_layout["middle"].split_row(
        Layout(name="left"),
        Layout(name="right"),
    )
    logging_layout["lower"].split_row(
        Layout(name="left"),
        Layout(name="right"),
    )

    metrics_table = self.make_logging_table(0, [0], [0], [0], [0])  # type: ignore
    training_detail_table = Table(title="Training Details")
    training_detail_table.add_column("Name", style="cyan", justify="center")
    training_detail_table.add_column("Value", style="cyan", justify="center")
    training_detail_table.add_row("Training Steps", str(n_steps))
    training_detail_table.add_row("Seed", str(self.seed))
    training_detail_table.add_row("Learning Rate", str(self.learning_rate))
    training_detail_table.add_row("Batch Size", str(self._batch_size))

    logging_layout["middle"]["left"].update(metrics_table)
    logging_layout["middle"]["right"].update(training_detail_table)

    progress = Progress(
        "[progress.description]{task.description}",
        BarColumn(),
        "[progress.percentage]{task.percentage:>3.0f}%",
        "Remaining:",
        TimeRemainingColumn(),
        "Elapsed:",
        TimeElapsedColumn(),
        disable=not self.render_progress,
    )
    steps_task = progress.add_task(
        "Train Steps",
        total=n_steps - self.steps,
        start=False,
        visible=False,
    )
    progress.start_task(steps_task)

    progress_table = Table.grid(expand=True)
    progress_table.add_row(
        Panel(
            progress,
            title="Training Progress",
            border_style="green",
            padding=(2, 2),
        ),
    )
    logging_layout["upper"].update(progress_table)
    update_multiplier = 0

    with Live(logging_layout, refresh_per_second=10, vertical_overflow="visible"):
        steps_since_eval = 0
        # FIXME: this is more of a question: are there cases where we don't want to reset this completely?
        # I can't think of any, can you? If yes, we should maybe add this as an optional argument
        metrics = {
            "env": self.env,
            "vf": self.value_function,  # type: ignore
            "policy": self.policy,
            "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, _ = self.env.reset()  # type: ignore
        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()
        progress.update(steps_task, visible=True)
        recent_episode_reward = []
        recent_step_reward = []
        recent_actions = []
        evaluation_reward = []
        eval_curve = [0]
        learning_curve = [0]
        curve_xs = [0]
        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)

            next_s, reward, terminated, truncated, _ = self.env.step(action)  # type: ignore
            dones = np.logical_or(terminated, truncated)

            episode_reward += reward

            # Log everything
            t = {
                "seed": self.seed,
                "step": self.steps,
                "reward": reward,
                "action": action,
                "state": curr_s,
                "next_state": next_s,
                "terminated": terminated.astype(int),
                "truncated": truncated.astype(int),
                "dones": dones.astype(int),
                "mean_episode_reward": last_episode_reward.mean().cpu().numpy().item(),
            }
            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:
                wandb.log(t)

            self.steps += len(action)
            metrics["step"] = self.steps
            steps_since_eval += 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()
                evaluation_reward = eval_metrics["eval_rewards"]

            # Log to command line
            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
            if (
                save_model_every_n_steps
                and self.steps % save_model_every_n_steps == 0
            ):
                self.save(self.steps)
                log_to_file(
                    self.output_dir,
                    self.result_buffer,
                    self.hp_buffer,
                    self.eval_buffer,
                    self.loss_buffer,
                )

            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
                if isinstance(self.env, DACENV) or isinstance(self.env, CARLENV):
                    instance = self.env.instance  # type: ignore
                else:
                    instance = None
                metrics["instance"] = instance
                episodes += 1
                for k in self.meta_modules:
                    self.meta_modules[k].post_episode(metrics)

                # Meta Module hooks
                for k in self.meta_modules:
                    self.meta_modules[k].pre_episode(metrics)
    log_to_file(
        self.output_dir,
        self.result_buffer,
        self.hp_buffer,
        self.eval_buffer,
        self.loss_buffer,
    )
    return metrics

update

update(metrics: Dict, update_kwargs: Dict) -> Dict

Update agent.

Source code in mighty/mighty_agents/base_agent.py

def update(self, metrics: Dict, update_kwargs: Dict) -> Dict:
    """Update agent."""
    for k in self.meta_modules:
        self.meta_modules[k].pre_update(metrics)

    batches = []
    for batches_left in reversed(range(self.n_gradient_steps)):
        batch = self.buffer.sample(self._batch_size)
        agent_update_metrics = self.update_agent(
            transition_batch=batch, batches_left=batches_left, **update_kwargs
        )

        # FIXME: EWRL: Move logging out of gradient steps

        metrics.update(agent_update_metrics)

        metrics = {k: np.array(v) for k, v in metrics.items()}
        metrics["step"] = self.steps

        # Wandb logging
        if self.log_wandb:
            import json

            import wandb

            # Only log relevant, serializable keys
            log_keys = [
                "step",
                "episode_reward",
                "Update/policy_loss",
                "Update/value_loss",
                "Update/entropy",
                "Update/approx_kl",
            ]
            serializable_metrics = {}
            for k in log_keys:
                if k in metrics:
                    v = metrics[k]
                    # Convert numpy arrays to scalars or lists
                    if isinstance(v, np.ndarray):
                        if v.size == 1:
                            v = v.item()
                        else:
                            v = v.tolist()
                    # Convert torch tensors to scalars or lists
                    if isinstance(v, torch.Tensor):
                        if v.numel() == 1:
                            v = v.item()
                        else:
                            v = v.cpu().numpy().tolist()
                    # Try to serialize, skip if not possible
                    try:
                        json.dumps(v)
                        serializable_metrics[k] = v
                    except TypeError:
                        print(f"Skipping non-serializable metric: {k}")

            wandb.log(serializable_metrics, step=self.steps)

        metrics["env"] = self.env
        metrics["vf"] = self.value_function  # type: ignore
        metrics["policy"] = self.policy
        batches.append(batch)

    metrics["update_batches"] = batches
    for k in self.meta_modules:
        self.meta_modules[k].post_update(metrics)
    del metrics["update_batches"]
    return metrics