feat(benchmark): add reproducible NeMo Gym routing comparison

benchmark/README.md

@@ -1,7 +1,12 @@
 <!-- SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. -->
 <!-- SPDX-License-Identifier: Apache-2.0 -->
 
-# Harbor Benchmarks
+# Switchyard Benchmarks
+
+For a fixed-model versus routed comparison with NeMo Gym, see the
+[NeMo Gym example](nemo_gym/README.md).
+
+## Harbor
 
 Use this guide to run Harbor Terminal-Bench Lite from a fresh Switchyard clone.
 It covers the two smoke paths most people need first:

benchmark/nemo_gym/README.md

@@ -0,0 +1,121 @@
+<!-- SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. -->
+<!-- SPDX-License-Identifier: Apache-2.0 -->
+
+# Compare Switchyard routes with NeMo Gym
+
+This example evaluates the same MMLU-Redux tasks through two routes in [`routes.toml`](routes.toml):
+
+- `strong-only` always uses the strong target.
+- `policy-model` uses the efficient model as a classifier, then routes to the efficient or strong
+  target.
+
+Gym owns the tasks, verifier, rewards, and rollout capture. Switchyard owns routing, model calls,
+and routing statistics. Gym's `switchyard_model` adapter joins them over HTTP, so neither project
+imports the other's core library. The script starts a fresh server from this Switchyard checkout
+for each condition to isolate its statistics.
+
+## Prerequisites
+
+Install Python 3.13.14 or newer, Rust 1.96.1, Cargo, `curl`, Git, and `uv`. You also
+need credentials for the model endpoints configured in [`routes.toml`](routes.toml). The bundled
+configuration uses two NVIDIA-hosted models and reads `NVIDIA_API_KEY`; create a key on the
+[NVIDIA API key page](https://build.nvidia.com/settings/api-keys). Switchyard can use another
+supported provider or compatible endpoint by changing the LLM client and targets in the TOML.
+
+Use this clean, pinned Gym checkout:
+
+```bash
+git clone https://github.com/NVIDIA-NeMo/Gym.git /path/to/Gym
+git -C /path/to/Gym checkout e044a8ca795ece2c69b053d30c0a8dea7fa3b9f3
+cd /path/to/Gym
+uv sync --frozen --no-dev
+```
+
+## Run
+
+From the Switchyard repository:
+
+```bash
+export NVIDIA_API_KEY="nvapi-..."
+export GYM_DIR=/path/to/Gym
+bash benchmark/nemo_gym/run.sh
+```
+
+To use another provider, copy `routes.toml`, retain the `strong-only` and `policy-model` route IDs,
+and update its LLM client, targets, and `api_key_env`. Export each credential named by the
+deployment, then point the runner at that TOML:
+
+```bash
+export OPENAI_API_KEY="..."
+export GYM_DIR=/path/to/Gym
+SWITCHYARD_CONFIG=/path/to/routes.toml bash benchmark/nemo_gym/run.sh
+```
+
+Switchyard validates the `api_key_env` entries when it loads the deployment. An unauthenticated
+endpoint does not need a credential variable.
+
+Run `bash benchmark/nemo_gym/run.sh --help` to see the optional environment overrides.
+
+The default run evaluates five tasks and writes a timestamped directory under
+`benchmark/nemo_gym/results/`. It is a workflow smoke test, not a benchmark result. The first run
+also downloads and prepares the dataset. Gym starts its serving environment for each condition.
+For a larger workflow check:
+
+```bash
+LIMIT=100 REPEATS=3 CONCURRENCY=4 RESULTS_DIR=/tmp/routing-eval \
+  bash benchmark/nemo_gym/run.sh
+```
+
+`LIMIT` takes the first tasks in Gym's prepared file, so a small limit is not a representative
+sample. The example above can make roughly 600 answer calls and 300 classifier calls before
+retries. Use a recorded stratified subset or the full benchmark for representative results. The
+script refuses to overwrite a result directory.
+
+## Read the result
+
+`comparison.json` pairs completed rollouts by task and repeat index and verifies identical inputs.
+It reports:
+
+- mean reward and routed-versus-baseline wins, ties, and losses;
+- missing and unpaired completions, so failed tasks are not silently scored or discarded;
+- paired answer-model tokens and endpoint latency from Gym;
+- classifier tokens, answer and classifier latency, routing overhead, model totals, and
+  classifier fail-open counts from Switchyard.
+
+Gym's endpoint latency covers the whole routed request. Switchyard routing overhead includes the
+classifier call, so routing overhead and classifier latency overlap and must not be added together.
+Answer and classifier tokens remain separate. Switchyard totals are condition-wide, while quality
+and answer usage are paired only over tasks completed by both conditions.
+
+The result directory contains the following artifacts. Gym may also write its best-effort
+`switchyard-stats.json` wrapper when its shutdown hook completes:
+
+| Artifact | Meaning |
+|---|---|
+| `comparison.json` | Paired quality and usage comparison for both conditions. |
+| `<condition>/rollouts.jsonl` | Completed Gym rollouts and rewards. |
+| `<condition>/rollouts_materialized_inputs.jsonl` | Exact task/repeat inputs used by Gym. |
+| `<condition>/rollouts_failures.jsonl` | Rollouts that failed before producing a scored result. |
+| `<condition>/rollouts_aggregate_metrics.json` | Gym's aggregate benchmark metrics. |
+| `<condition>/switchyard-condition.json` | Route and attached-proxy provenance written by Gym. |
+| `<condition>/switchyard-stats-raw.json` | Raw `/v1/stats` captured while the proxy is alive. |
+| `<condition>/switchyard-stats.json` | Best-effort Gym wrapper around `/v1/stats`. |
+| `<condition>/switchyard-metrics.prom` | Prometheus metrics, including classifier fail-open reasons. |
+| `<condition>/model-calls/` | Per-rollout model-call captures, including the served model. |
+| `<condition>/routes.toml` | Exact Switchyard deployment copied for the condition. |
+| `<condition>/switchyard.log` | Switchyard server output for diagnosis. |
+
+Gym excludes failure-sidecar rows from its reward calculation. `comparison.json` reports those
+rollouts as missing or unpaired instead of treating them as zero-reward answers.
+
+Keep the materialized inputs because Gym's MMLU-Redux loader does not pin a Hugging Face dataset
+revision. The workflow and evaluated inputs are reproducible; hosted model outputs, token counts,
+and latency can still change between runs. Run from a clean Switchyard checkout when you need to
+reproduce the exact server build. Dirty builds are labeled `-dirty`, but the source diff is not
+archived with the result.
+
+See the
+[NeMo Gym Switchyard model-server documentation](https://docs.nvidia.com/nemo/gym/main/model-server/switchyard/)
+for other benchmarks and hosted mode, and the
+[LLM classifier guide](../../docs/routing_algorithms/llm_classifier_routing.md) for the routing
+policy used here.

benchmark/nemo_gym/compare.py

@@ -0,0 +1,225 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Compare paired quality and usage for two attached NeMo Gym runs.
+
+Per-rollout measurements include only shared task/repeat pairs. Switchyard proxy
+statistics remain condition-wide.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import math
+from pathlib import Path
+from typing import Any
+
+RolloutKey = tuple[int, int]
+
+
+def _read_object(path: Path) -> dict[str, Any]:
+    try:
+        value = json.loads(path.read_text(encoding="utf-8"))
+    except json.JSONDecodeError as error:
+        raise ValueError(f"invalid JSON in {path}: {error.msg}") from error
+    if not isinstance(value, dict):
+        raise ValueError(f"expected a JSON object in {path}")
+    return value
+
+
+def _read_jsonl(path: Path) -> dict[RolloutKey, dict[str, Any]]:
+    rows: dict[RolloutKey, dict[str, Any]] = {}
+    with path.open(encoding="utf-8") as lines:
+        for line_number, line in enumerate(lines, start=1):
+            try:
+                row = json.loads(line)
+            except json.JSONDecodeError as error:
+                raise ValueError(f"{path}:{line_number}: invalid JSON: {error.msg}") from error
+            if not isinstance(row, dict):
+                raise ValueError(f"{path}:{line_number}: expected a JSON object")
+            try:
+                task_index = row["_ng_task_index"]
+                rollout_index = row["_ng_rollout_index"]
+            except KeyError as error:
+                raise ValueError(f"{path}:{line_number}: missing {error.args[0]}") from error
+            if type(task_index) is not int or type(rollout_index) is not int:
+                raise ValueError(f"{path}:{line_number}: rollout indices must be integers")
+            key = (task_index, rollout_index)
+            if key in rows:
+                raise ValueError(f"duplicate rollout key {key} in {path}")
+            rows[key] = row
+    return rows
+
+
+def _fail_opens(run_dir: Path) -> dict[str, int]:
+    """Aggregate Switchyard's fixed classifier fail-open metric by reason."""
+    path = run_dir / "switchyard-metrics.prom"
+    if not path.exists():
+        raise ValueError(f"missing {path}; capture /metrics before stopping the proxy")
+
+    counts: dict[str, int] = {}
+    prefix = "switchyard_classifier_fail_open_total{"
+    marker = 'reason="'
+    for line in path.read_text().splitlines():
+        if not line.startswith(prefix) or marker not in line:
+            continue
+        reason = line.split(marker, 1)[1].split('"', 1)[0]
+        value = int(float(line.rsplit(maxsplit=1)[1]))
+        counts[reason] = counts.get(reason, 0) + value
+    return counts
+
+
+def _models(stats: dict[str, Any]) -> dict[str, dict[str, int | float]]:
+    return {
+        name: {
+            "calls": model["calls"],
+            "errors": model["errors"],
+            "tokens": model["total_tokens"],
+            "model_call_latency_mean_ms": model["model_call_latency"]["avg_ms"],
+        }
+        for name, model in stats.items()
+    }
+
+
+def _reward(row: dict[str, Any], key: RolloutKey) -> float:
+    try:
+        value = float(row["reward"])
+    except (KeyError, TypeError, ValueError) as error:
+        raise ValueError(f"rollout {key} has no numeric reward") from error
+    if not math.isfinite(value):
+        raise ValueError(f"rollout {key} has a non-finite reward")
+    return value
+
+
+def _condition_summary(
+    run_dir: Path,
+    rows: dict[RolloutKey, dict[str, Any]],
+    shared: set[RolloutKey],
+) -> dict[str, Any]:
+    """Combine paired rollout measurements with condition-wide proxy statistics."""
+    stats = _read_object(run_dir / "switchyard-stats-raw.json")
+    rewards = [_reward(rows[key], key) for key in sorted(shared)]
+    try:
+        captures = [rows[key]["ng_model_call_capture"]["metrics"] for key in sorted(shared)]
+    except (KeyError, TypeError) as error:
+        raise ValueError(
+            "rollout is missing ng_model_call_capture.metrics; enable Gym observability"
+        ) from error
+
+    try:
+        answer_model_tokens = sum(int(item["tokens_total"]) for item in captures)
+        endpoint_latency_mean_ms = sum(float(item["latency_total_ms"]) for item in captures) / len(
+            captures
+        )
+    except (KeyError, TypeError, ValueError) as error:
+        raise ValueError(
+            "rollout capture metrics must include numeric tokens and latency"
+        ) from error
+
+    classifier = stats["classifier"]
+    return {
+        "paired": {
+            "mean_reward": sum(rewards) / len(rewards),
+            "answer_model_tokens": answer_model_tokens,
+            "endpoint_latency_mean_ms": endpoint_latency_mean_ms,
+        },
+        "condition_totals": {
+            "classifier_tokens": classifier["total_tokens"]["total"],
+            "routing_overhead_mean_ms": stats["routing_overhead"]["avg_ms"],
+            "classifier_fail_opens": _fail_opens(run_dir),
+            "answer_models": _models(stats["models"]),
+            "classifier_models": _models(classifier["models"]),
+        },
+    }
+
+
+def compare(baseline_dir: Path, routed_dir: Path) -> dict[str, Any]:
+    """Return paired quality and usage summaries for two result directories."""
+    conditions = {
+        "baseline": _read_object(baseline_dir / "switchyard-condition.json"),
+        "routed": _read_object(routed_dir / "switchyard-condition.json"),
+    }
+    if any(condition.get("mode") != "attached" for condition in conditions.values()):
+        raise ValueError("this example compares attached Switchyard runs")
+    provenances = {
+        name: condition.get("proxy_provenance") for name, condition in conditions.items()
+    }
+    required_revisions = ("gym_revision", "switchyard_revision")
+    if any(
+        not isinstance(provenance, dict)
+        or any(
+            not isinstance(provenance.get(key), str) or not provenance[key]
+            for key in required_revisions
+        )
+        for provenance in provenances.values()
+    ):
+        raise ValueError("the runs have incomplete Switchyard provenance")
+    provenance = provenances["baseline"]
+    if provenance != provenances["routed"]:
+        raise ValueError("the runs used different or incomplete Switchyard provenance")
+    if (baseline_dir / "routes.toml").read_bytes() != (routed_dir / "routes.toml").read_bytes():
+        raise ValueError("the runs used different Switchyard deployments")
+
+    runs = {
+        "baseline": _read_jsonl(baseline_dir / "rollouts.jsonl"),
+        "routed": _read_jsonl(routed_dir / "rollouts.jsonl"),
+    }
+    inputs = {
+        "baseline": _read_jsonl(baseline_dir / "rollouts_materialized_inputs.jsonl"),
+        "routed": _read_jsonl(routed_dir / "rollouts_materialized_inputs.jsonl"),
+    }
+    expected = set(inputs["baseline"])
+    if expected != set(inputs["routed"]):
+        raise ValueError("the runs materialized different rollout keys")
+    # Matching indices are insufficient because the source dataset revision is not pinned.
+    for key in sorted(expected):
+        if inputs["baseline"][key] != inputs["routed"][key]:
+            raise ValueError(f"paired rollout {key} used different materialized inputs")
+    for name, rows in runs.items():
+        if extra := set(rows) - expected:
+            raise ValueError(f"{name} completed unknown rollout keys: {sorted(extra)}")
+
+    shared = set(runs["baseline"]) & set(runs["routed"])
+    if not shared:
+        raise ValueError("the runs have no completed rollouts in common")
+    rewards = {
+        name: {key: _reward(rows[key], key) for key in shared} for name, rows in runs.items()
+    }
+    return {
+        "provenance": provenance,
+        "routes": {name: condition["route"] for name, condition in conditions.items()},
+        "coverage": {
+            "expected": len(expected),
+            "paired": len(shared),
+            "completed": {name: len(rows) for name, rows in runs.items()},
+            "unpaired": {name: len(set(rows) - shared) for name, rows in runs.items()},
+            "missing": {name: len(expected - set(rows)) for name, rows in runs.items()},
+        },
+        "routed_vs_baseline": {
+            "wins": sum(rewards["routed"][key] > rewards["baseline"][key] for key in shared),
+            "ties": sum(rewards["routed"][key] == rewards["baseline"][key] for key in shared),
+            "losses": sum(rewards["routed"][key] < rewards["baseline"][key] for key in shared),
+        },
+        "baseline": _condition_summary(baseline_dir, runs["baseline"], shared),
+        "routed": _condition_summary(routed_dir, runs["routed"], shared),
+    }
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("baseline_dir", type=Path, help="fixed-model condition directory")
+    parser.add_argument("routed_dir", type=Path, help="routed condition directory")
+    args = parser.parse_args()
+    try:
+        result = compare(args.baseline_dir, args.routed_dir)
+    except KeyError as error:
+        parser.exit(1, f"{parser.prog}: error: missing expected field {error}\n")
+    except (OSError, TypeError, ValueError) as error:
+        parser.exit(1, f"{parser.prog}: error: {error}\n")
+    print(json.dumps(result, indent=2))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())