Dict-like get for compatibility with code that uses skip.get('enable', False).

Move model parameters, optimizer states, or both to the specified device. Args: device: Target device identifier.

Pop a subset of the DataProto via `batch_keys` and `meta_info_keys` Args: batch_keys (list, optional): a list of strings indicati

Get the device type string based on available accelerators. Detects the available accelerator and returns the corresponding PyTorch device ty

Get the value associated with the given key. If the key does not exist, return the default value. Args: key (str): The attribute

split the dataset into num_splits sub-datasets Args: num_splits: specified number of splits Returns:

Time a code block.

Get parameter with default.

Update method for fixed KL controller (no-op). Args: current_kl (float): Current KL divergence value (unused). n_step

Move working copy back to CPU and free GPU memory.

Load the replay pool from a directory.

Get the PyTorch device module for the current accelerator. Returns the torch device namespace (e.g., torch.cuda, torch.npu) based on the dete

Get a rollout replica class by name.

Get the index of the current accelerator device. Returns: int: The current device index (e.g., 0 for 'cuda:0').

Convert an OmegaConf DictConfig to a dataclass. Args: config: The OmegaConf DictConfig or dict to convert. dataclass_type: T

Block until event completes.

Context manager for timing with platform markers. This utility function measures the execution time of code within its context, accumulates t

move the batch to device Args: device (torch.device, str): torch device Returns: DataProto: the current Data

Log GPU memory usage information. Args: head (str): A descriptive header for the memory usage log message. logger (logging.Logger

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

Create a tool instance. Args: instance_id: The instance id of the tool. Returns: The instance id of the tool

Select a subset of the DataProto via batch_keys and meta_info_keys Args: batch_keys (list, optional): a list of strings indicatin

Split the batch among dim=0 into chunks. The meta_info is passed to each DataProto after split. Args: chunks (int): the number of

Sleep each rollout server.

_summary_ Log a message with rank information using a logger. This function logs the message only if `log_only_rank_0` is False or if the rank

Split the batch among dim=0 into chunks. The meta_info is passed to each DataProto after split. Args: split_size (int): the size

Zero out accumulated gradients.

Repeat the batch data a specified number of times. Args: repeat_times (int): Number of times to repeat the data.

Return current device of a layer.

Concat a list of DataProto. The batch is concatenated among dim=0. The meta_info is merged, with special handling for metrics from different w

Create a huggingface pretrained tokenizer which correctness handles eos and pad tokens. Args: name (str): The name of the tokenizer.

Execute the main PPO training workflow. This method sets up the distributed training environment, initializes workers, datasets, and

Copy files/directories from HDFS to local cache with validation. Args: src (str): Source path - HDFS path (hdfs://...), local filesystem

Make this stream wait for an event.

A func to sync embedding input with alltoall in sequence parallel gather sequence dimension and scatter head dim: e.g. seq_dim: 1, head_d

Context manager for profiling a code block.

Sleep all rollout replica instances.

Init process group for checkpoint engine. Args: **kwargs: Keyword arguments from `build_topology`.

Save the replay pool to a directory.

Acquire tokens from the bucket, waiting if necessary. This method will block (using asyncio.sleep) until sufficient tokens are availa

Converts a sequence of ids in a string, using the tokenizer and vocabulary with options to remove special tokens and clean up tokeniz

Aggregate the loss across global batch to ensure the loss is invariant to fsdp/megatron parallelism. NOTE: The returned loss has different b

Batch generate sequences in sync mode. Args: prompts: The input prompts. Returns: The output sequences.

Get the resource pool of the worker_cls

If distributed is initialized, print only on rank 0.

Compute per-token log-probabilities for the given labels. Uses a Flash-Attention–based cross-entropy (if available) for efficient backward,

Backward pass with proper gradient computation. Returns: grad_x, grad_weight, None (for eps)

r"""Works like shutil.copy() for file, and shutil.copytree for dir, and supports hdfs. Copy data and mode bits ("cp src dst"). Return the file's

Normalize tokenizer outputs into a flat ``list[int]``. This handles Transformers 4/5 differences where ``apply_chat_template(tokenize=True)``

Apply chat template to messages with optional tools, images, and videos. Args: messages (list[dict]): Input messages.

Unified interface for computing IS weights and rejection masks. This function combines IS weight calculation (truncated) and rejection sampling (

Milliseconds between this event and end_event.

r"""Works like os.makedirs() but supports hdfs. Super-mkdir; create a leaf directory and all intermediate ones. Works like mkdir, except tha

Register the dp_rank for a given mesh name. This function is meant to be called by the worker Args: mesh_name (str):

More aggressive GPU memory cleanup function, tries to release PyTorch reserved but unallocated memory. Args: force_sync: Whether

Create a huggingface processor to process multimodal data. Args: name_or_path (str): The name of the processor. Returns: Opt

Spawn to a dictionary of worker groups, each with a subset of method with prefix. Args: prefix_set: Set of prefixes to create wor

Create a tool instance. Args: instance_id: The instance id of the tool. Returns: The instance id of the tool

Convert this DataProto to TensorDict. Note that this requires tensordict version at least 0.10 Returns: