File size: 3,286 Bytes
aa4269d
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
"""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)