"""Confidence scoring for answers and analyses. Combines four observable signals (no self-grading by the answering model): retrieval quality - fused retrieval scores of the evidence used cross-source agreement - share of verified claims where sources agree evidence reliability - source diversity (documents/pages backing the answer) citation coverage - whether the answer actually cites its evidence These are the four numbers surfaced in the UI panel. """ from __future__ import annotations import re from dataclasses import dataclass from src.retrieval.hybrid import RetrievedChunk _CITE_RE = re.compile(r"\[[^\[\]]+p\.\s*\d+\]") @dataclass class ConfidenceReport: answer_confidence: int retrieval_quality: int cross_source_agreement: int evidence_reliability: int citation_coverage: int def as_dict(self) -> dict[str, int]: return { "Answer Confidence": self.answer_confidence, "Retrieval Quality": self.retrieval_quality, "Cross-Source Agreement": self.cross_source_agreement, "Evidence Reliability": self.evidence_reliability, "Citation Coverage": self.citation_coverage, } def retrieval_quality(retrieved: list[RetrievedChunk]) -> int: if not retrieved: return 0 # both-retriever hits score higher than single-retriever hits per_chunk = [] for r in retrieved: both = r.semantic_rank is not None and r.keyword_rank is not None base = 0.9 if both else 0.6 rank = min(x for x in (r.semantic_rank, r.keyword_rank) if x is not None) per_chunk.append(base * (1.0 - 0.05 * rank)) return round(100 * max(0.0, min(1.0, sum(per_chunk) / len(per_chunk)))) def cross_source_agreement(findings: list[dict]) -> int: """From verifier evidence-matrix rows; 100 when every multi-source claim agrees.""" multi = [f for f in findings if f.get("status") in ("agree", "differ")] if not multi: return 50 # unknown — nothing was cross-checkable agree = sum(1 for f in multi if f["status"] == "agree") return round(100 * agree / len(multi)) def evidence_reliability(retrieved: list[RetrievedChunk]) -> int: if not retrieved: return 0 docs = {r.chunk.doc_id for r in retrieved} pages = {(r.chunk.doc_id, r.chunk.page) for r in retrieved} tables = sum(1 for r in retrieved if r.chunk.is_table) score = 0.4 + 0.15 * min(len(docs), 3) + 0.02 * min(len(pages), 5) score += 0.05 if tables else 0.0 return round(100 * min(score, 1.0)) def citation_coverage(answer: str) -> int: sentences = [s for s in re.split(r"(?<=[.!?])\s+", answer) if len(s) > 40] if not sentences: return 100 if _CITE_RE.search(answer) else 0 cited = sum(1 for s in sentences if _CITE_RE.search(s)) return round(100 * cited / len(sentences)) def score(answer: str, retrieved: list[RetrievedChunk], findings: list[dict] | None = None) -> ConfidenceReport: rq = retrieval_quality(retrieved) ag = cross_source_agreement(findings or []) er = evidence_reliability(retrieved) cc = citation_coverage(answer) overall = round(0.35 * rq + 0.25 * ag + 0.25 * er + 0.15 * cc) return ConfidenceReport(overall, rq, ag, er, cc)