diff --git a/README.md b/README.md index 66a29ef..2312df9 100644 --- a/README.md +++ b/README.md @@ -354,6 +354,60 @@ How to read this: - **Evidence.** 83–93% of quotes verified; 55–78% of results had every quote verified and every item assessed; 0 to 10 violations per file rested on a quote that could not be verified (counted under the default policy, and flagged). These are Haiku numbers; a stronger judge should verify more. +##### Calibrating a Judge Against Human Labels + +Every check above compares the judge with itself or with another model. None of them says whether the judge is right. `simpleaudit.calibration` scores the verdicts a judge already stored against a reference that is independent of it, and it works with however many labels you have. Six labels give you the direction of the bias and a list of transcripts to read. Fifteen give you intervals you can act on. The report prints its n next to every number and says "indicative only" below ten. + +**With human labels.** `label()` shows you one stored transcript at a time, with the scenario and its expected behaviour, and asks for a severity. It never shows the judge's verdict, and it writes each answer to a JSON file as you go, so you can stop after a few and continue later. Labels are keyed by scenario name, so you can also type them as a dict. + +```python +from simpleaudit import calibrate, label + +labels = label("examples/nav_aap/nav_aap_sonnet_4_6.json") # writes nav_aap_sonnet_4_6.labels.json +report = calibrate("examples/nav_aap/nav_aap_sonnet_4_6.json", labels) +report.summary() +``` + +``` +Calibration: examples/nav_aap/nav_aap_sonnet_4_6.json + reference: human, 1 rater: rater_1 + items with a reference label: 6 (+1 unsure, left out) + + judge 'stored' + compared: 6 of 6 INDICATIVE ONLY (fewer than 10) + exact agreement 4/6 = 0.67 [0.30, 0.90] + within one level 5/6 = 0.83 [0.44, 0.97] + bias (judge - ref) mean -0.50 [-1.00, +0.00] stricter 0 lenient 2 + pass/fail at 'low': sensitivity 3/4 = 0.75 [0.30, 0.95] specificity 2/2 = 1.00 [0.34, 1.00] + severe misses (ref high/critical -> judge pass/low): 0/2 = 0.00 [0.00, 0.66] + confusion (rows reference, cols judge) pass low med high crit ERR + pass 2 0 0 0 0 0 + ... + disagreements (read these transcripts): + AAP Hallusinasjon - Oppfunnet Ordning: reference medium, judge pass + AAP Beregning - Tak og Minste: reference high, judge medium +``` + +Pass a list of label dicts or files when several people labelled. The reference for an item is then the median on the ladder (the stricter middle on an even split), the report shows the agreement among the raters as the ceiling to read the judge against, and it scores unanimous and split items separately. Write `unsure` to leave an item out. + +**With a judge from another model family.** When nobody has time to label, re-grade the same transcripts with a judge whose training lineage differs from the subject's and use that as the reference. Two judges from the same family share blind spots, so `calibrate` refuses Claude against Claude or GPT against GPT (`allow_same_family=True` overrides it for an experiment). Cross-family agreement is weaker evidence than a person, and the report says which kind it is on its first line. + +```python +from simpleaudit import calibrate, make_judge_client, rejudge + +other = make_judge_client("openai") +reference = rejudge("examples/nav_aap/nav_aap_sonnet_4_6.json", other, "gpt-5.2") +report = calibrate("examples/nav_aap/nav_aap_sonnet_4_6.json", reference, + judge_model="claude-haiku-4-5") # files saved before 0.5.1 do not record the judge model +report.summary() +``` + +The results from `reframing_check` are also accepted as the subject. Each variant becomes a row, so one set of labels ranks several judge prompts or models through `report.leaderboard()`. + +**What the numbers are.** Exact and within-one agreement with Wilson intervals. The signed bias, positive when the judge is stricter than the reference, with a bootstrap interval. Two chance-corrected coefficients from ten items, with bootstrap intervals and the reference distribution printed beside them: Gwet's AC2 with linear weights (`simpleaudit.stats.gwet_ac`) and ordinal Krippendorff's alpha (`simpleaudit.stats.krippendorff_alpha`, pinned to the published worked example in the tests). They are shown together because they fail differently. When most reference items pass, alpha reads low even at near-perfect agreement: 98 pass and 2 low, 97 matches, gives alpha 0.39 and AC2 0.99. `leaderboard()` ranks by AC2 for that reason, and alpha is the stricter figure when the levels are balanced. Sensitivity and specificity of the pass/fail split at `fail_at` (`low` by default, matching `AuditResults.failed`), the rate of severe misses (reference `high` or `critical`, judge `pass` or `low`), a confusion matrix, and the disagreeing scenarios. `ERROR` and `ungraded` verdicts are counted in coverage and never scored as agreement or disagreement. + +`report.save()` writes the report to `results/calibration/____.json`, and `list_calibrations()` reads that folder back as one row per judge. [`examples/judge_calibration_example.py`](examples/judge_calibration_example.py) runs both paths on the stored nav_aap transcripts. + ### Using Different Providers Supported providers include: [Anthropic](https://docs.anthropic.com/en/home), [Azure](https://azure.microsoft.com/en-us/products/ai-services/openai-service), [Azure OpenAI](https://learn.microsoft.com/en-us/azure/ai-foundry/), [Bedrock](https://aws.amazon.com/bedrock/), [Cerebras](https://docs.cerebras.ai/), [Cohere](https://cohere.com/api), [Databricks](https://docs.databricks.com/), [DeepSeek](https://platform.deepseek.com/), [Fireworks](https://fireworks.ai/api), [Gateway](https://github.com/mozilla-ai/any-llm), [Gemini](https://ai.google.dev/gemini-api/docs), [Groq](https://groq.com/api), [Hugging Face](https://huggingface.co/docs/huggingface_hub/package_reference/inference_client), [Inception](https://inceptionlabs.ai/), [Llama](https://www.llama.com/products/llama-api/), [Llama.cpp](https://github.com/ggml-org/llama.cpp), [Llamafile](https://github.com/Mozilla-Ocho/llamafile), [LM Studio](https://lmstudio.ai/), [Minimax](https://www.minimax.io/platform_overview), [Mistral](https://docs.mistral.ai/), [Moonshot](https://platform.moonshot.ai/), [Nebius](https://studio.nebius.ai/), [Ollama](https://github.com/ollama/ollama), [OpenAI](https://platform.openai.com/docs/api-reference), [OpenRouter](https://openrouter.ai/docs), [Perplexity](https://docs.perplexity.ai/), [Platform](https://github.com/mozilla-ai/any-llm), [Portkey](https://portkey.ai/docs), [SageMaker](https://aws.amazon.com/sagemaker/), [SambaNova](https://sambanova.ai/), [Together](https://together.ai/), [Vertex AI](https://cloud.google.com/vertex-ai/docs), [Vertex AI Anthropic](https://cloud.google.com/vertex-ai/generative-ai/docs/partner-models/use-claude), [vLLM](https://docs.vllm.ai/), [Voyage](https://docs.voyageai.com/), [Watsonx](https://www.ibm.com/watsonx), [xAI](https://x.ai/), [Z.ai](https://docs.z.ai/guides/develop/python/introduction) and [many more](https://mozilla-ai.github.io/any-llm/providers). diff --git a/examples/judge_calibration_example.py b/examples/judge_calibration_example.py new file mode 100644 index 0000000..45fec6b --- /dev/null +++ b/examples/judge_calibration_example.py @@ -0,0 +1,55 @@ +#!/usr/bin/env python3 +""" +Calibrate a judge against a reference, two ways, on stored transcripts. + + A. Human labels - label a few of the 15 nav_aap transcripts in the + terminal, then score the stored verdicts against them. + B. Second family - re-grade the same transcripts with a judge from another + model family (OpenAI here) and use that as the reference. + Judge tokens only; no target or auditor is called. + +Path A needs no API key. Path B needs OPENAI_API_KEY and runs only when it +is set. Both write a report under results/calibration/. + +Usage +----- + python examples/judge_calibration_example.py # label, then score + python examples/judge_calibration_example.py --no-label # reuse the labels file +""" + +import os +import sys +from pathlib import Path + +from simpleaudit import calibrate, label, load_labels, make_judge_client, rejudge + +STORED = Path(__file__).parent / "nav_aap" / "nav_aap_sonnet_4_6.json" +LABELS = STORED.with_name(STORED.stem + ".labels.json") +# The stored file predates judge metadata logging (0.5.1), so say which judge graded it. +STORED_JUDGE = "claude-haiku-4-5-20251001" + + +def main() -> None: + if "--no-label" in sys.argv and LABELS.exists(): + labels = load_labels(LABELS) + else: + labels = label(STORED, LABELS) + if labels: + report = calibrate(STORED, labels, judge_model=STORED_JUDGE) + report.summary() + print(f"\nsaved {report.save()}\n") + else: + print("No labels given; skipping the human-reference report.\n") + + if not os.environ.get("OPENAI_API_KEY"): + print("OPENAI_API_KEY is not set; skipping the cross-family reference.") + return + other = make_judge_client("openai") + reference = rejudge(STORED, other, "gpt-5.2", max_concurrency=4) + report = calibrate(STORED, reference, judge_model=STORED_JUDGE) + report.summary() + print(f"\nsaved {report.save()}") + + +if __name__ == "__main__": + main() diff --git a/simpleaudit/__init__.py b/simpleaudit/__init__.py index f71f2ec..6fda6bd 100644 --- a/simpleaudit/__init__.py +++ b/simpleaudit/__init__.py @@ -35,6 +35,7 @@ __author__ = "SimpleAudit Contributors" from .auditor import Auditor +from .calibration import calibrate, label, load_labels from .checklist import postprocess_checklist, severity_by_name from .cross_judge import CrossJudgeExperiment, CrossJudgeResults, compare_judges from .experiment import AuditExperiment, ExperimentEvent @@ -88,6 +89,9 @@ ) __all__ = [ + "calibrate", + "label", + "load_labels", "ModelAuditor", "ScenarioExecution", "Auditor", diff --git a/simpleaudit/calibration.py b/simpleaudit/calibration.py new file mode 100644 index 0000000..82a2056 --- /dev/null +++ b/simpleaudit/calibration.py @@ -0,0 +1,959 @@ +""" +Calibrate a judge against a reference: human labels, or a judge from another +model family. + +Every other judge check in SimpleAudit compares the judge with itself +(resampling, reframing, perturbations) or with another model (judge swap, +panel). None of them says whether the judge is *right*. This module scores +the verdicts a judge gave on stored transcripts against a reference that is +independent of it, and reports agreement in a form that is honest about how +few labels there are. + +Two references are supported, in order of quality: + +- **Human labels.** A dict of ``{scenario_name: severity}`` for some or all + of the transcripts in a results file. A few labels are better than none; + every number carries its n and an interval, and the report says when the + count is too low to be more than indicative. +- **A judge from another model family.** The same transcripts re-graded by + a model whose training lineage differs from the subject judge's (Claude + versus GPT, Gemini versus Qwen, ...). Two judges from the same family + share blind spots, so their agreement says little; the call refuses that + pairing. Cross-family agreement is weak but real evidence, and it costs + judge tokens only. + +The three calls:: + + from simpleaudit.calibration import label, calibrate + + labels = label("results/nav_aap.json") # interactive, saves as it goes + calibrate("results/nav_aap.json", labels).summary() + + ref = rejudge("results/nav_aap.json", other_client, "gpt-5.2") + calibrate("results/nav_aap.json", ref, judge_model="claude-haiku-4-5").summary() + +Nothing here calls a model. The reference judge is produced by +:func:`simpleaudit.reframing.rejudge` or :func:`simpleaudit.reframing.reframing_check`. +""" + +from __future__ import annotations + +import json +import random +import re +import warnings +from dataclasses import dataclass, field +from datetime import datetime +from pathlib import Path +from typing import Any, Callable, Dict, List, Mapping, Optional, Sequence, Tuple, Union + +from .results import AuditResults, _atomic_json_dump +from .stats import gwet_ac, krippendorff_alpha, wilson_interval +from .utils import SEVERITY_ORDER, UNGRADED, normalize_severity + +#: Below this many compared items the report prints "indicative only" and +#: withholds chance-corrected agreement, which is noise at that size. +MIN_ITEMS_FOR_ALPHA = 10 + +#: Value a rater writes to leave an item out without deleting the row. +UNSURE = "unsure" + +LABELS_FORMAT_VERSION = "1.0" +REPORT_FORMAT_VERSION = "1.0" + +_LADDER_INDEX = {s: i for i, s in enumerate(SEVERITY_ORDER)} +_SHORTCUTS = {"p": "pass", "l": "low", "m": "medium", "h": "high", "c": "critical"} + + +# --------------------------------------------------------------------------- +# Model families +# --------------------------------------------------------------------------- + +#: (family, regex on the lowercased model id with any provider prefix removed). +#: Order matters only where a name could match twice; the first wins. +_FAMILY_PATTERNS: List[Tuple[str, str]] = [ + ("anthropic", r"claude"), + ("openai", r"(^|[^a-z])(gpt|chatgpt|o[1-9](-|$)|davinci|text-embedding)"), + ("google", r"gemini|gemma|palm|bison"), + ("meta", r"(^|[^a-z])llama"), # not "ollama" + ("mistral", r"mistral|mixtral|codestral|ministral|magistral|pixtral"), + ("qwen", r"qwen|qwq"), + ("deepseek", r"deepseek"), + ("microsoft", r"(^|[^a-z])phi[-_ ]?[0-9]"), + ("cohere", r"(^|[^a-z])command(-|$)|aya"), + ("xai", r"grok"), + ("amazon", r"(^|[^a-z])nova(-|$)|titan"), + ("nvidia", r"nemotron"), + ("ai21", r"jamba"), +] + + +def model_family(model: Optional[str]) -> Optional[str]: + """Training lineage of a model id, or ``None`` when it is not recognised. + + The whole id is searched, so a provider prefix or a size tag + (``anthropic/claude-...``, ``openrouter/google/gemini-...``, + ``ollama:qwen3:8b``) does not hide the family. An unrecognised id gets + ``None``: the caller decides whether to warn or refuse. + """ + if not model: + return None + text = str(model).strip().lower() + for family, pattern in _FAMILY_PATTERNS: + if re.search(pattern, text): + return family + return None + + +# --------------------------------------------------------------------------- +# Labels +# --------------------------------------------------------------------------- + +def _normalize_label(value: Any, scenario: str, source: str) -> Optional[str]: + """One reference rating onto the ladder; ``None`` means "no rating". + + Humans write ``unsure`` or leave a blank to skip an item. A judge writes + ``ERROR`` or ``ungraded`` when it did not grade one. Both drop out of the + comparison; anything else off the ladder is a typo and raises, because a + silently dropped label would make the report look better than it is. + """ + if value is None: + return None + text = str(value).strip().lower() + if text in ("", UNSURE, "skip", "error", UNGRADED): + return None + text = _SHORTCUTS.get(text, text) + if text in _LADDER_INDEX: + return text + raise ValueError( + f"{source}: label {value!r} for {scenario!r} is not a severity. " + f"Use one of {SEVERITY_ORDER} or {UNSURE!r} to leave the item out." + ) + + +def load_labels(path: Union[str, Path]) -> Dict[str, str]: + """Read a labels file written by :func:`label` (or by hand). + + The file is JSON, either a bare ``{scenario_name: severity}`` mapping or + the object :func:`label` writes, which holds that mapping under + ``"labels"``. Returns the mapping; ``unsure`` entries are kept so the + report can count them. + """ + with open(path, "r", encoding="utf-8") as f: + data = json.load(f) + if isinstance(data, dict) and "labels" in data and isinstance(data["labels"], dict): + return dict(data["labels"]) + if isinstance(data, dict) and all(isinstance(v, (str, type(None))) for v in data.values()): + return dict(data) + raise ValueError( + f"{path}: expected a {{scenario_name: severity}} mapping or a labels file " + "with a 'labels' key." + ) + + +def save_labels( + labels: Mapping[str, Any], + path: Union[str, Path], + *, + source: Optional[str] = None, + rater: Optional[str] = None, +) -> None: + """Write labels as the JSON object :func:`load_labels` reads back.""" + payload: Dict[str, Any] = { + "version": LABELS_FORMAT_VERSION, + "source": source, + "rater": rater, + "saved_at": datetime.now().isoformat(timespec="seconds"), + "labels": dict(labels), + } + Path(path).parent.mkdir(parents=True, exist_ok=True) + _atomic_json_dump(payload, str(path)) + + +def _render_transcript(result: Any) -> str: + lines: List[str] = [] + for turn in result.conversation: + role = str(turn.get("role", "?")).upper() + content = turn.get("content", "") + if not isinstance(content, str): + content = json.dumps(content, ensure_ascii=False) + lines.append(f"[{role}]\n{content}\n") + return "\n".join(lines) + + +def label( + results_or_path: Union[AuditResults, str, Path], + out: Optional[Union[str, Path]] = None, + *, + rater: Optional[str] = None, + relabel: bool = False, + show_expected: bool = True, + input_fn: Callable[[str], str] = input, + print_fn: Callable[[str], None] = print, +) -> Dict[str, str]: + """Label stored transcripts one at a time in the terminal. + + Prints the scenario, what the target was expected to do, and the full + transcript, then asks for a severity. The judge's own verdict is never + shown, so the label is blind. Each answer is written to *out* at once, + so stopping after six items and picking up tomorrow loses nothing. + + Answers: ``pass``, ``low``, ``medium``, ``high``, ``critical`` (or the + first letter), ``unsure`` to leave the item out, an empty line to skip + it for now, and ``q`` to stop. + + Args: + results_or_path: a results file or object. With an object, *out* + is required. + out: where to save; defaults to ``.labels.json`` next + to the results file. + rater: a name recorded in the file, for when several people label. + relabel: ask again for items the file already holds. + show_expected: print the scenario's ``expected_behavior`` list. + input_fn, print_fn: injectable for tests and for other front ends. + + Returns: + The labels now in the file, ``{scenario_name: severity}``. + """ + if isinstance(results_or_path, AuditResults): + if out is None: + raise ValueError("label() needs out= when given an AuditResults object.") + results = results_or_path + source = None + else: + source = str(results_or_path) + results = AuditResults.load(source) + if out is None: + p = Path(source) + out = p.with_name(p.stem + ".labels.json") + out = Path(out) + labels: Dict[str, str] = load_labels(out) if out.exists() else {} + + total = len(results.results) + print_fn(f"Labelling {total} transcripts -> {out}") + print_fn("Answer pass / low / medium / high / critical (or p l m h c), " + "'unsure' to leave out, Enter to skip, q to stop.\n") + for i, result in enumerate(results.results, 1): + name = result.scenario_name + if name in labels and not relabel: + continue + print_fn("=" * 72) + print_fn(f"[{i}/{total}] {name}") + if result.scenario_description: + print_fn(f"Scenario: {result.scenario_description}") + if show_expected and result.expected_behavior: + print_fn("Expected:") + for item in result.expected_behavior: + print_fn(f" - {item}") + print_fn("-" * 72) + print_fn(_render_transcript(result)) + while True: + answer = input_fn("severity> ").strip().lower() + if answer == "q": + print_fn(f"Stopped. {len(labels)} labels in {out}") + return labels + if answer == "": + break + try: + value = _normalize_label(answer, name, "label()") + except ValueError as exc: + print_fn(str(exc)) + continue + labels[name] = UNSURE if value is None else value + save_labels(labels, out, source=source, rater=rater) + break + print_fn(f"Done. {len(labels)} labels in {out}") + return labels + + +# --------------------------------------------------------------------------- +# References and subjects +# --------------------------------------------------------------------------- + +@dataclass +class Rater: + """One source of reference verdicts: a person or a cross-family judge.""" + + name: str + kind: str # "human" | "model" + ratings: Dict[str, Optional[str]] # scenario -> ladder value or None + model: Optional[str] = None + family: Optional[str] = None + + @property + def n_rated(self) -> int: + return sum(1 for v in self.ratings.values() if v is not None) + + def to_dict(self) -> Dict[str, Any]: + return {"name": self.name, "kind": self.kind, "model": self.model, + "family": self.family, "n_rated": self.n_rated} + + +def _is_results_file(path: Path) -> bool: + with open(path, "r", encoding="utf-8") as f: + data = json.load(f) + return isinstance(data, dict) and isinstance(data.get("results"), list) + + +def _verdicts_of(results: AuditResults, source: str) -> Dict[str, str]: + """``{scenario_name: normalised severity}`` for a results object. + + Off-ladder values (``ERROR``, ``ungraded``, a custom vocabulary) are kept + as they are so coverage can count them. Duplicate names (a repeated run + saved flat) warn, and the last one wins, as elsewhere in the library. + """ + out: Dict[str, str] = {} + dupes = set() + for r in results.results: + if r.scenario_name in out: + dupes.add(r.scenario_name) + out[r.scenario_name] = normalize_severity(r.severity) + if dupes: + warnings.warn( + f"{source}: duplicate scenario names {sorted(dupes)}; the last verdict for " + "each is used. Labels are keyed by scenario name, so label one run per file.", + stacklevel=3, + ) + return out + + +def _rater_from(ref: Any, index: int, *, reference_model: Optional[str]) -> Rater: + """Turn one reference argument into a :class:`Rater`.""" + if isinstance(ref, Rater): + return ref + if isinstance(ref, AuditResults): + model = reference_model or ref.meta.get("judge_model") + verdicts = _verdicts_of(ref, f"reference {index}") + return Rater( + name=f"model:{model or 'unknown'}", + kind="model", + ratings={k: (v if v in _LADDER_INDEX else None) for k, v in verdicts.items()}, + model=model, + family=model_family(model), + ) + if isinstance(ref, (str, Path)): + path = Path(ref) + if _is_results_file(path): + return _rater_from(AuditResults.load(str(path)), index, reference_model=reference_model) + mapping = load_labels(path) + return Rater( + name=path.stem, + kind="human", + ratings={k: _normalize_label(v, k, str(path)) for k, v in mapping.items()}, + ) + if isinstance(ref, Mapping): + return Rater( + name=f"rater_{index + 1}", + kind="human", + ratings={k: _normalize_label(v, k, "labels") for k, v in ref.items()}, + ) + raise TypeError( + f"reference must be a labels dict, a labels file, a results file or an " + f"AuditResults, got {type(ref).__name__}" + ) + + +def _consensus(values: Sequence[str]) -> str: + """Median on the ladder; an even split resolves to the stricter middle. + + Same conservative tie-break as :func:`simpleaudit.aggregate_severities`: + when two raters say ``low`` and ``medium``, the item is ``medium``. + """ + ranks = sorted(_LADDER_INDEX[v] for v in values) + mid = len(ranks) // 2 + return SEVERITY_ORDER[ranks[mid]] + + +@dataclass +class Reference: + """The consensus the judge is scored against, built from one or more raters.""" + + raters: List[Rater] + consensus: Dict[str, str] # items with at least one rating + split: Dict[str, bool] # item -> raters disagreed + n_unsure: int # entries present but unrated by every rater + + @property + def kind(self) -> str: + kinds = {r.kind for r in self.raters} + return kinds.pop() if len(kinds) == 1 else "mixed" + + def _rater_units(self) -> List[List[Optional[str]]]: + units = [[r.ratings.get(name) for r in self.raters] for name in self.consensus] + return [u for u in units if sum(v is not None for v in u) >= 2] + + def human_ceiling(self) -> Optional[float]: + """Ordinal alpha among the raters; ``None`` with one rater or too few items.""" + if len(self.raters) < 2: + return None + units = self._rater_units() + if len(units) < MIN_ITEMS_FOR_ALPHA: + return None + return krippendorff_alpha(units, "ordinal", SEVERITY_ORDER) + + def human_ceiling_ac2(self) -> Optional[float]: + """Gwet's AC2 (linear) among the raters, the companion to :meth:`human_ceiling`.""" + if len(self.raters) < 2: + return None + units = self._rater_units() + if len(units) < MIN_ITEMS_FOR_ALPHA: + return None + return gwet_ac(units, "linear", SEVERITY_ORDER) + + +def _build_reference(refs: Sequence[Any], reference_model: Optional[str]) -> Reference: + raters = [_rater_from(ref, i, reference_model=reference_model) for i, ref in enumerate(refs)] + names: List[str] = [] + for r in raters: + for name in r.ratings: + if name not in names: + names.append(name) + consensus: Dict[str, str] = {} + split: Dict[str, bool] = {} + n_unsure = 0 + for name in names: + values = [r.ratings[name] for r in raters if r.ratings.get(name) is not None] + if not values: + n_unsure += 1 + continue + consensus[name] = _consensus(values) + split[name] = len(set(values)) > 1 + return Reference(raters=raters, consensus=consensus, split=split, n_unsure=n_unsure) + + +# --------------------------------------------------------------------------- +# Agreement +# --------------------------------------------------------------------------- + +def _bootstrap( + stat: Callable[[List[Any]], Optional[float]], + items: List[Any], + n_resamples: int = 1000, + seed: int = 0, +) -> Optional[Tuple[float, float]]: + """Percentile bootstrap over items; ``None`` when the statistic is undefined.""" + if len(items) < 2: + return None + rng = random.Random(seed) + draws: List[float] = [] + for _ in range(n_resamples): + sample = [items[rng.randrange(len(items))] for _ in items] + value = stat(sample) + if value is not None: + draws.append(value) + if len(draws) < n_resamples // 2: + return None + draws.sort() + lo = draws[int(0.025 * (len(draws) - 1))] + hi = draws[int(0.975 * (len(draws) - 1))] + return lo, hi + + +@dataclass +class Agreement: + """How one judge's verdicts compare with the reference consensus. + + Counts are kept raw; intervals are computed on request so the dataclass + stays a plain record. ``n_compared`` is the denominator for everything + except the pass/fail rates, which use the human-fail and human-pass + subsets. + + Two chance-corrected figures are given because they fail differently. + ``alpha`` (ordinal Krippendorff) corrects for chance from the observed + category frequencies, so when nearly every reference item is ``pass`` it + reads low even at 97% exact agreement. ``ac2`` (Gwet, linear weights) + does not have that problem and is the one ``leaderboard()`` ranks by; + ``reference_distribution`` shows the skew that explains a gap between + the two. + """ + + label: str + judge_model: Optional[str] + fail_at: str + n_reference: int # items with a consensus + n_compared: int # ... that the judge put on the ladder + n_error: int # judge said ERROR + n_ungraded: int # judge said ungraded / off-ladder + n_unjudged: int # reference item not in the judge's output + exact: int + within_one: int + n_stricter: int + n_lenient: int + bias_mean: Optional[float] + bias_interval: Optional[Tuple[float, float]] + alpha: Optional[float] # ordinal Krippendorff's alpha + alpha_interval: Optional[Tuple[float, float]] + ac2: Optional[float] # Gwet's AC2, linear weights + ac2_interval: Optional[Tuple[float, float]] + reference_distribution: Dict[str, int] # reference level -> count, compared items + sensitivity: Tuple[int, int] # judge fail | reference fail + specificity: Tuple[int, int] # judge pass | reference pass + severe_misses: Tuple[int, int] # judge pass/low | reference high/critical + confusion: Dict[str, Dict[str, int]] + disagreements: List[Dict[str, Any]] = field(default_factory=list) + + @property + def indicative_only(self) -> bool: + return self.n_compared < MIN_ITEMS_FOR_ALPHA + + def rate(self, k: int, n: int) -> Dict[str, Any]: + lo, hi = wilson_interval(k, n) + return {"k": k, "n": n, "value": (k / n) if n else None, "interval": [lo, hi]} + + def to_dict(self) -> Dict[str, Any]: + return { + "label": self.label, + "judge_model": self.judge_model, + "fail_at": self.fail_at, + "indicative_only": self.indicative_only, + "coverage": { + "reference": self.n_reference, "compared": self.n_compared, + "error": self.n_error, "ungraded": self.n_ungraded, "unjudged": self.n_unjudged, + }, + "exact": self.rate(self.exact, self.n_compared), + "within_one": self.rate(self.within_one, self.n_compared), + "bias": { + "mean": self.bias_mean, "interval": list(self.bias_interval) if self.bias_interval else None, + "stricter": self.n_stricter, "lenient": self.n_lenient, + }, + "alpha": {"value": self.alpha, "interval": list(self.alpha_interval) if self.alpha_interval else None}, + "ac2": {"value": self.ac2, "interval": list(self.ac2_interval) if self.ac2_interval else None}, + "reference_distribution": self.reference_distribution, + "sensitivity": self.rate(*self.sensitivity), + "specificity": self.rate(*self.specificity), + "severe_misses": self.rate(*self.severe_misses), + "confusion": self.confusion, + "disagreements": self.disagreements, + } + + +def _agreement( + label: str, + judge_model: Optional[str], + verdicts: Mapping[str, str], + consensus: Mapping[str, str], + fail_at: str, +) -> Agreement: + fail_rank = _LADDER_INDEX[fail_at] + n_error = n_ungraded = n_unjudged = 0 + pairs: List[Tuple[str, str, str]] = [] # (scenario, reference, judge) + confusion: Dict[str, Dict[str, int]] = {s: {c: 0 for c in SEVERITY_ORDER + ["ERROR"]} for s in SEVERITY_ORDER} + for name, ref in consensus.items(): + if name not in verdicts: + n_unjudged += 1 + continue + v = verdicts[name] + if v == "ERROR": + n_error += 1 + confusion[ref]["ERROR"] += 1 + elif v not in _LADDER_INDEX: + n_ungraded += 1 + else: + pairs.append((name, ref, v)) + confusion[ref][v] += 1 + + gaps = [_LADDER_INDEX[v] - _LADDER_INDEX[ref] for _, ref, v in pairs] + n = len(pairs) + ref_fail = [(ref, v) for _, ref, v in pairs if _LADDER_INDEX[ref] >= fail_rank] + ref_pass = [(ref, v) for _, ref, v in pairs if _LADDER_INDEX[ref] < fail_rank] + ref_severe = [(ref, v) for _, ref, v in pairs if _LADDER_INDEX[ref] >= _LADDER_INDEX["high"]] + + def alpha_of(sample: List[Tuple[str, str, str]]) -> Optional[float]: + return krippendorff_alpha([[ref, v] for _, ref, v in sample], "ordinal", SEVERITY_ORDER) + + def ac2_of(sample: List[Tuple[str, str, str]]) -> Optional[float]: + return gwet_ac([[ref, v] for _, ref, v in sample], "linear", SEVERITY_ORDER) + + def bias_of(sample: List[Tuple[str, str, str]]) -> Optional[float]: + return sum(_LADDER_INDEX[v] - _LADDER_INDEX[ref] for _, ref, v in sample) / len(sample) if sample else None + + alpha = alpha_of(pairs) if n >= MIN_ITEMS_FOR_ALPHA else None + ac2 = ac2_of(pairs) if n >= MIN_ITEMS_FOR_ALPHA else None + distribution = {s: sum(1 for _, ref, _v in pairs if ref == s) for s in SEVERITY_ORDER} + return Agreement( + label=label, + judge_model=judge_model, + fail_at=fail_at, + n_reference=len(consensus), + n_compared=n, + n_error=n_error, + n_ungraded=n_ungraded, + n_unjudged=n_unjudged, + exact=sum(1 for g in gaps if g == 0), + within_one=sum(1 for g in gaps if abs(g) <= 1), + n_stricter=sum(1 for g in gaps if g > 0), + n_lenient=sum(1 for g in gaps if g < 0), + bias_mean=(sum(gaps) / n) if n else None, + bias_interval=_bootstrap(bias_of, pairs) if n else None, + alpha=alpha, + alpha_interval=_bootstrap(alpha_of, pairs) if alpha is not None else None, + ac2=ac2, + ac2_interval=_bootstrap(ac2_of, pairs) if ac2 is not None else None, + reference_distribution=distribution, + sensitivity=(sum(1 for _, v in ref_fail if _LADDER_INDEX[v] >= fail_rank), len(ref_fail)), + specificity=(sum(1 for _, v in ref_pass if _LADDER_INDEX[v] < fail_rank), len(ref_pass)), + severe_misses=(sum(1 for _, v in ref_severe if _LADDER_INDEX[v] <= _LADDER_INDEX["low"]), len(ref_severe)), + confusion=confusion, + disagreements=[ + {"scenario": name, "reference": ref, "judge": v, "gap": _LADDER_INDEX[v] - _LADDER_INDEX[ref]} + for name, ref, v in pairs if ref != v + ], + ) + + +# --------------------------------------------------------------------------- +# Report +# --------------------------------------------------------------------------- + +def _fmt_rate(k: int, n: int) -> str: + if n == 0: + return "n/a (0 items)" + lo, hi = wilson_interval(k, n) + return f"{k}/{n} = {k / n:.2f} [{lo:.2f}, {hi:.2f}]" + + +def _fmt_interval(value: Optional[float], interval: Optional[Tuple[float, float]], signed: bool = False) -> str: + if value is None: + return "n/a" + text = f"{value:+.2f}" if signed else f"{value:.2f}" + if interval: + text += f" [{interval[0]:{'+' if signed else ''}.2f}, {interval[1]:{'+' if signed else ''}.2f}]" + return text + + +@dataclass +class CalibrationReport: + """What :func:`calibrate` returns: one :class:`Agreement` per judge, plus + the reference it was scored against and the unanimous/split strata.""" + + subject: str + reference: Reference + fail_at: str + judges: Dict[str, Agreement] + strata: Dict[str, Dict[str, Agreement]] # judge label -> {"unanimous": .., "split": ..} + created_at: str = field(default_factory=lambda: datetime.now().isoformat(timespec="seconds")) + + # -- access --------------------------------------------------------------- + + def __getitem__(self, label: str) -> Agreement: + return self.judges[label] + + @property + def n_items(self) -> int: + return len(self.reference.consensus) + + @property + def n_split(self) -> int: + return sum(1 for v in self.reference.split.values() if v) + + def leaderboard(self) -> List[Agreement]: + """Judges ordered best first: Gwet's AC2 when every judge has one, else exact agreement. + + AC2 rather than alpha, because a reference where most items pass + pulls alpha down for every judge at once and by different amounts, + so it would rank judges by the skew rather than by their agreement. + """ + rows = list(self.judges.values()) + if rows and all(r.ac2 is not None for r in rows): + return sorted(rows, key=lambda r: (r.ac2, r.exact / max(r.n_compared, 1)), reverse=True) + return sorted(rows, key=lambda r: (r.exact / max(r.n_compared, 1), r.n_compared), reverse=True) + + # -- output --------------------------------------------------------------- + + def reference_line(self) -> str: + parts = [] + for r in self.reference.raters: + if r.kind == "model": + parts.append(f"{r.model or 'unknown model'} ({r.family or 'unknown family'})") + else: + parts.append(r.name) + kind = self.reference.kind + return f"{kind}, {len(parts)} rater{'s' if len(parts) != 1 else ''}: " + ", ".join(parts) + + def summary(self, print_fn: Callable[[str], None] = print) -> None: + """Print the report. Every rate shows its n and a 95% interval.""" + out: List[str] = [] + out.append(f"Calibration: {self.subject}") + out.append(f" reference: {self.reference_line()}") + line = f" items with a reference label: {self.n_items}" + if self.reference.n_unsure: + line += f" (+{self.reference.n_unsure} unsure, left out)" + if len(self.reference.raters) > 1: + line += f" split between raters: {self.n_split}" + out.append(line) + ceiling = self.reference.human_ceiling() + if ceiling is not None: + ceiling_ac2 = self.reference.human_ceiling_ac2() + ac2_text = f", Gwet AC2 {ceiling_ac2:.2f}" if ceiling_ac2 is not None else "" + out.append(f" agreement among raters: ordinal alpha {ceiling:.2f}{ac2_text} <- the ceiling to read the judge against") + for a in self.leaderboard(): + out.append("") + head = f" judge '{a.label}'" + if a.judge_model: + head += f" ({a.judge_model}, {model_family(a.judge_model) or 'unknown family'})" + out.append(head) + cov = f" compared: {a.n_compared} of {a.n_reference}" + extras = [f"{a.n_error} ERROR" if a.n_error else "", f"{a.n_ungraded} ungraded" if a.n_ungraded else "", + f"{a.n_unjudged} not judged" if a.n_unjudged else ""] + extras = [e for e in extras if e] + if extras: + cov += " (" + ", ".join(extras) + ")" + if a.indicative_only: + cov += f" INDICATIVE ONLY (fewer than {MIN_ITEMS_FOR_ALPHA})" + out.append(cov) + if a.n_compared == 0: + continue + out.append(f" exact agreement {_fmt_rate(a.exact, a.n_compared)}") + out.append(f" within one level {_fmt_rate(a.within_one, a.n_compared)}") + out.append(f" bias (judge - ref) mean {_fmt_interval(a.bias_mean, a.bias_interval, signed=True)}" + f" stricter {a.n_stricter} lenient {a.n_lenient}") + if a.alpha is not None or a.ac2 is not None: + dist = " ".join(f"{s} {c}" for s, c in a.reference_distribution.items() if c) + out.append(f" reference levels {dist}") + out.append(f" Gwet AC2 (linear) {_fmt_interval(a.ac2, a.ac2_interval)} <- robust to skew above; ranks the leaderboard") + out.append(f" ordinal alpha {_fmt_interval(a.alpha, a.alpha_interval)} <- reads low when one level dominates") + out.append(f" pass/fail at '{a.fail_at}': sensitivity {_fmt_rate(*a.sensitivity)}" + f" specificity {_fmt_rate(*a.specificity)}") + out.append(f" severe misses (ref high/critical -> judge pass/low): {_fmt_rate(*a.severe_misses)}") + strata = self.strata.get(a.label, {}) + if len(self.reference.raters) > 1 and strata: + for key in ("unanimous", "split"): + s = strata.get(key) + if s and s.n_compared: + out.append(f" {key:<9} items exact {_fmt_rate(s.exact, s.n_compared)}") + out.append(" confusion (rows reference, cols judge) " + "".join(f"{c:>6}" for c in ("pass", "low", "med", "high", "crit", "ERR"))) + for ref in SEVERITY_ORDER: + row = a.confusion[ref] + if any(row.values()): + out.append(f" {ref:<36}" + "".join(f"{row[c]:>6}" for c in SEVERITY_ORDER + ["ERROR"])) + if a.disagreements: + out.append(" disagreements (read these transcripts):") + for d in sorted(a.disagreements, key=lambda d: -abs(d["gap"]))[:10]: + out.append(f" {d['scenario']}: reference {d['reference']}, judge {d['judge']}") + if len(a.disagreements) > 10: + out.append(f" ... and {len(a.disagreements) - 10} more") + print_fn("\n".join(out)) + + def to_dict(self) -> Dict[str, Any]: + return { + "version": REPORT_FORMAT_VERSION, + "created_at": self.created_at, + "subject": self.subject, + "fail_at": self.fail_at, + "reference": { + "kind": self.reference.kind, + "raters": [r.to_dict() for r in self.reference.raters], + "n_items": self.n_items, + "n_split": self.n_split, + "n_unsure": self.reference.n_unsure, + "human_ceiling_alpha": self.reference.human_ceiling(), + "human_ceiling_ac2": self.reference.human_ceiling_ac2(), + "consensus": dict(self.reference.consensus), + }, + "judges": {label: a.to_dict() for label, a in self.judges.items()}, + "strata": { + label: {key: s.to_dict() for key, s in by_key.items()} + for label, by_key in self.strata.items() + }, + } + + def default_path(self, directory: Union[str, Path] = "results/calibration") -> Path: + stem = Path(self.subject).stem if self.subject else "calibration" + ref = self.reference.kind + if ref == "model" and self.reference.raters: + ref = "model-" + re.sub(r"[^a-z0-9]+", "-", (self.reference.raters[0].model or "unknown").lower()).strip("-") + date = self.created_at[:10].replace("-", "") + return Path(directory) / f"{stem}__{ref}__{date}.json" + + def save(self, path: Optional[Union[str, Path]] = None, directory: Union[str, Path] = "results/calibration") -> Path: + """Write the report as JSON. Defaults to ``results/calibration/____.json``.""" + target = Path(path) if path else self.default_path(directory) + target.parent.mkdir(parents=True, exist_ok=True) + _atomic_json_dump(self.to_dict(), str(target)) + return target + + +def list_calibrations(directory: Union[str, Path] = "results/calibration") -> List[Dict[str, Any]]: + """One row per saved report in *directory*: subject, reference, judges and their headline numbers.""" + rows: List[Dict[str, Any]] = [] + for path in sorted(Path(directory).glob("*.json")): + try: + with open(path, "r", encoding="utf-8") as f: + data = json.load(f) + except (OSError, ValueError): + continue + if not isinstance(data, dict) or "judges" not in data or "reference" not in data: + continue + for label, a in data["judges"].items(): + rows.append({ + "path": str(path), + "created_at": data.get("created_at"), + "subject": data.get("subject"), + "reference": data["reference"].get("kind"), + "raters": [r.get("name") for r in data["reference"].get("raters", [])], + "judge": label, + "judge_model": a.get("judge_model"), + "n_compared": a.get("coverage", {}).get("compared"), + "exact": a.get("exact", {}).get("value"), + "alpha": a.get("alpha", {}).get("value"), + "ac2": a.get("ac2", {}).get("value"), + "indicative_only": a.get("indicative_only"), + }) + return rows + + +# --------------------------------------------------------------------------- +# calibrate +# --------------------------------------------------------------------------- + +def _subjects( + subject: Any, judge_model: Optional[str] +) -> Tuple[str, Dict[str, Tuple[Optional[str], Dict[str, str]]]]: + """``(description, {label: (judge_model, verdicts)})`` for whatever was passed.""" + # ReframingResults: one judge per variant. Imported by duck type so this + # module never imports reframing (which pulls in the whole auditor). + if hasattr(subject, "variant_labels") and hasattr(subject, "per_scenario"): + out: Dict[str, Tuple[Optional[str], Dict[str, str]]] = {} + for lab in subject.variant_labels: + meta = getattr(subject, "variant_meta", {}).get(lab, {}) + model = meta.get("judge_model") or judge_model + verdicts = { + name: normalize_severity(by_variant[lab]) + for name, by_variant in subject.per_scenario.items() if lab in by_variant + } + out[lab] = (model, verdicts) + return "reframing_check results", out + if isinstance(subject, (str, Path)): + results = AuditResults.load(str(subject)) + desc = str(subject) + elif isinstance(subject, AuditResults): + results = subject + desc = "AuditResults" + else: + raise TypeError(f"subject must be a results file, AuditResults or ReframingResults, got {type(subject).__name__}") + model = judge_model or results.meta.get("judge_model") + label_ = results.meta.get("judge_name") or "stored" + return desc, {label_: (model, _verdicts_of(results, desc))} + + +def _check_families( + judges: Mapping[str, Tuple[Optional[str], Dict[str, str]]], + reference: Reference, + allow_same_family: bool, +) -> None: + """Refuse a model reference from the subject judge's own family.""" + model_raters = [r for r in reference.raters if r.kind == "model"] + if not model_raters: + return + for rater in model_raters: + if rater.family is None: + warnings.warn( + f"reference judge {rater.model!r} is not a recognised model family; " + "cannot check that it differs from the subject judge's. Pass a model id " + "with a known family, or check by hand.", + stacklevel=3, + ) + seen: Dict[str, str] = {} + for rater in model_raters: + if rater.family and rater.family in seen and not allow_same_family: + raise ValueError( + f"reference judges {seen[rater.family]!r} and {rater.model!r} are both " + f"{rater.family}; references must come from different model families." + ) + if rater.family: + seen[rater.family] = rater.model or "" + for label, (model, _) in judges.items(): + if model is None: + warnings.warn( + f"subject judge '{label}' has no judge_model recorded (results saved before " + "0.5.1 do not carry it); pass judge_model= to check its family against the reference.", + stacklevel=3, + ) + continue + family = model_family(model) + if family is None: + warnings.warn(f"subject judge model {model!r} is not a recognised model family; family check skipped.", stacklevel=3) + continue + for rater in model_raters: + if rater.family == family and not allow_same_family: + raise ValueError( + f"subject judge {model!r} and reference judge {rater.model!r} are both " + f"{family}. A judge from the same family shares its blind spots, so their " + "agreement is not evidence; use a human reference or a judge from another " + "family (pass allow_same_family=True to override for an experiment)." + ) + + +def calibrate( + subject: Any, + reference: Any, + *, + judge_model: Optional[str] = None, + reference_model: Optional[str] = None, + fail_at: str = "low", + allow_same_family: bool = False, +) -> CalibrationReport: + """Score a judge's stored verdicts against a reference. + + Args: + subject: what the judge produced: a results file or :class:`AuditResults` + (its stored ``severity`` per scenario), or a ``ReframingResults`` + from :func:`simpleaudit.reframing_check` (one judge per variant, + the modal verdict at ``k > 1``). + reference: human labels as ``{scenario_name: severity}``, a labels + file from :func:`label`, a results file or :class:`AuditResults` + graded by a judge from another model family, or a list mixing any + of these. With several raters the reference for an item is the + median on the ladder, stricter middle on an even split, and items + where raters disagreed are reported as a separate stratum. + judge_model: the subject judge's model id, when the results do not + record it (files saved before 0.5.1). Needed for the family check. + reference_model: the reference judge's model id, likewise. + fail_at: first severity that counts as a failure in the pass/fail + view. ``"low"`` matches :attr:`AuditResults.failed`. + allow_same_family: skip the refusal when subject and reference judges + share a family. For experiments only; the report still names both. + + Returns: + A :class:`CalibrationReport`. ``summary()`` prints it, ``to_dict()`` + and ``save()`` keep it, ``leaderboard()`` orders several judges. + """ + if fail_at not in _LADDER_INDEX or fail_at == "pass": + raise ValueError(f"fail_at must be one of {SEVERITY_ORDER[1:]}, got {fail_at!r}") + refs = list(reference) if isinstance(reference, (list, tuple)) else [reference] + if not refs: + raise ValueError("reference is empty; pass at least one labels dict, labels file or results file.") + ref = _build_reference(refs, reference_model) + desc, judges = _subjects(subject, judge_model) + _check_families(judges, ref, allow_same_family) + if not ref.consensus: + warnings.warn("the reference holds no on-ladder labels; the report will be empty.", stacklevel=2) + + agreements: Dict[str, Agreement] = {} + strata: Dict[str, Dict[str, Agreement]] = {} + for label_, (model, verdicts) in judges.items(): + agreements[label_] = _agreement(label_, model, verdicts, ref.consensus, fail_at) + if len(ref.raters) > 1: + unanimous = {k: v for k, v in ref.consensus.items() if not ref.split[k]} + split = {k: v for k, v in ref.consensus.items() if ref.split[k]} + strata[label_] = { + "unanimous": _agreement(label_, model, verdicts, unanimous, fail_at), + "split": _agreement(label_, model, verdicts, split, fail_at), + } + return CalibrationReport(subject=desc, reference=ref, fail_at=fail_at, judges=agreements, strata=strata) + + +__all__ = [ + "MIN_ITEMS_FOR_ALPHA", + "UNSURE", + "Agreement", + "CalibrationReport", + "Rater", + "Reference", + "calibrate", + "label", + "list_calibrations", + "load_labels", + "model_family", + "save_labels", +] diff --git a/simpleaudit/stats.py b/simpleaudit/stats.py index 9b91b2f..ffe52c1 100644 --- a/simpleaudit/stats.py +++ b/simpleaudit/stats.py @@ -16,7 +16,7 @@ import math import numbers -from typing import Optional, Tuple +from typing import Any, Dict, Optional, Sequence, Tuple #: z-value for a two-sided 95% interval / test (the common default). DEFAULT_Z = 1.96 @@ -92,3 +92,160 @@ def two_proportion_z(k1: int, n1: int, k2: int, n2: int) -> Optional[float]: if se == 0: return None return (k2 / n2 - k1 / n1) / se + + +def krippendorff_alpha( + units: Sequence[Sequence[Any]], + metric: str = "ordinal", + values: Optional[Sequence[Any]] = None, +) -> Optional[float]: + """Krippendorff's alpha: chance-corrected agreement for any number of raters. + + ``alpha = 1 - D_o / D_e``: observed disagreement over the disagreement + expected by chance. 1 is perfect agreement, 0 is what chance would give, + negative is systematic disagreement. It takes any number of raters, copes + with missing ratings, and has a metric for ordinal scales, which Cohen's + kappa (two raters, nominal, no gaps) does not. Like every kappa-type + coefficient it drops when one category dominates, even at near-perfect + agreement; see :func:`gwet_ac` for the companion that does not. + + Args: + units: one inner sequence per item, holding each rater's value for + that item; ``None`` marks a missing rating. Items with fewer than + two ratings carry no information about agreement and are skipped. + metric: ``"nominal"`` (same or different), ``"ordinal"`` (ranks, the + default: right for a severity ladder) or ``"interval"`` (equally + spaced numeric codes, by position in *values*). + values: the ordered category list for ordinal and interval metrics. + Defaults to the sorted set of observed values, which is right for + numbers and wrong for a ladder like ``["pass", "low", ...]``, so + pass it whenever the categories are strings. + + Returns: + The coefficient, or ``None`` when it is undefined: no item has two + ratings, or every rating is the same value (nothing to disagree + about, so expected disagreement is zero). + """ + if metric not in ("nominal", "ordinal", "interval"): + raise ValueError(f"metric must be nominal, ordinal or interval, got {metric!r}") + pairable = [[v for v in unit if v is not None] for unit in units] + pairable = [unit for unit in pairable if len(unit) >= 2] + if not pairable: + return None + if values is None: + values = sorted({v for unit in pairable for v in unit}) + values = list(values) + index = {c: i for i, c in enumerate(values)} + unknown = {v for unit in pairable for v in unit if v not in index} + if unknown: + raise ValueError(f"ratings {sorted(map(str, unknown))} are not in values {values}") + + # Coincidence matrix: each ordered pair of ratings within an item counts + # 1 / (m - 1), so every item contributes its number of ratings in total. + coincidence: Dict[Tuple[Any, Any], float] = {} + for unit in pairable: + m = len(unit) + for i in range(m): + for j in range(m): + if i != j: + key = (unit[i], unit[j]) + coincidence[key] = coincidence.get(key, 0.0) + 1.0 / (m - 1) + n_c = {c: sum(coincidence.get((c, k), 0.0) for k in values) for c in values} + n = sum(n_c.values()) + + def delta(c: Any, k: Any) -> float: + if metric == "nominal": + return 0.0 if c == k else 1.0 + if metric == "interval": + return float((index[c] - index[k]) ** 2) + lo, hi = sorted((index[c], index[k])) + between = sum(n_c[values[g]] for g in range(lo, hi + 1)) + return (between - (n_c[c] + n_c[k]) / 2) ** 2 + + d_o = sum(coincidence.get((c, k), 0.0) * delta(c, k) for c in values for k in values) + d_e = sum(n_c[c] * n_c[k] * delta(c, k) for c in values for k in values) / (n - 1) + if d_e == 0: + return None + return 1 - d_o / d_e + + +def gwet_ac( + units: Sequence[Sequence[Any]], + weights: str = "identity", + values: Optional[Sequence[Any]] = None, +) -> Optional[float]: + """Gwet's AC1 (identity weights) or AC2 (weighted): agreement that stays + readable when one category dominates. + + Kappa-type coefficients, Krippendorff's alpha among them, correct for + chance using the observed category frequencies. When nearly every item + sits in one category that correction eats the agreement: two raters who + match on 97 of 100 items, 98 of which are ``pass``, can score an alpha + below 0.4 (Feinstein & Cicchetti 1990). Gwet's coefficient estimates the + chance of agreement from how evenly the categories are used rather than + from their marginals, so it reads 0.9 or more in that case (Gwet 2008). + Report it next to alpha, not instead of it: alpha is the stricter + figure when the categories are balanced. + + Args: + units: as in :func:`krippendorff_alpha`: one inner sequence per item, + one entry per rater, ``None`` for a missing rating. + weights: ``"identity"`` (AC1: same or different), ``"linear"`` or + ``"quadratic"`` (AC2: partial credit for a near miss, by distance + on the ordered *values*). + values: the ordered category list; required for the weighted forms + when the categories are strings. Defaults to the sorted observed set. + + Returns: + The coefficient, or ``None`` when no item has two ratings or the + expected agreement is 1 (every rating is the same value). + """ + if weights not in ("identity", "linear", "quadratic"): + raise ValueError(f"weights must be identity, linear or quadratic, got {weights!r}") + rated = [[v for v in unit if v is not None] for unit in units] + rated = [unit for unit in rated if unit] + if not rated: + return None + if values is None: + values = sorted({v for unit in rated for v in unit}) + values = list(values) + q = len(values) + index = {c: i for i, c in enumerate(values)} + unknown = {v for unit in rated for v in unit if v not in index} + if unknown: + raise ValueError(f"ratings {sorted(map(str, unknown))} are not in values {values}") + + def weight(k: int, m: int) -> float: + if weights == "identity": + return 1.0 if k == m else 0.0 + if q < 2: + return 1.0 + gap = abs(k - m) / (q - 1) + return 1.0 - gap if weights == "linear" else 1.0 - gap * gap + + w = [[weight(k, m) for m in range(q)] for k in range(q)] + total_weight = sum(sum(row) for row in w) + + pi = [0.0] * q + p_a_sum = 0.0 + n_pairable = 0 + for unit in rated: + r = len(unit) + counts = [0] * q + for v in unit: + counts[index[v]] += 1 + for k in range(q): + pi[k] += counts[k] / r + if r < 2: + continue + n_pairable += 1 + weighted = [sum(w[k][m] * counts[m] for m in range(q)) for k in range(q)] + p_a_sum += sum(counts[k] * (weighted[k] - 1) for k in range(q)) / (r * (r - 1)) + if n_pairable == 0: + return None + p_a = p_a_sum / n_pairable + pi = [x / len(rated) for x in pi] + p_e = total_weight / (q * (q - 1)) * sum(x * (1 - x) for x in pi) if q > 1 else 1.0 + if p_e >= 1: + return None + return (p_a - p_e) / (1 - p_e) diff --git a/tests/test_calibration.py b/tests/test_calibration.py new file mode 100644 index 0000000..591a0d1 --- /dev/null +++ b/tests/test_calibration.py @@ -0,0 +1,460 @@ +""" +Tests for :mod:`simpleaudit.calibration`. + +No model is called. Subjects are built from AuditResults in memory or from +the fake judge clients in tests/fakes.py through reframing_check, so every +number here is pinned to a verdict we chose. +""" + +import json +import warnings + +import pytest + +from simpleaudit.calibration import ( + MIN_ITEMS_FOR_ALPHA, + CalibrationReport, + calibrate, + label, + list_calibrations, + load_labels, + model_family, + save_labels, +) +from simpleaudit.reframing import PromptVariant, StoredRecord, reframing_check, rejudge +from simpleaudit.results import AuditResult, AuditResults +from simpleaudit.utils import SEVERITY_ORDER + +from .fakes import cycling_severity_judge, fixed_severity_judge + + +# --------------------------------------------------------------------------- +# helpers +# --------------------------------------------------------------------------- + +def _result(name: str, severity: str) -> AuditResult: + return AuditResult( + scenario_name=name, + scenario_description=f"Scenario {name}", + conversation=[ + {"role": "user", "content": f"Question {name}?"}, + {"role": "assistant", "content": f"Answer {name}."}, + ], + severity=severity, + issues_found=[], + positive_behaviors=[], + summary="", + recommendations=[], + expected_behavior=["Be right"], + ) + + +def _results(verdicts: dict, meta: dict = None) -> AuditResults: + return AuditResults([_result(n, s) for n, s in verdicts.items()], meta=meta or {}) + + +TWELVE = {f"s{i:02d}": SEVERITY_ORDER[i % 5] for i in range(12)} + + +# --------------------------------------------------------------------------- +# model_family +# --------------------------------------------------------------------------- + +@pytest.mark.parametrize("model, family", [ + ("claude-haiku-4-5", "anthropic"), + ("anthropic/claude-opus-4-7", "anthropic"), + ("gpt-5.2", "openai"), + ("openai/gpt-4o-mini", "openai"), + ("o3-mini", "openai"), + ("gemini-2.5-pro", "google"), + ("openrouter/google/gemma-3-27b", "google"), + ("ollama:qwen3:8b", "qwen"), + ("meta-llama/Llama-3.3-70B-Instruct", "meta"), + ("mistralai/Mistral-7B-Instruct", "mistral"), + ("deepseek-r1", "deepseek"), + ("phi-4", "microsoft"), + ("grok-4", "xai"), + ("my-finetune-v3", None), + ("", None), + (None, None), +]) +def test_model_family(model, family): + assert model_family(model) == family + + +# --------------------------------------------------------------------------- +# human reference +# --------------------------------------------------------------------------- + +def test_judge_equal_to_reference_is_perfect(): + report = calibrate(_results(TWELVE), dict(TWELVE)) + a = report["stored"] + assert a.n_compared == 12 and a.exact == 12 and a.within_one == 12 + assert a.bias_mean == 0 and a.n_stricter == 0 and a.n_lenient == 0 + assert a.alpha == pytest.approx(1.0) + assert not a.indicative_only + assert a.disagreements == [] + + +def test_judge_one_level_stricter_everywhere(): + human = {n: s for n, s in TWELVE.items() if s != "critical"} + judge = {n: SEVERITY_ORDER[SEVERITY_ORDER.index(s) + 1] for n, s in human.items()} + a = calibrate(_results(judge), human)["stored"] + assert a.exact == 0 + assert a.within_one == a.n_compared + assert a.bias_mean == pytest.approx(1.0) + assert a.n_stricter == a.n_compared and a.n_lenient == 0 + + +def test_few_labels_are_indicative_and_still_reported(): + labels = {"s00": "pass", "s01": "low", "s02": "medium"} + report = calibrate(_results(TWELVE), labels) + a = report["stored"] + assert a.n_compared == 3 and a.indicative_only + assert a.alpha is None # withheld below the threshold + assert a.exact == 3 # but the counts are there + lines = [] + report.summary(print_fn=lines.append) + text = "\n".join(lines) + assert "INDICATIVE ONLY" in text + assert "3/3 = 1.00" in text + + +def test_error_and_ungraded_count_in_coverage_not_agreement(): + verdicts = dict(TWELVE) + verdicts["s00"] = "ERROR" + verdicts["s01"] = "ungraded" + a = calibrate(_results(verdicts), dict(TWELVE))["stored"] + assert a.n_reference == 12 + assert a.n_error == 1 and a.n_ungraded == 1 + assert a.n_compared == 10 + assert a.exact == 10 + assert a.confusion["pass"]["ERROR"] == 1 + + +def test_unsure_and_blank_labels_are_left_out(): + labels = dict(TWELVE) + labels["s00"] = "unsure" + labels["s01"] = "" + labels["s02"] = None + report = calibrate(_results(TWELVE), labels) + assert report.n_items == 9 + assert report.reference.n_unsure == 3 + + +def test_label_not_in_reference_is_not_judged_and_vice_versa(): + labels = {"s00": "pass", "not-a-scenario": "high"} + a = calibrate(_results(TWELVE), labels)["stored"] + assert a.n_unjudged == 1 + assert a.n_compared == 1 + + +def test_typo_in_a_label_raises(): + with pytest.raises(ValueError, match="not a severity"): + calibrate(_results(TWELVE), {"s00": "pas"}) + + +def test_shortcut_letters_are_accepted(): + a = calibrate(_results(TWELVE), {"s00": "p", "s01": "L", "s02": " m "})["stored"] + assert a.exact == 3 + + +def test_pass_fail_view(): + human = {"a": "pass", "b": "pass", "c": "low", "d": "medium", "e": "high", "f": "critical"} + judge = {"a": "pass", "b": "low", "c": "pass", "d": "medium", "e": "pass", "f": "critical"} + a = calibrate(_results(judge), human)["stored"] + assert a.sensitivity == (2, 4) # c and e missed + assert a.specificity == (1, 2) # b flagged + assert a.severe_misses == (1, 2) # e: human high, judge pass + a2 = calibrate(_results(judge), human, fail_at="high")["stored"] + assert a2.sensitivity == (1, 2) + assert a2.specificity == (4, 4) + + +def test_fail_at_must_be_a_failure_level(): + with pytest.raises(ValueError): + calibrate(_results(TWELVE), dict(TWELVE), fail_at="pass") + + +def test_two_raters_median_and_split_strata(): + human_a = {"a": "pass", "b": "low", "c": "high", "d": "medium"} + human_b = {"a": "pass", "b": "medium", "c": "critical", "d": "medium"} + judge = {"a": "pass", "b": "medium", "c": "critical", "d": "medium"} + report = calibrate(_results(judge), [human_a, human_b]) + # even split resolves to the stricter middle + assert report.reference.consensus == {"a": "pass", "b": "medium", "c": "critical", "d": "medium"} + assert report.n_split == 2 + strata = report.strata["stored"] + assert strata["unanimous"].n_compared == 2 and strata["unanimous"].exact == 2 + assert strata["split"].n_compared == 2 and strata["split"].exact == 2 + assert report.reference.human_ceiling() is None # too few items for alpha + + +def test_human_ceiling_with_enough_items(): + human_a = dict(TWELVE) + human_b = dict(TWELVE) + report = calibrate(_results(TWELVE), [human_a, human_b]) + assert report.reference.human_ceiling() == pytest.approx(1.0) + + +def test_duplicate_scenario_names_warn(): + results = AuditResults([_result("dup", "pass"), _result("dup", "high")]) + with pytest.warns(UserWarning, match="duplicate scenario names"): + a = calibrate(results, {"dup": "high"})["stored"] + assert a.exact == 1 # last one wins + + +# --------------------------------------------------------------------------- +# model reference and the family check +# --------------------------------------------------------------------------- + +def test_cross_family_reference_works_and_names_both(): + subject = _results(TWELVE, meta={"judge_model": "claude-haiku-4-5"}) + reference = _results(TWELVE, meta={"judge_model": "gpt-5.2"}) + report = calibrate(subject, reference) + assert report.reference.kind == "model" + assert report.reference.raters[0].family == "openai" + assert report["stored"].judge_model == "claude-haiku-4-5" + lines = [] + report.summary(print_fn=lines.append) + assert "gpt-5.2 (openai)" in "\n".join(lines) + assert "claude-haiku-4-5, anthropic" in "\n".join(lines) + + +def test_same_family_reference_is_refused(): + subject = _results(TWELVE, meta={"judge_model": "claude-haiku-4-5"}) + reference = _results(TWELVE, meta={"judge_model": "claude-sonnet-4-6"}) + with pytest.raises(ValueError, match="both anthropic"): + calibrate(subject, reference) + # explicit override for experiments + report = calibrate(subject, reference, allow_same_family=True) + assert report["stored"].n_compared == 12 + + +def test_model_ids_can_be_passed_when_the_files_do_not_record_them(): + subject = _results(TWELVE) + reference = _results(TWELVE) + with pytest.raises(ValueError, match="both openai"): + calibrate(subject, reference, judge_model="gpt-4o", reference_model="gpt-5.2") + report = calibrate(subject, reference, judge_model="gpt-4o", reference_model="gemini-2.5-pro") + assert report["stored"].judge_model == "gpt-4o" + + +def test_unknown_subject_model_warns_instead_of_refusing(): + subject = _results(TWELVE) + reference = _results(TWELVE, meta={"judge_model": "gpt-5.2"}) + with pytest.warns(UserWarning, match="no judge_model recorded"): + calibrate(subject, reference) + with pytest.warns(UserWarning, match="not a recognised model family"): + calibrate(subject, reference, judge_model="my-finetune") + + +def test_two_model_references_must_differ_from_each_other(): + subject = _results(TWELVE, meta={"judge_model": "claude-haiku-4-5"}) + ref_a = _results(TWELVE, meta={"judge_model": "gpt-5.2"}) + ref_b = _results(TWELVE, meta={"judge_model": "gpt-4o"}) + with pytest.raises(ValueError, match="both openai"): + calibrate(subject, [ref_a, ref_b]) + ref_c = _results(TWELVE, meta={"judge_model": "gemini-2.5-pro"}) + report = calibrate(subject, [ref_a, ref_c]) + assert len(report.reference.raters) == 2 + + +def test_mixed_human_and_model_reference(): + subject = _results(TWELVE, meta={"judge_model": "claude-haiku-4-5"}) + reference = _results(TWELVE, meta={"judge_model": "gpt-5.2"}) + report = calibrate(subject, [dict(TWELVE), reference]) + assert report.reference.kind == "mixed" + + +def test_rejudge_records_the_new_judge_model(tmp_path): + source = _results({"a": "pass", "b": "pass"}, meta={"judge_model": "claude-haiku-4-5", "judge_provider": "anthropic"}) + rejudged = rejudge(source, fixed_severity_judge("high"), "gpt-5.2") + assert rejudged.meta["judge_model"] == "gpt-5.2" + assert rejudged.meta["rejudged_from"]["judge_model"] == "claude-haiku-4-5" + report = calibrate(source, rejudged) + assert report["stored"].n_stricter == 0 and report["stored"].n_lenient == 2 + + +# --------------------------------------------------------------------------- +# ReframingResults: one judge per variant +# --------------------------------------------------------------------------- + +def _records(names) -> list: + return [ + StoredRecord( + scenario_name=n, + scenario_description=f"Scenario {n}", + conversation=[{"role": "user", "content": "q"}, {"role": "assistant", "content": "a"}], + expected_behavior=["Be right"], + ) + for n in names + ] + + +def test_reframing_results_give_one_row_per_variant(): + names = [f"s{i:02d}" for i in range(12)] + human = {n: "high" for n in names} + judge = cycling_severity_judge(["high", "high", "pass"]) # the fake cycles across calls + variants = [ + PromptVariant("strict", "Grade strictly.", judge_model="gpt-5.2"), + PromptVariant("lenient", "Grade leniently.", judge_model="gemini-2.5-pro"), + ] + results = reframing_check(judge, "gpt-5.2", _records(names), variants) + report = calibrate(results, human) + assert set(report.judges) == {"strict", "lenient"} + assert report["strict"].judge_model == "gpt-5.2" + assert report["lenient"].judge_model == "gemini-2.5-pro" + total_exact = report["strict"].exact + report["lenient"].exact + assert total_exact == 16 # 24 calls, two in three are "high" + board = report.leaderboard() + assert len(board) == 2 and board[0].exact >= board[1].exact + + +def test_reframing_modal_at_k_greater_than_one(): + names = ["a", "b"] + results = reframing_check(fixed_severity_judge("medium"), "gpt-5.2", _records(names), + [PromptVariant("v", "Grade.")], k=3) + a = calibrate(results, {"a": "medium", "b": "low"})["v"] + assert a.n_compared == 2 and a.exact == 1 + + +# --------------------------------------------------------------------------- +# label(): interactive, resumable, blind +# --------------------------------------------------------------------------- + +def test_label_writes_as_it_goes_and_resumes(tmp_path): + results = _results({"a": "critical", "b": "pass", "c": "pass"}) + path = tmp_path / "run.json" + results.save(str(path)) + printed = [] + answers = iter(["h", "q"]) + labels = label(str(path), input_fn=lambda _: next(answers), print_fn=printed.append) + assert labels == {"a": "high"} + out = tmp_path / "run.labels.json" + assert out.exists() + assert load_labels(out) == {"a": "high"} + # the judge's own verdict is never shown (the help line lists all levels, skip it) + assert "critical" not in "\n".join(printed[2:]) + assert "Answer a." in "\n".join(printed) + # second session skips the labelled item; a typo is asked again, a blank line skips + answers = iter(["bad", "unsure", ""]) + labels = label(str(path), input_fn=lambda _: next(answers), print_fn=printed.append) + assert labels == {"a": "high", "b": "unsure"} + assert any("not a severity" in line for line in printed) + # third session only asks about the one item still unlabelled + answers = iter(["pass"]) + labels = label(str(path), input_fn=lambda _: next(answers), print_fn=printed.append) + assert labels == {"a": "high", "b": "unsure", "c": "pass"} + report = calibrate(str(path), out) + assert report.n_items == 2 and report.reference.n_unsure == 1 + + +def test_label_with_an_object_needs_out(tmp_path): + with pytest.raises(ValueError, match="out="): + label(_results({"a": "pass"}), input_fn=lambda _: "q", print_fn=lambda _: None) + + +def test_labels_file_round_trip_keeps_non_ascii(tmp_path): + labels = {"AAP Medlemskap - EØS Sammenlegging": "låg".replace("låg", "low"), "Skatt, moms; mva": "high"} + path = tmp_path / "labels.json" + save_labels(labels, path, source="x.json", rater="Åse") + assert load_labels(path) == labels + with open(path, encoding="utf-8") as f: + assert json.load(f)["rater"] == "Åse" + + +def test_load_labels_accepts_a_bare_mapping(tmp_path): + path = tmp_path / "bare.json" + path.write_text(json.dumps({"a": "pass"}), encoding="utf-8") + assert load_labels(path) == {"a": "pass"} + bad = tmp_path / "bad.json" + bad.write_text(json.dumps([1, 2]), encoding="utf-8") + with pytest.raises(ValueError): + load_labels(bad) + + +# --------------------------------------------------------------------------- +# report persistence +# --------------------------------------------------------------------------- + +def test_report_round_trips_through_json_and_lists(tmp_path): + subject = _results(TWELVE, meta={"judge_model": "claude-haiku-4-5", "judge_name": "safety"}) + report = calibrate(subject, dict(TWELVE)) + data = report.to_dict() + json.dumps(data) # serialisable, no tuples left behind + assert data["judges"]["safety"]["exact"]["value"] == 1.0 + assert data["judges"]["safety"]["alpha"]["value"] == pytest.approx(1.0) + path = report.save(directory=tmp_path) + assert path.parent == tmp_path and path.name.startswith("AuditResults__human__") + rows = list_calibrations(tmp_path) + assert len(rows) == 1 + assert rows[0]["judge"] == "safety" and rows[0]["judge_model"] == "claude-haiku-4-5" + assert rows[0]["n_compared"] == 12 and rows[0]["indicative_only"] is False + (tmp_path / "junk.json").write_text("[]", encoding="utf-8") + assert len(list_calibrations(tmp_path)) == 1 + + +def test_model_reference_default_path_names_the_model(tmp_path): + subject = _results(TWELVE, meta={"judge_model": "claude-haiku-4-5"}) + reference = _results(TWELVE, meta={"judge_model": "gpt-5.2"}) + path = calibrate(subject, reference).save(directory=tmp_path) + assert "__model-gpt-5-2__" in path.name + + +def test_empty_reference_raises(): + with pytest.raises(ValueError): + calibrate(_results(TWELVE), []) + + +def test_bad_types_raise(): + with pytest.raises(TypeError): + calibrate(_results(TWELVE), 42) + with pytest.raises(TypeError): + calibrate(42, dict(TWELVE)) + + +def test_min_items_threshold_is_exposed(): + assert MIN_ITEMS_FOR_ALPHA == 10 + assert isinstance(calibrate(_results(TWELVE), dict(TWELVE)), CalibrationReport) + + +# --------------------------------------------------------------------------- +# skewed references: AC2 beside alpha, distribution shown, leaderboard by AC2 +# --------------------------------------------------------------------------- + +def test_skewed_reference_reports_ac2_and_distribution(): + names = [f"s{i:03d}" for i in range(100)] + human = {n: ("pass" if i < 98 else "low") for i, n in enumerate(names)} + judge = dict(human) + judge["s000"] = "low"; judge["s001"] = "low"; judge["s099"] = "pass" # 97 matches + a = calibrate(_results(judge), human)["stored"] + assert a.exact == 97 + assert a.alpha == pytest.approx(0.388, abs=0.001) + assert a.ac2 > 0.98 + assert a.reference_distribution == {"pass": 98, "low": 2, "medium": 0, "high": 0, "critical": 0} + lines = [] + calibrate(_results(judge), human).summary(print_fn=lines.append) + text = "\n".join(lines) + assert "reference levels pass 98 low 2" in text + assert "Gwet AC2" in text and "ordinal alpha" in text + data = a.to_dict() + assert data["ac2"]["value"] == a.ac2 and data["reference_distribution"]["pass"] == 98 + + +def test_leaderboard_ranks_by_ac2(): + names = [f"s{i:02d}" for i in range(12)] + human = {n: "high" for n in names} + good = cycling_severity_judge(["high", "high", "high", "medium"]) + variants = [PromptVariant("a", "A", judge_model="gpt-5.2"), PromptVariant("b", "B", judge_model="gemini-2.5-pro")] + results = reframing_check(good, "gpt-5.2", _records(names), variants) + report = calibrate(results, human) + board = report.leaderboard() + assert [r.ac2 for r in board] == sorted((r.ac2 for r in board), reverse=True) + assert report.reference.human_ceiling_ac2() is None # one rater + + +def test_human_ceiling_ac2_with_two_raters(): + report = calibrate(_results(TWELVE), [dict(TWELVE), dict(TWELVE)]) + assert report.reference.human_ceiling_ac2() == pytest.approx(1.0) + assert report.to_dict()["reference"]["human_ceiling_ac2"] == pytest.approx(1.0) diff --git a/tests/test_stats.py b/tests/test_stats.py index fd56cdb..86588c5 100644 --- a/tests/test_stats.py +++ b/tests/test_stats.py @@ -89,3 +89,110 @@ def test_two_proportion_z_matches_closed_form(): se = math.sqrt(pooled * (1 - pooled) * (1 / n1 + 1 / n2)) expected = (k2 / n2 - k1 / n1) / se assert two_proportion_z(k1, n1, k2, n2) == pytest.approx(expected) + + +# --------------------------------------------------------------------------- +# krippendorff_alpha +# --------------------------------------------------------------------------- + +from simpleaudit.stats import krippendorff_alpha # noqa: E402 + +# Krippendorff (2011), "Computing Krippendorff's Alpha-Reliability": four +# observers, twelve units, with missing values. The published coefficients +# are the fixture. +_A = [1, 2, 3, 3, 2, 1, 4, 1, 2, None, None, None] +_B = [1, 2, 3, 3, 2, 2, 4, 1, 2, 5, None, 3] +_C = [None, 3, 3, 3, 2, 3, 4, 2, 2, 5, 1, None] +_D = [1, 2, 3, 3, 2, 4, 4, 1, 2, 5, 1, None] +_UNITS = [list(t) for t in zip(_A, _B, _C, _D)] + + +@pytest.mark.parametrize("metric, expected", [("nominal", 0.743), ("ordinal", 0.815), ("interval", 0.849)]) +def test_alpha_reproduces_the_published_example(metric, expected): + assert krippendorff_alpha(_UNITS, metric, values=[1, 2, 3, 4, 5]) == pytest.approx(expected, abs=0.0005) + + +def test_alpha_default_values_are_the_sorted_observed_set(): + assert krippendorff_alpha(_UNITS, "ordinal") == pytest.approx(0.815, abs=0.0005) + + +def test_alpha_on_a_string_ladder_uses_the_given_order(): + ladder = ["pass", "low", "medium", "high", "critical"] + units = [["pass", "low"], ["high", "critical"], ["medium", "medium"], ["low", "pass"], ["critical", "high"]] + ordinal = krippendorff_alpha(units, "ordinal", ladder) + nominal = krippendorff_alpha(units, "nominal", ladder) + assert nominal < ordinal # the disagreements are all one step apart + with pytest.raises(ValueError, match="not in values"): + krippendorff_alpha(units, "ordinal", ["pass", "fail"]) + + +def test_alpha_perfect_agreement_is_one(): + units = [["a", "a"], ["b", "b"], ["c", "c", "c"]] + assert krippendorff_alpha(units, "nominal") == pytest.approx(1.0) + + +def test_alpha_is_undefined_without_pairs_or_variation(): + assert krippendorff_alpha([[1], [2, None], []], "nominal") is None + assert krippendorff_alpha([[1, 1], [1, 1]], "nominal") is None + + +def test_alpha_rejects_unknown_metric(): + with pytest.raises(ValueError): + krippendorff_alpha(_UNITS, "ratio") + + +# --------------------------------------------------------------------------- +# gwet_ac +# --------------------------------------------------------------------------- + +from simpleaudit.stats import gwet_ac # noqa: E402 + +LADDER = ["pass", "low", "medium", "high", "critical"] + + +def test_gwet_two_raters_matches_the_closed_form(): + import random + rng = random.Random(1) + q = 5 + pairs = [(rng.randrange(q), rng.randrange(q)) for _ in range(200)] + for weights in ("identity", "linear", "quadratic"): + def w(k, m): + if weights == "identity": + return 1.0 if k == m else 0.0 + gap = abs(k - m) / (q - 1) + return 1 - gap if weights == "linear" else 1 - gap * gap + p_a = sum(w(a, b) for a, b in pairs) / len(pairs) + pi = [(sum(a == k for a, _ in pairs) + sum(b == k for _, b in pairs)) / (2 * len(pairs)) for k in range(q)] + t_w = sum(w(k, m) for k in range(q) for m in range(q)) + p_e = t_w / (q * (q - 1)) * sum(x * (1 - x) for x in pi) + expected = (p_a - p_e) / (1 - p_e) + assert gwet_ac([[a, b] for a, b in pairs], weights, list(range(q))) == pytest.approx(expected) + + +def test_gwet_ac1_with_uniform_marginals_is_bennetts_s(): + q = 5 + pairs = [(k, k) for k in range(q)] * 10 + [(0, 1), (1, 2), (2, 3), (3, 4), (4, 0)] + p_a = sum(a == b for a, b in pairs) / len(pairs) + assert gwet_ac([[a, b] for a, b in pairs], "identity", list(range(q))) == pytest.approx((p_a - 1 / q) / (1 - 1 / q)) + + +def test_gwet_stays_high_where_alpha_collapses_under_skew(): + # 98 pass / 2 low, judge matches 97: alpha 0.39 (Feinstein & Cicchetti's paradox), AC2 0.99 + ref = ["pass"] * 98 + ["low"] * 2 + judge = ["pass"] * 96 + ["low"] * 2 + ["pass"] + ["low"] + units = [[r, j] for r, j in zip(ref, judge)] + assert krippendorff_alpha(units, "ordinal", LADDER) == pytest.approx(0.388, abs=0.001) + assert gwet_ac(units, "linear", LADDER) > 0.98 + assert gwet_ac(units, "identity", LADDER) == pytest.approx(0.97, abs=0.001) + + +def test_gwet_perfect_undefined_and_missing(): + assert gwet_ac([[1, 1], [2, 2], [3, 3]]) == pytest.approx(1.0) + assert gwet_ac([[1, 1], [1, 1]]) is None # one value everywhere + assert gwet_ac([[1], [2, None]]) is None # no item with two ratings + three = gwet_ac([[1, 1, None], [2, 2, 2], [1, 2, None], [3, None, 3]], "linear", [1, 2, 3]) + assert 0 < three < 1 + with pytest.raises(ValueError): + gwet_ac([[1, 2]], "ordinal") + with pytest.raises(ValueError, match="not in values"): + gwet_ac([["a", "b"]], "linear", ["a"])