Spaces:
Sleeping
Sleeping
File size: 3,112 Bytes
1239f44 | 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 | from __future__ import annotations
import json
from pathlib import Path
from typing import Any, cast
from src.evaluation.evaluation_dataset import RetrievalEvaluationRecord
def _recall_at_10(case: dict[str, Any]) -> float:
metrics = cast(dict[str, float] | None, case.get("metrics"))
return metrics.get("recall@10", 0.0) if metrics else 0.0
def analyze_retrieval_errors(
records: list[RetrievalEvaluationRecord],
baseline_cases: list[dict[str, Any]],
variant_cases: list[dict[str, Any]],
) -> dict[str, Any]:
records_by_id = {record.id: record for record in records}
variant_by_id = {str(case["id"]): case for case in variant_cases}
if set(records_by_id) != {str(case["id"]) for case in baseline_cases}:
raise ValueError("Baseline case ids do not match evaluation records.")
if set(records_by_id) != set(variant_by_id):
raise ValueError("Variant case ids do not match evaluation records.")
failures: list[dict[str, Any]] = []
category_counts: dict[str, int] = {}
recovered_count = 0
for baseline in baseline_cases:
if _recall_at_10(baseline) > 0:
continue
record = records_by_id[str(baseline["id"])]
retrieved = cast(list[dict[str, Any]], baseline["retrieved"])
same_source_hits = [
hit for hit in retrieved if str(hit.get("filename")) == record.source_file
]
category = "same_source_wrong_page" if same_source_hits else "source_absent_top10"
category_counts[category] = category_counts.get(category, 0) + 1
unique_pages = {
(str(hit.get("filename")), int(hit.get("page", 0))) for hit in retrieved
}
variant = variant_by_id[record.id]
recovered = _recall_at_10(variant) > 0
recovered_count += int(recovered)
failures.append(
{
"id": record.id,
"question": record.question,
"gold_source_file": record.source_file,
"gold_pages": record.gold_pages,
"category": category,
"same_source_pages_in_baseline_top10": sorted(
{int(hit["page"]) for hit in same_source_hits}
),
"duplicate_page_slots": len(retrieved) - len(unique_pages),
"recovered_by_variant_at_10": recovered,
}
)
return {
"baseline_miss_at_10": len(failures),
"recovered_by_variant_at_10": recovered_count,
"remaining_miss_at_10": len(failures) - recovered_count,
"category_counts": category_counts,
"failures": failures,
}
def load_cases(path: Path) -> list[dict[str, Any]]:
return [
cast(dict[str, Any], json.loads(line))
for line in path.read_text(encoding="utf-8").splitlines()
if line.strip()
]
def write_error_analysis(report: dict[str, Any], output_path: Path) -> None:
output_path.parent.mkdir(parents=True, exist_ok=True)
output_path.write_text(
json.dumps(report, ensure_ascii=False, indent=2) + "\n",
encoding="utf-8",
)
|