mirror of
https://github.com/elder-plinius/OBLITERATUS.git
synced 2026-08-29 22:20:36 +02:00
Merge pull request #125 from younger-plinius/capability-check-v2
capability-check CLI command
This commit is contained in:
pliny
+12
-1
@@ -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": []
|
||||
},
|
||||
{
|
||||
|
||||
@@ -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()
|
||||
@@ -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":
|
||||
|
||||
@@ -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)])
|
||||
Reference in New Issue
Block a user