Helper function containing the loading hf model logic

Log samples to wandb as a table

Log rollout data to disk. Args: batch (DataProto): The batch containing rollout data reward_extra_infos_dict (dict): A

Build a minimal ServerAdapter-like object without real init.

Create a numpy index array ensuring each group has at least 2 samples.

Process scales for each expert into Marlin format and stack.

Repack weight for each expert into Marlin format and stack.

Merge image embeddings into text hidden states at image token positions.

Move layer parameters to device.

Normalize answer expressions.

Pad an array to the target dimension with zeros along the specified axis.

Common post-processing logic for function outputs in TransferQueue bridge. This function handles the final return value based on whether data sho

Args: output: a dictionary containing loss, model_outputs and metrics Returns:

Issue prefetch for next layer.

Convert 2D padding mask [B, T] to 4D causal attention mask [B, 1, T, T].

Prepares input for action prediction by adding necessary tokens

Creates labels tensor for action prediction if not provided

Helper to get action masks from labels

Quantize one decoder layer's buffered params. Returns list of (name, tensor).

Process images through vision encoder + projector on GPU, then offload.

Process vision features with optional FiLM conditioning

Shared NPU routed-expert path for Qwen3Moe/Qwen3Next sparse MoE blocks. Returns: tuple: (flattened_input, routed_hidden_states,

Remove generation prompt IDs from input tensors if they are present at the end.

Remove all registered hooks.

Resolve device according to policy described in the module docstring. Priority: 1) explicit argument 2) VERL_FORCE_DEVICE env

Run diffusion-based action prediction

Simulate the sleep() method's tag selection logic and return chosen tags.

add noise to the action sampling process using Flow-SDE method. see https://arxiv.org/abs/2510.25889

Save actor, critic, and dataloader checkpoints to local (and optionally remote) storage.

Select modules to wrap individually with fully_shard in FSDP2. Matches transformer layers by class name, and embed_tokens/lm_head by name (wi

Determine whether the build should be paused

Start profiling for all worker groups including env workers.

Stop profiling for all worker groups including env workers.

Strip padding from (bs, seq_len, dim) embeds using the boolean mask and return nested tensors.

Update LRU order.

Unnormalize actions using dataset statistics

Convert a jagged nested tensor pair (embeds, mask) to dense padded tensors.

Update the actor network.

Update the critic network.

Validate the bounding box dimensions and aspect ratio.

Block until layer idx is ready.

Ensure layer is ready for compute.

Refresh worker's GPU modules from shared CPU master params.

Acquire a token from the bucket.

Get a GPU slot for tensor_id. Returns None if eviction needed.

Yield param pairs with a base-layer suffix added to the param name. Args: params: Iterator of (param_name, tensor) model_type: Th

Add a batch of samples into task-specific positive/negative pools.

Add a compute operation.

Simulate data stream

Accumulate gradient (handles device/dtype conversion).

allgather a dict into a dict of list Args: data: a dict group: the process group to allgather Returns: dict containing a lis

Setup bypass mode: Use rollout_log_probs as old_log_probs. Bypass mode skips expensive actor forward pass for old_log_prob computation b

Apply FSDP2 sharding to model.

Apply ModelOpt NVFP4 patches to support dynamic weight updates. Call before model loading.

Apply NPU patches for FSDP backend if NPU is available.

Apply QAT to a model by replacing nn.Linear with QATLinear.