wehub-skill-sync
24 KiB
24 KiB
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 与命中率
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（归一化折损累计增益）
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 框架
安装与设置
# 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 评估
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 指标详解
"""
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 评估
设置与基本用法
# 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"])
自定义反馈函数
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 作为评判者的评估
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}")
批量评估流水线
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
诊断检索效果不佳
检索诊断
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}")
查询分析
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
持续监控
生产指标仪表板
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())
质量退化告警
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 流水线 - 评估自动化
- 数据科学家 - 统计分析
- 监控专家 - 生产可观测性