RAG 评估体系 2026：从 RAGAS 到自定义指标

┌──────────────────────────────────────────────┐
│              端到端评估（L3）                 │
│  用户满意度 / 任务完成率 / 答案正确性         │
├──────────────────────────────────────────────┤
│              生成评估（L2）                   │
│  答案相关性 / 忠实度 / 完整性 / 引用准确性    │
├──────────────────────────────────────────────┤
│              检索评估（L1）                   │
│  召回率 / 精确率 / MRR / 上下文相关性        │
└──────────────────────────────────────────────┘

class RetrievalMetrics:
    @staticmethod
    def recall_at_k(retrieved_ids: list, relevant_ids: list, k: int = 5):
        """Top-K 召回率：相关文档是否出现在 Top-K 中"""
        retrieved_top_k = retrieved_ids[:k]
        hits = len(set(retrieved_top_k) & set(relevant_ids))
        return hits / len(relevant_ids) if relevant_ids else 0.0
    
    @staticmethod
    def precision_at_k(retrieved_ids: list, relevant_ids: list, k: int = 5):
        """Top-K 精确率"""
        retrieved_top_k = retrieved_ids[:k]
        hits = len(set(retrieved_top_k) & set(relevant_ids))
        return hits / k
    
    @staticmethod
    def mrr(retrieved_ids: list, relevant_ids: list):
        """Mean Reciprocal Rank：第一个相关文档的排名倒数"""
        for i, doc_id in enumerate(retrieved_ids):
            if doc_id in relevant_ids:
                return 1.0 / (i + 1)
        return 0.0
    
    @staticmethod
    def ndcg_at_k(retrieved_ids: list, relevance_scores: dict, k: int = 5):
        """Normalized Discounted Cumulative Gain"""
        dcg = sum(
            relevance_scores.get(doc_id, 0) / np.log2(i + 2)
            for i, doc_id in enumerate(retrieved_ids[:k])
        )
        ideal_scores = sorted(relevance_scores.values(), reverse=True)[:k]
        idcg = sum(s / np.log2(i + 2) for i, s in enumerate(ideal_scores))
        return dcg / idcg if idcg > 0 else 0.0

def context_relevance(question: str, contexts: list, llm) -> float:
    """评估检索到的上下文与问题的相关程度"""
    prompt = f"""
请评估以下检索上下文与问题的相关性。

问题：{question}

上下文：
{chr(10).join([f'[{i+1}] {c[:200]}' for i, c in enumerate(contexts)])}

对每条上下文打分（0-3）：
- 0: 完全无关
- 1: 部分相关，缺少关键信息
- 2: 相关，包含部分答案
- 3: 高度相关，直接回答问题

输出 JSON：{{"scores": [0-3, ...], "overall": 0.0-1.0}}
"""
    result = llm.generate(prompt, response_format="json")
    return result["overall"]

from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall,
    context_entity_recall,
    answer_similarity
)

# RAGAS 标准评估
def evaluate_with_ragas(dataset, llm, embeddings):
    result = evaluate(
        dataset=dataset,  # 包含 question, answer, contexts, ground_truth
        metrics=[
            faithfulness,
            answer_relevancy,
            context_precision,
            context_recall,
            context_entity_recall,
            answer_similarity
        ],
        llm=llm,
        embeddings=embeddings
    )
    return result

class CitationAccuracy:
    """评估答案中引用的准确性"""
    
    def evaluate(self, answer: str, contexts: list) -> dict:
        # 1. 提取答案中的所有引用
        citations = self._extract_citations(answer)
        
        # 2. 验证每个引用是否在上下文中
        correct = 0
        total = len(citations)
        
        for citation in citations:
            cited_content = self._get_cited_content(citation, contexts)
            claim = self._get_claim_before_citation(answer, citation)
            
            if self._verify_claim(claim, cited_content):
                correct += 1
        
        return {
            "citation_precision": correct / total if total > 0 else 0.0,
            "citation_coverage": total / max(count_claims(answer), 1),
            "total_citations": total,
            "correct_citations": correct
        }

class HallucinationDetector:
    def detect(self, question: str, answer: str, contexts: list) -> dict:
        # 将答案拆分为原子论断
        claims = self._decompose_claims(answer)
        
        hallucinated = []
        supported = []
        
        for claim in claims:
            # 检查每个论断是否有上下文支撑
            evidence = self._find_evidence(claim, contexts)
            if evidence["score"] < 0.5:
                hallucinated.append(claim)
            else:
                supported.append(claim)
        
        return {
            "hallucination_rate": len(hallucinated) / len(claims),
            "hallucinated_claims": hallucinated,
            "supported_claims": supported,
            "total_claims": len(claims)
        }
    
    def _decompose_claims(self, text: str) -> list:
        prompt = f"""
将以下文本拆分为原子论断（每个论断是一个可验证的事实陈述）：

文本：{text}

输出 JSON 列表：["论断1", "论断2", ...]
"""
        return llm.generate(prompt, response_format="json")

class TaskCompletionEvaluator:
    """评估 RAG 系统在特定任务上的完成质量"""
    
    def evaluate(self, question: str, answer: str, task_type: str) -> dict:
        evaluators = {
            "factual_qa": self._eval_factual,
            "summary": self._eval_summary,
            "comparison": self._eval_comparison,
            "analysis": self._eval_analysis,
            "code_gen": self._eval_code
        }
        
        evaluator = evaluators.get(task_type, self._eval_general)
        return evaluator(question, answer)
    
    def _eval_comparison(self, question: str, answer: str) -> dict:
        prompt = f"""
评估这个比较类回答的质量：

问题：{question}
回答：{answer}

评分维度（0-5分）：
1. 对比维度全面性：是否覆盖了所有关键对比维度？
2. 客观性：是否中立客观？
3. 数据支撑：是否有具体数据支持？
4. 结构清晰度：对比结构是否清晰？

输出 JSON：{{"dimensions": {{"completeness": 0, "objectivity": 0, "data_support": 0, "structure": 0}}, "overall": 0.0, "feedback": "..."}}
"""
        return llm.generate(prompt, response_format="json")

class EvalDatasetBuilder:
    """构建 RAG 评测数据集"""
    
    def build(self, documents: list, num_questions: int = 200):
        dataset = []
        
        for i in range(num_questions):
            # 1. 随机选取文档
            doc = random.choice(documents)
            
            # 2. 生成问题
            question = self._generate_question(doc)
            
            # 3. 生成 ground truth 答案
            gt_answer = self._generate_gt_answer(question, doc)
            
            # 4. 标记相关文档
            relevant_docs = self._find_relevant(question, documents)
            
            # 5. 分类问题类型
            q_type = self._classify_question(question)
            
            dataset.append({
                "id": f"eval_{i}",
                "question": question,
                "ground_truth": gt_answer,
                "relevant_docs": relevant_docs,
                "question_type": q_type,
                "difficulty": self._assess_difficulty(question)
            })
        
        return dataset

class RAGEvalPipeline:
    def __init__(self, rag_system, eval_dataset):
        self.rag = rag_system
        self.dataset = eval_dataset
    
    def run(self):
        results = []
        
        for item in self.dataset:
            # 1. 执行 RAG 查询
            rag_result = self.rag.query(item["question"])
            
            # 2. L1: 检索评估
            retrieval_scores = {
                "recall@5": RetrievalMetrics.recall_at_k(
                    rag_result.retrieved_ids, item["relevant_docs"], k=5
                ),
                "recall@10": RetrievalMetrics.recall_at_k(
                    rag_result.retrieved_ids, item["relevant_docs"], k=10
                ),
                "mrr": RetrievalMetrics.mrr(
                    rag_result.retrieved_ids, item["relevant_docs"]
                )
            }
            
            # 3. L2: 生成评估
            generation_scores = {
                "faithfulness": ragas_faithfulness(item["question"], rag_result.answer, rag_result.contexts),
                "answer_relevancy": ragas_answer_relevancy(item["question"], rag_result.answer),
                "hallucination_rate": HallucinationDetector().detect(
                    item["question"], rag_result.answer, rag_result.contexts
                )["hallucination_rate"]
            }
            
            # 4. L3: 端到端评估
            e2e_scores = TaskCompletionEvaluator().evaluate(
                item["question"], rag_result.answer, item["question_type"]
            )
            
            results.append({
                "id": item["id"],
                "question_type": item["question_type"],
                "retrieval": retrieval_scores,
                "generation": generation_scores,
                "e2e": e2e_scores
            })
        
        # 汇总
        return self._summarize(results)