YoVinchen
ecdb10c37a
Some checks failed
verify / verify (push) Has been cancelled
Separate local parsing from model indexing, bind review decisions to immutable manifests, persist vectors behind active profiles, and expose retrieval, chat, evaluation, and document workflows through the React workbench. Constraint: Live Bailian authentication currently fails for all three configured capabilities Rejected: Direct upload-to-embedding flow | bypasses local review and manifest binding Confidence: high Scope-risk: broad Directive: Keep private-data deployment blocked until authentication, RBAC, and separate database roles land Tested: make verify; fresh and replay Docker document smoke; worker recovery smoke; frozen synthetic evaluation; migration 0003-0004 roundtrip Not-tested: Successful live Bailian calls, OCR, real multi-user authorization
291 lines
10 KiB
Python
291 lines
10 KiB
Python
"""Deterministic, dependency-free RAG evaluation metrics and run freezing."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import math
|
|
import random
|
|
import re
|
|
from collections.abc import Mapping, Sequence
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
_SECRET_KEY_PATTERN = re.compile(
|
|
r"(?:api[_-]?key|secret|password|authorization|credential|access[_-]?token)",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
|
|
class EvaluationContractError(ValueError):
|
|
"""Raised when an evaluation would silently produce an invalid metric."""
|
|
|
|
|
|
class UnjudgedCandidateError(EvaluationContractError):
|
|
"""Raised instead of treating a pooled-but-unjudged candidate as irrelevant."""
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class RankingMetrics:
|
|
hit_at_k: float
|
|
recall_at_k: float
|
|
reciprocal_rank: float
|
|
ndcg_at_k: float
|
|
complete_hit_at_k: float
|
|
evidence_group_recall_at_k: float
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class CitationMetrics:
|
|
precision: float
|
|
recall: float
|
|
f1: float
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class RefusalMetrics:
|
|
precision: float
|
|
recall: float
|
|
f1: float
|
|
accuracy: float
|
|
true_positive: int
|
|
false_positive: int
|
|
false_negative: int
|
|
true_negative: int
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ConfidenceInterval:
|
|
mean: float
|
|
lower: float
|
|
upper: float
|
|
seed: int
|
|
iterations: int
|
|
|
|
|
|
def evaluate_ranking(
|
|
ranked_document_ids: Sequence[str],
|
|
*,
|
|
relevance: Mapping[str, float],
|
|
judged_document_ids: frozenset[str],
|
|
evidence_groups: Sequence[frozenset[str]],
|
|
k: int,
|
|
) -> RankingMetrics:
|
|
"""Score a ranking only when every candidate in the cutoff is judged.
|
|
|
|
A relevance value greater than zero is relevant. Evidence groups express
|
|
multi-part questions: complete hit requires at least one retrieved document
|
|
from every group.
|
|
"""
|
|
|
|
if isinstance(k, bool) or not isinstance(k, int) or k < 1:
|
|
raise EvaluationContractError("k must be a positive integer")
|
|
ranking = tuple(ranked_document_ids)
|
|
if any(not isinstance(item, str) or not item for item in ranking):
|
|
raise EvaluationContractError("ranking IDs must be non-empty strings")
|
|
if len(set(ranking)) != len(ranking):
|
|
raise EvaluationContractError("ranking IDs must be unique")
|
|
if any(
|
|
not isinstance(score, (int, float))
|
|
or isinstance(score, bool)
|
|
or not math.isfinite(float(score))
|
|
or float(score) < 0
|
|
for score in relevance.values()
|
|
):
|
|
raise EvaluationContractError("relevance values must be finite and non-negative")
|
|
if not set(relevance).issubset(judged_document_ids):
|
|
raise EvaluationContractError("every qrel document must be in the judgment pool")
|
|
if any(not group or not group.issubset(judged_document_ids) for group in evidence_groups):
|
|
raise EvaluationContractError("evidence groups must be non-empty and fully judged")
|
|
|
|
top_k = ranking[:k]
|
|
unjudged = [document_id for document_id in top_k if document_id not in judged_document_ids]
|
|
if unjudged:
|
|
raise UnjudgedCandidateError(f"top-{k} contains {len(unjudged)} unjudged candidate(s)")
|
|
|
|
gains = [float(relevance.get(document_id, 0.0)) for document_id in top_k]
|
|
positive_relevance = {
|
|
document_id for document_id, score in relevance.items() if float(score) > 0
|
|
}
|
|
relevant_retrieved = positive_relevance.intersection(top_k)
|
|
hit = 1.0 if relevant_retrieved else 0.0
|
|
recall = len(relevant_retrieved) / len(positive_relevance) if positive_relevance else 0.0
|
|
reciprocal_rank = next(
|
|
(1.0 / rank for rank, gain in enumerate(gains, start=1) if gain > 0),
|
|
0.0,
|
|
)
|
|
dcg = _dcg(gains)
|
|
ideal = sorted((float(value) for value in relevance.values()), reverse=True)[:k]
|
|
ideal_dcg = _dcg(ideal)
|
|
ndcg = dcg / ideal_dcg if ideal_dcg > 0 else 0.0
|
|
covered_groups = sum(bool(group.intersection(top_k)) for group in evidence_groups)
|
|
group_recall = covered_groups / len(evidence_groups) if evidence_groups else 0.0
|
|
complete_hit = 1.0 if evidence_groups and covered_groups == len(evidence_groups) else 0.0
|
|
return RankingMetrics(
|
|
hit_at_k=hit,
|
|
recall_at_k=recall,
|
|
reciprocal_rank=reciprocal_rank,
|
|
ndcg_at_k=ndcg,
|
|
complete_hit_at_k=complete_hit,
|
|
evidence_group_recall_at_k=group_recall,
|
|
)
|
|
|
|
|
|
def evaluate_citations(
|
|
cited_source_ids: Sequence[str],
|
|
*,
|
|
supported_source_ids: frozenset[str],
|
|
) -> CitationMetrics:
|
|
citations = tuple(cited_source_ids)
|
|
if any(not isinstance(item, str) or not item for item in citations):
|
|
raise EvaluationContractError("citation IDs must be non-empty strings")
|
|
if len(set(citations)) != len(citations):
|
|
raise EvaluationContractError("citation IDs must be unique")
|
|
cited = set(citations)
|
|
true_positive = len(cited.intersection(supported_source_ids))
|
|
precision = true_positive / len(cited) if cited else (1.0 if not supported_source_ids else 0.0)
|
|
recall = true_positive / len(supported_source_ids) if supported_source_ids else 1.0
|
|
f1 = _f1(precision, recall)
|
|
return CitationMetrics(precision=precision, recall=recall, f1=f1)
|
|
|
|
|
|
def evaluate_refusals(
|
|
predicted_refusals: Sequence[bool],
|
|
*,
|
|
answerable_labels: Sequence[bool],
|
|
) -> RefusalMetrics:
|
|
if len(predicted_refusals) != len(answerable_labels) or not predicted_refusals:
|
|
raise EvaluationContractError("refusal predictions and labels must be non-empty pairs")
|
|
if any(type(value) is not bool for value in (*predicted_refusals, *answerable_labels)):
|
|
raise EvaluationContractError("refusal predictions and labels must be booleans")
|
|
|
|
expected_refusals = tuple(not answerable for answerable in answerable_labels)
|
|
true_positive = sum(
|
|
predicted and expected
|
|
for predicted, expected in zip(predicted_refusals, expected_refusals, strict=True)
|
|
)
|
|
false_positive = sum(
|
|
predicted and not expected
|
|
for predicted, expected in zip(predicted_refusals, expected_refusals, strict=True)
|
|
)
|
|
false_negative = sum(
|
|
not predicted and expected
|
|
for predicted, expected in zip(predicted_refusals, expected_refusals, strict=True)
|
|
)
|
|
true_negative = len(predicted_refusals) - true_positive - false_positive - false_negative
|
|
precision = (
|
|
true_positive / (true_positive + false_positive) if true_positive + false_positive else 0.0
|
|
)
|
|
recall = (
|
|
true_positive / (true_positive + false_negative) if true_positive + false_negative else 0.0
|
|
)
|
|
return RefusalMetrics(
|
|
precision=precision,
|
|
recall=recall,
|
|
f1=_f1(precision, recall),
|
|
accuracy=(true_positive + true_negative) / len(predicted_refusals),
|
|
true_positive=true_positive,
|
|
false_positive=false_positive,
|
|
false_negative=false_negative,
|
|
true_negative=true_negative,
|
|
)
|
|
|
|
|
|
def bootstrap_mean_confidence_interval(
|
|
values: Sequence[float],
|
|
*,
|
|
seed: int,
|
|
iterations: int = 2_000,
|
|
confidence: float = 0.95,
|
|
) -> ConfidenceInterval:
|
|
if not values:
|
|
raise EvaluationContractError("bootstrap values must not be empty")
|
|
normalized = tuple(float(value) for value in values)
|
|
if any(not math.isfinite(value) for value in normalized):
|
|
raise EvaluationContractError("bootstrap values must be finite")
|
|
if isinstance(seed, bool) or not isinstance(seed, int):
|
|
raise EvaluationContractError("bootstrap seed must be an integer")
|
|
if isinstance(iterations, bool) or not isinstance(iterations, int) or iterations < 100:
|
|
raise EvaluationContractError("bootstrap iterations must be at least 100")
|
|
if not 0 < confidence < 1:
|
|
raise EvaluationContractError("confidence must be between zero and one")
|
|
|
|
mean = sum(normalized) / len(normalized)
|
|
if len(normalized) == 1:
|
|
return ConfidenceInterval(
|
|
mean=mean,
|
|
lower=mean,
|
|
upper=mean,
|
|
seed=seed,
|
|
iterations=iterations,
|
|
)
|
|
generator = random.Random(seed) # noqa: S311 - deterministic statistics, not security.
|
|
sample_means = sorted(
|
|
sum(generator.choice(normalized) for _ in normalized) / len(normalized)
|
|
for _ in range(iterations)
|
|
)
|
|
tail = (1.0 - confidence) / 2.0
|
|
lower = _percentile(sample_means, tail)
|
|
upper = _percentile(sample_means, 1.0 - tail)
|
|
return ConfidenceInterval(
|
|
mean=mean,
|
|
lower=lower,
|
|
upper=upper,
|
|
seed=seed,
|
|
iterations=iterations,
|
|
)
|
|
|
|
|
|
def freeze_run_config(config: Mapping[str, Any]) -> tuple[str, str]:
|
|
"""Return canonical JSON and SHA-256 while rejecting secret-shaped fields."""
|
|
|
|
_validate_frozen_value(config, path="config")
|
|
canonical = json.dumps(
|
|
config,
|
|
ensure_ascii=False,
|
|
sort_keys=True,
|
|
separators=(",", ":"),
|
|
allow_nan=False,
|
|
)
|
|
return canonical, hashlib.sha256(canonical.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def _validate_frozen_value(value: Any, *, path: str) -> None:
|
|
if isinstance(value, Mapping):
|
|
for key, item in value.items():
|
|
if not isinstance(key, str) or not key:
|
|
raise EvaluationContractError(f"{path} keys must be non-empty strings")
|
|
if _SECRET_KEY_PATTERN.search(key):
|
|
raise EvaluationContractError(f"{path} contains a forbidden secret-shaped key")
|
|
_validate_frozen_value(item, path=f"{path}.{key}")
|
|
return
|
|
if isinstance(value, (list, tuple)):
|
|
for index, item in enumerate(value):
|
|
_validate_frozen_value(item, path=f"{path}[{index}]")
|
|
return
|
|
if value is None or isinstance(value, (str, int, bool)):
|
|
return
|
|
if isinstance(value, float) and math.isfinite(value):
|
|
return
|
|
raise EvaluationContractError(f"{path} contains a non-canonical value")
|
|
|
|
|
|
def _dcg(gains: Sequence[float]) -> float:
|
|
return float(
|
|
sum((2.0**gain - 1.0) / math.log2(rank + 1) for rank, gain in enumerate(gains, start=1))
|
|
)
|
|
|
|
|
|
def _f1(precision: float, recall: float) -> float:
|
|
return 2 * precision * recall / (precision + recall) if precision + recall else 0.0
|
|
|
|
|
|
def _percentile(sorted_values: Sequence[float], probability: float) -> float:
|
|
position = probability * (len(sorted_values) - 1)
|
|
lower = math.floor(position)
|
|
upper = math.ceil(position)
|
|
if lower == upper:
|
|
return sorted_values[lower]
|
|
weight = position - lower
|
|
return sorted_values[lower] * (1.0 - weight) + sorted_values[upper] * weight
|