RAG 评估与优化

"如何衡量 RAG 系统的好坏"是最常被问到的问题。本章介绍系统化的评估方法和优化策略。

RAG 评估框架

检索质量评估

"""
检索质量评估工具
"""
from dataclasses import dataclass
@dataclass
class EvalResult:
"""评估结果"""
metric: str
score: float
details: dict
class RetrievalEvaluator:
"""检索质量评估器"""
def evaluate(
self,
queries: list[str],
retrieved_docs: list[list[str]],
ground_truth: list[list[str]]
) -> dict:
"""
评估检索质量
Args:
queries: 查询列表
retrieved_docs: 每个查询检索到的文档ID列表
ground_truth: 每个查询的标准答案文档ID列表
"""
results = {
"recall": self._calc_recall(retrieved_docs, ground_truth),
"precision": self._calc_precision(retrieved_docs, ground_truth),
"mrr": self._calc_mrr(retrieved_docs, ground_truth),
"hit_rate": self._calc_hit_rate(retrieved_docs, ground_truth),
}
print("=== 检索评估结果 ===")
for metric, score in results.items():
print(f"  {metric}: {score:.4f}")
return results
def _calc_recall(
self, retrieved: list[list[str]], truth: list[list[str]]
) -> float:
"""
召回率: 找到了多少相关文档
Recall = 检索到的相关文档数 / 全部相关文档数
"""
recalls = []
for ret, gt in zip(retrieved, truth):
if not gt:
continue
found = len(set(ret) & set(gt))
recalls.append(found / len(gt))
return sum(recalls) / len(recalls) if recalls else 0
def _calc_precision(
self, retrieved: list[list[str]], truth: list[list[str]]
) -> float:
"""
精确率: 检索结果中有多少是相关的
Precision = 检索到的相关文档数 / 检索到的总文档数
"""
precisions = []
for ret, gt in zip(retrieved, truth):
if not ret:
continue
found = len(set(ret) & set(gt))
precisions.append(found / len(ret))
return sum(precisions) / len(precisions) if precisions else 0
def _calc_mrr(
self, retrieved: list[list[str]], truth: list[list[str]]
) -> float:
"""
MRR (Mean Reciprocal Rank)
第一个相关文档的排名的倒数的平均值
"""
reciprocal_ranks = []
for ret, gt in zip(retrieved, truth):
gt_set = set(gt)
for rank, doc_id in enumerate(ret, 1):
if doc_id in gt_set:
reciprocal_ranks.append(1.0 / rank)
break
else:
reciprocal_ranks.append(0.0)
return (
sum(reciprocal_ranks) / len(reciprocal_ranks)
if reciprocal_ranks else 0
)
def _calc_hit_rate(
self, retrieved: list[list[str]], truth: list[list[str]]
) -> float:
"""
命中率: 有多少查询至少找到了一个相关文档
"""
hits = 0
for ret, gt in zip(retrieved, truth):
if set(ret) & set(gt):
hits += 1
return hits / len(retrieved) if retrieved else 0
# 使用
evaluator = RetrievalEvaluator()
results = evaluator.evaluate(
queries=["什么是RAG?", "如何优化检索?"],
retrieved_docs=[["doc1", "doc3", "doc5"], ["doc2", "doc4", "doc6"]],
ground_truth=[["doc1", "doc2"], ["doc4", "doc7"]]
)

LLM-as-a-Judge 评估

使用 LLM 自动评估生成质量，是目前最实用的方法：

"""
LLM-as-a-Judge 评估系统
使用 GPT-4 评估 RAG 回答的质量
"""
class LLMJudge:
"""LLM 评估器"""
def __init__(self, model: str = "gpt-4o"):
self.client = OpenAI()
self.model = model
def evaluate_faithfulness(
self, answer: str, context: str
) -> dict:
"""
评估忠实度: 回答是否忠实于参考资料
"""
prompt = f"""请评估以下回答是否忠实于参考资料。
## 评估标准
- 回答中的每个声明是否都能在参考资料中找到依据
- 是否有编造或添加参考资料中没有的信息
- 是否有曲解或误读参考资料
## 参考资料
{context}
## 回答
{answer}
## 评估
请给出 1-5 分的评分:
1分 = 完全不忠实，大量编造
3分 = 部分忠实，有一些无依据的信息
5分 = 完全忠实，所有信息都有依据
评分 (只输出数字):"""
response = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": prompt}],
temperature=0
)
try:
score = int(response.choices[0].message.content.strip()[0])
except (ValueError, IndexError):
score = 3
return {"metric": "faithfulness", "score": score, "max": 5}
def evaluate_relevance(
self, question: str, answer: str
) -> dict:
"""
评估相关性: 回答是否回答了用户的问题
"""
prompt = f"""请评估回答与问题的相关性。
## 评估标准
- 回答是否针对了用户的问题
- 是否包含了用户需要的信息
- 是否有大量与问题无关的内容
## 用户问题
{question}
## 回答
{answer}
## 评估
请给出 1-5 分:
1分 = 完全不相关
3分 = 部分相关
5分 = 高度相关，完整回答了问题
评分 (只输出数字):"""
response = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": prompt}],
temperature=0
)
try:
score = int(response.choices[0].message.content.strip()[0])
except (ValueError, IndexError):
score = 3
return {"metric": "relevance", "score": score, "max": 5}
def evaluate_all(
self, question: str, answer: str, context: str
) -> dict:
"""综合评估"""
faithfulness = self.evaluate_faithfulness(answer, context)
relevance = self.evaluate_relevance(question, answer)
avg_score = (faithfulness["score"] + relevance["score"]) / 2
return {
"faithfulness": faithfulness,
"relevance": relevance,
"average": avg_score,
}
# 使用
judge = LLMJudge()
result = judge.evaluate_all(
question="什么是 RAG？",
answer="RAG 是检索增强生成，它通过检索外部文档来增强 LLM 的回答能力。[1]",
context="RAG (Retrieval-Augmented Generation) 是一种结合检索和生成的技术。"
)
print(f"忠实度: {result['faithfulness']['score']}/5")
print(f"相关性: {result['relevance']['score']}/5")
print(f"综合评分: {result['average']:.1f}/5")

构建评估数据集

"""
构建标准评估数据集
"""
import json
class EvalDatasetBuilder:
"""评估数据集构建器"""
def __init__(self):
self.client = OpenAI()
self.dataset = []
def generate_qa_pairs(
self, documents: list[dict], pairs_per_doc: int = 3
) -> list[dict]:
"""
自动从文档中生成 QA 对
"""
for doc in documents:
prompt = f"""根据以下文档内容，生成 {pairs_per_doc} 个问答对。
要求：
1. 问题要自然、像真实用户会问的
2. 答案必须从文档内容中得出
3. 包含不同难度的问题
文档:
{doc['content'][:2000]}
请以 JSON 数组格式返回:
[
{{"question": "问题", "answer": "标准答案", "difficulty": "easy/medium/hard"}}
]
JSON:"""
response = self.client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
temperature=0.5,
response_format={"type": "json_object"}
)
try:
result = json.loads(response.choices[0].message.content)
pairs = result if isinstance(result, list) else result.get("pairs", [])
for pair in pairs:
pair["source_doc"] = doc.get("source", "")
self.dataset.append(pair)
except json.JSONDecodeError:
continue
print(f"生成了 {len(self.dataset)} 个评估 QA 对")
return self.dataset
def save(self, filepath: str) -> None:
"""保存评估数据集"""
with open(filepath, "w", encoding="utf-8") as f:
json.dump(self.dataset, f, ensure_ascii=False, indent=2)
print(f"数据集已保存到 {filepath}")
def load(self, filepath: str) -> list[dict]:
"""加载评估数据集"""
with open(filepath, "r", encoding="utf-8") as f:
self.dataset = json.load(f)
return self.dataset

常见优化方向

本章小结

• RAG 评估包含检索评估、生成评估和端到端评估
• LLM-as-a-Judge 是目前最实用的自动化评估方法
• 自动生成评估数据集可以降低人工标注成本
• 常见优化方向：Chunk 大小、Embedding 模型、重排序、Prompt
• 评估要持续进行，建立基线后才能衡量改进效果

下一章：我们将学习如何在生产环境部署 RAG 系统。