wehub-resource-sync
ba4be087d5
CICD NeMo / cicd-main-unit-tests (push) Blocked by required conditions
CICD NeMo / cicd-main-speech (push) Blocked by required conditions
CICD NeMo / cicd-test-container-build (push) Blocked by required conditions
CICD NeMo / cicd-import-tests (push) Blocked by required conditions
CICD NeMo / L0_Setup_Test_Data_And_Models (push) Blocked by required conditions
CICD NeMo / Nemo_CICD_Test (push) Blocked by required conditions
CICD NeMo / Coverage (e2e) (push) Blocked by required conditions
CICD NeMo / Coverage (unit-test) (push) Blocked by required conditions
CodeQL / Analyze (python) (push) Waiting to run
Create PR to main with cherry-pick from release / cherry-pick (push) Failing after 0s
CICD NeMo / pre-flight (push) Failing after 0s
CICD NeMo / configure (push) Has been skipped
Build, validate, and release Neural Modules / pre-flight (push) Failing after 1s
CICD NeMo / code-linting (push) Has been skipped
CICD NeMo / cicd-wait-in-queue (push) Waiting to run
Build, validate, and release Neural Modules / release (push) Has been skipped
Build, validate, and release Neural Modules / release-summary (push) Has been cancelled
1486 lines
67 KiB
Plaintext
1486 lines
67 KiB
Plaintext
{
|
|
"cells": [
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"id": "9caa4055",
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"\"\"\"\n",
|
|
"You can run either this notebook locally (if you have all the dependencies and a GPU) or on Google Colab.\n",
|
|
"\n",
|
|
"Instructions for setting up Colab are as follows:\n",
|
|
"1. Open a new Python 3 notebook.\n",
|
|
"2. Import this notebook from GitHub (File -> Upload Notebook -> \"GitHub\" tab -> copy/paste GitHub URL)\n",
|
|
"3. Connect to an instance with a GPU (Runtime -> Change runtime type -> select \"GPU\" for hardware accelerator)\n",
|
|
"4. Run this cell to set up dependencies.\n",
|
|
"5. Restart the runtime (Runtime -> Restart Runtime) for any upgraded packages to take effect\n",
|
|
"\n",
|
|
"\n",
|
|
"NOTE: User is responsible for checking the content of datasets and the applicable licenses and determining if they are suitable for the intended use.\n",
|
|
"\"\"\"\n",
|
|
"# If you're using Google Colab and not running locally, run this cell to install dependencies\n",
|
|
"\n",
|
|
"# Install dependencies\n",
|
|
"!pip install wget\n",
|
|
"!apt-get update && apt-get install -y sox libsndfile1 ffmpeg\n",
|
|
"!pip install text-unidecode\n",
|
|
"!pip install omegaconf\n",
|
|
"\n",
|
|
"BRANCH='main'\n",
|
|
"!python -m pip install \"nemo_toolkit[all] @ git+https://github.com/NVIDIA-NeMo/Speech.git@{BRANCH}\""
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"id": "i4mltwxzh1m",
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"# Optional dependencies:\n",
|
|
"# - nemo_text_processing: required for Inverse Text Normalization (ITN)\n",
|
|
"# - vllm: required for LLM-based Speech Translation\n",
|
|
"#\n",
|
|
"# For a complete Docker-based setup for speech translation with vLLM, see:\n",
|
|
"# https://github.com/NVIDIA/NeMo/blob/main/scripts/installers/Dockerfile.speech_translation_vllm\n",
|
|
"\n",
|
|
"!pip install nemo_text_processing\n",
|
|
"!pip install vllm==0.12.0"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"id": "9edd477c-6bdb-4191-a579-932925146384",
|
|
"metadata": {
|
|
"jp-MarkdownHeadingCollapsed": true
|
|
},
|
|
"source": [
|
|
"# Introduction\n",
|
|
"\n",
|
|
"This tutorial introduces the **ASR Inference Pipeline API** — a unified, streaming-first API for speech recognition that lives under `nemo.collections.asr.inference`. It provides a consistent interface for running inference with different ASR models (buffered CTC/RNNT/TDT, cache-aware CTC/RNNT) and optional post-processing features.\n",
|
|
"\n",
|
|
"## Table of Contents\n",
|
|
"\n",
|
|
"1. [Dataset Preparation](#Dataset-Preparation)\n",
|
|
"2. [Pipelines](#Pipelines)\n",
|
|
" - [Buffered CTC/RNNT/TDT Pipeline](#Buffered-CTC/RNNT/TDT-Pipeline)\n",
|
|
" - [Cache-Aware CTC/RNNT Pipeline](#Cache-Aware-CTC/RNNT-Pipeline)\n",
|
|
"3. [Advanced Features](#Advanced-Features)\n",
|
|
" - [Per-Stream Options](#Per-Stream-Options)\n",
|
|
" - [EoU Detection](#EoU-Detection)\n",
|
|
" - [Word Timestamps and Confidence Scores](#Word-Timestamps-and-Confidence-Scores)\n",
|
|
" - [Per-Stream Biasing](#Per-Stream-Biasing)\n",
|
|
" - [Inverse Text Normalization](#Inverse-Text-Normalization)\n",
|
|
" - [Speech Translation](#Speech-Translation)\n",
|
|
"4. [Implementation Details](#Implementation-Details)\n",
|
|
" - [Frame](#Frame)\n",
|
|
" - [Stream](#Stream)\n",
|
|
" - [Multiple Streams](#Multiple-Streams)\n",
|
|
" - [Continuous Batching](#Continuous-Batching)"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"id": "f7a4c877-9792-4cd3-9a7b-be0ed7fc2b16",
|
|
"metadata": {},
|
|
"source": [
|
|
"# Dataset Preparation\n",
|
|
"\n",
|
|
"We use **Mini LibriSpeech** (`dev_clean_2`) — a small subset of LibriSpeech that is also used in other NeMo ASR tutorials. The download script creates a NeMo manifest JSON file that lists every audio file path together with its transcript and duration."
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"id": "5dcd7de7-9b98-4013-a585-eb53619462a9",
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"%%bash\n",
|
|
"DATA_ROOT=\"datasets/mini-librispeech\"\n",
|
|
"MANIFEST=\"$DATA_ROOT/dev_clean_2.json\"\n",
|
|
"\n",
|
|
"if [ -f \"$MANIFEST\" ]; then\n",
|
|
" echo \"Dataset already exists at '$DATA_ROOT', skipping download.\"\n",
|
|
"else\n",
|
|
" echo \"Downloading Mini LibriSpeech dataset...\"\n",
|
|
" mkdir -p \"$DATA_ROOT\"\n",
|
|
" python ../../scripts/dataset_processing/get_librispeech_data.py \\\n",
|
|
" --data_root \"$DATA_ROOT/\" \\\n",
|
|
" --data_sets dev_clean_2\n",
|
|
" echo \"Download complete.\"\n",
|
|
"fi"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"id": "af3dae25-6236-438c-bc9b-20281e0a2ef9",
|
|
"metadata": {
|
|
"scrolled": true
|
|
},
|
|
"outputs": [],
|
|
"source": [
|
|
"import json\n",
|
|
"import os\n",
|
|
"\n",
|
|
"MANIFEST_PATH = \"datasets/mini-librispeech/dev_clean_2.json\"\n",
|
|
"LIMIT = 5\n",
|
|
"\n",
|
|
"audio_filepaths = []\n",
|
|
"reference_texts = []\n",
|
|
"with open(MANIFEST_PATH) as f:\n",
|
|
" for line in f:\n",
|
|
" item = json.loads(line)\n",
|
|
" audio_filepaths.append(item[\"audio_filepath\"])\n",
|
|
" reference_texts.append(item[\"text\"])\n",
|
|
" if len(audio_filepaths) == LIMIT:\n",
|
|
" break\n",
|
|
"\n",
|
|
"print(f\"Loaded {len(audio_filepaths)} audio files\")\n",
|
|
"for path, ref in zip(audio_filepaths, reference_texts):\n",
|
|
" print(f\" {path.split('/')[-1]} → {ref}\")"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"id": "3b2544c0-ac45-4921-9582-6f1305abca4a",
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"! wget -nc https://dldata-public.s3.us-east-2.amazonaws.com/2086-149220-0033.wav\n",
|
|
"! wget -nc https://nemo-public.s3.us-east-2.amazonaws.com/an4_diarize_test.wav"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"id": "a4fc476a-91dd-4710-91ea-9da125ec39d3",
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"# Demo files\n",
|
|
"demo_audio_filepath = \"2086-149220-0033.wav\"\n",
|
|
"itn_demo_audio_filepath = \"an4_diarize_test.wav\"\n",
|
|
"biasing_demo_audio_file = audio_filepaths[2]\n",
|
|
"\n",
|
|
"missing = [f for f in [demo_audio_filepath, itn_demo_audio_filepath, biasing_demo_audio_file] if not os.path.exists(f)]\n",
|
|
"if missing:\n",
|
|
" raise FileNotFoundError(f\"Demo files not found: {missing} — re-run the download cell above.\")\n",
|
|
"print(\"All demo files exist.\")"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"id": "751e745f-5c94-4a80-ab1c-9e92aa465848",
|
|
"metadata": {},
|
|
"source": [
|
|
"# Pipelines\n",
|
|
"\n",
|
|
"#### How it works?\n",
|
|
"A `Pipeline` is the central inference engine. It wraps the ASR model, feature extraction, buffer/state/cache management, and optional text post-processing (ITN, translation) into a single object. The primary method is `transcribe_step()`, which accepts a batch of `Frame` objects and immediately returns one `TranscribeStepOutput` per frame.\n",
|
|
"\n",
|
|
"`PipelineBuilder.build_pipeline(cfg)` is a factory that reads an OmegaConf config and instantiates the appropriate pipeline variant.\n",
|
|
"\n",
|
|
"#### Key parameters\n",
|
|
"`TranscribeStepOutput` fields returned by `transcribe_step()`:\n",
|
|
"- `stream_id`: integer identifying which stream this output belongs to\n",
|
|
"- `final_transcript`: text finalized at the last EoU boundary; empty on most steps, populated only when EoU is detected\n",
|
|
"- `partial_transcript`: accumulated text since the previous EoU boundary; updated every step and may change as future frames arrive\n",
|
|
"- `current_step_transcript`: text decoded from the current frame only\n",
|
|
"- `final_segments`: list of `TextSegment` objects (word or segment metadata) corresponding to `final_transcript`\n",
|
|
"\n",
|
|
"#### Warnings and Notes\n",
|
|
"- `final_transcript` is only non-empty at EoU boundaries. On all other steps it is an empty string — use `partial_transcript` to track in-progress output.\n",
|
|
"- `partial_transcript` is not stable and will be revised as more audio arrives. Only `final_transcript` should be treated as authoritative output.\n",
|
|
"- A pipeline session must be opened with `open_session()` before calling `transcribe_step()` and closed with `close_session()` afterward. The convenience method `pipeline.run()` handles this automatically."
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"id": "3faf0168-d1fd-4d2e-8b24-1e4cc0c9fa58",
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"# This cell contains helper imports and utilities\n",
|
|
"\n",
|
|
"from IPython.display import display, HTML, Audio\n",
|
|
"from collections import defaultdict\n",
|
|
"from time import sleep\n",
|
|
"\n",
|
|
"import re\n",
|
|
"import gc\n",
|
|
"import os\n",
|
|
"import torch\n",
|
|
"import librosa\n",
|
|
"import numpy as np\n",
|
|
"import matplotlib.pyplot as plt\n",
|
|
"\n",
|
|
"# Allow loading trusted checkpoints that hold non-tensor objects (torch>=2.6 weights_only default).\n",
|
|
"os.environ[\"TORCH_FORCE_NO_WEIGHTS_ONLY_LOAD\"] = \"1\"\n",
|
|
"\n",
|
|
"from omegaconf import OmegaConf\n",
|
|
"from nemo.collections.asr.inference.factory.pipeline_builder import PipelineBuilder, BasePipeline\n",
|
|
"from nemo.collections.asr.inference.streaming.framing.request_options import ASRRequestOptions\n",
|
|
"from nemo.collections.asr.inference.utils.enums import ASROutputGranularity\n",
|
|
"from nemo.collections.asr.inference.utils.text_segment import TextSegment\n",
|
|
"from nemo.collections.asr.parts.context_biasing.biasing_multi_model import BiasingRequestItemConfig\n",
|
|
"from nemo.collections.asr.parts.context_biasing.boosting_graph_batched import BoostingTreeModelConfig\n",
|
|
"from nemo.collections.asr.parts.utils.eval_utils import remove_punctuations\n",
|
|
"from nemo.collections.asr.metrics.wer import word_error_rate\n",
|
|
"\n",
|
|
"\n",
|
|
"_PALETTE = [\"#4a9eff\", \"#ff7043\", \"#26a69a\", \"#ab47bc\", \"#66bb6a\", \"#ffa726\", \"#ef5350\", \"#42a5f5\"]\n",
|
|
"\n",
|
|
"\n",
|
|
"def create_pipeline(config_path: str, cfg_overrides: dict = None):\n",
|
|
" cfg = OmegaConf.load(config_path)\n",
|
|
" if cfg_overrides:\n",
|
|
" _MISSING = object()\n",
|
|
" for key, value in cfg_overrides.items():\n",
|
|
" if OmegaConf.select(cfg, key, default=_MISSING) is _MISSING:\n",
|
|
" raise KeyError(f\"cfg_overrides key '{key}' does not exist in config '{config_path}'\")\n",
|
|
" OmegaConf.update(cfg, key, value)\n",
|
|
" return PipelineBuilder.build_pipeline(cfg)\n",
|
|
"\n",
|
|
"\n",
|
|
"def compute_wer(hypotheses: list[str], references: list[str]):\n",
|
|
" \"\"\"WER with punctuation and capitalisation ignored (mirrors pipeline_eval defaults).\"\"\"\n",
|
|
" hyps = [remove_punctuations(h).lower() for h in hypotheses]\n",
|
|
" refs = [remove_punctuations(r).lower() for r in references]\n",
|
|
" return word_error_rate(hypotheses=hyps, references=refs)\n",
|
|
"\n",
|
|
"\n",
|
|
"def do_streaming(\n",
|
|
" pipeline: BasePipeline,\n",
|
|
" audio_filepaths: list[str],\n",
|
|
" reference_texts: list[str] = None,\n",
|
|
" options: list[ASRRequestOptions] = None,\n",
|
|
" wait: float = 0.0,\n",
|
|
"):\n",
|
|
" sep = pipeline.get_sep()\n",
|
|
" request_generator = pipeline.get_request_generator()\n",
|
|
"\n",
|
|
" if options is None:\n",
|
|
" options = [ASRRequestOptions() for _ in audio_filepaths]\n",
|
|
" request_generator.set_audio_filepaths(audio_filepaths, options)\n",
|
|
"\n",
|
|
" has_translation = any(opt.enable_nmt for opt in options)\n",
|
|
" COLORS = [_PALETTE[i % len(_PALETTE)] for i in range(len(audio_filepaths))]\n",
|
|
"\n",
|
|
" accumulated = defaultdict(str)\n",
|
|
" cur_partial = defaultdict(str)\n",
|
|
" accumulated_translation = defaultdict(str)\n",
|
|
" cur_partial_translation = defaultdict(str)\n",
|
|
" pipeline_name = type(pipeline).__name__\n",
|
|
"\n",
|
|
" def render_html():\n",
|
|
" parts = []\n",
|
|
" for i in range(len(audio_filepaths)):\n",
|
|
" filename = audio_filepaths[i].split(\"/\")[-1]\n",
|
|
" color = COLORS[i]\n",
|
|
" final = accumulated[i]\n",
|
|
" partial = cur_partial[i]\n",
|
|
"\n",
|
|
" meta = f'stream {i} · {pipeline_name} · {filename}'\n",
|
|
" if has_translation and options[i].target_language:\n",
|
|
" meta += f' · {options[i].source_language} → {options[i].target_language}'\n",
|
|
"\n",
|
|
" translation_row = \"\"\n",
|
|
" if has_translation:\n",
|
|
" final_tr = accumulated_translation[i]\n",
|
|
" partial_tr = f'<span style=\"color:#aaa;\">{cur_partial_translation[i]}</span>' if cur_partial_translation[i] else \"\"\n",
|
|
" translation_row = (\n",
|
|
" f'<div style=\"margin-top:6px;padding-top:6px;border-top:1px solid #e0e0e0;'\n",
|
|
" f'color:#555;font-style:italic;\">{final_tr}{partial_tr}</div>'\n",
|
|
" )\n",
|
|
"\n",
|
|
" parts.append(\n",
|
|
" f'<div style=\"margin:4px 0;padding:10px 14px;border-left:4px solid {color};'\n",
|
|
" f'background:#fafafa;border-radius:0 4px 4px 0;font-family:serif;\">'\n",
|
|
" f'<div style=\"font-size:0.75em;color:#888;margin-bottom:4px;font-family:monospace;\">'\n",
|
|
" f'{meta}</div>'\n",
|
|
" f'<div>{final}{partial}</div>'\n",
|
|
" f'{translation_row}'\n",
|
|
" f'</div>'\n",
|
|
" )\n",
|
|
" return \"\".join(parts)\n",
|
|
"\n",
|
|
" handle = display(HTML(render_html()), display_id=True)\n",
|
|
"\n",
|
|
" pipeline.open_session()\n",
|
|
" for requests in request_generator:\n",
|
|
" step_outputs = pipeline.transcribe_step(requests)\n",
|
|
" for out in step_outputs:\n",
|
|
" sid = out.stream_id\n",
|
|
" if out.final_transcript:\n",
|
|
" text = out.final_transcript if accumulated[sid] else out.final_transcript.lstrip(sep)\n",
|
|
" accumulated[sid] += text\n",
|
|
" cur_partial[sid] = out.partial_transcript\n",
|
|
"\n",
|
|
" if has_translation:\n",
|
|
" if out.final_translation:\n",
|
|
" accumulated_translation[sid] += out.final_translation\n",
|
|
" cur_partial_translation[sid] = out.partial_translation\n",
|
|
"\n",
|
|
" handle.update(HTML(render_html()))\n",
|
|
" if wait > 0:\n",
|
|
" sleep(wait)\n",
|
|
" pipeline.close_session()\n",
|
|
"\n",
|
|
" handle.update(HTML(render_html()))\n",
|
|
"\n",
|
|
" if reference_texts is not None:\n",
|
|
" hypotheses = [accumulated[i] for i in range(len(audio_filepaths))]\n",
|
|
" wer = compute_wer(hypotheses, reference_texts)\n",
|
|
" print(f\"\\nWER for {pipeline_name} is: {wer:.2%}\")\n",
|
|
"\n",
|
|
"\n",
|
|
"def log_text_segments(segments: list[TextSegment], title: str = None):\n",
|
|
" has_class = any(getattr(seg, \"semiotic_class\", None) is not None for seg in segments)\n",
|
|
"\n",
|
|
" header_cells = [\"#\", \"Start\", \"End\", \"Dur\", \"Conf\"]\n",
|
|
" if has_class:\n",
|
|
" header_cells.append(\"Class\")\n",
|
|
" header_cells.append(\"Text\")\n",
|
|
"\n",
|
|
" th = \"\".join(\n",
|
|
" f'<th style=\"padding:4px 10px;text-align:{\"left\" if h==\"Text\" else \"right\"};'\n",
|
|
" f'color:#888;font-weight:normal;border-bottom:1px solid #ddd;\">{h}</th>'\n",
|
|
" for h in header_cells\n",
|
|
" )\n",
|
|
"\n",
|
|
" rows = []\n",
|
|
" for i, seg in enumerate(segments):\n",
|
|
" bg = \"#fafafa\" if i % 2 == 0 else \"#fff\"\n",
|
|
" cells = [\n",
|
|
" f'<td style=\"padding:3px 10px;text-align:right;color:#aaa;\">{i}</td>',\n",
|
|
" f'<td style=\"padding:3px 10px;text-align:right;\">{seg.start:.2f}</td>',\n",
|
|
" f'<td style=\"padding:3px 10px;text-align:right;\">{seg.end:.2f}</td>',\n",
|
|
" f'<td style=\"padding:3px 10px;text-align:right;\">{seg.duration:.2f}</td>',\n",
|
|
" f'<td style=\"padding:3px 10px;text-align:right;\">{seg.conf:.3f}</td>',\n",
|
|
" ]\n",
|
|
" if has_class:\n",
|
|
" cls = getattr(seg, \"semiotic_class\", None) or \"\"\n",
|
|
" cells.append(f'<td style=\"padding:3px 10px;text-align:right;color:#888;\">{cls}</td>')\n",
|
|
" cells.append(f'<td style=\"padding:3px 10px;\">{seg.text}</td>')\n",
|
|
" rows.append(\n",
|
|
" f'<tr style=\"background:{bg};\">{\"\".join(cells)}</tr>'\n",
|
|
" )\n",
|
|
"\n",
|
|
" count_label = f'{len(segments)} {\"word\" if hasattr(segments[0], \"word\") else \"segment\"}s' if segments else \"0 segments\"\n",
|
|
" title_html = f'<div style=\"font-weight:bold;margin-bottom:2px;\">{title}</div>' if title else \"\"\n",
|
|
" html = (\n",
|
|
" f'<div style=\"font-family:monospace;font-size:0.88em;margin:6px 0;\">'\n",
|
|
" f'{title_html}'\n",
|
|
" f'<div style=\"color:#888;margin-bottom:4px;\">{count_label}</div>'\n",
|
|
" f'<table style=\"border-collapse:collapse;width:auto;\">'\n",
|
|
" f'<thead><tr>{th}</tr></thead>'\n",
|
|
" f'<tbody>{\"\".join(rows)}</tbody>'\n",
|
|
" f'</table></div>'\n",
|
|
" )\n",
|
|
" display(HTML(html))\n",
|
|
"\n",
|
|
"\n",
|
|
"def visualize_word_timestamps(audio_filepath, word_segments):\n",
|
|
" samples, sr = librosa.load(audio_filepath, sr=None)\n",
|
|
" duration = len(samples) / sr\n",
|
|
" t = np.linspace(0, duration, len(samples))\n",
|
|
"\n",
|
|
" fig, ax = plt.subplots(figsize=(16, 4))\n",
|
|
" ax.plot(t, samples, color=\"#4a9eff\", linewidth=0.4, alpha=0.7)\n",
|
|
"\n",
|
|
" colors = plt.cm.tab20.colors\n",
|
|
" for i, word in enumerate(word_segments):\n",
|
|
" color = colors[i % len(colors)]\n",
|
|
" ax.axvspan(word.start, word.end, alpha=0.25, color=color)\n",
|
|
" mid = (word.start + word.end) / 2\n",
|
|
" y_top = ax.get_ylim()[1]\n",
|
|
" ax.text(mid, y_top * 0.88, word.text,\n",
|
|
" ha=\"center\", va=\"top\", fontsize=7.5,\n",
|
|
" color=color, fontweight=\"bold\", rotation=45)\n",
|
|
"\n",
|
|
" ax.set_xlabel(\"Time (s)\")\n",
|
|
" ax.set_ylabel(\"Amplitude\")\n",
|
|
" ax.set_title(\"Audio waveform with word-level timestamps\")\n",
|
|
" ax.margins(x=0)\n",
|
|
" plt.tight_layout()\n",
|
|
" plt.show()\n",
|
|
"\n",
|
|
" display(Audio(audio_filepath))"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"id": "fc7c18a2-ac21-45a6-9e1f-7dc5243486c7",
|
|
"metadata": {},
|
|
"source": [
|
|
"## Buffered CTC/RNNT/TDT Pipeline"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"id": "w3g13fasvtc",
|
|
"metadata": {},
|
|
"source": [
|
|
"<img src=\"./images/buffered_pipeline.png\" alt=\"Buffered Streaming Pipeline\" style=\"max-width:100%;margin:12px 0;\" />\n",
|
|
"\n",
|
|
"#### How it works?\n",
|
|
"The buffered pipeline adapts any standard offline ASR model (CTC, RNNT, TDT) for streaming without modifying the model architecture. At each step, the pipeline assembles a three-part padded buffer and runs a full forward pass through the model:\n",
|
|
"\n",
|
|
"- **Left context** — audio from previous steps, providing stable past context so the model \"remembers\" what came before.\n",
|
|
"- **Middle chunk** — the newly arrived audio; the primary segment whose tokens will be emitted.\n",
|
|
"- **Right context** (lookahead) — future audio already received, letting the model resolve ambiguities at the right edge of the middle chunk.\n",
|
|
"\n",
|
|
"After each forward pass, the pipeline scans the output token sequence looking for **N consecutive blank tokens (EoU)**. When found, all tokens from the start of the middle chunk up to that point are emitted as output.\n",
|
|
"\n",
|
|
"#### Key parameters\n",
|
|
"- `left_padding_size`: seconds of past audio prepended as left context\n",
|
|
"- `chunk_size`: duration of the middle chunk in seconds; also equal to the frame size\n",
|
|
"- `right_padding_size`: seconds of lookahead audio appended as right context\n",
|
|
"- `batch_size`: number of streams processed in parallel\n",
|
|
"- `stateful` (RNNT only): if `True`, the RNNT decoder state is carried over between steps for better transcript continuity; if `False`, the decoder resets at each step\n",
|
|
"\n",
|
|
"#### Warnings and Notes\n",
|
|
"- Maximum theoretical latency = `chunk_size` + `right_padding_size`. In the worst case, the pipeline must wait for the full chunk and then buffer the lookahead before emitting tokens.\n",
|
|
"- Each audio frame is re-encoded together with its left and right context — the same audio samples are processed multiple times; this is the key difference from cache-aware streaming, which processes each frame only once."
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"id": "727c71cc-c4d3-43f5-97a6-96f240a2146d",
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"batch_size = 8\n",
|
|
"log_level = 40\n",
|
|
"\n",
|
|
"# Amount of past audio prepended as left context so the model can condition on what came before the current chunk\n",
|
|
"left_padding_size = 1.6\n",
|
|
"\n",
|
|
"# Duration of the middle chunk. Also, equal to frame size\n",
|
|
"chunk_size = 0.54\n",
|
|
"\n",
|
|
"# Amount of future (lookahead) audio appended as right context, helping resolve ambiguities at the right edge of the chunk.\n",
|
|
"right_padding_size = 1.6\n",
|
|
"\n",
|
|
"# (RNNT only) - When True, the RNNT decoder state is carried over from the previous step, improving transcript continuity across chunks;\n",
|
|
"# when False, the decoder state is reset at each step (stateless decoding).\n",
|
|
"stateful = True\n",
|
|
"\n",
|
|
"buffered_rnnt_pipeline = create_pipeline(\n",
|
|
" \"../../examples/asr/conf/asr_streaming_inference/buffered_rnnt.yaml\",\n",
|
|
" cfg_overrides={\n",
|
|
" \"asr.model_name\": \"nvidia/parakeet-rnnt-1.1b\",\n",
|
|
" \"streaming.left_padding_size\": left_padding_size,\n",
|
|
" \"streaming.chunk_size\": chunk_size,\n",
|
|
" \"streaming.right_padding_size\": right_padding_size,\n",
|
|
" \"streaming.batch_size\": batch_size,\n",
|
|
" \"streaming.stateful\": stateful,\n",
|
|
" \"asr_decoding_type\": \"rnnt\",\n",
|
|
" \"log_level\": log_level,\n",
|
|
" },\n",
|
|
")\n",
|
|
"\n",
|
|
"buffered_ctc_pipeline = create_pipeline(\n",
|
|
" \"../../examples/asr/conf/asr_streaming_inference/buffered_ctc.yaml\",\n",
|
|
" cfg_overrides={\n",
|
|
" \"asr.model_name\": \"nvidia/parakeet-ctc-1.1b\",\n",
|
|
" \"streaming.left_padding_size\": left_padding_size,\n",
|
|
" \"streaming.chunk_size\": chunk_size,\n",
|
|
" \"streaming.right_padding_size\": right_padding_size,\n",
|
|
" \"streaming.batch_size\": batch_size,\n",
|
|
" \"asr_decoding_type\": \"ctc\",\n",
|
|
" \"log_level\": log_level,\n",
|
|
" },\n",
|
|
")"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"id": "2a2311f0-6a64-481d-a5df-b14b87cb9fa8",
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"for pipeline in [buffered_rnnt_pipeline, buffered_ctc_pipeline]:\n",
|
|
" do_streaming(pipeline, audio_filepaths, reference_texts)"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"id": "bdb3d0d8-6a95-41ca-b68d-46fe8311aea4",
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"del buffered_rnnt_pipeline, buffered_ctc_pipeline\n",
|
|
"gc.collect()\n",
|
|
"torch.cuda.empty_cache()"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"id": "d04466cf-a392-4525-8e84-1f610ee31da1",
|
|
"metadata": {},
|
|
"source": [
|
|
"## Cache-Aware CTC/RNNT Pipeline"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"id": "p09ox0y4dyb",
|
|
"metadata": {},
|
|
"source": [
|
|
"<img src=\"./images/cache_aware_pipeline.png\" alt=\"Cache-Aware Streaming Pipeline\" style=\"max-width:100%;margin:12px 0;\"/>\n",
|
|
"\n",
|
|
"#### How it works?\n",
|
|
"Unlike buffered inference — which slides an overlapping window and re-encodes the same audio frames repeatedly — cache-aware streaming processes **each audio frame once**. The FastConformer encoder maintains a **bounded cache of intermediate representations** across all layers (both self-attention KV states and convolutional states). When a new chunk arrives, only that chunk is encoded; past context is read from the cache rather than recomputed. This eliminates redundant computation and enables stable, predictable memory usage regardless of utterance length.\n",
|
|
"\n",
|
|
"The chunk size and latency are determined by the attention context configuration `[left_context, right_context]`, where chunk size = `right_context + 1` frames (for FastConformer each frame = 80 ms):\n",
|
|
"- `[70, 0]` → chunk size = 1 (0.08 s)\n",
|
|
"- `[70, 1]` → chunk size = 2 (0.16 s)\n",
|
|
"- `[70, 6]` → chunk size = 7 (0.56 s)\n",
|
|
"- `[70, 13]` → chunk size = 14 (1.12 s)\n",
|
|
"\n",
|
|
"#### Key parameters\n",
|
|
"- `batch_size`: number of concurrent streams processed in each forward pass\n",
|
|
"- `num_slots`: total number of pre-allocated cache slots; must be greater than or equal to `batch_size`\n",
|
|
"- `att_context_size`: list `[left_context, right_context]` controlling attention span and chunk size\n",
|
|
"\n",
|
|
"#### Warnings and Notes\n",
|
|
"- `num_slots` pre-allocates GPU memory for all cache slots at startup. Set it to the maximum number of concurrent streams expected, not just the current `batch_size`.\n",
|
|
"- Cache-aware streaming requires a FastConformer model trained with streaming context. Standard offline models are not compatible.\n",
|
|
"- EoU detection may cause punctuation to be dropped or misplaced at segment boundaries for models trained to emit punctuation after silence."
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"id": "uuk1yvul2b9",
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"batch_size = 8\n",
|
|
"log_level = 40\n",
|
|
"\n",
|
|
"# Maximum number of concurrent stream slots in the cache manager. Must be ≥ batch_size. Controls how many slots should be pre-allocated.\n",
|
|
"num_slots = 64\n",
|
|
"\n",
|
|
"# Number of frames the self-attention layers can attend to on the left (past) and right (lookahead) sides. \n",
|
|
"# Smaller right context lowers latency.\n",
|
|
"att_context_size = [70, 13]\n",
|
|
"\n",
|
|
"ca_rnnt_pipeline = create_pipeline(\n",
|
|
" \"../../examples/asr/conf/asr_streaming_inference/cache_aware_rnnt.yaml\",\n",
|
|
" cfg_overrides={\n",
|
|
" \"asr.model_name\": \"nvidia/nemotron-speech-streaming-en-0.6b\",\n",
|
|
" \"streaming.batch_size\": batch_size,\n",
|
|
" \"streaming.num_slots\": num_slots,\n",
|
|
" \"streaming.att_context_size\": att_context_size,\n",
|
|
" \"log_level\": log_level,\n",
|
|
" \"asr_decoding_type\": \"rnnt\"\n",
|
|
" },\n",
|
|
")\n",
|
|
"\n",
|
|
"ca_ctc_pipeline = create_pipeline(\n",
|
|
" \"../../examples/asr/conf/asr_streaming_inference/cache_aware_ctc.yaml\",\n",
|
|
" cfg_overrides={\n",
|
|
" \"asr.model_name\": \"stt_en_fastconformer_hybrid_large_streaming_multi\",\n",
|
|
" \"streaming.batch_size\": batch_size,\n",
|
|
" \"streaming.num_slots\": num_slots,\n",
|
|
" \"streaming.att_context_size\": att_context_size,\n",
|
|
" \"asr_decoding_type\": \"ctc\",\n",
|
|
" \"log_level\": log_level,\n",
|
|
" },\n",
|
|
")"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"id": "ybwudhtx1l",
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"for pipeline in [ca_rnnt_pipeline, ca_ctc_pipeline]:\n",
|
|
" do_streaming(pipeline, audio_filepaths, reference_texts)"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"id": "8bb3ad96-f829-4176-bef1-0b2718791c44",
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"del ca_rnnt_pipeline, ca_ctc_pipeline\n",
|
|
"gc.collect()\n",
|
|
"torch.cuda.empty_cache()"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"id": "uswct47ipp",
|
|
"metadata": {},
|
|
"source": [
|
|
"# Advanced Features\n",
|
|
"\n",
|
|
"The following sections explore the per-request options and post-processing features available in the Pipeline API. Most examples use a single audio file for clarity, but all features work in batch mode as well."
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"id": "kofdkteqafq",
|
|
"metadata": {},
|
|
"source": [
|
|
"## Per-Stream Options\n",
|
|
"\n",
|
|
"#### How it works?\n",
|
|
"`ASRRequestOptions` is an immutable dataclass that lets you configure inference behaviour **per audio stream**. Every field defaults to `None`, meaning \"use the pipeline default\". Passing an explicit value overrides the pipeline default for that request only — so different streams in the same batch can have entirely different settings.\n",
|
|
"\n",
|
|
"#### Key parameters\n",
|
|
"- `enable_itn`: apply Inverse Text Normalization (e.g. `\"twenty dollars\"` → `\"$20\"`)\n",
|
|
"- `stop_history_eou`: silence window in milliseconds that triggers an EoU boundary; set to `-1` to disable EoU detection\n",
|
|
"- `asr_output_granularity`: `ASROutputGranularity.SEGMENT` (default) or `ASROutputGranularity.WORD` — controls the granularity of `final_segments` timestamps\n",
|
|
"- `language_code`: language hint for prompt-enabled multilingual models\n",
|
|
"- `enable_nmt`: enable Neural Machine Translation post-processing\n",
|
|
"- `source_language`: NMT source language, effective only when `enable_nmt=True`\n",
|
|
"- `target_language`: NMT target language, effective only when `enable_nmt=True`\n",
|
|
"- `biasing_cfg`: per-stream context biasing configuration — key phrases and their boosting weight\n",
|
|
"\n",
|
|
"#### Warnings and Notes\n",
|
|
"- `enable_itn` and `enable_nmt` are per-request toggles, but the corresponding modules (ITN grammar, NMT model) must be loaded at **pipeline construction time** via the config. Toggling them per-request without loading at startup has no effect.\n",
|
|
"- `ASRRequestOptions` is immutable — create a new instance for each request rather than modifying an existing one."
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"id": "hci7ftoxeio",
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"# All fields are None → pipeline defaults apply for every option\n",
|
|
"default_opts = ASRRequestOptions()\n",
|
|
"print(\"Default options:\", default_opts)\n",
|
|
"\n",
|
|
"# Override individual fields per request\n",
|
|
"custom_opts = ASRRequestOptions(\n",
|
|
" enable_itn=True, # Inverse Text Normalization\n",
|
|
" stop_history_eou=400, # EoU silence window (ms)\n",
|
|
" asr_output_granularity=ASROutputGranularity.WORD, # word-level timestamps\n",
|
|
")\n",
|
|
"print(\"Custom options: \", custom_opts)"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"id": "ziumnhh0kqq",
|
|
"metadata": {},
|
|
"source": [
|
|
"## EoU Detection\n",
|
|
"\n",
|
|
"#### How it works?\n",
|
|
"EoU (End-of-Utterance) detection enables the pipeline to automatically segment a continuous audio stream into discrete, finalized utterances — without requiring explicit stream boundaries from the caller. After each forward pass the pipeline inspects the token sequence in the buffer. When a long-enough run of **silent (blank) tokens** is observed at the trailing edge, the pipeline concludes that the speaker has paused and marks an EoU boundary. At that moment:\n",
|
|
"- `final_transcript` is populated with all text accumulated since the previous EoU boundary.\n",
|
|
"- `partial_transcript` is reset to an empty string.\n",
|
|
"\n",
|
|
"#### Key parameters\n",
|
|
"- `stop_history_eou` (in `ASRRequestOptions`): silence window in milliseconds; a pause of at least this duration triggers an EoU boundary. Set to `-1` to disable EoU detection entirely.\n",
|
|
"\n",
|
|
"#### Warnings and Notes\n",
|
|
"- Different pipeline types implement different EoU detection logic — the exact blank-counting mechanism may vary between buffered and cache-aware pipelines.\n",
|
|
"- `final_transcript` is an empty string on all steps where no EoU boundary is detected. Always check `partial_transcript` for the current in-progress transcription.\n",
|
|
"- Setting `stop_history_eou` too low may cause premature EoU boundaries mid-sentence; setting it too high increases latency before a segment is finalized."
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"id": "vxnzuopyoib",
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"generic_pipeline = create_pipeline(\n",
|
|
" \"../../examples/asr/conf/asr_streaming_inference/buffered_rnnt.yaml\",\n",
|
|
" cfg_overrides={\n",
|
|
" \"asr.model_name\": \"nvidia/parakeet-rnnt-1.1b\",\n",
|
|
" \"streaming.left_padding_size\": 1.6,\n",
|
|
" \"streaming.chunk_size\": 0.54,\n",
|
|
" \"streaming.right_padding_size\": 1.6,\n",
|
|
" \"streaming.batch_size\": 8,\n",
|
|
" \"streaming.stateful\": True,\n",
|
|
" \"asr_decoding_type\": \"rnnt\",\n",
|
|
" \"log_level\": 40,\n",
|
|
" \"asr.decoding.greedy.preserve_frame_confidence\": True, # enable per-token confidence scores\n",
|
|
" },\n",
|
|
")"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"id": "le69wzcbur",
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"def transcribe_single(pipeline, audio_filepath, options=None, verbose=False):\n",
|
|
" \"\"\"\n",
|
|
" Stream a single audio file through *pipeline* and print each EoU event\n",
|
|
" together with the partial transcript updates along the way.\n",
|
|
" \"\"\"\n",
|
|
" request_generator = pipeline.get_request_generator()\n",
|
|
" if options is None:\n",
|
|
" options = ASRRequestOptions()\n",
|
|
" request_generator.set_audio_filepaths([audio_filepath], [options])\n",
|
|
"\n",
|
|
" sep = pipeline.get_sep()\n",
|
|
" accumulated_final = \"\"\n",
|
|
" segments = []\n",
|
|
" text_to_show = \"\"\n",
|
|
"\n",
|
|
" pipeline.open_session()\n",
|
|
" for step, requests in enumerate(request_generator):\n",
|
|
" step_outputs = pipeline.transcribe_step(requests)\n",
|
|
" for out in step_outputs:\n",
|
|
" if out.final_transcript:\n",
|
|
" # Strip leading separator on the very first segment\n",
|
|
" text = out.final_transcript if accumulated_final else out.final_transcript.lstrip(sep)\n",
|
|
" accumulated_final += text + \"[EoU-🎬]\"\n",
|
|
"\n",
|
|
" final_segments = out.final_segments\n",
|
|
" if len(final_segments) > 0:\n",
|
|
" # Strip leading separator on the very first segment\n",
|
|
" first_segment = final_segments[0]\n",
|
|
" first_segment.text = first_segment.text.lstrip(sep)\n",
|
|
" segments.extend(final_segments)\n",
|
|
" \n",
|
|
" partial_transcript = out.partial_transcript\n",
|
|
" if verbose:\n",
|
|
" text_to_show = f\"{accumulated_final}{partial_transcript}\".strip()\n",
|
|
" print(f\"Step#{step:<3} -> {text_to_show}\")\n",
|
|
" pipeline.close_session()\n",
|
|
"\n",
|
|
" final_text = accumulated_final.replace(\"[EoU-🎬]\", \"\")\n",
|
|
" if verbose:\n",
|
|
" print(\"-\" * 100)\n",
|
|
" print(f\"Full transcription: {final_text}\")\n",
|
|
" return final_text, segments"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"id": "af5cfcdb-98c5-47fd-a2bb-d8c8f0e4fa6b",
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"# Enable EoU per-request: 400 ms silence window (≈ 5 blank tokens) triggers an EoU boundary\n",
|
|
"options = ASRRequestOptions(stop_history_eou=400)\n",
|
|
"final_text, segments = transcribe_single(generic_pipeline, demo_audio_filepath, options=options, verbose=True)"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"id": "rqp1dezrzx",
|
|
"metadata": {},
|
|
"source": [
|
|
"## Word Timestamps and Confidence Scores\n",
|
|
"\n",
|
|
"Each `TranscribeStepOutput` carries a `final_segments` list that is populated at every EoU boundary. By default the list contains one `TextSegment` per finalized utterance (segment-level granularity). Switching to **word-level** granularity returns one `Word` object per recognized word, enabling precise alignment of each token to its position in the audio timeline.\n",
|
|
"\n",
|
|
"#### How it works?\n",
|
|
"Granularity is controlled per-request via `ASRRequestOptions`:\n",
|
|
"\n",
|
|
"```python\n",
|
|
"from nemo.collections.asr.inference.utils.enums import ASROutputGranularity\n",
|
|
"from nemo.collections.asr.inference.streaming.framing.request_options import ASRRequestOptions\n",
|
|
"\n",
|
|
"# Segment-level (default)\n",
|
|
"ASRRequestOptions(asr_output_granularity=ASROutputGranularity.SEGMENT)\n",
|
|
"\n",
|
|
"# Word-level\n",
|
|
"ASRRequestOptions(asr_output_granularity=ASROutputGranularity.WORD)\n",
|
|
"```\n",
|
|
"\n",
|
|
"For **RNNT pipelines**, word-level confidence scores require enabling `preserve_frame_confidence` at pipeline construction time:\n",
|
|
"```python\n",
|
|
"cfg_overrides={\"asr.decoding.greedy.preserve_frame_confidence\": True}\n",
|
|
"```\n",
|
|
"\n",
|
|
"#### Key parameters\n",
|
|
"- `asr_output_granularity`: `ASROutputGranularity.SEGMENT` returns one `TextSegment` per EoU segment; `ASROutputGranularity.WORD` returns one `Word` per recognized token with individual start/end timestamps\n",
|
|
"\n",
|
|
"#### Warnings and Notes\n",
|
|
"- `final_segments` is only populated at EoU boundaries. It is empty on all other steps."
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"id": "mzf6j8ysb1n",
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"options = ASRRequestOptions(\n",
|
|
" stop_history_eou=400, \n",
|
|
" asr_output_granularity=ASROutputGranularity.SEGMENT\n",
|
|
")\n",
|
|
"_, text_segments = transcribe_single(generic_pipeline, demo_audio_filepath, options=options)\n",
|
|
"log_text_segments(text_segments, title=\"Segment-level timestamps\")"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"id": "0t0fudp0wsi",
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"options = ASRRequestOptions(\n",
|
|
" stop_history_eou=400, \n",
|
|
" asr_output_granularity=ASROutputGranularity.WORD\n",
|
|
")\n",
|
|
"_, word_segments = transcribe_single(generic_pipeline, demo_audio_filepath, options=options)\n",
|
|
"\n",
|
|
"log_text_segments(word_segments, title=\"Word-level timestamps\")\n",
|
|
"visualize_word_timestamps(demo_audio_filepath, word_segments)"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"id": "gzs6daxfn0m",
|
|
"metadata": {},
|
|
"source": [
|
|
"## Per-Stream Biasing\n",
|
|
"\n",
|
|
"#### How it works?\n",
|
|
"Context biasing (phrase boosting) steers the decoder toward a list of expected words or phrases. This is useful when domain-specific terms — product names, proper nouns, technical jargon — are unlikely to be recognized correctly out of the box. A compact token-level prefix tree (boosting tree) is built from the supplied key phrases. At each decoding step the tree contributes a log-probability bonus (`alpha`) to tokens that advance a phrase prefix, raising those tokens above competitors. Each stream in a batch can carry a **different** biasing config, so you can boost different phrases per caller within the same inference pass.\n",
|
|
"\n",
|
|
"#### Key parameters\n",
|
|
"- `key_phrases_list` (in `BoostingTreeModelConfig`): list of strings to boost during decoding\n",
|
|
"- `boosting_model_alpha` (in `BiasingRequestItemConfig`): log-probability bonus applied to tokens that match a boosted phrase prefix; higher values produce stronger boosting\n",
|
|
"\n",
|
|
"#### Warnings and Notes\n",
|
|
"- Context biasing currently works with greedy decoding. It will be more effective with beam search, which is under development.\n",
|
|
"- Setting `boosting_model_alpha` too high can cause the decoder to force a boosted phrase even when it is clearly not present in the audio. Tune this value empirically.\n",
|
|
"- The boosting tree is built at request time — large key phrase lists with long phrases may add overhead at session startup."
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"id": "lvw77603on",
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"biasing_options = [\n",
|
|
" # No biasing\n",
|
|
" ASRRequestOptions(stop_history_eou=400),\n",
|
|
"\n",
|
|
" # key_phrases_list=[\"Yoolka\"]\n",
|
|
" ASRRequestOptions(\n",
|
|
" stop_history_eou=400,\n",
|
|
" biasing_cfg=BiasingRequestItemConfig(\n",
|
|
" boosting_model_cfg=BoostingTreeModelConfig(key_phrases_list=[\"Yoolka\"]),\n",
|
|
" boosting_model_alpha=10.0,\n",
|
|
" ),\n",
|
|
" ),\n",
|
|
" # key_phrases_list=[\"yoolka\", \"Antoniya\"]\n",
|
|
" ASRRequestOptions(\n",
|
|
" stop_history_eou=400,\n",
|
|
" biasing_cfg=BiasingRequestItemConfig(\n",
|
|
" boosting_model_cfg=BoostingTreeModelConfig(key_phrases_list=[\"yoolka\", \"Antoniya\"]),\n",
|
|
" boosting_model_alpha=3.0,\n",
|
|
" ),\n",
|
|
" ),\n",
|
|
"]\n",
|
|
"\n",
|
|
"# pipeline.run() is a convenience wrapper around open_session/transcribe_step/close_session.\n",
|
|
"# It accepts a list of audio file paths and per-stream options, and returns a list of dicts\n",
|
|
"# with 'text' and 'segments' keys — one entry per input stream.\n",
|
|
"# Note: biasing currently works with greedy decoding and will be more effective with beam search.\n",
|
|
"output = generic_pipeline.run([biasing_demo_audio_file]*len(biasing_options), options=biasing_options)\n",
|
|
"\n",
|
|
"\n",
|
|
"def highlight_phrases(text, phrases, color=\"#ff7043\"):\n",
|
|
" \"\"\"Wrap each occurrence of a phrase (case-insensitive) in a highlighted span.\"\"\"\n",
|
|
" if not phrases:\n",
|
|
" return text\n",
|
|
" pattern = re.compile(\n",
|
|
" r'\\b(' + '|'.join(re.escape(p) for p in phrases) + r')\\b',\n",
|
|
" re.IGNORECASE,\n",
|
|
" )\n",
|
|
" return pattern.sub(\n",
|
|
" lambda m: (\n",
|
|
" f'<span style=\"background:{color}33;color:{color};font-weight:bold;'\n",
|
|
" f'border-radius:3px;padding:1px 5px;\">{m.group()}</span>'\n",
|
|
" ),\n",
|
|
" text,\n",
|
|
" )\n",
|
|
"\n",
|
|
"\n",
|
|
"rows = []\n",
|
|
"for i, opt in enumerate(biasing_options):\n",
|
|
" bcfg = opt.biasing_cfg\n",
|
|
" phrases = bcfg.boosting_model_cfg.key_phrases_list if bcfg else []\n",
|
|
" keywords_str = \", \".join(f'<code>{p}</code>' for p in phrases) if phrases else \"—\"\n",
|
|
" border_color = \"#4a9eff\" if phrases else \"#ccc\"\n",
|
|
" highlighted = highlight_phrases(output[i]['text'], phrases)\n",
|
|
"\n",
|
|
" rows.append(\n",
|
|
" f'<div style=\"margin:6px 0;padding:10px 14px;border-left:4px solid {border_color};'\n",
|
|
" f'background:#fafafa;border-radius:0 4px 4px 0;font-family:serif;\">'\n",
|
|
" f'<div style=\"font-size:0.75em;color:#888;margin-bottom:5px;font-family:monospace;\">'\n",
|
|
" f'stream {i} · bias: {keywords_str}</div>'\n",
|
|
" f'<div style=\"line-height:1.7;\">{highlighted}</div>'\n",
|
|
" f'</div>'\n",
|
|
" )\n",
|
|
"\n",
|
|
"display(HTML(\"\".join(rows)))"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"id": "062879bf-5bcc-4e8f-9c16-2f8479f31500",
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"del generic_pipeline\n",
|
|
"gc.collect()\n",
|
|
"torch.cuda.empty_cache()"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"id": "2efed1dd-31c5-4e0e-84a0-08fde0c77708",
|
|
"metadata": {},
|
|
"source": [
|
|
"## Inverse Text Normalization\n",
|
|
"\n",
|
|
"#### How it works?\n",
|
|
"**Inverse Text Normalization (ITN)** converts ASR output from spoken form to written form, turning numeric expressions, dates, currency amounts, and other entities into their canonical notation. ITN runs as a post-processing step inside `transcribe_step()` and is applied to each finalized EoU segment before it appears in `final_transcript`. Word-level alignment is preserved, so timestamps remain correct even after normalization.\n",
|
|
"\n",
|
|
"#### Key parameters\n",
|
|
"- `enable_itn` (pipeline config): must be `true` at pipeline construction time to compile and load the ITN grammar\n",
|
|
"- `lang` (pipeline config): language code for the ITN grammar (e.g. `\"en\"`)\n",
|
|
"\n",
|
|
"#### Warnings and Notes\n",
|
|
"- ITN requires the `nemo_text_processing` package with `pynini`, included in the NeMo ASR extras.\n",
|
|
"- The first run compiles and caches `.far` grammar files to `cache_dir`. Subsequent runs reuse the cache and are fast.\n",
|
|
"- `enable_itn: true` must be set in the **pipeline config** at startup. Per-request toggling via `ASRRequestOptions` only takes effect when the grammar is already loaded."
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"id": "dd8d5671-38cc-4847-8c7d-d88ff98998ee",
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"# Build a pipeline with ITN loaded (enable_itn=True in the config).\n",
|
|
"# The grammar is compiled once at startup; per-request toggling is cheap after that.\n",
|
|
"itn_pipeline = create_pipeline(\n",
|
|
" \"../../examples/asr/conf/asr_streaming_inference/buffered_rnnt.yaml\",\n",
|
|
" cfg_overrides={\n",
|
|
" \"asr.model_name\": \"nvidia/parakeet-rnnt-1.1b\",\n",
|
|
" \"streaming.left_padding_size\": 1.6,\n",
|
|
" \"streaming.chunk_size\": 0.54,\n",
|
|
" \"streaming.right_padding_size\": 1.6,\n",
|
|
" \"streaming.batch_size\": 8,\n",
|
|
" \"streaming.stateful\": True,\n",
|
|
" \"asr_decoding_type\": \"rnnt\",\n",
|
|
" \"log_level\": 40,\n",
|
|
" \"enable_itn\": True, # compile & load ITN grammar at startup\n",
|
|
" \"lang\": \"en\", # language code for the ITN grammar; required when enable_itn=True\n",
|
|
" },\n",
|
|
")"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"id": "cmk77ef92f",
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"# Run the same audio twice in a single batch:\n",
|
|
"# stream 0 → ITN OFF (raw ASR output, spoken form preserved)\n",
|
|
"# stream 1 → ITN ON (spoken numbers/dates/currency converted to written notation)\n",
|
|
"# For audio that contains numbers, dates, or currency you will see the difference.\n",
|
|
"itn_options = [\n",
|
|
" ASRRequestOptions(enable_itn=False, stop_history_eou=400, asr_output_granularity=ASROutputGranularity.WORD),\n",
|
|
" ASRRequestOptions(enable_itn=True, stop_history_eou=400, asr_output_granularity=ASROutputGranularity.WORD),\n",
|
|
"]\n",
|
|
"\n",
|
|
"output = itn_pipeline.run([itn_demo_audio_filepath, itn_demo_audio_filepath], options=itn_options)\n",
|
|
"for i in range(len(itn_options)):\n",
|
|
" itn_enabled = [\"OFF\", \"ON\"][int(itn_options[i].enable_itn)]\n",
|
|
" log_text_segments(output[i][\"segments\"], title=f\"ITN is {itn_enabled}\")"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"id": "732e53e8-3f8b-46a7-9cf8-f1dc658b3e58",
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"del itn_pipeline\n",
|
|
"gc.collect()\n",
|
|
"torch.cuda.empty_cache()"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"id": "1mr4buo1ypg",
|
|
"metadata": {},
|
|
"source": [
|
|
"## Speech Translation\n",
|
|
"\n",
|
|
"#### How it works?\n",
|
|
"The Pipeline API supports **LLM-based streaming translation** as an optional post-processing step. It first performs streaming ASR, then simultaneously translates the transcribed source language into any target language. Translation runs automatically inside `transcribe_step()` — no extra calls are needed. Each stream in a batch can target a **different language** independently.\n",
|
|
"\n",
|
|
"Users will see **partial translations** that may be revised as new audio chunks arrive. After some point, a portion of the translation prefix becomes fixed and will no longer change. The LLM prompt used at each step:\n",
|
|
"```\n",
|
|
"Translate the following {src_lang} source text to {tgt_lang}. Always output text in the {tgt_lang} language:\n",
|
|
"{src_lang}: {asr_predicted_text}\n",
|
|
"{tgt_lang}: {translation_prefix}\n",
|
|
"```\n",
|
|
"The model completes the prompt from `{translation_prefix}`, producing an updated translation consistent with prior output. At each step:\n",
|
|
"\n",
|
|
"1. Run the LLM on the current ASR transcript, seeded with the translation prefix from the previous step\n",
|
|
"2. Compare the new translation output with the previous one\n",
|
|
"3. Extend or retain the prefix using one of two policies:\n",
|
|
" - **wait-k**: specifies the maximum number of words the translation is allowed to lag behind the ASR transcript. If the translation falls more than `wait_k` words behind, the prefix is automatically extended so that `len(prefix.split()) >= len(asr_transcript.split()) - wait_k`. Larger values of `wait_k` yield better translation quality at the cost of higher latency — more ASR tokens must arrive before the LLM is invoked.\n",
|
|
" - **LCP** (Longest Common Prefix): the prefix is the longest common prefix shared between the current and previous translations. The prefix is never forced to grow. Enable LCP by setting `wait_k=-1`.\n",
|
|
"\n",
|
|
"#### Key parameters\n",
|
|
"Per-stream configuration via `ASRRequestOptions`:\n",
|
|
"- `enable_nmt`: set to `True` to activate translation for this stream\n",
|
|
"- `source_language`: language of the audio (e.g. `\"English\"`)\n",
|
|
"- `target_language`: desired output language (e.g. `\"German\"`)\n",
|
|
"\n",
|
|
"Translation output fields in `TranscribeStepOutput`:\n",
|
|
"- `final_translation`: finalized translation of `final_transcript`, populated at EoU boundaries\n",
|
|
"- `partial_translation`: running translation of `partial_transcript`, updated every step and may change\n",
|
|
"\n",
|
|
"#### Warnings and Notes\n",
|
|
"- `enable_nmt: true` must be set in the **pipeline config** at construction time so the NMT model is loaded. Per-request toggling only works after the model is loaded.\n",
|
|
"- Translation requires the **vLLM** package: `pip install vllm`\n",
|
|
"- The NMT model requires significant GPU memory (3+ GB for a 1.7B model). Use `select_devices()` to route ASR and NMT to separate GPUs when available.\n",
|
|
"- `partial_translation` is not stable and will be revised as more audio arrives. Only `final_translation` should be treated as authoritative translated output.\n",
|
|
"- Higher `wait_k` values improve BLEU/COMET scores at the cost of increased latency (LAAL).\n",
|
|
"- `LCP` typically matches or slightly exceeds the best wait-k BLEU score, but also yields the highest latency.\n",
|
|
"- Better ASR transcripts produce better translations — ASR errors propagate directly into the translation output."
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"id": "5d0db159-c199-4e83-a149-821a8c769b30",
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"# del nmt_pipeline if it's already defined to avoid OOM\n",
|
|
"if \"nmt_pipeline\" in globals():\n",
|
|
" del nmt_pipeline\n",
|
|
" gc.collect()\n",
|
|
" torch.cuda.empty_cache()"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"id": "hl9226cfwhu",
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"def select_devices(min_single_gpu_gb=80, min_dual_gpu_gb=30):\n",
|
|
" \"\"\"\n",
|
|
" Priority:\n",
|
|
" 1. Single GPU with >= min_single_gpu_gb GB -> both ASR and NMT on GPU 0\n",
|
|
" 2. Two GPUs each with >= min_dual_gpu_gb GB -> ASR on GPU 0, NMT on GPU 1\n",
|
|
" 3. Fallback -> ASR on CPU (device_id=-1), NMT on GPU 0\n",
|
|
" \"\"\"\n",
|
|
" n = torch.cuda.device_count()\n",
|
|
"\n",
|
|
" def gb(i):\n",
|
|
" return torch.cuda.get_device_properties(i).total_memory / 1024 ** 3\n",
|
|
"\n",
|
|
" print(f\"Available GPUs: {n}\")\n",
|
|
" for i in range(n):\n",
|
|
" props = torch.cuda.get_device_properties(i)\n",
|
|
" print(f\" GPU {i}: {props.name} — {gb(i):.1f} GB\")\n",
|
|
"\n",
|
|
" if n >= 1 and gb(0) >= min_single_gpu_gb:\n",
|
|
" asr_device, asr_device_id = \"cuda\", 0\n",
|
|
" nmt_device, nmt_device_id = \"cuda\", 0\n",
|
|
" print(f\"\\n-> Single large GPU detected. Using GPU 0 ({gb(0):.1f} GB) for both ASR and NMT.\")\n",
|
|
" elif n >= 2 and gb(0) >= min_dual_gpu_gb and gb(1) >= min_dual_gpu_gb:\n",
|
|
" asr_device, asr_device_id = \"cuda\", 0\n",
|
|
" nmt_device, nmt_device_id = \"cuda\", 1\n",
|
|
" print(f\"\\n-> Two GPUs detected. ASR on GPU 0 ({gb(0):.1f} GB), NMT on GPU 1 ({gb(1):.1f} GB).\")\n",
|
|
" else:\n",
|
|
" asr_device, asr_device_id = \"cpu\", -1\n",
|
|
" nmt_device, nmt_device_id = \"cuda\", 0\n",
|
|
" gpu_info = f\"GPU 0 ({gb(0):.1f} GB)\" if n >= 1 else \"no GPU found\"\n",
|
|
" print(f\"\\n-> Insufficient GPU memory. ASR on CPU, NMT on {gpu_info}.\")\n",
|
|
"\n",
|
|
" return asr_device, asr_device_id, nmt_device, nmt_device_id\n",
|
|
"\n",
|
|
"\n",
|
|
"asr_device, asr_device_id, nmt_device, nmt_device_id = select_devices()"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"id": "1fdqk4yw3w5",
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"nmt_pipeline = create_pipeline(\n",
|
|
" \"../../examples/asr/conf/asr_streaming_inference/buffered_rnnt.yaml\",\n",
|
|
" cfg_overrides={\n",
|
|
" \"asr.model_name\": \"nvidia/parakeet-rnnt-1.1b\",\n",
|
|
" \"asr.device\": asr_device,\n",
|
|
" \"asr.device_id\": asr_device_id,\n",
|
|
" \"streaming.left_padding_size\": 1.6,\n",
|
|
" \"streaming.chunk_size\": 0.54,\n",
|
|
" \"streaming.right_padding_size\": 1.6,\n",
|
|
" \"streaming.batch_size\": 8,\n",
|
|
" \"streaming.stateful\": True,\n",
|
|
" \"asr_decoding_type\": \"rnnt\",\n",
|
|
" \"log_level\": 40,\n",
|
|
" \"enable_nmt\": True,\n",
|
|
" \"nmt.model_name\": \"utter-project/EuroLLM-1.7B-Instruct\",\n",
|
|
" \"nmt.device\": nmt_device,\n",
|
|
" \"nmt.device_id\": nmt_device_id,\n",
|
|
" \"nmt.batch_size\": 8,\n",
|
|
" \"nmt.waitk\": -1, \n",
|
|
" },\n",
|
|
")"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"id": "0cgb3ps4ev9f",
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"TARGET_LANGUAGES = [\"German\", \"French\", \"Spanish\", \"Russian\", \"Serbian\", \"Croatian\", \"English\"]\n",
|
|
"\n",
|
|
"# Use the same English audio file for each target language\n",
|
|
"audio_files = [demo_audio_filepath] * len(TARGET_LANGUAGES)\n",
|
|
"\n",
|
|
"options = [\n",
|
|
" ASRRequestOptions(\n",
|
|
" enable_nmt=True,\n",
|
|
" source_language=\"English\",\n",
|
|
" target_language=lang,\n",
|
|
" stop_history_eou=400,\n",
|
|
" )\n",
|
|
" for lang in TARGET_LANGUAGES\n",
|
|
"]"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"id": "sx2gzhmezya",
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"do_streaming(nmt_pipeline, audio_files, options=options)"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"id": "r0k3k4sl0aq",
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"del nmt_pipeline\n",
|
|
"gc.collect()\n",
|
|
"torch.cuda.empty_cache()"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"id": "impl-details-section-header",
|
|
"metadata": {},
|
|
"source": [
|
|
"# Implementation Details\n",
|
|
"\n",
|
|
"This section describes the low-level building blocks that underpin the pipeline API. Understanding these concepts is useful for advanced usage, custom integrations, or debugging.\n",
|
|
"\n",
|
|
"1. **Frame**: An immutable container representing a single chunk of audio.\n",
|
|
"2. **Stream**: An iterator over an audio source that yields `Frame` objects.\n",
|
|
"3. **Multiple Streams**: A higher-level wrapper (`MultiStream`) that manages multiple streams simultaneously and interleaves their frames into batches.\n",
|
|
"4. **Continuous Batching**: Keeps the active batch at full capacity (`ContinuousBatchedFrameStreamer`) by automatically replacing finished streams with new ones."
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"id": "8778a9e9-9206-4ba8-b462-8038baaff4ba",
|
|
"metadata": {},
|
|
"source": [
|
|
"## Frame\n",
|
|
"\n",
|
|
"#### How it works?\n",
|
|
"A `Frame` is a frozen dataclass that wraps a short slice of raw audio samples together with metadata. Frames are the atomic unit of data flowing through the pipeline — each call to `transcribe_step()` receives a batch of `Frame` objects.\n",
|
|
"\n",
|
|
"#### Key parameters\n",
|
|
"- `samples`: 1-D Tensor of raw audio samples\n",
|
|
"- `stream_id`: integer identifier for the audio stream this frame belongs to\n",
|
|
"- `is_first`: boolean flag marking the first chunk of a stream\n",
|
|
"- `is_last`: boolean flag marking the last chunk of a stream\n",
|
|
"- `length`: number of valid samples in the tensor; the remainder may be zero-padding for the last chunk. `-1` means all samples are valid\n",
|
|
"\n",
|
|
"#### Warnings and Notes\n",
|
|
"- `Frame` is immutable (frozen dataclass) — do not attempt to modify its fields in-place."
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"id": "f053f3b4-e3c7-4ce2-88bd-0c1600c8342b",
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"import torch\n",
|
|
"from nemo.collections.asr.inference.streaming.framing.request import Frame\n",
|
|
"\n",
|
|
"# Simulate a 160-ms chunk at 16 kHz\n",
|
|
"SAMPLE_RATE = 16_000\n",
|
|
"FRAME_SIZE_SECS = 0.16\n",
|
|
"n_samples = int(SAMPLE_RATE * FRAME_SIZE_SECS) # 2560 samples\n",
|
|
"\n",
|
|
"frame = Frame(\n",
|
|
" samples=torch.rand(n_samples),\n",
|
|
" stream_id=0,\n",
|
|
" is_first=True,\n",
|
|
" is_last=False,\n",
|
|
" length=n_samples, # all samples are valid (no padding)\n",
|
|
")\n",
|
|
"\n",
|
|
"print(f\"stream_id : {frame.stream_id}\")\n",
|
|
"print(f\"is_first : {frame.is_first}\")\n",
|
|
"print(f\"is_last : {frame.is_last}\")\n",
|
|
"print(f\"length : {frame.length}\")"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"id": "88a81851-647f-4dc6-be88-7077d421d477",
|
|
"metadata": {},
|
|
"source": [
|
|
"## Stream\n",
|
|
"\n",
|
|
"#### How it works?\n",
|
|
"A `Stream` is an iterator that reads an audio source and emits `Frame` objects. The concrete implementation used here is `MonoStream`, which reads a mono WAV file and slices it into fixed-size, non-overlapping chunks. Each iteration yields a one-element list containing a single `Frame`.\n",
|
|
"\n",
|
|
"#### Key parameters\n",
|
|
"- `rate`: target sample rate in Hz; the audio file is resampled automatically if its native rate differs\n",
|
|
"- `frame_size_in_secs`: duration of each emitted frame in seconds\n",
|
|
"- `stream_id`: unique integer identifier assigned to every frame produced by this stream\n",
|
|
"- `pad_last_frame`: if `True`, the final (potentially shorter) frame is zero-padded to the full `frame_size`; the `valid_size` field on that frame reflects the true number of meaningful samples\n",
|
|
"\n",
|
|
"#### Warnings and Notes\n",
|
|
"- Frames produced by `MonoStream` do not overlap — each audio sample appears in exactly one frame.\n",
|
|
"- All frames have the same `size`, but only the last frame may have `valid_size < size`."
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"id": "1b695e6d-ce10-461a-849b-e7bece07105f",
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"from nemo.collections.asr.inference.streaming.framing.mono_stream import MonoStream\n",
|
|
"\n",
|
|
"FRAME_SIZE_SECS = 1.0\n",
|
|
"\n",
|
|
"stream = MonoStream(\n",
|
|
" rate=SAMPLE_RATE,\n",
|
|
" frame_size_in_secs=FRAME_SIZE_SECS,\n",
|
|
" stream_id=0,\n",
|
|
" pad_last_frame=True,\n",
|
|
")\n",
|
|
"stream.load_audio(audio_filepaths[0])\n",
|
|
"\n",
|
|
"# Iterate through all frames to observe size, valid_size, and boundary flags\n",
|
|
"print(f\"Audio file : {audio_filepaths[0]}\")\n",
|
|
"print(f\"Frame size : {FRAME_SIZE_SECS}s ({int(SAMPLE_RATE * FRAME_SIZE_SECS)} samples)\")\n",
|
|
"print()\n",
|
|
"\n",
|
|
"total_frames = 0\n",
|
|
"for frames_list in stream:\n",
|
|
" frame = frames_list[0] # MonoStream yields a one-element list\n",
|
|
" total_frames += 1\n",
|
|
" tag = \"FIRST\" if frame.is_first else (\"LAST\" if frame.is_last else \"\")\n",
|
|
" print(f\" Frame {total_frames:1d} size={frame.size:<5d} valid_size={frame.valid_size:<5d} {tag}\")\n",
|
|
"print(f\"\\nTotal frames: {total_frames}\")"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"id": "313caefc-2a5c-4836-8422-9e2e88d8bb9c",
|
|
"metadata": {},
|
|
"source": [
|
|
"## Multiple Streams\n",
|
|
"\n",
|
|
"#### How it works?\n",
|
|
"When transcribing more than one audio source simultaneously, use `MultiStream` — a higher-level wrapper that manages a fixed collection of `MonoStream` objects and interleaves their frames. At each iteration step it pulls `n_frames_per_stream` frames from every active stream and returns them as a flat batch. Iteration continues until all streams are exhausted; as shorter files finish first, the batch size naturally shrinks.\n",
|
|
"\n",
|
|
"#### Key parameters\n",
|
|
"- `n_frames_per_stream`: number of frames drawn from each stream per iteration step\n",
|
|
"\n",
|
|
"#### Warnings and Notes\n",
|
|
"- All streams must be added before iteration begins — `MultiStream` does not support adding streams mid-iteration.\n",
|
|
"- The batch size decreases as shorter streams finish. If you need a consistently full batch throughout, use `ContinuousBatchedFrameStreamer` instead (see the next section)."
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"id": "94dd9b4c-0f83-41e1-a115-684d1e7b1528",
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"from nemo.collections.asr.inference.streaming.framing.multi_stream import MultiStream\n",
|
|
"\n",
|
|
"multi_streamer = MultiStream(n_frames_per_stream=1)\n",
|
|
"for audio_id, filepath in enumerate(audio_filepaths):\n",
|
|
" stream = MonoStream(\n",
|
|
" rate=SAMPLE_RATE,\n",
|
|
" frame_size_in_secs=FRAME_SIZE_SECS,\n",
|
|
" stream_id=audio_id,\n",
|
|
" pad_last_frame=True,\n",
|
|
" )\n",
|
|
" stream.load_audio(filepath)\n",
|
|
" multi_streamer.add_stream(stream=stream, stream_id=stream.stream_id)\n",
|
|
"\n",
|
|
"print(f\"Active streams: {len(multi_streamer)}\")\n",
|
|
"print(f\"{'Step':<6} {'Batch':>5} Stream_ids\")\n",
|
|
"print(\"-\" * 40)\n",
|
|
"\n",
|
|
"frame_counts = {}\n",
|
|
"for step, batch in enumerate(multi_streamer):\n",
|
|
" # batch is a list of Frame objects.\n",
|
|
" # shorter files finish earlier, so the batch size shrinks as streams exhaust.\n",
|
|
" for frame in batch:\n",
|
|
" frame_counts[frame.stream_id] = frame_counts.get(frame.stream_id, 0) + 1\n",
|
|
"\n",
|
|
" ids = [f.stream_id for f in batch]\n",
|
|
" print(f\"{step:<6} {len(batch):>5} {ids}\")\n",
|
|
"\n",
|
|
"print(f\"\\nPer-stream frame counts:\")\n",
|
|
"for sid in sorted(frame_counts):\n",
|
|
" print(f\" stream {sid}: {frame_counts[sid]:>3} frames\")"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"id": "34549f3f-dae1-4f11-95c9-581a5aa75af7",
|
|
"metadata": {},
|
|
"source": [
|
|
"## Continuous Batching\n",
|
|
"\n",
|
|
"#### How it works?\n",
|
|
"With `MultiStream`, all streams are loaded upfront and the batch shrinks as shorter files finish, leaving GPU capacity underutilised. `ContinuousBatchedFrameStreamer` fixes this by automatically starting a new stream whenever one completes, keeping the active batch at `batch_size` for as long as there are files remaining. It wraps `MultiStream` internally and refills it to capacity at every step.\n",
|
|
"\n",
|
|
"#### Key parameters\n",
|
|
"- `sample_rate`: target sample rate in Hz for all streams\n",
|
|
"- `frame_size_in_secs`: duration of each frame in seconds\n",
|
|
"- `batch_size`: maximum number of concurrently active streams; new streams are added automatically to maintain this limit\n",
|
|
"- `n_frames_per_stream`: number of frames pulled from each active stream per iteration step\n",
|
|
"- `pad_last_frame`: if `True`, the final frame of each stream is zero-padded to the full frame size\n",
|
|
"\n",
|
|
"#### Warnings and Notes\n",
|
|
"- The batch size only drops below `batch_size` when fewer than `batch_size` files remain in the queue overall.\n",
|
|
"- `set_audio_filepaths()` must be called before iteration begins; it accepts a list of file paths and a corresponding list of `ASRRequestOptions`."
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"id": "21air7eg9sr",
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"from nemo.collections.asr.inference.streaming.framing.multi_stream import ContinuousBatchedFrameStreamer\n",
|
|
"from nemo.collections.asr.inference.streaming.framing.request_options import ASRRequestOptions\n",
|
|
"\n",
|
|
"SAMPLE_RATE = 16_000\n",
|
|
"FRAME_SIZE_SECS = 1.0\n",
|
|
"BATCH_SIZE = 3 # at most 3 streams active simultaneously\n",
|
|
"\n",
|
|
"streamer = ContinuousBatchedFrameStreamer(\n",
|
|
" sample_rate=SAMPLE_RATE,\n",
|
|
" frame_size_in_secs=FRAME_SIZE_SECS,\n",
|
|
" batch_size=BATCH_SIZE,\n",
|
|
" n_frames_per_stream=1,\n",
|
|
" pad_last_frame=True,\n",
|
|
")\n",
|
|
"\n",
|
|
"# define per-stream options\n",
|
|
"options = [ASRRequestOptions() for _ in audio_filepaths]\n",
|
|
"streamer.set_audio_filepaths(audio_filepaths, options)\n",
|
|
"\n",
|
|
"print(f\"Total files : {len(audio_filepaths)}\")\n",
|
|
"print(f\"Batch size : {BATCH_SIZE} (active streams)\")\n",
|
|
"print()\n",
|
|
"print(f\"{'Step':<6} {'Batch':>5} Stream_ids\")\n",
|
|
"print(\"-\" * 40)\n",
|
|
"\n",
|
|
"frame_counts = {}\n",
|
|
"for step, batch in enumerate(streamer):\n",
|
|
" # batch shrinks only when fewer than batch_size files remain overall;\n",
|
|
" # otherwise it stays at batch_size even as individual streams finish.\n",
|
|
" for frame in batch:\n",
|
|
" frame_counts[frame.stream_id] = frame_counts.get(frame.stream_id, 0) + 1\n",
|
|
"\n",
|
|
" ids = [f.stream_id for f in batch]\n",
|
|
" print(f\"{step:<6} {len(batch):>5} {ids}\")\n",
|
|
"\n",
|
|
"print(f\"\\nPer-stream frame counts:\")\n",
|
|
"for sid in sorted(frame_counts):\n",
|
|
" print(f\" stream {sid}: {frame_counts[sid]:>3} frames\")"
|
|
]
|
|
}
|
|
],
|
|
"metadata": {
|
|
"kernelspec": {
|
|
"display_name": "nemo-env",
|
|
"language": "python",
|
|
"name": "env"
|
|
},
|
|
"language_info": {
|
|
"codemirror_mode": {
|
|
"name": "ipython",
|
|
"version": 3
|
|
},
|
|
"file_extension": ".py",
|
|
"mimetype": "text/x-python",
|
|
"name": "python",
|
|
"nbconvert_exporter": "python",
|
|
"pygments_lexer": "ipython3",
|
|
"version": "3.12.12"
|
|
}
|
|
},
|
|
"nbformat": 4,
|
|
"nbformat_minor": 5
|
|
}
|