各位来宾，各位技术同仁，大家好！
今天，我们齐聚一堂，探讨一个在人工智能，特别是大型语言模型（LLM）应用领域中日益突出且至关重要的议题——幻觉（Hallucination）。我们都知道，LLM 的强大之处在于其惊人的文本生成能力，但随之而来的挑战是，它们有时会“编造”事实，提供听起来合理但实际上错误的答案。这对于追求准确性、可靠性和可信度的企业级应用而言，是不可接受的。
为了应对这一挑战，我们引入了检索增强生成（Retrieval Augmented Generation, RAG）架构。RAG 的核心思想是通过将 LLM 的生成能力与外部知识库的检索能力相结合，旨在提供更准确、更可溯源的答案。然而，即使是 RAG，也并非万无一失。它仍然可能产生幻觉，尽管形式可能有所不同——例如，误读检索到的信息、过度概括，或者在多个来源之间做出错误的连接。
因此，今天我将为大家深入解析一个概念性的，但极具实践指导意义的框架：“幻觉过滤器电路”（The Hallucination Filter Circuit）。我们将利用多步验证节点，对 RAG 召回并生成的所有事实进行严谨的“来源一致性检查”。这不仅仅是一个比喻，更是一种系统性地提升 RAG 系统可靠性的工程方法。
RAG 的光芒与阴影——幻觉的挑战
首先，让我们快速回顾一下 RAG 的基本工作原理，并剖析幻觉问题为何如此顽固。
RAG 的基本流程
RAG 架构通常遵循以下步骤：
- 用户查询（User Query）：用户提出问题。
- 检索（Retrieval）：系统根据用户查询，从一个大型的、外部的知识库（如文档库、数据库、网页）中检索出最相关的文档片段（chunks）。这通常通过向量相似性搜索完成。
- 上下文组装（Context Assembly）：将检索到的文档片段作为上下文，与用户查询一起输入给大型语言模型。
- 生成（Generation）：LLM 基于提供的上下文和用户查询，生成一个连贯且信息丰富的答案。
这种方法显著提高了 LLM 答案的准确性和可溯源性，因为它能够利用最新、最具体的外部信息，而不是仅仅依赖模型训练时的数据。
幻觉的本质与RAG中的表现
尽管 RAG 带来了巨大进步，但幻觉依然存在。在 RAG 场景下，幻觉通常表现为：
- 误读或误解上下文：LLM 可能错误地解释了检索到的文档内容，导致生成了与原文不符的答案。
- 过度推理或联想：LLM 在给定上下文的基础上进行了超出事实范围的推断，将可能性当成事实。
- 上下文冲突：检索到的多个文档片段之间存在细微差异或矛盾，LLM 未能妥善处理，生成了不一致的答案。
- 信息填充：当检索到的信息不足以回答问题时，LLM 可能会“填补空白”，创造出看似合理但无事实依据的内容。
- 忽略关键信息：LLM 在生成答案时，可能忽略了上下文中的某些关键细节，导致答案不完整或误导。
这些幻觉，无论是轻微的偏差还是完全的虚构，都会严重损害 RAG 系统的可信度。在医疗、金融、法律等对准确性要求极高的领域，甚至是灾难性的。
我们的目标，正是要构建一个能够系统性地识别、验证并纠正这些幻觉的机制。
RAG 的基石：召回与生成机制回顾
为了更好地理解幻觉过滤器电路，我们首先需要一个 RAG 的基本实现作为起点。我们这里会用 Python 模拟一个简化的 RAG 流程，不依赖复杂的第三方库（如 LangChain 或 LlamaIndex 的完整链），而是展示其核心逻辑。
假设我们有一个简单的向量数据库，存储了一些关于“人工智能历史”的文档片段。
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity
from typing import List, Dict, Any
# 模拟一个向量存储
class MockVectorStore:
def __init__(self, documents: List[Dict[str, str]]):
self.documents = documents
self.embeddings = self._generate_mock_embeddings(documents)
def _generate_mock_embeddings(self, docs: List[Dict[str, str]]) -> np.ndarray:
# 实际应用中会使用预训练的嵌入模型，如 OpenAI text-embedding-ada-002 或 Sentence-BERT
# 这里为了演示，我们随机生成向量
print("Generating mock embeddings...")
return np.random.rand(len(docs), 128) # 假设嵌入维度为128
def search(self, query_embedding: np.ndarray, k: int = 3) -> List[Dict[str, Any]]:
# 计算查询嵌入与所有文档嵌入的余弦相似度
similarities = cosine_similarity(query_embedding.reshape(1, -1), self.embeddings)[0]
# 获取最相似的k个文档索引
top_k_indices = np.argsort(similarities)[::-1][:k]
results = []
for idx in top_k_indices:
doc = self.documents[idx]
results.append({
"content": doc["content"],
"source_id": doc.get("source_id", f"doc_{idx}"),
"similarity": similarities[idx]
})
return results
# 模拟一个LLM接口
class MockLLM:
def invoke(self, prompt: str) -> str:
# 实际应用中会调用 OpenAI, Anthropic 等 API
# 这里我们模拟一个简单的响应逻辑，可能包含一些预设的“幻觉”
print(f"n--- LLM Input Prompt ---n{prompt}n--- End LLM Input ---")
if "capital of France" in prompt:
return "The capital of France is Paris, a city known for its beautiful architecture and the famous Eiffel Tower. It was founded in the 10th century by Julius Caesar." # Julius Caesar founded Paris is a hallucination
elif "first computer" in prompt:
return "The first computer was invented by Charles Babbage in the 19th century. His Analytical Engine is considered a precursor to modern computers. However, the first *electronic* computer was the ENIAC, developed in the 1940s."
elif "Python programming language" in prompt in prompt:
return "Python was created by Guido van Rossum and first released in 1991. It's a popular language for web development and artificial intelligence."
else:
return "I'm sorry, I cannot provide a definitive answer based on the information I have. Here's a general statement: The requested information is related to some historical event or concept."
# 模拟一个嵌入模型
class MockEmbeddingModel:
def embed_query(self, text: str) -> np.ndarray:
# 实际中会调用嵌入模型API
return np.random.rand(128) # 返回一个随机向量
def embed_documents(self, texts: List[str]) -> List[np.ndarray]:
return [np.random.rand(128) for _ in texts]
# 准备一些模拟数据
documents_data = [
{"content": "Paris is the capital and most populous city of France, with an area of 105 square kilometers. It is a major European city and a global center for art, fashion, gastronomy and culture.", "source_id": "wiki_paris"},
{"content": "The Eiffel Tower is a wrought-iron lattice tower on the Champ de Mars in Paris, France. It was constructed from 1887–1889 as the entrance to the 1889 World's Fair.", "source_id": "wiki_eiffel"},
{"content": "The city of Paris has a history dating back to the 3rd century BC with the Parisii, a Celtic tribe. The Romans conquered the Paris Basin in 52 BC and established a Gallo-Roman town called Lutetia.", "source_id": "wiki_paris_history"},
{"content": "Charles Babbage, an English polymath, originated the concept of a programmable computer. Considered the 'father of the computer', Babbage's designs for the Difference Engine and Analytical Engine were pioneering.", "source_id": "wiki_babbage"},
{"content": "ENIAC (Electronic Numerical Integrator and Computer) was the first electronic general-purpose digital computer. It was a Turing-complete, digital computer capable of being reprogrammed to solve a full range of computing problems. ENIAC was designed and built by John Mauchly and J. Presper Eckert of the University of Pennsylvania between 1943 and 1945.", "source_id": "wiki_eniac"},
{"content": "Guido van Rossum began working on Python in the late 1980s as a successor to the ABC programming language and first released it in 1991. Python is an interpreted, high-level, general-purpose programming language.", "source_id": "wiki_python"},
{"content": "Artificial intelligence (AI) is intelligence demonstrated by machines, in contrast to the natural intelligence displayed by humans and other animals. Leading AI applications include natural language processing, expert systems, and machine vision.", "source_id": "wiki_ai"}
]
# 初始化模拟组件
embedding_model = MockEmbeddingModel()
vector_store = MockVectorStore(documents_data)
# 模拟的 RAG Chain
def simple_rag_chain(query: str, k: int = 3) -> Dict[str, Any]:
query_embedding = embedding_model.embed_query(query)
retrieved_docs = vector_store.search(query_embedding, k=k)
context = "n".join([doc["content"] for doc in retrieved_docs])
sources = [{"content": doc["content"], "source_id": doc["source_id"]} for doc in retrieved_docs]
prompt = f"""Based on the following context, answer the question accurately and concisely.
Context:
{context}
Question: {query}
Answer:"""
llm = MockLLM()
response = llm.invoke(prompt)
return {
"question": query,
"answer": response,
"sources": sources
}
# 运行一个示例 RAG 查询
print("n--- Running Simple RAG Example ---")
rag_output = simple_rag_chain("What is the capital of France?")
print(f"nQuestion: {rag_output['question']}")
print(f"Answer: {rag_output['answer']}")
print("Sources retrieved:")
for source in rag_output['sources']:
print(f" - [{source['source_id']}] {source['content'][:50]}...")
# 另一个示例，展示幻觉
print("n--- Running Another RAG Example (Potential Hallucination) ---")
rag_output_babbage = simple_rag_chain("Who invented the first computer?")
print(f"nQuestion: {rag_output_babbage['question']}")
print(f"Answer: {rag_output_babbage['answer']}")
print("Sources retrieved:")
for source in rag_output_babbage['sources']:
print(f" - [{source['source_id']}] {source['content'][:50]}...")代码解析：
MockVectorStore：模拟了向量数据库的功能，可以存储文档及其嵌入向量，并根据查询向量进行相似性搜索。这里为了简化，嵌入向量是随机生成的。MockLLM：模拟了大型语言模型的invoke方法。它的响应是硬编码的，并且故意在“The capital of France is Paris…”的例子中插入了一个幻觉：“It was founded in the 10th century by Julius Caesar.”（实际是罗马人于公元前52年建立了卢泰西亚，且与凯撒并非直接关系）。MockEmbeddingModel：模拟了将文本转换为向量的嵌入模型。simple_rag_chain：这是核心 RAG 逻辑，它获取查询，通过嵌入模型转换为向量，在向量存储中检索相关文档，然后将这些文档作为上下文传递给 LLM 以生成答案。它还返回了检索到的sources。
这个基础 RAG 演示了问题所在：LLM 即使有上下文，也可能因为其生成特性而添加不准确的信息。我们的目标就是在这个基础上，构建一个能够捕获并处理这些不准确信息的“过滤器电路”。
幻觉的根源：源头一致性缺失
幻觉之所以产生，根本原因在于 RAG 流程中的源头一致性缺失。LLM 在接收到检索到的上下文后，会进行复杂的推理和生成。在这个过程中，它可能：
- “相信”自己而非上下文：LLM 强大的内部知识有时会凌驾于提供的上下文之上，尤其当上下文信息模糊或不完整时。
- 过度概括或简化：将复杂的上下文信息概括为简单的陈述，但在这个过程中丢失了关键的准确性。
- 缝合矛盾信息：当检索到的多个片段包含略微冲突的信息时，LLM 可能会尝试“融合”它们，结果却创造出新的、不准确的事实。
- 遗漏关键限定词：例如，把“可能是”变成“是”，把“部分”变成“所有”。
为了解决这些问题，我们必须确保 RAG 最终生成的答案中的每一个原子事实，都能被独立地、明确地追溯到其原始来源，并且与来源内容保持高度一致。这就是“幻觉过滤器电路”的核心理念。
揭秘“幻觉过滤器电路”：架构与核心组件
现在，让我们深入探讨“幻觉过滤器电路”的架构。我们可以将其想象成一个数据处理流水线，其中每个“节点”都负责执行特定的验证任务。数据（RAG 的初步答案及其来源）流经这些节点，在每个阶段都被审查、核对，并最终打上置信度标签。
这个电路主要由四个核心组件构成：
- 初始召回与上下文组装节点 (Initial Recall & Context Assembly Node)
- 多步验证节点 (Multi-Step Verification Nodes) – 这是电路的心脏，包含多个子节点。
- 聚合与决策节点 (Aggregation & Decision Node)
- 反馈与优化回路 (Feedback & Refinement Loop)
我们将详细讲解每个组件的功能和实现。
核心组件一：初始召回与上下文组装节点
这个节点的功能与我们前面展示的 simple_rag_chain 类似，但它会更明确地将检索到的原始文档片段连同最终答案一起输出。这是后续验证步骤的基础。
# 初始召回与上下文组装节点 (Initial Recall & Context Assembly Node)
# 沿用之前的 simple_rag_chain 函数，但我们更明确地将其视为这个节点。
# 它返回的字典中包含了 'answer' 和 'sources'，为后续节点提供输入。
# 示例调用
initial_rag_output = simple_rag_chain("What is the capital of France?")
print("n--- Initial Recall & Context Assembly Node Output ---")
print(f"RAG Answer: {initial_rag_output['answer']}")
print("Associated Sources:")
for src in initial_rag_output['sources']:
print(f" - [{src['source_id']}] {src['content'][:100]}...")功能：
- 根据用户查询从知识库中检索最相关的文档片段。
- 将这些片段组织成上下文。
- 将上下文和查询提供给 LLM，生成初步答案。
- 关键输出： LLM 生成的初步答案 (
answer) 和用于生成该答案的原始文档片段列表 (sources)。
核心组件二：多步验证节点
这是“幻觉过滤器电路”的核心所在，它不满足于 LLM 给出的一次性答案，而是对其进行细致的拆解和针对性的验证。
子节点 2.1：事实提取节点 (Fact Extraction Node)
这个节点的目标是将 RAG 生成的连贯答案分解成一系列独立的、原子化的事实陈述。验证的粒度越细，我们发现幻觉的可能性就越大。
实现策略： 使用另一个 LLM 调用，通过精心设计的提示词，指示 LLM 从给定文本中提取出事实列表。
# 子节点 2.1：事实提取节点 (Fact Extraction Node)
class FactExtractor:
def __init__(self, llm: MockLLM):
self.llm = llm
def extract_facts(self, text: str) -> List[str]:
prompt = f"""
Extract all distinct, atomic factual statements from the following text.
Each statement should be a self-contained fact. List them one per line.
Example:
Text: "The capital of France is Paris, and it's known for the Eiffel Tower. It was founded in the 10th century by Julius Caesar."
Facts:
- The capital of France is Paris.
- Paris is known for the Eiffel Tower.
- Paris was founded in the 10th century by Julius Caesar.
Text: "{text}"
Facts:
"""
response = self.llm.invoke(prompt)
# 简单的后处理，将 LLM 生成的列表转换为 Python 列表
facts = [line.strip('- ').strip() for line in response.split('n') if line.strip()]
return facts
# 实例化事实提取器
fact_extractor_llm = MockLLM() # 可以是与主 RAG LLM 不同的实例或配置
fact_extractor = FactExtractor(fact_extractor_llm)
# 针对 RAG 答案进行事实提取
print("n--- Fact Extraction Node Output ---")
rag_answer_to_verify = initial_rag_output['answer']
extracted_facts = fact_extractor.extract_facts(rag_answer_to_verify)
print(f"Original RAG Answer: {rag_answer_to_verify}")
print("Extracted Facts:")
for i, fact in enumerate(extracted_facts):
print(f" {i+1}. {fact}")代码解析：
FactExtractor类封装了事实提取逻辑。extract_facts方法构建了一个带有“few-shot”示例的提示词，指导 LLM 如何将一个段落分解成原子事实。- LLM 的响应被简单解析（按行分割，移除“-”和空格）以获得事实列表。
子节点 2.2：源头查询节点 (Source Query Node – Per Fact)
对于每一个从 RAG 答案中提取出的原子事实，我们需要再次独立地查询原始文档集合。这次查询不是为了生成答案，而是为了寻找直接支持或反驳该特定事实的证据。这与初始 RAG 召回是不同的，因为初始召回是基于整个用户查询，而这里是基于一个微小的、具体的原子事实。
实现策略： 针对每个事实，使用嵌入模型生成其向量，然后再次在向量存储中进行搜索，查找最相关的文档片段。
# 子节点 2.2：源头查询节点 (Source Query Node - Per Fact)
class FactSourceQuerier:
def __init__(self, embedding_model: MockEmbeddingModel, vector_store: MockVectorStore):
self.embedding_model = embedding_model
self.vector_store = vector_store
def query_for_fact(self, fact: str, k: int = 2) -> List[Dict[str, Any]]:
fact_embedding = self.embedding_model.embed_query(fact)
# 搜索与该事实最相关的文档片段
relevant_docs = self.vector_store.search(fact_embedding, k=k)
return relevant_docs
# 实例化源头查询器
fact_querier = FactSourceQuerier(embedding_model, vector_store)
print("n--- Source Query Node Output (Per Fact) ---")
fact_evidences = {}
for i, fact in enumerate(extracted_facts):
print(f"nVerifying Fact {i+1}: '{fact}'")
evidences = fact_querier.query_for_fact(fact)
fact_evidences[fact] = evidences
if evidences:
print(" Found potential evidences:")
for ev in evidences:
print(f" - [{ev['source_id']}] Similarity: {ev['similarity']:.2f} Content: {ev['content'][:70]}...")
else:
print(" No direct evidence found.")代码解析：
FactSourceQuerier类封装了针对单个事实进行源头查询的逻辑。query_for_fact方法将单个事实转换为嵌入向量，然后在vector_store中搜索最相似的文档。这里的k值可能需要根据实际情况调整，以平衡召回率和效率。
子节点 2.3：一致性检查节点 (Consistency Check Node – Per Fact)
这个节点是判断幻觉的关键。它比较 RAG 答案中的原子事实与“源头查询节点”找到的证据。判断两者是否一致。
挑战： 一致性不仅仅是简单的字符串匹配。我们需要处理语义上的等价性、同义词、语序变化，甚至是信息量的差异。
实现策略： 再次利用 LLM 的强大理解能力，让它来判断一个事实与一个或多个证据之间的语义一致性。
一致性检查的类型与策略
| 检查类型 | 描述 | 适用场景 | 挑战 |
|---|---|---|---|
| 词汇匹配 | 简单字符串包含、子串匹配。 | 事实表述非常直接，不涉及复杂语义。 | 容易漏报（同义词、语序不同）、误报（巧合匹配）。 |
| 语义相似度 | 使用嵌入模型计算事实与证据的语义相似度得分。 | 事实表述多样，但语义核心一致。 | 阈值设定困难，无法判断逻辑上的支持或反驳。 |
| LLM 推理判断 | 利用 LLM 理解能力，判断事实是否能从证据中推断出、是否矛盾。 | 最灵活、准确，能处理复杂语义和逻辑。 | 计算成本高，需要精心设计的提示词，LLM 本身可能出错。 |
| 规则匹配 | 定义特定模式或正则表达式，检查事实中是否存在某些关键词或结构。 | 针对特定业务规则或数据模式。 | 维护成本高，覆盖范围有限。 |
我们主要采用 LLM 推理判断，辅以语义相似度作为辅助。
# 子节点 2.3：一致性检查节点 (Consistency Check Node - Per Fact)
class ConsistencyChecker:
def __init__(self, llm: MockLLM):
self.llm = llm
def check_consistency(self, fact: str, evidences: List[Dict[str, Any]]) -> Dict[str, Any]:
if not evidences:
return {"status": "No Evidence", "explanation": "No relevant sources found to verify this fact."}
evidence_texts = [ev["content"] for ev in evidences]
combined_evidence = "n---n".join(evidence_texts)
prompt = f"""
You are an expert fact-checker. Your task is to determine if the following statement (the 'Fact') is directly supported by or consistent with the provided 'Evidence'.
Fact: "{fact}"
Evidence:
{combined_evidence}
Answer with "YES", "NO", or "PARTIAL". Then, provide a concise explanation for your decision, specifically pointing out any discrepancies or supporting phrases.
Example Output Format:
STATUS: YES
EXPLANATION: The fact 'The capital of France is Paris' is directly supported by the evidence stating 'Paris is the capital and most populous city of France'.
STATUS: NO
EXPLANATION: The fact 'The Eiffel Tower was built in the 17th century' is contradicted by the evidence stating 'It was constructed from 1887–1889'.
STATUS: PARTIAL
EXPLANATION: The fact 'Paris is known for its beautiful architecture and the famous Eiffel Tower' is partially supported. The evidence mentions the Eiffel Tower, but doesn't explicitly state 'beautiful architecture' or confirm it's 'known for' it.
Your analysis:
"""
response = self.llm.invoke(prompt)
status_line = next((line for line in response.split('n') if line.startswith('STATUS:')), "STATUS: UNKNOWN")
explanation_line = next((line for line in response.split('n') if line.startswith('EXPLANATION:')), "EXPLANATION: No explanation provided.")
status = status_line.replace("STATUS:", "").strip()
explanation = explanation_line.replace("EXPLANATION:", "").strip()
return {"status": status, "explanation": explanation}
# 实例化一致性检查器
consistency_checker_llm = MockLLM() # 可以与 Fact Extractor 使用同一个 LLM 实例
consistency_checker = ConsistencyChecker(consistency_checker_llm)
print("n--- Consistency Check Node Output (Per Fact) ---")
fact_consistency_results = {}
for fact, evidences in fact_evidences.items():
print(f"nChecking consistency for fact: '{fact}'")
consistency_result = consistency_checker.check_consistency(fact, evidences)
fact_consistency_results[fact] = consistency_result
print(f" Status: {consistency_result['status']}")
print(f" Explanation: {consistency_result['explanation']}")
代码解析：
ConsistencyChecker类负责执行一致性检查。check_consistency方法构建了一个详细的提示词，要求 LLM 判断一个事实与提供的证据之间的关系（YES, NO, PARTIAL）。- 提示词中包含了明确的示例输出格式，这对于引导 LLM 给出结构化的答案至关重要。
- 对 LLM 的响应进行解析，提取状态和解释。
子节点 2.4：置信度评分节点 (Confidence Scoring Node)
根据一致性检查的结果，为每个事实分配一个量化的置信度分数。这使得我们能够对 RAG 答案的整体可靠性进行评估。
考虑因素：
- 一致性状态： "YES" 应该获得高分，"NO" 获得低分，"PARTIAL" 获得中等分数。
- 证据数量/强度： 多个强有力的支持证据可以提升分数。
- 来源可靠性： （高级功能）如果能对不同来源分配信任分数，可以进一步加权。
# 子节点 2.4：置信度评分节点 (Confidence Scoring Node)
class ConfidenceScorer:
def score_fact(self, consistency_result: Dict[str, Any]) -> float:
status = consistency_result["status"].upper()
if status == "YES":
return 1.0 # 完全支持
elif status == "PARTIAL":
return 0.5 # 部分支持或需要进一步澄清
elif status == "NO":
return 0.0 # 明确矛盾
elif status == "NO EVIDENCE":
return 0.2 # 没有证据支持，但也没有明确矛盾，风险较高
else:
return 0.1 # 未知状态，保守处理
# 实例化置信度评分器
confidence_scorer = ConfidenceScorer()
print("n--- Confidence Scoring Node Output (Per Fact) ---")
fact_scores = {}
for fact, consistency_result in fact_consistency_results.items():
score = confidence_scorer.score_fact(consistency_result)
fact_scores[fact] = score
print(f"Fact: '{fact}'")
print(f" Consistency Status: {consistency_result['status']}")
print(f" Confidence Score: {score:.2f}")
代码解析：
ConfidenceScorer类根据consistency_result中的status分配一个浮点分数。- 分数范围通常在 0.0 到 1.0 之间，1.0 表示完全可信，0.0 表示完全不可信。
- “NO EVIDENCE”的情况被赋予一个较低但非零的分数（0.2），表示该事实未经证实，具有风险。
核心组件三：聚合与决策节点 (Aggregation & Decision Node)
这个节点负责汇总所有原子事实的置信度分数，并根据预设的阈值做出最终决策，决定如何处理 RAG 系统的输出。
决策策略：
- 接受 (Accept)：所有事实都高度可信。
- 标记 (Flag)：部分事实可疑，需要人工审查或向用户发出警告。
- 修改 (Modify)：自动移除或修正低置信度事实（需要更高级的生成能力）。
- 拒绝 (Reject)：大部分事实不可信，不提供答案或请求用户重新提问。
# 核心组件三：聚合与决策节点 (Aggregation & Decision Node)
class AggregationDecisionNode:
def __init__(self, high_confidence_threshold: float = 0.8, flag_threshold: float = 0.5):
self.high_confidence_threshold = high_confidence_threshold
self.flag_threshold = flag_threshold
def make_decision(self, fact_scores: Dict[str, float], original_rag_answer: str) -> Dict[str, Any]:
if not fact_scores:
return {"decision": "UNKNOWN", "explanation": "No facts extracted for verification."}
scores = list(fact_scores.values())
# 计算平均置信度
overall_average_score = np.mean(scores)
# 找到最低置信度
min_score = np.min(scores)
# 识别低置信度事实
low_confidence_facts = [
fact for fact, score in fact_scores.items() if score < self.high_confidence_threshold
]
decision = "ACCEPTED"
explanation = "All facts appear to be highly consistent with sources."
if min_score < self.flag_threshold:
decision = "REJECTED"
explanation = f"Critically low confidence in some facts (min score {min_score:.2f}). Potential hallucination detected. Specific facts with low confidence: {', '.join(low_confidence_facts)}"
elif overall_average_score < self.high_confidence_threshold or low_confidence_facts:
decision = "FLAGGED_FOR_REVIEW"
explanation = f"Some facts have moderate to low confidence (average {overall_average_score:.2f}, min {min_score:.2f}). Review recommended for facts: {', '.join(low_confidence_facts)}"
return {
"decision": decision,
"overall_average_score": overall_average_score,
"min_score": min_score,
"low_confidence_facts": low_confidence_facts,
"explanation": explanation,
"original_rag_answer": original_rag_answer
}
# 实例化聚合与决策节点
aggregation_node = AggregationDecisionNode()
print("n--- Aggregation & Decision Node Output ---")
final_decision = aggregation_node.make_decision(fact_scores, rag_answer_to_verify)
print(f"Overall Decision: {final_decision['decision']}")
print(f"Explanation: {final_decision['explanation']}")
print(f"Overall Average Confidence: {final_decision['overall_average_score']:.2f}")
print(f"Minimum Fact Confidence: {final_decision['min_score']:.2f}")
if final_decision['low_confidence_facts']:
print(f"Facts requiring attention: {final_decision['low_confidence_facts']}")
代码解析：
AggregationDecisionNode类根据传入的事实分数列表做出最终决策。- 它计算了平均分数和最低分数，并根据预设的阈值（
high_confidence_threshold和flag_threshold）判断最终状态。 - 返回的字典包含了决策、解释以及相关的统计数据。
核心组件四：反馈与优化回路 (Feedback & Refinement Loop)
这是一个可选但极其重要的组件。被标记为“不一致”或“低置信度”的案例，不应仅仅被丢弃。它们是宝贵的学习机会。
实现方式：
- 人工标注：将低置信度案例提交给人工专家进行审查和纠正。这些人工反馈数据可以用来：
- 改进嵌入模型的训练。
- 优化 RAG 的检索策略（例如，调整
k值、更改 chunking 策略）。 - 微调 LLM 的提示词，使其在事实提取和一致性检查方面更准确。
- 甚至可以用于微调生成 LLM 本身，使其减少特定类型的幻觉。
- 强化学习：在更复杂的系统中，可以使用强化学习来优化整个链条的参数，以最大化最终答案的准确性和召回率，同时最小化幻觉。
这个回路通常不会直接体现在一次 RAG 调用的代码中，而是作为一个外部系统或流程来管理。
实践中的考量与进阶技术
构建“幻觉过滤器电路”并非没有挑战。我们需要考虑以下几个方面：
-
语义与词汇一致性：
- 如前所述，简单词汇匹配不足。LLM 驱动的语义一致性检查是关键。
- 可以结合使用句向量相似度（如
sentence-transformers）作为 LLM 判断的辅助信号或预过滤步骤。如果语义相似度过低，甚至可以不调用 LLM 进行细致检查，直接标记为“不一致”。
-
多源冲突解决：
- 当针对一个事实检索到多个来源，且这些来源彼此矛盾时，如何处理？
- 策略：
- 源头权重： 为不同来源分配信任分数（例如，官方文档 > 维基百科 > 论坛帖子）。高权重来源的证据优先。
- 多数投票： 如果多个来源支持一个版本，而少数来源支持另一个，则倾向于多数。
- 请求澄清： 如果冲突严重且无法自动解决，可以将该事实标记为“需要澄清”，并可能向用户提供所有冲突信息，让他们自行判断。
- LLM 仲裁： 再次使用 LLM，提供所有冲突的证据，要求它判断哪个更可信，或指出无法判断。
-
源头信任度：
- 不仅仅是内容的相似度，来源本身的可靠性也是一个重要因素。
- 可以维护一个
source_trust_score数据库，在置信度评分节点中将此分数纳入考量。例如，来自gov.cn的文档比来自个人博客的文档具有更高的信任分数。
-
计算开销与性能优化：
- 多步验证意味着更多的 LLM 调用和向量搜索，这会显著增加延迟和成本。
- 优化策略：
- 并行处理： 事实提取后的每个事实的源头查询和一致性检查可以并行执行。
- 缓存机制： 缓存常见事实的验证结果。
- 阈值剪枝： 如果 RAG 初始答案非常短或非常简单，或者如果某个事实在检索阶段就已经匹配度极高，可以跳过部分验证步骤。
- 模型选择： 对不同验证任务选择不同大小和速度的 LLM。例如，事实提取可以使用更小、更快的模型。
-
验证提示工程：
- LLM 在事实提取和一致性检查中的表现高度依赖于提示词的质量。
- 技巧：
- Few-shot Examples： 提供清晰、多样的示例，指导 LLM 的行为。
- 角色扮演： 让 LLM 扮演“专家事实核查员”或“公正的仲裁者”。
- 链式思考 (Chain-of-Thought)： 要求 LLM 在给出最终判断前，先解释其推理过程，这有助于提高透明度和准确性。
- 输出格式约束： 使用 JSON 或其他结构化格式来约束 LLM 的输出，便于程序解析。
“幻觉过滤器电路”的系统级实现
现在，我们将所有组件整合到一个统一的 HallucinationFilterCircuit 类中，展示如何在一个实际系统中编排这些步骤。我们将使用一个更接近 LangChain Runnable 或 LlamaIndex Agent 概念的结构。
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity
from typing import List, Dict, Any, Optional
# 沿用之前的 Mock 组件定义
# MockVectorStore, MockLLM, MockEmbeddingModel, documents_data, embedding_model, vector_store
class HallucinationFilterCircuit:
def __init__(self,
llm_main: MockLLM,
llm_fact_extraction: MockLLM,
llm_consistency_check: MockLLM,
embedding_model: MockEmbeddingModel,
vector_store: MockVectorStore,
high_confidence_threshold: float = 0.8,
flag_threshold: float = 0.5,
k_retrieval_initial: int = 3,
k_retrieval_fact: int = 2):
self.llm_main = llm_main # 用于RAG主生成
self.llm_fact_extraction = llm_fact_extraction # 用于事实提取
self.llm_consistency_check = llm_consistency_check # 用于一致性检查
self.embedding_model = embedding_model
self.vector_store = vector_store
self.fact_extractor = FactExtractor(llm_fact_extraction)
self.fact_querier = FactSourceQuerier(embedding_model, vector_store)
self.consistency_checker = ConsistencyChecker(llm_consistency_check)
self.confidence_scorer = ConfidenceScorer()
self.aggregation_node = AggregationDecisionNode(high_confidence_threshold, flag_threshold)
self.k_retrieval_initial = k_retrieval_initial
self.k_retrieval_fact = k_retrieval_fact
def _initial_rag_recall_and_assemble(self, query: str) -> Dict[str, Any]:
"""
核心组件一：初始召回与上下文组装节点
执行标准的RAG召回和初步生成。
"""
query_embedding = self.embedding_model.embed_query(query)
retrieved_docs = self.vector_store.search(query_embedding, k=self.k_retrieval_initial)
context = "n".join([doc["content"] for doc in retrieved_docs])
sources = [{"content": doc["content"], "source_id": doc["source_id"]} for doc in retrieved_docs]
prompt = f"""Based on the following context, answer the question accurately and concisely.
Context:
{context}
Question: {query}
Answer:"""
response = self.llm_main.invoke(prompt)
return {
"question": query,
"answer": response,
"sources": sources # 包含原始来源，便于调试和追溯
}
def run(self, query: str) -> Dict[str, Any]:
"""
运行整个幻觉过滤器电路。
"""
print(f"n--- Running Hallucination Filter Circuit for Query: '{query}' ---")
# 1. 初始召回与上下文组装
initial_rag_output = self._initial_rag_recall_and_assemble(query)
rag_answer = initial_rag_output['answer']
print(f"n[Step 1: Initial RAG Answer] -> {rag_answer}")
# 2.1 事实提取
extracted_facts = self.fact_extractor.extract_facts(rag_answer)
print(f"n[Step 2.1: Extracted Facts] -> {len(extracted_facts)} facts: {extracted_facts}")
if not extracted_facts:
print("[Circuit Debug] No facts extracted, skipping verification.")
return {
"final_answer": rag_answer,
"verification_status": "UNKNOWN",
"explanation": "Could not extract facts for verification.",
"original_rag_output": initial_rag_output,
"fact_details": []
}
fact_details = []
fact_scores = {}
# 2.2, 2.3, 2.4 多步验证节点 (并行处理每个事实)
print("n[Step 2.2-2.4: Verifying Each Fact]")
for i, fact in enumerate(extracted_facts):
print(f" Verifying Fact {i+1}: '{fact}'...")
# 2.2 源头查询
evidences = self.fact_querier.query_for_fact(fact, k=self.k_retrieval_fact)
# 2.3 一致性检查
consistency_result = self.consistency_checker.check_consistency(fact, evidences)
# 2.4 置信度评分
score = self.confidence_scorer.score_fact(consistency_result)
fact_details.append({
"fact": fact,
"evidences_retrieved": [{"source_id": ev["source_id"], "content": ev["content"][:100] + "..."} for ev in evidences],
"consistency_check": consistency_result,
"confidence_score": score
})
fact_scores[fact] = score
print(f" -> Status: {consistency_result['status']}, Score: {score:.2f}")
# 3. 聚合与决策
final_decision = self.aggregation_node.make_decision(fact_scores, rag_answer)
print(f"n[Step 3: Aggregation & Decision] -> Decision: {final_decision['decision']}, Score: {final_decision['overall_average_score']:.2f}")
print(f" Explanation: {final_decision['explanation']}")
# 4. 反馈与优化回路 (此处仅为概念性表示，实际实现通常是外部系统)
# if final_decision['decision'] == "REJECTED" or final_decision['decision'] == "FLAGGED_FOR_REVIEW":
# self._log_for_feedback_loop(query, initial_rag_output, fact_details, final_decision)
# print("n[Step 4: Feedback Loop] -> Case logged for review/refinement.")
# 返回结果，可能包括原始答案和验证状态
return {
"final_answer": rag_answer, # 这里可以根据decision修改答案，但为了演示，先返回原答案
"verification_status": final_decision['decision'],
"explanation": final_decision['explanation'],
"overall_confidence": final_decision['overall_average_score'],
"original_rag_output": initial_rag_output,
"fact_details": fact_details
}
# 实例化整个电路
circuit = HallucinationFilterCircuit(
llm_main=MockLLM(),
llm_fact_extraction=MockLLM(),
llm_consistency_check=MockLLM(),
embedding_model=embedding_model,
vector_store=vector_store
)
# 运行一个查询，观察幻觉如何被捕获
print("n--- Running Hallucination Filter Circuit for 'What is the capital of France?' ---")
result_france = circuit.run("What is the capital of France?")
print("nFinal Result for France:")
print(f" Answer: {result_france['final_answer']}")
print(f" Verification Status: {result_france['verification_status']}")
print(f" Explanation: {result_france['explanation']}")
print(f" Overall Confidence: {result_france['overall_confidence']:.2f}")
print("n Fact Details:")
for detail in result_france['fact_details']:
print(f" - Fact: '{detail['fact']}'")
print(f" Status: {detail['consistency_check']['status']}, Score: {detail['confidence_score']:.2f}")
print(f" Explanation: {detail['consistency_check']['explanation']}")
if 'Julius Caesar founded Paris' in detail['fact']:
print(" ^^^^^ This fact is the hallucination we're targeting! ^^^^^")
print("n--- Running Hallucination Filter Circuit for 'Who invented the first computer?' ---")
result_computer = circuit.run("Who invented the first computer?")
print("nFinal Result for Computer:")
print(f" Answer: {result_computer['final_answer']}")
print(f" Verification Status: {result_computer['verification_status']}")
print(f" Explanation: {result_computer['explanation']}")
print(f" Overall Confidence: {result_computer['overall_confidence']:.2f}")
print("n Fact Details:")
for detail in result_computer['fact_details']:
print(f" - Fact: '{detail['fact']}'")
print(f" Status: {detail['consistency_check']['status']}, Score: {detail['confidence_score']:.2f}")
print(f" Explanation: {detail['consistency_check']['explanation']}")
代码解析：
HallucinationFilterCircuit类将所有之前定义的组件实例化并作为其成员。__init__方法允许为不同的任务（主生成、事实提取、一致性检查）指定不同的 LLM 实例，这在实际中非常有用，可以根据任务的复杂度和成本选择合适的模型。_initial_rag_recall_and_assemble方法执行 RAG 的初步步骤，返回原始答案和来源。run方法是整个电路的入口点。它按顺序调用各个节点的方法：- 首先是初始 RAG。
- 然后是事实提取。
- 接着，在一个循环中，对每个提取出的事实执行源头查询、一致性检查和置信度评分。这里可以很容易地并行化。
- 最后，聚合所有事实的分数并做出最终决策。
- 返回的结果包含了原始答案、验证状态、解释和每个事实的详细验证信息，这对于调试和透明度至关重要。
案例分析：当 RAG 尝试“编故事”时
让我们再次审视我们关于“法国首都是哪里？”的例子。
无过滤器 RAG 的输出（回顾）:
The capital of France is Paris, a city known for its beautiful architecture and the famous Eiffel Tower. It was founded in the 10th century by Julius Caesar.
应用过滤器电路：
-
初始召回与上下文组装：
- RAG 检索到关于巴黎、埃菲尔铁塔、巴黎历史的文档。
- LLM 基于这些上下文生成上述答案。
-
事实提取节点：
- 从 RAG 答案中提取出以下原子事实：
The capital of France is Paris.Paris is a city known for its beautiful architecture.Paris is known for the famous Eiffel Tower.Paris was founded in the 10th century by Julius Caesar.
- 从 RAG 答案中提取出以下原子事实：
-
源头查询节点 (Per Fact)：
- 针对
The capital of France is Paris.：检索到wiki_paris文档，其中明确提到“Paris is the capital…of France.” - 针对
Paris is known for its beautiful architecture.：可能检索到wiki_paris，但该文档可能没有直接提及“beautiful architecture”，或只是间接提及。 - 针对
Paris is known for the famous Eiffel Tower.：检索到wiki_eiffel，明确提到“The Eiffel Tower is…in Paris…” - 针对
Paris was founded in the 10th century by Julius Caesar.：这是关键！ 再次查询时，系统会检索到wiki_paris_history文档，其中提到“The Romans conquered the Paris Basin in 52 BC and established a Gallo-Roman town called Lutetia.”，但没有提及“10th century”或“Julius Caesar founded”的直接证据。
- 针对
-
一致性检查节点 (Per Fact)：
The capital of France is Paris.：YES (与wiki_paris一致)Paris is known for its beautiful architecture.：PARTIAL (证据可能不直接，或 LLM 认为“beautiful architecture”是主观评价而非严格事实)Paris is known for the famous Eiffel Tower.：YES (与wiki_eiffel一致)Paris was founded in the 10th century by Julius Caesar.：NO (证据明确指出是公元前52年罗马人，而非10世纪的凯撒，存在严重矛盾)
-
置信度评分节点：
- 事实1：1.0
- 事实2：0.5
- 事实3：1.0
- 事实4：0.0 (因为是“NO”)
-
聚合与决策节点：
- 计算所有事实的平均分，发现最低分是 0.0。
- 根据预设阈值（例如
flag_threshold = 0.5），由于存在一个明确矛盾的事实，最终决策将是REJECTED或FLAGGED_FOR_REVIEW。 - 系统会明确指出
Paris was founded in the 10th century by Julius Caesar.是一个低置信度或矛盾的事实。
通过这个电路，我们成功地识别并标记了 RAG 答案中的幻觉。最终的输出可以是一个带有警告的答案，或者直接拒绝该答案并要求用户重试，或者，更高级地，可以尝试在最终输出中移除或修正这些低置信度的事实。
挑战与展望
尽管“幻觉过滤器电路”提供了一个强大的框架，但仍然存在挑战：
- 复杂推理与模糊概念： 对于需要复杂多跳推理或涉及模糊概念（如“什么是创新？”）的问题，事实提取和一致性检查会变得非常困难。
- 实时性要求： 对于需要毫秒级响应的应用，多步验证的额外延迟可能难以接受。需要更高效的模型和并行化策略。
- LLM 本身的局限性： 验证过程中使用的 LLM 自身也可能产生幻觉或判断错误，这要求我们对验证 LLM 的性能进行持续监控和评估。
- 动态知识库： 当底层知识库频繁更新时，如何确保验证过程能够及时反映最新信息，避免“过期”的验证结果。
未来，我们可以期待更智能、更自适应的验证策略：
- 自我修正模型： RAG 系统可能在发现幻觉后，自动向 LLM 发出修正指令，尝试生成一个无幻觉的版本。
- 更智能的验证策略： 结合符号推理、知识图谱与 LLM 共同进行事实核查，提高准确性和可解释性。
- 集成人类反馈： 更紧密地将人工审核整合到反馈循环中，以半监督或全监督的方式持续改进系统。
- 可解释性与透明度： 不仅要指出幻觉，还要能够清晰地解释为什么某个事实被认为是幻觉，这对于用户信任至关重要。
结束语
通过今天对“幻觉过滤器电路”的深入解析，我们看到，利用多步验证节点对 RAG 召回的所有事实进行“来源一致性检查”，是构建一个更加可靠、可信赖的 RAG 系统的关键路径。它将 RAG 的能力从简单的信息聚合与生成，提升到了一个能够自我批判、自我验证的智能系统层面。虽然前方仍有挑战，但这一框架为我们提供了一条清晰的工程化路线，引领我们走向更负责任、更强大的 AI 应用未来。
感谢大家！