diff --git a/tests/test_distributions.py b/tests/test_distributions.py new file mode 100644 index 0000000..47651e4 --- /dev/null +++ b/tests/test_distributions.py @@ -0,0 +1,20 @@ +import numpy as np +import torch as th + +from torchy_baselines.common.distributions import DiagGaussianDistribution, SquashedDiagGaussianDistribution,\ + CategoricalDistribution, TanhBijector + +# TODO: more tests for the other distributions +def test_bijector(): + """ + Test TanhBijector + """ + actions = th.ones(5) * 2.0 + + bijector = TanhBijector() + + squashed_actions = bijector.forward(actions) + # Check that the boundaries are not violated + assert th.max(th.abs(squashed_actions)) <= 1.0 + # Check the inverse method + assert th.isclose(TanhBijector.inverse(squashed_actions), actions).all() diff --git a/torchy_baselines/common/distributions.py b/torchy_baselines/common/distributions.py index 82a07a2..9d67b39 100644 --- a/torchy_baselines/common/distributions.py +++ b/torchy_baselines/common/distributions.py @@ -46,9 +46,10 @@ class Distribution(object): class DiagGaussianDistribution(Distribution): """ - Gaussian distribution with diagonal covariance matrix. + Gaussian distribution with diagonal covariance matrix, + for continuous actions. - :param action_dim: (int) Number of actions + :param action_dim: (int) Number of continuous actions """ def __init__(self, action_dim): super(DiagGaussianDistribution, self).__init__() @@ -65,6 +66,7 @@ class DiagGaussianDistribution(Distribution): :param latent_dim: (int) Dimension og the last layer of the policy (before the action layer) :param log_std_init: (float) Initial value for the log standard deviation + :return: (nn.Linear, nn.Parameter) """ mean_actions = nn.Linear(latent_dim, self.action_dim) # TODO: allow action dependent std @@ -111,6 +113,14 @@ class DiagGaussianDistribution(Distribution): return action, log_prob def log_prob(self, action): + """ + Get the log probabilty of an action given a distribution. + Note that you must call `proba_distribution()` method + before. + + :param action: (th.Tensor) + :return: (th.Tensor) + """ log_prob = self.distribution.log_prob(action) if len(log_prob.shape) > 1: log_prob = log_prob.sum(axis=1) @@ -120,6 +130,13 @@ class DiagGaussianDistribution(Distribution): class SquashedDiagGaussianDistribution(DiagGaussianDistribution): + """ + Gaussian distribution with diagonal covariance matrix, + followed by a squashing function (tanh) to ensure bounds. + + :param action_dim: (int) Number of continuous actions + :param epsilon: (float) small value to avoid NaN due to numerical imprecision. + """ def __init__(self, action_dim, epsilon=1e-6): super(SquashedDiagGaussianDistribution, self).__init__(action_dim) # Avoid NaN (prevents division by zero or log of zero) @@ -146,27 +163,40 @@ class SquashedDiagGaussianDistribution(DiagGaussianDistribution): def log_prob(self, action, gaussian_action=None): # Inverse tanh - # Naive implementation (not stable): 0.5 * torch.log((1 + x ) / (1 - x)) + # Naive implementation (not stable): 0.5 * torch.log((1 + x) / (1 - x)) # We use numpy to avoid numerical instability if gaussian_action is None: - # Clip to avoid NaN - clipped_action = np.clip(action.cpu().numpy(), -1.0 + self.epsilon, 1.0 + self.epsilon) - gaussian_action = th.from_numpy(np.arctanh(clipped_action)).to(action.device) + # It will be clipped to avoid NaN when inversing tanh + gaussian_action = TanhBijector.inverse(action) # Log likelihood for a gaussian distribution log_prob = super(SquashedDiagGaussianDistribution, self).log_prob(gaussian_action) # Squash correction (from original SAC implementation) + # this comes from the fact that tanh is bijective and differentiable log_prob -= th.sum(th.log(1 - action ** 2 + self.epsilon), dim=1) return log_prob class CategoricalDistribution(Distribution): + """ + Categorical distribution for discrete actions. + + :param action_dim: (int) Number of discrete actions + """ def __init__(self, action_dim): super(CategoricalDistribution, self).__init__() self.distribution = None self.action_dim = action_dim def proba_distribution_net(self, latent_dim): + """ + Create the layer that represents the distribution: + it will be the logits of the Categorical distribution. + You can then get probabilties using a softmax. + + :param latent_dim: (int) Dimension og the last layer of the policy (before the action layer) + :return: (nn.Linear) + """ action_logits = nn.Linear(latent_dim, self.action_dim) return action_logits @@ -198,6 +228,19 @@ class CategoricalDistribution(Distribution): class StateDependentNoiseDistribution(Distribution): + """ + Distribution class for using State Dependent Exploration (SDE). + It is used to create the noise exploration matrix and + compute the log probabilty of an action with that noise. + + :param action_dim: (int) Number of continuous actions + :param use_expln: (bool) Use `expln()` function instead of `exp()` to ensure + a positive standard deviation (cf paper). It allows to keep variance + above zero and prevent it from growing too fast. In practice, `exp()` is usually enough. + :param squash_output: (bool) Whether to squash the output using a tanh function, + this allows to ensure boundaries. + :param epsilon: (float) small value to avoid NaN due to numerical imprecision. + """ def __init__(self, action_dim, use_expln=False, squash_output=False, epsilon=1e-6): super(StateDependentNoiseDistribution, self).__init__() @@ -215,6 +258,13 @@ class StateDependentNoiseDistribution(Distribution): self.bijector = None def get_std(self, log_std): + """ + Get the standard deviation from the learned parameter + (log of it by default). This ensures that the std is positive. + + :param log_std: (th.Tensor) + :return: (th.Tensor) + """ if self.use_expln: # From SDE paper, it allows to keep variance # above zero and prevent it from growing too fast @@ -223,19 +273,44 @@ class StateDependentNoiseDistribution(Distribution): else: return th.log(log_std + 1.0) + 1.0 else: + # Use normal exponential return th.exp(log_std) def sample_weights(self, log_std): + """ + Sample weights for the noise exploration matrix, + using a centered gaussian distribution. + + :param log_std: (th.Tensor) + """ + # TODO: reduce the number of learned dimensions (cf TD3) self.weights_dist = Normal(th.zeros_like(log_std), self.get_std(log_std)) self.exploration_mat = self.weights_dist.rsample() def proba_distribution_net(self, latent_dim, log_std_init=0.0): + """ + Create the layers and parameter that represent the distribution: + one output will be the deterministic action, the other parameter will be the + standard deviation of the distribution that control the weights of the noise matrix. + + :param latent_dim: (int) Dimension og the last layer of the policy (before the action layer) + :param log_std_init: (float) Initial value for the log standard deviation + :return: (nn.Linear, nn.Parameter) + """ mean_actions = nn.Linear(latent_dim, self.action_dim) log_std = nn.Parameter(th.ones(latent_dim, self.action_dim) * log_std_init) self.sample_weights(log_std) return mean_actions, log_std def proba_distribution(self, mean_actions, log_std, latent_pi, deterministic=False): + """ + Create and sample for the distribution given its parameters (mean, std) + + :param mean_actions: (th.Tensor) + :param log_std: (th.Tensor) + :param deterministic: (bool) + :return: (th.Tensor) + """ variance = th.mm(latent_pi.detach() ** 2, self.get_std(log_std) ** 2) self.distribution = Normal(mean_actions, th.sqrt(variance)) @@ -287,6 +362,13 @@ class StateDependentNoiseDistribution(Distribution): class TanhBijector(object): + """ + Bijective transformation of a probabilty distribution + using a squashing function (tanh) + TODO: use Pyro instead (https://pyro.ai/) + + :param epsilon: (float) small value to avoid NaN due to numerical imprecision. + """ def __init__(self, epsilon=1e-6): super(TanhBijector, self).__init__() self.epsilon = epsilon @@ -294,23 +376,27 @@ class TanhBijector(object): def forward(self, x): return th.tanh(x) - def inverse(self, action): + @staticmethod + def atanh(x): + """ + Inverse of Tanh + + Taken from pyro: https://github.com/pyro-ppl/pyro + 0.5 * torch.log((1 + x ) / (1 - x)) + """ + return 0.5 * (x.log1p() - (-x).log1p()) + + @staticmethod + def inverse(y): """ Inverse tanh. - From https://github.com/tensorflow/agents: - 0.99999997 is the maximum value such that atanh(x) is valid for both - float32 and float64 - - :param action: (th.Tensor) + :param y: (th.Tensor) :return: (th.Tensor) """ - # Inverse tanh - # Naive implementation (not stable): 0.5 * torch.log((1 + x ) / (1 - x)) - # We use numpy to avoid numerical instability - # Note: Using numpy, we do not keep the gradient - clipped_action = np.clip(action.cpu().numpy(), -0.99999997, 0.99999997) - return th.from_numpy(np.arctanh(clipped_action)).to(action.device) + eps = th.finfo(y.dtype).eps + # Clip the action to avoid NaN + return TanhBijector.atanh(y.clamp(min=-1. + eps, max=1. - eps)) def log_prob_correction(self, x): # Squash correction (from original SAC implementation)