近零幻觉RAG Pipeline实战：千万级文档构建全攻略
面对千万级文档规模，RAG系统如何逼近零幻觉？本文详解了一套从数据预处理到自我验证的端到端技术架构。
- 混合检索索引构建与高效查询机制
- 带引文生成与逐句验证的反幻觉策略
- 千万级向量扩展与性能基准测试
注入RAG系统的文档越多，它产生虚构内容的倾向就越强——当语料库膨胀至数百万、接近千万甚至更高量级时，幻觉问题会呈指数级恶化。要在如此规模下维持答案的可信度，核心是构建一条让智能体自我校验证据、并为每个论断提供引用的处理流水线，其设计思路与Claude采用引文机制如出一辙。
以下便是这条流水线的完整拆解，按自上而下的顺序依次构建每个功能模块。
- 环境配置与数据获取：下载语料库，评估其规模与真实样本，固定所有随机种子以确保结果可复现。
- 数据清洗与文本切片：执行文本规范化，利用MinHash LSH算法剔除近似重复内容，按结构感知策略切分文本块，并为每块添加上下文前缀。
- 构建混合索引：将每个文本块作为稠密向量存入LanceDB，同时维护一份稀疏的BM25倒排索引，将全部数据落盘以支撑千万级向量查询。
- 检索与重排序：采用倒数排名融合算法合并稠密与稀疏检索结果，随后将150个候选结果重排序至前20个。
- 路由与问题分解：对每个问题进行分类，在检索前将多跳问题拆解为若干子问题。
- 带引文生成：严格基于上下文作答，每个句子附上引文标识，否则输出弃权令牌。
- 逐句校验：将答案拆解为原子化论断，使用忠实度判别模型对照其引用的文本逐一检查每个论断。
- 不确定性下的拒答机制：融合多种信号形成校准后的决策，在证据支撑不足时直接拒答。
- 智能体串联：将所有组件整合为自我纠错的CRAG循环，在证据薄弱时触发重新检索。
- 评估与扩展：在包含200个问题的黄金测试集上评估幻觉率，随后将检索索引扩展到真实的1000万向量规模，并外推至1亿量级。
全部代码均已开源至GitHub仓库（理论+代码）：
https://github.com/FareedKhan-dev/rag-zero-hallucinations
目录
- 近零，而非绝对零
- 项目初始化
- 数据获取
- 语料清洗
- 文本切块与上下文构建
- 检索模型加载
- 混合索引构建
- 倒数排名融合
- 路由与问题分解
- 带引文生成
- 校验关卡
- 适时拒答
- 智能体设计
- 有效性验证
- 黄金测试集构建
- 幻觉的量化呈现
- 安全性的权衡代价
- 判别模型质量评估
- 向量规模扩展至千万级
- 真实的千万级向量索引
- 千万级查询18毫秒，及外推至1亿
- 时间开销分布分析
- 当前范围与后续方向
近零，而非绝对零
核心要解决的问题并非“让模型更聪明”。更大的模型在检索为空时依然会进行猜测，因为猜测本就是文本生成的天性。
因此，与其追求完美模型，不如将常规模型封装进一个仅具备安全失效模式的系统中。当证据缺失时，正确的输出不应是流畅的猜测，而是明确弃权。
该系统提供了四层控制，以下各节将分别对应每一层展开。
- 检索正确证据：采用稠密检索加BM25搜索的混合方案，结合上下文感知的文本块与重排序技术。
- 约束式生成：仅基于检索上下文作答，为每个句子引用段落编号，否则直接拒答。
- 原子化论断校验：使用忠实度判别模型对照引文来源，逐一检查每个论断。
- 拒答决策：当论断支撑度或检索置信度低于校准阈值时，执行拒答操作。
系统同时追求两大目标。其一是可信度，在系统选择作答的问题上实现近乎零的幻觉率。其二是可扩展性，检索后端必须容纳千万级向量，并仍能在毫秒级时间内给出响应。前者依赖校验逻辑，后者依赖索引架构。两者均需构建。
项目初始化
所有逻辑执行之前，先完成项目环境搭建。计划包含：导入依赖库、固定所有随机种子以保证可复现性、检查唯一GPU可用性、将轻量客户端指向文本生成器、冻结配置参数、确保无头运行每次行为一致。
首先导入所需模块，并定义一个为每个随机数生成器设置种子的函数。
import json, os, random, subprocess, time
from dataclasses import dataclass, asdict, field
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)
一开始就固定种子，因为不可复现的RAG评估根本算不上评估，同时整篇文章的核心也在于信任最终的数字。该Notebook采用参数化设计，因此一个单元会解析运行配置文件，并打印当前构建所依据的参数。
#### OUTPUT ####
profile=FULL slice=20000 eval=100+100 artifacts=/mnt/data/artifacts
这里是完整运行模式：2万段落以及100+100个评估问题，而非最初用于低成本排查代码错误的简易模式。单张GPU，显存是硬预算限制，而非运行时意外。通过nvidia-smi读取显卡信息，并断言其符合预期。
def gpu_report() -> dict:
"""返回GPU名称/显存/驱动信息，并确认运行在80GB H100上。"""
name = _smi("name")[0]
total = float(_smi("memory.total")[0]) / 1024.0 # GiB
rep = {"name": name, "total_gb": round(total, 1),
"free_gb": round(float(_smi("memory.free")[0]) / 1024.0, 1),
"driver": _smi("driver_version")[0]}
print(json.dumps(rep, indent=2))
assert "H100" in name and total >= 79 # 单张80GB H100，不得低于此规格
return rep
#### OUTPUT ####
{
"name": "NVIDIA H100 PCIe",
"total_gb": 79.6,
"free_gb": 32.8,
"driver": "570.195.03"
}
所用硬件为80 GB NVIDIA H100，主机配备180 GB RAM和750 GB NVMe硬盘，这在后期索引规模增大时至关重要。32B参数的文本生成器并未运行在本Notebook进程中，而是部署在独立的vLLM服务器上，通过一个小型兼容OpenAI API的客户端与之通信。让生成器在独立进程中保持热状态，可以反复重跑Notebook而无需重新加载模型。
class LocalLLM:
"""面向已预热vLLM OpenAI兼容服务器的轻量客户端。"""
def __init__(self, endpoint: str, model: str, thinking: bool = False):
self.endpoint, self.model, self.thinking = endpoint.rstrip("/"), model, thinking
def chat(self, system: str, user: str, temperature: float = 0.0, max_tokens: int = 512) -> str:
body = {"model": self.model, "temperature": temperature, "max_tokens": max_tokens,
"messages": [{"role": "system", "content": system},
{"role": "user", "content": user}]}
if not self.thinking: # Qwen3: 跳过思考过程以降低延迟
body["chat_template_kwargs"] = {"enable_thinking": False}
r = requests.post(f"{self.endpoint}/chat/completions", json=body, timeout=120)
r.raise_for_status()
return r.json()["choices"][0]["message"]["content"]
llm = LocalLLM("http://localhost:8000/v1", "Qwen/Qwen3-32B")
print(f"[llm] up={llm.is_up()}")
#### OUTPUT ####
[llm] up=True
服务器已启动。最后一个初始化步骤是将所有参数冻结到一个配置对象中并打印，这样驱动文章其余部分的数字便集中在一个地方。
#### OUTPUT ####
{
"gen_model": "Qwen/Qwen3-32B",
"embed_offline": "Qwen/Qwen3-Embedding-4B",
"rerank_model": "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, "seed": 42
}
检索150个候选，重排序至20个，允许智能体最多进行3次纠错跳转，支撑度阈值设为0.3，后续将进一步校准。文本生成器为Qwen3–32B，嵌入模型与重排序模型均为4B参数的Qwen3，忠实度判别模型同样使用32B版本。生成器温度设为0并关闭思考过程，以获得可复现、低延迟的答案，同时避免随机采样导致模型偏离证据。所有模型均为本地开源的Qwen3，因为整个前提在于没有任何文档或查询会离开这台机器，这正使该流水线适用于私有语料库。工具准备就绪，接下来获取数据。
数据获取
流水线的质量取决于底层语料库，因此第一个关键步骤是下载数据集并审视其内容。选择HotpotQA的干扰项设置版本，原因有二：每个问题都附带句子级别的黄金支持事实，这是评估检索召回率最干净的方式；同时它绑定的维基百科段落免费提供了一个真实语料库。在测试的另一端，拉取SQuAD v2中不可回答的问题，并手写少量前提错误的问题，因为衡量幻觉的唯一方式，就是询问语料库无法回答的问题，并检查系统是否保持沉默。
第三个数据集HaluBench将在接近尾声时出现，仅用于验证校验器本身。HotpotQA是用于构建和搜索的语料库。
from datasets import load_dataset
def load_hotpotqa(split: str = "validation"):
# datasets 3.x 需要命名空间的仓库ID
return load_dataset("hotpotqa/hotpot_qa", "distractor", split=split, cache_dir=DS_CACHE)
hotpot = load_hotpotqa()
print(f"[data] hotpotqa(validation) = {len(hotpot)} questions")
#### OUTPUT ####
[data] hotpotqa(validation) = 7405 questions
7,405 个问题，每个问题都捆绑着来源的维基百科段落。定义段落和问题的数据结构，然后编写一个构建器，将每个问题的上下文段落合并成一个语料库，同时记录哪些段落属于黄金证据。
@dataclass
class Passage:
id: str
title: str
text: str
is_gold_for: list[str] = field(default_factory=list) # 该段落作为黄金证据的问题ID列表
@dataclass
class QAItem:
qid: str
question: str
answer: str
answerable: bool
gold_titles: list[str] = field(default_factory=list)
gold_sentences: list[str] = field(default_factory=list)
qtype: str = "" # bridge | comparison | unanswerable | false_premise
class CorpusBuilder:
"""从HotpotQA干扰项上下文中构建段落语料库及问答条目。"""
def build(self, qa, n_passages: int):
passages, qa_items = {}, []
for ex in qa:
gold = list(dict.fromkeys(ex["supporting_facts"]["title"])) # 黄金证据标题
for t, ss in zip(ex["context"]["title"], ex["context"]["sentences"]):
para = " ".join(s.strip() for s in ss).strip()
if len(para) < 40:
continue
p = passages.setdefault(_pid(t, 0), Passage(_pid(t, 0), t, para))
if t in gold:
p.is_gold_for.append(ex["id"])
qa_items.append(QAItem(ex["id"], ex["question"], ex["answer"], True,
gold_titles=gold, qtype=ex.get("type", "")))
if len(passages) >= n_passages:
break
return list(passages.values()), qa_items
corpus, qa_items = CorpusBuilder().build(hotpot, SLICE_SIZE)
print(f"[corpus] passages={len(corpus)} qa_items={len(qa_items)} "
f"gold-bearing passages={sum(1 for p in corpus if p.is_gold_for)}")
#### OUTPUT ####
[corpus] passages=20007 qa_items=2073 gold-bearing passages=4072
当前拥有20,007 个段落和2,073个问题，其中4,072个段落被标记为某些问题的黄金证据。在构建任何组件之前，应当仔细查看数据，包括规模分布及一个真实示例。
import pandas as pd
tok_lens = [len(p.text.split()) for p in corpus]
print(pd.Series(tok_lens, name="passage_word_count").describe().round(1).to_string())
ex = qa_items[0]
print(f"nSample question:n Q: {ex.question}n A: {ex.answer} (type={ex.qtype})")
print(f" gold titles: {ex.gold_titles}")
for s in ex.gold_sentences:
print(f" - {s}")
#### OUTPUT ####
count 20007.0
mean 89.2
std 53.4
min 7.0
25% 54.0
50% 80.0
75% 113.0
max 1378.0
Sample question:
Q: Were Scott Derrickson and Ed Wood of the same nationality?
A: yes (type=comparison)
gold titles: ['Scott Derrickson', 'Ed Wood']
- Scott Derrickson (born July 16, 1966) is an American director, screenwriter and producer.
- Edward Da vis Wood Jr. was an American filmmaker, actor, writer, producer, and director.
段落平均约89 个词，既足够短以容纳至单个提示中，又足够长以承载一个完整事实。该样本为一个比较问题：“Scott Derrickson和Ed Wood是否来自同一国籍？”，其两条黄金句子已包含答案：两人均为美国人。这将是全文跟踪的问题，因为观察一个真实问题穿过整个流水线，能使每个组件的功能变得具体明确。这里已可看出两个层次：可回答问题用于衡量正确证据是否被检索到；而后加入的不可回答问题则用于衡量幻觉率——一个能回答语料库中无支持问题的系统，就是在编造内容。
语料清洗
垃圾输入意味着幻觉输出，因此在构建索引之前，先对文本进行清洗。两个低成本步骤带来的收益远超投入。规范化处理让分词器对每个段落的表现一致，而近似重复内容剔除则阻止复制或转发过的段落占据顶部检索结果，从而在不增加任何新证据的情况下虚高检索召回率。
import re, unicodedata
def normalize_text(s: str) -> str:
s = unicodedata.normalize("NFKC", s) # 规范的Unicode形式
s = s.replace("u00ad", "") # 删除软连字符
s = re.sub(r"[ t]+", " ", s) # 合并连续空格
return s.strip()
首先执行NFKC规范化，因为BM25基于原始字符进行分词，一个连字或一串多余空格会悄悄将一个词拆成两个，或将两个词合成一个，从而损害召回率。在一个凌乱的字符串上，这个函数正好做了所需的事情。
#### OUTPUT ####
>>> normalize_text("the final reporttwas ready")
'the final report was ready'
连字“fi”变为普通“fi”，制表符与连续空格折叠为单个空格，两个仅在不可见字符上不同的段落现在会被分词器视为相同。
去重部分则更为有趣。必须选择MinHash LSH这类近似方法，而非逐对比较，因为精确的逐对比较是二次时间复杂度，在语料库规模下永远无法完成，而带LSH索引的MinHash能以近似线性时间找到近似重复内容。剔除这些重复内容同时服务于两个目标：在迈向千万级向量时使索引更小；防止同一段落的三个副本挤满顶部结果——这是检索器向模型注入冗余上下文、诱使其过度信任单一来源的隐蔽方式。
class Deduper:
"""通过MinHash LSH对词级别shingle进行近似重复检测，剔除重复段落。"""
def __init__(self, threshold: float = 0.9, num_perm: int = 64):
self.threshold, self.num_perm = threshold, num_perm
def fit_transform(self, passages: list[Passage]):
lsh = MinHashLSH(threshold=self.threshold, num_perm=self.num_perm)
kept, dropped = [], 0
for p in passages:
m = self._mh(p.text)
if lsh.query(m): # 已保留近似重复
dropped += 1
continue
lsh.insert(p.id, m)
kept.append(p)
return kept, {"kept": len(kept), "dropped_near_dup": dropped}
#### OUTPUT ####
{
"kept": 19987,
"dropped_near_dup": 19,
"input": 20007,
"after_quality": 20006,
"after_dedup": 19987
}
剔除19个近似重复和1个过短片段后，保留19,987 个段落。该语料库虽为精心挑选的切片，但清洗步骤完全相同，无论输入是2万段落还是2000万段落。
文本切块与上下文构建
现在将段落切分为文本块。固定大小切块是简单选择，却也是错误选择，因为它会将包含实体的句子与用于消歧的上下文割裂，这对多跳问题而言是致命的。因此，将完整句子打包至token预算内，设置少量重叠，并使用生成器自身的分词器计算token数，使预算与模型实际看到的内容保持一致。这是一个隐藏在切块细节中的幻觉问题。如果一个文本块超出预算后被静默截断，包含答案的句子可能会消失，于是问题看似无缘无故地变得不可回答。因此，宁可尊重句子边界，并接受略多一些的切块数量。
class StructureAwareChunker:
def __init__(self, tokenizer, target_tokens: int = 256, overlap: int = 32):
self.tok, self.target, self.overlap = tokenizer, target_tokens, overlap
def chunk(self, passage: Passage) -> list[Chunk]:
sents = split_sentences(passage.text) or [passage.text]
chunks, cur, cur_tok = [], [], 0
for s in sents:
st = self._ntok(s)
if cur and cur_tok + st > self.target:
chunks.append(self._make(passage, cur))
cur, cur_tok = ([cur[-1]], self._ntok(cur[-1])) if self.overlap else ([], 0)
cur.append(s)
cur_tok += st
if cur:
chunks.append(self._make(passage, cur))
return chunks
#### OUTPUT ####
[chunk] 19987 passages -> 21259 chunks (tokens: mean=125 p95=236)
得到21,259 个文本块，平均125 token，舒适地低于256预算。在建立索引之前，还有一个问题需要解决。
像“该季度收入增长3%”这样的文本块本身难以检索，因为无法得知是谁的收入、哪个季度。因此在建立索引前，为每个文本块前面添加一句定位性的句子，这便是上下文检索思路，只不过使用本地Qwen3而非托管模型来生成这句话。
CONTEXTUALIZE_PROMPT = (
"以下是标题为'{title}'的文档：nn{doc}n nn"
"以下是其中的一个片段：nn{chunk}n nn"
"请用一句简短、单句的上下文（<=25词）描述该片段在文档中的位置，"
"使其能够独立被检索到。仅输出该句子。"
)
该方法将按文本块调用的请求分发至线程池，因为这些调用彼此独立，vLLM会在服务端对它们进行批处理，比逐个文本块处理快得多。同时还会对结果设置检查点，重新运行时将跳过整个步骤。
class Contextualizer:
def contextualize(self, chunks, doc_lookup, workers: int = 32):
def _one(c):
user = CONTEXTUALIZE_PROMPT.format(title=c.title,
doc=doc_lookup.get(c.passage_id, c.text)[:4000],
chunk=c.text)
ctx = self.llm.chat("你负责编写简洁的检索上下文。", user, max_tokens=64).strip()
c.contextual_text = (ctx + "n" + c.text) if ctx else c.text # 添加前缀，保留原文
with ThreadPoolExecutor(max_workers=workers) as ex:
list(ex.map(_one, chunks)) # 同时处理32个请求
return chunks
#### OUTPUT ####
处理前：
Ed Wood是一部1994年美国传记时代喜剧片，由蒂姆·伯顿执导和制作，约翰尼·德普饰演邪典电影制作人埃德·伍德...
处理后（添加上下文前缀）：
该片段介绍1994年电影《Ed Wood》，由蒂姆·伯顿执导，概述其主要题材与演员阵容。
Ed Wood是一部1994年美国传记时代喜剧片，由蒂姆·伯顿执导和制作，约翰尼·德普饰演邪典电影制作人埃德·伍德...
这句额外的句子成本很低，每个文本块仅需一次短文本生成，它告诉检索器该文本块的主题，即使文本块本身可能存在歧义。这种提升是召回率最终如此之高的主要原因。召回率是整个幻觉故事的基础，因为下游校验器只能将答案锚定在检索实际找到的证据上，因此在这里获得的每一点召回率，都意味着更多可回答而非拒绝的问题。
检索模型加载
文本块准备好后，加载将它们转化为可搜索证据并在后续检查答案的模型。三个模型与生成器共享这块GPU，因此在每次加载后快照显存使用情况，并保持在预算范围内。这里加载重排序模型和忠实度判别模型，嵌入模型稍后在构建索引时再加载。
不希望三步之后因内存溢出崩溃才发现超预算，因此每次加载都记录来自nvidia-smi的整卡数值，以及来自PyTorch的仅内核数值。
def vram_snapshot(tag: str) -> dict:
"""每次加载步骤后记录GPU整体及仅内核的显存使用情况。"""
kernel = round(torch.cuda.memory_allocated() / 1024**3, 2) # 仅本进程
used = round(float(_smi("memory.used")[0]) / 1024.0, 2) # 整张GPU，包括两个进程
print(f"[vram] {tag:22} gpu_used={used}GB kernel={kernel}GB")
return {"tag": tag, "gpu_used_gb": used, "kernel_gb": kernel}
重排序模型是一个小型因果模型，用作是与否的判别。每个查询与文档对都会被包装进固定模板，分数直接从下一个token的logits中读取，重排序对每个候选仅需一次前向传播。加载专用的交叉编码器重排序模型，而非相信嵌入分数，因为嵌入器会将整个段落压缩为一个向量，这足以快速扫描语料库，但会模糊“仅提及实体的段落”与“真正回答问题的段落”之间的差别，而这种差别正是将错误证据挡在提示和答案之外的关键。
class Qwen3Reranker:
"""通过模型赋予'是'token的概率来对（查询，文档）对进行评分。"""
@torch.no_grad()
def score(self, query: str, docs: list[str], batch_size: int = 16) -> list[float]:
out = []
for i in range(0, len(docs), batch_size):
batch = [self._fmt(query, d) for d in docs[i:i + batch_size]]
enc = self.tok(batch, return_tensors="pt", padding=True,
truncation=True, max_length=1024).to(self.model.device)
logits = self.model(**enc).logits[:, -1, :] # 最后一个token的logits
yn = logits[:, [self.no_id, self.yes_id]] # 比较'否'与'是'
probs = torch.softmax(yn.float(), dim=-1)[:, 1] # 保留P('是')
out.extend(probs.cpu().tolist())
return out
忠实度判别模型使用32B生成器本身，通过提示返回某个论断对某段上下文的单一支撑分数。选择本地32B作为判别模型，是因为RAG中的忠实度检查意味着同时读取一个论断和几段长文本，这正是小型句子对NLI模型容易出错的地方；同时，这个判别模型是将自信的错误答案转化为弃权决策的单一组件。它是近乎零幻觉论断的核心，因此宁愿将最强的模型用在这里。NLI交叉编码器和MiniCheck仍作为轻量替代方案接入，但本次运行使用LLM判别模型。
JUDGE_PROMPT = (
"你是一个严格的事实核查员。判断上下文是否支持该论断。nn"
"上下文：n{context}nn论断：{claim}nn"
"仅输出一个数字：如果上下文明确陈述或蕴含该论断，输出1.0；"
"如果矛盾或未提及，输出0.0；或输出介于两者之间的值。"
)
class JudgeVerifier:
def _score(self, claim: str, context: str) -> float:
out = self.llm.chat("你是一个严格的忠实度评分员。",
JUDGE_PROMPT.format(context=context[:6000], claim=claim), max_tokens=8)
m = re.search(r"[01](?:.d+)?", out)
return min(1.0, float(m.group())) if m else 0.0
#### OUTPUT ####
[vram] reranker gpu_used=54.3GB kernel=7.49GB
[verifier] using the local LLM as faithfulness judge
[vram] whole-GPU used=54.3GB / 80.0GB (need >= 3.0GB headroom)
整个模型栈占用H100 80 GB中的54.3 GB，为接下来的索引工作留下了余量。判别模型无需额外显存，因为它复用已在vLLM服务器中运行的生成器。所有组件均在一台机器上，没有任何内容访问外部API。
混合索引构建
现在开始索引构建。这里的问题是，没有任何单一检索器足够好。稠密嵌入能捕捉同义改写，这正是当查询和答案使用不同词语时所需的能力。BM25则能捕捉名称、ID和编号等精确token，而这些正是稠密模型容易模糊的内容。因此，对两者都建立索引，以文本块ID为键，作用于已添加上下文的文本。
仅在索引时加载嵌入模型，对每个文本块进行嵌入，然后在用更小的在线嵌入模型提供查询服务之前释放它。向量将被归一化，余弦相似度即为普通点积。
def embed_texts(embedder, texts, is_query: bool = False) -> np.ndarray:
kw = {"normalize_embeddings": True, "convert_to_numpy": True, "batch_size": 64}
if is_query:
kw["prompt_name"] = "query"
return embedder.encode(texts, **kw).astype("float32")
加载查询嵌入模型是最后一个推高显存用量的步骤，快照显示最终落点。
#### OUTPUT ####
[vram] embedder(online) gpu_used=61.85GB kernel=15.04GB
峰值约为80 GB中的62 GB，仍在预算内，并在索引构建后立即释放更重的离线嵌入模型，查询时仅小型在线嵌入模型常驻。必须为稠密端选择LanceDB，因为它是嵌入式、基于磁盘、基于NVMe且无需运行服务器的数据库，这意味着同一条代码路径可以承载远超内存大小的索引，正是这个属性使该设计在后续无需修改一行代码就能达到千万级向量。
class LanceVectorStore:
def search(self, qvec: np.ndarray, k: int) -> list[tuple[str, float]]:
res = self.tbl.search(qvec).metric("cosine").limit(k).to_list()
return [(r["id"], 1.0 - r["_distance"] / 2.0) for r in res]
在向量旁边保留一个词法BM25索引，因为稠密嵌入恰好会将稀有名称、ID或编号模糊到邻近内容中，而事实性问题往往就取决于这些token，因此稀疏端是防止自信答案建立在近似命中段落上的保险。稀疏端会以与文档相同的方式对查询进行词干化，然后按BM25分数返回顶部匹配结果。
class BM25Index:
def search(self, query: str, k: int) -> list[tuple[str, float]]:
q = bm25s.tokenize(query, stemmer=self.stemmer)
idx, scores = self.retriever.retrieve(q, k=min(k, len(self.ids)))
return [(self.ids[int(i)], float(s)) for i, s in zip(idx[0], scores[0])]
#### OUTPUT ####
[index] LanceDB on-disk: /mnt/data/artifacts/lancedb | bm25 over 21259 chunks
这21,259个文本块的整个索引在磁盘上约11.1 MB，非常小，但重点在于形状而非大小。LanceDB将向量存放在NVMe而非RAM中，因此同一条代码路径能承载远超内存大小的索引。这就是文章最后将设计推到千万级向量时所依赖的属性。
检索：融合与重排序
倒数排名融合
现在有两个排名列表，一个稠密，一个稀疏，必须组合起来。陷阱在于它们的分数不可比较，因为BM25分数和余弦相似度处于不同量级。倒数排名融合完全绕开这个问题——忽略分数，仅使用排名，为每个结果赋予一个1除以k加排名的权重，然后在两个列表上将这些权重相加。
def rrf_fuse(rankings: list[list[str]], k: int = 60) -> list[tuple[str, float]]:
scores: dict[str, float] = {}
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 x: -x[1])
看起来比描述起来更简单。取两个短排名列表，其中稠密列表和稀疏列表意见不同，观察融合做了什么。
#### OUTPUT ####
>>> rrf_fuse([["a", "b", "c"], ["b", "c", "a"]])
[('b', 0.03252), ('a', 0.03227), ('c', 0.03200)]
文档b胜出，尽管没有任何一个列表将其排在第一，因为它在两个列表中都排名靠前。这就是全部要点——两个不同检索器都同意的结果，会排在仅一个检索器非常喜欢的结果之上。之所以需要融合，是因为这两个检索器的失败方式不同。稠密搜索会漏掉一个在嵌入空间中不靠近任何已见内容的稀有专有名词，而稀疏搜索会漏掉一个与查询没有共享词的同义改写，融合它们可以找回单独使用任一检索器都会丢掉的文档。混合检索器将它们连接起来：对查询进行一次嵌入，以相同宽度运行两种搜索，然后将两个ID排名融合成一个。
class HybridRetriever:
def retrieve(self, query: str, k: int) -> list[RetrievedChunk]:
qvec = embed_texts(self.embedder, [query], is_query=True)[0]
dense = self.vec.search(qvec, k) # 稠密端捕捉同义改写与语义
sparse = self.bm25.search(query, k) # 稀疏端捕捉精确名称、ID、编号
fused = rrf_fuse([[i for i, _ in dense], [i for i, _ in sparse]], self.rrf_k)[:k]
return [c for c in (self._mk(cid, s, "hybrid") for cid, s in fused) if c]
融合后的列表是召回阶段，故意放宽至150个候选，因为下一阶段才是用召回率换取精确度的地方。
重排序
召回率便宜，精确度昂贵，因此按此顺序执行。前面加载的重排序模型会同时阅读查询与候选文档，并给出它们匹配程度的分数，这比双编码器嵌入准确得多，但也慢得多，无法在整个语料库上运行。仅在150个融合候选上运行它，便处于甜点区。将昂贵模型运行在150个候选而非整个语料库上，也是一个扩展性决策——无论索引包含2万文本块还是1000万文本块，这个成本都固定为150个配对。
class RerankerStage:
def rerank(self, query, cands, top_n):
scores = self.reranker.score(query, [c.text for c in cands])
ranked = sorted(zip(cands, scores), key=lambda x: -x[1])[:top_n]
out = []
for c, s in ranked:
c.score, c.source = float(s), "reranked"
out.append(c)
return out
可以通过对照HotpotQA黄金标题衡量段落召回率，来证明仅稠密、混合再到重排序的提升，并用正在跟踪的例子来展示。
#### OUTPUT ###
Q: Were Scott Derrickson and Ed Wood of the same nationality?
gold titles: ['Scott Derrickson', 'Ed Wood']
recall@20: dense=1.00 hybrid=1.00 reranked=1.00
top-3 reranked:
[a9ec406223bd] (0.999) Scott Derrickson
[2d2201c92ac5] (0.996) Ed Wood
[b7dbb0e190b4] (0.796) Ed Wood (电影)
两个黄金段落均进入前三，重排序分数分别为0.999和0.996，相关性较弱的电影条目以0.796排在更后。在整个评估中，这套检索栈达到0.97的上下文召回率——当问题可回答时，证据几乎总是在那里。检索问题已解决。后续所有工作都是关于如何不滥用它。
路由与问题分解
并非每个查询都值得走完整流水线。问候不需要检索，简单查找需要一个跳转，比较则需要多个跳转。因此智能体做的第一件事，是将问题路由到三个标签之一，仅在有必要的地方消耗计算资源。
ROUTER_PROMPT = (
"将问题分类为以下一个标签：n"
"- no_retrieval：问候/观点或任何文档语料库都无法回答的问题n"
"- single_hop：通过查找一个事实即可回答n"
"- multi_hop：需要从多个文档中组合事实n"
"问题：{q}n仅输出标签。"
)
class QueryRouter:
LABELS = {"no_retrieval", "single_hop", "multi_hop"}
def route(self, query: str) -> str:
out = self.llm.chat("你是一个精确的查询分类器。",
ROUTER_PROMPT.format(q=query), max_tokens=8).strip().lower()
for lbl in self.LABELS:
if lbl in out:
return lbl
return "single_hop"
分解器和前提错误检测也一样小巧。分解器要求返回两个或三个自包含的子问题，检测器则直接问一个是否问题：查询是否假设了某个可能不为真的前提。
DECOMPOSE_PROMPT = (
"将这个多跳问题拆分为2-3个有序的、自包含的子问题，"
"每个子问题一行，无需编号。如果问题已足够简单，则原样返回。n问题：{q}"
)
def detect_false_premise(query: str, llm: LocalLLM) -> bool:
out = llm.chat("你负责检测错误预设。",
FALSE_PREMISE_PROMPT.format(q=query), max_tokens=4)
return out.strip().lower().startswith("y")
#### OUTPUT ####
route('Were Scott Derrickson and Ed Wood of the same nationality?...') -> single_hop
decompose ->
• 斯科特·德瑞克森拥有哪国国籍？
• 埃德·伍德拥有哪国国籍？
同一路由器在另外两类问题上展示了其他分支。
#### OUTPUT ####
route('最好的编程语言是什么？') -> no_retrieval
route('谁执导了《Ed Wood》，这位导演还因什么而闻名？') -> multi_hop
一个观点性问题得到no_retrieval，这本身就是一条弃权路径——系统会拒绝回答，而不是搜索没有任何文档持有答案的问题。一个真正的双事实问题得到multi_hop，会在之后将智能体送入纠错循环。路由器将运行示例判为单跳，因为重排序后的段落已直接回答了它，而分解器仍然展示了如果第一轮检索结果薄弱，它会如何将比较问题拆解为两个干净的查找。路由非常便宜，仅需一次短分类调用，它的价值在于将昂贵的检索和校验工作从不必要的问题上移除，这在扩展时也很重要——每跳过一次检索，都是省下的延迟。
带引文生成
这是第一道幻觉防火墙。系统提示禁止使用外部知识，要求每个句子都有内联引文，并给模型一个明确的令牌，让它在上下文不包含答案时输出。仅告诉模型引用还不够，因此还会验证引文，并删除模型编造的任何引文。
ABSTAIN_TOKEN = "INSUFFICIENT_EVIDENCE"
GENERATION_SYSTEM_PROMPT = (
"你仅从编号的上下文段落中回答。规则：n"
"1. 仅使用段落中的事实，绝不用外部知识。n"
f"2. 如果段落中不包含答案，请准确回复：{ABSTAIN_TOKEN}n"
"3. 每个句子必须以引文结尾，指向其使用的段落ID，格式如[abc123def456]。n"
"4. 保持简洁和事实性。"
)
生成之后，解析引文标记，仅保留与真实文本块ID匹配的引文，这样虚构的引文永远不会出现在用户面前。
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
在一个引用了一个真实段落和一个模型编造段落的句子上运行它，虚假引文会直接消失。
#### OUTPUT ####
>>> text = "巴黎是法国的首都[a1b2c3d4e5f6]。卢浮宫于1793年开放[deadbeef0000]。"
>>> parse_citations(text, valid_ids={"a1b2c3d4e5f6"})
(['a1b2c3d4e5f6'], '巴黎是法国的首都[a1b2c3d4e5f6]。卢浮宫于1793年开放。')
有效ID保留，编造的[deadbeef0000]被移除，仅真实引文会进入下一阶段。这很重要，因为最危险的幻觉是一个自信的句子披着它并不配拥有的引文，而这里它在任何人看到之前就消失了。生成器会用ID格式化检索到的段落，调用模型一次，然后返回弃权信号或一个经过解析和引文检查的答案。
class CitedGenerator:
def generate(self, question, chunks) -> CitedAnswer:
user = f"上下文段落：n{format_context(chunks)}nn问题：{question}nn答案："
raw = self.llm.chat(GENERATION_SYSTEM_PROMPT, user, max_tokens=400).strip()
if ABSTAIN_TOKEN in raw:
return CitedAnswer(text="", cited_ids=[], abstained=True, raw=raw)
cited, cleaned = parse_citations(raw, {c.id for c in chunks})
return CitedAnswer(text=cleaned.strip(), cited_ids=cited, abstained=False, raw=raw)
#### OUTPUT ####
Q: Were Scott Derrickson and Ed Wood of the same nationality?
abstained=False citations=['a9ec406223bd', '2d2201c92ac5']
A: Yes, Scott Derrickson and Ed Wood were of the same nationality; both were American. [a9ec406223bd] [2d2201c92ac5]
答案引用了检索到的两个段落，两个ID均为真实，没有任何内容被移除。此时有了一个流畅、带引文的答案，但引文只能证明模型指向了某个段落，不能证明段落真的支持它说的话。模型可能引用真实段落，却仍然误读它，因此引文是必要条件但不充分。下一道防火墙正是要关闭这个缺口。
校验关卡
这是决定性的防火墙。将草拟的答案拆解为原子化论断，然后用前面加载的忠实度判别模型对照其引用的上下文检查每个论断。分数低于阈值的论断被视为无支撑，如果任何论断失败，整个答案都会降级为弃权。
论断提取器将答案拆解为原子化、可独立检查的陈述，先删除引文标记，使论断成为干净的文本。
class ClaimExtractor:
def extract(self, answer: str) -> list[str]:
clean = _CITE_RE.sub("", answer).strip()
out = self.llm.chat("你将提取原子化的事实性论断。",
CLAIM_DECOMP_PROMPT.format(a=clean), max_tokens=300)
claims = [re.sub(r"^s*d+[.)]s*", "", ln).strip(" -t")
for ln in out.splitlines() if ln.strip()]
return [c for c in claims if len(c) > 3]
关卡提取论断，对照引用的段落为每个论断打分，仅当所有论断都超过阈值时才通过。
class VerificationGate:
def check(self, cited: CitedAnswer, chunks: list[RetrievedChunk]) -> GateResult:
claims = self.extractor.extract(cited.text)
used = [c for c in chunks if c.id in set(cited.cited_ids)] or chunks
context = "nn".join(c.text for c in used)
verdicts = []
for cl in claims:
s = self.verifier.support(cl, context)
verdicts.append(ClaimVerdict(cl, s["score"], s["score"] >= self.tau,
s["nli"], s["minicheck"]))
min_support = min((v.score for v in verdicts), default=0.0)
passed = len(verdicts) > 0 and all(v.supported for v in verdicts)
return GateResult(passed, verdicts, min_support, len(verdicts))
#### OUTPUT ####
claims=3 passed=True min_support=1.00
[OK 1.00] Scott Derrickson is American.
[OK 1.00] Ed Wood is American.
[OK 1.00] Scott Derrickson and Ed Wood share the same nationality.
这个一句话答案被拆解为3个可检查的论断，每个论断对照引用的段落都得到了完整的1.00分，关卡以1.00的最小支撑度通过。在论断级别而非整个答案级别进行检查，是其严格性的关键。一段长答案可以有80% grounded，却仍然夹带一个编造的事实，而答案级别的分数会放它通过；论断级别的关卡则能隔离那个句子并让它失败。关键设计选择是关卡报告的是最弱的论断，而非平均值，因为答案的可信度只取决于它最缺乏支撑的句子。
这个最弱论断规则在触发时最容易看清。下面是同一关卡处理某个前提错误问题的草稿时的情形，模型曾试图配合回答。
#### OUTPUT ####
claims=2 passed=False min_support=0.20
[OK 0.95] Marie Curie was a physicist.
[XX 0.20] Marie Curie traveled to the Moon.
第一个论断支撑充分，但第二个分数为0.20，远低于0.3的阈值，因为没有任何段落提及此事。一个失败论断会将passed翻为False，整个答案被丢弃，问题变成弃权，而不是自信的错误陈述。这正是幻觉被捕获并转化为安全拒答的时刻。
对于边界答案，不直接丢弃它。链式校验过程给它一次修复机会，重写上下文不支持的任何句子并保留引文，然后关卡会在修订文本上再次运行。
COVE_PROMPT = (
"修订答案，使每个句子都直接由上下文支持。"
"移除或弱化任何不被上下文支持的论断。保留引文[id]。nn"
"上下文：n{ctx}nn答案：n{ans}nn修订后的答案："
)
def cove_revise(answer: str, chunks, llm: LocalLLM) -> str:
ctx = format_context(chunks)
return llm.chat("你负责使答案严格忠实于上下文。",
COVE_PROMPT.format(ctx=ctx, ans=answer), max_tokens=400).strip()
适时拒答
弃权是正确答案，而非失败，因此将其作为一等输出。这是实现近乎零幻觉的关键动作。无法阻止模型在语料库中没有答案的问题上出错，但可以让系统拒绝该问题，将一个无界失败（自信的谎言）变成一个可界定的失败（可见、可测、可调的弃权）。策略会将信号合并成一个决策。如果路由器判断为无需检索，或模型输出了弃权令牌，或校验关卡失败，就弃权；否则用已验证的文本回答。
每个输出都是一条严格、可审计的记录，评估可以无歧义地解析已回答和已弃权。
@dataclass
class FinalAnswer:
status: str # "answered" or "abstained"
answer: str
citations: list[str]
min_support: float
reason: str # 哪个关卡触发，或"verified"
class AbstentionPolicy:
def decide(self, route, false_premise, cited, gate) -> FinalAnswer:
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 or gate.min_support < self.tau:
return self._abstain("unsupported_claims", gate)
return FinalAnswer("answered", cited.text, cited.cited_ids,
gate.min_support, "verified", {})
#### OUTPUT ####
AbstentionPolicy ready; reasons = {routed_no_retrieval, false_premise, model_abstained, unsupported_claims, verified}
有一个细节值得指出。前提错误标志作为信号记录，但不是硬关卡，因为一个小型的二元检测器噪声太大，不能单独信任。让评分加论断校验组成的证据路径做真正决策，当没有段落支持这些问题时，它无论如何都会捕获前提错误问题。当系统弃权时，返回一条朴素的信息：“我在现有来源中找不到足够的支撑证据来自信地回答这个问题”，而不是猜测。
智能体设计
现在每个组件都已构建，最后一步是把它们接成一个会自我纠错的图，因为幻觉最大的单一原因就是从错误上下文中生成。这个循环用LangGraph构建，选择它是因为控制流真的是一个图，而非一条直线：路由可以跳过检索，评分可以循环回到优化，校验可以将答案降级为弃权。路由、检索，然后评分证据。如果证据很强，就生成；如果弱，就优化查询并重新检索，直到达到跳转上限；如果无望，就在从未生成的情况下弃权。
智能体在节点之间传递一个状态对象——一个类型化字典，累积分发结果、证据、评分、草稿、关卡结果和运行中的延迟统计。
class AgentState(TypedDict, total=False):
question: str
route: str
query: str
evidence: list
grade: float
draft: Any
gate: Any
final: Any
hops: int
latencies: dict
每个节点只做一件事。评分器对当前段落针对问题的回答程度打分，而优化节点是纠错步骤，增加跳转计数器，分解问题，在再次检索前扩展查询。
def grade_evidence(query: str, chunks, llm: LocalLLM) -> float:
ctx = "n".join(f"- {c.text[:200]}" for c in chunks[:8])
out = llm.chat("你负责评估检索充分性。",
GRADE_PROMPT.format(q=query, ctx=ctx), max_tokens=8)
m = re.search(r"[01](?:.d+)?", out)
return float(m.group()) if m else 0.5
def n_refine(state: AgentState) -> AgentState:
state["hops"] = state.get("hops", 0) + 1
subs = decomposer.decompose(state["question"])
state["query"] = " ".join(subs)
return state
一个小路由函数将评分转为下一步动作，图通过优化步骤循环回到检索的方式连接各个节点。
def _after_grade(state: AgentState) -> str:
g = state.get("grade", 0.0)
if g >= CRAG_OK: # 0.7+，证据充分，直接回答
return "generate"
if g < CRAG_BAD or state.get("hops", 0) >= MAX_HOPS:
return "generate" if g >= CRAG_BAD else "finalize" # 过弱，弃权
return "refine" # 边界情况，优化查询并重试
def build_agent_graph():
g = StateGraph(AgentState)
for name, fn in [("route", n_route), ("retrieve", n_retrieve), ("grade", n_grade),
("refine", n_refine), ("generate", n_generate),
("verify", n_verify), ("finalize", n_finalize)]:
g.add_node(name, fn)
g.set_entry_point("route")
g.add_conditional_edges("grade", _after_grade,
{"generate": "generate", "refine": "refine", "finalize": "finalize"})
g.add_edge("refine", "retrieve") # 纠错循环
g.add_edge("generate", "verify")
g.add_edge("verify", "finalize")
return g.compile()
在运行示例上跑完整智能体，展示每个阶段及其耗时。
#### OUTPUT ####
Q: Were Scott Derrickson and Ed Wood of the same nationality?
route=single_hop hops=0 grade=1.00 status=answered reason=verified
A: Yes, Scott Derrickson and Ed Wood were of the same nationality; both were American.
latencies(s): {'route': 0.16, 'retrieve': 2.4, 'grade': 0.13, 'generate': 0.94, 'verify': 0.97, 'total': 4.6}
评分返回1.00，智能体直接进入生成，最终状态为answered，原因为verified，意味着它通过了每一道关卡。跳转计数器保持为0，但在检索薄弱时，它会爬升到3后才放弃。有界循环既让延迟保持在预算内，也允许第二次和第三次尝试。两行就能体现整个设计的对比——给智能体一个语料库中没有答案的问题，同一图会得出相反但正确的结论。
#### OUTPUT ####
Q: Which programming language did Isaac Newton invent in 1700?
route=single_hop hops=0 grade=0.15 status=abstained reason=unsupported_claims
A: I do not have enough supporting evidence in the available sources to answer this confidently.
latencies(s): {'route': 0.17, 'retrieve': 2.9, 'grade': 0.14, 'total': 3.3}
检索找不到任何关于牛顿发明编程语言的内容，评分返回0.15，低于crag_bad下限0.4，智能体直接进入最终阶段并弃权，完全不生成。这种早期退出也是弃权路径更快的原因（3.3秒 vs 可答情况的4.6秒）——一旦系统知道证据不存在，就不会在生成或校验上花费任何资源。不可答集合中100个问题有98个弃权，就是这样逐个问题实现的。
有效性验证
黄金测试集构建
要衡量这一切，需要一个有两个层次的测试集。可答层次来自HotpotQA，不可答层次来自SQuAD v2的不可回答问题，再加少量手工构建的前提错误问题。不可答的一半是重要部分，因为普通RAG系统会在这里悄悄虚张声势。构建的一切——引文规则、论断关卡、弃权策略——都是为了让这一半保持沉默，因此这个层次才是真正衡量近乎零幻觉论断的部分，而可答的一半则衡量检索是否完成了它的工作。
def build_false_premise_set() -> list[EvalItem]:
qs = [
"爱因斯坦在哪一年获得了他的第二个诺贝尔物理学奖？",
"玛丽·居里飞往月球的宇宙飞船叫什么名字？",
"威廉·莎士比亚在奥运会上获得了多少枚金牌？",
"艾萨克·牛顿在1700年发明了哪种编程语言？",
]
return [EvalItem(f"fp_{i}", q, "", [], False, "false_premise") for i, q in enumerate(qs)]
#### OUTPUT ####
[golden] 200 items (answerable=100, unanswerable=100)
最终得到一个平衡的200 问题集，一半可答，一半不可答。前提错误问题刻意荒谬，问牛顿在1700年发明了哪种语言，因为一个会回答这些问题的系统，也会为任何听起来自信的问题编造事实。平衡两半很重要，如果集合主要是可答的，一个系统即使在困难案例上虚张声势，也可能得分不错。这个集合的一半存在的唯一目的，就是衡量克制。
幻觉的量化呈现
现在在全部200个问题上运行智能体，把结果打成一个2x2的表格。行是可答或不可答，列是已回答或已弃权，唯一危险的单元格是不可答且已回答，因为根据定义这就是幻觉。
def confusion_2x2(results, items) -> np.ndarray:
cm = np.zeros((2, 2), dtype=int)
for r, it in zip(results, items):
i = 0 if it.answerable else 1
j = 0 if r.final.status == "answered" else 1
cm[i, j] += 1
return cm
#### OUTPUT ####
confusion (rows ans/unans, cols answered/abstained):
[[46 54]
[ 2 98]]
hallucinations (unanswerable answered): 2 / 100 unanswerable
看底部行——在100个不可回答问题中，系统对98个弃权，仅回答了2个，也就是专门用来诱捕它的问题上2%的幻觉率。没有校验关卡的普通RAG系统会点亮那个单元格，因为没有任何东西阻止它回答一个语料库无法支持的问题。矩阵的顶部行是这种安全性的代价，接下来看它。
安全性的权衡代价
2x2表格使用了固定阈值，但阈值是一个旋钮。调高它，系统会更多弃权，从而降低幻觉率，但也降低覆盖度。为了有意识地选择它，扫描阈值并画出风险‑覆盖曲线，然后选择一个在幻觉率保持低于预算的同时尽可能多回答的点。
def pick_tau(df, max_halluc: float = 0.05) -> float:
ok = df[df["hallucination_rate"] <= max_halluc]
return float(ok.sort_values("coverage", ascending=False).iloc[0]["tau"]) if len(ok) else 1.0
#### OUTPUT ####
chosen τ* (halluc<=5%): 1.0
metrics: {
"faithfulness": 0.908,
"answer_relevancy": 0.817,
"context_recall@k": 0.97,
"answerable_accuracy": 0.58
}
在已回答的问题上，得到0.908的忠实度和0.97的上下文召回率，说明证据在那里，答案也保持 grounded。代价是矩阵的顶部行。100个可回答问题中，回答了46个，其余弃权，覆盖度为0.46。这是有意的权衡——宁可在本可以回答的问题上保持沉默，也不冒险给出自信但错误的答案。在这条曲线上的位置是产品决策，而非模型决策，可以根据错误答案在对应领域中的代价按语料库设置。
判别模型质量评估
还有一个漏洞需要补上。整个关卡都依赖校验器，一个未经校验的校验器只是把幻觉从答案转移到了评分卡。单独在HaluBench上测试校验器——一个由人工标注忠实和幻觉答案的集合，并报告ROC曲线下的面积。
def eval_verifier(verifier, n: int = 300) -> dict:
hb = load_halubench().shuffle(seed=SEED).select(range(n))
scores, labels = [], []
for ex in hb:
scores.append(verifier.nli_score(ex["answer"], ex["passage"]))
labels.append(1 if str(ex["label"]).upper().startswith("PASS") else 0)
from sklearn.metrics import roc_auc_score
return {"auroc": round(float(roc_auc_score(labels, scores)), 3), "n": len(labels)}
#### OUTPUT ####
[verifier] AUROC=0.702 over n=300 HaluBench items
校验器在300个条目上拿到AUROC 0.702，明显好于随机，但远非完美。直言这一点，因为它是整个关卡真正的上限。更强的校验器是能进一步提升上述数字的单一改变，而架构的构建方式允许不动其他部分就替换进去。关卡不需要完美校验器才有用，它需要的是足够经常地把有支撑的论断排在无支撑的论断之上，以移动操作点，0.702达到了这个门槛，同时仍有大量提升空间。
向量规模扩展至千万级
真实的千万级向量索引
质量流水线已在精心挑选的切片上得到证明。现在必须字面证明扩展性的论断，因为标题说的是千万级+文档，而基准测试是唯一能定论的东西。因此在10万、100万和1000万向量上构建LanceDB索引，带真实的近似最近邻索引，测量每一步的构建时间、磁盘占用和查询延迟。必须使用近似IVF_PQ索引，而非精确搜索，因为精确扫描会将查询与每个向量比较，复杂度随n线性增长，这正是1000万时会爆炸的成本；而近似索引仅访问少量分区，并将每个向量量化到几个字节，用一点召回率换取随着语料库增长几乎不动的延迟。
为了保持这是一个纯向量搜索基准测试，这里的向量是合成的1024维单位向量，通过Arrow导入它们，让路径可以承载数千万行。主机有180 GB RAM和750 GB NVMe磁盘，一个1000万向量的索引可以舒适地放在单机上，这正是磁盘存储的全部意义。
class ScaleBench:
def run(self, sizes: list[int]) -> "pd.DataFrame":
rows = []
for n in sizes:
vecs = make_synthetic_vectors(n, self.dim)
db = lancedb.connect(str(SCRATCH_DIR / f"scale_{n}"))
t0 = time.time()
tbl = db.create_table("v", data=self._arrow(vecs), mode="overwrite")
if n >= 100_000:
tbl.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(tbl, vecs, build_s))
return pd.DataFrame(rows)
#### OUTPUT ####
[scale] building n=100,000 with IVF_PQ ANN index ...
-> {'n': 100000, 'build_s': 41.82, 'disk_gb': 0.39, 'p50_ms': 8.5, 'p95_ms': 10.59, 'recall@10': 0.135}
[scale] building n=1,000,000 with IVF_PQ ANN index ...
-> {'n': 1000000, 'build_s': 81.22, 'disk_gb': 3.884, 'p50_ms': 11.34, 'p95_ms': 14.46, 'recall@10': 0.105}
[scale] building n=10,000,000 with IVF_PQ ANN index ...
-> {'n': 10000000, 'build_s': 347.04, 'disk_gb': 38.825, 'p50_ms': 16.91, 'p95_ms': 18.48, 'recall@10': 0.105}
关键数据在最后一行。一个1000万向量索引以18.48毫秒的p95回答，而规模小100倍的索引回答时间为10.59毫秒。数据增长一百倍，延迟增长不到一倍。磁盘线性增长，从0.39 GB到38.8 GB，这正是想要的——磁盘便宜，而这种规模的内存索引并不合适。构建时间也以温和的方式增长，从10万向量的42秒到1000万的不到6分钟，每一个字节都留在单机NVMe磁盘上。
千万级查询18毫秒，及外推至1亿
延迟几乎不动的原因是近似索引的性质。IVF_PQ索引搜索少量分区，而非整个空间，查询成本随分区数增长，而非随向量数增长；磁盘线性增长，每个向量仍然必须存储。拟合这一趋势，外推到1亿。
def fit_and_extrapolate(df, target: int = 100_000_000) -> dict:
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
#### OUTPUT ####
projection to 100M: {
"build_s": 3075.1,
"disk_gb": 388.23,
"p95_ms": 77.58
}
在1亿向量时，投影落在77.58毫秒的p95，索引为388 GB，仍能放在单台机器的NVMe磁盘上。需要明确一个注意事项：此处recall@10接近0.1，只是因为向量是随机的，近似索引几乎没有真正可找的东西，这次运行衡量的是延迟和吞吐量，而非检索质量。在真实语料库上，同一索引会保持高召回率，延迟数字才是扩展时保持成立的部分。
时间开销分布分析
扩展是容易的部分。昂贵的是每个查询的智能体，因此按阶段归因延迟，看看预算实际花在哪里。
def aggregate_latencies(results) -> "pd.DataFrame":
stages = {}
for r in results:
for k, v in r.latencies.items():
stages.setdefault(k, []).append(v)
rows = [{"stage": k, "p50_s": round(np.percentile(v, 50), 3),
"p95_s": round(np.percentile(v, 95), 3),
"mean_s": round(np.mean(v), 3)} for k, v in stages.items()]
return pd.DataFrame(rows).sort_values("mean_s", ascending=False)
#### OUTPUT ####
stage p50_s p95_s mean_s
total 4.001 17.668 5.823
retrieve 3.074 11.393 4.166
verify 1.534 3.878 1.758
generate 1.451 2.484 1.619
refine 1.471 2.888 1.575
route 0.168 0.206 0.170
grade 0.127 0.431 0.161
典型问题在约中位数4秒完成，慢尾p95达到17.7秒。检索占据主导，因为它运行嵌入器、两种搜索，并让交叉编码器重排序模型处理150个候选；在困难问题上，还会通过纠错循环运行不止一次。向量搜索本身是便宜的部分，这也是扩展实验教给我们的同一结论——索引不是瓶颈，围绕它的语言模型调用才是。在优化前值得知道，收益来自减少模型调用、批处理重排序或缓存评分，而非换一个更快的向量数据库。
当前范围与后续方向
最后需要明确这是什么，以及它不是什么。幻觉率是不可答集合上2%，而非零，因为从生成式模型出发不可能实现字面上的零。可答问题上的覆盖度为0.46，这是为这种安全性有意付出的代价，风险‑覆盖曲线是在两者之间权衡的旋钮。1000万运行是基于合成向量的向量搜索基准测试，它证明索引在延迟和磁盘上可以扩展，真实语料库才是在同等速度下保持高召回率的东西。校验器的AUROC为0.702，不错但不算优秀，它是下一步最值得改进的部分。
从这里开始，有几个方向值得投入。
- 更强的校验器：关卡的质量取决于判别模型，更好的忠实度模型会一次性提升所有下游指标。
- 真实嵌入的规模化：在真实文档向量上重新运行扩展实验，确认召回率保持的同时18毫秒延迟仍然成立。
- 分片与量化：超过单机后，索引会拆分到分片上，正确性逻辑完全不变。
- 覆盖度校准：按领域调整阈值，高风险语料库更多弃权，常规语料库更多回答。
这些后续方向都不会改变设计主干。索引可以增长，校验器可以改进，阈值可以移动，但核心契约保持不变——到达用户的每个句子，都是系统能够在检索文本中找到支撑的句子，其余全部变成弃权。
整件事就是一个想法贯彻到底：不试图让模型永远不出错，而是构建一个系统，只说它能证明的内容，否则就拒答。索引以18毫秒扩展到1000万向量，答案以0.908忠实度保持grounded，而它无法支持的问题会返