diff --git a/ci/test-risk-map.json b/ci/test-risk-map.json index 3218398..37e8a5c 100644 --- a/ci/test-risk-map.json +++ b/ci/test-risk-map.json @@ -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": [] }, { diff --git a/obliteratus/capability_check.py b/obliteratus/capability_check.py new file mode 100644 index 0000000..a966de8 --- /dev/null +++ b/obliteratus/capability_check.py @@ -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() diff --git a/obliteratus/cli.py b/obliteratus/cli.py index 64ca4cd..811aafc 100644 --- a/obliteratus/cli.py +++ b/obliteratus/cli.py @@ -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": diff --git a/tests/test_capability_check.py b/tests/test_capability_check.py new file mode 100644 index 0000000..9ea88f4 --- /dev/null +++ b/tests/test_capability_check.py @@ -0,0 +1,180 @@ +"""Tests for obliteratus.capability_check.""" + +from __future__ import annotations + +import json +from unittest.mock import MagicMock, patch + +import pytest + + +class TestCapabilityCheckImport: + def test_import(self): + from obliteratus.capability_check import capability_check # noqa: F401 + + def test_quick_subjects_defined(self): + from obliteratus.capability_check import QUICK_SUBJECTS + + assert len(QUICK_SUBJECTS) >= 3 + assert all(s.startswith("mmlu_") for s in QUICK_SUBJECTS) + + +class TestRunLmEval: + @patch("obliteratus.capability_check.subprocess.run") + def test_calls_lm_eval(self, mock_run, tmp_path): + from obliteratus.capability_check import _run_lm_eval + + # Create fake results + results_dir = tmp_path / "results" / "fake_model" + results_dir.mkdir(parents=True) + results_file = results_dir / "results_2026.json" + results_file.write_text(json.dumps({ + "results": {"mmlu": {"acc,none": 0.85}}, + })) + + mock_run.return_value = MagicMock(returncode=0, stdout="", stderr="") + + result = _run_lm_eval("fake/model", "mmlu", 5, "cpu", str(tmp_path / "results")) + assert result["results"]["mmlu"]["acc,none"] == 0.85 + mock_run.assert_called_once() + + @patch("obliteratus.capability_check.subprocess.run") + def test_raises_on_failure(self, mock_run, tmp_path): + from obliteratus.capability_check import _run_lm_eval + + mock_run.return_value = MagicMock(returncode=1, stdout="", stderr="error") + + with pytest.raises(RuntimeError, match="lm-eval exited"): + _run_lm_eval("fake/model", "mmlu", 5, "cpu", str(tmp_path)) + + +class TestCapabilityCheck: + @patch("obliteratus.capability_check._run_lm_eval") + def test_computes_delta(self, mock_lm_eval, tmp_path): + from obliteratus.capability_check import capability_check + + mock_lm_eval.side_effect = [ + {"results": {"mmlu": {"acc,none": 0.81}}}, # abliterated + {"results": {"mmlu": {"acc,none": 0.87}}}, # stock + ] + + result = capability_check( + "fake/abliterated", "fake/stock", + device="cpu", output_dir=str(tmp_path), + ) + + assert result["abliterated_acc"] == 0.81 + assert result["stock_acc"] == 0.87 + assert result["delta_pp"] == -6.0 + + @patch("obliteratus.capability_check._run_lm_eval") + def test_quick_mode(self, mock_lm_eval, tmp_path): + from obliteratus.capability_check import QUICK_SUBJECTS, capability_check + + mock_lm_eval.side_effect = [ + {"results": {s: {"acc,none": 0.8} for s in QUICK_SUBJECTS}}, + {"results": {s: {"acc,none": 0.9} for s in QUICK_SUBJECTS}}, + ] + + result = capability_check( + "fake/abliterated", "fake/stock", + device="cpu", quick=True, output_dir=str(tmp_path), + ) + + assert result["tasks"] == ",".join(QUICK_SUBJECTS) + assert abs(result["abliterated_acc"] - 0.8) < 0.01 + + @patch("obliteratus.capability_check._run_lm_eval") + def test_saves_summary(self, mock_lm_eval, tmp_path): + from obliteratus.capability_check import capability_check + + mock_lm_eval.side_effect = [ + {"results": {"mmlu": {"acc,none": 0.85}}}, + {"results": {"mmlu": {"acc,none": 0.87}}}, + ] + + capability_check( + "fake/abliterated", "fake/stock", + device="cpu", output_dir=str(tmp_path), + ) + + summary_path = tmp_path / "capability_summary.json" + assert summary_path.exists() + summary = json.loads(summary_path.read_text()) + assert "delta_pp" in summary + assert summary["method"] == "lm-eval-harness 0-shot log-likelihood" + + @patch("obliteratus.capability_check._run_lm_eval") + def test_custom_subjects(self, mock_lm_eval, tmp_path): + from obliteratus.capability_check import capability_check + + mock_lm_eval.side_effect = [ + {"results": {"mmlu_physics": {"acc,none": 0.75}}}, + {"results": {"mmlu_physics": {"acc,none": 0.80}}}, + ] + + result = capability_check( + "fake/abliterated", "fake/stock", + device="cpu", subjects=["mmlu_physics"], + output_dir=str(tmp_path), + ) + + assert result["tasks"] == "mmlu_physics" + assert result["abliterated_acc"] == 0.75 + + @patch("obliteratus.capability_check._run_lm_eval") + def test_individual_subjects_mean(self, mock_lm_eval, tmp_path): + """Test mean computation when no aggregate mmlu key exists.""" + from obliteratus.capability_check import capability_check + + # Results don't have "mmlu" aggregate key or match tasks.split(",")[0] + mock_lm_eval.side_effect = [ + {"results": {"sub_a": {"acc,none": 0.6}, "sub_b": {"acc,none": 0.8}}}, + {"results": {"sub_a": {"acc,none": 0.7}, "sub_b": {"acc,none": 0.9}}}, + ] + + result = capability_check( + "fake/abliterated", "fake/stock", + device="cpu", subjects=["mmlu_a", "mmlu_b"], + output_dir=str(tmp_path), + ) + + assert abs(result["abliterated_acc"] - 0.7) < 0.01 + assert abs(result["stock_acc"] - 0.8) < 0.01 + + +class TestMainCLI: + @patch("obliteratus.capability_check.capability_check") + def test_main_runs(self, mock_check): + from obliteratus.capability_check import main + + mock_check.return_value = { + "stock_acc": 0.87, "abliterated_acc": 0.81, + "delta_pp": -6.0, "tasks": "mmlu", "method": "test", + } + + import sys + old_argv = sys.argv + sys.argv = ["prog", "--abliterated", "fake/abl", "--stock", "fake/stock", "--device", "cpu"] + try: + main() + finally: + sys.argv = old_argv + + mock_check.assert_called_once() + + @patch("obliteratus.capability_check._run_lm_eval") + def test_cli_dispatch(self, mock_lm_eval, tmp_path): + """Test that 'obliteratus capability-check' dispatches correctly.""" + from obliteratus.cli import main as cli_main + + mock_lm_eval.side_effect = [ + {"results": {"mmlu": {"acc,none": 0.81}}}, + {"results": {"mmlu": {"acc,none": 0.87}}}, + ] + + cli_main(["capability-check", + "--abliterated", "fake/abl", + "--stock", "fake/stock", + "--device", "cpu", + "--output-dir", str(tmp_path)])