RAG 质量评估与持续改进体系从人工抽检到自动化评测闭环一、上线后不管的 RAG 质量漂移一个内部知识库 RAG 系统上线时回答准确率约 85%。三个月后用户投诉搜索发票流程找不到。排查发现三个月间知识库新增了 200 篇文档但向量索引没有重新构建新文档的检索排名很低。同时模型 API 从 GPT-4-0613 升级到了 GPT-4-0125同样的 Prompt 在新模型上输出的格式略有差异——原来用### 步骤开头现在变成了**步骤**导致下游的格式化函数解析失败。RAG 系统的质量不是一个静态快照而是随时间持续漂移的动态过程。知识库在变化、模型 API 在升级、用户的提问方式在演变。需要一个持续运行的评估体系来发现和纠正质量漂移。二、RAG 质量的四层评估模型graph TB subgraph 第一层检索质量 R1[召回率br/相关文档是否被找到] R2[精确率br/找到的文档是否都相关] R3[MRRbr/相关文档的排名位置] end subgraph 第二层生成质量 G1[忠实度br/回答是否基于检索文档] G2[相关性br/回答是否切题] G3[完整性br/是否覆盖所有要点] end subgraph 第三层端到端质量 E1[答案正确率br/人工标注对比] E2[用户满意度br// 反馈] E3[任务完成率br/用户是否达成目标] end subgraph 第四层系统健康 S1[检索延迟 P95] S2[Token 消耗] S3[缓存命中率] end R1 -- DASH[RAG 质量看板] G1 -- DASH E1 -- DASH S1 -- DASH第一层评估找到了什么是 RAG 的根基。第二层评估说了什么衡量大模型生成的质量。第三层评估用户得到了什么是最终价值的体现。第四层监控系统运行状况。三、自动化评估实现 RAG 质量自动化评估——定期运行基线测试集。 设计意图每次知识库更新或 Prompt 变更后自动运行 对比基线指标检测质量退化。 from dataclasses import dataclass from typing import List, Dict, Optional from datetime import datetime import json dataclass class QAPair: 问答对——评估的黄金标准 question: str # 参考答案——可以是精确答案或关键词列表 expected_answer: Optional[str] None expected_keywords: List[str] None # 期望的相关文档 ID 列表 expected_doc_ids: List[str] None # 问题分类 category: str general dataclass class RetrievalMetrics: 检索层指标 recall_at_5: float # Top-5 召回率 precision_at_5: float mrr: float # 平均倒数排名 dataclass class GenerationMetrics: 生成层指标 faithfulness: float # 忠实度 answer_relevance: float keyword_coverage: float dataclass class EvalReport: 评估报告 timestamp: datetime test_suite_name: str total_questions: int retrieval: RetrievalMetrics generation: GenerationMetrics # 与基线的对比 baseline_comparison: Dict[str, float] # 失败案例列表 failed_cases: List[Dict] class RAGEvaluator: RAG 质量评估器 def __init__(self, test_suite: List[QAPair]): self.test_suite test_suite self.baseline: Optional[EvalReport] None def evaluate( self, retrieval_fn, # 检索函数 (query) - List[(doc_id, score)] generation_fn, # 生成函数 (query, docs) - str ) - EvalReport: 运行全量评估 retrieval_metrics_list [] generation_metrics_list [] failed_cases [] for qa in self.test_suite: # 检索评估 retrieved_docs retrieval_fn(qa.question) retrieved_ids [doc_id for doc_id, _ in retrieved_docs[:5]] # 计算检索指标 if qa.expected_doc_ids: recall len( set(retrieved_ids) set(qa.expected_doc_ids) ) / len(qa.expected_doc_ids) precision len( set(retrieved_ids) set(qa.expected_doc_ids) ) / len(retrieved_ids) if retrieved_ids else 0 # MRR——第一个相关文档的排名倒数 mrr 0 for i, doc_id in enumerate(retrieved_ids, 1): if doc_id in qa.expected_doc_ids: mrr 1.0 / i break else: recall precision mrr None # 生成评估 answer generation_fn(qa.question, retrieved_docs) # 关键词覆盖——简化版的忠实度检查 if qa.expected_keywords: keyword_coverage sum( 1 for kw in qa.expected_keywords if kw.lower() in answer.lower() ) / len(qa.expected_keywords) else: keyword_coverage None # 收集失败案例 if recall is not None and recall 0.5: failed_cases.append({ question: qa.question, category: qa.category, reason: f检索召回率过低: {recall:.2f}, retrieved: retrieved_ids[:5], expected: qa.expected_doc_ids, }) elif keyword_coverage is not None and keyword_coverage 0.6: failed_cases.append({ question: qa.question, category: qa.category, reason: f关键词覆盖率低: {keyword_coverage:.2f}, answer: answer[:200], }) retrieval_metrics_list.append((recall, precision, mrr)) generation_metrics_list.append(keyword_coverage) # 聚合指标 valid_retrieval [ (r, p, m) for r, p, m in retrieval_metrics_list if r is not None ] valid_generation [ g for g in generation_metrics_list if g is not None ] report EvalReport( timestampdatetime.now(), test_suite_nameRAG Baseline Test Suite v1, total_questionslen(self.test_suite), retrievalRetrievalMetrics( recall_at_5( sum(r for r, _, _ in valid_retrieval) / len(valid_retrieval) if valid_retrieval else 0 ), precision_at_5( sum(p for _, p, _ in valid_retrieval) / len(valid_retrieval) if valid_retrieval else 0 ), mrr( sum(m for _, _, m in valid_retrieval) / len(valid_retrieval) if valid_retrieval else 0 ), ), generationGenerationMetrics( faithfulness0, # 需接入专用评测模型 answer_relevance0, keyword_coverage( sum(valid_generation) / len(valid_generation) if valid_generation else 0 ), ), baseline_comparisonself._compare_with_baseline(report) if self.baseline else {}, failed_casesfailed_cases, ) # 如果是首次评估——设为基线 if self.baseline is None: self.baseline report return report def _compare_with_baseline(self, current: EvalReport) - Dict[str, float]: 与基线对比——检测质量退化 if not self.baseline: return {} return { recall_delta: ( current.retrieval.recall_at_5 - self.baseline.retrieval.recall_at_5 ), mrr_delta: ( current.retrieval.mrr - self.baseline.retrieval.mrr ), keyword_delta: ( current.generation.keyword_coverage - self.baseline.generation.keyword_coverage ), } def is_quality_regression(self, report: EvalReport) - bool: 判断是否存在质量退化 if not self.baseline: return False comp report.baseline_comparison # 任一指标下降超过 5% 即告警 thresholds { recall_delta: -0.05, mrr_delta: -0.05, keyword_delta: -0.05, } for metric, delta in comp.items(): if metric in thresholds and delta thresholds[metric]: print(f质量退化告警: {metric} {delta:.3f}) return True return False四、持续改进的运作流程周度评估每周自动运行一次全量评估对比基线指标。任一指标下降超过 5% 触发告警。月度审查每月 Review 一次测试用例集——添加用户反馈中暴露的新类型问题移除不再适用的旧用例。变更触发知识库更新、模型切换、Prompt 变更时立即触发评估结果与基线对比。退化超过阈值时阻断变更。人工抽检的补充自动化评估能检测答非所问但无法检测答案虽然相关但措辞不当。每月随机抽样 50 条实际用户对话进行人工评分。五、总结RAG 质量持续改进的四层体系检索质量——召回率、精确率、MRR生成质量——忠实度、相关性、关键词覆盖端到端质量——答案正确率、用户满意度系统健康——延迟、Token 消耗、缓存命中率。落地建议先建立 30-50 条黄金测试用例集覆盖主要业务场景实现自动化评估脚本每周自动运行并对比基线知识库更新时触发评估质量退化则阻断上线每月通过用户反馈补充测试用例保持测试集代表性。