nemo_rl.data.energon.multimodal.task_encoders.generic_sft

Module Contents

Classes

SFTProcessorAdapter	Boundary between canonical and model-specific SFT data.
HFMultimodalSFTProcessorAdapter	Hugging Face implementation of the generic processor boundary.
GenericSFTTaskEncoder	Encode, group, and batch complete multimodal SFT conversations.

Functions

_normalize_messages	Validate the message structure and attach each part’s media.
build_processor_adapter	Build the configured model processor adapter.

Data

__all__

API

class nemo_rl.data.energon.multimodal.task_encoders.generic_sft.SFTProcessorAdapter

Bases: typing.Protocol

Boundary between canonical and model-specific SFT data.

property fingerprint: str

encode(

sample: nemo_rl.data.energon.multimodal.types.CanonicalSFTSample,

) → nemo_rl.data.energon.multimodal.types.EncodedSFTSample

nemo_rl.data.energon.multimodal.task_encoders.generic_sft._normalize_messages(

sample: nemo_rl.data.energon.multimodal.types.CanonicalSFTSample,

*,

materialize: bool = True,

) → list[dict[str, Any]]

Validate the message structure and attach each part’s media.

Parameters:

• sample – The cooked conversation.

• materialize – Decode each media value and attach the payload. Set False to attach the MediaRef instead.

Model-specific renderers may replace media parts with text built from metadata and overwrite message["content"] wholesale, so decoding for them is pure waste. It is also waste paid at the wrong time: this runs in pre-encode, before select_samples_to_pack, so rows that selection discards are decoded too. Model-specific encoders can defer media work to post-encode by setting materialize=False.

Only GenericSFTTaskEncoder.encode consumes the payload, via get_formatted_message_log, so it keeps the default.

class nemo_rl.data.energon.multimodal.task_encoders.generic_sft.HFMultimodalSFTProcessorAdapter(

*,

processor: Any,

max_sequence_length: int,

add_bos: bool,

add_eos: bool,

add_generation_prompt: bool,

)

Hugging Face implementation of the generic processor boundary.

Initialization

property fingerprint: str

encode(

sample: nemo_rl.data.energon.multimodal.types.CanonicalSFTSample,

) → nemo_rl.data.energon.multimodal.types.EncodedSFTSample

class nemo_rl.data.energon.multimodal.task_encoders.generic_sft.GenericSFTTaskEncoder(

*,

adapter: nemo_rl.data.energon.multimodal.task_encoders.generic_sft.SFTProcessorAdapter,

cooker_functions: collections.abc.Sequence[nemo_rl.data.energon.multimodal.task_encoders.base.SFTCooker],

include_source_ids: bool,

packer: nemo_rl.data.packing.SequencePacker | None = None,

tokenizer: Any | None = None,

sequence_length_pad_multiple: int = 1,

only_unmask_final: bool = False,

loss_mask_mode: Literal[precomputed] | None = None,

)

Bases: nemo_rl.data.energon.multimodal.task_encoders.base.BaseSFTTaskEncoder

Encode, group, and batch complete multimodal SFT conversations.

Initialization

Configure conversation encoding and optional sequence packing.

Parameters:

• adapter – Processor adapter that encodes each conversation.

• cooker_functions – Cookers that produce canonical SFT samples.

• include_source_ids – Include source identifiers in each batch.

• packer – Sequence packer, or None to batch individual conversations.

• tokenizer – Tokenizer used to pad packed batches.

• sequence_length_pad_multiple – Alignment for each source in a pack.

• only_unmask_final – Train only on the final assistant turn in default mode.

• loss_mask_mode – None builds assistant-role loss masks. “precomputed” preserves each message’s token_loss_mask, which must be a binary one-dimensional tensor matching token_ids. Precomputed masks do not support only_unmask_final=True. Both modes support packed and unpacked conversations.

__default_failure_tolerance__

1

decoder

‘SampleDecoder(…)’

preencode_sample(

sample: nemo_rl.data.energon.multimodal.types.CanonicalSFTSample,

) → nemo_rl.data.energon.multimodal.types.EncodedSFTSample

postencode_sample(

sample: nemo_rl.data.energon.multimodal.types.EncodedSFTSample,

) → nemo_rl.data.energon.multimodal.types.EncodedSFTSample

batch_group_criterion(

sample: nemo_rl.data.energon.multimodal.types.EncodedSFTSample | nemo_rl.data.energon.multimodal.types.PackedSFTSample,

) → tuple[tuple[Any, ...], None]

select_samples_to_pack(

samples: list[nemo_rl.data.energon.multimodal.types.EncodedSFTSample],

) → list[list[nemo_rl.data.energon.multimodal.types.EncodedSFTSample]]

pack_selected_samples(

samples: list[nemo_rl.data.energon.multimodal.types.EncodedSFTSample],

) → nemo_rl.data.energon.multimodal.types.PackedSFTSample

batch(

samples: list[nemo_rl.data.energon.multimodal.types.EncodedSFTSample | nemo_rl.data.energon.multimodal.types.PackedSFTSample],

) → nemo_rl.distributed.batched_data_dict.BatchedDataDict[Any]

encode_batch(

batch: nemo_rl.distributed.batched_data_dict.BatchedDataDict[Any],

) → nemo_rl.distributed.batched_data_dict.BatchedDataDict[Any]

nemo_rl.data.energon.multimodal.task_encoders.generic_sft.build_processor_adapter(

*,

processor_adapter: str,

processor: Any,

max_sequence_length: int,

add_bos: bool,

add_eos: bool,

add_generation_prompt: bool,

) → nemo_rl.data.energon.multimodal.task_encoders.generic_sft.SFTProcessorAdapter

Build the configured model processor adapter.

nemo_rl.data.energon.multimodal.task_encoders.generic_sft.__all__

[‘GenericSFTTaskEncoder’, ‘HFMultimodalSFTProcessorAdapter’, ‘SFTProcessorAdapter’, ‘build_processor…