Spaces:

Qalam
/

Nuclear-Intelligence

Running

Nuclear-Intelligence / core /evaluation_enhanced.py

Deploy canonical runtime d8375ee from GitHub Actions

1028375 verified 3 months ago

12.2 kB

"""
Nuclear Intelligence v4.0 - Enhanced Evaluation Engine ⚛️📊
═══════════════════════════════════════════════════════════════════
Upgrades the single-shot LLM-as-judge with:
1. **Self-Consistency Check (NLI-style)**:
   - Generate N independent assessments (default N=3).
   - Compute median per dimension → robust to single-pass variance.
   - Inter-rater agreement (1 - normalized std) → bonus.
2. **Citation Quality Scoring**:
   - Reward answers that cite IAEA, NRC, peer-reviewed journals,
     DOE/INL/ORNL reports.
   - Penalize unsourced claims and circular self-reference.
3. **Novelty vs Knowledge Graph**:
   - Compare new answer's n-gram + embedding distance to existing KG
     entities. High distance → truly novel (vs slight rewording).
4. **Tokenization-Readiness Score**:
   - Composite of Accuracy ≥ 93%, Novelty ≥ 75%, Usefulness ≥ 80%,
     Self-Consistency ≥ 80%, Citation-Quality ≥ 0.5.
   - Returns a binary "ready_to_mint" flag plus per-dimension detail.
This module is *additive* — existing callers can keep using
EvaluationScore.overall_score() unchanged.
═══════════════════════════════════════════════════════════════════
"""

from __future__ import annotations

import re
import statistics
from collections import Counter
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Tuple
from loguru import logger

from core.nuclear_intelligence_v4 import EvaluationScore


# ─── Trusted source whitelist ────────────────────────────────────────
TRUSTED_SOURCES = {
    "iaea.org", "iaea", "nrc.gov", "ans.org", "ieee.org", "doi.org",
    "nature.com", "sciencedirect", "springer", "wiley",
    "energy.gov", "inl.gov", "ornl.gov", "anl.gov", "lanl.gov",
    "bnl.gov", "sandia.gov", "iter.org", "iter", "iter newsline",
    "cern", "iae", "usnc", "researchgate", "arxiv.org", "pubmed",
    "oecd-nea.org", "nea.fr", "world-nuclear.org", "wna",
    "rosatom", "cea.fr", "jaea.go.jp", "kaeri.re.kr",
    "wikipedia.org",  # only as last-resort cross-check
}

# Anti-patterns: vague, promotional, or circular references
WEAK_CITATION_PATTERNS = [
    re.compile(r"\bbroadly\s+speaking\b", re.I),
    re.compile(r"\bsome\s+experts\s+(?:say|believe|claim)\b", re.I),
    re.compile(r"\bit\s+is\s+(?:widely|generally)\s+(?:believed|known|accepted)\s+that\b", re.I),
    re.compile(r"\baccording\s+to\s+(?:some|many|most)\s+(?:sources|reports)\b", re.I),
]


@dataclass
class CitationQuality:
    """Per-answer citation assessment."""
    citation_count: int
    trusted_ratio: float  # 0..1
    has_doi: bool
    has_year: bool
    weak_phrase_count: int
    score: float  # 0..100

    def to_dict(self) -> Dict:
        return {
            "citation_count": self.citation_count,
            "trusted_ratio": round(self.trusted_ratio, 3),
            "has_doi": self.has_doi,
            "has_year": self.has_year,
            "weak_phrase_count": self.weak_phrase_count,
            "score": round(self.score, 2),
        }


def assess_citation_quality(answer_text: str, citations: List[str]) -> CitationQuality:
    """Score the citation backbone of an answer."""
    if not answer_text:
        return CitationQuality(0, 0.0, False, False, 0, 0.0)

    text_lower = answer_text.lower()
    cites_lower = [c.lower() for c in (citations or [])]

    # Citation count: explicit references + inline markers
    inline_cites = len(re.findall(r"\[\d+\]|\(\d{4}\)|et al\.", text_lower))
    citation_count = inline_cites + len(cites_lower)

    # Trusted ratio
    if cites_lower:
        trusted = sum(1 for c in cites_lower if any(t in c for t in TRUSTED_SOURCES))
        trusted_ratio = trusted / max(1, len(cites_lower))
    else:
        trusted_ratio = 0.0

    # DOI / year presence
    has_doi = bool(re.search(r"\b10\.\d{4,9}/[-._;()/:A-Z0-9]+", text_lower, re.I))
    has_year = bool(re.search(r"\b(19|20)\d{2}\b", answer_text))

    # Weak phrases
    weak = sum(1 for p in WEAK_CITATION_PATTERNS if p.search(answer_text))

    # Score (out of 100)
    score = 0.0
    score += min(40, citation_count * 5)            # up to 40 pts for citations
    score += 30 * trusted_ratio                     # up to 30 pts
    score += 10 if has_doi else 0                   # DOI bonus
    score += 5 if has_year else 0                   # year bonus
    score -= min(20, weak * 8)                      # penalty

    return CitationQuality(
        citation_count=citation_count,
        trusted_ratio=trusted_ratio,
        has_doi=has_doi,
        has_year=has_year,
        weak_phrase_count=weak,
        score=max(0.0, min(100.0, score)),
    )


# ─── Self-consistency via N independent evaluations ─────────────────

@dataclass
class ConsistencyReport:
    n_samples: int
    accuracy_median: float
    novelty_median: float
    usefulness_median: float
    completeness_median: float
    accuracy_std: float
    novelty_std: float
    usefulness_std: float
    completeness_std: float
    agreement: float            # 1 - mean(normalized std)
    passed: bool

    def to_dict(self) -> Dict:
        return {
            "n_samples": self.n_samples,
            "accuracy_median": round(self.accuracy_median, 2),
            "novelty_median": round(self.novelty_median, 2),
            "usefulness_median": round(self.usefulness_median, 2),
            "completeness_median": round(self.completeness_median, 2),
            "accuracy_std": round(self.accuracy_std, 3),
            "novelty_std": round(self.novelty_std, 3),
            "usefulness_std": round(self.usefulness_std, 3),
            "completeness_std": round(self.completeness_std, 3),
            "agreement": round(self.agreement, 3),
            "passed": self.passed,
        }


def consistency_report(scores: List[EvaluationScore], agreement_threshold: float = 0.80) -> ConsistencyReport:
    """Aggregate N independent EvaluationScore samples.
    `agreement` = 1 - mean(coeff_of_variation) across dimensions.
    `passed` = agreement >= threshold.
    """
    if not scores:
        return ConsistencyReport(0, 0, 0, 0, 0, 1, 1, 1, 1, 0.0, False)

    acc = [s.scientific_accuracy for s in scores]
    nov = [s.novelty_score for s in scores]
    use = [s.usefulness_score for s in scores]
    comp = [s.completeness for s in scores]

    def _safe_std(xs: List[float]) -> float:
        return statistics.pstdev(xs) if len(xs) > 1 else 0.0

    s_acc, s_nov, s_use, s_comp = _safe_std(acc), _safe_std(nov), _safe_std(use), _safe_std(comp)

    def _cv(std: float, mean: float) -> float:
        return (std / mean) if mean > 0 else 0.0

    cvs = [_cv(s_acc, statistics.mean(acc)), _cv(s_nov, statistics.mean(nov)),
           _cv(s_use, statistics.mean(use)), _cv(s_comp, statistics.mean(comp))]

    agreement = max(0.0, 1.0 - statistics.mean(cvs))

    return ConsistencyReport(
        n_samples=len(scores),
        accuracy_median=statistics.median(acc),
        novelty_median=statistics.median(nov),
        usefulness_median=statistics.median(use),
        completeness_median=statistics.median(comp),
        accuracy_std=s_acc,
        novelty_std=s_nov,
        usefulness_std=s_use,
        completeness_std=s_comp,
        agreement=agreement,
        passed=agreement >= agreement_threshold,
    )


# ─── Novelty vs existing Knowledge Graph entities ──────────────────

def novelty_against_kg(answer_text: str, kg_entity_questions: List[str], ngram_n: int = 3) -> float:
    """Estimate textual novelty vs prior knowledge-graph entries.
    Returns 0..100 — higher means more genuinely new (less paraphrase
    of existing questions).
    Cheap heuristic: 3-gram Jaccard distance against all prior questions.
    """
    if not answer_text or not kg_entity_questions:
        return 75.0  # neutral if we have no baseline

    def ngrams(text: str, n: int) -> Counter:
        tokens = re.findall(r"\w+", text.lower())
        return Counter(tuple(tokens[i:i + n]) for i in range(len(tokens) - n + 1))

    a_ngrams = ngrams(answer_text[:5000], ngram_n)
    if not a_ngrams:
        return 75.0

    max_sim = 0.0
    a_total = sum(a_ngrams.values())
    for prev_q in kg_entity_questions[-200:]:  # cap for cost
        p_ngrams = ngrams(prev_q, ngram_n)
        if not p_ngrams:
            continue
        overlap = sum((a_ngrams & p_ngrams).values())
        p_total = sum(p_ngrams.values())
        denom = a_total + p_total - overlap
        if denom <= 0:
            continue
        jac = overlap / denom
        if jac > max_sim:
            max_sim = jac

    # Map similarity -> novelty (50..100)
    novelty = 100.0 - 50.0 * max_sim
    return max(0.0, min(100.0, novelty))


# ─── Tokenization-Readiness composite score ────────────────────────

@dataclass
class TokenizationReadiness:
    accuracy: float
    novelty: float
    usefulness: float
    self_consistency: float      # agreement fraction
    citation_quality: float
    overall: float
    ready_to_mint: bool
    notes: List[str] = field(default_factory=list)

    THRESHOLDS = {
        "accuracy": 93.0,
        "novelty": 75.0,
        "usefulness": 80.0,
        "self_consistency": 0.80,
        "citation_quality": 50.0,
    }
    OVERALL_MIN = 85.0

    def to_dict(self) -> Dict:
        return {
            "accuracy": round(self.accuracy, 2),
            "novelty": round(self.novelty, 2),
            "usefulness": round(self.usefulness, 2),
            "self_consistency": round(self.self_consistency, 3),
            "citation_quality": round(self.citation_quality, 2),
            "overall": round(self.overall, 2),
            "ready_to_mint": self.ready_to_mint,
            "notes": self.notes,
        }


def tokenization_readiness(
    evaluation: EvaluationScore,
    consistency: Optional[ConsistencyReport] = None,
    citation_quality: Optional[CitationQuality] = None,
    weights: Optional[Dict[str, float]] = None,
) -> TokenizationReadiness:
    """Compute the composite Tokenization-Readiness Score.
    `weights` lets operators tune without forking the codebase.
    Defaults to the canonical NI v4 weights from the system prompt.
    """
    w = weights or {"accuracy": 0.40, "novelty": 0.20, "usefulness": 0.20,
                    "self_consistency": 0.10, "citation_quality": 0.10}

    sc = consistency.agreement if consistency else 1.0
    cq = citation_quality.score if citation_quality else 0.0

    overall = (
        evaluation.scientific_accuracy * w["accuracy"]
        + evaluation.novelty_score * w["novelty"]
        + evaluation.usefulness_score * w["usefulness"]
        + sc * 100.0 * w["self_consistency"]
        + cq * w["citation_quality"]
    )

    notes: List[str] = []
    T = TokenizationReadiness.THRESHOLDS

    if evaluation.scientific_accuracy < T["accuracy"]:
        notes.append(f"accuracy {evaluation.scientific_accuracy:.1f} < {T['accuracy']}")
    if evaluation.novelty_score < T["novelty"]:
        notes.append(f"novelty {evaluation.novelty_score:.1f} < {T['novelty']}")
    if evaluation.usefulness_score < T["usefulness"]:
        notes.append(f"usefulness {evaluation.usefulness_score:.1f} < {T['usefulness']}")
    if sc < T["self_consistency"]:
        notes.append(f"self_consistency {sc:.3f} < {T['self_consistency']}")
    if cq < T["citation_quality"]:
        notes.append(f"citation_quality {cq:.1f} < {T['citation_quality']}")

    ready = overall >= TokenizationReadiness.OVERALL_MIN and not notes

    return TokenizationReadiness(
        accuracy=evaluation.scientific_accuracy,
        novelty=evaluation.novelty_score,
        usefulness=evaluation.usefulness_score,
        self_consistency=sc,
        citation_quality=cq,
        overall=overall,
        ready_to_mint=ready,
        notes=notes,
    )

	"""
	Nuclear Intelligence v4.0 - Enhanced Evaluation Engine ⚛️📊
	═══════════════════════════════════════════════════════════════════
	Upgrades the single-shot LLM-as-judge with:

	1. Self-Consistency Check (NLI-style):
	- Generate N independent assessments (default N=3).
	- Compute median per dimension → robust to single-pass variance.
	- Inter-rater agreement (1 - normalized std) → bonus.

	2. Citation Quality Scoring:
	- Reward answers that cite IAEA, NRC, peer-reviewed journals,
	DOE/INL/ORNL reports.
	- Penalize unsourced claims and circular self-reference.

	3. Novelty vs Knowledge Graph:
	- Compare new answer's n-gram + embedding distance to existing KG
	entities. High distance → truly novel (vs slight rewording).

	4. Tokenization-Readiness Score:
	- Composite of Accuracy ≥ 93%, Novelty ≥ 75%, Usefulness ≥ 80%,
	Self-Consistency ≥ 80%, Citation-Quality ≥ 0.5.
	- Returns a binary "ready_to_mint" flag plus per-dimension detail.

	This module is additive — existing callers can keep using
	EvaluationScore.overall_score() unchanged.
	═══════════════════════════════════════════════════════════════════
	"""

	from __future__ import annotations

	import re
	import statistics
	from collections import Counter
	from dataclasses import dataclass, field
	from typing import Dict, List, Optional, Tuple
	from loguru import logger

	from core.nuclear_intelligence_v4 import EvaluationScore


	# ─── Trusted source whitelist ────────────────────────────────────────
	TRUSTED_SOURCES = {
	"iaea.org", "iaea", "nrc.gov", "ans.org", "ieee.org", "doi.org",
	"nature.com", "sciencedirect", "springer", "wiley",
	"energy.gov", "inl.gov", "ornl.gov", "anl.gov", "lanl.gov",
	"bnl.gov", "sandia.gov", "iter.org", "iter", "iter newsline",
	"cern", "iae", "usnc", "researchgate", "arxiv.org", "pubmed",
	"oecd-nea.org", "nea.fr", "world-nuclear.org", "wna",
	"rosatom", "cea.fr", "jaea.go.jp", "kaeri.re.kr",
	"wikipedia.org", # only as last-resort cross-check
	}

	# Anti-patterns: vague, promotional, or circular references
	WEAK_CITATION_PATTERNS = [
	re.compile(r"\bbroadly\s+speaking\b", re.I),
	re.compile(r"\bsome\s+experts\s+(?:say\|believe\|claim)\b", re.I),
	re.compile(r"\bit\s+is\s+(?:widely\|generally)\s+(?:believed\|known\|accepted)\s+that\b", re.I),
	re.compile(r"\baccording\s+to\s+(?:some\|many\|most)\s+(?:sources\|reports)\b", re.I),
	]


	@dataclass
	class CitationQuality:
	"""Per-answer citation assessment."""
	citation_count: int
	trusted_ratio: float # 0..1
	has_doi: bool
	has_year: bool
	weak_phrase_count: int
	score: float # 0..100

	def to_dict(self) -> Dict:
	return {
	"citation_count": self.citation_count,
	"trusted_ratio": round(self.trusted_ratio, 3),
	"has_doi": self.has_doi,
	"has_year": self.has_year,
	"weak_phrase_count": self.weak_phrase_count,
	"score": round(self.score, 2),
	}


	def assess_citation_quality(answer_text: str, citations: List[str]) -> CitationQuality:
	"""Score the citation backbone of an answer."""
	if not answer_text:
	return CitationQuality(0, 0.0, False, False, 0, 0.0)

	text_lower = answer_text.lower()
	cites_lower = [c.lower() for c in (citations or [])]

	# Citation count: explicit references + inline markers
	inline_cites = len(re.findall(r"\[\d+\]\|\(\d{4}\)\|et al\.", text_lower))
	citation_count = inline_cites + len(cites_lower)

	# Trusted ratio
	if cites_lower:
	trusted = sum(1 for c in cites_lower if any(t in c for t in TRUSTED_SOURCES))
	trusted_ratio = trusted / max(1, len(cites_lower))
	else:
	trusted_ratio = 0.0

	# DOI / year presence
	has_doi = bool(re.search(r"\b10\.\d{4,9}/[-._;()/:A-Z0-9]+", text_lower, re.I))
	has_year = bool(re.search(r"\b(19\|20)\d{2}\b", answer_text))

	# Weak phrases
	weak = sum(1 for p in WEAK_CITATION_PATTERNS if p.search(answer_text))

	# Score (out of 100)
	score = 0.0
	score += min(40, citation_count * 5) # up to 40 pts for citations
	score += 30 * trusted_ratio # up to 30 pts
	score += 10 if has_doi else 0 # DOI bonus
	score += 5 if has_year else 0 # year bonus
	score -= min(20, weak * 8) # penalty

	return CitationQuality(
	citation_count=citation_count,
	trusted_ratio=trusted_ratio,
	has_doi=has_doi,
	has_year=has_year,
	weak_phrase_count=weak,
	score=max(0.0, min(100.0, score)),
	)


	# ─── Self-consistency via N independent evaluations ─────────────────

	@dataclass
	class ConsistencyReport:
	n_samples: int
	accuracy_median: float
	novelty_median: float
	usefulness_median: float
	completeness_median: float
	accuracy_std: float
	novelty_std: float
	usefulness_std: float
	completeness_std: float
	agreement: float # 1 - mean(normalized std)
	passed: bool

	def to_dict(self) -> Dict:
	return {
	"n_samples": self.n_samples,
	"accuracy_median": round(self.accuracy_median, 2),
	"novelty_median": round(self.novelty_median, 2),
	"usefulness_median": round(self.usefulness_median, 2),
	"completeness_median": round(self.completeness_median, 2),
	"accuracy_std": round(self.accuracy_std, 3),
	"novelty_std": round(self.novelty_std, 3),
	"usefulness_std": round(self.usefulness_std, 3),
	"completeness_std": round(self.completeness_std, 3),
	"agreement": round(self.agreement, 3),
	"passed": self.passed,
	}


	def consistency_report(scores: List[EvaluationScore], agreement_threshold: float = 0.80) -> ConsistencyReport:
	"""Aggregate N independent EvaluationScore samples.

	`agreement` = 1 - mean(coeff_of_variation) across dimensions.
	`passed` = agreement >= threshold.
	"""
	if not scores:
	return ConsistencyReport(0, 0, 0, 0, 0, 1, 1, 1, 1, 0.0, False)

	acc = [s.scientific_accuracy for s in scores]
	nov = [s.novelty_score for s in scores]
	use = [s.usefulness_score for s in scores]
	comp = [s.completeness for s in scores]

	def _safe_std(xs: List[float]) -> float:
	return statistics.pstdev(xs) if len(xs) > 1 else 0.0

	s_acc, s_nov, s_use, s_comp = _safe_std(acc), _safe_std(nov), _safe_std(use), _safe_std(comp)

	def _cv(std: float, mean: float) -> float:
	return (std / mean) if mean > 0 else 0.0

	cvs = [_cv(s_acc, statistics.mean(acc)), _cv(s_nov, statistics.mean(nov)),
	_cv(s_use, statistics.mean(use)), _cv(s_comp, statistics.mean(comp))]

	agreement = max(0.0, 1.0 - statistics.mean(cvs))

	return ConsistencyReport(
	n_samples=len(scores),
	accuracy_median=statistics.median(acc),
	novelty_median=statistics.median(nov),
	usefulness_median=statistics.median(use),
	completeness_median=statistics.median(comp),
	accuracy_std=s_acc,
	novelty_std=s_nov,
	usefulness_std=s_use,
	completeness_std=s_comp,
	agreement=agreement,
	passed=agreement >= agreement_threshold,
	)


	# ─── Novelty vs existing Knowledge Graph entities ──────────────────

	def novelty_against_kg(answer_text: str, kg_entity_questions: List[str], ngram_n: int = 3) -> float:
	"""Estimate textual novelty vs prior knowledge-graph entries.

	Returns 0..100 — higher means more genuinely new (less paraphrase
	of existing questions).

	Cheap heuristic: 3-gram Jaccard distance against all prior questions.
	"""
	if not answer_text or not kg_entity_questions:
	return 75.0 # neutral if we have no baseline

	def ngrams(text: str, n: int) -> Counter:
	tokens = re.findall(r"\w+", text.lower())
	return Counter(tuple(tokens[i:i + n]) for i in range(len(tokens) - n + 1))

	a_ngrams = ngrams(answer_text[:5000], ngram_n)
	if not a_ngrams:
	return 75.0

	max_sim = 0.0
	a_total = sum(a_ngrams.values())
	for prev_q in kg_entity_questions[-200:]: # cap for cost
	p_ngrams = ngrams(prev_q, ngram_n)
	if not p_ngrams:
	continue
	overlap = sum((a_ngrams & p_ngrams).values())
	p_total = sum(p_ngrams.values())
	denom = a_total + p_total - overlap
	if denom <= 0:
	continue
	jac = overlap / denom
	if jac > max_sim:
	max_sim = jac

	# Map similarity -> novelty (50..100)
	novelty = 100.0 - 50.0 * max_sim
	return max(0.0, min(100.0, novelty))


	# ─── Tokenization-Readiness composite score ────────────────────────

	@dataclass
	class TokenizationReadiness:
	accuracy: float
	novelty: float
	usefulness: float
	self_consistency: float # agreement fraction
	citation_quality: float
	overall: float
	ready_to_mint: bool
	notes: List[str] = field(default_factory=list)

	THRESHOLDS = {
	"accuracy": 93.0,
	"novelty": 75.0,
	"usefulness": 80.0,
	"self_consistency": 0.80,
	"citation_quality": 50.0,
	}
	OVERALL_MIN = 85.0

	def to_dict(self) -> Dict:
	return {
	"accuracy": round(self.accuracy, 2),
	"novelty": round(self.novelty, 2),
	"usefulness": round(self.usefulness, 2),
	"self_consistency": round(self.self_consistency, 3),
	"citation_quality": round(self.citation_quality, 2),
	"overall": round(self.overall, 2),
	"ready_to_mint": self.ready_to_mint,
	"notes": self.notes,
	}


	def tokenization_readiness(
	evaluation: EvaluationScore,
	consistency: Optional[ConsistencyReport] = None,
	citation_quality: Optional[CitationQuality] = None,
	weights: Optional[Dict[str, float]] = None,
	) -> TokenizationReadiness:
	"""Compute the composite Tokenization-Readiness Score.

	`weights` lets operators tune without forking the codebase.
	Defaults to the canonical NI v4 weights from the system prompt.
	"""
	w = weights or {"accuracy": 0.40, "novelty": 0.20, "usefulness": 0.20,
	"self_consistency": 0.10, "citation_quality": 0.10}

	sc = consistency.agreement if consistency else 1.0
	cq = citation_quality.score if citation_quality else 0.0

	overall = (
	evaluation.scientific_accuracy * w["accuracy"]
	+ evaluation.novelty_score * w["novelty"]
	+ evaluation.usefulness_score * w["usefulness"]
	+ sc * 100.0 * w["self_consistency"]
	+ cq * w["citation_quality"]
	)

	notes: List[str] = []
	T = TokenizationReadiness.THRESHOLDS

	if evaluation.scientific_accuracy < T["accuracy"]:
	notes.append(f"accuracy {evaluation.scientific_accuracy:.1f} < {T['accuracy']}")
	if evaluation.novelty_score < T["novelty"]:
	notes.append(f"novelty {evaluation.novelty_score:.1f} < {T['novelty']}")
	if evaluation.usefulness_score < T["usefulness"]:
	notes.append(f"usefulness {evaluation.usefulness_score:.1f} < {T['usefulness']}")
	if sc < T["self_consistency"]:
	notes.append(f"self_consistency {sc:.3f} < {T['self_consistency']}")
	if cq < T["citation_quality"]:
	notes.append(f"citation_quality {cq:.1f} < {T['citation_quality']}")

	ready = overall >= TokenizationReadiness.OVERALL_MIN and not notes

	return TokenizationReadiness(
	accuracy=evaluation.scientific_accuracy,
	novelty=evaluation.novelty_score,
	usefulness=evaluation.usefulness_score,
	self_consistency=sc,
	citation_quality=cq,
	overall=overall,
	ready_to_mint=ready,
	notes=notes,
	)