diff --git a/fast_llm/layers/language_model/loss/config.py b/fast_llm/layers/language_model/loss/config.py index 027d53f7a..9d0fa7892 100644 --- a/fast_llm/layers/language_model/loss/config.py +++ b/fast_llm/layers/language_model/loss/config.py @@ -211,7 +211,7 @@ def loss_class(self) -> "type[LanguageModelZLoss]": class PolicyMetricsLevel(enum.StrEnum): none = "none" basic = "basic" - with_entropy = "with_entropy" + auto = "auto" @config_class() @@ -223,12 +223,13 @@ class LanguageModelPolicyGradientLossConfig(LanguageModelLossConfig): epsilon_low: float = Field(default=0.2, desc="Lower clip parameter for ratio of log probs") epsilon_high: float = Field(default=0.2, desc="Upper clip parameter for ratio of log probs") metrics: PolicyMetricsLevel = Field( - default=PolicyMetricsLevel.none, + default=PolicyMetricsLevel.auto, desc=( - "Additional diagnostic metrics to log. " - "`basic`: importance-ratio, KL and advantage statistics. " - "`with_entropy`: also log the policy entropy. " - "Not supported with pipeline_parallel > 1." + "Diagnostic metrics to log. " + "`basic`: importance-ratio, KL, advantage statistics and the policy entropy. " + "`auto`: `basic` when `pipeline_parallel == 1`, else `none`. " + "`none`: disable, adding no cost. " + "`basic` is not supported with `pipeline_parallel > 1`." ), hint=FieldHint.feature, ) @@ -286,10 +287,6 @@ def _validate(self) -> None: ) if loss.use_triton is not None: raise ValueError(f"Loss `{name}` sets `use_triton`, which has no effect on a fused child loss.") - # GSPO's per-segment metrics need the eager segment aggregation, which the shared-softmax path - # defers to the backward seam; the composite emits only its loss, so metrics would be dropped. - if isinstance(loss, LanguageModelGSPOLossConfig) and loss.metrics != PolicyMetricsLevel.none: - raise ValueError(f"Loss `{name}` requests GSPO metrics, which are unavailable in a monolithic loss.") # A single softmax serves one effective scale (stacked with the common model scale). Assert.eq(len({loss.logits_scale_factor for loss in self.losses.values()}), 1) diff --git a/fast_llm/layers/language_model/loss/policy_gradient.py b/fast_llm/layers/language_model/loss/policy_gradient.py index c2c36749a..d1112cb24 100644 --- a/fast_llm/layers/language_model/loss/policy_gradient.py +++ b/fast_llm/layers/language_model/loss/policy_gradient.py @@ -77,12 +77,19 @@ def __init__( weight=weight, register_loss=register_loss, ) - # The extra metrics need a second softmax over the full logits, which pipeline parallelism splits. - Assert.custom( - lambda metrics, pipeline_parallel: metrics == PolicyMetricsLevel.none or pipeline_parallel == 1, - config.metrics, - distributed_config.pipeline_parallel, - ) + # The metrics need the full-logits pass, which pipeline parallelism splits, so they require + # `pipeline_parallel == 1`. `auto` enables them only when that holds; an explicit level must satisfy it. + if config.metrics == PolicyMetricsLevel.auto: + self._metrics_level = ( + PolicyMetricsLevel.basic if distributed_config.pipeline_parallel == 1 else PolicyMetricsLevel.none + ) + else: + Assert.custom( + lambda metrics, pipeline_parallel: metrics == PolicyMetricsLevel.none or pipeline_parallel == 1, + config.metrics, + distributed_config.pipeline_parallel, + ) + self._metrics_level = config.metrics def _register_new_logprobs( self, @@ -97,9 +104,9 @@ def _register_new_logprobs( ) def _policy_metric_definitions(self, *extra: LossDef) -> list[LossDef]: - if self._config.metrics == PolicyMetricsLevel.none: + if self._metrics_level == PolicyMetricsLevel.none: return [] - defs = [ + return [ LossDef(f"{self._name}_old_logprobs"), LossDef(f"{self._name}_ratio_new_old"), LossDef(f"{self._name}_ratio_new_old_sum"), @@ -111,10 +118,8 @@ def _policy_metric_definitions(self, *extra: LossDef) -> list[LossDef]: LossDef(f"{self._name}_min_advantage", reduction=ReductionType.minimum), *extra, LossDef(f"{self._name}_num_tokens"), + LossDef(f"{self._name}_entropy"), ] - if self._config.metrics == PolicyMetricsLevel.with_entropy: - defs.append(LossDef(f"{self._name}_entropy")) - return defs def _register_policy_metrics( self, metrics: "PolicyMetrics", kwargs: dict[str, typing.Any], losses: dict | None @@ -176,7 +181,6 @@ def _forward_backward( epsilon_high, num_labels_in_seq, compute_metrics, - _, ) = arguments loss, grad, new_logprobs_mean = triton_grpo_loss_forward_backward( logits, @@ -214,8 +218,7 @@ def get_inputs(self, kwargs: dict[str, typing.Any], split_index: int, register: self._config.epsilon_low, self._config.epsilon_high, (self._prepare_target(kwargs[LanguageModelLossKwargs.label_counts], split_index) if register else None), - register and self._config.metrics != PolicyMetricsLevel.none, - register and self._config.metrics == PolicyMetricsLevel.with_entropy, + register and self._metrics_level != PolicyMetricsLevel.none, ) @staticmethod @@ -241,7 +244,6 @@ def fused_core( epsilon_high, num_labels_in_seq, compute_metrics, - compute_entropy, ) = arguments loss_mask = target >= 0 predicted_logits, target_masked, target_mask = predicted_logits_from_labels( @@ -269,9 +271,6 @@ def fused_core( metrics = ( grpo_metrics_core( - logits_norm, - exp_logits, - sum_exp_logits, new_log_probs, advantages, old_log_probabilities, @@ -279,8 +278,7 @@ def fused_core( num_labels_in_seq, epsilon_low, epsilon_high, - group, - compute_entropy, + _policy_entropy_per_token(logits_norm, exp_logits, sum_exp_logits, group), ) if compute_metrics else None @@ -335,7 +333,6 @@ def _register_extra_metrics( self._config.epsilon_high, self._logits_scale_factor, group=self._parallel_dim.group if self._vocab_parallel else None, - compute_entropy=self._config.metrics == PolicyMetricsLevel.with_entropy, ) self._register_policy_metrics(metrics, kwargs, losses) @@ -435,7 +432,7 @@ def _forward_backward( self._register_new_logprobs(new_logprobs_mean, kwargs, losses) # Skip the extra softmax pass when there is nothing to register. - if losses is not None and self._config.metrics != PolicyMetricsLevel.none: + if losses is not None and self._metrics_level != PolicyMetricsLevel.none: self._register_extra_metrics(logits, kwargs, losses, split_index, document_index_zero_based, num_segments) return loss, grad @@ -463,12 +460,11 @@ def _register_extra_metrics( group=self._parallel_dim.group if self._vocab_parallel else None, sdp_group=self._sequence_data_dim.group if self._sequence_data_active else None, sp_group=self._parallel_dim.group if self._sequence_parallel else None, - compute_entropy=self._config.metrics == PolicyMetricsLevel.with_entropy, ) self._register_policy_metrics(metrics, kwargs, losses) def get_inputs(self, kwargs: dict[str, typing.Any], split_index: int, register: bool) -> tuple: - return (self._get_labels(kwargs, split_index),) + return (self._get_labels(kwargs, split_index), register and self._metrics_level != PolicyMetricsLevel.none) @staticmethod def fused_core( @@ -480,37 +476,41 @@ def fused_core( logits_scale_factor: float, arguments: tuple, ) -> tuple[None, None, tuple]: - """GSPO forward over the shared softmax: the per-token log-probs plus the softmax tensors its seam and - backward need. Returns `(None, None, forward_state)` — GSPO's loss and gradient can't be produced here - (the segment aggregation is eager), so both are deferred to `finish`.""" - (target,) = arguments + """GSPO forward over the shared softmax: the per-token log-probs plus the softmax tensors its seam, + backward and (when requested) metrics need. Returns `(None, None, forward_state)` — GSPO's loss and + gradient can't be produced here (the segment aggregation is eager), so both are deferred to `finish`.""" + target, compute_metrics = arguments loss_mask = target >= 0 predicted_logits, target_masked, target_mask = predicted_logits_from_labels( logits_norm, target, loss_mask, group ) new_log_probs = predicted_logits - sum_exp_logits.log() - return None, None, (new_log_probs, loss_mask, exp_logits, sum_exp_logits, target_masked, target_mask) + return ( + None, + None, + ( + new_log_probs, + loss_mask, + exp_logits, + sum_exp_logits, + target_masked, + target_mask, + logits_norm, + compute_metrics, + ), + ) - def finish( - self, - loss: torch.Tensor | None, - extra: tuple, - kwargs: dict[str, typing.Any], - split_index: int, - grad_logits: torch.Tensor | None, - logits_dtype: torch.dtype, + def _run_segment_seam( + self, new_log_probs: torch.Tensor, loss_mask: torch.Tensor, kwargs: dict[str, typing.Any], split_index: int ) -> tuple[torch.Tensor, torch.Tensor, torch.Tensor | None]: - """Run the eager segment seam and the compiled backward over the shared softmax deferred by - `fused_core`, accumulating GSPO's gradient into `grad_logits`. Returns the loss and the `new_logprobs` - metric (registered by `register_combinable_extras`).""" - new_log_probs, loss_mask, exp_logits, sum_exp_logits, target_masked, target_mask = extra - document_index_zero_based = self._document_index_zero_based(kwargs, split_index) - loss, new_logprobs_mean, effective_grad = gspo_segment_seam( + """Run the eager segment seam from per-token new log-probs, pulling its remaining inputs from `kwargs`. + Returns the loss, the `new_logprobs` metric, and the per-token backward coefficient.""" + return gspo_segment_seam( new_log_probs, loss_mask, self._prepare_target(kwargs[LanguageModelLossKwargs.advantages], split_index), self._prepare_target(kwargs[LanguageModelLossKwargs.old_log_probabilities], split_index), - document_index_zero_based, + self._document_index_zero_based(kwargs, split_index), kwargs[BlockKwargs.num_documents_in_sequence], self._prepare_target(kwargs[LanguageModelLossKwargs.label_counts], split_index), kwargs[LanguageModelKwargs.num_documents_in_batch], @@ -521,6 +521,31 @@ def finish( self._config.epsilon_high, self._logits_scale_factor, ) + + def finish( + self, + loss: torch.Tensor | None, + extra: tuple, + kwargs: dict[str, typing.Any], + split_index: int, + grad_logits: torch.Tensor | None, + logits_dtype: torch.dtype, + ) -> tuple[torch.Tensor, tuple, torch.Tensor | None]: + """Run the eager segment seam and the compiled backward over the shared softmax deferred by + `fused_core`, accumulating GSPO's gradient into `grad_logits`. Returns the loss and the + `(new_logprobs, metrics)` extras (metrics from the same shared softmax, `None` when not requested), + both registered by `register_combinable_extras`.""" + ( + new_log_probs, + loss_mask, + exp_logits, + sum_exp_logits, + target_masked, + target_mask, + logits_norm, + compute_metrics, + ) = extra + loss, new_logprobs_mean, effective_grad = self._run_segment_seam(new_log_probs, loss_mask, kwargs, split_index) if effective_grad is not None: grad_logits = gspo_backward_core( exp_logits, @@ -532,12 +557,36 @@ def finish( logits_dtype, grad_logits, ) - return loss, new_logprobs_mean, grad_logits + metrics = ( + gspo_metrics_core( + new_log_probs, + self._prepare_target(kwargs[LanguageModelLossKwargs.advantages], split_index), + self._prepare_target(kwargs[LanguageModelLossKwargs.old_log_probabilities], split_index), + loss_mask, + self._document_index_zero_based(kwargs, split_index), + kwargs[BlockKwargs.num_documents_in_sequence], + self._prepare_target(kwargs[LanguageModelLossKwargs.label_counts], split_index), + self._config.epsilon_low, + self._config.epsilon_high, + _policy_entropy_per_token( + logits_norm, + exp_logits, + sum_exp_logits, + self._parallel_dim.group if self._vocab_parallel else None, + ), + self._sequence_data_dim.group if self._sequence_data_active else None, + self._parallel_dim.group if self._sequence_parallel else None, + ) + if compute_metrics + else None + ) + return loss, (new_logprobs_mean, metrics), grad_logits - def register_combinable_extras( - self, extra: torch.Tensor | None, kwargs: dict[str, typing.Any], losses: dict | None - ) -> None: - self._register_new_logprobs(extra, kwargs, losses) + def register_combinable_extras(self, extra: tuple, kwargs: dict[str, typing.Any], losses: dict | None) -> None: + new_logprobs_mean, metrics = extra + self._register_new_logprobs(new_logprobs_mean, kwargs, losses) + if metrics is not None: + self._register_policy_metrics(metrics, kwargs, losses) def get_loss_definitions(self) -> list[LossDef]: return super().get_loss_definitions() + self._policy_metric_definitions(LossDef(f"{self._name}_num_segments")) @@ -582,10 +631,22 @@ def _policy_metrics_reduce( ) -def grpo_metrics_core( +def _policy_entropy_per_token( logits_norm: torch.Tensor, # (*batch, vocab) exp_logits: torch.Tensor, # (*batch, vocab) sum_exp_logits: torch.Tensor, # (*batch,) + group: torch.distributed.ProcessGroup | None, +) -> torch.Tensor: + """Per-token policy entropy from a student softmax: `log(Σ exp) - E_softmax[logits_norm]`. `exp_logits` + and `logits_norm` are local vocab slices, so the expectation sums over the local slice then all-reduces + across the tensor-parallel group before dividing by the already-global `sum_exp_logits`.""" + weighted_logits_sum = (exp_logits * logits_norm).sum(-1) + if group is not None: + all_reduce(weighted_logits_sum, op=ReduceOp.SUM, group=group) + return sum_exp_logits.log() - weighted_logits_sum / sum_exp_logits + + +def grpo_metrics_core( new_log_probs: torch.Tensor, # (*batch,) — predicted_logits - log(sum_exp_logits) advantages: torch.Tensor, # (*batch,) old_log_probabilities: torch.Tensor, # (*batch,) @@ -593,23 +654,11 @@ def grpo_metrics_core( label_counts: torch.Tensor, # (*batch,) — global per-sequence count broadcast per token epsilon_low: float, epsilon_high: float, - group: "torch.distributed.ProcessGroup | None" = None, - compute_entropy: bool = False, + entropy_per_token: torch.Tensor, # (*batch,) ) -> PolicyMetrics: - """Per-token metric family from a precomputed student softmax and per-token new log-probs, so the metrics - reuse a loss kernel's softmax instead of recomputing it. Un-compiled so it inlines into a `@torch.compile` - boundary. The importance ratio's clip / KL are token-level, and the ratio variance is over tokens - (`variance_weight` = the token mask). Entropy is the only term needing the full vocab axis (a sum over the - local slice plus a tensor-parallel all-reduce).""" - entropy_per_token: torch.Tensor | None = None - if compute_entropy: - # exp_logits and logits_norm are local vocab slices — sum over the local slice, then all-reduce - # across the tensor-parallel group to recover the global expectation before dividing by the - # already-global sum_exp_logits. - weighted_logits_sum = (exp_logits * logits_norm).sum(-1) - if group is not None: - all_reduce(weighted_logits_sum, op=ReduceOp.SUM, group=group) - entropy_per_token = sum_exp_logits.log() - weighted_logits_sum / sum_exp_logits + """Per-token metric family from per-token new log-probs and a precomputed entropy. The importance ratio's + clip / KL are token-level, and the ratio variance is over tokens (`variance_weight` = the token mask). + Un-compiled so it inlines into a `@torch.compile` boundary.""" log_ratio = new_log_probs - old_log_probabilities mask = loss_mask.to(log_ratio.dtype) return _policy_metrics_reduce( @@ -639,7 +688,6 @@ def compute_grpo_metrics( epsilon_high: float = 0.2, logits_scale_factor: float = 1.0, group: torch.distributed.ProcessGroup | None = None, - compute_entropy: bool = False, ) -> PolicyMetrics: """Standalone per-token diagnostics: one softmax over the logits, then the shared `grpo_metrics_core`.""" loss_mask = target >= 0 @@ -647,9 +695,6 @@ def compute_grpo_metrics( predicted_logits, _, _ = fused_predicted_logits_from_labels(logits_norm, target, loss_mask, group) new_log_probs = predicted_logits - sum_exp_logits.log() return grpo_metrics_core( - logits_norm, - exp_logits, - sum_exp_logits, new_log_probs, advantages, old_log_probabilities, @@ -657,46 +702,32 @@ def compute_grpo_metrics( label_counts, epsilon_low, epsilon_high, - group, - compute_entropy, + _policy_entropy_per_token(logits_norm, exp_logits, sum_exp_logits, group), ) # Not @torch.compile for the same reason as `fused_gspo_loss_forward_backward`: the Python-int -# `num_segments` argument trips dynamo. Metrics run only on logging steps, so eager is fine. -def compute_gspo_metrics( - logits: torch.Tensor, # (*batch, vocab_local) - target: torch.Tensor, # (*batch,) +# `num_segments` argument in `index_add_` trips dynamo. Metrics run only on logging steps, so eager is fine. +def gspo_metrics_core( + new_log_probs: torch.Tensor, # (*batch,) — predicted_logits - log(sum_exp_logits) advantages: torch.Tensor, # (*batch,) old_log_probabilities: torch.Tensor, # (*batch,) + loss_mask: torch.Tensor, # (*batch,) bool, == target >= 0 document_index_zero_based: torch.Tensor, # (*batch,) int — segment ID per token, 0-based num_segments: int, label_counts: torch.Tensor, # (*batch,) — per-document labeled-token count broadcast per token - epsilon_low: float = 0.2, - epsilon_high: float = 0.2, - logits_scale_factor: float = 1.0, - group: torch.distributed.ProcessGroup | None = None, + epsilon_low: float, + epsilon_high: float, + entropy_per_token: torch.Tensor, # (*batch,) sdp_group: torch.distributed.ProcessGroup | None = None, sp_group: torch.distributed.ProcessGroup | None = None, - compute_entropy: bool = False, ) -> PolicyMetrics: - """Segment-level diagnostics (clipping is per document/segment): the ratio is the per-segment - geometric mean, broadcast back to tokens, and the ratio variance is over segments - (`variance_weight` = `document_weight`). The per-segment log-ratio / advantage are token-weighted - by `document_weight` (which sums to 1 per document across SDP/SP ranks) then all-reduced, so they - partition correctly across ranks and the token-level reduction matches the per-token loss.""" - loss_mask = target >= 0 - logits_norm, exp_logits, sum_exp_logits, _ = fused_softmax_base(logits, logits_scale_factor, group) - predicted_logits, _, _ = fused_predicted_logits_from_labels(logits_norm, target, loss_mask, group) - log_ratio = predicted_logits - sum_exp_logits.log() - old_log_probabilities - - entropy_per_token: torch.Tensor | None = None - if compute_entropy: - weighted_logits_sum = (exp_logits * logits_norm).sum(-1) - if group is not None: - all_reduce(weighted_logits_sum, op=ReduceOp.SUM, group=group) - entropy_per_token = sum_exp_logits.log() - weighted_logits_sum / sum_exp_logits - + """Segment-level metric family from per-token new log-probs and a precomputed entropy. Clipping is per + document/segment: the ratio is the per-segment geometric mean, broadcast back to tokens, and the ratio + variance is over segments (`variance_weight` = `document_weight`). The per-segment log-ratio / advantage + are token-weighted by `document_weight` (which sums to 1 per document across SDP/SP ranks) then + all-reduced, so they partition correctly across ranks and the token-level reduction matches the loss.""" + log_ratio = new_log_probs - old_log_probabilities flat_document_index = document_index_zero_based.reshape(-1).long() flat_mask = loss_mask.reshape(-1).to(log_ratio.dtype) document_weight = flat_mask / label_counts.reshape(-1).to(log_ratio.dtype).clamp(min=1) @@ -728,11 +759,47 @@ def compute_gspo_metrics( variance_weight=document_weight, epsilon_low=epsilon_low, epsilon_high=epsilon_high, - entropy_per_token=None if entropy_per_token is None else entropy_per_token.reshape(-1), + entropy_per_token=entropy_per_token.reshape(-1), num_segments=document_weight.sum(), ) +def compute_gspo_metrics( + logits: torch.Tensor, # (*batch, vocab_local) + target: torch.Tensor, # (*batch,) + advantages: torch.Tensor, # (*batch,) + old_log_probabilities: torch.Tensor, # (*batch,) + document_index_zero_based: torch.Tensor, # (*batch,) int — segment ID per token, 0-based + num_segments: int, + label_counts: torch.Tensor, # (*batch,) — per-document labeled-token count broadcast per token + epsilon_low: float = 0.2, + epsilon_high: float = 0.2, + logits_scale_factor: float = 1.0, + group: torch.distributed.ProcessGroup | None = None, + sdp_group: torch.distributed.ProcessGroup | None = None, + sp_group: torch.distributed.ProcessGroup | None = None, +) -> PolicyMetrics: + """Standalone segment-level diagnostics: one softmax over the logits, then the shared `gspo_metrics_core`.""" + loss_mask = target >= 0 + logits_norm, exp_logits, sum_exp_logits, _ = fused_softmax_base(logits, logits_scale_factor, group) + predicted_logits, _, _ = fused_predicted_logits_from_labels(logits_norm, target, loss_mask, group) + new_log_probs = predicted_logits - sum_exp_logits.log() + return gspo_metrics_core( + new_log_probs, + advantages, + old_log_probabilities, + loss_mask, + document_index_zero_based, + num_segments, + label_counts, + epsilon_low, + epsilon_high, + _policy_entropy_per_token(logits_norm, exp_logits, sum_exp_logits, group), + sdp_group, + sp_group, + ) + + @torch.compile def _gspo_forward_core( logits: torch.Tensor, # (*batch, vocab) diff --git a/tests/layers/test_lm_head.py b/tests/layers/test_lm_head.py index 8ceb30b41..9c233765b 100644 --- a/tests/layers/test_lm_head.py +++ b/tests/layers/test_lm_head.py @@ -13,7 +13,12 @@ from fast_llm.layers.language_model.loss.config import LanguageModelLossKwargs from fast_llm.models.gpt.config import GPTModelConfig from fast_llm.utils import Assert -from tests.layers.test_lm_losses import reference_grpo_loss, reference_grpo_metrics, reference_gspo_loss +from tests.layers.test_lm_losses import ( + reference_grpo_loss, + reference_grpo_metrics, + reference_gspo_loss, + reference_gspo_metrics, +) from tests.utils.utils import get_base_model, get_stage NUM_TOKENS = 200 @@ -42,6 +47,7 @@ class LMHeadTestConfig: gspo_document_lengths: tuple[int, ...] | None = None loss_implementation: str = "per_loss" grpo_metrics: str | None = None + gspo_metrics: str | None = None @property def actual_label_loss(self): @@ -80,13 +86,13 @@ def get_config(self) -> GPTModelConfig: if isinstance(self.z_loss, float): losses["z_loss"]["weight"] = self.z_loss if self.grpo_loss is not False: - losses["grpo_loss"] = {"type": "grpo"} + # Metrics default to `auto` (→ `basic` at `pipeline_parallel == 1`), so pin `none` explicitly + # unless the test exercises them. + losses["grpo_loss"] = {"type": "grpo", "metrics": self.grpo_metrics or "none"} if isinstance(self.grpo_loss, float): losses["grpo_loss"]["weight"] = self.grpo_loss - if self.grpo_metrics is not None: - losses["grpo_loss"]["metrics"] = self.grpo_metrics if self.gspo_loss is not False: - losses["gspo_loss"] = {"type": "gspo"} + losses["gspo_loss"] = {"type": "gspo", "metrics": self.gspo_metrics or "none"} if isinstance(self.gspo_loss, float): losses["gspo_loss"]["weight"] = self.gspo_loss if self.loss_implementation == "fused" and losses: @@ -297,7 +303,6 @@ def get_reference_outputs( epsilon_low=0.2, epsilon_high=0.2, logits_scale_factor=1.0, - compute_entropy=self.grpo_metrics == "with_entropy", ) num_documents = kwargs[LanguageModelKwargs.num_documents_in_batch] for attr in ("old_logprobs", "ratio_new_old", "kl_new_old", "clipped_ratio_fraction", "advantage"): @@ -306,8 +311,7 @@ def get_reference_outputs( names_losses_weights.append((f"grpo_loss_{attr}", getattr(metrics, attr), 0.0)) names_losses_weights.append(("grpo_loss_max_advantage", metrics.max_advantage, 0.0)) names_losses_weights.append(("grpo_loss_min_advantage", metrics.min_advantage, 0.0)) - if metrics.entropy is not None: - names_losses_weights.append(("grpo_loss_entropy", metrics.entropy / num_documents, 0.0)) + names_losses_weights.append(("grpo_loss_entropy", metrics.entropy / num_documents, 0.0)) if self.gspo_loss is not False: gspo_loss, _ = reference_gspo_loss( @@ -339,6 +343,30 @@ def get_reference_outputs( names_losses_weights.append(("gspo_loss", gspo_loss, float(self.gspo_loss))) names_losses_weights.append(("gspo_loss_new_logprobs", new_logprobs, 0.0)) + if self.gspo_metrics is not None: + # `logits` is already scaled above, so pass logits_scale_factor=1.0. + metrics = reference_gspo_metrics( + logits, + labels, + kwargs[LanguageModelLossKwargs.advantages][head._prediction_distance - 1], + kwargs[LanguageModelLossKwargs.old_log_probabilities][head._prediction_distance - 1], + kwargs[BlockKwargs.global_document_index_q].long() - 1, + kwargs[BlockKwargs.num_documents_in_sequence], + kwargs[LanguageModelLossKwargs.label_counts][head._prediction_distance - 1], + epsilon_low=0.2, + epsilon_high=0.2, + logits_scale_factor=1.0, + ) + num_documents = kwargs[LanguageModelKwargs.num_documents_in_batch] + for attr in ("old_logprobs", "ratio_new_old", "kl_new_old", "clipped_ratio_fraction", "advantage"): + names_losses_weights.append((f"gspo_loss_{attr}", getattr(metrics, attr) / num_documents, 0.0)) + for attr in ("ratio_new_old_sum", "ratio_new_old_squared_sum", "num_tokens"): + names_losses_weights.append((f"gspo_loss_{attr}", getattr(metrics, attr), 0.0)) + names_losses_weights.append(("gspo_loss_max_advantage", metrics.max_advantage, 0.0)) + names_losses_weights.append(("gspo_loss_min_advantage", metrics.min_advantage, 0.0)) + names_losses_weights.append(("gspo_loss_num_segments", metrics.num_segments, 0.0)) + names_losses_weights.append(("gspo_loss_entropy", metrics.entropy / num_documents, 0.0)) + actual_losses = [loss * weight for _, loss, weight in names_losses_weights if weight != 0.0] total_loss = sum(actual_losses) total_loss.backward() @@ -456,22 +484,31 @@ def _add_configs(base_name: str, **kwargs): loss_implementation="fused", ) ) -# GRPO metric family. Single-split only: per-split metric partials reduce across splits, which the -# whole-sequence reference doesn't model. +# GRPO/GSPO metric families (`basic` includes entropy) on the standalone and compiled shared-softmax paths. +# Single-split only: per-split metric partials reduce across splits, which the whole-sequence reference +# doesn't model. for _loss_implementation in ("per_loss", "fused"): _prefix = "" if _loss_implementation == "per_loss" else "fused_" - for _metrics in ("basic", "with_entropy"): - _suffix = "metrics" if _metrics == "basic" else "entropy" - for _loss_masking in (False, True): - _lm_head_test_configs.append( - LMHeadTestConfig( - f"{_prefix}grpo_loss_{_suffix}{'_masked' if _loss_masking else ''}", - grpo_loss=True, - grpo_metrics=_metrics, - loss_masking=_loss_masking, - loss_implementation=_loss_implementation, - ) + for _loss_masking in (False, True): + _mask_suffix = "_masked" if _loss_masking else "" + _lm_head_test_configs.append( + LMHeadTestConfig( + f"{_prefix}grpo_loss_metrics{_mask_suffix}", + grpo_loss=True, + grpo_metrics="basic", + loss_masking=_loss_masking, + loss_implementation=_loss_implementation, ) + ) + _lm_head_test_configs.append( + LMHeadTestConfig( + f"{_prefix}gspo_loss_metrics{_mask_suffix}", + gspo_loss=True, + gspo_metrics="basic", + loss_masking=_loss_masking, + loss_implementation=_loss_implementation, + ) + ) # The metric family co-resides with z-loss in the shared softmax pass. Single-split (metrics can't be split). for _loss_masking in (False, True): _lm_head_test_configs.append( @@ -484,6 +521,9 @@ def _add_configs(base_name: str, **kwargs): loss_implementation="fused", ) ) +# `auto` metrics resolve to `basic` when pipeline_parallel == 1 (all head tests), covering the default path. +_lm_head_test_configs.append(LMHeadTestConfig("grpo_loss_metrics_auto", grpo_loss=True, grpo_metrics="auto")) +_lm_head_test_configs.append(LMHeadTestConfig("gspo_loss_metrics_auto", gspo_loss=True, gspo_metrics="auto")) @pytest.mark.slow diff --git a/tests/layers/test_lm_losses.py b/tests/layers/test_lm_losses.py index 50e4327ce..03c0c6bd6 100644 --- a/tests/layers/test_lm_losses.py +++ b/tests/layers/test_lm_losses.py @@ -80,8 +80,13 @@ def _compare_losses_and_grads( ref_grad: torch.Tensor | None, threshold=1e-5, group: torch.distributed.ProcessGroup | None = None, + loss_min_threshold: float = 1e-6, ): - Assert.rms_close_relative(loss, ref_loss, threshold, 1e-6) + # `loss_min_threshold` raises the absolute floor of the scalar-loss comparison. GSPO's per-segment + # geometric-mean reduction accumulates more float32 error than a per-token sum, and its loss can be near + # zero, so the fused-vs-reference difference sits at the abs floor rather than the relative band (measured + # worst ~6e-6 over 300 random draws); the default `1e-6` floor flakes there. The gradient stays tight. + Assert.rms_close_relative(loss, ref_loss, threshold, loss_min_threshold) if has_grad: Assert.rms_close_relative( grad, split_op(ref_grad, group, -1), threshold, 1e-8 if grad.dtype == torch.float32 else 1e-7 @@ -142,7 +147,6 @@ def reference_grpo_metrics( epsilon_low: float, epsilon_high: float, logits_scale_factor: float, - compute_entropy: bool, ) -> PolicyMetrics: log_softmax = torch.nn.functional.log_softmax(logits.float() * logits_scale_factor, dim=-1) loss_mask = target >= 0 @@ -155,10 +159,8 @@ def reference_grpo_metrics( clipped = (ratio < 1.0 - epsilon_low) | (ratio > 1.0 + epsilon_high) kl = ratio - log_ratio - 1.0 - entropy = None - if compute_entropy: - entropy_per_token = -(log_softmax.exp() * log_softmax).sum(-1) - entropy = (entropy_per_token * masked).sum() + entropy_per_token = -(log_softmax.exp() * log_softmax).sum(-1) + entropy = (entropy_per_token * masked).sum() return PolicyMetrics( old_logprobs=(old_log_probabilities.float() * masked).sum(), @@ -187,7 +189,6 @@ def reference_gspo_metrics( epsilon_low: float, epsilon_high: float, logits_scale_factor: float, - compute_entropy: bool, ) -> PolicyMetrics: log_softmax = torch.nn.functional.log_softmax(logits.float() * logits_scale_factor, dim=-1) loss_mask = target >= 0 @@ -223,11 +224,9 @@ def reference_gspo_metrics( old_logprobs = old_logprobs + flat_old[in_segment].sum() / count_float num_segments_count = num_segments_count + 1.0 - entropy = None - if compute_entropy: - entropy_per_token = -(log_softmax.exp() * log_softmax).sum(-1).reshape(-1) - masked = flat_mask.float() / num_labels_in_seq.reshape(-1).float().clamp(min=1) - entropy = (entropy_per_token * masked).sum() + entropy_per_token = -(log_softmax.exp() * log_softmax).sum(-1).reshape(-1) + masked = flat_mask.float() / num_labels_in_seq.reshape(-1).float().clamp(min=1) + entropy = (entropy_per_token * masked).sum() return PolicyMetrics( old_logprobs=old_logprobs, @@ -446,7 +445,7 @@ def _test_grpo_loss( split_op(logits, group, -1), group, local_previous_grad.clone() if accumulate else None, - (target, advantages, old_log_probabilities, grad_output, divisor, 0.2, 0.2, num_labels_in_seq, False, False), + (target, advantages, old_log_probabilities, grad_output, divisor, 0.2, 0.2, num_labels_in_seq, False), ) _compare_losses_and_grads(out_fused, out_ref, grad_output is not None, grad_fused, grad_ref, group=group) @@ -534,7 +533,9 @@ def _test_gspo_loss( logits_scale_factor=logits_scale_factor, num_labels_in_seq=num_labels_in_seq, ) - _compare_losses_and_grads(out_fused, out_ref, grad_output is not None, grad_fused, grad_ref, group=group) + _compare_losses_and_grads( + out_fused, out_ref, grad_output is not None, grad_fused, grad_ref, group=group, loss_min_threshold=1e-5 + ) Assert.rms_close_relative(new_logprobs_fused, ref_new_logprobs, 1e-5, 1e-6) if not triton_available: @@ -553,7 +554,9 @@ def _test_gspo_loss( group=group, logits_scale_factor=logits_scale_factor, ) - _compare_losses_and_grads(out_triton, out_ref, grad_output is not None, grad_triton, grad_ref, group=group) + _compare_losses_and_grads( + out_triton, out_ref, grad_output is not None, grad_triton, grad_ref, group=group, loss_min_threshold=1e-5 + ) Assert.rms_close_relative(new_logprobs_triton, new_logprobs_fused, 1e-5, 1e-6) @@ -567,9 +570,7 @@ def _check_policy_metrics(ref: PolicyMetrics, got: PolicyMetrics, threshold: flo Assert.rms_close_relative(got_value, ref_value, threshold, 1e-6) -def _test_grpo_metrics( - batch_shape, num_columns, logits_scale_factor, loss_masking, dtype, compute_entropy, group=None -): +def _test_grpo_metrics(batch_shape, num_columns, logits_scale_factor, loss_masking, dtype, group=None): logits, target, advantages, old_log_probabilities = _get_grpo_loss_inputs( num_columns, loss_masking, batch_shape, dtype ) @@ -587,7 +588,6 @@ def _test_grpo_metrics( epsilon_low=0.2, epsilon_high=0.2, logits_scale_factor=logits_scale_factor, - compute_entropy=compute_entropy, ) got = compute_grpo_metrics( split_op(logits, group, -1).contiguous(), @@ -599,14 +599,11 @@ def _test_grpo_metrics( epsilon_high=0.2, logits_scale_factor=logits_scale_factor, group=group, - compute_entropy=compute_entropy, ) _check_policy_metrics(ref, got, threshold=5e-5 if dtype == DataType.float32 else 1e-4) -def _test_gspo_metrics( - batch_shape, num_columns, logits_scale_factor, loss_masking, dtype, compute_entropy, num_segments, group=None -): +def _test_gspo_metrics(batch_shape, num_columns, logits_scale_factor, loss_masking, dtype, num_segments, group=None): logits, target, advantages, old_log_probabilities = _get_grpo_loss_inputs( num_columns, loss_masking, batch_shape, dtype ) @@ -633,7 +630,6 @@ def _test_gspo_metrics( epsilon_low=0.2, epsilon_high=0.2, logits_scale_factor=logits_scale_factor, - compute_entropy=compute_entropy, ) got = compute_gspo_metrics( split_op(logits, group, -1).contiguous(), @@ -647,7 +643,6 @@ def _test_gspo_metrics( epsilon_high=0.2, logits_scale_factor=logits_scale_factor, group=group, - compute_entropy=compute_entropy, ) _check_policy_metrics(ref, got, threshold=5e-5 if dtype == DataType.float32 else 1e-4) @@ -862,7 +857,6 @@ def test_gspo_loss( ("num_columns", "grad_output", "logits_scale_factor", "loss_masking", "dtype", "block_size", "accumulate"), _LOSS_PARAMETERS, ) -@pytest.mark.parametrize("compute_entropy", (False, True)) def test_grpo_metrics( batch_shape, num_columns, @@ -872,9 +866,8 @@ def test_grpo_metrics( dtype, block_size, accumulate, - compute_entropy, ): - _test_grpo_metrics(batch_shape, num_columns, logits_scale_factor, loss_masking, dtype, compute_entropy) + _test_grpo_metrics(batch_shape, num_columns, logits_scale_factor, loss_masking, dtype) @pytest.mark.slow @@ -883,7 +876,6 @@ def test_grpo_metrics( ("num_columns", "grad_output", "logits_scale_factor", "loss_masking", "dtype", "num_segments", "accumulate"), _GSPO_PARAMETERS, ) -@pytest.mark.parametrize("compute_entropy", (False, True)) def test_gspo_metrics( batch_shape, num_columns, @@ -893,11 +885,8 @@ def test_gspo_metrics( dtype, num_segments, accumulate, - compute_entropy, ): - _test_gspo_metrics( - batch_shape, num_columns, logits_scale_factor, loss_masking, dtype, compute_entropy, num_segments - ) + _test_gspo_metrics(batch_shape, num_columns, logits_scale_factor, loss_masking, dtype, num_segments) @pytest.mark.skip(reason="DPO loss is broken") @@ -993,35 +982,31 @@ def _run_lm_loss_distributed(test_context: DistributedTestContext, base_path: pa test_context.group, ) # GRPO metrics - for compute_entropy in (False, True): - with test_context.subtest(base_path, f"grpo_metrics-{compute_entropy}-{suffix}", 2) as subtest: - if subtest.do_run: - torch.manual_seed((seed + hash(subtest.name)) % 2**32) - _test_grpo_metrics( - batch_shape, - num_columns, - logits_scale_factor, - loss_masking, - dtype, - compute_entropy, - test_context.group, - ) + with test_context.subtest(base_path, f"grpo_metrics-{suffix}", 2) as subtest: + if subtest.do_run: + torch.manual_seed((seed + hash(subtest.name)) % 2**32) + _test_grpo_metrics( + batch_shape, + num_columns, + logits_scale_factor, + loss_masking, + dtype, + test_context.group, + ) # GSPO metrics num_segments = 4 - for compute_entropy in (False, True): - with test_context.subtest(base_path, f"gspo_metrics-{compute_entropy}-{suffix}", 2) as subtest: - if subtest.do_run: - torch.manual_seed((seed + hash(subtest.name)) % 2**32) - _test_gspo_metrics( - batch_shape, - num_columns, - logits_scale_factor, - loss_masking, - dtype, - compute_entropy, - num_segments, - test_context.group, - ) + with test_context.subtest(base_path, f"gspo_metrics-{suffix}", 2) as subtest: + if subtest.do_run: + torch.manual_seed((seed + hash(subtest.name)) % 2**32) + _test_gspo_metrics( + batch_shape, + num_columns, + logits_scale_factor, + loss_masking, + dtype, + num_segments, + test_context.group, + ) # Monolithic composite: multiple losses share one tensor-parallel-reduced softmax. with test_context.subtest(base_path, f"monolithic-{suffix}", 2) as subtest: if subtest.do_run: @@ -1077,10 +1062,8 @@ def test_run_lm_loss_distributed(run_parallel_script, result_path): "z_loss", "grpo", "gspo", - "grpo_metrics-False", - "grpo_metrics-True", - "gspo_metrics-False", - "gspo_metrics-True", + "grpo_metrics", + "gspo_metrics", "monolithic", ), ) diff --git a/tests/utils/model_configs.py b/tests/utils/model_configs.py index 4f2da8b54..63b02e66a 100644 --- a/tests/utils/model_configs.py +++ b/tests/utils/model_configs.py @@ -708,7 +708,9 @@ def update_and_add_testing_config( # Tests mixture of experts, mixtral converter. "llama", "llama_grpo", - updates={("model", "base_model", "head", "losses"): {"grpo": {"type": "grpo"}}}, + # Metrics default to `auto` (→ `basic` when pipeline_parallel == 1); pin `none` so this loss-mechanics + # config doesn't register the metric family (metrics are covered by the subtests in `test_lm_losses`). + updates={("model", "base_model", "head", "losses"): {"grpo": {"type": "grpo", "metrics": "none"}}}, groups={ ModelTestingGroup.basic: ModelTestingGroupAction.normal, ModelTestingGroup.checkpoint: ModelTestingGroupAction.not_implemented, @@ -724,7 +726,9 @@ def update_and_add_testing_config( update_and_add_testing_config( "llama", "llama_gspo", - updates={("model", "base_model", "head", "losses"): {"gspo": {"type": "gspo"}}}, + # Metrics default to `auto`; pin `none` for the same reason as `llama_grpo` (keep this config focused + # on loss mechanics; metrics are covered by `test_lm_losses`). + updates={("model", "base_model", "head", "losses"): {"gspo": {"type": "gspo", "metrics": "none"}}}, # `ms*` (micro_batch_splits>1) and `ce*` (cross_entropy_splits>1) both produce # multiple kernel calls per micro-batch; GSPO's per-document geometric mean can't be # reconstructed from per-fragment `exp(mean)` values, so we skip these variants.