diff --git a/examples/sglang_multiturn/run_qwen2.5-3b_gsm8k_multiturn.sh b/examples/sglang_multiturn/run_qwen2.5-3b_gsm8k_multiturn.sh index 662723df445..e5ab236a4ad 100644 --- a/examples/sglang_multiturn/run_qwen2.5-3b_gsm8k_multiturn.sh +++ b/examples/sglang_multiturn/run_qwen2.5-3b_gsm8k_multiturn.sh @@ -8,6 +8,12 @@ ulimit -n 65535 PROJECT_DIR="$(pwd)" CONFIG_PATH="$PROJECT_DIR/examples/sglang_multiturn/config" +function now() { + date '+%d-%H-%M' +} + +EXPERIMENT_NAME="qwen2.5-3b_baseline_$(now)" + python3 -m verl.trainer.main_ppo \ --config-path="$CONFIG_PATH" \ --config-name='gsm8k_multiturn_grpo' \ @@ -31,21 +37,24 @@ python3 -m verl.trainer.main_ppo \ actor_rollout_ref.actor.fsdp_config.param_offload=False \ actor_rollout_ref.actor.fsdp_config.optimizer_offload=False \ actor_rollout_ref.rollout.log_prob_micro_batch_size_per_gpu=32 \ - actor_rollout_ref.rollout.tensor_model_parallel_size=2 \ + actor_rollout_ref.rollout.tensor_model_parallel_size=1 \ actor_rollout_ref.rollout.name=sglang \ - actor_rollout_ref.rollout.gpu_memory_utilization=0.5 \ + actor_rollout_ref.rollout.gpu_memory_utilization=0.85 \ + actor_rollout_ref.rollout.multi_stage_wake_up=True \ actor_rollout_ref.rollout.n=16 \ actor_rollout_ref.ref.log_prob_micro_batch_size_per_gpu=32 \ actor_rollout_ref.ref.fsdp_config.param_offload=True \ + actor_rollout_ref.rollout.over_sample_rate=0 \ algorithm.use_kl_in_reward=False \ trainer.critic_warmup=0 \ trainer.logger='["console","wandb"]' \ - trainer.project_name='gsm8k_async_rl' \ - trainer.experiment_name='qwen2.5-3b_function_rm-gsm8k-sgl-multi-w-tool-verify-n16' \ + trainer.project_name='multi-turn-grpo-qwen2.5-3b-sglang' \ + trainer.experiment_name=$EXPERIMENT_NAME \ trainer.n_gpus_per_node=8 \ trainer.nnodes=1 \ trainer.save_freq=-1 \ trainer.test_freq=20 \ + trainer.val_before_train=True \ data.train_files=$HOME/data/gsm8k/train.parquet \ data.val_files=$HOME/data/gsm8k/test.parquet \ actor_rollout_ref.rollout.multi_turn.tool_config_path="$PROJECT_DIR/examples/sglang_multiturn/config/tool_config/gsm8k_tool_config.yaml" \ diff --git a/examples/sglang_multiturn/run_qwen2_3b_dapo_multiturn.sh b/examples/sglang_multiturn/run_qwen3_4b_dapo_multiturn.sh similarity index 74% rename from examples/sglang_multiturn/run_qwen2_3b_dapo_multiturn.sh rename to examples/sglang_multiturn/run_qwen3_4b_dapo_multiturn.sh index 9eb02c05e00..53f856cca27 100644 --- a/examples/sglang_multiturn/run_qwen2_3b_dapo_multiturn.sh +++ b/examples/sglang_multiturn/run_qwen3_4b_dapo_multiturn.sh @@ -17,6 +17,10 @@ hf download \ --repo-type dataset \ --local-dir $HOME/data/Maxwell-Jia/AIME_2024 +# Note that this script is using AgentLoop instead of SGLang Multi-Turn +# We are concerned that the reward is not actually converge, since the +# reward of retool is encouraging the model to generate more turns to +# call more tools. The answers are not actually correct. python3 -m verl.trainer.main_ppo \ algorithm.adv_estimator=grpo \ @@ -34,7 +38,7 @@ python3 -m verl.trainer.main_ppo \ data.custom_cls.name=CustomRLHFDataset \ custom_reward_function.path=$PROJECT_DIR/recipe/retool/retool.py \ custom_reward_function.name=compute_score \ - actor_rollout_ref.model.path=Qwen/Qwen2.5-3B-Instruct \ + actor_rollout_ref.model.path=Qwen/Qwen3-4B-Instruct-2507 \ actor_rollout_ref.model.use_remove_padding=True \ actor_rollout_ref.model.enable_gradient_checkpointing=True \ actor_rollout_ref.actor.use_kl_loss=False \ @@ -43,13 +47,16 @@ python3 -m verl.trainer.main_ppo \ actor_rollout_ref.actor.clip_ratio_high=0.28 \ actor_rollout_ref.actor.clip_ratio_c=10.0 \ actor_rollout_ref.actor.optim.lr=1e-6 \ - actor_rollout_ref.actor.use_dynamic_bsz=True \ - actor_rollout_ref.actor.ppo_mini_batch_size=8 \ - actor_rollout_ref.actor.ppo_max_token_len_per_gpu=1024 \ + actor_rollout_ref.actor.ppo_mini_batch_size=32 \ + actor_rollout_ref.actor.ppo_micro_batch_size_per_gpu=8 \ + actor_rollout_ref.actor.ppo_max_token_len_per_gpu=32768 \ + actor_rollout_ref.ref.log_prob_micro_batch_size_per_gpu=8 \ + actor_rollout_ref.rollout.log_prob_micro_batch_size_per_gpu=8 \ actor_rollout_ref.rollout.name=sglang \ actor_rollout_ref.rollout.mode=async \ actor_rollout_ref.rollout.tensor_model_parallel_size=1 \ - actor_rollout_ref.rollout.gpu_memory_utilization=0.85 \ + actor_rollout_ref.rollout.gpu_memory_utilization=0.80 \ + actor_rollout_ref.rollout.update_weights_bucket_megabytes=512 \ actor_rollout_ref.rollout.multi_stage_wake_up=True \ actor_rollout_ref.rollout.multi_turn.enable=True \ actor_rollout_ref.rollout.multi_turn.max_user_turns=16 \ @@ -62,8 +69,8 @@ python3 -m verl.trainer.main_ppo \ actor_rollout_ref.rollout.val_kwargs.n=30 \ trainer.logger=['console','wandb'] \ trainer.project_name=sglang-dapo-multiturn \ - trainer.experiment_name=qwen2_5-3b_dapo_multiturn \ - trainer.n_gpus_per_node=4 \ + trainer.experiment_name=qwen3-4b_dapo_multiturn \ + trainer.n_gpus_per_node=8 \ trainer.log_val_generations=20 \ trainer.val_before_train=True \ trainer.nnodes=1 \ diff --git a/verl/trainer/config/_generated_ppo_megatron_trainer.yaml b/verl/trainer/config/_generated_ppo_megatron_trainer.yaml index 1994038ba60..b030fddac06 100644 --- a/verl/trainer/config/_generated_ppo_megatron_trainer.yaml +++ b/verl/trainer/config/_generated_ppo_megatron_trainer.yaml @@ -163,6 +163,7 @@ actor_rollout_ref: disable_log_stats: true do_sample: true 'n': 1 + over_sample_rate: 0 multi_stage_wake_up: false engine_kwargs: vllm: diff --git a/verl/trainer/config/_generated_ppo_trainer.yaml b/verl/trainer/config/_generated_ppo_trainer.yaml index ced6c49299f..f90422bb5b2 100644 --- a/verl/trainer/config/_generated_ppo_trainer.yaml +++ b/verl/trainer/config/_generated_ppo_trainer.yaml @@ -138,6 +138,7 @@ actor_rollout_ref: disable_log_stats: true do_sample: true 'n': 1 + over_sample_rate: 0 multi_stage_wake_up: false engine_kwargs: vllm: diff --git a/verl/trainer/config/rollout/rollout.yaml b/verl/trainer/config/rollout/rollout.yaml index be903a79f27..acb88728b0d 100644 --- a/verl/trainer/config/rollout/rollout.yaml +++ b/verl/trainer/config/rollout/rollout.yaml @@ -82,7 +82,13 @@ do_sample: True # number of responses (i.e. num sample times). > 1 for grpo n: 1 -# Whether to wake up inference engine in multi-stage to reduce peak memory during training-rollout transition. +# The over_sample_rate parameter controls the early termination threshold for training rollouts, +# where the system will abort remaining requests when (1 - over_sample_rate) * total_requests completions are reached. +over_sample_rate: 0 + +# Whether to wake up inference engine in multi-stage for SGLang +# to reduce peak memory during training-rollout transition. +# This is only effective for SGLang rollout. multi_stage_wake_up: false # Extra inference engine arguments (vllm, sglang). diff --git a/verl/trainer/ppo/metric_utils.py b/verl/trainer/ppo/metric_utils.py index 72d07b80390..3be65ba978c 100644 --- a/verl/trainer/ppo/metric_utils.py +++ b/verl/trainer/ppo/metric_utils.py @@ -118,6 +118,20 @@ def compute_data_metrics(batch: DataProto, use_critic: bool = True) -> dict[str, prompt_length = response_info["prompt_length"] response_length = response_info["response_length"] + aborted_mask = (response_length == 0).bool() + non_aborted_mask = ~aborted_mask + + non_aborted_sequence_score = sequence_score[non_aborted_mask] + non_aborted_sequence_reward = sequence_reward[non_aborted_mask] + + score_mean = torch.mean(non_aborted_sequence_score).detach().item() + score_max = torch.max(non_aborted_sequence_score).detach().item() + score_min = torch.min(non_aborted_sequence_score).detach().item() + + reward_mean = torch.mean(non_aborted_sequence_reward).detach().item() + reward_max = torch.max(non_aborted_sequence_reward).detach().item() + reward_min = torch.min(non_aborted_sequence_reward).detach().item() + valid_adv = torch.masked_select(advantages, response_mask) valid_returns = torch.masked_select(returns, response_mask) @@ -127,15 +141,30 @@ def compute_data_metrics(batch: DataProto, use_critic: bool = True) -> dict[str, return_diff_var = torch.var(valid_returns - valid_values) return_var = torch.var(valid_returns) + # Aborted samples and non-aborted response length statistics + # response_length_non_aborted/*: statistics computed on non-aborted samples only + aborted_ratio = torch.mean(aborted_mask.float()).detach().item() + + non_aborted_response_length = response_length[non_aborted_mask] + if non_aborted_response_length.numel() > 0: + non_aborted_response_length_mean = torch.mean(non_aborted_response_length).detach().item() + non_aborted_response_length_max = torch.max(non_aborted_response_length).detach().item() + non_aborted_response_length_min = torch.min(non_aborted_response_length).detach().item() + non_aborted_response_length_clip_ratio = ( + torch.mean(torch.eq(non_aborted_response_length, max_response_length).float()).detach().item() + ) + else: + raise ValueError("All samples are aborted, this should not happen.") + metrics = { # score - "critic/score/mean": torch.mean(sequence_score).detach().item(), - "critic/score/max": torch.max(sequence_score).detach().item(), - "critic/score/min": torch.min(sequence_score).detach().item(), + "critic/score/mean": score_mean, + "critic/score/max": score_max, + "critic/score/min": score_min, # reward - "critic/rewards/mean": torch.mean(sequence_reward).detach().item(), - "critic/rewards/max": torch.max(sequence_reward).detach().item(), - "critic/rewards/min": torch.min(sequence_reward).detach().item(), + "critic/rewards/mean": reward_mean, + "critic/rewards/max": reward_max, + "critic/rewards/min": reward_min, # adv "critic/advantages/mean": torch.mean(valid_adv).detach().item(), "critic/advantages/max": torch.max(valid_adv).detach().item(), @@ -163,6 +192,15 @@ def compute_data_metrics(batch: DataProto, use_critic: bool = True) -> dict[str, "response_length/clip_ratio": torch.mean(torch.eq(response_length, max_response_length).float()) .detach() .item(), + # response length (non-aborted only) + # These statistics exclude aborted samples to avoid skew from zeros + "response_length_non_aborted/mean": non_aborted_response_length_mean, + "response_length_non_aborted/max": non_aborted_response_length_max, + "response_length_non_aborted/min": non_aborted_response_length_min, + "response_length_non_aborted/clip_ratio": non_aborted_response_length_clip_ratio, + # aborted ratio + # Fraction of samples whose response length is zero + "response/aborted_ratio": aborted_ratio, # prompt length "prompt_length/mean": torch.mean(prompt_length).detach().item(), "prompt_length/max": torch.max(prompt_length).detach().item(), diff --git a/verl/workers/config/rollout.py b/verl/workers/config/rollout.py index 370a61f928c..0b5e21dbfbe 100644 --- a/verl/workers/config/rollout.py +++ b/verl/workers/config/rollout.py @@ -86,6 +86,10 @@ class RolloutConfig(BaseConfig): do_sample: bool = True n: int = 1 + # Early termination threshold for multi-turn rollout in sglang. + # Abort remaining requests when (1 - over_sample_rate) * total_requests are completed. + over_sample_rate: float = 0.0 + prompt_length: int = 512 response_length: int = 512 diff --git a/verl/workers/rollout/sglang_rollout/sglang_rollout.py b/verl/workers/rollout/sglang_rollout/sglang_rollout.py index cd0712fc8f3..383b98ce9d6 100644 --- a/verl/workers/rollout/sglang_rollout/sglang_rollout.py +++ b/verl/workers/rollout/sglang_rollout/sglang_rollout.py @@ -159,6 +159,20 @@ async def update_weights_from_tensor(self, update_weights_request: UpdateWeights async def flush_cache(self): return await self.tokenizer_manager.flush_cache() + async def abort_request(self, rid: str = "", abort_all: bool = False): + """Abort a specific request or all requests. + + Args: + rid: The request ID to abort. If empty and abort_all is False, no action is taken. + abort_all: If True, abort all running requests regardless of rid. + """ + try: + result = self.tokenizer_manager.abort_request(rid=rid, abort_all=abort_all) + return result if result is not None else {"status": "aborted"} + except Exception as e: + logger.error(f"Failed to abort requests: {e}") + raise + # NOTE(sgm): add for verl. We can optimize it by making # the dataloader yield List[int] without padding. @@ -450,16 +464,17 @@ def _init_inference_engine(self, trust_remote_code, actor_module, port): # NOTE(linjunrong): add rank to prevent SGLang generate same port inside PortArgs.init_new # when random.seed is being set during training port=30000 + rank, - # NOTE(Chenyang): if you want to debug the SGLang engine output - # please set the following parameters - # Otherwise, it will make the engine run too slow - # log_level="INFO", + # NOTE(Chenyang): turn on log_level to see the decoding speed of SGLang Engine + # log_level="INFO" + # NOTE(Chenyang): turn the following lines to see the input and output of each request # log_requests=True, # log_requests_level=2, + # NOTE(Chenyang): turn on max_running_requests to set the max concurrent running requests # max_running_requests=1, mm_attention_backend="fa3", attention_backend=attention_backend if attention_backend is not None else "fa3", - # In async mode, we want token in token out. + # In async mode for AgentLoop, SGLang support token in token out to avoid the tokenizer + # inconsistency issue. skip_tokenizer_init=self.config.mode == "async", ) else: @@ -861,7 +876,9 @@ async def _async_rollout_a_request( # Only continue the conversation if the prompt length is not greater than max_model_len - 1, # since SGLang raises an error when max_new_tokens + 1 is greater to max_model_len (the extra # token accounts for the EOS token). - if len(_req.get_generation_prompt_ids(self.processing_class)) + 1 >= self.config.max_model_len: + prompt_length = len(_req.get_generation_prompt_ids(self.processing_class)) + + if prompt_length + 1 >= self.config.max_model_len: finish_reason_type = FinishReasonTypeEnum.LENGTH break @@ -989,7 +1006,6 @@ async def calc_reward_and_release_fn(name: str, tool: BaseTool): all_rewards = {**tool_reward_scores, **{"user_turn_rewards": user_turn_rewards}} _req.finalize(self.processing_class, all_rewards, finish_reason_type) if self.config.calculate_log_probs: - # 把input_ids输入sglang内生成一遍，并设置max_new_tokens=0，以生成log_probs debug_sampling_params = {**self.sampling_params} debug_sampling_params["max_new_tokens"] = 0 output = await self._engine.async_generate( @@ -1013,9 +1029,11 @@ async def _handle_engine_generate( self, generation_prompt_ids: list[int], sampling_params: dict, image_data: Optional[list[Any]] = None ) -> dict: max_new_tokens = min(self.config.response_length, self.config.max_model_len - len(generation_prompt_ids) - 1) + kwargs = sampling_params.copy() kwargs["max_new_tokens"] = max_new_tokens kwargs["n"] = 1 # group size is supported in preprocess + output = await self._engine.async_generate( input_ids=generation_prompt_ids, sampling_params=kwargs, @@ -1048,16 +1066,6 @@ async def _handle_pending_state(self, _req: AsyncRolloutRequest) -> AsyncRollout interaction = self.interaction_map[interaction_name] await interaction.start_interaction(_req.request_id, **interaction_kwargs) - @GPUMemoryLogger(role="sglang rollout", logger=logger) - @torch.no_grad() - def generate_sequences_with_tools(self, prompts: DataProto, **kwargs) -> DataProto: - logger.warning( - "`generate_sequences_with_tools` is deprecated, please use `generate_sequences(...)`", - DeprecationWarning, - stacklevel=2, - ) - return self._req_level_generate_sequences(prompts, **kwargs) - @GPUMemoryLogger(role="sglang rollout", logger=logger) @torch.no_grad() def _req_level_generate_sequences(self, prompts: DataProto, **kwargs) -> DataProto: @@ -1071,16 +1079,74 @@ def _req_level_generate_sequences(self, prompts: DataProto, **kwargs) -> DataPro do_sample = prompts.meta_info.get("do_sample", True) is_validate = prompts.meta_info.get("validate", False) tgt_device = prompts.batch["input_ids"].device + if self._tp_rank == 0: req_list = self._preprocess_prompt_to_async_rollout_requests( prompts, ) - loop = asyncio.get_event_loop() - output_req_list = loop.run_until_complete( - asyncio.gather( - *[self._async_rollout_a_request(req, do_sample, is_validate, **kwargs) for req in req_list], + + # distinguish training and validation + if is_validate: + # Validation mode: process all requests without abort + loop = asyncio.get_event_loop() + output_req_list = loop.run_until_complete( + asyncio.gather( + *[self._async_rollout_a_request(req, do_sample, is_validate, **kwargs) for req in req_list], + ) ) - ) + else: + # add progress monitoring and abort function + total_requests = len(req_list) + target_completion = int(total_requests * (1 - self.config.over_sample_rate)) + # abort when target_completion of requests are completed + + completed_count = 0 + aborted_requests = [] + all_tasks = [] + + async def rollout_a_request_with_cancellation_handler(req): + try: + result = await self._async_rollout_a_request(req, do_sample, is_validate, **kwargs) + return result + except asyncio.CancelledError: + # request is cancelled, return padding + logger.info(f"Request {req.request_id} was cancelled, creating padding") + aborted_requests.append(req.request_id) + return self._create_padding_request(req) + + async def run_with_cancellation(): + nonlocal all_tasks + nonlocal completed_count + all_tasks = [ + asyncio.create_task(rollout_a_request_with_cancellation_handler(req)) for req in req_list + ] + + # Wait for target_completion tasks to complete + try: + for completed_task in asyncio.as_completed(all_tasks): + await completed_task + completed_count += 1 + if completed_count >= target_completion: + break + finally: + # Cancel remaining tasks + for t in all_tasks: + if not t.done(): + t.cancel() + + # Wait for all tasks to finish (including cancelled ones) + final_results = await asyncio.gather(*all_tasks, return_exceptions=True) + + try: + await self._engine.abort_request(abort_all=True) + except Exception as e: + logger.error(f"Failed to send abort signal to SGLang engine: {e}") + + return final_results + + loop = asyncio.get_event_loop() + output_req_list = loop.run_until_complete(run_with_cancellation()) + sorted_output_req_list = sorted(output_req_list, key=lambda x: (x.batch_data_id, x.rollout_offset)) else: sorted_output_req_list = None @@ -1268,6 +1334,88 @@ def _req_level_generate_sequences(self, prompts: DataProto, **kwargs) -> DataPro non_tensor_batch=non_tensor_batch, ) + def _create_padding_request(self, original_req: AsyncRolloutRequest) -> AsyncRolloutRequest: + # create a padding request to replace the aborted request + # the padding request has the following characteristics: + # 1. state is COMPLETED, but contains empty response + # 2. response_loss_mask is all 0, ensuring it is ignored in loss calculation + # 3. keep the original request structure, but the content is empty + # create padding response_ids (all pad_token_id) + padding_response_length = self.config.response_length + padding_response_ids = torch.full( + (1, padding_response_length), + self.pad_token_id, + dtype=torch.long, + device=original_req.input_ids.device if original_req.input_ids is not None else "cpu", + ) + + # create padding attention_mask (all 0) + padding_response_attention_mask = torch.zeros( + (1, padding_response_length), + dtype=torch.long, + device=original_req.attention_mask.device if original_req.attention_mask is not None else "cpu", + ) + + # create padding position_ids + if original_req.position_ids is not None: + prompt_length = original_req.prompt_ids.shape[-1] if original_req.prompt_ids is not None else 0 + padding_response_position_ids = torch.arange( + prompt_length, prompt_length + padding_response_length, dtype=torch.long + ).unsqueeze(0) + if original_req.position_ids.dim() == 2: + # if it is a 2D tensor (e.g. qwen2vl) + padding_response_position_ids = padding_response_position_ids.repeat( + original_req.position_ids.shape[0], 1 + ) + else: + padding_response_position_ids = None + + # create padding loss_mask (all 0, ensuring it is ignored) + padding_response_loss_mask = torch.zeros( + (1, padding_response_length), + dtype=torch.long, + device=original_req.loss_mask.device if original_req.loss_mask is not None else "cpu", + ) + + padding_req = AsyncRolloutRequest( + batch_data_id=original_req.batch_data_id, + rollout_offset=original_req.rollout_offset, + request_id=original_req.request_id, + state=AsyncRolloutRequestStateEnum.COMPLETED, + messages=original_req.messages, + multi_modal_keys=original_req.multi_modal_keys, + multi_modal_data=original_req.multi_modal_data, + multi_modal_inputs=original_req.multi_modal_inputs, + tool_schemas=original_req.tool_schemas, + tools_kwargs=original_req.tools_kwargs, + interaction_kwargs=original_req.interaction_kwargs, + input_ids=original_req.input_ids, + prompt_ids=original_req.prompt_ids, + response_ids=padding_response_ids, + attention_mask=original_req.attention_mask, + prompt_attention_mask=original_req.prompt_attention_mask, + response_attention_mask=padding_response_attention_mask, + position_ids=original_req.position_ids, + prompt_position_ids=original_req.prompt_position_ids, + response_position_ids=padding_response_position_ids, + loss_mask=original_req.loss_mask, + prompt_loss_mask=original_req.prompt_loss_mask, + response_loss_mask=padding_response_loss_mask, + reward_scores={}, + max_prompt_len=original_req.max_prompt_len, + max_response_len=original_req.max_response_len, + metrics={}, + output_token_ids=None, + rollout_log_probs=None, + use_inference_chat_template=original_req.use_inference_chat_template, + tokenization_sanity_check_mode=original_req.tokenization_sanity_check_mode, + generation_prompt_ids=original_req.generation_prompt_ids, + base_conv_wo_gen_prompt_end_pos=original_req.base_conv_wo_gen_prompt_end_pos, + base_conv_with_gen_prompt_end_pos=original_req.base_conv_with_gen_prompt_end_pos, + processing_class=self.processing_class, + ) + return padding_req + def _preprocess_prompt_to_async_rollout_requests(self, prompts: DataProto, n: int = 1) -> list[AsyncRolloutRequest]: assert "raw_prompt" in prompts.non_tensor_batch, ( "need data.return_raw_chat=True, due to no official way do parse_messages"