# RAG 评估

---

## 评估框架概览

| 框架 | 侧重点 | 优势 | 适用场景 |
|-----------|-------|----------|----------|
| **RAGAS** | RAG 专用指标 | 忠实度、相关性 | 生产环境 RAG 评估 |
| **TruLens** | LLM 应用可观测性 | 追踪、反馈函数 | 调试与监控 |
| **LangSmith** | LangChain 生态系统 | 追踪、数据集、测试 | LangChain 项目 |
| **自定义** | 特定需求 | 完全掌控 | 领域特定需求 |

---

## 核心指标

### 检索指标

| 指标 | 公式 | 衡量内容 |
|--------|---------|------------------|
| **Precision@k** | 前 k 个结果中相关数 / k | 检索到的文档是否相关？ |
| **Recall@k** | 前 k 个结果中相关数 / 总相关数 | 是否检索到了所有相关文档？ |
| **MRR** | 1 / 首个相关文档的排名 | 找到相关文档的速度有多快？ |
| **NDCG@k** | DCG@k / IDCG@k | 排序顺序是否正确？ |
| **命中率** | 前 k 个结果中含相关文档的查询数 / 总查询数 | 二值成功率 |

### 生成指标

| 指标 | 衡量内容 |
|--------|------------------|
| **忠实度** | 答案是否基于检索到的上下文？ |
| **答案相关性** | 答案是否回答了问题？ |
| **上下文相关性** | 检索到的上下文是否与问题相关？ |
| **上下文利用率** | 实际使用了多少上下文？ |

---

## 实现核心指标

### Precision、Recall 与命中率

```python
from dataclasses import dataclass
from typing import Set

@dataclass
class RetrievalMetrics:
    precision_at_k: float
    recall_at_k: float
    hit_rate: float
    mrr: float

def calculate_retrieval_metrics(
    retrieved_ids: list[str],
    relevant_ids: set[str],
    k: int
) -> RetrievalMetrics:
    """计算核心检索指标。"""
    top_k = retrieved_ids[:k]
    top_k_set = set(top_k)

    # Precision@k：前 k 个结果中相关数 / k
    relevant_in_top_k = len(top_k_set & relevant_ids)
    precision = relevant_in_top_k / k if k > 0 else 0

    # Recall@k：前 k 个结果中相关数 / 总相关数
    recall = relevant_in_top_k / len(relevant_ids) if relevant_ids else 0

    # 命中率：前 k 个结果中有相关文档则为 1，否则为 0
    hit_rate = 1.0 if relevant_in_top_k > 0 else 0.0

    # MRR：1 / 首个相关结果的排名
    mrr = 0.0
    for i, doc_id in enumerate(top_k, 1):
        if doc_id in relevant_ids:
            mrr = 1.0 / i
            break

    return RetrievalMetrics(
        precision_at_k=precision,
        recall_at_k=recall,
        hit_rate=hit_rate,
        mrr=mrr
    )

# 使用示例
retrieved = ["doc1", "doc2", "doc3", "doc4", "doc5"]
relevant = {"doc2", "doc5", "doc7"}  # 真实标注

metrics = calculate_retrieval_metrics(retrieved, relevant, k=5)
print(f"Precision@5: {metrics.precision_at_k:.2f}")  # 2/5 = 0.40
print(f"Recall@5: {metrics.recall_at_k:.2f}")        # 2/3 = 0.67
print(f"MRR: {metrics.mrr:.2f}")                     # 1/2 = 0.50
```

### NDCG（归一化折损累计增益）

```python
import numpy as np

def dcg_at_k(relevance_scores: list[float], k: int) -> float:
    """计算折损累计增益。"""
    relevance_scores = np.array(relevance_scores[:k])
    if len(relevance_scores) == 0:
        return 0.0

    # DCG = sum(rel_i / log2(i + 1)) for i in 1..k
    discounts = np.log2(np.arange(2, len(relevance_scores) + 2))
    return np.sum(relevance_scores / discounts)

def ndcg_at_k(
    retrieved_ids: list[str],
    relevance_scores: dict[str, float],
    k: int
) -> float:
    """
    计算 NDCG@k。
    relevance_scores：将文档 ID 映射到相关性分数的字典（例如 0、1、2、3）
    """
    # 获取检索到的文档的相关性分数
    retrieved_relevance = [
        relevance_scores.get(doc_id, 0)
        for doc_id in retrieved_ids[:k]
    ]

    # 计算检索顺序的 DCG
    dcg = dcg_at_k(retrieved_relevance, k)

    # 计算理想 DCG（完美排序）
    ideal_relevance = sorted(relevance_scores.values(), reverse=True)[:k]
    idcg = dcg_at_k(ideal_relevance, k)

    return dcg / idcg if idcg > 0 else 0.0

# 使用示例（分级相关性）
retrieved = ["doc1", "doc2", "doc3", "doc4", "doc5"]
relevance = {
    "doc1": 0,   # 不相关
    "doc2": 3,   # 高度相关
    "doc3": 1,   # 部分相关
    "doc5": 2,   # 相关
    "doc7": 3,   # 高度相关（未检索到）
}

ndcg = ndcg_at_k(retrieved, relevance, k=5)
print(f"NDCG@5: {ndcg:.3f}")
```

---

## RAGAS 框架

### 安装与设置

```python
# pip install ragas

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

# 准备评估数据集
eval_data = {
    "question": [
        "法国的首都是哪里？",
        "如何安装 Python？"
    ],
    "answer": [
        "法国的首都是巴黎。",
        "你可以从 python.org 下载安装 Python。"
    ],
    "contexts": [
        ["巴黎是法国的首都和最大城市。"],
        ["Python 可以从官方网站 python.org 安装。",
         "你也可以使用 brew 或 apt 等包管理器。"]
    ],
    "ground_truth": [
        "巴黎是法国的首都。",
        "从 python.org 安装 Python，或使用包管理器。"
    ]
}

dataset = Dataset.from_dict(eval_data)

# 运行评估
results = evaluate(
    dataset,
    metrics=[
        faithfulness,
        answer_relevancy,
        context_precision,
        context_recall,
    ]
)

print(results)
# {'faithfulness': 0.95, 'answer_relevancy': 0.88, ...}
```

### 自定义 RAGAS 评估

```python
from ragas.metrics import Metric
from ragas.llms import LangchainLLM
from langchain_openai import ChatOpenAI

# 使用自定义 LLM
custom_llm = LangchainLLM(llm=ChatOpenAI(model="gpt-4o-mini"))

# 使用自定义设置进行评估
results = evaluate(
    dataset,
    metrics=[faithfulness, answer_relevancy],
    llm=custom_llm,
    raise_exceptions=False  # 出错时继续执行
)

# 逐样本得分
for i, row in enumerate(results.to_pandas().itertuples()):
    print(f"Q{i+1}: Faithfulness={row.faithfulness:.2f}, "
          f"Relevancy={row.answer_relevancy:.2f}")
```

### RAGAS 指标详解

```python
"""
RAGAS 核心指标：

1. 忠实度（0-1）：
   - 衡量答案是否基于上下文
   - LLM 从答案中提取声明，并与上下文进行验证
   - 高分 = 答案没有幻觉

2. 答案相关性（0-1）：
   - 衡量答案是否回应了问题
   - 从答案生成问题，与原始问题进行比较
   - 高分 = 答案切题

3. 上下文精确度（0-1）：
   - 衡量检索到的上下文是否相关
   - 按相关性对上下文排序，计算每个排序位置的精确度
   - 高分 = 靠前的上下文最相关

4. 上下文召回率（0-1）：
   - 衡量所有真实标注信息是否都在上下文中
   - 检查真实标注的句子是否被上下文支持
   - 高分 = 上下文包含所需信息
"""

# 诊断低分原因
def diagnose_ragas_scores(results_df):
    """识别有问题的样本。"""
    issues = []

    for idx, row in results_df.iterrows():
        if row.get('faithfulness', 1) < 0.5:
            issues.append({
                "index": idx,
                "issue": "忠实度过低——答案可能包含幻觉",
                "question": row['question'],
                "answer": row['answer'][:200]
            })

        if row.get('context_recall', 1) < 0.5:
            issues.append({
                "index": idx,
                "issue": "上下文召回率过低——检索遗漏了相关文档",
                "question": row['question']
            })

    return issues
```

---

## TruLens 评估

### 设置与基本用法

```python
# pip install trulens-eval

from trulens_eval import Tru, TruChain, Feedback
from trulens_eval.feedback import Groundedness
from trulens_eval.feedback.provider import OpenAI as fOpenAI

# 初始化 TruLens
tru = Tru()

# 创建反馈提供器
provider = fOpenAI()

# 定义反馈函数
f_groundedness = Feedback(
    provider.groundedness_measure_with_cot_reasons,
    name="Groundedness"
).on(
    TruChain.select_context().node.text  # 检索到的上下文
).on_output()

f_relevance = Feedback(
    provider.relevance_with_cot_reasons,
    name="Answer Relevance"
).on_input().on_output()

f_context_relevance = Feedback(
    provider.context_relevance_with_cot_reasons,
    name="Context Relevance"
).on_input().on(
    TruChain.select_context().node.text
)

# 包装你的 RAG 链
from langchain.chains import RetrievalQA

rag_chain = RetrievalQA.from_chain_type(
    llm=llm,
    retriever=vector_store.as_retriever()
)

tru_recorder = TruChain(
    rag_chain,
    app_id="rag-v1",
    feedbacks=[f_groundedness, f_relevance, f_context_relevance]
)

# 记录运行结果
with tru_recorder as recording:
    response = rag_chain.invoke({"query": "如何配置身份验证？"})

# 查看结果
tru.run_dashboard()  # 打开 Web UI
# 或通过编程方式获取
records = tru.get_records_and_feedback(app_ids=["rag-v1"])
```

### 自定义反馈函数

```python
from trulens_eval import Feedback, Select

def custom_citation_check(response: str, context: str) -> float:
    """检查响应是否引用了上下文中的来源。"""
    # 从响应中提取引用（例如 [1]、[Source: X]）
    import re
    citations = re.findall(r'\[[\d\w\s:]+\]', response)

    if not citations:
        return 0.0  # 没有引用

    # 验证引用是否确实指向实际上下文
    valid_citations = sum(1 for c in citations if c.lower() in context.lower())
    return valid_citations / len(citations)

f_citation = Feedback(
    custom_citation_check,
    name="Citation Accuracy"
).on_output().on(Select.RecordCalls.retriever.get_relevant_documents.rets.page_content)
```

---

## 构建自定义评估流水线

### LLM 作为评判者的评估

```python
from openai import OpenAI
from dataclasses import dataclass
from typing import Literal

client = OpenAI()

@dataclass
class EvalResult:
    score: float
    reasoning: str
    criteria: str

def evaluate_with_llm(
    question: str,
    answer: str,
    context: str,
    criteria: Literal["faithfulness", "relevance", "completeness"]
) -> EvalResult:
    """使用 LLM 作为评判者进行评估。"""

    criteria_prompts = {
        "faithfulness": """
            评估答案是否完全由提供的上下文支持。
            如果答案中的每个声明都能从上下文中得到验证，则得 1.0 分。
            如果大多数声明有支持但部分没有，则得 0.5 分。
            如果答案包含大量无法支持的声明，则得 0.0 分。
        """,
        "relevance": """
            评估答案是否直接回应了问题。
            如果答案完全回应了问题，则得 1.0 分。
            如果答案部分回应了问题，则得 0.5 分。
            如果答案离题或未回应问题，则得 0.0 分。
        """,
        "completeness": """
            评估答案是否涵盖了问题的所有方面。
            如果答案全面且完整，则得 1.0 分。
            如果答案涵盖了主要要点但遗漏了细节，则得 0.5 分。
            如果答案明显不完整，则得 0.0 分。
        """
    }

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": f"""你是一名 RAG 系统的专家评估员。
                {criteria_prompts[criteria]}

                请以 JSON 格式回复：
                {{"score": <0.0-1.0>, "reasoning": "<解释>"}}"""
            },
            {
                "role": "user",
                "content": f"""问题：{question}

上下文：
{context}

答案：{answer}

请评估答案的{criteria}："""
            }
        ],
        response_format={"type": "json_object"}
    )

    import json
    result = json.loads(response.choices[0].message.content)

    return EvalResult(
        score=result["score"],
        reasoning=result["reasoning"],
        criteria=criteria
    )

# 使用示例
eval_result = evaluate_with_llm(
    question="如何配置 OAuth2？",
    answer="在 config.yaml 中设置 client_id 和 client_secret 即可配置 OAuth2。",
    context="OAuth2 配置需要在 config.yaml 中设置 client_id、client_secret 和 redirect_uri。",
    criteria="faithfulness"
)
print(f"忠实度：{eval_result.score:.2f}")
print(f"推理过程：{eval_result.reasoning}")
```

### 批量评估流水线

```python
import asyncio
from tqdm.asyncio import tqdm_asyncio

async def evaluate_batch(
    test_cases: list[dict],
    retriever,
    generator,
    metrics: list[str] = ["precision", "faithfulness", "relevance"]
) -> dict:
    """对测试用例运行批量评估。"""

    results = {
        "per_sample": [],
        "aggregated": {}
    }

    async def evaluate_single(case: dict) -> dict:
        # 检索
        retrieved = await retriever.aretrieve(case["question"])
        retrieved_ids = [r.id for r in retrieved]

        # 生成
        answer = await generator.agenerate(
            question=case["question"],
            context=[r.text for r in retrieved]
        )

        # 计算指标
        sample_result = {
            "question": case["question"],
            "answer": answer,
            "retrieved_ids": retrieved_ids
        }

        if "relevant_ids" in case and "precision" in metrics:
            retrieval_metrics = calculate_retrieval_metrics(
                retrieved_ids,
                set(case["relevant_ids"]),
                k=5
            )
            sample_result["precision@5"] = retrieval_metrics.precision_at_k
            sample_result["recall@5"] = retrieval_metrics.recall_at_k

        if "faithfulness" in metrics:
            faith_eval = evaluate_with_llm(
                case["question"],
                answer,
                "\n".join([r.text for r in retrieved]),
                "faithfulness"
            )
            sample_result["faithfulness"] = faith_eval.score

        return sample_result

    # 并发运行评估
    tasks = [evaluate_single(case) for case in test_cases]
    results["per_sample"] = await tqdm_asyncio.gather(*tasks)

    # 聚合结果
    for metric in ["precision@5", "recall@5", "faithfulness"]:
        scores = [r.get(metric) for r in results["per_sample"] if r.get(metric) is not None]
        if scores:
            results["aggregated"][metric] = {
                "mean": sum(scores) / len(scores),
                "min": min(scores),
                "max": max(scores)
            }

    return results
```

---

## 诊断检索效果不佳

### 检索诊断

```python
def diagnose_retrieval(
    query: str,
    retrieved_docs: list,
    expected_docs: list,
    embedding_model
) -> dict:
    """诊断检索可能失败的原因。"""

    query_embedding = embedding_model.encode(query)
    retrieved_embeddings = [embedding_model.encode(d) for d in retrieved_docs]
    expected_embeddings = [embedding_model.encode(d) for d in expected_docs]

    from sklearn.metrics.pairwise import cosine_similarity
    import numpy as np

    diagnosis = {
        "query": query,
        "issues": []
    }

    # 检查查询-文档相似度
    for i, (doc, emb) in enumerate(zip(retrieved_docs, retrieved_embeddings)):
        sim = cosine_similarity([query_embedding], [emb])[0][0]
        if sim < 0.5:
            diagnosis["issues"].append({
                "type": "low_similarity",
                "doc_index": i,
                "similarity": float(sim),
                "doc_preview": doc[:100]
            })

    # 检查预期文档是否得分更高
    for i, (doc, emb) in enumerate(zip(expected_docs, expected_embeddings)):
        sim = cosine_similarity([query_embedding], [emb])[0][0]
        retrieved_max_sim = max(
            cosine_similarity([query_embedding], [e])[0][0]
            for e in retrieved_embeddings
        )

        if sim > retrieved_max_sim:
            diagnosis["issues"].append({
                "type": "missed_better_doc",
                "expected_doc_index": i,
                "expected_sim": float(sim),
                "best_retrieved_sim": float(retrieved_max_sim),
                "doc_preview": doc[:100]
            })

    # 检查词汇不匹配
    query_terms = set(query.lower().split())
    for i, doc in enumerate(retrieved_docs):
        doc_terms = set(doc.lower().split())
        overlap = query_terms & doc_terms
        if len(overlap) < len(query_terms) * 0.3:
            diagnosis["issues"].append({
                "type": "vocabulary_mismatch",
                "doc_index": i,
                "query_terms": list(query_terms),
                "overlapping_terms": list(overlap)
            })

    return diagnosis

# 使用示例
diagnosis = diagnose_retrieval(
    query="如何配置 OAuth 身份验证",
    retrieved_docs=retrieved_texts,
    expected_docs=expected_texts,
    embedding_model=sentence_transformer
)

for issue in diagnosis["issues"]:
    print(f"问题：{issue['type']}")
    print(f"详情：{issue}")
```

### 查询分析

```python
def analyze_query_performance(
    query_logs: list[dict],
    threshold_precision: float = 0.6
) -> dict:
    """分析查询模式以发现系统性问题。"""

    analysis = {
        "total_queries": len(query_logs),
        "low_performing": [],
        "patterns": {}
    }

    for log in query_logs:
        if log.get("precision@5", 1.0) < threshold_precision:
            analysis["low_performing"].append(log)

    # 分析表现不佳的查询
    if analysis["low_performing"]:
        # 检查常见模式
        low_perf_queries = [l["query"] for l in analysis["low_performing"]]

        # 查询长度分析
        avg_length = sum(len(q.split()) for q in low_perf_queries) / len(low_perf_queries)
        analysis["patterns"]["avg_low_perf_query_length"] = avg_length

        # 失败查询中的常见词汇
        from collections import Counter
        all_terms = []
        for q in low_perf_queries:
            all_terms.extend(q.lower().split())
        analysis["patterns"]["common_failing_terms"] = Counter(all_terms).most_common(10)

        # 问题类型分析
        question_words = ["how", "what", "why", "when", "where", "who"]
        question_types = Counter()
        for q in low_perf_queries:
            for qw in question_words:
                if q.lower().startswith(qw):
                    question_types[qw] += 1
                    break
            else:
                question_types["other"] += 1
        analysis["patterns"]["failing_question_types"] = dict(question_types)

    return analysis
```

---

## 持续监控

### 生产指标仪表板

```python
import time
from dataclasses import dataclass, field
from collections import deque
from threading import Lock

@dataclass
class RAGMetricsCollector:
    """在生产环境中收集和追踪 RAG 指标。"""

    window_size: int = 1000
    _latencies: deque = field(default_factory=lambda: deque(maxlen=1000))
    _retrieval_scores: deque = field(default_factory=lambda: deque(maxlen=1000))
    _generation_scores: deque = field(default_factory=lambda: deque(maxlen=1000))
    _lock: Lock = field(default_factory=Lock)

    def record_query(
        self,
        latency_ms: float,
        retrieval_score: float | None = None,
        generation_score: float | None = None
    ):
        """记录单个查询的指标。"""
        with self._lock:
            self._latencies.append(latency_ms)
            if retrieval_score is not None:
                self._retrieval_scores.append(retrieval_score)
            if generation_score is not None:
                self._generation_scores.append(generation_score)

    def get_summary(self) -> dict:
        """获取当前指标摘要。"""
        with self._lock:
            import numpy as np

            summary = {
                "queries_in_window": len(self._latencies),
                "latency": {
                    "p50": np.percentile(self._latencies, 50) if self._latencies else 0,
                    "p95": np.percentile(self._latencies, 95) if self._latencies else 0,
                    "p99": np.percentile(self._latencies, 99) if self._latencies else 0,
                },
                "retrieval_score": {
                    "mean": np.mean(self._retrieval_scores) if self._retrieval_scores else 0,
                    "std": np.std(self._retrieval_scores) if self._retrieval_scores else 0,
                },
                "generation_score": {
                    "mean": np.mean(self._generation_scores) if self._generation_scores else 0,
                    "std": np.std(self._generation_scores) if self._generation_scores else 0,
                }
            }

            return summary

# 使用示例
metrics = RAGMetricsCollector()

# 在 RAG 端点中
start = time.time()
response = rag_pipeline.query(question)
latency = (time.time() - start) * 1000

metrics.record_query(
    latency_ms=latency,
    retrieval_score=response.get("retrieval_score"),
    generation_score=response.get("generation_score")
)

# 定期检查
print(metrics.get_summary())
```

### 质量退化告警

```python
class RAGQualityMonitor:
    """监控 RAG 质量并在退化时发出告警。"""

    def __init__(
        self,
        baseline_precision: float = 0.8,
        alert_threshold: float = 0.1,  # 下降超过 10% 即告警
        window_size: int = 100
    ):
        self.baseline = baseline_precision
        self.threshold = alert_threshold
        self.window_size = window_size
        self.recent_scores = deque(maxlen=window_size)

    def record_score(self, precision: float) -> dict | None:
        """记录分数，如果质量退化则返回告警。"""
        self.recent_scores.append(precision)

        if len(self.recent_scores) < self.window_size // 2:
            return None  # 数据不足

        current_mean = sum(self.recent_scores) / len(self.recent_scores)
        degradation = self.baseline - current_mean

        if degradation > self.threshold:
            return {
                "alert": "QUALITY_DEGRADATION",
                "baseline": self.baseline,
                "current": current_mean,
                "degradation": degradation,
                "window_size": len(self.recent_scores)
            }

        return None

# 使用示例
monitor = RAGQualityMonitor(baseline_precision=0.85)

for query_result in production_queries:
    alert = monitor.record_score(query_result["precision@5"])
    if alert:
        send_alert(alert)  # 发送至 Slack、PagerDuty 等
```

---

## 评估最佳实践

| 实践 | 描述 |
|----------|-------------|
| **黄金测试集** | 维护 50-200 个带有真实标注的精选问答对 |
| **分层采样** | 测试集中包含多样化的查询类型 |
| **人工基线** | 将 LLM 评判者与人工标注者进行比较 |
| **版本控制** | 将评估结果与模型版本一同追踪 |
| **定期重新评估** | 每次检索变更时重新运行黄金测试 |
| **A/B 测试** | 在线上流量中比较新的检索策略 |

---

## 快速参考

| 目标 | 指标 | 目标值 |
|------|--------|--------|
| 文档是否相关？ | Precision@5 | > 0.7 |
| 是否获取了所有文档？ | Recall@5 | > 0.8 |
| 排序是否良好？ | NDCG@5 | > 0.7 |
| 答案是否基于上下文？ | 忠实度 | > 0.9 |
| 答案是否切题？ | 答案相关性 | > 0.8 |
| 上下文是否有用？ | 上下文相关性 | > 0.7 |

| 框架 | 最适合 |
|-----------|----------|
| RAGAS | 快速 RAG 专用评估 |
| TruLens | 生产监控与追踪 |
| 自定义 LLM 评判者 | 领域特定标准 |
| 人工标注 | 真实标注创建 |

## 相关技能

- **RAG 架构师** - 系统设计
- **ML 流水线** - 评估自动化
- **数据科学家** - 统计分析
- **监控专家** - 生产可观测性