From f8d3b3b1760e7df54f88a2771bbc189ef438adf7 Mon Sep 17 00:00:00 2001 From: zhenzhudingzhen528-cpu <316532047+zhenzhudingzhen528-cpu@users.noreply.github.com> Date: Thu, 13 Aug 2026 20:31:00 +0800 Subject: [PATCH] Batch hallucination-rate evaluation and reporting over a dataset --- scripts/report_hallucination_rates.py | 667 ++++++++++++++++++ .../test_report_hallucination_rates_pytest.py | 444 ++++++++++++ 2 files changed, 1111 insertions(+) create mode 100644 scripts/report_hallucination_rates.py create mode 100644 tests/test_report_hallucination_rates_pytest.py diff --git a/scripts/report_hallucination_rates.py b/scripts/report_hallucination_rates.py new file mode 100644 index 0000000..c092c92 --- /dev/null +++ b/scripts/report_hallucination_rates.py @@ -0,0 +1,667 @@ +"""Report hallucination rates for a JSONL or CSV dataset. + +Each input row must contain ``context`` (a string or a list of strings) and +``answer``. ``question`` and an identifier such as ``id`` are optional. CSV +contexts may be plain text or JSON-encoded string arrays. Use ``--group-by`` +to name any additional column whose values should receive separate rates; +blank or missing values are reported as ````. + +Example:: + + python scripts/report_hallucination_rates.py \ + --input sample.jsonl \ + --method transformer \ + --model-path KRLabsOrg/lettucedect-v2-mmbert-base \ + --taxonomy-head KRLabsOrg/lettucedect-v2-taxonomy-head \ + --min-confidence 0.7 \ + --group-by source \ + --out-json report.json \ + --out-md report.md + +The reported hallucination rate is the fraction of scored answers with at +least one flagged span. It is not a gold-label detector-quality metric. +""" + +from __future__ import annotations + +import argparse +import csv +import json +import logging +import math +import tempfile +from collections import Counter +from collections.abc import Mapping, Sequence +from pathlib import Path +from typing import Any, Protocol, TypedDict, cast + +MISSING_GROUP = "" +CONFIDENCE_BIN_COUNT = 10 +logger = logging.getLogger(__name__) + + +class InputRecord(TypedDict): + """Normalized input row passed to a detector.""" + + row_number: int + context: list[str] + answer: str + question: str | None + identifier: str | None + group: str | None + + +class PredictDetector(Protocol): + """Detector surface needed by the reporting runner.""" + + def predict( + self, + context: list[str], + answer: str, + question: str | None = None, + output_format: str = "tokens", + min_confidence: float = 0.0, + ) -> list[dict[str, Any]]: + """Return hallucination predictions for one record.""" + ... + + +class ScoredRecord(TypedDict): + """One normalized input row paired with its predicted spans.""" + + record: InputRecord + spans: list[dict[str, Any]] + + +def positive_int(value: str) -> int: + """Parse a strictly positive integer for argparse.""" + try: + parsed = int(value) + except ValueError as exc: + raise argparse.ArgumentTypeError("must be an integer") from exc + if parsed < 1: + raise argparse.ArgumentTypeError("must be at least 1") + return parsed + + +def non_negative_int(value: str) -> int: + """Parse a non-negative integer for argparse.""" + try: + parsed = int(value) + except ValueError as exc: + raise argparse.ArgumentTypeError("must be an integer") from exc + if parsed < 0: + raise argparse.ArgumentTypeError("must be zero or greater") + return parsed + + +def confidence_threshold(value: str) -> float: + """Parse a finite confidence threshold in the closed interval [0, 1].""" + try: + parsed = float(value) + except ValueError as exc: + raise argparse.ArgumentTypeError("must be a number between 0 and 1") from exc + if not math.isfinite(parsed) or not 0.0 <= parsed <= 1.0: + raise argparse.ArgumentTypeError("must be a finite number between 0 and 1") + return parsed + + +def _normalize_context(value: object, location: str, *, from_csv: bool) -> list[str]: + """Normalize a context string or string array with a location-aware error.""" + if from_csv and isinstance(value, str) and value.lstrip().startswith("["): + try: + value = json.loads(value) + except json.JSONDecodeError: + pass + + if isinstance(value, str): + if not value.strip(): + raise ValueError(f"{location}: context must not be blank") + return [value] + if not isinstance(value, list) or not value: + raise ValueError(f"{location}: context must be a non-empty string or list of strings") + if any(not isinstance(item, str) or not item.strip() for item in value): + raise ValueError(f"{location}: every context item must be a non-blank string") + return list(value) + + +def _normalize_optional_text(value: object, location: str, field: str) -> str | None: + """Normalize an optional scalar text field.""" + if value is None or value == "": + return None + if not isinstance(value, str): + raise ValueError(f"{location}: {field} must be a string when present") + return value + + +def _normalize_row( + row: Mapping[str, Any], + *, + path: Path, + row_number: int, + group_by: str | None, + from_csv: bool, +) -> InputRecord: + """Validate and normalize one JSONL or CSV input row.""" + location = f"{path}:{row_number}" + missing = [field for field in ("context", "answer") if field not in row] + if missing: + raise ValueError(f"{location}: missing required field(s): {', '.join(missing)}") + + answer = row["answer"] + if not isinstance(answer, str) or not answer.strip(): + raise ValueError(f"{location}: answer must be a non-blank string") + + raw_group = row.get(group_by) if group_by is not None else None + if raw_group is None or raw_group == "": + group = MISSING_GROUP if group_by is not None else None + elif isinstance(raw_group, (str, int, float, bool)): + group = str(raw_group) + else: + raise ValueError(f"{location}: grouping field {group_by!r} must be a scalar value") + + raw_identifier = row.get("id") + if raw_identifier is None or raw_identifier == "": + identifier = None + elif isinstance(raw_identifier, (str, int, float, bool)): + identifier = str(raw_identifier) + else: + raise ValueError(f"{location}: id must be a scalar value when present") + + return { + "row_number": row_number, + "context": _normalize_context(row["context"], location, from_csv=from_csv), + "answer": answer, + "question": _normalize_optional_text(row.get("question"), location, "question"), + "identifier": identifier, + "group": group, + } + + +def load_records( + path: Path, *, group_by: str | None = None, limit: int | None = None +) -> list[InputRecord]: + """Load and normalize JSONL or CSV records, optionally stopping after ``limit`` rows.""" + if limit is not None and limit < 1: + raise ValueError("limit must be at least 1") + suffix = path.suffix.lower() + if suffix not in {".jsonl", ".csv"}: + raise ValueError(f"{path}: input must have a .jsonl or .csv extension") + + records: list[InputRecord] = [] + if suffix == ".jsonl": + with path.open(encoding="utf-8") as handle: + for line_number, line in enumerate(handle, start=1): + if not line.strip(): + raise ValueError(f"{path}:{line_number}: blank JSONL rows are not allowed") + try: + raw = json.loads(line) + except json.JSONDecodeError as exc: + raise ValueError(f"{path}:{line_number}: invalid JSON: {exc.msg}") from exc + if not isinstance(raw, dict): + raise ValueError(f"{path}:{line_number}: each JSONL row must be an object") + records.append( + _normalize_row( + raw, + path=path, + row_number=line_number, + group_by=group_by, + from_csv=False, + ) + ) + if limit is not None and len(records) >= limit: + break + else: + with path.open(encoding="utf-8-sig", newline="") as handle: + reader = csv.DictReader(handle) + if reader.fieldnames is None: + raise ValueError(f"{path}: CSV must have a header row") + missing_columns = {"context", "answer"} - set(reader.fieldnames) + if missing_columns: + fields = ", ".join(sorted(missing_columns)) + raise ValueError(f"{path}: CSV is missing required column(s): {fields}") + if group_by is not None and group_by not in reader.fieldnames: + raise ValueError(f"{path}: CSV is missing grouping column {group_by!r}") + for row_number, raw in enumerate(reader, start=2): + records.append( + _normalize_row( + raw, + path=path, + row_number=row_number, + group_by=group_by, + from_csv=True, + ) + ) + if limit is not None and len(records) >= limit: + break + + if not records: + raise ValueError(f"{path}: input does not contain any records") + return records + + +def score_records( + detector: PredictDetector, + records: Sequence[InputRecord], + *, + min_confidence: float, + progress: bool = True, +) -> list[ScoredRecord]: + """Score records sequentially and return validated span lists.""" + if not math.isfinite(min_confidence) or not 0.0 <= min_confidence <= 1.0: + raise ValueError("min_confidence must be a finite number between 0 and 1") + + scored: list[ScoredRecord] = [] + total = len(records) + for index, record in enumerate(records, start=1): + predictions = detector.predict( + context=record["context"], + answer=record["answer"], + question=record["question"], + output_format="spans", + min_confidence=min_confidence, + ) + if not isinstance(predictions, list) or any( + not isinstance(span, Mapping) for span in predictions + ): + raise TypeError(f"record {index}: detector must return a list of span objects") + scored.append({"record": record, "spans": [dict(span) for span in predictions]}) + if progress: + logger.info("Scored %d/%d records", index, total) + return scored + + +def _rate(numerator: int, denominator: int) -> dict[str, int | float]: + """Build a rate object with its explicit numerator and denominator.""" + return { + "numerator": numerator, + "denominator": denominator, + "value": numerator / denominator if denominator else 0.0, + } + + +def _valid_confidence(value: object) -> float | None: + """Return a valid finite confidence in [0, 1], excluding booleans.""" + if isinstance(value, bool) or not isinstance(value, (int, float)): + return None + number = float(value) + return number if math.isfinite(number) and 0.0 <= number <= 1.0 else None + + +def _confidence_bin(value: float) -> str: + """Return a stable 0.1-wide confidence-bin label.""" + index = min(int(value * CONFIDENCE_BIN_COUNT), CONFIDENCE_BIN_COUNT - 1) + lower = index / CONFIDENCE_BIN_COUNT + upper = (index + 1) / CONFIDENCE_BIN_COUNT + closing = "]" if index == CONFIDENCE_BIN_COUNT - 1 else ")" + return f"[{lower:.1f}, {upper:.1f}{closing}" + + +def _preview(value: str, length: int = 240) -> str: + """Collapse whitespace and truncate text for a compact report example.""" + compact = " ".join(value.split()) + return compact if len(compact) <= length else compact[: length - 1] + "…" + + +def _report_span(span: Mapping[str, Any]) -> dict[str, Any]: + """Copy a span while replacing invalid confidence values with JSON-safe null.""" + result = dict(span) + if "confidence" in result and result["confidence"] is not None: + if _valid_confidence(result["confidence"]) is None: + result["confidence"] = None + result["confidence_valid"] = False + return result + + +def aggregate_results( + pairs: Sequence[ScoredRecord], + *, + group_by: str | None = None, + top_n: int = 10, +) -> dict[str, Any]: + """Aggregate detector output without loading or invoking a model.""" + if top_n < 0: + raise ValueError("top_n must be zero or greater") + + total = len(pairs) + flagged = sum(bool(pair["spans"]) for pair in pairs) + span_counts = Counter(len(pair["spans"]) for pair in pairs) + category_counts: Counter[str] = Counter() + subcategory_counts: Counter[str] = Counter() + confidence_counts: Counter[str] = Counter() + missing_confidence = 0 + invalid_confidence = 0 + valid_confidences = 0 + + bin_labels = [_confidence_bin(index / CONFIDENCE_BIN_COUNT) for index in range(10)] + confidence_counts.update({label: 0 for label in bin_labels}) + + for pair in pairs: + for span in pair["spans"]: + category = span.get("category") + if isinstance(category, str) and category.strip(): + category_counts[category] += 1 + subcategory = span.get("subcategory") + if isinstance(subcategory, str) and subcategory.strip(): + subcategory_counts[subcategory] += 1 + + if "confidence" not in span or span["confidence"] is None: + missing_confidence += 1 + continue + confidence = _valid_confidence(span["confidence"]) + if confidence is None: + invalid_confidence += 1 + continue + valid_confidences += 1 + confidence_counts[_confidence_bin(confidence)] += 1 + + groups: dict[str, Any] | None = None + if group_by is not None: + group_totals: Counter[str] = Counter() + group_flagged: Counter[str] = Counter() + for pair in pairs: + group = pair["record"]["group"] or MISSING_GROUP + group_totals[group] += 1 + if pair["spans"]: + group_flagged[group] += 1 + rates = { + group: _rate(group_flagged[group], denominator) + for group, denominator in sorted(group_totals.items()) + } + groups = { + "column": group_by, + "missing_value_label": MISSING_GROUP, + "denominator_sum": sum(group_totals.values()), + "rates": rates, + } + + ranked: list[tuple[int, float | None, ScoredRecord]] = [] + for index, pair in enumerate(pairs): + if not pair["spans"]: + continue + confidences = [ + confidence + for span in pair["spans"] + if (confidence := _valid_confidence(span.get("confidence"))) is not None + ] + ranked.append((index, max(confidences) if confidences else None, pair)) + ranked.sort(key=lambda item: (item[1] is None, -(item[1] or 0.0), item[0])) + + examples: list[dict[str, Any]] = [] + for _, max_confidence, pair in ranked[:top_n]: + record = pair["record"] + examples.append( + { + "row_number": record["row_number"], + "id": record["identifier"], + "group": record["group"], + "question": record["question"], + "answer_preview": _preview(record["answer"]), + "context_preview": _preview("\n".join(record["context"])), + "span_count": len(pair["spans"]), + "max_confidence": max_confidence, + "spans": [_report_span(span) for span in pair["spans"]], + } + ) + + return { + "schema_version": 1, + "summary": { + "rows_in": total, + "rows_scored": total, + "flagged_answers": flagged, + "hallucination_rate": _rate(flagged, total), + }, + "groups": groups, + "span_count_histogram": { + str(count): frequency for count, frequency in sorted(span_counts.items()) + }, + "span_confidence_histogram": { + "bins": dict(confidence_counts), + "valid_confidence_spans": valid_confidences, + "missing_confidence_spans": missing_confidence, + "invalid_confidence_spans": invalid_confidence, + "all_spans": sum(span_counts[count] * count for count in span_counts), + }, + "category_counts": dict(sorted(category_counts.items())), + "subcategory_counts": dict(sorted(subcategory_counts.items())), + "top_flagged_examples": examples, + } + + +def _markdown_cell(value: object) -> str: + """Escape an arbitrary scalar for one Markdown table cell.""" + if value is None: + return "—" + return str(value).replace("|", "\\|").replace("\n", " ") + + +def render_markdown(report: Mapping[str, Any]) -> str: + """Render a human-readable Markdown view of a canonical report object.""" + summary = report["summary"] + rate = summary["hallucination_rate"] + lines = [ + "# Hallucination rate report", + "", + "The rate is the fraction of scored answers with at least one flagged span.", + "", + "## Overall", + "", + "| Metric | Value |", + "| --- | ---: |", + f"| Rows in | {summary['rows_in']} |", + f"| Rows scored | {summary['rows_scored']} |", + f"| Flagged answers | {summary['flagged_answers']} |", + f"| Hallucination rate | {rate['value']:.2%} ({rate['numerator']}/{rate['denominator']}) |", + ] + + groups = report["groups"] + if groups is not None: + lines.extend( + [ + "", + f"## Rates by `{groups['column']}`", + "", + "| Group | Flagged | Denominator | Rate |", + "| --- | ---: | ---: | ---: |", + ] + ) + for group, group_rate in groups["rates"].items(): + lines.append( + f"| {_markdown_cell(group)} | {group_rate['numerator']} | " + f"{group_rate['denominator']} | {group_rate['value']:.2%} |" + ) + lines.append(f"\nGroup denominators sum to **{groups['denominator_sum']}**.") + + lines.extend( + [ + "", + "## Span count histogram", + "", + "| Spans per answer | Answers |", + "| ---: | ---: |", + ] + ) + for count, frequency in report["span_count_histogram"].items(): + lines.append(f"| {count} | {frequency} |") + + confidence = report["span_confidence_histogram"] + lines.extend( + [ + "", + "## Span confidence histogram", + "", + "Bins are left-inclusive and right-exclusive, except `[0.9, 1.0]`.", + "", + "| Confidence | Spans |", + "| --- | ---: |", + ] + ) + for label, count in confidence["bins"].items(): + lines.append(f"| `{label}` | {count} |") + lines.extend( + [ + f"| Missing confidence | {confidence['missing_confidence_spans']} |", + f"| Invalid confidence | {confidence['invalid_confidence_spans']} |", + f"| **All spans** | **{confidence['all_spans']}** |", + ] + ) + + for title, key in ( + ("Category counts", "category_counts"), + ("Subcategory counts", "subcategory_counts"), + ): + lines.extend(["", f"## {title}", ""]) + counts = report[key] + if not counts: + lines.append("No typed spans were reported.") + else: + lines.extend(["| Label | Spans |", "| --- | ---: |"]) + for label, count in counts.items(): + lines.append(f"| {_markdown_cell(label)} | {count} |") + + lines.extend(["", "## Top flagged examples", ""]) + examples = report["top_flagged_examples"] + if not examples: + lines.append("No answers were flagged.") + else: + lines.extend( + [ + "| Row | ID | Group | Max confidence | Spans | Answer preview |", + "| ---: | --- | --- | ---: | ---: | --- |", + ] + ) + for example in examples: + maximum = ( + "—" if example["max_confidence"] is None else f"{example['max_confidence']:.3f}" + ) + lines.append( + f"| {example['row_number']} | {_markdown_cell(example['id'])} | " + f"{_markdown_cell(example['group'])} | {maximum} | {example['span_count']} | " + f"{_markdown_cell(example['answer_preview'])} |" + ) + lines.append("") + lines.append("```json") + lines.append( + json.dumps(example["spans"], ensure_ascii=False, indent=2, allow_nan=False) + ) + lines.append("```") + + return "\n".join(lines).rstrip() + "\n" + + +def _atomic_write(path: Path, content: str) -> None: + """Replace a text output atomically after its full content is ready.""" + path.parent.mkdir(parents=True, exist_ok=True) + temporary_path: Path | None = None + try: + with tempfile.NamedTemporaryFile( + mode="w", + encoding="utf-8", + newline="", + dir=path.parent, + prefix=f".{path.name}.", + suffix=".tmp", + delete=False, + ) as handle: + handle.write(content) + temporary_path = Path(handle.name) + temporary_path.replace(path) + finally: + if temporary_path is not None and temporary_path.exists(): + temporary_path.unlink() + + +def write_reports(report: Mapping[str, Any], json_path: Path, markdown_path: Path) -> None: + """Write JSON and Markdown representations of the same report object.""" + json_content = json.dumps(report, ensure_ascii=False, indent=2, allow_nan=False) + "\n" + markdown_content = render_markdown(report) + _atomic_write(json_path, json_content) + _atomic_write(markdown_path, markdown_content) + + +def build_parser() -> argparse.ArgumentParser: + """Build the command-line parser.""" + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--input", type=Path, required=True, help="Input .jsonl or .csv file.") + parser.add_argument( + "--method", choices=["transformer", "llm"], default="transformer", help="Detector type." + ) + parser.add_argument( + "--model-path", + required=True, + help="Transformer model path/HF id, or the model name for --method llm.", + ) + parser.add_argument( + "--taxonomy-head", help="Optional transformer taxonomy-head path or Hugging Face id." + ) + parser.add_argument( + "--min-confidence", type=confidence_threshold, default=0.0, help="Span threshold in [0, 1]." + ) + parser.add_argument( + "--lang", default="en", help="Detector prompt/model language (default: en)." + ) + parser.add_argument("--group-by", help="Optional input column to aggregate rates by.") + parser.add_argument("--limit", type=positive_int, help="Score only the first N records.") + parser.add_argument( + "--top-n", + type=non_negative_int, + default=10, + help="Number of flagged examples (default: 10).", + ) + parser.add_argument("--out-json", type=Path, required=True, help="JSON report destination.") + parser.add_argument("--out-md", type=Path, required=True, help="Markdown report destination.") + return parser + + +def create_detector( + method: str, model_path: str, lang: str, taxonomy_head: str | None +) -> PredictDetector: + """Create a detector lazily with method-appropriate constructor arguments.""" + if method == "llm" and taxonomy_head is not None: + raise ValueError("--taxonomy-head is only supported with --method transformer") + + from lettucedetect.models.inference import HallucinationDetector + + kwargs: dict[str, Any] = {"lang": lang} + if method == "transformer": + kwargs["model_path"] = model_path + if taxonomy_head is not None: + kwargs["taxonomy_head"] = taxonomy_head + else: + kwargs["model"] = model_path + return cast(PredictDetector, HallucinationDetector(method=method, **kwargs)) + + +def main(argv: Sequence[str] | None = None, *, detector: PredictDetector | None = None) -> int: + """Run dataset scoring and write both report formats.""" + parser = build_parser() + args = parser.parse_args(argv) + + try: + records = load_records(args.input, group_by=args.group_by, limit=args.limit) + active_detector = detector or create_detector( + args.method, args.model_path, args.lang, args.taxonomy_head + ) + scored = score_records( + active_detector, records, min_confidence=args.min_confidence, progress=True + ) + report = aggregate_results(scored, group_by=args.group_by, top_n=args.top_n) + report["run"] = { + "input": str(args.input), + "method": args.method, + "model_path": args.model_path, + "taxonomy_head": args.taxonomy_head, + "min_confidence": args.min_confidence, + "lang": args.lang, + "group_by": args.group_by, + "limit": args.limit, + } + write_reports(report, args.out_json, args.out_md) + except (OSError, UnicodeError, ValueError, TypeError) as exc: + parser.exit(1, f"error: {exc}\n") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_report_hallucination_rates_pytest.py b/tests/test_report_hallucination_rates_pytest.py new file mode 100644 index 0000000..4009cce --- /dev/null +++ b/tests/test_report_hallucination_rates_pytest.py @@ -0,0 +1,444 @@ +"""Tests for the dataset-level hallucination-rate reporting script.""" + +import importlib.util +import json +import sys +from pathlib import Path +from types import ModuleType + +import pytest + + +def load_report_module() -> ModuleType: + """Load the reporting script without importing detector dependencies.""" + script = Path(__file__).parents[1] / "scripts" / "report_hallucination_rates.py" + spec = importlib.util.spec_from_file_location("report_hallucination_rates", script) + assert spec is not None + module = importlib.util.module_from_spec(spec) + assert spec.loader is not None + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +class StubDetector: + """Network-free detector that flags answers containing the word hallucinated.""" + + def __init__(self, *, fail_at: int | None = None) -> None: + """Initialize the stub and optional one-based failure position.""" + self.calls: list[dict] = [] + self.fail_at = fail_at + + def predict(self, **kwargs): + """Record the invocation and return a deterministic span list.""" + self.calls.append(kwargs) + if self.fail_at == len(self.calls): + raise ValueError("stub detector failed") + answer = kwargs["answer"] + if "hallucinated" not in answer: + return [] + start = answer.index("hallucinated") + return [ + { + "start": start, + "end": start + len("hallucinated"), + "text": "hallucinated", + "confidence": 0.9, + "category": "factual", + "subcategory": "contradiction", + } + ] + + +def make_record(module, index, group=None): + """Build a canonical record for pure aggregation tests.""" + return { + "row_number": index + 1, + "context": [f"context {index}"], + "answer": f"answer {index}", + "question": None, + "identifier": str(index), + "group": group, + } + + +class TestInputLoading: + """Input normalization tests for both supported file formats.""" + + def test_jsonl_normalizes_string_and_list_context_and_honors_limit(self, tmp_path): + """JSONL string/list contexts become the same canonical shape.""" + module = load_report_module() + path = tmp_path / "records.jsonl" + rows = [ + {"id": 1, "context": "one", "answer": "a", "source": "alpha"}, + { + "id": 2, + "context": ["two", "three"], + "question": "q?", + "answer": "b", + }, + {"id": 3, "context": "ignored", "answer": "c"}, + ] + path.write_text("\n".join(json.dumps(row) for row in rows), encoding="utf-8") + + records = module.load_records(path, group_by="source", limit=2) + + assert [record["context"] for record in records] == [["one"], ["two", "three"]] + assert records[0]["identifier"] == "1" + assert records[1]["question"] == "q?" + assert records[1]["group"] == module.MISSING_GROUP + + def test_csv_accepts_plain_and_json_array_context(self, tmp_path): + """CSV context cells support plain text and encoded string arrays.""" + module = load_report_module() + path = tmp_path / "records.csv" + path.write_text( + 'context,answer,source\nplain context,one,A\n"[""first"", ""second""]",two,B\n', + encoding="utf-8", + ) + + records = module.load_records(path, group_by="source") + + assert records[0]["context"] == ["plain context"] + assert records[1]["context"] == ["first", "second"] + assert [record["group"] for record in records] == ["A", "B"] + + def test_csv_keeps_bracket_prefixed_prose_as_literal_context(self, tmp_path): + """A bracket-prefixed prose context is not required to be a JSON array.""" + module = load_report_module() + path = tmp_path / "records.csv" + path.write_text( + "context,answer\n[Introduction] The source says hello,answer\n", + encoding="utf-8", + ) + + records = module.load_records(path) + + assert records[0]["context"] == ["[Introduction] The source says hello"] + + @pytest.mark.parametrize( + ("content", "match"), + [ + ('{"context": ["ok", 3], "answer": "a"}\n', "every context item"), + ('{"context": "ok"}\n', "missing required field.*answer"), + ("not json\n", "invalid JSON"), + ("\n", "blank JSONL"), + ], + ) + def test_jsonl_errors_include_line_number(self, tmp_path, content, match): + """Malformed rows fail loudly with their physical input line.""" + module = load_report_module() + path = tmp_path / "bad.jsonl" + path.write_text(content, encoding="utf-8") + + with pytest.raises(ValueError, match=match) as error: + module.load_records(path) + + assert f"{path}:1" in str(error.value) + + def test_csv_rejects_non_string_context_array_members(self, tmp_path): + """A JSON-like CSV array cannot smuggle non-string passages.""" + module = load_report_module() + path = tmp_path / "bad.csv" + path.write_text('context,answer\n"[""ok"", 2]",answer\n', encoding="utf-8") + + with pytest.raises(ValueError, match="every context item") as error: + module.load_records(path) + + assert f"{path}:2" in str(error.value) + + +class TestAggregation: + """Pure aggregation tests independent of any detector or model.""" + + def test_rates_histograms_typed_counts_and_groups_reconcile(self): + """Every requested count uses the scored-row population consistently.""" + module = load_report_module() + pairs = [ + {"record": make_record(module, 0, "A"), "spans": []}, + { + "record": make_record(module, 1, "A"), + "spans": [ + { + "text": "x", + "confidence": 0.0, + "category": "factual", + "subcategory": "contradiction", + } + ], + }, + { + "record": make_record(module, 2, module.MISSING_GROUP), + "spans": [ + {"text": "y", "confidence": 0.1, "category": "factual"}, + {"text": "z", "confidence": 1.0, "subcategory": "fabrication"}, + ], + }, + ] + + report = module.aggregate_results(pairs, group_by="source", top_n=5) + + assert report["summary"] == { + "rows_in": 3, + "rows_scored": 3, + "flagged_answers": 2, + "hallucination_rate": {"numerator": 2, "denominator": 3, "value": 2 / 3}, + } + assert report["groups"]["rates"]["A"]["denominator"] == 2 + assert report["groups"]["rates"]["A"]["numerator"] == 1 + assert report["groups"]["rates"][module.MISSING_GROUP]["denominator"] == 1 + assert report["groups"]["denominator_sum"] == 3 + assert report["span_count_histogram"] == {"0": 1, "1": 1, "2": 1} + assert report["span_confidence_histogram"]["bins"]["[0.0, 0.1)"] == 1 + assert report["span_confidence_histogram"]["bins"]["[0.1, 0.2)"] == 1 + assert report["span_confidence_histogram"]["bins"]["[0.9, 1.0]"] == 1 + assert report["span_confidence_histogram"]["all_spans"] == 3 + assert report["category_counts"] == {"factual": 2} + assert report["subcategory_counts"] == {"contradiction": 1, "fabrication": 1} + + def test_confidence_accounting_handles_missing_and_invalid_values(self): + """Bad confidence values are counted separately and remain valid JSON.""" + module = load_report_module() + pairs = [ + { + "record": make_record(module, 0), + "spans": [ + {"text": "a"}, + {"text": "b", "confidence": None}, + {"text": "c", "confidence": float("nan")}, + {"text": "d", "confidence": True}, + {"text": "e", "confidence": 1.2}, + {"text": "f", "confidence": 0.55}, + ], + } + ] + + report = module.aggregate_results(pairs, top_n=1) + histogram = report["span_confidence_histogram"] + + assert histogram["valid_confidence_spans"] == 1 + assert histogram["missing_confidence_spans"] == 2 + assert histogram["invalid_confidence_spans"] == 3 + assert histogram["all_spans"] == 6 + assert report["top_flagged_examples"][0]["spans"][2]["confidence"] is None + json.dumps(report, allow_nan=False) + + def test_top_n_ranks_confidence_stably_then_keeps_missing_confidence(self): + """Flagged examples use maximum confidence and stable source-order ties.""" + module = load_report_module() + pairs = [ + { + "record": make_record(module, 0), + "spans": [{"confidence": 0.6}, {"confidence": 0.9}], + }, + {"record": make_record(module, 1), "spans": [{"confidence": 0.9}]}, + {"record": make_record(module, 2), "spans": [{"text": "no score"}]}, + {"record": make_record(module, 3), "spans": []}, + ] + + report = module.aggregate_results(pairs, top_n=99) + examples = report["top_flagged_examples"] + + assert [example["id"] for example in examples] == ["0", "1", "2"] + assert examples[2]["max_confidence"] is None + assert module.aggregate_results(pairs, top_n=0)["top_flagged_examples"] == [] + + def test_markdown_uses_the_same_report_values(self): + """Markdown exposes the canonical JSON counts, denominator, and examples.""" + module = load_report_module() + pair = { + "record": make_record(module, 0, "docs"), + "spans": [{"text": "answer", "confidence": 0.8, "category": "factual"}], + } + report = module.aggregate_results([pair], group_by="source", top_n=1) + + markdown = module.render_markdown(report) + + assert "100.00% (1/1)" in markdown + assert "| docs | 1 | 1 | 100.00% |" in markdown + assert "| `[0.8, 0.9)` | 1 |" in markdown + assert "factual" in markdown + assert "answer" in markdown + + +class TestRunnerAndCli: + """Network-free scoring and CLI-shaped integration tests.""" + + def test_score_records_forwards_every_required_predict_argument(self): + """Each selected row produces exactly one span prediction.""" + module = load_report_module() + detector = StubDetector() + records = [make_record(module, index) for index in range(3)] + + scored = module.score_records(detector, records, min_confidence=0.42, progress=False) + + assert len(scored) == 3 + assert len(detector.calls) == 3 + assert all(call["output_format"] == "spans" for call in detector.calls) + assert all(call["min_confidence"] == 0.42 for call in detector.calls) + assert detector.calls[0]["context"] == ["context 0"] + + def test_ten_row_cli_shaped_run_writes_json_and_markdown(self, tmp_path, caplog): + """The accepted command shape completes over ten rows without a model.""" + module = load_report_module() + input_path = tmp_path / "sample.jsonl" + output_json = tmp_path / "nested" / "report.json" + output_markdown = tmp_path / "nested" / "report.md" + rows = [ + { + "id": index, + "context": [f"source passage {index}"], + "question": f"question {index}", + "answer": ( + f"hallucinated answer {index}" if index % 2 else f"supported answer {index}" + ), + "source": "odd" if index % 2 else "even", + } + for index in range(10) + ] + input_path.write_text("\n".join(json.dumps(row) for row in rows), encoding="utf-8") + detector = StubDetector() + + with caplog.at_level(module.logging.INFO, logger=module.__name__): + result = module.main( + [ + "--input", + str(input_path), + "--method", + "transformer", + "--model-path", + "unused-stub-model", + "--min-confidence", + "0.7", + "--group-by", + "source", + "--out-json", + str(output_json), + "--out-md", + str(output_markdown), + ], + detector=detector, + ) + + assert result == 0 + assert len(detector.calls) == 10 + report = json.loads(output_json.read_text(encoding="utf-8")) + assert report["summary"]["rows_in"] == 10 + assert report["summary"]["rows_scored"] == 10 + assert report["summary"]["hallucination_rate"]["denominator"] == 10 + assert report["summary"]["hallucination_rate"]["numerator"] == 5 + assert report["groups"]["denominator_sum"] == 10 + assert "Hallucination rate report" in output_markdown.read_text(encoding="utf-8") + assert caplog.messages[-1] == "Scored 10/10 records" + + def test_detector_failure_does_not_leave_reports(self, tmp_path): + """A failed scoring run never writes a successful-looking partial artifact.""" + module = load_report_module() + input_path = tmp_path / "sample.jsonl" + output_json = tmp_path / "report.json" + output_markdown = tmp_path / "report.md" + input_path.write_text( + "\n".join( + json.dumps({"context": "c", "answer": f"answer {index}"}) for index in range(3) + ), + encoding="utf-8", + ) + + with pytest.raises(SystemExit) as error: + module.main( + [ + "--input", + str(input_path), + "--model-path", + "unused", + "--out-json", + str(output_json), + "--out-md", + str(output_markdown), + ], + detector=StubDetector(fail_at=2), + ) + + assert error.value.code == 1 + assert not output_json.exists() + assert not output_markdown.exists() + + @pytest.mark.parametrize("value", ["-0.1", "1.1", "nan", "inf"]) + def test_parser_rejects_invalid_min_confidence(self, value): + """The CLI validates the detector threshold before any work begins.""" + module = load_report_module() + + with pytest.raises(SystemExit): + module.build_parser().parse_args( + [ + "--input", + "in.jsonl", + "--model-path", + "model", + "--min-confidence", + value, + "--out-json", + "out.json", + "--out-md", + "out.md", + ] + ) + + @pytest.mark.parametrize(("flag", "value"), [("--limit", "0"), ("--top-n", "-1")]) + def test_parser_rejects_invalid_count_options(self, flag, value): + """Limit is positive and top-N is non-negative by contract.""" + module = load_report_module() + + with pytest.raises(SystemExit): + module.build_parser().parse_args( + [ + "--input", + "in.jsonl", + "--model-path", + "model", + flag, + value, + "--out-json", + "out.json", + "--out-md", + "out.md", + ] + ) + + def test_create_detector_rejects_transformer_only_taxonomy_head(self): + """Detector construction owns validation of method-specific flags.""" + module = load_report_module() + + with pytest.raises(ValueError, match="only supported"): + module.create_detector("llm", "model", "en", "head") + + def test_main_surfaces_detector_configuration_value_error(self, tmp_path, monkeypatch): + """A detector configuration ValueError becomes a clean CLI exit code 1.""" + module = load_report_module() + input_path = tmp_path / "sample.jsonl" + input_path.write_text('{"context": "c", "answer": "a"}\n', encoding="utf-8") + monkeypatch.setattr( + module, + "create_detector", + lambda *_args: (_ for _ in ()).throw(ValueError("invalid detector configuration")), + ) + + with pytest.raises(SystemExit) as error: + module.main( + [ + "--input", + str(input_path), + "--method", + "llm", + "--model-path", + "model", + "--taxonomy-head", + "head", + "--out-json", + str(tmp_path / "out.json"), + "--out-md", + str(tmp_path / "out.md"), + ], + ) + + assert error.value.code == 1