> For clean Markdown of any page, append .md to the page URL.
> For a complete documentation index, see https://docs.nvidia.com/nemo/curator/llms.txt.
> For AI client integration (Claude Code, Cursor, etc.), connect to the MCP server at https://docs.nvidia.com/nemo/curator/_mcp/server.

# nemo_curator.pipeline.pipeline

## Module Contents

### Classes

| Name                                                   | Description                                                      |
| ------------------------------------------------------ | ---------------------------------------------------------------- |
| [`Pipeline`](#nemo_curator-pipeline-pipeline-Pipeline) | User-facing pipeline definition for composing processing stages. |

### Functions

| Name                                                                           | Description                                            |
| ------------------------------------------------------------------------------ | ------------------------------------------------------ |
| [`assign_root_task_ids`](#nemo_curator-pipeline-pipeline-assign_root_task_ids) | Assign root `task_id`s to user-provided initial tasks. |

### API

```python
class nemo_curator.pipeline.pipeline.Pipeline(
    name: str,
    description: str | None = None,
    stages: list[nemo_curator.stages.base.ProcessingStage] | None = None,
    config: dict[str, typing.Any] | None = None
)
```

User-facing pipeline definition for composing processing stages.

**`config`** `= config or {}`

---

**`stages`** `list[ProcessingStage] = stages or []`

---

```python
nemo_curator.pipeline.pipeline.Pipeline.__repr__() -> str
```

String representation of the pipeline.

```python
nemo_curator.pipeline.pipeline.Pipeline._assign_source_sink_roles() -> None
```

```python
nemo_curator.pipeline.pipeline.Pipeline._decompose_stages(
    stages: list[nemo_curator.stages.base.ProcessingStage | nemo_curator.stages.base.CompositeStage]
) -> tuple[list[nemo_curator.stages.base.ProcessingStage], dict[str, list[str]]]
```

Decompose composite stages into execution stages.

**Parameters:**

**`stages`** `list[ProcessingStage | CompositeStage]`

List of stages that may include composite stages

---

**Returns:** `tuple[list[ProcessingStage], dict[str, list[str]]]`

tuple\[list\[ProcessingStage], dict\[str, list\[str]]]: Tuple of (execution stages, decomposition info dict)

**Raises:**

* `TypeError`: If a composite stage is decomposed into another composite stage

```python
nemo_curator.pipeline.pipeline.Pipeline._run_with_resumability(
    executor: nemo_curator.backends.base.BaseExecutor,
    initial_tasks: list[nemo_curator.tasks.Task] | None,
    checkpoint_path: pathlib.Path
) -> list[nemo_curator.tasks.Task] | None
```

Run with resumability around a pre-existing Ray cluster (e.g. one
started by `RayClient`).

We briefly connect with `with ray.init()` to spawn the detached
checkpoint actor, then disconnect *before* `executor.execute` so the
executor's own `ray.init` runs un-nested — a nested
`ray.init(runtime_env=...)` is silently dropped, so the executor's env
vars wouldn't propagate otherwise. The detached actor lives in the
cluster across the executor's separate Ray session; a final
`with ray.init()` closes and kills it. The cluster must pre-exist: had
we started it, the first `with`-exit shutdown would tear it down and
take the actor with it.

```python
nemo_curator.pipeline.pipeline.Pipeline.add_stage(
    stage: nemo_curator.stages.base.ProcessingStage
) -> nemo_curator.pipeline.pipeline.Pipeline
```

Add a stage to the pipeline.

**Parameters:**

**`stage`** `ProcessingStage`

Processing stage to add

---

**Returns:** `Pipeline`

Self (Pipeline) for method chaining

```python
nemo_curator.pipeline.pipeline.Pipeline.build() -> None
```

Build an execution plan from the pipeline.

**Raises:**

* `ValueError`: If the pipeline has no stages

```python
nemo_curator.pipeline.pipeline.Pipeline.describe() -> str
```

Get a detailed description of the pipeline stages and their requirements.

```python
nemo_curator.pipeline.pipeline.Pipeline.run(
    executor: nemo_curator.backends.base.BaseExecutor | None = None,
    initial_tasks: list[nemo_curator.tasks.Task] | None = None,
    checkpoint_path: str | pathlib.Path | None = None
) -> list[nemo_curator.tasks.Task] | None
```

Run the pipeline.

**Parameters:**

**`executor`** `BaseExecutor` — default: None

Executor to use

---

**`initial_tasks`** `list[Task]` — default: None

Initial tasks to start the pipeline with. Defaults to None.

---

**`checkpoint_path`** `str | Path` — default: None

Resumability directory. Must
be a LOCAL filesystem path (the LMDB state is written locally),
not a remote/cloud URI. When set, completed source partitions are
tracked (in a `.nemo_curator_metadata` subdir) and skipped on
rerun. Multiple runs (e.g. a SLURM array) may share the directory
— each writes its own LMDB file, so there is no contention.

---

**Returns:** `list[Task] | None`

list\[Task] | None: List of tasks

```python
nemo_curator.pipeline.pipeline.assign_root_task_ids(
    initial_tasks: list[nemo_curator.tasks.Task]
) -> list[nemo_curator.tasks.Task]
```

Assign root `task_id`s to user-provided initial tasks.

Every task in a run descends from the implicit root `"0"` (the id of
:class:`EmptyTask`). User-provided initial tasks are its direct
children, so they get `"0_0"`, `"0_1"`, … `EmptyTask` instances
are skipped (already `"0"`). All downstream `task_id` assignment
happens in `BaseStageAdapter`.

NOTE: we deliberately use the positional index here, NOT
`get_deterministic_id()`, even for content-bearing tasks like
`FileGroupTask`. The source stage is the single place content-based
ids are assigned (to its outputs); hashing here too would put the
content hash at two levels of the id path (`"0_<hashA>_<hashB>"`).
Passing initial tasks directly is rare; if you need reorder-stable
source ids, let a source stage emit them.