Compare commits

..
11 Commits
14 changed files with 428 additions and 292 deletions
+1
View File
@@ -1,5 +1,6 @@
__pycache__ __pycache__
.venv .venv
.vscode
wandb wandb
*.egg-info/ *.egg-info/
test.py test.py
+1 -1
View File
@@ -49,7 +49,7 @@ The TRPL implementation in Fancy RL includes projections based on the Kullback-L
To run the test suite: To run the test suite:
```bash ```bash
pytest test/test_ppo.py pytest test/
``` ```
## Status ## Status
+45 -19
View File
@@ -1,9 +1,13 @@
import torch import torch
import gymnasium as gym import gymnasium as gym
from torchrl.envs.libs.gym import GymWrapper from torchrl.envs import GymEnv, TransformedEnv, Compose, RewardSum, StepCounter, SerialEnv
from torchrl.record import VideoRecorder from torchrl.record import VideoRecorder
from abc import ABC from abc import ABC
import pdb
import numpy as np
from tensordict import TensorDict from tensordict import TensorDict
from torchrl.envs import GymWrapper, TransformedEnv
from torchrl.envs import BatchSizeTransform
from fancy_rl.loggers import TerminalLogger from fancy_rl.loggers import TerminalLogger
@@ -47,18 +51,37 @@ class Algo(ABC):
self.eval_episodes = eval_episodes self.eval_episodes = eval_episodes
def make_env(self, eval=False): def make_env(self, eval=False):
"""Creates an environment and wraps it if necessary."""
env_spec = self.env_spec_eval if eval else self.env_spec env_spec = self.env_spec_eval if eval else self.env_spec
env = self._wrap_env(env_spec)
env.reset()
return env
def _wrap_env(self, env_spec):
# If given an existing env, ensure it's properly batched
if isinstance(env_spec, (GymEnv, GymWrapper)):
if not env_spec.batch_size:
raise ValueError("Environment must be batched")
return env_spec
# Handle callable without wrapping the recursive call
if callable(env_spec):
return self._wrap_env(env_spec())
# Create new batched environment using SerialEnv
if isinstance(env_spec, str): if isinstance(env_spec, str):
env = gym.make(env_spec) env = SerialEnv(1, lambda: GymEnv(env_spec, device=self.device))
env = GymWrapper(env).to(self.device) elif isinstance(env_spec, gym.Env):
elif callable(env_spec): wrapped_env = GymWrapper(env_spec, device=self.device)
env = env_spec() if wrapped_env.batch_size:
if not (isinstance(env, gym.Env) or isinstance(env, gym.core.Wrapper)): env = wrapped_env
raise ValueError("env_spec must be a string or a callable that returns an environment. Was a callable that returned a {}".format(type(env))) else:
env = GymWrapper(env).to(self.device) env = SerialEnv(1, lambda: wrapped_env)
else: else:
raise ValueError("env_spec must be a string or a callable that returns an environment. Was a {}".format(type(env_spec))) raise ValueError(
f"env_spec must be a string, callable, Gymnasium environment, or GymEnv, "
f"got {type(env_spec)}"
)
return env return env
def train_step(self, batch): def train_step(self, batch):
@@ -72,18 +95,21 @@ class Algo(ABC):
def predict( def predict(
self, self,
observation, tensordict,
state=None, state=None,
deterministic=False deterministic=False
): ):
with torch.no_grad(): with torch.no_grad():
obs_tensor = torch.as_tensor(observation, device=self.device).unsqueeze(0) # If numpy array, convert to TensorDict
td = TensorDict({"observation": obs_tensor}, batch_size=[1]) if isinstance(tensordict, np.ndarray):
tensordict = TensorDict(
{"observation": torch.from_numpy(tensordict).float()},
batch_size=[]
)
action_td = self.prob_actor(td) # Move to device
action = action_td["action"] tensordict = tensordict.to(self.device)
# We're not using recurrent policies, so we'll always return None for the state # Get action from policy
next_state = None action_td = self.prob_actor(tensordict)
return action_td
return action.squeeze(0).cpu().numpy(), next_state
+16 -8
View File
@@ -2,9 +2,9 @@ import torch
from torchrl.modules import ProbabilisticActor from torchrl.modules import ProbabilisticActor
from torchrl.objectives import ClipPPOLoss from torchrl.objectives import ClipPPOLoss
from torchrl.objectives.value.advantages import GAE from torchrl.objectives.value.advantages import GAE
from torchrl.data.tensor_specs import DiscreteTensorSpec
from fancy_rl.algos.on_policy import OnPolicy from fancy_rl.algos.on_policy import OnPolicy
from fancy_rl.policy import Actor, Critic from fancy_rl.policy import Actor, Critic
from fancy_rl.utils import is_discrete_space
class PPO(OnPolicy): class PPO(OnPolicy):
def __init__( def __init__(
@@ -41,17 +41,25 @@ class PPO(OnPolicy):
# Initialize environment to get observation and action space sizes # Initialize environment to get observation and action space sizes
self.env_spec = env_spec self.env_spec = env_spec
env = self.make_env() env = self.make_env()
obs_space = env.observation_space
act_space = env.action_space # Get spaces from specs for parallel env
self.obs_space = env.observation_spec
self.act_space = env.action_spec
self.discrete = isinstance(self.act_space, DiscreteTensorSpec)
self.discrete = is_discrete_space(act_space) self.critic = Critic(self.obs_space, critic_hidden_sizes, critic_activation_fn, device)
self.actor = Actor(self.obs_space, self.act_space, actor_hidden_sizes, actor_activation_fn, device, full_covariance=full_covariance)
self.critic = Critic(obs_space, critic_hidden_sizes, critic_activation_fn, device)
self.actor = Actor(obs_space, act_space, actor_hidden_sizes, actor_activation_fn, device, full_covariance=full_covariance)
if self.discrete: if self.discrete:
distribution_class = torch.distributions.Categorical distribution_class = torch.distributions.Categorical
distribution_kwargs = {"logits": "action_logits"} self.prob_actor = ProbabilisticActor(
module=self.actor,
distribution_class=distribution_class,
return_log_prob=True,
in_keys=["logits"],
out_keys=["action"],
)
else: else:
if full_covariance: if full_covariance:
distribution_class = torch.distributions.MultivariateNormal distribution_class = torch.distributions.MultivariateNormal
+43 -8
View File
@@ -1,6 +1,7 @@
import torch import torch
from torch import nn from torch import nn
from typing import Dict, Any, Optional from typing import Dict, Any, Optional
from torchrl.data.tensor_specs import DiscreteTensorSpec
from torchrl.modules import ProbabilisticActor, ValueOperator from torchrl.modules import ProbabilisticActor, ValueOperator
from torchrl.objectives import ClipPPOLoss from torchrl.objectives import ClipPPOLoss
from torchrl.collectors import SyncDataCollector from torchrl.collectors import SyncDataCollector
@@ -10,10 +11,11 @@ from fancy_rl.algos.on_policy import OnPolicy
from fancy_rl.policy import Actor, Critic from fancy_rl.policy import Actor, Critic
from fancy_rl.projections import get_projection, BaseProjection from fancy_rl.projections import get_projection, BaseProjection
from fancy_rl.objectives import TRPLLoss from fancy_rl.objectives import TRPLLoss
from fancy_rl.utils import is_discrete_space
from copy import deepcopy from copy import deepcopy
from tensordict.nn import TensorDictModule from tensordict.nn import TensorDictModule
from tensordict import TensorDict from tensordict import TensorDict
from torch.distributions import Categorical, MultivariateNormal, Normal
class ProjectedActor(TensorDictModule): class ProjectedActor(TensorDictModule):
def __init__(self, raw_actor, old_actor, projection): def __init__(self, raw_actor, old_actor, projection):
@@ -26,6 +28,8 @@ class ProjectedActor(TensorDictModule):
self.raw_actor = raw_actor self.raw_actor = raw_actor
self.old_actor = old_actor self.old_actor = old_actor
self.projection = projection self.projection = projection
self.discrete = raw_actor.discrete
self.full_covariance = raw_actor.full_covariance
class CombinedModule(nn.Module): class CombinedModule(nn.Module):
def __init__(self, raw_actor, old_actor, projection): def __init__(self, raw_actor, old_actor, projection):
@@ -35,12 +39,41 @@ class ProjectedActor(TensorDictModule):
self.projection = projection self.projection = projection
def forward(self, tensordict): def forward(self, tensordict):
# Convert the tuple outputs to TensorDict
raw_params = self.raw_actor(tensordict) raw_params = self.raw_actor(tensordict)
if isinstance(raw_params, tuple):
raw_params = TensorDict({
"loc": raw_params[0],
"scale": raw_params[1]
}, batch_size=[raw_params[0].shape[0]]) # Use the first dimension of the tensor as batch size
old_params = self.old_actor(tensordict) old_params = self.old_actor(tensordict)
combined_params = TensorDict({**raw_params, **{f"old_{key}": value for key, value in old_params.items()}}, batch_size=tensordict.batch_size) if isinstance(old_params, tuple):
old_params = TensorDict({
"loc": old_params[0],
"scale": old_params[1]
}, batch_size=[old_params[0].shape[0]]) # Use the first dimension of the tensor as batch size
# Now combine them
combined_params = TensorDict({
**raw_params,
**{f"old_{key}": value for key, value in old_params.items()}
}, batch_size=[raw_params["loc"].shape[0]]) # Use the first dimension of loc tensor as batch size
projected_params = self.projection(combined_params) projected_params = self.projection(combined_params)
return projected_params return projected_params
def get_dist(self, tensordict):
# Forward the observation through the network
out = self.forward(tensordict)
if self.discrete:
return Categorical(logits=out["logits"])
else:
if self.full_covariance:
return MultivariateNormal(loc=out["loc"], scale_tril=out["scale_tril"])
else:
return Normal(loc=out["loc"], scale=out["scale"])
class TRPL(OnPolicy): class TRPL(OnPolicy):
def __init__( def __init__(
self, self,
@@ -77,14 +110,16 @@ class TRPL(OnPolicy):
# Initialize environment to get observation and action space sizes # Initialize environment to get observation and action space sizes
self.env_spec = env_spec self.env_spec = env_spec
env = self.make_env() env = self.make_env()
obs_space = env.observation_space
act_space = env.action_space
assert not is_discrete_space(act_space), "TRPL does not support discrete action spaces" # Get spaces from specs for parallel env
self.obs_space = env.observation_spec
self.act_space = env.action_spec
self.critic = Critic(obs_space, critic_hidden_sizes, critic_activation_fn, device) assert not isinstance(self.act_space, DiscreteTensorSpec), "TRPL does not support discrete action spaces"
self.raw_actor = Actor(obs_space, act_space, actor_hidden_sizes, actor_activation_fn, device, full_covariance=full_covariance)
self.old_actor = Actor(obs_space, act_space, actor_hidden_sizes, actor_activation_fn, device, full_covariance=full_covariance) self.critic = Critic(self.obs_space, critic_hidden_sizes, critic_activation_fn, device)
self.raw_actor = Actor(self.obs_space, self.act_space, actor_hidden_sizes, actor_activation_fn, device, full_covariance=full_covariance)
self.old_actor = Actor(self.obs_space, self.act_space, actor_hidden_sizes, actor_activation_fn, device, full_covariance=full_covariance)
# Handle projection_class # Handle projection_class
if isinstance(projection_class, str): if isinstance(projection_class, str):
+28 -10
View File
@@ -1,14 +1,18 @@
import torch.nn as nn import torch.nn as nn
from tensordict.nn import TensorDictModule from tensordict.nn import TensorDictModule
from torchrl.modules import MLP from torchrl.modules import MLP
from torchrl.data.tensor_specs import DiscreteTensorSpec
from tensordict.nn.distributions import NormalParamExtractor from tensordict.nn.distributions import NormalParamExtractor
from fancy_rl.utils import is_discrete_space, get_space_shape
from tensordict import TensorDict from tensordict import TensorDict
from torch.distributions import Categorical, MultivariateNormal, Normal
class Actor(TensorDictModule): class Actor(TensorDictModule):
def __init__(self, obs_space, act_space, hidden_sizes, activation_fn, device, full_covariance=False): def __init__(self, obs_space, act_space, hidden_sizes, activation_fn, device, full_covariance=False):
self.discrete = is_discrete_space(act_space) self.discrete = isinstance(act_space, DiscreteTensorSpec)
act_space_shape = get_space_shape(act_space)
obs_space = obs_space["observation"]
act_space_shape = act_space.shape[1:]
obs_space_shape = obs_space.shape[1:]
if self.discrete and full_covariance: if self.discrete and full_covariance:
raise ValueError("Full covariance is not applicable for discrete action spaces.") raise ValueError("Full covariance is not applicable for discrete action spaces.")
@@ -16,18 +20,18 @@ class Actor(TensorDictModule):
self.full_covariance = full_covariance self.full_covariance = full_covariance
if self.discrete: if self.discrete:
out_features = act_space_shape[-1] out_features = act_space_shape[0]
out_keys = ["action_logits"] out_keys = ["logits"]
else: else:
if full_covariance: if full_covariance:
out_features = act_space_shape[-1] + (act_space_shape[-1] * (act_space_shape[-1] + 1)) // 2 out_features = act_space_shape[0] + (act_space_shape[0] * (act_space_shape[0] + 1)) // 2
out_keys = ["loc", "scale_tril"] out_keys = ["loc", "scale_tril"]
else: else:
out_features = act_space_shape[-1] * 2 out_features = act_space_shape[0] * 2
out_keys = ["loc", "scale"] out_keys = ["loc", "scale"]
actor_module = MLP( actor_module = MLP(
in_features=get_space_shape(obs_space)[-1], in_features=obs_space_shape[0],
out_features=out_features, out_features=out_features,
num_cells=hidden_sizes, num_cells=hidden_sizes,
activation_class=getattr(nn, activation_fn), activation_class=getattr(nn, activation_fn),
@@ -36,7 +40,7 @@ class Actor(TensorDictModule):
if not self.discrete: if not self.discrete:
if full_covariance: if full_covariance:
param_extractor = FullCovarianceNormalParamExtractor(act_space_shape[-1]) param_extractor = FullCovarianceNormalParamExtractor(act_space_shape[0])
else: else:
param_extractor = NormalParamExtractor() param_extractor = NormalParamExtractor()
actor_module = nn.Sequential(actor_module, param_extractor) actor_module = nn.Sequential(actor_module, param_extractor)
@@ -47,6 +51,17 @@ class Actor(TensorDictModule):
out_keys=out_keys out_keys=out_keys
) )
def get_dist(self, tensordict):
# Forward the observation through the network
out = self.forward(tensordict)
if self.discrete:
return Categorical(logits=out["logits"])
else:
if self.full_covariance:
return MultivariateNormal(loc=out["loc"], scale_tril=out["scale_tril"])
else:
return Normal(loc=out["loc"], scale=out["scale"])
class FullCovarianceNormalParamExtractor(nn.Module): class FullCovarianceNormalParamExtractor(nn.Module):
def __init__(self, action_dim): def __init__(self, action_dim):
super().__init__() super().__init__()
@@ -62,8 +77,11 @@ class FullCovarianceNormalParamExtractor(nn.Module):
class Critic(TensorDictModule): class Critic(TensorDictModule):
def __init__(self, obs_space, hidden_sizes, activation_fn, device): def __init__(self, obs_space, hidden_sizes, activation_fn, device):
obs_space = obs_space["observation"]
obs_space_shape = obs_space.shape[1:]
critic_module = MLP( critic_module = MLP(
in_features=get_space_shape(obs_space)[-1], in_features=obs_space_shape[0],
out_features=1, out_features=1,
num_cells=hidden_sizes, num_cells=hidden_sizes,
activation_class=getattr(nn, activation_fn), activation_class=getattr(nn, activation_fn),
+23 -2
View File
@@ -1,7 +1,7 @@
from abc import ABC, abstractmethod from abc import ABC, abstractmethod
import torch import torch
from torch import nn from torch import nn
from typing import Dict, List from typing import Dict, List, Tuple
class BaseProjection(nn.Module, ABC): class BaseProjection(nn.Module, ABC):
def __init__(self, in_keys: List[str], out_keys: List[str], trust_region_coeff: float = 1.0, mean_bound: float = 0.01, cov_bound: float = 0.01, contextual_std: bool = True): def __init__(self, in_keys: List[str], out_keys: List[str], trust_region_coeff: float = 1.0, mean_bound: float = 0.01, cov_bound: float = 0.01, contextual_std: bool = True):
@@ -68,4 +68,25 @@ class BaseProjection(nn.Module, ABC):
if not self.full_cov: if not self.full_cov:
return torch.sqrt(cov.diagonal(dim1=-2, dim2=-1)) return torch.sqrt(cov.diagonal(dim1=-2, dim2=-1))
else: else:
return torch.linalg.cholesky(cov) return torch.linalg.cholesky(cov)
def _mean_projection(self, mean: torch.Tensor, old_mean: torch.Tensor, mean_part: torch.Tensor) -> torch.Tensor:
"""Project mean based on the Mahalanobis objective and trust region.
Args:
mean: Current mean vectors
old_mean: Old mean vectors
mean_part: Mahalanobis/Euclidean distance between the two mean vectors
Returns:
Projected mean that satisfies the trust region
"""
mask = mean_part > self.mean_bound
omega = torch.ones_like(mean_part, device=mean.device)
omega = torch.where(mask, torch.sqrt(mean_part / self.mean_bound) - 1., omega)
omega = torch.maximum(-omega, omega)[..., None]
# Use matrix operations instead of boolean indexing
m = (mean + omega * old_mean) / (1. + omega + 1e-16)
mask_matrix = mask[..., None].to(mean.dtype)
return mask_matrix * m + (1 - mask_matrix) * mean
+40 -19
View File
@@ -1,7 +1,7 @@
import torch import torch
from .base_projection import BaseProjection from .base_projection import BaseProjection
from tensordict.nn import TensorDictModule from tensordict.nn import TensorDictModule
from typing import Dict from typing import Dict, Tuple
class FrobeniusProjection(BaseProjection): class FrobeniusProjection(BaseProjection):
def __init__(self, in_keys: list[str], out_keys: list[str], trust_region_coeff: float = 1.0, mean_bound: float = 0.01, cov_bound: float = 0.01, scale_prec: bool = False, contextual_std: bool = True): def __init__(self, in_keys: list[str], out_keys: list[str], trust_region_coeff: float = 1.0, mean_bound: float = 0.01, cov_bound: float = 0.01, scale_prec: bool = False, contextual_std: bool = True):
@@ -12,16 +12,23 @@ class FrobeniusProjection(BaseProjection):
mean = policy_params["loc"] mean = policy_params["loc"]
old_mean = old_policy_params["loc"] old_mean = old_policy_params["loc"]
# Convert to covariance representation
cov = self._calc_covariance(policy_params) cov = self._calc_covariance(policy_params)
old_cov = self._calc_covariance(old_policy_params) old_cov = self._calc_covariance(old_policy_params)
mean_part, cov_part = self._gaussian_frobenius((mean, cov), (old_mean, old_cov)) if not self.contextual_std:
cov = cov[:1]
old_cov = old_cov[:1]
mean_part, cov_part = self._gaussian_frobenius((mean, cov), (old_mean, old_cov))
proj_mean = self._mean_projection(mean, old_mean, mean_part) proj_mean = self._mean_projection(mean, old_mean, mean_part)
proj_cov = self._cov_projection(cov, old_cov, cov_part) proj_cov = self._cov_projection(cov, old_cov, cov_part)
scale_or_scale_tril = self._calc_scale_or_scale_tril(proj_cov) scale_or_tril = self._calc_scale_or_scale_tril(proj_cov)
return {"loc": proj_mean, self.out_keys[1]: scale_or_scale_tril} if not self.contextual_std:
scale_or_tril = scale_or_tril.expand(mean.shape[0], *scale_or_tril.shape[1:])
return {"loc": proj_mean, self.out_keys[1]: scale_or_tril}
def get_trust_region_loss(self, policy_params: Dict[str, torch.Tensor], proj_policy_params: Dict[str, torch.Tensor]) -> torch.Tensor: def get_trust_region_loss(self, policy_params: Dict[str, torch.Tensor], proj_policy_params: Dict[str, torch.Tensor]) -> torch.Tensor:
mean = policy_params["loc"] mean = policy_params["loc"]
@@ -35,34 +42,48 @@ class FrobeniusProjection(BaseProjection):
return (mean_diff + cov_diff).mean() * self.trust_region_coeff return (mean_diff + cov_diff).mean() * self.trust_region_coeff
def _gaussian_frobenius(self, p, q): def _gaussian_frobenius(self, p: Tuple[torch.Tensor, torch.Tensor], q: Tuple[torch.Tensor, torch.Tensor]) -> Tuple[torch.Tensor, torch.Tensor]:
mean, cov = p mean, cov = p
old_mean, old_cov = q old_mean, old_cov = q
if self.scale_prec: if self.scale_prec:
prec_old = torch.inverse(old_cov) if self.full_cov:
mean_part = torch.sum(torch.matmul(mean - old_mean, prec_old) * (mean - old_mean), dim=-1) # Use triangular solve instead of inverse for stability
cov_part = torch.sum(prec_old * cov, dim=(-2, -1)) - torch.logdet(torch.matmul(prec_old, cov)) - mean.shape[-1] diff = mean - old_mean
solved = torch.triangular_solve(diff.unsqueeze(-1), old_cov, upper=False)[0].squeeze(-1)
mean_part = torch.sum(torch.square(solved), dim=-1)
else:
# Diagonal case - direct division is stable
mean_part = torch.sum(torch.square((mean - old_mean) / old_cov), dim=-1)
else: else:
mean_part = torch.sum(torch.square(mean - old_mean), dim=-1) mean_part = torch.sum(torch.square(mean - old_mean), dim=-1)
cov_part = torch.sum(torch.square(cov - old_cov), dim=(-2, -1))
# Covariance part
if self.full_cov:
diff = old_cov - cov
cov_part = torch.sum(torch.square(diff), dim=(-2, -1))
else:
diff = old_cov - cov
cov_part = torch.sum(torch.square(diff), dim=-1)
return mean_part, cov_part return mean_part, cov_part
def _mean_projection(self, mean: torch.Tensor, old_mean: torch.Tensor, mean_part: torch.Tensor) -> torch.Tensor:
diff = mean - old_mean
norm = torch.sqrt(mean_part)
return torch.where(norm > self.mean_bound, old_mean + diff * self.mean_bound / norm.unsqueeze(-1), mean)
def _cov_projection(self, cov: torch.Tensor, old_cov: torch.Tensor, cov_part: torch.Tensor) -> torch.Tensor: def _cov_projection(self, cov: torch.Tensor, old_cov: torch.Tensor, cov_part: torch.Tensor) -> torch.Tensor:
batch_shape = cov.shape[:-2] batch_shape = cov.shape[:-2] if cov.ndim > 2 else cov.shape[:-1]
cov_mask = cov_part > self.cov_bound cov_mask = cov_part > self.cov_bound
eta = torch.ones(batch_shape, dtype=cov.dtype, device=cov.device) eta = torch.ones(batch_shape, dtype=cov.dtype, device=cov.device)
eta[cov_mask] = torch.sqrt(cov_part[cov_mask] / self.cov_bound) - 1. eta = torch.where(cov_mask, torch.sqrt(cov_part / self.cov_bound) - 1., eta)
eta = torch.max(-eta, eta) eta = torch.maximum(-eta, eta)
new_cov = (cov + torch.einsum('i,ijk->ijk', eta, old_cov)) / (1. + eta + 1e-16)[..., None, None] if self.full_cov:
proj_cov = torch.where(cov_mask[..., None, None], new_cov, cov) new_cov = (cov + torch.einsum('...,...ij->...ij', eta, old_cov)) / \
(1. + eta + 1e-16)[..., None, None]
mask_matrix = cov_mask[..., None, None].to(cov.dtype)
proj_cov = torch.where(mask_matrix, new_cov, cov)
else:
new_cov = (cov + eta[..., None] * old_cov) / (1. + eta + 1e-16)[..., None]
mask_matrix = cov_mask[..., None].to(cov.dtype)
proj_cov = torch.where(mask_matrix, new_cov, cov)
return proj_cov return proj_cov
+69 -45
View File
@@ -1,5 +1,9 @@
import torch import torch
import cpp_projection try:
import cpp_projection
cpp_projection_available = True
except ImportError:
cpp_projection_available = False
import numpy as np import numpy as np
from .base_projection import BaseProjection from .base_projection import BaseProjection
from tensordict.nn import TensorDictModule from tensordict.nn import TensorDictModule
@@ -15,8 +19,17 @@ class KLProjection(BaseProjection):
super().__init__(in_keys=in_keys, out_keys=out_keys, trust_region_coeff=trust_region_coeff, mean_bound=mean_bound, cov_bound=cov_bound, contextual_std=contextual_std) super().__init__(in_keys=in_keys, out_keys=out_keys, trust_region_coeff=trust_region_coeff, mean_bound=mean_bound, cov_bound=cov_bound, contextual_std=contextual_std)
def project(self, policy_params: Dict[str, torch.Tensor], old_policy_params: Dict[str, torch.Tensor]) -> Dict[str, torch.Tensor]: def project(self, policy_params: Dict[str, torch.Tensor], old_policy_params: Dict[str, torch.Tensor]) -> Dict[str, torch.Tensor]:
mean, scale_or_tril = policy_params["loc"], policy_params[self.in_keys[1]] self._validate_inputs(policy_params, old_policy_params)
old_mean, old_scale_or_tril = old_policy_params["loc"], old_policy_params[self.in_keys[1]]
mean = policy_params["loc"]
old_mean = old_policy_params["loc"]
if self.full_cov:
scale_or_tril = policy_params["scale_tril"]
old_scale_or_tril = old_policy_params["scale_tril"]
else:
scale_or_tril = policy_params["scale"]
old_scale_or_tril = old_policy_params["scale"]
mean_part, cov_part = self._gaussian_kl((mean, scale_or_tril), (old_mean, old_scale_or_tril)) mean_part, cov_part = self._gaussian_kl((mean, scale_or_tril), (old_mean, old_scale_or_tril))
@@ -31,11 +44,22 @@ class KLProjection(BaseProjection):
if not self.contextual_std: if not self.contextual_std:
proj_scale_or_tril = proj_scale_or_tril.expand(mean.shape[0], *proj_scale_or_tril.shape[1:]) proj_scale_or_tril = proj_scale_or_tril.expand(mean.shape[0], *proj_scale_or_tril.shape[1:])
return {"loc": proj_mean, self.out_keys[1]: proj_scale_or_tril} if self.full_cov:
return {"loc": proj_mean, "scale_tril": proj_scale_or_tril}
else:
return {"loc": proj_mean, "scale": proj_scale_or_tril}
def get_trust_region_loss(self, policy_params: Dict[str, torch.Tensor], proj_policy_params: Dict[str, torch.Tensor]) -> torch.Tensor: def get_trust_region_loss(self, policy_params: Dict[str, torch.Tensor], proj_policy_params: Dict[str, torch.Tensor]) -> torch.Tensor:
mean, scale_or_tril = policy_params["loc"], policy_params[self.in_keys[1]] mean = policy_params["loc"]
proj_mean, proj_scale_or_tril = proj_policy_params["loc"], proj_policy_params[self.out_keys[1]] proj_mean = proj_policy_params["loc"]
if self.full_cov:
scale_or_tril = policy_params["scale_tril"]
proj_scale_or_tril = proj_policy_params["scale_tril"]
else:
scale_or_tril = policy_params["scale"]
proj_scale_or_tril = proj_policy_params["scale"]
kl = sum(self._gaussian_kl((mean, scale_or_tril), (proj_mean, proj_scale_or_tril))) kl = sum(self._gaussian_kl((mean, scale_or_tril), (proj_mean, proj_scale_or_tril)))
return kl.mean() * self.trust_region_coeff return kl.mean() * self.trust_region_coeff
@@ -50,7 +74,9 @@ class KLProjection(BaseProjection):
det_term_other = self._log_determinant(scale_or_tril_other) det_term_other = self._log_determinant(scale_or_tril_other)
if self.full_cov: if self.full_cov:
trace_part = self._torch_batched_trace_square(torch.linalg.solve_triangular(scale_or_tril_other, scale_or_tril, upper=False)) trace_part = self._batched_trace_square(
torch.triangular_solve(scale_or_tril, scale_or_tril_other, upper=False)[0]
)
else: else:
trace_part = torch.sum((scale_or_tril / scale_or_tril_other) ** 2, dim=-1) trace_part = torch.sum((scale_or_tril / scale_or_tril_other) ** 2, dim=-1)
@@ -61,62 +87,60 @@ class KLProjection(BaseProjection):
def _maha(self, x: torch.Tensor, y: torch.Tensor, scale_or_tril: torch.Tensor) -> torch.Tensor: def _maha(self, x: torch.Tensor, y: torch.Tensor, scale_or_tril: torch.Tensor) -> torch.Tensor:
diff = x - y diff = x - y
if self.full_cov: if self.full_cov:
return torch.sum(torch.square(torch.triangular_solve(diff.unsqueeze(-1), scale_or_tril, upper=False)[0].squeeze(-1)), dim=-1) solved = torch.triangular_solve(diff.unsqueeze(-1), scale_or_tril, upper=False)[0]
return torch.sum(torch.square(solved.squeeze(-1)), dim=-1)
else: else:
return torch.sum(torch.square(diff / scale_or_tril), dim=-1) return torch.sum(torch.square(diff / scale_or_tril), dim=-1)
def _log_determinant(self, scale_or_tril: torch.Tensor) -> torch.Tensor: def _log_determinant(self, scale_or_tril: torch.Tensor) -> torch.Tensor:
if self.full_cov: if self.full_cov:
return 2 * torch.log(scale_or_tril.diagonal(dim1=-2, dim2=-1)).sum(-1) return 2 * torch.sum(torch.log(torch.diagonal(scale_or_tril, dim1=-2, dim2=-1)), dim=-1)
else: else:
return 2 * torch.log(scale_or_tril).sum(-1) return 2 * torch.sum(torch.log(scale_or_tril), dim=-1)
def _torch_batched_trace_square(self, x: torch.Tensor) -> torch.Tensor: def _batched_trace_square(self, x: torch.Tensor) -> torch.Tensor:
return torch.sum(x.pow(2), dim=(-2, -1)) return torch.sum(x ** 2, dim=(-2, -1))
def _mean_projection(self, mean: torch.Tensor, old_mean: torch.Tensor, mean_part: torch.Tensor) -> torch.Tensor:
return old_mean + (mean - old_mean) * torch.sqrt(self.mean_bound / (mean_part + 1e-8)).unsqueeze(-1)
def _cov_projection(self, scale_or_tril: torch.Tensor, old_scale_or_tril: torch.Tensor, cov_part: torch.Tensor) -> torch.Tensor: def _cov_projection(self, scale_or_tril: torch.Tensor, old_scale_or_tril: torch.Tensor, cov_part: torch.Tensor) -> torch.Tensor:
if self.full_cov: if self.full_cov:
cov = torch.matmul(scale_or_tril, scale_or_tril.transpose(-1, -2)) cov = torch.matmul(scale_or_tril, scale_or_tril.transpose(-1, -2))
old_cov = torch.matmul(old_scale_or_tril, old_scale_or_tril.transpose(-1, -2)) old_cov = torch.matmul(old_scale_or_tril, old_scale_or_tril.transpose(-1, -2))
else: else:
cov = scale_or_tril.pow(2) cov = scale_or_tril ** 2
old_cov = old_scale_or_tril.pow(2) old_cov = old_scale_or_tril ** 2
mask = cov_part > self.cov_bound mask = cov_part > self.cov_bound
proj_scale_or_tril = torch.zeros_like(scale_or_tril) proj_scale_or_tril = scale_or_tril # Start with original scale
proj_scale_or_tril[~mask] = scale_or_tril[~mask]
if mask.any():
try: if self.full_cov:
if mask.any(): proj_cov = project_full_covariance(cov, scale_or_tril, old_scale_or_tril, self.cov_bound)
if self.full_cov: is_invalid = torch.isnan(proj_cov.mean(dim=(-2, -1)))
proj_cov = KLProjectionGradFunctionCovOnly.apply(cov, scale_or_tril.detach(), old_scale_or_tril, self.cov_bound) proj_scale_or_tril = torch.where(is_invalid[..., None, None], old_scale_or_tril, scale_or_tril)
is_invalid = proj_cov.mean([-2, -1]).isnan() & mask mask = mask & ~is_invalid
if is_invalid.any(): chol = torch.linalg.cholesky(proj_cov)
proj_scale_or_tril[is_invalid] = old_scale_or_tril[is_invalid] proj_scale_or_tril = torch.where(mask[..., None, None], chol, proj_scale_or_tril)
mask &= ~is_invalid else:
proj_scale_or_tril[mask], failed_mask = torch.linalg.cholesky_ex(proj_cov[mask]) proj_cov = project_diag_covariance(cov, old_cov, self.cov_bound)
failed_mask = failed_mask.bool() is_invalid = (torch.isnan(proj_cov.mean(dim=-1)) |
if failed_mask.any(): torch.isinf(proj_cov.mean(dim=-1)) |
proj_scale_or_tril[failed_mask] = old_scale_or_tril[failed_mask] (proj_cov.min(dim=-1).values < 0))
else: proj_scale_or_tril = torch.where(is_invalid[..., None], old_scale_or_tril, scale_or_tril)
proj_cov = KLProjectionGradFunctionDiagCovOnly.apply(cov, old_cov, self.cov_bound) mask = mask & ~is_invalid
is_invalid = (proj_cov.mean(dim=-1).isnan() | proj_cov.mean(dim=-1).isinf() | (proj_cov.min(dim=-1).values < 0)) & mask proj_scale_or_tril = torch.where(mask[..., None], torch.sqrt(proj_cov), scale_or_tril)
if is_invalid.any():
proj_scale_or_tril[is_invalid] = old_scale_or_tril[is_invalid]
mask &= ~is_invalid
proj_scale_or_tril[mask] = proj_cov[mask].sqrt()
except Exception as e:
import logging
logging.error('Projection failed, taking old scale_or_tril for projection.')
print("Projection failed, taking old scale_or_tril for projection.")
proj_scale_or_tril = old_scale_or_tril
raise e
return proj_scale_or_tril return proj_scale_or_tril
def _validate_inputs(self, policy_params, old_policy_params):
if self.full_cov:
required_keys = ["loc", "scale_tril"]
else:
required_keys = ["loc", "scale"]
for key in required_keys:
if key not in policy_params or key not in old_policy_params:
raise KeyError(f"Missing required key '{key}' in policy parameters")
class KLProjectionGradFunctionCovOnly(torch.autograd.Function): class KLProjectionGradFunctionCovOnly(torch.autograd.Function):
projection_op = None projection_op = None
+113 -38
View File
@@ -1,6 +1,5 @@
import torch import torch
from .base_projection import BaseProjection from .base_projection import BaseProjection
from tensordict.nn import TensorDictModule
from typing import Dict, Tuple from typing import Dict, Tuple
def scale_tril_to_sqrt(scale_tril: torch.Tensor) -> torch.Tensor: def scale_tril_to_sqrt(scale_tril: torch.Tensor) -> torch.Tensor:
@@ -12,75 +11,151 @@ def scale_tril_to_sqrt(scale_tril: torch.Tensor) -> torch.Tensor:
""" """
return scale_tril return scale_tril
def gaussian_wasserstein_commutative(policy, p: Tuple[torch.Tensor, torch.Tensor], def gaussian_wasserstein_commutative(p: Tuple[torch.Tensor, torch.Tensor],
q: Tuple[torch.Tensor, torch.Tensor], scale_prec=False) -> Tuple[torch.Tensor, torch.Tensor]: q: Tuple[torch.Tensor, torch.Tensor],
scale_prec: bool = False) -> Tuple[torch.Tensor, torch.Tensor]:
mean, scale_or_sqrt = p mean, scale_or_sqrt = p
mean_other, scale_or_sqrt_other = q mean_other, scale_or_sqrt_other = q
mean_part = torch.sum(torch.square(mean - mean_other), dim=-1) mean_part = torch.sum(torch.square(mean - mean_other), dim=-1)
if scale_or_sqrt.dim() == mean.dim(): # Diagonal case if scale_or_sqrt.dim() == mean.dim(): # Diagonal case
cov = scale_or_sqrt.pow(2)
cov_other = scale_or_sqrt_other.pow(2)
if scale_prec: if scale_prec:
identity = torch.eye(mean.shape[-1], dtype=scale_or_sqrt.dtype, device=scale_or_sqrt.device) # More stable implementation for precision scaling
sqrt_inv_other = 1 / scale_or_sqrt_other scale_part = torch.sum(
c = sqrt_inv_other.pow(2) * cov scale_or_sqrt_other**2 + scale_or_sqrt**2 -
cov_part = torch.sum(identity + c - 2 * sqrt_inv_other * scale_or_sqrt, dim=-1) 2 * scale_or_sqrt_other * scale_or_sqrt,
dim=-1
)
else: else:
cov_part = torch.sum(cov_other + cov - 2 * scale_or_sqrt_other * scale_or_sqrt, dim=-1) # Standard W2 for diagonal case
scale_part = torch.sum(
scale_or_sqrt_other**2 + scale_or_sqrt**2 -
2 * scale_or_sqrt_other * scale_or_sqrt,
dim=-1
)
else: # Full covariance case else: # Full covariance case
# Note: scale_or_sqrt is treated as the matrix square root, not Cholesky decomposition # Note: scale_or_sqrt is treated as the matrix square root, not Cholesky decomposition
cov = torch.matmul(scale_or_sqrt, scale_or_sqrt.transpose(-1, -2))
cov_other = torch.matmul(scale_or_sqrt_other, scale_or_sqrt_other.transpose(-1, -2))
if scale_prec: if scale_prec:
# More stable implementation using triangular solve
identity = torch.eye(mean.shape[-1], dtype=scale_or_sqrt.dtype, device=scale_or_sqrt.device) identity = torch.eye(mean.shape[-1], dtype=scale_or_sqrt.dtype, device=scale_or_sqrt.device)
sqrt_inv_other = torch.linalg.solve(scale_or_sqrt_other, identity) sqrt_inv_other = torch.triangular_solve(identity, scale_or_sqrt_other, upper=False)[0]
c = sqrt_inv_other @ cov @ sqrt_inv_other.transpose(-1, -2) c = torch.matmul(sqrt_inv_other, scale_or_sqrt)
cov_part = torch.trace(identity + c - 2 * sqrt_inv_other @ scale_or_sqrt) scale_part = torch.sum(identity**2 + c**2 - 2 * c, dim=(-2, -1))
else: else:
cov_part = torch.trace(cov_other + cov - 2 * scale_or_sqrt_other @ scale_or_sqrt) # Standard W2 for full covariance
scale_part = torch.sum(
scale_or_sqrt_other**2 + scale_or_sqrt**2 -
2 * torch.matmul(scale_or_sqrt_other, scale_or_sqrt.transpose(-1, -2)),
dim=(-2, -1)
)
return mean_part, cov_part return mean_part, scale_part
class WassersteinProjection(BaseProjection): class WassersteinProjection(BaseProjection):
def __init__(self, in_keys: list[str], out_keys: list[str], trust_region_coeff: float = 1.0, mean_bound: float = 0.01, cov_bound: float = 0.01, scale_prec: bool = False, contextual_std: bool = True): def __init__(self, in_keys: list[str], out_keys: list[str], trust_region_coeff: float = 1.0,
super().__init__(in_keys=in_keys, out_keys=out_keys, trust_region_coeff=trust_region_coeff, mean_bound=mean_bound, cov_bound=cov_bound, contextual_std=contextual_std) mean_bound: float = 0.01, cov_bound: float = 0.01, scale_prec: bool = False,
contextual_std: bool = True):
super().__init__(in_keys=in_keys, out_keys=out_keys, trust_region_coeff=trust_region_coeff,
mean_bound=mean_bound, cov_bound=cov_bound, contextual_std=contextual_std)
self.scale_prec = scale_prec self.scale_prec = scale_prec
def project(self, policy_params: Dict[str, torch.Tensor], old_policy_params: Dict[str, torch.Tensor]) -> Dict[str, torch.Tensor]: def project(self, policy_params: Dict[str, torch.Tensor], old_policy_params: Dict[str, torch.Tensor]) -> Dict[str, torch.Tensor]:
mean = policy_params["loc"] mean = policy_params["loc"]
old_mean = old_policy_params["loc"] old_mean = old_policy_params["loc"]
scale_or_sqrt = scale_tril_to_sqrt(policy_params[self.in_keys[1]])
old_scale_or_sqrt = scale_tril_to_sqrt(old_policy_params[self.in_keys[1]]) # scale_tril is already Cholesky of matrix sqrt
scale_sqrt = policy_params[self.in_keys[1]]
old_scale_sqrt = old_policy_params[self.in_keys[1]]
mean_part, cov_part = gaussian_wasserstein_commutative(None, (mean, scale_or_sqrt), (old_mean, old_scale_or_sqrt), self.scale_prec) if not self.contextual_std:
scale_sqrt = scale_sqrt[:1]
old_scale_sqrt = old_scale_sqrt[:1]
mean_part, scale_part = self._gaussian_wasserstein(
(mean, scale_sqrt),
(old_mean, old_scale_sqrt)
)
proj_mean = self._mean_projection(mean, old_mean, mean_part) proj_mean = self._mean_projection(mean, old_mean, mean_part)
proj_scale_or_sqrt = self._cov_projection(scale_or_sqrt, old_scale_or_sqrt, cov_part) proj_scale_sqrt = self._scale_projection(scale_sqrt, old_scale_sqrt, scale_part)
return {"loc": proj_mean, self.out_keys[1]: proj_scale_or_sqrt} if not self.contextual_std:
proj_scale_sqrt = proj_scale_sqrt.expand(mean.shape[0], *proj_scale_sqrt.shape[1:])
return {"loc": proj_mean, self.out_keys[1]: proj_scale_sqrt}
def _gaussian_wasserstein(self, p: Tuple[torch.Tensor, torch.Tensor],
q: Tuple[torch.Tensor, torch.Tensor]) -> Tuple[torch.Tensor, torch.Tensor]:
mean, scale_sqrt = p
mean_other, scale_sqrt_other = q
mean_part = torch.sum(torch.square(mean - mean_other), dim=-1)
if not self.full_cov:
# Diagonal case is simpler
scale_part = torch.sum(
scale_sqrt_other**2 + scale_sqrt**2 -
2 * scale_sqrt_other * scale_sqrt,
dim=-1
)
else:
# Full covariance case uses matrix operations
scale_part = torch.sum(
scale_sqrt_other**2 + scale_sqrt**2 -
2 * torch.matmul(scale_sqrt_other, scale_sqrt.transpose(-1, -2)),
dim=(-2, -1)
)
return mean_part, scale_part
def get_trust_region_loss(self, policy_params: Dict[str, torch.Tensor], proj_policy_params: Dict[str, torch.Tensor]) -> torch.Tensor: def get_trust_region_loss(self, policy_params: Dict[str, torch.Tensor], proj_policy_params: Dict[str, torch.Tensor]) -> torch.Tensor:
mean = policy_params["loc"] mean = policy_params["loc"]
proj_mean = proj_policy_params["loc"] proj_mean = proj_policy_params["loc"]
scale_or_sqrt = scale_tril_to_sqrt(policy_params[self.in_keys[1]]) scale_or_sqrt = scale_tril_to_sqrt(policy_params[self.in_keys[1]])
proj_scale_or_sqrt = scale_tril_to_sqrt(proj_policy_params[self.out_keys[1]]) proj_scale_or_sqrt = scale_tril_to_sqrt(proj_policy_params[self.out_keys[1]])
mean_part, cov_part = gaussian_wasserstein_commutative(None, (mean, scale_or_sqrt), (proj_mean, proj_scale_or_sqrt), self.scale_prec)
w2 = mean_part + cov_part mean_part, scale_part = gaussian_wasserstein_commutative(
(mean, scale_or_sqrt),
(proj_mean, proj_scale_or_sqrt),
self.scale_prec
)
w2 = mean_part + scale_part
return w2.mean() * self.trust_region_coeff return w2.mean() * self.trust_region_coeff
def _mean_projection(self, mean: torch.Tensor, old_mean: torch.Tensor, mean_part: torch.Tensor) -> torch.Tensor: def _scale_projection(self, scale_or_sqrt: torch.Tensor, old_scale_or_sqrt: torch.Tensor, scale_part: torch.Tensor) -> torch.Tensor:
diff = mean - old_mean """Project scale parameters using multiplicative update."""
norm = torch.sqrt(mean_part)
return torch.where(norm > self.mean_bound, old_mean + diff * self.mean_bound / norm.unsqueeze(-1), mean)
def _cov_projection(self, scale_or_sqrt: torch.Tensor, old_scale_or_sqrt: torch.Tensor, cov_part: torch.Tensor) -> torch.Tensor:
if scale_or_sqrt.dim() == old_scale_or_sqrt.dim() == 2: # Diagonal case if scale_or_sqrt.dim() == old_scale_or_sqrt.dim() == 2: # Diagonal case
diff = scale_or_sqrt - old_scale_or_sqrt return self._diagonal_scale_projection(scale_or_sqrt, old_scale_or_sqrt, scale_part)
norm = torch.sqrt(cov_part)
return torch.where(norm > self.cov_bound, old_scale_or_sqrt + diff * self.cov_bound / norm.unsqueeze(-1), scale_or_sqrt)
else: # Full covariance case else: # Full covariance case
diff = scale_or_sqrt - old_scale_or_sqrt return self._full_cov_scale_projection(scale_or_sqrt, old_scale_or_sqrt, scale_part)
norm = torch.norm(diff, dim=(-2, -1), keepdim=True)
return torch.where(norm > self.cov_bound, old_scale_or_sqrt + diff * self.cov_bound / norm, scale_or_sqrt) def _diagonal_scale_projection(self, scale: torch.Tensor, old_scale: torch.Tensor, scale_part: torch.Tensor) -> torch.Tensor:
cov_mask = scale_part > self.cov_bound
batch_shape = scale.shape[:-1]
eta = torch.ones(batch_shape, dtype=scale.dtype, device=scale.device)
eta = torch.where(cov_mask,
torch.sqrt(scale_part / self.cov_bound) - 1.,
eta)
eta = torch.maximum(-eta, eta)
new_scale = (scale + eta[..., None] * old_scale) / \
(1. + eta + 1e-16)[..., None]
mask_matrix = cov_mask[..., None].to(scale.dtype)
return torch.where(mask_matrix, new_scale, scale)
def _full_cov_scale_projection(self, scale_sqrt: torch.Tensor, old_scale_sqrt: torch.Tensor, scale_part: torch.Tensor) -> torch.Tensor:
cov_mask = scale_part > self.cov_bound
batch_shape = scale_sqrt.shape[:-2]
eta = torch.ones(batch_shape, dtype=scale_sqrt.dtype, device=scale_sqrt.device)
eta = torch.where(cov_mask,
torch.sqrt(scale_part / self.cov_bound) - 1.,
eta)
eta = torch.maximum(-eta, eta)
new_scale = (scale_sqrt + torch.einsum('...,...ij->...ij', eta, old_scale_sqrt)) / \
(1. + eta + 1e-16)[..., None, None]
mask_matrix = cov_mask[..., None, None].to(scale_sqrt.dtype)
return torch.where(mask_matrix, new_scale, scale_sqrt)
-61
View File
@@ -1,61 +0,0 @@
import gymnasium
from gymnasium.spaces import Discrete as GymnasiumDiscrete, MultiDiscrete as GymnasiumMultiDiscrete, MultiBinary as GymnasiumMultiBinary, Box as GymnasiumBox
from torchrl.data.tensor_specs import (
DiscreteTensorSpec, OneHotDiscreteTensorSpec, MultiDiscreteTensorSpec,
BinaryDiscreteTensorSpec, BoundedTensorSpec, UnboundedContinuousTensorSpec
)
try:
import gym
from gym.spaces import Discrete as GymDiscrete, MultiDiscrete as GymMultiDiscrete, MultiBinary as GymMultiBinary, Box as GymBox
gym_available = True
except ImportError:
gym_available = False
def is_discrete_space(action_space):
discrete_types = (
GymnasiumDiscrete, GymnasiumMultiDiscrete, GymnasiumMultiBinary,
DiscreteTensorSpec, OneHotDiscreteTensorSpec, MultiDiscreteTensorSpec, BinaryDiscreteTensorSpec
)
continuous_types = (
GymnasiumBox, BoundedTensorSpec, UnboundedContinuousTensorSpec
)
if gym_available:
discrete_types += (GymDiscrete, GymMultiDiscrete, GymMultiBinary)
continuous_types += (GymBox,)
if isinstance(action_space, discrete_types):
return True
elif isinstance(action_space, continuous_types):
return False
else:
raise ValueError(f"Unsupported action space type: {type(action_space)}")
def get_space_shape(action_space):
discrete_types = (GymnasiumDiscrete, GymnasiumMultiDiscrete, GymnasiumMultiBinary,
DiscreteTensorSpec, OneHotDiscreteTensorSpec, MultiDiscreteTensorSpec, BinaryDiscreteTensorSpec)
continuous_types = (GymnasiumBox, BoundedTensorSpec, UnboundedContinuousTensorSpec)
if gym_available:
discrete_types += (GymDiscrete, GymMultiDiscrete, GymMultiBinary)
continuous_types += (GymBox,)
if isinstance(action_space, discrete_types):
if isinstance(action_space, (GymnasiumDiscrete, DiscreteTensorSpec, OneHotDiscreteTensorSpec)):
return (action_space.n,)
elif isinstance(action_space, (GymnasiumMultiDiscrete, MultiDiscreteTensorSpec)):
return (sum(action_space.nvec),)
elif isinstance(action_space, (GymnasiumMultiBinary, BinaryDiscreteTensorSpec)):
return (action_space.n,)
elif gym_available:
if isinstance(action_space, GymDiscrete):
return (action_space.n,)
elif isinstance(action_space, GymMultiDiscrete):
return (sum(action_space.nvec),)
elif isinstance(action_space, GymMultiBinary):
return (action_space.n,)
elif isinstance(action_space, continuous_types):
return action_space.shape
raise ValueError(f"Unsupported action space type: {type(action_space)}")
+4 -2
View File
@@ -8,7 +8,7 @@
description = "Minimalistic and efficient implementations of PPO and TRPL for torchrl" description = "Minimalistic and efficient implementations of PPO and TRPL for torchrl"
authors = [{name = "Dominik Roth", email = "mail@dominik-roth.eu"}] authors = [{name = "Dominik Roth", email = "mail@dominik-roth.eu"}]
readme = "README.md" readme = "README.md"
requires-python = ">=3.7,<3.12" requires-python = ">=3.7"
classifiers = [ classifiers = [
"Development Status :: 3 - Alpha", "Development Status :: 3 - Alpha",
"Intended Audience :: Developers", "Intended Audience :: Developers",
@@ -23,9 +23,10 @@
dependencies = [ dependencies = [
"numpy", "numpy",
"torch", "torch",
"gymnasium", "gymnasium<1.0",
"tensordict", "tensordict",
"torchrl", "torchrl",
"pytest",
] ]
[project.urls] [project.urls]
@@ -33,3 +34,4 @@
[project.optional-dependencies] [project.optional-dependencies]
dev = ["pytest"] dev = ["pytest"]
box2d = ["swig", "gymnasium[box2d]"]
+23 -47
View File
@@ -2,9 +2,12 @@ import pytest
import numpy as np import numpy as np
from fancy_rl import PPO from fancy_rl import PPO
import gymnasium as gym import gymnasium as gym
from torchrl.envs import GymEnv
import torch as th
from tensordict import TensorDict
def simple_env(): def simple_env():
return gym.make('LunarLander-v2', continuous=True) return gym.make('LunarLander-v2')
def test_ppo_instantiation(): def test_ppo_instantiation():
ppo = PPO(simple_env) ppo = PPO(simple_env)
@@ -14,65 +17,38 @@ def test_ppo_instantiation_from_str():
ppo = PPO('CartPole-v1') ppo = PPO('CartPole-v1')
assert isinstance(ppo, PPO) assert isinstance(ppo, PPO)
@pytest.mark.parametrize("learning_rate", [1e-4, 3e-4, 1e-3])
@pytest.mark.parametrize("n_steps", [1024, 2048])
@pytest.mark.parametrize("batch_size", [32, 64, 128])
@pytest.mark.parametrize("n_epochs", [5, 10])
@pytest.mark.parametrize("gamma", [0.95, 0.99])
@pytest.mark.parametrize("clip_range", [0.1, 0.2, 0.3])
def test_ppo_initialization_with_different_hps(learning_rate, n_steps, batch_size, n_epochs, gamma, clip_range):
ppo = PPO(
simple_env,
learning_rate=learning_rate,
n_steps=n_steps,
batch_size=batch_size,
n_epochs=n_epochs,
gamma=gamma,
clip_range=clip_range
)
assert ppo.learning_rate == learning_rate
assert ppo.n_steps == n_steps
assert ppo.batch_size == batch_size
assert ppo.n_epochs == n_epochs
assert ppo.gamma == gamma
assert ppo.clip_range == clip_range
def test_ppo_predict(): def test_ppo_predict():
ppo = PPO(simple_env) ppo = PPO(simple_env)
env = ppo.make_env() env = ppo.make_env()
obs, _ = env.reset() obs = env.reset()
action, _ = ppo.predict(obs) action = ppo.predict(obs)
assert isinstance(action, np.ndarray) assert isinstance(action, TensorDict)
assert action.shape == env.action_space.shape
# Handle both single and composite action spaces
def test_ppo_learn(): if isinstance(env.action_space, list):
ppo = PPO(simple_env, n_steps=64, batch_size=32) expected_shape = (len(env.action_space),) + env.action_space[0].shape
env = ppo.make_env() else:
obs, _ = env.reset() expected_shape = env.action_space.shape
for _ in range(64):
action, _ = ppo.predict(obs) assert action["action"].shape == expected_shape
obs, reward, done, truncated, _ = env.step(action)
if done or truncated:
obs, _ = env.reset()
def test_ppo_training(): def test_ppo_training():
ppo = PPO(simple_env, total_timesteps=10000) ppo = PPO(simple_env, total_timesteps=100)
env = ppo.make_env() env = ppo.make_env()
initial_performance = evaluate_policy(ppo, env) initial_performance = evaluate_policy(ppo, env)
ppo.train() ppo.train()
final_performance = evaluate_policy(ppo, env) final_performance = evaluate_policy(ppo, env)
assert final_performance > initial_performance, "PPO should improve performance after training"
def evaluate_policy(policy, env, n_eval_episodes=10): def evaluate_policy(policy, env, n_eval_episodes=3):
total_reward = 0 total_reward = 0
for _ in range(n_eval_episodes): for _ in range(n_eval_episodes):
obs, _ = env.reset() tensordict = env.reset()
done = False done = False
while not done: while not done:
action, _ = policy.predict(obs) action = policy.predict(tensordict)
obs, reward, terminated, truncated, _ = env.step(action) next_tensordict = env.step(action).get("next")
total_reward += reward total_reward += next_tensordict["reward"]
done = terminated or truncated done = next_tensordict["done"]
tensordict = next_tensordict
return total_reward / n_eval_episodes return total_reward / n_eval_episodes
+22 -32
View File
@@ -2,9 +2,10 @@ import pytest
import numpy as np import numpy as np
from fancy_rl import TRPL from fancy_rl import TRPL
import gymnasium as gym import gymnasium as gym
from tensordict import TensorDict
def simple_env(): def simple_env():
return gym.make('LunarLander-v2', continuous=True) return gym.make('Pendulum-v1')
def test_trpl_instantiation(): def test_trpl_instantiation():
trpl = TRPL(simple_env) trpl = TRPL(simple_env)
@@ -34,52 +35,41 @@ def test_trpl_initialization_with_different_hps(learning_rate, n_steps, batch_si
assert trpl.n_steps == n_steps assert trpl.n_steps == n_steps
assert trpl.batch_size == batch_size assert trpl.batch_size == batch_size
assert trpl.gamma == gamma assert trpl.gamma == gamma
assert trpl.projection.trust_region_bound_mean == trust_region_bound_mean assert trpl.projection.mean_bound == trust_region_bound_mean
assert trpl.projection.trust_region_bound_cov == trust_region_bound_cov assert trpl.projection.cov_bound == trust_region_bound_cov
def test_trpl_predict(): def test_trpl_predict():
trpl = TRPL(simple_env) trpl = TRPL(simple_env)
env = trpl.make_env() env = trpl.make_env()
obs, _ = env.reset() obs = env.reset()
action, _ = trpl.predict(obs) action = trpl.predict(obs)
assert isinstance(action, np.ndarray) assert isinstance(action, TensorDict)
assert action.shape == env.action_space.shape
def test_trpl_learn():
trpl = TRPL(simple_env, n_steps=64, batch_size=32)
env = trpl.make_env()
obs, _ = env.reset()
for _ in range(64):
action, _ = trpl.predict(obs)
next_obs, reward, done, truncated, _ = env.step(action)
trpl.store_transition(obs, action, reward, done, next_obs)
obs = next_obs
if done or truncated:
obs, _ = env.reset()
loss = trpl.learn() # Handle both single and composite action spaces
assert isinstance(loss, dict) if isinstance(env.action_space, list):
assert "policy_loss" in loss expected_shape = (len(env.action_space),) + env.action_space[0].shape
assert "value_loss" in loss else:
expected_shape = env.action_space.shape
assert action["action"].shape == expected_shape
def test_trpl_training(): def test_trpl_training():
trpl = TRPL(simple_env, total_timesteps=10000) trpl = TRPL(simple_env, total_timesteps=100)
env = trpl.make_env() env = trpl.make_env()
initial_performance = evaluate_policy(trpl, env) initial_performance = evaluate_policy(trpl, env)
trpl.train() trpl.train()
final_performance = evaluate_policy(trpl, env) final_performance = evaluate_policy(trpl, env)
assert final_performance > initial_performance, "TRPL should improve performance after training"
def evaluate_policy(policy, env, n_eval_episodes=10): def evaluate_policy(policy, env, n_eval_episodes=3):
total_reward = 0 total_reward = 0
for _ in range(n_eval_episodes): for _ in range(n_eval_episodes):
obs, _ = env.reset() tensordict = env.reset()
done = False done = False
while not done: while not done:
action, _ = policy.predict(obs) action = policy.predict(tensordict)
obs, reward, terminated, truncated, _ = env.step(action) next_tensordict = env.step(action).get("next")
total_reward += reward total_reward += next_tensordict["reward"]
done = terminated or truncated done = next_tensordict["done"]
tensordict = next_tensordict
return total_reward / n_eval_episodes return total_reward / n_eval_episodes