From fbead29897ed2d7bd2b59fd3b54d896991b2ac01 Mon Sep 17 00:00:00 2001 From: YeonwooSung Date: Wed, 19 Aug 2026 11:04:03 +0900 Subject: [PATCH 1/2] [trainer, data] fix: do not drop_last SFT validation batches Fixes #7464 Co-authored-by: Grok Signed-off-by: YeonwooSung --- tests/trainer/test_sft_val_batch_on_cpu.py | 35 ++++++++++++++++++++ verl/trainer/config/sft_trainer_engine.yaml | 6 +++- verl/trainer/sft_trainer.py | 36 ++++++++++++++------- verl/trainer/sft_trainer_ray.py | 28 +++++++++++----- verl/trainer/sft_val_utils.py | 28 ++++++++++++++++ 5 files changed, 113 insertions(+), 20 deletions(-) create mode 100644 tests/trainer/test_sft_val_batch_on_cpu.py create mode 100644 verl/trainer/sft_val_utils.py diff --git a/tests/trainer/test_sft_val_batch_on_cpu.py b/tests/trainer/test_sft_val_batch_on_cpu.py new file mode 100644 index 00000000000..c201c349fe8 --- /dev/null +++ b/tests/trainer/test_sft_val_batch_on_cpu.py @@ -0,0 +1,35 @@ +# Copyright 2026 Bytedance Ltd. and/or its affiliates +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from verl.trainer.sft_val_utils import resolve_sft_val_batch_size + + +def test_resolve_prefers_explicit_val_batch_size(): + assert resolve_sft_val_batch_size({"val_batch_size": 16, "micro_batch_size_per_gpu": 4}, 256, 200) == 16 + + +def test_resolve_falls_back_to_micro_batch_not_train_batch(): + assert resolve_sft_val_batch_size({"micro_batch_size_per_gpu": 4}, 256, 200) == 4 + + +def test_resolve_uses_val_len_when_no_micro_batch(): + assert resolve_sft_val_batch_size({}, 256, 200) == 200 + + +def test_small_val_set_is_not_empty_with_resolved_batch(): + train_bs = 256 + n = 200 + assert n // train_bs == 0 # old train-batch + drop_last path + batch = resolve_sft_val_batch_size({"micro_batch_size_per_gpu": 4}, train_bs, n) + assert (n + batch - 1) // batch > 0 diff --git a/verl/trainer/config/sft_trainer_engine.yaml b/verl/trainer/config/sft_trainer_engine.yaml index 8fe10e63ee1..146641113b9 100644 --- a/verl/trainer/config/sft_trainer_engine.yaml +++ b/verl/trainer/config/sft_trainer_engine.yaml @@ -15,7 +15,11 @@ defaults: data: train_batch_size: 256 # global batch size - micro_batch_size_per_gpu: 4 # this is also val batch size + micro_batch_size_per_gpu: 4 + + # Validation batch size per DP rank. Null uses micro_batch_size_per_gpu. + val_batch_size: null + max_token_len_per_gpu: 8192 use_dynamic_bsz: True train_files: ~/data/gsm8k/train.parquet diff --git a/verl/trainer/sft_trainer.py b/verl/trainer/sft_trainer.py index 0c4aa8e08d8..e5d12f79b2e 100644 --- a/verl/trainer/sft_trainer.py +++ b/verl/trainer/sft_trainer.py @@ -31,6 +31,7 @@ from torchdata.stateful_dataloader import StatefulDataLoader from tqdm import tqdm +from verl.trainer.sft_val_utils import resolve_sft_val_batch_size from verl.utils import tensordict_utils as tu from verl.utils.checkpoint import CheckpointHandler from verl.utils.dataset.dataset_utils import SFTTensorCollator @@ -253,17 +254,20 @@ def _build_dataloader(self): ) if self.val_dataset: + val_batch_size = resolve_sft_val_batch_size( + config.data, self.train_batch_size_per_dp, len(self.val_dataset) + ) self.val_sampler = DistributedSampler( - self.val_dataset, shuffle=False, num_replicas=dp_size, rank=dp_rank, drop_last=True + self.val_dataset, shuffle=False, num_replicas=dp_size, rank=dp_rank, drop_last=False ) self.val_dataloader = StatefulDataLoader( dataset=self.val_dataset, - batch_size=self.train_batch_size_per_dp, + batch_size=val_batch_size, sampler=self.val_sampler, collate_fn=self.collate_fn, num_workers=self.config.data.num_workers, pin_memory=False, - drop_last=True, + drop_last=False, pin_memory_device=device_name, ) else: @@ -425,16 +429,26 @@ def fit(self): val_losses.append(metrics["loss"]) if self.engine.is_mp_src_rank_with_outputs(): - val_loss = torch.mean(torch.tensor(val_losses, device=self.device_name)) - # average over data parallel group + n_val = torch.tensor(float(len(val_losses)), device=self.device_name) + sum_val = torch.tensor( + float(sum(val_losses)) if val_losses else 0.0, device=self.device_name + ) dp_group = self.engine.get_data_parallel_group() if dp_group is not None: - torch.distributed.all_reduce(val_loss, op=torch.distributed.ReduceOp.AVG, group=dp_group) - - if is_logging: - metric = {"val/loss": val_loss.detach().item()} - tracking.log(data=metric, step=global_step) - last_valid_metric = metric + torch.distributed.all_reduce(n_val, op=torch.distributed.ReduceOp.SUM, group=dp_group) + torch.distributed.all_reduce(sum_val, op=torch.distributed.ReduceOp.SUM, group=dp_group) + if n_val.item() <= 0: + log_with_rank( + "Validation produced no batches; skip val/loss rather than logging NaN.", + logger=logger, + rank=self.rank, + level=logging.WARNING, + log_only_rank_0=True, + ) + elif is_logging: + metric = {"val/loss": (sum_val / n_val).detach().item()} + tracking.log(data=metric, step=global_step) + last_valid_metric = metric torch.distributed.barrier() if is_last_step or (self.save_freq > 0 and is_save_step): diff --git a/verl/trainer/sft_trainer_ray.py b/verl/trainer/sft_trainer_ray.py index 5f2d68d0e54..e9b2653013b 100644 --- a/verl/trainer/sft_trainer_ray.py +++ b/verl/trainer/sft_trainer_ray.py @@ -32,6 +32,7 @@ from torchdata.stateful_dataloader import StatefulDataLoader from tqdm import tqdm +from verl.trainer.sft_val_utils import resolve_sft_val_batch_size from verl.utils import tensordict_utils as tu from verl.utils.checkpoint import CheckpointHandler, OrchestrationMode from verl.utils.dataset.dataset_utils import SFTTensorCollator @@ -201,17 +202,20 @@ def _build_dataloader(self): ) if self.val_dataset: + val_batch_size = resolve_sft_val_batch_size( + config.data, self.train_batch_size_per_dp, len(self.val_dataset) + ) self.val_sampler = DistributedSampler( - self.val_dataset, shuffle=False, num_replicas=dp_size, rank=dp_rank, drop_last=True + self.val_dataset, shuffle=False, num_replicas=dp_size, rank=dp_rank, drop_last=False ) self.val_dataloader = StatefulDataLoader( dataset=self.val_dataset, - batch_size=self.train_batch_size_per_dp, + batch_size=val_batch_size, sampler=self.val_sampler, collate_fn=self.collate_fn, num_workers=8, pin_memory=False, - drop_last=True, + drop_last=False, pin_memory_device=device_name, ) else: @@ -364,11 +368,19 @@ def fit(self): metrics = tu.get(output, "metrics") val_losses.append(metrics["loss"]) - val_loss = torch.mean(torch.tensor(val_losses, device=self.device_name)) - - metric = {"val/loss": val_loss.detach().item()} - tracking.log(data=metric, step=global_step) - last_valid_metric = metric + if not val_losses: + log_with_rank( + "Validation produced no batches; skip val/loss rather than logging NaN.", + logger=logger, + rank=0, + level=logging.WARNING, + log_only_rank_0=True, + ) + else: + val_loss = torch.mean(torch.tensor(val_losses, device=self.device_name)) + metric = {"val/loss": val_loss.detach().item()} + tracking.log(data=metric, step=global_step) + last_valid_metric = metric if is_last_step or (self.save_freq > 0 and is_save_step): self.ckpt_handler.save_checkpoint(step=global_step) diff --git a/verl/trainer/sft_val_utils.py b/verl/trainer/sft_val_utils.py new file mode 100644 index 00000000000..22ca288d397 --- /dev/null +++ b/verl/trainer/sft_val_utils.py @@ -0,0 +1,28 @@ +# Copyright 2026 Bytedance Ltd. and/or its affiliates +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + + +def resolve_sft_val_batch_size(data_config, train_batch_size_per_dp: int, val_dataset_len: int) -> int: + """Pick a validation batch size that does not swallow small val sets. + + Preference: ``data.val_batch_size`` > ``data.micro_batch_size_per_gpu`` > + the full val set. Falls back to the train per-DP batch only if nothing else + is available. + """ + val_batch_size = data_config.get("val_batch_size", None) + if val_batch_size is None: + val_batch_size = data_config.get("micro_batch_size_per_gpu", None) + if val_batch_size is None: + val_batch_size = val_dataset_len if val_dataset_len > 0 else train_batch_size_per_dp + return max(1, int(val_batch_size)) From 596bff0493768373032fe6d90f1d377783a075af Mon Sep 17 00:00:00 2001 From: YeonwooSung Date: Wed, 26 Aug 2026 14:28:22 +0900 Subject: [PATCH 2/2] [trainer, data] fix: default SFT val batch to the full val set Address review on #7467: match PPO (val_batch_size or len(val_dataset)), weight val/loss by sample count, assert the val loader is non-empty at build time, and add a dataloader regression test for drop_last=False. Fixes #7464 Co-authored-by: Grok Signed-off-by: YeonwooSung --- tests/trainer/test_sft_val_batch_on_cpu.py | 52 ++++++++++++++++----- verl/trainer/config/sft_trainer_engine.yaml | 2 +- verl/trainer/sft_trainer.py | 19 ++++---- verl/trainer/sft_trainer_ray.py | 18 +++---- verl/trainer/sft_val_utils.py | 33 +++++++++---- 5 files changed, 86 insertions(+), 38 deletions(-) diff --git a/tests/trainer/test_sft_val_batch_on_cpu.py b/tests/trainer/test_sft_val_batch_on_cpu.py index c201c349fe8..f801c65bc68 100644 --- a/tests/trainer/test_sft_val_batch_on_cpu.py +++ b/tests/trainer/test_sft_val_batch_on_cpu.py @@ -12,24 +12,52 @@ # See the License for the specific language governing permissions and # limitations under the License. -from verl.trainer.sft_val_utils import resolve_sft_val_batch_size +from torch.utils.data import DataLoader, Dataset, DistributedSampler +from torchdata.stateful_dataloader import StatefulDataLoader + +from verl.trainer.sft_val_utils import reduce_sft_val_loss, resolve_sft_val_batch_size + + +class _Toy(Dataset): + def __len__(self): + return 200 + + def __getitem__(self, i): + return i def test_resolve_prefers_explicit_val_batch_size(): - assert resolve_sft_val_batch_size({"val_batch_size": 16, "micro_batch_size_per_gpu": 4}, 256, 200) == 16 + assert resolve_sft_val_batch_size({"val_batch_size": 16}, 200) == 16 + + +def test_resolve_defaults_to_full_val_set(): + assert resolve_sft_val_batch_size({}, 200) == 200 + assert resolve_sft_val_batch_size({"micro_batch_size_per_gpu": 4}, 200) == 200 + + +def test_reduce_sft_val_loss_is_sample_weighted(): + # Batches of 4, 4, 4, 1 must not treat the tail as 25% of the mean. + assert reduce_sft_val_loss([(1.0, 4), (1.0, 4), (1.0, 4), (13.0, 1)]) == (3 * 4 * 1.0 + 13.0) / 13 + assert reduce_sft_val_loss([]) is None -def test_resolve_falls_back_to_micro_batch_not_train_batch(): - assert resolve_sft_val_batch_size({"micro_batch_size_per_gpu": 4}, 256, 200) == 4 +def _make_val_loader(*, drop_last: bool) -> DataLoader: + dataset = _Toy() + return StatefulDataLoader( + dataset, + batch_size=256, + sampler=DistributedSampler(dataset, num_replicas=1, rank=0, shuffle=False, drop_last=drop_last), + drop_last=drop_last, + ) -def test_resolve_uses_val_len_when_no_micro_batch(): - assert resolve_sft_val_batch_size({}, 256, 200) == 200 +def test_drop_last_true_with_train_batch_is_empty(): + assert len(_make_val_loader(drop_last=True)) == 0 -def test_small_val_set_is_not_empty_with_resolved_batch(): - train_bs = 256 - n = 200 - assert n // train_bs == 0 # old train-batch + drop_last path - batch = resolve_sft_val_batch_size({"micro_batch_size_per_gpu": 4}, train_bs, n) - assert (n + batch - 1) // batch > 0 +def test_drop_last_false_keeps_short_val_set(): + """Regression test for #7464: 200 samples, train-sized batch 256 must still yield a batch.""" + loader = _make_val_loader(drop_last=False) + assert len(loader) >= 1 + seen = sum(len(batch) for batch in loader) + assert seen == 200 diff --git a/verl/trainer/config/sft_trainer_engine.yaml b/verl/trainer/config/sft_trainer_engine.yaml index 146641113b9..ffe7ac2070b 100644 --- a/verl/trainer/config/sft_trainer_engine.yaml +++ b/verl/trainer/config/sft_trainer_engine.yaml @@ -17,7 +17,7 @@ data: train_batch_size: 256 # global batch size micro_batch_size_per_gpu: 4 - # Validation batch size per DP rank. Null uses micro_batch_size_per_gpu. + # Validation dataloader batch size. Null sends the full val set in one batch (same as PPO). val_batch_size: null max_token_len_per_gpu: 8192 diff --git a/verl/trainer/sft_trainer.py b/verl/trainer/sft_trainer.py index e5d12f79b2e..55c43d8cd0d 100644 --- a/verl/trainer/sft_trainer.py +++ b/verl/trainer/sft_trainer.py @@ -31,7 +31,7 @@ from torchdata.stateful_dataloader import StatefulDataLoader from tqdm import tqdm -from verl.trainer.sft_val_utils import resolve_sft_val_batch_size +from verl.trainer.sft_val_utils import resolve_sft_val_batch_size, sft_val_num_samples from verl.utils import tensordict_utils as tu from verl.utils.checkpoint import CheckpointHandler from verl.utils.dataset.dataset_utils import SFTTensorCollator @@ -254,9 +254,7 @@ def _build_dataloader(self): ) if self.val_dataset: - val_batch_size = resolve_sft_val_batch_size( - config.data, self.train_batch_size_per_dp, len(self.val_dataset) - ) + val_batch_size = resolve_sft_val_batch_size(config.data, len(self.val_dataset)) self.val_sampler = DistributedSampler( self.val_dataset, shuffle=False, num_replicas=dp_size, rank=dp_rank, drop_last=False ) @@ -270,6 +268,7 @@ def _build_dataloader(self): drop_last=False, pin_memory_device=device_name, ) + assert len(self.val_dataloader) >= 1, "Validation dataloader is empty!" else: self.val_dataloader = None @@ -419,19 +418,23 @@ def fit(self): # early exit or validation step if is_last_step and self.val_dataloader is not None or (self.test_freq > 0 and is_valid_step): # Perform validation - val_losses = [] + val_losses_and_counts = [] for val_data in self.val_dataloader: val_data = tu.get_tensordict(tensor_dict=val_data, non_tensor_dict=meta_info) + n_samples = sft_val_num_samples(val_data) output = self.training_client.infer_batch(val_data) if self.engine.is_mp_src_rank_with_outputs(): metrics = tu.get(output, "metrics") - val_losses.append(metrics["loss"]) + val_losses_and_counts.append((metrics["loss"], n_samples)) if self.engine.is_mp_src_rank_with_outputs(): - n_val = torch.tensor(float(len(val_losses)), device=self.device_name) + n_val = torch.tensor( + float(sum(n for _, n in val_losses_and_counts)), device=self.device_name + ) sum_val = torch.tensor( - float(sum(val_losses)) if val_losses else 0.0, device=self.device_name + float(sum(float(loss) * n for loss, n in val_losses_and_counts)), + device=self.device_name, ) dp_group = self.engine.get_data_parallel_group() if dp_group is not None: diff --git a/verl/trainer/sft_trainer_ray.py b/verl/trainer/sft_trainer_ray.py index e9b2653013b..2db9ecb92cb 100644 --- a/verl/trainer/sft_trainer_ray.py +++ b/verl/trainer/sft_trainer_ray.py @@ -32,7 +32,7 @@ from torchdata.stateful_dataloader import StatefulDataLoader from tqdm import tqdm -from verl.trainer.sft_val_utils import resolve_sft_val_batch_size +from verl.trainer.sft_val_utils import reduce_sft_val_loss, resolve_sft_val_batch_size, sft_val_num_samples from verl.utils import tensordict_utils as tu from verl.utils.checkpoint import CheckpointHandler, OrchestrationMode from verl.utils.dataset.dataset_utils import SFTTensorCollator @@ -202,9 +202,7 @@ def _build_dataloader(self): ) if self.val_dataset: - val_batch_size = resolve_sft_val_batch_size( - config.data, self.train_batch_size_per_dp, len(self.val_dataset) - ) + val_batch_size = resolve_sft_val_batch_size(config.data, len(self.val_dataset)) self.val_sampler = DistributedSampler( self.val_dataset, shuffle=False, num_replicas=dp_size, rank=dp_rank, drop_last=False ) @@ -218,6 +216,7 @@ def _build_dataloader(self): drop_last=False, pin_memory_device=device_name, ) + assert len(self.val_dataloader) >= 1, "Validation dataloader is empty!" else: self.val_dataloader = None @@ -360,15 +359,17 @@ def fit(self): # early exit or validation step if is_last_step and self.val_dataloader is not None or (self.test_freq > 0 and is_valid_step): # Perform validation - val_losses = [] + val_losses_and_counts = [] for val_data in self.val_dataloader: val_data = tu.get_tensordict(tensor_dict=val_data, non_tensor_dict=meta_info) + n_samples = sft_val_num_samples(val_data) output = self.training_client.infer_batch(val_data) output = output.get() metrics = tu.get(output, "metrics") - val_losses.append(metrics["loss"]) + val_losses_and_counts.append((metrics["loss"], n_samples)) - if not val_losses: + val_loss = reduce_sft_val_loss(val_losses_and_counts) + if val_loss is None: log_with_rank( "Validation produced no batches; skip val/loss rather than logging NaN.", logger=logger, @@ -377,8 +378,7 @@ def fit(self): log_only_rank_0=True, ) else: - val_loss = torch.mean(torch.tensor(val_losses, device=self.device_name)) - metric = {"val/loss": val_loss.detach().item()} + metric = {"val/loss": val_loss} tracking.log(data=metric, step=global_step) last_valid_metric = metric diff --git a/verl/trainer/sft_val_utils.py b/verl/trainer/sft_val_utils.py index 22ca288d397..1d90b7e0b54 100644 --- a/verl/trainer/sft_val_utils.py +++ b/verl/trainer/sft_val_utils.py @@ -12,17 +12,34 @@ # See the License for the specific language governing permissions and # limitations under the License. +from typing import Optional -def resolve_sft_val_batch_size(data_config, train_batch_size_per_dp: int, val_dataset_len: int) -> int: - """Pick a validation batch size that does not swallow small val sets. - Preference: ``data.val_batch_size`` > ``data.micro_batch_size_per_gpu`` > - the full val set. Falls back to the train per-DP batch only if nothing else - is available. +def resolve_sft_val_batch_size(data_config, val_dataset_len: int) -> int: + """Pick the SFT validation dataloader batch size. + + Matches PPO: ``data.val_batch_size`` if set, otherwise the full val set. + ``micro_batch_size_per_gpu`` is an engine split size, not a dataloader knob. """ val_batch_size = data_config.get("val_batch_size", None) if val_batch_size is None: - val_batch_size = data_config.get("micro_batch_size_per_gpu", None) - if val_batch_size is None: - val_batch_size = val_dataset_len if val_dataset_len > 0 else train_batch_size_per_dp + val_batch_size = val_dataset_len return max(1, int(val_batch_size)) + + +def sft_val_num_samples(batch) -> int: + """Number of sequences in a collated SFT val batch.""" + batch_size = getattr(batch, "batch_size", None) + if batch_size: + return int(batch_size[0]) + if hasattr(batch, "__contains__") and "input_ids" in batch: + return int(batch["input_ids"].shape[0]) + return 1 + + +def reduce_sft_val_loss(losses_and_counts: list[tuple[float, int]]) -> Optional[float]: + """Sample-weighted mean of per-batch val losses. None if there were no samples.""" + total_n = sum(n for _, n in losses_and_counts) + if total_n <= 0: + return None + return sum(float(loss) * n for loss, n in losses_and_counts) / total_n