"""Shared data containers for token-level hallucination baselines.""" from __future__ import annotations from dataclasses import dataclass from enum import Enum from math import isfinite from numbers import Integral from typing import Any, Mapping from evaluation.protocols import PREDICTION_SCHEMA_VERSION class BaselineDetectorName(str, Enum): """Supported baseline detector names for CLI/API selection.""" EPR = "epr" LOOKBACK = "lookback" ATTENTION_ENTROPY = "attention_entropy" HIDDEN_STATE_PROBE = "hidden_state_probe" SAPLMA = "saplma" HAMI = "hami" MULTI_VIEW_ATTENTION = "multi_view_attention" MVA_FEATURE_LOGISTIC = "mva_feature_logistic_baseline" MULTI_VIEW_ATTENTION_TRANSFORMER_CRF = "multi_view_attention_transformer_crf" def _coerce_labels(labels, name: str, allowed: set[int]) -> list[int] | None: if labels is None: return None allowed_str = {str(v) for v in allowed} text = "/".join(str(v) for v in sorted(allowed)) values = [] for label in labels: if isinstance(label, str): if label not in allowed_str: raise ValueError(f"{name} must contain only {text} values") values.append(int(label)) elif isinstance(label, Integral): value = int(label) if value not in allowed: raise ValueError(f"{name} must contain only {text} values") values.append(value) else: raise ValueError(f"{name} must contain only {text} values") return values def _coerce_binary_labels(labels: list[int] | None, name: str = "labels") -> list[int] | None: """Strict 0/1 labels — used by baseline example/prediction structs.""" return _coerce_labels(labels, name, {0, 1}) def _coerce_token_class_labels(labels: list[int] | None, name: str = "labels") -> list[int] | None: """Token-eval labels: 0=grounded, 1=hallucinated, 2=uncertain. The uncertain class is only emitted under the "uncertain" unknown-policy; other policies still yield 0/1 only. """ return _coerce_labels(labels, name, {0, 1, 2}) def _coerce_finite_floats(values, name: str) -> list[float]: coerced = [] for value in values: score = float(value) if not isfinite(score): raise ValueError(f"{name} must contain only finite values") coerced.append(score) return coerced @dataclass class BaselineExample: """One prompt/answer pair for training or evaluation.""" prompt: str answer: str labels: list[int] | None = None metadata: dict[str, Any] | None = None def __post_init__(self) -> None: if not isinstance(self.prompt, str): raise TypeError("prompt must be a string") if not isinstance(self.answer, str): raise TypeError("answer must be a string") if self.labels is not None and not isinstance(self.labels, list): raise TypeError("labels must be a list of ints or None") self.labels = _coerce_binary_labels(self.labels) if self.metadata is None: self.metadata = {} if not isinstance(self.metadata, dict): raise TypeError("metadata must be a dict") @dataclass class HallucinatedSpan: """A hallucinated character span inside a generated answer.""" start: int end: int text: str | None = None note: str | None = None def __post_init__(self) -> None: self.start = int(self.start) self.end = int(self.end) if self.start < 0: raise ValueError("span start must be nonnegative") if self.end < self.start: raise ValueError("span end must be >= start") if self.text is not None and not isinstance(self.text, str): raise TypeError("span text must be a string or None") if self.note is not None and not isinstance(self.note, str): raise TypeError("span note must be a string or None") @classmethod def from_json_dict(cls, span: Mapping[str, Any], answer: str) -> "HallucinatedSpan": if "start" in span and "end" in span: start = int(span["start"]) end = int(span["end"]) text = span.get("text") elif "text" in span: text = span["text"] start = answer.find(text) if start < 0: raise ValueError(f"hallucinated span text not found in answer: {text!r}") end = start + len(text) else: raise ValueError("hallucinated span requires either start/end or text") if end > len(answer): raise ValueError(f"span end exceeds answer length: {end}") if text is None: text = answer[start:end] elif text != answer[start:end]: raise ValueError("hallucinated span text does not match answer bounds") return cls(start=start, end=end, text=text, note=span.get("note")) def to_json_dict(self) -> dict[str, Any]: row = {"start": self.start, "end": self.end} if self.text is not None: row["text"] = self.text if self.note is not None: row["note"] = self.note return row @dataclass class LabelCounts: """Counts of auto-labelled token classes.""" grounded: int = 0 hallucinated: int = 0 unknown: int = 0 @classmethod def from_labels(cls, labels: list[str] | None) -> "LabelCounts": counts = cls() for label in labels or []: if label == "G": counts.grounded += 1 elif label == "H": counts.hallucinated += 1 elif label == "U": counts.unknown += 1 return counts @classmethod def from_json_dict(cls, row: Mapping[str, Any] | None) -> "LabelCounts | None": if row is None: return None return cls( grounded=int(row.get("G", 0)), hallucinated=int(row.get("H", 0)), unknown=int(row.get("U", 0)), ) def to_json_dict(self) -> dict[str, int]: row = {} if self.grounded: row["G"] = self.grounded if self.hallucinated: row["H"] = self.hallucinated if self.unknown: row["U"] = self.unknown return row @dataclass class TokenEvalMetadata: """Metadata carried with token-evaluation rows.""" source: str | None = None reference_answer: str | None = None generator_model: str | None = None generator_backend: str | None = None auto_label_method: str | None = None judge_model: str | None = None judge_rationale: str | None = None extra: dict[str, Any] | None = None generated_tokens: list[str] | None = None generated_token_ids: list[int] | None = None generated_logprobs: list[float] | None = None token_label_reasons: list[str] | None = None t_star_index: int | None = None original_label_counts: LabelCounts | None = None @classmethod def from_json_dict(cls, row: Mapping[str, Any] | None) -> "TokenEvalMetadata": if row is None: return cls() t_star_index = row.get("t_star_index") return cls( source=row.get("source"), reference_answer=row.get("reference_answer"), generator_model=row.get("generator_model"), generator_backend=row.get("generator_backend"), auto_label_method=row.get("auto_label_method"), judge_model=row.get("judge_model"), judge_rationale=row.get("judge_rationale"), extra=dict(row["extra"]) if isinstance(row.get("extra"), Mapping) else None, generated_tokens=list(row["generated_tokens"]) if row.get("generated_tokens") is not None else None, generated_token_ids=[int(token_id) for token_id in row["generated_token_ids"]] if row.get("generated_token_ids") is not None else None, generated_logprobs=[float(logprob) for logprob in row["generated_logprobs"]] if row.get("generated_logprobs") is not None else None, token_label_reasons=[str(reason) for reason in row["token_label_reasons"]] if row.get("token_label_reasons") is not None else None, t_star_index=None if t_star_index is None else int(t_star_index), original_label_counts=LabelCounts.from_json_dict(row.get("original_label_counts")), ) def to_json_dict(self) -> dict[str, Any]: row = {} if self.source is not None: row["source"] = self.source if self.reference_answer is not None: row["reference_answer"] = self.reference_answer if self.generator_model is not None: row["generator_model"] = self.generator_model if self.generator_backend is not None: row["generator_backend"] = self.generator_backend if self.auto_label_method is not None: row["auto_label_method"] = self.auto_label_method if self.judge_model is not None: row["judge_model"] = self.judge_model if self.judge_rationale is not None: row["judge_rationale"] = self.judge_rationale if self.extra is not None: row["extra"] = self.extra if self.generated_tokens is not None: row["generated_tokens"] = [str(token) for token in self.generated_tokens] if self.generated_token_ids is not None: row["generated_token_ids"] = [int(token_id) for token_id in self.generated_token_ids] if self.generated_logprobs is not None: row["generated_logprobs"] = [float(logprob) for logprob in self.generated_logprobs] if self.token_label_reasons is not None: row["token_label_reasons"] = [str(reason) for reason in self.token_label_reasons] if self.t_star_index is not None: row["t_star_index"] = self.t_star_index if self.original_label_counts is not None: row["original_label_counts"] = self.original_label_counts.to_json_dict() return row @dataclass class TokenOffset: """Character offset for one generated token.""" start: int end: int def __post_init__(self) -> None: self.start = int(self.start) self.end = int(self.end) if self.start < 0: raise ValueError("token offset start must be nonnegative") if self.end < self.start: raise ValueError("token offset end must be >= start") @dataclass class AutoLabelledRow: """One raw row from auto_labelled.jsonl after JSON-boundary parsing.""" id: str source: str | None question: str reference_answer: str | None generated_answer: str generator_model: str | None generator_backend: str | None auto_label_method: str | None t_star_index: int | None token_labels: list[str] | None gen_token_offsets: list[TokenOffset] | None error: str | None = None judge_model: str | None = None judge_rationale: str | None = None extra: dict[str, Any] | None = None gen_tokens: list[str] | None = None gen_token_ids: list[int] | None = None gen_logprobs: list[float] | None = None token_label_reasons: list[str] | None = None @classmethod def from_json_dict(cls, row: Mapping[str, Any]) -> "AutoLabelledRow": offsets = None if row.get("gen_token_offsets") is not None: offsets = [ TokenOffset(start=offset[0], end=offset[1]) for offset in row["gen_token_offsets"] ] t_star_index = row.get("t_star_index") return cls( id=row["id"], source=row.get("source"), question=row["question"], reference_answer=row.get("reference_answer"), generated_answer=row.get("generated_answer") or "", generator_model=row.get("generator_model"), generator_backend=row.get("generator_backend"), auto_label_method=row.get("auto_label_method"), t_star_index=None if t_star_index is None else int(t_star_index), token_labels=list(row["token_labels"]) if row.get("token_labels") is not None else None, gen_token_offsets=offsets, error=row.get("error"), judge_model=row.get("judge_model"), judge_rationale=row.get("judge_rationale"), extra=dict(row["extra"]) if isinstance(row.get("extra"), Mapping) else None, gen_tokens=list(row["gen_tokens"]) if row.get("gen_tokens") is not None else None, gen_token_ids=[int(token_id) for token_id in row["gen_token_ids"]] if row.get("gen_token_ids") is not None else None, gen_logprobs=[float(logprob) for logprob in row["gen_logprobs"]] if row.get("gen_logprobs") is not None else None, token_label_reasons=[str(reason) for reason in row["token_label_reasons"]] if row.get("token_label_reasons") is not None else None, ) def has_unknown_labels(self) -> bool: return self.token_labels is not None and "U" in self.token_labels @dataclass class TokenEvalRow: """One prompt/answer row for token-level baseline evaluation.""" prompt: str answer: str id: str | None = None labels: list[int] | None = None hallucinated_spans: list[HallucinatedSpan] | None = None span_labels_provided: bool | None = None metadata: TokenEvalMetadata | None = None def __post_init__(self) -> None: if not isinstance(self.prompt, str): raise TypeError("prompt must be a string") if not isinstance(self.answer, str): raise TypeError("answer must be a string") if self.id is not None and not isinstance(self.id, str): raise TypeError("id must be a string or None") if self.labels is not None: if not isinstance(self.labels, list): raise TypeError("labels must be a list of ints or None") self.labels = _coerce_token_class_labels(self.labels) if self.span_labels_provided is None: self.span_labels_provided = self.hallucinated_spans is not None else: self.span_labels_provided = bool(self.span_labels_provided) if self.hallucinated_spans is None: self.hallucinated_spans = [] self.hallucinated_spans = [ span if isinstance(span, HallucinatedSpan) else HallucinatedSpan.from_json_dict(span, self.answer) for span in self.hallucinated_spans ] if self.metadata is None: self.metadata = TokenEvalMetadata() elif not isinstance(self.metadata, TokenEvalMetadata): self.metadata = TokenEvalMetadata.from_json_dict(self.metadata) @classmethod def from_json_dict(cls, row: Mapping[str, Any]) -> "TokenEvalRow": if "prompt" not in row or "answer" not in row: raise ValueError("evaluation rows require prompt and answer") return cls( id=row.get("id"), prompt=row["prompt"], answer=row["answer"], labels=row.get("labels"), hallucinated_spans=row.get("hallucinated_spans"), span_labels_provided=row.get("hallucinated_spans") is not None, metadata=row.get("metadata"), ) def to_json_dict(self) -> dict[str, Any]: row = { "prompt": self.prompt, "answer": self.answer, "metadata": self.metadata.to_json_dict(), } if self.span_labels_provided or self.hallucinated_spans: row["hallucinated_spans"] = [span.to_json_dict() for span in self.hallucinated_spans] if self.id is not None: row["id"] = self.id if self.labels is not None: row["labels"] = self.labels return row def has_label_signal(self) -> bool: return self.labels is not None or bool(self.span_labels_provided) @dataclass class TokenPredictionRow: """One method's token-level predictions for one evaluation row. Prediction artifacts preserve the evaluator's full token-label vocabulary: 0=grounded, 1=hallucinated, and 2=uncertain. Binary metric/training consumers are responsible for masking uncertain tokens rather than changing their recorded labels. """ index: int id: str | None method: str tokens: list[str] scores: list[float] labels: list[int] | None seconds: float labels_aligned: bool metadata: dict[str, Any] | None = None prompt: str | None = None answer: str | None = None hallucinated_spans: list[HallucinatedSpan] | None = None example_metadata: TokenEvalMetadata | None = None token_ids: list[int] | None = None token_indices: list[int] | None = None score_valid_mask: list[bool] | None = None comparison_mask: list[bool] | None = None prediction_schema_version: int = PREDICTION_SCHEMA_VERSION def __post_init__(self) -> None: self.index = int(self.index) if self.id is not None and not isinstance(self.id, str): raise TypeError("id must be a string or None") if not isinstance(self.method, str): raise TypeError("method must be a string") self.tokens = [str(token) for token in self.tokens] self.scores = _coerce_finite_floats(self.scores, "scores") if len(self.tokens) != len(self.scores): raise ValueError("tokens and scores must have the same length") if self.labels is not None: self.labels = _coerce_token_class_labels(self.labels) self.seconds = float(self.seconds) self.labels_aligned = bool(self.labels_aligned) if self.labels_aligned: if self.labels is None: raise ValueError("labels_aligned=True requires labels") if len(self.labels) != len(self.scores): raise ValueError("aligned labels must have the same length as scores") if self.metadata is None: self.metadata = {} if not isinstance(self.metadata, dict): raise TypeError("metadata must be a dict") if self.prompt is not None and not isinstance(self.prompt, str): raise TypeError("prompt must be a string or None") if self.answer is not None and not isinstance(self.answer, str): raise TypeError("answer must be a string or None") if self.hallucinated_spans is None: self.hallucinated_spans = [] else: self.hallucinated_spans = [ self._coerce_prediction_span(span) for span in self.hallucinated_spans ] if self.example_metadata is None: self.example_metadata = TokenEvalMetadata() elif not isinstance(self.example_metadata, TokenEvalMetadata): self.example_metadata = TokenEvalMetadata.from_json_dict(self.example_metadata) n = len(self.scores) if self.token_ids is not None: self.token_ids = [int(token_id) for token_id in self.token_ids] if len(self.token_ids) != n: raise ValueError("token_ids must have the same length as scores") if self.token_indices is not None: self.token_indices = [int(token_index) for token_index in self.token_indices] if len(self.token_indices) != n: raise ValueError("token_indices must have the same length as scores") if len(set(self.token_indices)) != n: raise ValueError("token_indices must be unique within a prediction row") if self.score_valid_mask is not None: self.score_valid_mask = [bool(value) for value in self.score_valid_mask] if len(self.score_valid_mask) != n: raise ValueError("score_valid_mask must have the same length as scores") if self.comparison_mask is not None: self.comparison_mask = [bool(value) for value in self.comparison_mask] if len(self.comparison_mask) != n: raise ValueError("comparison_mask must have the same length as scores") if self.score_valid_mask is not None and any( compare and not valid for compare, valid in zip(self.comparison_mask, self.score_valid_mask) ): raise ValueError("comparison_mask cannot include invalid score slots") self.prediction_schema_version = int(self.prediction_schema_version) if self.prediction_schema_version not in {1, PREDICTION_SCHEMA_VERSION}: raise ValueError("unsupported prediction_schema_version") def _coerce_prediction_span(self, span: HallucinatedSpan | Mapping[str, Any]) -> HallucinatedSpan: if isinstance(span, HallucinatedSpan): return span if self.answer is not None: return HallucinatedSpan.from_json_dict(span, self.answer) if "start" not in span or "end" not in span: raise ValueError("prediction span text lookup requires answer context") return HallucinatedSpan( start=span["start"], end=span["end"], text=span.get("text"), note=span.get("note"), ) @classmethod def from_json_dict(cls, row: Mapping[str, Any]) -> "TokenPredictionRow": return cls( index=row["index"], id=row.get("id"), method=row["method"], tokens=list(row["tokens"]), scores=list(row["scores"]), labels=list(row["labels"]) if row.get("labels") is not None else None, seconds=row.get("seconds", 0.0), labels_aligned=row.get("labels_aligned", False), metadata=row.get("metadata"), prompt=row.get("prompt"), answer=row.get("answer"), hallucinated_spans=row.get("hallucinated_spans"), example_metadata=row.get("example_metadata"), token_ids=list(row["token_ids"]) if row.get("token_ids") is not None else None, token_indices=list(row["token_indices"]) if row.get("token_indices") is not None else None, score_valid_mask=list(row["score_valid_mask"]) if row.get("score_valid_mask") is not None else None, comparison_mask=list(row["comparison_mask"]) if row.get("comparison_mask") is not None else None, prediction_schema_version=row.get("prediction_schema_version", 1), ) def to_json_dict(self) -> dict[str, Any]: row = { "index": self.index, "id": self.id, "method": self.method, "tokens": self.tokens, "scores": self.scores, "labels": self.labels, "seconds": self.seconds, "labels_aligned": self.labels_aligned, } if self.prediction_schema_version >= PREDICTION_SCHEMA_VERSION: row["prediction_schema_version"] = self.prediction_schema_version if self.metadata: row["metadata"] = self.metadata if self.prompt is not None: row["prompt"] = self.prompt if self.answer is not None: row["answer"] = self.answer if self.hallucinated_spans: row["hallucinated_spans"] = [span.to_json_dict() for span in self.hallucinated_spans] if self.example_metadata and self.example_metadata.to_json_dict(): row["example_metadata"] = self.example_metadata.to_json_dict() if self.token_ids is not None: row["token_ids"] = self.token_ids if self.token_indices is not None: row["token_indices"] = self.token_indices if self.score_valid_mask is not None: row["score_valid_mask"] = self.score_valid_mask if self.comparison_mask is not None: row["comparison_mask"] = self.comparison_mask return row @dataclass class ResponsePredictionRow: """One response-level prediction for native EPR/WEPR evaluation.""" index: int id: str | None method: str score: float label: int | None seconds: float = 0.0 score_valid: bool = True comparison_valid: bool = True metadata: dict[str, Any] | None = None prediction_schema_version: int = PREDICTION_SCHEMA_VERSION def __post_init__(self) -> None: self.index = int(self.index) if self.id is not None and not isinstance(self.id, str): raise TypeError("id must be a string or None") if not isinstance(self.method, str): raise TypeError("method must be a string") self.score = float(self.score) if not isfinite(self.score): raise ValueError("score must be finite; use score_valid=False with a finite placeholder") if self.label is not None: self.label = int(self.label) if self.label not in {0, 1, 2}: raise ValueError("label must be 0, 1, 2, or None") self.seconds = float(self.seconds) self.score_valid = bool(self.score_valid) self.comparison_valid = bool(self.comparison_valid) if self.comparison_valid and (not self.score_valid or self.label not in {0, 1}): raise ValueError("comparison_valid requires a valid score and binary label") if self.metadata is None: self.metadata = {} if not isinstance(self.metadata, dict): raise TypeError("metadata must be a dict") self.prediction_schema_version = int(self.prediction_schema_version) if self.prediction_schema_version != PREDICTION_SCHEMA_VERSION: raise ValueError("unsupported response prediction schema version") @classmethod def from_json_dict(cls, row: Mapping[str, Any]) -> "ResponsePredictionRow": return cls( index=row["index"], id=row.get("id"), method=row["method"], score=row["score"], label=row.get("label"), seconds=row.get("seconds", 0.0), score_valid=row.get("score_valid", True), comparison_valid=row.get("comparison_valid", True), metadata=row.get("metadata"), prediction_schema_version=row.get( "prediction_schema_version", PREDICTION_SCHEMA_VERSION ), ) def to_json_dict(self) -> dict[str, Any]: row = { "prediction_schema_version": self.prediction_schema_version, "index": self.index, "id": self.id, "method": self.method, "score": self.score, "label": self.label, "seconds": self.seconds, "score_valid": self.score_valid, "comparison_valid": self.comparison_valid, } if self.metadata: row["metadata"] = self.metadata return row @dataclass class MethodScoreAggregate: """Accumulated scores and binary metric labels for one baseline method. All scores contribute to score-distribution statistics. Labelled metric arrays contain only aligned G/H labels; U labels (class 2) are deliberately excluded from binary AUPRC/AUROC/F1 inputs. """ method: str scores: list[float] | None = None labels: list[int] | None = None labelled_scores: list[float] | None = None def __post_init__(self) -> None: if self.scores is None: self.scores = [] if self.labels is None: self.labels = [] if self.labelled_scores is None: self.labelled_scores = [] def extend( self, scores: list[float], labels: list[int] | None, comparison_mask: list[bool] | None = None, ) -> None: score_values = _coerce_finite_floats(scores, "scores") label_values = None if labels is not None: label_values = _coerce_token_class_labels(list(labels)) or [] if len(label_values) != len(score_values): raise ValueError("aggregate labels must align with scores") if comparison_mask is not None: comparison_mask = [bool(value) for value in comparison_mask] if len(comparison_mask) != len(score_values): raise ValueError("aggregate comparison mask must align with scores") self.scores.extend(score_values) if label_values is not None: mask = comparison_mask or [True] * len(score_values) for score, label, keep in zip(score_values, label_values, mask): if label == 2 or not keep: continue self.labelled_scores.append(score) self.labels.append(label) def labels_or_none(self) -> list[int] | None: return self.labels if self.labels else None def labelled_scores_or_none(self) -> list[float] | None: return self.labelled_scores if self.labelled_scores else None @dataclass class ScoreSummary: """Aggregate statistics for one method's token scores.""" method: str n_tokens: int mean: float std: float min: float max: float p95: float auprc: float | None = None @dataclass class TokenScoreResult: """Per-token hallucination-risk scores for a generated answer.""" tokens: list[str] scores: list[float] method: str metadata: dict[str, Any] | None = None def __post_init__(self) -> None: self.tokens = [str(token) for token in self.tokens] self.scores = _coerce_finite_floats(self.scores, "scores") if len(self.tokens) != len(self.scores): raise ValueError("tokens and scores must have the same length") if not isinstance(self.method, str): raise TypeError("method must be a string") if self.metadata is None: self.metadata = {} if not isinstance(self.metadata, dict): raise TypeError("metadata must be a dict") @dataclass class LogisticConfig: """Configuration for sklearn logistic-regression heads.""" max_iter: int = 1000 C: float = 1.0 class_weight: str | dict | None = None solver: str = "lbfgs" random_state: int | None = None penalty: str | None = "l2"