Implementation:NVIDIA NeMo Curator Task Base

Knowledge Sources	NVIDIA NeMo Curator
Domains	Data Curation, Pipeline Framework, Pipeline Tasks
Last Updated	2026-02-14 00:00 GMT

Overview

The Task abstract base class defines the fundamental unit of work flowing through the NeMo Curator processing pipeline, providing a generic container for data with built-in performance tracking, validation, and metadata attachment.

Description

Task is a generic ABC parameterized by a data type T (via Generic[T]), implemented as a dataclass. It defines the contract that all task types in the pipeline must fulfill.

Core fields:

task_id (str): Unique identifier for this task instance.
dataset_name (str): Name of the dataset this task belongs to.
data (T): The actual payload data, typed generically.
_stage_perf (list[StagePerfStats]): Performance statistics collected as the task passes through pipeline stages. Defaults to an empty list.
_metadata (dict[str, Any]): Arbitrary metadata dictionary. Defaults to an empty dict.
_uuid (str): Auto-generated UUID (not included in __init__), created via uuid.uuid4() in the default factory.

Abstract interface:

validate() (abstract method): Subclasses must implement data validation logic. This method is automatically called in __post_init__, ensuring every task is validated upon creation.
num_items (abstract property): Subclasses must return the count of items in the task.

Performance tracking: The add_stage_perf(perf_stats) method appends a StagePerfStats instance to the _stage_perf list, enabling per-stage performance tracking as the task flows through the pipeline.

Sentinel task: The module also defines _EmptyTask, a concrete subclass with data=None and zero items, and a pre-instantiated module-level EmptyTask singleton used as a sentinel for list-type (ls) stages that do not process real data.

Usage

Task is never instantiated directly. Instead, it is subclassed by modality-specific task types such as DocumentBatch (text), ImageBatch (images), and AudioBatch (audio). Every piece of data moving through processing stages is wrapped in a Task subclass, enabling uniform performance tracking, validation, and metadata attachment across all modalities.

Code Reference

Source Location

Repository: NeMo-Curator
File: nemo_curator/tasks/tasks.py
Lines: 1-80

Signature

T = TypeVar("T")

@dataclass
class Task(ABC, Generic[T]):
    task_id: str
    dataset_name: str
    data: T
    _stage_perf: list[StagePerfStats] = field(default_factory=list)
    _metadata: dict[str, Any] = field(default_factory=dict)
    _uuid: str = field(init=False, default_factory=lambda: str(uuid.uuid4()))

    def __post_init__(self) -> None: ...

    @property
    @abstractmethod
    def num_items(self) -> int: ...

    def add_stage_perf(self, perf_stats: StagePerfStats) -> None: ...
    def __repr__(self) -> str: ...

    @abstractmethod
    def validate(self) -> bool: ...


@dataclass
class _EmptyTask(Task[None]):
    @property
    def num_items(self) -> int: ...
    def validate(self) -> bool: ...

EmptyTask = _EmptyTask(task_id="empty", dataset_name="empty", data=None)

Import

from nemo_curator.tasks.tasks import Task, EmptyTask
# or
from nemo_curator.tasks import Task, EmptyTask

I/O Contract

Task Fields

Name	Type	Required	Description
task_id	str	Yes	Unique identifier for this task instance
dataset_name	str	Yes	Name of the dataset this task belongs to
data	T (generic)	Yes	The payload data, typed by the subclass
_stage_perf	list[StagePerfStats]	No	Performance stats accumulated per stage (default: empty list)
_metadata	dict[str, Any]	No	Arbitrary metadata dictionary (default: empty dict)
_uuid	str	No	Auto-generated UUID, not settable via __init__

Abstract Methods

Name	Return Type	Description
num_items	int	Number of items in the task (abstract property)
validate()	bool	Validate the task data (abstract method, called on construction)

Usage Examples

Subclassing Task

from dataclasses import dataclass
from nemo_curator.tasks.tasks import Task

@dataclass
class MyCustomBatch(Task[list[dict]]):
    @property
    def num_items(self) -> int:
        return len(self.data)

    def validate(self) -> bool:
        return self.data is not None and len(self.data) > 0

# Usage
batch = MyCustomBatch(
    task_id="custom_001",
    dataset_name="my_dataset",
    data=[{"key": "value"}],
)
print(batch.num_items)  # 1
print(batch._uuid)      # Auto-generated UUID

Performance Tracking

from nemo_curator.utils.performance_utils import StagePerfStats

# Stages add performance stats as the task flows through the pipeline
perf = StagePerfStats(stage_name="my_stage")
batch.add_stage_perf(perf)
print(len(batch._stage_perf))  # 1

Using EmptyTask

from nemo_curator.tasks.tasks import EmptyTask

# EmptyTask is a pre-instantiated singleton for ls-type stages
print(EmptyTask.task_id)       # "empty"
print(EmptyTask.dataset_name)  # "empty"
print(EmptyTask.num_items)     # 0

Related Pages

Environment:NVIDIA_NeMo_Curator_Python_Linux_Base
NVIDIA_NeMo_Curator_DocumentBatch - Text document task subclass
NVIDIA_NeMo_Curator_AudioBatch - Audio data task subclass
NVIDIA_NeMo_Curator_ImageBatch - Image data task subclass
NVIDIA_NeMo_Curator_TaskPerfUtils - Utilities for aggregating performance metrics from tasks

Page Connections

Double-click a node to navigate. Hold to expand connections.

Principle

Implementation

Heuristic

Environment