"""Run selected token-score baselines on a HuggingFace causal LM. Examples: python scripts/eval_baselines_dgx.py \ --model Qwen/Qwen2.5-0.5B-Instruct \ --prompt "What is the capital of France?" \ --answer "The capital is Paris." \ --detector epr python scripts/eval_baselines_dgx.py \ --model Qwen/Qwen2.5-3B-Instruct \ --input examples.jsonl \ --device-map auto \ --output predictions.jsonl \ --detector epr \ --detector lookback Input JSONL rows should contain: {"prompt": "...", "answer": "...", "labels": [0, 1, ...]} Labels are optional. If provided, they must already align to the tokenizer's answer tokens for AUPRC to be computed. Rows may also provide `hallucinated_spans`, which are answer-text spans that this script converts to token labels with tokenizer offset mappings. Example span-labeled row: { "prompt": "What is the capital of France?", "answer": "The capital is Canberra.", "hallucinated_spans": [{"text": "Canberra."}] } """ from __future__ import annotations import argparse import hashlib import json import os import re import sys import time from pathlib import Path import numpy as np REPO_ROOT = Path(__file__).resolve().parents[1] if str(REPO_ROOT) not in sys.path: sys.path.insert(0, str(REPO_ROOT)) from baselines import ( AttentionEntropyDetector, BaselineDetectorName, EPRDetector, HallucinatedSpan, HiddenStateProbeDetector, LookbackLensDetector, MethodScoreAggregate, MultiViewAttentionDetector, ScoreSummary, TokenEvalRow, TokenPredictionRow, ) from metrics import auprc PROMPT_TEMPLATE_RAW = "raw" PROMPT_TEMPLATE_TANAZZAH_LLAMA_CHAT = "tanazzah-llama-chat" TANAZZAH_SYSTEM_PROMPT = "Answer the question concisely and factually." TANAZZAH_PROMPT_MAX_LENGTH = 2048 TANAZZAH_CONTEXT_TOKEN_BUDGET = 1600 TANAZZAH_CONTEXT_KEYS = ( "context", "knowledge", "passage", "evidence", "supporting_context", ) IMMUTABLE_REVISION_RE = re.compile(r"[0-9a-fA-F]{40,64}") def detector_choices() -> list[str]: return [detector.value for detector in BaselineDetectorName] def parse_args(argv=None): parser = argparse.ArgumentParser(description="Evaluate token-score baselines on DGX/HF models.") parser.add_argument("--model", required=True, help="HuggingFace model id or local path.") parser.add_argument( "--model-revision", default=None, help="Revision loaded for both model and tokenizer (prefer an immutable commit SHA).", ) parser.add_argument( "--allow-unpinned-model", action="store_true", help="diagnostic only: allow a missing or mutable model revision", ) parser.add_argument("--input", help="Optional JSONL file with prompt/answer rows.") parser.add_argument("--prompt", help="Prompt for a single-example run.") parser.add_argument("--answer", help="Answer for a single-example run.") parser.add_argument("--output", help="Optional JSONL path for per-example predictions.") parser.add_argument( "--detector", action="append", required=True, choices=detector_choices(), help="Baseline detector to run. Repeat to run multiple detectors.", ) parser.add_argument("--top-k", type=int, default=15, help="Top-k probabilities for EPR.") parser.add_argument("--window-size", type=int, default=8, help="Lookback Lens trailing window size.") parser.add_argument("--hidden-layer", type=int, default=-1, help="Hidden layer index for hidden-state probe features.") parser.add_argument("--device", default="cuda", help="Device for non-device-map loading.") parser.add_argument("--device-map", default=None, help='Pass "auto" for sharded/Accelerate loading.') parser.add_argument("--dtype", default="bfloat16", choices=["auto", "float32", "float16", "bfloat16"]) parser.add_argument("--attn-implementation", default="eager", help="HF attention implementation; eager is safest for attentions.") parser.add_argument( "--prompt-template", default=PROMPT_TEMPLATE_RAW, choices=[PROMPT_TEMPLATE_RAW, PROMPT_TEMPLATE_TANAZZAH_LLAMA_CHAT], help="Prompt wrapper for scoring. Use tanazzah-llama-chat to replay Tanazzah's LLaMA generation prompt.", ) parser.add_argument("--trust-remote-code", action="store_true") parser.add_argument("--require-labels", action="store_true", help="Fail if an input row has no labels or hallucinated_spans.") parser.add_argument("--strict-labels", action="store_true", help="Fail instead of warning when labels do not align to answer tokens.") parser.add_argument( "--allow-token-id-model-mismatch", action="store_true", help="When saved generated_token_ids came from another model, re-tokenize answer text instead of failing.", ) parser.add_argument( "--allow-unverified-replay", action="store_true", help=( "diagnostic only: permit missing/unusable saved answer IDs to be " "retokenized and reuse saved IDs when generation-time model/" "tokenizer or prompt provenance is missing or differs" ), ) return parser.parse_args(argv) def load_examples(args): if args.input: examples = [] with open(args.input, "r", encoding="utf-8") as f: for line_no, line in enumerate(f, start=1): if not line.strip(): continue row = json.loads(line) if "prompt" not in row or "answer" not in row: raise ValueError(f"{args.input}:{line_no} requires prompt and answer") examples.append(TokenEvalRow.from_json_dict(row)) return examples if args.prompt is None or args.answer is None: raise ValueError("provide either --input or both --prompt and --answer") return [TokenEvalRow(prompt=args.prompt, answer=args.answer)] def labels_from_hallucinated_spans(answer: str, tokenizer, spans): """Convert answer character spans into tokenizer-aligned 0/1 labels.""" try: encoded = tokenizer(answer, add_special_tokens=False, return_offsets_mapping=True) except (NotImplementedError, TypeError) as exc: raise ValueError("hallucinated_spans require a fast tokenizer with offset mappings") from exc offsets = encoded.get("offset_mapping") if offsets is None: raise ValueError("tokenizer did not return offset mappings for hallucinated_spans") span_bounds = [ (span.start, span.end) for span in ( span if isinstance(span, HallucinatedSpan) else HallucinatedSpan.from_json_dict(span, answer) for span in spans ) ] for start, end in span_bounds: if start < 0 or end < start or end > len(answer): raise ValueError(f"invalid hallucinated span bounds: start={start}, end={end}") labels = [] for offset in offsets: start, end = int(offset[0]), int(offset[1]) if end <= start: labels.append(0) continue overlaps_span = any(max(start, span_start) < min(end, span_end) for span_start, span_end in span_bounds) labels.append(1 if overlaps_span else 0) return labels def resolve_labels(example, tokenizer): if example.labels is not None: return [int(label) for label in example.labels] if example.span_labels_provided: return labels_from_hallucinated_spans(example.answer, tokenizer, example.hallucinated_spans) return None def prompt_add_special_tokens(prompt_template: str) -> bool: if prompt_template in {PROMPT_TEMPLATE_RAW, PROMPT_TEMPLATE_TANAZZAH_LLAMA_CHAT}: return True raise ValueError(f"unknown prompt template: {prompt_template}") def prompt_max_length(prompt_template: str) -> int | None: if prompt_template == PROMPT_TEMPLATE_TANAZZAH_LLAMA_CHAT: return TANAZZAH_PROMPT_MAX_LENGTH if prompt_template == PROMPT_TEMPLATE_RAW: return None raise ValueError(f"unknown prompt template: {prompt_template}") def tanazzah_generation_context(extra) -> str | None: """Mirror 02_generate_answers._extract_context exactly.""" if not isinstance(extra, dict): return None for key in TANAZZAH_CONTEXT_KEYS: value = extra.get(key) if value: text = str(value).strip() if text: return text return None def fit_tanazzah_generation_context(context: str, tokenizer) -> str: """Mirror the generator's 1,600-token context-only truncation.""" encoded = tokenizer(context, add_special_tokens=False) token_ids = encoded["input_ids"] if hasattr(token_ids, "tolist"): token_ids = token_ids.tolist() if token_ids and isinstance(token_ids[0], list): token_ids = token_ids[0] if len(token_ids) <= TANAZZAH_CONTEXT_TOKEN_BUDGET: return context return tokenizer.decode( token_ids[:TANAZZAH_CONTEXT_TOKEN_BUDGET], skip_special_tokens=True, ) def scoring_prompt(example: TokenEvalRow, tokenizer, prompt_template: str) -> str: if prompt_template == PROMPT_TEMPLATE_RAW: return example.prompt if prompt_template != PROMPT_TEMPLATE_TANAZZAH_LLAMA_CHAT: raise ValueError(f"unknown prompt template: {prompt_template}") system_prompt = TANAZZAH_SYSTEM_PROMPT extra = getattr(example.metadata, "extra", None) context = tanazzah_generation_context(extra) if context: system_prompt += ( f"\nContext: {fit_tanazzah_generation_context(context, tokenizer)}" ) messages = [ {"role": "system", "content": system_prompt}, {"role": "user", "content": example.prompt}, ] template_kwargs = {} v5_3 = extra.get("v5_3") if isinstance(extra, dict) else None replay = v5_3.get("replay") if isinstance(v5_3, dict) else None if isinstance(replay, dict) and replay.get("canonical_date_string"): template_kwargs["date_string"] = str(replay["canonical_date_string"]) try: # Generation (02_generate_answers.py) passes enable_thinking=False for # tokenizers that support it (Qwen3), with a TypeError fallback for the # rest. Replay must render the *identical* prompt string, so the same # attempt-then-fallback is mirrored here: Qwen3 with the default # enable_thinking=True renders a different prompt than generation used, # which would shift every token position. try: return tokenizer.apply_chat_template( messages, tokenize=False, add_generation_prompt=True, enable_thinking=False, **template_kwargs, ) except TypeError: return tokenizer.apply_chat_template( messages, tokenize=False, add_generation_prompt=True, **template_kwargs, ) except Exception as exc: raise ValueError("tanazzah-llama-chat requires a tokenizer with apply_chat_template") from exc def saved_answer_token_ids( example: TokenEvalRow, model_id: str | None = None, model_revision: str | None = None, allow_model_mismatch: bool = False, allow_unverified_revision: bool = False, require_saved: bool = False, ) -> list[int] | None: token_ids = getattr(example.metadata, "generated_token_ids", None) if not token_ids: if require_saved: raise ValueError( "reportable dataset replay requires saved generated_token_ids " f"for id={example.id!r}; use --allow-unverified-replay only " "for a clearly marked diagnostic retokenization" ) return None generator_model = getattr(example.metadata, "generator_model", None) if model_id and generator_model and model_id != generator_model: if allow_model_mismatch: if require_saved: raise ValueError( "saved generated_token_ids cannot be used because their " f"generator model is {generator_model!r}, while the scoring " f"model is {model_id!r}; add --allow-unverified-replay only " "for a clearly marked diagnostic retokenization" ) return None raise ValueError( "saved generated_token_ids were produced by " f"{generator_model!r}, but the scoring model is {model_id!r}; " "use the generator model for exact replay or pass " "--allow-token-id-model-mismatch to re-tokenize answer text" ) if model_revision is not None and not allow_unverified_revision: extra = getattr(example.metadata, "extra", None) generation = extra.get("generation") if isinstance(extra, dict) else None v5_3 = extra.get("v5_3") if isinstance(extra, dict) else None replay = v5_3.get("replay") if isinstance(v5_3, dict) else None generated_model_revision = ( generation.get("model_revision") if isinstance(generation, dict) else None ) generated_tokenizer_revision = ( generation.get("tokenizer_revision") if isinstance(generation, dict) else None ) if isinstance(replay, dict) and replay.get("status") == "canonical_teacher_forced_replay": if str(replay.get("model_id") or "") != str(model_id or ""): raise ValueError( "v5.3 replay model does not match the scoring model: " f"replay={replay.get('model_id')!r}, scoring={model_id!r}" ) if str(replay.get("model_revision") or "") != str(model_revision): raise ValueError( "v5.3 replay model revision does not match the scoring revision" ) if str(replay.get("tokenizer_id") or "") != str(model_id or ""): raise ValueError( "v5.3 replay tokenizer does not come from the scoring model" ) if str(replay.get("tokenizer_revision") or "") != str(model_revision): raise ValueError( "v5.3 replay tokenizer revision does not match the scoring revision" ) if replay.get("answer_decode_verified") is not True: raise ValueError("v5.3 replay has not verified answer-token decoding") expected_ids_sha256 = replay.get("answer_token_ids_sha256") actual_ids_sha256 = hashlib.sha256( json.dumps( [int(token_id) for token_id in token_ids], separators=(",", ":") ).encode("utf-8") ).hexdigest() if str(expected_ids_sha256 or "") != actual_ids_sha256: raise ValueError("v5.3 replay answer token IDs have changed") return [int(token_id) for token_id in token_ids] if str(generated_model_revision or "") != str(model_revision): raise ValueError( "saved generated_token_ids cannot be verified against the " "scoring model revision: " f"generated={generated_model_revision!r}, " f"scoring={model_revision!r}; use --allow-unverified-replay " "only for a clearly marked diagnostic run" ) if str(generated_tokenizer_revision or "") != str(model_revision): raise ValueError( "saved generated_token_ids do not carry a matching tokenizer " "revision: " f"generated={generated_tokenizer_revision!r}, " f"scoring={model_revision!r}; exact token replay is unproven. " "Use --allow-unverified-replay only for a clearly marked " "diagnostic run" ) return [int(token_id) for token_id in token_ids] def validate_saved_answer_decoding( example: TokenEvalRow, tokenizer, token_ids: list[int] ) -> None: """Reject token IDs that the loaded tokenizer does not decode to the answer.""" kwargs = { "skip_special_tokens": False, "clean_up_tokenization_spaces": False, } try: decoded = tokenizer.decode(token_ids, **kwargs) except TypeError: decoded = tokenizer.decode(token_ids, skip_special_tokens=False) if decoded != example.answer: raise ValueError( "saved generated_token_ids do not decode to the recorded answer " f"for id={example.id!r}; scoring this row would misalign labels" ) def validate_generation_prompt_replay( example: TokenEvalRow, tokenizer, prompt: str, prompt_template: str, *, allow_unverified: bool, ) -> str: """Verify the effective scoring prompt token IDs when provenance exists.""" if not getattr(example.metadata, "generated_token_ids", None): return "not_applicable_no_saved_generation" extra = getattr(example.metadata, "extra", None) generation = extra.get("generation") if isinstance(extra, dict) else None v5_3 = extra.get("v5_3") if isinstance(extra, dict) else None replay = v5_3.get("replay") if isinstance(v5_3, dict) else None if isinstance(replay, dict) and replay.get("status") == "canonical_teacher_forced_replay": if str(replay.get("prompt_template") or "") != prompt_template: raise ValueError( "v5.3 replay prompt template does not match the scoring template" ) expected = replay.get("effective_prompt_token_ids_sha256") expected_ids = replay.get("effective_prompt_token_ids") verified_status = "verified_v5_3_canonical_prompt_token_ids" actual_identity = v5_3_tokenizer_identity(tokenizer) for field in ( "tokenizer_backend_sha256", "chat_template_sha256", "special_tokens_sha256", ): if str(replay.get(field) or "") != str(actual_identity[field] or ""): raise ValueError( f"loaded tokenizer does not match the sealed v5.3 {field}" ) else: expected = ( generation.get("effective_prompt_token_ids_sha256") if isinstance(generation, dict) else None ) expected_ids = None verified_status = "verified_generation_prompt_token_ids" if expected is None: if allow_unverified: return "unverified_missing_generation_prompt_hash" raise ValueError( "generation-time effective prompt token IDs were not recorded for " f"id={example.id!r}; use --allow-unverified-replay only for a " "clearly marked diagnostic run" ) kwargs = { "add_special_tokens": prompt_add_special_tokens(prompt_template), } max_length = prompt_max_length(prompt_template) if max_length is not None: kwargs.update({"truncation": True, "max_length": max_length}) encoded = tokenizer(prompt, **kwargs)["input_ids"] if hasattr(encoded, "tolist"): encoded = encoded.tolist() if encoded and isinstance(encoded[0], list): encoded = encoded[0] token_ids = [int(value) for value in encoded] if expected_ids is not None and [int(value) for value in expected_ids] != token_ids: if allow_unverified: return "unverified_generation_prompt_mismatch" raise ValueError( "scoring prompt token IDs do not match the sealed v5.3 replay " f"for id={example.id!r}" ) actual = hashlib.sha256( json.dumps(token_ids, separators=(",", ":")).encode("utf-8") ).hexdigest() if str(expected) != actual: if allow_unverified: return "unverified_generation_prompt_mismatch" raise ValueError( "scoring prompt token IDs do not match generation-time provenance " f"for id={example.id!r}" ) return verified_status def v5_3_tokenizer_identity(tokenizer) -> dict[str, str | None]: """Content-address the tokenizer once per process for v5.3 replay.""" cached = getattr(tokenizer, "_latticespike_v5_3_identity", None) if isinstance(cached, dict): return cached try: chat_template = tokenizer.get_chat_template() except (AttributeError, TypeError, ValueError): # ValueError: newer transformers raises when no template is set. A # tokenizer without one is legitimate -- --prompt-template raw needs no # chat template -- and "absent" is a fact worth recording, not a crash. chat_template = getattr(tokenizer, "chat_template", None) backend = getattr(tokenizer, "backend_tokenizer", None) backend_json = backend.to_str() if backend is not None else None special = { "all_special_ids": [ int(value) for value in (getattr(tokenizer, "all_special_ids", None) or []) ], "bos_token_id": getattr(tokenizer, "bos_token_id", None), "eos_token_id": getattr(tokenizer, "eos_token_id", None), "pad_token_id": getattr(tokenizer, "pad_token_id", None), } identity = { "tokenizer_backend_sha256": None if backend_json is None else hashlib.sha256(backend_json.encode("utf-8")).hexdigest(), "chat_template_sha256": None if not isinstance(chat_template, str) else hashlib.sha256(chat_template.encode("utf-8")).hexdigest(), "special_tokens_sha256": hashlib.sha256( json.dumps( special, sort_keys=True, separators=(",", ":"), ensure_ascii=False, ).encode("utf-8") ).hexdigest(), } setattr(tokenizer, "_latticespike_v5_3_identity", identity) return identity def validate_model_revision( revision: str | None, *, allow_unpinned: bool ) -> None: if allow_unpinned: return if IMMUTABLE_REVISION_RE.fullmatch(str(revision or "")) is None: raise ValueError( "--model-revision must be an immutable 40-64 digit hexadecimal " "commit hash for reportable runs; use --allow-unpinned-model only " "for diagnostics" ) def resolved_component_revision(component) -> str | None: """Best-effort resolved Hugging Face commit for run provenance.""" for holder in ( component, getattr(component, "config", None), getattr(component, "init_kwargs", None), ): if holder is None: continue if isinstance(holder, dict): value = holder.get("_commit_hash") or holder.get("revision") else: value = getattr(holder, "_commit_hash", None) or getattr( holder, "revision", None ) if value: return str(value) for attribute in ("vocab_file", "name_or_path"): value = getattr(component, attribute, None) if isinstance(value, str) and "snapshots" in value: parts = value.split(os.sep) try: return parts[parts.index("snapshots") + 1] except (ValueError, IndexError): pass return None def torch_dtype(dtype_name): import torch if dtype_name == "auto": return "auto" if dtype_name == "float32": return torch.float32 if dtype_name == "float16": return torch.float16 if dtype_name == "bfloat16": return torch.bfloat16 raise ValueError(f"unknown dtype: {dtype_name}") def load_model_and_tokenizer(args): from transformers import AutoModelForCausalLM, AutoTokenizer revision = getattr(args, "model_revision", None) tokenizer = AutoTokenizer.from_pretrained( args.model, revision=revision, trust_remote_code=args.trust_remote_code, ) if args.prompt_template == PROMPT_TEMPLATE_TANAZZAH_LLAMA_CHAT: # The generator's final 2,048-token safety net drops from the left so # the question and assistant marker survive pathological prompts. tokenizer.truncation_side = "left" model_kwargs = { "revision": revision, "dtype": torch_dtype(args.dtype), "trust_remote_code": args.trust_remote_code, "attn_implementation": args.attn_implementation, } if args.device_map: model_kwargs["device_map"] = args.device_map try: model = AutoModelForCausalLM.from_pretrained(args.model, **model_kwargs) except TypeError: # Older Transformers/model classes may reject `dtype` (which they spell # `torch_dtype`) or `attn_implementation`. Retry the rename on its own # first: dropping attn_implementation as well would silently fall back # to sdpa, which returns no attention tensors, and every attention-based # baseline would then fail much later with an unrelated-looking error. model_kwargs["torch_dtype"] = model_kwargs.pop("dtype") try: model = AutoModelForCausalLM.from_pretrained(args.model, **model_kwargs) except TypeError: model_kwargs.pop("attn_implementation", None) model = AutoModelForCausalLM.from_pretrained(args.model, **model_kwargs) if not args.device_map and args.device: model.to(args.device) model.eval() if tokenizer.pad_token is None and tokenizer.eos_token is not None: tokenizer.pad_token = tokenizer.eos_token # Some Transformers tokenizer classes do not retain `_commit_hash` even # when `from_pretrained(..., revision=)` was used. The # requested SHA is therefore the truthful resolution fallback after a # successful pinned load; content hashes below provide an independent # identity check for sealed v5.3 replay. args.resolved_model_revision = resolved_component_revision(model) or revision args.resolved_tokenizer_revision = ( resolved_component_revision(tokenizer) or revision ) args.resolved_tokenizer_identity = v5_3_tokenizer_identity(tokenizer) resolved = [ value for value in ( args.resolved_model_revision, args.resolved_tokenizer_revision, ) if value is not None ] if revision is not None and any(value != str(revision) for value in resolved): raise ValueError( "loaded model/tokenizer revision differs from --model-revision: " f"requested={revision!r}, model={args.resolved_model_revision!r}, " f"tokenizer={args.resolved_tokenizer_revision!r}" ) return model, tokenizer def summarize(method, scores, labels=None, labelled_scores=None): arr = np.asarray(scores, dtype=float) auprc_value = None if labels is not None: auprc_scores = np.asarray(labelled_scores if labelled_scores is not None else scores, dtype=float) if len(labels) == len(auprc_scores): auprc_value = auprc(auprc_scores, labels) return ScoreSummary( method=method, n_tokens=int(arr.size), mean=float(arr.mean()) if arr.size else float("nan"), std=float(arr.std()) if arr.size else float("nan"), min=float(arr.min()) if arr.size else float("nan"), max=float(arr.max()) if arr.size else float("nan"), p95=float(np.percentile(arr, 95)) if arr.size else float("nan"), auprc=auprc_value, ) def print_table(rows: list[ScoreSummary]): method_width = max([19] + [len(row.method) for row in rows]) print(f"\n{'method':<{method_width}} tokens mean p95 max auprc") print(f"{'-' * method_width} ------- -------- -------- -------- --------") for row in rows: auprc_val = row.auprc if row.auprc is not None else float("nan") print( f"{row.method:<{method_width}} " f"{row.n_tokens:>7d} " f"{row.mean:>8.4f} " f"{row.p95:>8.4f} " f"{row.max:>8.4f} " f"{auprc_val:>8.4f}" ) def aggregate_for_method(aggregates: list[MethodScoreAggregate], method: str) -> MethodScoreAggregate: for aggregate in aggregates: if aggregate.method == method: return aggregate aggregate = MethodScoreAggregate(method=method) aggregates.append(aggregate) return aggregate def selected_detector_names(args) -> list[BaselineDetectorName]: names = [] seen = set() for detector in args.detector: name = BaselineDetectorName(detector) if name in seen: continue seen.add(name) names.append(name) return names def build_detectors(args, model, tokenizer, device_for_detectors): detectors = [] add_special_tokens = prompt_add_special_tokens(args.prompt_template) max_length = prompt_max_length(args.prompt_template) for name in selected_detector_names(args): if name == BaselineDetectorName.EPR: detectors.append( EPRDetector( model=model, tokenizer=tokenizer, top_k=args.top_k, device=device_for_detectors, prompt_add_special_tokens=add_special_tokens, prompt_max_length=max_length, ) ) elif name == BaselineDetectorName.LOOKBACK: detectors.append( LookbackLensDetector( model=model, tokenizer=tokenizer, window_size=args.window_size, device=device_for_detectors, prompt_add_special_tokens=add_special_tokens, prompt_max_length=max_length, ) ) elif name == BaselineDetectorName.ATTENTION_ENTROPY: detectors.append( AttentionEntropyDetector( model=model, tokenizer=tokenizer, device=device_for_detectors, prompt_add_special_tokens=add_special_tokens, prompt_max_length=max_length, ) ) elif name == BaselineDetectorName.HIDDEN_STATE_PROBE: detectors.append( HiddenStateProbeDetector( model=model, tokenizer=tokenizer, hidden_layer=args.hidden_layer, device=device_for_detectors, prompt_add_special_tokens=add_special_tokens, prompt_max_length=max_length, ) ) elif name in {BaselineDetectorName.MULTI_VIEW_ATTENTION, BaselineDetectorName.MVA_FEATURE_LOGISTIC}: detectors.append( MultiViewAttentionDetector( model=model, tokenizer=tokenizer, device=device_for_detectors, prompt_add_special_tokens=add_special_tokens, prompt_max_length=max_length, ) ) elif name == BaselineDetectorName.MULTI_VIEW_ATTENTION_TRANSFORMER_CRF: raise ValueError("multi_view_attention_transformer_crf is trainable; use scripts/train_faithful_baselines_dgx.py") else: # pragma: no cover - argparse enum choices prevent this. raise ValueError(f"unsupported detector: {name}") return detectors def prediction_metadata( args, result_metadata: dict, used_saved_answer_token_ids: bool, prompt_replay_status: str | None = None, ) -> dict: """Metadata that makes a prediction row auditable after the DGX run.""" return { **result_metadata, "scoring_model": args.model, "scoring_model_revision": getattr(args, "model_revision", None), "resolved_scoring_model_revision": getattr( args, "resolved_model_revision", None ), "resolved_scoring_tokenizer_revision": getattr( args, "resolved_tokenizer_revision", None ), "prompt_template": args.prompt_template, "prompt_add_special_tokens": prompt_add_special_tokens(args.prompt_template), "prompt_max_length": prompt_max_length(args.prompt_template), "used_saved_answer_token_ids": bool(used_saved_answer_token_ids), "allow_token_id_model_mismatch": bool(args.allow_token_id_model_mismatch), "allow_unverified_replay": bool(args.allow_unverified_replay), "prompt_replay_status": prompt_replay_status, "dtype": args.dtype, "attn_implementation": args.attn_implementation, } def main(argv=None): args = parse_args(argv) validate_model_revision( args.model_revision, allow_unpinned=args.allow_unpinned_model ) examples = load_examples(args) model, tokenizer = load_model_and_tokenizer(args) device_for_detectors = None if args.device_map else args.device detectors = build_detectors(args, model, tokenizer, device_for_detectors) aggregates = [] prediction_rows = [] for idx, example in enumerate(examples): prompt = scoring_prompt(example, tokenizer, args.prompt_template) prompt_replay_status = validate_generation_prompt_replay( example, tokenizer, prompt, args.prompt_template, allow_unverified=args.allow_unverified_replay, ) answer = example.answer answer_token_ids = saved_answer_token_ids( example, model_id=args.model, model_revision=args.model_revision, allow_model_mismatch=args.allow_token_id_model_mismatch, allow_unverified_revision=args.allow_unverified_replay, require_saved=bool(args.input) and not args.allow_unverified_replay, ) if answer_token_ids is not None: validate_saved_answer_decoding(example, tokenizer, answer_token_ids) labels = resolve_labels(example, tokenizer) if args.require_labels and labels is None: raise ValueError(f"example {idx} requires labels or hallucinated_spans") scored = [] for detector in detectors: start = time.perf_counter() result = detector.score(prompt, answer, answer_token_ids=answer_token_ids) scored.append((result, time.perf_counter() - start)) for result, seconds in scored: if labels is not None and len(labels) == len(result.scores): aggregate_for_method(aggregates, result.method).extend(result.scores, labels) elif labels is not None: message = ( f"labels for example {idx} have length {len(labels)} " f"but {result.method} returned {len(result.scores)} token scores" ) if args.strict_labels: raise ValueError(message) print(f"warning: {message}; skipping AUPRC labels") aggregate_for_method(aggregates, result.method).extend(result.scores, None) else: aggregate_for_method(aggregates, result.method).extend(result.scores, None) prediction_rows.append( TokenPredictionRow( index=idx, id=example.id, method=result.method, tokens=result.tokens, scores=result.scores, labels=labels, seconds=seconds, labels_aligned=labels is not None and len(labels) == len(result.scores), metadata=prediction_metadata( args, result.metadata, answer_token_ids is not None, prompt_replay_status, ), prompt=example.prompt, answer=example.answer, hallucinated_spans=example.hallucinated_spans, example_metadata=example.metadata, ) ) n_tokens = len(scored[0][0].tokens) if scored else 0 print(f"[{idx + 1}/{len(examples)}] scored {n_tokens} answer tokens") table_rows = [ summarize(aggregate.method, aggregate.scores, aggregate.labels_or_none(), aggregate.labelled_scores_or_none()) for aggregate in aggregates ] print_table(table_rows) if args.output: out_path = Path(args.output) with open(out_path, "w", encoding="utf-8") as f: for row in prediction_rows: f.write(json.dumps(row.to_json_dict()) + "\n") print(f"\nwrote predictions: {out_path}") if __name__ == "__main__": main()