Merge branch 'verl-project:main' into main

Author: khazic

docs/ascend_tutorial/ascend_sglang_quick_start.rst

@@ -76,7 +76,8 @@ Atlas 800T A3
     git clone https://github.com/volcengine/verl.git  
     # Make sure you have activated verl conda env
     # NPU_DEVICE=A3 or A2 depends on your device
-    NPU_DEVICE=A3 bash verl/scripts/install_sglang_mcore_npu.sh
+    # USE_MEGATRON=1 if you need to install megatron backend
+    NPU_DEVICE=A3 USE_MEGATRON=1 bash verl/scripts/install_sglang_mcore_npu.sh
 
 **4. 安装verl**
 

docs/ascend_tutorial/examples/ascend_sglang_best_practices.rst

@@ -43,18 +43,11 @@ SGLang 是当前主流的高性能开源推理引擎, 昇腾已经全面原生
 ^^^^^^^^^^^
 **下载模型权重**
 
---local-dir: 模型保存路径
-
-.. code-block:: bash
-
-  export HF_ENDPOINT=https://hf-mirror.com
-  hf download --resume-download Qwen/Qwen3-30B-A3B --local-dir /path/to/local_dir
+Qwen3-30B: https://huggingface.co/Qwen/Qwen3-30B-A3B
 
 **下载数据集**
 
-.. code-block:: bash
-
-  git clone https://www.modelscope.cn/datasets/AI-ModelScope/DAPO-Math-17k.git
+DAPO-Math-17k: https://huggingface.co/datasets/BytedTsinghua-SIA/DAPO-Math-17k
 
 **HuggingFace To Megatron权重转换(可选)**
 

examples/dppo_trainer/run_qwen30b_dppo.sh

@@ -58,7 +58,7 @@ bypass_mode=True
 # We recommand using Dr.GRPO to remove the length and difficulty bias in original GRPO.
 # See Section 3.1 in https://arxiv.org/pdf/2503.20783 for more details.
 norm_adv_by_std_in_grpo=False               # remove the difficulty bias
-loss_agg_mode="seq-mean-token-sum"          # remove the length bias
+loss_agg_mode="seq-mean-token-sum-norm"     # remove the length bias
 
 # reference policy
 use_kl_in_reward=False

scripts/install_sglang_mcore_npu.sh

@@ -1,6 +1,7 @@
 #!/bin/bash
 set -e
 NPU_DEVICE=${NPU_DEVICE:=A3}
+USE_MEGATRON=${USE_MEGATRON:-1}
 
 export MAX_JOBS=32
 

verl/models/mcore/util.py

@@ -13,14 +13,19 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import logging
 import math
+import os
 
 import torch
 from megatron.core import parallel_state as mpu
 from megatron.core.packed_seq_params import PackedSeqParams
 
 from verl.utils.model import CausalLMOutputForPPO
 
+logger = logging.getLogger(__file__)
+logger.setLevel(os.getenv("VERL_LOGGING_LEVEL", "WARN"))
+
 
 def preprocess_packed_seqs(
     input_ids: torch.Tensor, attention_mask: torch.Tensor, pre_process: bool = True, use_fp8_padding=False
@@ -333,6 +338,19 @@ def preprocess_thd_no_padding(
             start_idx = cu_seqlens_padded_cpu[i] // cp_size
             # split to 2 chunks
             d = input_ids[i]
+            # If the number of elements in `d` is smaller than the required
+            # alignment size, pad the tensor with zeros so that its total
+            # length matches `align_size`. This ensures size alignment for
+            # downstream operations (e.g., communication or memory alignment).
+            if d.numel() < align_size:
+                original_size = d.numel()
+                pad = torch.zeros(align_size - d.numel(), dtype=d.dtype, device=d.device)
+                d = torch.cat([d, pad], dim=0)
+                logger.warning_once(
+                    f"Padding tensor for context parallel alignment, original_size={original_size}, "
+                    f"align_size={align_size}"
+                )
+
             input_ids_rmpad[start_idx : start_idx + half_seqlen] = d[
                 half_seqlen * cp_rank : half_seqlen * (cp_rank + 1)
             ]

verl/trainer/ppo/core_algos.py

@@ -1058,14 +1058,19 @@ def agg_loss(
                 raise ValueError("(global) batch_num_tokens is required when dp_size > 1")
             batch_num_tokens = loss_mask.sum()
         loss = verl_F.masked_sum(loss_mat, loss_mask) / batch_num_tokens * dp_size
-    elif loss_agg_mode == "seq-mean-token-sum":
+    elif loss_agg_mode in ["seq-mean-token-sum", "seq-mean-token-sum-norm"]:
         seq_losses = torch.sum(loss_mat * loss_mask, dim=-1)  # token-sum
         seq_mask = (torch.sum(loss_mask, dim=-1) > 0).float()  # exclude fully masked sequences
         if global_batch_size is None:
             if dp_size > 1:
                 raise ValueError("global_batch_size is required when dp_size > 1")
             global_batch_size = seq_mask.sum()
         loss = verl_F.masked_sum(seq_losses, seq_mask) / global_batch_size * dp_size  # seq-mean
+        if loss_agg_mode == "seq-mean-token-sum-norm":
+            if loss_scale_factor is None:
+                horizon = loss_mask.shape[-1]
+                loss_scale_factor = horizon
+            loss /= loss_scale_factor
     elif loss_agg_mode == "seq-mean-token-mean":
         seq_mask = torch.sum(loss_mask, dim=-1)  # per-sequence token count
         seq_losses = torch.sum(loss_mat * loss_mask, dim=-1) / (seq_mask + 1e-8)  # token-mean
@@ -1075,14 +1080,6 @@ def agg_loss(
                 raise ValueError("global_batch_size is required when dp_size > 1")
             global_batch_size = seq_mask.sum()
         loss = verl_F.masked_sum(seq_losses, seq_mask) / global_batch_size * dp_size  # seq-mean
-    elif loss_agg_mode == "seq-mean-token-sum-norm":
-        if loss_scale_factor is None:
-            raise ValueError(
-                f"{loss_agg_mode=} but {loss_scale_factor=}. "
-                'If not intented for custom scaling factor, try setting loss_agg_mode="seq-mean-token-sum".'
-            )
-        seq_losses = torch.sum(loss_mat * loss_mask, dim=-1)
-        loss = torch.sum(seq_losses) / loss_scale_factor * dp_size
     else:
         raise ValueError(f"Invalid loss_agg_mode: {loss_agg_mode}")
 

verl/utils/megatron_utils.py

@@ -442,6 +442,11 @@ def offload_megatron_model_to_cpu(models):
                         # if the grad_data size is already zero, we assume that it is already offloaded
                         buffer.grad_data_size = buffer.grad_data.storage().size()
                         buffer.grad_data.storage().resize_(0)
+            # Offload frozen parameters not in DDP buffers (e.g. base model in LoRA/PEFT)
+            # DDP buffers only contain requires_grad=True params, so frozen params must be offloaded separately.
+            for param in model_chunk.module.parameters():
+                if not param.requires_grad and param.device.type != "cpu":
+                    param.data = param.data.to("cpu", non_blocking=True)
         else:
             # we need this for ref module
             for _, param in model_chunk.named_parameters():
@@ -453,7 +458,14 @@ def offload_megatron_model_to_cpu(models):
 
 
 @torch.no_grad()
-def load_megatron_model_to_gpu(models, load_grad=True):
+def load_megatron_model_to_gpu(models, load_grad=True, load_frozen_params=True):
+    """
+    Load megatron model to GPU.
+    Args:
+        models: The model to load.
+        load_grad: Whether to load gradients.
+        load_frozen_params: Whether to load frozen parameters.
+    """
     for model_chunk in models:
         if isinstance(model_chunk, DDP):
             model_chunk_all_buffers = [model_chunk.buffers, model_chunk.expert_parallel_buffers]
@@ -468,6 +480,13 @@ def load_megatron_model_to_gpu(models, load_grad=True):
                         buffer.param_data.storage().resize_(buffer.param_data_size)
                         # copy data from cpu to cuda
                         buffer.param_data.copy_(buffer.param_data.cpu_data, non_blocking=True)
+
+            # Load frozen parameters that were offloaded (e.g. base model in LoRA/PEFT)
+            if load_frozen_params:
+                device_id = get_device_id()
+                for param in model_chunk.module.parameters():
+                    if not param.requires_grad and param.device.type == "cpu":
+                        param.data = param.data.to(device_id, non_blocking=True)
         else:
             # we need this for ref module
             device_id = get_device_id()

verl/utils/profiler/torch_profile.py

@@ -14,6 +14,7 @@
 
 import functools
 import os
+from datetime import datetime, timezone
 from typing import Callable, Optional
 
 import torch
@@ -34,7 +35,11 @@ def get_torch_profiler(
 
     os.makedirs(save_path, exist_ok=True)
 
-    save_file_name = f"prof_rank-{rank}.json.gz"
+    current_time = datetime.now(tz=timezone.utc).astimezone()
+    timestamp = current_time.strftime("%Y%m%d%H%M%S%f")[:-3]
+    pid = os.getpid()
+
+    save_file_name = f"prof_rank-{rank}_{pid}_{timestamp}.json.gz"
     if save_file_prefix:
         save_file_name = f"{save_file_prefix}_{save_file_name}"
     save_path = os.path.join(save_path, save_file_name)

verl/workers/engine/megatron/transformer_impl.py

@@ -319,6 +319,8 @@ def initialize(self):
         if self.engine_config.forward_only:
             self.optimizer = None
             self.lr_scheduler = None
+            self.to(device="cpu", model=self._is_offload_param, optimizer=False, grad=False)
+            log_gpu_memory_usage("After offload model during init (forward_only)", logger=logger)
             return
 
         self.optimizer = self._build_optimizer()
@@ -602,12 +604,14 @@ def forward_backward_batch(self, data: TensorDict, loss_function: Callable, forw
             return {}
 
     def get_per_tensor_param(self, base_sync_done=False, **kwargs):
-        load_megatron_model_to_gpu(self.module, load_grad=False)
         peft_config = None
         non_merge_lora_sync = self.peft_cls is not None and not self.model_config.lora.get("merge", False)
+        adapter_only = base_sync_done and non_merge_lora_sync
+        # when lora adapter only, we only load adapter weights when base sync is done, otherwise load all weights
+        load_megatron_model_to_gpu(self.module, load_grad=False, load_frozen_params=not adapter_only)
         if self.vanilla_bridge:
             per_tensor_param = self.bridge.export_weights(self.module)
-        elif base_sync_done and non_merge_lora_sync:
+        elif adapter_only:
             # Only export adapter weights
             peft_config = build_peft_config_for_vllm(self.model_config.lora)
             per_tensor_param = self.bridge.export_adapter_weights(self.module)

verl/workers/rollout/vllm_rollout/vllm_async_server.py

@@ -180,6 +180,12 @@ def get_server_address(self):
         assert self._server_port is not None, "http server is not launched, port is None"
         return self._server_address, self._server_port
 
+    @property
+    def lora_as_adapter(self) -> bool:
+        return (
+            self.model_config.lora_rank > 0 or self.model_config.lora.get("rank", 0) > 0
+        ) and not self.model_config.lora.get("merge", False)
+
     async def collective_rpc(
         self,
         method: str | Callable,
@@ -543,9 +549,7 @@ async def generate(
 
         # Add lora request
         lora_request = None
-        if (
-            self.model_config.lora_rank > 0 or self.model_config.lora.get("rank", 0) > 0
-        ) and not self.model_config.lora.get("merge", False):
+        if self.lora_as_adapter:
             # Make sure we also check that the lora is already loaded in the engine
             lora_loaded = VLLM_LORA_INT_ID in await self.engine.list_loras()
             if lora_loaded:
@@ -618,7 +622,12 @@ async def sleep(self):
 
         if self.rollout_mode == RolloutMode.HYBRID:
             # Don't use engine.sleep(level=2) here
-            await self.engine.collective_rpc("sleep", kwargs={"level": 2})
+            # lora only update adapter weights, so set sleep level to 1
+            if self.lora_as_adapter:
+                sleep_level = 1
+            else:
+                sleep_level = 2
+            await self.engine.collective_rpc("sleep", kwargs={"level": sleep_level})
 
             # clear encoder cache: https://github.com/vllm-project/vllm/pull/33452
             # await self.engine.reset_encoder_cache()