nemo_curator.backends.base

Module Contents

Classes

Name	Description
`BaseExecutor`	Executor for a pipeline.
`BaseStageAdapter`	Adapts ProcessingStage to an execution backend, if needed.
`NodeInfo`	Generic node information for setup_on_node calls across backends.
`WorkerMetadata`	Generic worker metadata for setup_on_node calls across backends.

Functions

Name	Description
`_is_sentinel`	A payload-less marker (NoneTask/FailedTask), stripped before the next stage.

API

class nemo_curator.backends.base.BaseExecutor(
    config: dict[str, typing.Any] | None = None,
    ignore_head_node: bool = False
)

Abstract

Executor for a pipeline.

config

= config or {}

ignore_head_node

= ignore_head_node or ignore_ray_head_node()

nemo_curator.backends.base.BaseExecutor.execute(
    stages: list[nemo_curator.stages.base.ProcessingStage],
    initial_tasks: list[nemo_curator.tasks.Task] | None = None
) -> None

abstract

Execute the pipeline.

class nemo_curator.backends.base.BaseStageAdapter(
    stage: nemo_curator.stages.base.ProcessingStage
)

Adapts ProcessingStage to an execution backend, if needed.

nemo_curator.backends.base.BaseStageAdapter._apply_resumability_counters(
    input_tasks: list[nemo_curator.tasks.Task],
    output_tasks: list[nemo_curator.tasks.Task]
) -> list[nemo_curator.tasks.Task]

nemo_curator.backends.base.BaseStageAdapter._post_process_task_ids(
    input_tasks: list[nemo_curator.tasks.Task],
    output_tasks: list[nemo_curator.tasks.Task]
) -> list[nemo_curator.tasks.Task]

Assign a deterministic task_id (parent id + own segment) to every emitted task. Runs once per stage on every backend, so process vs process_batch makes no difference; ids are re-derived at each stage boundary, so one object passing through N stages gets N ids.

single input → fan-out: each output is parent_<seg>
len(output) == len(input) → positional 1:1: parent_i_<seg>; a NoneTask slot means input i was filtered (kept for alignment, then dropped from the result)
any other cardinality → a random "r"-prefixed uuid (non-deterministic, ancestry-not-tracked; see Task.task_id)

seg is the content id (get_deterministic_id()) for a source stage, else the positional index. A stage that both filters and fans out in one batch can’t be mapped positionally and falls to the "r" case — return one value (or None) per input to stay positional.

nemo_curator.backends.base.BaseStageAdapter._source_counters(
    output_tasks: list[nemo_curator.tasks.Task]
) -> list[nemo_curator.tasks.Task]

Source stage: each output is a source partition; its _source_id is Task.get_source_id(). Drop already-completed sources; each survivor fires +1.

nemo_curator.backends.base.BaseStageAdapter.process_batch(
    tasks: list[nemo_curator.tasks.Task]
) -> list[nemo_curator.tasks.Task]

Process a batch of tasks.

Parameters:

tasks

list[Task]

List of tasks to process

Returns: list[Task]

list[Task]: List of processed tasks

nemo_curator.backends.base.BaseStageAdapter.setup(
    worker_metadata: nemo_curator.backends.base.WorkerMetadata | None = None
) -> None

Setup the stage once per actor.

Parameters:

worker_metadata

WorkerMetadataDefaults to None

Information about the worker

nemo_curator.backends.base.BaseStageAdapter.setup_on_node(
    node_info: nemo_curator.backends.base.NodeInfo | None = None,
    worker_metadata: nemo_curator.backends.base.WorkerMetadata | None = None
) -> None

Setup the stage on a node.

Parameters:

node_info

NodeInfoDefaults to None

Information about the node

worker_metadata

WorkerMetadataDefaults to None

Information about the worker

nemo_curator.backends.base.BaseStageAdapter.teardown() -> None

Teardown the stage once per actor.

class nemo_curator.backends.base.NodeInfo(
    node_id: str = ''
)

Dataclass

Generic node information for setup_on_node calls across backends. Simplified to match Xenna’s structure.

node_id

str = ''

class nemo_curator.backends.base.WorkerMetadata(
    worker_id: str = '',
    allocation: typing.Any = None
)

Dataclass

Generic worker metadata for setup_on_node calls across backends. Simplified to match Xenna’s structure. The allocation field can contain backend-specific allocation information.

worker_id

str = ''

nemo_curator.backends.base._is_sentinel(
    task: nemo_curator.tasks.Task
) -> bool

A payload-less marker (NoneTask/FailedTask), stripped before the next stage.