`nemo_rl.data.interfaces`#

Module Contents#

Classes#

`NemoGymSourceIdentity`	Filesystem identity used to cache agent names from one Gym dataset.
`DatumSpec`
`PreferenceDatumSpec`
`TaskDataSpec`
`TaskDataProcessFnCallable`	A callable that processes a loaded datum dictionary into a DatumSpec.
`TaskDataPreProcessFnCallable`	A callable that processes a loaded raw datum dictionary into a dictionary with required format for further processing.

Data#

`LLMMessageLogType`
`VLMMessageLogType`
`FlatMessagesType`
`PathLike`
`TokenizerType`

API#

nemo_rl.data.interfaces.LLMMessageLogType#: None

nemo_rl.data.interfaces.VLMMessageLogType#: None

nemo_rl.data.interfaces.FlatMessagesType#: None

nemo_rl.data.interfaces.PathLike#: None

nemo_rl.data.interfaces.TokenizerType#: None

class nemo_rl.data.interfaces.NemoGymSourceIdentity#

Filesystem identity used to cache agent names from one Gym dataset.

path: str#: None

device: int#: None

inode: int#: None

ctime_ns: int#: None

mtime_ns: int#: None

size: int#: None

classmethod from_stat( path: str, stat: os.stat_result, ) → nemo_rl.data.interfaces.NemoGymSourceIdentity#: Build a source identity snapshot from a resolved path and stat result.

matches(stat: os.stat_result) → bool#: Return whether stat still describes the same source contents.

class nemo_rl.data.interfaces.DatumSpec#

Bases: typing.TypedDict

message_log: nemo_rl.data.interfaces.LLMMessageLogType | nemo_rl.data.interfaces.VLMMessageLogType#: None

length: int#: None

extra_env_info: Optional[dict[str, Any]]#: None

loss_multiplier: float#: None

idx: int#: None

task_name: NotRequired[str]#: None

message_loss_mask: NotRequired[list[int]]#: None

tools: NotRequired[list[dict[str, Any]]]#: None

stop_strings: NotRequired[list[str]]#: None

__extra__: NotRequired[Any]#: None

class nemo_rl.data.interfaces.PreferenceDatumSpec#

Bases: typing.TypedDict

message_log_chosen: nemo_rl.data.interfaces.LLMMessageLogType | nemo_rl.data.interfaces.VLMMessageLogType#: None

message_log_rejected: nemo_rl.data.interfaces.LLMMessageLogType | nemo_rl.data.interfaces.VLMMessageLogType#: None

length_chosen: int#: None

length_rejected: int#: None

loss_multiplier: float#: None

idx: int#: None

task_name: NotRequired[str]#: None

class nemo_rl.data.interfaces.TaskDataSpec#

task_name: Optional[str]#: None

prompt_file: Optional[nemo_rl.data.interfaces.PathLike]#: None

system_prompt_file: Optional[nemo_rl.data.interfaces.PathLike]#: None

num_frames: Optional[int]#: None

video_sampling_style: Optional[str]#: None

video_target_num_patches: Optional[int]#: None

video_temporal_patch_size: Optional[int]#: None

video_maintain_aspect_ratio: Optional[bool]#: None

pad_audio_to_hop_length: Optional[bool]#: None

min_generation_tokens: Optional[int]#: None

image_max_num_tiles: Optional[int]#: None

__post_init__() → None#

copy_defaults(from_spec: nemo_rl.data.interfaces.TaskDataSpec) → None#: Apply default values from another Task instance for any None attributes.

class nemo_rl.data.interfaces.TaskDataProcessFnCallable#

Bases: typing.Protocol

A callable that processes a loaded datum dictionary into a DatumSpec.

abstractmethod __call__( datum_dict: dict[str, Any], task_data_spec: nemo_rl.data.interfaces.TaskDataSpec, tokenizer: nemo_rl.data.interfaces.TokenizerType, max_seq_length: int | None, idx: int, ) → nemo_rl.data.interfaces.DatumSpec#

class nemo_rl.data.interfaces.TaskDataPreProcessFnCallable#