feat: Granary v2 pipeline — vLLM 0.19.1, throughput metadata export, standardized audio stages

[mohammadaaftabv]

Summary

• Upgrade vLLM to 0.19.1 with torch==2.10.0 for Qwen3.5 MoE architecture support
• Add throughput metadata export: per-stage inference timing via _log_metrics() in InferenceQwenOmniStage and PnCRestorationStage, with perf_summary.json aggregate output in ShardedManifestWriterStage
• Pin dependencies (vllm==0.19.1, torch==2.10.0+cu128, transformers==5.7.0) for stable ABI compatibility
• Fix config paths (/src/Curator -> /opt/Curator) for Docker image alignment
• Add pipeline architecture documentation

Changes

• nemo_curator/stages/audio/inference/qwen_omni.py — add inference timing + utterance count metrics
• nemo_curator/stages/audio/text_filtering/pnc_restoration.py — add inference timing + restoration count metrics
• nemo_curator/stages/audio/io/sharded_manifest_writer.py — add perf_summary.json + per-shard _perf.jsonl export
• tutorials/audio/qwen_omni_inprocess/main.py — log perf summary after pipeline completion
• pyproject.toml / uv.lock — dependency pins

Test plan

• Verified pipeline completes successfully on Kratos with 17,197 utterances (25.5 min)
• Docker image built and pushed with correct dependency versions
• Verify perf_summary.json output in next Kratos run

[sarahyurick]

Hi @mohammadaaftabv since the PR is very large, are you splitting to split it up by stage/module?

[greptile-apps]

Mock return value unpacks to wrong arity — tests will fail

QwenOmni.generate() returns a 3-tuple (pred_texts, disfluency_texts, skipped_indices), but every test mock here supplies only a 2-tuple. process_batch unpacks with pred_texts, disfluency_texts, skipped_indices = self._model.generate(...), so all affected tests will raise ValueError: not enough values to unpack (expected 3, got 2).

Affected mocks (at minimum lines 62, 74, 81, 99, 112) should each be updated to include the third element:

stage._model.generate.return_value = (["hello world"], [""], set())

[greptile-apps]

ITN model silently skipped when skip_pnc=True and models share default ID

_prefetch_models avoids a double-download by comparing itn_model against pnc_model_id regardless of whether PnC itself is being fetched. When skip_pnc=True and itn_model_id equals pnc_model_id (both default to "Qwen/Qwen3.5-35B-A3B-FP8"), the PnC download is skipped, the comparison itn_model != cfg.get("pnc_model_id", ...) evaluates to False, and the ITN model is never queued. The parallel pre-fetch is silently a no-op for the ITN model. The pipeline falls back to setup_on_node() downloading it sequentially, erasing the 10-15 min advantage the pre-fetch was designed to provide.

Suggested change

	if cfg.get("enable_itn", False):
	itn_model = cfg.get("itn_model_id", "Qwen/Qwen3.5-35B-A3B-FP8")
	if itn_model != cfg.get("pnc_model_id", "Qwen/Qwen3.5-35B-A3B-FP8"):
	tasks.append((f"snapshot:{itn_model}", lambda m=itn_model: snapshot_download(m)))
	if cfg.get("enable_itn", False):
	itn_model = cfg.get("itn_model_id", "Qwen/Qwen3.5-35B-A3B-FP8")
	already_queued = any(name == f"snapshot:{itn_model}" for name, _ in tasks)
	if not already_queued:
	tasks.append((f"snapshot:{itn_model}", lambda m=itn_model: snapshot_download(m)))

[greptile-apps]

ManifestWriterStage missing IS_ACTOR_STAGE spec for Ray

ManifestWriterStage defines num_workers() = 1 and an Xenna-spec {"num_workers": 1}, but has no ray_stage_spec() override. In the Ray Data backend, without IS_ACTOR_STAGE: True, the executor may instantiate multiple concurrent workers for this stage. Each worker appends to the same output path, producing interleaved JSONL lines that corrupt the output file. ShardedManifestWriterStage (the parallel writer in this PR) correctly pairs num_workers() = 1 with ray_stage_spec() = {RayStageSpecKeys.IS_ACTOR_STAGE: True}; ManifestWriterStage needs the same.

[greptile-apps]

followup_prompt_file is resolved inside _create_model() but outputs() and process_batch() both gate on self.followup_prompt (the raw dataclass field, which stays None). If a user sets only followup_prompt_file, QwenOmni will run full two-turn inference consuming GPU time and tokens, but disfluency_text_key is never written to any task and is not declared as an output — downstream stages relying on it receive tasks with the field entirely absent.

Suggested change

	def outputs(self) -> tuple[list[str], list[str]]:
	keys = [self.pred_text_key, self.skip_me_key]
	if self.followup_prompt:
	keys.append(self.disfluency_text_key)
	return [], keys
	def outputs(self) -> tuple[list[str], list[str]]:
	keys = [self.pred_text_key, self.skip_me_key]
	if self.followup_prompt or self.followup_prompt_file:
	keys.append(self.disfluency_text_key)
	return [], keys