ExperimentValidator(encoding_pipeline, algorithm, decoder, observation_space, action_space, loss, is_tokenized=False, tokenized_obs_keys=None, image_norm_type=None, quantization_config=None, training_config=None)
Validates experiment configuration consistency.
Validates encoder-observation consistency, decoder-encoder compatibility,
and loss key validation.
When training_config.stages is provided, the validator also owns checks required by multi-stage training:
Initialize validator with policy components.
Parameters:
| Name |
Type |
Description |
Default |
encoding_pipeline
|
EncodingPipeline
|
The encoding pipeline with configured encoders.
|
required
|
algorithm
|
DecodingAlgorithm
|
The decoding algorithm (BC, diffusion, etc.).
|
required
|
decoder
|
ActionDecoder
|
The action decoder architecture.
|
required
|
observation_space
|
ObservationSpace
|
Task observation space configuration.
|
required
|
action_space
|
ActionSpace
|
Task action space configuration.
|
required
|
loss
|
BaseLoss
|
The loss module for training.
|
required
|
is_tokenized
|
bool
|
Whether observations are tokenized.
|
False
|
tokenized_obs_keys
|
set[str] | None
|
Keys of observations that are tokenized.
|
None
|
image_norm_type
|
str | None
|
RGB image normalization type for pretrained vision
encoder validation.
|
None
|
quantization_config
|
BaseQuantizationWorkflow | None
|
Optional quantization configuration to validate.
|
None
|
training_config
|
TrainingConfig | None
|
Training configuration. When provided with
stages, the validator also checks stage ordering, optimizer
group references, and loss-weight paths against the loss tree.
|
None
|
Source code in src/versatil/validation.py
| def __init__(
self,
encoding_pipeline: EncodingPipeline,
algorithm: DecodingAlgorithm,
decoder: ActionDecoder,
observation_space: ObservationSpace,
action_space: ActionSpace,
loss: BaseLoss,
is_tokenized: bool = False,
tokenized_obs_keys: set[str] | None = None,
image_norm_type: str | None = None,
quantization_config: BaseQuantizationWorkflow | None = None,
training_config: TrainingConfig | None = None,
):
"""Initialize validator with policy components.
Args:
encoding_pipeline: The encoding pipeline with configured encoders.
algorithm: The decoding algorithm (BC, diffusion, etc.).
decoder: The action decoder architecture.
observation_space: Task observation space configuration.
action_space: Task action space configuration.
loss: The loss module for training.
is_tokenized: Whether observations are tokenized.
tokenized_obs_keys: Keys of observations that are tokenized.
image_norm_type: RGB image normalization type for pretrained vision
encoder validation.
quantization_config: Optional quantization configuration to validate.
training_config: Training configuration. When provided with
``stages``, the validator also checks stage ordering, optimizer
group references, and loss-weight paths against the loss tree.
"""
self.encoding_pipeline = encoding_pipeline
self.algorithm = algorithm
self.decoder = decoder
self.observation_space = observation_space
self.action_space = action_space
self.loss = loss
self.is_tokenized = is_tokenized
self.tokenized_obs_keys = tokenized_obs_keys or set()
self.image_norm_type = image_norm_type
self.quantization_config = quantization_config
self.training_config = training_config
self.errors: list[str] = []
self.warnings: list[str] = []
|
validate_all(validate_loss_keys=True)
Run all validation checks and raise if any fail.
Parameters:
| Name |
Type |
Description |
Default |
validate_loss_keys
|
bool
|
Whether to validate loss keys against action heads.
|
True
|
Source code in src/versatil/validation.py
| def validate_all(self, validate_loss_keys: bool = True) -> None:
"""Run all validation checks and raise if any fail.
Args:
validate_loss_keys: Whether to validate loss keys against action heads.
"""
self.validate_encoder_observation_consistency()
self.validate_decoder_encoder_compatibility()
self.validate_loss_algorithm_compatibility()
if validate_loss_keys:
self.validate_loss_keys()
if self.quantization_config is not None:
self.validate_quantization()
if self.training_config is not None and self.training_config.stages:
self.validate_stage_ordering()
self.validate_stage_group_references()
self.validate_stage_loss_paths()
if self.errors:
error_msg = "\n".join([f" - {err}" for err in self.errors])
raise ExperimentValidationError(
f"Policy validation failed with {len(self.errors)} error(s):\n{error_msg}"
)
if self.warnings:
warning_msg = "\n".join([f" - {warn}" for warn in self.warnings])
logging.warning(
msg=f"Policy validation warnings ({len(self.warnings)}):\n{warning_msg}"
)
|
validate_encoder_observation_consistency()
Validate that encoder inputs match available observations.
Source code in src/versatil/validation.py
| def validate_encoder_observation_consistency(self) -> None:
"""Validate that encoder inputs match available observations."""
has_language = (
ObsKey.LANGUAGE.value in self.observation_space.observations_metadata
)
if has_language and not self.is_tokenized:
self.errors.append(
"Language observations are enabled but tokenization is disabled. "
"Language observations require tokenization to be enabled."
)
return
available_keys = self._available_observation_keys()
if (
has_language
and self.is_tokenized
and self.tokenized_obs_keys
and ObsKey.LANGUAGE.value not in self.tokenized_obs_keys
):
self.errors.append(
f"Language observations are enabled but '{ObsKey.LANGUAGE.value}' is not in "
f"observation_tokenizer.observation_keys: {self.tokenized_obs_keys}"
)
configured_encoder_inputs = set()
for encoder_name, encoder in self.encoding_pipeline.encoders.items():
encoder: EncodingMixin
self._validate_encoder_image_normalization(
encoder_name=encoder_name,
encoder=encoder,
)
input_keys = encoder.input_specification.keys
if isinstance(input_keys, str):
input_keys = [input_keys]
configured_encoder_inputs.update(input_keys)
missing = set(input_keys) - available_keys
if missing:
self.errors.append(
f"Encoder '{encoder_name}' requires keys {missing} "
f"which are not in observation space. "
f"Available keys: {available_keys}. "
f"Please either add them to the observation space or modify encoder configuration."
)
self._validate_camera_modality_constraints(
owner_name=f"Encoder '{encoder_name}'",
input_specification=encoder.input_specification,
)
for key in input_keys:
metadata = self.observation_space.observations_metadata.get(key)
if metadata is None:
continue
error = encoder.validate_input_metadata(key=key, metadata=metadata)
if error:
self.errors.append(f"Encoder '{encoder_name}': {error}")
configured_encoder_inputs.update(
self._validate_decoder_observation_inputs(available_keys=available_keys)
)
uncovered_keys = available_keys - configured_encoder_inputs
uncovered_keys -= {
SampleKey.TOKENIZED_OBSERVATIONS.value,
SampleKey.IS_PAD_OBSERVATION.value,
}
if uncovered_keys:
self.warnings.append(
f"Observation space contains keys {uncovered_keys} "
f"but no encoder is configured to process them."
)
|
validate_decoder_encoder_compatibility()
Validate that decoder inputs match encoder outputs or raw observations.
Source code in src/versatil/validation.py
| def validate_decoder_encoder_compatibility(self) -> None:
"""Validate that decoder inputs match encoder outputs or raw observations."""
available_features = self.encoding_pipeline.get_features()
available_observation_keys = self._available_observation_keys()
available_feature_names = set(available_features.keys())
available_input_names = (
available_feature_names
| available_observation_keys
| self.algorithm.injected_feature_keys()
)
decoder_input_keys = self.decoder.decoder_input.keys
for expected_feature in decoder_input_keys:
if expected_feature not in available_input_names:
self.errors.append(
f"Action decoder expects input key '{expected_feature}' but it "
"is neither a raw observation nor produced by any encoder or "
"fusion layer. Available raw observations: "
f"{sorted(available_observation_keys)}. Available encoded "
f"features: {sorted(available_feature_names)}"
)
self.decoder.decoder_input.validate_feature_types(
available_features={
**available_features,
**self._observation_feature_metadata(),
}
)
if isinstance(self.decoder, MoEDecoder):
self._validate_moe_gating_feature(sorted(available_feature_names))
|
validate_loss_algorithm_compatibility()
Validate that no loss module requires action-space targets when the algorithm predicts outside it.
Source code in src/versatil/validation.py
| def validate_loss_algorithm_compatibility(self) -> None:
"""Validate that no loss module requires action-space targets when the algorithm predicts outside it."""
if self.algorithm.predicts_in_action_space:
return
algorithm_name = type(self.algorithm).__name__
for name, loss_module in self.loss.loss_modules.items():
if loss_module.requires_action_space_targets:
self.errors.append(
f"Loss module '{name}' requires action-space targets "
f"but algorithm '{algorithm_name}' predicts outside the "
f"action space (e.g. velocity or noise). Use a regression "
f"loss (MSE/L1) instead."
)
|
Validate that loss keys reference valid action heads or auxiliary keys.
Source code in src/versatil/validation.py
| def validate_loss_keys(self) -> None:
"""Validate that loss keys reference valid action heads or auxiliary keys."""
valid_loss_keys: set[str] = set()
valid_loss_keys.update(self.decoder.get_loss_output_keys())
for key, meta in self.action_space.actions_metadata.items():
if not meta.requires_prediction_head:
valid_loss_keys.add(key)
valid_loss_keys.update(self.algorithm.get_auxiliary_output_keys())
valid_loss_keys.update(self.decoder.get_auxiliary_output_keys())
required_keys = self.loss.get_required_keys()
invalid_keys = required_keys - valid_loss_keys
if invalid_keys:
self.errors.append(
f"Loss module references keys {invalid_keys} that are not "
f"defined in the action space or auxiliary keys. "
f"Valid loss keys: {valid_loss_keys}. "
f"Please update your loss configuration or decoder."
)
|
Validate that quantization config is present and is a valid strategy.
Source code in src/versatil/validation.py
| def validate_quantization(self) -> None:
"""Validate that quantization config is present and is a valid strategy."""
config = self.quantization_config
if config is None:
return
if not isinstance(config, BaseQuantizationWorkflow):
self.errors.append(
f"Quantization config is type {type(config).__name__}, "
"expected a quantization workflow with quantization_mode. "
f"This may indicate incorrect Hydra instantiation."
)
|
validate_stage_ordering()
Validate stage names are unique and start epochs are strictly ordered.
Note
Gaps between stages are allowed. Those epochs fall back to the cached
base regime at runtime.
Source code in src/versatil/validation.py
| def validate_stage_ordering(self) -> None:
"""Validate stage names are unique and start epochs are strictly ordered.
Note:
Gaps between stages are allowed. Those epochs fall back to the cached
base regime at runtime.
"""
stages = self.training_config.stages
names = [stage.name for stage in stages]
duplicates = sorted({name for name in names if names.count(name) > 1})
if duplicates:
self.errors.append(f"Training stage names must be unique: {duplicates}.")
for previous, current in zip(stages, stages[1:], strict=False):
if current.start_epoch <= previous.start_epoch:
self.errors.append(
"training.stages must be listed in strictly increasing "
"start_epoch order."
)
break
for previous, current in zip(stages, stages[1:], strict=False):
if (
previous.end_epoch is not None
and previous.end_epoch > current.start_epoch
):
self.errors.append("training.stages intervals must not overlap.")
break
|
validate_stage_group_references()
Validate that every staged group name exists in the optimizer layout.
The reserved unmatched group is always considered available because
LightningPolicy injects it when building optimizer parameter groups.
Source code in src/versatil/validation.py
| def validate_stage_group_references(self) -> None:
"""Validate that every staged group name exists in the optimizer layout.
The reserved unmatched group is always considered available because
``LightningPolicy`` injects it when building optimizer parameter groups.
"""
available_groups = {OPTIMIZER_UNMATCHED_GROUPS_NAME}
available_groups.update(
group.name for group in self.training_config.optimizer.param_groups
)
for stage in self.training_config.stages:
referenced = (
set(stage.trainable_groups)
| set(stage.frozen_groups)
| set(stage.group_lrs)
| set(stage.group_weight_decays)
)
missing = sorted(referenced - available_groups)
if missing:
self.errors.append(
f"Training stage '{stage.name}' references unknown optimizer "
f"groups {missing}. Available groups: "
f"{sorted(available_groups)}."
)
|
validate_stage_loss_paths()
Validate every staged loss_weights patch against the loss tree.
stage.loss_weights must be a nested partial tree compatible with
policy.loss_module.weights. Unknown keys and dict/scalar shape
mismatches are rejected here so training fails before the callback ever
mutates runtime state.
Source code in src/versatil/validation.py
| def validate_stage_loss_paths(self) -> None:
"""Validate every staged ``loss_weights`` patch against the loss tree.
``stage.loss_weights`` must be a nested partial tree compatible with
``policy.loss_module.weights``. Unknown keys and dict/scalar shape
mismatches are rejected here so training fails before the callback ever
mutates runtime state.
"""
loss_tree = self.loss.weights
uses_loss_weights = any(
stage.loss_weights for stage in self.training_config.stages
)
if uses_loss_weights and not loss_tree:
self.errors.append(
"training.stages declare loss_weights overrides but the loss "
"module exposes no tunable weights."
)
return
for stage in self.training_config.stages:
if not stage.loss_weights:
continue
try:
_merge_weights(
existing_weights=loss_tree,
override_weights=stage.loss_weights,
)
except (KeyError, TypeError) as exc:
self.errors.append(f"Training stage '{stage.name}' loss_weights: {exc}")
|