src/error_analysis.py - master - luvtrippin/RAG-Diploma

src/error_analysis.py
88 строк3 KB
Timur
Prepare reproducible RAG experiment project
25 июн 2026, 18:20
25 июн 2026, 18:20b279d80
ERROR_CATEGORIES = (
    "Retrieval Failure",
    "Context Dilution",
    "Reasoning Hallucination",
    "Unsupported Claim",
)


def _metric(row, name, default=0.0):
    if row.get(name) is not None:
        try:
            return float(row[name])
        except (TypeError, ValueError):
            pass
    legacy = name.replace("_at_k", "_at_5")
    if legacy != name and row.get(legacy) is not None:
        try:
            return float(row[legacy])
        except (TypeError, ValueError):
            pass
    return float(default)


def _has_unsupported_fact(row):
    for item in row.get("fact_checks") or []:
        if item.get("verdict") in {"contradicted", "not_found"}:
            return True
    return False


def classify_errors(row):
    categories = []
    recall = _metric(row, "recall_at_k")
    precision = _metric(row, "precision_at_k")

    if recall == 0.0:
        categories.append("Retrieval Failure")

    if precision < 0.4 and recall > 0.0:
        categories.append("Context Dilution")

    if (
        recall > 0.0
        and _metric(row, "context_relevance") >= 0.5
        and _metric(row, "answer_correctness") < 0.5
    ):
        categories.append("Reasoning Hallucination")

    if _metric(row, "hallucination_rate") >= 0.5 or _has_unsupported_fact(row):
        categories.append("Unsupported Claim")

    return categories


def collect_error_examples(results_by_method, min_examples=10):
    examples = []
    for method, rows in results_by_method.items():
        grouped = {category: [] for category in ERROR_CATEGORIES}
        for row in rows:
            for category in classify_errors(row):
                grouped[category].append(row)

        for category, category_rows in grouped.items():
            category_rows = sorted(
                category_rows,
                key=lambda row: (
                    _metric(row, "answer_correctness"),
                    -_metric(row, "hallucination_rate"),
                ),
            )
            for row in category_rows[:min_examples]:
                examples.append({
                    "method": method,
                    "category": category,
                    "question_id": row.get("question_id"),
                    "question": row.get("question"),
                    "ground_truth": row.get("ground_truth"),
                    "generated_answer": row.get("generated_answer"),
                    "retrieval_eval_k": row.get("retrieval_eval_k", row.get("selected_context_count")),
                    "precision_at_k": _metric(row, "precision_at_k"),
                    "recall_at_k": _metric(row, "recall_at_k"),
                    "mrr": row.get("mrr"),
                    "faithfulness": row.get("faithfulness"),
                    "answer_correctness": row.get("answer_correctness"),
                    "hallucination_rate": row.get("hallucination_rate"),
                    "retrieved_context_ids": row.get("retrieved_context_ids"),
                })
    return examples