10M+ 文档，RAG 幻觉压到 2%，这次又行了？
数据STUDIO约 18 分钟文档越多，RAG 越容易遇到一个不太好看的事实：模型并不会因为语料库变大就自动变得更谨慎。相反，检索漏掉证据、上下文重复、引用指向错误，都会给生成器留下“补一个看起来合理的答案”的空间。
要把 RAG 的幻觉真正压下来，思路其实很朴素：先找证据，把生成限制在证据里，再逐条检查答案，最后把没有支持的结果变成拒答。
下面云朵君就从数据准备开始，把检索、生成、验证、拒答和规模测试一层层拆开。
01先把“近零”说清楚
这套系统解决的不是“如何让模型永远正确”，而是如何让错误尽量只有一种可见的形态：证据不够时拒答，而不是用流畅语言补齐事实。
整套设计可以压成四层：
找对证据：dense embedding、BM25、上下文感知 chunk 和 reranker 一起提高召回与精度。 限制生成：模型只能使用给定上下文，每个句子都要带 passage id；上下文没有答案时输出 INSUFFICIENT_EVIDENCE。验证原子声明：把答案拆成可以单独核对的 claims，逐条和引用的文本比较。 必要时拒答：只要检索置信度或某条 claim 的支持度不够，就不把答案交给用户。
这里有两个目标同时存在。第一是信任：在它选择回答的问题上尽量少说错。第二是规模：检索骨架要能承载 10M 以上向量，并且保持可接受的查询延迟。前者依赖验证与拒答，后者依赖索引和磁盘布局，两件事不能相互替代。
02先固定运行条件
第一步先把会影响结果的旋钮集中到一份配置里，并固定随机种子。这样做不只是为了让代码“能再次运行”，也是为了让后面的 risk-coverage、召回率和延迟数字有可比口径。
import json, os, random
import numpy as np
def set_determinism(seed: int) -> None:
random.seed(seed)
np.random.seed(seed)
os.environ["PYTHONHASHSEED"] = str(seed)
try:
import torch
torch.manual_seed(seed)
torch.cuda.manual_seed_all(seed)
except Exception:
pass
set_determinism(42)
运行记录里的完整 profile 是 slice=20000、eval=100+100，运行在一张 NVIDIA H100 PCIe 上，显存约 79.6 GiB。生成器放在单独的 vLLM 服务中，客户端通过 OpenAI-compatible 接口访问本地服务；生成器是 Qwen/Qwen3-32B，嵌入模型和重排模型分别是 Qwen/Qwen3-Embedding-4B 与 Qwen/Qwen3-Reranker-4B。
关键配置如下：
chunk_tokens=256 chunk_overlap=32
retrieve_k=150 rerank_top_n=20
rrf_k=60 max_hops=3
crag_ok=0.7 crag_bad=0.4
tau_claim=0.3 tau_abstain=0.3
temperature=0 seed=42
生成器保持 temperature 0，并关闭 Qwen3 的 thinking trace。这样做主要是为了减少采样漂移，同时把更强的模型能力留给 faithfulness judge。整个设计的前提也是本地模型：文档和问题不离开这台机器。
03先构建一个能评测的语料
语料不是随手找一批文档就开始嵌入。这里没有把所有数据集混在一起，而是给它们分别安排不同角色：
HotpotQA distractor validation：既提供 Wikipedia 段落，也提供带有 sentence-level gold supporting facts 的可回答问题，用来评估检索召回。 SQuAD v2 impossible questions：补充不可回答问题。 手写 false-premise questions：故意提出“艾因斯坦第二次获得诺贝尔物理学奖是哪一年”“牛顿在 1700 年发明了哪种编程语言”这类问题，检查系统会不会顺着错误前提编故事。 HaluBench：最后单独用来评估 verifier，而不是拿它直接构建主语料。
HotpotQA validation 一共读取 7,405 个问题，再把问题上下文中的段落合并成 corpus，同时保留每个 passage 对应的 gold question id。
from datasets import load_dataset
def load_hotpotqa(split: str = "validation"):
return load_dataset(
"hotpotqa/hotpot_qa",
"distractor",
split=split,
cache_dir=DS_CACHE,
)
hotpot = load_hotpotqa()
print(f"[data] hotpotqa(validation) = {len(hotpot)} questions")
运行输出是：
[data] hotpotqa(validation) = 7405 questions
[corpus] passages=20007 qa_items=2073 gold-bearing passages=4072
样本 passage 平均约 89.2 个词，p95 约 236 个词。这里用“Were Scott Derrickson and Ed Wood of the same nationality?”作为贯穿示例：两个 gold passage 分别说明两人都是 American。这个例子很适合跟完整管道，因为它既不是简单关键词查找，也没有必要靠长篇生成才能回答。
04清洗、分块，再补一层上下文
清洗先做两件事：归一化文本，删除近重复。
import re
import unicodedata
def normalize_text(s: str) -> str:
s = unicodedata.normalize("NFKC", s)
s = s.replace("\u00ad", "")
s = re.sub(r"[ \t]+", " ", s)
return s.strip()
NFKC 可以把兼容字符统一成同一形态，连续空格和 tab 也会收成一个空格。对 BM25 来说，这不是表面清洁：不同的 Unicode 形式可能让同一个词被分成不同 token，直接损失召回。
近重复用 MinHash LSH，而不是逐对比较。这里的配置是 threshold=0.9、num_perm=64。输入的 20,007 个 passage 经过一个过短片段的质量过滤和 19 个近重复删除后，保留 19,987 个。
这一步看起来只是清洗，实际上会直接影响后面的检索。索引当然会小一点，更重要的是，相同段落不会反复挤占 top results。三份内容一样的 passage 并不是三份独立证据，反而可能让生成器误以为某个事实得到了多次确认。
接下来不是直接把文本按固定字符数切开。这里改成按句子打包，并使用生成器自己的 tokenizer 计算 token：
class StructureAwareChunker:
def __init__(self, tokenizer, target_tokens=256, overlap=32):
self.tok = tokenizer
self.target = target_tokens
self.overlap = overlap
def chunk(self, passage):
sents = split_sentences(passage.text) or [passage.text]
chunks, cur, cur_tok = [], [], 0
for sent in sents:
sent_tok = self._ntok(sent)
if cur and cur_tok + sent_tok > self.target:
chunks.append(self._make(passage, cur))
cur, cur_tok = (
([cur[-1]], self._ntok(cur[-1]))
if self.overlap else ([], 0)
)
cur.append(sent)
cur_tok += sent_tok
if cur:
chunks.append(self._make(passage, cur))
return chunks
这样做是为了避免把实体所在的句子和解释它的句子切开，也避免静默截断刚好包含答案的那一行。19,987 个 passage 最终生成 21,259 个 chunk，平均 125 tokens，p95 为 236 tokens。
不过，句子边界仍然解决不了另一种问题。比如一个 chunk 只剩下“revenue grew 3 percent that quarter”，读者知道它说了什么，检索器却不知道是谁的 revenue、哪个 quarter。因此还要给每个 chunk 前置一条不超过 25 个词的上下文说明：
CONTEXTUALIZE_PROMPT = (
"Here is a document titled '{title}':\n<document>\n{doc}\n</document>\n\n"
"Here is a chunk from it:\n<chunk>\n{chunk}\n</chunk>\n\n"
"Give a short, single-sentence context (<=25 words) that situates this chunk "
"within the document so it can be retrieved on its own. Answer with the sentence only."
)
上下文生成通过 32 个并发 worker 执行，结果 checkpoint 后再进入索引。原文示例里，原本只有 “Ed Wood is a 1994 American biographical...” 的 chunk，前面多了一句“This chunk introduces the 1994 film Ed Wood...” 。原文不丢掉 chunk 本体，只把这句 situating sentence 放在它前面。
05检索不是一个模型的工作
这里没有让 embedding 模型单独承担全部检索。原因很简单：dense embedding 擅长语义改写，但可能把罕见人名、ID、数字压进相近语义里；BM25 对这些精确 token 更敏感，却可能漏掉完全换了说法的 paraphrase。
因此每个 contextualized chunk 会进入两条索引：
dense vector 存在 LanceDB 的 on-disk 表中； sparse posting 存在 BM25 索引中。
先用 embedding 得到归一化向量，再把 LanceDB 的 cosine distance 转成 0 到 1 的相似度。BM25 则用和文档相同的 stemmer 处理查询。
def rrf_fuse(rankings, k=60):
scores = {}
for ranking in rankings:
for rank, cid in enumerate(ranking):
scores[cid] = scores.get(cid, 0.0) + 1.0 / (k + rank + 1)
return sorted(scores.items(), key=lambda item: -item[1])
RRF 不比较 BM25 分数和 cosine 分数，因为两种分数不在同一个尺度上。它只看排名：同一个 chunk 同时出现在两路结果的前面，就得到两份靠前排名的贡献。
一个很小的例子就能看出它的效果：
>>> rrf_fuse([["a", "b", "c"], ["b", "c", "a"]])
[('b', 0.03252), ('a', 0.03227), ('c', 0.03200)]
b 没有在任何一路排第一，却因为两路都靠前而胜出。混合检索阶段故意取宽一点的 150 个候选，随后再用 cross-encoder reranker 收窄到 20 个：
class RerankerStage:
def rerank(self, query, candidates, top_n):
scores = self.reranker.score(
query, [candidate.text for candidate in candidates]
)
ranked = sorted(
zip(candidates, scores),
key=lambda item: -item[1],
)[:top_n]
return [
self._mark(candidate, float(score), "reranked")
for candidate, score in ranked
]
cross-encoder 比 bi-encoder 更慢，但它仍按固定宽度处理 150 个候选。变化的是底层语料或索引规模：它可以从约 20,000 级别扩展到 10M；候选窗口并不会随语料规模一起变大。在这个贯穿示例中，两个 gold passage 都进入 reranked top-3，分数分别为 0.999 和 0.996，相关性较弱的 “Ed Wood (film)” 排在后面。
完整评估里的 context_recall@k=0.97。这个数字说明，在它的可回答评测集和配置上，后面的验证器通常有证据可以检查；它不等于所有语料都能得到同样的召回率。
06路由、引用生成与第一道拒答
不是所有问题都值得完整跑一遍检索。router 先把问题分成 no_retrieval、single_hop 和 multi_hop：
ROUTER_PROMPT = (
"Classify the question into exactly one label:\n"
"- no_retrieval: greetings/opinions or questions no document corpus could answer\n"
"- single_hop: answerable by finding one fact\n"
"- multi_hop: needs combining facts from multiple documents\n"
"Question: {q}\nReply with only the label."
)
multi_hop 问题会被拆成 2 到 3 个自洽子问题。false-premise detector 也会运行，但这里没有把它设成单独的硬门：一个小的 yes/no 检测器可能误判，最终仍要由证据 grade 和 claim verification 决定是否回答。
生成阶段是第一道真正的内容约束。system prompt 要求模型只使用编号 passage；每个句子末尾都要有引用；上下文没有答案时只能输出 INSUFFICIENT_EVIDENCE。
ABSTAIN_TOKEN = "INSUFFICIENT_EVIDENCE"
GENERATION_SYSTEM_PROMPT = (
"You answer strictly from the numbered context passages. Rules:\n"
"1. Use ONLY facts in the passages, never outside knowledge.\n"
f"2. If the passages do not contain the answer, reply with exactly: {ABSTAIN_TOKEN}\n"
"3. Every sentence MUST end with a citation to the passage id(s) it uses.\n"
"4. Be concise and factual."
)
模型输出之后还要过 citation parser。它的处理顺序有三步：先从文本中收集所有 citation，再把存在于当前候选 chunk 集合中的 id 与无效 id 分开，最后从原文清掉无效 citation，并返回有效 id 列表和清理后的文本。任何模型自己编出来的 [deadbeef0000] 都会被删除。
def parse_citations(text: str, valid_ids: set[str]) -> tuple[list[str], str]:
found = _CITE_RE.findall(text)
valid = [c for c in dict.fromkeys(found) if c in valid_ids]
invalid = [c for c in dict.fromkeys(found) if c not in valid_ids]
cleaned = text
for bad in invalid:
cleaned = cleaned.replace(f"[{bad}]", "")
return valid, cleaned
例如，一条 citation 指向真实 passage，另一条使用模型虚构的 id：
>>> text = "Paris is the capital of France [a1b2c3d4e5f6]. The Louvre opened in 1793 [deadbeef0000]."
>>> parse_citations(text, valid_ids={"a1b2c3d4e5f6"})
(['a1b2c3d4e5f6'], 'Paris is the capital of France [a1b2c3d4e5f6]. The Louvre opened in 1793 .')
但到这里还远远不够。
citation parser 只能保证“这个引用 id 确实存在”，不能保证 passage 真正支持这句话。模型完全可能引用一段真实文本，然后把它读错。所以，有引用只是必要条件，不是充分条件。
07验证门：引用存在不等于说对
验证门把答案拆成 atomic claims，再逐条检查。这里最关键的判断是：答案不能用平均支持度通过，只要有一条声明明显不受支持，整条答案就要降级。
class VerificationGate:
def check(self, cited_answer, chunks):
claims = self.extractor.extract(cited_answer.text)
used = [
chunk for chunk in chunks
if chunk.id in set(cited_answer.cited_ids)
] or chunks
context = "\n\n".join(chunk.text for chunk in used)
verdicts = []
for claim in claims:
score = self.verifier.support(claim, context)
verdicts.append({
"claim": claim,
"score": score,
"supported": score >= self.tau,
})
min_support = min(
(item["score"] for item in verdicts),
default=0.0,
)
passed = bool(verdicts) and all(
item["supported"] for item in verdicts
)
return passed, min_support, verdicts
tau 设为 0.3。贯穿示例的答案被拆成三条 claim：Scott Derrickson 是 American、Ed Wood 是 American、两人国籍相同，三条都得到 1.00，于是通过。
另一个 false-premise 示例里，模型试图回答“Marie Curie 飞到月球的飞船叫什么”。“Marie Curie was a physicist”得到 0.95，但“Marie Curie traveled to the Moon”只有 0.20。单条 claim 失败，整条答案被丢弃，最终输出拒答。
边界案例可以先经过一次 Chain-of-Verification 修订，让模型删除或弱化上下文不支持的句子，再重新运行验证。这里仍然是“重写后必须复查”，而不是相信修订模型自己说它已经修好了。
拒答因此是一个正式状态，而不是异常分支：
@dataclass
class FinalAnswer:
status: str
answer: str
citations: list[str]
min_support: float
reason: str
class AbstentionPolicy:
def decide(self, route, cited, gate):
if route == "no_retrieval":
return self._abstain("routed_no_retrieval", gate)
if cited.abstained:
return self._abstain("model_abstained", gate)
if gate is None or not gate.passed:
return self._abstain("unsupported_claims", gate)
if gate.min_support < self.tau:
return self._abstain("unsupported_claims", gate)
return FinalAnswer(
"answered",
cited.text,
cited.cited_ids,
gate.min_support,
"verified",
)
示例里的拒答文本是：“I do not have enough supporting evidence in the available sources to answer this confidently.” 这句话的价值不在于措辞，而在于系统把它当成可统计、可调参的最终状态。
08用 CRAG 对弱证据重检索
如果前面的组件只是排成一条直线，第一次检索很弱时，系统就只能立刻拒答。
但有些问题不是没有答案，只是第一次 query 没找对。这时更合适的做法，是用 LangGraph 把它们连成一张带条件边的图：
route → retrieve → grade
├─ strong → generate → verify → finalize
├─ weak → finalize
└─ borderline → refine → retrieve
AgentState 里保存 question、route、evidence、grade、draft、gate、final、hops 和每个阶段的 latency。grade 节点先判断当前 passages 是否足够：
def after_grade(state):
grade = state.get("grade", 0.0)
if grade >= CRAG_OK:
return "generate"
if grade < CRAG_BAD or state.get("hops", 0) >= MAX_HOPS:
return "generate" if grade >= CRAG_BAD else "finalize"
return "refine"
这里 crag_ok=0.7，高于它就生成；低于 crag_bad=0.4 就走 finalize，进入拒答路径。中间区域会拆分子问题、拓宽 query，再回到 retrieve；如果已经达到 3 个 hop，分支会停止继续 refine，此时仍处在可生成区间就生成，只有低于拒答阈值才 finalize。
贯穿示例的 grade 是 1.00，所以不需要重检索，最终 status=answered、reason=verified，端到端约 4.6 秒。另一个关于“牛顿在 1700 年发明了哪种编程语言”的问题，grade 只有 0.15，系统在生成之前就结束，返回拒答。
这个 loop 的重点不只是“多试几次”。它把重检索限制在明确的 hop budget 里，避免证据不足时无止境调用模型，也让每次重试都有可追踪的原因。
09评测：安全不是免费得到的
评测集用 200 个问题组成 balanced golden set：100 个 answerable，100 个 unanswerable。这个比例很重要。如果测试集几乎都是可回答问题，系统即使在不可回答问题上胡说，也可能拿到一个好看的总体分数。
运行结果的 2×2 矩阵是：
confusion (rows ans/unans, cols answered/abstained):
[[46 54]
[ 2 98]]
hallucinations (unanswerable answered): 2 / 100 unanswerable
底部一行说明，在 100 个不可回答问题里，系统拒答了 98 个，回答了 2 个。对应的 hallucination rate 是 2%。这不是 zero，而且只对这套测试集、模型和阈值成立。
代价在上面一行：100 个可回答问题只回答了 46 个，coverage 是 0.46。也就是说，这个系统宁愿放弃一部分本来可能答对的问题，也不愿把支持不足的答案交出去。
阈值不是只能凭感觉设置。阈值也不是凭感觉设置，而是通过 risk-coverage sweep，先找到 hallucination 不超过 5% 的 operating point，再从中选择 coverage 更高的阈值。其他指标如下：
faithfulness 0.908
answer_relevancy 0.817
context_recall@k 0.970
answerable_accuracy 0.580
coverage 0.460
这组数字共同说明：安全门不是免费层。提高阈值通常会带来更多拒答，降低错误风险，但也会减少覆盖率。高风险语料和普通知识问答，不应该共享同一组阈值。
还有一个不能跳过的检查：验证器自己是否可靠。verifier 还单独用 HaluBench 的 300 个样本做了评估，AUROC=0.702。这个结果高于随机，但离完美还有明显距离。
所以 verifier 是整个安全门的上限。它不需要完美到能证明每条 claim 绝对正确，但至少要足够经常把受支持声明排在不受支持声明前面，才能把 operating point 往安全方向移动。后续换更强的 verifier，可以不改动检索、生成和图结构。
1010M 向量的规模实验
质量管道和 20,000 级别的 curated slice 跑通之后，还要单独看规模问题。因为标题里写了 10M+，就必须把“能不能放下、查询有多慢”拿出来测，而不是只从小索引推断。
这次使用的是 1024 维随机单位向量，不是真实文档 embedding。每个规模都通过 Arrow 写入 LanceDB，并建立 IVF_PQ approximate nearest-neighbor index：
class ScaleBench:
def run(self, sizes):
rows = []
for n in sizes:
vecs = make_synthetic_vectors(n, dim=1024)
db = lancedb.connect(str(SCRATCH_DIR / f"scale_{n}"))
t0 = time.time()
table = db.create_table(
"v",
data=self._arrow(vecs),
mode="overwrite",
)
if n >= 100_000:
table.create_index(
metric="cosine",
num_partitions=int(min(4096, max(256, n ** 0.5))),
num_sub_vectors=64,
)
build_s = time.time() - t0
rows.append(self._measure(table, vecs, build_s))
return pd.DataFrame(rows)
结果如下：
10M 向量的 p95 是 18.48ms。相对于 100k 的 10.59ms，数据量扩大 100 倍，查询尾延迟不到两倍；磁盘则从 0.39GB 线性增长到 38.8GB。这个结果说明 on-disk ANN 的规模路径成立，但不说明真实语料的检索质量已经被证明。
特别是 recall@10 只有约 0.1。这里要特别注意：随机向量没有真实语义结构，近似索引几乎没有“正确邻居”可找。因此这部分 benchmark 主要测延迟、构建时间和磁盘占用，不能拿来替代真实 embedding 的 recall 评估。
在 10M 实测之后，又用 np.polyfit 线性外推到 100M：
def fit_and_extrapolate(df, target=100_000_000):
n = df["n"].values.astype(float)
out = {"target": target}
for col in ["build_s", "disk_gb", "p95_ms"]:
a, b = np.polyfit(n, df[col].values, 1)
out[col] = round(float(a * target + b), 2)
return out
projection 得到：
build_s = 3075.1
disk_gb = 388.23
p95_ms = 77.58
这里必须把“projection”和“实测”分开。100M 没有实际构建，线性拟合只能说明这是根据三个测量点做出的估算，不能保证真实系统在更大规模仍按同一斜率增长。
11真正慢的地方可能不在向量库
规模实验测的是向量搜索，端到端 agent 还要把 route、embedding、BM25、reranker、grade、generate、verify 和可能发生的 refine 都算进去。
一次完整运行的阶段延迟汇总如下：
在这组运行条件下，retrieve 是平均耗时和 p95 都最高的阶段。这里的 retrieve 不只是向量库查询，还包括 query embedding、dense search、BM25 和对 150 个候选的 reranking；难题还可能重复经过纠错 loop。
所以真正要优化端到端延迟时，第一反应未必应该是“再换一个更快的向量数据库”。更直接的方向可能是减少模型调用、批处理 reranker、缓存 grade，或者减少不必要的重检索。
向量索引在 10M 上可以做到十几毫秒，但一次完整回答仍然可能花几秒。两者根本不是一个量级的问题。
12写在最后
这套结果真正能说明的是一条工程路径，而不是“RAG 已经被解决”：
数据清洗和结构感知分块，先减少检索阶段的噪声； dense 与 BM25 互补，再用 RRF 和 reranker 把候选收窄； 生成器不能只被要求“请引用”，还要检查引用 id 是否真实； 一条答案要拆成 claims，任何一条不受支持都应该让整条答案进入拒答； CRAG loop 可以给边界证据一次有限的重检索机会； 200 题测试里，2% 的错误回答换来了 0.46 的可回答覆盖率； verifier AUROC=0.702，说明验证器有效但仍是主要上限； 10M 向量的 18.48ms p95 说明 on-disk IVF_PQ 可以把索引规模推上去，但随机向量 benchmark 不能代表真实语料质量； 端到端延迟主要花在检索周边的模型调用，而不是单独的向量查找。
所以“近零幻觉”不是某个更大的模型参数带来的特性，而是一份更严格的输出契约：每个到达用户的句子，都要能回到检索文本；回不去的，就停在拒答。
后面还有四个很自然的优化方向：换更强的 verifier、用真实 embedding 重跑规模实验、在单机之外做 sharding 和 quantization，以及按领域重新校准 coverage。它们会改变性能和 operating point，但不会改变这条主线。
参考材料
配套代码 — GitHub: FareedKhan-dev/rag-zero-hallucinations