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.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)

_source_id

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

_stage_perf

list[StagePerfStats] = field(default_factory=list)

data

dataset_name

str

num_items

int

Get the number of items in this task.

task_id

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

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

Post-initialization hook.

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

nemo_curator.tasks.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 :py:meth:get_deterministic_id) for source-stage emissions where stability across input reordering matters.

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

Add performance stats for a stage.

nemo_curator.tasks.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 :class: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.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 :py:meth:_set_task_id so the "_" id-path encoding lives in one place.

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

abstract

Validate the task data.

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