Fix AppTek long-form audio evals

Author: pzelasko

nemo_skills/dataset/apptek-callcenter-dialogues/prepare.py

@@ -57,7 +57,15 @@
 
 def _metadata_paths(data_dir: Path, accent_codes: Iterable[str]) -> list[Path]:
     """Return expected metadata paths for selected accents."""
-    return [data_dir / accent / "metadata.jsonl" for accent in accent_codes]
+    return [_metadata_path(data_dir, accent) for accent in accent_codes]
+
+
+def _metadata_path(data_dir: Path, accent_code: str) -> Path:
+    """Return the metadata path, supporting both old and current HF layouts."""
+    direct_path = data_dir / accent_code / "metadata.jsonl"
+    if direct_path.exists():
+        return direct_path
+    return data_dir / "test" / accent_code / "metadata.jsonl"
 
 
 def _metadata_complete(data_dir: Path, accent_codes: Iterable[str]) -> bool:
@@ -68,7 +76,7 @@ def _metadata_complete(data_dir: Path, accent_codes: Iterable[str]) -> bool:
 def _iter_metadata_rows(data_dir: Path, accent_codes: Iterable[str]) -> Iterable[tuple[str, dict]]:
     """Yield ``(accent_code, row)`` pairs from downloaded metadata files."""
     for accent_code in accent_codes:
-        metadata_path = data_dir / accent_code / "metadata.jsonl"
+        metadata_path = _metadata_path(data_dir, accent_code)
         with metadata_path.open(encoding="utf-8") as fin:
             for line in fin:
                 line = line.strip()
@@ -81,7 +89,10 @@ def _resolve_audio_path(data_dir: Path, accent_code: str, file_name: str) -> Pat
     direct_path = data_dir / file_name
     if direct_path.exists():
         return direct_path
-    return data_dir / accent_code / file_name
+    old_layout_path = data_dir / accent_code / file_name
+    if old_layout_path.exists():
+        return old_layout_path
+    return data_dir / "test" / accent_code / file_name
 
 
 def _audio_complete(data_dir: Path, accent_codes: Iterable[str]) -> bool:
@@ -103,9 +114,9 @@ def _allow_patterns(accent_codes: Iterable[str], with_audio: bool) -> list[str]:
     """
     patterns = ["README.md", "score.py", "word_mappings.py"]
     for accent_code in accent_codes:
-        patterns.append(f"{accent_code}/metadata.jsonl")
+        patterns.append(f"test/{accent_code}/metadata.jsonl")
         if with_audio:
-            patterns.append(f"{accent_code}/audio/*.wav")
+            patterns.append(f"test/{accent_code}/audio/*.wav")
     return patterns
 
 

nemo_skills/dataset/audiobench/nonjudge/__init__.py

@@ -26,5 +26,7 @@
 # Evaluation settings
 EVAL_ARGS = "++eval_type=audio ++eval_config.normalization_mode=audiobench"
 
-# Generation settings - OpenAI format for audio-language models
-GENERATION_ARGS = "++prompt_format=openai ++enable_audio=true"
+# Generation settings - OpenAI format for audio-language models.
+# Soft-fail keeps one over-limit or malformed audio request from aborting an
+# entire chunk; audio metrics count the empty generation as incorrect.
+GENERATION_ARGS = "++prompt_format=openai ++enable_audio=true ++server.enable_soft_fail=true"

nemo_skills/dataset/audiobench/prepare.py

@@ -25,6 +25,7 @@
 """
 
 import argparse
+import copy
 import json
 import os
 import shutil
@@ -155,12 +156,14 @@ def create_manifest_entry(
         Manifest entry dict with proper format for nemo-skills
     """
     instruction = sample.get("instruction", sample.get("text", "Process the audio"))
+    if isinstance(instruction, dict):
+        instruction = instruction.get("text") or "Process the audio"
     reference = sample.get("reference", sample.get("answer", ""))
     task_type = sample.get("task_type", "unknown")
 
-    # Create absolute audio path with /data/ prefix for cluster deployment
-    # Format: /data/audiobench/{category}/audio/{dataset_name}/{filename}
-    audio_rel_path = f"/data/audiobench/{category}/audio/{dataset_name}/{audio_filename}"
+    # Paths are resolved relative to the manifest directory by the inference code.
+    # The combined category manifest lives at audiobench/{category}/test.jsonl.
+    audio_rel_path = f"audio/{dataset_name}/{audio_filename}"
 
     # Create audio metadata (both singular and plural forms for compatibility)
     audio_metadata = {"path": audio_rel_path, "duration": duration}
@@ -203,6 +206,28 @@ def create_manifest_entry(
     return entry
 
 
+def make_dataset_manifest_entry(entry: Dict) -> Dict:
+    """Adjust category-relative audio paths for per-dataset manifests."""
+    entry = copy.deepcopy(entry)
+
+    def rewrite(path: str) -> str:
+        return f"../{path}" if path.startswith("audio/") else path
+
+    if isinstance(entry.get("audio_path"), list):
+        entry["audio_path"] = [rewrite(path) for path in entry["audio_path"]]
+    elif isinstance(entry.get("audio_path"), str):
+        entry["audio_path"] = rewrite(entry["audio_path"])
+
+    for message in entry.get("messages", []):
+        if "audio" in message and "path" in message["audio"]:
+            message["audio"]["path"] = rewrite(message["audio"]["path"])
+        for audio in message.get("audios", []):
+            if "path" in audio:
+                audio["path"] = rewrite(audio["path"])
+
+    return entry
+
+
 def process_dataset(
     dataset_name: str,
     output_dir: Path,
@@ -473,7 +498,7 @@ def process_dataset(
     manifest_path = dataset_dir / f"{split}.jsonl"
     with open(manifest_path, "w", encoding="utf-8") as f:
         for entry in manifest_entries:
-            f.write(json.dumps(entry, ensure_ascii=False) + "\n")
+            f.write(json.dumps(make_dataset_manifest_entry(entry), ensure_ascii=False) + "\n")
 
     print(f"✓ Saved {successful} samples to {manifest_path}")
     if failed > 0:
@@ -566,6 +591,7 @@ def main():
 
     total_samples = 0
     total_datasets = 0
+    combined_entries = {"judge": [], "nonjudge": []}
 
     for name in target_datasets:
         # Normalize dataset name: allow passing without _test suffix
@@ -576,10 +602,10 @@ def main():
                 dataset_name = f"{dataset_name}_test"
 
         # Determine category for logging
-        category = "judge" if name in JUDGE_DATASETS else "nonjudge"
+        category = "judge" if dataset_name in JUDGE_DATASETS else "nonjudge"
 
         try:
-            num_samples, _ = process_dataset(
+            num_samples, manifest_entries = process_dataset(
                 dataset_name=dataset_name,
                 output_dir=output_dir,
                 save_audio=args.save_audio,
@@ -588,11 +614,21 @@ def main():
             )
             total_samples += num_samples
             total_datasets += 1
+            combined_entries[category].extend(manifest_entries)
             print(f"✓ Completed {dataset_name}: {num_samples} samples")
         except Exception as e:
             print(f"✗ Failed {dataset_name}: {e}")
             continue
 
+    for category, entries in combined_entries.items():
+        if not entries:
+            continue
+        combined_manifest = output_dir / category / f"{args.split}.jsonl"
+        with open(combined_manifest, "w", encoding="utf-8") as f:
+            for entry in entries:
+                f.write(json.dumps(entry, ensure_ascii=False) + "\n")
+        print(f"✓ Saved combined {category} manifest with {len(entries)} samples to {combined_manifest}")
+
     print("\n" + "=" * 60)
     print("AudioBench Preparation Summary")
     print("=" * 60)

nemo_skills/dataset/mmau-pro/closed_form/__init__.py

@@ -13,8 +13,7 @@
 # limitations under the License.
 METRICS_TYPE = "mmau_pro_closed_form"
 SCORE_MODULE = "nemo_skills.evaluation.metrics.mmau_pro_metrics"
-GENERATION_ARGS = "++prompt_format=openai ++enable_audio=true"
-EVAL_ARGS = "++eval_type=mmau-pro"
+GENERATION_ARGS = "++prompt_format=openai ++enable_audio=true ++server.enable_soft_fail=true"
 
 # NVEmbed judge configuration for closed-form evaluation
 JUDGE_PIPELINE_ARGS = {

nemo_skills/dataset/mmau-pro/prepare.py

@@ -86,9 +86,11 @@ def format_entry(entry, with_audio=False):
         audio_path = entry["audio_path"]
 
         if isinstance(audio_path, list) and audio_path:
-            user_message["audios"] = [{"path": path, "duration": 10.0} for path in audio_path]
+            formatted_entry["audio_path"] = [f"../{path}" for path in audio_path]
+            user_message["audios"] = [{"path": path, "duration": 10.0} for path in formatted_entry["audio_path"]]
         elif isinstance(audio_path, str):
-            user_message["audio"] = {"path": audio_path, "duration": 10.0}
+            formatted_entry["audio_path"] = f"../{audio_path}"
+            user_message["audio"] = {"path": formatted_entry["audio_path"], "duration": 10.0}
 
     formatted_entry["messages"] = [user_message]
     return formatted_entry

nemo_skills/evaluation/evaluator/audio.py

@@ -764,12 +764,28 @@ def eval_audio(cfg):
     asyncio.run(evaluator.eval_full())
 
 
+def coerce_text_field(value: Any) -> str:
+    """Convert loose AudioBench text/reference fields to plain text."""
+    if value is None:
+        return ""
+    if isinstance(value, str):
+        return value.strip()
+    if isinstance(value, dict):
+        for key in ("text", "expected_answer", "answer", "transcript", "reference"):
+            if key in value and value[key] is not None:
+                return coerce_text_field(value[key])
+        return " ".join(coerce_text_field(item) for item in value.values() if item is not None).strip()
+    if isinstance(value, (list, tuple)):
+        return " ".join(coerce_text_field(item) for item in value if item is not None).strip()
+    return str(value).strip()
+
+
 def evaluate_sample(sample: dict[str, Any], config: AudioEvaluatorConfig) -> dict[str, Any]:
     """Evaluate single sample based on task_type. Returns dict of updates to merge."""
     updates = {}
     task_type = sample.get("task_type", "unknown")
-    generation = sample["generation"].strip()
-    expected_answer = sample.get("expected_answer", "").strip()
+    generation = coerce_text_field(sample.get("generation", ""))
+    expected_answer = coerce_text_field(sample.get("expected_answer", ""))
 
     # Extract ASR text from generation
     # E.g Qwen ASR uses <asr_text> tags to indicate the ASR text

nemo_skills/evaluation/evaluator/mmau_pro.py

@@ -56,6 +56,10 @@ def eval_mmau_pro(cfg):
 
 def evaluate_instruction_following_sample(sample: dict[str, Any]) -> dict[str, Any]:
     """Evaluate a single instruction following sample."""
+    if sample is None:
+        LOG.warning("Encountered null MMAU-Pro sample; marking it incorrect")
+        return {"is_correct": False, "error": "null_sample"}
+
     sample = sample.copy()
     generation = sample.get("generation", "").strip()
 

nemo_skills/evaluation/evaluator/nvembed_judge.py

@@ -42,7 +42,7 @@ def install_packages():
     """Install required packages for NVEmbed evaluation."""
     LOG.info("Installing required packages...")
     subprocess.run(
-        ["pip", "install", "-q", "datasets", "einops", "transformers==4.42.4", "tqdm"],
+        ["pip", "install", "-q", "numpy<2", "datasets", "einops", "transformers==4.42.4", "tqdm"],
         check=True,
         capture_output=True,
         text=True,
@@ -124,11 +124,18 @@ def evaluate_sample_with_nvembed(sample: dict[str, Any], model_name: str = "nvid
     choices = sample.get("choices", [])
     expected_answer = sample.get("expected_answer", "")
 
-    # Fail fast if data is malformed - this indicates a pipeline error
+    # Empty model outputs are valid failed predictions. Keep malformed-data
+    # checks strict, but do not let one blank generation abort the whole eval.
     if not generation:
-        raise ValueError(
-            f"Sample missing generation field or has empty generation. Sample ID: {sample.get('id', 'unknown')}"
+        sample.update(
+            {
+                "nvembed_matched_choice": "",
+                "nvembed_confidence": 0.0,
+                "is_correct": False,
+                "nvembed_error": "empty_generation",
+            }
         )
+        return sample
 
     if not choices:
         raise ValueError(

nemo_skills/inference/generate.py

@@ -912,11 +912,27 @@ def restore_async_order(self):
         with open(self.cfg.output_file + "-async", "rt", encoding="utf-8") as fin:
             generations = [json.loads(line) for line in fin]
 
-        ordered_generations = [None] * len(generations)
+        if not generations:
+            ordered_generations = []
+        else:
+            max_position = max(gen_dict[self.cfg.async_position_key] for gen_dict in generations)
+            ordered_generations = [None] * (max_position + 1)
         for gen_dict in generations:
             async_pos = gen_dict.pop(self.cfg.async_position_key)
+            if async_pos < 0:
+                raise ValueError(f"Invalid async output position {async_pos}")
             ordered_generations[async_pos] = gen_dict
 
+        missing_positions = [idx for idx, gen_dict in enumerate(ordered_generations) if gen_dict is None]
+        if missing_positions:
+            preview = ", ".join(map(str, missing_positions[:20]))
+            if len(missing_positions) > 20:
+                preview += ", ..."
+            raise RuntimeError(
+                f"Missing async outputs for {len(missing_positions)} positions: {preview}. "
+                "Refusing to write null rows to the merged output."
+            )
+
         with open(self.cfg.output_file, "wt", encoding="utf-8") as fout:
             for gen_dict in ordered_generations:
                 fout.write(json.dumps(gen_dict) + "\n")

nemo_skills/inference/model/vllm_multimodal.py

@@ -321,6 +321,14 @@ def content_text_to_list(self, message: dict) -> dict:
         content = result["content"]
         if isinstance(content, str):
             result["content"] = [{"type": "text", "text": content}]
+        elif isinstance(content, dict):
+            if content.get("type") == "text" and "text" in content:
+                result["content"] = [content]
+            elif "text" in content:
+                text = content["text"]
+                result["content"] = [] if text is None else [{"type": "text", "text": str(text)}]
+            else:
+                raise TypeError(f"Unexpected content dict keys: {sorted(content)}")
         elif not isinstance(content, list):
             raise TypeError(f"Unexpected content type: {type(content)}")
 
@@ -433,6 +441,14 @@ async def _generate_with_chunking(
                     content = msg_copy["content"]
                     if isinstance(content, str):
                         text_content = [{"type": "text", "text": content}]
+                    elif isinstance(content, dict):
+                        if content.get("type") == "text" and "text" in content:
+                            text_content = [content]
+                        elif "text" in content:
+                            text = content["text"]
+                            text_content = [] if text is None else [{"type": "text", "text": str(text)}]
+                        else:
+                            raise TypeError(f"Unexpected content dict keys: {sorted(content)}")
                     else:
                         text_content = content