> This page is for version 26.07 · v1.3.0.
> For other versions, use one of these documentation indexes:
> - Latest · v1.4.0 (26.09) (default): https://docs.nvidia.com/nemo/curator/latest/llms.txt
> - Main · preview: https://docs.nvidia.com/nemo/curator/main/llms.txt
> - 26.09 · v1.4.0: https://docs.nvidia.com/nemo/curator/v26.09/llms.txt
> - 26.07 · v1.3.0: https://docs.nvidia.com/nemo/curator/v26.07/llms.txt
> - 26.04 · v1.2.0: https://docs.nvidia.com/nemo/curator/v26.04/llms.txt
> - 26.02 · v1.1.0: https://docs.nvidia.com/nemo/curator/v26.02/llms.txt
> - 25.09 · v1.0.0: https://docs.nvidia.com/nemo/curator/v25.09/llms.txt

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

# AudioTask

> API reference for AudioTask - the task type for audio processing

`AudioTask` is the task type for audio processing in NeMo Curator. Each `AudioTask` holds a single manifest entry as a dict, matching the convention used by `VideoTask` and `FileGroupTask`.

## Import

```python
from nemo_curator.tasks import AudioTask
```

## Class Definition

```python
from dataclasses import dataclass

@dataclass
class AudioTask(Task[dict]):
    """Task containing a single audio manifest entry for processing.

    Attributes:
        task_id: Framework-managed deterministic lineage identifier.
        dataset_name: Name of the source dataset.
        data: Audio manifest entry (single dict, stored as _AttrDict).
    """

    dataset_name: str
    data: dict  # _AttrDict subclass — supports attribute-style access
    # task_id is inherited from Task with init=False.
```

## Audio Manifest Format

Audio data follows the NeMo manifest format:

```json
{
  "audio_filepath": "/path/to/audio.wav",
  "duration": 5.2,
  "text": "Transcription text...",
  "speaker": "speaker_001",
  "metadata": {
    "sample_rate": 16000,
    "channels": 1
  }
}
```

## Properties

### `num_items`

Always returns `1` — each `AudioTask` holds exactly one manifest entry.

```python
@property
def num_items(self) -> int:
    """Returns 1."""
```

## Creating AudioTask

```python
from nemo_curator.tasks import AudioTask

# Single manifest entry
task = AudioTask(
    dataset_name="speech_dataset",
    data={
        "audio_filepath": "/data/audio/sample.wav",
        "duration": 5.2,
        "text": "Hello world",
    },
)

# Access fields via attribute or dict style
task.data["audio_filepath"]   # "/data/audio/sample.wav"
task.data.audio_filepath      # "/data/audio/sample.wav"
```

Do not pass or derive `task_id`. Treat it as read-only even when working with a task subclass that exposes a compatibility default; the adapter overwrites it with framework lineage at the next stage boundary.

## Usage in Stages

All audio stages subclass `ProcessingStage[AudioTask, AudioTask]` directly — there is no intermediate base class.

### CPU Stage (per-task processing)

```python
from dataclasses import dataclass
from nemo_curator.stages.base import ProcessingStage
from nemo_curator.tasks import AudioTask

@dataclass
class DurationFilterStage(ProcessingStage[AudioTask, AudioTask]):
    """Filter audio by duration."""

    name: str = "DurationFilter"
    min_duration: float = 1.0
    max_duration: float = 30.0

    def inputs(self) -> tuple[list[str], list[str]]:
        return ["data"], []

    def outputs(self) -> tuple[list[str], list[str]]:
        return ["data"], []

    def process(self, task: AudioTask) -> AudioTask | None:
        duration = task.data.get("duration", 0)
        if self.min_duration <= duration <= self.max_duration:
            return task
        return None
```

### Batch Stage (GPU/IO processing)

```python
from dataclasses import dataclass, field
from nemo_curator.stages.resources import Resources

@dataclass
class MyGpuStage(ProcessingStage[AudioTask, AudioTask]):
    """GPU stage using process_batch."""

    name: str = "MyGpuStage"
    resources: Resources = field(
        default_factory=lambda: Resources(gpu_memory_gb=16.0)
    )

    def process(self, task: AudioTask) -> AudioTask:
        raise NotImplementedError("Use process_batch for GPU stages")

    def process_batch(self, tasks: list[AudioTask]) -> list[AudioTask]:
        # Batched GPU inference
        paths = [t.data["audio_filepath"] for t in tasks]
        results = self.model.infer(paths)
        for task, result in zip(tasks, results):
            task.data["pred_text"] = result
        return tasks
```

## Common Operations

### ASR Transcription

```python
def process(self, task: AudioTask) -> AudioTask:
    audio_path = task.data["audio_filepath"]
    task.data["pred_text"] = self.asr_model.transcribe(audio_path)
    return task
```

### Quality Scoring

```python
def process(self, task: AudioTask) -> AudioTask:
    if "text" in task.data and "pred_text" in task.data:
        task.data["wer"] = compute_wer(task.data["text"], task.data["pred_text"])
    return task
```

## Source Code

[View source on GitHub](https://github.com/NVIDIA-NeMo/Curator/blob/main/nemo_curator/tasks/audio_task.py)