From 93e9c838bb6ab13017397216fb33f15d58dc700b Mon Sep 17 00:00:00 2001 From: Nucleic Date: Thu, 30 Jul 2026 05:04:34 -0700 Subject: [PATCH] Merge nucleic/sleek-ember-seal-uady into dev --- ENERGY_AND_RESIDENCY.md | 46 +++ README.md | 52 +++- audit_data.py | 531 ++++++++++++++++++++++++++++++++++ data/curation-review-v1.json | 84 ++++++ data/dataset-v1-manifest.json | 51 ++-- data/frozen-test-v1.jsonl | 1 - data/semantic-audit-v1.json | 97 +++++++ eval.py | 246 +++++++++++++--- export.py | 334 +++++++++++++++++++++ prepare_data.py | 48 ++- purpose_data.py | 67 +++++ requirements-base.txt | 3 + tests/test_audit_data.py | 92 ++++++ tests/test_eval.py | 27 ++ tests/test_purpose_data.py | 20 ++ tests/test_train.py | 42 +++ train.py | 144 ++++++++- 17 files changed, 1802 insertions(+), 83 deletions(-) create mode 100644 ENERGY_AND_RESIDENCY.md create mode 100644 audit_data.py create mode 100644 data/curation-review-v1.json create mode 100644 data/semantic-audit-v1.json create mode 100644 export.py create mode 100644 tests/test_audit_data.py create mode 100644 tests/test_eval.py diff --git a/ENERGY_AND_RESIDENCY.md b/ENERGY_AND_RESIDENCY.md new file mode 100644 index 0000000..fe59da9 --- /dev/null +++ b/ENERGY_AND_RESIDENCY.md @@ -0,0 +1,46 @@ +# Purpose-classifier energy and accelerator-residency check + +This is the required per-model-version hardware spot check from +`docs/PURPOSE_CLASSIFIER.md` §8.5. It is deliberately not CI: the numbers are meaningful +only on named physical hardware with a stable power source, thermal state, and runtime +placement. + +## Common protocol + +1. Record the artifact SHA-256, app/build commit, machine model, OS build, battery/AC + state, ambient power mode, and runtime compute policy. +2. Warm the model with 100 batch-one classifications, then classify the same fixed + 1,000-prompt sequence. Tokenization is included. Disable network activity and other + foreground workloads. +3. Run at least five trials per compute policy, alternating policy order. Report median + wall time, p95 inference latency, average package power, and joules per inference. +4. Capture the runtime placement evidence alongside the power trace. A fast CPU fallback + is still a residency failure for `purpose-deep`. + +## macOS + +- Run the release Core ML artifact once with `.cpuAndNeuralEngine`, recording the + `MLComputePlan` ANE operation share and the artifact's required floor. +- Repeat with a diagnostic CPU-only configuration on the same Mac and power source. +- Sample the 1,000-inference region with `sudo powermetrics --samplers cpu_power,gpu_power` + at one-second intervals. Store the raw trace outside git and put its summarized table in + the model metrics report. +- The default ANE policy is accepted only when it uses less average power and fewer joules + per inference than CPU-only. On AC, a `.all` GPU retry must separately meet its declared + GPU operation-share floor before it can activate a deep model. + +## Windows + +- Record the Windows ML execution provider and assigned device after AOT compilation. + Deep requires QNN NPU, OpenVINO NPU/GPU, NvTensorRT-RTX, or MIGraphX; bare CPU is only + valid for lite. +- On battery compare `MAX_EFFICIENCY` with an explicit CPU session. On AC compare the + selected NPU, otherwise the validated vendor GPU EP, with CPU. +- Capture the 1,000-inference window with HWiNFO sensor logging or the vendor's documented + NPU/GPU telemetry. Include package/device average power and energy per inference in the + model metrics report. +- Verify that unplugging skips the discrete-GPU rung and that an idle session unloads + without keeping the dGPU awake. + +The report is incomplete if it gives latency without placement evidence and energy, or if +it compares different prompt sequences between policies. diff --git a/README.md b/README.md index 80342a3..a3913ac 100644 --- a/README.md +++ b/README.md @@ -54,7 +54,9 @@ python3 ml/purpose-classifier/prepare_data.py --refresh-frozen-test Use a dedicated virtual environment. The base model is pinned to a specific `sentence-transformers/all-MiniLM-L6-v2` commit: a 6-layer, 384-dimensional encoder. The training collator always pads/truncates to 128 tokens so the later ONNX/Core ML export -can expose a fixed `1 x 128` runtime shape. +can expose a fixed `1 x 128` runtime shape. Long prompts preserve both ends as +`[CLS]` + 63 head tokens + `[SEP]` + 62 tail tokens + `[SEP]`; this keeps the ask when it +follows a pasted log or stack trace while retaining enough leading context to interpret it. ```bash python3 -m venv ml/purpose-classifier/.venv @@ -68,8 +70,9 @@ accelerator training, install the platform's `torch==2.13.0` build using PyTorch platform selector, then install `requirements-base.txt`. Training writes a local checkpoint, `calibration.json`, and `metrics.json` under -`outputs/purpose-lite-v1/`. It fits one validation-only temperature and derives nested -HIGH/MEDIUM/LOW cutoffs from calibrated top-one probability plus top-two margin. +`outputs/purpose-lite-v1/`. It selects checkpoints and fits temperature on label-scorable +validation records. When deriving nested HIGH/MEDIUM/LOW cutoffs, every `vague-eval` +record counts as an abstention miss even if its synthetic label happens to match. For a wiring smoke test, use a small deterministic prefix: @@ -85,10 +88,39 @@ ml/purpose-classifier/.venv/bin/python ml/purpose-classifier/train.py \ ml/purpose-classifier/.venv/bin/python ml/purpose-classifier/eval.py ``` -The command returns failure unless frozen accuracy is at least 95%, every purpose recall -is at least 85%, and measured batch-one p95 is at most 20 ms. Use `--no-gate` only for -diagnostic runs. Accelerator residency, ONNX export/quantization, tokenizer golden tests, -tier-drift evaluation, and Core ML parity remain follow-on work. Before calling dataset -work item 2 complete, also run a semantic embedding duplicate audit and record the planned -10% human label spot-check; the current dependency-free word-trigram pass is deliberately -conservative. +The command returns failure unless label-scorable frozen accuracy is at least 95%, every +purpose recall is at least 85%, at least 90% of the deliberately context-free +`vague-eval` slice resolves LOW, every misroute stays within one routing cost tier, and +measured batch-one p95 is at most 20 ms. Use `--no-gate` only for diagnostic runs. + +## Export and score ONNX + +```bash +ml/purpose-classifier/.venv/bin/python ml/purpose-classifier/export.py +ml/purpose-classifier/.venv/bin/python ml/purpose-classifier/eval.py \ + --onnx-model ml/purpose-classifier/outputs/purpose-lite-v1/export/purpose-lite-v1-int8-qdq.onnx \ + --report ml/purpose-classifier/outputs/purpose-lite-v1/export/int8-frozen-eval.json +``` + +`export.py` emits fixed-shape opset-17 fp16 and int8-QDQ graphs, a tokenizer/ +normalization contract, golden tokenizations, shared calibration config, graph checks, +artifact hashes, and a size report. The int8 graph is the ≤25 MiB shipping candidate; +the fp16 graph remains the accelerator-oriented conversion input. + +## Audit curation + +Run the semantic embedding duplicate audit. It also emits the deterministic, +purpose/slice/language-stratified 10% human label-and-difficulty review CSV: + +```bash +ml/purpose-classifier/.venv/bin/python ml/purpose-classifier/audit_data.py +``` + +The semantic pass uses the same commit-pinned MiniLM encoder and fixed 128-token input as +`purpose-lite`. Similarity only proposes review candidates; it never edits source data or +the frozen split automatically. The 18 word-trigram exclusions and the human-review +completion rule are recorded in `data/curation-review-v1.json`; the semantic report is +versioned as `data/semantic-audit-v1.json`. + +The one-time, hardware-bound energy and accelerator-residency procedure is in +`ENERGY_AND_RESIDENCY.md`. diff --git a/audit_data.py b/audit_data.py new file mode 100644 index 0000000..2e8a894 --- /dev/null +++ b/audit_data.py @@ -0,0 +1,531 @@ +#!/usr/bin/env python3 +"""Audit semantic duplicates and emit the frozen human-review sample.""" + +from __future__ import annotations + +import argparse +import csv +import hashlib +import json +import math +import sys +from collections import Counter, defaultdict +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Sequence + +import numpy as np + +from prepare_data import ( + DEFAULT_CURATION_REVIEW, + DEFAULT_FIXTURES, + DEFAULT_FROZEN_TEST, + default_source_paths, + load_reviewed_semantic_exclusions, +) +from purpose_data import ( + LABELS, + DataError, + SourceRecord, + curate_records, + exclude_reviewed_duplicates, + load_classifiable_fixtures, + load_sources, + normalize_prompt, + prompt_hash, + write_json, +) +from train import ( + DEFAULT_MODEL, + DEFAULT_MODEL_REVISION, + HEAD_TOKENS, + MAX_LENGTH, + TAIL_TOKENS, + encode_fixed_shape, +) + + +SCRIPT_DIR = Path(__file__).resolve().parent +REPOSITORY_ROOT = SCRIPT_DIR.parent.parent +DEFAULT_REPORT = SCRIPT_DIR / "data" / "semantic-audit-v1.json" +DEFAULT_REVIEW_CSV = SCRIPT_DIR / ".artifacts" / "human-review-v1.csv" +DEFAULT_SAMPLE_SEED = 0xA11D17 +DEFAULT_SAMPLE_FRACTION = 0.10 + + +@dataclass(frozen=True) +class AuditRecord: + prompt: str + purpose: str + prompt_hash: str + origin: str + source: str + line: int | None + + +@dataclass(frozen=True) +class SemanticCandidate: + left: int + right: int + similarity: float + + +def _relative(path: Path) -> str: + try: + return str(path.resolve().relative_to(REPOSITORY_ROOT)) + except ValueError: + return str(path.resolve()) + + +def _stable_rank(seed: int, record: SourceRecord) -> str: + material = f"{seed}\0{prompt_hash(record.value['prompt'])}".encode("utf-8") + return hashlib.sha256(material).hexdigest() + + +def _review_stratum(record: SourceRecord) -> tuple[str, str, str]: + language = record.value["lang"].split("-", 1)[0].casefold() + return record.value["purpose"], record.value["slice"], language + + +def stratified_review_sample( + records: Sequence[SourceRecord], + *, + fraction: float = DEFAULT_SAMPLE_FRACTION, + seed: int = DEFAULT_SAMPLE_SEED, +) -> list[SourceRecord]: + """Choose exactly round(N*fraction), apportioned across purpose/slice/language.""" + + if not records: + raise DataError("cannot sample an empty review population") + if not 0.0 < fraction <= 1.0: + raise DataError("review fraction must be in (0, 1]") + + target = round(len(records) * fraction) + groups: dict[tuple[str, str, str], list[SourceRecord]] = defaultdict(list) + for record in records: + groups[_review_stratum(record)].append(record) + + # Hamilton apportionment preserves small language/slice strata while still producing + # the exact requested global sample size. + allocations: dict[tuple[str, str, str], int] = {} + remainders: list[tuple[float, str, tuple[str, str, str]]] = [] + allocated = 0 + for key in sorted(groups): + quota = len(groups[key]) * target / len(records) + base = math.floor(quota) + allocations[key] = base + allocated += base + tie_break = hashlib.sha256(f"{seed}\0{key}".encode("utf-8")).hexdigest() + remainders.append((quota - base, tie_break, key)) + for _, _, key in sorted(remainders, reverse=True)[: target - allocated]: + allocations[key] += 1 + + selected: list[SourceRecord] = [] + for key in sorted(groups): + ordered = sorted(groups[key], key=lambda record: _stable_rank(seed, record)) + selected.extend(ordered[: allocations[key]]) + return sorted(selected, key=lambda record: _stable_rank(seed + 1, record)) + + +def write_review_csv(path: Path, records: Sequence[SourceRecord]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + with path.open("w", encoding="utf-8", newline="") as handle: + writer = csv.DictWriter( + handle, + fieldnames=[ + "promptHash", + "source", + "line", + "prompt", + "generatedPurpose", + "generatedSecondary", + "generatedMixed", + "generatedDifficulty", + "generatedSlice", + "generatedLanguage", + "reviewedPurpose", + "reviewedSecondary", + "reviewedDifficulty", + "reviewStatus", + "reviewNotes", + ], + ) + writer.writeheader() + for record in records: + value = record.value + writer.writerow( + { + "promptHash": prompt_hash(value["prompt"]), + "source": _relative(record.source), + "line": record.line, + "prompt": normalize_prompt(value["prompt"]), + "generatedPurpose": value["purpose"], + "generatedSecondary": value["secondary"] or "", + "generatedMixed": str(value["mixed"]).lower(), + "generatedDifficulty": value["difficulty"], + "generatedSlice": value["slice"], + "generatedLanguage": value["lang"], + "reviewedPurpose": "", + "reviewedSecondary": "", + "reviewedDifficulty": "", + "reviewStatus": "", + "reviewNotes": "", + } + ) + + +def semantic_candidates( + embeddings: np.ndarray, + purposes: Sequence[str], + *, + same_label_threshold: float, + cross_label_threshold: float, + neighbors: int, + block_size: int, +) -> list[SemanticCandidate]: + """Return de-duplicated high-cosine nearest-neighbor pairs.""" + + if embeddings.ndim != 2 or embeddings.shape[0] != len(purposes): + raise ValueError("embeddings and purposes must have matching record counts") + if len(purposes) < 2: + return [] + if not ( + 0.0 < same_label_threshold <= 1.0 + and 0.0 < cross_label_threshold <= 1.0 + ): + raise ValueError("semantic thresholds must be in (0, 1]") + if neighbors <= 0 or block_size <= 0: + raise ValueError("neighbors and block size must be positive") + + norms = np.linalg.norm(embeddings, axis=1, keepdims=True) + if np.any(norms == 0): + raise ValueError("semantic embeddings must be non-zero") + normalized = embeddings.astype(np.float32, copy=False) / norms + top_count = min(neighbors, len(purposes) - 1) + found: dict[tuple[int, int], float] = {} + for start in range(0, len(purposes), block_size): + stop = min(start + block_size, len(purposes)) + similarities = normalized[start:stop] @ normalized.T + local_rows = np.arange(stop - start) + similarities[local_rows, np.arange(start, stop)] = -1.0 + candidate_columns = np.argpartition( + similarities, -top_count, axis=1 + )[:, -top_count:] + for local_index, columns in enumerate(candidate_columns): + left = start + local_index + for right in columns: + similarity = float(similarities[local_index, right]) + threshold = ( + same_label_threshold + if purposes[left] == purposes[right] + else cross_label_threshold + ) + if similarity < threshold: + continue + pair = (min(left, int(right)), max(left, int(right))) + found[pair] = max(found.get(pair, -1.0), similarity) + return [ + SemanticCandidate(left, right, similarity) + for (left, right), similarity in sorted( + found.items(), key=lambda item: (-item[1], item[0]) + ) + ] + + +def _select_device(torch: Any, requested: str) -> Any: + if requested != "auto": + return torch.device(requested) + if torch.cuda.is_available(): + return torch.device("cuda") + if hasattr(torch.backends, "mps") and torch.backends.mps.is_available(): + return torch.device("mps") + return torch.device("cpu") + + +def embed_prompts( + records: Sequence[AuditRecord], + *, + model_name: str, + model_revision: str, + device_name: str, + batch_size: int, +) -> tuple[np.ndarray, str]: + try: + import torch + from transformers import AutoModel, AutoTokenizer + except ImportError as exc: + raise DataError( + "semantic-audit dependencies are missing; install requirements.txt" + ) from exc + + device = _select_device(torch, device_name) + tokenizer = AutoTokenizer.from_pretrained( + model_name, revision=model_revision, use_fast=True + ) + model = AutoModel.from_pretrained(model_name, revision=model_revision).to(device) + model.eval() + chunks: list[np.ndarray] = [] + with torch.inference_mode(): + for start in range(0, len(records), batch_size): + batch = records[start : start + batch_size] + encoded = encode_fixed_shape( + tokenizer, + [record.prompt for record in batch], + torch, + ) + encoded = {key: value.to(device) for key, value in encoded.items()} + hidden = model(**encoded).last_hidden_state + mask = encoded["attention_mask"].unsqueeze(-1).expand(hidden.size()).float() + pooled = (hidden * mask).sum(dim=1) / mask.sum(dim=1).clamp(min=1e-9) + pooled = torch.nn.functional.normalize(pooled, p=2, dim=1) + chunks.append(pooled.cpu().numpy()) + return np.concatenate(chunks), str(device) + + +def _audit_records( + curated: Sequence[SourceRecord], + fixtures: Sequence[dict[str, str]], + fixtures_path: Path, +) -> list[AuditRecord]: + records = [ + AuditRecord( + prompt=record.value["prompt"], + purpose=record.value["purpose"], + prompt_hash=prompt_hash(record.value["prompt"]), + origin="synthetic", + source=_relative(record.source), + line=record.line, + ) + for record in curated + ] + records.extend( + AuditRecord( + prompt=fixture["prompt"], + purpose=fixture["purpose"], + prompt_hash=prompt_hash(fixture["prompt"]), + origin="shipped-fixture", + source=_relative(fixtures_path), + line=None, + ) + for fixture in fixtures + ) + return records + + +def _split_membership() -> dict[str, str]: + membership: dict[str, str] = {} + artifact_dir = SCRIPT_DIR / ".artifacts" / "dataset-v1" + for split in ("train", "validation"): + path = artifact_dir / f"{split}.jsonl" + if not path.exists(): + continue + for line in path.read_text(encoding="utf-8").splitlines(): + value = json.loads(line) + membership[prompt_hash(value["prompt"])] = split + for line in DEFAULT_FROZEN_TEST.read_text(encoding="utf-8").splitlines(): + value = json.loads(line) + membership[prompt_hash(value["prompt"])] = "test" + return membership + + +def audit(args: argparse.Namespace) -> dict[str, Any]: + sources = ( + [path.resolve() for path in args.source] + if args.source + else default_source_paths() + ) + fixtures = load_classifiable_fixtures(args.fixtures.resolve()) + source_records = load_sources(sources) + lexical_curation = curate_records( + source_records, + fixtures, + near_duplicate_threshold=args.word_duplicate_threshold, + ) + curated = exclude_reviewed_duplicates( + lexical_curation.records, + load_reviewed_semantic_exclusions(args.curation_review.resolve()), + ) + sample = stratified_review_sample( + curated.records, + fraction=args.review_fraction, + seed=args.review_seed, + ) + write_review_csv(args.review_csv.resolve(), sample) + + audit_records = _audit_records( + curated.records, fixtures, args.fixtures.resolve() + ) + embeddings, device = embed_prompts( + audit_records, + model_name=args.model, + model_revision=args.model_revision, + device_name=args.device, + batch_size=args.batch_size, + ) + candidates = semantic_candidates( + embeddings, + [record.purpose for record in audit_records], + same_label_threshold=args.same_label_threshold, + cross_label_threshold=args.cross_label_threshold, + neighbors=args.neighbors, + block_size=args.block_size, + ) + membership = _split_membership() + + candidate_values = [] + leakage_candidates = 0 + for candidate in candidates: + left = audit_records[candidate.left] + right = audit_records[candidate.right] + left_split = ( + "shipped-fixture" + if left.origin == "shipped-fixture" + else membership.get(left.prompt_hash, "unknown") + ) + right_split = ( + "shipped-fixture" + if right.origin == "shipped-fixture" + else membership.get(right.prompt_hash, "unknown") + ) + crosses_frozen_boundary = ( + left_split == "train" + and right_split in {"validation", "test", "shipped-fixture"} + ) or ( + right_split == "train" + and left_split in {"validation", "test", "shipped-fixture"} + ) + leakage_candidates += int(crosses_frozen_boundary) + candidate_values.append( + { + "similarity": round(candidate.similarity, 6), + "sameLabel": left.purpose == right.purpose, + "crossesFrozenBoundary": crosses_frozen_boundary, + "left": { + "promptHash": left.prompt_hash, + "purpose": left.purpose, + "origin": left.origin, + "split": left_split, + "source": left.source, + "line": left.line, + }, + "right": { + "promptHash": right.prompt_hash, + "purpose": right.purpose, + "origin": right.origin, + "split": right_split, + "source": right.source, + "line": right.line, + }, + } + ) + + report = { + "schemaVersion": 1, + "auditVersion": "purpose-semantic-audit-v1", + "embedding": { + "model": args.model, + "revision": args.model_revision, + "fixedInputShape": [1, MAX_LENGTH], + "truncation": { + "strategy": "head-tail-pair", + "headTokens": HEAD_TOKENS, + "tailTokens": TAIL_TOKENS, + }, + "pooling": "attention-mask mean pooling followed by L2 normalization", + "device": device, + }, + "population": { + "canonicalSourceRecords": len(source_records), + "curatedSyntheticRecords": len(curated.records), + "classifiableShippedFixtures": len(fixtures), + "auditedRecords": len(audit_records), + }, + "candidatePolicy": { + "sameLabelCosineThreshold": args.same_label_threshold, + "crossLabelCosineThreshold": args.cross_label_threshold, + "nearestNeighborsPerRecord": args.neighbors, + "automaticExclusion": False, + "note": ( + "Embedding similarity proposes review candidates only. It cannot safely " + "distinguish deliberate boundary pairs from duplicated labels." + ), + }, + "summary": { + "candidates": len(candidate_values), + "sameLabelCandidates": sum(item["sameLabel"] for item in candidate_values), + "crossLabelCandidates": sum( + not item["sameLabel"] for item in candidate_values + ), + "frozenBoundaryCandidates": leakage_candidates, + }, + "humanReviewSample": { + "status": "planned", + "fraction": args.review_fraction, + "seed": args.review_seed, + "populationRecords": len(curated.records), + "sampleRecords": len(sample), + "strata": ["purpose", "slice", "primary language"], + "generatedCSV": _relative(args.review_csv), + "samplePurposeCounts": dict( + sorted(Counter(record.value["purpose"] for record in sample).items()) + ), + "sampleSliceCounts": dict( + sorted(Counter(record.value["slice"] for record in sample).items()) + ), + "requiredFields": [ + "reviewedPurpose", + "reviewedSecondary", + "reviewedDifficulty", + "reviewStatus", + ], + }, + "candidates": candidate_values, + } + write_json(args.report.resolve(), report) + return report + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--source", action="append", type=Path) + parser.add_argument("--fixtures", type=Path, default=DEFAULT_FIXTURES) + parser.add_argument("--report", type=Path, default=DEFAULT_REPORT) + parser.add_argument("--review-csv", type=Path, default=DEFAULT_REVIEW_CSV) + parser.add_argument( + "--curation-review", type=Path, default=DEFAULT_CURATION_REVIEW + ) + parser.add_argument("--review-fraction", type=float, default=DEFAULT_SAMPLE_FRACTION) + parser.add_argument("--review-seed", type=int, default=DEFAULT_SAMPLE_SEED) + parser.add_argument("--model", default=DEFAULT_MODEL) + parser.add_argument("--model-revision", default=DEFAULT_MODEL_REVISION) + parser.add_argument("--device", default="auto") + parser.add_argument("--batch-size", type=int, default=64) + parser.add_argument("--block-size", type=int, default=256) + parser.add_argument("--neighbors", type=int, default=8) + parser.add_argument("--same-label-threshold", type=float, default=0.97) + parser.add_argument("--cross-label-threshold", type=float, default=0.985) + parser.add_argument("--word-duplicate-threshold", type=float, default=0.92) + return parser + + +def main(argv: Sequence[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + if args.batch_size <= 0 or args.block_size <= 0 or args.neighbors <= 0: + parser.error("batch size, block size, and neighbors must be positive") + try: + report = audit(args) + except (DataError, OSError, UnicodeError, ValueError) as exc: + print(f"error: {exc}", file=sys.stderr) + return 1 + print( + f"Semantic audit: {report['population']['auditedRecords']} records, " + f"{report['summary']['candidates']} candidates " + f"({report['summary']['frozenBoundaryCandidates']} cross a frozen boundary); " + f"{report['humanReviewSample']['sampleRecords']} prompts sampled for review." + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/data/curation-review-v1.json b/data/curation-review-v1.json new file mode 100644 index 0000000..f686f5f --- /dev/null +++ b/data/curation-review-v1.json @@ -0,0 +1,84 @@ +{ + "schemaVersion": 1, + "datasetVersion": "purpose-dataset-v1", + "wordTrigramExclusionReview": { + "status": "complete", + "reviewedRecords": 18, + "confirmedTemplateDuplicates": 18, + "rejectedExclusions": 0, + "decision": "All 18 pairs preserve the same task, requested outcome, and purpose label; differences are generated identifiers, ticket numbers, cosmetic context prefixes, or difficulty/slice metadata. Keep the earliest member and exclude the later template copy.", + "reviewedDroppedPromptHashes": [ + "f29de1d646055017f80346055bb3ac1468c49af7ab0c6249f45f19cf386f9fa7", + "dabbcd0a19d2918199b5d0d35fac49fd5413cb17a5164252b789114fc68ec43b", + "ac3a166e07a1db67475c773ef3d996d12fe627d11b7bb91a7c524c9c67c841b0", + "d5f9d735490874ded856105e10df99ed3c2c703c09ec7909fd07aa1e9e7a0c91", + "6376e5f6f0d7c3d6faae9e50975034113f43a7f959bcbbea67531d3b4f0cfd16", + "10933534254a1f73163e0ba7883225e2d0c896b565590753698c1052743af664", + "cf4deb437c380216601b176e238964bfce79bf2645d54e1eb659f8479b980186", + "fe6b2e05da51a12ea07d6fa015933196931fb1023b62c0eb267b80318f083ba0", + "c0fddb025279caf6438453b53c99b266ccfb61c6c19609b43c6f731d88812813", + "15c28d9f67ba103378db711c82193aba7330acbe602d9df54995a5db20e1bbb1", + "b3583a130811ba7d2f425042c41e1f69244c622764a0c952ed42bb5114e18735", + "bba760782054c4ed405611cf8b55bc99d710424bb3f9bab0682dc2c1cf4e2a52", + "bd4d2d8ea68a4e56eaed5c6465815abefa6c8ae78fb533eb81547dedac9c71ed", + "8c1ab1a866cadd294c3d2a6c1d6eefbede62549220c81e1dd28a886717aa285d", + "c074ce20e7cd81c28d390aed919bab30cabe9211817103e52c9b229e6a3c94ff", + "fdc04376376dbffcaf9928e9edcef29376203640b49c200c2c117f78891125a1", + "432e67ead52d6cd57e4a5034cd8bcde14bd2da13f88cb5532c42fd8fbd8daf05", + "24f3759ff4a3fb9fc77bed4c8525f30f9af2479a6cffbb0c4842043f99a7d8f1" + ] + }, + "semanticDuplicateReview": { + "status": "complete", + "auditVersion": "purpose-semantic-audit-v1", + "reviewedCandidates": 4, + "excludedCandidates": 3, + "retainedCandidates": 1, + "decision": "Exclude three same-purpose generated template copies. One pair ('make the screen nicer' / 'Make this screen nicer') remains because it is a deliberate vague-language variant, lives entirely in validation, and does not contaminate the frozen evaluation boundary.", + "excluded": [ + { + "droppedPromptHash": "7348d764d602cfff72dd2b1d369ad0375a96ea52b38174aa0bd6168a548bcc2f", + "matchedPromptHash": "3ad9cfbb664504f0e7ae84b905520937e7812d4e879fa706415782341f2cd400", + "similarity": 0.984897 + }, + { + "droppedPromptHash": "a0c6e7984e92adc11ea6855c39688d733bf7665f5f965939325b15f3ba39e937", + "matchedPromptHash": "17f79a890a323c08770193a94b100dadaa00116405bdb7d14c1b89abc3c3066a", + "similarity": 0.980998 + }, + { + "droppedPromptHash": "bfd7791069fd04c13797f39ba99224ad68ca2deebbc912f08b10815954a7d223", + "matchedPromptHash": "dc48031efb18f25a56a3beddcdd814f26473b23cd496482ce1884ae6f697417a", + "similarity": 0.973329 + } + ], + "retained": [ + { + "leftPromptHash": "70944e70e0062e947476f73b704ec01ecee19420d06fd174c58a41c04e8112f6", + "rightPromptHash": "cb3e63992cbf5830a222933dc7b44be39933baedcb46e335be02e3c379685517", + "similarity": 0.970376 + } + ] + }, + "humanLabelAndDifficultyReview": { + "status": "planned", + "populationRecords": 12193, + "sampleFraction": 0.1, + "sampleRecords": 1219, + "seed": 10558743, + "strata": [ + "purpose", + "slice", + "primary language" + ], + "reviewFields": [ + "purpose", + "secondary", + "difficulty", + "review status", + "notes" + ], + "generatedArtifact": "ml/purpose-classifier/.artifacts/human-review-v1.csv", + "completionRule": "Every sampled row must be marked accept, relabel, or reject by a human reviewer; relabel/reject decisions are applied to canonical source data before dataset-v1 can be called fully curated." + } +} diff --git a/data/dataset-v1-manifest.json b/data/dataset-v1-manifest.json index 82de8a9..c8cb287 100644 --- a/data/dataset-v1-manifest.json +++ b/data/dataset-v1-manifest.json @@ -1,12 +1,15 @@ { "curation": { "excludedDuplicates": { - "near": 18 + "near": 18, + "semantic-reviewed": 3 }, "inputRecords": 12214, "nearDuplicateMethod": "word-trigram Jaccard after SimHash LSH candidate search", "nearDuplicateThreshold": 0.92, - "retainedRecords": 12196, + "retainedRecords": 12193, + "reviewPath": "ml/purpose-classifier/data/curation-review-v1.json", + "reviewSha256": "2db763df027b545b4fd654108dcadb0717f9c9c0a63555bf8cea26b791cda370", "vagueEvalPolicy": "validation/test only" }, "datasetVersion": "purpose-dataset-v1", @@ -23,7 +26,7 @@ "shippedFixturesPath": "Tests/NucleicCoreTests/Fixtures/purpose-prompts.json", "shippedFixturesSha256": "6c8f0eedce35f9da7a407934009e87737b5e85c7371f6a118982d9fae2ea7aae", "syntheticPath": "ml/purpose-classifier/data/frozen-test-v1.jsonl", - "syntheticSha256": "e5a6b501a43ce256092932ab05b0af01ff9137aee42d85b7d58bb4ce65f2943f" + "syntheticSha256": "5d34c8ac3d39c5b4e2e739c4c9058093beb6f790a42a1397cedefae6271654ab" }, "ratios": { "test": 0.1, @@ -50,7 +53,7 @@ "distribution": { "language": { "de": 7, - "en": 1111, + "en": 1110, "es": 9, "fr": 5, "ja": 3, @@ -63,7 +66,7 @@ "frontendImpl": 157, "planning": 132, "quickFix": 140, - "refactor": 137, + "refactor": 136, "review": 127, "writing": 129 }, @@ -72,19 +75,19 @@ "core": 507, "mixed": 80, "pasted-context": 84, - "vague-eval": 292 + "vague-eval": 291 } }, - "hardSyntheticRecords": 635, - "logicalRecords": 1229, - "sha256": "e5a6b501a43ce256092932ab05b0af01ff9137aee42d85b7d58bb4ce65f2943f", - "syntheticRecords": 1142 + "hardSyntheticRecords": 634, + "logicalRecords": 1228, + "sha256": "5d34c8ac3d39c5b4e2e739c4c9058093beb6f790a42a1397cedefae6271654ab", + "syntheticRecords": 1141 }, "train": { "distribution": { "language": { "de": 100, - "en": 9326, + "en": 9324, "es": 100, "fr": 81, "ja": 77, @@ -92,24 +95,24 @@ "zh": 72 }, "purpose": { - "backendImpl": 1225, + "backendImpl": 1226, "debugging": 1245, - "frontendImpl": 1235, + "frontendImpl": 1234, "planning": 1223, "quickFix": 1230, - "refactor": 1225, + "refactor": 1223, "review": 1230, "writing": 1213 }, "slice": { - "boundary": 2065, - "core": 5820, + "boundary": 2066, + "core": 5818, "mixed": 906, - "pasted-context": 1035 + "pasted-context": 1034 } }, - "records": 9826, - "sha256": "ca0ec9fe6fef6bc4013bf09bdfed5877327f69c98f8e5552c082cbb4e76d9b7d" + "records": 9824, + "sha256": "40a5064c2a95d4394d71b3edbad7d03dce7d67b2e5ba9ab9320476e35e3a4953" }, "validation": { "distribution": { @@ -123,25 +126,25 @@ "zh": 7 }, "purpose": { - "backendImpl": 184, + "backendImpl": 183, "debugging": 157, "frontendImpl": 168, "planning": 141, "quickFix": 151, - "refactor": 148, + "refactor": 149, "review": 140, "writing": 139 }, "slice": { - "boundary": 196, + "boundary": 195, "core": 541, "mixed": 83, "pasted-context": 95, - "vague-eval": 313 + "vague-eval": 314 } }, "records": 1228, - "sha256": "7eca32f98105eec1c1a34c324851d52004d8dc720ca5853a1e91fa0a0dd620f4" + "sha256": "3d0a3730931db360dd8346cf12d03901ac139acf5be1d9e84975ac591186c588" } } } diff --git a/data/frozen-test-v1.jsonl b/data/frozen-test-v1.jsonl index 81ae576..537fa2b 100644 --- a/data/frozen-test-v1.jsonl +++ b/data/frozen-test-v1.jsonl @@ -747,7 +747,6 @@ {"prompt":"add a compact density option to the data grid — user setting, persists, rows go 44 -> 32 and the font drops a step","purpose":"frontendImpl","secondary":null,"mixed":false,"difficulty":0.5,"slice":"core","lang":"en"} {"prompt":"Ticket OPS-52152: retire the legacy replay path for LongbowFernSnapshotCoordinator\n\nBackground\nThe mobile clients now send monotonic cursors, but the worker still supports the pre-2024 offset token. That compatibility branch performs an extra database lookup on every page and owns a separate retry counter. Support confirmed that the oldest active client is version 8.14, which uses cursors.\n\nAcceptance notes\n- existing cursor tokens must remain valid during a two-release overlap\n- operators need one metric showing legacy traffic by client version\n- a rollback must not require restoring deleted columns\n- tenant isolation remains enforced in the storage query\n- SDK examples should show how to persist the continuation token\n- the public route and current 429 envelope cannot change\n\nOpen questions\nDo we delete the offset decoder immediately or quarantine it behind a flag? Who owns the mobile version gate? Can the database index be removed in the same deploy? Security wants evidence that forged cursors cannot select another tenant.\n\nSlack excerpt\nMina 09:42: “I can provide client adoption numbers, but not a guarantee about sideloaded builds.”\nRavi 09:47: “Please make the rollback trigger explicit; queue lag alone is too noisy.”\n\nAssess LongbowFernSnapshotCoordinator for durability, tenant isolation, races, and misleading observability. Separate blockers from questions and do not produce a patch.","purpose":"review","secondary":null,"mixed":false,"difficulty":0.79,"slice":"pasted-context","lang":"en"} {"prompt":"Propose an architecture for claim-event analytics with privacy, late corrections, financial snapshots, model features, retention, and reproducibility.","purpose":"planning","secondary":null,"mixed":false,"difficulty":0.91,"slice":"core","lang":"en"} -{"prompt":"CalderaQuartzPlayerCoordinator: could this be clearer","purpose":"refactor","secondary":null,"mixed":false,"difficulty":0.58,"slice":"vague-eval","lang":"en"} {"prompt":"Create an operator runbook for draining a NATS JetStream consumer, replaying a dead-letter subject, and confirming that lag has recovered.","purpose":"writing","secondary":null,"mixed":false,"difficulty":0.4,"slice":"core","lang":"en"} {"prompt":"Investigate why expired offers remain visible in saved-search results.","purpose":"debugging","secondary":null,"mixed":false,"difficulty":0.61,"slice":"boundary","lang":"en"} {"prompt":"Zephyr: Incident timeline — INC-49147\n\n08:02 deploy IsotopeOspreyJobFlow 6.18.0 begins in eu-west\n08:06 write latency rises from 22 ms to 410 ms; error rate remains below 0.2%\n08:08 autoscaler adds three workers; queue lag continues rising\n08:11 on-call disables feature flag `parallel_commit_v2`\n08:13 latency returns to 31 ms, but duplicate-event warnings increase\n08:19 traffic moved to the previous worker pool\n08:27 queue lag clears; no customer data loss observed\n\nWhat changed\nThe release moved metrics emission into the same Promise.all as the durable write and added NFC normalization to event payloads. Database CPU peaked at 64%, Redis stayed normal, and the downstream consumer reported 183 duplicate keys that its uniqueness constraint ignored.\n\nConstraints from incident command\n- no emergency schema change\n- preserve tenant ordering\n- canary must include a forced consumer rebalance\n- rollback decision must use two independent signals\n- ownership between storage and ingestion teams must be explicit\n\nThe next regular release window is in six days.\n\nMap a safe route from the current IsotopeOspreyJobFlow behavior to the desired one, comparing two approaches and naming telemetry, failure drills, and rollback responsibility.","purpose":"planning","secondary":null,"mixed":false,"difficulty":0.91,"slice":"pasted-context","lang":"en"} diff --git a/data/semantic-audit-v1.json b/data/semantic-audit-v1.json new file mode 100644 index 0000000..f13a20f --- /dev/null +++ b/data/semantic-audit-v1.json @@ -0,0 +1,97 @@ +{ + "auditVersion": "purpose-semantic-audit-v1", + "candidatePolicy": { + "automaticExclusion": false, + "crossLabelCosineThreshold": 0.985, + "nearestNeighborsPerRecord": 8, + "note": "Embedding similarity proposes review candidates only. It cannot safely distinguish deliberate boundary pairs from duplicated labels.", + "sameLabelCosineThreshold": 0.97 + }, + "candidates": [ + { + "crossesFrozenBoundary": false, + "left": { + "line": 4639, + "origin": "synthetic", + "promptHash": "70944e70e0062e947476f73b704ec01ecee19420d06fd174c58a41c04e8112f6", + "purpose": "frontendImpl", + "source": "ml/purpose-classifier/data/purpose-prompts.jsonl", + "split": "validation" + }, + "right": { + "line": 8435, + "origin": "synthetic", + "promptHash": "cb3e63992cbf5830a222933dc7b44be39933baedcb46e335be02e3c379685517", + "purpose": "frontendImpl", + "source": "ml/purpose-classifier/data/purpose-prompts.jsonl", + "split": "validation" + }, + "sameLabel": true, + "similarity": 0.970376 + } + ], + "embedding": { + "device": "cpu", + "fixedInputShape": [ + 1, + 128 + ], + "model": "sentence-transformers/all-MiniLM-L6-v2", + "pooling": "attention-mask mean pooling followed by L2 normalization", + "revision": "1110a243fdf4706b3f48f1d95db1a4f5529b4d41", + "truncation": { + "headTokens": 63, + "strategy": "head-tail-pair", + "tailTokens": 62 + } + }, + "humanReviewSample": { + "fraction": 0.1, + "generatedCSV": "ml/purpose-classifier/.artifacts/human-review-v1.csv", + "populationRecords": 12193, + "requiredFields": [ + "reviewedPurpose", + "reviewedSecondary", + "reviewedDifficulty", + "reviewStatus" + ], + "samplePurposeCounts": { + "backendImpl": 158, + "debugging": 154, + "frontendImpl": 155, + "planning": 152, + "quickFix": 150, + "refactor": 149, + "review": 150, + "writing": 151 + }, + "sampleRecords": 1219, + "sampleSliceCounts": { + "boundary": 245, + "core": 689, + "mixed": 108, + "pasted-context": 117, + "vague-eval": 60 + }, + "seed": 10558743, + "status": "planned", + "strata": [ + "purpose", + "slice", + "primary language" + ] + }, + "population": { + "auditedRecords": 12280, + "canonicalSourceRecords": 12214, + "classifiableShippedFixtures": 87, + "curatedSyntheticRecords": 12193 + }, + "schemaVersion": 1, + "summary": { + "candidates": 1, + "crossLabelCandidates": 0, + "frozenBoundaryCandidates": 0, + "sameLabelCandidates": 1 + } +} diff --git a/eval.py b/eval.py index 4889a4d..0c2aea6 100644 --- a/eval.py +++ b/eval.py @@ -19,13 +19,14 @@ from purpose_data import ( DataError, load_classifiable_fixtures, load_jsonl, - normalize_prompt, + prompt_hash, write_json, ) from train import ( MAX_LENGTH, classification_metrics, confidence_score, + encode_fixed_shape, expected_calibration_error, ) @@ -43,6 +44,20 @@ DEFAULT_FIXTURES = ( / "purpose-prompts.json" ) DEFAULT_REPORT = SCRIPT_DIR / "outputs" / "purpose-lite-v1" / "frozen-eval.json" +ROUTING_LEVELS = ("quick", "light", "balanced", "deep", "max") +# Cost tiers mirrored from IntelligenceRouter.matrix for the two provider lanes. Keep this +# table in sync with Sources/NucleicCore/IntelligenceRouting.swift; the eval report names +# every violating prompt/level/lane so a matrix change cannot fail opaquely. +ROUTING_COST_TIERS = { + "planning": {"claude": (1, 2, 2, 3, 3), "codex": (1, 1, 2, 2, 2)}, + "backendImpl": {"claude": (1, 1, 2, 3, 3), "codex": (0, 1, 2, 2, 2)}, + "frontendImpl": {"claude": (1, 1, 2, 2, 3), "codex": (0, 1, 1, 2, 2)}, + "quickFix": {"claude": (1, 1, 1, 2, 2), "codex": (0, 0, 1, 1, 2)}, + "refactor": {"claude": (1, 1, 1, 2, 3), "codex": (0, 1, 1, 2, 2)}, + "debugging": {"claude": (1, 1, 2, 3, 3), "codex": (1, 1, 2, 2, 2)}, + "review": {"claude": (0, 1, 1, 2, 2), "codex": (0, 1, 1, 2, 2)}, + "writing": {"claude": (0, 1, 1, 2, 2), "codex": (0, 0, 1, 1, 1)}, +} def _device(torch: Any, requested: str) -> Any: @@ -85,6 +100,46 @@ def _synchronize(torch: Any, device: Any) -> None: torch.mps.synchronize() +def routing_tier_drift( + records: Sequence[dict[str, Any]], + actual: Sequence[int], + predicted: Sequence[int], +) -> dict[str, Any]: + violations = [] + maximum = 0 + checked_misroutes = 0 + for record, expected_index, predicted_index in zip(records, actual, predicted): + if expected_index == predicted_index: + continue + checked_misroutes += 1 + expected = LABELS[expected_index] + got = LABELS[predicted_index] + for lane in ("claude", "codex"): + for level_index, level in enumerate(ROUTING_LEVELS): + drift = abs( + ROUTING_COST_TIERS[expected][lane][level_index] + - ROUTING_COST_TIERS[got][lane][level_index] + ) + maximum = max(maximum, drift) + if drift > 1: + violations.append( + { + "promptHash": prompt_hash(record["prompt"]), + "expected": expected, + "predicted": got, + "lane": lane, + "level": level, + "tierDrift": drift, + } + ) + return { + "checkedMisroutes": checked_misroutes, + "maximumTierDrift": maximum, + "violations": violations, + "passed": not violations, + } + + def evaluate(args: argparse.Namespace) -> dict[str, Any]: try: import torch @@ -114,29 +169,63 @@ def evaluate(args: argparse.Namespace) -> dict[str, Any]: high_threshold = float(calibration["confidence"]["high"]["minimumScore"]) medium_threshold = float(calibration["confidence"]["medium"]["minimumScore"]) label_to_id = {label: index for index, label in enumerate(LABELS)} - device = _device(torch, args.device) tokenizer = AutoTokenizer.from_pretrained(args.model_dir, local_files_only=True) - model = AutoModelForSequenceClassification.from_pretrained( - args.model_dir, local_files_only=True - ).to(device) - model.eval() + onnx_session = None + if args.onnx_model is not None: + try: + import onnxruntime as ort + except ImportError as exc: + raise DataError( + "ONNX evaluation requires onnxruntime from requirements.txt" + ) from exc + if args.device not in ("auto", "cpu"): + raise DataError("ONNX evaluation currently measures the CPU provider") + session_options = ort.SessionOptions() + session_options.graph_optimization_level = ort.GraphOptimizationLevel.ORT_ENABLE_ALL + onnx_session = ort.InferenceSession( + str(args.onnx_model), + sess_options=session_options, + providers=["CPUExecutionProvider"], + ) + onnx_input_names = {item.name for item in onnx_session.get_inputs()} + device = torch.device("cpu") + model = None + runtime_name = "onnxruntime-cpu" + else: + device = _device(torch, args.device) + model = AutoModelForSequenceClassification.from_pretrained( + args.model_dir, local_files_only=True + ).to(device) + model.eval() + onnx_input_names = set() + runtime_name = "pytorch" + + def predict_logits(encoded: dict[str, Any]) -> Any: + if onnx_session is not None: + inputs = { + key: value.numpy() + for key, value in encoded.items() + if key in onnx_input_names + } + return torch.from_numpy(onnx_session.run(["logits"], inputs)[0]) + moved = {key: value.to(device) for key, value in encoded.items()} + return model(**moved).logits.cpu() actual: list[int] = [] predicted: list[int] = [] probabilities: list[float] = [] confidences: list[str] = [] + margins: list[float] = [] + inference_batch_size = 1 if onnx_session is not None else args.batch_size with torch.inference_mode(): - for start in range(0, len(records), args.batch_size): - batch = records[start : start + args.batch_size] - encoded = tokenizer( - [normalize_prompt(record["prompt"]) for record in batch], - padding="max_length", - truncation=True, - max_length=MAX_LENGTH, - return_tensors="pt", + for start in range(0, len(records), inference_batch_size): + batch = records[start : start + inference_batch_size] + encoded = encode_fixed_shape( + tokenizer, + [record["prompt"] for record in batch], + torch, ) - encoded = {key: value.to(device) for key, value in encoded.items()} - logits = model(**encoded).logits.cpu() / temperature + logits = predict_logits(encoded) / temperature distribution = torch.softmax(logits, dim=-1) top = torch.topk(distribution, k=2, dim=-1) batch_probabilities = top.values[:, 0].tolist() @@ -156,6 +245,7 @@ def evaluate(args: argparse.Namespace) -> dict[str, Any]: actual.append(label_to_id[record["purpose"]]) predicted.append(prediction) probabilities.append(probability) + margins.append(margin) confidences.append(confidence) metrics = classification_metrics(actual, predicted) @@ -169,6 +259,24 @@ def evaluate(args: argparse.Namespace) -> dict[str, Any]: [actual[index] for index in hard_indexes], [predicted[index] for index in hard_indexes], ) + scored_indexes = [ + index + for index, record in enumerate(records) + if record.get("slice") != "vague-eval" + ] + scored_metrics = classification_metrics( + [actual[index] for index in scored_indexes], + [predicted[index] for index in scored_indexes], + ) + scored_hard_indexes = [ + index + for index in hard_indexes + if records[index].get("slice") != "vague-eval" + ] + scored_hard_metrics = classification_metrics( + [actual[index] for index in scored_hard_indexes], + [predicted[index] for index in scored_hard_indexes], + ) fixture_indexes = [ index for index, record in enumerate(records) @@ -187,40 +295,95 @@ def evaluate(args: argparse.Namespace) -> dict[str, Any]: if accepted_indexes else 1.0 ) + + def subset_report(indexes: Sequence[int]) -> dict[str, Any]: + subset_correct = [correctness[index] for index in indexes] + subset_accepted = [ + index for index in indexes if confidences[index] != "low" + ] + return { + **classification_metrics( + [actual[index] for index in indexes], + [predicted[index] for index in indexes], + ), + "confidenceCounts": dict( + sorted(Counter(confidences[index] for index in indexes).items()) + ), + "acceptedPrecision": ( + sum(correctness[index] for index in subset_accepted) + / len(subset_accepted) + if subset_accepted + else 1.0 + ), + "acceptedCoverage": len(subset_accepted) / len(indexes), + "meanTopProbability": statistics.mean( + probabilities[index] for index in indexes + ), + "meanTopTwoMargin": statistics.mean(margins[index] for index in indexes), + "expectedCalibrationError": expected_calibration_error( + [probabilities[index] for index in indexes], + subset_correct, + ), + } + + slice_reports = {} + for slice_name in sorted({record.get("slice", "unknown") for record in records}): + indexes = [ + index + for index, record in enumerate(records) + if record.get("slice", "unknown") == slice_name + ] + slice_reports[slice_name] = subset_report(indexes) + + misclassifications = [ + { + "promptHash": prompt_hash(records[index]["prompt"]), + "slice": records[index].get("slice", "unknown"), + "expected": LABELS[actual[index]], + "predicted": LABELS[predicted[index]], + "confidence": confidences[index], + "topProbability": round(probabilities[index], 6), + "topTwoMargin": round(margins[index], 6), + } + for index in range(len(records)) + if not correctness[index] + ] latency_samples: list[float] = [] latency_records = records[: args.latency_samples] if latency_records: with torch.inference_mode(): for record in latency_records[: min(5, len(latency_records))]: - encoded = tokenizer( - normalize_prompt(record["prompt"]), - padding="max_length", - truncation=True, - max_length=MAX_LENGTH, - return_tensors="pt", + encoded = encode_fixed_shape( + tokenizer, + [record["prompt"]], + torch, ) - model(**{key: value.to(device) for key, value in encoded.items()}) + predict_logits(encoded) _synchronize(torch, device) for record in latency_records: started = time.perf_counter() - encoded = tokenizer( - normalize_prompt(record["prompt"]), - padding="max_length", - truncation=True, - max_length=MAX_LENGTH, - return_tensors="pt", + encoded = encode_fixed_shape( + tokenizer, + [record["prompt"]], + torch, ) - model(**{key: value.to(device) for key, value in encoded.items()}) + predict_logits(encoded) _synchronize(torch, device) latency_samples.append((time.perf_counter() - started) * 1000) + tier_drift = routing_tier_drift(records, actual, predicted) report = { "modelVersion": calibration.get("modelVersion", args.model_dir.name), "device": str(device), + "runtime": runtime_name, + "artifact": str(args.onnx_model or args.model_dir), "fixedInputShape": [1, MAX_LENGTH], "overall": metrics, + "scoredClassification": scored_metrics, "hardSlice": hard_metrics, + "scoredHardSlice": scored_hard_metrics, "shippedFixtures": fixture_metrics, + "bySlice": slice_reports, "calibration": { "temperature": temperature, "expectedCalibrationError": expected_calibration_error( @@ -235,16 +398,26 @@ def evaluate(args: argparse.Namespace) -> dict[str, Any]: "median": statistics.median(latency_samples) if latency_samples else 0.0, "p95": _percentile(latency_samples, 0.95), }, + "routingTierDrift": tier_drift, "gates": { - "accuracyAtLeast95Percent": metrics["accuracy"] >= 0.95, + "scoredAccuracyAtLeast95Percent": scored_metrics["accuracy"] >= 0.95, "everyPurposeRecallAtLeast85Percent": min( - metrics["perPurposeRecall"].values() + scored_metrics["perPurposeRecall"].values() ) >= 0.85, + "vagueEvalLowConfidenceAtLeast90Percent": ( + slice_reports.get("vague-eval", {}).get("confidenceCounts", {}).get( + "low", 0 + ) + / max(1, slice_reports.get("vague-eval", {}).get("records", 0)) + >= 0.90 + ), "latencyP95AtMost20Milliseconds": ( not latency_samples or _percentile(latency_samples, 0.95) <= 20.0 ), + "misroutesStayWithinOneCostTier": tier_drift["passed"], }, + "misclassifications": misclassifications, } write_json(args.report, report) return report @@ -253,6 +426,11 @@ def evaluate(args: argparse.Namespace) -> dict[str, Any]: def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--model-dir", type=Path, default=DEFAULT_MODEL_DIR) + parser.add_argument( + "--onnx-model", + type=Path, + help="score a fixed-shape ONNX artifact instead of the PyTorch checkpoint", + ) parser.add_argument("--calibration", type=Path, default=DEFAULT_CALIBRATION) parser.add_argument("--test", type=Path, default=DEFAULT_TEST) parser.add_argument("--fixtures", type=Path, default=DEFAULT_FIXTURES) @@ -279,8 +457,8 @@ def main(argv: Sequence[str] | None = None) -> int: print(f"error: {exc}", file=sys.stderr) return 1 print( - f"Frozen eval: accuracy={report['overall']['accuracy']:.4%}, " - f"hard={report['hardSlice']['accuracy']:.4%}, " + f"Frozen eval: scored={report['scoredClassification']['accuracy']:.4%}, " + f"scored-hard={report['scoredHardSlice']['accuracy']:.4%}, " f"p95={report['latencyMilliseconds']['p95']:.2f} ms." ) if not args.no_gate and not all(report["gates"].values()): diff --git a/export.py b/export.py new file mode 100644 index 0000000..2c68351 --- /dev/null +++ b/export.py @@ -0,0 +1,334 @@ +#!/usr/bin/env python3 +"""Export purpose-lite to fixed-shape fp16 and int8-QDQ ONNX artifacts.""" + +from __future__ import annotations + +import argparse +import hashlib +import shutil +import sys +import tempfile +from pathlib import Path +from typing import Any, Sequence + +import numpy as np + +from purpose_data import DataError, load_jsonl, normalize_prompt, write_json +from train import ( + HEAD_TOKENS, + MAX_LENGTH, + TAIL_TOKENS, + encode_fixed_shape, +) + + +SCRIPT_DIR = Path(__file__).resolve().parent +DEFAULT_MODEL_DIR = SCRIPT_DIR / "outputs" / "purpose-lite-v1" / "model" +DEFAULT_CALIBRATION = SCRIPT_DIR / "outputs" / "purpose-lite-v1" / "calibration.json" +DEFAULT_VALIDATION = SCRIPT_DIR / ".artifacts" / "dataset-v1" / "validation.jsonl" +DEFAULT_OUTPUT_DIR = SCRIPT_DIR / "outputs" / "purpose-lite-v1" / "export" +MODEL_VERSION = "purpose-lite-v1" +SHIPPING_BUDGET_BYTES = 25 * 1024 * 1024 +GOLDEN_PROMPTS = ( + "Fix the typo in the README.", + " Cafe\u0301\tdeploy\nnow ", + ( + "EXPLAIN ANALYZE shows a sequential scan before the incident notes. " + + "context " * 180 + + "Find the root cause and explain which query plan evidence proves it." + ), + "レビューだけして、コードは変更しないでください。", +) + + +def _sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def _fixed_shape_inputs(model: Any) -> list[str]: + names = ["input_ids", "attention_mask"] + if getattr(model.config, "type_vocab_size", 0) > 1: + names.append("token_type_ids") + return names + + +def _validate_graph(path: Path, input_names: Sequence[str]) -> dict[str, Any]: + import onnx + + model = onnx.load(path) + onnx.checker.check_model(model) + opset = max( + item.version for item in model.opset_import if item.domain in ("", "ai.onnx") + ) + if opset < 17: + raise DataError(f"{path}: ONNX opset {opset} is below 17") + shapes = {} + for value in model.graph.input: + dimensions = [item.dim_value for item in value.type.tensor_type.shape.dim] + shapes[value.name] = dimensions + expected = {name: [1, MAX_LENGTH] for name in input_names} + if shapes != expected: + raise DataError(f"{path}: expected fixed inputs {expected}, got {shapes}") + custom_domains = sorted( + { + node.domain + for node in model.graph.node + if node.domain not in ("", "ai.onnx") + } + ) + if custom_domains: + raise DataError(f"{path}: custom ONNX op domains are not allowed: {custom_domains}") + return { + "opset": opset, + "inputs": shapes, + "nodes": len(model.graph.node), + "operators": sorted({node.op_type for node in model.graph.node}), + } + + +def _write_tokenizer_contract( + output_dir: Path, model_dir: Path, tokenizer: Any, torch: Any +) -> dict[str, Any]: + vocab_destination = output_dir / "vocab.txt" + vocabulary = tokenizer.get_vocab() + ordered_vocabulary = sorted(vocabulary.items(), key=lambda item: item[1]) + if [token_id for _, token_id in ordered_vocabulary] != list( + range(len(ordered_vocabulary)) + ): + raise DataError("tokenizer vocabulary IDs are not contiguous") + vocab_destination.write_text( + "".join(f"{token}\n" for token, _ in ordered_vocabulary), + encoding="utf-8", + ) + tokenizer_json_source = model_dir / "tokenizer.json" + if not tokenizer_json_source.exists(): + raise DataError(f"{tokenizer_json_source}: tokenizer JSON is missing") + tokenizer_json_destination = output_dir / "tokenizer.json" + shutil.copy2(tokenizer_json_source, tokenizer_json_destination) + + golden_values = [] + for prompt in GOLDEN_PROMPTS: + encoded = encode_fixed_shape(tokenizer, [prompt], torch) + golden_values.append( + { + "prompt": prompt, + "normalizedPrompt": normalize_prompt(prompt), + "inputIds": encoded["input_ids"][0].tolist(), + "attentionMask": encoded["attention_mask"][0].tolist(), + "tokenTypeIds": encoded.get( + "token_type_ids", torch.zeros_like(encoded["input_ids"]) + )[0].tolist(), + } + ) + write_json(output_dir / "tokenizer-goldens.json", golden_values) + contract = { + "schemaVersion": 1, + "modelVersion": MODEL_VERSION, + "tokenizer": { + "family": "BERT WordPiece", + "vocabFile": vocab_destination.name, + "vocabSha256": _sha256(vocab_destination), + "tokenizerJSON": tokenizer_json_destination.name, + "tokenizerJSONSha256": _sha256(tokenizer_json_destination), + "lowercase": bool(getattr(tokenizer, "do_lower_case", True)), + }, + "normalization": ["Unicode NFKC", "collapse Unicode whitespace", "trim"], + "input": { + "shape": [1, MAX_LENGTH], + "padding": "right", + "longPromptStrategy": "BERT sentence pair: head and tail", + "headTokens": HEAD_TOKENS, + "tailTokens": TAIL_TOKENS, + "specialTokenLayout": "[CLS] head [SEP] tail [SEP]", + }, + "specialTokenIds": { + "padding": tokenizer.pad_token_id, + "unknown": tokenizer.unk_token_id, + "classification": tokenizer.cls_token_id, + "separator": tokenizer.sep_token_id, + }, + "goldens": "tokenizer-goldens.json", + } + write_json(output_dir / "tokenizer-spec.json", contract) + return contract + + +def export(args: argparse.Namespace) -> dict[str, Any]: + try: + import onnx + import torch + from onnxconverter_common import float16 + from onnxruntime.quantization import ( + CalibrationDataReader, + QuantFormat, + QuantType, + quantize_static, + ) + from transformers import AutoModelForSequenceClassification, AutoTokenizer + except ImportError as exc: + raise DataError( + "export dependencies are missing; install requirements.txt" + ) from exc + + output_dir: Path = args.output_dir + output_dir.mkdir(parents=True, exist_ok=True) + tokenizer = AutoTokenizer.from_pretrained(args.model_dir, local_files_only=True) + model = AutoModelForSequenceClassification.from_pretrained( + args.model_dir, local_files_only=True + ) + model.eval() + input_names = _fixed_shape_inputs(model) + example = encode_fixed_shape(tokenizer, ["Plan a safe cache migration."], torch) + + class LogitsModel(torch.nn.Module): + def __init__(self, inner: Any) -> None: + super().__init__() + self.inner = inner + + def forward(self, *values: Any) -> Any: + inputs = dict(zip(input_names, values)) + return self.inner(**inputs).logits + + fp16_path = output_dir / f"{MODEL_VERSION}-fp16.onnx" + int8_path = output_dir / f"{MODEL_VERSION}-int8-qdq.onnx" + with tempfile.TemporaryDirectory() as temporary: + fp32_path = Path(temporary) / f"{MODEL_VERSION}-fp32.onnx" + torch.onnx.export( + LogitsModel(model), + tuple(example[name] for name in input_names), + fp32_path, + input_names=input_names, + output_names=["logits"], + opset_version=args.opset, + do_constant_folding=True, + dynamo=False, + ) + fp32_model = onnx.load(fp32_path) + fp16_model = float16.convert_float_to_float16( + fp32_model, + keep_io_types=True, + disable_shape_infer=False, + ) + onnx.save(fp16_model, fp16_path) + + validation = load_jsonl(args.validation) + + class Reader(CalibrationDataReader): + def __init__(self) -> None: + self.index = 0 + self.samples = validation[: args.calibration_records] + + def get_next(self) -> dict[str, np.ndarray] | None: + if self.index >= len(self.samples): + return None + record = self.samples[self.index] + self.index += 1 + encoded = encode_fixed_shape(tokenizer, [record["prompt"]], torch) + return { + name: encoded[name].numpy().astype(np.int64, copy=False) + for name in input_names + } + + quantize_static( + fp32_path, + int8_path, + Reader(), + quant_format=QuantFormat.QDQ, + activation_type=QuantType.QUInt8, + weight_type=QuantType.QInt8, + per_channel=True, + # Gather is essential: MiniLM's 30k x 384 embedding table is more than half + # the checkpoint. Quantizing only MatMul/Gemm leaves a ~47 MB fp32 table and + # cannot meet the 25 MB parity-floor artifact budget. + op_types_to_quantize=["Gather", "MatMul", "Gemm"], + extra_options={ + "ActivationSymmetric": False, + "WeightSymmetric": True, + }, + ) + + graph_reports = { + "fp16": _validate_graph(fp16_path, input_names), + "int8QDQ": _validate_graph(int8_path, input_names), + } + int8_size = int8_path.stat().st_size + if int8_size > args.shipping_budget_bytes: + raise DataError( + f"{int8_path}: {int8_size} bytes exceeds the " + f"{args.shipping_budget_bytes}-byte shipping budget" + ) + + tokenizer_contract = _write_tokenizer_contract( + output_dir, args.model_dir, tokenizer, torch + ) + calibration_destination = output_dir / "calibration.json" + shutil.copy2(args.calibration, calibration_destination) + artifacts = {} + for name, path in (("fp16", fp16_path), ("int8QDQ", int8_path)): + artifacts[name] = { + "path": path.name, + "bytes": path.stat().st_size, + "sha256": _sha256(path), + } + report = { + "schemaVersion": 1, + "modelVersion": MODEL_VERSION, + "sourceModel": str(args.model_dir), + "opset": args.opset, + "fixedInputShape": [1, MAX_LENGTH], + "inputNames": input_names, + "calibrationRecords": min(args.calibration_records, len(validation)), + "shippingArtifact": "int8QDQ", + "shippingBudgetBytes": args.shipping_budget_bytes, + "shippingBudgetPassed": int8_size <= args.shipping_budget_bytes, + "artifacts": artifacts, + "graphs": graph_reports, + "tokenizerSpecSha256": _sha256(output_dir / "tokenizer-spec.json"), + "tokenizerContract": tokenizer_contract["input"], + "calibrationSha256": _sha256(calibration_destination), + } + write_json(output_dir / "export-metrics.json", report) + return report + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--model-dir", type=Path, default=DEFAULT_MODEL_DIR) + parser.add_argument("--calibration", type=Path, default=DEFAULT_CALIBRATION) + parser.add_argument("--validation", type=Path, default=DEFAULT_VALIDATION) + parser.add_argument("--output-dir", type=Path, default=DEFAULT_OUTPUT_DIR) + parser.add_argument("--opset", type=int, default=17) + parser.add_argument("--calibration-records", type=int, default=256) + parser.add_argument( + "--shipping-budget-bytes", type=int, default=SHIPPING_BUDGET_BYTES + ) + return parser + + +def main(argv: Sequence[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + if ( + args.opset < 17 + or args.calibration_records <= 0 + or args.shipping_budget_bytes <= 0 + ): + parser.error("opset must be >=17 and record/budget values must be positive") + try: + report = export(args) + except (DataError, OSError, ValueError) as exc: + print(f"error: {exc}", file=sys.stderr) + return 1 + print( + f"Exported {MODEL_VERSION}: fp16={report['artifacts']['fp16']['bytes']} bytes, " + f"int8-QDQ={report['artifacts']['int8QDQ']['bytes']} bytes." + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/prepare_data.py b/prepare_data.py index 55ddcb1..08187d0 100644 --- a/prepare_data.py +++ b/prepare_data.py @@ -13,10 +13,12 @@ from typing import Any, Sequence from purpose_data import ( HARD_SLICES, LABELS, + CurationResult, DataError, SourceRecord, curate_records, distribution, + exclude_reviewed_duplicates, file_sha256, jsonl_bytes, load_classifiable_fixtures, @@ -42,6 +44,7 @@ DEFAULT_FIXTURES = ( DEFAULT_OUTPUT_DIR = SCRIPT_DIR / ".artifacts" / "dataset-v1" DEFAULT_FROZEN_TEST = DATA_DIR / "frozen-test-v1.jsonl" DEFAULT_SPLIT_MANIFEST = DATA_DIR / "dataset-v1-manifest.json" +DEFAULT_CURATION_REVIEW = DATA_DIR / "curation-review-v1.json" DATASET_VERSION = "purpose-dataset-v1" DEFAULT_SEED = 0xC1A551F1 @@ -85,6 +88,21 @@ def _sha256_bytes(value: bytes) -> str: return hashlib.sha256(value).hexdigest() +def load_reviewed_semantic_exclusions(path: Path) -> list[dict[str, Any]]: + try: + value = json.loads(path.read_text(encoding="utf-8")) + review = value["semanticDuplicateReview"] + decisions = review["excluded"] + status = review["status"] + except (OSError, UnicodeError, json.JSONDecodeError, KeyError, TypeError) as exc: + raise DataError(f"{path}: cannot read semantic exclusion review: {exc}") from exc + if status != "complete": + raise DataError(f"{path}: semantic duplicate review is not complete") + if not isinstance(decisions, list): + raise DataError(f"{path}: semantic excluded must be an array") + return decisions + + def _records(values: Sequence[SourceRecord]) -> list[dict[str, Any]]: return [record.value for record in values] @@ -102,8 +120,9 @@ def _build_manifest( output_hashes: dict[str, str], seed: int, near_duplicate_threshold: float, + curation_review_path: Path | None, ) -> dict[str, Any]: - return { + manifest = { "schemaVersion": 1, "datasetVersion": DATASET_VERSION, "seed": seed, @@ -159,6 +178,10 @@ def _build_manifest( }, }, } + if curation_review_path is not None: + manifest["curation"]["reviewPath"] = _relative(curation_review_path) + manifest["curation"]["reviewSha256"] = file_sha256(curation_review_path) + return manifest def prepare( @@ -171,6 +194,7 @@ def prepare( refresh_frozen_test: bool, seed: int, near_duplicate_threshold: float, + curation_review_path: Path | None = None, ) -> dict[str, Any]: records = load_sources(sources) fixtures = load_classifiable_fixtures(fixtures_path) @@ -178,11 +202,23 @@ def prepare( if classifiable_fixture_count != len(fixtures): raise DataError("fixture accounting mismatch") - curated = curate_records( + lexical_curation = curate_records( records, fixtures, near_duplicate_threshold=near_duplicate_threshold, ) + reviewed_curation = ( + exclude_reviewed_duplicates( + lexical_curation.records, + load_reviewed_semantic_exclusions(curation_review_path), + ) + if curation_review_path is not None + else CurationResult(records=lexical_curation.records, duplicates=[]) + ) + curated = CurationResult( + records=reviewed_curation.records, + duplicates=lexical_curation.duplicates + reviewed_curation.duplicates, + ) splits = split_records( curated.records, fixture_count=classifiable_fixture_count, @@ -246,6 +282,7 @@ def prepare( output_hashes=output_hashes, seed=seed, near_duplicate_threshold=near_duplicate_threshold, + curation_review_path=curation_review_path, ) # The frozen path can be overridden in tests or experiments. manifest["frozenEval"]["syntheticPath"] = _relative(frozen_test_path) @@ -285,6 +322,12 @@ def build_parser() -> argparse.ArgumentParser: ) parser.add_argument("--seed", type=int, default=DEFAULT_SEED) parser.add_argument("--near-duplicate-threshold", type=float, default=0.92) + parser.add_argument( + "--curation-review", + type=Path, + default=DEFAULT_CURATION_REVIEW, + help="completed semantic duplicate review applied after lexical deduplication", + ) return parser @@ -305,6 +348,7 @@ def main(argv: Sequence[str] | None = None) -> int: refresh_frozen_test=args.refresh_frozen_test, seed=args.seed, near_duplicate_threshold=args.near_duplicate_threshold, + curation_review_path=args.curation_review.resolve(), ) except (DataError, OSError, UnicodeError, json.JSONDecodeError) as exc: print(f"error: {exc}", file=sys.stderr) diff --git a/purpose_data.py b/purpose_data.py index 1e73bb4..8015a2d 100644 --- a/purpose_data.py +++ b/purpose_data.py @@ -364,6 +364,73 @@ def curate_records( return CurationResult(records=kept, duplicates=duplicates) +def exclude_reviewed_duplicates( + records: Sequence[SourceRecord], + decisions: Sequence[dict[str, Any]], +) -> CurationResult: + """Apply human-reviewed semantic exclusions, failing closed on corpus drift.""" + + by_hash = {prompt_hash(record.value["prompt"]): record for record in records} + if len(by_hash) != len(records): + raise DataError("reviewed exclusions require an exact-deduplicated population") + + drops: dict[str, tuple[str, float]] = {} + for index, decision in enumerate(decisions, 1): + try: + dropped_hash = decision["droppedPromptHash"] + matched_hash = decision["matchedPromptHash"] + similarity = float(decision["similarity"]) + except (KeyError, TypeError, ValueError) as exc: + raise DataError( + f"semantic exclusion {index}: invalid decision fields" + ) from exc + if ( + not isinstance(dropped_hash, str) + or not isinstance(matched_hash, str) + or len(dropped_hash) != 64 + or len(matched_hash) != 64 + or not 0.0 <= similarity <= 1.0 + ): + raise DataError(f"semantic exclusion {index}: invalid hashes/similarity") + if dropped_hash == matched_hash: + raise DataError(f"semantic exclusion {index}: cannot match itself") + if dropped_hash in drops: + raise DataError(f"semantic exclusion {index}: duplicate dropped hash") + drops[dropped_hash] = (matched_hash, similarity) + + missing = sorted((set(drops) | {item[0] for item in drops.values()}) - set(by_hash)) + if missing: + raise DataError( + "reviewed semantic exclusion no longer matches the curated corpus: " + + ", ".join(missing) + ) + + kept = [] + duplicates = [] + for record in records: + dropped_hash = prompt_hash(record.value["prompt"]) + decision = drops.get(dropped_hash) + if decision is None: + kept.append(record) + continue + matched_hash, similarity = decision + matched = by_hash[matched_hash] + if record.value["purpose"] != matched.value["purpose"]: + raise DataError( + "reviewed semantic duplicate labels no longer agree: " + f"{dropped_hash} vs {matched_hash}" + ) + duplicates.append( + Duplicate( + dropped=record, + matched_prompt_hash=matched_hash, + kind="semantic-reviewed", + similarity=similarity, + ) + ) + return CurationResult(records=kept, duplicates=duplicates) + + def _stable_digest(seed: int, prompt: str) -> str: material = f"{seed}\0{normalized_key(prompt)}".encode("utf-8") return hashlib.sha256(material).hexdigest() diff --git a/requirements-base.txt b/requirements-base.txt index 5a55444..ccc1ea5 100644 --- a/requirements-base.txt +++ b/requirements-base.txt @@ -1,2 +1,5 @@ numpy==2.5.1 +onnx==1.22.0 +onnxconverter-common==1.16.0 +onnxruntime==1.28.0 transformers==5.14.1 diff --git a/tests/test_audit_data.py b/tests/test_audit_data.py new file mode 100644 index 0000000..fb6ec79 --- /dev/null +++ b/tests/test_audit_data.py @@ -0,0 +1,92 @@ +import sys +import tempfile +import unittest +from pathlib import Path + +import numpy as np + + +MODULE_DIR = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(MODULE_DIR)) + +import audit_data +from purpose_data import SourceRecord + + +def record(index, purpose, slice_name="core", language="en"): + return SourceRecord( + value={ + "prompt": f"prompt {index}", + "purpose": purpose, + "secondary": None, + "mixed": False, + "difficulty": 0.5, + "slice": slice_name, + "lang": language, + }, + source=Path("source.jsonl"), + line=index + 1, + ) + + +class ReviewSampleTests(unittest.TestCase): + def test_sample_is_exact_and_deterministic(self): + records = [ + record(index, audit_data.LABELS[index % len(audit_data.LABELS)]) + for index in range(101) + ] + first = audit_data.stratified_review_sample(records, fraction=0.1, seed=42) + second = audit_data.stratified_review_sample(records, fraction=0.1, seed=42) + self.assertEqual(10, len(first)) + self.assertEqual( + [item.value["prompt"] for item in first], + [item.value["prompt"] for item in second], + ) + + def test_review_csv_has_blank_reviewer_fields(self): + records = [record(0, "planning")] + with tempfile.TemporaryDirectory() as temporary: + path = Path(temporary) / "review.csv" + audit_data.write_review_csv(path, records) + text = path.read_text(encoding="utf-8") + self.assertIn("reviewedPurpose", text) + self.assertIn("prompt 0", text) + + +class SemanticCandidateTests(unittest.TestCase): + def test_threshold_depends_on_label_agreement(self): + embeddings = np.asarray( + [ + [1.0, 0.0], + [0.98, 0.2], + [0.98, -0.2], + ], + dtype=np.float32, + ) + candidates = audit_data.semantic_candidates( + embeddings, + ["planning", "planning", "writing"], + same_label_threshold=0.97, + cross_label_threshold=0.99, + neighbors=2, + block_size=2, + ) + pairs = {(item.left, item.right) for item in candidates} + self.assertIn((0, 1), pairs) + self.assertNotIn((0, 2), pairs) + + def test_candidate_pairs_are_deduplicated(self): + embeddings = np.asarray([[1.0, 0.0], [1.0, 0.0]], dtype=np.float32) + candidates = audit_data.semantic_candidates( + embeddings, + ["review", "review"], + same_label_threshold=0.9, + cross_label_threshold=0.9, + neighbors=1, + block_size=1, + ) + self.assertEqual(1, len(candidates)) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_eval.py b/tests/test_eval.py new file mode 100644 index 0000000..f7a12b2 --- /dev/null +++ b/tests/test_eval.py @@ -0,0 +1,27 @@ +import sys +import unittest +from pathlib import Path + + +MODULE_DIR = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(MODULE_DIR)) + +import eval as purpose_eval + + +class TierDriftTests(unittest.TestCase): + def test_current_routing_matrix_bounds_every_label_pair(self): + records = [{"prompt": f"prompt {index}"} for index in range(8 * 8)] + actual = [] + predicted = [] + for expected in range(8): + for got in range(8): + actual.append(expected) + predicted.append(got) + report = purpose_eval.routing_tier_drift(records, actual, predicted) + self.assertTrue(report["passed"]) + self.assertLessEqual(report["maximumTierDrift"], 1) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_purpose_data.py b/tests/test_purpose_data.py index 2f5bf37..e3cbdaf 100644 --- a/tests/test_purpose_data.py +++ b/tests/test_purpose_data.py @@ -84,6 +84,26 @@ class CurationTests(unittest.TestCase): near_duplicate_threshold=0.92, ) + def test_reviewed_semantic_exclusions_fail_closed_on_corpus_drift(self): + first = source(example(0, prompt="Plan the cache migration", purpose="planning"), 1) + second = source( + example(1, prompt="Propose a cache migration plan", purpose="planning"), 2 + ) + decision = { + "droppedPromptHash": purpose_data.prompt_hash(second.value["prompt"]), + "matchedPromptHash": purpose_data.prompt_hash(first.value["prompt"]), + "similarity": 0.98, + } + result = purpose_data.exclude_reviewed_duplicates( + [first, second], [decision] + ) + self.assertEqual([first], result.records) + self.assertEqual("semantic-reviewed", result.duplicates[0].kind) + with self.assertRaisesRegex( + purpose_data.DataError, "no longer matches the curated corpus" + ): + purpose_data.exclude_reviewed_duplicates([first], [decision]) + class SplitTests(unittest.TestCase): def test_split_is_deterministic_stratified_and_keeps_vague_out_of_train(self): diff --git a/tests/test_train.py b/tests/test_train.py index daee3bc..73665a9 100644 --- a/tests/test_train.py +++ b/tests/test_train.py @@ -2,6 +2,8 @@ import sys import unittest from pathlib import Path +import torch + MODULE_DIR = Path(__file__).resolve().parents[1] sys.path.insert(0, str(MODULE_DIR)) @@ -42,5 +44,45 @@ class MetricsTests(unittest.TestCase): ) +class FixedShapeTokenizerTests(unittest.TestCase): + class Tokenizer: + pad_token_id = 0 + cls_token_id = 1 + sep_token_id = 2 + padding_side = "right" + model_input_names = ["input_ids", "attention_mask", "token_type_ids"] + + def __call__(self, texts, **_): + return { + "input_ids": [ + list(range(10, 10 + int(text.split()[-1]))) for text in texts + ] + } + + @staticmethod + def num_special_tokens_to_add(pair=False): + return 3 if pair else 2 + + def test_long_input_keeps_head_and_tail_in_fixed_pair_shape(self): + encoded = train.encode_fixed_shape( + self.Tokenizer(), ["tokens 200"], torch + ) + self.assertEqual((1, train.MAX_LENGTH), tuple(encoded["input_ids"].shape)) + row = encoded["input_ids"][0].tolist() + self.assertEqual(list(range(10, 10 + train.HEAD_TOKENS)), row[1:64]) + self.assertEqual(2, row[64]) + self.assertEqual( + list(range(10 + 200 - train.TAIL_TOKENS, 10 + 200)), + row[65:127], + ) + self.assertEqual(2, row[127]) + self.assertEqual(1, encoded["token_type_ids"][0, 65].item()) + + def test_short_input_is_padded_as_one_sequence(self): + encoded = train.encode_fixed_shape(self.Tokenizer(), ["tokens 3"], torch) + self.assertEqual([1, 10, 11, 12, 2], encoded["input_ids"][0, :5].tolist()) + self.assertEqual(0, encoded["attention_mask"][0, 5].item()) + + if __name__ == "__main__": unittest.main() diff --git a/train.py b/train.py index fca6686..d6ab54f 100644 --- a/train.py +++ b/train.py @@ -22,12 +22,95 @@ DEFAULT_MODEL = "sentence-transformers/all-MiniLM-L6-v2" # Reproducibility requires a model commit, not a mutable `main` branch. DEFAULT_MODEL_REVISION = "1110a243fdf4706b3f48f1d95db1a4f5529b4d41" MAX_LENGTH = 128 +HEAD_TAIL_SPECIAL_TOKENS = 3 +HEAD_TOKENS = (MAX_LENGTH - HEAD_TAIL_SPECIAL_TOKENS + 1) // 2 +TAIL_TOKENS = MAX_LENGTH - HEAD_TAIL_SPECIAL_TOKENS - HEAD_TOKENS def prepare_text(prompt: str) -> str: return normalize_prompt(prompt) +def encode_fixed_shape( + tokenizer: Any, + texts: Sequence[str], + torch: Any, +) -> dict[str, Any]: + """Tokenize to 1x128 while retaining both context and a tail-buried request. + + Pasted logs and stack traces frequently put the actual ask after the context. Plain + right truncation made generated boundary examples identical even when their final + request — and therefore their label — differed. Long inputs use BERT's sentence-pair + framing: [CLS] first 63 content tokens [SEP] last 62 content tokens [SEP]. + """ + + normalized = [prepare_text(text) for text in texts] + raw = tokenizer( + normalized, + add_special_tokens=False, + padding=False, + truncation=False, + return_attention_mask=False, + return_token_type_ids=False, + verbose=False, + ) + if not isinstance(raw.get("input_ids"), list): + raise DataError("tokenizer did not return input_ids") + if tokenizer.pad_token_id is None: + raise DataError("purpose-lite tokenizer must define a padding token") + if tokenizer.cls_token_id is None or tokenizer.sep_token_id is None: + raise DataError("purpose-lite tokenizer must define BERT CLS and SEP tokens") + if tokenizer.padding_side != "right": + raise DataError("purpose-lite tokenizer must use right padding") + + input_rows: list[list[int]] = [] + mask_rows: list[list[int]] = [] + type_rows: list[list[int]] = [] + include_token_types = "token_type_ids" in tokenizer.model_input_names + single_budget = MAX_LENGTH - tokenizer.num_special_tokens_to_add(pair=False) + pair_budget = MAX_LENGTH - tokenizer.num_special_tokens_to_add(pair=True) + if pair_budget != HEAD_TOKENS + TAIL_TOKENS: + raise DataError( + "purpose-lite tokenizer special-token layout changed; expected three " + "tokens for head-tail inputs" + ) + + for content in raw["input_ids"]: + if len(content) <= single_budget: + first = content + second = None + else: + first = content[:HEAD_TOKENS] + second = content[-TAIL_TOKENS:] + if second is None: + input_ids = [tokenizer.cls_token_id] + first + [tokenizer.sep_token_id] + token_types = [0] * len(input_ids) + else: + input_ids = ( + [tokenizer.cls_token_id] + + first + + [tokenizer.sep_token_id] + + second + + [tokenizer.sep_token_id] + ) + token_types = [0] * (len(first) + 2) + [1] * (len(second) + 1) + if len(input_ids) > MAX_LENGTH: + raise DataError("fixed-shape tokenizer exceeded its 128-token contract") + padding = MAX_LENGTH - len(input_ids) + input_rows.append(input_ids + [tokenizer.pad_token_id] * padding) + mask_rows.append([1] * len(input_ids) + [0] * padding) + if include_token_types: + type_rows.append(token_types + [0] * padding) + + encoded = { + "input_ids": torch.tensor(input_rows, dtype=torch.long), + "attention_mask": torch.tensor(mask_rows, dtype=torch.long), + } + if include_token_types: + encoded["token_type_ids"] = torch.tensor(type_rows, dtype=torch.long) + return encoded + + def classification_metrics( actual: Sequence[int], predicted: Sequence[int] ) -> dict[str, Any]: @@ -267,6 +350,11 @@ def train(args: argparse.Namespace) -> dict[str, Any]: config.purpose_classifier_version = "purpose-lite-v1" config.purpose_classifier_max_length = MAX_LENGTH config.purpose_classifier_fixed_shape = [1, MAX_LENGTH] + config.purpose_classifier_truncation = { + "strategy": "head-tail-pair", + "headTokens": HEAD_TOKENS, + "tailTokens": TAIL_TOKENS, + } model.to(device) class PromptDataset(Dataset): @@ -282,13 +370,7 @@ def train(args: argparse.Namespace) -> dict[str, Any]: def collate(items: Sequence[tuple[str, int]]) -> dict[str, Any]: texts, labels = zip(*items) - encoded = tokenizer( - list(texts), - padding="max_length", - truncation=True, - max_length=MAX_LENGTH, - return_tensors="pt", - ) + encoded = encode_fixed_shape(tokenizer, list(texts), torch) encoded["labels"] = torch.tensor(labels, dtype=torch.long) return encoded @@ -311,6 +393,10 @@ def train(args: argparse.Namespace) -> dict[str, Any]: num_workers=args.workers, pin_memory=device.type == "cuda", ) + validation_scorable = torch.tensor( + [record.get("slice") != "vague-eval" for record in validation_records], + dtype=torch.bool, + ) optimizer = torch.optim.AdamW( model.parameters(), lr=args.learning_rate, weight_decay=args.weight_decay @@ -350,7 +436,15 @@ def train(args: argparse.Namespace) -> dict[str, Any]: logits, labels = _evaluate(torch, model, validation_loader, device) predictions = logits.argmax(dim=-1).tolist() - metrics = classification_metrics(labels.tolist(), predictions) + scored_labels = labels[validation_scorable].tolist() + scored_predictions = [ + prediction + for prediction, scorable in zip( + predictions, validation_scorable.tolist() + ) + if scorable + ] + metrics = classification_metrics(scored_labels, scored_predictions) metrics["epoch"] = epoch metrics["meanTrainingLoss"] = running_loss / len(train_loader) history.append(metrics) @@ -367,15 +461,23 @@ def train(args: argparse.Namespace) -> dict[str, Any]: model = AutoModelForSequenceClassification.from_pretrained(best_dir).to(device) logits, labels = _evaluate(torch, model, validation_loader, device) - temperature = _fit_temperature(torch, logits, labels) + temperature = _fit_temperature( + torch, + logits[validation_scorable], + labels[validation_scorable], + ) calibrated = torch.softmax(logits / temperature, dim=-1) top = torch.topk(calibrated, k=2, dim=-1) top_probabilities = top.values[:, 0].tolist() margins = (top.values[:, 0] - top.values[:, 1]).tolist() predictions = top.indices[:, 0].tolist() correct = [ - prediction == actual - for prediction, actual in zip(predictions, labels.tolist()) + prediction == actual and scorable + for prediction, actual, scorable in zip( + predictions, + labels.tolist(), + validation_scorable.tolist(), + ) ] thresholds = choose_confidence_thresholds( top_probabilities, @@ -397,12 +499,30 @@ def train(args: argparse.Namespace) -> dict[str, Any]: "baseModel": args.model, "baseModelRevision": args.model_revision, "fixedInputShape": [1, MAX_LENGTH], + "truncation": { + "strategy": "head-tail-pair", + "headTokens": HEAD_TOKENS, + "tailTokens": TAIL_TOKENS, + }, "device": str(device), "trainingSeconds": time.perf_counter() - started, "trainRecords": len(train_records), "validationRecords": len(validation_records), + "scoredValidationRecords": int(validation_scorable.sum().item()), + "vagueAbstentionValidationRecords": int( + (~validation_scorable).sum().item() + ), "bestValidationAccuracy": best_accuracy, - "bestValidation": classification_metrics(labels.tolist(), predictions), + "bestValidation": classification_metrics( + labels[validation_scorable].tolist(), + [ + prediction + for prediction, scorable in zip( + predictions, validation_scorable.tolist() + ) + if scorable + ], + ), "history": history, "calibration": calibration, }