diff --git a/dev/benchmark/benchmark_loaders.py b/dev/benchmark/benchmark_loaders.py index af1e48d..ca62261 100644 --- a/dev/benchmark/benchmark_loaders.py +++ b/dev/benchmark/benchmark_loaders.py @@ -1,4 +1,4 @@ -from magnet.helm_outputs import HelmRun +from magnet.backends.helm.helm_outputs import HelmRun import ubelt as ub import timerit from typing import Iterator diff --git a/dev/oneoff/check_results_overlap.py b/dev/oneoff/check_results_overlap.py deleted file mode 100644 index 7b5cf3e..0000000 --- a/dev/oneoff/check_results_overlap.py +++ /dev/null @@ -1,371 +0,0 @@ -""" -!uv pip install kaleido plotly -""" - -import pandas as pd -import ubelt as ub -import kwutil -from magnet.backends.helm.helm_outputs import HelmRun -from magnet.backends.helm.helm_outputs import HelmOutputs -from magnet.backends.helm.helm_run_analysis import HelmRunAnalysis -from magnet.backends.helm.helm_run_diff import HelmRunDiff -from magnet.utils import sankey - -""" -!python ~/code/aiq-magnet/dev/poc/inspect_historic_helm_runs.py /data/crfm-helm-public --out_fpath run_specs.yaml --out_detail_fpath run_details.yaml -""" -helm_rows = kwutil.Yaml.load('run_details.yaml') - -if 0: - # Debug HelmRunAnalysis - run_dir = '/data/crfm-helm-public/classic/benchmark_output/runs/v0.3.0/wikifact:k=5,subject=symptoms_and_signs,model=lmsys_vicuna-7b-v1.3' - helm_run = HelmRun.coerce(run_dir) - self = HelmRunAnalysis(helm_run) - self.summary(level=10) - run_dir = '/data/crfm-helm-public/classic/benchmark_output/runs/v0.2.4/boolq:model=eleutherai_pythia-2.8b-v0,data_augmentation=canonical/' - helm_run = HelmRun.coerce(run_dir) - self = HelmRunAnalysis(helm_run) - self.summary(level=10) - run_dir = '/data/crfm-helm-public/capabilities/benchmark_output/runs/v1.12.0/ifeval:model=openai_gpt-oss-20b/' - helm_run = HelmRun.coerce(run_dir) - self = HelmRunAnalysis(helm_run) - self.summary(level=10) - - for helm_row in helm_rows: - run_dir = ub.Path(helm_row['run_dir']) - helm_run = HelmRun.coerce(run_dir) - self = HelmRunAnalysis(helm_run) - -finished_jobs = list( - ub.Path('/home/local/KHQ/jon.crall/code/aiq-magnet/results/helm').glob( - '*/DONE' - ) -) -kwdagger_rows = [] -for fpath in finished_jobs: - config = kwutil.Json.coerce(fpath.parent / 'job_config.json') - run_spec_name = config['helm.run_entry'] - dpath = fpath.parent - runs = HelmOutputs.coerce(dpath / 'benchmark_output').suites()[0].runs() - assert len(runs) == 1 - run = runs[0] - kwdagger_rows.append( - { - 'dpath': dpath, - 'run_spec_name': run_spec_name, - 'run': run, - } - ) -kwdagger_lut = {r['run_spec_name']: r for r in kwdagger_rows} -print(f'len(helm_rows)={len(helm_rows)}') -print(f'len(kwdagger_rows)={len(kwdagger_rows)}') - - -def sankey_stats(rd: HelmRunDiff) -> dict: - """ - Return a small, stable set of fields intended for building Sankey tables. - """ - s = self.summary_dict(level=1) - va = (s.get("value_agreement") or {}) - by_class = (va.get("by_class") or {}) - overall = (va.get("overall") or {}) - - core = (by_class.get("core") or {}) - book = (by_class.get("bookkeeping") or {}) - - core_ratio = core.get("agree_ratio", None) - book_ratio = book.get("agree_ratio", None) - overall_ratio = overall.get("agree_ratio", None) - - spec_ok = bool(s.get("run_spec_dict_ok", False)) - scen_ok = s.get("scenario_ok", None) # True/False/None - - spec_status = "spec match" if spec_ok else "spec mismatch" - if scen_ok is None: - scenario_status = "scenario unknown" - else: - scenario_status = "scenario match" if scen_ok else "scenario mismatch" - - # stats_name_status is cheap and very useful as “schema drift” indicator - cov = (s.get("stats_coverage_by_name") or {}) - stats_name_status = "stats names match" if (cov.get("only_a", 0) == 0 and cov.get("only_b", 0) == 0) else "stats names mismatch" - - def _bucket_ratio(x: float | None, *, good=0.995, ok=0.95) -> str: - if x is None: - return "unknown" - if x >= good: - return "high" - if x >= ok: - return "medium" - return "low" - - core_b = _bucket_ratio(core_ratio) - book_b = _bucket_ratio(book_ratio) - - if core_b == "high" and (book_b in {"high", "unknown"}): - agreement_quality = "match" - elif core_b == "high" and book_b in {"medium", "low"}: - agreement_quality = "core match, bookkeeping differs" - elif core_b == "medium": - agreement_quality = "core partial" - elif core_b == "low": - agreement_quality = "core mismatch" - else: - agreement_quality = "unknown" - - return { - # orthogonal statuses - "spec_status": spec_status, - "scenario_status": scenario_status, - "stats_name_status": stats_name_status, - "agreement_quality": agreement_quality, - - # numeric signals - "run_agree_ratio_core": core_ratio, - "run_agree_ratio_bookkeeping": book_ratio, - "run_agree_ratio_overall": overall_ratio, - "comparable_core": core.get("comparable", None), - "mismatched_core": core.get("mismatched", None), - "comparable_bookkeeping": book.get("comparable", None), - "mismatched_bookkeeping": book.get("mismatched", None), - "comparable_overall": overall.get("comparable", None), - "mismatched_overall": overall.get("mismatched", None), - } - return out - -sankey_rows = [] - -rundiff_lut = {} -for helm_row in ub.ProgIter(helm_rows, desc='compare runs'): - run_dir = ub.Path(helm_row['run_dir']) - suite_name = run_dir.parent.name - benchmark_name = run_dir.parent.parent.parent.parent.name - assert run_dir.parent.parent.parent.name == 'benchmark_output' - assert run_dir.parent.parent.name == 'runs' - helm_row['suite_name'] = suite_name - helm_row['benchmark_name'] = benchmark_name - run_dir = ub.Path(helm_row['run_dir']) - run_spec_name = helm_row['run_spec_name'] - - kwrow = kwdagger_lut.get(run_spec_name) - helm_row['reproduced_step1'] = kwrow is not None - - # Base row (always emitted, even if not attempted) - out = { - "run_spec_name": run_spec_name, - "run_dir": str(run_dir), - "suite_name": suite_name, - "benchmark_name": benchmark_name, - "model_name": helm_row['model'], - - # default pipeline fields - "reproduced_step1": False, - "attempt_status": None, - "attempt_error": None, - - # default agreement fields - "agreement_bucket": None, - "agreement_bucket_instances": None, - "run_agree_ratio_core": None, - "run_agree_ratio_bookkeeping": None, - "run_agree_ratio_overall": None, - "inst_agree_ratio_unperturbed": None, - "inst_agree_ratio_perturbed": None, - "n_instance_mismatched": None, - - # signatures (fill opportunistically) - "sig_run_spec": None, - "sig_scenario": None, - "sig_stats_name": None, - } - out["reproduced_step1"] = (kwrow is not None) - sankey_rows.append(out) - - if kwrow is None: - out["attempt_status"] = "not attempted" - out["agreement_bucket"] = "not attempted" - - helm_row['agreement_bucket_base_task'] = 'not attempted' - continue - - # raise Exception - - # Attempt exists: try to compare - out["attempt_status"] = "compared" - try: - helm_run = HelmRun.coerce(run_dir) - kwdg_run = kwrow["run"] - - a = HelmRunAnalysis(helm_run, name="HELM") - b = HelmRunAnalysis(kwdg_run, name="KWDG") - - # Light signatures: cheap to compute and very useful for debugging. - sa = a.summary_dict(level=0) - sb = b.summary_dict(level=0) - out["sig_run_spec"] = f"{sa['signatures'].get('run_spec_sig')}|{sb['signatures'].get('run_spec_sig')}" - out["sig_scenario"] = f"{sa['signatures'].get('scenario_sig')}|{sb['signatures'].get('scenario_sig')}" - out["sig_stats_name"] = f"{sa['signatures'].get('stats_name_sig')}|{sb['signatures'].get('stats_name_sig')}" - - rd = HelmRunDiff(run_a=helm_run, run_b=kwdg_run, a_name="HELM", b_name="KWDG") - rundiff_lut[run_spec_name] = rd # save for later drilldown - out.update(sankey_stats(rd)) - - except Exception as ex: - raise - out["attempt_status"] = "error" - out["attempt_error"] = repr(ex) - out["agreement_bucket"] = "error" - - # # raise Exception - - # helm_run = HelmRun.coerce(run_dir) - # kwdg_run = kwrow['run'] - - # a = HelmRunAnalysis(helm_run) - # b = HelmRunAnalysis(kwdg_run) - - # if 0: - # a.summary(level=10) - # b.summary(level=10) - - # rd = HelmRunDiff( - # run_a=helm_run, run_b=kwdg_run, a_name='HELM', b_name='KWDG' - # ) - # self = rd # NOQA - # rd.summary(level=1) - # rd.summarize_instances() - - # if 0: - # table1 = rd.a.joined_instance_stat_table() - # table2 = rd.a.joined_instance_stat_table() - - # instance_id = 'id1237' - # keys1 = table1.variant_keys_for_instance(instance_id) - # keys2 = table2.variant_keys_for_instance(instance_id) - # print(f'keys1 = {ub.urepr(keys1, nl=1)}') - # print(f'keys2 = {ub.urepr(keys2, nl=1)}') - # assert set(keys1) == set(keys2) - # for k1 in keys1: - # table1.rows_by_variant[k1] - # table2.rows_by_variant[k1] - - # print(rd.drilldown_core_metric_instances()) - # rd.lookup_instance(('instance_id', 'id14045'), which='a') - # rd.lookup_instance(('instance_id', 'id14045'), which='b') - - # raise Exception - - # helm_row.update(rd.summary_base_task()) - # helm_row.update(rd.summary_core()) - - # helm_stats = helm_run.json.stats() - # kwdg_stats = kwdg_run.json.stats() - - # # out = compare.compare_run_pair(helm_stats, kwdg_stats, rel_tol=1e-4, abs_tol=1e-8) - # helm_row.update(out) - - -for rd in rundiff_lut.values(): - rd.summary(level=0) - rd.summary() - a = rd.a - b = rd.b - rd = HelmRunDiff(run_a=a, run_b=b, a_name="HELM", b_name="KWDG") - spec_a = rd.a.run_spec() - spec_b = rd.b.run_spec() - print(f'spec_a = {ub.urepr(spec_a, nl=3)}') - print(f'spec_b = {ub.urepr(spec_b, nl=3)}') - - if 0: - rd.summarize_instances() - -df = pd.DataFrame(sankey_rows) - -df = df[df['attempt_status'] == 'compared'] -print(df.value_counts(['benchmark_name', 'reproduced_step1'])) -print(df.value_counts(['benchmark_name', 'reproduced_step1', 'spec_status', 'agreement_quality'])) -print(df.value_counts(['attempt_status', 'agreement_bucket']).sort_index()) - - -def attempt_status(row: dict[str, object]) -> str: - return ( - 'attempted' if row.get('reproduced_step1', False) else 'not_attempted' - ) - - -def attempt_label(row: dict[str, object]) -> str: - # We already computed this in the table builder - return str(row.get('attempt_status', 'unknown')) - - -def agreement_label(row: dict[str, object]) -> str: - # Used in the sankey plan; keep it stable. - return row.get('agreement_bucket_base_task', 'unknown') - - -plan = sankey.Plan( - sankey.Root('Attempted Set'), - sankey.Group('benchmark', by='benchmark_name'), - sankey.Bucket('spec', by='spec_status'), - sankey.Bucket('agreement', by='agreement_quality'), -) - -print(plan.to_text()) - -G = plan.build_sankey(helm_rows, label_fmt='{name}: {value}') -print(G.summarize(max_edges=150)) - -fig = G.to_plotly(title='HELM Reproduction Funnel') -fpath = 'helm_repro_sankey.jpg' -fig.write_image(fpath) -print(f'Wrote helm_repro_sankey: {fpath}') - -if 1: - print(ub.codeblock( - f''' - # On Host - rm -rf {fpath} - # Run the wormhole command - ''')) - ub.cmd(f'wormhole send {fpath}', verbose=3) - """ - !wormhole send helm_repro_sankey.jpg - """ - -# --- Per-benchmark drilldown sankeys (deeper, but still run-level) --- -# Here we drill by model within each benchmark (only if model_name exists). -# If model_name is missing, you can swap this to group by run_spec_name prefix, etc. -bench_groups = ub.group_items(sankey_rows, key=lambda r: r.get('benchmark_name', 'unknown')) - -out_dpath = ub.Path('benchmark_sankeys').ensuredir() -for bench, rows in bench_groups.items(): - if bench in {None, 'unknown'}: - continue - - # Skip tiny groups if you want: - # if len(rows) < 5: continue - - plan_bench = sankey.Plan( - sankey.Root(f'{bench}'), - sankey.Group('model', by='model_name'), - sankey.Bucket('spec', by='spec_status'), - sankey.Bucket('agreement', by='agreement_quality'), - ) - - Gb = plan_bench.build_sankey(rows, label_fmt='{name}: {value}') - figb = Gb.to_plotly(title=f'HELM Repro Funnel: {bench}') - - # sanitize bench for filename - bench_slug = ub.Path(str(bench)).name.replace('/', '_') - fpath_b = out_dpath / f'{bench_slug}.jpg' - figb.write_image(str(fpath_b)) - print(f'Wrote benchmark sankey: {fpath_b}') - -if 1: - print(ub.codeblock( - f''' - # On Host - rm -rf {out_dpath} - # Run the wormhole command - ''')) - ub.cmd(f'wormhole send {out_dpath}', verbose=3) diff --git a/dev/poc/inspect_historic_helm_runs.py b/dev/poc/inspect_historic_helm_runs.py deleted file mode 100644 index 0d60d4f..0000000 --- a/dev/poc/inspect_historic_helm_runs.py +++ /dev/null @@ -1,327 +0,0 @@ -r""" -Compile a reproduction list from existing HELM outputs on disk. - -Given one or more roots that contain HELM outputs, discover all run directories -and emit a list of run specs you can feed into kwdagger / helm-run. - -Outputs are structured so you can: -- reproduce exact run directories (by using run_entry == run directory name) -- optionally include max_eval_instances inferred from per_instance_stats.json - -Ignore: - - ls /data/crfm-helm-public/thaiexam/benchmark_output/runs/v1.1.0/thai_exam:exam=tpat1,method=multiple_choice_joint,model=aisingapore_llama3-8b-cpt-sea-lionv2.1-instruct - - python ~/code/aiq-magnet/dev/poc/inspect_historic_helm_runs.py /data/crfm-helm-public --out_fpath run_specs.yaml --out_detail_fpath run_details.yaml - - cat run_specs.yaml | grep -v together > run_specs2.yaml - - python ~/code/aiq-magnet/dev/poc/inspect_historic_helm_runs.py /data/Public/AIQ/crfm-helm-public/ - - # we need fully featured helm installed - uv pip install crfm-helm[all] -U - - # Need to login to huggingface can pass token via --token - hf auth login - - # Need TogetherAPI credentials - - kwdagger schedule \ - --params=" - pipeline: 'magnet.backends.helm.pipeline.helm_single_run_pipeline()' - matrix: - helm.run_entry: - - __include__: run_specs2.yaml - helm.max_eval_instances: - - 1000 - helm.precomputed_root: null - " \ - --devices="0,1,2,3" \ - --tmux_workers=4 \ - --root_dpath=$PWD/results \ - --backend=tmux \ - --skip_existing=1 \ - --run=1 -""" - -from __future__ import annotations -from pathlib import Path -from typing import Iterable, Any - -import ubelt as ub -import kwutil -import scriptconfig as scfg -from loguru import logger - -from magnet.backends.helm.helm_outputs import HelmOutputs, HelmRun - -# Reuse your existing discovery + inference logic -from magnet.backends.helm.materialize_helm_run import ( - discover_benchmark_output_dirs, - infer_num_instances, - is_complete_run_dir, -) - - -class CompileHelmReproListConfig(scfg.DataConfig): - roots = scfg.Value( - ['/data/crfm-helm-public'], - nargs="+", - help=( - "One or more roots that either ARE a benchmark_output dir, contain " - "benchmark_output dirs, or contain suite/benchmark_output dirs." - ), - position=1, - ) - - suite_pattern = scfg.Value( - "*", - help="Glob applied to benchmark_output/runs/ directories.", - ) - - run_pattern = scfg.Value( - "*:*", - help="Glob applied within each suite to select runs (default selects HELM run dirs).", - ) - - require_per_instance_stats = scfg.Value( - False, - help="If True, only include runs that have per_instance_stats.json.", - ) - - include_max_eval_instances = scfg.Value( - False, - help="If True, infer max_eval_instances from per_instance_stats.json when possible. CAN BE VERY SLOW", - ) - - out_fpath = scfg.Value( - None, - help="Where to write output. If omitted, prints to stdout.", - ) - - out_detail_fpath = scfg.Value( - None, - help="Where to write detailed output.", - ) - - dedupe = scfg.Value( - True, - help="If True, dedupe identical (suite, run_entry, max_eval_instances) rows.", - ) - - @classmethod - def main(cls, argv=None, **kwargs): - """ - Example: - >>> # It's a good idea to setup a doctest. - >>> import sys, ubelt - >>> sys.path.append(ubelt.expandpath('~/code/aiq-magnet/dev/poc')) - >>> from inspect_historic_helm_runs import * # NOQA - >>> argv = False - >>> kwargs = dict() - >>> cls = CompileHelmReproListConfig - >>> config = cls(**kwargs) - >>> cls.main(argv=argv, **config) - """ - config = cls.cli(argv=argv, data=kwargs, verbose="auto") - roots = [Path(r).expanduser() for r in config.roots] - if not roots: - raise SystemExit("Must provide at least one root") - - suite_pattern = config.suite_pattern - run_pattern = config.run_pattern - require_per_instance_stats = config.require_per_instance_stats - include_max_eval_instances = config.include_max_eval_instances - - runs = gather_runs( - roots=roots, - suite_pattern=suite_pattern, - run_pattern=run_pattern, - require_per_instance_stats=require_per_instance_stats, - include_max_eval_instances=include_max_eval_instances, - ) - rows = build_run_table(runs) - - scenario_histo = ub.dict_hist([r['scenario_class'] for r in rows]) - model_histo = ub.dict_hist([r['model'] for r in rows]) - scenario_histo = ub.udict.sorted_values(scenario_histo) - model_histo = ub.udict.sorted_values(model_histo) - print(f'scenario_histo = {ub.urepr(scenario_histo, nl=1)}') - print(f'model_histo = {ub.urepr(model_histo, nl=1)}') - - from helm.benchmark import config_registry - from helm.benchmark import model_deployment_registry - config_registry.register_builtin_configs_from_helm_package() - model_rows = [] - for model_name, count in model_histo.items(): - try: - model_meta = model_deployment_registry.get_model_metadata(model_name) - model_row = model_meta.__dict__ | {'count': count} - if model_meta.deployment_names: - for deploy_name in model_meta.deployment_names: - deploy_info = model_deployment_registry.get_model_deployment(deploy_name) - model_row['client'] = deploy_info.client_spec.class_name - model_rows.append(model_row) - except (TypeError, ValueError) as ex: - logger.warning(f'missing: model_name = {ub.urepr(model_name, nl=1)} {ex}') - - if 0: - ub.dict_hist([r.get('client') for r in model_rows]) - - require_tags = { - 'FULL_FUNCTIONALITY_TEXT_MODEL_TAG' - } - MAX_PARAMS = 10e9 - # MAX_PARAMS = 200e9 - - # Filter to text models that will fit in memory - chosen_model_rows = [ - r for r in model_rows if ( - set(r['tags']).issuperset(require_tags) and - (r['num_parameters'] is not None and r['num_parameters'] <= MAX_PARAMS) and - (r['access'] == 'open') and - (r.get('client') == 'helm.clients.huggingface_client.HuggingFaceClient') - ) - ] - chosen_model_names = {r['name'] for r in chosen_model_rows} - logger.info('Filter to {} / {} models', len(chosen_model_rows), len(model_rows)) - - chosen_rows = [r for r in rows if r['model'] in chosen_model_names] - logger.info('Filter to {} / {} runs', len(chosen_rows), len(rows)) - # logger.info(f'chosen_rows = {ub.urepr(chosen_rows, nl=1)}') - - if 1: - # Show filtered histograms - scenario_histo = ub.dict_hist([r['scenario_class'] for r in chosen_rows]) - model_histo = ub.dict_hist([r['model'] for r in chosen_rows]) - scenario_histo = ub.udict.sorted_values(scenario_histo) - model_histo = ub.udict.sorted_values(model_histo) - logger.info(f'scenario_histo = {ub.urepr(scenario_histo, nl=1)}') - logger.info(f'model_histo = {ub.urepr(model_histo, nl=1)}') - - if config.out_detail_fpath: - text = kwutil.Yaml.dumps(chosen_rows) - Path(config.out_detail_fpath).write_text(text) - logger.success("Wrote {}", config.out_detail_fpath) - - run_spec_names = [r["run_spec_name"] for r in chosen_rows] - text = kwutil.Yaml.dumps(run_spec_names) - if config.out_fpath: - Path(config.out_fpath).write_text(text) - logger.success("Wrote {}", config.out_fpath) - else: - print(text, end="") - - -def gather_runs( - roots: Iterable[Path], - suite_pattern: str = "*", - run_pattern: str = "*:*", - require_per_instance_stats: bool = False, - include_max_eval_instances: bool = True, -) -> list[HelmRun]: - - # Discover all benchmark_output dirs under provided roots - logger.info('Discover benchmarks') - bo_dirs = list(ub.ProgIter(discover_benchmark_output_dirs(roots), desc='discovering benchmarks', verbose=3, homogeneous=False)) - logger.info('Finished Discover benchmarks') - if not bo_dirs: - logger.warning("No benchmark_output dirs found under roots={}", roots) - - runs: list[HelmRun] = [] - for bo in ub.ProgIter(bo_dirs, desc='Check dirs'): - try: - outputs = HelmOutputs.coerce(bo) - except Exception: - continue - - for suite in outputs.suites(pattern=suite_pattern): - for run in suite.runs(pattern=run_pattern): - run_dir = Path(run.path) - - run = HelmRun(run_dir) - - # TODO: if not run.exists(): - # ... - # Only include if it looks “complete enough” - if not is_complete_run_dir(run_dir, require_per_instance_stats=require_per_instance_stats): - continue - - runs.append(run) - - # Stable order - logger.info('Found {} run directories', len(runs)) - return runs - - -def build_run_table(runs: list[HelmRun]) -> list[dict]: - rows = [] - - include_max_eval_instances = False - mismatches = [] - for run in ub.ProgIter(runs, desc='Extract run spec info'): - max_eval_instances = None - if include_max_eval_instances: - max_eval_instances = infer_num_instances(run.path) - - # Not sure if there is an advantage to msgspec or json here - # ZFS is likely messing up my timings. - if 1: - run_spec = run.json.run_spec() - scenario_class = run_spec['scenario_spec']['class_name'] - model = run_spec['adapter_spec']['model'] - run_spec_name = run_spec['name'] - else: - run_spec = run.msgspec.run_spec() - scenario_class = run_spec.scenario_spec.class_name - model = run_spec.adapter_spec.model - run_spec_name = run_spec.name - - if run.path.name != run_spec_name.replace('/', '_'): - mismatches.append({ - 'run.path.parent': run.path.parent, - 'run.path.name': run.path.name, - 'run_spec_name': run_spec_name, - }) - - # Hack: run spec names sometimes don't correctly encode the model - FIX_RUN_SPEC_NAME = True - if FIX_RUN_SPEC_NAME: - normalized_model = model.replace('/', '_') - run_spec_name = run_spec_name.replace(normalized_model, model) - - rows.append({ - # "benchmark_output_dir": str(Path(outputs.root_dir)), - # "suite": suite.name, - # # Use run directory name as the canonical "run_entry" to reproduce. - # # This is faithful even if HELM normalized defaults into the name. - - # Use run directory name as the canonical "run_entry" to reproduce. - # This is faithful even if HELM normalized defaults into the name. - "run_spec_name": run_spec_name, - "run_dir": str(run.path), - "max_eval_instances": max_eval_instances, - 'model': model, - 'scenario_class': scenario_class, - }) - logger.warning(f'mismatches = {ub.urepr(mismatches, nl=2, align=":")}') - rows.sort(key=lambda r: (r["run_dir"])) - return rows - - -def dedupe_rows(rows: list[dict[str, Any]]) -> list[dict[str, Any]]: - seen = set() - out = [] - for r in rows: - key = (r["suite"], r["run_entry"], r.get("max_eval_instances", None)) - if key in seen: - continue - seen.add(key) - out.append(r) - return out - - -__cli__ = CompileHelmReproListConfig - -if __name__ == "__main__": - __cli__.main() diff --git a/magnet/backends/helm/cli/materialize_helm_run.py b/magnet/backends/helm/cli/materialize_helm_run.py index 898a613..c7c252c 100755 --- a/magnet/backends/helm/cli/materialize_helm_run.py +++ b/magnet/backends/helm/cli/materialize_helm_run.py @@ -136,6 +136,7 @@ import os import time +import shutil from dataclasses import dataclass from pathlib import Path from typing import Iterable, Iterator, Optional @@ -150,7 +151,104 @@ # We rely on MAGNET's HELM output exploration helpers. # These are already present in aiq-magnet and know how to load / validate # the standard json files produced by helm-run. -from magnet.helm_outputs import HelmOutputs +from magnet.backends.helm.helm_outputs import HelmOutputs + + +def _normalize_optional_pathish(value): + """ + Normalize common "unset" placeholder values emitted by schedulers / CLIs. + + Args: + value: raw parsed CLI/config value + + Returns: + The original value, or ``None`` for empty / null-like placeholders. + """ + if value is None: + return None + if isinstance(value, str): + text = value.strip() + if text == '': + return None + if text.lower() in {'none', 'null'}: + return None + return text + return value + + +def _safe_config_dict(config) -> dict: + try: + return config.asdict() + except Exception: + try: + return dict(config) + except Exception: + return {} + + +def _query_nvidia_smi() -> dict | None: + """ + Best-effort GPU query for reproducibility metadata. + """ + try: + cmd = [ + 'nvidia-smi', + '--query-gpu=index,name,memory.total,driver_version', + '--format=csv,noheader,nounits', + ] + info = ub.cmd(cmd, verbose=0, system=False, check=False) + if info.returncode != 0: + return None + gpus = [] + for line in info.stdout.strip().splitlines(): + parts = [p.strip() for p in line.split(',')] + if len(parts) != 4: + continue + idx, name, mem_total, driver = parts + gpus.append({ + 'index': int(idx), + 'name': name, + 'memory_total_mb': int(mem_total), + 'driver_version': driver, + }) + return {'gpus': gpus} + except FileNotFoundError: + return None + except Exception as ex: + return {'error': repr(ex)} + + +def _capture_process_context(out_dpath: Path, config) -> dict: + from kwutil.process_context import ProcessContext + + process_context_fpath = out_dpath / 'process_context.json' + extra = { + 'env': { + 'CUDA_VISIBLE_DEVICES': os.environ.get('CUDA_VISIBLE_DEVICES'), + 'HOSTNAME': os.environ.get('HOSTNAME'), + } + } + ctx = ProcessContext( + name='materialize_helm_run', + config=_safe_config_dict(config), + extra=extra, + output_fpath=process_context_fpath, + ) + ctx.start() + ctx.stop() + try: + ctx.add_disk_info(out_dpath) + except Exception: + pass + gpu_info = _query_nvidia_smi() + if gpu_info: + ctx.properties.setdefault('extra', {}) + ctx.properties['extra']['nvidia_smi'] = gpu_info + try: + process_context_fpath.write_text(kwutil.Json.dumps(ctx.obj, indent=2)) + except Exception: + pass + return ctx.obj class MaterializeHelmRunConfig(scfg.DataConfig): @@ -218,6 +316,43 @@ class MaterializeHelmRunConfig(scfg.DataConfig): tags=['perf_param'], ) + local_path = scfg.Value( + 'prod_env', + type=str, + help='Passed to helm-run --local-path. Relative paths are resolved inside out_dpath.', + tags=['perf_param'], + ) + + model_deployments_fpath = scfg.Value( + None, + type=str, + help=( + 'Optional path to a HELM model_deployments.yaml file that will be copied ' + 'into /model_deployments.yaml before invoking helm-run.' + ), + tags=['algo_param'], + ) + + enable_huggingface_models = scfg.Value( + None, + type=str, + help=( + 'Optional YAML-encoded list passed through to helm-run ' + '--enable-huggingface-models. Example: \'[repo-a, repo-b]\'' + ), + tags=['algo_param'], + ) + + enable_local_huggingface_models = scfg.Value( + None, + type=str, + help=( + 'Optional YAML-encoded list passed through to helm-run ' + '--enable-local-huggingface-models. Example: \'[/models/a, /models/b]\'' + ), + tags=['algo_param'], + ) + # extra_helm_args = scfg.Value( # [], # nargs='*', @@ -260,9 +395,9 @@ def main(cls, argv=None, **kwargs) -> dict: Example: >>> # This doctest is illustrative only; it requires helm-run installed. >>> # xdoctest: +REQUIRES(env:HELM_RUN_AVAILABLE) - >>> from magnet.backends.helm.cli.materialize_helm_run import main + >>> from magnet.backends.helm.cli.materialize_helm_run import MaterializeHelmRunConfig >>> dpath = ub.Path.appdir('magnet/tests/materialize').delete().ensuredir() - >>> main([ + >>> MaterializeHelmRunConfig.main([ ... '--run-entry', 'mmlu:subject=philosophy,model=openai/gpt2', ... '--suite', 'my-suite', ... '--max-eval-instances', '2', @@ -273,6 +408,16 @@ def main(cls, argv=None, **kwargs) -> dict: config = MaterializeHelmRunConfig.cli( argv=argv, data=kwargs, verbose='auto' ) + config.precomputed_root = _normalize_optional_pathish(config.precomputed_root) + config.model_deployments_fpath = _normalize_optional_pathish( + config.model_deployments_fpath + ) + config.enable_huggingface_models = kwutil.Yaml.coerce( + config.enable_huggingface_models + ) + config.enable_local_huggingface_models = kwutil.Yaml.coerce( + config.enable_local_huggingface_models + ) if config.run_entry is None: raise SystemExit('Missing required --run-entry') @@ -320,6 +465,10 @@ def main(cls, argv=None, **kwargs) -> dict: 'require_per_instance_stats': config.require_per_instance_stats, 'mode': config.mode, 'materialize': config.materialize, + 'local_path': config.local_path, + 'model_deployments_fpath': config.model_deployments_fpath, + 'enable_huggingface_models': list(config.enable_huggingface_models or []), + 'enable_local_huggingface_models': list(config.enable_local_huggingface_models or []), }, 'status': None, 'reuse': None, @@ -327,6 +476,9 @@ def main(cls, argv=None, **kwargs) -> dict: 'out_dpath': str(out_dpath), 'timestamp': time.time(), } + process_context = _capture_process_context(out_dpath, config) + manifest['process_context_fpath'] = str(out_dpath / 'process_context.json') + manifest['process_context'] = process_context # 1) Try reuse match = None @@ -389,14 +541,22 @@ def main(cls, argv=None, **kwargs) -> dict: # Ensure benchmark_output exists (helm-run will create, but pre-creating is fine) (out_dpath / 'benchmark_output').mkdir(exist_ok=True) + prepared_local_path = prepare_local_helm_config( + out_dpath=out_dpath, + local_path=config.local_path, + model_deployments_fpath=config.model_deployments_fpath, + ) logger.info('No reusable run found; running helm-run') run_helm( requested_desc=config.run_entry, suite=config.suite, out_dpath=out_dpath, + local_path=prepared_local_path, max_eval_instances=config.max_eval_instances, num_threads=config.num_threads, + enable_huggingface_models=list(config.enable_huggingface_models or []), + enable_local_huggingface_models=list(config.enable_local_huggingface_models or []), extra_args=[], ) @@ -993,12 +1153,113 @@ def ensure_copytree(src: Path, dst: Path) -> None: ub.copytree(src, dst) +def resolve_local_path(out_dpath: Path, local_path: str | os.PathLike[str]) -> Path: + """ + Resolve HELM's local config path. + + HELM defaults to ``prod_env`` relative to its current working directory, so + we mirror that here by resolving relative paths inside ``out_dpath``. + """ + path = Path(local_path) + if path.is_absolute(): + return path + return out_dpath / path + + +def prepare_local_helm_config( + out_dpath: Path, + local_path: str | os.PathLike[str], + model_deployments_fpath: str | os.PathLike[str] | None = None, +) -> Path: + """ + Prepare the local HELM config directory used by ``helm-run``. + + Currently this only materializes an optional ``model_deployments.yaml`` + override file, but keeping it centralized makes future config additions + straightforward. + """ + local_path_abs = resolve_local_path(out_dpath, local_path) + local_path_abs.mkdir(parents=True, exist_ok=True) + + if model_deployments_fpath: + src = Path(model_deployments_fpath).expanduser().resolve() + if not src.exists(): + raise FileNotFoundError( + f'model_deployments_fpath does not exist: {src}' + ) + dst = local_path_abs / 'model_deployments.yaml' + shutil.copy2(src, dst) + logger.info('Copied model deployments override: {} -> {}', src, dst) + + return local_path_abs + + +def write_helm_log_config(out_dpath: Path) -> Path: + """ + Write a HELM logging config file into ``out_dpath``. + + This keeps the ``helm-run`` console logs and file logs colocated with the + kwdagger node outputs so they survive rsync / artifact transfer. + """ + log_fpath = out_dpath / 'helm-run.log' + debug_log_fpath = out_dpath / 'helm-run.debug.log' + config = { + 'version': 1, + 'disable_existing_loggers': False, + 'formatters': { + 'console': { + 'datefmt': '%Y-%m-%dT%H:%M:%S', + 'format': '%(asctime)s %(levelname)-8s %(message)s', + }, + 'detailed': { + 'datefmt': '%Y-%m-%dT%H:%M:%S', + 'format': '%(asctime)s %(levelname)-8s %(name)s %(message)s', + }, + }, + 'handlers': { + 'stdout': { + 'class': 'logging.StreamHandler', + 'stream': 'ext://sys.stdout', + 'formatter': 'console', + 'level': 'INFO', + }, + 'file_info': { + 'class': 'logging.FileHandler', + 'filename': os.fspath(log_fpath), + 'formatter': 'detailed', + 'level': 'INFO', + 'mode': 'w', + }, + 'file_debug': { + 'class': 'logging.FileHandler', + 'filename': os.fspath(debug_log_fpath), + 'formatter': 'detailed', + 'level': 'DEBUG', + 'mode': 'w', + }, + }, + 'loggers': { + 'helm': { + 'handlers': ['stdout', 'file_info', 'file_debug'], + 'level': 'DEBUG', + 'propagate': False, + } + }, + } + config_fpath = out_dpath / 'helm_log_config.yaml' + config_fpath.write_text(kwutil.Yaml.dumps(config)) + return config_fpath + + def run_helm( requested_desc: str, suite: str, out_dpath: Path, + local_path: Path, max_eval_instances: Optional[int], num_threads: int, + enable_huggingface_models: Optional[list[str]] = None, + enable_local_huggingface_models: Optional[list[str]] = None, extra_args: Optional[list[str]] = None, ) -> None: """ @@ -1006,11 +1267,25 @@ def run_helm( We do not run helm-summarize by design. """ - cmd = ['helm-run', '--run-entries', requested_desc, '--suite', suite] + cmd = [ + 'helm-run', + '--run-entries', + requested_desc, + '--suite', + suite, + '--local-path', + os.fspath(local_path), + '--log-config', + os.fspath(write_helm_log_config(out_dpath)), + ] if max_eval_instances is not None: cmd += ['--max-eval-instances', str(max_eval_instances)] if num_threads is not None: cmd += ['--num-threads', str(num_threads)] + if enable_huggingface_models: + cmd += ['--enable-huggingface-models', *map(str, enable_huggingface_models)] + if enable_local_huggingface_models: + cmd += ['--enable-local-huggingface-models', *map(str, enable_local_huggingface_models)] cmd += list(extra_args or []) logger.info('Executing: {}', ' '.join(map(str, cmd))) ub.cmd(cmd, cwd=out_dpath, verbose=3, system=True).check_returncode() diff --git a/magnet/backends/helm/helm_run_diff.py b/magnet/backends/helm/helm_run_diff.py index c93c57f..81e462e 100644 --- a/magnet/backends/helm/helm_run_diff.py +++ b/magnet/backends/helm/helm_run_diff.py @@ -17,86 +17,47 @@ xdoctest -m magnet.backends.helm.helm_run_diff __doc__ Example: - >>> import ubelt as ub - >>> import kwutil - >>> from magnet.backends.helm.helm_outputs import HelmRun + >>> import json + >>> from magnet.backends.helm.helm_run_analysis import HelmRunAnalysis >>> from magnet.backends.helm.helm_run_diff import HelmRunDiff - - >>> run_a = HelmRun.demo() - >>> dpath = ub.Path.appdir('magnet/tests/helm/helm_run_diff').delete().ensuredir() - - >>> # --- Case 1: identical copy -> perfect agreement ----------------- - >>> same_path = dpath / (run_a.path.name + '_same') - >>> run_a.path.copy(same_path) - >>> run_b = HelmRun(same_path) - - >>> rd = HelmRunDiff(run_a, run_b, a_name='orig', b_name='same') - >>> info = rd.summary_dict(level=10) - >>> assert info['run_spec_dict_ok'] is True - >>> assert info['scenario_ok'] in {True, None} - >>> assert info['value_agreement']['overall']['mismatched'] == 0 - >>> assert info['value_agreement']['overall']['agree_ratio'] == 1.0 - - >>> line = rd.summary_text(level=0) - >>> assert 'orig' in line and 'same' in line - >>> rd.summary(level=1000) - - >>> # --- Case 2: perturb a single RUN-level stat mean ---------------- - >>> stats_path = dpath / (run_a.path.name + '_statsmod') - >>> run_a.path.copy(stats_path) - >>> stat_fpath = stats_path / 'stats.json' - >>> stats = kwutil.Json.loads(stat_fpath.read_text()) - >>> old_mean = float(stats[0].get('mean', 0.0)) - >>> stats[0]['mean'] = old_mean + 1.23 - >>> stat_fpath.write_text(kwutil.Json.dumps(stats)) - - >>> run_b2 = HelmRun(stats_path) - >>> rd2 = HelmRunDiff(run_a, run_b2, a_name='orig', b_name='stats+1.23') - >>> info2 = rd2.summary_dict(level=10) - >>> assert info2['value_agreement']['overall']['mismatched'] >= 1 - >>> rd.summary(level=1000) - - >>> # --- Case 3: perturb ONE per-instance stat mean ------------------ - >>> inst_path = dpath / (run_a.path.name + '_perinstmod') - >>> run_a.path.copy(inst_path) - >>> pi_fpath = inst_path / 'per_instance_stats.json' - >>> if pi_fpath.exists(): - ... perinst = kwutil.Json.loads(pi_fpath.read_text()) - ... # deterministically modify the first mean-bearing stat for the first entry - ... ei, sj = 0, None - ... for j, s in enumerate(perinst[ei]['stats']): - ... if int(s.get('count', 0) or 0) and ('mean' in s): - ... sj = j - ... break - ... assert sj is not None - ... old = float(perinst[ei]['stats'][sj]['mean']) - ... perinst[ei]['stats'][sj]['mean'] = old + 9.0 - ... pi_fpath.write_text(kwutil.Json.dumps(perinst)) - ... run_bi = HelmRun(inst_path) - ... rd_i = HelmRunDiff(run_a, run_bi, a_name='orig', b_name='perinst+9') - ... inst_info = rd_i.instance_summary_dict(top_n=5) - ... assert inst_info['means']['mismatched'] >= 1 - ... rd_i.summary(level=1000) - - >>> # --- Case 4: run spec diff ---------------- - >>> new_dpath = dpath / (run_a.path.name + '_runspec_mod') - >>> run_a.path.copy(new_dpath) - >>> spec_fpath = new_dpath / 'run_spec.json' - >>> run_spec = kwutil.Json.loads(spec_fpath.read_text()) - >>> run_spec['adapter_spec']['model_deployment'] = 'someotherdeploy/gpt2' - >>> spec_fpath.write_text(kwutil.Json.dumps(run_spec)) - >>> run_b4 = HelmRun(new_dpath) - >>> rd = HelmRunDiff(run_a, run_b4, a_name='orig', b_name='runspec_mod') - >>> rd.summary(level=1000) - >>> info = rd.summary_dict(level=10) - >>> assert not info['run_spec_dict_ok'] + >>> class _DummyJoined: + ... def __init__(self): + ... self.row_by_key = {} + ... def __iter__(self): + ... return iter(self.row_by_key.values()) + >>> def _ana(run_spec, stats, request_states): + ... a = HelmRunAnalysis.__new__(HelmRunAnalysis) + ... a._raw_cache = {} + ... a._cache = {} + ... a.run = None + ... a.name = None + ... a.run_spec = lambda: run_spec + ... a.scenario = lambda: {'class_name': 'ToyScenario', 'output_path': 'tmp/a'} + ... a.scenario_state = lambda: {'request_states': request_states} + ... a.stats = lambda: stats + ... a.joined_instance_stat_table = lambda *args, **kwargs: _DummyJoined() + ... return a + >>> rs = [{'instance': {'id': 'id1', 'split': 'test', 'input': {'text': 'Q'}}, 'train_trial_index': 0, 'request': {'prompt': 'P'}, 'result': {'completions': [{'text': 'A'}]}}] + >>> stats_a = [{'name': {'name': 'exact_match', 'split': 'test'}, 'count': 1, 'mean': 1.0}] + >>> stats_b = [{'name': {'name': 'exact_match', 'split': 'test'}, 'count': 1, 'mean': 0.0}] + >>> spec_a = {'name': 'toy', 'adapter_spec': {'model': 'm'}, 'metric_specs': [{'class_name': 'M0', 'args': {}}]} + >>> spec_b = {'name': 'toy', 'adapter_spec': {'model': 'm', 'model_deployment': 'huggingface/m'}, 'metric_specs': [{'class_name': 'M1', 'args': {}}]} + >>> rd = HelmRunDiff(_ana(spec_a, stats_a, rs), _ana(spec_b, stats_b, rs), a_name='A', b_name='B') + >>> info = rd.summary_dict(level=20) + >>> assert info['run_spec_name_ok'] is True + >>> assert info['dataset_overlap']['base_iou'] == 1.0 + >>> assert info['value_agreement']['overall']['mismatched'] == 1 + >>> assert isinstance(info['diagnosis']['label'], str) + >>> _ = json.dumps(info, allow_nan=False) """ from __future__ import annotations +import math import ubelt as ub +from collections import Counter from dataclasses import dataclass from magnet.backends.helm.util import helm_hashers from magnet.backends.helm.util import helm_metrics @@ -117,7 +78,23 @@ def _safe_float(x: Any) -> float | None: return None -def _walker_diff(a: Any, b: Any, *, max_paths: int = 12) -> list[str]: +def _quantile(values: list[float], q: float) -> float | None: + if not values: + return None + if q <= 0: + return values[0] + if q >= 1: + return values[-1] + pos = (len(values) - 1) * q + lo = math.floor(pos) + hi = math.ceil(pos) + if lo == hi: + return values[lo] + alpha = pos - lo + return values[lo] * (1 - alpha) + values[hi] * alpha + + +def _walker_diff(a: Any, b: Any, *, max_paths: int = 12) -> dict[str, Any]: """ Return a dict with formatted lines for: @@ -198,6 +175,30 @@ def _truncate(lines: list[str], max_items: int) -> list[str]: return out +def _walker_diff_paths(a: Any, b: Any) -> dict[str, list[str]]: + """Return full path-level differences (untruncated), path-only. + + The output is intentionally JSON-friendly and stable for diagnostics. + """ + walker_a = ub.IndexableWalker(a) + walker_b = ub.IndexableWalker(b) + info = walker_a.diff(walker_b) + + def _format_path(path: Iterable[Any]) -> str: + return '.'.join(map(str, path)) + + unique1 = sorted(_format_path(p) for p in info.get('unique1', [])) + unique2 = sorted(_format_path(p) for p in info.get('unique2', [])) + faillist = sorted( + _format_path(d.path) for d in info.get('faillist', []) + ) + return { + 'unique1': unique1, + 'unique2': unique2, + 'faillist': faillist, + } + + def _default_writer(writer=None) -> Callable[[str], Any]: if writer is not None: return writer @@ -262,6 +263,242 @@ def _short_urepr(obj: Any, max_chars: int = 140) -> str: return _smart_truncate(s, max_chars) +def _coerce_path_token(tok: str) -> str | int: + if tok.isdigit(): + try: + return int(tok) + except Exception: + return tok + return tok + + +def _path_get(obj: Any, path: str) -> tuple[Any, bool]: + """Best-effort dotted-path getter supporting dict/list traversal.""" + cur = obj + for raw_tok in path.split('.'): + tok = _coerce_path_token(raw_tok) + if isinstance(cur, dict): + if tok in cur: + cur = cur[tok] + elif isinstance(tok, int) and str(tok) in cur: + cur = cur[str(tok)] + else: + return None, False + elif isinstance(cur, (list, tuple)): + if isinstance(tok, int) and 0 <= tok < len(cur): + cur = cur[tok] + else: + return None, False + else: + return None, False + return cur, True + + +def _path_value_examples( + a_obj: Any, + b_obj: Any, + paths: list[str], + *, + max_items: int = 20, +) -> list[dict[str, Any]]: + """Return path-level value pairs for selected diff paths.""" + examples: list[dict[str, Any]] = [] + for p in sorted(paths): + rec: dict[str, Any] = {'path': p} + va, oka = _path_get(a_obj, p) + vb, okb = _path_get(b_obj, p) + rec['a'] = va if oka else None + rec['b'] = vb if okb else None + rec['a_found'] = bool(oka) + rec['b_found'] = bool(okb) + examples.append(rec) + if len(examples) >= max_items: + break + return _json_compatible(examples) + + +def _json_compatible(obj: Any) -> Any: + """Recursively coerce to strict JSON-compatible types. + + Notably: + - tuples/sets -> lists + - non-finite floats -> None + - unknown objects -> string repr + """ + if obj is None or isinstance(obj, (str, int, bool)): + return obj + if isinstance(obj, float): + return obj if math.isfinite(obj) else None + if isinstance(obj, dict): + return {str(k): _json_compatible(v) for k, v in obj.items()} + if isinstance(obj, (list, tuple, set)): + return [_json_compatible(v) for v in obj] + try: + # common dataclass / custom key cases + if hasattr(obj, 'as_tuple') and callable(getattr(obj, 'as_tuple')): + return _json_compatible(list(obj.as_tuple())) + except Exception: + pass + try: + return ub.urepr(obj, nl=0, compact=1) + except Exception: + return str(obj) + + +def _preview_list(items: list[str], *, limit: int = 20) -> list[str]: + """Return a stable preview list with an optional '' suffix.""" + if limit <= 0 or len(items) <= limit: + return items + remain = len(items) - limit + return items[:limit] + [f'<{remain} more not shown>'] + + +_RUNSPEC_EXEC_ADAPTER_NOISE_FIELDS = { + # Added in newer HELM formats; often default/no-op in practice. + 'chain_of_thought_prefix', + 'chain_of_thought_suffix', + 'global_suffix', + 'num_trials', +} + + +def _classify_run_spec_path(path: str) -> str: + """Classify run-spec diff paths into semantic buckets.""" + if path.startswith('metric_specs') or path.startswith('groups'): + return 'evaluation' + if path.startswith('adapter_spec.'): + parts = path.split('.') + field = parts[1] if len(parts) > 1 else '' + if field in _RUNSPEC_EXEC_ADAPTER_NOISE_FIELDS: + return 'nonsemantic' + return 'execution' + if path.startswith('scenario_spec') or path.startswith('data_augmenter_spec'): + return 'execution' + if path in {'name'}: + return 'nonsemantic' + return 'other' + + +def _classify_scenario_path(path: str) -> str: + """Classify scenario diff paths into semantic buckets.""" + # scenario.output_path is environment-local and should not affect content + if path == 'output_path' or path.endswith('.output_path'): + return 'nonsemantic' + return 'semantic' + + +def _canonicalize_metric_spec_for_semantic_diff(metric_spec: Any) -> Any: + """Normalize one metric spec for order-insensitive semantic comparison.""" + if not isinstance(metric_spec, dict): + return helm_hashers.canonicalize_for_hashing(metric_spec) + out = { + 'class_name': metric_spec.get('class_name', None), + 'args': helm_hashers.canonicalize_for_hashing( + metric_spec.get('args', None) + ), + } + return out + + +def _canonicalize_run_spec_for_semantic_diff(run_spec: dict[str, Any]) -> dict[str, Any]: + """Canonicalize run_spec with order-insensitive handling for select lists.""" + spec = helm_hashers.canonicalize_for_hashing(run_spec) + if not isinstance(spec, dict): + return {'_invalid_spec': spec} + spec = dict(spec) + + metric_specs = spec.get('metric_specs', None) + if isinstance(metric_specs, list): + canon_items = [ + _canonicalize_metric_spec_for_semantic_diff(ms) + for ms in metric_specs + ] + canon_items = sorted( + canon_items, key=lambda x: helm_hashers.stable_hash36(x) + ) + spec['metric_specs'] = canon_items + + groups = spec.get('groups', None) + if isinstance(groups, list): + spec['groups'] = sorted(groups, key=lambda x: str(x)) + + return spec + + +def _metric_specs_multiset_delta( + metric_specs_a: Any, + metric_specs_b: Any, + *, + short_hash: int = 12, + max_items: int = 20, +) -> dict[str, Any]: + """Order-insensitive multiset delta for run_spec.metric_specs.""" + specs_a = metric_specs_a if isinstance(metric_specs_a, list) else [] + specs_b = metric_specs_b if isinstance(metric_specs_b, list) else [] + + def _make_id(ms: Any) -> tuple[str, dict[str, Any]]: + canon = _canonicalize_metric_spec_for_semantic_diff(ms) + sid = helm_hashers.stable_hash36(canon)[:short_hash] + if isinstance(canon, dict): + class_name = canon.get('class_name', None) + args = canon.get('args', None) + else: + class_name = None + args = canon + rec = { + 'id': sid, + 'class_name': class_name, + 'args': args, + 'preview': _short_urepr(canon, max_chars=160), + } + return sid, rec + + id_to_rec: dict[str, dict[str, Any]] = {} + a_ids: list[str] = [] + b_ids: list[str] = [] + for ms in specs_a: + sid, rec = _make_id(ms) + id_to_rec.setdefault(sid, rec) + a_ids.append(sid) + for ms in specs_b: + sid, rec = _make_id(ms) + id_to_rec.setdefault(sid, rec) + b_ids.append(sid) + + a_counter = Counter(a_ids) + b_counter = Counter(b_ids) + keys = sorted(set(a_counter) | set(b_counter)) + added = [] + removed = [] + for sid in keys: + ca = a_counter.get(sid, 0) + cb = b_counter.get(sid, 0) + if cb > ca: + added.append(id_to_rec[sid] | {'count': cb - ca}) + if ca > cb: + removed.append(id_to_rec[sid] | {'count': ca - cb}) + + added = sorted(added, key=lambda r: (str(r.get('class_name')), r['id'])) + removed = sorted(removed, key=lambda r: (str(r.get('class_name')), r['id'])) + return _json_compatible( + { + 'n_a': len(specs_a), + 'n_b': len(specs_b), + 'n_added': sum(r['count'] for r in added), + 'n_removed': sum(r['count'] for r in removed), + 'added': _preview_list( + [ub.urepr(r, nl=0, compact=1) for r in added], limit=max_items + ), + 'removed': _preview_list( + [ub.urepr(r, nl=0, compact=1) for r in removed], limit=max_items + ), + 'added_structured': added[:max_items], + 'removed_structured': removed[:max_items], + 'equal_as_multiset': (len(added) == 0 and len(removed) == 0), + } + ) + + @dataclass(frozen=True) class Coverage: """Coverage bookkeeping for two key-sets.""" @@ -295,6 +532,229 @@ def _fmt(x: Any) -> str: return str(x) +def _key_to_serializable(key: Any) -> Any: + """Convert various key types (dataclasses, tuples) into JSON-friendly types. + + - If object has ``as_tuple()``, use that and return a list. + - If it's a tuple, return a list (JSON will accept either but list is explicit). + - Otherwise fallback to string repr. + """ + # dataclass-like keys (InstanceStatKey, InstanceVariantKey) implement as_tuple + try: + if hasattr(key, 'as_tuple') and callable(getattr(key, 'as_tuple')): + return list(key.as_tuple()) + except Exception: + pass + if isinstance(key, tuple): + return list(key) + # lists are already JSON-safe + if isinstance(key, list): + return key + # fallback: use a stable repr + try: + return ub.urepr(key, nl=0, compact=1) + except Exception: + return str(key) + + +def dataset_overlap_from_request_states( + request_states_a: list[dict[str, Any]], + request_states_b: list[dict[str, Any]], + *, + short_hash: int = 16, + max_examples: int = 5, +) -> dict[str, Any]: + """Compare two request_state lists at dataset/prompt/completion level. + + This is a pure function used by :meth:`HelmRunDiff.dataset_overlap_summary`. + + Example: + >>> rs_a = [ + ... { + ... 'instance': {'id': 'id1', 'split': 'test', 'input': {'text': 'Q1'}}, + ... 'train_trial_index': 0, + ... 'request': {'prompt': 'P1'}, + ... 'result': {'completions': [{'text': 'A1'}]}, + ... }, + ... { + ... 'instance': { + ... 'id': 'id1', 'split': 'test', 'input': {'text': 'Q1'}, + ... 'perturbation': {'name': 'dialect', 'prob': 1.0}, + ... }, + ... 'train_trial_index': 0, + ... 'request': {'prompt': 'P1-d'}, + ... 'result': {'completions': [{'text': 'A1d'}]}, + ... }, + ... ] + >>> rs_b = [ + ... { + ... 'instance': {'id': 'id1', 'split': 'test', 'input': {'text': 'Q1'}}, + ... 'train_trial_index': 0, + ... 'request': {'prompt': 'P1x'}, + ... 'result': {'completions': [{'text': 'A1'}]}, + ... }, + ... ] + >>> info = dataset_overlap_from_request_states(rs_a, rs_b, max_examples=2) + >>> assert info['base_coverage']['n_isect'] == 1 + >>> assert info['variant_coverage']['only_a'] == 1 + >>> assert info['content_equality']['prompt']['equal_ratio'] == 0.0 + >>> assert isinstance(info['mismatch_examples']['prompt'], list) + """ + + def _coerce_int(x: Any) -> int | None: + try: + if x is None: + return None + if isinstance(x, bool): + return int(x) + if isinstance(x, int): + return x + if isinstance(x, float) and x.is_integer(): + return int(x) + if isinstance(x, str) and x.isdigit(): + return int(x) + except Exception: + pass + return None + + def _base_key(rs: dict[str, Any]) -> tuple[Any, ...]: + inst = rs.get('instance') or {} + return ( + inst.get('id', None), + _coerce_int(rs.get('train_trial_index', None)), + inst.get('split', None), + ) + + def _variant_key(rs: dict[str, Any]) -> tuple[Any, ...]: + inst = rs.get('instance') or {} + pid = helm_hashers.perturbation_id( + inst.get('perturbation', None), short_hash=short_hash + ) + return _base_key(rs) + (pid,) + + def _index_unique( + rows: list[dict[str, Any]], key_fn + ) -> tuple[dict[tuple[Any, ...], dict[str, Any]], int]: + out: dict[tuple[Any, ...], dict[str, Any]] = {} + duplicates = 0 + for rs in rows: + k = key_fn(rs) + if k in out: + duplicates += 1 + continue + out[k] = rs + return out, duplicates + + def _extract_input(rs: dict[str, Any]) -> Any: + inst = rs.get('instance') or {} + inp = inst.get('input', None) + if isinstance(inp, dict) and 'text' in inp: + return inp.get('text', None) + return inp + + def _extract_prompt(rs: dict[str, Any]) -> Any: + req = rs.get('request') or {} + return req.get('prompt', None) + + def _extract_completion(rs: dict[str, Any]) -> Any: + res = rs.get('result') or {} + comps = res.get('completions') or [] + if not comps: + return None + first = comps[0] + if isinstance(first, dict): + return first.get('text', None) + return first + + def _summarize( + map_a: dict[tuple[Any, ...], dict[str, Any]], + map_b: dict[tuple[Any, ...], dict[str, Any]], + keys: set[tuple[Any, ...]], + *, + extractor, + ) -> tuple[dict[str, Any], list[dict[str, Any]]]: + def _keysort(k: tuple[Any, ...]) -> str: + try: + return ub.urepr(k, nl=0, compact=1) + except Exception: + return str(k) + + comparable = 0 + mismatched = 0 + examples: list[dict[str, Any]] = [] + for k in sorted(keys, key=_keysort): + va = extractor(map_a[k]) + vb = extractor(map_b[k]) + comparable += 1 + if va != vb: + mismatched += 1 + if len(examples) < max_examples: + examples.append( + { + 'key': _key_to_serializable(k), + 'a': _short_urepr(va, max_chars=180), + 'b': _short_urepr(vb, max_chars=180), + } + ) + return ( + { + 'comparable': comparable, + 'mismatched': mismatched, + 'equal_ratio': ratio(comparable, mismatched), + }, + examples, + ) + + base_a, dup_base_a = _index_unique(request_states_a, _base_key) + base_b, dup_base_b = _index_unique(request_states_b, _base_key) + var_a, dup_var_a = _index_unique(request_states_a, _variant_key) + var_b, dup_var_b = _index_unique(request_states_b, _variant_key) + + cov_base = Coverage.from_sets(set(base_a), set(base_b)) + cov_var = Coverage.from_sets(set(var_a), set(var_b)) + isect_base = set(base_a) & set(base_b) + isect_var = set(var_a) & set(var_b) + + base_iou = ( + cov_base.n_isect / cov_base.n_union if cov_base.n_union else None + ) + variant_iou = ( + cov_var.n_isect / cov_var.n_union if cov_var.n_union else None + ) + + input_eq, ex_input = _summarize( + base_a, base_b, isect_base, extractor=_extract_input + ) + prompt_eq, ex_prompt = _summarize( + var_a, var_b, isect_var, extractor=_extract_prompt + ) + completion_eq, ex_completion = _summarize( + var_a, var_b, isect_var, extractor=_extract_completion + ) + + out = { + 'base_coverage': cov_base.__dict__, + 'variant_coverage': cov_var.__dict__, + 'base_iou': base_iou, + 'variant_iou': variant_iou, + 'content_equality': { + 'input': input_eq, + 'prompt': prompt_eq, + 'completion': completion_eq, + }, + 'duplicates': { + 'a': {'base': dup_base_a, 'variant': dup_var_a}, + 'b': {'base': dup_base_b, 'variant': dup_var_b}, + }, + 'mismatch_examples': { + 'input': ex_input, + 'prompt': ex_prompt, + 'completion': ex_completion, + }, + } + return _json_compatible(out) + + class HelmRunDiff(ub.NiceRepr): """Compare two HELM runs. @@ -382,7 +842,7 @@ def summary_dict(self, *, level: int = 10) -> dict[str, Any]: a_run_name is not None ) - # 2) run spec dict hash + # 2) run spec dict hash (strict) spec_hash_a = helm_hashers.stable_hash36( helm_hashers.canonicalize_for_hashing(a_spec) ) @@ -390,6 +850,14 @@ def summary_dict(self, *, level: int = 10) -> dict[str, Any]: helm_hashers.canonicalize_for_hashing(b_spec) ) run_spec_dict_ok = spec_hash_a == spec_hash_b + if run_spec_dict_ok: + spec_path_info: dict[str, list[str]] = { + 'unique1': [], + 'unique2': [], + 'faillist': [], + } + else: + spec_path_info = _walker_diff_paths(a_spec, b_spec) if level == 0: spec_diff_paths = None else: @@ -397,13 +865,42 @@ def summary_dict(self, *, level: int = 10) -> dict[str, Any]: {} if run_spec_dict_ok else _walker_diff(a_spec, b_spec) ) + # 2b) run spec semantic hash (order-insensitive for metric lists) + a_spec_sem = _canonicalize_run_spec_for_semantic_diff(a_spec) + b_spec_sem = _canonicalize_run_spec_for_semantic_diff(b_spec) + spec_sem_hash_a = helm_hashers.stable_hash36(a_spec_sem) + spec_sem_hash_b = helm_hashers.stable_hash36(b_spec_sem) + run_spec_semantic_dict_ok = spec_sem_hash_a == spec_sem_hash_b + if run_spec_semantic_dict_ok: + spec_sem_path_info: dict[str, list[str]] = { + 'unique1': [], + 'unique2': [], + 'faillist': [], + } + else: + spec_sem_path_info = _walker_diff_paths(a_spec_sem, b_spec_sem) + if level == 0: + spec_sem_diff_paths = None + else: + spec_sem_diff_paths = ( + {} + if run_spec_semantic_dict_ok + else _walker_diff(a_spec_sem, b_spec_sem) + ) + run_spec_semantic = self._run_spec_semantic_summary( + path_info=spec_sem_path_info, + a_spec=a_spec, + b_spec=b_spec, + ) + # 3) scenario check with unknown semantics scen_known = bool(a_scen) and bool(b_scen) if not scen_known: scenario_ok: bool | None = None scenario_hash_a = None scenario_hash_b = None - scen_diff_paths: list[str] = [] + scen_diff_paths = [] + scen_path_info: dict[str, list[str]] | None = None else: scenario_hash_a = helm_hashers.stable_hash36( helm_hashers.canonicalize_for_hashing(a_scen) @@ -412,12 +909,19 @@ def summary_dict(self, *, level: int = 10) -> dict[str, Any]: helm_hashers.canonicalize_for_hashing(b_scen) ) scenario_ok = scenario_hash_a == scenario_hash_b + if scenario_ok: + scen_path_info = {'unique1': [], 'unique2': [], 'faillist': []} + else: + scen_path_info = _walker_diff_paths(a_scen, b_scen) if level == 0: scen_diff_paths = None else: scen_diff_paths = ( {} if scenario_ok else _walker_diff(a_scen, b_scen) ) + scenario_semantic = self._scenario_semantic_summary( + scenario_ok=scenario_ok, path_info=scen_path_info + ) # 4/5) stats coverage a_stats = self.a.stats() or [] @@ -458,6 +962,19 @@ def summary_dict(self, *, level: int = 10) -> dict[str, Any]: # 6) value agreement (means) on intersecting keys value_summary = self._value_agreement_summary() + dataset_summary: dict[str, Any] | None = None + if level >= 5: + try: + dataset_summary = self.dataset_overlap_summary(max_examples=5) + except Exception as ex: # nocover + dataset_summary = {'error': repr(ex)} + diagnosis = self._diagnose_repro( + run_spec_name_ok=run_spec_name_ok, + run_spec_semantic=run_spec_semantic, + scenario_semantic=scenario_semantic, + dataset_overlap=dataset_summary, + value_summary=value_summary, + ) out: dict[str, Any] = { 'a': self._lite_run_dict(self.a), @@ -469,13 +986,21 @@ def summary_dict(self, *, level: int = 10) -> dict[str, Any]: 'run_spec_hash_a': spec_hash_a, 'run_spec_hash_b': spec_hash_b, 'run_spec_diff_paths': spec_diff_paths, + 'run_spec_semantic_dict_ok': run_spec_semantic_dict_ok, + 'run_spec_semantic_hash_a': spec_sem_hash_a, + 'run_spec_semantic_hash_b': spec_sem_hash_b, + 'run_spec_diff_paths_semantic': spec_sem_diff_paths, + 'run_spec_semantic': run_spec_semantic, 'scenario_ok': scenario_ok, 'scenario_hash_a': scenario_hash_a, 'scenario_hash_b': scenario_hash_b, 'scenario_diff_paths': scen_diff_paths, + 'scenario_semantic': scenario_semantic, 'stats_coverage_by_name': cov_name.__dict__, 'stats_coverage_by_name_count': cov_name_count.__dict__, 'value_agreement': value_summary, + 'dataset_overlap': dataset_summary, + 'diagnosis': diagnosis, } if level >= 20: @@ -487,6 +1012,7 @@ def summary_dict(self, *, level: int = 10) -> dict[str, Any]: except Exception as ex: # nocover out['instance_value_agreement'] = {'error': repr(ex)} + out = _json_compatible(out) self._cache[cache_key] = out return out @@ -524,7 +1050,7 @@ def summary(self, *, level: int = 10, writer=None) -> None: f'{_format_bool(ok)} {self.a_name} vs {self.b_name} {line_name} ' f'spec={_format_bool(info["run_spec_dict_ok"])} ' f'stats={cov["n_isect"]}/{cov["n_union"]} ' - f'agree={agree:.3f}' + f'agree={_fmt(agree)}' ) if level > 0: @@ -549,8 +1075,21 @@ def summary(self, *, level: int = 10, writer=None) -> None: ) if level >= 15: - if not info['run_spec_dict_ok']: - writer(f' diff: {ub.urepr(info["run_spec_diff_paths"])}') + if not info.get('run_spec_semantic_dict_ok', True): + writer( + f' semantic diff: {ub.urepr(info["run_spec_diff_paths_semantic"])}' + ) + if (not info['run_spec_dict_ok']) and level >= 20: + writer( + f' strict diff: {ub.urepr(info["run_spec_diff_paths"])}' + ) + rs_sem = info.get('run_spec_semantic', {}) or {} + dep = rs_sem.get('deployment', {}) or {} + if dep.get('changed', False): + writer( + f' deployment: A={_short_urepr(dep.get("a", None), 80)} ' + f'B={_short_urepr(dep.get("b", None), 80)}' + ) if info['scenario_ok'] is None: writer( @@ -565,6 +1104,11 @@ def summary(self, *, level: int = 10, writer=None) -> None: writer( f' diff: {ub.urepr(info["scenario_diff_paths"])}' ) + scen_sem = info.get('scenario_semantic', {}) or {} + if scen_sem.get('known', False) and level >= 15: + writer( + f'Scenario semantic: {_format_bool(bool(scen_sem.get("semantic_ok", False)))}' + ) writer('') cov2 = info['stats_coverage_by_name_count'] @@ -583,13 +1127,13 @@ def summary(self, *, level: int = 10, writer=None) -> None: ov = info['value_agreement']['overall'] writer( f' overall: comparable={ov["comparable"]} mismatched={ov["mismatched"]} ' - f'agree_ratio={ov["agree_ratio"]:.3f}' + f'agree_ratio={_fmt(ov["agree_ratio"])}' ) for cls in ('core', 'bookkeeping', 'untracked'): s = info['value_agreement']['by_class'][cls] writer( f' {cls:11s}: comparable={s["comparable"]} mismatched={s["mismatched"]} ' - f'agree_ratio={s["agree_ratio"]:.3f}' + f'agree_ratio={_fmt(s["agree_ratio"])}' ) if level >= 20: @@ -601,6 +1145,34 @@ def summary(self, *, level: int = 10, writer=None) -> None: f' {r["key"]} A={_fmt(r["a"])} B={_fmt(r["b"])} |Δ|={_fmt(r["abs_delta"])}' ) + if level >= 15: + ds = info.get('dataset_overlap', None) + if isinstance(ds, dict) and 'error' not in ds: + writer('') + writer('Dataset overlap:') + writer( + f' base_iou={_fmt(ds.get("base_iou"))} ' + f'variant_iou={_fmt(ds.get("variant_iou"))}' + ) + ce = ds.get('content_equality', {}) or {} + for field in ('input', 'prompt', 'completion'): + row = ce.get(field, {}) or {} + writer( + f' {field:10s}: comparable={row.get("comparable")} ' + f'mismatched={row.get("mismatched")} equal_ratio={_fmt(row.get("equal_ratio"))}' + ) + elif isinstance(ds, dict) and 'error' in ds: + writer(f'Dataset overlap: ⚠️ {ds["error"]}') + + if level >= 10: + diag = info.get('diagnosis', {}) or {} + writer('') + writer(f'Diagnosis: {diag.get("label", "unknown")}') + if level >= 20: + reasons = diag.get('reasons', []) + if reasons: + writer(f' reasons: {ub.urepr(reasons, nl=0)}') + if level >= 30: writer('') try: @@ -611,8 +1183,8 @@ def summary(self, *, level: int = 10, writer=None) -> None: means = inst['means'] writer( f'Instance-level means: comparable={means["comparable"]} mismatched={means["mismatched"]} ' - f'agree={means["agree_ratio"]:.3f} (unpert={means["agree_ratio_unperturbed"]:.3f}, ' - f'pert={means["agree_ratio_perturbed"]:.3f})' + f'agree={_fmt(means["agree_ratio"])} (unpert={_fmt(means["agree_ratio_unperturbed"])}, ' + f'pert={_fmt(means["agree_ratio_perturbed"])})' ) def _analysis_summary_line( @@ -650,6 +1222,405 @@ def _lite_run_dict(self, ana: HelmRunAnalysis) -> dict[str, Any]: spec = ana.run_spec() or {} return {'run_spec_name': spec.get('name', None)} + def _run_spec_semantic_summary( + self, + *, + path_info: dict[str, list[str]], + a_spec: dict[str, Any], + b_spec: dict[str, Any], + ) -> dict[str, Any]: + """Classify run-spec differences into semantic buckets.""" + all_paths = sorted( + set(path_info.get('unique1', [])) + | set(path_info.get('unique2', [])) + | set(path_info.get('faillist', [])) + ) + by_class: dict[str, list[str]] = { + 'execution': [], + 'evaluation': [], + 'nonsemantic': [], + 'other': [], + } + for p in all_paths: + by_class[_classify_run_spec_path(p)].append(p) + + deployment_paths = [ + p + for p in all_paths + if p.startswith('adapter_spec.model_deployment') + ] + deployment_a = ( + (a_spec.get('adapter_spec', {}) or {}).get( + 'model_deployment', None + ) + if isinstance(a_spec, dict) + else None + ) + deployment_b = ( + (b_spec.get('adapter_spec', {}) or {}).get( + 'model_deployment', None + ) + if isinstance(b_spec, dict) + else None + ) + deployment_changed = (deployment_a != deployment_b) or any( + p.startswith('adapter_spec.model_deployment') for p in all_paths + ) + metric_specs_delta = _metric_specs_multiset_delta( + (a_spec or {}).get('metric_specs', None), + (b_spec or {}).get('metric_specs', None), + short_hash=self.short_hash, + max_items=20, + ) + evaluation_changed = bool(by_class['evaluation']) or ( + not bool(metric_specs_delta.get('equal_as_multiset', True)) + ) + execution_ok = len(by_class['execution']) == 0 + evaluation_only = ( + (len(all_paths) > 0) + and execution_ok + and (evaluation_changed or len(by_class['nonsemantic']) > 0) + ) + return _json_compatible( + { + 'n_total_paths': len(all_paths), + 'execution_ok': execution_ok, + 'evaluation_only': evaluation_only, + 'evaluation_changed': evaluation_changed, + 'deployment_changed': deployment_changed, + 'deployment': { + 'a': deployment_a, + 'b': deployment_b, + 'changed': deployment_changed, + }, + 'counts': { + k: len(v) + for k, v in by_class.items() + }, + 'deployment_paths': _preview_list(deployment_paths, limit=20), + 'execution_paths': _preview_list( + by_class['execution'], limit=20 + ), + 'execution_value_examples': _path_value_examples( + a_spec, b_spec, by_class['execution'], max_items=20 + ), + 'evaluation_paths': _preview_list( + by_class['evaluation'], limit=20 + ), + 'metric_specs_multiset_delta': metric_specs_delta, + 'nonsemantic_paths': _preview_list( + by_class['nonsemantic'], limit=20 + ), + 'other_paths': _preview_list(by_class['other'], limit=20), + } + ) + + def _scenario_semantic_summary( + self, + *, + scenario_ok: bool | None, + path_info: dict[str, list[str]] | None, + ) -> dict[str, Any]: + """Classify scenario differences into semantic/nonsemantic buckets.""" + if scenario_ok is None: + return { + 'known': False, + 'strict_ok': None, + 'semantic_ok': None, + 'counts': {'semantic': 0, 'nonsemantic': 0}, + 'semantic_paths': [], + 'nonsemantic_paths': [], + } + + path_info = path_info or {'unique1': [], 'unique2': [], 'faillist': []} + all_paths = sorted( + set(path_info.get('unique1', [])) + | set(path_info.get('unique2', [])) + | set(path_info.get('faillist', [])) + ) + semantic_paths = [ + p for p in all_paths if _classify_scenario_path(p) == 'semantic' + ] + nonsemantic_paths = [ + p for p in all_paths if _classify_scenario_path(p) == 'nonsemantic' + ] + semantic_ok = bool(scenario_ok) or (len(semantic_paths) == 0) + return _json_compatible( + { + 'known': True, + 'strict_ok': bool(scenario_ok), + 'semantic_ok': semantic_ok, + 'counts': { + 'semantic': len(semantic_paths), + 'nonsemantic': len(nonsemantic_paths), + }, + 'semantic_paths': _preview_list(semantic_paths, limit=20), + 'nonsemantic_paths': _preview_list( + nonsemantic_paths, limit=20 + ), + } + ) + + def dataset_overlap_summary(self, *, max_examples: int = 5) -> dict[str, Any]: + """Compare scenario_state request datasets between runs. + + Example: + >>> from magnet.backends.helm.helm_run_analysis import HelmRunAnalysis + >>> ana = HelmRunAnalysis.__new__(HelmRunAnalysis) + >>> ana._raw_cache = {} + >>> ana._cache = {} + >>> ana.run = None + >>> ana.name = None + >>> ana.scenario_state = lambda: {'request_states': [ + ... {'instance': {'id': 'id1', 'split': 'test', 'input': {'text': 'Q1'}}, + ... 'train_trial_index': 0, + ... 'request': {'prompt': 'P1'}, + ... 'result': {'completions': [{'text': 'A1'}]}}, + ... ]} + >>> rd = HelmRunDiff(ana, ana) + >>> ds = rd.dataset_overlap_summary(max_examples=2) + >>> assert ds['base_iou'] == 1.0 + >>> assert ds['variant_iou'] == 1.0 + >>> assert ds['content_equality']['input']['equal_ratio'] == 1.0 + """ + cache_key = ('dataset_overlap_summary', max_examples, self.short_hash) + if cache_key in self._cache: + return self._cache[cache_key] + rs_a = (self.a.scenario_state() or {}).get('request_states', []) or [] + rs_b = (self.b.scenario_state() or {}).get('request_states', []) or [] + out = dataset_overlap_from_request_states( + rs_a, + rs_b, + short_hash=self.short_hash, + max_examples=max_examples, + ) + out = _json_compatible(out) + self._cache[cache_key] = out + return out + + def _diagnose_repro( + self, + *, + run_spec_name_ok: bool, + run_spec_semantic: dict[str, Any], + scenario_semantic: dict[str, Any], + dataset_overlap: dict[str, Any] | None, + value_summary: dict[str, Any], + ) -> dict[str, Any]: + """High-level diagnosis for reproducibility triage. + + Returns a primary label plus a full list of contributing reasons. + Lower ``priority`` is earlier / more significant in the pipeline. + """ + reasons: list[dict[str, Any]] = [] + + def add_reason(name: str, priority: int, details: dict[str, Any]) -> None: + reasons.append( + { + 'name': name, + 'priority': int(priority), + 'details': _json_compatible(details), + } + ) + + # Priority 0: run pairing / spec-level execution blockers + if not run_spec_name_ok: + add_reason( + 'wrong_run_pair', + 0, + {'run_spec_name_ok': False}, + ) + + execution_ok = bool(run_spec_semantic.get('execution_ok', False)) + execution_paths = run_spec_semantic.get('execution_paths', []) or [] + deployment_paths = run_spec_semantic.get('deployment_paths', []) or [] + non_deployment_execution_paths = [ + p + for p in execution_paths + if not str(p).startswith('adapter_spec.model_deployment') + ] + if not execution_ok and non_deployment_execution_paths: + add_reason( + 'execution_spec_drift', + 0, + { + 'execution_paths': execution_paths, + 'execution_value_examples': run_spec_semantic.get( + 'execution_value_examples', [] + ), + 'counts': run_spec_semantic.get('counts', {}), + }, + ) + + if bool(run_spec_semantic.get('deployment_changed', False)): + dep = run_spec_semantic.get('deployment', {}) or {} + add_reason( + 'deployment_drift', + 0, + { + 'a_value': dep.get('a', None), + 'b_value': dep.get('b', None), + 'execution_paths': [ + p + for p in ( + execution_paths + ) + if str(p).startswith('adapter_spec.model_deployment') + ] + or deployment_paths, + }, + ) + + scen_known = bool(scenario_semantic.get('known', False)) + scen_semantic_ok = scenario_semantic.get('semantic_ok', None) + if scen_known and not bool(scen_semantic_ok): + add_reason( + 'scenario_spec_drift', + 0, + { + 'semantic_paths': scenario_semantic.get( + 'semantic_paths', [] + ), + 'counts': scenario_semantic.get('counts', {}), + }, + ) + + # Priority 1: dataset/request-state drift + if isinstance(dataset_overlap, dict): + if 'error' in dataset_overlap: + add_reason( + 'dataset_overlap_error', + 1, + {'error': dataset_overlap.get('error', None)}, + ) + else: + base_iou = dataset_overlap.get('base_iou', None) + variant_iou = dataset_overlap.get('variant_iou', None) + if base_iou is not None and base_iou < 1.0: + add_reason( + 'dataset_instance_drift', + 1, + { + 'base_iou': base_iou, + 'base_coverage': dataset_overlap.get( + 'base_coverage', {} + ), + }, + ) + if variant_iou is not None and variant_iou < 1.0: + add_reason( + 'dataset_variant_drift', + 1, + { + 'variant_iou': variant_iou, + 'variant_coverage': dataset_overlap.get( + 'variant_coverage', {} + ), + }, + ) + + ce = dataset_overlap.get('content_equality', {}) or {} + mex = dataset_overlap.get('mismatch_examples', {}) or {} + for field, reason_name, pr in [ + ('input', 'dataset_input_drift', 1), + ('prompt', 'request_prompt_drift', 1), + ('completion', 'completion_content_drift', 2), + ]: + row = ce.get(field, {}) or {} + eq = row.get('equal_ratio', None) + if eq is not None and eq < 1.0: + details = dict(row) + examples = mex.get(field, None) + if examples: + details['examples'] = examples + add_reason(reason_name, pr, details) + + # Priority 2: evaluation schema / metric set drift + metric_specs_delta = ( + run_spec_semantic.get('metric_specs_multiset_delta', {}) or {} + ) + eval_paths = run_spec_semantic.get('evaluation_paths', []) or [] + evaluation_changed = bool(eval_paths) or ( + not bool(metric_specs_delta.get('equal_as_multiset', True)) + ) + if evaluation_changed: + details = {'evaluation_paths': eval_paths} + if not bool(metric_specs_delta.get('equal_as_multiset', True)): + details['metric_specs_multiset_delta'] = metric_specs_delta + add_reason( + 'evaluation_spec_drift', + 2, + details, + ) + + # Priority 3: value-level drift (may be downstream effect) + core = ((value_summary.get('by_class') or {}).get('core') or {}) + book = ((value_summary.get('by_class') or {}).get('bookkeeping') or {}) + core_ratio = core.get('agree_ratio', None) + book_ratio = book.get('agree_ratio', None) + + if core_ratio is None: + add_reason( + 'no_comparable_core_metrics', + 3, + {'core': core}, + ) + else: + if core_ratio < 0.995: + add_reason( + 'core_metric_drift', + 3, + { + 'core_agree_ratio': core_ratio, + 'core': core, + }, + ) + elif (book_ratio is not None) and (book_ratio < 0.95): + add_reason( + 'bookkeeping_metric_drift', + 3, + { + 'core_agree_ratio': core_ratio, + 'bookkeeping_agree_ratio': book_ratio, + 'bookkeeping': book, + }, + ) + + if not reasons: + add_reason( + 'no_detected_drift', + 0, + { + 'core_agree_ratio': core_ratio, + 'bookkeeping_agree_ratio': book_ratio, + }, + ) + + reasons = sorted( + reasons, + key=lambda r: ( + int(r.get('priority', 999)), + str(r.get('name', '')), + ), + ) + min_priority = min(int(r['priority']) for r in reasons) + primary_reason_names = [ + r['name'] for r in reasons if int(r['priority']) == min_priority + ] + if primary_reason_names == ['no_detected_drift']: + label = 'reproduced' + elif len(primary_reason_names) == 1: + label = primary_reason_names[0] + else: + label = 'multiple_primary_reasons' + + return { + 'label': label, + 'primary_priority': min_priority, + 'primary_reason_names': primary_reason_names, + 'reasons': reasons, + } + # --------------------------------------------------------------------- # Run-level mean agreement @@ -733,6 +1704,94 @@ def agrees(x: float, y: float) -> bool: 'top_mismatches': top, } + out = _json_compatible(out) + self._cache[cache_key] = out + return out + + def value_distance_profile( + self, + *, + top_n: int = 12, + ) -> dict[str, Any]: + """Programmatic raw distance summary for intersecting run-level stats. + + Unlike :meth:`_value_agreement_summary`, this does not threshold values + into matched / mismatched. It reports absolute / relative deltas and + their distributions so tolerance policies can be applied later without + recomputing the underlying joins. + """ + cache_key = ('value_distance_profile', top_n, self.short_hash) + if cache_key in self._cache: + return self._cache[cache_key] + + idx_a = self.a.stat_index( + drop_zero_count=True, require_mean=True, short_hash=self.short_hash + ) + idx_b = self.b.stat_index( + drop_zero_count=True, require_mean=True, short_hash=self.short_hash + ) + keys = set(idx_a.keys()) & set(idx_b.keys()) + + by_class: dict[str, list[dict[str, Any]]] = { + 'core': [], + 'bookkeeping': [], + 'untracked': [], + } + all_rows: list[dict[str, Any]] = [] + for k in keys: + a = idx_a[k] + b = idx_b[k] + if a.mean is None or b.mean is None: + continue + abs_delta = abs(a.mean - b.mean) + denom = max(abs(a.mean), abs(b.mean), 1e-12) + rel_delta = abs_delta / denom + row = { + 'key': k, + 'a': a.mean, + 'b': b.mean, + 'abs_delta': abs_delta, + 'rel_delta': rel_delta, + 'metric_class': a.metric_class, + } + all_rows.append(row) + by_class.setdefault(a.metric_class, []).append(row) + + def summarize(rows: list[dict[str, Any]]) -> dict[str, Any]: + if not rows: + return { + 'count': 0, + 'abs_delta': {'min': None, 'p50': None, 'p90': None, 'p99': None, 'max': None}, + 'rel_delta': {'min': None, 'p50': None, 'p90': None, 'p99': None, 'max': None}, + 'top_abs_deltas': [], + } + abs_vals = sorted(float(r['abs_delta']) for r in rows) + rel_vals = sorted(float(r['rel_delta']) for r in rows) + top = sorted(rows, key=lambda r: r['abs_delta'], reverse=True)[:top_n] + return { + 'count': len(rows), + 'abs_delta': { + 'min': abs_vals[0], + 'p50': _quantile(abs_vals, 0.50), + 'p90': _quantile(abs_vals, 0.90), + 'p99': _quantile(abs_vals, 0.99), + 'max': abs_vals[-1], + }, + 'rel_delta': { + 'min': rel_vals[0], + 'p50': _quantile(rel_vals, 0.50), + 'p90': _quantile(rel_vals, 0.90), + 'p99': _quantile(rel_vals, 0.99), + 'max': rel_vals[-1], + }, + 'top_abs_deltas': top, + } + + out = { + 'overall': summarize(all_rows), + 'by_class': {cls: summarize(rows) for cls, rows in by_class.items()}, + } + out = _json_compatible(out) self._cache[cache_key] = out return out @@ -761,9 +1820,9 @@ def instance_summary_dict( - agree_ratio: 1 - mismatched / comparable - agree_ratio_unperturbed / agree_ratio_perturbed * top_mismatches_by_group: - Mapping from ``(metric_class, metric_name)`` to a list of mismatch - items (sorted by |Δ|), each containing: - key, a, b, abs_delta, signed_delta + List of objects; each element has fields ``metric_class`` and + ``metric`` and an ``items`` list of mismatches (sorted by |Δ|). + Each mismatch item contains: key, a, b, abs_delta, signed_delta. Notes ----- @@ -834,6 +1893,8 @@ def agrees(x: float, y: float) -> bool: 'perturbed': {'comparable': 0, 'mismatched': 0}, } + # use a temporary map for bookkeeping, but the final result will be a + # list of objects so the summary dict is JSON-serializable. grouped: dict[tuple[str, str | None], list[dict[str, Any]]] = {} for k in set_a & set_b: @@ -909,7 +1970,7 @@ def agrees(x: float, y: float) -> bool: mismatched += 1 var_stats[variant]['mismatched'] += 1 item = { - 'key': k, + 'key': _key_to_serializable(k), 'a': ma, 'b': mb, 'abs_delta': abs(ma - mb), @@ -936,15 +1997,369 @@ def agrees(x: float, y: float) -> bool: ), } + # convert to a JSON-friendly structure: list of group objects + group_list: list[dict[str, Any]] = [] + for (mclass, metric), items in grouped.items(): + group_list.append( + { + 'metric_class': mclass, + 'metric': metric, + 'items': items, + } + ) out = { 'coverage': cov.__dict__, 'means': means, - 'top_mismatches_by_group': grouped, + 'top_mismatches_by_group': group_list, } + out = _json_compatible(out) + self._cache[cache_key] = out + return out + + def instance_distance_profile( + self, + *, + top_n: int = 10, + ) -> dict[str, Any]: + """Programmatic raw distance summary for per-instance stat means.""" + cache_key = ('instance_distance_profile', top_n, self.short_hash) + if cache_key in self._cache: + return self._cache[cache_key] + + joined_a = self.a.joined_instance_stat_table( + assert_assumptions=False, short_hash=self.short_hash + ) + joined_b = self.b.joined_instance_stat_table( + assert_assumptions=False, short_hash=self.short_hash + ) + map_a = getattr(joined_a, 'row_by_key', None) + map_b = getattr(joined_b, 'row_by_key', None) + + def _iter_rows(joined) -> Iterable[Any]: + if map_a is not None and joined is joined_a: + return map_a.values() + if map_b is not None and joined is joined_b: + return map_b.values() + if isinstance(joined, dict): + return joined.values() + if hasattr(joined, '__iter__'): + return joined + return [] + + def _row_key(row: Any) -> Any: + return ( + getattr(row, 'key', None) + or getattr(row, 'stat_key', None) + or getattr(row, 'row_key', None) + or row + ) + + if map_a is None: + map_a = {_row_key(r): r for r in _iter_rows(joined_a)} + if map_b is None: + map_b = {_row_key(r): r for r in _iter_rows(joined_b)} + + all_rows: list[dict[str, Any]] = [] + by_group: dict[tuple[str, str | None], list[dict[str, Any]]] = {} + for k in set(map_a) & set(map_b): + ra = map_a[k] + rb = map_b[k] + sa = ( + getattr(ra, 'stat', None) + if hasattr(ra, 'stat') + else (ra.get('stat', None) if isinstance(ra, dict) else None) + ) + sb = ( + getattr(rb, 'stat', None) + if hasattr(rb, 'stat') + else (rb.get('stat', None) if isinstance(rb, dict) else None) + ) + ma = _safe_float( + (sa or {}).get('mean', None) + if isinstance(sa, dict) + else getattr(sa, 'mean', None) + ) + mb = _safe_float( + (sb or {}).get('mean', None) + if isinstance(sb, dict) + else getattr(sb, 'mean', None) + ) + ca = ( + int((sa or {}).get('count', 0) or 0) + if isinstance(sa, dict) + else int(getattr(sa, 'count', 0) or 0) + ) + cb = ( + int((sb or {}).get('count', 0) or 0) + if isinstance(sb, dict) + else int(getattr(sb, 'count', 0) or 0) + ) + if ma is None or mb is None or ca == 0 or cb == 0: + continue + name_obj = ( + (sa or {}).get('name', None) + if isinstance(sa, dict) + else getattr(sa, 'name_obj', None) + ) + metric = ( + name_obj.get('name', None) + if isinstance(name_obj, dict) + else None + ) + if metric is None and sa is not None and not isinstance(sa, dict): + metric = getattr(sa, 'metric', None) + metric_class, _ = helm_metrics.classify_metric(metric) + abs_delta = abs(ma - mb) + denom = max(abs(ma), abs(mb), 1e-12) + rel_delta = abs_delta / denom + item = { + 'key': _key_to_serializable(k), + 'a': ma, + 'b': mb, + 'abs_delta': abs_delta, + 'rel_delta': rel_delta, + 'signed_delta': (mb - ma), + 'metric_class': metric_class, + 'metric': metric, + } + all_rows.append(item) + by_group.setdefault((metric_class, metric), []).append(item) + + def summarize(items: list[dict[str, Any]]) -> dict[str, Any]: + if not items: + return { + 'count': 0, + 'abs_delta': {'min': None, 'p50': None, 'p90': None, 'p99': None, 'max': None}, + 'rel_delta': {'min': None, 'p50': None, 'p90': None, 'p99': None, 'max': None}, + } + abs_vals = sorted(float(r['abs_delta']) for r in items) + rel_vals = sorted(float(r['rel_delta']) for r in items) + return { + 'count': len(items), + 'abs_delta': { + 'min': abs_vals[0], + 'p50': _quantile(abs_vals, 0.50), + 'p90': _quantile(abs_vals, 0.90), + 'p99': _quantile(abs_vals, 0.99), + 'max': abs_vals[-1], + }, + 'rel_delta': { + 'min': rel_vals[0], + 'p50': _quantile(rel_vals, 0.50), + 'p90': _quantile(rel_vals, 0.90), + 'p99': _quantile(rel_vals, 0.99), + 'max': rel_vals[-1], + }, + } + + grouped_rows = [] + for (metric_class, metric), items in sorted(by_group.items()): + items = sorted(items, key=lambda r: r['abs_delta'], reverse=True) + grouped_rows.append( + { + 'metric_class': metric_class, + 'metric': metric, + 'summary': summarize(items), + 'top_abs_deltas': items[:top_n], + } + ) + out = { + 'overall': summarize(all_rows), + 'by_metric': grouped_rows, + } + out = _json_compatible(out) + self._cache[cache_key] = out + return out + + def instance_agreement_profile( + self, + *, + abs_tol: float = 0.0, + rel_tol: float = 0.0, + ) -> dict[str, Any]: + """Programmatic agreement summary grouped by per-instance metric.""" + cache_key = ('instance_agreement_profile', abs_tol, rel_tol, self.short_hash) + if cache_key in self._cache: + return self._cache[cache_key] + + joined_a = self.a.joined_instance_stat_table( + assert_assumptions=False, short_hash=self.short_hash + ) + joined_b = self.b.joined_instance_stat_table( + assert_assumptions=False, short_hash=self.short_hash + ) + map_a = getattr(joined_a, 'row_by_key', None) + map_b = getattr(joined_b, 'row_by_key', None) + + def _iter_rows(joined) -> Iterable[Any]: + if map_a is not None and joined is joined_a: + return map_a.values() + if map_b is not None and joined is joined_b: + return map_b.values() + if isinstance(joined, dict): + return joined.values() + if hasattr(joined, '__iter__'): + return joined + return [] + + def _row_key(row: Any) -> Any: + return ( + getattr(row, 'key', None) + or getattr(row, 'stat_key', None) + or getattr(row, 'row_key', None) + or row + ) + + if map_a is None: + map_a = {_row_key(r): r for r in _iter_rows(joined_a)} + if map_b is None: + map_b = {_row_key(r): r for r in _iter_rows(joined_b)} + + def agrees(x: float, y: float) -> bool: + if abs_tol == 0.0 and rel_tol == 0.0: + return x == y + return abs(x - y) <= max(abs_tol, rel_tol * max(abs(x), abs(y))) + + overall = {'comparable': 0, 'mismatched': 0} + by_group: dict[tuple[str, str | None], dict[str, Any]] = {} + + for k in set(map_a) & set(map_b): + ra = map_a[k] + rb = map_b[k] + sa = ( + getattr(ra, 'stat', None) + if hasattr(ra, 'stat') + else (ra.get('stat', None) if isinstance(ra, dict) else None) + ) + sb = ( + getattr(rb, 'stat', None) + if hasattr(rb, 'stat') + else (rb.get('stat', None) if isinstance(rb, dict) else None) + ) + ma = _safe_float( + (sa or {}).get('mean', None) + if isinstance(sa, dict) + else getattr(sa, 'mean', None) + ) + mb = _safe_float( + (sb or {}).get('mean', None) + if isinstance(sb, dict) + else getattr(sb, 'mean', None) + ) + ca = ( + int((sa or {}).get('count', 0) or 0) + if isinstance(sa, dict) + else int(getattr(sa, 'count', 0) or 0) + ) + cb = ( + int((sb or {}).get('count', 0) or 0) + if isinstance(sb, dict) + else int(getattr(sb, 'count', 0) or 0) + ) + if ma is None or mb is None or ca == 0 or cb == 0: + continue + name_obj = ( + (sa or {}).get('name', None) + if isinstance(sa, dict) + else getattr(sa, 'name_obj', None) + ) + metric = ( + name_obj.get('name', None) + if isinstance(name_obj, dict) + else None + ) + if metric is None and sa is not None and not isinstance(sa, dict): + metric = getattr(sa, 'metric', None) + metric_class, _ = helm_metrics.classify_metric(metric) + key = (metric_class, metric) + group = by_group.setdefault( + key, + { + 'metric_class': metric_class, + 'metric': metric, + 'comparable': 0, + 'mismatched': 0, + }, + ) + overall['comparable'] += 1 + group['comparable'] += 1 + if not agrees(ma, mb): + overall['mismatched'] += 1 + group['mismatched'] += 1 + + grouped_rows = [] + for _, group in sorted(by_group.items()): + grouped_rows.append( + { + 'metric_class': group['metric_class'], + 'metric': group['metric'], + 'comparable': group['comparable'], + 'mismatched': group['mismatched'], + 'agree_ratio': ratio(group['comparable'], group['mismatched']), + } + ) + + out = { + 'overall': { + 'comparable': overall['comparable'], + 'mismatched': overall['mismatched'], + 'agree_ratio': ratio(overall['comparable'], overall['mismatched']), + }, + 'by_metric': grouped_rows, + } + out = _json_compatible(out) self._cache[cache_key] = out return out + def tolerance_sweep_summary( + self, + *, + run_tolerances: list[dict[str, Any]] | None = None, + instance_tolerances: list[dict[str, Any]] | None = None, + ) -> dict[str, Any]: + """Evaluate multiple tolerance policies without recomputing run loading.""" + if run_tolerances is None: + run_tolerances = [ + {'name': 'strict', 'abs_tol': 0.0, 'rel_tol': 0.0}, + {'name': 'tiny', 'abs_tol': 1e-12, 'rel_tol': 1e-6}, + {'name': 'small', 'abs_tol': 1e-9, 'rel_tol': 1e-4}, + {'name': 'medium', 'abs_tol': 1e-6, 'rel_tol': 1e-3}, + {'name': 'loose', 'abs_tol': 1e-3, 'rel_tol': 1e-2}, + ] + if instance_tolerances is None: + instance_tolerances = list(run_tolerances) + + run_results = [] + for cfg in run_tolerances: + summary = self._value_agreement_summary( + abs_tol=float(cfg.get('abs_tol', 0.0) or 0.0), + rel_tol=float(cfg.get('rel_tol', 0.0) or 0.0), + ) + run_results.append({ + 'name': cfg.get('name', 'unnamed'), + 'abs_tol': float(cfg.get('abs_tol', 0.0) or 0.0), + 'rel_tol': float(cfg.get('rel_tol', 0.0) or 0.0), + 'summary': summary, + }) + + instance_results = [] + for cfg in instance_tolerances: + summary = self.instance_summary_dict( + abs_tol=float(cfg.get('abs_tol', 0.0) or 0.0), + rel_tol=float(cfg.get('rel_tol', 0.0) or 0.0), + ) + instance_results.append({ + 'name': cfg.get('name', 'unnamed'), + 'abs_tol': float(cfg.get('abs_tol', 0.0) or 0.0), + 'rel_tol': float(cfg.get('rel_tol', 0.0) or 0.0), + 'summary': summary, + }) + return _json_compatible({ + 'run_level': run_results, + 'instance_level': instance_results, + }) + def summarize_instances( self, *, @@ -980,24 +2395,22 @@ def summarize_instances( ) writer( f' means: comparable={means["comparable"]} mismatched={means["mismatched"]} ' - f'agree_ratio={means["agree_ratio"]:.3f} (unpert={means["agree_ratio_unperturbed"]:.3f}, ' - f'pert={means["agree_ratio_perturbed"]:.3f})' + f'agree_ratio={_fmt(means["agree_ratio"])} (unpert={_fmt(means["agree_ratio_unperturbed"])}, ' + f'pert={_fmt(means["agree_ratio_perturbed"])})' ) - grouped: dict[tuple[str, str | None], list[dict[str, Any]]] = ( - info.get('top_mismatches_by_group', {}) or {} - ) + grouped: list[dict[str, Any]] = info.get('top_mismatches_by_group', []) or [] + # grouped is now a list of group objects; convert to list for sorting # Choose groups to show: core first, bookkeeping second - def _group_rank( - item: tuple[tuple[str, str | None], list[dict[str, Any]]], - ) -> tuple[int, float]: - (cls, _metric), items = item + def _group_rank(group: dict[str, Any]) -> tuple[int, float]: + cls = group.get('metric_class') + items = group.get('items', []) cls_rank = {'core': 0, 'bookkeeping': 1, 'untracked': 2}.get(cls, 9) max_abs = items[0]['abs_delta'] if items else 0.0 return (cls_rank, -max_abs) - groups_sorted = sorted(grouped.items(), key=_group_rank) + groups_sorted = sorted(grouped, key=_group_rank) # Decide which metric classes are eligible at this level allowed_classes = {'core'} @@ -1007,14 +2420,14 @@ def _group_rank( allowed_classes |= {'bookkeeping'} # Filter groups by allowed class - filtered = [g for g in groups_sorted if g[0][0] in allowed_classes] + filtered = [g for g in groups_sorted if g.get('metric_class') in allowed_classes] # If we filtered everything out (e.g. no core diffs), fall back to showing *something* if not filtered and groups_sorted: # Prefer untracked, then bookkeeping, then whatever exists pref_order = ['untracked', 'bookkeeping', 'core'] for cls in pref_order: - filtered = [g for g in groups_sorted if g[0][0] == cls] + filtered = [g for g in groups_sorted if g.get('metric_class') == cls] if filtered: break if not filtered: @@ -1028,12 +2441,12 @@ def _group_rank( max_groups = 6 if max_groups is not None: # Ensure we show at least some core and some bookkeeping groups when possible. - core = [g for g in groups_sorted if g[0][0] == 'core'] - book = [g for g in groups_sorted if g[0][0] == 'bookkeeping'] + core = [g for g in groups_sorted if g.get('metric_class') == 'core'] + book = [g for g in groups_sorted if g.get('metric_class') == 'bookkeeping'] other = [ g for g in groups_sorted - if g[0][0] not in {'core', 'bookkeeping'} + if g.get('metric_class') not in {'core', 'bookkeeping'} ] keep = [] keep.extend(core[: max_groups // 2]) @@ -1062,8 +2475,13 @@ def _group_rank( ) A_map = getattr(A_join, 'row_by_key', None) B_map = getattr(B_join, 'row_by_key', None) + A_getrow = getattr(A_join, 'get_row', None) + B_getrow = getattr(B_join, 'get_row', None) - for (cls, metric), items in groups_sorted: + for group in groups_sorted: + cls = group.get('metric_class') + metric = group.get('metric') + items = group.get('items', []) writer(f' [bold]top mismatches ({(cls, metric)!r}):[/bold]') for rank, item in enumerate(items[:top_n], start=1): k = item['key'] @@ -1072,9 +2490,9 @@ def _group_rank( abs_d = float(item['abs_delta']) signed_d = float(item['signed_delta']) - # Try to extract split/sub_split info from key if it is tuple-like + # Try to extract split/sub_split info from key if it is tuple/list-like split = None - if isinstance(k, tuple) and len(k) >= 5: + if (isinstance(k, tuple) or isinstance(k, list)) and len(k) >= 5: # (id, tti, pert_id, metric, split, ...) split = k[4] @@ -1097,8 +2515,29 @@ def _group_rank( and A_map is not None and B_map is not None ): - ra = A_map.get(k, None) - rb = B_map.get(k, None) + # Try to resolve row objects from the join tables. + # Item keys were serialized (lists); attempt to use the + # table's `get_row` API with a tuple form, which will + # reconstruct InstanceStatKey when appropriate. + ra = None + rb = None + if A_getrow is not None: + try: + lookup_key = tuple(k) if isinstance(k, list) else k + ra = A_getrow(lookup_key) + except Exception: + ra = None + if ra is None and A_map is not None: + ra = A_map.get(k, None) + + if B_getrow is not None: + try: + lookup_key = tuple(k) if isinstance(k, list) else k + rb = B_getrow(lookup_key) + except Exception: + rb = None + if rb is None and B_map is not None: + rb = B_map.get(k, None) rs_a = ( getattr(ra, 'request_state', None) @@ -1210,5 +2649,5 @@ def _completion(rs: Any) -> str: writer('') -def ratio(c: int, m: int) -> float: - return 1.0 - (m / c) if c else float('nan') +def ratio(c: int, m: int) -> float | None: + return (1.0 - (m / c)) if c else None diff --git a/magnet/backends/helm/pipeline.py b/magnet/backends/helm/pipeline.py index a765633..d4e5ef5 100644 --- a/magnet/backends/helm/pipeline.py +++ b/magnet/backends/helm/pipeline.py @@ -35,6 +35,8 @@ from __future__ import annotations +import shlex + import kwdagger @@ -84,6 +86,9 @@ class MaterializeHelmRunNode(kwdagger.ProcessNode): 'suite': 'my-suite', 'max_eval_instances': None, 'require_per_instance_stats': True, + 'model_deployments_fpath': None, + 'enable_huggingface_models': None, + 'enable_local_huggingface_models': None, # Behavior toggles that change how/what we materialize 'mode': 'compute_if_missing', # reuse_only | compute_if_missing | force_recompute 'materialize': 'symlink', # symlink | copy @@ -93,11 +98,44 @@ class MaterializeHelmRunNode(kwdagger.ProcessNode): # These are recorded, but ideally should not change the “meaning” of outputs. perf_params = { # Your shared precomputed root: - 'precomputed_root': '/data/crfm-helm-public', + 'precomputed_root': None, # helm-run perf knobs: 'num_threads': 1, + 'local_path': 'prod_env', } + @property + def command(self) -> str: + """ + Render CLI args while omitting unset optional values. + + This keeps kwdagger-facing params in a clean key/value style and avoids + emitting placeholders such as ``--foo=None`` or bare flags for empty + list defaults. + """ + filtered_config = {} + for key, value in self.final_config.items(): + if value is None: + continue + if isinstance(value, list) and len(value) == 0: + continue + filtered_config[key] = value + parts = [] + for key, value in filtered_config.items(): + if isinstance(value, dict): + from kwutil.util_yaml import Yaml + value_text = shlex.quote(Yaml.dumps(value)) + if '\n' in value_text and value_text[0] == "'": + value_text = "'\n" + value_text[1:] + parts.append(f' --{key}={value_text} \\') + else: + value_text = shlex.quote(str(value)) + parts.append(f' --{key}={value_text} \\') + argstr = '\n'.join(parts).lstrip().rstrip('\\') + if argstr: + return self.executable + ' \\\n ' + argstr + return self.executable + # Optional: You can define load_result if you want kwdagger aggregate to read # something out of this node. Usually this node is an “adapter/materializer” # and a downstream eval node would implement load_result instead. diff --git a/magnet/utils/sankey.py b/magnet/utils/sankey.py index 118b86d..705b99f 100644 --- a/magnet/utils/sankey.py +++ b/magnet/utils/sankey.py @@ -8,6 +8,7 @@ from __future__ import annotations from dataclasses import dataclass, field + from typing import Any, Callable, Dict, Iterable, List, Optional, Union import typing @@ -29,6 +30,34 @@ def update_layout( @dataclass(frozen=True) class Root: label: str + def builder(self) -> PlanBuilder: + return PlanBuilder(Plan(self)) + def group(self, *, by: Grouper, name: Optional[str] = None) -> PlanBuilder: + return self.builder().group(by=by, name=name) + def split(self, *, by: Grouper, name: Optional[str] = None) -> SplitBuilder: + return self.builder().split(by=by, name=name) + + +# A terminal label convenience (primarily for branch specs) +@dataclass(frozen=True) +class Node: + label: str + + +@dataclass(frozen=True) +class Branch: + """ + A Split branch: + - label: optional override for the *split outcome node label* + - plan: plan to continue after the split outcome node + - terminal: if True, stop at the outcome node (ignore plan) + """ + plan: "Plan" = field(default_factory=lambda: Plan()) + label: Optional[str] = None + terminal: bool = False + + +BranchLike = Union["Plan", Branch, Node, str] # str => terminal label @dataclass(frozen=True) @@ -47,12 +76,103 @@ class Bucket: class Split: name: str by: Grouper - branches: Dict[Any, 'Plan'] - default: Optional['Plan'] = None + branches: Dict[Any, BranchLike] + default: Optional[BranchLike] = None + + def __post_init__(self): + def coerce(x: BranchLike) -> Branch: + if isinstance(x, Branch): + return x + if isinstance(x, Node): + return Branch(label=x.label, terminal=True) + if isinstance(x, str): + return Branch(label=x, terminal=True) + if isinstance(x, Plan): + return Branch(plan=x) + raise TypeError(f"Invalid branch spec: {type(x)}") + + # frozen dataclass: use object.__setattr__ to normalize + object.__setattr__( + self, "branches", {k: coerce(v) for k, v in self.branches.items()} + ) + if self.default is not None: + object.__setattr__(self, "default", coerce(self.default)) + + +class PlanBuilder: + def __init__(self, plan: Optional[Plan] = None): + self.plan = plan or Plan() + + def build(self) -> Plan: + return self.plan + + def group(self, *, by: Grouper, name: Optional[str] = None) -> "PlanBuilder": + if name is None: + name = by if isinstance(by, str) else "group" + self.plan.steps.append(Group(name, by)) + return self + + def bucket(self, *, by: Grouper, name: Optional[str] = None) -> "PlanBuilder": + if name is None: + name = by if isinstance(by, str) else "bucket" + self.plan.steps.append(Bucket(name, by)) + return self + + def split(self, *, by: Grouper, name: Optional[str] = None) -> "SplitBuilder": + if name is None: + name = by if isinstance(by, str) else "split" + sp = Split(name, by, branches={}) + self.plan.steps.append(sp) + return SplitBuilder(sp) + + +class CaseBuilder(PlanBuilder): + def __init__(self, branch_ref: Dict[Any, Branch], key: Any): + # branch_ref is the Split.branches dict (already normalized to Branch) + self._branch_ref = branch_ref + self._key = key + super().__init__(plan=self._branch_ref[key].plan) + def set_label(self, label: str) -> "CaseBuilder": + br = self._branch_ref[self._key] + self._branch_ref[self._key] = Branch(plan=br.plan, label=label, terminal=br.terminal) + return self + + def terminal(self, label: Optional[str] = None) -> "CaseBuilder": + br = self._branch_ref[self._key] + self._branch_ref[self._key] = Branch( + plan=br.plan, + label=(label if label is not None else br.label), + terminal=True, + ) + return self + + +class SplitBuilder: + def __init__(self, split: Split): + self.split = split + # Split.__post_init__ normalized branches to Branch already + self.cases: Dict[Any, CaseBuilder] = {} -def _eval_by(by: Grouper, row: Row): - return by(row) if callable(by) else row.get(by) + def add_case(self, *, value: Any) -> CaseBuilder: + if value not in self.split.branches: + # create an empty branch by default + self.split.branches[value] = Branch(plan=Plan()) + cb = CaseBuilder(self.split.branches, value) + self.cases[value] = cb + return cb + + def __getitem__(self, value: Any) -> CaseBuilder: + # defaultdict-ish + return self.add_case(value=value) + + def set_default(self) -> CaseBuilder: + if self.split.default is None: + # store default as a Branch (not in branches dict) + object.__setattr__(self.split, "default", Branch(plan=Plan())) + # expose via a pseudo key + # (or provide a dedicated DefaultCaseBuilder) + raise NotImplementedError("Implement a DefaultCaseBuilder if you want this.") def _by_repr(by: Grouper) -> str: @@ -64,10 +184,6 @@ def _by_repr(by: Grouper) -> str: return f'' -def _label(stage: str, value: Any, fmt: str): - return fmt.format(name=stage, value=value) - - @dataclass class Plan: """ @@ -128,15 +244,19 @@ def rec(plan: 'Plan', indent: str = ''): f'{indent}BUCKET {st.name!r} by={_by_repr(st.by)}' ) elif isinstance(st, Split): - lines.append( - f'{indent}SPLIT {st.name!r} by={_by_repr(st.by)}' - ) - for k, sub in st.branches.items(): - lines.append(f'{indent} BRANCH {k!r}:') - rec(sub, indent + ' ') + lines.append(f"{indent}SPLIT {st.name!r} by={_by_repr(st.by)}") + for k, br in st.branches.items(): + extra = [] + if br.label is not None: + extra.append(f"label={br.label!r}") + if br.terminal: + extra.append("terminal=True") + extra_s = (" [" + ", ".join(extra) + "]") if extra else "" + lines.append(f"{indent} BRANCH {k!r}:{extra_s}") + rec(br.plan, indent + " ") if st.default is not None: - lines.append(f'{indent} DEFAULT:') - rec(st.default, indent + ' ') + # same idea for default + ... else: lines.append(f'{indent}{type(st).__name__} (?)') @@ -149,6 +269,7 @@ def trace(self, row: Row, *, label_fmt='{name}: {value}') -> List[str]: Example: >>> # Trace the path a row takes through the plan + >>> from magnet.utils import sankey >>> plan = Plan( ... Root("ROOT_NODE"), ... Group("dataset", "dataset"), @@ -170,6 +291,9 @@ def trace(self, row: Row, *, label_fmt='{name}: {value}') -> List[str]: root = self._find_root_label(default='ROOT_NODE') path = [root] + def _eval_by(by: Grouper, row: Row): + return by(row) if callable(by) else row.get(by) + def run(plan: 'Plan', cur: str): node = cur for st in plan.steps: @@ -177,18 +301,30 @@ def run(plan: 'Plan', cur: str): continue elif isinstance(st, (Group, Bucket)): val = _eval_by(st.by, row) - nxt = _label(st.name, val, label_fmt) + nxt = label_fmt.format(name=st.name, value=val) path.append(nxt) node = nxt elif isinstance(st, Split): key = _eval_by(st.by, row) - split_node = _label(st.name, key, label_fmt) - path.append(split_node) - node = split_node - branch = st.branches.get(key) or st.default - if branch is None: + + br = st.branches.get(key) or st.default + if br is None: + return node + + # Decide the split outcome node label + split_label = ( + br.label + if (br.label is not None) + else label_fmt.format(name=st.name, value=key) + ) + path.append(split_label) + node = split_label + + # Terminal branch stops here + if br.terminal: return node - node = run(branch, node) + + node = run(br.plan, node) else: raise TypeError(f'Unknown step type: {type(st)}') return node @@ -453,3 +589,70 @@ def to_plotly(self, *, title: str = 'Sankey') -> PlotlyFigureLike: ) fig.update_layout(title_text=title, font_size=14) return fig + + +def demo(): + """ + """ + import kwutil + import ubelt as ub + rows = kwutil.Yaml.loads(ub.codeblock( + ''' + - {run: run1, suite: suite1, retcode: 1, retmsg: 'unsupported', spec_diagnosis: null, metric_iou: null} + - {run: run2, suite: suite1, retcode: 1, retmsg: 'unsupported', spec_diagnosis: null, metric_iou: null} + - {run: run3, suite: suite1, retcode: 1, retmsg: 'unsupported', spec_diagnosis: null, metric_iou: null} + - {run: run4, suite: suite1, retcode: 1, retmsg: 'unsupported', spec_diagnosis: null, metric_iou: null} + - {run: run5, suite: suite1, retcode: 1, retmsg: 'unsupported', spec_diagnosis: null, metric_iou: null} + + - {run: run1, suite: suite2, retcode: 0, retmsg: '', spec_diagnosis: 'agree', metric_iou: 1.0} + - {run: run2, suite: suite2, retcode: 0, retmsg: '', spec_diagnosis: 'agree', metric_iou: 1.0} + - {run: run3, suite: suite2, retcode: 0, retmsg: '', spec_diagnosis: 'agree', metric_iou: 1.0} + - {run: run4, suite: suite2, retcode: 0, retmsg: '', spec_diagnosis: 'agree', metric_iou: 1.0} + - {run: run5, suite: suite2, retcode: 0, retmsg: '', spec_diagnosis: 'agree', metric_iou: 1.0} + + - {run: run1, suite: suite3, retcode: 0, retmsg: '', spec_diagnosis: 'agree', metric_iou: 0.9} + - {run: run2, suite: suite3, retcode: 0, retmsg: '', spec_diagnosis: 'agree', metric_iou: 0.7} + - {run: run3, suite: suite3, retcode: 0, retmsg: '', spec_diagnosis: 'agree', metric_iou: 0.9} + - {run: run4, suite: suite3, retcode: 0, retmsg: '', spec_diagnosis: 'agree', metric_iou: 1.0} + - {run: run5, suite: suite3, retcode: 1, retmsg: 'oom', spec_diagnosis: null, metric_iou: null} + - {run: run6, suite: suite3, retcode: 1, retmsg: 'oom', spec_diagnosis: null, metric_iou: null} + + - {run: run1, suite: suite4, retcode: 0, retmsg: '', spec_diagnosis: 'disagree-deploy', metric_iou: 0.0} + - {run: run2, suite: suite4, retcode: 0, retmsg: '', spec_diagnosis: 'disagree-deploy', metric_iou: 0.0} + - {run: run3, suite: suite4, retcode: 0, retmsg: '', spec_diagnosis: 'disagree-input', metric_iou: 0.3} + - {run: run4, suite: suite4, retcode: 0, retmsg: '', spec_diagnosis: 'disagree-input', metric_iou: 0.1} + ''')) + + from magnet.utils import sankey_builder + root = sankey_builder.Root(label="All Attempts") + bench = root.group(by="suite", name="benchmark") + + splits = bench.group(by="retcode", name="Ran") + splits[1].label = 'Failed' + splits[0].label = 'Ran' + + # Do something unique on the fail branch + splits[1].group(by='retmsg') + + def iou_grouper(row): + value = row['metric_iou'] + if value == 1: + return '1' + elif value > 0.5: + return '0.5 - 1' + elif value > 0: + return '0 - 0.5' + else: + return '0' + + diagnosis_buckets = splits[0].group(by="agreement") + iou_groups = diagnosis_buckets.group(by=iou_grouper) + + # Connect only some of the underlying buckets + iou_groups['0.5 - 1'].connect('Priority Analysis') + iou_groups['0 - 0.5'].connect('Priority Analysis') + + graph = root.build_sankey(rows) + print(root.to_text()) + import networkx as nx + nx.write_network_text(graph) diff --git a/magnet/utils/sankey_builder.py b/magnet/utils/sankey_builder.py new file mode 100644 index 0000000..a55421d --- /dev/null +++ b/magnet/utils/sankey_builder.py @@ -0,0 +1,711 @@ +""" +sankey_builder.py + +Fluent Sankey spec builder (mutable DAG), focused on: +- Nodes: Root / Constant / Group +- Group becomes a "split" when cases are configured +- connect() coerces strings to Constant nodes +- trace(row) for debugging +- trace_batch(rows) for fast aggregated execution +- build_sankey(rows) to produce a networkx graph with node+edge flow totals +- to_text() for readable structural representation + +Semantics: +- Each row follows ONE path through the spec (row -> labels). +- The sankey graph is the union of these paths (fan-out and fan-in supported). +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Callable, Dict, Iterable, Iterator, List, Optional, Sequence, Tuple, Union + +import itertools +import networkx as nx +import typing + +Row = Dict[str, Any] +By = Union[str, Callable[[Row], Any]] + +if typing.TYPE_CHECKING: + from typing import Protocol + + Row = Dict[str, Any] + Grouper = Union[str, Callable[[Row], Any]] + + class PlotlyFigureLike(Protocol): + def write_image(self, *args, **kwargs): ... + def update_layout( + self, dict1=None, overwrite=False, **kwargs + ) -> 'PlotlyFigureLike': ... + + +def _eval(by: By, row: Row) -> Any: + return by(row) if callable(by) else row.get(by) + + +def _by_repr(by: By) -> str: + if isinstance(by, str): + return repr(by) + name = getattr(by, "__name__", None) + if name: + return f"" + return f"" + + +def _coerce_target(target: Union["Node", str, None]) -> Optional["Node"]: + """ + Coerce a connect() target: + - Node -> itself + - str -> new Constant(label=str) + - None -> None (terminal) + """ + if target is None: + return None + if isinstance(target, Node): + return target + if isinstance(target, str): + return Constant(label=target) + raise TypeError(f"Cannot connect to target of type {type(target)}") + + +BatchItem = Tuple[Row, float] +NodeFlow = Dict[str, float] +EdgeFlow = Dict[Tuple[str, str], float] + + +@dataclass +class BatchTraceResult: + node_flow: NodeFlow + edge_flow: EdgeFlow + + +@dataclass +class Case: + """ + Case config for a parent Group. + + - label: override emitted label for that key + - then: override continuation for that key + """ + label: Optional[str] = None + then: Optional["Node"] = None + + def group(self, *, by: By, name: Optional[str] = None) -> "Group": + g = Group(name=name or (by if isinstance(by, str) else "group"), by=by) + if self.then is not None: + raise RuntimeError("This case already has a `.then` continuation") + self.then = g + return g + + def connect(self, target: Union["Node", str, None]) -> Optional["Node"]: + self.then = _coerce_target(target) + return self.then + + +@dataclass +class CaseSet: + """ + Handle for bulk operations on multiple cases. + """ + cases: List[Case] + + def connect(self, target: Union["Node", str, None]) -> Optional["Node"]: + # If target is a string, create ONE Constant and share it. + tgt = _coerce_target(target) + for c in self.cases: + c.then = tgt + return tgt + + def set_label(self, label: str) -> "CaseSet": + for c in self.cases: + c.label = label + return self + + +@dataclass +class Node: + """ + Base node in the spec. + + Every Node: + - emits exactly one label per row + - chooses a continuation Node (or terminates) + + `next` is the fallback continuation. + """ + next: Optional["Node"] = None + + # ---- fluent building ---- + + def group(self, *, by: By, name: Optional[str] = None) -> "Group": + g = Group(name=name or (by if isinstance(by, str) else "group"), by=by) + if self.next is not None: + raise RuntimeError("This node already has a `.next` continuation") + self.next = g + return g + + def connect(self, target: Union["Node", str, None]) -> Optional["Node"]: + self.next = _coerce_target(target) + return self.next + + # ---- debug execution ---- + + def trace(self, row: Row, *, label_fmt: str = "{name}: {value}", max_steps: int = 10_000) -> List[str]: + """ + Debug trace: realize the label path for a single row. + """ + labels: List[str] = [] + cur: Optional[Node] = self + visited: set[int] = set() + steps = 0 + while cur is not None: + steps += 1 + if steps > max_steps: + raise RuntimeError(f"Exceeded max_steps={max_steps}; possible cycle") + oid = id(cur) + if oid in visited: + raise RuntimeError("Cycle detected while tracing (node revisited)") + visited.add(oid) + + label, cur = cur._step_row(row, label_fmt=label_fmt) + labels.append(label) + return labels + + # ---- fast batch execution ---- + + def trace_batch( + self, + rows: Iterable[Row], + *, + weight: Union[float, Callable[[Row], float]] = 1.0, + label_fmt: str = "{name}: {value}", + max_depth: int = 10_000, + ) -> BatchTraceResult: + """ + Fast batch tracing: aggregate node/edge flows without tracing each row into a full path. + + Returns: + BatchTraceResult(node_flow, edge_flow) + """ + weight_fn = weight if callable(weight) else (lambda r: float(weight)) + batch: List[BatchItem] = [(r, float(weight_fn(r))) for r in rows] + + node_flow: NodeFlow = {} + edge_flow: EdgeFlow = {} + + def bump_node(label: str, wsum: float) -> None: + node_flow[label] = node_flow.get(label, 0.0) + wsum + + def bump_edge(u: str, v: str, wsum: float) -> None: + edge_flow[(u, v)] = edge_flow.get((u, v), 0.0) + wsum + + def walk(node: Node, in_batch: List[BatchItem], parent_label: Optional[str], stack: List[int]) -> None: + if not in_batch: + return + if len(stack) > max_depth: + raise RuntimeError(f"Exceeded max_depth={max_depth}; possible cycle") + + oid = id(node) + if oid in stack: + raise RuntimeError("Cycle detected in spec while batch-tracing") + stack.append(oid) + try: + for label, out_batch, wsum, nxt in node._step_batch(in_batch, label_fmt=label_fmt): + bump_node(label, wsum) + if parent_label is not None: + bump_edge(parent_label, label, wsum) + if nxt is not None: + walk(nxt, out_batch, label, stack) + finally: + stack.pop() + + walk(self, batch, parent_label=None, stack=[]) + return BatchTraceResult(node_flow=node_flow, edge_flow=edge_flow) + + def build_sankey( + self, + rows: Iterable[Row], + *, + weight: Union[float, Callable[[Row], float]] = 1.0, + edge_attr: str = "value", + node_attr: str = "count", + label_fmt: str = "{name}: {value}", + ) -> SankeyDiGraph: + """ + Build a SankeyDiGraph using batch tracing. + """ + res = self.trace_batch(rows, weight=weight, label_fmt=label_fmt) + G = SankeyDiGraph(edge_attr=edge_attr, node_attr=node_attr) + + for n, c in res.node_flow.items(): + G.add_node(n, **{node_attr: c}) + for (u, v), val in res.edge_flow.items(): + # ensure nodes exist even if someone mutates later + if not G.has_node(u): + G.add_node(u, **{node_attr: 0.0}) + if not G.has_node(v): + G.add_node(v, **{node_attr: 0.0}) + G.add_edge(u, v, **{edge_attr: val}) + return G + + # ---- structure printing ---- + + def to_text(self) -> str: + dumper = _TextDumper() + dumper.rec(self, "") + return "\n".join(dumper.lines) + + # ---- subclass hooks ---- + + def _step_row(self, row: Row, *, label_fmt: str) -> Tuple[str, Optional["Node"]]: + raise NotImplementedError + + def _step_batch( + self, + batch: List[BatchItem], + *, + label_fmt: str, + ) -> Iterator[Tuple[str, List[BatchItem], float, Optional["Node"]]]: + """ + Yield (label, out_batch, weight_sum, next_node) for each outgoing route. + """ + raise NotImplementedError + + def _text_head(self, nid: int) -> str: + return f"{type(self).__name__}#{nid}" + + def _text_body(self, dumper: "_TextDumper", indent: str, nid: int) -> None: + if self.next is not None: + # dumper.lines.append(f"{indent} NEXT:") + dumper.rec(self.next, indent + " ") + + +@dataclass +class Constant(Node): + label: str = "CONST" + + def _step_row(self, row: Row, *, label_fmt: str) -> Tuple[str, Optional[Node]]: + return self.label, self.next + + def _step_batch( + self, + batch: List[BatchItem], + *, + label_fmt: str, + ) -> Iterator[Tuple[str, List[BatchItem], float, Optional[Node]]]: + wsum = sum(w for _, w in batch) + yield (self.label, batch, wsum, self.next) + + def _text_head(self, nid: int) -> str: + return f"CONST#{nid} {self.label!r}" + + +@dataclass +class Root(Constant): + def _text_head(self, nid: int) -> str: + return f"ROOT#{nid} {self.label!r}" + + +@dataclass +class Group(Node): + """ + A group computes value = by(row), emits a label, and optionally routes based on that value. + + - cases[value] can override: + - label (Case.label) + - next node (Case.then) + - fallback continuation is Group.next + """ + name: str = "group" + by: By = "" + cases: Dict[Any, Case] = field(default_factory=dict) + default: Optional[Case] = None + + def __getitem__(self, key: Union[Any, Sequence[Any]]) -> Union[Case, CaseSet]: + if isinstance(key, (list, tuple, set)): + vals = list(key) + return CaseSet([self.cases.setdefault(v, Case()) for v in vals]) + return self.cases.setdefault(key, Case()) + + def set_default(self) -> Case: + if self.default is None: + self.default = Case() + return self.default + + def _label_for(self, value: Any, case: Optional[Case], *, label_fmt: str) -> str: + if case is not None and case.label is not None: + return case.label + return label_fmt.format(name=self.name, value=value) + + def _next_for(self, case: Optional[Case]) -> Optional[Node]: + if case is not None and case.then is not None: + return case.then + return self.next + + def _step_row(self, row: Row, *, label_fmt: str) -> Tuple[str, Optional[Node]]: + v = _eval(self.by, row) + case = self.cases.get(v) or self.default + label = self._label_for(v, case, label_fmt=label_fmt) + nxt = self._next_for(case) + return label, nxt + + def _step_batch( + self, + batch: List[BatchItem], + *, + label_fmt: str, + ) -> Iterator[Tuple[str, List[BatchItem], float, Optional[Node]]]: + # Partition incoming batch by computed value + groups: Dict[Any, List[BatchItem]] = {} + sums: Dict[Any, float] = {} + for r, w in batch: + v = _eval(self.by, r) + groups.setdefault(v, []).append((r, w)) + sums[v] = sums.get(v, 0.0) + w + + for v, out_batch in groups.items(): + case = self.cases.get(v) or self.default + label = self._label_for(v, case, label_fmt=label_fmt) + nxt = self._next_for(case) + yield (label, out_batch, sums[v], nxt) + + def _text_head(self, nid: int) -> str: + return f"GROUP#{nid} {self.name!r} by={_by_repr(self.by)}" + + def _text_body(self, dumper: "_TextDumper", indent: str, nid: int) -> None: + has_branches = bool(self.cases) or (self.default is not None) + # Cases first + if self.cases: + for k in sorted(self.cases.keys(), key=lambda x: (str(type(x)), repr(x))): + c = self.cases[k] + extra = [] + if c.label is not None: + extra.append(f"label={c.label!r}") + hdr = f"{indent} CASE {k!r}" + if extra: + hdr += " [" + ", ".join(extra) + "]" + dumper.lines.append(hdr + ":") + if c.then is not None: + dumper.rec(c.then, indent + " ") + else: + dumper.lines.append(f"{indent} (fallthrough to NEXT)") + + if self.default is not None: + c = self.default + extra = [] + if c.label is not None: + extra.append(f"label={c.label!r}") + hdr = f"{indent} DEFAULT" + if extra: + hdr += " [" + ", ".join(extra) + "]" + dumper.lines.append(hdr + ":") + if c.then is not None: + dumper.rec(c.then, indent + " ") + else: + dumper.lines.append(f"{indent} (fallthrough to NEXT)") + + # Fallback continuation + if self.next is not None: + if has_branches: + dumper.lines.append(f"{indent} ELSE:") + dumper.rec(self.next, indent + " ") + else: + dumper.rec(self.next, indent + " ") + + +class _TextDumper: + """ + DAG-aware dumper used by Node.to_text(). + """ + def __init__(self) -> None: + self.lines: List[str] = [] + self._ids: Dict[int, int] = {} + self._expanded: set[int] = set() + self._counter = itertools.count(1) + + def _get_id(self, obj: object) -> int: + oid = id(obj) + if oid not in self._ids: + self._ids[oid] = next(self._counter) + return self._ids[oid] + + def rec(self, node: Node, indent: str) -> None: + nid = self._get_id(node) + head = node._text_head(nid) + self.lines.append(f"{indent}{head}") + + oid = id(node) + if oid in self._expanded: + self.lines[-1] += " (ref)" + return + self._expanded.add(oid) + + node._text_body(self, indent, nid) + + +class SankeyDiGraph(nx.DiGraph): + """ + A DiGraph with convenience methods for Sankey rendering / exporting. + + Notes: + - Flow is stored on edges in `edge_attr` (default: "value") + - Plotly node ordering is topological when possible, otherwise insertion order. + """ + + def __init__(self, *args, edge_attr: str = 'value', **kwargs): + super().__init__(*args, **kwargs) + self.edge_attr = edge_attr + + @classmethod + def demo(cls, n=200, seed=0) -> SankeyDiGraph: + """ + Demodata for tests + """ + import random + + r = random.Random(seed) + + rows = [ + dict( + dataset=r.choice(['coco', 'openimages', 'cityscapes']), + backend=r.choice(['cuda', 'cpu']), + status=('fail' if r.random() < 0.15 else 'ok'), + ) + for _ in range(n) + ] + for row in rows: + row['reason'] = ( + r.choice(['oom', 'timeout']) + if row['status'] == 'fail' + else None + ) + + plan = Plan( + Root('All Runs'), + Group('dataset', 'dataset'), + Split( + 'status', + 'status', + branches={ + 'ok': Plan(Group('backend', 'backend')), + 'fail': Plan( + Bucket('reason', 'reason'), Group('backend', 'backend') + ), + }, + ), + ) + self = plan.build_sankey(rows) + return self + + # ---- light reporting helpers (optional, but nice) ---- + + def summarize( + self, + *, + edge_attr: Optional[str] = None, + max_edges: Optional[int] = 200, + sort: str = 'value_desc', + ) -> str: + """ + Like Plan.graph_to_text, but bound to the graph. + + Example: + >>> # xdoctest: +REQUIRES(module:plotly) + >>> import plotly + >>> from magnet.utils.sankey import * # NOQA + >>> self = SankeyDiGraph.demo() + >>> print(self.summarize()) + Nodes: 10 Edges: 17 + ... + Top nodes by outflow/inflow: + All Runs out=200 in=0 + status: ok out=171 in=171 + dataset: cityscapes out=71 in=71 + ... + Edges: + status: ok -> backend: cuda value=94 + status: ok -> backend: cpu value=77 + All Runs -> dataset: cityscapes value=71 + ... + """ + edge_attr = edge_attr or self.edge_attr + lines: List[str] = [] + lines.append( + f'Nodes: {self.number_of_nodes()} Edges: {self.number_of_edges()}' + ) + lines.append('') + + def outflow(n): + return sum(self[n][v].get(edge_attr, 0) for v in self.successors(n)) + + def inflow(n): + return sum( + self[u][n].get(edge_attr, 0) for u in self.predecessors(n) + ) + + nodes_sorted = sorted( + self.nodes, key=lambda n: (outflow(n), inflow(n)), reverse=True + ) + lines.append('Top nodes by outflow/inflow:') + for n in nodes_sorted[:20]: + lines.append(f' {n} out={outflow(n):g} in={inflow(n):g}') + lines.append('') + + edges = [(u, v, self[u][v].get(edge_attr, 0)) for u, v in self.edges] + if sort == 'value_desc': + edges.sort(key=lambda t: t[2], reverse=True) + elif sort == 'lex': + edges.sort(key=lambda t: (str(t[0]), str(t[1]))) + + lines.append('Edges:') + shown = edges if max_edges is None else edges[:max_edges] + for u, v, val in shown: + lines.append(f' {u} -> {v} {edge_attr}={val:g}') + if max_edges is not None and len(edges) > max_edges: + lines.append(f'... ({len(edges) - max_edges} more edges)') + return '\n'.join(lines) + + # ---- core conversions ---- + + def _to_sankey_data( + self, + ) -> tuple[List[Any], List[int], List[int], List[float]]: + """ + Convert into (nodes, source, target, value) for Plotly Sankey. + + Example: + >>> # Convert nx graph to plotly sankey arrays + >>> from magnet.utils.sankey import * # NOQA + >>> import networkx as nx + >>> G = SankeyDiGraph() + >>> G.add_edge("A", "B", value=2) + >>> G.add_edge("A", "C", value=3) + >>> nodes, source, target, value = G._to_sankey_data() + >>> set(nodes) == {"A", "B", "C"} + True + >>> len(source) == len(target) == len(value) == 2 + True + >>> sorted(value) + [2.0, 3.0] + """ + try: + nodes = list(nx.topological_sort(self)) + except nx.NetworkXUnfeasible: + nodes = list(self.nodes) + + idx = {n: i for i, n in enumerate(nodes)} + source: List[int] = [] + target: List[int] = [] + value: List[float] = [] + + for u, v, data in self.edges(data=True): + source.append(idx[u]) + target.append(idx[v]) + value.append(float(data.get(self.edge_attr, 0))) + + node_labels = [self.nodes[n].get('label', n) for n in nodes] + + return node_labels, source, target, value + + def to_plotly(self, *, title: str = 'Sankey') -> PlotlyFigureLike: + """ + Build a publishable Plotly Sankey figure. + + Example: + >>> # xdoctest: +REQUIRES(module:plotly) + >>> import plotly + >>> from magnet.utils.sankey import * # NOQA + >>> G = SankeyDiGraph.demo(n=20) + >>> fig = G.to_plotly(title='Demo') + >>> assert fig.layout.title.text == 'Demo' + >>> # xdoctest: +REQUIRES(module:kaleido) + >>> # xdoctest: +REQUIRES(module:kwplot) + >>> # xdoctest: +REQUIRES(--show) + >>> import kwplot + >>> kwplot.autompl() + >>> import tempfile + >>> import os + >>> with tempfile.TemporaryDirectory() as d: + ... fpath = os.path.join(d, "sankey_demo.png") + ... fig.write_image(fpath, scale=1) + ... assert os.path.exists(fpath) + ... kwplot.imshow(fpath) + """ + import plotly.graph_objects as go + + nodes, source, target, value = self._to_sankey_data() + sankey = go.Sankey( + node=dict(label=nodes, pad=15, thickness=18), + link=dict(source=source, target=target, value=value), + ) + fig = go.Figure(sankey) + fig.update_layout(title_text=title, font_size=14) + return fig + + +def demo(): + """ + """ + import kwutil + import ubelt as ub + rows = kwutil.Yaml.loads(ub.codeblock( + ''' + - {run: run1, suite: suite1, retcode: 1, retmsg: 'unsupported', spec_diagnosis: null, metric_iou: null} + - {run: run2, suite: suite1, retcode: 1, retmsg: 'unsupported', spec_diagnosis: null, metric_iou: null} + - {run: run3, suite: suite1, retcode: 1, retmsg: 'unsupported', spec_diagnosis: null, metric_iou: null} + - {run: run4, suite: suite1, retcode: 1, retmsg: 'unsupported', spec_diagnosis: null, metric_iou: null} + - {run: run5, suite: suite1, retcode: 1, retmsg: 'unsupported', spec_diagnosis: null, metric_iou: null} + + - {run: run1, suite: suite2, retcode: 0, retmsg: '', spec_diagnosis: 'agree', metric_iou: 1.0} + - {run: run2, suite: suite2, retcode: 0, retmsg: '', spec_diagnosis: 'agree', metric_iou: 1.0} + - {run: run3, suite: suite2, retcode: 0, retmsg: '', spec_diagnosis: 'agree', metric_iou: 1.0} + - {run: run4, suite: suite2, retcode: 0, retmsg: '', spec_diagnosis: 'agree', metric_iou: 1.0} + - {run: run5, suite: suite2, retcode: 0, retmsg: '', spec_diagnosis: 'agree', metric_iou: 1.0} + + - {run: run1, suite: suite3, retcode: 0, retmsg: '', spec_diagnosis: 'agree', metric_iou: 0.9} + - {run: run2, suite: suite3, retcode: 0, retmsg: '', spec_diagnosis: 'agree', metric_iou: 0.7} + - {run: run3, suite: suite3, retcode: 0, retmsg: '', spec_diagnosis: 'agree', metric_iou: 0.9} + - {run: run4, suite: suite3, retcode: 0, retmsg: '', spec_diagnosis: 'agree', metric_iou: 1.0} + - {run: run5, suite: suite3, retcode: 1, retmsg: 'oom', spec_diagnosis: null, metric_iou: null} + - {run: run6, suite: suite3, retcode: 1, retmsg: 'oom', spec_diagnosis: null, metric_iou: null} + + - {run: run1, suite: suite4, retcode: 0, retmsg: '', spec_diagnosis: 'disagree-deploy', metric_iou: 0.0} + - {run: run2, suite: suite4, retcode: 0, retmsg: '', spec_diagnosis: 'disagree-deploy', metric_iou: 0.0} + - {run: run3, suite: suite4, retcode: 0, retmsg: '', spec_diagnosis: 'disagree-input', metric_iou: 0.3} + - {run: run4, suite: suite4, retcode: 0, retmsg: '', spec_diagnosis: 'disagree-input', metric_iou: 0.1} + ''')) + + from magnet.utils import sankey_builder + root = sankey_builder.Root(label="All Attempts") + bench = root.group(by="suite", name="benchmark") + + splits = bench.group(by="retcode", name="Ran") + splits[1].label = 'Failed' + splits[0].label = 'Ran' + + # Do something unique on the fail branch + splits[1].group(by='retmsg') + + def iou_grouper(row): + value = row['metric_iou'] + if value == 1: + return '1' + elif value > 0.5: + return '0.5 - 1' + elif value > 0: + return '0 - 0.5' + else: + return '0' + + diagnosis_buckets = splits[0].group(by="agreement") + iou_groups = diagnosis_buckets.group(by=iou_grouper) + + # Connect only some of the underlying buckets + iou_groups['0.5 - 1'].connect('Priority Analysis') + iou_groups['0 - 0.5'].connect('Priority Analysis') + + graph = root.build_sankey(rows) + print(root.to_text()) + import networkx as nx + nx.write_network_text(graph) diff --git a/tests/test_helm_output_loader.py b/tests/test_helm_output_loader.py new file mode 100644 index 0000000..5a083ed --- /dev/null +++ b/tests/test_helm_output_loader.py @@ -0,0 +1,18 @@ +def _create_small_helm_output(): + # Do we want a temp dir, or cache outputs for faster test rerun? + # import tempfile + # tempdir = tempfile.TemporaryDirectory(prefix='helm-output-test') + + import ubelt as ub + + dpath = ub.Path.appdir('magnet/tests/helm_output').ensuredir() + + res = ub.cmd( + 'helm-run --run-entries mmlu:subject=philosophy,model=openai/gpt2 --suite my-suite --max-eval-instances 1 --num-threads 1', + cwd=dpath, + verbose=3, + ) + res.check_returncode() + + res = ub.cmd('helm-summarize --suite my-suite', cwd=dpath, verbose=3) + res.check_returncode() diff --git a/tests/test_helm_run_diff_heavy.py b/tests/test_helm_run_diff_heavy.py new file mode 100644 index 0000000..83db77c --- /dev/null +++ b/tests/test_helm_run_diff_heavy.py @@ -0,0 +1,93 @@ +import json +import subprocess + +import kwutil +import pytest +import ubelt as ub + +from magnet.backends.helm.helm_outputs import HelmRun +from magnet.backends.helm.helm_run_diff import HelmRunDiff + + +def _coerce_demo_run() -> HelmRun: + """Best-effort demo run fetch; skip if environment cannot materialize it.""" + try: + return HelmRun.demo() + except subprocess.CalledProcessError as ex: + pytest.skip(f'Unable to materialize HelmRun.demo(): {ex!r}') + + +def test_helm_run_diff_heavy_demo_workflow(): + """Heavyweight regression test for end-to-end HelmRunDiff behavior.""" + run_a = _coerce_demo_run() + dpath = ub.Path.appdir('magnet/tests/helm/helm_run_diff_heavy').delete().ensuredir() + + # Case 1: identical copy + same_path = dpath / (run_a.path.name + '_same') + run_a.path.copy(same_path) + run_b = HelmRun(same_path) + rd = HelmRunDiff(run_a, run_b, a_name='orig', b_name='same') + info = rd.summary_dict(level=20) + assert info['run_spec_dict_ok'] is True + assert info['scenario_ok'] in {True, None} + assert info['value_agreement']['overall']['mismatched'] == 0 + assert info['value_agreement']['overall']['agree_ratio'] == 1.0 + assert info['dataset_overlap']['base_iou'] == 1.0 + assert info['diagnosis']['label'] in {'reproduced', 'core_match_bookkeeping_drift'} + json.dumps(info, allow_nan=False) + + # Case 2: perturb one run-level stat mean + stats_path = dpath / (run_a.path.name + '_statsmod') + run_a.path.copy(stats_path) + stat_fpath = stats_path / 'stats.json' + stats = kwutil.Json.loads(stat_fpath.read_text()) + old_mean = float(stats[0].get('mean', 0.0)) + stats[0]['mean'] = old_mean + 1.23 + stat_fpath.write_text(kwutil.Json.dumps(stats)) + + rd2 = HelmRunDiff(run_a, HelmRun(stats_path), a_name='orig', b_name='stats+1.23') + info2 = rd2.summary_dict(level=20) + assert info2['value_agreement']['overall']['mismatched'] >= 1 + assert info2['diagnosis']['label'] in { + 'core_metric_drift', + 'core_match_bookkeeping_drift', + 'reproduced', + } + json.dumps(info2, allow_nan=False) + + # Case 3: perturb one per-instance stat mean (if file exists) + inst_path = dpath / (run_a.path.name + '_perinstmod') + run_a.path.copy(inst_path) + pi_fpath = inst_path / 'per_instance_stats.json' + if pi_fpath.exists(): + perinst = kwutil.Json.loads(pi_fpath.read_text()) + ei, sj = 0, None + for j, stat in enumerate(perinst[ei]['stats']): + if int(stat.get('count', 0) or 0) and ('mean' in stat): + sj = j + break + assert sj is not None + old = float(perinst[ei]['stats'][sj]['mean']) + perinst[ei]['stats'][sj]['mean'] = old + 9.0 + pi_fpath.write_text(kwutil.Json.dumps(perinst)) + + rd_i = HelmRunDiff(run_a, HelmRun(inst_path), a_name='orig', b_name='perinst+9') + inst_info = rd_i.instance_summary_dict(top_n=5) + assert inst_info['means']['mismatched'] >= 1 + json.dumps(inst_info, allow_nan=False) + + # Case 4: run-spec deployment change + spec_path = dpath / (run_a.path.name + '_runspec_mod') + run_a.path.copy(spec_path) + spec_fpath = spec_path / 'run_spec.json' + run_spec = kwutil.Json.loads(spec_fpath.read_text()) + run_spec.setdefault('adapter_spec', {}) + run_spec['adapter_spec']['model_deployment'] = 'someotherdeploy/gpt2' + spec_fpath.write_text(kwutil.Json.dumps(run_spec)) + + rd4 = HelmRunDiff(run_a, HelmRun(spec_path), a_name='orig', b_name='runspec_mod') + info4 = rd4.summary_dict(level=20) + assert info4['run_spec_dict_ok'] is False + assert info4['run_spec_semantic']['deployment_changed'] is True + assert info4['diagnosis']['label'] in {'deployment_drift', 'execution_spec_drift'} + json.dumps(info4, allow_nan=False) diff --git a/tests/test_helm_run_diff_serializable.py b/tests/test_helm_run_diff_serializable.py new file mode 100644 index 0000000..dd29bf7 --- /dev/null +++ b/tests/test_helm_run_diff_serializable.py @@ -0,0 +1,178 @@ +import json +from typing import Any, Iterable + +from magnet.backends.helm.helm_run_analysis import HelmRunAnalysis +from magnet.backends.helm.helm_run_diff import HelmRunDiff +from magnet.backends.helm.helm_run_diff import dataset_overlap_from_request_states + + +class DummyJoined: + def __init__(self, rows: Iterable[dict[str, Any]]): + # rows should each have a 'key' field + self.row_by_key = {r['key']: r for r in rows} + + def __iter__(self): + return iter(self.row_by_key.values()) + + +class DummyRun: + def __init__(self, joined: DummyJoined): + self._joined = joined + + def joined_instance_stat_table(self, *, assert_assumptions: bool = False, short_hash: int = 0): + return self._joined + + +def _dummy_analysis(run_spec: dict[str, Any], stats: list[dict[str, Any]]): + ana = HelmRunAnalysis.__new__(HelmRunAnalysis) + ana._raw_cache = {} + ana._cache = {} + ana.run = None + ana.name = None + ana.run_spec = lambda: run_spec + ana.scenario = lambda: {'class_name': 'ToyScenario', 'output_path': '/tmp/a'} + ana.scenario_state = lambda: {'request_states': []} + ana.stats = lambda: stats + ana.joined_instance_stat_table = lambda *args, **kwargs: DummyJoined([]) + return ana + + +def test_instance_summary_serializable(): + """The output of ``instance_summary_dict`` must be JSON serializable. + + In particular the previously-used tuple keys were converted to a list of + objects with explicit ``metric_class``/``metric`` fields. Exercise the + computation path using minimal dummy data so we don't depend on HELM or + pandas. + """ + # build two joined tables that share a key but have different means + key = ("id", 0, None, "m", "split", None, None) + row_a = { + 'key': key, + 'stat': {'mean': 1.0, 'count': 1, 'name': {'name': 'm'}}, + 'request_state': {}, + } + row_b = { + 'key': key, + 'stat': {'mean': 2.0, 'count': 1, 'name': {'name': 'm'}}, + 'request_state': {}, + } + + # create bare HelmRunAnalysis instances and override their joiners + ana_a = HelmRunAnalysis.__new__(HelmRunAnalysis) + ana_b = HelmRunAnalysis.__new__(HelmRunAnalysis) + # give them the minimal attributes expected by other methods + for ana in (ana_a, ana_b): + ana._raw_cache = {} + ana._cache = {} + ana.run = None + ana.name = None + ana.scenario = lambda: {} + ana.run_spec = lambda: {} + ana.stats = lambda: [] + ana_a.joined_instance_stat_table = lambda *args, **kwargs: DummyJoined([row_a]) + ana_b.joined_instance_stat_table = lambda *args, **kwargs: DummyJoined([row_b]) + rd = HelmRunDiff(ana_a, ana_b) + info = rd.instance_summary_dict(top_n=1) + # also check that the higher-level summary_dict exposes the same list + sdict = rd.summary_dict(level=20) + iva = sdict.get('instance_value_agreement', {}) + assert isinstance(iva.get('top_mismatches_by_group'), list) + # strict JSON: no NaN / Infinity + json.dumps(sdict, allow_nan=False) + + +def test_dataset_overlap_json_serializable(): + """Pure request_state overlap helper should stay strictly JSON-compatible.""" + rs_a = [ + { + 'instance': {'id': 'id1', 'split': 'test', 'input': {'text': 'Q1'}}, + 'train_trial_index': 0, + 'request': {'prompt': 'P1'}, + 'result': {'completions': [{'text': 'A1'}]}, + }, + { + 'instance': { + 'id': 'id1', + 'split': 'test', + 'input': {'text': 'Q1'}, + 'perturbation': {'name': 'dialect', 'prob': 1.0}, + }, + 'train_trial_index': 0, + 'request': {'prompt': 'P1-d'}, + 'result': {'completions': [{'text': 'A1d'}]}, + }, + ] + rs_b = [ + { + 'instance': {'id': 'id1', 'split': 'test', 'input': {'text': 'Q1'}}, + 'train_trial_index': 0, + 'request': {'prompt': 'P1x'}, + 'result': {'completions': [{'text': 'A1'}]}, + }, + ] + info = dataset_overlap_from_request_states(rs_a, rs_b, max_examples=3) + assert info['base_coverage']['n_isect'] == 1 + assert info['variant_coverage']['only_a'] == 1 + assert info['content_equality']['prompt']['equal_ratio'] == 0.0 + json.dumps(info, allow_nan=False) + + +def test_run_spec_metric_order_semantic_and_deployment_reason_values(): + """Metric list order should not create semantic eval drift; deployment reason should include values.""" + stats = [ + { + 'name': {'name': 'exact_match', 'split': 'test'}, + 'count': 1, + 'mean': 1.0, + } + ] + spec_a = { + 'name': 'toy', + 'adapter_spec': {'model': 'm', 'model_deployment': 'dep/A'}, + 'metric_specs': [ + {'class_name': 'M0', 'args': {'x': 0}}, + {'class_name': 'M1', 'args': {'x': 1}}, + ], + } + spec_b_order_only = { + 'name': 'toy', + 'adapter_spec': {'model': 'm', 'model_deployment': 'dep/A'}, + 'metric_specs': [ + {'class_name': 'M1', 'args': {'x': 1}}, + {'class_name': 'M0', 'args': {'x': 0}}, + ], + } + rd_order = HelmRunDiff( + _dummy_analysis(spec_a, stats), + _dummy_analysis(spec_b_order_only, stats), + a_name='A', + b_name='B', + ) + info_order = rd_order.summary_dict(level=20) + assert info_order['run_spec_dict_ok'] is False + assert info_order['run_spec_semantic_dict_ok'] is True + assert info_order['run_spec_semantic']['metric_specs_multiset_delta']['equal_as_multiset'] is True + reason_names_order = [r['name'] for r in info_order['diagnosis']['reasons']] + assert 'evaluation_spec_drift' not in reason_names_order + json.dumps(info_order, allow_nan=False) + + spec_b_deploy = { + 'name': 'toy', + 'adapter_spec': {'model': 'm', 'model_deployment': 'dep/B'}, + 'metric_specs': spec_a['metric_specs'], + } + rd_dep = HelmRunDiff( + _dummy_analysis(spec_a, stats), + _dummy_analysis(spec_b_deploy, stats), + a_name='A', + b_name='B', + ) + info_dep = rd_dep.summary_dict(level=20) + reasons = {r['name']: r for r in info_dep['diagnosis']['reasons']} + assert 'deployment_drift' in reasons + assert reasons['deployment_drift']['details']['a_value'] == 'dep/A' + assert reasons['deployment_drift']['details']['b_value'] == 'dep/B' + exec_examples = info_dep['run_spec_semantic']['execution_value_examples'] + assert any(ex.get('path') == 'adapter_spec.model_deployment' for ex in exec_examples) + json.dumps(info_dep, allow_nan=False)