From 746c6e5a4ca5d2c39edb06c55224bddd3890f7f3 Mon Sep 17 00:00:00 2001 From: Gaia Di Lorenzo Date: Fri, 31 Jul 2026 14:23:37 +0200 Subject: [PATCH 01/14] feat(insights): add Harbor testbed adapter Signed-off-by: Gaia Di Lorenzo --- plugins/nemo-insights/testbed/README.md | 23 + plugins/nemo-insights/testbed/adapters.py | 435 +++++++++++++++++- plugins/nemo-insights/testbed/cli.py | 13 +- plugins/nemo-insights/testbed/otlp_ingest.py | 11 +- plugins/nemo-insights/testbed/testbeds.toml | 17 + .../tests/testbed/test_adapters.py | 211 ++++++++- .../nemo-insights/tests/testbed/test_cli.py | 24 + .../tests/testbed/test_otlp_ingest.py | 16 + .../tests/testbed/test_registry.py | 18 +- 9 files changed, 750 insertions(+), 18 deletions(-) diff --git a/plugins/nemo-insights/testbed/README.md b/plugins/nemo-insights/testbed/README.md index a547fbc625..e1cd3b6d90 100644 --- a/plugins/nemo-insights/testbed/README.md +++ b/plugins/nemo-insights/testbed/README.md @@ -13,6 +13,7 @@ uv run python -m testbed analyze all # refresh every pinned ben uv run python -m testbed list uv run python -m testbed doctor # fresh clone? run this first uv run python -m testbed run tau2-airline # produce: tau2 -> ingest -> record the run (expensive, once) +uv run python -m testbed run tau3-airline-harbor # produce: Harbor -> ingest -> record the run uv run python -m testbed analyze tau2-airline --live # analyze the recorded run's live traces (no restore) uv run python -m testbed analyze glamr --live # intake: analyze existing live traces uv run python -m testbed snapshot tau2-airline # export the subject's workspaces (read API) into a portable bundle @@ -198,16 +199,38 @@ exactly what to install/set (`✓ ready` or `✗ needs: …`). Subjects live in `testbeds.toml` — one table per subject, keyed by `type`: - `type = "intake"` — analyze an agent's existing Intake traces (config: `agent`, `workspace`, `base_url`, optional `since`). - `type = "benchmark"` — run a benchmark to produce traces, ingest them into Intake, then analyze (config: `domain`, `base_url`, `workspace`, `agent_llm`, `user_llm`, `task_split_name`, `num_trials`, `max_concurrency`, `seed`, optional `num_tasks`/`timeout`/`include_rewards`). +- `type = "harbor"` — run a Harbor dataset against an importable agent wrapper, enrich and ingest its OTLP traces, then analyze the recorded evaluation. Use exactly one of `dataset`, `dataset_ref`, or `dataset_id`. `--since` (analyze `--live`, snapshot) accepts `Nd`/`Nh`/`Nm` (days/hours/minutes) or an ISO date; `--since ''` means no lower bound (the epoch). Insights are written to `testbed/tmp/insights_.yaml`. +## Harbor benchmark (Tau3 Airline example) + +The checked-in subject runs `sierra-research/tau3-bench@1` from Harbor Hub +against the Experimentalist's Tau3 NOOA example agent. Harbor and its container +runtime are required. + +```bash +export NMP_BASE_URL=http://localhost:8080 +export INFERENCE_API_KEY=sk-... +export OPENAI_API_KEY="$INFERENCE_API_KEY" +export OPENAI_BASE_URL=https://inference-api.nvidia.com/v1 + +uv run python -m testbed run tau3-airline-harbor --base "$NMP_BASE_URL" +uv run python -m testbed analyze tau3-airline-harbor --live +``` + +The run creates an Evaluation in `canonical-tau3-airline`, associates every +Harbor trial through `nemo.experiment.id`, adds `nemo.test_case.id`, and posts +each numeric verifier reward as an evaluator result. + ## Config split: secrets in `.env`, everything else in `testbeds.toml` On startup the CLI auto-loads `testbed/.env` (gitignored) as `KEY=VALUE` lines. Keep **only secrets/endpoints** there — `INFERENCE_API_KEY` (analyst) and `OPENAI_API_KEY`/`OPENAI_API_BASE` (the proxy litellm uses for the benchmark sim LLMs). +The Harbor example agent uses `OPENAI_API_KEY`/`OPENAI_BASE_URL` instead. GLAMR live analysis additionally reads `GLAMR_INTAKE_USER` and `GLAMR_INTAKE_PASSWORD` from `.env`; `testbeds.toml` stores only those environment-variable names, never their credential values. diff --git a/plugins/nemo-insights/testbed/adapters.py b/plugins/nemo-insights/testbed/adapters.py index 54a0d7c108..c7826e8b32 100644 --- a/plugins/nemo-insights/testbed/adapters.py +++ b/plugins/nemo-insights/testbed/adapters.py @@ -2,18 +2,23 @@ # SPDX-License-Identifier: Apache-2.0 """Per-type testbed adapters that turn a subject into analyst Insights.""" +import base64 +import json import os import shutil import sys import time +from collections.abc import Iterator, Mapping from datetime import datetime, timezone from pathlib import Path -from typing import Protocol +from typing import Any, Protocol import httpx +from google.protobuf.json_format import ParseDict from nemo_insights_plugin.analyst.run import run_analyst from nemo_insights_plugin.client import make_client from nemo_platform import AsyncNeMoPlatform +from opentelemetry.proto.collector.trace.v1.trace_service_pb2 import ExportTraceServiceRequest from testbed.ingest import ( create_experiment, ensure_experiment_group, @@ -23,7 +28,7 @@ ) from testbed.intake_client import build_basic_auth_intake_client from testbed.otlp_build import session_id_for, sim_to_spans -from testbed.otlp_ingest import export_spans, post_evaluator_results, trace_id_for +from testbed.otlp_ingest import export_spans, export_trace_request, post_evaluator_results, trace_id_for from testbed.registry import Subject from testbed.tau2run import load_tasks, policy_version, read_policy, resolve_paths, run_tau2 @@ -54,7 +59,7 @@ def __init__(self, subject: Subject) -> None: def check(self) -> list[str]: """Unmet prerequisites for this subject (empty list = ready to run).""" cfg = self.subject.config - missing = [f"config key '{k}'" for k in ("agent", "workspace", "base_url") if not cfg.get(k)] + missing: list[str] = [f"config key '{k}'" for k in ("agent", "workspace", "base_url") if not cfg.get(k)] if cfg.get("auth") == "basic": missing.extend(self._missing_basic_auth()) return missing @@ -307,9 +312,431 @@ async def analyze( ) -_ADAPTERS: dict[str, type[IntakeAdapter] | type[BenchmarkAdapter]] = { +def _export_harbor_trace_files( + base_url: str, + workspace: str, + trace_dir: Path, + agent_name: str, + *, + evaluation_id: str, + agent_version: str | None = None, + client: httpx.Client | None = None, +) -> tuple[int, int, set[str]]: + """Enrich Harbor OTLP-JSONL traces and export them to Intake.""" + headers: dict[str, str] = {} + if api_key := os.environ.get("INFERENCE_API_KEY"): + headers["Authorization"] = f"Bearer {api_key}" + + owns_client = client is None + active_client = client or httpx.Client(timeout=30.0) + sent = errors = 0 + session_ids: set[str] = set() + try: + for path in sorted(trace_dir.rglob("*.jsonl")): + parsed: list[tuple[int, dict[str, Any]]] = [] + for line_number, line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1): + if not line.strip(): + continue + try: + body = json.loads(line) + except json.JSONDecodeError as exc: + print(f"Harbor trace {path.name} line {line_number}: JSON error — {exc}", file=sys.stderr) + errors += 1 + continue + if not isinstance(body, dict): + print( + f"Harbor trace {path.name} line {line_number}: expected a JSON object", + file=sys.stderr, + ) + errors += 1 + continue + parsed.append((line_number, body)) + + test_case_id, input_value, output_value, rewards = _harbor_trace_context(path, trace_dir, parsed) + evaluator_posted = False + for line_number, body in parsed: + resource_spans = body.get("resourceSpans", []) + if not isinstance(resource_spans, list): + errors += 1 + continue + session_id = _find_session_id(resource_spans) or path.stem + session_ids.add(session_id) + _inject_resource_attributes( + resource_spans, + { + "gen_ai.agent.name": agent_name, + "gen_ai.agent.id": agent_name, + "gen_ai.agent.version": agent_version, + "session.id": session_id, + "gen_ai.conversation.id": session_id, + "nemo.experiment.id": evaluation_id, + "nemo.optimizer.workspace": workspace, + }, + ) + _inject_harbor_root_attributes( + resource_spans, + { + "nemo.test_case.id": test_case_id, + "input.value": input_value, + "input.mime_type": "text/plain" if input_value else None, + "output.value": output_value, + "output.mime_type": "text/markdown" if output_value else None, + }, + ) + root_span_id = _find_harbor_root_span_id(resource_spans) + try: + request = ParseDict(body, ExportTraceServiceRequest()) + export_trace_request( + base_url, + workspace, + request, + client=active_client, + headers=headers, + ) + if root_span_id is not None and not evaluator_posted: + for reward_name, reward_value in rewards.items(): + post_evaluator_results( + base_url, + workspace, + span_id=root_span_id, + session_id=session_id, + score=reward_value, + name=reward_name, + client=active_client, + ) + evaluator_posted = True + except Exception as exc: + print(f"Harbor trace {path.name} line {line_number}: export error — {exc}", file=sys.stderr) + errors += 1 + else: + sent += 1 + finally: + if owns_client: + active_client.close() + return sent, errors, session_ids + + +def _harbor_trace_context( + path: Path, + trace_dir: Path, + parsed: list[tuple[int, dict[str, Any]]], +) -> tuple[str | None, str | None, str | None, dict[str, float]]: + """Resolve task, input, output, and verifier rewards for one Harbor trial.""" + relative = path.relative_to(trace_dir) + trial_dir = trace_dir / relative.parts[0] if len(relative.parts) > 1 else None + test_case_id = None + output_value = None + rewards: dict[str, float] = {} + if trial_dir is not None: + result_path = trial_dir / "result.json" + if result_path.is_file(): + try: + result = json.loads(result_path.read_text(encoding="utf-8")) + test_case_id = str(result["task_name"]) + raw_rewards = ( + {} if result.get("exception_info") else (result.get("verifier_result") or {}).get("rewards") or {} + ) + rewards = { + str(name): float(value) + for name, value in raw_rewards.items() + if isinstance(value, int | float) and not isinstance(value, bool) + } + except (json.JSONDecodeError, KeyError, TypeError, ValueError): + pass + artifact_root = trial_dir / "artifacts" / "logs" / "artifacts" + outputs = sorted(artifact_root.glob("output.*")) if artifact_root.is_dir() else [] + if outputs: + output_value = outputs[0].read_text(encoding="utf-8", errors="replace") + + input_value, trace_output_value = _root_values_from_trace(parsed) + if trace_output_value is not None: + output_value = trace_output_value + return test_case_id, input_value, output_value, rewards + + +def _root_values_from_trace(parsed: list[tuple[int, dict[str, Any]]]) -> tuple[str | None, str | None]: + """Return input and output values from root AGENT spans.""" + input_value = None + output_value = None + for _, body in parsed: + root = _find_harbor_root_span(body.get("resourceSpans", [])) + if root is None: + continue + attributes = {item.get("key"): item.get("value", {}) for item in root.get("attributes", [])} + if input_value is None: + input_value = attributes.get("input.value", {}).get("stringValue") + if output_value is None: + output_value = attributes.get("output.value", {}).get("stringValue") + if input_value is not None and output_value is not None: + break + return input_value, output_value + + +def _inject_harbor_root_attributes(resource_spans: list[dict[str, Any]], extra: Mapping[str, str | None]) -> None: + for span in _harbor_root_spans(resource_spans): + attributes = span.setdefault("attributes", []) + existing = {item["key"] for item in attributes} + for key, value in extra.items(): + if value is not None and key not in existing: + attributes.append({"key": key, "value": {"stringValue": value}}) + + +def _find_harbor_root_span_id(resource_spans: list[dict[str, Any]]) -> str | None: + """Return root AGENT span ID in Intake's hexadecimal form.""" + root = _find_harbor_root_span(resource_spans) + if root is None: + return None + return base64.b64decode(root["spanId"]).hex() + + +def _find_harbor_root_span(resource_spans: list[dict[str, Any]]) -> dict[str, Any] | None: + return next(_harbor_root_spans(resource_spans), None) + + +def _harbor_root_spans(resource_spans: list[dict[str, Any]]) -> Iterator[dict[str, Any]]: + """Yield root AGENT spans from OTLP JSON.""" + for resource_span in resource_spans: + for scope_spans in resource_span.get("scopeSpans", []): + for span in scope_spans.get("spans", []): + kind = next( + ( + item.get("value", {}).get("stringValue") + for item in span.get("attributes", []) + if item.get("key") == "openinference.span.kind" + ), + None, + ) + if not span.get("parentSpanId") and kind == "AGENT": + yield span + + +def _inject_resource_attributes(resource_spans: list[dict[str, Any]], extra: Mapping[str, str | None]) -> None: + for resource_span in resource_spans: + attributes = resource_span.setdefault("resource", {}).setdefault("attributes", []) + existing = {attribute["key"] for attribute in attributes} + for key, value in extra.items(): + if value is not None and key not in existing: + attributes.append({"key": key, "value": {"stringValue": value}}) + + +def _find_session_id(resource_spans: list[dict[str, Any]]) -> str | None: + for resource_span in resource_spans: + for scope_spans in resource_span.get("scopeSpans", []): + for span in scope_spans.get("spans", []): + for attribute in span.get("attributes", []): + if attribute.get("key") == "session.id": + return attribute.get("value", {}).get("stringValue") + return None + + +class HarborAdapter: + """Run Harbor, ingest collected OTLP traces, then analyze that evaluation.""" + + def __init__(self, subject: Subject) -> None: + self.subject = subject + + @staticmethod + def _repo_path(value: object, *, repo_root: Path) -> Path: + path = Path(str(value)) + return (repo_root / path).resolve() if not path.is_absolute() else path.resolve() + + @classmethod + def _build_dataset_config(cls, cfg: Mapping[str, object], *, repo_root: Path): + from harbor.models.job.config import DatasetConfig # noqa: PLC0415 + + dataset_path = cfg.get("dataset") + dataset_ref = cfg.get("dataset_ref") + dataset_id = cfg.get("dataset_id") + selected = [value for value in (dataset_path, dataset_ref, dataset_id) if value] + if len(selected) != 1: + raise ValueError("exactly one of config keys 'dataset', 'dataset_ref', or 'dataset_id' is required") + if cfg.get("registry_path") and cfg.get("registry_url"): + raise ValueError("config keys 'registry_path' and 'registry_url' are mutually exclusive") + + n_tasks = int(str(cfg["num_tasks"])) if cfg.get("num_tasks") is not None else None + if dataset_path: + return DatasetConfig(path=cls._repo_path(dataset_path, repo_root=repo_root), n_tasks=n_tasks) + + registry_path = cfg.get("registry_path") + registry_url = str(cfg["registry_url"]) if cfg.get("registry_url") else None + resolved_registry_path = cls._repo_path(registry_path, repo_root=repo_root) if registry_path else None + if dataset_ref: + dataset_name, separator, dataset_version = str(dataset_ref).rpartition("@") + if not separator: + dataset_name = str(dataset_ref) + dataset_version = "" + return DatasetConfig( + name=dataset_name, + version=dataset_version or None, + registry_url=registry_url, + registry_path=resolved_registry_path, + n_tasks=n_tasks, + ) + return DatasetConfig( + name=str(dataset_id), + version=str(cfg["dataset_version"]) if cfg.get("dataset_version") else None, + registry_url=registry_url, + registry_path=resolved_registry_path, + n_tasks=n_tasks, + ) + + @classmethod + def _build_job_config(cls, cfg: Mapping[str, object], *, run_id: str, repo_root: Path): + from harbor.models.job.config import AgentConfig, EnvironmentConfig, JobConfig, VerifierConfig # noqa: PLC0415 + + environment_env: dict[str, str] = {} + if user_llm := cfg.get("user_llm"): + environment_env["TAU2_USER_MODEL"] = str(user_llm) + for config_key, env_key in ( + ("user_reasoning_effort", "TAU2_USER_REASONING_EFFORT"), + ("user_temperature", "TAU2_USER_TEMPERATURE"), + ("user_llm_args_json", "TAU2_USER_LLM_ARGS_JSON"), + ): + if (value := cfg.get(config_key)) is not None: + environment_env[env_key] = str(value) + + verifier_env: dict[str, str] = {} + if verifier_llm := cfg.get("verifier_llm"): + verifier_env["TAU2_NL_ASSERTIONS_MODEL"] = str(verifier_llm) + timeout = float(str(cfg["timeout"])) if cfg.get("timeout") is not None else None + return JobConfig( + job_name=run_id, + jobs_dir=cls._repo_path(cfg.get("jobs_dir", "testbed/tmp/jobs"), repo_root=repo_root), + n_attempts=int(str(cfg.get("num_trials", 1))), + datasets=[cls._build_dataset_config(cfg, repo_root=repo_root)], + agents=[ + AgentConfig( + import_path=str(cfg.get("agent_import_path", "harbor_wrapper:WrappedAgent")), + model_name=str(cfg["agent_llm"]) if cfg.get("agent_llm") else None, + override_timeout_sec=timeout, + ) + ], + environment=EnvironmentConfig(env=environment_env), + verifier=VerifierConfig(env=verifier_env), + n_concurrent_trials=int(str(cfg.get("max_concurrency", 2))), + ) + + def check(self) -> list[str]: + cfg = self.subject.config + missing: list[str] = [f"config key '{key}'" for key in ("base_url", "workspace") if not cfg.get(key)] + if not (agent_dir_value := cfg.get("agent_dir")): + missing.append("config key 'agent_dir'") + elif not self._repo_path(agent_dir_value, repo_root=REPO_ROOT).is_dir(): + missing.append(f"agent_dir '{self._repo_path(agent_dir_value, repo_root=REPO_ROOT)}' is not a directory") + if not os.environ.get("INFERENCE_API_KEY"): + missing.append("env INFERENCE_API_KEY") + try: + self._build_job_config(cfg, run_id="preflight", repo_root=REPO_ROOT) + except ModuleNotFoundError: + missing.append("Harbor is not installed; sync the experimentalist dependency group") + except (TypeError, ValueError) as exc: + missing.append(str(exc)) + return missing + + async def produce(self) -> dict[str, object]: + cfg = self.subject.config + if missing := self.check(): + raise SystemExit(f"harbor testbed '{self.subject.name}' is missing: " + "; ".join(missing)) + + agent_dir = self._repo_path(cfg["agent_dir"], repo_root=REPO_ROOT) + base_url = str(cfg["base_url"]) + workspace = str(cfg["workspace"]) + agent_name = str(cfg.get("agent_name", "agent-0")) + run_id = mint_agent_id(workspace) + created_at = datetime.now(timezone.utc).isoformat() + raw_dataset = str(cfg.get("dataset_ref") or cfg.get("dataset_id") or Path(str(cfg["dataset"])).name) + dataset_name, separator, ref_version = raw_dataset.rpartition("@") + if not separator: + dataset_name = raw_dataset + dataset_version = str(cfg.get("dataset_version") or ref_version or "unversioned") + + ensure_workspace(base_url, workspace) + group_id = ensure_experiment_group(base_url, workspace, workspace) + create_experiment( + base_url, + workspace, + name=run_id, + experiment_group_id=group_id, + dataset_name=dataset_name, + dataset_version=dataset_version, + metadata={"agent": agent_name, "model": str(cfg.get("agent_llm", "")), "producer": "harbor"}, + ) + + agent_dir_string = str(agent_dir) + if agent_dir_string not in sys.path: + sys.path.insert(0, agent_dir_string) + + from harbor.job import Job # noqa: PLC0415 + + job_config = self._build_job_config(cfg, run_id=run_id, repo_root=REPO_ROOT) + job = await Job.create(job_config) + await job.run() + + trace_dir = job_config.jobs_dir / run_id + sent, errors, session_ids = _export_harbor_trace_files( + base_url, + workspace, + trace_dir, + agent_name, + evaluation_id=run_id, + agent_version=str(cfg.get("agent_version", "")) or None, + ) + if sent == 0: + raise SystemExit(f"harbor testbed '{self.subject.name}': no OTLP traces found under {trace_dir}") + if errors: + print(f"warning: {errors} Harbor trace upload error(s) for '{self.subject.name}'.", file=sys.stderr) + if len(session_ids) < 3: + print(f"warning: only {len(session_ids)} session(s) ingested; the analyst needs 3+.", file=sys.stderr) + if session_ids: + visible = poll_visible(base_url, workspace, session_ids) + if len(visible) < len(session_ids): + print( + f"warning: only {len(visible)}/{len(session_ids)} session(s) visible in Intake.", + file=sys.stderr, + ) + + return { + "agent": agent_name, + "workspace": workspace, + "base_url": base_url, + "run_id": run_id, + "experiment_id": run_id, + "experiment_group": workspace, + "dataset_name": dataset_name, + "dataset_version": dataset_version, + "created_at": created_at, + } + + async def analyze( + self, + *, + record: dict[str, object] | None, + since: datetime | None, + verbose: bool, + out_path: Path, + ) -> str: + if record is None: + raise SystemExit( + f"no recorded run for '{self.subject.name}' — run " + f"`uv run python -m testbed run {self.subject.name}` first" + ) + return await run_analyst( + agent=str(record["agent"]), + agent_spec=None, + workspace=str(record["workspace"]), + base_url=str(record["base_url"]), + client=make_client(str(record["base_url"])), + insights_output=str(out_path), + verbose=verbose, + since=since, + evaluation_id=str(record["experiment_id"]), + ) + + +_ADAPTERS: dict[str, type[IntakeAdapter] | type[BenchmarkAdapter] | type[HarborAdapter]] = { "intake": IntakeAdapter, "benchmark": BenchmarkAdapter, + "harbor": HarborAdapter, } diff --git a/plugins/nemo-insights/testbed/cli.py b/plugins/nemo-insights/testbed/cli.py index d781155dfc..9d983fa166 100644 --- a/plugins/nemo-insights/testbed/cli.py +++ b/plugins/nemo-insights/testbed/cli.py @@ -865,7 +865,7 @@ def main() -> None: p_doc.add_argument("name", nargs="?", help="Subject name; omit to check every subject.") p_run = sub.add_parser( "run", - help="Produce traces for a subject (benchmark: run tau2 + ingest), then record the run.", + help="Produce and ingest traces for a benchmark or Harbor subject, then record the run.", ) p_run.add_argument("name", help="Subject name from testbeds.toml.") p_run.add_argument( @@ -903,9 +903,14 @@ def main() -> None: record = asyncio.run(build_adapter(subject).produce()) TMP.mkdir(parents=True, exist_ok=True) save_run(TMP / f"{args.name}.run.json", record) - ws_line = f"realistic ws '{record['realistic_workspace']}'" - if record.get("oracle_workspace"): - ws_line += f" + oracle ws '{record['oracle_workspace']}'" + if realistic_workspace := record.get("realistic_workspace"): + ws_line = f"realistic ws '{realistic_workspace}'" + if record.get("oracle_workspace"): + ws_line += f" + oracle ws '{record['oracle_workspace']}'" + elif workspace := record.get("workspace"): + ws_line = f"ws '{workspace}'" + else: + ws_line = "workspace not recorded" print( f"✓ recorded run '{record['agent']}' ({ws_line}) — analyze with: " f"uv run python -m testbed analyze {args.name} --live" diff --git a/plugins/nemo-insights/testbed/otlp_ingest.py b/plugins/nemo-insights/testbed/otlp_ingest.py index ea1b618eed..4b7d9a97dc 100644 --- a/plugins/nemo-insights/testbed/otlp_ingest.py +++ b/plugins/nemo-insights/testbed/otlp_ingest.py @@ -6,8 +6,8 @@ ``opentelemetry.proto`` ``ExportTraceServiceRequest`` and POSTs it to Intake's permissive OTLP route. OTLP ingest does **not** auto-create the queryable ``evaluator_results`` rows the Analyst reads, so :func:`post_evaluator_results` -separately POSTs the reward (mirroring exactly the row the ATIF importer used to -create: ``name="reward"``, ``NUMERIC``, targeting the EVALUATOR span). +separately POSTs verifier rewards as ``NUMERIC`` evaluator rows targeting the +relevant root span. The protobuf build mirrors nemo-platform's own ``services/intake/tests/integration/spans/conftest.py::make_otlp_request`` helper, @@ -85,19 +85,20 @@ def post_evaluator_results( span_id: str, session_id: str, score: float, + name: str = "reward", client: httpx.Client | None = None, ) -> None: """POST the reward row the OTLP path doesn't auto-create. - Reproduces the ATIF importer's row exactly: ``name="reward"``, ``NUMERIC``, - targeting the EVALUATOR span — so the Analyst reads the reward unchanged. + ``name`` defaults to ``reward`` for Tau2 compatibility; Harbor adapters pass + each verifier criterion's name. The Analyst reads every value unchanged. Raises ``RuntimeError`` on any non-2xx. """ url = f"{base_url.rstrip('/')}/apis/intake/v2/workspaces/{workspace}/evaluator-results" body = { "span_id": span_id, "session_id": session_id, - "name": "reward", + "name": name, "value": score, "data_type": "NUMERIC", } diff --git a/plugins/nemo-insights/testbed/testbeds.toml b/plugins/nemo-insights/testbed/testbeds.toml index ed684fb102..e016bc8eb9 100644 --- a/plugins/nemo-insights/testbed/testbeds.toml +++ b/plugins/nemo-insights/testbed/testbeds.toml @@ -30,6 +30,23 @@ agent = "nemo-oo-airline" workspace = "tau2-airline-20260710-152942-4754" base_url = "http://localhost:8080" +[tau3-airline-harbor] +type = "harbor" +agent_dir = "../nemo-experimentalist/examples/tau3-nooa-agent" +base_url = "http://localhost:8080" +workspace = "canonical-tau3-airline" +agent_name = "nemo-experimentalist-tau3-nooa" +agent_version = "1.0.0" +agent_llm = "openai/openai/openai/gpt-5-mini" +user_llm = "openai/openai/openai/gpt-5-mini" +verifier_llm = "openai/openai/openai/gpt-5-mini" +dataset_ref = "sierra-research/tau3-bench@1" +registry_url = "https://hub.harborframework.com" +num_tasks = 6 +num_trials = 1 +max_concurrency = 6 +timeout = 3600 + [tau2-airline] type = "benchmark" domain = "airline" diff --git a/plugins/nemo-insights/tests/testbed/test_adapters.py b/plugins/nemo-insights/tests/testbed/test_adapters.py index 04645bdf17..10ba434e08 100644 --- a/plugins/nemo-insights/tests/testbed/test_adapters.py +++ b/plugins/nemo-insights/tests/testbed/test_adapters.py @@ -5,7 +5,15 @@ from typing import Any import pytest -from testbed.adapters import BenchmarkAdapter, IntakeAdapter, build_adapter +from testbed.adapters import ( + BenchmarkAdapter, + HarborAdapter, + IntakeAdapter, + _export_harbor_trace_files, + _harbor_trace_context, + _root_values_from_trace, + build_adapter, +) from testbed.registry import Subject _CFG = { @@ -37,6 +45,149 @@ }, ] +_HARBOR_OTLP_JSON = ( + '{"resourceSpans":[{"resource":{"attributes":[{"key":"service.name","value":{"stringValue":"harbor"}}]},' + '"scopeSpans":[{"spans":[{"traceId":"AAAAAAAAAAAAAAAAAAAAAQ==","spanId":"AAAAAAAAAAE=",' + '"name":"agent","startTimeUnixNano":"1","endTimeUnixNano":"2",' + '"attributes":[{"key":"session.id","value":{"stringValue":"sess-1"}}]}]}]}]}' +) + + +def test_harbor_trace_conversion_enriches_and_exports(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + (tmp_path / "fallback-session.jsonl").write_text(f"{_HARBOR_OTLP_JSON}\n", encoding="utf-8") + monkeypatch.setenv("INFERENCE_API_KEY", "secret") + calls: list[dict[str, object]] = [] + + def capture_export(base_url, workspace, request, *, client=None, headers=None): + calls.append({"base_url": base_url, "workspace": workspace, "request": request, "headers": headers}) + + monkeypatch.setattr("testbed.adapters.export_trace_request", capture_export) + + result = _export_harbor_trace_files( + "http://x", + "ws", + tmp_path, + "agent-0", + evaluation_id="evaluation-1", + ) + + assert result == (1, 0, {"sess-1"}) + assert calls[0]["headers"] == {"Authorization": "Bearer secret"} + request = calls[0]["request"] + resource_attrs = { + item.key: getattr(item.value, item.value.WhichOneof("value")) + for item in request.resource_spans[0].resource.attributes + } + assert resource_attrs == { + "service.name": "harbor", + "gen_ai.agent.name": "agent-0", + "gen_ai.agent.id": "agent-0", + "session.id": "sess-1", + "gen_ai.conversation.id": "sess-1", + "nemo.experiment.id": "evaluation-1", + "nemo.optimizer.workspace": "ws", + } + + +def test_harbor_trace_conversion_adds_task_fields_and_rewards( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + trial_dir = tmp_path / "task-1__trial" + trace_dir = trial_dir / "artifacts" / "logs" / "artifacts" / "traces" + trace_dir.mkdir(parents=True) + (trial_dir / "result.json").write_text( + json.dumps({"task_name": "task-1", "verifier_result": {"rewards": {"score": 0.75, "reward": 1.0}}}), + encoding="utf-8", + ) + output_path = trial_dir / "artifacts" / "logs" / "artifacts" / "output.md" + output_path.write_text("# Final answer", encoding="utf-8") + body = json.loads(_HARBOR_OTLP_JSON) + root = body["resourceSpans"][0]["scopeSpans"][0]["spans"][0] + root["name"] = "method.solve" + root["attributes"].extend( + [ + {"key": "openinference.span.kind", "value": {"stringValue": "AGENT"}}, + {"key": "input.value", "value": {"stringValue": "Research this topic"}}, + ] + ) + (trace_dir / "trace.jsonl").write_text(json.dumps(body), encoding="utf-8") + exports: list[object] = [] + rewards: list[dict[str, object]] = [] + monkeypatch.setattr( + "testbed.adapters.export_trace_request", + lambda base_url, workspace, request, *, client=None, headers=None: exports.append(request), + ) + monkeypatch.setattr( + "testbed.adapters.post_evaluator_results", + lambda base_url, workspace, **kwargs: rewards.append( + {key: value for key, value in kwargs.items() if key != "client"} + ), + ) + + assert _export_harbor_trace_files("http://x", "ws", tmp_path, "agent-0", evaluation_id="evaluation-1") == ( + 1, + 0, + {"sess-1"}, + ) + exported_root = exports[0].resource_spans[0].scope_spans[0].spans[0] + attributes = {item.key: getattr(item.value, item.value.WhichOneof("value")) for item in exported_root.attributes} + assert attributes["nemo.test_case.id"] == "task-1" + assert attributes["input.value"] == "Research this topic" + assert attributes["output.value"] == "# Final answer" + assert rewards == [ + {"span_id": exported_root.span_id.hex(), "session_id": "sess-1", "score": 0.75, "name": "score"}, + {"span_id": exported_root.span_id.hex(), "session_id": "sess-1", "score": 1.0, "name": "reward"}, + ] + + +def test_harbor_trace_context_drops_rewards_for_failed_trial(tmp_path: Path) -> None: + trial_dir = tmp_path / "task-1__trial" + trace_dir = trial_dir / "artifacts" / "logs" / "artifacts" / "traces" + trace_dir.mkdir(parents=True) + path = trace_dir / "trace.jsonl" + path.write_text("", encoding="utf-8") + (trial_dir / "result.json").write_text( + json.dumps( + { + "task_name": "task-1", + "exception_info": {"exception_type": "RuntimeError"}, + "verifier_result": {"rewards": {"reward": 1.0}}, + } + ), + encoding="utf-8", + ) + + test_case_id, _, _, rewards = _harbor_trace_context(path, tmp_path, []) + + assert test_case_id == "task-1" + assert rewards == {} + + +def test_root_values_from_trace_reads_only_root_agent_span() -> None: + body = json.loads(_HARBOR_OTLP_JSON) + spans = body["resourceSpans"][0]["scopeSpans"][0]["spans"] + spans[0]["attributes"].extend( + [ + {"key": "openinference.span.kind", "value": {"stringValue": "AGENT"}}, + {"key": "input.value", "value": {"stringValue": "root input"}}, + {"key": "output.value", "value": {"stringValue": "root output"}}, + ] + ) + spans.append( + { + "traceId": "AAAAAAAAAAAAAAAAAAAAAQ==", + "spanId": "AAAAAAAAAAI=", + "parentSpanId": "AAAAAAAAAAE=", + "name": "acompletion", + "startTimeUnixNano": "1", + "endTimeUnixNano": "2", + "attributes": [{"key": "input.value", "value": {"stringValue": "child input"}}], + } + ) + + assert _root_values_from_trace([(1, body)]) == ("root input", "root output") + def _intake_subject(**overrides) -> Subject: config = {"agent": "a", "workspace": "w", "base_url": "u", **overrides} @@ -169,6 +320,64 @@ def test_build_adapter_dispatches_benchmark(): assert isinstance(adapter, BenchmarkAdapter) +def test_build_adapter_dispatches_harbor(): + adapter = build_adapter(Subject("tau3-airline-harbor", "harbor", {})) + assert isinstance(adapter, HarborAdapter) + + +def test_harbor_dataset_config_accepts_hub_ref(): + config = HarborAdapter._build_dataset_config( + { + "dataset_ref": "sierra-research/tau3-bench@1", + "registry_url": "https://hub.harborframework.com", + "num_tasks": 6, + }, + repo_root=Path("/repo"), + ) + + assert config.name == "sierra-research/tau3-bench" + assert config.version == "1" + assert config.registry_url == "https://hub.harborframework.com" + assert config.n_tasks == 6 + + +def test_harbor_dataset_config_rejects_ambiguous_source(): + with pytest.raises(ValueError, match="exactly one"): + HarborAdapter._build_dataset_config( + {"dataset": "local", "dataset_ref": "org/data@1"}, + repo_root=Path("/repo"), + ) + + +async def test_harbor_analyze_uses_record(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + seen: dict[str, object] = {} + + async def fake_run_analyst(**kwargs: object) -> str: + seen.update(kwargs) + return "REPORT" + + monkeypatch.setattr("testbed.adapters.run_analyst", fake_run_analyst) + monkeypatch.setattr("testbed.adapters.make_client", lambda base_url: object()) + record = { + "agent": "nemo-experimentalist-tau3-nooa", + "workspace": "canonical-tau3-airline", + "base_url": "http://localhost:8080", + "experiment_id": "canonical-tau3-airline-20260731-120000-abcd", + } + + report = await HarborAdapter(Subject("tau3-airline-harbor", "harbor", {})).analyze( + record=record, + since=None, + verbose=True, + out_path=tmp_path / "insights.yaml", + ) + + assert report == "REPORT" + assert seen["agent"] == "nemo-experimentalist-tau3-nooa" + assert seen["workspace"] == "canonical-tau3-airline" + assert seen["evaluation_id"] == record["experiment_id"] + + async def test_benchmark_preflight_lists_missing(monkeypatch): monkeypatch.delenv("OPENAI_API_KEY", raising=False) monkeypatch.delenv("OPENAI_API_BASE", raising=False) diff --git a/plugins/nemo-insights/tests/testbed/test_cli.py b/plugins/nemo-insights/tests/testbed/test_cli.py index 7a982c2d59..8949b2c013 100644 --- a/plugins/nemo-insights/tests/testbed/test_cli.py +++ b/plugins/nemo-insights/tests/testbed/test_cli.py @@ -934,6 +934,30 @@ async def fake_produce(self): assert load_run(tmp_path / "tau2-airline.run.json")["agent"] == "tau2-airline-xyz" +def test_run_harbor_records_and_prints_workspace(monkeypatch, tmp_path, capsys): + monkeypatch.setattr(cli, "TMP", tmp_path) + + async def fake_produce(self): + return { + "agent": "nemo-experimentalist-tau3-nooa", + "workspace": "canonical-tau3-airline", + "base_url": "http://localhost:8080", + "experiment_id": "eval-1", + } + + monkeypatch.setattr("testbed.adapters.HarborAdapter.produce", fake_produce) + monkeypatch.setattr(sys, "argv", ["testbed", "run", "tau3-airline-harbor"]) + + cli.main() + + out = capsys.readouterr().out + assert "ws 'canonical-tau3-airline'" in out + assert "analyze tau3-airline-harbor --live" in out + from testbed.runstore import load_run + + assert load_run(tmp_path / "tau3-airline-harbor.run.json")["experiment_id"] == "eval-1" + + def test_analyze_live_passes_record_to_analyze(monkeypatch, tmp_path): from testbed.runstore import save_run diff --git a/plugins/nemo-insights/tests/testbed/test_otlp_ingest.py b/plugins/nemo-insights/tests/testbed/test_otlp_ingest.py index 03e27b1db6..a758d46420 100644 --- a/plugins/nemo-insights/tests/testbed/test_otlp_ingest.py +++ b/plugins/nemo-insights/tests/testbed/test_otlp_ingest.py @@ -189,6 +189,22 @@ def test_post_evaluator_results_route_and_body(): } +def test_post_evaluator_results_accepts_named_criterion(): + stub = _JsonStub(status=201) + + post_evaluator_results( + "http://x", + "ws", + span_id="sp1", + session_id="sess-1", + score=0.75, + name="policy_compliance", + client=stub, + ) + + assert stub.calls[0][1]["name"] == "policy_compliance" + + def test_post_evaluator_results_raises_on_error(): with pytest.raises(RuntimeError): post_evaluator_results("http://x", "ws", span_id="s", session_id="z", score=0.0, client=_JsonStub(status=500)) diff --git a/plugins/nemo-insights/tests/testbed/test_registry.py b/plugins/nemo-insights/tests/testbed/test_registry.py index dd7b286aa3..4a1fbce4ff 100644 --- a/plugins/nemo-insights/tests/testbed/test_registry.py +++ b/plugins/nemo-insights/tests/testbed/test_registry.py @@ -35,8 +35,9 @@ def test_registry_contains_only_expected_analyzable_subjects() -> None: "tau2-airline", "tau2-retail", "tau2-telecom", + "tau3-airline-harbor", } - assert all(subject.type in ("benchmark", "intake") for subject in subjects.values()) + assert all(subject.type in ("benchmark", "harbor", "intake") for subject in subjects.values()) assert subjects["nvq"].config["agent"] == "content-dedup" @@ -64,6 +65,14 @@ def test_tau2_telecom_uses_small_split() -> None: assert telecom.config["task_split_name"] == "small" +def test_tau3_airline_harbor_uses_checked_in_agent_and_hub_dataset() -> None: + subject = load_registry(cli.REGISTRY_PATH)["tau3-airline-harbor"] + + assert subject.type == "harbor" + assert subject.config["agent_dir"] == "../nemo-experimentalist/examples/tau3-nooa-agent" + assert subject.config["dataset_ref"] == "sierra-research/tau3-bench@1" + + def test_every_analyzable_subject_has_expected_state_pin() -> None: expected = { "glamr": "state-v8", @@ -74,6 +83,7 @@ def test_every_analyzable_subject_has_expected_state_pin() -> None: "tau2-telecom": "state-v10", } - assert { - name: release.lock_ref(cli.HERE / "state.lock", name) for name in sorted(load_registry(cli.REGISTRY_PATH)) - } == expected + subjects = load_registry(cli.REGISTRY_PATH) + reproducible = sorted(name for name, subject in subjects.items() if subject.type in ("benchmark", "intake")) + + assert {name: release.lock_ref(cli.HERE / "state.lock", name) for name in reproducible} == expected From ffe3ec3b0185f50482046a237abf8266cec5cbfe Mon Sep 17 00:00:00 2001 From: Gaia Di Lorenzo Date: Fri, 31 Jul 2026 14:23:46 +0200 Subject: [PATCH 02/14] feat(experimentalist): add Tau3 quality workflow Signed-off-by: Gaia Di Lorenzo --- .../benchmarks/configs/tau3-quality.yaml | 3 - .../experimentalist-quality.yaml | 22 ++++ .../tau3-nooa-agent/prepare-airline-smoke.sh | 75 ------------- .../tau3-nooa-agent/prepare-airline.sh | 103 ++++++++++++++++++ 4 files changed, 125 insertions(+), 78 deletions(-) create mode 100644 plugins/nemo-experimentalist/examples/tau3-nooa-agent/experimentalist-quality.yaml delete mode 100755 plugins/nemo-experimentalist/examples/tau3-nooa-agent/prepare-airline-smoke.sh create mode 100755 plugins/nemo-experimentalist/examples/tau3-nooa-agent/prepare-airline.sh diff --git a/plugins/nemo-experimentalist/benchmarks/configs/tau3-quality.yaml b/plugins/nemo-experimentalist/benchmarks/configs/tau3-quality.yaml index ae953389a3..7fd4164a40 100644 --- a/plugins/nemo-experimentalist/benchmarks/configs/tau3-quality.yaml +++ b/plugins/nemo-experimentalist/benchmarks/configs/tau3-quality.yaml @@ -21,9 +21,6 @@ optimizer: disable_convergence_check: false evaluator: n_attempts: 2 - # Keep at 3 or lower: each task requests 8192 MB and parallel image builds - # have triggered Docker Hub rate limiting and DNS failures. - n_concurrent_trials: 3 quiet: true agent_setup_timeout_multiplier: 2.0 # tau3 builds two images per task, and the first build clones and installs diff --git a/plugins/nemo-experimentalist/examples/tau3-nooa-agent/experimentalist-quality.yaml b/plugins/nemo-experimentalist/examples/tau3-nooa-agent/experimentalist-quality.yaml new file mode 100644 index 0000000000..cbf9d111c3 --- /dev/null +++ b/plugins/nemo-experimentalist/examples/tau3-nooa-agent/experimentalist-quality.yaml @@ -0,0 +1,22 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +# Agent-level equivalent of benchmarks/configs/tau3-quality.yaml's optimizer +# section, for direct `nemo experimentalist run --config ...` usage. +max_rounds: 3 +min_rounds_before_stopping: 2 +max_survivors: 2 +max_candidates: 3 +max_trajectory_tasks: 8 +max_train_batch_tasks: 16 +train_batch_seed: 20260727 +disable_trajectory_scoring: false +disable_convergence_check: false +evaluator: + n_attempts: 2 + quiet: true + agent_setup_timeout_multiplier: 2.0 + environment_build_timeout_multiplier: 3.0 +eval_author: + max_traces: 10 + max_validation_repair_attempts: 5 diff --git a/plugins/nemo-experimentalist/examples/tau3-nooa-agent/prepare-airline-smoke.sh b/plugins/nemo-experimentalist/examples/tau3-nooa-agent/prepare-airline-smoke.sh deleted file mode 100755 index f57f437e6a..0000000000 --- a/plugins/nemo-experimentalist/examples/tau3-nooa-agent/prepare-airline-smoke.sh +++ /dev/null @@ -1,75 +0,0 @@ -#!/usr/bin/env bash -# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. -# SPDX-License-Identifier: Apache-2.0 - -set -euo pipefail -shopt -s nullglob - -SCRIPT_DIR="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)" -PLUGIN_ROOT="$(cd -- "${SCRIPT_DIR}/../.." && pwd)" -OUTPUT_ROOT="${1:-${PLUGIN_ROOT}/tmp/tau3-airline-smoke}" -SOURCE_ROOT="${OUTPUT_ROOT}/source" -DATASET_ROOT="${SOURCE_ROOT}/tau3-bench" -TRAIN_ROOT="${OUTPUT_ROOT}/train" -VALIDATION_ROOT="${OUTPUT_ROOT}/validation" -DATASET_REF="sierra-research/tau3-bench@1" - -TRAIN_TASKS=( - "tau3-bench__tau3-airline-0" - "tau3-bench__tau3-airline-20" - "tau3-bench__tau3-airline-39" -) -VALIDATION_TASKS=( - "tau3-bench__tau3-airline-3" - "tau3-bench__tau3-airline-36" -) - -if [[ ! -f "${DATASET_ROOT}/${TRAIN_TASKS[0]}/task.toml" ]]; then - mkdir -p "${SOURCE_ROOT}" - ( - cd "${PLUGIN_ROOT}" - uv run --frozen harbor download "${DATASET_REF}" \ - --output-dir "${SOURCE_ROOT}" \ - --export \ - --overwrite - ) -fi - -prepare_split() { - local split_root="$1" - shift - local task_names=("$@") - - if [[ -d "${split_root}" ]]; then - local existing_tasks=("${split_root}"/*/task.toml) - if [[ ${#existing_tasks[@]} -eq ${#task_names[@]} ]]; then - for task_name in "${task_names[@]}"; do - if [[ ! -f "${split_root}/${task_name}/task.toml" ]]; then - echo "Existing split is not the expected dataset: ${split_root}" >&2 - exit 1 - fi - done - echo "Reusing ${split_root}" - return - fi - echo "Existing split is incomplete: ${split_root}" >&2 - exit 1 - fi - - mkdir -p "${split_root}" - for task_name in "${task_names[@]}"; do - local source_task="${DATASET_ROOT}/${task_name}" - if [[ ! -f "${source_task}/task.toml" ]]; then - echo "Downloaded dataset is missing ${task_name}" >&2 - exit 1 - fi - cp -R "${source_task}" "${split_root}/${task_name}" - done -} - -prepare_split "${TRAIN_ROOT}" "${TRAIN_TASKS[@]}" -prepare_split "${VALIDATION_ROOT}" "${VALIDATION_TASKS[@]}" - -echo "Tau3 Airline smoke datasets are ready:" -echo " train: ${TRAIN_ROOT}" -echo " validation: ${VALIDATION_ROOT}" diff --git a/plugins/nemo-experimentalist/examples/tau3-nooa-agent/prepare-airline.sh b/plugins/nemo-experimentalist/examples/tau3-nooa-agent/prepare-airline.sh new file mode 100755 index 0000000000..c30aa3131b --- /dev/null +++ b/plugins/nemo-experimentalist/examples/tau3-nooa-agent/prepare-airline.sh @@ -0,0 +1,103 @@ +#!/usr/bin/env bash +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +set -euo pipefail +shopt -s nullglob + +SCRIPT_DIR="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)" +PLUGIN_ROOT="$(cd -- "${SCRIPT_DIR}/../.." && pwd)" +OUTPUT_ROOT="${1:-${PLUGIN_ROOT}/tmp/tau3-airline-smoke}" +SOURCE_ROOT="${OUTPUT_ROOT}/source" +DATASET_ROOT="${SOURCE_ROOT}/tau3-bench" +TRAIN_ROOT="${OUTPUT_ROOT}/train" +VALIDATION_ROOT="${OUTPUT_ROOT}/validation" +DATASET_REF="sierra-research/tau3-bench@1" + +TRAIN_TASKS=( + "tau3-bench__tau3-airline-0" + "tau3-bench__tau3-airline-1" + "tau3-bench__tau3-airline-4" + "tau3-bench__tau3-airline-5" + "tau3-bench__tau3-airline-9" + "tau3-bench__tau3-airline-10" + "tau3-bench__tau3-airline-12" + "tau3-bench__tau3-airline-14" + "tau3-bench__tau3-airline-17" + "tau3-bench__tau3-airline-20" + "tau3-bench__tau3-airline-23" + "tau3-bench__tau3-airline-27" + "tau3-bench__tau3-airline-33" + "tau3-bench__tau3-airline-34" + "tau3-bench__tau3-airline-38" + "tau3-bench__tau3-airline-39" + "tau3-bench__tau3-airline-41" + "tau3-bench__tau3-airline-42" + "tau3-bench__tau3-airline-46" + "tau3-bench__tau3-airline-47" +) +VALIDATION_TASKS=( + "tau3-bench__tau3-airline-3" + "tau3-bench__tau3-airline-7" + "tau3-bench__tau3-airline-11" + "tau3-bench__tau3-airline-15" + "tau3-bench__tau3-airline-21" + "tau3-bench__tau3-airline-28" + "tau3-bench__tau3-airline-36" + "tau3-bench__tau3-airline-40" + "tau3-bench__tau3-airline-43" + "tau3-bench__tau3-airline-49" +) + +if [[ ! -f "${DATASET_ROOT}/${TRAIN_TASKS[0]}/task.toml" ]]; then + mkdir -p "${SOURCE_ROOT}" + ( + cd "${PLUGIN_ROOT}" + uv run --frozen harbor download "${DATASET_REF}" \ + --output-dir "${SOURCE_ROOT}" \ + --export \ + --overwrite + ) +fi + +prepare_split() { + local split_root="$1" + shift + local task_names=("$@") + + mkdir -p "${split_root}" + local existing_task_path existing_task_name expected + for existing_task_path in "${split_root}"/*/task.toml; do + existing_task_name="$(basename -- "$(dirname -- "${existing_task_path}")")" + expected=false + for task_name in "${task_names[@]}"; do + if [[ "${existing_task_name}" == "${task_name}" ]]; then + expected=true + break + fi + done + if [[ "${expected}" != true ]]; then + echo "Existing split contains unexpected task ${existing_task_name}: ${split_root}" >&2 + exit 1 + fi + done + + for task_name in "${task_names[@]}"; do + if [[ -f "${split_root}/${task_name}/task.toml" ]]; then + continue + fi + local source_task="${DATASET_ROOT}/${task_name}" + if [[ ! -f "${source_task}/task.toml" ]]; then + echo "Downloaded dataset is missing ${task_name}" >&2 + exit 1 + fi + cp -R "${source_task}" "${split_root}/${task_name}" + done +} + +prepare_split "${TRAIN_ROOT}" "${TRAIN_TASKS[@]}" +prepare_split "${VALIDATION_ROOT}" "${VALIDATION_TASKS[@]}" + +echo "Tau3 Airline quality datasets are ready:" +echo " train: ${TRAIN_ROOT} (${#TRAIN_TASKS[@]} tasks)" +echo " validation: ${VALIDATION_ROOT} (${#VALIDATION_TASKS[@]} tasks)" From 5f87ca55a4231c7b634656af82282469eb3141a7 Mon Sep 17 00:00:00 2001 From: Gaia Di Lorenzo Date: Fri, 31 Jul 2026 14:23:52 +0200 Subject: [PATCH 03/14] fix(experimentalist): clean Harbor on SIGINT Signed-off-by: Gaia Di Lorenzo --- .../components/evaluator/harbor.py | 102 +++++++++++++++++- .../experimentalist/test_evaluator_harbor.py | 90 ++++++++++++++++ 2 files changed, 190 insertions(+), 2 deletions(-) diff --git a/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/experimentalist/components/evaluator/harbor.py b/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/experimentalist/components/evaluator/harbor.py index 24a60ffa37..1fcfa565af 100644 --- a/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/experimentalist/components/evaluator/harbor.py +++ b/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/experimentalist/components/evaluator/harbor.py @@ -32,6 +32,7 @@ from harbor.models.task.task import Task as HarborTaskModel from harbor.models.trial.config import ServiceVolumeConfig from harbor.models.trial.paths import EnvironmentPaths, TrialPaths +from harbor.trial.hooks import TrialHookEvent from nemo_experimentalist_plugin.experimentalist.components.evaluator.base import ( Evaluator, EvaluatorConfig, @@ -113,6 +114,8 @@ class HarborResourceSpec(TypedDict): _TRACE_ARTIFACT_SOURCE = "/app/traces" _TRACE_ARTIFACT_DESTINATION = "traces" _SHELL_SYNTAX_TIMEOUT_SEC = 10.0 +_DOCKER_CLEANUP_TIMEOUT_SEC = 30.0 +_DOCKER_COMPOSE_PROJECT_LABEL = "com.docker.compose.project" _AGENT_IMPORT_ROOT = "_nemo_experimentalist_eval_agents" _IDENTIFIER_RE = re.compile(r"\W+") _TRIAL_LOG_DESCRIPTIONS = { @@ -129,6 +132,88 @@ class HarborResourceSpec(TypedDict): logger = logging.getLogger(__name__) +def _sanitize_compose_project_name(name: str) -> str: + normalized = name.lower() + if not re.match(r"^[a-z0-9]", normalized): + normalized = f"0{normalized}" + return re.sub(r"[^a-z0-9_-]", "-", normalized) + + +async def _docker_cleanup_command(args: Sequence[str]) -> str: + docker_path = shutil.which("docker", path=os.defpath) + if docker_path is None: + raise FileNotFoundError("docker executable not found on the system path") + + process = await asyncio.create_subprocess_exec( + docker_path, + *args, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.STDOUT, + ) + try: + stdout, _ = await asyncio.wait_for( + process.communicate(), + timeout=_DOCKER_CLEANUP_TIMEOUT_SEC, + ) + except (TimeoutError, asyncio.CancelledError): + if process.returncode is None: + try: + process.kill() + except ProcessLookupError: + pass + await process.communicate() + raise + + output = stdout.decode("utf-8", errors="replace").strip() + if process.returncode != 0: + raise RuntimeError(output or f"docker {' '.join(args)} exited with status {process.returncode}") + return output + + +async def _cleanup_cancelled_harbor_projects(trial_names: set[str]) -> None: + """Remove Docker Compose resources left by cancelled Harbor trials.""" + if not trial_names: + return + + project_prefixes = tuple(f"{_sanitize_compose_project_name(name)}__" for name in trial_names) + resource_specs = ( + ("container", ".ID", ("container", "rm", "--force")), + ("network", ".ID", ("network", "rm")), + ("volume", ".Name", ("volume", "rm", "--force")), + ) + + for resource, id_template, remove_command in resource_specs: + list_command = [resource, "ls"] + if resource == "container": + list_command.append("--all") + list_command.extend( + ( + "--filter", + f"label={_DOCKER_COMPOSE_PROJECT_LABEL}", + "--format", + f'{{{{{id_template}}}}}\t{{{{.Label "{_DOCKER_COMPOSE_PROJECT_LABEL}"}}}}', + ) + ) + try: + rows = await _docker_cleanup_command(list_command) + except Exception as exc: + logger.warning("Could not list Harbor %s resources during cancellation cleanup: %s", resource, exc) + continue + + resource_ids = [] + for row in rows.splitlines(): + resource_id, separator, project = row.partition("\t") + if separator and any(project.startswith(prefix) for prefix in project_prefixes): + resource_ids.append(resource_id) + if not resource_ids: + continue + + try: + await _docker_cleanup_command((*remove_command, *resource_ids)) + except Exception as exc: + logger.warning("Could not remove cancelled Harbor %s resources: %s", resource, exc) + + @dataclass(frozen=True) class HarborVerifierValidationFailure: """Syntax failure found in one task's Harbor verifier.""" @@ -1247,9 +1332,11 @@ class HarborEvaluator(Evaluator): def __init__(self, options: HarborEvaluatorConfig | None = None, experiment_dir: Path | None = None) -> None: super().__init__(options or HarborEvaluatorConfig(), experiment_dir=experiment_dir) - async def _run(self, agent: Path, dataset: Dataset, options: HarborEvaluatorConfig) -> Sequence[TrialResult]: + async def _run(self, agent: Path, dataset: Dataset, options: EvaluatorConfig) -> Sequence[TrialResult]: if not isinstance(dataset, HarborDataset): raise ValueError("Dataset must be a Harbor dataset") + if not isinstance(options, HarborEvaluatorConfig): + raise ValueError("Options must be a Harbor evaluator config") if dataset.source is None: raise ValueError("Harbor dataset source is required") @@ -1282,7 +1369,18 @@ async def _run(self, agent: Path, dataset: Dataset, options: HarborEvaluatorConf try: job = await Job.create(job_config) - await job.run() + started_trial_names: set[str] = set() + + async def track_started_trial(event: TrialHookEvent) -> None: + started_trial_names.add(event.trial_name) + + job.on_trial_started(track_started_trial) + try: + await job.run() + except asyncio.CancelledError: + cleanup_task = asyncio.create_task(_cleanup_cancelled_harbor_projects(started_trial_names)) + await asyncio.shield(cleanup_task) + raise finally: _cleanup_scoped_imports(scoped_package) diff --git a/plugins/nemo-experimentalist/tests/experimentalist/test_evaluator_harbor.py b/plugins/nemo-experimentalist/tests/experimentalist/test_evaluator_harbor.py index f869a74586..955aeb3db6 100644 --- a/plugins/nemo-experimentalist/tests/experimentalist/test_evaluator_harbor.py +++ b/plugins/nemo-experimentalist/tests/experimentalist/test_evaluator_harbor.py @@ -20,6 +20,7 @@ HarborEvaluatorConfig, HarborVerifierValidationError, _chmod_path_chain, + _cleanup_cancelled_harbor_projects, _cleanup_scoped_imports, _ensure_package, _python_syntax_failure, @@ -95,6 +96,7 @@ class RecordingJob: def __init__(self, config) -> None: self.config = config self.job_dir = job_dir + self.started_hook = None @classmethod async def create(cls, config): @@ -106,6 +108,10 @@ async def run(self): type(self).run_calls += 1 return SimpleNamespace(id="job-id", stats=None) + def on_trial_started(self, callback): + self.started_hook = callback + return self + return RecordingJob @@ -720,6 +726,10 @@ async def create(cls, config): async def run(self): return SimpleNamespace(id="job-id", stats=FakeStats()) + def on_trial_started(self, callback): + self.started_hook = callback + return self + monkeypatch.setattr("nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor.Job", FakeJob) result = await evaluator.run( @@ -975,6 +985,82 @@ async def test_harbor_evaluator_accepts_valid_python_verifier( assert fake_job.run_calls == 1 +@pytest.mark.asyncio +async def test_harbor_evaluator_cleans_started_projects_on_cancellation( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + agent_dir = tmp_path / "agent" + agent_dir.mkdir() + task_dir = tmp_path / "task-a" + _write(task_dir / "task.toml", "") + dataset = HarborDataset.from_path(task_dir) + cleaned: list[set[str]] = [] + + class CancelledJob: + def __init__(self, config) -> None: + self.config = config + self.job_dir = tmp_path / "jobs" / "cancelled" + self.started_hook = None + + @classmethod + async def create(cls, config): + return cls(config) + + def on_trial_started(self, callback): + self.started_hook = callback + return self + + async def run(self): + assert self.started_hook is not None + await self.started_hook(SimpleNamespace(trial_name="task-a__abc123")) + raise asyncio.CancelledError + + async def fake_cleanup(trial_names: set[str]) -> None: + cleaned.append(trial_names) + + monkeypatch.setattr("nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor.Job", CancelledJob) + monkeypatch.setattr( + "nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor._cleanup_cancelled_harbor_projects", + fake_cleanup, + ) + + with pytest.raises(asyncio.CancelledError): + await HarborEvaluator()._run(agent_dir, dataset, HarborEvaluatorConfig()) + + assert cleaned == [{"task-a__abc123"}] + + +@pytest.mark.asyncio +async def test_cancelled_harbor_cleanup_removes_only_matching_compose_projects( + monkeypatch: pytest.MonkeyPatch, +) -> None: + calls: list[tuple[str, ...]] = [] + + async def fake_docker_command(args) -> str: + command = tuple(args) + calls.append(command) + if command[:2] == ("container", "ls"): + return "c1\ttask-a__abc123__env\nc2\tother-task__xyz__env" + if command[:2] == ("network", "ls"): + return "n1\ttask-a__abc123__env" + if command[:2] == ("volume", "ls"): + return "v1\ttask-a__abc123__verifier__step" + return "" + + monkeypatch.setattr( + "nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor._docker_cleanup_command", + fake_docker_command, + ) + + await _cleanup_cancelled_harbor_projects({"task-a__abc123"}) + + assert ("container", "rm", "--force", "c1") in calls + assert ("network", "rm", "n1") in calls + assert ("volume", "rm", "--force", "v1") in calls + assert all("c2" not in command for command in calls) + + @pytest.mark.asyncio async def test_harbor_evaluator_rejects_invalid_configured_test_sh_before_job_create( tmp_path: Path, @@ -1419,6 +1505,10 @@ async def create(cls, config): async def run(self): return SimpleNamespace(id="job-id", stats=None) + def on_trial_started(self, callback): + self.started_hook = callback + return self + monkeypatch.setattr("nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor.Job", FakeJob) trials = await evaluator._run( From 4af98ca36ca655e8675751da9a0bc99077125491 Mon Sep 17 00:00:00 2001 From: Gaia Di Lorenzo Date: Fri, 31 Jul 2026 14:23:56 +0200 Subject: [PATCH 04/14] fix(experimentalist): disable Gemini cache hints Signed-off-by: Gaia Di Lorenzo --- .../components/model_config.py | 20 ++++++++++++++++--- .../experimentalist/test_model_config.py | 17 +++++++++++++++- 2 files changed, 33 insertions(+), 4 deletions(-) diff --git a/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/experimentalist/components/model_config.py b/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/experimentalist/components/model_config.py index f2eeb8d044..f38b4d8bb3 100644 --- a/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/experimentalist/components/model_config.py +++ b/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/experimentalist/components/model_config.py @@ -18,6 +18,20 @@ def _optional_env(name: str, default: str) -> str: return os.environ.get(name, "").strip() or default +def _completion_client(name: str, *, api_base: str, api_key: str) -> CompletionClient: + if "/gemini-" in name.lower(): + # NoOA injects explicit prompt-cache breakpoints by default. Vertex AI + # rejects those breakpoints when the cacheable prefix is below its + # minimum token count, so Gemini must use provider-managed caching. + return CompletionClient( + name, + api_base=api_base, + api_key=api_key, + cache_control_injection_points=[], + ) + return CompletionClient(name, api_base=api_base, api_key=api_key) + + @functools.cache def get_smart_model() -> CompletionClient: """Return the cached smart (high-capability) LLM client configured from environment variables. @@ -32,7 +46,7 @@ def get_smart_model() -> CompletionClient: api_base = _required_env("EXPERIMENTALIST_API_BASE") api_key = _required_env("EXPERIMENTALIST_API_KEY") name = _optional_env("EXPERIMENTALIST_SMART_MODEL_NAME", "openai/openai/openai/gpt-5.5") - return CompletionClient(name, api_base=api_base, api_key=api_key) + return _completion_client(name, api_base=api_base, api_key=api_key) @functools.cache @@ -49,7 +63,7 @@ def get_mid_model() -> CompletionClient: api_base = _required_env("EXPERIMENTALIST_API_BASE") api_key = _required_env("EXPERIMENTALIST_API_KEY") name = _optional_env("EXPERIMENTALIST_MID_MODEL_NAME", "openai/gcp/google/gemini-3.5-flash") - return CompletionClient(name, api_base=api_base, api_key=api_key) + return _completion_client(name, api_base=api_base, api_key=api_key) @functools.cache @@ -66,7 +80,7 @@ def get_fast_model() -> CompletionClient: api_base = _required_env("EXPERIMENTALIST_API_BASE") api_key = _required_env("EXPERIMENTALIST_API_KEY") name = _optional_env("EXPERIMENTALIST_FAST_MODEL_NAME", "openai/openai/openai/gpt-5-mini") - return CompletionClient(name, api_base=api_base, api_key=api_key) + return _completion_client(name, api_base=api_base, api_key=api_key) def _mask_key(value: str) -> str: diff --git a/plugins/nemo-experimentalist/tests/experimentalist/test_model_config.py b/plugins/nemo-experimentalist/tests/experimentalist/test_model_config.py index b3eddfa1bc..15fe96608c 100644 --- a/plugins/nemo-experimentalist/tests/experimentalist/test_model_config.py +++ b/plugins/nemo-experimentalist/tests/experimentalist/test_model_config.py @@ -1,7 +1,11 @@ # SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 -from nemo_experimentalist_plugin.experimentalist.components.model_config import get_mid_model, log_model_config +from nemo_experimentalist_plugin.experimentalist.components.model_config import ( + _completion_client, + get_mid_model, + log_model_config, +) def test_mid_model_default_uses_openai_provider_for_gateway(monkeypatch) -> None: @@ -16,4 +20,15 @@ def test_mid_model_default_uses_openai_provider_for_gateway(monkeypatch) -> None get_mid_model.cache_clear() assert client.model == "openai/gcp/google/gemini-3.5-flash" + assert client.cache_control_injection_points == [] assert "mid model: openai/gcp/google/gemini-3.5-flash" in log_model_config() + + +def test_openai_model_keeps_nooa_cache_control_defaults() -> None: + client = _completion_client( + "openai/openai/openai/gpt-5-mini", + api_base="https://example.test/v1", + api_key="test-key", + ) + + assert client.cache_control_injection_points From 6834fa8e4691f6505d49c469eebc2e7d3c0451b2 Mon Sep 17 00:00:00 2001 From: Gaia Di Lorenzo Date: Fri, 31 Jul 2026 14:24:00 +0200 Subject: [PATCH 05/14] docs: fix agent optimization guide Signed-off-by: Gaia Di Lorenzo --- docs/get-started/example-agent.mdx | 115 ++++++++++++++++++++--------- 1 file changed, 80 insertions(+), 35 deletions(-) diff --git a/docs/get-started/example-agent.mdx b/docs/get-started/example-agent.mdx index 226f334cb9..6c71afcdac 100644 --- a/docs/get-started/example-agent.mdx +++ b/docs/get-started/example-agent.mdx @@ -3,7 +3,10 @@ title: "Get started with an example agent" description: "Run the optimization loop from end to end on τ-Bench" --- -This provides a guide for how you can quickly load agent traces into your NeMo Platform, run the analyst agent to discover issues with that agent, and launch an experimentalist to fix those issues. This is meant as an example to show how the whole platform works. You can also jump straight to setting up your agent with NeMo Platform. +This guide runs one Tau3 Airline agent through the full loop: Harbor evaluates +the checked-in example agent, the Analyst discovers issues from its traces, and +the Experimentalist uses one discovered Insight to improve the same agent. +You can also jump straight to setting up your agent with NeMo Platform. ## Prerequisites @@ -32,35 +35,40 @@ services/intake/scripts/spans/run_clickhouse.sh uv run nemo services start --config packages/nmp_platform/config/local.yaml ``` +In order to visualize experiment results, you need to enable the `experiment` feature flag. You can do this by editing the `packages/nmp_platform/config/local.yaml` file and setting the `feature_flags.experiment` to `true`. + +```yaml +feature_flags: + experiment: true +``` + You should now be able to navigate to `http://localhost:8080` and see the NeMo Platform web UI. -## 3. Prepare Tau2 and run the airline agent +## 3. Run the Tau3 Airline agent with Harbor -Now that we have a running NeMo Platform, it's time to load up some data! This example uses the τ-Bench from Sierra. It's a great representation of a simplified agent that performs a business-critical task. Our first step will be to download the tau2 repository and install the dependencies to run it. +Now that NeMo Platform is running, evaluate the checked-in Tau3 NOOA example +agent against `sierra-research/tau3-bench@1` from Harbor Hub. The Insights +testbed downloads the dataset through Harbor, runs six Airline tasks, and +uploads the resulting traces and verifier rewards into NeMo Platform. -Then, we'll run a script to go through 30 of the tasks from the Tau 2 Airline dataset. In order to make our example more realistic and representative of the kind of agent data we typically see in production, we won't record whether the tasks passed verification – we want to see if we can gain useful insights even without ground truth data. This could take up to 20 minutes, and will cost about a dollar. +This can take 20 minutes or longer because Harbor builds task containers and +runs model-backed user simulation. ```bash -git init tmp/tau2-bench -git -C tmp/tau2-bench remote add origin https://github.com/sierra-research/tau2-bench.git -git -C tmp/tau2-bench fetch --depth 1 origin 8ebb7499622fc2be9b9d510d6f7a7653461f4f29 -git -C tmp/tau2-bench checkout --detach FETCH_HEAD -uv --directory tmp/tau2-bench sync --frozen -uv --directory tmp/tau2-bench run --frozen tau2 check-data - export NMP_BASE_URL=http://localhost:8080 export INFERENCE_API_KEY=sk-... export OPENAI_API_KEY="$INFERENCE_API_KEY" -export OPENAI_API_BASE=https://inference-api.nvidia.com/v1 +export OPENAI_BASE_URL=https://inference-api.nvidia.com/v1 uv run --directory plugins/nemo-insights --frozen \ - python -m testbed run tau2-airline \ - --base "$NMP_BASE_URL" \ - --set include_rewards=false \ - --set tau2_repo="$PWD/tmp/tau2-bench" + python -m testbed run tau3-airline-harbor \ + --base "$NMP_BASE_URL" ``` -The testbed runs the 30-task airline train split and ingests realistic traces into the `tau2-airline` workspace. At this point you can navigate to the traces view and see the traces from the agent. +The testbed records the run under +`plugins/nemo-insights/testbed/tmp/tau3-airline-harbor.run.json` and ingests it +into the `canonical-tau3-airline` workspace. In Studio, open that workspace's +Experiments area to inspect task results, rewards, and traces. ## 4. Run the Analyst @@ -68,26 +76,29 @@ Now that we have data in our system, we can run the analyst agent to understand If the analyst discovers issues in your application, it will create what we call an 'insight'. An insight is a human-readable description of a problem in your agentic system. The closest analogy is a bug report. They don't try to describe why an issue happened or the code you should change to fix it, and they should be understandable to a user of your agent. -Let's run it: +Run the testbed's live analysis command. It reads the recorded evaluation ID, +so the Analyst only examines traces from the Harbor run above: ```bash -uv run --frozen nemo insights analyze \ - --agent tau2-airline \ - --workspace tau2-airline \ - --base-url "$NMP_BASE_URL" +uv run --directory plugins/nemo-insights --frozen \ + python -m testbed analyze tau3-airline-harbor --live ``` -This will take a few minutes to run. Once it's done, you can navigate to the insights page to see the issues the analyst discovered in the τ-Bench Airline Agent. +This writes +`plugins/nemo-insights/testbed/tmp/insights_tau3-airline-harbor.yaml`. In +Studio, open the `canonical-tau3-airline` workspace's Insights area to review +the same persisted Insights. -## 5. Prepare a Tau3 Airline smoke dataset +## 5. Prepare the Tau3 Airline quality dataset -The next step is to improve the τ-Bench agent using the experimentalist. The -experimentalist will run your evals, debug failures using trace data, understand -the root cause of the failure and attempt to fix it. After it makes the change, -it will run the evals again to validate whether the change improved the -performance of your agent on the evalaution. +The Experimentalist now improves the same checked-in Tau3 agent. It uses the +first Insight produced above as its failure lens, generates candidate changes, +and evaluates them on explicit train and validation splits from the same Harbor +dataset. -This is a shortened example that only uses a few tasks, but it can still take up to an hour to finish. First, set your environment variables and set up the task dataset: +The quality partition contains 20 training tasks and 10 validation tasks. A full +optimization can take several hours and make many model calls. First, set your +environment variables and prepare the dataset: ```bash export OPENAI_API_KEY="$INFERENCE_API_KEY" @@ -95,29 +106,31 @@ export OPENAI_BASE_URL=https://inference-api.nvidia.com/v1 export TAU2_USER_MODEL=openai/openai/openai/gpt-5-mini export TAU2_NL_ASSERTIONS_MODEL=openai/openai/openai/gpt-5-mini export AUT_MODEL_NAME=openai/openai/openai/gpt-5-mini -export EXPERIMENTALIST_SMART_MODEL_NAME=openai/openai/openai/gpt-5-mini -export EXPERIMENTALIST_MID_MODEL_NAME=openai/openai/openai/gpt-5-mini +export EXPERIMENTALIST_SMART_MODEL_NAME=openai/openai/openai/gpt-5.3-codex +export EXPERIMENTALIST_MID_MODEL_NAME=openai/gcp/google/gemini-3.5-flash export EXPERIMENTALIST_FAST_MODEL_NAME=openai/openai/openai/gpt-5-mini uv run --frozen nemo workspaces create canonical-tau3-airline \ --description "Tau3 Airline Experimentalist runs" \ --exist-ok -plugins/nemo-experimentalist/examples/tau3-nooa-agent/prepare-airline-smoke.sh +plugins/nemo-experimentalist/examples/tau3-nooa-agent/prepare-airline.sh ``` Now we're ready to run the experimentalist! ```bash uv run --frozen nemo experimentalist run \ - --no-insight \ + --insight plugins/nemo-insights/testbed/tmp/insights_tau3-airline-harbor.yaml \ + --insight-id 0 \ --agent plugins/nemo-experimentalist/examples/tau3-nooa-agent \ --agent-spec plugins/nemo-experimentalist/examples/tau3-nooa-agent/AGENT-SPEC.md \ --train-dataset plugins/nemo-experimentalist/tmp/tau3-airline-smoke/train \ --validation-dataset plugins/nemo-experimentalist/tmp/tau3-airline-smoke/validation \ + --task-template plugins/nemo-experimentalist/tmp/tau3-airline-smoke/source/tau3-bench/tau3-bench__tau3-airline-0 \ --workspace canonical-tau3-airline \ --framework-skills plugins/nemo-experimentalist/framework-skills/nooa \ - --config plugins/nemo-experimentalist/examples/tau3-nooa-agent/experimentalist-smoke.yaml \ + --config plugins/nemo-experimentalist/examples/tau3-nooa-agent/experimentalist-quality.yaml \ --experiment-dir plugins/nemo-experimentalist/tmp/tau3-airline-experimentalist \ --base-url "$NMP_BASE_URL" ``` @@ -126,3 +139,35 @@ The trace records include the Experimentalist evaluation ID and Tau3 task ID. After the run completes, inspect `eval-and-optimize/run.json` for the selected winner and compare the `agent-0` and `agent-1` directories to review the code change that was evaluated. + +## Optimize without insights + +If your dataset already includes objective evaluation metrics, you can run the +Experimentalist in Mode 2 without an Analyst-generated Insight. In this mode, +the Harbor verifier rewards from the training dataset drive candidate +selection, while the validation dataset measures whether the selected changes +generalize. + +Pass `--no-insight` explicitly. This prevents the Experimentalist from loading +an Insight supplied by a profile or left over from an earlier Analyst run: + +```bash +uv run --frozen nemo experimentalist run \ + --no-insight \ + --agent plugins/nemo-experimentalist/examples/tau3-nooa-agent \ + --agent-spec plugins/nemo-experimentalist/examples/tau3-nooa-agent/AGENT-SPEC.md \ + --train-dataset plugins/nemo-experimentalist/tmp/tau3-airline-smoke/train \ + --validation-dataset plugins/nemo-experimentalist/tmp/tau3-airline-smoke/validation \ + --workspace canonical-tau3-airline \ + --framework-skills plugins/nemo-experimentalist/framework-skills/nooa \ + --config plugins/nemo-experimentalist/examples/tau3-nooa-agent/experimentalist-quality.yaml \ + --experiment-dir plugins/nemo-experimentalist/tmp/tau3-airline-experimentalist-mode2 \ + --base-url "$NMP_BASE_URL" +``` + +Mode 2 skips the production-issue lens provided by an Insight; it does not skip +evaluation or failure analysis inside the optimization loop. The +Experimentalist still runs Harbor tasks, examines failing trajectories, +proposes agent changes, and compares candidates using the dataset's verifier +metrics. The quality dataset prepared above contains 20 training tasks and 10 +validation tasks. From 3a7fe9175e3dba8762d3cb193fa60bf7d95d25c6 Mon Sep 17 00:00:00 2001 From: Gaia Di Lorenzo Date: Fri, 31 Jul 2026 14:45:19 +0200 Subject: [PATCH 06/14] fix(insights): select Tau3 airline tasks Signed-off-by: Gaia Di Lorenzo --- plugins/nemo-insights/testbed/README.md | 2 +- plugins/nemo-insights/testbed/adapters.py | 19 ++++++++++++++++++- plugins/nemo-insights/testbed/testbeds.toml | 9 ++++++++- .../tests/testbed/test_adapters.py | 4 ++-- .../tests/testbed/test_registry.py | 8 ++++++++ 5 files changed, 37 insertions(+), 5 deletions(-) diff --git a/plugins/nemo-insights/testbed/README.md b/plugins/nemo-insights/testbed/README.md index e1cd3b6d90..e0b6f876aa 100644 --- a/plugins/nemo-insights/testbed/README.md +++ b/plugins/nemo-insights/testbed/README.md @@ -199,7 +199,7 @@ exactly what to install/set (`✓ ready` or `✗ needs: …`). Subjects live in `testbeds.toml` — one table per subject, keyed by `type`: - `type = "intake"` — analyze an agent's existing Intake traces (config: `agent`, `workspace`, `base_url`, optional `since`). - `type = "benchmark"` — run a benchmark to produce traces, ingest them into Intake, then analyze (config: `domain`, `base_url`, `workspace`, `agent_llm`, `user_llm`, `task_split_name`, `num_trials`, `max_concurrency`, `seed`, optional `num_tasks`/`timeout`/`include_rewards`). -- `type = "harbor"` — run a Harbor dataset against an importable agent wrapper, enrich and ingest its OTLP traces, then analyze the recorded evaluation. Use exactly one of `dataset`, `dataset_ref`, or `dataset_id`. +- `type = "harbor"` — run a Harbor dataset against an importable agent wrapper, enrich and ingest its OTLP traces and verifier rewards, then analyze the recorded evaluation. Use exactly one of `dataset`, `dataset_ref`, or `dataset_id`; optionally select tasks with `task_names` or `num_tasks`. `--since` (analyze `--live`, snapshot) accepts `Nd`/`Nh`/`Nm` (days/hours/minutes) or an ISO date; `--since ''` means no lower bound (the epoch). Insights are diff --git a/plugins/nemo-insights/testbed/adapters.py b/plugins/nemo-insights/testbed/adapters.py index c7826e8b32..d0dac5a068 100644 --- a/plugins/nemo-insights/testbed/adapters.py +++ b/plugins/nemo-insights/testbed/adapters.py @@ -554,8 +554,23 @@ def _build_dataset_config(cls, cfg: Mapping[str, object], *, repo_root: Path): raise ValueError("config keys 'registry_path' and 'registry_url' are mutually exclusive") n_tasks = int(str(cfg["num_tasks"])) if cfg.get("num_tasks") is not None else None + raw_task_names = cfg.get("task_names") + if raw_task_names is not None and ( + not isinstance(raw_task_names, list) + or not all(isinstance(task_name, str) and task_name for task_name in raw_task_names) + ): + raise ValueError("config key 'task_names' must be a list of non-empty strings") + task_names = ( + [task_name for task_name in raw_task_names if isinstance(task_name, str)] + if isinstance(raw_task_names, list) + else None + ) if dataset_path: - return DatasetConfig(path=cls._repo_path(dataset_path, repo_root=repo_root), n_tasks=n_tasks) + return DatasetConfig( + path=cls._repo_path(dataset_path, repo_root=repo_root), + n_tasks=n_tasks, + task_names=task_names, + ) registry_path = cfg.get("registry_path") registry_url = str(cfg["registry_url"]) if cfg.get("registry_url") else None @@ -571,6 +586,7 @@ def _build_dataset_config(cls, cfg: Mapping[str, object], *, repo_root: Path): registry_url=registry_url, registry_path=resolved_registry_path, n_tasks=n_tasks, + task_names=task_names, ) return DatasetConfig( name=str(dataset_id), @@ -578,6 +594,7 @@ def _build_dataset_config(cls, cfg: Mapping[str, object], *, repo_root: Path): registry_url=registry_url, registry_path=resolved_registry_path, n_tasks=n_tasks, + task_names=task_names, ) @classmethod diff --git a/plugins/nemo-insights/testbed/testbeds.toml b/plugins/nemo-insights/testbed/testbeds.toml index e016bc8eb9..1074e0d0a2 100644 --- a/plugins/nemo-insights/testbed/testbeds.toml +++ b/plugins/nemo-insights/testbed/testbeds.toml @@ -42,7 +42,14 @@ user_llm = "openai/openai/openai/gpt-5-mini" verifier_llm = "openai/openai/openai/gpt-5-mini" dataset_ref = "sierra-research/tau3-bench@1" registry_url = "https://hub.harborframework.com" -num_tasks = 6 +task_names = [ + "tau3-bench__tau3-airline-0", + "tau3-bench__tau3-airline-1", + "tau3-bench__tau3-airline-4", + "tau3-bench__tau3-airline-5", + "tau3-bench__tau3-airline-9", + "tau3-bench__tau3-airline-10", +] num_trials = 1 max_concurrency = 6 timeout = 3600 diff --git a/plugins/nemo-insights/tests/testbed/test_adapters.py b/plugins/nemo-insights/tests/testbed/test_adapters.py index 10ba434e08..8c786d0c87 100644 --- a/plugins/nemo-insights/tests/testbed/test_adapters.py +++ b/plugins/nemo-insights/tests/testbed/test_adapters.py @@ -330,7 +330,7 @@ def test_harbor_dataset_config_accepts_hub_ref(): { "dataset_ref": "sierra-research/tau3-bench@1", "registry_url": "https://hub.harborframework.com", - "num_tasks": 6, + "task_names": ["tau3-bench__tau3-airline-0", "tau3-bench__tau3-airline-1"], }, repo_root=Path("/repo"), ) @@ -338,7 +338,7 @@ def test_harbor_dataset_config_accepts_hub_ref(): assert config.name == "sierra-research/tau3-bench" assert config.version == "1" assert config.registry_url == "https://hub.harborframework.com" - assert config.n_tasks == 6 + assert config.task_names == ["tau3-bench__tau3-airline-0", "tau3-bench__tau3-airline-1"] def test_harbor_dataset_config_rejects_ambiguous_source(): diff --git a/plugins/nemo-insights/tests/testbed/test_registry.py b/plugins/nemo-insights/tests/testbed/test_registry.py index 4a1fbce4ff..4b6fafcacd 100644 --- a/plugins/nemo-insights/tests/testbed/test_registry.py +++ b/plugins/nemo-insights/tests/testbed/test_registry.py @@ -71,6 +71,14 @@ def test_tau3_airline_harbor_uses_checked_in_agent_and_hub_dataset() -> None: assert subject.type == "harbor" assert subject.config["agent_dir"] == "../nemo-experimentalist/examples/tau3-nooa-agent" assert subject.config["dataset_ref"] == "sierra-research/tau3-bench@1" + assert subject.config["task_names"] == [ + "tau3-bench__tau3-airline-0", + "tau3-bench__tau3-airline-1", + "tau3-bench__tau3-airline-4", + "tau3-bench__tau3-airline-5", + "tau3-bench__tau3-airline-9", + "tau3-bench__tau3-airline-10", + ] def test_every_analyzable_subject_has_expected_state_pin() -> None: From b415fb648845a81f3e73b789e52184de69dc7b7c Mon Sep 17 00:00:00 2001 From: Gaia Di Lorenzo Date: Fri, 31 Jul 2026 14:45:25 +0200 Subject: [PATCH 07/14] docs: explain Tau3 Harbor dataset Signed-off-by: Gaia Di Lorenzo --- docs/get-started/example-agent.mdx | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/docs/get-started/example-agent.mdx b/docs/get-started/example-agent.mdx index 6c71afcdac..56b130c243 100644 --- a/docs/get-started/example-agent.mdx +++ b/docs/get-started/example-agent.mdx @@ -47,9 +47,14 @@ You should now be able to navigate to `http://localhost:8080` and see the NeMo P ## 3. Run the Tau3 Airline agent with Harbor Now that NeMo Platform is running, evaluate the checked-in Tau3 NOOA example -agent against `sierra-research/tau3-bench@1` from Harbor Hub. The Insights -testbed downloads the dataset through Harbor, runs six Airline tasks, and -uploads the resulting traces and verifier rewards into NeMo Platform. +agent against version 1 of the +[`sierra-research/tau3-bench`](https://hub.harborframework.com/datasets/sierra-research/tau3-bench/latest) +dataset on Harbor Hub. Tau3 simulates customer-service conversations across +airline, retail, telecom, and banking domains. Each task gives the agent a user +request, domain policy, and domain-specific tools, then uses a verifier to score +the outcome. This guide selects six Airline tasks from the larger dataset. The +Insights testbed runs them and uploads their traces and verifier rewards into +NeMo Platform. This can take 20 minutes or longer because Harbor builds task containers and runs model-backed user simulation. From 8ec2c6f026c0c19ec58b2449f2391e6ae88dadce Mon Sep 17 00:00:00 2001 From: Gaia Di Lorenzo Date: Fri, 31 Jul 2026 14:47:50 +0200 Subject: [PATCH 08/14] docs: explain Tau3 NOOA example Signed-off-by: Gaia Di Lorenzo --- docs/get-started/example-agent.mdx | 19 +++++++++++-------- 1 file changed, 11 insertions(+), 8 deletions(-) diff --git a/docs/get-started/example-agent.mdx b/docs/get-started/example-agent.mdx index 56b130c243..8df4dee514 100644 --- a/docs/get-started/example-agent.mdx +++ b/docs/get-started/example-agent.mdx @@ -46,15 +46,18 @@ You should now be able to navigate to `http://localhost:8080` and see the NeMo P ## 3. Run the Tau3 Airline agent with Harbor -Now that NeMo Platform is running, evaluate the checked-in Tau3 NOOA example -agent against version 1 of the +Now that NeMo Platform is running, evaluate the +[checked-in Tau3 NOOA example agent](https://github.com/NVIDIA-NeMo/nemo-platform/tree/main/plugins/nemo-experimentalist/examples/tau3-nooa-agent) +against version 1 of the [`sierra-research/tau3-bench`](https://hub.harborframework.com/datasets/sierra-research/tau3-bench/latest) -dataset on Harbor Hub. Tau3 simulates customer-service conversations across -airline, retail, telecom, and banking domains. Each task gives the agent a user -request, domain policy, and domain-specific tools, then uses a verifier to score -the outcome. This guide selects six Airline tasks from the larger dataset. The -Insights testbed runs them and uploads their traces and verifier rewards into -NeMo Platform. +dataset on Harbor Hub. The example is a customer-service agent built with +NVIDIA-labs OO Agents (NOOA) and a CodeAct strategy. It receives the Airline +policy at runtime and uses MCP tools to look up records, talk to the simulated +customer, and complete permitted actions. Its `AGENT-SPEC.md` defines the +behavior that the Experimentalist later optimizes. + +The Insights testbed selects six Airline tasks, runs the agent through Harbor, +and uploads the resulting traces and verifier rewards into NeMo Platform. This can take 20 minutes or longer because Harbor builds task containers and runs model-backed user simulation. From 5ad0d3b796bc69799a16933d24505023e5d927b5 Mon Sep 17 00:00:00 2001 From: Gaia Di Lorenzo Date: Fri, 31 Jul 2026 14:49:08 +0200 Subject: [PATCH 09/14] docs: direct optimization review to Studio Signed-off-by: Gaia Di Lorenzo --- docs/get-started/example-agent.mdx | 28 +++++++++++----------------- 1 file changed, 11 insertions(+), 17 deletions(-) diff --git a/docs/get-started/example-agent.mdx b/docs/get-started/example-agent.mdx index 8df4dee514..ef8ac897ef 100644 --- a/docs/get-started/example-agent.mdx +++ b/docs/get-started/example-agent.mdx @@ -73,10 +73,8 @@ uv run --directory plugins/nemo-insights --frozen \ --base "$NMP_BASE_URL" ``` -The testbed records the run under -`plugins/nemo-insights/testbed/tmp/tau3-airline-harbor.run.json` and ingests it -into the `canonical-tau3-airline` workspace. In Studio, open that workspace's -Experiments area to inspect task results, rewards, and traces. +In Studio, open the `canonical-tau3-airline` workspace's Experiments area to +inspect the task results, rewards, and traces. ## 4. Run the Analyst @@ -92,15 +90,14 @@ uv run --directory plugins/nemo-insights --frozen \ python -m testbed analyze tau3-airline-harbor --live ``` -This writes -`plugins/nemo-insights/testbed/tmp/insights_tau3-airline-harbor.yaml`. In -Studio, open the `canonical-tau3-airline` workspace's Insights area to review -the same persisted Insights. +In Studio, open the `canonical-tau3-airline` workspace's Insights area to +review the persisted Insights. Select the Insight you want to optimize and copy +its ID for the next command. ## 5. Prepare the Tau3 Airline quality dataset The Experimentalist now improves the same checked-in Tau3 agent. It uses the -first Insight produced above as its failure lens, generates candidate changes, +Insight selected above as its failure lens, generates candidate changes, and evaluates them on explicit train and validation splits from the same Harbor dataset. @@ -128,9 +125,10 @@ plugins/nemo-experimentalist/examples/tau3-nooa-agent/prepare-airline.sh Now we're ready to run the experimentalist! ```bash +export PLATFORM_INSIGHT_ID=insight-id-from-studio + uv run --frozen nemo experimentalist run \ - --insight plugins/nemo-insights/testbed/tmp/insights_tau3-airline-harbor.yaml \ - --insight-id 0 \ + --insight "$PLATFORM_INSIGHT_ID" \ --agent plugins/nemo-experimentalist/examples/tau3-nooa-agent \ --agent-spec plugins/nemo-experimentalist/examples/tau3-nooa-agent/AGENT-SPEC.md \ --train-dataset plugins/nemo-experimentalist/tmp/tau3-airline-smoke/train \ @@ -139,14 +137,11 @@ uv run --frozen nemo experimentalist run \ --workspace canonical-tau3-airline \ --framework-skills plugins/nemo-experimentalist/framework-skills/nooa \ --config plugins/nemo-experimentalist/examples/tau3-nooa-agent/experimentalist-quality.yaml \ - --experiment-dir plugins/nemo-experimentalist/tmp/tau3-airline-experimentalist \ --base-url "$NMP_BASE_URL" ``` -The trace records include the Experimentalist evaluation ID and Tau3 task ID. -After the run completes, inspect `eval-and-optimize/run.json` for the selected -winner and compare the `agent-0` and `agent-1` directories to review the code -change that was evaluated. +After the run completes, open the workspace's Experiments area in Studio to +review its evaluations, compare candidates, and inspect the selected winner. ## Optimize without insights @@ -169,7 +164,6 @@ uv run --frozen nemo experimentalist run \ --workspace canonical-tau3-airline \ --framework-skills plugins/nemo-experimentalist/framework-skills/nooa \ --config plugins/nemo-experimentalist/examples/tau3-nooa-agent/experimentalist-quality.yaml \ - --experiment-dir plugins/nemo-experimentalist/tmp/tau3-airline-experimentalist-mode2 \ --base-url "$NMP_BASE_URL" ``` From a8ec8edcdc904ad350da2458b37dc02cd440f469 Mon Sep 17 00:00:00 2001 From: Gaia Di Lorenzo Date: Fri, 31 Jul 2026 14:50:36 +0200 Subject: [PATCH 10/14] docs: publish example Insights to Studio Signed-off-by: Gaia Di Lorenzo --- docs/get-started/example-agent.mdx | 16 +++++++++------- 1 file changed, 9 insertions(+), 7 deletions(-) diff --git a/docs/get-started/example-agent.mdx b/docs/get-started/example-agent.mdx index ef8ac897ef..83f1eb2593 100644 --- a/docs/get-started/example-agent.mdx +++ b/docs/get-started/example-agent.mdx @@ -82,17 +82,19 @@ Now that we have data in our system, we can run the analyst agent to understand If the analyst discovers issues in your application, it will create what we call an 'insight'. An insight is a human-readable description of a problem in your agentic system. The closest analogy is a bug report. They don't try to describe why an issue happened or the code you should change to fix it, and they should be understandable to a user of your agent. -Run the testbed's live analysis command. It reads the recorded evaluation ID, -so the Analyst only examines traces from the Harbor run above: +Run the Analyst against the agent's traces in the dedicated workspace: ```bash -uv run --directory plugins/nemo-insights --frozen \ - python -m testbed analyze tau3-airline-harbor --live +uv run --frozen nemo insights analyze \ + --agent nemo-experimentalist-tau3-nooa \ + --workspace canonical-tau3-airline \ + --base-url "$NMP_BASE_URL" ``` -In Studio, open the `canonical-tau3-airline` workspace's Insights area to -review the persisted Insights. Select the Insight you want to optimize and copy -its ID for the next command. +Because this command does not set `--insights-file-output`, it publishes the +Insights to NeMo Platform. In Studio, open the `canonical-tau3-airline` +workspace's Insights area to review them. Select the Insight you want to +optimize and copy its ID for the next command. ## 5. Prepare the Tau3 Airline quality dataset From 3a2eb44d9ceb03d577c67c31bce0c075a1012385 Mon Sep 17 00:00:00 2001 From: Gaia Di Lorenzo Date: Fri, 31 Jul 2026 14:57:21 +0200 Subject: [PATCH 11/14] docs: link NOOA example framework Signed-off-by: Gaia Di Lorenzo --- docs/get-started/example-agent.mdx | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/docs/get-started/example-agent.mdx b/docs/get-started/example-agent.mdx index 83f1eb2593..da752b5a5d 100644 --- a/docs/get-started/example-agent.mdx +++ b/docs/get-started/example-agent.mdx @@ -51,7 +51,8 @@ Now that NeMo Platform is running, evaluate the against version 1 of the [`sierra-research/tau3-bench`](https://hub.harborframework.com/datasets/sierra-research/tau3-bench/latest) dataset on Harbor Hub. The example is a customer-service agent built with -NVIDIA-labs OO Agents (NOOA) and a CodeAct strategy. It receives the Airline +[NVIDIA-labs OO Agents (NOOA)](https://github.com/NVIDIA-NeMo/labs-OO-Agents) +and a CodeAct strategy. It receives the Airline policy at runtime and uses MCP tools to look up records, talk to the simulated customer, and complete permitted actions. Its `AGENT-SPEC.md` defines the behavior that the Experimentalist later optimizes. From ac322022d9b957080c4de96b77fa0274dac74833 Mon Sep 17 00:00:00 2001 From: Gaia Di Lorenzo Date: Fri, 31 Jul 2026 15:03:01 +0200 Subject: [PATCH 12/14] fix(experimentalist): use CPU concurrency for smoke Signed-off-by: Gaia Di Lorenzo --- plugins/nemo-experimentalist/benchmarks/configs/tau3-smoke.yaml | 1 - .../examples/tau3-nooa-agent/experimentalist-smoke.yaml | 1 - 2 files changed, 2 deletions(-) diff --git a/plugins/nemo-experimentalist/benchmarks/configs/tau3-smoke.yaml b/plugins/nemo-experimentalist/benchmarks/configs/tau3-smoke.yaml index c4be604fb3..1184c51b14 100644 --- a/plugins/nemo-experimentalist/benchmarks/configs/tau3-smoke.yaml +++ b/plugins/nemo-experimentalist/benchmarks/configs/tau3-smoke.yaml @@ -21,7 +21,6 @@ optimizer: disable_convergence_check: true evaluator: n_attempts: 1 - n_concurrent_trials: 1 quiet: true agent_setup_timeout_multiplier: 2.0 # tau3 builds two images per task, and the first build clones and installs diff --git a/plugins/nemo-experimentalist/examples/tau3-nooa-agent/experimentalist-smoke.yaml b/plugins/nemo-experimentalist/examples/tau3-nooa-agent/experimentalist-smoke.yaml index 2dfa4f9721..e1ea789f89 100644 --- a/plugins/nemo-experimentalist/examples/tau3-nooa-agent/experimentalist-smoke.yaml +++ b/plugins/nemo-experimentalist/examples/tau3-nooa-agent/experimentalist-smoke.yaml @@ -12,7 +12,6 @@ disable_trajectory_scoring: true disable_convergence_check: true evaluator: n_attempts: 1 - n_concurrent_trials: 1 quiet: true agent_setup_timeout_multiplier: 2.0 environment_build_timeout_multiplier: 3.0 From 8387d6653f6ec875688a20dd984b30b10d237c10 Mon Sep 17 00:00:00 2001 From: Gaia Di Lorenzo Date: Fri, 31 Jul 2026 15:03:07 +0200 Subject: [PATCH 13/14] docs: use smoke optimizer in getting started Signed-off-by: Gaia Di Lorenzo --- docs/get-started/example-agent.mdx | 29 ++++++++++++++--------------- 1 file changed, 14 insertions(+), 15 deletions(-) diff --git a/docs/get-started/example-agent.mdx b/docs/get-started/example-agent.mdx index da752b5a5d..029bc0f34e 100644 --- a/docs/get-started/example-agent.mdx +++ b/docs/get-started/example-agent.mdx @@ -48,7 +48,7 @@ You should now be able to navigate to `http://localhost:8080` and see the NeMo P Now that NeMo Platform is running, evaluate the [checked-in Tau3 NOOA example agent](https://github.com/NVIDIA-NeMo/nemo-platform/tree/main/plugins/nemo-experimentalist/examples/tau3-nooa-agent) -against version 1 of the +against the [`sierra-research/tau3-bench`](https://hub.harborframework.com/datasets/sierra-research/tau3-bench/latest) dataset on Harbor Hub. The example is a customer-service agent built with [NVIDIA-labs OO Agents (NOOA)](https://github.com/NVIDIA-NeMo/labs-OO-Agents) @@ -92,10 +92,9 @@ uv run --frozen nemo insights analyze \ --base-url "$NMP_BASE_URL" ``` -Because this command does not set `--insights-file-output`, it publishes the -Insights to NeMo Platform. In Studio, open the `canonical-tau3-airline` -workspace's Insights area to review them. Select the Insight you want to -optimize and copy its ID for the next command. +The Insights are published to NeMo Platform. In Studio, open the +`canonical-tau3-airline` workspace's Insights area to review them. Select an +Insight and copy its ID for the Experimentalist command below. ## 5. Prepare the Tau3 Airline quality dataset @@ -104,8 +103,9 @@ Insight selected above as its failure lens, generates candidate changes, and evaluates them on explicit train and validation splits from the same Harbor dataset. -The quality partition contains 20 training tasks and 10 validation tasks. A full -optimization can take several hours and make many model calls. First, set your +The partition contains 20 training tasks and 10 validation tasks. The smoke +configuration used below runs one optimization round and samples at most four +training tasks, keeping this getting-started run bounded. First, set your environment variables and prepare the dataset: ```bash @@ -139,17 +139,17 @@ uv run --frozen nemo experimentalist run \ --task-template plugins/nemo-experimentalist/tmp/tau3-airline-smoke/source/tau3-bench/tau3-bench__tau3-airline-0 \ --workspace canonical-tau3-airline \ --framework-skills plugins/nemo-experimentalist/framework-skills/nooa \ - --config plugins/nemo-experimentalist/examples/tau3-nooa-agent/experimentalist-quality.yaml \ + --config plugins/nemo-experimentalist/examples/tau3-nooa-agent/experimentalist-smoke.yaml \ --base-url "$NMP_BASE_URL" ``` After the run completes, open the workspace's Experiments area in Studio to review its evaluations, compare candidates, and inspect the selected winner. -## Optimize without insights +## 6. (Optional) Optimize without insights If your dataset already includes objective evaluation metrics, you can run the -Experimentalist in Mode 2 without an Analyst-generated Insight. In this mode, +Experimentalist without an Analyst-generated Insight. In this mode, the Harbor verifier rewards from the training dataset drive candidate selection, while the validation dataset measures whether the selected changes generalize. @@ -166,13 +166,12 @@ uv run --frozen nemo experimentalist run \ --validation-dataset plugins/nemo-experimentalist/tmp/tau3-airline-smoke/validation \ --workspace canonical-tau3-airline \ --framework-skills plugins/nemo-experimentalist/framework-skills/nooa \ - --config plugins/nemo-experimentalist/examples/tau3-nooa-agent/experimentalist-quality.yaml \ + --config plugins/nemo-experimentalist/examples/tau3-nooa-agent/experimentalist-smoke.yaml \ --base-url "$NMP_BASE_URL" ``` -Mode 2 skips the production-issue lens provided by an Insight; it does not skip -evaluation or failure analysis inside the optimization loop. The +The Experimentalist skips the production-issue lens provided by an Insight; it +does not skip evaluation or failure analysis inside the optimization loop. The Experimentalist still runs Harbor tasks, examines failing trajectories, proposes agent changes, and compares candidates using the dataset's verifier -metrics. The quality dataset prepared above contains 20 training tasks and 10 -validation tasks. +metrics. From fb036100a5202b05bda07cb9122b38e703575926 Mon Sep 17 00:00:00 2001 From: Gaia Di Lorenzo Date: Fri, 31 Jul 2026 15:06:25 +0200 Subject: [PATCH 14/14] revert(experimentalist): remove Harbor SIGINT cleanup Signed-off-by: Gaia Di Lorenzo --- .../components/evaluator/harbor.py | 102 +----------------- .../experimentalist/test_evaluator_harbor.py | 90 ---------------- 2 files changed, 2 insertions(+), 190 deletions(-) diff --git a/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/experimentalist/components/evaluator/harbor.py b/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/experimentalist/components/evaluator/harbor.py index 1fcfa565af..24a60ffa37 100644 --- a/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/experimentalist/components/evaluator/harbor.py +++ b/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/experimentalist/components/evaluator/harbor.py @@ -32,7 +32,6 @@ from harbor.models.task.task import Task as HarborTaskModel from harbor.models.trial.config import ServiceVolumeConfig from harbor.models.trial.paths import EnvironmentPaths, TrialPaths -from harbor.trial.hooks import TrialHookEvent from nemo_experimentalist_plugin.experimentalist.components.evaluator.base import ( Evaluator, EvaluatorConfig, @@ -114,8 +113,6 @@ class HarborResourceSpec(TypedDict): _TRACE_ARTIFACT_SOURCE = "/app/traces" _TRACE_ARTIFACT_DESTINATION = "traces" _SHELL_SYNTAX_TIMEOUT_SEC = 10.0 -_DOCKER_CLEANUP_TIMEOUT_SEC = 30.0 -_DOCKER_COMPOSE_PROJECT_LABEL = "com.docker.compose.project" _AGENT_IMPORT_ROOT = "_nemo_experimentalist_eval_agents" _IDENTIFIER_RE = re.compile(r"\W+") _TRIAL_LOG_DESCRIPTIONS = { @@ -132,88 +129,6 @@ class HarborResourceSpec(TypedDict): logger = logging.getLogger(__name__) -def _sanitize_compose_project_name(name: str) -> str: - normalized = name.lower() - if not re.match(r"^[a-z0-9]", normalized): - normalized = f"0{normalized}" - return re.sub(r"[^a-z0-9_-]", "-", normalized) - - -async def _docker_cleanup_command(args: Sequence[str]) -> str: - docker_path = shutil.which("docker", path=os.defpath) - if docker_path is None: - raise FileNotFoundError("docker executable not found on the system path") - - process = await asyncio.create_subprocess_exec( - docker_path, - *args, - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.STDOUT, - ) - try: - stdout, _ = await asyncio.wait_for( - process.communicate(), - timeout=_DOCKER_CLEANUP_TIMEOUT_SEC, - ) - except (TimeoutError, asyncio.CancelledError): - if process.returncode is None: - try: - process.kill() - except ProcessLookupError: - pass - await process.communicate() - raise - - output = stdout.decode("utf-8", errors="replace").strip() - if process.returncode != 0: - raise RuntimeError(output or f"docker {' '.join(args)} exited with status {process.returncode}") - return output - - -async def _cleanup_cancelled_harbor_projects(trial_names: set[str]) -> None: - """Remove Docker Compose resources left by cancelled Harbor trials.""" - if not trial_names: - return - - project_prefixes = tuple(f"{_sanitize_compose_project_name(name)}__" for name in trial_names) - resource_specs = ( - ("container", ".ID", ("container", "rm", "--force")), - ("network", ".ID", ("network", "rm")), - ("volume", ".Name", ("volume", "rm", "--force")), - ) - - for resource, id_template, remove_command in resource_specs: - list_command = [resource, "ls"] - if resource == "container": - list_command.append("--all") - list_command.extend( - ( - "--filter", - f"label={_DOCKER_COMPOSE_PROJECT_LABEL}", - "--format", - f'{{{{{id_template}}}}}\t{{{{.Label "{_DOCKER_COMPOSE_PROJECT_LABEL}"}}}}', - ) - ) - try: - rows = await _docker_cleanup_command(list_command) - except Exception as exc: - logger.warning("Could not list Harbor %s resources during cancellation cleanup: %s", resource, exc) - continue - - resource_ids = [] - for row in rows.splitlines(): - resource_id, separator, project = row.partition("\t") - if separator and any(project.startswith(prefix) for prefix in project_prefixes): - resource_ids.append(resource_id) - if not resource_ids: - continue - - try: - await _docker_cleanup_command((*remove_command, *resource_ids)) - except Exception as exc: - logger.warning("Could not remove cancelled Harbor %s resources: %s", resource, exc) - - @dataclass(frozen=True) class HarborVerifierValidationFailure: """Syntax failure found in one task's Harbor verifier.""" @@ -1332,11 +1247,9 @@ class HarborEvaluator(Evaluator): def __init__(self, options: HarborEvaluatorConfig | None = None, experiment_dir: Path | None = None) -> None: super().__init__(options or HarborEvaluatorConfig(), experiment_dir=experiment_dir) - async def _run(self, agent: Path, dataset: Dataset, options: EvaluatorConfig) -> Sequence[TrialResult]: + async def _run(self, agent: Path, dataset: Dataset, options: HarborEvaluatorConfig) -> Sequence[TrialResult]: if not isinstance(dataset, HarborDataset): raise ValueError("Dataset must be a Harbor dataset") - if not isinstance(options, HarborEvaluatorConfig): - raise ValueError("Options must be a Harbor evaluator config") if dataset.source is None: raise ValueError("Harbor dataset source is required") @@ -1369,18 +1282,7 @@ async def _run(self, agent: Path, dataset: Dataset, options: EvaluatorConfig) -> try: job = await Job.create(job_config) - started_trial_names: set[str] = set() - - async def track_started_trial(event: TrialHookEvent) -> None: - started_trial_names.add(event.trial_name) - - job.on_trial_started(track_started_trial) - try: - await job.run() - except asyncio.CancelledError: - cleanup_task = asyncio.create_task(_cleanup_cancelled_harbor_projects(started_trial_names)) - await asyncio.shield(cleanup_task) - raise + await job.run() finally: _cleanup_scoped_imports(scoped_package) diff --git a/plugins/nemo-experimentalist/tests/experimentalist/test_evaluator_harbor.py b/plugins/nemo-experimentalist/tests/experimentalist/test_evaluator_harbor.py index 955aeb3db6..f869a74586 100644 --- a/plugins/nemo-experimentalist/tests/experimentalist/test_evaluator_harbor.py +++ b/plugins/nemo-experimentalist/tests/experimentalist/test_evaluator_harbor.py @@ -20,7 +20,6 @@ HarborEvaluatorConfig, HarborVerifierValidationError, _chmod_path_chain, - _cleanup_cancelled_harbor_projects, _cleanup_scoped_imports, _ensure_package, _python_syntax_failure, @@ -96,7 +95,6 @@ class RecordingJob: def __init__(self, config) -> None: self.config = config self.job_dir = job_dir - self.started_hook = None @classmethod async def create(cls, config): @@ -108,10 +106,6 @@ async def run(self): type(self).run_calls += 1 return SimpleNamespace(id="job-id", stats=None) - def on_trial_started(self, callback): - self.started_hook = callback - return self - return RecordingJob @@ -726,10 +720,6 @@ async def create(cls, config): async def run(self): return SimpleNamespace(id="job-id", stats=FakeStats()) - def on_trial_started(self, callback): - self.started_hook = callback - return self - monkeypatch.setattr("nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor.Job", FakeJob) result = await evaluator.run( @@ -985,82 +975,6 @@ async def test_harbor_evaluator_accepts_valid_python_verifier( assert fake_job.run_calls == 1 -@pytest.mark.asyncio -async def test_harbor_evaluator_cleans_started_projects_on_cancellation( - tmp_path: Path, - monkeypatch: pytest.MonkeyPatch, -) -> None: - agent_dir = tmp_path / "agent" - agent_dir.mkdir() - task_dir = tmp_path / "task-a" - _write(task_dir / "task.toml", "") - dataset = HarborDataset.from_path(task_dir) - cleaned: list[set[str]] = [] - - class CancelledJob: - def __init__(self, config) -> None: - self.config = config - self.job_dir = tmp_path / "jobs" / "cancelled" - self.started_hook = None - - @classmethod - async def create(cls, config): - return cls(config) - - def on_trial_started(self, callback): - self.started_hook = callback - return self - - async def run(self): - assert self.started_hook is not None - await self.started_hook(SimpleNamespace(trial_name="task-a__abc123")) - raise asyncio.CancelledError - - async def fake_cleanup(trial_names: set[str]) -> None: - cleaned.append(trial_names) - - monkeypatch.setattr("nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor.Job", CancelledJob) - monkeypatch.setattr( - "nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor._cleanup_cancelled_harbor_projects", - fake_cleanup, - ) - - with pytest.raises(asyncio.CancelledError): - await HarborEvaluator()._run(agent_dir, dataset, HarborEvaluatorConfig()) - - assert cleaned == [{"task-a__abc123"}] - - -@pytest.mark.asyncio -async def test_cancelled_harbor_cleanup_removes_only_matching_compose_projects( - monkeypatch: pytest.MonkeyPatch, -) -> None: - calls: list[tuple[str, ...]] = [] - - async def fake_docker_command(args) -> str: - command = tuple(args) - calls.append(command) - if command[:2] == ("container", "ls"): - return "c1\ttask-a__abc123__env\nc2\tother-task__xyz__env" - if command[:2] == ("network", "ls"): - return "n1\ttask-a__abc123__env" - if command[:2] == ("volume", "ls"): - return "v1\ttask-a__abc123__verifier__step" - return "" - - monkeypatch.setattr( - "nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor._docker_cleanup_command", - fake_docker_command, - ) - - await _cleanup_cancelled_harbor_projects({"task-a__abc123"}) - - assert ("container", "rm", "--force", "c1") in calls - assert ("network", "rm", "n1") in calls - assert ("volume", "rm", "--force", "v1") in calls - assert all("c2" not in command for command in calls) - - @pytest.mark.asyncio async def test_harbor_evaluator_rejects_invalid_configured_test_sh_before_job_create( tmp_path: Path, @@ -1505,10 +1419,6 @@ async def create(cls, config): async def run(self): return SimpleNamespace(id="job-id", stats=None) - def on_trial_started(self, callback): - self.started_hook = callback - return self - monkeypatch.setattr("nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor.Job", FakeJob) trials = await evaluator._run(