这是我学习 Evaluation-driven RAG 的完整记录。通过 3 个递进式 demo，从零跑通 Evaluation-driven RAG 的核心循环：评估 → 分析 → 优化 → 迭代。
前言
前面我学了 GraphRAG（关系链检索）、Agentic RAG（多轮流程）、Verification RAG（结果验证），系统已经很完善了。但我发现了一个新问题：系统推荐完就结束了，没有评估效果，无法持续改进。
比如系统推荐了 3 套房子，但用户最后买的是第 4 套。系统无法知道为什么推荐失败，也无法学习和改进。
一句话定位：
- GraphRAG 是"检索能力升级"（从相似度 → 关系链）
- Agentic RAG 是"流程能力升级"（从一次检索 → 多轮流程）
- Verification RAG 是"质量能力升级"（从相信结果 → 验证结果）
- Evaluation-driven RAG 是"优化能力升级"（从被动接受 → 主动评估优化）
Evaluation-driven RAG 不是"更强的检索"，而是用评估指标驱动系统持续改进，让系统能够自我学习和优化。
本文记录了我从零开始学习 Evaluation-driven RAG 的完整过程，包括 3 个本地 demo 和对应的讲解。
第二部分：Verification RAG 的瓶颈 + Evaluation-driven RAG 的改进
Verification RAG 的问题
场景： 系统推荐了 3 套房子，用户最后买的是第 4 套
Verification RAG 的做法：
Step 1: Plan（拆解需求）
Step 2: Retrieve（多路检索）
Step 3: Verify（验证关键信息）
Step 4: Reflect（查缺口）
Step 5: Iterate（补充信息）
Step 6: Answer（生成推荐）
推荐完成 → 结束
Verification RAG 的问题：
❌ 问题 1：无法评估推荐质量
系统推荐完就结束了，不知道推荐是否有效
例如：用户最后买的房子和推荐的不一样
结果：系统无法知道失败原因
❌ 问题 2：无法检测系统瓶颈
系统运行了 1000 次，但不知道哪个环节有问题
例如：是 Retrieve 不准？还是 Verify 有漏洞？
结果：无法针对性优化
❌ 问题 3：无法持续改进
系统参数固定不变，无法根据实际效果调整
例如：max_rounds = 3 是否最优？
结果：系统效果停滞不前
❌ 问题 4：无法对比不同方案
不知道"加验证"是否真的有效
例如：Verification RAG vs Agentic RAG，哪个更好？
结果：无法做出数据驱动的决策
Evaluation-driven RAG 的改进方案
核心思想： 在 Verification RAG 的基础上，加一层"评估反馈循环"
Verification RAG 的流程：
Plan → Retrieve → Verify → Reflect → Iterate → Answer
Evaluation-driven RAG 的流程：
Plan → Retrieve → Verify → Reflect → Iterate → Answer
↓
Evaluate（新增）
↓
Analyze（新增）
↓
Optimize（新增）
↓
Feedback Loop（新增）
Evaluation-driven RAG 的 3 个改进
改进 1：评估推荐质量 + 反馈循环
Verification RAG：
推荐 → 结束
问题：不知道推荐是否有效
Evaluation-driven RAG：
推荐 → 评估（用户反馈、实际成交）
→ 计算评估指标（Recall、Precision、MRR、NDCG）
→ 反馈给系统
优势：系统知道推荐的质量，可以学习
改进 2：检测系统瓶颈 + 针对性优化
Verification RAG：
系统运行 1000 次，但不知道哪个环节有问题
Evaluation-driven RAG：
系统运行 1000 次 → 分析每个环节的效果
– Retrieve 的准确率：85%
– Verify 的冲突检测率：92%
– Reflect 的缺口检测率：78%
发现瓶颈 → Reflect 的缺口检测率最低
针对性优化 → 改进 Reflect 的缺口检测逻辑
优势：系统能自动发现瓶颈，针对性优化
改进 3：参数自适应 + 持续优化
Verification RAG：
max_rounds = 3（固定）
max_retrieval_per_round = 3（固定）
问题：这些参数是否最优？
Evaluation-driven RAG：
初始参数：max_rounds = 3
运行 100 次 → 评估效果
分析：max_rounds = 2 时准确率 92%，max_rounds = 3 时准确率 91%
优化：调整 max_rounds = 2
继续运行 100 次 → 再次评估
分析：max_rounds = 2 时准确率 93%
优化：保持 max_rounds = 2
优势：系统能根据实际效果自动调整参数
第三部分：3 个递进式 Demo（基于 Verification RAG 的改进）
Demo 1：在 Verification RAG 的 Answer 后加 Evaluate
目标： 理解"怎么评估推荐质量"
Verification RAG 的流程：
Plan → Retrieve → Verify → Reflect → Iterate → Answer
改进后的流程：
Plan → Retrieve → Verify → Reflect → Iterate → Answer → Evaluate（新增）
↓
收集反馈
评估推荐质量的代码：
def evaluate_recommendations(recommendations, user_feedback):
"""
评估推荐质量
输入：
– recommendations：系统推荐的 Top3 房源
– user_feedback：用户反馈（最后买的房子、满意度等）
输出：
– 评估指标（Recall、Precision、MRR、NDCG）
"""
evaluation = {
"recall": 0.0,
"precision": 0.0,
"mrr": 0.0,
"ndcg": 0.0,
"user_satisfaction": 0.0,
"feedback": user_feedback
}
# 指标 1：Recall（召回率）
# 定义：用户最后买的房子是否在推荐列表中
recommended_ids = [r["id"] for r in recommendations]
if user_feedback["purchased_house_id"] in recommended_ids:
evaluation["recall"] = 1.0
else:
evaluation["recall"] = 0.0
# 指标 2：Precision（精确率）
# 定义：推荐列表中有多少比例的房源用户满意
satisfied_count = sum(1 for r in recommendations if r["id"] in user_feedback.get("satisfied_houses", []))
evaluation["precision"] = satisfied_count / len(recommendations) if recommendations else 0.0
# 指标 3：MRR（Mean Reciprocal Rank）
# 定义：用户最后买的房子在推荐列表中的排名倒数
for rank, recommendation in enumerate(recommendations, 1):
if recommendation["id"] == user_feedback["purchased_house_id"]:
evaluation["mrr"] = 1.0 / rank
break
# 指标 4：NDCG（Normalized Discounted Cumulative Gain）
# 定义：考虑排名位置的累积增益
dcg = 0.0
for rank, recommendation in enumerate(recommendations, 1):
if recommendation["id"] in user_feedback.get("satisfied_houses", []):
dcg += 1.0 / (rank + 1)
# 理想情况：所有推荐都满足
idcg = sum(1.0 / (i + 1) for i in range(len(recommendations)))
evaluation["ndcg"] = dcg / idcg if idcg > 0 else 0.0
# 指标 5：用户满意度
evaluation["user_satisfaction"] = user_feedback.get("satisfaction_score", 0.0)
return evaluation
# 使用示例
recommendations = [
{"id": "h1", "name": "望京新城", "rank": 1},
{"id": "h2", "name": "望京花园", "rank": 2},
{"id": "h3", "name": "朝阳门", "rank": 3}
]
user_feedback = {
"purchased_house_id": "h2", # 用户最后买的是第 2 个推荐
"satisfied_houses": ["h1", "h2"], # 用户满意的房源
"satisfaction_score": 0.8 # 满意度 80%
}
evaluation = evaluate_recommendations(recommendations, user_feedback)
print(f"Recall: {evaluation['recall']:.2%}")
print(f"Precision: {evaluation['precision']:.2%}")
print(f"MRR: {evaluation['mrr']:.4f}")
print(f"NDCG: {evaluation['ndcg']:.4f}")
print(f"用户满意度: {evaluation['user_satisfaction']:.2%}")
关键点：
- 用多个评估指标衡量推荐质量
- 收集用户反馈
- 为后续优化提供数据支撑
Demo 2：批量评估 + 分析系统瓶颈
目标： 理解"怎么检测系统瓶颈"
场景： 系统运行了 100 次，现在要分析每个环节的效果
代码实现：
def batch_evaluate_system(query_results, user_feedbacks):
"""
批量评估系统效果
输入：
– query_results：100 次查询的结果
– user_feedbacks：100 次查询的用户反馈
输出：
– 系统级别的评估指标
– 每个环节的效果分析
"""
# 第 1 步：计算整体指标
all_recalls = []
all_precisions = []
all_mrrs = []
all_ndcgs = []
for query_result, user_feedback in zip(query_results, user_feedbacks):
evaluation = evaluate_recommendations(query_result["recommendations"], user_feedback)
all_recalls.append(evaluation["recall"])
all_precisions.append(evaluation["precision"])
all_mrrs.append(evaluation["mrr"])
all_ndcgs.append(evaluation["ndcg"])
system_metrics = {
"avg_recall": sum(all_recalls) / len(all_recalls),
"avg_precision": sum(all_precisions) / len(all_precisions),
"avg_mrr": sum(all_mrrs) / len(all_mrrs),
"avg_ndcg": sum(all_ndcgs) / len(all_ndcgs),
}
# 第 2 步：分析每个环节的效果
pipeline_analysis = {
"retrieve": analyze_retrieve_stage(query_results),
"verify": analyze_verify_stage(query_results),
"reflect": analyze_reflect_stage(query_results),
"iterate": analyze_iterate_stage(query_results),
}
# 第 3 步：识别瓶颈
bottlenecks = identify_bottlenecks(pipeline_analysis)
return {
"system_metrics": system_metrics,
"pipeline_analysis": pipeline_analysis,
"bottlenecks": bottlenecks
}
def analyze_retrieve_stage(query_results):
"""分析 Retrieve 阶段的效果"""
analysis = {
"avg_candidates": 0.0,
"avg_recall_at_k": 0.0,
"precision_at_k": 0.0
}
candidate_counts = []
recalls = []
for result in query_results:
candidates = result.get("candidates", [])
candidate_counts.append(len(candidates))
# 计算 Recall@K
relevant_count = sum(1 for c in candidates if c.get("is_relevant", False))
recall = relevant_count / len(candidates) if candidates else 0.0
recalls.append(recall)
analysis["avg_candidates"] = sum(candidate_counts) / len(candidate_counts)
analysis["avg_recall_at_k"] = sum(recalls) / len(recalls)
return analysis
def analyze_verify_stage(query_results):
"""分析 Verify 阶段的效果"""
analysis = {
"avg_confidence": 0.0,
"conflict_detection_rate": 0.0,
"avg_conflicts_per_query": 0.0
}
confidences = []
conflicts_detected = 0
total_conflicts = 0
for result in query_results:
verifications = result.get("verifications", {})
for field, verification in verifications.items():
confidences.append(verification.get("confidence", 0.0))
if verification.get("conflicts"):
conflicts_detected += 1
total_conflicts += len(verification["conflicts"])
analysis["avg_confidence"] = sum(confidences) / len(confidences) if confidences else 0.0
analysis["conflict_detection_rate"] = conflicts_detected / len(query_results) if query_results else 0.0
analysis["avg_conflicts_per_query"] = total_conflicts / len(query_results) if query_results else 0.0
return analysis
def analyze_reflect_stage(query_results):
"""分析 Reflect 阶段的效果"""
analysis = {
"gap_detection_rate": 0.0,
"avg_gaps_per_query": 0.0,
"gap_resolution_rate": 0.0
}
gaps_detected = 0
total_gaps = 0
gaps_resolved = 0
for result in query_results:
gaps = result.get("gaps", [])
gaps_detected += len(gaps)
total_gaps += len(gaps)
# 检查缺口是否被解决
for gap in gaps:
if gap.get("resolved", False):
gaps_resolved += 1
analysis["gap_detection_rate"] = gaps_detected / len(query_results) if query_results else 0.0
analysis["avg_gaps_per_query"] = total_gaps / len(query_results) if query_results else 0.0
analysis["gap_resolution_rate"] = gaps_resolved / total_gaps if total_gaps > 0 else 0.0
return analysis
def identify_bottlenecks(pipeline_analysis):
"""识别系统瓶颈"""
bottlenecks = []
# 检查 Retrieve 阶段
if pipeline_analysis["retrieve"]["avg_recall_at_k"] < 0.7:
bottlenecks.append({
"stage": "retrieve",
"issue": "召回率低",
"metric": pipeline_analysis["retrieve"]["avg_recall_at_k"],
"recommendation": "改进检索算法或增加候选数量"
})
# 检查 Verify 阶段
if pipeline_analysis["verify"]["avg_confidence"] < 0.8:
bottlenecks.append({
"stage": "verify",
"issue": "置信度低",
"metric": pipeline_analysis["verify"]["avg_confidence"],
"recommendation": "增加数据源或改进冲突解决策略"
})
# 检查 Reflect 阶段
if pipeline_analysis["reflect"]["gap_resolution_rate"] < 0.8:
bottlenecks.append({
"stage": "reflect",
"issue": "缺口解决率低",
"metric": pipeline_analysis["reflect"]["gap_resolution_rate"],
"recommendation": "改进缺口检测或补检索逻辑"
})
return bottlenecks
# 使用示例
query_results = [...] # 100 次查询的结果
user_feedbacks = [...] # 100 次查询的用户反馈
evaluation = batch_evaluate_system(query_results, user_feedbacks)
print(f"系统指标：")
print(f" Avg Recall: {evaluation['system_metrics']['avg_recall']:.2%}")
print(f" Avg Precision: {evaluation['system_metrics']['avg_precision']:.2%}")
print(f" Avg MRR: {evaluation['system_metrics']['avg_mrr']:.4f}")
print(f" Avg NDCG: {evaluation['system_metrics']['avg_ndcg']:.4f}")
print(f"\\n瓶颈分析：")
for bottleneck in evaluation["bottlenecks"]:
print(f" {bottleneck['stage']}: {bottleneck['issue']}")
print(f" 建议：{bottleneck['recommendation']}")
关键点：
- 批量评估系统效果
- 分析每个环节的性能
- 自动识别瓶颈
Demo 3：根据评估结果优化参数
目标： 理解"怎么根据评估结果调整系统参数"
场景： 发现 Reflect 阶段的缺口解决率低，需要优化
代码实现：
def optimize_parameters_based_evaluation(evaluation_results, current_params):
"""
根据评估结果优化参数
"""
optimized_params = current_params.copy()
optimization_log = []
# 优化 1：如果 Retrieve 的召回率低，增加候选数量
if evaluation_results["pipeline_analysis"]["retrieve"]["avg_recall_at_k"] < 0.7:
old_value = optimized_params.get("max_candidates", 50)
optimized_params["max_candidates"] = int(old_value * 1.2) # 增加 20%
optimization_log.append({
"parameter": "max_candidates",
"old_value": old_value,
"new_value": optimized_params["max_candidates"],
"reason": "Retrieve 召回率低"
})
# 优化 2：如果 Verify 的置信度低，增加数据源
if evaluation_results["pipeline_analysis"]["verify"]["avg_confidence"] < 0.8:
old_value = optimized_params.get("num_data_sources", 3)
optimized_params["num_data_sources"] = old_value + 1
optimization_log.append({
"parameter": "num_data_sources",
"old_value": old_value,
"new_value": optimized_params["num_data_sources"],
"reason": "Verify 置信度低"
})
# 优化 3：如果 Reflect 的缺口解决率低，增加迭代轮数
if evaluation_results["pipeline_analysis"]["reflect"]["gap_resolution_rate"] < 0.8:
old_value = optimized_params.get("max_iterations", 3)
optimized_params["max_iterations"] = old_value + 1
optimization_log.append({
"parameter": "max_iterations",
"old_value": old_value,
"new_value": optimized_params["max_iterations"],
"reason": "Reflect 缺口解决率低"
})
# 优化 4：如果整体 Recall 低，调整 Reflect 的缺口检测阈值
if evaluation_results["system_metrics"]["avg_recall"] < 0.7:
old_value = optimized_params.get("gap_detection_threshold", 0.7)
optimized_params["gap_detection_threshold"] = old_value – 0.1 # 降低阈值，更容易检测缺口
optimization_log.append({
"parameter": "gap_detection_threshold",
"old_value": old_value,
"new_value": optimized_params["gap_detection_threshold"],
"reason": "整体 Recall 低"
})
return optimized_params, optimization_log
def run_optimization_loop(initial_params, num_iterations=5):
"""
运行优化循环
"""
current_params = initial_params.copy()
optimization_history = []
for iteration in range(num_iterations):
print(f"\\n=== 优化循环 {iteration + 1} ===")
# 第 1 步：用当前参数运行系统
query_results = run_system_with_params(current_params, num_queries=100)
user_feedbacks = collect_user_feedbacks(query_results)
# 第 2 步：评估系统效果
evaluation = batch_evaluate_system(query_results, user_feedbacks)
print(f"Avg Recall: {evaluation['system_metrics']['avg_recall']:.2%}")
print(f"Avg Precision: {evaluation['system_metrics']['avg_precision']:.2%}")
# 第 3 步：根据评估结果优化参数
optimized_params, optimization_log = optimize_parameters_based_evaluation(evaluation, current_params)
# 第 4 步：记录优化历史
optimization_history.append({
"iteration": iteration + 1,
"metrics": evaluation["system_metrics"],
"optimizations": optimization_log,
"new_params": optimized_params
})
# 第 5 步：更新参数
current_params = optimized_params
# 打印优化日志
for opt in optimization_log:
print(f" 优化：{opt['parameter']} {opt['old_value']} → {opt['new_value']}")
return optimization_history
# 使用示例
initial_params = {
"max_candidates": 50,
"num_data_sources": 3,
"max_iterations": 3,
"gap_detection_threshold": 0.7
}
history = run_optimization_loop(initial_params, num_iterations=5)
print(f"\\n=== 优化总结 ===")
for h in history:
print(f"循环 {h['iteration']}: Recall {h['metrics']['avg_recall']:.2%}, Precision {h['metrics']['avg_precision']:.2%}")
关键点：
- 根据评估指标自动调整参数
- 运行多轮优化循环
- 持续改进系统效果
第五部分：现实代价（相比 Verification RAG 的额外成本）
难：需要收集用户反馈
Verification RAG 的做法：
推荐完就结束了，不需要用户反馈
Evaluation-driven RAG 的做法：
需要收集用户反馈（最后买的房子、满意度等）
成本高，但能评估系统效果
工程成本： 需要"反馈收集系统"的工作
贵：评估本身有成本
成本公式：
Verification RAG 成本：
≈ 39x（相比普通 RAG）
Evaluation-driven RAG 成本：
= Verification RAG 成本
+ 评估成本（计算评估指标）
+ 分析成本（分析瓶颈）
+ 优化成本（调整参数）
≈ 39x + 5x
≈ 44x
结论： Evaluation-driven RAG 的成本是 Verification RAG 的 1.1 倍（增加 10%）
慢：评估需要时间
延迟公式：
Verification RAG 延迟：
≈ 4050ms（4 秒）
Evaluation-driven RAG 延迟：
= Verification RAG 延迟
+ 评估延迟（计算指标）
≈ 4050ms + 50ms
≈ 4100ms（4.1 秒）
结论： Evaluation-driven RAG 的延迟增加 1%，但能持续改进系统
第六部分：企业级迁移（基于 Verification RAG 的改进）
改进 1：建立评估框架
class EvaluationFramework:
"""评估框架"""
def __init__(self):
self.metrics = {}
self.history = []
def register_metric(self, name, metric_func):
"""注册评估指标"""
self.metrics[name] = metric_func
def evaluate(self, recommendations, user_feedback):
"""评估推荐"""
results = {}
for name, metric_func in self.metrics.items():
results[name] = metric_func(recommendations, user_feedback)
self.history.append(results)
return results
def get_average_metrics(self, window_size=100):
"""获取平均指标"""
if len(self.history) < window_size:
window_size = len(self.history)
recent_history = self.history[–window_size:]
avg_metrics = {}
for metric_name in self.metrics.keys():
values = [h[metric_name] for h in recent_history]
avg_metrics[metric_name] = sum(values) / len(values)
return avg_metrics
改进 2：建立反馈收集系统
class FeedbackCollector:
"""反馈收集系统"""
def __init__(self):
self.feedbacks = []
def collect_feedback(self, query_id, recommendations, user_action):
"""收集用户反馈"""
feedback = {
"query_id": query_id,
"recommendations": recommendations,
"purchased_house_id": user_action.get("purchased_house_id"),
"satisfied_houses": user_action.get("satisfied_houses", []),
"satisfaction_score": user_action.get("satisfaction_score", 0.0),
"timestamp": time.time()
}
self.feedbacks.append(feedback)
return feedback
def get_recent_feedbacks(self, num_feedbacks=100):
"""获取最近的反馈"""
return self.feedbacks[–num_feedbacks:]
改进 3：建立优化引擎
class OptimizationEngine:
"""优化引擎"""
def __init__(self, initial_params):
self.current_params = initial_params.copy()
self.optimization_history = []
def optimize(self, evaluation_results):
"""根据评估结果优化参数"""
optimized_params, optimization_log = optimize_parameters_based_evaluation(
evaluation_results,
self.current_params
)
self.optimization_history.append({
"metrics": evaluation_results["system_metrics"],
"optimizations": optimization_log,
"new_params": optimized_params
})
self.current_params = optimized_params
return optimized_params
def get_optimization_history(self):
"""获取优化历史"""
return self.optimization_history
改进 4：建立监控系统
class MonitoringSystem:
"""监控系统"""
def __init__(self, evaluation_framework, feedback_collector):
self.evaluation_framework = evaluation_framework
self.feedback_collector = feedback_collector
self.alerts = []
def check_health(self):
"""检查系统健康状态"""
avg_metrics = self.evaluation_framework.get_average_metrics()
# 检查 Recall
if avg_metrics.get("avg_recall", 1.0) < 0.7:
self.alerts.append({
"level": "warning",
"message": f"Recall 低于 70%：{avg_metrics['avg_recall']:.2%}"
})
# 检查 Precision
if avg_metrics.get("avg_precision", 1.0) < 0.6:
self.alerts.append({
"level": "warning",
"message": f"Precision 低于 60%：{avg_metrics['avg_precision']:.2%}"
})
# 检查用户满意度
if avg_metrics.get("avg_satisfaction", 1.0) < 0.7:
self.alerts.append({
"level": "critical",
"message": f"用户满意度低于 70%：{avg_metrics['avg_satisfaction']:.2%}"
})
return self.alerts
第七部分：常见问题（基于 Verification RAG 的改进）
Q1：怎么收集用户反馈？
现象： 不知道用户最后买的房子是哪一套
解决方案：
# 方案 1：主动询问
def ask_user_feedback(user_id, recommendations):
"""主动询问用户反馈"""
feedback = {
"purchased_house_id": input("你最后买的房子是哪一套？"),
"satisfied_houses": input("你满意的房源有哪些？").split(","),
"satisfaction_score": float(input("你的满意度是多少（0-1）？"))
}
return feedback
# 方案 2：被动收集
def collect_passive_feedback(user_id, recommendations):
"""从用户行为被动收集反馈"""
# 从用户的浏览历史、点击记录等推断
feedback = {
"purchased_house_id": infer_from_user_behavior(user_id),
"satisfied_houses": infer_satisfied_houses(user_id),
"satisfaction_score": infer_satisfaction_score(user_id)
}
return feedback
# 方案 3：混合方式
def collect_hybrid_feedback(user_id, recommendations):
"""混合主动和被动反馈"""
passive = collect_passive_feedback(user_id, recommendations)
# 如果被动反馈不确定，主动询问
if passive["satisfaction_score"] < 0.5:
active = ask_user_feedback(user_id, recommendations)
return merge_feedbacks(passive, active)
return passive
Q2：优化参数时怎么避免过度优化？
现象： 参数一直在变化，系统不稳定
解决方案：
# 方案 1：设置最小改进阈值
def should_optimize(old_metrics, new_metrics, threshold=0.01):
"""判断是否应该优化"""
improvement = new_metrics["avg_recall"] – old_metrics["avg_recall"]
return improvement > threshold
# 方案 2：限制参数变化幅度
def limit_parameter_change(old_param, new_param, max_change=0.2):
"""限制参数变化幅度"""
change_ratio = abs(new_param – old_param) / old_param
if change_ratio > max_change:
# 只改变 max_change 的幅度
new_param = old_param * (1 + max_change * (1 if new_param > old_param else –1))
return new_param
# 方案 3：使用平滑优化
def smooth_optimization(old_params, new_params, smoothing_factor=0.7):
"""平滑优化：新参数 = 旧参数 × smoothing_factor + 新参数 × (1 – smoothing_factor)"""
smoothed_params = {}
for key in old_params:
smoothed_params[key] = old_params[key] * smoothing_factor + new_params[key] * (1 – smoothing_factor)
return smoothed_params
Q3：怎么知道优化是否有效？
现象： 不知道优化后系统是否真的变好了
解决方案：
# 方案 1：A/B 测试
def ab_test(old_params, new_params, num_queries=1000):
"""A/B 测试"""
# 50% 用旧参数，50% 用新参数
old_results = run_system_with_params(old_params, num_queries // 2)
new_results = run_system_with_params(new_params, num_queries // 2)
old_metrics = evaluate_results(old_results)
new_metrics = evaluate_results(new_results)
improvement = new_metrics["avg_recall"] – old_metrics["avg_recall"]
return {
"old_metrics": old_metrics,
"new_metrics": new_metrics,
"improvement": improvement,
"is_significant": improvement > 0.02 # 改进 > 2% 才认为显著
}
# 方案 2：统计显著性检验
def statistical_significance_test(old_results, new_results):
"""统计显著性检验"""
from scipy import stats
old_recalls = [r["recall"] for r in old_results]
new_recalls = [r["recall"] for r in new_results]
t_stat, p_value = stats.ttest_ind(old_recalls, new_recalls)
return {
"t_stat": t_stat,
"p_value": p_value,
"is_significant": p_value < 0.05 # p < 0.05 认为显著
}
总结
Evaluation-driven RAG 相比 Verification RAG 的改进：
Verification RAG：
✅ 多轮流程 + 验证机制
❌ 无法评估效果、无法持续改进
Evaluation-driven RAG：
✅ 多轮流程 + 验证机制 + 评估反馈循环
✅ 能评估推荐质量、能检测瓶颈、能自动优化
❌ 成本增加 10%、需要收集用户反馈
什么时候用 Evaluation-driven RAG？
✅ 用 Evaluation-driven RAG：
– 需要持续改进系统效果
– 有充足的用户反馈数据
– 能接受 4.1 秒的延迟
– 想要数据驱动的决策
❌ 不用 Evaluation-driven RAG：
– 系统已经稳定，不需要改进
– 没有用户反馈数据
– 延迟要求 < 4 秒
– 不关心系统优化
怎么快速上手？
第 1 步：从 Demo 1 开始
理解"怎么评估推荐质量"
第 2 步：到 Demo 2
理解"怎么检测系统瓶颈"
第 3 步：到 Demo 3
理解"怎么根据评估结果优化参数"
第 4 步：理解代价
反馈收集、评估成本的权衡
完整的 RAG 系列回顾：
GraphRAG（第 1 篇）
↓ 解决"关系链检索"
Agentic RAG（第 2 篇）
↓ 解决"多轮流程编排"
Verification RAG（第 3 篇）
↓ 解决"结果验证"
Evaluation-driven RAG（第 4 篇）
↓ 解决"持续优化"
完整的房产推荐系统 ✅
最后一句话：
Evaluation-driven RAG 解决的是"怎么持续改进系统"。
它在 Verification RAG 的基础上加一层"评估反馈循环"，让系统从"静态推荐"升级到"动态优化"。
这四篇博客形成了一个完整的"构建→优化"闭环，从基础的关系链检索，到多轮流程编排，再到结果验证，最后到持续优化。
这就是我的 RAG 系列学习笔记。希望能帮助你快速上手！
