夜雨聆风构建近零幻觉的千万级文档 RAG 管线
检索、约束、验证、放弃回答
往一个 RAG 系统里塞的文档越多，它编造内容的可能性就越大。当语料库膨胀到百万、千万乃至更大规模时，幻觉问题只会越来越严重。要在这种规模上保持答案可信，就需要一条让 Agent 自己核对证据、并为每一条陈述附上引用的管线。这正是 Claude 用来生成引用的同一种思路。
完整管线：从问题到带引用的答案，或经过校准后的放弃回答（作者：AI拉呱）
下面是这条管线的全部内容，我们将自上而下逐个组件构建：
• 环境准备与数据获取：下载语料库，查看其规模与真实样本，固定每一个随机种子，确保运行可复现。 • 清洗与切块：对文本做归一化，用 MinHash LSH 去除近似重复，再按结构感知的方式切成带一行上下文前缀的小块。 • 构建混合索引：把每个切块同时存入 LanceDB 的稠密向量与 BM25 稀疏倒排，全部落盘，以便扩展到一千万+向量。 • 检索与重排：用倒数排名融合（RRF）把稠密与稀疏排序融合在一起，再把 150 个候选重排到 20 个。 • 路由与分解：对每个问题先分类，把多跳问题拆成子问题再去检索。 • 带引用的生成：严格基于上下文回答，每句话都附引用；否则输出弃答标记。 • 逐条验证：把答案拆成原子断言，用 faithfulness judge 拿每条断言去核对引用的原文。 • 拿不准就放弃：把所有信号折叠成一次校准过的决策，证据不足时拒绝回答。 • 串成 Agent：把所有组件接成一个自纠错的 CRAG 循环，在证据薄弱时重新检索。 • 评估与扩展：在 200 道题的金集上打分，再把索引规模拉到真实的 1000 万向量并外推到 1 亿。
完整代码在我的 GitHub 仓库（理论 + 代码）：
GitHub - 用 RAG 处理 1000 万+ 文档且实现零幻觉
目录
• 近零，不是绝对零 • 项目环境搭建 • 获取数据 • 清洗语料 • 切块与上下文 • 加载检索模型 • 构建混合索引 • 检索：融合与重排 • 路由与分解 • 带引用的生成 • 验证关卡 • 何时放弃回答 • Agent • 它工作得怎么样？ • 扩展到 1000 万+ 向量 • 范围与下一步
近零，不是绝对零
我们要解决的问题不是"把模型做得更聪明"。一个更大的模型在检索什么都没找到时一样会瞎猜，因为瞎猜本就是生成的本能。
所以与其追求一个完美的模型，不如用一个只有一种安全失败模式的系统把普通模型包裹起来。当证据缺失时，正确的输出不是一个流畅的猜测，而是一次放弃回答。
整套系统遵循的原则：检索证据、把生成约束在证据上、验证每条断言、缺乏支持时放弃回答（作者：AI拉呱）
这给我们四层控制，下面每个章节对应其中一层。
1. 检索到正确的证据：稠密 + BM25 混合检索、上下文化切块、重排。 2. 约束生成：只从上下文回答，每句话引用相应段落 id，否则放弃。 3. 逐条核对原子断言：用 faithfulness judge 把每条断言与引用文本核对。 4. 放弃：当断言支持度或检索置信度低于校准过的阈值时。
我们同时追求两个目标。第一个是可信，意思是在我们选择回答的问题上幻觉率近乎为零。
第二个是规模，意思是检索骨干需要承载 1000 万+ 向量，并且仍能在毫秒级返回。第一个目标依赖验证逻辑，第二个目标依赖索引设计。
我们两个都要建。
项目环境搭建
在写任何逻辑之前，先把项目搭起来。计划是：导入库、固定每个随机种子保证可复现、检查我们手上这一块 GPU、用一个轻客户端连到生成器，最后冻结配置以便无头运行每次行为一致。
环境准备：导入库与种子、检查 GPU、连接已预热的生成器、冻结配置（作者：AI拉呱）
首先是导入和一个用于给每个随机数生成器植入种子的函数。
import json, os, random, subprocess, time
from dataclasses import dataclass, asdict, field
import numpy as np
def set_determinism(seed: int) -> None:
"""给我们用到的每个 RNG 都植入种子，保证可复现。"""
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"
}我们用的是一块 80GB 显存的 NVIDIA H100，宿主机配 180GB 内存和 750GB NVMe 硬盘，这在索引变大时才显其重要。32B 的生成器不住在这个 notebook 里。
它住在另一个 vLLM 服务里，我们用一个小型 OpenAI 兼容客户端去访问。让它在独立进程里保持预热意味着我们可以重复运行这个 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:为了低延迟跳过 <think> 推理过程
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服务已就绪。最后一步是把所有旋钮冻结到一个 config 对象里并打印出来，让支撑后续全文的数字集中在一处。
#### 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 个，允许 Agent 最多 3 次纠错跳数，把两个支持度阈值设在 0.3 并稍后校准。生成器是 Qwen3-32B，嵌入器和重排器是 4B 版本的 Qwen3，faithfulness judge 用 32B 本身。
我把生成器温度固定为 0、关闭思考过程，因为我要可复现、低延迟的答案，而且采样是又一个让模型偏离证据的地方。这里所有模型都是本地开源权重的 Qwen3，原因是整条管线的前提就是文档和查询都不出这台机器，这正是它能在私有语料上落地的关键。
工具就绪，可以去拿数据了。
获取数据
管线的质量上限就是底下语料的质量上限，所以第一步是下载一个数据集然后看看它。我选了 HotpotQA 的 distractor 切片，有两个理由。
每道题都自带句级别的黄金支持证据，这是最干净的方式来给检索召回率打分；而它捆绑的 Wikipedia 段落白送我一个真实语料库。对面的测试集，我从 SQuAD v2 的 impossible 部分以及手写的几个伪前提问题中取，因为衡量幻觉的唯一办法就是问语料库无法回答的问题，看系统是否能保持沉默。
第三个数据集 HaluBench 留到最后纯粹用于验证验证器本身。HotpotQA 是我们要建索引并检索的语料。
每个数据集各有分工：HotpotQA 是语料和可回答集，SQuAD v2 + 伪前提问题是不可回答集，HaluBench 测验证器（作者：AI拉呱）
from datasets import load_dataset
def load_hotpotqa(split: str = "validation"):
# datasets 3.x 需要带命名空间的 repo 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总共 7405 道题，每道题都自带它出自的 Wikipedia 段落。我们定义段落（passage）和问题（question）的样子，再写一个 builder，把每道题的上下文段落汇成一个语料库，同时记录哪些段落是黄金证据。
@dataclass
class Passage:
id: str
title: str
text: str
is_gold_for: list[str] = field(default_factory=list) # 这是哪几道题的黄金证据
@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_premiseclass CorpusBuilder:
"""从 HotpotQA distractor 上下文构建段落语料库 + QA 条目。"""
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"])
# （完整版 builder 还会记录每道题的黄金支持句）
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现在我们手上有 20007 个段落、2073 道题，其中 4072 个段落被标记为某道题的黄金证据。在往上面盖任何东西之前，我们应该真正看一眼数据：尺寸分布和一个真实样例。
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 Davis Wood Jr. was an American filmmaker, actor, writer, producer, and director.段落平均 89 个词，短到能在一个 prompt 里放下几段，又长到能承载一个事实。样例是一道比较型问题，"Scott Derrickson 和 Ed Wood 是否同一国籍？"，它的两条黄金句已经包含答案，两人都是美国人。
这就是我们要跟随穿过管线每一步的问题，看一道真实问题穿过整条管线能让每个组件具象化。两个分层在这里已经显现。
像这样的可回答问题让我能衡量正确证据是否被召回，而我后加的不可回答问题用来衡量幻觉——一个对语料中没支持的问题给答案的系统就是一个会编造的系统。
清洗语料
进来的是垃圾出去就是幻觉，所以索引前先清洗文本。两个便宜的步骤回报远超投入：归一化让分词器对每个段落表现一致；近似去重让被复制或转发的段落不会在结果顶端挤成一团，从而虚抬召回但没增加新证据。
清洗：对每个段落归一化，再用 MinHash LSH 去除近似重复，剩下 19987 个段落（作者：AI拉呱）
import re, unicodedata
def normalize_text(s: str) -> str:
s = unicodedata.normalize("NFKC", s) # 规范化 Unicode 形式
s = s.replace("", "") # 去除软连字符
s = re.sub(r"[ \t]+", " ", s) # 合并连续空白
return s.strip()我先做 NFKC 归一化，因为 BM25 是按原始字符分词的，所以连字符或一串野生空格会把一个词拆成两个，或者把两个并成一个，悄悄拉低召回。下面是一个杂乱字符串上函数的效果。
#### OUTPUT ####
>>> normalize_text("the final report\twas ready")
'the final report was ready'连字符 "fi" 变成普通 "fi"，制表符和成串空格合并为单空格，于是只在不可见字符上有差异的两个段落现在能被同样地分词。
去重那一步是关键。我必须选择 MinHash LSH 这类近似方法而不是两两比较，因为穷举两两比较是平方复杂度，在语料规模上永远跑不完；而 MinHash 配 LSH 索引能近线性地找出近似重复。
去掉它们同时服务于两个目标：让索引规模在向 1000 万向量迈进时更小；防止同一段落的三个副本挤在结果顶端——那是检索器悄悄给模型喂冗余上下文、诱使其过度信赖单一来源的方式。
class Deduper:
"""通过基于词 shingle 的 MinHash LSH 去除近似重复段落。"""
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 个过短片段，剩 19987 个段落。这个语料是精选切片，但清洗步骤无论输入是 2 万还是 2000 万段都一样能用。
切块与上下文
现在把段落切成切块（chunk）。定长切块是省事的选择，也是错误的选择，因为它会把携带实体的句子和消歧上下文割开，这对多跳问题是致命的。
所以我们把整句按 token 预算打包，再加一点重叠，并且用生成器自己的分词器计数，预算才能与模型真正看到的对齐。这是一个藏在切块细节里的幻觉问题。
如果切块溢出预算被悄悄截断，恰好携带答案的那句话就消失了，问题就无端"看起来"无答案，所以我宁愿尊重句界，多付出一些切块的代价。
切块按 token 预算打包整句，之后上下文化器在前面拼一行情境句（作者：AI拉呱）
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)
# 一旦加上这句会超 token 预算，就开新切块
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)切出 21259 个切块，平均 125 token，舒服地低于 256 的预算。还有一个问题要在索引前解决。
像"那季度营收增长了 3%"这种切块单独是不可检索的，因为是谁的营收、哪一季度都不在了。所以索引前给每个切块前面拼一行情境句，这是上下文检索的思路，只是我们用本地 Qwen3 来写这句，而不是托管模型。
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."
)这个方法把每切块的调用扇出到线程池，因为调用之间相互独立，vLLM 在服务端会自动批处理，比逐切块跑快得多。我们还把结果做了 checkpoint，重跑时跳过整个步骤。
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("You write concise retrieval context.", 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 ####
Before:
Ed Wood is a 1994 American biographical period comedy-drama film directed and
produced by Tim Burton, and starring Johnny Depp as cult filmmaker Ed Wood...
After (context-prefixed):
This chunk introduces the 1994 film *Ed Wood*, directed by Tim Burton, and
outlines its main subject and cast.
Ed Wood is a 1994 American biographical period comedy-drama film...多出来的这句很便宜，每个切块一次短生成，它告诉检索器这个切块讲的是什么——哪怕切块单独看是歧义的。这种提升解释了为什么最终召回这么高。召回是整个反幻觉故事的基石，因为下游验证器只能在检索真正找到的证据上做接地，所以这里每多买到一个百分点的召回，就是多一道我能回答而不是回避的问题。
加载检索模型
切块准备好了，接下来加载把它们变成可检索证据并稍后核对答案的模型。三个模型与生成器共享这张 GPU，所以每次加载后我们都拍一次显存快照，确保不超预算。这里加载重排器和 faithfulness judge，嵌入器在构建索引时再加载。
一张 H100 同时承载 vLLM 中的 32B 生成器，以及内核里的嵌入器、重排器、judge（作者：AI拉呱）
我们不希望显存超用以三步后 OOM 的形式被发现，所以每次加载都同时记录 nvidia-smi 的全卡占用和 torch 的内核占用。
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) # 全卡，两个进程
print(f"[vram] {tag:22} gpu_used={used}GB kernel={kernel}GB")
return {"tag": tag, "gpu_used_gb": used, "kernel_gb": kernel}重排器是一个用作"是/否"判定器的小型因果模型。每个 (query, doc) 对被装进固定模板，得分直接从下一个 token 的 logits 里读出，所以重排一次就是每个候选一次前向。我用一个专门的 cross-encoder 重排器而不是嵌入相似度，是因为嵌入器把整段压成一个向量，扫库够快但模糊了"只是提到那几个实体"和"真的回答了问题"之间的区别——而这正是把错证据挡在 prompt 之外、挡在答案之外的关键。
class Qwen3Reranker:
"""通过模型给 'yes' token 的概率给 (query, doc) 对打分。"""
@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]] # 比较 'no' 与 'yes'
probs = torch.softmax(yn.float(), dim=-1)[:, 1] # 保留 P('yes')
out.extend(probs.cpu().tolist())
return outfaithfulness judge 就是 32B 的生成器本身，提示它给某条断言对应一段上下文返回一个支持度分数。我把 judge 用本地 32B，是因为 RAG 里的忠实性核对意味着一次性把一条断言对几段长文本一起读，这恰是小型句对 NLI 模型脆弱的地方，也因为这个 judge 是唯一把"自信的错答案"转成"放弃回答"的组件。
它是近零幻觉论断的心脏，所以我宁愿把手头最强的模型砸在这里。NLI cross-encoder 和 MiniCheck 仍作为更轻的备选挂着，但这次跑用的是 LLM judge。
JUDGE_PROMPT = (
"You are a strict fact-checker. Decide whether the CONTEXT supports the CLAIM.\n\n"
"CONTEXT:\n{context}\n\nCLAIM: {claim}\n\n"
"Output ONLY a number: 1.0 if the context clearly states or entails the claim, "
"0.0 if it contradicts or does not mention it, or a value in between."
)
class JudgeVerifier:
def _score(self, claim: str, context: str) -> float:
out = self.llm.chat("You are a strict faithfulness grader.",
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)整套栈占用 80GB 中的 54.3GB，给后面建索引留了余地。judge 不需要额外显存，它复用已经在 vLLM 服务里跑的生成器。一切都留在一台机器上，没有任何东西外发到第三方 API。
构建混合索引
现在建索引，这里的问题是单一检索器不够用。稠密向量擅长 paraphrase，当问题与答案用不同词时它有用。
BM25 擅长精确 token，比如人名、id 和数字，正是稠密模型会糊掉的部分。所以我们把两者都建索引，按切块 id 关联，建在上下文化后的文本上。
混合索引为每个切块存两份：LanceDB 中的稠密向量与 BM25 稀疏倒排（作者：AI拉呱）
我们只为建索引加载嵌入器，把每个切块都嵌入完，然后释放，再用一个更小的在线嵌入器服务查询。向量做了归一化，所以余弦相似度就是普通点积。
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: # Qwen3-Embedding 在查询时要带 prompt
kw["prompt_name"] = "query"
return embedder.encode(texts, **kw).astype("float32")加载查询嵌入器是显存的最后一道压力，快照能看到落点。
#### OUTPUT ####
[vram] embedder(online) gpu_used=61.85GB kernel=15.04GB显存峰值大约 80GB 中的 62GB，仍在预算内；索引完后我立即释放更重的离线嵌入器，只让小的在线嵌入器留在显存里服务查询。我必须选 LanceDB，是因为它是嵌入式的、落盘在 NVMe 上、没有服务进程要起，意味着同一段代码能容纳远大于内存的索引——而这一点是这个设计最终能扩展到 1000 万+ 向量、不改一行代码的关键。
稠密那一边是它的薄包装，唯一的小技巧是把余弦距离换回 0-1 范围的相似度。
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()
# 余弦距离在 [0, 2] 区间，转成 [0, 1] 的相似度
return [(r["id"], 1.0 - r["_distance"] / 2.0) for r in res]我同时保留一份 bm25s 的词汇索引，因为稠密嵌入恰好就会把一个生僻名字、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 chunks21259 个切块整个索引在磁盘上大约 11.1 MB，很小，但重点是形状不是大小。LanceDB 把向量留在 NVMe 上而非内存里，所以同一段代码可以容纳远大于内存的索引。这就是文末把这套设计推到 1000 万向量时所依靠的属性。
检索：融合与重排
倒数排名融合
我们现在有两个排名列表，一个稠密一个稀疏，得把它们合起来。陷阱在于两者的分数不可直接比较，因为 BM25 分数和余弦相似度活在不同尺度上。
倒数排名融合（Reciprocal Rank Fusion，RRF）完全绕过这点。它忽略分数，只用排名，给每个结果一个 1/(k + 排名) 的权重，再把两个列表的权重加起来。
RRF 按排名融合稠密与稀疏列表，无需分数归一化（作者：AI拉呱）
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 胜出，尽管它在哪个列表里都不是第一，因为它在两个列表里都靠前。这就是要点。
两个检索器都同意的结果跑赢只有一个检索器看好的结果。我们之所以融合，正是因为两种检索器以不同方式失败。
稠密会漏掉一个生僻专有名词，它在嵌入空间里离任何见过的东西都远；稀疏会漏掉一个与查询零词共现的 paraphrase；融合恢复了任何一方单独会丢掉的文档。检索器把这两端拧到一起：把查询编一次嵌入、同宽度并行跑两个搜索、再把两个 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) # 稠密抓 paraphrase 与语义
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 个候选，因为下一个阶段会把这种召回换成精度。
重排
召回便宜，精度贵，所以我们按这个顺序跑。前面加载好的重排器一起读查询和一个候选，给它们的匹配打分，这远比 bi-encoder 嵌入精确，但慢到不可能跑遍整个语料库。
只跑 150 个融合候选是甜区。把贵模型跑在 150 个候选上、而不是全语料库上，也是一个扩展性选择：无论索引有 2 万切块还是 1000 万切块，这部分成本固定在 150 对上。
一个薄阶段封装这个模型，给每个候选打分、保留前 20 个。
重排取 150 个融合候选、按重排器分保留 20 个最好的（作者：AI拉呱）
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 (film)两段黄金段落都落在前三，重排分别为 0.999 和 0.996，相关性较低的电影词条排在 0.796。在整个评估上，这套检索栈达到 0.97 的上下文召回，意味着在问题可回答的情况下证据几乎总能找到。
检索算解决了。后面所有事都是为了不滥用它。
路由与分解
不是每个查询都该走完整管线。一句问候不需要检索，简单查找走一跳，比较型问题需要好几跳。所以 Agent 做的第一件事就是把问题路由到三种标签之一，让算力只用在该用的地方。
路由器把每个问题送进 no_retrieval、single_hop 或 multi_hop 路径，并独立做一次伪前提检测（作者：AI拉呱）
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."
)
class QueryRouter:
LABELS = {"no_retrieval", "single_hop", "multi_hop"}
def route(self, query: str) -> str:
out = self.llm.chat("You are a precise query classifier.",
ROUTER_PROMPT.format(q=query), max_tokens=8).strip().lower()
for lbl in self.LABELS:
if lbl in out:
return lbl
return "single_hop" # 模型啰嗦时的安全兜底分解器和伪前提检测器同样小巧。分解器要 2-3 个自包含的子问题，检测器问一个直白的"是否假设了某件可能不真的事"。
DECOMPOSE_PROMPT = (
"Break this multi-hop question into 2-3 ordered, self-contained sub-questions, "
"one per line, no numbering. If it is already simple, return it unchanged.\nQuestion: {q}"
)
def detect_false_premise(query: str, llm: LocalLLM) -> bool:
out = llm.chat("You detect false presuppositions.",
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 ->
• What is the nationality of Scott Derrickson?
• What is the nationality of Ed Wood?同一个路由器对另外两类问题的分支：
#### OUTPUT ####
route('What is the best programming language?') -> no_retrieval
route('Who directed Ed Wood, and what is that director also known for?') -> multi_hop意见类问题落到 no_retrieval，本身就是一条放弃路径，因为系统选择拒绝而不是去搜一个文档不会持有的答案。一个真的两跳问题落到 multi_hop，它会驱使 Agent 走纠错循环。
路由对我们这个示例答的是 single_hop，因为重排后的段落已经直接回答了它，分解器仍能展示如果第一遍召回单薄它会怎么把这个比较拆成两次干净的查找。路由很便宜，一次短分类调用，它存在的意义在于不让昂贵的检索和验证花在不需要它的问题上，这在规模上同样关键，因为我跳过的每一次检索都是我没花的延迟。
带引用的生成
这是第一道幻觉防火墙。系统提示禁止外部知识、要求每句话都带行内引用，并给模型一个明确的 token，在上下文不含答案时直接抛出。光让模型引用还不够，所以我们还要校验引用、剔除模型编造的引用。
带引用的生成：只从上下文回答、每句一引用，否则放弃，并剔除任何无效引用（作者：AI拉呱）
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, like [abc123def456].\n"
"4. Be concise and factual."
)生成后我们解析引用标记，只保留与真实切块 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 = "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 .')有效 id 留下，捏造的 [deadbeef0000] 被剔除，所以只有真实引用能到达下一阶段。这点之所以重要，是因为最危险的幻觉就是一句穿戴着不属于它的引用、显得很自信的话；而在这里，那个引用在任何人看到之前就被去掉了。生成器把检索到的段落带 id 格式化好，模型调用一次，要么返回弃答信号，要么返回一个解析过、引用核对过的答案。
class CitedGenerator:
def generate(self, question, chunks) -> CitedAnswer:
user = f"Context passages:\n{format_context(chunks)}\n\nQuestion: {question}\n\nAnswer:"
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 都是真实的，所以没有任何东西被剔除。到这一步我们有一个流畅、带引用的答案，但引用只能证明模型指向某段，不能证明那段真的支持它说的。
模型可以引用真段落却把它读错，所以引用是必要不充分。这道空缺由下一道防火墙补上。
验证关卡
这是决定性的防火墙。我们把草稿答案拆成原子断言，再用前面加载的 faithfulness judge 把每条断言对其引用的上下文核对。任何低于阈值的断言视为不被支持，只要有一条断言失败，整个答案降级为放弃回答。
验证关卡把答案拆成原子断言，对其引用上下文打分，只要有一条低于 tau 就放弃回答（作者：AI拉呱）
断言抽取器把答案拆成原子的、可独立核对的陈述，先把引用标记去掉让断言是干净文本。
class ClaimExtractor:
def extract(self, answer: str) -> list[str]:
clean = _CITE_RE.sub("", answer).strip() # 先去掉 [id] 标记
out = self.llm.chat("You extract atomic factual claims.",
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 = "\n\n".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.这一句答案被拆成三条可核对断言，每条都拿到 1.00 满分，关卡以最小支持度 1.00 通过。在断言级而不是答案级核对，这才是它的严苛之处。
一个长答案可以 80% 都接地，但仍能塞进一条编造的事实，答案级打分会放它过去；断言级关卡会孤立那一句并因它失败。关键设计选择是：关卡报告最弱的那条断言，而不是平均，因为一个答案的可信度取决于它最不被支持的那句话。
最弱断言原则在它真触发时最容易看清楚。下面是同一道关卡作用于一道伪前提问题的草稿。
#### 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，整答案被丢掉，问题变成放弃回答而不是一句自信的伪陈述。这就是一次幻觉被抓住并转成安全拒答的瞬间。
对于边缘答案我们不会直接扔掉。一次 chain-of-verification 给它一次自我修复机会，把上下文不支持的句子改写或软化，并保留引用，然后关卡再核一次。
COVE_PROMPT = (
"Revise the answer so EVERY sentence is directly supported by the context. "
"Remove or soften any claim not supported. Keep citations [id].\n\n"
"Context:\n{ctx}\n\nAnswer:\n{ans}\n\nRevised answer:"
)
def cove_revise(answer: str, chunks, llm: LocalLLM) -> str:
ctx = format_context(chunks)
return llm.chat("You make answers strictly faithful to context.",
COVE_PROMPT.format(ctx=ctx, ans=answer), max_tokens=400).strip()何时放弃回答
放弃是正确答案，不是失败，所以我们让它成为一等结果。这一招让"近零幻觉"在原则上成为可能。
我无法阻止模型在语料里没答案的问题上犯错，但我可以让系统拒绝那个问题，把无界的失败模式（自信撒谎）转成有界的（可见的弃答），后者我能衡量、能调。策略把所有信号折叠成一个决定。
如果路由说不检索、模型自己抛了弃答 token、或验证关卡失败，我们就放弃；否则用经过验证的文本回答。
弃答策略：三道硬关卡触发弃答，验证路径触发回答，伪前提作为信号被记录（作者：AI拉呱）
每个结果都是严格、可审计的一条记录，所以评估时区分回答与弃答没有任何猜测空间。
@dataclass
class FinalAnswer:
status: str # "answered" 或 "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}有一个细节值得点出。伪前提标志被记成一个信号，但不是硬关卡，因为一个小型是/否检测器单独不够可靠。
我们让"打分 + 断言验证"的证据路径作真正决策，它本来就能抓住伪前提（没段落支持就过不了关）。当系统真要弃答时，它返回一句平实的"我在可用的资料里没有足够证据自信地回答"，而不是一个猜测。
Agent
到这里我们造完了所有组件，最后一步是把它们接成一张能自纠错的图，因为幻觉最大的来源就是基于劣质上下文生成。这个循环用 LangGraph 搭建，我选它是因为控制流确实是图而不是直线：路由能跳过检索、打分能回到精炼步、验证能把答案降级为弃答，这些分支我宁愿声明出来，而不是埋在嵌套条件里。
我们路由、检索、然后给证据打分。证据强就生成；弱就精炼查询再检索，最多重复到跳数上限；无望就直接弃答，不进入生成。
CRAG 循环给自己证据打分，证据弱时精炼并重新检索，只在证据足够强时才生成（作者：AI拉呱）
Agent 在节点之间传一个状态对象，一个有类型的字典，累积路由、证据、打分、草稿、关卡结果以及实时延迟。
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每个节点只做一件事。打分器衡量当前段落对问题的回答能力；refine 节点是纠错步，它增加跳数计数器、分解问题、扩宽查询，再次检索。
def grade_evidence(query: str, chunks, llm: LocalLLM) -> float:
ctx = "\n".join(f"- {c.text[:200]}" for c in chunks[:8])
out = llm.chat("You grade retrieval sufficiency.",
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一个小路由函数把分数变成下一步动作，图把节点接起来，refine 步回环到 retrieve。
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()把完整 Agent 跑在样例上能看到每一步以及它的耗时。
#### 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，Agent 直接进生成，最终状态 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，Agent 直接 finalize 到弃答，根本没进入生成。这也是为什么弃答路径更快，3.3 秒对比已回答情形的 4.6 秒——一旦知道证据缺失，系统就不在生成与验证上花一分钱。这就是不可回答集 100 题里 98 次弃答的样子，一题一题地。
它工作得怎么样？
金集
要衡量任何这些，需要一个有两个分层的测试集。可回答分层来自 HotpotQA，不可回答分层来自 SQuAD v2 impossible 加几道手写的伪前提。
不可回答那一半是关键，因为那是普通 RAG 系统会悄悄说瞎话的地方。我们建的所有东西，引用规则、断言关卡、弃答策略，都是为了让那一半保持沉默；所以这一半实际给"近零幻觉"打分，而可回答那一半给检索打分。
金集是 100 道可回答与 100 道不可回答，包含手写的伪前提问题（作者：AI拉呱）
def build_false_premise_set() -> list[EvalItem]:
qs = [
"In what year did Albert Einstein win his second Nobel Prize in Physics?",
"What was the name of the spaceship Marie Curie flew to the Moon?",
"How many gold medals did William Shakespeare win at the Olympics?",
"Which programming language did Isaac Newton invent in 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 年发明哪种编程语言——因为一个会回答这种问题的系统就会对任何"听起来自信"的问题编造事实。
平衡两半很重要，因为以可回答为主的集合会让系统看似得分高，却仍在难题上吹牛。这一半存在的唯一目的就是衡量克制力。
幻觉只活在一格里
现在把 Agent 跑在 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 cm2x2 混淆矩阵，幻觉就是"不可答-已答"格中的两例（作者：AI拉呱）
#### 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风险-覆盖率曲线，所选工作点把幻觉压在预算下（作者：AI拉呱）
#### 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 上下文召回，意味着证据在场、答案接地。代价是矩阵的顶行。
100 道可答问题里我们答了 46 道，其余弃答，覆盖率 0.46。这是刻意的取舍。
我们宁愿在本可回答的问题上保持沉默，也不愿冒一个自信错答的风险。你最终落在曲线上的哪一点是产品决策、不是模型决策，并可以按语料调整：错答代价高的领域多弃答，轻量场景多回答。
Judge 自己靠谱吗？
还有一个洞要补。整道关卡都倚赖验证器，所以一个没验证过的验证器只是把幻觉从答案搬到记分牌。我们用 HaluBench 单独测验证器，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"])) # judge 的支持度分数
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)}HaluBench 上验证器的 ROC 曲线，AUC 为 0.702（作者：AI拉呱）
#### OUTPUT ####
[verifier] AUROC=0.702 over n=300 HaluBench items验证器在 300 例上拿到 AUROC 0.702，明显好过随机但远未完美。我想直白地说出这一点，因为它是整道关卡真正的天花板。
更强的验证器是单一最有价值的下一步改动，它会同时拉高上面所有数字；而架构本身让你可以原地替换它而不动其他部分。关卡不需要完美的验证器，它只需要一个把"被支持的断言"排在"未被支持的断言"之上的次数足以移动工作点的验证器，0.702 跨过这道线还留有不少成长空间。
扩展到 1000 万+ 向量
一个真实的 1000 万向量索引
质量管线在精选切片上得到了证明。现在我们必须把规模这一论断变成字面意义上的事实，因为标题写的是 1000 万+ 文档，能定论的只有基准测试。
我们用 LanceDB 在 10 万、100 万、1000 万向量规模上分别建一次真正的近似最近邻索引，并测量构建时间、磁盘占用与查询延迟。我必须用近似的 IVF_PQ 索引而不是精确搜索，因为精确扫描把查询和每一个向量比，复杂度对 n 线性，正是 1000 万规模上爆炸的成本；而近似索引只访问少量分区并把每个向量量化到几个字节，用一点召回换一个几乎不动的延迟。
为了让它是干净的向量搜索基准，这里的向量是合成的 1024 维单位向量，并通过 Arrow 入库以承载几千万行。宿主机有 180GB 内存和 750GB NVMe，1000 万向量的索引能舒服地装在一台机器上，这正是落盘存储的全部意义。
规模实验在 10w、100w、1000w 合成向量上构建真实的 IVF_PQ 索引并测 p95 延迟（作者：AI拉呱）
class ScaleBench:
def run(self, sizes: list[int]) -> "pd.DataFrame":
rows = []
for n in sizes:
vecs = make_synthetic_vectors(n, self.dim) # 1024 维单位向量
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: # 建一个真实的 ANN 索引
tbl.create_index(metric="cosine",
num_partitions=int(min(4096, max(256, n ** 0.5))),
num_sub_vectors=64)
build_s = time.time() - t0
# 然后跑 50 次查询测 p50/p95，对比暴力搜索看 recall@10
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 ms p95 返回，而百倍小的索引是 10.59 ms。
数据百倍增长，延迟还不到翻倍。磁盘线性增长，从 0.39GB 到 38.8GB，正合期望，因为磁盘便宜，而同尺寸的内存索引就贵了。
构建时间也以同样温和的速度增长，从 10w 时的 42 秒到 1000w 时的不到 6 分钟，全部都在一台机器的 NVMe 盘上。
1000 万 18ms，外推到 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) # 对 n 做线性拟合
out[col] = round(float(a * target + b), 2)
return out10w 至 1000w 的实测 p95 延迟、磁盘和构建时间，红色为 1 亿外推（作者：AI拉呱）
#### OUTPUT ####
projection to 100M: {
"build_s": 3075.1,
"disk_gb": 388.23,
"p95_ms": 77.58
}在 1 亿向量上外推落在 77.58 ms p95、388GB 索引，仍能装在一台机器的 NVMe 上。一个直白的注意点。
这里 recall@10 大约 0.1 只是因为向量是随机的，给了近似索引几乎没东西可找；所以这次跑测的是延迟和吞吐，不是检索质量。在真实语料上同样的索引能保持高召回，而延迟数字会保持。
时间花在哪
规模是简单的部分。贵的是每次查询的 Agent，所以我们把延迟按阶段归因，看预算到底花在哪里。
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)各阶段 p95 延迟，检索主导端到端预算（作者：AI拉呱）
#### 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 个候选跑 cross-encoder 重排，难题上还会经过纠错循环再来一次。
向量搜索本身是便宜部分，与规模实验给的同一个教训。索引不是瓶颈，围绕它的语言模型调用才是。
这点在优化前值得知道，因为收益在削减模型调用、批处理重排、缓存打分上，不在更快的向量库上。
范围与下一步
在结尾我想坦率说清这是什么、不是什么。幻觉率是 不可答集上 2%，而不是零，因为生成模型字面意义上的零并不可达。
可答问题上的覆盖率是 0.46，这是我们为安全付的刻意代价，风险-覆盖率曲线就是用来在两者间调的旋钮。1000 万跑的是合成向量的向量搜索基准，所以它证明索引在延迟和磁盘上能扩展，而真实语料才是让召回保持在同样速度下的关键。
验证器停在 AUROC 0.702，好但不出色，是下一步最值得改的部分。
往后几个方向值得投入：
1. 更强的验证器：关卡的上限就是 judge 的上限，更好的忠实性模型能同时拉高下游所有数字。 2. 真实嵌入下的扩展：在真实文档向量上重跑规模实验，确认在 18ms 延迟下召回保持。 3. 分片与量化：超过单机后，索引切分成片，正确性逻辑完全不变。 4. 校准的覆盖率：按领域调阈值，高风险语料更倾向弃答、轻量语料更倾向回答。
这些下一步不会动设计的脊梁。索引可以增长、验证器可以变强、阈值可以平移，但契约不变。每一句到达用户的话都是系统能在检索文本里指给你看的，其余一切转成弃答。
整套东西是一个想法贯彻到底。我们不去试图让模型永不出错，而是建一个只说能证明的东西、其余都放弃的系统。索引扩展到 1000 万向量、18ms；答案以 0.908 忠实性接地；它无法支持的问题以一句平实的"我没有足够证据"返回，而不是一个自信的猜测。
AI拉呱：洞察AI技术前沿
