RAG 从 Demo 到生产：八个绕不过去的问题

"""RAG 评测：分别测量召回质量与答案质量。

分开测很重要。如果只看最终答案准确率，
分不清是「检索没找到」还是「找到了但生成错了」，优化就没有方向。
"""

from __future__ import annotations

import json
from dataclasses import dataclass, field
from typing import Any, Callable


@dataclass
class EvalCase:
    question: str
    expected_answer: str
    expected_sources: list[str]          # 应该命中的文档标识
    should_refuse: bool = False           # 是否属于应拒答的问题


@dataclass
class EvalResult:
    total: int = 0
    recall_hit: int = 0                   # 召回命中数
    answer_correct: int = 0               # 答案正确数
    refuse_correct: int = 0               # 拒答正确数
    hallucinated: int = 0                 # 编造数
    failures: list[dict[str, Any]] = field(default_factory=list)

    @property
    def recall_rate(self) -> float:
        return self.recall_hit / self.total if self.total else 0.0

    @property
    def accuracy(self) -> float:
        return self.answer_correct / self.total if self.total else 0.0


def evaluate(
    cases: list[EvalCase],
    retrieve: Callable[[str], list[dict]],
    generate: Callable[[str, list[dict]], dict],
    judge: Callable[[str, str, str], bool],
) -> EvalResult:
    """judge 用一个较强的模型做答案一致性判定（LLM-as-judge）。"""
    result = EvalResult()

    for case in cases:
        result.total += 1
        chunks = retrieve(case.question)
        hit_sources = {c["source"] for c in chunks}

        # 召回评估：期望的来源是否至少命中一个
        recalled = bool(set(case.expected_sources) & hit_sources)
        if recalled or case.should_refuse:
            result.recall_hit += 1

        output = generate(case.question, chunks)
        answer = output["answer"]
        refused = any(k in answer for k in ("无法回答", "没有找到", "资料不足"))

        if case.should_refuse:
            if refused:
                result.refuse_correct += 1
                result.answer_correct += 1
            else:
                result.hallucinated += 1
                result.failures.append({
                    "question": case.question,
                    "issue": "应拒答但给出了答案",
                    "answer": answer[:200],
                })
            continue

        if refused:
            result.failures.append({
                "question": case.question,
                "issue": "过度拒答",
                "recalled": recalled,
            })
            continue

        if judge(case.question, case.expected_answer, answer):
            result.answer_correct += 1
        else:
            result.failures.append({
                "question": case.question,
                "issue": "答案不正确",
                "recalled": recalled,          # 关键：区分召回问题还是生成问题
                "expected": case.expected_answer[:150],
                "actual": answer[:200],
            })

    return result


def report(result: EvalResult) -> str:
    lines = [
        f"总题数：{result.total}",
        f"召回率：{result.recall_rate:.1%}",
        f"答案准确率：{result.accuracy:.1%}",
        f"编造次数：{result.hallucinated}",
       ",
        "失败案例归因：",
    ]
    # 按「召回失败」和「生成失败」分类，指明优化方向
    retrieval_fail = sum(1 for f in result.failures if f.get("recalled") is False)
    generation_fail = sum(1 for f in result.failures if f.get("recalled") is True)
    lines.append(f"  召回阶段问题：{retrieval_fail} 例 → 优化切分与检索")
    lines.append(f"  生成阶段问题：{generation_fail} 例 → 优化提示词与重排")
    return "\n".join(lines)