RAG 评估体系：召回、忠实度与自动化指标

(question, retrieved_contexts, answer, ground_truth)

import numpy as np

def recall_at_k(retrieved: list[str], relevant: set[str], k: int) -> float:
    if not relevant:
        return 0.0
    return len(set(retrieved[:k]) & relevant) / len(relevant)

def precision_at_k(retrieved: list[str], relevant: set[str], k: int) -> float:
    top = retrieved[:k]
    return len(set(top) & relevant) / k if k else 0.0

def mrr(retrieved: list[str], relevant: set[str]) -> float:
    for rank, doc_id in enumerate(retrieved, start=1):
        if doc_id in relevant:
            return 1.0 / rank
    return 0.0

def ndcg_at_k(retrieved: list[str], relevant: set[str], k: int) -> float:
    dcg = sum(
        1.0 / np.log2(rank + 1)
        for rank, doc_id in enumerate(retrieved[:k], start=1)
        if doc_id in relevant
    )
    idcg = sum(1.0 / np.log2(rank + 1) for rank in range(1, min(len(relevant), k) + 1))
    return dcg / idcg if idcg else 0.0

Faithfulness = 能被上下文支撑的断言数 / 总断言数

上下文：本产品支持 7 天无理由退货，运费由买家承担。
回答：本产品支持 7 天无理由退货，运费由卖家承担。

断言 1「支持 7 天无理由退货」→ 被支撑 ✓
断言 2「运费由卖家承担」    → 与上下文矛盾 ✗
Faithfulness = 1/2 = 0.5

Answer Relevance = 用回答反向生成问题的相似度均值

def context_recall(contexts: list[str], ground_truth: str, judge) -> float:
    sentences = split_sentences(ground_truth)
    if not sentences:
        return 0.0
    hits = sum(
        1 for s in sentences
        if judge.binary(f"下面的上下文能否支撑这句话？\n上下文：{contexts}\n句子：{s}")
    )
    return hits / len(sentences)

from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall,
    answer_correctness,
)
from datasets import Dataset

data = Dataset.from_dict({
    "question": [
        "退款需要几天到账？",
        "支持哪些支付方式？",
    ],
    "answer": [
        "退款一般在 3~5 个工作日到账。",
        "支持微信、支付宝和银行卡。",
    ],
    "contexts": [
        ["退款将在审核通过后 3~5 个工作日内退回原支付渠道。"],
        ["支持微信支付、支付宝、银行卡三种方式。"],
    ],
    "ground_truth": [
        "审核通过后 3~5 个工作日退回原渠道。",
        "支持微信、支付宝和银行卡。",
    ],
})

result = evaluate(
    dataset=data,
    metrics=[
        faithfulness,
        answer_relevancy,
        context_precision,
        context_recall,
        answer_correctness,
    ],
)
print(result)
# {'faithfulness': 0.95, 'answer_relevancy': 0.91,
#  'context_precision': 0.88, 'context_recall': 0.90, 'answer_correctness': 0.84}
df = result.to_pandas()
df.to_csv("rag_eval_report.csv", index=False)

import asyncio

async def evaluate_rag_system(rag, golden: list[dict]) -> dict:
    rows = []
    for case in golden:
        retrieved = await rag.retrieve(case["question"])
        answer = await rag.generate(case["question"], retrieved)
        rows.append({
            "question": case["question"],
            "answer": answer,
            "contexts": [c["text"] for c in retrieved],
            "ground_truth": case["ground_truth"],
        })
    ds = Dataset.from_list(rows)
    return evaluate(ds, metrics=[faithfulness, answer_relevancy,
                                 context_precision, context_recall])

JUDGE_PROMPT = """你是一个严格的事实核查员。

任务：判断「回答」中的每一条事实性断言是否被「上下文」支持。

规则：
1. 只依据上下文判断，不使用你的先验知识。
2. 上下文未提及的内容，一律判为「不支持」。
3. 回答长度与评分无关。

上下文：
{context}

回答：
{answer}

以 JSON 输出：{"claims": [{"claim": "...", "supported": true|false}], "score": 0.0~1.0}
"""

P50/P95 延迟、Token 成本/次、检索命中缓存率、拒答率、用户追问率

GEN_PROMPT = """基于下面的文档片段，生成 {n} 个用户可能提出的问题及其标准答案。

要求：
1. 问题必须能仅凭该片段回答。
2. 问题要口语化，模拟真实用户。
3. 覆盖不同类型的提问：事实查询、对比、条件判断。

文档片段：
{chunk}

以 JSON 输出：[{{"question": "...", "ground_truth": "..."}}]
"""

def build_eval_set(chunks: list[str], n_per_chunk: int = 3) -> list[dict]:
    cases = []
    for chunk in chunks:
        raw = llm.complete(GEN_PROMPT.format(n=n_per_chunk, chunk=chunk))
        cases.extend(json.loads(raw))
    return dedup(cases)

import hashlib

def assign_bucket(user_id: str, salt: str = "rag-v2") -> str:
    """稳定分流：同一用户始终进同一组。"""
    h = hashlib.md5(f"{user_id}:{salt}".encode()).hexdigest()
    return "treatment" if int(h[:8], 16) % 100 < 50 else "control"

# .github/workflows/rag-eval.yml
name: RAG Evaluation
on:
  pull_request:
    paths: ["src/rag/**", "configs/retrieval.yaml"]
jobs:
  evaluate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - run: pip install -r requirements.txt
      - run: python scripts/eval_rag.py --golden data/golden.jsonl --out report.json
      - run: python scripts/check_gate.py --report report.json --max-drop 0.02