feat(benchmark): add family-local qualification (#1204)

* feat(benchmark): add family-local qualification

Auto-discover family-owned Accuracy and Performance cases through pytest. Exercise public TRTMC commands, keep benchmark definitions shared, and support family-specific reference environments.

Signed-off-by: chaofengw <chaofengw@nvidia.com>

* refactor(benchmark): separate internal qualification

Keep trtmc-bench as the public user tool while moving model discovery, datasets, reference environments, metrics, and reports into a repository-only driver.

Generate candidate descriptors from public build inputs so internal qualification does not depend on the E2E manifest registry. Load that registry lazily for explicit descriptors and retain compiled-to-eager fallback evidence.

Signed-off-by: chaofengw <chaofengw@nvidia.com>

* fix(benchmark): adapt qualification to Task contracts

Expose repository Python packages to internal benchmark subprocesses so source-tree bundle inspection works. Keep explicit reference timing in shared benchmark definitions and carry it into generated performance suites required by the semantic Task API.

Signed-off-by: chaofengw <chaofengw@nvidia.com>

* fix(benchmark): forward runtime to bundle preparation

Pass the selected runtime root to the user-facing benchmark during Accuracy bundle preparation. This lets source-tree qualification use the exact native CLI for bundle inspection before candidate execution.

Signed-off-by: chaofengw <chaofengw@nvidia.com>

---------

Signed-off-by: chaofengw <chaofengw@nvidia.com>

Author: chaofengw-nv

apps/benchmark/native/benchmark_worker.cpp

@@ -647,7 +647,8 @@ Json run_solve(trtmc::ITask& task, const Json& request, const Timing& timing) {
         [](const trtmc::ForecastResult& result) {
             return Json{{"windows", 1},
                         {"forecast_elements", result.values.size()},
-                        {"shape", result.shape}};
+                        {"shape", result.shape},
+                        {"values", result.values}};
         });
 }
 

apps/benchmark/performance/README.md

@@ -28,6 +28,16 @@ trtmc-bench run --model distilgpt2 --runtime-root /opt/trtmc/lib -o results/dist
 trtmc-bench run apps/benchmark/example.yaml -o results/example
 ```
 
+Without another workload, `--model` uses that model's E2E testcase. To benchmark
+user input, pass public Task request JSON or JSONL. Each JSONL row may be a raw
+request or `{"name": "case-name", "request": {...}}`; relative `*_path` values
+are resolved from the data file.
+
+```bash
+trtmc-bench run --model gpt2-125m --data requests.jsonl \
+  --runtime-root /opt/trtmc/lib -o results/gpt2-data
+```
+
 Missing bundles are built through the public build command and cached. Pass
 `--no-build` when every selected bundle must already exist.
 
@@ -317,3 +327,30 @@ Install benchmark-only dependencies without adding them to a model family:
 ```bash
 python -m pip install -r apps/benchmark/performance/requirements.txt
 ```
+
+## Internal Accuracy and Performance qualification
+
+The installed `trtmc-bench` remains a user application. Repository CI and QA
+use the separate, non-packaged `tools/model_benchmark.py` driver. A family opts
+in by adding one YAML file under `families/<family>/tests/benchmark/`; the file
+may contain multiple `accuracy` and `performance` cases. Families without that
+file, including L0-only models, are not discovered.
+
+Run every discovered case or select an exact model:
+
+```bash
+python3 tools/model_benchmark.py list
+python3 tools/model_benchmark.py run --all --kind performance \
+  --runtime-root /opt/trtmc/lib --worker /opt/trtmc/bin/trtmc_benchmark_worker
+python3 tools/model_benchmark.py run --model gpt2-125m --kind accuracy \
+  --dataset mmlu-five-shot=/data/mmlu_dataset.json \
+  --runtime-root /opt/trtmc/lib --worker /opt/trtmc/bin/trtmc_benchmark_worker
+```
+
+Manual or restricted datasets are supplied as `--dataset ID=PATH`; public
+download definitions may instead materialize into `--data-root`. Model files do
+not select a GPU. Both Accuracy and Performance invoke the installed
+`trtmc-bench` through its public bundle and Task path. Performance tries the
+declared compiled reference first and uses eager only when that reference fails
+to execute. A completed reference whose output disagrees with the candidate is
+a failed contract and never triggers fallback.

apps/benchmark/performance/baselines/task_reference.py

@@ -110,6 +110,7 @@ class Session:
     timing_scope: str = "task-model-call-wall"
     input_preparation_included: bool = False
     asset_loading_included: bool = False
+    compile_evidence: dict[str, Any] | None = None
 
 
 def build_parser() -> argparse.ArgumentParser:
@@ -125,7 +126,9 @@ def build_parser() -> argparse.ArgumentParser:
     parser.add_argument("--adapter-options-json", default="{}")
     parser.add_argument("--timing-contract-json", default="{}")
     parser.add_argument("--precision", required=True, choices=("fp16", "fp32", "bf16"))
-    parser.add_argument("--mode", required=True, choices=("hf-eager", "pytorch-eager"))
+    parser.add_argument(
+        "--mode", required=True, choices=("hf-eager", "pytorch-eager", "torch-compile")
+    )
     parser.add_argument("--padding", default="longest")
     parser.add_argument("--trust-remote-code", action="store_true")
     parser.add_argument("--local-files-only", action="store_true")
@@ -285,10 +288,25 @@ def _tensor_summary(value: Any) -> dict[str, Any]:
     }
 
 
+def _flatten_tensor_values(value: Any) -> list[float]:
+    flattened: list[float] = []
+
+    def visit(item: Any) -> None:
+        if isinstance(item, (list, tuple)):
+            for child in item:
+                visit(child)
+        else:
+            flattened.append(float(item))
+
+    visit(value.detach().float().cpu().tolist())
+    return flattened
+
+
 def _forecast_summary(
     value: Any, task: str, quantile_levels: Sequence[float] = ()
 ) -> dict[str, Any]:
     summary = _tensor_summary(value)
+    summary["values"] = _flatten_tensor_values(value)
     if task not in {"series_to_point_forecast", "series_to_quantile_forecast"}:
         return summary
     shape = summary["shape"]
@@ -1352,6 +1370,9 @@ def _load_timeseries(
         chronos_options = _processor_kwargs(arguments)
         chronos_options.update({"device_map": str(device), "dtype": dtype})
         model = ChronosBoltPipeline.from_pretrained(arguments.model, **chronos_options)
+        compile_evidence = None
+        if arguments.mode == "torch-compile":
+            compile_evidence = _compile_forward(model.model)
         raw = _numeric_values(request, "past_values")
         observed = _observed_values(request, len(raw))
         context = torch.tensor([value if mask > 0 else float("nan")
@@ -1367,7 +1388,7 @@ def invoke() -> Mapping[str, Any]:
             quantiles = model.model.config.chronos_config["quantiles"]
             return _forecast_summary(value, task_id, quantiles)
 
-        return Session(invoke, "chronos")
+        return Session(invoke, "chronos", compile_evidence=compile_evidence)
 
     config = transformers.AutoConfig.from_pretrained(
         arguments.model, **_processor_kwargs(arguments)
@@ -2288,18 +2309,60 @@ def _synchronize() -> None:
         return
 
 
+def _compile_forward(model: Any) -> dict[str, Any]:
+    import torch
+    from torch._dynamo.backends.registry import lookup_backend
+
+    evidence = {"compiled_graph_count": 0}
+    inductor = lookup_backend("inductor")
+
+    def compile_graph(graph: Any, inputs: Any, **options: Any) -> Any:
+        compiled = inductor(graph, inputs, **options)
+        evidence["compiled_graph_count"] += 1
+        return compiled
+
+    model.forward = torch.compile(
+        model.forward,
+        backend=compile_graph,
+        fullgraph=False,
+        dynamic=False,
+    )
+    evidence.update(
+        {
+            "api": "torch.compile",
+            "target": "model.forward",
+            "backend": "inductor",
+            "mode": "default",
+            "fullgraph": False,
+            "dynamic": False,
+            "applied": True,
+        }
+    )
+    return evidence
+
+
 def _measure(session: Session, warmup: int, iterations: int) -> tuple[list[float], dict[str, Any]]:
     output: Mapping[str, Any] = {}
     for _ in range(warmup):
         output = session.invoke()
         _synchronize()
+    compiled_graphs = None
+    if session.compile_evidence is not None:
+        compiled_graphs = int(session.compile_evidence["compiled_graph_count"])
+        if compiled_graphs < 1:
+            raise RuntimeError("warmup did not execute a compiled graph")
     samples = []
     for _ in range(iterations):
         _synchronize()
         started = time.perf_counter()
         output = session.invoke()
         _synchronize()
         samples.append((time.perf_counter() - started) * 1000.0)
+    if (
+        compiled_graphs is not None
+        and int(session.compile_evidence["compiled_graph_count"]) != compiled_graphs
+    ):
+        raise RuntimeError("model compilation occurred inside timed samples")
     return samples, dict(output)
 
 
@@ -2715,8 +2778,13 @@ def run(arguments: argparse.Namespace) -> int:
     if arguments.warmup < 0 or arguments.iterations <= 0:
         raise ValueError("warmup must be non-negative and iterations must be positive")
     expected_mode = "pytorch-eager" if arguments.adapter in PYTORCH_ADAPTERS else "hf-eager"
-    if arguments.mode != expected_mode:
-        raise ValueError(f"adapter {arguments.adapter} requires mode {expected_mode}")
+    supported_modes = {expected_mode}
+    if arguments.adapter == "pytorch-timeseries" and arguments.family == "chronos_bolt":
+        supported_modes.add("torch-compile")
+    if arguments.mode not in supported_modes:
+        raise ValueError(
+            f"adapter {arguments.adapter} requires one of {sorted(supported_modes)}"
+        )
     request = flatten_config(_json_object(arguments.request_json, "--request-json"))
     options = _json_object(arguments.adapter_options_json, "--adapter-options-json")
     configured_timing = _json_object(arguments.timing_contract_json, "--timing-contract-json")
@@ -2729,6 +2797,7 @@ def run(arguments: argparse.Namespace) -> int:
         expected_timing = {name: declared[name] for name in fields}
     load_started = time.perf_counter()
     load_seconds: float | None = None
+    compile_evidence: dict[str, Any] | None = None
     if arguments.adapter == "upstream-elf":
         samples, output_summary, framework, timing_scope, input_included, asset_included = _run_elf(
             arguments, request, options
@@ -2762,6 +2831,7 @@ def run(arguments: argparse.Namespace) -> int:
         ) = _run_sana_wm(arguments, request, options)
     else:
         session = LOADERS[arguments.adapter](arguments, request, options)
+        compile_evidence = session.compile_evidence
         load_seconds = time.perf_counter() - load_started
         framework = session.framework
         timing_scope = session.timing_scope
@@ -2821,8 +2891,8 @@ def run(arguments: argparse.Namespace) -> int:
         "precision": arguments.precision,
         "padding": arguments.padding,
         "experts_implementation": None,
-        "compile_scope": None,
-        "compile_evidence": None,
+        "compile_scope": "model.forward" if arguments.mode == "torch-compile" else None,
+        "compile_evidence": compile_evidence,
         "timing_scope": timing_scope,
         "input_preparation_included": input_included,
         "asset_loading_included": asset_included,
@@ -2839,6 +2909,7 @@ def run(arguments: argparse.Namespace) -> int:
             "input_preparation_included": input_included,
             "asset_loading_included": asset_included,
             "model_load_excluded": True,
+            "compile_excluded": True,
             "warmup_excluded": True,
             "output_materialization_included": True,
         },
@@ -2856,6 +2927,9 @@ def run(arguments: argparse.Namespace) -> int:
         "environment": _environment(),
         "finished_at": datetime.now(timezone.utc).isoformat(),
     }
+    if compile_evidence is not None:
+        compile_evidence["warmup_completed"] = True
+        compile_evidence["timed_callable_uses_compiled_target"] = True
     if not all(math.isfinite(float(value)) and float(value) > 0.0 for value in samples):
         raise RuntimeError("reference produced an invalid timing sample")
     arguments.output.parent.mkdir(parents=True, exist_ok=True)

apps/benchmark/trtmc_benchmark/catalog.py

@@ -41,12 +41,16 @@ def default_manifest_root() -> Path:
 
 class ManifestCatalog:
     def __init__(self, root: Path | None = None) -> None:
-        self.root = (root or default_manifest_root()).expanduser().resolve()
+        self.root = root.expanduser().resolve() if root is not None else None
+
+    def _root(self) -> Path:
+        root = self.root or default_manifest_root()
+        if not root.is_dir():
+            raise BenchmarkError(f"manifest root does not exist: {root}")
+        return root
 
     def _manifest_paths(self) -> tuple[Path, ...]:
-        if not self.root.is_dir():
-            raise BenchmarkError(f"manifest root does not exist: {self.root}")
-        return tuple(sorted(self.root.glob("*/tests/manifests/*.json")))
+        return tuple(sorted(self._root().glob("*/tests/manifests/*.json")))
 
     def entries(self) -> tuple[CatalogEntry, ...]:
         entries: list[CatalogEntry] = []
@@ -121,7 +125,7 @@ def resolve(self, selector: str) -> ModelDescriptor:
             if selector in {path.stem, model.name, model.hf_id}:
                 matches.append(model)
         if not matches:
-            raise BenchmarkError(f"unknown model {selector!r} under {self.root}")
+            raise BenchmarkError(f"unknown model {selector!r} under {self._root()}")
         if len(matches) != 1:
             paths = ", ".join(str(model.manifest_path) for model in matches)
             raise BenchmarkError(f"ambiguous model {selector!r}: {paths}")

apps/benchmark/trtmc_benchmark/cli.py

@@ -42,6 +42,13 @@ def build_parser() -> argparse.ArgumentParser:
     run.add_argument("--rebuild", action="store_true")
     run.add_argument("--manifest-root", type=Path)
     run.add_argument("--case", action="append", default=[])
+    run.add_argument(
+        "--data",
+        action="append",
+        default=[],
+        metavar="[MODEL=]PATH",
+        help="Use public Task request JSON or JSONL instead of the manifest testcase input",
+    )
     run.add_argument("--operation")
     run.add_argument("--task", dest="selected_task", help="Select a Task bound by the model")
     run.add_argument("--set", dest="sets", action="append", default=[], metavar="FIELD=VALUE")
@@ -204,6 +211,7 @@ def _resolve_cases(
 ) -> tuple[ResolvedCase, ...]:
     entries = _model_entries(arguments.model, spec)
     bundles = _path_arguments(arguments.bundle)
+    data_paths = _path_arguments(arguments.data)
     configured_roots = spec.get("bundle_roots", [])
     if not isinstance(configured_roots, list):
         raise BenchmarkError("bundle_roots must be a list")
@@ -225,6 +233,8 @@ def _resolve_cases(
         selector = str(entry["model"])
         model = catalog.resolve(selector)
         explicit = _entry_path(entry.get("bundle"), selector, bundles, len(entries))
+        data_path = _entry_path(entry.get("data"), selector, data_paths, len(entries))
+        data_requests = _load_data_requests(data_path) if data_path is not None else ()
         bundle = find_bundle(model, explicit=explicit, roots=roots)
         if bundle is None:
             bundle = builder.provisional_path(model)
@@ -258,7 +268,24 @@ def _resolve_cases(
                 name=display, runtime_root=runtime_root, bundle_is_explicit=explicit is not None
             )
             sweeps = _merge_sweeps(case_spec.get("sweep", {}), cli_sweeps)
-            resolved.extend(expand_sweeps(base, sweeps))
+            swept = expand_sweeps(base, sweeps)
+            if not data_requests:
+                resolved.extend(swept)
+                continue
+            for swept_case in swept:
+                for data_name, request in data_requests:
+                    merged = {**swept_case.request, **request}
+                    sources = {
+                        **swept_case.sources,
+                        **{name: f"data file {data_path}" for name in request},
+                    }
+                    resolved.append(
+                        swept_case.with_values(
+                            name=f"{swept_case.name}/{data_name}",
+                            request=merged,
+                            sources=sources,
+                        )
+                    )
     if selected_names and arguments.config:
         missing = selected_names - matched_names
         if missing:
@@ -385,6 +412,67 @@ def _path_arguments(values: list[str]) -> dict[str, Path]:
     return result
 
 
+def _load_data_requests(path: Path) -> tuple[tuple[str, dict[str, Any]], ...]:
+    path = _absolute(path)
+    if not path.is_file():
+        raise BenchmarkError(f"benchmark data does not exist: {path}")
+    try:
+        text = path.read_text(encoding="utf-8")
+    except OSError as error:
+        raise BenchmarkError(f"cannot read benchmark data {path}: {error}") from error
+    try:
+        parsed = json.loads(text)
+    except json.JSONDecodeError:
+        values = []
+        for line_number, line in enumerate(text.splitlines(), start=1):
+            if not line.strip():
+                continue
+            try:
+                values.append(json.loads(line))
+            except json.JSONDecodeError as error:
+                raise BenchmarkError(
+                    f"benchmark data {path}:{line_number} is not valid JSON: {error}"
+                ) from error
+    else:
+        values = parsed if isinstance(parsed, list) else [parsed]
+    if not values:
+        raise BenchmarkError(f"benchmark data contains no requests: {path}")
+
+    requests = []
+    seen: set[str] = set()
+    for index, value in enumerate(values, start=1):
+        if not isinstance(value, Mapping):
+            raise BenchmarkError(f"benchmark data request {index} must be an object")
+        raw_request = value.get("request", value)
+        if not isinstance(raw_request, Mapping):
+            raise BenchmarkError(f"benchmark data request {index}.request must be an object")
+        raw_name = value.get("name") if "request" in value else None
+        name = str(raw_name or f"data-{index:04d}")
+        if not name or name in seen:
+            raise BenchmarkError(f"benchmark data request name is empty or repeated: {name!r}")
+        seen.add(name)
+        request = _resolve_data_paths(dict(raw_request), path.parent)
+        requests.append((name, request))
+    return tuple(requests)
+
+
+def _resolve_data_paths(value: Any, root: Path) -> Any:
+    if isinstance(value, Mapping):
+        result = {}
+        for name, nested in value.items():
+            if isinstance(nested, str) and str(name).endswith("_path"):
+                candidate = Path(nested).expanduser()
+                result[str(name)] = str(
+                    candidate if candidate.is_absolute() else (root / candidate).resolve()
+                )
+            else:
+                result[str(name)] = _resolve_data_paths(nested, root)
+        return result
+    if isinstance(value, list):
+        return [_resolve_data_paths(item, root) for item in value]
+    return value
+
+
 def _entry_path(
     configured: Any,
     selector: str,