We compute dense reward here so that we can directly train RL without SFT

Computes throughput metrics for PPO training. This function calculates performance metrics related to token processing speed, including

Computes timing metrics for different processing stages in PPO training. This function calculates both raw timing metrics (in seconds) and p

Token embedding module. Infinity execution: prefetch(adapter.embedding) hidden = adapter.embedding(input_ids

Create a DataProto from a TensorDict. This assumes that 1. All the tensor in tensor_dict have the same dim0 2. Only dim0 is the batch

Generate text using the SGLang server. Args: prompt (Optional[str], optional): Text prompt for generation. Defaults to None.

Return a tensor with the input `shape` as a view into the 1-D data starting at `start_index`.

Return the available VL auto model class across transformers versions.

Create a constant LR schedule with a linear warmup phase. Args: optimizer (Optimizer): Wrapped optimizer. num_warmup_steps (

Get all registered dispatch and collect dp_ranks. Returns: dict[str, int]: A dictionary mapping mesh names to the

Get the index offset of any pipeline stage, given the level of pipelining. Make pipeline_rank and vp_stage as two arguments to make it more

Get the environment variable name for visible device selection. Returns the appropriate environment variable name based on the available acce

Check if a path is a non-local (HDFS) path. Args: path (str): The path to check. Returns: bool: True if the path is an HDFS

Return True if the current router_replay_action is RECORD (R2) for the local router instances. This inspects the first local RouterReplay ins

Load model, optimizer, and scheduler states from a checkpoint. Args: local_path: Local filesystem path of the checkpoint

Applies a monkey patch to the forward method of a given model class to enable Ulysses sequence parallelism input slicing.

Postprocess packed sequences

Reset the model engine to the initial state. If the engine is not initialized, we initialize it. Otherwise, reload ckpt and reset sta

Resume rollout weights or kv cache in GPU memory. Args: tags: weights or kv_cache.

Resume GPU memory occupation. Args: tags (Optional[List[str]], optional): List of tags to specify which memory to resume.

Save model, optimizer, and scheduler states to a checkpoint. Args: local_path: Local filesystem path to save checkpoint.

Set the dispatch and collect dp_ranks for all registered meshes. Args: mesh_name (str): Mesh name to set dispatch and collect dp_

Shutdown the HTTP server and clean up resources. This method performs the following cleanup operations: 1. Unregisters the worker fro

Move model parameters, optimizer states, or both to the specified device. Note that this function executes irrespective of offload co

Validate actor configuration with runtime parameters.

Return names of modules selected for wrapping.

Apply KL penalty to the token-level rewards. This function computes the KL divergence between the reference policy and current policy, then a

Convert arbitrary group labels to a contiguous 1-D torch.long tensor (0..G-1). Args: index: Any iterable of labels or tensor/ndarray

Build an optimizer based on the configuration. Dynamically imports and instantiates an optimizer class from the specified module. Args:

calculate rollout vs actor logprobs diff, for debugging purpose Args: data: DataProto the data batch to calculate

Check and disable incompatible configurations for older Megatron version. Args: original_config (dict): The original model configura

Compute off-policy diagnostic metrics (helper function). This helper function operates on raw tensors and is used internally by: - compute_ro

Compute the clipped policy objective and related metrics for PPO. Adapted from https://github.com/huggingface/trl/blob/main/trl/trainer/

Generates a memory snapshot and saves it as a pickle file in a specified directory. The files are organized by timestamp in subdirect

Helper function for reproducibility in distributed training. See https://pytorch.org/docs/stable/notes/randomness.html for details.

Args: tensor: a tensor with shape (bsz,) nested_tensor: a nested tensor with shape (bsz, xxx) Returns: a tensor wit

Return the most recent checkpoint directory based on a tracker file. Args: path (str): Base directory containing the checkpoint trac

Fallback implementation for older transformers versions. Returns False to disable features that require this function.

Perform a forward pass and optionally a backward pass on a batch of data. Args: data: The input data for the forward pas

Forward-only pass that returns logits. Used for inference (rollout, ref policy, etc.). In multi-GPU mode, splits the batch across workers for

Spawn agent loop for each sample in the batch without waiting for the results.

Create a schedule with a learning rate that decreases following the values of the cosine function between the initial lr set in the optimizer

Extract a sub-dictionary from a TensorDict based on a given prefix. Args: tensordict: The input TensorDict containing various keys.

Return RoPE base frequency theta. Most configs expose ``rope_theta`` on the root. Newer models (e.g. Qwen3 in transformers>=5) store it under

Get the policy loss with a given name. Args: name: `(str)` The name of the policy loss. Returns: `(callable)`: T

Convert huggingface PretrainedConfig to mcore TransformerConfig. Args: hf_config: The huggingface PretrainedConfig. dtype: The dt

Launch an SGLang HTTP server process and wait for it to be ready. This function starts a new process running an SGLang HTTP server, then waits

Normalize a final answer to a quantitative reasoning question. Args: final_answer: The answer string to normalize Returns: N

Pad a DataProto to size divisible by size_divisor Args: size_divisor (int): size divisor Returns: data: (DataProto): the pad

postprocess the output of a forward_backward_batch. output_lst is a list of dict containing outputs for each micro-batch reorder entropy and o

Postprocess packed sequences

Preprocess bshd sequences return "input_ids, attention_mask, position_ids"

Process validation metrics into a structured format with statistical analysis. This function organizes validation metrics by data source and

Extract response from nested model output. Args: tensor: a nested tensor with shape (bsz, prompt_len + response_len) response_mas

Apply RMSNorm with custom gradient.

Safely create a task with exception handling Args: coro: The coroutine to run name: Name for the task task_set: Optional

Select specific indices from the DataProto. Args: idxs (torch.Tensor or numpy.ndarray or list): Indices to select

Sets the router replay action for all router instances.

Sets the router replay action for this layer.

Move FSDP model and/or optimizer to CPU or GPU with offload support. Note that this function executes irrespective of offload config.

Perform a training step on a batch of data. Args: data: The input data for training, typically containing tensors and me

Unpack the output from the visual encoder, handling both tuple and object return types. Newer versions of transformers return an object with `poo

Unpad the data proto with pad_size. i.e. `data[:-pad_size]`