[misc] fix data proto (#458)

Author: hiyouga

examples/config.yaml

@@ -34,8 +34,8 @@ algorithm:
 worker:
   actor:
     global_batch_size: 128  # equivalent to verl's actor.ppo_mini_batch_size
-    micro_batch_size_per_device_for_update: 4  # equivalent to verl's actor.ppo_micro_batch_size_per_gpu
-    micro_batch_size_per_device_for_experience: 16  # equivalent to verl's rollout.log_prob_micro_batch_size_per_gpu
+    micro_batch_size_per_device_for_update: 1  # equivalent to verl's actor.ppo_micro_batch_size_per_gpu
+    micro_batch_size_per_device_for_experience: 2  # equivalent to verl's rollout.log_prob_micro_batch_size_per_gpu
     max_grad_norm: 1.0
     padding_free: true
     dynamic_batching: true

examples/qwen2_5_vl_32b_geo3k_grpo.sh

@@ -11,8 +11,6 @@ python3 -m verl.trainer.main \
     data.train_files=hiyouga/geometry3k@train \
     data.val_files=hiyouga/geometry3k@test \
     worker.actor.model.model_path=${MODEL_PATH} \
-    worker.actor.micro_batch_size_per_device_for_update=1 \
-    worker.actor.micro_batch_size_per_device_for_experience=8 \
     worker.actor.fsdp.torch_dtype=bf16 \
     worker.actor.optim.strategy=adamw_bf16 \
     worker.rollout.tensor_parallel_size=8 \

verl/protocol.py

@@ -204,12 +204,14 @@ def __getitem__(
         raise TypeError(f"Indexing with {type(item)} is not supported.")
 
     def __getstate__(self) -> tuple[bytes, dict[str, NDArray], dict[str, Any]]:
-        buffer = io.BytesIO()
         if self.batch is not None:
-            self.batch: TensorDict = self.batch.contiguous()
-            self.batch: TensorDict = self.batch.consolidate()
+            batch_to_save: TensorDict = self.batch.contiguous()
+            batch_to_save: TensorDict = batch_to_save.consolidate()
+        else:
+            batch_to_save = None
 
-        torch.save(self.batch, buffer)
+        buffer = io.BytesIO()
+        torch.save(batch_to_save, buffer)
         buffer_bytes = buffer.getvalue()
         return buffer_bytes, self.non_tensor_batch, self.meta_info
 

verl/trainer/ray_trainer.py

@@ -39,7 +39,7 @@
 from ..utils import torch_functional as VF
 from ..utils.checkpoint import CHECKPOINT_TRACKER, find_latest_ckpt, remove_obsolete_ckpt
 from ..utils.logger import Tracker
-from ..utils.py_functional import convert_dict_to_str, timer
+from ..utils.py_functional import convert_dict_to_str, timer, unflatten_dict
 from ..utils.seqlen_balancing import get_seqlen_balanced_partitions, log_seqlen_unbalance
 from ..workers.fsdp_workers import FSDPWorker
 from ..workers.reward import FunctionRewardManager
@@ -694,7 +694,7 @@ def fit(self):
                 val_metrics = self._validate()
                 self.logger.log(data=val_metrics, step=self.global_step)
 
-            print(f"Final validation metrics: {convert_dict_to_str(val_metrics)}")
+            print(f"Final validation metrics:\n{convert_dict_to_str(unflatten_dict(val_metrics))}")
 
         if self.config.trainer.save_freq <= 0 or self.global_step % self.config.trainer.save_freq != 0:
             self._save_checkpoint()

verl/workers/fsdp_workers.py

@@ -456,7 +456,7 @@ def _process_multi_modal_inputs(self, data: DataProto):
             multi_modal_inputs_cache = {}  # avoid repeated processing for n > 1 samples
             for index, multi_modal_data in zip(
                 data.non_tensor_batch["uid"], data.non_tensor_batch["multi_modal_data"]
-            ): # process multi modal data per sample
+            ):  # process multi modal data per sample
                 if index not in multi_modal_inputs_cache:
                     images, videos = [], []
                     if "images" in multi_modal_data: