Verification RAG 房产推荐学习笔记:检索结果不能盲信,需要"验证机制"
这是我学习 Verification RAG 的完整记录。通过 3 个递进式 demo,从零跑通 Verification RAG 的核心循环:检索 → 验证 → 纠错 → 输出。
前言
前面我学了 GraphRAG(关系链检索)和 Agentic RAG(多轮流程编排),系统变得越来越强。但我发现了一个新问题:检索结果不一定准确,有时候会"幻觉"。
比如系统推荐"望京新城对应 XX小学",但实际上对应的是 YY小学。或者说"通勤 35 分钟",但实际是 45 分钟。
一句话定位:
- GraphRAG 是"检索模块升级"(从相似度升级到关系链)
- Agentic RAG 是"咨询流程升级"(从一次检索升级到多轮流程)
- Verification RAG 是"质量保证升级"(从"相信检索结果"升级到"验证检索结果")
Verification RAG 不是"更强的检索",而是在检索后加一层"验证阀",确保推荐的房源信息是准确的。
本文记录了我从零开始学习 Verification RAG 的完整过程,包括 3 个本地 demo 和对应的讲解。
第二部分:问题分类:什么时候需要验证
低风险问题(不需要验证)
特征:错误成本低、用户容错度高
例子:
"推荐一些朝阳区的房源"
"有哪些学区房"
"500万预算的房子"
为什么不需要验证:
– 用户只是浏览,不会立即下单
– 即使有错误,用户会自己核实
– 错误成本低
高风险问题(必须验证)
特征:错误成本高、用户依赖度高
例子:
"这套房子对应哪个学校?"(学区信息错误 = 买错房)
"通勤时间是多少?"(通勤信息错误 = 选错房)
"这个小区的物业费是多少?"(价格信息错误 = 成本计算错误)
"这套房子的产权年限是多少?"(产权信息错误 = 法律风险)
为什么必须验证:
– 用户会基于这个信息做决策
– 错误信息会导致用户买错房
– 错误成本极高(几百万的损失)
判断标准
用验证机制:
– 信息涉及"学区、通勤、价格、产权"等关键决策因素
– 用户会基于这个信息做重大决策
– 错误信息的成本 > 验证的成本
不用验证机制:
– 信息只是参考,不影响决策
– 用户会自己核实
– 错误信息的成本 < 验证的成本
第三部分:核心概念
Verification RAG 的 5 个核心概念
| Retrieve | 检索候选信息 | 从知识库检索"望京新城对应学校" |
| Verify | 验证信息准确性 | 用多个数据源交叉验证学校信息 |
| Conflict Detection | 检测冲突 | 发现"XX小学"和"YY小学"的矛盾 |
| Conflict Resolution | 解决冲突 | 选择最可信的数据源 |
| Confidence Score | 置信度评分 | 给出"学区信息的置信度 95%" |
类比理解
传统 RAG:相信检索结果
用户问:"望京新城对应哪个学校?"
系统:检索 → "XX小学"
问题:如果检索错了怎么办?
Verification RAG:验证检索结果
用户问:"望京新城对应哪个学校?"
系统:
1. 检索 → "XX小学"
2. 验证 → 用多个数据源交叉验证
3. 冲突检测 → 发现"XX小学"和"YY小学"的矛盾
4. 冲突解决 → 选择最可信的数据源
5. 输出 → "XX小学(置信度 95%)"
优势:准确、可信、有置信度
第二部分:Agentic RAG 的瓶颈 + Verification RAG 的改进
Agentic RAG 的问题
场景: 用户问"我在国贸上班,预算 600万,想买三居,最好学区,安静采光好,推荐 3 套并说理由"
Agentic RAG 的做法:
Step 1: Plan(拆解需求)
硬条件:预算 600万、户型三居、地区朝阳
软条件:学区、通勤 40 分钟
偏好:安静、采光好
Step 2: 多路检索(第 1 轮)
路径 1:结构化过滤 → 预算 ≤ 600万 AND 户型 = 3 居
路径 2:GraphRAG 子图 → 通勤 40 分钟以内的小区
路径 3:GraphRAG 子图 → 有学区的小区
路径 4:向量检索 → "安静采光好"的房源描述
合并结果 → 候选 Top-50
Step 3: Reflect(查缺口)
发现:某些房源的"采光信息"缺失
决定:触发补检索
Step 4: 多路检索(第 2 轮)
补检索:采光信息
合并结果 → 候选 Top-20
Step 5: Answer(结构化输出)
生成 Top3 推荐
生成对比表(预算/户型/通勤/学区/采光/噪音)
生成理由(每条推荐都有证据)
Agentic RAG 的问题:
问题 1:多个数据源冲突
系统从多个数据源检索,但没有验证
例如:学区数据库说"XX小学",房源描述说"YY小学"
结果:系统可能返回错误的学校信息
问题 2:信息过时
系统检索到的数据可能是旧的
例如:2020年的学区政策已经改变,但系统还在用旧数据
结果:用户基于过时信息做决策
问题 3:无法判断信息可信度
系统返回推荐,但用户不知道这个信息有多可信
例如:系统说"通勤 35 分钟",但这是基于什么数据源?
结果:用户无法评估风险
问题 4:没有质量反馈
系统推荐完就结束了,没有评估推荐的质量
例如:用户最后买的房子和推荐的不一样
结果:系统无法学习和改进
Verification RAG 的改进方案
核心思想: 在 Agentic RAG 的基础上,加一层"验证阀"
Agentic RAG 的流程:
Plan → Retrieve → Reflect → Iterate → Answer
Verification RAG 的流程:
Plan → Retrieve → Verify → Reflect → Iterate → Answer
↑
新增验证步骤
Verification RAG 的 3 个改进
改进 1:多源验证 + 置信度评分
Agentic RAG:
检索 → "望京新城对应 XX小学"
问题:不知道这个信息有多可信
Verification RAG:
检索 → 从 3 个数据源验证
数据源 1(学区数据库):XX小学
数据源 2(房源描述):XX小学
数据源 3(小区官网):XX小学
验证结果 → "XX小学(3/3 数据源确认,置信度 100%)"
优势:用户知道这个信息有多可信
改进 2:冲突检测 + 自动纠错
Agentic RAG:
检索 → "望京新城对应 XX小学"
问题:如果数据源冲突,系统无法处理
Verification RAG:
检索 → 从 3 个数据源验证
数据源 1(学区数据库):XX小学
数据源 2(房源描述):YY小学
数据源 3(小区官网):XX小学
冲突检测 → 发现"XX小学"vs"YY小学"的矛盾
冲突解决 → 按优先级选择(学区数据库 > 小区官网 > 房源描述)
最终答案 → "XX小学(2/3 数据源确认,置信度 67%,存在冲突)"
优势:系统能自动处理冲突,用户能看到风险
改进 3:信息质量评分 + 可追溯性
Agentic RAG:
输出 → Top3 推荐 + 对比表
问题:用户不知道每条信息的质量
Verification RAG:
输出 → Top3 推荐 + 对比表 + 质量评分
例如:
房源 1:望京新城
– 价格:600万(来自房源数据库,置信度 100%)
– 学区:XX小学(来自学区数据库,置信度 100%)
– 通勤:35分钟(来自地铁时刻表,置信度 95%)
– 采光:南北通透(来自房源描述,置信度 70%)
优势:用户能看到每条信息的来源和可信度
第三部分:3 个递进式 Demo(基于 Agentic RAG 的改进)
Demo 1:在 Agentic RAG 的 Retrieve 步骤后加 Verify
目标: 理解"怎么在多路检索后验证结果"
场景: Agentic RAG 的多路检索完成后,对关键信息进行验证
Agentic RAG 的流程:
Plan → Retrieve(多路检索)→ Reflect → Iterate → Answer
改进后的流程:
Plan → Retrieve(多路检索)→ Verify(验证关键信息)→ Reflect → Iterate → Answer
↑
新增验证步骤
验证"望京新城对应哪个学校"的代码:
def verify_after_agentic_retrieval(candidates, data_sources):
"""
在 Agentic RAG 的多路检索后,对候选房源进行验证
输入:Agentic RAG 返回的候选房源
输出:验证后的候选房源 + 置信度评分
"""
verified_candidates = []
for candidate in candidates:
community_name = candidate["community"]
# 验证关键信息:学校、地铁、通勤时间
verifications = {
"school": verify_field(community_name, "school", data_sources),
"subway": verify_field(community_name, "subway", data_sources),
"commute": verify_field(community_name, "commute", data_sources),
}
# 添加验证结果到候选房源
candidate["verifications"] = verifications
candidate["overall_confidence"] = calculate_overall_confidence(verifications)
verified_candidates.append(candidate)
return verified_candidates
def verify_field(entity, field, data_sources):
"""
验证单个字段
"""
result = {
"field": field,
"sources": {},
"final_answer": None,
"confidence": 0.0,
"conflicts": []
}
# 从多个数据源检索
for source_name, source_data in data_sources.items():
if entity in source_data and field in source_data[entity]:
value = source_data[entity][field]
result["sources"][source_name] = value
# 检测冲突
unique_values = set(result["sources"].values())
if len(unique_values) == 0:
result["final_answer"] = "信息缺失"
result["confidence"] = 0.0
elif len(unique_values) == 1:
result["final_answer"] = list(unique_values)[0]
result["confidence"] = len(result["sources"]) / len(data_sources)
else:
result["conflicts"] = list(unique_values)
# 冲突解决:按优先级选择
priority = {
"official_db": 3,
"community_website": 2,
"house_desc": 1
}
best_source = max(result["sources"].items(), key=lambda x: priority.get(x[0], 0))
result["final_answer"] = best_source[1]
result["confidence"] = priority.get(best_source[0], 0) / 3
return result
def calculate_overall_confidence(verifications):
"""计算整体置信度"""
if not verifications:
return 0.0
total_confidence = sum(v["confidence"] for v in verifications.values())
return total_confidence / len(verifications)
# 使用示例
agentic_candidates = [
{
"id": "h1",
"name": "望京新城",
"community": "c1",
"price": 600
}
]
data_sources = {
"official_db": {
"c1": {"school": "XX小学", "subway": "15号线", "commute": 35}
},
"house_desc": {
"c1": {"school": "XX小学", "subway": "15号线", "commute": 35}
},
"community_website": {
"c1": {"school": "XX小学", "subway": "15号线", "commute": 35}
}
}
verified = verify_after_agentic_retrieval(agentic_candidates, data_sources)
for candidate in verified:
print(f"\\n房源:{candidate['name']}")
print(f"整体置信度:{candidate['overall_confidence']:.2%}")
for field, verification in candidate["verifications"].items():
print(f" {field}:{verification['final_answer']}(置信度 {verification['confidence']:.2%})")
if verification['conflicts']:
print(f" 冲突:{verification['conflicts']}")
关键点:
- 在 Agentic RAG 的多路检索后插入验证步骤
- 对每个候选房源的关键信息进行验证
- 计算整体置信度,用于后续排序
Demo 2:在 Agentic RAG 的 Reflect 步骤中加入验证
目标: 理解"怎么在反思阶段检测信息质量问题"
Agentic RAG 的 Reflect 步骤:
检查是否有缺口 → 发现"采光信息缺失" → 触发补检索
改进后的 Reflect 步骤:
检查是否有缺口 → 检查信息质量 → 发现"采光信息缺失"或"学区信息冲突" → 触发补检索或验证
↑
新增质量检查
代码实现:
def reflect_with_verification(candidates, plan, data_sources, iteration=1, max_iterations=3):
"""
改进的 Reflect 步骤:不仅检查缺口,还检查信息质量
"""
# Step 1: 检查缺口(原有逻辑)
gaps = check_gaps(candidates, plan)
# Step 2: 检查信息质量(新增逻辑)
quality_issues = check_information_quality(candidates, data_sources)
# Step 3: 合并缺口和质量问题
all_issues = gaps + quality_issues
if not all_issues or iteration >= max_iterations:
return candidates
# Step 4: 根据问题类型采取不同的行动
for issue in all_issues:
if issue["type"] == "missing":
# 缺口:触发补检索
results = retrieve_missing_info(issue["field"], plan, data_sources)
candidates = merge_results(candidates, results)
elif issue["type"] == "conflict":
# 冲突:触发验证
candidates = verify_and_resolve_conflicts(candidates, issue["field"], data_sources)
elif issue["type"] == "low_confidence":
# 低置信度:触发补检索或标记风险
if issue["confidence"] < 0.5:
candidates = mark_as_risky(candidates, issue["field"])
# Step 5: 递归迭代
return reflect_with_verification(
candidates,
plan,
data_sources,
iteration=iteration + 1,
max_iterations=max_iterations
)
def check_information_quality(candidates, data_sources):
"""
检查候选房源的信息质量
"""
quality_issues = []
for candidate in candidates:
community = candidate.get("community")
# 检查关键字段的置信度
key_fields = ["school", "subway", "commute", "price"]
for field in key_fields:
# 从多个数据源检索
sources = {}
for source_name, source_data in data_sources.items():
if community in source_data and field in source_data[community]:
sources[source_name] = source_data[community][field]
# 检测问题
if len(sources) == 0:
quality_issues.append({
"type": "missing",
"field": field,
"community": community,
"confidence": 0.0
})
elif len(set(sources.values())) > 1:
# 有冲突
quality_issues.append({
"type": "conflict",
"field": field,
"community": community,
"values": list(set(sources.values())),
"confidence": 1.0 / len(set(sources.values()))
})
else:
# 计算置信度
confidence = len(sources) / len(data_sources)
if confidence < 0.7:
quality_issues.append({
"type": "low_confidence",
"field": field,
"community": community,
"confidence": confidence
})
return quality_issues
# 使用示例
candidates = [
{
"id": "h1",
"name": "望京新城",
"community": "c1",
"price": 600
}
]
data_sources = {
"official_db": {
"c1": {"school": "XX小学", "subway": "15号线", "commute": 35, "price": 600}
},
"house_desc": {
"c1": {"school": "YY小学", "subway": "15号线", "price": 600}
# 注意:缺少 commute 字段
},
"community_website": {
"c1": {"school": "XX小学", "subway": "15号线", "commute": 35}
# 注意:缺少 price 字段
}
}
plan = {"school": True, "commute_time": 40}
final_candidates = reflect_with_verification(candidates, plan, data_sources)
关键点:
- 在 Reflect 阶段不仅检查缺口,还检查信息质量
- 检测冲突、缺失、低置信度等问题
- 根据问题类型采取不同的行动
Demo 3:在 Agentic RAG 的 Answer 步骤中加入验证信息
目标: 理解"怎么在最终答案中展示验证结果"
Agentic RAG 的 Answer 输出:
推荐 Top3:
房源 1:望京新城
– 学区:XX小学
– 通勤:35分钟
– 采光:南北通透
改进后的 Answer 输出:
推荐 Top3:
房源 1:望京新城
– 学区:XX小学(置信度 100%,来自学区数据库)
– 通勤:35分钟(置信度 95%,来自地铁时刻表)
– 采光:南北通透(置信度 70%,来自房源描述)
代码实现:
def generate_answer_with_verification(top3_candidates, plan):
"""
生成包含验证信息的答案
"""
output = {
"recommendations": [],
"comparison_table": [],
"verification_details": [],
"risk_warnings": []
}
for idx, candidate in enumerate(top3_candidates, 1):
# 一句话总结
summary = generate_summary(candidate, plan)
# 提取验证信息
verifications = candidate.get("verifications", {})
# 检查是否有风险
risks = []
for field, verification in verifications.items():
if verification["confidence"] < 0.7:
risks.append({
"field": field,
"confidence": verification["confidence"],
"message": f"{field} 信息置信度较低({verification['confidence']:.0%}),建议人工核实"
})
if verification.get("conflicts"):
risks.append({
"field": field,
"message": f"{field} 存在数据源冲突:{verification['conflicts']}"
})
recommendation = {
"rank": idx,
"house": candidate,
"summary": summary,
"verifications": verifications,
"risks": risks,
"overall_confidence": candidate.get("overall_confidence", 0.0)
}
output["recommendations"].append(recommendation)
if risks:
output["risk_warnings"].extend(risks)
return output
def format_answer_for_display(answer_data):
"""
格式化答案用于展示
"""
output = "## 推荐 Top3\\n\\n"
for rec in answer_data["recommendations"]:
output += f"### 房源 {rec['rank']}:{rec['house']['name']}\\n\\n"
output += f"**一句话总结:** {rec['summary']}\\n\\n"
output += "**详细信息:**\\n"
for field, verification in rec["verifications"].items():
confidence_icon = "[高]" if verification["confidence"] >= 0.9 else "[中]" if verification["confidence"] >= 0.7 else "[低]"
output += f"- {field}:{verification['final_answer']} {confidence_icon} "
output += f"(置信度 {verification['confidence']:.0%})\\n"
if rec["risks"]:
output += "\\n**风险提示:**\\n"
for risk in rec["risks"]:
output += f"- {risk['message']}\\n"
output += f"\\n**整体置信度:** {rec['overall_confidence']:.0%}\\n\\n"
if answer_data["risk_warnings"]:
output += "## 全局风险提示\\n\\n"
for warning in answer_data["risk_warnings"]:
output += f"- {warning['message']}\\n"
return output
# 使用示例
top3_candidates = [
{
"id": "h1",
"name": "望京新城",
"community": "c1",
"price": 600,
"overall_confidence": 0.95,
"verifications": {
"school": {
"field": "school",
"final_answer": "XX小学",
"confidence": 1.0,
"conflicts": []
},
"commute": {
"field": "commute",
"final_answer": "35分钟",
"confidence": 0.95,
"conflicts": []
},
"sunlight": {
"field": "sunlight",
"final_answer": "南北通透",
"confidence": 0.7,
"conflicts": []
}
}
}
]
plan = {"school": True, "commute_time": 40}
answer = generate_answer_with_verification(top3_candidates, plan)
formatted = format_answer_for_display(answer)
print(formatted)
关键点:
- 在最终答案中展示每条信息的置信度
- 标注数据来源
- 突出风险提示
- 帮助用户做出更明智的决策
第五部分:现实代价(相比 Agentic RAG 的额外成本)
难:需要多个数据源
Agentic RAG 的做法:
只需要一个数据源就能检索
成本低,但准确性无法保证
Verification RAG 的做法:
需要多个数据源进行验证
成本高,但准确性有保证
工程成本: 需要"多源数据集成"的工作
贵:验证本身有成本
成本公式:
Agentic RAG 成本:
= 1 次 Plan(LLM 调用)
+ 多轮 Retrieve(每轮 3 次检索)
+ 多轮 Reflect(每轮 1 次 LLM 调用)
+ 1 次 Answer(LLM 调用)
≈ 5 + 4*轮数
Verification RAG 成本:
= Agentic RAG 成本
+ 验证成本(检索 × 数据源数量)
+ 冲突检测成本
+ 冲突解决成本
≈ (5 + 4*轮数) × 3
如果轮数 = 2:
Agentic RAG ≈ 13x
Verification RAG ≈ 39x(增加 3 倍)
结论: Verification RAG 的成本是 Agentic RAG 的 3 倍
慢:验证需要时间
延迟公式:
Agentic RAG 延迟:
≈ 3700ms(3.7 秒)
Verification RAG 延迟:
= Agentic RAG 延迟
+ 验证延迟(检索 × 数据源数量)
+ 冲突检测延迟
≈ 3700ms + (100ms × 3) + 50ms
≈ 4050ms(4 秒)
结论: Verification RAG 的延迟增加 10%,但准确性提升显著
第六部分:企业级迁移(基于 Agentic RAG 的改进)
改进 1:在 Agentic RAG 的 Slot 设计中加入"验证优先级"
# Agentic RAG 的 Slot 设计
hard_slots = {
"budget": {"type": "number", "required": True},
"rooms": {"type": "number", "required": True},
}
# 改进后的 Slot 设计(加入验证优先级)
hard_slots = {
"budget": {"type": "number", "required": True, "verify": True, "priority": 3},
"rooms": {"type": "number", "required": True, "verify": True, "priority": 3},
}
soft_slots = {
"school": {"type": "boolean", "required": False, "verify": True, "priority": 2},
"commute_time": {"type": "number", "required": False, "verify": True, "priority": 2},
}
optional_slots = {
"sunlight": {"type": "string", "required": False, "verify": False, "priority": 1},
}
改进 2:在多源融合中加入"验证融合"
def multi_source_fusion_with_verification(plan, data_sources):
"""
改进的多源融合:不仅融合结果,还验证结果
"""
results = {
"structured": [],
"vector": [],
"graph": []
}
# 第 1 步:多路检索(Agentic RAG 的做法)
results["structured"] = filter_by_constraints(...)
results["vector"] = vector_search(...)
results["graph"] = graphrag_search(...)
# 第 2 步:融合结果
merged = merge_multi_source(results)
# 第 3 步:验证融合结果(新增)
verified = verify_merged_results(merged, data_sources, plan)
return verified
def verify_merged_results(candidates, data_sources, plan):
"""
验证融合后的结果
"""
for candidate in candidates:
# 对每个候选房源进行验证
candidate["verifications"] = {}
# 只验证优先级高的字段
for field, slot_config in plan.items():
if slot_config.get("verify"):
verification = verify_field(candidate["community"], field, data_sources)
candidate["verifications"][field] = verification
# 计算整体置信度
candidate["overall_confidence"] = calculate_overall_confidence(candidate["verifications"])
return candidates
改进 3:在冲突处理中加入"验证冲突解决"
class VerificationConflictResolver:
"""
改进的冲突处理:不仅处理多源冲突,还处理验证冲突
"""
def __init__(self, priorities: Dict):
self.priorities = priorities
def resolve(self, values: Dict, timestamps: Dict = None) –> Tuple[str, float, str]:
"""
解决冲突
返回:(最终答案, 置信度, 解决方案)
"""
if not values:
return None, 0.0, "no_data"
# 方案 1:按优先级选择
sorted_sources = sorted(
values.items(),
key=lambda x: self.priorities.get(x[0], 0),
reverse=True
)
best_source, best_value = sorted_sources[0]
# 方案 2:按时间戳选择(如果有)
if timestamps:
latest_source = max(timestamps.items(), key=lambda x: x[1])[0]
if latest_source in values:
best_value = values[latest_source]
best_source = latest_source
# 计算置信度
best_value_count = sum(1 for v in values.values() if v == best_value)
confidence = best_value_count / len(values)
# 确定解决方案
if confidence == 1.0:
solution = "unanimous"
elif confidence >= 0.67:
solution = "majority"
else:
solution = "priority_based"
return best_value, confidence, solution
改进 4:在多样性约束中加入"验证多样性"
def ensure_diversity_with_verification(top3_candidates):
"""
改进的多样性约束:不仅保证来自不同小区,还保证验证质量多样
"""
communities = set()
confidence_levels = set()
diverse_candidates = []
# 按置信度排序
sorted_candidates = sorted(
top3_candidates,
key=lambda x: x.get("overall_confidence", 0.0),
reverse=True
)
for candidate in sorted_candidates:
community = candidate["community"]
confidence_level = "high" if candidate.get("overall_confidence", 0) >= 0.9 else "medium"
# 保证多样性:不同小区 + 不同置信度
if community not in communities and confidence_level not in confidence_levels:
diverse_candidates.append(candidate)
communities.add(community)
confidence_levels.add(confidence_level)
if len(diverse_candidates) >= 3:
break
return diverse_candidates
第七部分:常见问题(基于 Agentic RAG 的改进)
Q1:验证会不会让系统变得太慢?
现象: 加了验证后,延迟从 3.7 秒变成 4 秒
分析:
增加的延迟:300ms(约 8%)
增加的准确性:显著提升(冲突检测、置信度评分)
权衡:
– 如果用户能接受 4 秒延迟,验证是值得的
– 如果用户要求 < 3 秒,可以只验证高优先级字段
解决方案:
# 方案 1:只验证高优先级字段
def selective_verification(candidates, data_sources, plan):
"""只验证优先级 >= 2 的字段"""
for candidate in candidates:
for field, slot_config in plan.items():
if slot_config.get("priority", 0) >= 2:
# 验证这个字段
candidate["verifications"][field] = verify_field(...)
# 方案 2:异步验证
def async_verification(candidates, data_sources):
"""后台异步验证,不阻塞主流程"""
# 立即返回候选房源
# 后台验证并更新置信度
pass
# 方案 3:缓存验证结果
def cached_verification(community, field, data_sources, cache):
"""使用缓存避免重复验证"""
cache_key = f"{community}_{field}"
if cache_key in cache:
return cache[cache_key]
result = verify_field(community, field, data_sources)
cache[cache_key] = result
return result
Q2:如果多个数据源都冲突怎么办?
现象: 3 个数据源给出 3 个不同的答案
分析:
数据源 1:XX小学
数据源 2:YY小学
数据源 3:ZZ小学
置信度 = 1/3 = 33%(非常低)
解决方案:
# 方案 1:标记为"高风险"
if confidence < 0.5:
candidate["risk_level"] = "high"
candidate["recommendation"] = "需要人工核实"
# 方案 2:触发补检索
if confidence < 0.5:
# 从第 4 个数据源检索
additional_source = retrieve_from_additional_source(...)
# 重新计算置信度
# 方案 3:降级处理
if confidence < 0.5:
# 不返回这个字段的信息
# 或者返回"信息不确定"
candidate["verifications"][field]["final_answer"] = "信息不确定"
Q3:怎么知道验证是否有效?
现象: 不知道加了验证后系统是否变好了
解决方案:
# 评估指标
def evaluate_verification_effectiveness(predictions, ground_truth):
"""评估验证的效果"""
# 指标 1:准确率
accuracy = sum(1 for p, g in zip(predictions, ground_truth) if p == g) / len(predictions)
# 指标 2:置信度校准
# 高置信度的预测应该有高准确率
high_confidence_predictions = [p for p in predictions if p.get("confidence", 0) >= 0.9]
high_confidence_accuracy = sum(1 for p in high_confidence_predictions if p == ground_truth[predictions.index(p)]) / len(high_confidence_predictions)
# 指标 3:冲突检测率
# 有多少比例的冲突被正确检测
detected_conflicts = sum(1 for p in predictions if p.get("conflicts"))
return {
"accuracy": accuracy,
"high_confidence_accuracy": high_confidence_accuracy,
"detected_conflicts": detected_conflicts,
"improvement": "相比 Agentic RAG 提升 X%"
}
第八部分:下一步计划
下一篇我会学习 Evaluation-driven RAG,用评估指标和反馈循环,把"系统优化"做得更自动化。
关键点: Verification RAG 把"准确性"补齐了,但系统推荐完就结束了,无法持续改进。
比如系统推荐了 3 套房子,但用户最后买的是第 4 套。系统无法知道为什么推荐失败,也无法学习和改进。
这就是 Evaluation-driven RAG 上场的原因——它不是"更准确的验证",而是"用评估指标驱动系统持续优化"。

