Refactor SACPolicy for improved readability and action dimension handling

- Cleaned up code formatting for better readability, including consistent spacing and removal of unnecessary blank lines.
- Consolidated continuous action dimension calculation to enhance clarity and maintainability.
- Simplified loss return statements in the forward method to improve code structure.
- Ensured grasp critic parameters are included conditionally based on configuration settings.
This commit is contained in:
AdilZouitine 2025-04-01 15:43:29 +00:00
parent 451a7b01db
commit 699d374d89
2 changed files with 37 additions and 36 deletions

View File

@ -35,6 +35,7 @@ from lerobot.common.policies.utils import get_device_from_parameters
DISCRETE_DIMENSION_INDEX = -1 # Gripper is always the last dimension DISCRETE_DIMENSION_INDEX = -1 # Gripper is always the last dimension
class SACPolicy( class SACPolicy(
PreTrainedPolicy, PreTrainedPolicy,
): ):
@ -50,6 +51,10 @@ class SACPolicy(
config.validate_features() config.validate_features()
self.config = config self.config = config
continuous_action_dim = config.output_features["action"].shape[0]
if config.num_discrete_actions is not None:
continuous_action_dim -= 1
if config.dataset_stats is not None: if config.dataset_stats is not None:
input_normalization_params = _convert_normalization_params_to_tensor(config.dataset_stats) input_normalization_params = _convert_normalization_params_to_tensor(config.dataset_stats)
self.normalize_inputs = Normalize( self.normalize_inputs = Normalize(
@ -117,10 +122,7 @@ class SACPolicy(
self.grasp_critic = None self.grasp_critic = None
self.grasp_critic_target = None self.grasp_critic_target = None
continuous_action_dim = config.output_features["action"].shape[0]
if config.num_discrete_actions is not None: if config.num_discrete_actions is not None:
continuous_action_dim -= 1
# Create grasp critic # Create grasp critic
self.grasp_critic = GraspCritic( self.grasp_critic = GraspCritic(
encoder=encoder_critic, encoder=encoder_critic,
@ -142,7 +144,6 @@ class SACPolicy(
self.grasp_critic = torch.compile(self.grasp_critic) self.grasp_critic = torch.compile(self.grasp_critic)
self.grasp_critic_target = torch.compile(self.grasp_critic_target) self.grasp_critic_target = torch.compile(self.grasp_critic_target)
self.actor = Policy( self.actor = Policy(
encoder=encoder_actor, encoder=encoder_actor,
network=MLP(input_dim=encoder_actor.output_dim, **asdict(config.actor_network_kwargs)), network=MLP(input_dim=encoder_actor.output_dim, **asdict(config.actor_network_kwargs)),
@ -162,11 +163,14 @@ class SACPolicy(
self.temperature = self.log_alpha.exp().item() self.temperature = self.log_alpha.exp().item()
def get_optim_params(self) -> dict: def get_optim_params(self) -> dict:
return { optim_params = {
"actor": self.actor.parameters_to_optimize, "actor": self.actor.parameters_to_optimize,
"critic": self.critic_ensemble.parameters_to_optimize, "critic": self.critic_ensemble.parameters_to_optimize,
"temperature": self.log_alpha, "temperature": self.log_alpha,
} }
if self.config.num_discrete_actions is not None:
optim_params["grasp_critic"] = self.grasp_critic.parameters_to_optimize
return optim_params
def reset(self): def reset(self):
"""Reset the policy""" """Reset the policy"""
@ -283,18 +287,21 @@ class SACPolicy(
return {"loss_critic": loss_critic} return {"loss_critic": loss_critic}
if model == "actor": if model == "actor":
return {"loss_actor": self.compute_loss_actor( return {
"loss_actor": self.compute_loss_actor(
observations=observations, observations=observations,
observation_features=observation_features, observation_features=observation_features,
)} )
}
if model == "temperature": if model == "temperature":
return {"loss_temperature": self.compute_loss_temperature( return {
"loss_temperature": self.compute_loss_temperature(
observations=observations, observations=observations,
observation_features=observation_features, observation_features=observation_features,
)} )
}
raise ValueError(f"Unknown model type: {model}") raise ValueError(f"Unknown model type: {model}")
@ -413,9 +420,7 @@ class SACPolicy(
# Use gather to select Q-values for best actions # Use gather to select Q-values for best actions
target_next_grasp_q = torch.gather( target_next_grasp_q = torch.gather(
target_next_grasp_qs, target_next_grasp_qs, dim=1, index=best_next_grasp_action.unsqueeze(-1)
dim=1,
index=best_next_grasp_action.unsqueeze(-1)
).squeeze(-1) ).squeeze(-1)
# Compute target Q-value with Bellman equation # Compute target Q-value with Bellman equation
@ -425,11 +430,7 @@ class SACPolicy(
predicted_grasp_qs = self.grasp_critic_forward(observations=observations, use_target=False) predicted_grasp_qs = self.grasp_critic_forward(observations=observations, use_target=False)
# Use gather to select Q-values for taken actions # Use gather to select Q-values for taken actions
predicted_grasp_q = torch.gather( predicted_grasp_q = torch.gather(predicted_grasp_qs, dim=1, index=actions.unsqueeze(-1)).squeeze(-1)
predicted_grasp_qs,
dim=1,
index=actions.unsqueeze(-1)
).squeeze(-1)
# Compute MSE loss between predicted and target Q-values # Compute MSE loss between predicted and target Q-values
grasp_critic_loss = F.mse_loss(input=predicted_grasp_q, target=target_grasp_q) grasp_critic_loss = F.mse_loss(input=predicted_grasp_q, target=target_grasp_q)

View File

@ -405,7 +405,7 @@ def add_actor_information_and_train(
optimizers["critic"].step() optimizers["critic"].step()
# Grasp critic optimization (if available) # Grasp critic optimization (if available)
if "loss_grasp_critic" in critic_output and hasattr(policy, "grasp_critic"): if "loss_grasp_critic" in critic_output:
loss_grasp_critic = critic_output["loss_grasp_critic"] loss_grasp_critic = critic_output["loss_grasp_critic"]
optimizers["grasp_critic"].zero_grad() optimizers["grasp_critic"].zero_grad()
loss_grasp_critic.backward() loss_grasp_critic.backward()