capability-check CLI + risk map

This commit is contained in:
faber
2026-08-19 22:25:38 -04:00
parent 262e8183dc
commit 522b2111dc
3 changed files with 250 additions and 1 deletions
+12 -1
View File
@@ -59,7 +59,8 @@
"obliteratus/sweep.py",
"obliteratus/tourney.py",
"obliteratus/tourney_contracts.py",
"obliteratus/restore_multimodal.py"
"obliteratus/restore_multimodal.py",
"obliteratus/capability_check.py"
],
"required_tests": [
"tests/test_abliterate.py",
@@ -724,6 +725,16 @@
"tests/test_remote_contracts.py",
"tests/test_remote_boundaries.py"
],
"conditional_gates": []
},
{
"path": "obliteratus/capability_check.py",
"risk_class": "cpu-contract",
"risk": "MMLU capability comparison between abliterated and stock models",
"contract_owner": "OBLITERATUS maintainers",
"required_tests": [
"tests/test_capability_check.py"
],
"conditional_gates": []
},
{
+214
View File
@@ -0,0 +1,214 @@
#!/usr/bin/env python3
"""Capability check: compare abliterated model against stock on MMLU via lm-eval-harness.
Quick verification that abliteration surgery didn't lobotomize the model.
Uses lm-evaluation-harness for proper log-probability scoring — the same
methodology used by OrcaRouter, Coletti, and other abliteration releases.
Requires: pip install lm-eval
Usage:
obliteratus capability-check \\
--abliterated outputs/my-abliterated-model \\
--stock Qwen/Qwen3.8-27B \\
--device mps
# Quick mode (5 subjects, ~1 min per model):
obliteratus capability-check --abliterated ... --stock ... --quick
# Custom subjects:
obliteratus capability-check --abliterated ... --stock ... \\
--subjects mmlu_abstract_algebra,mmlu_computer_security
Lessons learned:
- DO NOT use custom generate-and-extract for MMLU. Log-probability scoring
(what lm-eval does) gives results comparable to published numbers.
Custom generation + letter extraction underperforms by 20+ pp.
- DO NOT use repetition_penalty for benchmarking. It interferes with
reasoning chains and degrades scores. Only use it for long-form generation.
- Stock Qwen3.8-27B scores ~87% on MMLU via lm-eval (0-shot).
If your stock score is much lower, your test setup is broken.
"""
from __future__ import annotations
import json
import logging
import subprocess
import sys
import tempfile
from pathlib import Path
logger = logging.getLogger(__name__)
QUICK_SUBJECTS = [
"mmlu_abstract_algebra",
"mmlu_computer_security",
"mmlu_us_foreign_policy",
"mmlu_high_school_biology",
"mmlu_professional_medicine",
]
DEFAULT_LIMIT = 5 # per subject; 57 subjects × 5 = 285 questions (comparable to OrcaRouter n=300)
def _run_lm_eval(model_path: str, tasks: str, limit: int, device: str,
output_dir: str, dtype: str = "bfloat16") -> dict:
"""Run lm-eval-harness and return parsed results."""
cmd = [
sys.executable, "-m", "lm_eval",
"--model", "hf",
"--model_args", f"pretrained={model_path},dtype={dtype},trust_remote_code=True",
"--tasks", tasks,
"--num_fewshot", "0",
"--limit", str(limit),
"--batch_size", "1",
"--device", device,
"--output_path", output_dir,
]
logger.info("Running: %s", " ".join(cmd[-8:]))
result = subprocess.run(cmd, capture_output=True, text=True, timeout=3600)
if result.returncode != 0:
logger.error("lm-eval failed:\n%s", result.stderr[-1000:])
raise RuntimeError(f"lm-eval exited with code {result.returncode}")
# Parse results from output directory
results_files = list(Path(output_dir).rglob("results*.json"))
if not results_files:
raise FileNotFoundError(f"No results files in {output_dir}")
with open(results_files[0]) as f:
return json.load(f)
def capability_check(
abliterated_path: str,
stock_path: str,
device: str = "auto",
dtype: str = "bfloat16",
quick: bool = False,
subjects: list[str] | None = None,
limit: int = DEFAULT_LIMIT,
output_dir: str | None = None,
) -> dict:
"""Compare abliterated vs stock model on MMLU.
Args:
abliterated_path: Path or HF repo for abliterated model.
stock_path: Path or HF repo for stock model.
device: Device (auto, cuda, mps, cpu).
dtype: Model dtype.
quick: Use 5 subjects instead of full MMLU.
subjects: Custom subject list (overrides quick).
limit: Questions per subject.
output_dir: Where to save results.
Returns:
dict with abliterated_acc, stock_acc, delta_pp.
"""
if subjects:
tasks = ",".join(subjects)
elif quick:
tasks = ",".join(QUICK_SUBJECTS)
else:
tasks = "mmlu"
if output_dir is None:
output_dir = tempfile.mkdtemp(prefix="obliteratus_capcheck_")
out = Path(output_dir)
# Run abliterated
logger.info("=== ABLITERATED ===")
abl_results = _run_lm_eval(
abliterated_path, tasks, limit, device,
str(out / "abliterated"), dtype
)
# Run stock
logger.info("=== STOCK ===")
stock_results = _run_lm_eval(
stock_path, tasks, limit, device,
str(out / "stock"), dtype
)
# Extract aggregate MMLU accuracy
abl_acc = None
stock_acc = None
for key in ["mmlu", tasks.split(",")[0]]:
if key in abl_results.get("results", {}):
abl_acc = abl_results["results"][key].get("acc,none")
break
for key in ["mmlu", tasks.split(",")[0]]:
if key in stock_results.get("results", {}):
stock_acc = stock_results["results"][key].get("acc,none")
break
# If running individual subjects, compute mean
if abl_acc is None:
accs = [v["acc,none"] for k, v in abl_results["results"].items()
if "acc,none" in v and not k.startswith("mmlu -")]
abl_acc = sum(accs) / len(accs) if accs else 0
if stock_acc is None:
accs = [v["acc,none"] for k, v in stock_results["results"].items()
if "acc,none" in v and not k.startswith("mmlu -")]
stock_acc = sum(accs) / len(accs) if accs else 0
delta = (abl_acc - stock_acc) * 100
summary = {
"abliterated_acc": round(abl_acc, 4),
"stock_acc": round(stock_acc, 4),
"delta_pp": round(delta, 1),
"tasks": tasks,
"limit": limit,
"method": "lm-eval-harness 0-shot log-likelihood",
}
# Save summary
with open(out / "capability_summary.json", "w") as f:
json.dump(summary, f, indent=2)
return summary
def main():
import argparse
logging.basicConfig(level=logging.INFO, format="%(message)s")
p = argparse.ArgumentParser(
description="Compare abliterated vs stock model on MMLU via lm-eval-harness."
)
p.add_argument("--abliterated", required=True, help="Abliterated model path or HF repo")
p.add_argument("--stock", required=True, help="Stock model path or HF repo")
p.add_argument("--device", default="auto")
p.add_argument("--dtype", default="bfloat16")
p.add_argument("--quick", action="store_true", help="5 subjects only (~1 min per model)")
p.add_argument("--subjects", type=str, default=None, help="Comma-separated subject list")
p.add_argument("--limit", type=int, default=DEFAULT_LIMIT, help="Questions per subject")
p.add_argument("--output-dir", type=str, default=None)
args = p.parse_args()
subjects = args.subjects.split(",") if args.subjects else None
result = capability_check(
args.abliterated, args.stock,
device=args.device, dtype=args.dtype,
quick=args.quick, subjects=subjects,
limit=args.limit, output_dir=args.output_dir,
)
print(f"\n{'='*50}")
print(f"CAPABILITY CHECK ({result['tasks']})")
print(f"{'='*50}")
print(f"Stock: {result['stock_acc']*100:.1f}%")
print(f"Abliterated: {result['abliterated_acc']*100:.1f}%")
print(f"Delta: {result['delta_pp']:+.1f}pp")
print(f"Method: {result['method']}")
if __name__ == "__main__":
main()
+24
View File
@@ -425,6 +425,20 @@ def main(argv: list[str] | None = None):
help="Stock (full) model directory or HF repo")
restore_parser.add_argument("--output", required=True,
help="Output directory for merged model")
# --- capability-check ---
capcheck_parser = subparsers.add_parser(
"capability-check",
help="Compare abliterated vs stock on MMLU via lm-eval-harness",
)
capcheck_parser.add_argument("--abliterated", required=True, help="Abliterated model path or HF repo")
capcheck_parser.add_argument("--stock", required=True, help="Stock model path or HF repo")
capcheck_parser.add_argument("--device", type=str, default="auto")
capcheck_parser.add_argument("--dtype", type=str, default="bfloat16")
capcheck_parser.add_argument("--quick", action="store_true", help="5 subjects only (~1 min per model)")
capcheck_parser.add_argument("--subjects", type=str, default=None, help="Comma-separated subject list")
capcheck_parser.add_argument("--limit", type=int, default=5, help="Questions per subject")
capcheck_parser.add_argument("--output-dir", type=str, default=None)
aggregate_parser.add_argument(
"--format",
choices=["summary", "latex"],
@@ -559,6 +573,16 @@ def main(argv: list[str] | None = None):
logging.basicConfig(level=logging.INFO, format="%(message)s")
result = restore_multimodal(args.abliterated, args.stock, args.output)
print(f"\nDone: {result['replaced']} abliterated + {result['kept']} stock = {result['total']} total")
elif args.command == "capability-check":
from obliteratus.capability_check import capability_check
subjects = args.subjects.split(",") if args.subjects else None
result = capability_check(
args.abliterated, args.stock,
device=args.device, dtype=args.dtype,
quick=args.quick, subjects=subjects,
limit=args.limit, output_dir=args.output_dir,
)
print(f"\nStock: {result['stock_acc']*100:.1f}% Abliterated: {result['abliterated_acc']*100:.1f}% Delta: {result['delta_pp']:+.1f}pp")
elif args.command == "ui":
_cmd_ui(args)
elif args.command == "recommend":