feat(calibration): LLM-judge calibration certificate (#12)

src/agenteval/calibration.py

@@ -0,0 +1,99 @@
+"""LLM-judge calibration certificate (#12; builds on the suite hash from #11).
+
+Quantifies how well an LLM-judge's verdicts agree with reference (human) labels
+on a calibration set, scoped to the EXACT dataset (the #11 suite provenance hash)
+and the reference label distribution. A judge's calibration is only valid for
+that dataset/distribution — and the certificate is content-hashed (optionally
+HMAC-signed via AGENTEVAL_CALIBRATION_SIGNING_KEY), so the calibration claim is
+*notarized*, not merely asserted (cf. DeepEval/Braintrust).
+"""
+
+from __future__ import annotations
+
+import hashlib
+import hmac
+import json
+import os
+from datetime import datetime, timezone
+from typing import Any, Dict, Optional, Sequence
+
+
+def _canonical(obj: Any) -> str:
+    return json.dumps(obj, sort_keys=True, separators=(",", ":"), default=str)
+
+
+def calibration_metrics(predicted: Sequence[bool], reference: Sequence[bool]) -> Dict[str, Any]:
+    """Agreement, Cohen's kappa (chance-corrected) and the confusion matrix
+    between an LLM-judge's pass/fail verdicts and reference labels."""
+    if len(predicted) != len(reference):
+        raise ValueError("predicted and reference must be the same length")
+    n = len(predicted)
+    if n == 0:
+        return {"n": 0, "agreement": None, "cohenKappa": None,
+                "confusion": {"tp": 0, "tn": 0, "fp": 0, "fn": 0}, "precision": None, "recall": None}
+
+    tp = tn = fp = fn = 0
+    for p_raw, r_raw in zip(predicted, reference):
+        p, r = bool(p_raw), bool(r_raw)
+        if p and r:
+            tp += 1
+        elif not p and not r:
+            tn += 1
+        elif p and not r:
+            fp += 1
+        else:
+            fn += 1
+
+    agreement = (tp + tn) / n
+    # Cohen's kappa: (po - pe) / (1 - pe), pe = chance agreement.
+    p_pred_pos = (tp + fp) / n
+    p_ref_pos = (tp + fn) / n
+    pe = p_pred_pos * p_ref_pos + (1 - p_pred_pos) * (1 - p_ref_pos)
+    if agreement == 1.0:
+        kappa = 1.0
+    elif pe >= 1.0:  # degenerate (all one class) — no chance-corrected signal
+        kappa = 0.0
+    else:
+        kappa = (agreement - pe) / (1 - pe)
+
+    return {
+        "n": n,
+        "agreement": round(agreement, 4),
+        "cohenKappa": round(kappa, 4),
+        "confusion": {"tp": tp, "tn": tn, "fp": fp, "fn": fn},
+        "precision": round(tp / (tp + fp), 4) if (tp + fp) else None,
+        "recall": round(tp / (tp + fn), 4) if (tp + fn) else None,
+    }
+
+
+def build_calibration_certificate(
+    *,
+    judge_model: str,
+    dataset: str,
+    predicted: Sequence[bool],
+    reference: Sequence[bool],
+    suite_hash: Optional[str] = None,
+    generated_at: Optional[datetime] = None,
+) -> Dict[str, Any]:
+    """A notarized LLM-judge calibration certificate (pure). Content-hashed and,
+    when ``AGENTEVAL_CALIBRATION_SIGNING_KEY`` is set, HMAC-signed."""
+    metrics = calibration_metrics(predicted, reference)
+    ref_pos = sum(1 for r in reference if bool(r))
+    core: Dict[str, Any] = {
+        "kind": "agenteval.llm-judge-calibration/v1",
+        "judgeModel": judge_model,
+        "dataset": dataset,
+        "suiteHash": suite_hash,
+        "referenceDistribution": {"positive": ref_pos, "negative": len(reference) - ref_pos},
+        "metrics": metrics,
+        "generatedAt": (generated_at or datetime.now(timezone.utc)).isoformat(),
+    }
+    canon = _canonical(core)
+    core["contentHash"] = "sha256:" + hashlib.sha256(canon.encode("utf-8")).hexdigest()
+    key = os.environ.get("AGENTEVAL_CALIBRATION_SIGNING_KEY")
+    if key:
+        value = hmac.new(key.encode("utf-8"), canon.encode("utf-8"), hashlib.sha256).hexdigest()
+        core["signature"] = {"type": "hmac", "alg": "sha256", "value": value}
+    else:
+        core["signature"] = None
+    return core

src/agenteval/commands/__init__.py

@@ -6,6 +6,7 @@
 
 from agenteval.commands import (
     baseline,
+    calibrate,
     ci,
     compare,
     coverage,
@@ -48,6 +49,7 @@
     trends,
     verify,
     suite_hash,
+    calibrate,
 ]
 
 

src/agenteval/commands/calibrate.py

@@ -0,0 +1,77 @@
+"""The 'calibrate' command — LLM-judge calibration certificate (#12)."""
+
+from __future__ import annotations
+
+import json
+
+import click
+
+
+def register(cli: click.Group, helpers: dict) -> None:
+    """Register the calibrate command on the CLI group."""
+
+    @cli.command()
+    @click.argument("run_id")
+    @click.option("--labels", "labels_file", required=True, type=click.Path(exists=True),
+                  help='JSON mapping of reference labels: {"case_name": true|false}.')
+    @click.option("--judge-model", default="unknown", help="The judge model under calibration.")
+    @click.option("--suite-file", default=None, type=click.Path(exists=True),
+                  help="Suite file → provenance hash (#11) scoping the cert to an exact dataset version.")
+    @click.option("--db", default="agenteval.db", show_default=True, help="SQLite database path.")
+    @click.option("--format", "fmt", default="json", type=click.Choice(["json", "text"]), show_default=True)
+    def calibrate(run_id, labels_file, judge_model, suite_file, db, fmt):
+        """Build an LLM-judge calibration certificate for a run.
+
+        Compares the run's per-case verdicts against reference (human) labels and
+        records agreement + Cohen's kappa, scoped to the dataset (suite hash) and
+        the reference distribution. Notarized via content hash (+ optional HMAC).
+
+          agenteval calibrate RUN_ID --labels labels.json --judge-model claude-...
+        """
+        import agenteval.cli as _cli_mod
+        _fail = _cli_mod._fail
+
+        from agenteval.calibration import build_calibration_certificate
+        from agenteval.store import ResultStore
+
+        with open(labels_file, encoding="utf-8") as f:
+            labels = json.load(f)
+        if not isinstance(labels, dict):
+            _fail("--labels must be a JSON object {case_name: bool}.")
+
+        store = ResultStore(db)
+        try:
+            run = store.get_run(run_id)
+            if run is None:
+                _fail(f"Run '{run_id}' not found.")
+        finally:
+            store.close()
+
+        assert run is not None  # _fail() raises otherwise
+        predicted, reference = [], []
+        for r in run.results:
+            if r.case_name in labels:
+                predicted.append(bool(r.passed))
+                reference.append(bool(labels[r.case_name]))
+        if not predicted:
+            _fail("No run cases matched the provided labels.")
+
+        suite_hash = None
+        if suite_file:
+            from agenteval.loader import load_suite
+            from agenteval.provenance import suite_content_hash
+            suite_hash = suite_content_hash(load_suite(suite_file))
+
+        cert = build_calibration_certificate(
+            judge_model=judge_model, dataset=run.suite,
+            predicted=predicted, reference=reference, suite_hash=suite_hash,
+        )
+
+        if fmt == "json":
+            click.echo(json.dumps(cert, indent=2))
+        else:
+            m = cert["metrics"]
+            click.echo(
+                f"judge={cert['judgeModel']} dataset={cert['dataset']} n={m['n']} "
+                f"agreement={m['agreement']} kappa={m['cohenKappa']} hash={cert['contentHash'][:16]}…"
+            )

tests/test_calibration.py

@@ -0,0 +1,76 @@
+"""LLM-judge calibration certificate (#12)."""
+
+import pytest
+
+from agenteval.calibration import build_calibration_certificate, calibration_metrics
+
+
+def test_perfect_agreement():
+    m = calibration_metrics([True, False, True], [True, False, True])
+    assert m["n"] == 3
+    assert m["agreement"] == 1.0
+    assert m["cohenKappa"] == 1.0
+    assert m["confusion"] == {"tp": 2, "tn": 1, "fp": 0, "fn": 0}
+
+
+def test_total_disagreement_has_negative_kappa():
+    m = calibration_metrics([True, True, False, False], [False, False, True, True])
+    assert m["agreement"] == 0.0
+    assert m["cohenKappa"] < 0  # worse than chance
+
+
+def test_confusion_and_precision_recall():
+    # predicted P,P,P,N vs reference P,N,P,P → tp=2, fp=1, fn=1, tn=0
+    m = calibration_metrics([True, True, True, False], [True, False, True, True])
+    assert m["confusion"] == {"tp": 2, "tn": 0, "fp": 1, "fn": 1}
+    assert m["precision"] == round(2 / 3, 4)
+    assert m["recall"] == round(2 / 3, 4)
+
+
+def test_length_mismatch_raises():
+    with pytest.raises(ValueError):
+        calibration_metrics([True], [True, False])
+
+
+def test_empty_is_null_metrics():
+    m = calibration_metrics([], [])
+    assert m["n"] == 0 and m["agreement"] is None and m["cohenKappa"] is None
+
+
+def test_degenerate_single_class_kappa_zero():
+    # all predicted+reference negative → agreement 1.0 short-circuits to kappa 1.0
+    assert calibration_metrics([False, False], [False, False])["cohenKappa"] == 1.0
+    # predicted all P, reference all P but one N → pe high; not perfect
+    m = calibration_metrics([True, True], [True, False])
+    assert m["agreement"] == 0.5
+
+
+def test_certificate_shape_and_hash(monkeypatch):
+    monkeypatch.delenv("AGENTEVAL_CALIBRATION_SIGNING_KEY", raising=False)
+    c = build_calibration_certificate(
+        judge_model="claude-haiku-4-5", dataset="pii-suite",
+        predicted=[True, False], reference=[True, True], suite_hash="sha256:abc",
+    )
+    assert c["kind"] == "agenteval.llm-judge-calibration/v1"
+    assert c["judgeModel"] == "claude-haiku-4-5"
+    assert c["suiteHash"] == "sha256:abc"
+    assert c["referenceDistribution"] == {"positive": 2, "negative": 0}
+    assert c["metrics"]["n"] == 2
+    assert c["contentHash"].startswith("sha256:")
+    assert c["signature"] is None
+
+
+def test_certificate_signed_with_key(monkeypatch):
+    monkeypatch.setenv("AGENTEVAL_CALIBRATION_SIGNING_KEY", "calib-key-at-least-16-chars")
+    c = build_calibration_certificate(judge_model="m", dataset="d", predicted=[True], reference=[True])
+    assert c["signature"]["type"] == "hmac" and len(c["signature"]["value"]) == 64
+
+
+def test_command_registered():
+    from click.testing import CliRunner
+
+    from agenteval.cli import cli
+
+    res = CliRunner().invoke(cli, ["calibrate", "--help"])
+    assert res.exit_code == 0
+    assert "calibration certificate" in res.output.lower()