From 344d1582fe110a0f24958710a69a1cd849ee6624 Mon Sep 17 00:00:00 2001 From: Marcos Lohmann Date: Wed, 7 Oct 2026 14:47:13 -0300 Subject: [PATCH] update config and docs --- analysis/__init__.py | 0 analysis/metrics.py | 77 ++++++++++++++++++ harness/evaluate.py | 178 ++++++++++++++++++++++++++++++++++++++++++ infra/ds/collect.py | 79 +++++++++++++++++++ infra/ds/gpu_queue.py | 4 +- tests/test_metrics.py | 29 +++++++ 6 files changed, 366 insertions(+), 1 deletion(-) create mode 100644 analysis/__init__.py create mode 100644 analysis/metrics.py create mode 100644 harness/evaluate.py create mode 100644 infra/ds/collect.py create mode 100644 tests/test_metrics.py diff --git a/analysis/__init__.py b/analysis/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/analysis/metrics.py b/analysis/metrics.py new file mode 100644 index 0000000..65f8907 --- /dev/null +++ b/analysis/metrics.py @@ -0,0 +1,77 @@ +"""Métricas de qualidade e desempenho (tarefas 67–69).""" +from __future__ import annotations + +import math +from collections import Counter, defaultdict + +FALLBACK = "sem_correspondencia" + + +def percentile(values: list[float], q: float) -> float | None: + if not values: + return None + v = sorted(values) + k = (len(v) - 1) * q + lo, hi = math.floor(k), math.ceil(k) + return v[lo] + (v[hi] - v[lo]) * (k - lo) + + +def latency_summary(values: list[float]) -> dict: + return {"n": len(values), "p50": percentile(values, 0.5), "p95": percentile(values, 0.95), + "p99": percentile(values, 0.99), "mean": sum(values) / len(values) if values else None} + + +def classification(rows: list[dict]) -> dict: + """rows: dicts com `gold`, `pred` (rótulos). Ignora linhas com erro (pred None).""" + ok = [r for r in rows if r.get("pred") is not None] + if not ok: + return {"n": 0} + labels = sorted({r["gold"] for r in ok} | {r["pred"] for r in ok}) + tp, fp, fn = Counter(), Counter(), Counter() + for r in ok: + if r["pred"] == r["gold"]: + tp[r["gold"]] += 1 + else: + fp[r["pred"]] += 1 + fn[r["gold"]] += 1 + per = {} + for lab in labels: + p = tp[lab] / (tp[lab] + fp[lab]) if tp[lab] + fp[lab] else 0.0 + rc = tp[lab] / (tp[lab] + fn[lab]) if tp[lab] + fn[lab] else 0.0 + per[lab] = {"precision": p, "recall": rc, "f1": 2 * p * rc / (p + rc) if p + rc else 0.0, + "support": tp[lab] + fn[lab]} + gold_labels = [lab for lab in labels if per[lab]["support"] > 0] + return { + "n": len(ok), "errors": len(rows) - len(ok), + "accuracy": sum(r["pred"] == r["gold"] for r in ok) / len(ok), + "macro_f1": sum(per[lab]["f1"] for lab in gold_labels) / len(gold_labels), + "macro_precision": sum(per[lab]["precision"] for lab in gold_labels) / len(gold_labels), + "macro_recall": sum(per[lab]["recall"] for lab in gold_labels) / len(gold_labels), + "per_label": per, + } + + +def calibration(rows: list[dict], bins: int = 15) -> dict: + """rows: `probabilities` (dict), `gold`, `pred`. Brier multiclasse e ECE sobre a confiança max(p).""" + ok = [r for r in rows if r.get("probabilities")] + if not ok: + return {} + brier = sum(sum((p - (k == r["gold"])) ** 2 for k, p in r["probabilities"].items()) for r in ok) / len(ok) + nll = -sum(math.log(max(r["probabilities"].get(r["gold"], 0.0), 1e-12)) for r in ok) / len(ok) + buckets = defaultdict(list) + for r in ok: + conf = max(r["probabilities"].values()) + buckets[min(int(conf * bins), bins - 1)].append((conf, r["pred"] == r["gold"])) + ece = sum(len(b) / len(ok) * abs(sum(c for c, _ in b) / len(b) - sum(a for _, a in b) / len(b)) + for b in buckets.values()) + return {"brier": brier, "nll": nll, "ece": ece} + + +def selective(rows: list[dict], threshold: float) -> dict: + """Abstém (encaminha para desambiguação) quando max(p) < limiar.""" + ok = [r for r in rows if r.get("probabilities")] + kept = [r for r in ok if max(r["probabilities"].values()) >= threshold] + wrong_confident = sum(r["pred"] != r["gold"] for r in kept) + return {"threshold": threshold, "coverage": len(kept) / len(ok) if ok else None, + "selective_accuracy": (len(kept) - wrong_confident) / len(kept) if kept else None, + "confident_wrong_rate": wrong_confident / len(ok) if ok else None} diff --git a/harness/evaluate.py b/harness/evaluate.py new file mode 100644 index 0000000..1333b91 --- /dev/null +++ b/harness/evaluate.py @@ -0,0 +1,178 @@ +"""Executa um candidato sobre uma partição congelada (tarefas 52–57, 59, 63). + +- Confere o SHA-256 da partição contra data/splits/MANIFEST.json. +- Amostra determinística (seed) quando --limit é usado; ordem fixa por id. +- Mesma política de truncamento para todos (dmb.canonical.fit_state) e contagem de tokens por entrada. +- Lote 1, sem paralelismo; os primeiros --warmup exemplos ficam marcados e fora das estatísticas de latência. +- Mede tempo de carga, latência por decisão, pico de RAM (VmHWM) e de memória/uso de GPU (nvidia-smi). +- A partição de teste só roda com a configuração congelada (config/frozen.json, tarefa 62). + +Uso: python -m harness.evaluate --candidate laya --models-root ... --device cpu --out DIR +Parâmetros também por ambiente: DMB_SPLIT (dev), DMB_LIMIT (todos), DMB_SEED, DMB_WARMUP, DMB_STATE_BUDGET. +""" +import argparse +import hashlib +import json +import os +import random +import subprocess +import sys +import threading +import time +from pathlib import Path + +from dmb.offline import enforce_offline + +ROOT = Path(__file__).resolve().parents[1] +SPLITS = ROOT / "data" / "splits" +FROZEN = ROOT / "config" / "frozen.json" + + +class GpuSampler(threading.Thread): + """Amostra memória usada e utilização da GPU a cada 0,5 s (nvidia-smi).""" + + def __init__(self): + super().__init__(daemon=True) + self.samples, self.stop_flag = [], threading.Event() + + def run(self): + while not self.stop_flag.is_set(): + try: + out = subprocess.run(["nvidia-smi", "--query-gpu=memory.used,utilization.gpu", + "--format=csv,noheader,nounits"], capture_output=True, text=True, + timeout=5).stdout.strip().splitlines()[0] + mem, util = (float(x) for x in out.split(",")) + self.samples.append((mem, util)) + except Exception: + return + self.stop_flag.wait(0.5) + + def summary(self) -> dict: + if not self.samples: + return {} + return {"gpu_mem_peak_mib": max(m for m, _ in self.samples), + "gpu_util_mean_pct": sum(u for _, u in self.samples) / len(self.samples)} + + +def rss_peak_mib() -> float | None: + try: + for line in Path("/proc/self/status").read_text().splitlines(): + if line.startswith("VmHWM:"): + return int(line.split()[1]) / 1024 + except OSError: + return None + return None + + +def load_split(name: str) -> list[dict]: + manifest = json.loads((SPLITS / "MANIFEST.json").read_text()) + info = manifest["splits"][name] + path = SPLITS / info["file"] + if hashlib.sha256(path.read_bytes()).hexdigest() != info["sha256"]: + raise SystemExit(f"partição {name} diverge do MANIFEST (SHA-256)") + return [json.loads(l) for l in path.open()] + + +def model_precision(adapter) -> str | None: + for attr in ("model", "agent"): + obj = getattr(adapter, attr, None) + for cand in (obj, getattr(obj, "model", None)): + if cand is not None and hasattr(cand, "parameters"): + try: + return str(next(cand.parameters()).dtype).replace("torch.", "") + except StopIteration: + pass + return "gguf-q8_0" if adapter.name == "rizzo_flow" else None + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("--candidate", required=True) + ap.add_argument("--models-root", default="models") + ap.add_argument("--device", default="auto") + ap.add_argument("--out", required=True) + ap.add_argument("--split", default=os.environ.get("DMB_SPLIT", "dev")) + ap.add_argument("--limit", type=int, default=int(os.environ.get("DMB_LIMIT", "0")) or None) + ap.add_argument("--seed", type=int, default=int(os.environ.get("DMB_SEED", "20261007"))) + ap.add_argument("--warmup", type=int, default=int(os.environ.get("DMB_WARMUP", "5"))) + ap.add_argument("--state-budget", type=int, default=int(os.environ.get("DMB_STATE_BUDGET", "512"))) + args = ap.parse_args() + + if args.split == "test" and not FROZEN.is_file(): + raise SystemExit("partição de teste bloqueada: configuração ainda não congelada (tarefa 62)") + + enforce_offline() + from dmb.offline import self_test + guard = self_test() + if not guard["process_guard"].startswith("ok"): + raise SystemExit(f"bloqueio de rede inativo: {guard}") + + from analysis.metrics import calibration, classification, latency_summary + from dmb.canonical import to_jev_request + from dmb.jev import JevError + from dmb.server import systemone + from harness.candidates import build_adapter + + examples = sorted(load_split(args.split), key=lambda e: e["id"]) + if args.limit: + examples = sorted(random.Random(args.seed).sample(examples, min(args.limit, len(examples))), + key=lambda e: e["id"]) + random.Random(args.seed).shuffle(examples) # ordem fixa, mas sem agrupar por intenção + + gpu = GpuSampler() + gpu.start() + adapter = build_adapter(args.candidate, args.models_root, args.device) + t0 = time.perf_counter() + adapter.load() + load_s = time.perf_counter() - t0 + + out = Path(args.out) + out.mkdir(parents=True, exist_ok=True) + rows = [] + with (out / f"predictions-{args.candidate}-{args.device}-{args.split}.jsonl").open("w") as f: + for i, ex in enumerate(examples): + request, info = to_jev_request(ex, adapter.model_id, n_turns=0, budget=args.state_budget, + count_tokens=adapter.count_tokens) + row = {"id": ex["id"], "gold": ex["gold_label"], "warmup": i < args.warmup, + "state_tokens": info.get("tokens"), "truncated": info.get("truncated", False), + "n_options": len(ex["options"]), "meta": ex["metadata"]} + started = time.perf_counter() + try: + resp = systemone(adapter, request) + ans = resp["answers"][ex["task"]] + row.update(pred=ans["choice"], probabilities=ans["probabilities"], + confidence=ans.get("confidence"), input_tokens=resp["usage"]["input_tokens"], + latency_ms=resp["metadata"]["latency_ms"]) + except JevError as error: + row.update(pred=None, error=error.error_type, error_message=error.message[:200], + latency_ms=(time.perf_counter() - started) * 1000) + rows.append(row) + f.write(json.dumps(row, ensure_ascii=False) + "\n") + if (i + 1) % 50 == 0: + print(f"{i + 1}/{len(examples)}", flush=True) + gpu.stop_flag.set() + + measured = [r for r in rows if not r["warmup"]] + summary = { + "candidate": adapter.describe(), "precision": model_precision(adapter), "device": args.device, + "split": args.split, "examples": len(rows), "seed": args.seed, "warmup": args.warmup, + "batch_size": 1, "state_budget_tokens": args.state_budget, **guard, + "load_seconds": round(load_s, 2), + "warmup_latency_ms": [round(r["latency_ms"], 1) for r in rows if r["warmup"]], + "latency_ms": latency_summary([r["latency_ms"] for r in measured if r.get("pred") is not None]), + "errors": {k: sum(r.get("error") == k for r in rows) for k in {r.get("error") for r in rows} if k}, + "truncated": sum(r["truncated"] for r in rows), + "state_tokens_max": max((r["state_tokens"] or 0) for r in rows), + "rss_peak_mib": rss_peak_mib(), **gpu.summary(), + "quality": {k: v for k, v in classification(rows).items() if k != "per_label"}, + "calibration_raw": calibration([r for r in rows if r.get("pred")]), + } + (out / f"summary-{args.candidate}-{args.device}-{args.split}.json").write_text( + json.dumps(summary, indent=2, ensure_ascii=False) + "\n") + print(json.dumps({k: summary[k] for k in ("examples", "load_seconds", "latency_ms", "errors", "quality")}, + ensure_ascii=False)) + sys.exit(0) + + +if __name__ == "__main__": + main() diff --git a/infra/ds/collect.py b/infra/ds/collect.py new file mode 100644 index 0000000..25195e6 --- /dev/null +++ b/infra/ds/collect.py @@ -0,0 +1,79 @@ +"""Coleta os resumos de avaliação (dmb-resultados e dmb-resultados-gru) para reports//. + +Para cada (candidato, dispositivo, partição) usa a execução mais recente bem-sucedida com o +número de exemplos pedido. Gera os JSONs, as predições e um README.md com a tabela comparativa. + +Uso: PYTHONPATH=infra/oci .venv/bin/python infra/ds/collect.py --split dev --examples 50 --dest reports/pilot +""" +import argparse +import json +from pathlib import Path + +import oci + +import provision +from session import load_config + +ROOT = Path(__file__).resolve().parents[2] +SOURCES = [(provision.REGION, "dmb-resultados"), (provision.GPU_REGION, f"dmb-resultados{provision.GPU_SUFFIX}")] + + +def fmt(v, nd=1): + return "—" if v is None else f"{v:.{nd}f}" + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("--split", default="dev") + ap.add_argument("--examples", type=int) + ap.add_argument("--dest", required=True) + args = ap.parse_args() + dest = ROOT / args.dest + dest.mkdir(parents=True, exist_ok=True) + latest = {} + for region, bucket in SOURCES: + c = oci.object_storage.ObjectStorageClient(load_config(region=region)) + ns = c.get_namespace().data + names = [o.name for o in oci.pagination.list_call_get_all_results( + c.list_objects, ns, bucket, prefix="runs/").data.objects] + for n in names: + fname = n.rsplit("/", 1)[-1] + if not (fname.startswith("summary-") and fname.endswith(f"-{args.split}.json")): + continue + s = json.loads(c.get_object(ns, bucket, n).data.content) + if args.examples and s["examples"] != args.examples: + continue + key = (s["candidate"]["name"], s["device"]) + run_id = n.split("/")[1] + if key not in latest or run_id > latest[key][0]: + pred = n.replace("summary-", "predictions-").replace(".json", ".jsonl") + latest[key] = (run_id, s, c, ns, bucket, pred, region) + rows = [] + for (cand, dev), (run_id, s, c, ns, bucket, pred, region) in sorted(latest.items()): + s["oci"] = {"region": region, "run_id": run_id} + (dest / f"summary-{cand}-{dev}.json").write_text(json.dumps(s, indent=2, ensure_ascii=False) + "\n") + try: + (dest / f"predictions-{cand}-{dev}.jsonl").write_bytes(c.get_object(ns, bucket, pred).data.content) + except oci.exceptions.ServiceError: + pass + q, lat, cal = s["quality"], s["latency_ms"], s.get("calibration_raw", {}) + rows.append(f"| {cand} | {dev} | {s.get('precision') or '—'} | {q.get('n', 0)} | " + f"{fmt(q.get('accuracy', 0) * 100)}% | {fmt(q.get('macro_f1', 0) * 100)}% | " + f"{fmt(cal.get('ece'), 3)} | {fmt(lat['p50'])} | {fmt(lat['p95'])} | {fmt(lat['p99'])} | " + f"{fmt(s['load_seconds'])} | {fmt(s.get('rss_peak_mib'), 0)} | {fmt(s.get('gpu_mem_peak_mib'), 0)} | " + f"{sum(s['errors'].values())} | {s['truncated']} |") + table = ("| Candidato | Disp. | Precisão | n | Acurácia | Macro F1 | ECE (bruto) | p50 ms | p95 ms | p99 ms | " + "Carga s | RAM pico MiB | GPU pico MiB | Erros | Truncados |\n" + "|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|\n" + "\n".join(rows)) + readme = (f"# Resultados — partição `{args.split}`" + + (f", {args.examples} exemplos" if args.examples else "") + "\n\n" + "Gerado por `infra/ds/collect.py` a partir dos buckets de resultados. CPU: us-chicago-1 " + "(`VM.Standard.E4.Flex`, 8 OCPU, sub-rede privada). GPU: sa-saopaulo-1 (`VM.GPU.A10.1`). " + "Lote 1; latências excluem os exemplos de aquecimento. Probabilidades ainda sem calibração.\n\n" + + table + "\n") + (dest / "README.md").write_text(readme) + print(readme) + + +if __name__ == "__main__": + main() diff --git a/infra/ds/gpu_queue.py b/infra/ds/gpu_queue.py index b1715be..3c6f3cd 100644 --- a/infra/ds/gpu_queue.py +++ b/infra/ds/gpu_queue.py @@ -21,6 +21,7 @@ def main(): ap.add_argument("--after") ap.add_argument("--shape", default="VM.GPU.A10.1") ap.add_argument("--storage", type=int, default=100) + ap.add_argument("--env", nargs="*", default=[], help="variáveis extras K=V (ex.: DMB_SPLIT=dev DMB_LIMIT=50)") args = ap.parse_args() LOG.parent.mkdir(exist_ok=True) if args.after: @@ -29,7 +30,8 @@ def main(): while True: try: run_id = jobs.run_job(f"{args.mode}-{cand}-gpu", "infra/ds/run_job.sh", args.shape, None, None, - False, {"DMB_CANDIDATE": cand, "DMB_MODE": args.mode, "DMB_DEVICE": "auto"}, + False, {"DMB_CANDIDATE": cand, "DMB_MODE": args.mode, "DMB_DEVICE": "auto", + **dict(kv.split("=", 1) for kv in args.env)}, storage_gb=args.storage, region=provision.GPU_REGION) break except jobs.oci.exceptions.ServiceError as error: diff --git a/tests/test_metrics.py b/tests/test_metrics.py new file mode 100644 index 0000000..a6c77aa --- /dev/null +++ b/tests/test_metrics.py @@ -0,0 +1,29 @@ +import pytest + +from analysis.metrics import calibration, classification, latency_summary, selective + +ROWS = [ + {"gold": "a", "pred": "a", "probabilities": {"a": 0.9, "b": 0.1}}, + {"gold": "a", "pred": "b", "probabilities": {"a": 0.4, "b": 0.6}}, + {"gold": "b", "pred": "b", "probabilities": {"a": 0.2, "b": 0.8}}, + {"gold": "b", "pred": None, "error": "timeout"}, +] + + +def test_classification(): + m = classification(ROWS) + assert m["n"] == 3 and m["errors"] == 1 + assert m["accuracy"] == pytest.approx(2 / 3) + assert m["per_label"]["a"]["recall"] == pytest.approx(0.5) + assert m["per_label"]["b"]["precision"] == pytest.approx(0.5) + + +def test_calibration_and_selective(): + c = calibration(ROWS) + assert 0 <= c["ece"] <= 1 and c["brier"] > 0 + s = selective(ROWS, 0.7) + assert s["coverage"] == pytest.approx(2 / 3) and s["selective_accuracy"] == 1.0 + + +def test_latency(): + assert latency_summary([1, 2, 3, 4])["p50"] == pytest.approx(2.5)