finpy1789's picture
Agentic Financial Document Analyst: multi-agent RAG + MCP, agent-coloured UI
aa4269d verified
Raw
History Blame Contribute Delete
3.29 kB
"""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)