69 lines
2.5 KiB
Python
69 lines
2.5 KiB
Python
"""Pure data contract for RAGAS evaluation of frozen ADC answers."""
|
|
from __future__ import annotations
|
|
|
|
import math
|
|
from collections import defaultdict
|
|
|
|
METRICS = ("faithfulness", "answer_relevancy", "context_recall", "answer_correctness")
|
|
|
|
|
|
def sample_from_answer(row, case):
|
|
contexts = []
|
|
for ref in row.get("response", {}).get("references") or []:
|
|
content = ref.get("content") or []
|
|
if isinstance(content, str):
|
|
raise ValueError("Expected a list of source chunks, not a single string")
|
|
for text in content:
|
|
if not isinstance(text, str):
|
|
raise ValueError("Source chunks must be strings")
|
|
if text.strip() and text not in contexts:
|
|
contexts.append(text)
|
|
return {
|
|
"user_input": case["question"],
|
|
"response": row.get("response", {}).get("response", ""),
|
|
"reference": case["reference_answer"],
|
|
"retrieved_contexts": contexts,
|
|
}
|
|
|
|
|
|
def skip_reason(row, case, metric, sample):
|
|
if row["status"] != "ok":
|
|
return "generation_failed"
|
|
if case["expected"] == "abstain_or_request_details":
|
|
return "abstention_scenario_evaluated_separately"
|
|
if metric in {"faithfulness", "context_recall"}:
|
|
if row["mode"] == "bypass":
|
|
return "no_retrieval_in_baseline"
|
|
if not sample["retrieved_contexts"]:
|
|
return "missing_source_contexts"
|
|
return None
|
|
|
|
|
|
def checked_score(metric, value):
|
|
value = float(value)
|
|
if not math.isfinite(value):
|
|
raise ValueError("RAGAS returned a nonfinite score")
|
|
minimum = -1 if metric == "answer_relevancy" else 0
|
|
if not minimum <= value <= 1:
|
|
raise ValueError("RAGAS score outside its mathematical range")
|
|
return value
|
|
|
|
|
|
def aggregate(rows):
|
|
# Each pair/metric contributes at most one latest attempt.
|
|
latest = {(r["case_id"], r["mode"], r["metric"]): r for r in rows}
|
|
groups = defaultdict(list)
|
|
for row in latest.values():
|
|
groups[(row["mode"], row["metric"])].append(row)
|
|
summary = {}
|
|
for (mode, metric), values in groups.items():
|
|
good = [checked_score(metric, r["value"]) for r in values if r["status"] == "ok"]
|
|
summary.setdefault(mode, {})[metric] = {
|
|
"mean": sum(good) / len(good) if good else None,
|
|
"n_ok": len(good),
|
|
"n_error": sum(r["status"] == "error" for r in values),
|
|
"n_undefined": sum(r["status"] == "undefined" for r in values),
|
|
"n_skipped": sum(r["status"] == "skipped" for r in values),
|
|
}
|
|
return summary
|