nemo_curator.tasks.tasks

Module Contents

Classes

Name	Description
`Task`	Abstract base class for tasks in the pipeline.

Data

T

API

class nemo_curator.tasks.Task(
    dataset_name: str,
    data: nemo_curator.tasks.tasks.T,
    _stage_perf: list[nemo_curator.utils.performance_utils.StagePerfStats] = list(),
    _metadata: dict[str, typing.Any] = dict()
)

DataclassAbstract

Bases: Generic[T]

Abstract base class for tasks in the pipeline.

A task represents a batch of data to be processed. Different modalities (text, audio, video) can implement their own task types.

_metadata

dict[str, Any] = field(default_factory=dict)

Free-form metadata carried alongside the task.

_source_id

str = field(init=False, default='')

Source (input partition) this task descends from. Stamped at the source stage, inherited downstream; used only by the opt-in resumability layer. Empty for pre-source tasks.

_stage_perf

list[StagePerfStats] = field(default_factory=list)

Per-stage perf stats this task has accumulated.

data

The task’s payload (modality-specific).

dataset_name

str

Name of the dataset this task belongs to.

num_items

int

Get the number of items in this task.

task_id

str = field(init=False, default='')

Deterministic identifier for this task. NOT user-settable — the framework assigns it via _set_task_id at every stage boundary. It is an underscore-joined id path through the pipeline DAG — the parents’ ids plus this task’s own segment (e.g. "abc123_0_5" = source abc123, then child 0, then grandchild 5). Using the readable path directly (rather than a hash of it) keeps task ids easy to debug. Empty string until the first stage runs; two runs of the same pipeline on the same inputs produce byte-identical task_ids across all tasks.

A task_id that starts with "r" (followed by a uuid) is a fallback assigned when the parent→child mapping could NOT be derived — e.g. a stage that overrides process_batch with an ambiguous batch fan-out (M inputs → K≠M outputs). Such ids are NON-deterministic (differ across runs).

nemo_curator.tasks.Task.__post_init__() -> None

Post-initialization hook.

nemo_curator.tasks.Task.__repr__() -> str

nemo_curator.tasks.Task._set_task_id(
    parent_task_id: str,
    current_task_id_suffix: str | int
) -> None

Assign this task’s deterministic task_id from its parent.

The task_id is the parent id and this task’s own segment joined by "_" — e.g. parent "abc123" + suffix 0 → "abc123_0". Always overwrites task_id; there is no idempotency check — each stage transition re-derives it, so the same physical Python object passing through N stages gets N distinct task_ids (one per stage boundary). The dedup keys used by resumability are captured BEFORE this method runs on a given output, so the rewrite is safe.

Only a single parent id is taken: the supported mappings (1→1, 1→N fan-out, N→N positional) each give an output exactly one parent. N→1 aggregations don’t track ancestry — those outputs get a random "r"-prefixed id in the adapter instead of calling this.

Parameters:

parent_task_id

str

task_id of the parent. An empty string (an unassigned / EmptyTask parent) is dropped so it doesn’t contribute a leading "_" to the path.

current_task_id_suffix

str | int

This task’s own segment of the id path — appended after the parent id. Either a positional index (int → coerced to str) for plain emissions, or a string id (e.g. a content-based hash from get_deterministic_id) for source-stage emissions where stability across input reordering matters.

nemo_curator.tasks.Task.add_stage_perf(
    perf_stats: nemo_curator.utils.performance_utils.StagePerfStats
) -> None

Add performance stats for a stage.

nemo_curator.tasks.Task.get_deterministic_id() -> str | None

Return a content-based identifier for this task as a source, or None to fall back to the positional index.

Override in subclasses that have stable content. The canonical example is FileGroupTask, which hashes its sorted file paths so that adding or removing files between runs doesn’t shift the identifiers of unchanged source partitions.

Only called by source-stage adapters; non-source stages ignore this and always use positional indices.

nemo_curator.tasks.Task.get_source_id() -> str

This task’s source-partition identity: the trailing segment of task_id (the id-path leaf). At a source stage that segment is the partition’s own id (content id or index); the resumability layer stamps it onto _source_id and inherits it downstream. Kept here next to _set_task_id so the "_" id-path encoding lives in one place.

nemo_curator.tasks.Task.validate() -> bool

abstract

Validate the task data.

nemo_curator.tasks.tasks.T = TypeVar('T')