在构建基于大语言模型（LLM）的应用时，你是否遇到过这样的困扰：即便使用了检索增强生成（RAG）技术，模型仍然会一本正经地胡说八道，或者因为检索到了低质量的文档而给出错误的答案？这就是传统的 RAG 架构在面对噪声数据、过时信息或模糊查询时的局限性。为了解决这一痛点，我们将深入探讨一种更先进的架构——纠正性检索增强生成（Corrective Retrieval Augmented Generation，简称 CRAG）。
在这篇文章中，我们将不仅理解 CRAG 的核心概念，还将通过实际的代码示例，一步步构建一个具备自我纠正能力的 RAG 系统。你将学到如何评估检索质量、如何在知识不足时触发网络搜索，以及如何融合多源信息来确保回答的准确性。
什么是 CRAG？
CRAG 通过引入一个自我纠正机制改进了检索增强生成（RAG）。传统的 RAG 就像是一个只会“拿来主义”的助手，它从知识库中抓取文档，无论这些文档是好是坏，直接喂给大模型生成答案。而 CRAG 则更像是一个严谨的研究员，它不仅检索文档，还会停下来“反思”：这些文档真的靠谱吗？它们真的回答了用户的问题吗？
这种机制能够评估并优化检索到的知识，从而减少错误并提高准确性。它可以与多种基于 RAG 的方法相结合。它的核心在于一个反馈循环，持续评估检索到的文档质量，并根据评估结果动态调整生成策略。
传统的 RAG 架构通常是线性的：查询 -> 检索 -> 生成。而 CRAG 在中间插入了一个关键的质量评估与纠正步骤。
为什么我们需要 CRAG
在我们开始写代码之前，让我们先明确一下，为什么我们需要在传统的 RAG 基础上增加这一层复杂性。以下是 CRAG 在改进系统方面至关重要的几个关键原因：
- 解决无关检索问题：传统的向量检索往往基于语义相似度。有时候，检索到的文档虽然词频或语义相似，但核心内容并不相关。CRAG 能够过滤掉那些“看似相关实则无用”的文档。
- 消除噪声与错误：互联网或企业知识库中的数据往往充满噪声。CRAG 能够检测并移除过时或低质量的信息，防止“垃圾进，垃圾出”。
- 抑制幻觉问题：当检索到的上下文不足或不准确时，大模型倾向于编造答案。CRAG 通过验证上下文，强制模型只在有据可依的情况下生成答案。
- 提升系统可靠性：对于医疗、法律等敏感领域，确保信息的准确性和上下文的正确性是至关重要的，容不得半点马虎。
- 智能文档排序：CRAG 不仅仅是接受检索器的排序，它会对文档进行重新排序（Rerank），确保最相关的文档被优先处理。
- 利用动态知识：当内部知识库无法回答问题时，CRAG 能够意识到这一点，并转向外部知识源（如网络搜索），确保信息的及时性。
举个例子：考拉吃什么？
为了让你更直观地理解，让我们来看一个具体的场景。
假设用户问：“考拉吃什么？”
- 传统 RAG 的表现：检索器可能会找到几篇关于澳大利亚动物的文档。其中一篇提到了考拉吃桉树叶（正确），但另一篇得分也很高的文档是关于“有袋类动物习性”的，里面详细描述了袋鼠吃草和熊猫吃竹子。传统 RAG 可能会将这些混合在一起给大模型，导致模型回答：“考拉吃树叶，有时候也吃竹子或草。”——这就是被噪声干扰了。
- CRAG 的表现：CRAG 会在生成答案前先评估。它会发现关于袋鼠和熊猫的文档虽然和“动物”相关，但和“考拉的食物”这个具体查询并不直接匹配（置信度低）。于是，它会过滤掉这些干扰项，只保留关于桉树叶的高分文档。如果所有内部文档质量都很差，它甚至会触发网络搜索去获取最新的确凿知识。最终，它会明确指出：“考拉主要以桉树叶为食。”
CRAG 的工作原理
CRAG 的工作流程比传统 RAG 更加复杂，但也更加健壮。让我们把整个流程拆解开来，一步步看看它是如何运作的。
1. 输入查询
一切始于用户的输入。例如：“考拉吃什么？”。这个查询将作为后续所有步骤的锚点。
2. 初步检索
这是标准 RAG 的第一步。系统根据输入查询，从知识库（向量数据库）中基于相似度选择出前 K 个文档。注意，此时我们并不确定这些文档的质量。
3. 检索评估
这是 CRAG 的核心。我们需要一个“评估器”来给刚才检索到的文档打分。这个评估器通常也是一个经过微调的小型语言模型或特定的评分模型，它会判断每个文档相对于输入查询的相关性和质量。
4. 决策判定
基于上一步的评估结果，系统会做出决策，通常分为三种情况：
- 正确：如果至少有一个文档具有高相关性得分（例如超过某个阈值），系统认为检索质量良好。
- 不正确：如果所有文档的相关性得分都很低，系统认为检索失败，现有的知识库无法回答该问题。
- 模糊：如果得分不高也不低，处于中间地带，意味着对整体质量存在不确定性。
5. 动态纠正路径
根据上述判定，CRAG 会走不同的路径来纠正信息：
- 路径 A（正确） -> 知识精炼：
如果文档被判定为高质量，我们还是不能大意。我们会进行重排，结合相似度、质量和新鲜度对文档重新加权，确保最相关的片段排在最前面。同时进行去重，防止重复内容浪费 Token。
- 路径 B（不正确） -> 网络搜索：
如果文档被判定为不正确（即知识库里没有答案），系统将触发网络搜索（如 Google Search API），从互联网上检索额外的相关信息。这一步极大地扩展了系统的知识边界，使其具有动态性。
- 路径 C（模糊） -> 知识融合：
如果判定为模糊，系统会采取“骑墙”策略。它会同时利用初始检索的内部知识和来自网络搜索的外部知识，试图融合这两部分信息来补全答案。
6. 最终生成
最后，也是用户看到的一步。LLM 仅使用经过纠正、优化或新检索的信息来生成响应。此时的上下文已经是经过“清洗”和“提纯”的，因此生成的答案更加准确且符合事实。
CRAG 的 Python 实现指南
理论讲得差不多了，让我们撸起袖子写点代码。为了让你能够彻底理解并在自己的项目中应用，我们将构建一个最小可行产品（MVP）级别的 CRAG 系统。
注意：为了保持代码的清晰和可运行性，我们将模拟一些组件（如网络搜索和 LLM 调用），但这足以展示 CRAG 的核心逻辑。
步骤 1：准备环境和库
首先，我们需要导入一些基础的 Python 库。我们将使用 INLINECODEe531e9bd 来处理数据，INLINECODEa366d98b 用于计算余弦相似度（作为评估器的基础），以及 re 进行文本清洗。
import math
import re
from collections import Counter
from datetime import datetime
# 模拟一个简单的日期工具
def get_current_time():
return datetime.now().strftime("%Y-%m-%d")
步骤 2：构建向量空间模型与评估器
在真实的 CRAG 中，评估器通常是一个专门的 BERT 模型。为了演示，我们将实现一个基于 TF-IDF（词频-逆文档频率） 和 余弦相似度 的简单评估器。这将帮助我们要计算“查询”与“文档”之间的匹配度。
def cosine_similarity(vec1, vec2):
"""计算两个向量之间的余弦相似度"""
intersection = set(vec1.keys()) & set(vec2.keys())
numerator = sum(vec1[x] * vec2[x] for x in intersection)
sum1 = sum(vec1[x]**2 for x in vec1.keys())
sum2 = sum(vec2[x]**2 for x in vec2.keys())
denominator = math.sqrt(sum1) * math.sqrt(sum2)
if not denominator:
return 0.0
else:
return float(numerator) / denominator
def text_to_vector(text):
"""简单的文本向量化（词袋模型）"""
word_count = Counter(re.findall(r‘\w+‘, text.lower()))
return word_count
def evaluate_retrieval(query, document):
"""
评估检索质量的核心函数。
在真实场景中，这里会调用 T5 或 BERT 模型来打分。
这里我们使用简单的词重叠率作为相关性得分。
"""
query_vec = text_to_vector(query)
doc_vec = text_to_vector(document)
score = cosine_similarity(query_vec, doc_vec)
return score
步骤 3：定义知识库与决策逻辑
现在，让我们创建一个知识库。请注意，这个知识库中既包含完美的答案，也包含噪声（干扰信息）。CRAG 的任务就是区分它们。
# 模拟的知识库 (KB)
KNOWLEDGE_BASE = [
{
"id": "doc1",
"text": "Koalas eat eucalyptus leaves as their primary food source. They are very picky eaters.",
"metadata": {"date": "2023-01-01", "source": "wildlife_journal"}
},
{
"id": "doc2",
"text": "Pandas eat mostly bamboo shoots and leaves. They live in China.",
"metadata": {"date": "2019-05-20", "source": "zoo_digest"}
},
{
"id": "doc3",
"text": "Kangaroos graze on grasses and shrubs. They can jump very high.",
"metadata": {"date": "2021-11-15", "source": "australia_fauna"}
},
{
"id": "doc4",
"text": "Koalas are marsupials. They carry their babies in pouches.",
"metadata": {"date": "2022-08-10", "source": "animal_facts"}
}
]
def decide_action(retrieved_docs, query, threshold_correct=0.2, threshold_ambiguous=0.05):
"""
根据评估得分决定行动策略。
返回: ‘correct‘, ‘ambiguous‘, 或 ‘incorrect‘
"""
max_score = 0
best_doc = None
# 遍历所有检索到的文档，找到得分最高的那个
for doc in retrieved_docs:
score = evaluate_retrieval(query, doc[‘text‘])
print(f"评估文档 {doc[‘id‘]} 得分: {score:.4f}")
if score > max_score:
max_score = score
best_doc = doc
# 决策逻辑
if max_score >= threshold_correct:
return ‘correct‘, best_doc
elif max_score >= threshold_ambiguous:
return ‘ambiguous‘, best_doc
else:
return ‘incorrect‘, None
步骤 4：实现纠正机制（网络搜索与重排）
这是 CRAG 的精髓所在。我们需要根据上一步的决策，执行不同的代码路径。
def web_search_mock(query):
"""
模拟网络搜索功能。
在实际应用中，你会调用 SerpAPI 或 Google Search API。
"""
print(f"
[系统触发] 正在搜索引擎上查找: ‘{query}‘...")
return {
"text": "According to online sources, Koalas strictly eat eucalyptus leaves and occasional mistletoe.",
"source": "web_search"
}
def rerank_docs(docs, query):
"""
如果判定为 ‘Correct‘，我们对文档进行重排和去重。
这里我们按照得分重新排序。
"""
scored_docs = []
for doc in docs:
score = evaluate_retrieval(query, doc[‘text‘])
scored_docs.append((score, doc))
# 按得分降序排列
scored_docs.sort(key=lambda x: x[0], reverse=True)
return [doc for score, doc in scored_docs]
def crag_generation(query, knowledge_base):
print(f"
用户查询: {query}")
print("-" * 30)
# 1. 初步检索 (假设我们检索到了前3个文档)
# 在真实场景中，这里会是向量数据库查询
retrieved_docs = knowledge_base[:3]
print(f"初步检索到 {len(retrieved_docs)} 个文档。")
# 2. 评估与决策
action, best_doc = decide_action(retrieved_docs, query)
final_context = []
# 3. 纠正路径
if action == ‘correct‘:
print("
决策: 文档质量良好 (Correct)。执行知识精炼与重排...")
refined_docs = rerank_docs(retrieved_docs, query)
final_context.append(f"[Verified Doc] {refined_docs[0][‘text‘]}")
elif action == ‘incorrect‘:
print("
决策: 文档质量低/不相关。触发网络搜索...")
web_result = web_search_mock(query)
final_context.append(f"[Web Search] {web_result[‘text‘]}")
else: # ambiguous
print("
决策: 文档质量模糊。执行知识融合...")
# 结合最好的内部文档和网络信息
web_result = web_search_mock(query)
final_context.append(f"[Internal Doc] {best_doc[‘text‘]}")
final_context.append(f"[Web Supplement] {web_result[‘text‘]}")
# 4. 最终生成 (模拟)
print("
=== 最终生成阶段 ===")
print(f"使用的上下文信息: {‘ | ‘.join(final_context)}")
print(f"LLM 生成答案: 基于上述信息，回答您的问题... ")
return final_context
# 运行示例
if __name__ == "__main__":
user_query = "What does a koala eat?"
crag_generation(user_query, KNOWLEDGE_BASE)
实战中的最佳实践与性能优化
通过上面的代码，我们已经搭建起了一个 CRAG 系统的骨架。但在实际的生产环境中，你还需要考虑以下几点来确保系统的稳健性和高性能：
1. 如何优化评估器
我们在示例中使用了简单的 TF-IDF，但在实际中，这远远不够。建议使用专门训练的评估模型，例如 BGE-Reranker 或 Cohere Rerank API。这些模型能理解更深层的语义关系，能区分“考拉吃树叶”和“袋鼠吃草”之间的细微差别，从而大幅提升决策的准确性。
2. 处理网络搜索的延迟
触发网络搜索是 CRAG 中最耗时的部分。为了保证用户体验，你可以：
- 异步处理：在评估检索质量的同时，后台预先启动网络搜索线程，以此减少延迟。
- 缓存机制：对于常见的“不正确”查询，直接从 Redis 缓存中返回之前搜索过的结果，避免重复调用昂贵的搜索 API。
3. 知识融合的技巧
当状态为“模糊”时，如何融合内部知识和网络搜索结果是一门艺术。简单的拼接可能会导致逻辑冲突。建议使用 LLM 再次进行一次“综合”，提示词可以是：“基于以下内部文档和网络搜索结果，提炼出最一致的回答。”
总结
CRAG 不仅仅是一个 RAG 的变体，它代表了一种从“被动检索”到“主动验证”的思维转变。通过引入自我纠正、动态决策和知识融合，我们构建了一个能够自我进化的系统。它能意识到自己的无知，并通过网络搜索来弥补；它也能警惕信息的过载，通过重排来去伪存真。
现在，你已经拥有了构建 CRAG 系统的核心知识和代码框架。你可以尝试将其集成到你的 LangChain 或 LlamaIndex 项目中，看看它是否能帮你解决那些令人头疼的幻觉问题。祝你在探索生成式 AI 的道路上玩得开心！