Maskable PPO

Implementation of invalid action masking for the Proximal Policy Optimization (PPO) algorithm. Other than adding support for action masking, the behavior is the same as in SB3’s core PPO algorithm.

Available Policies

MlpPolicy

alias of MaskableActorCriticPolicy

CnnPolicy

alias of MaskableActorCriticCnnPolicy

MultiInputPolicy

alias of MaskableMultiInputActorCriticPolicy

Notes

Can I use?

  • Recurrent policies: ❌

  • Multi processing: ✔️

  • Gym spaces:

Space

Action

Observation

Discrete

✔️

✔️

Box

✔️

MultiDiscrete

✔️

✔️

MultiBinary

✔️

✔️

Dict

✔️

Warning

You must use MaskableEvalCallback from sb3_contrib.common.maskable.callbacks instead of the base EvalCallback to properly evaluate a model with action masks. Similarly, you must use evaluate_policy from sb3_contrib.common.maskable.evaluation instead of the SB3 one.

Warning

In order to use SubprocVecEnv with MaskablePPO, you must implement the action_masks inside the environment (ActionMasker cannot be used). You can have a look at the built-in environments with invalid action masks to have a working example.

Example

Train a PPO agent on InvalidActionEnvDiscrete. InvalidActionEnvDiscrete has a action_masks method that returns the invalid action mask (True if the action is valid, False otherwise).

from sb3_contrib import MaskablePPO
from sb3_contrib.common.envs import InvalidActionEnvDiscrete
from sb3_contrib.common.maskable.evaluation import evaluate_policy
from sb3_contrib.common.maskable.utils import get_action_masks
# This is a drop-in replacement for EvalCallback
from sb3_contrib.common.maskable.callbacks import MaskableEvalCallback


env = InvalidActionEnvDiscrete(dim=80, n_invalid_actions=60)
model = MaskablePPO("MlpPolicy", env, gamma=0.4, seed=32, verbose=1)
model.learn(5_000)

evaluate_policy(model, env, n_eval_episodes=20, reward_threshold=90, warn=False)

model.save("ppo_mask")
del model # remove to demonstrate saving and loading

model = MaskablePPO.load("ppo_mask")

obs, _ = env.reset()
while True:
    # Retrieve current action mask
    action_masks = get_action_masks(env)
    action, _states = model.predict(obs, action_masks=action_masks)
    obs, reward, terminated, truncated, info = env.step(action)

If the environment implements the invalid action mask but using a different name, you can use the ActionMasker to specify the name (see PR #25):

Note

If you are using a custom environment and you want to debug it with check_env, it will execute the method step passing a random action to it (using action_space.sample()), without taking into account the invalid actions mask (see issue #145).

import gymnasium as gym
import numpy as np

from sb3_contrib.common.maskable.policies import MaskableActorCriticPolicy
from sb3_contrib.common.wrappers import ActionMasker
from sb3_contrib.ppo_mask import MaskablePPO


def mask_fn(env: gym.Env) -> np.ndarray:
    # Do whatever you'd like in this function to return the action mask
    # for the current env. In this example, we assume the env has a
    # helpful method we can rely on.
    return env.valid_action_mask()


env = ...  # Initialize env
env = ActionMasker(env, mask_fn)  # Wrap to enable masking

# MaskablePPO behaves the same as SB3's PPO unless the env is wrapped
# with ActionMasker. If the wrapper is detected, the masks are automatically
# retrieved and used when learning. Note that MaskablePPO does not accept
# a new action_mask_fn kwarg, as it did in an earlier draft.
model = MaskablePPO(MaskableActorCriticPolicy, env, verbose=1)
model.learn()

# Note that use of masks is manual and optional outside of learning,
# so masking can be "removed" at testing time
model.predict(observation, action_masks=valid_action_array)

Results

Results are shown for two MicroRTS benchmarks: MicrortsMining4x4F9-v0 (600K steps) and MicrortsMining10x10F9-v0 (1.5M steps). For each, models were trained with and without masking, using 3 seeds.

4x4

No masking

../_images/4x4_no_mask.png

With masking

../_images/4x4_mask.png

Combined

../_images/4x4_combined.png

10x10

No masking

../_images/10x10_no_mask.png

With masking


../_images/10x10_mask.png

Combined

../_images/10x10_combined.png

More information may be found in the associated PR.

How to replicate the results?

Clone the repo for the experiment:

git clone git@github.com:kronion/microrts-ppo-comparison.git
cd microrts-ppo-comparison

Install dependencies:

# Install MicroRTS:
rm -fR ~/microrts && mkdir ~/microrts && \
 wget -O ~/microrts/microrts.zip http://microrts.s3.amazonaws.com/microrts/artifacts/202004222224.microrts.zip && \
 unzip ~/microrts/microrts.zip -d ~/microrts/

# You may want to make a venv before installing packages
pip install -r requirements.txt

Train several times with various seeds, with and without masking:

# python sb/train_ppo.py [output dir] [MicroRTS map size] [--mask] [--seed int]

# 4x4 unmasked
python sb3/train_ppo.py zoo 4 --seed 42
python sb3/train_ppo.py zoo 4 --seed 43
python sb3/train_ppo.py zoo 4 --seed 44

# 4x4 masked
python sb3/train_ppo.py zoo 4 --mask --seed 42
python sb3/train_ppo.py zoo 4 --mask --seed 43
python sb3/train_ppo.py zoo 4 --mask --seed 44

# 10x10 unmasked
python sb3/train_ppo.py zoo 10 --seed 42
python sb3/train_ppo.py zoo 10 --seed 43
python sb3/train_ppo.py zoo 10 --seed 44

# 10x10 masked
python sb3/train_ppo.py zoo 10 --mask --seed 42
python sb3/train_ppo.py zoo 10 --mask --seed 43
python sb3/train_ppo.py zoo 10 --mask --seed 44

View the tensorboard log output:

# For 4x4 environment
 tensorboard --logdir zoo/4x4/runs

 # For 10x10 environment
 tensorboard --logdir zoo/10x10/runs

Parameters

class sb3_contrib.ppo_mask.MaskablePPO(policy, env, learning_rate=0.0003, n_steps=2048, batch_size=64, n_epochs=10, gamma=0.99, gae_lambda=0.95, clip_range=0.2, clip_range_vf=None, normalize_advantage=True, ent_coef=0.0, vf_coef=0.5, max_grad_norm=0.5, rollout_buffer_class=None, rollout_buffer_kwargs=None, target_kl=None, stats_window_size=100, tensorboard_log=None, policy_kwargs=None, verbose=0, seed=None, device='auto', _init_setup_model=True)[source]

Proximal Policy Optimization algorithm (PPO) (clip version) with Invalid Action Masking.

Based on the original Stable Baselines 3 implementation.

Introduction to PPO: https://spinningup.openai.com/en/latest/algorithms/ppo.html Background on Invalid Action Masking: https://arxiv.org/abs/2006.14171

Parameters:
  • policy (MaskableActorCriticPolicy) – The policy model to use (MlpPolicy, CnnPolicy, …)

  • env (Env | VecEnv | str) – The environment to learn from (if registered in Gym, can be str)

  • learning_rate (float | Callable[[float], float]) – The learning rate, it can be a function of the current progress remaining (from 1 to 0)

  • n_steps (int) – The number of steps to run for each environment per update (i.e. batch size is n_steps * n_env where n_env is number of environment copies running in parallel)

  • batch_size (int | None) – Minibatch size

  • n_epochs (int) – Number of epoch when optimizing the surrogate loss

  • gamma (float) – Discount factor

  • gae_lambda (float) – Factor for trade-off of bias vs variance for Generalized Advantage Estimator

  • clip_range (float | Callable[[float], float]) – Clipping parameter, it can be a function of the current progress remaining (from 1 to 0).

  • clip_range_vf (None | float | Callable[[float], float]) – Clipping parameter for the value function, it can be a function of the current progress remaining (from 1 to 0). This is a parameter specific to the OpenAI implementation. If None is passed (default), no clipping will be done on the value function. IMPORTANT: this clipping depends on the reward scaling.

  • normalize_advantage (bool) – Whether to normalize or not the advantage

  • ent_coef (float) – Entropy coefficient for the loss calculation

  • vf_coef (float) – Value function coefficient for the loss calculation

  • max_grad_norm (float) – The maximum value for the gradient clipping

  • target_kl (float | None) – Limit the KL divergence between updates, because the clipping is not enough to prevent large update see issue #213 (cf https://github.com/hill-a/stable-baselines/issues/213) By default, there is no limit on the kl div.

  • stats_window_size (int) – Window size for the rollout logging, specifying the number of episodes to average the reported success rate, mean episode length, and mean reward over

  • tensorboard_log (str | None) – the log location for tensorboard (if None, no logging)

  • policy_kwargs (Dict[str, Any] | None) – additional arguments to be passed to the policy on creation

  • verbose (int) – the verbosity level: 0 no output, 1 info, 2 debug

  • seed (int | None) – Seed for the pseudo random generators

  • device (device | str) – Device (cpu, cuda, …) on which the code should be run. Setting it to auto, the code will be run on the GPU if possible.

  • _init_setup_model (bool) – Whether or not to build the network at the creation of the instance

  • rollout_buffer_class (Type[RolloutBuffer] | None) –

  • rollout_buffer_kwargs (Dict[str, Any] | None) –

collect_rollouts(env, callback, rollout_buffer, n_rollout_steps, use_masking=True)[source]

Collect experiences using the current policy and fill a RolloutBuffer. The term rollout here refers to the model-free notion and should not be used with the concept of rollout used in model-based RL or planning.

This method is largely identical to the implementation found in the parent class.

Parameters:
  • env (VecEnv) – The training environment

  • callback (BaseCallback) – Callback that will be called at each step (and at the beginning and end of the rollout)

  • rollout_buffer (RolloutBuffer) – Buffer to fill with rollouts

  • n_steps – Number of experiences to collect per environment

  • use_masking (bool) – Whether or not to use invalid action masks during training

  • n_rollout_steps (int) –

Returns:

True if function returned with at least n_rollout_steps collected, False if callback terminated rollout prematurely.

Return type:

bool

get_env()

Returns the current environment (can be None if not defined).

Returns:

The current environment

Return type:

VecEnv | None

get_parameters()

Return the parameters of the agent. This includes parameters from different networks, e.g. critics (value functions) and policies (pi functions).

Returns:

Mapping of from names of the objects to PyTorch state-dicts.

Return type:

Dict[str, Dict]

get_vec_normalize_env()

Return the VecNormalize wrapper of the training env if it exists.

Returns:

The VecNormalize env.

Return type:

VecNormalize | None

learn(total_timesteps, callback=None, log_interval=1, tb_log_name='PPO', reset_num_timesteps=True, use_masking=True, progress_bar=False)[source]

Return a trained model.

Parameters:
  • total_timesteps (int) – The total number of samples (env steps) to train on

  • callback (None | Callable | List[BaseCallback] | BaseCallback) – callback(s) called at every step with state of the algorithm.

  • log_interval (int) – The number of episodes before logging.

  • tb_log_name (str) – the name of the run for TensorBoard logging

  • reset_num_timesteps (bool) – whether or not to reset the current timestep number (used in logging)

  • progress_bar (bool) – Display a progress bar using tqdm and rich.

  • self (SelfMaskablePPO) –

  • use_masking (bool) –

Returns:

the trained model

Return type:

SelfMaskablePPO

classmethod load(path, env=None, device='auto', custom_objects=None, print_system_info=False, force_reset=True, **kwargs)

Load the model from a zip-file. Warning: load re-creates the model from scratch, it does not update it in-place! For an in-place load use set_parameters instead.

Parameters:
  • path (str | Path | BufferedIOBase) – path to the file (or a file-like) where to load the agent from

  • env (Env | VecEnv | None) – the new environment to run the loaded model on (can be None if you only need prediction from a trained model) has priority over any saved environment

  • device (device | str) – Device on which the code should run.

  • custom_objects (Dict[str, Any] | None) – Dictionary of objects to replace upon loading. If a variable is present in this dictionary as a key, it will not be deserialized and the corresponding item will be used instead. Similar to custom_objects in keras.models.load_model. Useful when you have an object in file that can not be deserialized.

  • print_system_info (bool) – Whether to print system info from the saved model and the current system info (useful to debug loading issues)

  • force_reset (bool) – Force call to reset() before training to avoid unexpected behavior. See https://github.com/DLR-RM/stable-baselines3/issues/597

  • kwargs – extra arguments to change the model when loading

Returns:

new model instance with loaded parameters

Return type:

SelfBaseAlgorithm

property logger: Logger

Getter for the logger object.

predict(observation, state=None, episode_start=None, deterministic=False, action_masks=None)[source]

Get the policy action from an observation (and optional hidden state). Includes sugar-coating to handle different observations (e.g. normalizing images).

Parameters:
  • observation (ndarray | Dict[str, ndarray]) – the input observation

  • state (Tuple[ndarray, ...] | None) – The last hidden states (can be None, used in recurrent policies)

  • episode_start (ndarray | None) – The last masks (can be None, used in recurrent policies) this correspond to beginning of episodes, where the hidden states of the RNN must be reset.

  • deterministic (bool) – Whether or not to return deterministic actions.

  • action_masks (ndarray | None) –

Returns:

the model’s action and the next hidden state (used in recurrent policies)

Return type:

Tuple[ndarray, Tuple[ndarray, …] | None]

save(path, exclude=None, include=None)

Save all the attributes of the object and the model parameters in a zip-file.

Parameters:
  • path (str | Path | BufferedIOBase) – path to the file where the rl agent should be saved

  • exclude (Iterable[str] | None) – name of parameters that should be excluded in addition to the default ones

  • include (Iterable[str] | None) – name of parameters that might be excluded but should be included anyway

Return type:

None

set_env(env, force_reset=True)

Checks the validity of the environment, and if it is coherent, set it as the current environment. Furthermore wrap any non vectorized env into a vectorized checked parameters: - observation_space - action_space

Parameters:
Return type:

None

set_logger(logger)

Setter for for logger object.

Warning

When passing a custom logger object, this will overwrite tensorboard_log and verbose settings passed to the constructor.

Parameters:

logger (Logger) –

Return type:

None

set_parameters(load_path_or_dict, exact_match=True, device='auto')

Load parameters from a given zip-file or a nested dictionary containing parameters for different modules (see get_parameters).

Parameters:
  • load_path_or_iter – Location of the saved data (path or file-like, see save), or a nested dictionary containing nn.Module parameters used by the policy. The dictionary maps object names to a state-dictionary returned by torch.nn.Module.state_dict().

  • exact_match (bool) – If True, the given parameters should include parameters for each module and each of their parameters, otherwise raises an Exception. If set to False, this can be used to update only specific parameters.

  • device (device | str) – Device on which the code should run.

  • load_path_or_dict (str | Dict[str, Tensor]) –

Return type:

None

set_random_seed(seed=None)

Set the seed of the pseudo-random generators (python, numpy, pytorch, gym, action_space)

Parameters:

seed (int | None) –

Return type:

None

train()[source]

Update policy using the currently gathered rollout buffer.

Return type:

None

MaskablePPO Policies

sb3_contrib.ppo_mask.MlpPolicy

alias of MaskableActorCriticPolicy

class sb3_contrib.common.maskable.policies.MaskableActorCriticPolicy(observation_space, action_space, lr_schedule, net_arch=None, activation_fn=<class 'torch.nn.modules.activation.Tanh'>, ortho_init=True, features_extractor_class=<class 'stable_baselines3.common.torch_layers.FlattenExtractor'>, features_extractor_kwargs=None, share_features_extractor=True, normalize_images=True, optimizer_class=<class 'torch.optim.adam.Adam'>, optimizer_kwargs=None)[source]

Policy class for actor-critic algorithms (has both policy and value prediction). Used by A2C, PPO and the likes.

Parameters:
  • observation_space (Space) – Observation space

  • action_space (Space) – Action space

  • lr_schedule (Callable[[float], float]) – Learning rate schedule (could be constant)

  • net_arch (List[int] | Dict[str, List[int]] | None) – The specification of the policy and value networks.

  • activation_fn (Type[Module]) – Activation function

  • ortho_init (bool) – Whether to use or not orthogonal initialization

  • features_extractor_class (Type[BaseFeaturesExtractor]) – Features extractor to use.

  • features_extractor_kwargs (Dict[str, Any] | None) – Keyword arguments to pass to the features extractor.

  • share_features_extractor (bool) – If True, the features extractor is shared between the policy and value networks.

  • normalize_images (bool) – Whether to normalize images or not, dividing by 255.0 (True by default)

  • optimizer_class (Type[Optimizer]) – The optimizer to use, th.optim.Adam by default

  • optimizer_kwargs (Dict[str, Any] | None) – Additional keyword arguments, excluding the learning rate, to pass to the optimizer

evaluate_actions(obs, actions, action_masks=None)[source]

Evaluate actions according to the current policy, given the observations.

Parameters:
  • obs (Tensor) – Observation

  • actions (Tensor) – Actions

  • action_masks (Tensor | None) –

Returns:

estimated value, log likelihood of taking those actions and entropy of the action distribution.

Return type:

Tuple[Tensor, Tensor, Tensor | None]

extract_features(obs, features_extractor=None)[source]

Preprocess the observation if needed and extract features.

Parameters:
  • obs (Tensor | Dict[str, Tensor]) – Observation

  • features_extractor (BaseFeaturesExtractor | None) – The features extractor to use. If None, then self.features_extractor is used.

Returns:

The extracted features. If features extractor is not shared, returns a tuple with the features for the actor and the features for the critic.

Return type:

Tensor | Tuple[Tensor, Tensor]

forward(obs, deterministic=False, action_masks=None)[source]

Forward pass in all the networks (actor and critic)

Parameters:
  • obs (Tensor) – Observation

  • deterministic (bool) – Whether to sample or use deterministic actions

  • action_masks (ndarray | None) – Action masks to apply to the action distribution

Returns:

action, value and log probability of the action

Return type:

Tuple[Tensor, Tensor, Tensor]

get_distribution(obs, action_masks=None)[source]

Get the current policy distribution given the observations.

Parameters:
  • obs (Tensor | Dict[str, Tensor]) – Observation

  • action_masks (ndarray | None) – Actions’ mask

Returns:

the action distribution.

Return type:

MaskableDistribution

predict(observation, state=None, episode_start=None, deterministic=False, action_masks=None)[source]

Get the policy action from an observation (and optional hidden state). Includes sugar-coating to handle different observations (e.g. normalizing images).

Parameters:
  • observation (ndarray | Dict[str, ndarray]) – the input observation

  • state (Tuple[ndarray, ...] | None) – The last states (can be None, used in recurrent policies)

  • episode_start (ndarray | None) – The last masks (can be None, used in recurrent policies)

  • deterministic (bool) – Whether or not to return deterministic actions.

  • action_masks (ndarray | None) – Action masks to apply to the action distribution

Returns:

the model’s action and the next state (used in recurrent policies)

Return type:

Tuple[ndarray, Tuple[ndarray, …] | None]

predict_values(obs)[source]

Get the estimated values according to the current policy given the observations.

Parameters:

obs (Tensor | Dict[str, Tensor]) – Observation

Returns:

the estimated values.

Return type:

Tensor

sb3_contrib.ppo_mask.CnnPolicy

alias of MaskableActorCriticCnnPolicy

class sb3_contrib.common.maskable.policies.MaskableActorCriticCnnPolicy(observation_space, action_space, lr_schedule, net_arch=None, activation_fn=<class 'torch.nn.modules.activation.Tanh'>, ortho_init=True, features_extractor_class=<class 'stable_baselines3.common.torch_layers.NatureCNN'>, features_extractor_kwargs=None, share_features_extractor=True, normalize_images=True, optimizer_class=<class 'torch.optim.adam.Adam'>, optimizer_kwargs=None)[source]

CNN policy class for actor-critic algorithms (has both policy and value prediction). Used by A2C, PPO and the likes.

Parameters:
  • observation_space (Space) – Observation space

  • action_space (Space) – Action space

  • lr_schedule (Callable[[float], float]) – Learning rate schedule (could be constant)

  • net_arch (List[int] | Dict[str, List[int]] | None) – The specification of the policy and value networks.

  • activation_fn (Type[Module]) – Activation function

  • ortho_init (bool) – Whether to use or not orthogonal initialization

  • features_extractor_class (Type[BaseFeaturesExtractor]) – Features extractor to use.

  • features_extractor_kwargs (Dict[str, Any] | None) – Keyword arguments to pass to the features extractor.

  • share_features_extractor (bool) – If True, the features extractor is shared between the policy and value networks.

  • normalize_images (bool) – Whether to normalize images or not, dividing by 255.0 (True by default)

  • optimizer_class (Type[Optimizer]) – The optimizer to use, th.optim.Adam by default

  • optimizer_kwargs (Dict[str, Any] | None) – Additional keyword arguments, excluding the learning rate, to pass to the optimizer

sb3_contrib.ppo_mask.MultiInputPolicy

alias of MaskableMultiInputActorCriticPolicy

class sb3_contrib.common.maskable.policies.MaskableMultiInputActorCriticPolicy(observation_space, action_space, lr_schedule, net_arch=None, activation_fn=<class 'torch.nn.modules.activation.Tanh'>, ortho_init=True, features_extractor_class=<class 'stable_baselines3.common.torch_layers.CombinedExtractor'>, features_extractor_kwargs=None, share_features_extractor=True, normalize_images=True, optimizer_class=<class 'torch.optim.adam.Adam'>, optimizer_kwargs=None)[source]

MultiInputActorClass policy class for actor-critic algorithms (has both policy and value prediction). Used by A2C, PPO and the likes.

Parameters:
  • observation_space (Dict) – Observation space (Tuple)

  • action_space (Space) – Action space

  • lr_schedule (Callable[[float], float]) – Learning rate schedule (could be constant)

  • net_arch (List[int] | Dict[str, List[int]] | None) – The specification of the policy and value networks.

  • activation_fn (Type[Module]) – Activation function

  • ortho_init (bool) – Whether to use or not orthogonal initialization

  • features_extractor_class (Type[BaseFeaturesExtractor]) – Uses the CombinedExtractor

  • features_extractor_kwargs (Dict[str, Any] | None) – Keyword arguments to pass to the feature extractor.

  • share_features_extractor (bool) – If True, the features extractor is shared between the policy and value networks.

  • normalize_images (bool) – Whether to normalize images or not, dividing by 255.0 (True by default)

  • optimizer_class (Type[Optimizer]) – The optimizer to use, th.optim.Adam by default

  • optimizer_kwargs (Dict[str, Any] | None) – Additional keyword arguments, excluding the learning rate, to pass to the optimizer