向量检索
从表示到证据

10.1.1 Embedding 到底表示什么

Embedding 可以写成一个函数 fθ(x) → v：输入文本、图像或其他对象 x，输出固定长度的向量 v。向量的每一维通常没有独立的人类可读含义，含义来自整个空间中的相对位置。两个文本靠近，表示模型认为它们在训练目标下有相似的使用场景；这不是一句“它们事实相同”的保证。

最早的词表示常用 one-hot：词表有一万个词，每个词用一万个维度表示，只有一个位置是 1。one-hot 能区分身份，却不能表达“猫”和“狗”比“猫”和“发动机”更接近。稠密向量用较少维度承载统计关系，距离由训练数据塑造，因此也会继承语料中的领域偏见和缺失。

训练信号决定空间的形状。Skip-gram 让词向量预测上下文，分类任务让表示区分标签，双塔检索让问题向量靠近正确文档、远离负例，句向量模型还会使用正负句对和对比损失。相同一句话用不同目标训练，得到的向量未必适合相同的检索任务。

向量的“相似”是条件性的。一个新闻语料训练的模型可能把“苹果”更多理解为公司，把食品语料训练的模型更多理解为水果；技术文档里的“接口”与社交文本里的“接口”也可能落在不同邻域。使用前先说明领域、语言、时间范围和目标任务，不要把通用模型当作永远正确的语义地图。

Embedding 还会受输入粒度影响。词向量适合查词和组成短语，句向量适合段落级检索，文档向量适合粗粒度聚类；把整本书压成一个向量，虽然省空间，却很难定位具体证据。RAG 通常需要文档级元数据和 chunk 级向量同时存在，不能只保留一个“整篇文档向量”。

向量维度不是越大越好。维度增加可能提高表达容量，也会增加存储、内存带宽和索引成本；低维模型更快，但可能把细粒度差异压在一起。比较模型时同时记录维度、dtype、归一化方式、最大输入长度和每千条文本的计算时间，才能讨论“质量提高是否值得”。

import numpy as np

texts = ["退款需要订单号", "订单退款要提供编号", "今天适合散步"]
# 这里用可解释的 toy 向量模拟编码器输出
vectors = np.array([
    [0.90, 0.80, 0.10],
    [0.88, 0.77, 0.12],
    [0.05, 0.10, 0.95],
], dtype=np.float32)

print("count, dimension:", vectors.shape)
print("dtype:", vectors.dtype)
print("text/vector:", list(zip(texts, vectors.tolist())))

示例故意不用真实模型，先让你检查最基本的契约：有多少条文本、每条多少维、数据类型是什么、文本和向量是否按同一顺序对应。接入真实编码器后，再补充最大输入长度、批次、设备、归一化和模型版本；这些字段缺失时，向量库里的分数很难复现。

10.1.2 词向量、句向量与多语言表示

词向量通常把一个词或 Token 作为单位，但现代文本编码器往往先把词拆成多个子词，再通过上下文得到 hidden state。同一个词在不同句子里可以有不同表示，这叫上下文表示；句向量则需要把多个 Token 的信息聚合成一个固定长度结果。聚合方式可能是特殊 Token、平均池化、加权池化或模型专门训练的 pooling head。

句向量模型常见的训练信号是对比学习：给定一个查询和正确文本作为正对，再给出其他文本作为负例，让正对的相似度更高。负例怎么选非常关键。随机负例容易区分，难负例更接近真实检索边界；如果负例本身也正确，模型会被迫把两个正确答案推远，造成空间扭曲。

多语言模型希望不同语言中表达相同意图的句子靠近，但对齐不是自动完美的。语言覆盖、脚本、形态变化、翻译质量和训练数据比例都会影响结果。中文查询配英文文档、简体配繁体、代码注释配自然语言时，都要用自己的跨语言样本测试，不要只看模型卡上的语言列表。

Embedding 的各向异性是一个常见现象：大量向量集中在空间的一小块区域，导致任意两段文本都显得相似。中心化、白化、归一化或模型专门的训练目标可能改善分布，但处理方式要在验证集上比较。盲目做后处理可能提高平均分，却伤害某些长尾类别。

句向量还会受到长度和格式影响。把标题、正文、表格和页脚直接拼在一起，模型可能把页码、导航词当作语义；过长输入被截断后，关键结论可能完全没有进入向量。入库前需要保留标题路径、文档类型和字符范围，必要时分别编码标题与正文，再在检索阶段融合。

选择 pooling 时不要凭接口名称判断质量。平均池化可能对短句稳定，特殊 Token 可能依赖模型训练方式，最后一个 Token 对因果模型有时更自然，但每个模型的预训练和后训练不同。最小做法是取一个小验证集，比较不同 pooling、归一化和长度策略的 Recall@k 与延迟。

import numpy as np

# [token_count, hidden_size] 的 toy hidden states
hidden = np.array([
    [1.0, 0.0, 0.0],
    [0.8, 0.2, 0.0],
    [0.0, 0.8, 0.2],
], dtype=np.float32)

mean_vector = hidden.mean(axis=0)
mean_vector = mean_vector / np.linalg.norm(mean_vector)
print("mean pooled:", np.round(mean_vector, 3))

# 只保留有效 Token，避免 padding 参与平均
valid = hidden[:2]
masked = valid.mean(axis=0)
masked = masked / np.linalg.norm(masked)
print("masked pooled:", np.round(masked, 3))

这里的关键不是某个 pooling 永远最好，而是 padding 和有效长度必须明确。实际批处理会补齐不同长度，如果把补齐向量也平均进去，短文本会被系统性拉向某个方向。记录 attention mask、有效 Token 数和 pooling 版本，才能解释两个实现为什么分数不同。

10.1.3 选择模型与建立向量契约

选 Embedding 模型先从任务反推。查询是短问题还是完整段落？文档是中文政策还是代码？需要跨语言吗？索引规模和更新频率如何？是否允许把数据发送到外部服务？这些问题比“排行榜第几名”更能缩小选择范围。一个在通用语义相似度上高分的模型，可能不理解你的产品术语。

向量契约至少写出 model_id、版本或 hash、维度、dtype、归一化、最大输入 Token、pooling、批次、语言、相似度函数和更新时间。查询端与文档端必须使用兼容契约；只要更换模型、维度或归一化，就不能把新向量直接写进旧索引。迁移时保留双写或双索引，比较一段时间后再切换。

外部 API 的向量还要记录供应商、区域、请求版本和失败重试。服务返回的浮点精度、截断策略和模型别名可能变化，调用方如果只保存一个友好的模型名称，就无法重现旧结果。对敏感资料要考虑数据出境、保留策略、加密和删除接口，Embedding 并非自动匿名化。

本地模型则要关注权重许可、Tokenizer 文件、硬件、量化和并行。量化可能改变向量方向和近邻顺序，不能只比较生成质量。首次部署时跑一个固定向量基准，保存若干文本的向量摘要、范数、近邻 ID 和耗时，后续升级才能做回归。

契约还应规定异常输入。空字符串、只有标点、超长文本、未知语言、包含秘密的文本和编码错误应该分别处理。是返回零向量、报错、跳过入库还是切成多个 chunk，必须由上层流程决定；悄悄把异常输入送进模型会让索引混入无法解释的点。

最后建立“质量—资源—风险”三维选择表。质量包括召回和切片表现，资源包括单条编码时间、显存、存储和查询成本，风险包括许可、隐私和供应商依赖。只有当三维条件都满足，模型才适合进入生产，而不是因为一个分数漂亮就上线。

import json
from pathlib import Path

manifest = {
    "model_id": "example-embedding-v2",
    "model_revision": "sha256:demo",
    "dimension": 768,
    "dtype": "float32",
    "normalize": True,
    "pooling": "mean_mask",
    "max_input_tokens": 512,
    "similarity": "cosine",
}

Path("embedding_manifest.json").write_text(
    json.dumps(manifest, ensure_ascii=False, indent=2),
    encoding="utf-8",
)
print("contract saved:", manifest["model_id"])

把 manifest 当作数据的一部分保存，而不是只放在部署文档里。索引中的每条记录都可以引用 manifest_id；查询日志也保存它，出现排名变化时能回答“是不是模型或归一化变了”。当你准备切换模型时，先用离线集比较新旧近邻和下游答案，再决定是否迁移。

import numpy as np

query = np.array([0.8, 0.6], dtype=np.float32)
docs = {
    "refund": np.array([0.79, 0.61], dtype=np.float32),
    "shipping": np.array([0.60, 0.80], dtype=np.float32),
    "weather": np.array([-0.2, 0.1], dtype=np.float32),
}

def cosine(a, b):
    return float(a @ b / (np.linalg.norm(a) * np.linalg.norm(b)))

ranked = sorted(
    ((name, cosine(query, vector)) for name, vector in docs.items()),
    key=lambda item: item[1],
    reverse=True,
)
print(ranked)

练习时把排序结果与人工标签并排展示。若“shipping”因为词面相似排在“refund”前面，说明模型或查询需要更多上下文、关键词过滤或重排，而不是简单把阈值调高。每一次修复都保留前后的近邻，避免只报告改进后的样例。

import numpy as np

def audit(vectors):
    vectors = np.asarray(vectors, dtype=np.float32)
    norms = np.linalg.norm(vectors, axis=1)
    pairwise = vectors @ vectors.T
    return {
        "shape": vectors.shape,
        "norm_min": float(norms.min()),
        "norm_max": float(norms.max()),
        "mean_per_dim": np.round(vectors.mean(axis=0), 4).tolist(),
        "self_similarity_ok": bool(np.allclose(np.diag(pairwise), 1.0, atol=1e-3)),
    }

vectors = np.array([[1, 0], [0.8, 0.6], [-1, 0]], dtype=np.float32)
vectors = vectors / np.linalg.norm(vectors, axis=1, keepdims=True)
print(audit(vectors))

这个审计器只检查几何和形状，不判断文本是否真的相关。它适合放在入库前，及早发现空向量、维度错误、未归一化或向量塌缩。语义质量仍要通过带人工标签的检索集验证。

再做一个可解释性对照：把每个向量的最近邻和最远邻都保存下来。最近邻告诉你模型把什么视为相似，最远邻能暴露异常方向、空文本或编码失败。检查这些极端样本，通常比只看平均范数更快发现问题。

如果你的语料中有同一事实的多个版本，向量相似并不表示版本可以互换。把日期、地区、适用对象和生效状态作为元数据保存；后续检索要先做权限和版本过滤，再用向量排序。向量只负责发现候选，不负责替你决定哪一版有效。

记录 Embedding 的失败成本。一次批量编码失败可能阻塞索引，一条异常文本可能污染近邻，模型升级可能让所有分数重新分布。为批处理设置重试、坏记录清单和断点续跑，确保修复单条文本时不用从头猜测整个数据集发生了什么。

把实验结果写成可迁移的规则：短查询用句向量，长资料用 chunk；精确编号走关键词，语义改写走向量；敏感租户先过滤，公共语料再排序。规则带条件和证据，未来换模型或换数据库仍然成立。

不要把向量可视化图当作质量证明。降维会改变距离，二维图只适合观察大致簇和异常点；真正的检索质量仍要在原始维度上用 Top-k、标签和声明级结果评估。可视化是诊断工具，不是最终指标。

本节的最终产物是一份向量卡：模型契约、输入处理、近邻示例、失败样本、资源账本、隐私边界和迁移办法。下一节会在这张卡上加入距离函数、阈值和召回评价。

再把向量卡交给一个没有参与实验的同伴阅读：他能否只根据卡片判断向量维度、距离方向、输入清洗和版本迁移？如果不能，说明契约还依赖口头知识。把疑问改成字段和示例，后续的索引、召回和评估才有共同语言。

向量表示的学习也应留下反例。保存一个词面相似却事实冲突的近邻、一个跨语言正确近邻、一个空文本异常和一个版本过期近邻；解释模型为什么可能做出这些判断，并写出数据或规则层面的补救。

10.2.1 点积、余弦与欧氏距离

点积是两个向量对应维度相乘后求和：a·b=Σaᵢbᵢ。它同时受到方向和长度影响。一个向量范数很大，即使方向一般，也可能获得较高点积；在某些双塔模型中，范数确实携带训练信号，但在另一些检索任务中，长度差异只是噪声。使用点积前要确认编码器和索引的约定。

余弦相似度把点积除以两个向量的范数：cos(a,b)=(a·b)/(||a||||b||)，只比较方向。两个完全同向的向量得分接近 1，正交接近 0，反向接近 -1。文本向量经 L2 归一化后，点积和余弦排序等价，但原始范数信息也会被舍弃。

欧氏距离是坐标差的平方和开根号：||a-b||₂。对已归一化的向量，欧氏距离和余弦排序有固定关系；对未归一化向量，两者可能给出不同近邻。不要只看数据库的参数名叫“cosine”还是“L2”，实际检查它是否自动归一化、返回的是距离还是相似度，以及排序方向是否正确。

距离的数值范围决定阈值解释。余弦通常在 -1 到 1 之间，归一化后的点积也在这个范围；非归一化点积可能没有固定上限；L2 距离随着维度和范数变化。把一个模型的 0.8 阈值搬到另一个模型，往往会造成大量漏召回或误召回。阈值必须在自己的验证集上校准。

高维空间会出现距离集中：随机向量之间的距离差异变小，最近邻和普通邻居不再明显分开。向量维度、数据规模、归一化和分布都会影响这种现象。除了看 Top-1，还要看 Top-10 与随机候选的分数分布、最近邻间隔和不同查询的排名稳定性。

数值稳定性也不能忽视。零向量无法做余弦，极小范数会放大浮点误差，float16 可能在长向量点积时丢精度。编码器输出进入索引前先检查 NaN、Inf、范数和维度；查询时遇到异常要返回明确错误，不要让数据库悄悄把坏向量排在前面。

import numpy as np

a = np.array([3.0, 4.0], dtype=np.float32)
b = np.array([6.0, 8.0], dtype=np.float32)
c = np.array([4.0, -3.0], dtype=np.float32)

def cosine(x, y):
    nx, ny = np.linalg.norm(x), np.linalg.norm(y)
    if nx == 0 or ny == 0:
        raise ValueError("zero vector")
    return float(x @ y / (nx * ny))

def l2(x, y):
    return float(np.linalg.norm(x - y))

for name, item in [("b", b), ("c", c)]:
    print(name, "dot=", float(a @ item),
          "cos=", round(cosine(a, item), 3),
          "l2=", round(l2(a, item), 3))

代码中的 a 和 b 方向完全相同，点积很大但余弦为 1；这正好说明“长”与“同向”是两个不同信号。接入真实向量后，把未归一化和归一化结果都打印一次，确认数据库排序和本地计算一致。

10.2.2 归一化、Top-k 与阈值

Top-k 表示返回分数最高的 k 个候选，但 k 不是质量保证。k 太小，正确证据可能排在后面；k 太大，上下文会被无关内容挤满，生成成本和冲突概率上升。k 应与后续重排能力、上下文预算、证据粒度和任务风险一起决定。

阈值与 Top-k 是两种不同控制。Top-k 保证“至少返回 k 个”（如果索引有结果），阈值保证“只有分数足够高才返回”。实践中常用两者组合：先取较大的候选集，再按校准阈值、元数据和去重过滤，最后保留不超过预算的证据。资料不足时允许返回少于 k 个，不能为了凑数塞入低相关文本。

归一化通常有两层：编码器输出的 L2 归一化，以及数据库存储或查询时的自动归一化。只做一边会让分数不可比，做两边通常结果相同但可能浪费计算。把归一化责任写进契约，并用一对已知向量验证索引返回的分数。

阈值需要按切片校准。短查询、长查询、中文、英文、代码、数字和低资源语言的分数分布可能不同；同一个全局阈值会偏爱某些切片。可以用验证集画 precision-recall 曲线，选择满足风险预算的点，或为不同查询类型设置不同门槛，并记录门槛版本。

稳定排序也很重要。分数相同或非常接近时，数据库返回顺序可能随分片、并发和索引重建变化。为每条记录加入稳定的 tie-breaker，例如 doc_id；日志保存原始分数和排序前后列表。否则一次升级后排名变化，无法判断是模型变了还是并列顺序变了。

检索结果还要做去重和多样性控制。同一文档的相邻 chunk 可能占据 Top-k，重复内容挤掉其他来源。可以按文档配额、标题路径、来源和版本去重，也可以使用 MMR 在相关性和新颖性之间折中。去重规则本身要评估，过度去重会删除互相补充的证据。

candidates = [
    {"id": "a", "score": 0.91, "doc": "policy", "tokens": 120},
    {"id": "b", "score": 0.89, "doc": "policy", "tokens": 130},
    {"id": "c", "score": 0.87, "doc": "faq", "tokens": 90},
    {"id": "d", "score": 0.62, "doc": "old", "tokens": 110},
]

threshold = 0.80
budget = 250
selected, used_docs, used_tokens = [], set(), 0
for item in candidates:
    if item["score"] < threshold:
        continue
    if item["doc"] in used_docs:
        continue
    if used_tokens + item["tokens"] > budget:
        continue
    selected.append(item)
    used_docs.add(item["doc"])
    used_tokens += item["tokens"]
print("selected:", [x["id"] for x in selected])

这个示例把阈值、文档去重和 Token 预算放到同一个选择器中。生产版本要先做权限过滤，再做分数排序；示例中的“每文档一个”只是教学规则，是否合理应由任务验证。任何被过滤或截断的候选都应该写入调试日志。

10.2.3 语义召回的基线与陷阱

任何向量检索都应和简单基线比较。关键词、BM25、标题匹配或规则过滤在编号、专有名词、错误代码和精确版本上经常胜过语义检索。一个向量方案只有在相同测试集上超过或补足基线，才值得增加复杂度；如果基线已经能解决任务，就不要为“用了 Embedding”而强行加入向量。

语义相似容易被否定、数字和实体差异欺骗。“允许退款”和“不允许退款”词面几乎一样，向量可能把它们排在一起；“版本 2.1”和“版本 2.10”只差一个字符，却可能代表不同规则；人名、产品名和地区需要精确匹配。对这些场景加入关键词、字段过滤或重排器，而不是单纯提高向量维度。

长查询往往混合多个意图。把整句编码成一个向量，会让一个主题遮住另一个主题；可以先拆成子查询，分别召回，再合并证据并去重。拆分也可能放大错误，必须记录原查询、子查询和各自结果，确保能够解释最终上下文来自哪一步。

查询和文档的语言风格不对称时，Embedding 也可能退化。用户说“怎么退钱”，文档写“售后款项返还政策”；训练好的模型通常能连接两者，但公司内部缩写、产品代号和新词未必能。把真实查询、同义改写和内部术语加入验证集，必要时训练领域适配器或维护术语表。

检索错误要分成漏召回和误召回。漏召回是正确文档没进候选集，应该查编码、分块、查询和索引；误召回是无关文档排名靠前，应该查距离、关键词、过滤和重排。两者的修复方向不同，不能只看最终答案“错了”就调一个阈值。

建立难负例比随机负例更有价值。随机文档很容易被排除，难负例要与查询共享词面或主题，却缺少关键事实。每次发现错误近邻，把它加入回归集；但要确认它确实是负例，避免把两个都正确的资料错误标成互斥。

def lexical_score(query, document):
    q = set(query.lower().split())
    d = set(document.lower().split())
    return len(q & d) / max(len(q), 1)

def recall_at_k(results, gold_ids, k):
    hit = sum(item["id"] in gold_ids for item in results[:k])
    return hit / max(len(gold_ids), 1)

query = "版本 2.1 的退款期限"
docs = [
    {"id": "semantic-1", "text": "售后款项返还政策"},
    {"id": "exact-2.1", "text": "版本 2.1 退款期限为七天"},
]
ranked = sorted(
    docs,
    key=lambda item: lexical_score(query, item["text"]),
    reverse=True,
)
print(ranked)

基线代码很简单，却能提醒你精确词面在数字和版本任务中的价值。实际系统可以把 lexical、dense、metadata 和业务规则的结果并列记录，再由重排器或融合算法排序。先让每条通道独立可测，调试时才知道某条证据是从哪里进入候选集的。

def recall_at_k(ranked_ids, gold_ids, k):
    return int(bool(set(ranked_ids[:k]) & set(gold_ids)))

def reciprocal_rank(ranked_ids, gold_ids):
    gold = set(gold_ids)
    for rank, item_id in enumerate(ranked_ids, start=1):
        if item_id in gold:
            return 1.0 / rank
    return 0.0

cases = [
    (["d2", "d1", "d8"], {"d1"}),
    (["d4", "d5", "d6"], {"d9"}),
]
for ranked, gold in cases:
    print({
        "recall@2": recall_at_k(ranked, gold, 2),
        "rr": reciprocal_rank(ranked, gold),
    })

这个报告只计算最小指标，真正项目还要按查询类型和风险切片聚合。先确认 gold 标签质量，再解释数字；如果标签不完整，低 recall 可能是标注问题。指标、原始排名和人工判断必须同时保留。

做距离实验时，把同一批向量分别以 float32 和 float16 保存，比较排序是否改变。若只有边界样本改变，记录可接受波动；若大量近邻改变，先检查量化和归一化，再决定是否需要更高精度。

做阈值实验时，不要把测试集拿来反复调门槛。用训练/校准集选择阈值，用锁定的测试集报告最终结果。这样才能知道门槛是在适应真实分布，还是只记住了几条样本。

做基线实验时，保持同样的 gold 集、同样的权限过滤和同样的输出预算。否则 dense 方案使用更多上下文、lexical 方案使用更少上下文，比较出来的不是检索算法差异。

如果一个候选没有来源或版本信息，即使相似度很高，也不能进入最终上下文。来源、版本和权限是证据的组成部分，不是检索完成后的装饰字段。

记录查询失败而不是只记录成功。空查询、编码异常、索引超时、低于阈值和权限过滤都应该有状态码；失败分布会告诉你系统需要更好的用户提示、降级通道还是数据修复。

掌握这些约定后，你就能把“搜索感觉不准”拆成数学、数据、排序和产品动作四层问题，下一步才有必要引入 ANN 索引来提高规模和速度。

相似度报告还应保留排序间隔。Top-1 为 0.81、Top-2 为 0.80 与 Top-1 为 0.81、Top-2 为 0.42 的风险不同；前者需要重排或人工确认，后者通常有更清晰的候选。用 margin、分数分布和人工标签一起判断。

当多个来源都相关时，保留它们的出处和覆盖范围。不要为了提高 precision 只留一条最像的文本；问答可能需要政策条款、例外条款和定义条款共同支持。最终选择器应按事实单元而不是只按文档数量做配额。

查询中的时间词也需要显式处理。“现在”“去年”“最新”不能只交给 Embedding；先解析时间范围，再用 metadata 过滤和新鲜度排序。否则模型可能把旧版本的语义近邻送进上下文，生成一段过时但流畅的答案。

在多租户系统中，分数比较必须发生在权限过滤之后。不同租户的向量可以在数学上相似，但不代表可以互相暴露；先限定可见集合，再做 Top-k，日志只记录当前租户允许看到的 ID。

本节的结业标准是能用一页表格解释：距离如何计算、分数如何校准、候选如何去重、基线如何比较、资料不足如何处理，以及一次错误近邻怎样被加入回归集。

检查报告中是否明确写出了“相关性”的定义：是同一主题、同一答案、同一版本，还是能支持某一条声明。定义不清，模型分数再高也无法指导下一步。

把阈值、k 和排序函数记录成配置而不是散落在代码里；后续切换索引或模型时，配置和回放事件会告诉你哪些结果仍然可以比较。

一个好的召回协议还会规定失败如何对用户说：低分时说明资料不足，权限过滤时不泄露被过滤内容，索引故障时切换到关键词或人工路径。技术分数只有转化为清晰动作，才真正改善体验。

def decide(scores, threshold=0.72, margin=0.06):
    if not scores or scores[0] < threshold:
        return "clarify"
    if len(scores) > 1 and scores[0] - scores[1] < margin:
        return "keep_multiple"
    return "continue"

for scores in ([0.88, 0.71], [0.79, 0.77], [0.54, 0.51]):
    print(scores, decide(scores))

这段策略不是通用真理，而是一个可被数据推翻的起点。把 threshold、margin、模型、距离和数据切片写入实验配置；当策略改变时，保留旧配置和回放结果，避免只凭一次线上样例就修改门槛。

完成这个实验后，距离函数就不再是孤立的数学选择，而是连接检索质量、上下文预算和用户体验的接口。下一章的 ANN 索引可以加速候选发现，但不能替代这里建立的校准与证据协议。

10.3.1 ANN、HNSW、IVF 与 PQ

精确最近邻的做法是把查询向量与所有文档逐一计算距离，再排序。这在小数据集上最容易理解，也应该作为质量基线；但候选数量增加后，计算、内存访问和网络返回都会成为瓶颈。ANN，也就是近似最近邻，用少量召回损失换取更低延迟，关键是要知道损失发生在哪里。

HNSW 把向量看成图上的节点，每个节点连接若干近邻，并在不同层构建稀疏的导航结构。查询从高层的少量节点开始，逐步下降到更密的底层，在局部扩展候选。M 控制连接数，ef_construction 影响建图质量和构建成本，ef_search 影响查询时探索范围；参数越大通常召回更好，但内存或延迟也会上升。

IVF 先用聚类把向量分到若干倒排桶，查询时只探查最接近的几个桶。桶数、训练样本、nprobe 决定搜索范围。桶太少，候选多、速度慢；桶太多，边界附近的正确向量可能落在未探查的桶里。IVF 适合数据量较大且可以接受训练索引的场景，更新策略要额外设计。

PQ 将向量分段并把每段量化为较短的码，用码表近似计算距离，从而节省内存和带宽。压缩会改变边界样本的排序，适合先粗召回、再用原始向量重排。不要把 PQ 的距离直接拿去和未压缩向量的阈值比较，先做一组压缩前后的近邻一致性实验。

ANN 的调参目标不是追求某个参数最大，而是在召回、p95 延迟、内存、构建时间和更新成本之间找到满足门槛的点。用固定查询集扫描 ef_search、nprobe 或压缩率，记录 Recall@k 和资源。平均延迟之外要看长尾，因为一次慢查询可能拖住整个 RAG 请求。

索引还会受到硬件和访问模式影响。随机内存访问使 CPU cache 不友好，GPU 索引可能提高吞吐却增加搬运和构建复杂度；批量查询能摊薄开销，但在线单查询仍要满足交互延迟。压测要区分冷启动、热缓存、单查询、批量和并发，不能用一个数字代表全部场景。

settings = [
    {"name": "hnsw-fast", "ef_search": 32, "expected_recall": 0.90},
    {"name": "hnsw-balanced", "ef_search": 96, "expected_recall": 0.95},
    {"name": "hnsw-careful", "ef_search": 192, "expected_recall": 0.98},
]

budget_ms = 80
for item in settings:
    latency = 20 + item["ef_search"] * 0.25
    status = "ok" if latency <= budget_ms else "over_budget"
    print(item["name"], round(latency, 1), status)

代码只是假设延迟模型，目的是展示如何把索引参数和预算放在一张实验表里。真实压测要记录硬件、数据量、向量维度、并发、缓存状态和实际召回。用固定查询集与精确 Flat 结果做对照，才知道“expected_recall”是否真的成立。

10.3.2 向量库 Schema、过滤与租户

一条向量记录不应只有一个浮点数组。最小 Schema 包括稳定的 chunk_id、父文档 doc_id、向量、原文或可回查位置、标题路径、语言、来源、版本、生效时间、权限标签和模型 manifest。把这些字段放入索引或可联查的存储，检索结果才能从坐标回到可引用的证据。

过滤条件与向量搜索的执行顺序会影响结果。如果先在全库找近邻，再过滤权限，可能得到不足 k 个候选，甚至把不该看到的 ID 暴露在调试日志；如果先过滤可见集合，搜索空间变小但索引可能需要支持高选择率过滤。向量库的文档要确认 pre-filter、post-filter 和过滤索引的实际语义。

多租户系统把 tenant_id 作为硬边界。最安全的做法是查询时由服务端从会话取得租户和用户权限，模型或客户端不能自行传入任意 tenant_id。大租户可以独立索引，小租户可以共享索引但必须有强制过滤；无论哪种方式，都要用交叉租户测试证明不会泄露。

权限不只是一列标签。文档可能对部门、角色、地区、项目或时间段开放，chunk 继承权限但也可能有更严格的局部规则。权限更新后，旧向量和缓存需要失效；查询结果要保存当时的授权决策版本，方便审计。宁可少召回，也不能把不可见资料拼进上下文。

元数据还能帮助新鲜度和多样性控制。查询“最新政策”时先按生效时间和状态过滤，再按向量排序；同一文档的相邻 chunk 可以按父文档配额；不同来源的证据可以要求至少覆盖两个独立文档。元数据不是对向量的替代，而是把业务约束放回正确的位置。

Schema 设计要考虑演进。向量维度或模型版本变化时，可以用 collection_v1、collection_v2 和 alias 做双读；字段新增时给默认值并更新解析器；删除时保留 tombstone 或操作日志，防止离线重建又把已删除文档带回来。所有 schema 变更都写迁移说明和回滚路径。

REQUIRED = {
    "chunk_id", "doc_id", "vector", "text_ref",
    "tenant_id", "acl", "manifest_id",
}

def validate_record(record, dimension):
    missing = sorted(REQUIRED - record.keys())
    if missing:
        return False, "missing:" + ",".join(missing)
    if len(record["vector"]) != dimension:
        return False, "dimension_mismatch"
    if not record["chunk_id"].startswith(record["doc_id"] + ":"):
        return False, "unstable_chunk_id"
    return True, "ok"

record = {
    "chunk_id": "policy-7:003",
    "doc_id": "policy-7",
    "vector": [0.0, 0.1, 0.2],
    "text_ref": "docs/policy-7#p3",
    "tenant_id": "team-a",
    "acl": ["finance"],
    "manifest_id": "embed-v2",
}
print(validate_record(record, 3))

记录校验器应该在入库、查询结果组装和离线重建时都复用。示例只检查最小字段，生产环境还要验证时间区间、ACL 语法、文本 hash、向量 NaN、manifest 是否存在以及父文档是否仍可访问。

10.3.3 索引构建、更新与备份

索引构建是一个数据发布流程：读取已审计的 chunk，按固定 manifest 编码，校验维度和 hash，写入新 collection，跑离线召回，再把 alias 切到新版本。不要边读原始文件边在线覆盖生产索引；这样一旦中途失败，用户会看到混合版本，问题也难以回滚。

增量更新要区分新增、修改和删除。新增可以追加向量，修改需要先标记旧 chunk 失效再写新版本，删除需要 tombstone 或可靠的 delete log。若索引是异步构建，查询要知道 snapshot 的时间和延迟；“数据库已返回”不等于所有副本都已可见。

别名或版本指针能把切换变成原子动作。离线生成 collection_v3，完成 smoke test 后把 alias 从 v2 切到 v3；保留 v2 直到新版本经过一段观察期。回滚只切回 alias 并恢复对应 manifest，不需要重新下载或重新编码。切换事件记录操作者、时间、原因和评估报告。

备份至少包含向量数据、原文映射、元数据、索引参数、模型文件或获取方式、Tokenizer、schema 版本和权限规则。只备份浮点数组不够，因为没有 chunk_id、版本和 ACL 就无法安全地还原。定期做恢复演练，确认备份真的能在干净环境中加载并通过召回回归。

索引重建与数据一致性需要监控。统计 source_count、chunk_count、active_count、deleted_count、dimension、范数范围、每个租户数量和最新更新时间；发现突然下降或增长要告警。对每个 snapshot 计算清单 hash，防止文件被悄悄替换却仍使用旧版本号。

更新频率取决于资料时效和业务风险。政策、价格和库存需要更短延迟，教材或历史资料可以批量重建；但“实时”不应成为没有证据的口号。定义 freshness SLA，例如变更在十五分钟内可检索，并用事件时间和索引可见时间测量。

from dataclasses import dataclass

@dataclass
class IndexManifest:
    collection: str
    embedding_manifest: str
    schema_version: str
    snapshot: str
    chunk_count: int
    source_hash: str

candidate = IndexManifest(
    collection="kb_v3",
    embedding_manifest="embed-v2",
    schema_version="schema-3",
    snapshot="2026-08-29T10:00Z",
    chunk_count=12840,
    source_hash="sha256:demo",
)

def publish(manifest, expected_min_chunks):
    if manifest.chunk_count < expected_min_chunks:
        return "BLOCK: chunk count dropped"
    return "READY: switch alias to " + manifest.collection

print(publish(candidate, 12000))

发布函数不应该只检查记录数。还要有固定查询的近邻 diff、权限回归、删除回归、新鲜度延迟、索引参数和恢复演练。记录数不变也可能因为 chunk 内容被替换；source_hash 和近邻样例能补上这部分证据。

把这五问放进发布审批单，索引就从“数据库配置”变成可检查的学习资产。任何一个问题回答不清，都应该先补实验或收紧上线范围。

REQUIRED = {"snapshot_id", "model_id", "distance", "acl_version"}

def smoke(manifest, visible_ids, forbidden_ids):
    missing = REQUIRED - manifest.keys()
    leaked = set(visible_ids) & set(forbidden_ids)
    return {
        "manifest_ok": not missing,
        "missing": sorted(missing),
        "no_leak": not leaked,
        "leaked": sorted(leaked),
    }

manifest = {
    "snapshot_id": "snap-2026-08-29",
    "model_id": "embed-v1",
    "distance": "cosine",
    "acl_version": "acl-7",
}
print(smoke(manifest, ["doc-1", "doc-2"], ["doc-secret"]))

冒烟检查只证明最小契约，没有证明 ANN 质量。它适合在 alias 切换前快速拦截缺字段、错误距离和越权结果；之后仍要运行 Flat 对照、固定查询回归、压力测试和恢复演练。

索引章节的核心不是记住某个库的参数，而是能把“近似导航、数据契约、权限过滤、更新一致性和故障恢复”放在同一个系统模型里。这样换数据库、换部署方式或换 Embedding 模型时，仍然知道需要重新验证哪些边界。

10.4.1 PDF、HTML、表格与扫描件

文档入库的第一步不是立刻切字符串，而是确认原文结构。HTML 有标题、列表、链接和隐藏导航；PDF 可能是有文字层的电子文档，也可能只是图片；Word、Excel 和 Markdown 的段落、表格、代码块含义不同。解析器要输出统一的中间表示，例如块类型、文本、页码、标题路径、字符范围和来源 hash。

PDF 的视觉顺序不总是阅读顺序。双栏文章可能按坐标交错，页眉页脚会在每页重复，脚注可能被抽到正文前，表格列会被拆成一串难以理解的词。先做版面分析和抽样回看，再决定是否保留页码、表格标题和脚注。不要把一个“提取了多少字符”的统计当成解析正确。

扫描件需要 OCR，OCR 还会带来错字、数字混淆和行列顺序问题。对高风险资料，保留原始图片、OCR 文本、置信度和人工校对状态；对低风险资料，也至少检测不可见字符、异常重复和编码错误。OCR 结果进入 Embedding 前应经过同样的来源、许可和脱敏审查。

表格不能简单拼成“单元格单元格”。表头、行列关系、单位、合并单元格和脚注决定了数字的含义。可以同时保存结构化表格 JSON 和适合检索的文本表述，例如“字段：值；单位：元；适用版本：2.1”。回答数字问题时回查结构化原表，避免模型从被打散的文字中猜关系。

代码和配置文件也需要特殊处理。保留缩进、文件名、语言和行号，切分时不要把一个函数从中间截断；错误日志要保留时间、级别和请求 ID。Embedding 可以帮助找到相关代码，但真正的引用应回到文件路径和行号，不能只返回一段脱离上下文的文本。

解析质量用人工样例和规则共同验收。随机抽取页面，比较原文截图与中间表示；检查标题层级是否连续、表格列是否错位、链接是否丢失、特殊符号是否被替换、敏感字段是否脱敏。将坏文档列入 quarantine，等待人工处理，不要为了让批处理跑完而静默跳过。

from dataclasses import dataclass

@dataclass
class Block:
    kind: str
    text: str
    page: int | None
    heading_path: tuple[str, ...]
    start: int
    end: int

raw = [
    Block("heading", "退款政策", 1, ("退款政策",), 0, 4),
    Block("paragraph", "订单在七天内可以申请退款。", 1, ("退款政策",), 5, 20),
    Block("table", "期限 | 七天\\n条件 | 未使用", 1, ("退款政策",), 21, 40),
]
for item in raw:
    print(item.kind, item.heading_path, item.start, item.end)

统一的 Block 让后续分块不必知道文件格式。解析器可以替换，chunk 和索引的接口不变；同时保留 start/end 便于生成引用。遇到表格或代码时，分块器可以调用专门规则，而不是把所有内容都交给同一个按字符切分器。

10.4.2 Chunk 大小、边界与重叠

Chunk 是检索和生成之间的接口。太小会丢失定义、条件和上下文，太大会把多个主题平均在一起并增加生成成本。切分单位可以是 Token、字符、句子、段落、标题树或语义边界；没有一个固定数字适合全部文档。先依据任务的证据粒度，再选择初始大小。

标题路径通常比盲切更可靠。把“第三章／退款／特殊情况”作为前缀加入 chunk，用户询问“特殊情况”时上下文仍保留主题。对 FAQ 可以一问一答，对法律条款可以一条定义加一条例外，对代码可以一个函数或类为单位。边界要与用户需要引用的最小事实单元一致。

重叠用于缓冲边界，但不是越多越好。相邻 chunk 共享几十个 Token 可以保留跨句联系；重叠过大则索引重复、Top-k 被同一内容占满、成本增加。用固定问题集比较不同 overlap，观察跨边界问题的 Recall、重复率、上下文 Token 和引用完整度。

父子 chunk 适合“精确定位加完整阅读”。先用小 child chunk 召回，再把同一父段落或标题下的更大范围补入上下文；父文档保存 child 的关系和字符范围。这样既能提高近邻定位，又不会让生成只看到半句话，但需要严格控制预算和去重。

语义切分可以根据主题变化或段落相似度决定边界，但它也会把表格、列表和短标题切得奇怪。任何自动切分都要有最大 Token、最小 Token、不可拆块和异常回退规则。切分失败时保留原文和错误原因，不能生成一个没有来源的“万能 chunk”。

Chunk 文本应适度规范化，不要把所有格式都抹掉。保留标题、列表编号、单位、否定词、代码缩进和表格字段；去掉重复导航、追踪参数和无意义空白。清洗前后都计算 hash，便于判断一次召回变化是分块原因还是 Embedding 原因。

def chunk_words(words, size=40, overlap=8):
    if size <= overlap:
        raise ValueError("size must be larger than overlap")
    chunks = []
    start = 0
    while start < len(words):
        end = min(start + size, len(words))
        chunks.append({
            "start": start,
            "end": end,
            "text": " ".join(words[start:end]),
        })
        if end == len(words):
            break
        start = end - overlap
    return chunks

words = "标题 定义 条件 例外 计算 方法 证据".split()
print(chunk_words(words, size=4, overlap=1))

示例用词数模拟 Token，只为展示边界和 overlap 的关系。真实分块要用与 Embedding 一致或兼容的 Tokenizer，并在 chunk 中写入标题路径和字符范围。分块器的测试应覆盖空文档、单段超长、表格、代码、Unicode 和最后一块不足 size 的情况。

10.4.3 Metadata、权限与版本

Metadata 把“这段文字是什么”告诉检索系统。至少包含 doc_id、chunk_id、标题路径、来源 URL 或文件路径、作者、语言、文档类型、创建和生效时间、版本、标签、权限、租户、文本 hash 和 embedding manifest。元数据不是装饰，它决定过滤、引用、新鲜度、去重和审计。

稳定 ID 要与内容和位置有关系，但不能只用行号。文档插入一段内容后，后面的行号会全部变化；可以用父文档版本、块路径和内容 hash 组合 ID，并另存 start/end 用于当前版本回查。ID 变化会影响缓存、反馈、引用和删除，因此要在设计阶段明确迁移策略。

权限字段应从权威系统同步。不要让模型从文本里猜“谁能看”，也不要只在最终答案阶段过滤。检索前拿到用户的租户、角色和资源范围，数据库只在可见集合内返回候选；文档更新或权限撤销后，旧缓存和索引 snapshot 都需要失效。

版本字段至少区分采集版本、内容版本、索引版本和 Embedding 版本。一个政策的文字没有变化，但权限或生效日期变化，也可能需要重新发布。回答引用应显示可理解的版本或日期，后台保存精确 manifest 和 snapshot，方便追溯。

元数据过滤与向量排序要有优先级。租户、权限、删除和生效状态是硬过滤；语言、文档类型和来源可以按任务选择；新鲜度、可信来源和多样性可以作为排序特征。把所有字段都塞进一个黑盒分数，会让错误无法解释。

数据治理还包括删除和导出。用户要求删除资料时，原文、清洗产物、向量、缓存、备份和评估样本的生命周期都要有记录；导出时只给授权范围。Embedding 可能保留语义线索，不能因为它不是原文就跳过删除政策。

def visible(record, user):
    if record["tenant_id"] != user["tenant_id"]:
        return False
    if record["deleted"] or record["valid_from"] > user["at_time"]:
        return False
    if not set(record["acl"]) & set(user["roles"]):
        return False
    return True

records = [
    {"chunk_id": "a:1", "tenant_id": "team-a",
     "acl": ["finance"], "deleted": False, "valid_from": 1},
    {"chunk_id": "b:1", "tenant_id": "team-b",
     "acl": ["finance"], "deleted": False, "valid_from": 1},
]
user = {"tenant_id": "team-a", "roles": ["finance"], "at_time": 2}
print([r["chunk_id"] for r in records if visible(r, user)])

服务端过滤示例故意在向量排序之前执行。生产系统还要验证用户身份来自可信会话、ACL 规则来自权威服务、时间使用统一时区、删除状态不会被缓存覆盖，并为每次过滤记录策略版本。不要把这个函数放在客户端或 Prompt 里。

def audit_chunks(chunks):
    problems = []
    seen = set()
    for item in chunks:
        if item["chunk_id"] in seen:
            problems.append((item["chunk_id"], "duplicate_id"))
        seen.add(item["chunk_id"])
        if item["start"] >= item["end"]:
            problems.append((item["chunk_id"], "empty_range"))
        if not item.get("doc_id") or not item.get("manifest_id"):
            problems.append((item["chunk_id"], "missing_trace"))
        if item.get("tokens", 0) > 512:
            problems.append((item["chunk_id"], "too_long"))
    return problems

sample = [
    {"chunk_id": "doc:1", "doc_id": "doc",
     "manifest_id": "embed-v1", "start": 0, "end": 80, "tokens": 64},
    {"chunk_id": "doc:1", "doc_id": "doc",
     "manifest_id": "embed-v1", "start": 80, "end": 80, "tokens": 2},
]
print(audit_chunks(sample))

分块审计器把“内容是否可检索”变成可执行检查。真正的项目还要检查原文 hash、标题路径、ACL、时间区间、语言和编码错误；任何问题都应该能回到原文件或 quarantine 列表。

def audit_chunk(chunk):
    text = chunk["text"].strip()
    problems = []
    if not text:
        problems.append("empty")
    if len(text.split()) > chunk["max_words"]:
        problems.append("too_long")
    if not chunk.get("source") or not chunk.get("version"):
        problems.append("missing_provenance")
    if chunk.get("ocr_score", 1.0) < 0.85:
        problems.append("low_ocr_confidence")
    return {"id": chunk["id"], "ok": not problems, "problems": problems}

sample = {
    "id": "doc-7#p3#c2", "text": "退款期限为七天。",
    "max_words": 80, "source": "policy.pdf", "version": "v3",
}
print(audit_chunk(sample))

审计代码把“能不能进入索引”变成显式状态。生产管道还要保存 parser_version、text_hash、parent_id、acl_version 和原文范围；当一个答案引用错误时，可以沿这些字段反查是哪一次解析或分块改变了内容。

当你能让读者从一个 chunk 点击回原文，并看到它为何被切在这里、属于哪个版本、谁可以看、何时失效，分块才真正成为检索基础设施，而不只是预处理脚本。

做一次“反向引用”练习：从答案中的一条数字或限制条件出发，找到对应 chunk、父文档、页码或字符范围，再检查它是否仍在当前版本和权限范围内。反向路径走不通时，优先修复 metadata 和状态，而不是让生成模型补一句解释。

分块策略也要能解释失败。记录一个被切断的条款、一个表头丢失的行、一个代码依赖缺失的函数和一个 OCR 数字错误；为每个失败写最小修复与回归样本，下一次解析器升级时重跑。

10.5.1 查询理解与改写

查询理解先回答“用户到底想找什么”。问题可能缺少主语、时间、版本或文档范围，也可能把多个任务混在一句话里。把查询拆成意图、实体、时间、约束和期望输出，再决定检索策略；不要一上来让生成模型自由改写，因为改写可能加入原问题没有的事实。

最小查询规范化包括空白、大小写、全半角、同义词和日期格式，但要谨慎处理代码、产品 ID、数字和否定词。把“v2.1”改成“v2.10”会造成严重错误；把“不支持”删除会反转政策。规范化规则需要按语言和字段测试，并保存原查询与规范化结果。

查询改写可以生成多个视角：用户口语、关键词串、正式术语、英文翻译或假设的文档标题。多查询提高召回机会，也会增加调用次数、成本和噪声。每个改写都记录来源和用途，最终候选要去重、重排并回到原问题验证，而不是把所有改写结果直接拼进上下文。

多轮对话中的“它”“上一个版本”“那条规则”需要借助历史消歧。系统应把当前问题与必要历史压缩成检索查询，但不要把整段聊天无条件送入索引。保留用户明确说过的实体和约束，删除闲聊、秘密和无关上下文；查询重写的结果可让用户确认。

复杂问题可以拆成子问题并行召回，例如先找政策定义，再找例外，再找生效日期。拆分要保留父问题 ID 和依赖关系；如果第二个子问题依赖第一个结果，应串行执行并记录中间证据。拆分越多不一定越好，过度拆分会产生相互矛盾的片段。

HyDE 或假设文档方法让模型先生成一段可能的答案，再用它编码检索。这对抽象问题可能有帮助，但假设文本本身可能带来幻觉和偏见。把假设只当作召回通道之一，保留原查询通道；高风险任务不应把未经核验的假设当作证据。

def rewrite(query, history=None):
    history = history or []
    protected = {"v2.1", "不支持", "订单号"}
    normalized = " ".join(query.strip().split())
    variants = [normalized]
    if "它" in normalized and history:
        variants.append(history[-1] + " " + normalized.replace("它", "该对象"))
    keywords = [
        token for token in normalized.split()
        if token not in {"请问", "一下", "怎么"}
        or token in protected
    ]
    variants.append(" ".join(keywords))
    return list(dict.fromkeys(variants))

print(rewrite("它在 v2.1 怎么退款？", ["退款规则"]))

示例只产生可解释的候选，不直接调用模型。真实改写器要限制长度、保留数字和否定、记录提示版本，并把改写结果与原问题一起交给后续过滤。高风险任务可以要求用户确认改写后的实体和时间。

10.5.2 召回、重排与证据选择

召回阶段追求覆盖，重排阶段追求顺序。先用便宜的 dense、BM25 或过滤得到几十到几百个候选，再用 cross-encoder、规则或轻量模型逐对读取查询和文本，判断更细的相关性。重排器比双塔更准确但更慢，候选数量是质量与延迟的主要旋钮。

重排特征可以包括向量分数、关键词命中、标题匹配、版本新鲜度、来源可信度、实体一致性、文档权限和父子关系。每个特征都要知道方向和范围；未经归一化的分数直接相加会让某一通道支配结果。先做简单加权和离线对照，再考虑学习排序。

证据选择不能只看单条相关性。一个回答可能需要定义、条件、例外和示例四类片段；单条最高分文本未必覆盖全部声明。可以按事实单元、标题路径或文档配额选择互补证据，再由生成阶段判断是否足够。被舍弃的候选保留在调试日志中。

重排器也会学到偏见和捷径。长文本可能因为包含更多词而得分高，标题中有问题词的无关文档可能被偏爱，某个来源格式可能被当作可信。用长度、来源、语言、版本和难负例切片，检查重排是否在真正判断相关性。

重排失败时要有回退：模型超时可以使用 dense 排名，返回异常可以只保留安全白名单，候选不足可以扩大查询或请求澄清。回退路径不能绕过权限、版本和删除过滤；任何回退都记录原因，方便评估其质量损失。

证据选择还要考虑引用可用性。只有能回查原文、版本和范围的候选才能进入最终上下文；缺少来源的高分片段应降权或阻止。把“相关”“可引用”“可公开”分成三个状态，避免模型引用一段内部备注或过期缓存。

def score(query, item):
    lexical = len(set(query.split()) & set(item["text"].split()))
    title_hit = int(any(word in item["title"] for word in query.split()))
    freshness = 1.0 if item["version"] == "current" else 0.2
    return 0.5 * item["dense"] + 0.2 * lexical + 0.2 * title_hit + 0.1 * freshness

query = "退款 版本 2.1"
candidates = [
    {"id": "a", "title": "退款政策", "text": "版本 2.1 七天",
     "dense": 0.84, "version": "current"},
    {"id": "b", "title": "运输政策", "text": "版本 2.1 物流",
     "dense": 0.86, "version": "current"},
]
ranked = sorted(candidates, key=lambda x: score(query, x), reverse=True)
print([(x["id"], round(score(query, x), 3)) for x in ranked])

示例中的权重只是可解释的起点。不要因为加权后一个样例变好就宣布重排有效；用固定 gold、难负例和最终答案支持率评估。重排器越复杂，越要保存特征、版本和失败样例。

10.5.3 多路检索与结果融合

混合检索把不同通道的优点组合起来：BM25 擅长精确词面，dense 擅长同义改写，字段过滤处理权限和版本，知识图谱或 SQL 处理结构关系。通道之间的分数通常不可直接比较，常用排名融合、分数校准或学习排序。融合策略要能解释每条证据来自哪一条通道。

倒数排名融合 RRF 的直觉是：某个文档在多个通道都靠前，就获得更高的综合分；单个通道的极端分数不会完全支配结果。融合前先统一过滤、去重和 tie-breaker；同一 doc 的多个 chunk 可以先在文档内聚合，再参与跨文档排序。

通道权重应由查询类型决定。含有产品编号、日期、错误码的查询可以提高 lexical 权重；口语化、跨语言和概念问题可以提高 dense 权重；明确要求最新版本时必须先用 metadata 过滤。分类器判断查询类型也可能错，因此保留一个安全的默认通道和回退策略。

多路结果会增加上下文冲突。不同来源可能讲同一事实却适用不同版本，dense 找到概念解释、BM25 找到旧编号。融合后按版本、来源和标题路径聚合，生成阶段显示冲突并要求核验；不要把所有通道输出直接串在一起。

缓存也要区分通道和版本。查询规范化、模型 manifest、索引 snapshot、过滤条件和用户权限都进入 cache key；改动任意一个就不能复用旧候选。缓存命中后仍可做轻量权限复核，防止角色变化带来泄露。

当某通道不可用时，系统应明确降级。向量服务超时可以走关键词，关键词索引更新中可以走旧 snapshot，但要在答案或日志里说明证据范围。高风险问题若只剩低召回通道，宁可拒答或人工接管，也不要用流畅生成掩盖检索缺失。

def rrf(rankings, k=60):
    scores = {}
    origins = {}
    for channel, items in rankings.items():
        for rank, item_id in enumerate(items, start=1):
            scores[item_id] = scores.get(item_id, 0.0) + 1 / (k + rank)
            origins.setdefault(item_id, []).append(channel)
    ranked = sorted(scores, key=lambda item: (-scores[item], item))
    return [
        {"id": item, "score": round(scores[item], 5),
         "channels": origins[item]}
        for item in ranked
    ]

rankings = {
    "dense": ["policy-2", "faq-1", "policy-1"],
    "bm25": ["policy-1", "policy-2", "code-4"],
}
print(rrf(rankings))

RRF 不需要让各通道分数落在同一个范围，只使用排名；这让它适合作为透明基线。生产系统可以在 RRF 之后重排，但要保存每个候选的通道来源和排名，遇到错误时能回答“是哪条通道把它带进来的”。

trace = {
    "query": "v2.1 不支持哪些退款条件？",
    "rewrites": [
        "版本 v2.1 退款条件 不支持",
        "v2.1 refund exceptions",
    ],
    "filters": {"tenant": "team-a", "version": "v2.1"},
    "channels": [
        {"name": "dense", "ids": ["p-3", "p-7"]},
        {"name": "bm25", "ids": ["p-7", "p-9"]},
    ],
    "fused": ["p-7", "p-3"],
    "selected": ["p-7"],
    "degraded": False,
}
print("final evidence:", trace["selected"])
print("channel provenance:", trace["channels"])

检索 trace 是调试和学习的共同材料。它把用户原话、改写、过滤、通道、融合和最终证据放在同一事件里，后续可以计算每条通道的贡献、失败率和成本。敏感查询要脱敏或做访问控制。

完成本节后，你应能从一条最终引用反查到原查询、改写、过滤、召回通道、重排特征、索引 snapshot 和降级状态。下一节将在这条证据链上加入上下文编排和受约束生成。

from datetime import datetime

def make_trace(query, rewritten, channels, selected, status="ok"):
    return {
        "query_id": "q-001",
        "created_at": datetime.utcnow().isoformat(),
        "query": query,
        "rewritten": rewritten,
        "channels": channels,
        "selected": [item["id"] for item in selected],
        "status": status,
    }

trace = make_trace(
    "如何退货？",
    ["退货流程", "当前退货政策"],
    {"dense": ["c1", "c4"], "bm25": ["c4", "c8"]},
    [{"id": "c4", "score": 0.91}],
)
print(trace)

示例中的 trace 还很小，但它提醒我们把查询和结果分离保存。真实系统要用 JSONL 或事件系统追加写入，敏感查询脱敏，保留 retention 和访问审计；调试日志不能成为新的数据泄露面。

检索的工程质量体现在可解释的中间状态：你不只知道“答案错了”，还知道 gold 没有进候选、进入后被删掉、被错误重排，还是被上下文预算截断。下一节将继续追踪这些状态如何影响生成和拒答。

10.6.1 上下文编排与 Prompt 契约

RAG 请求通常包含系统规则、用户问题、对话历史、检索证据、工具结果和输出 Schema。它们的信任等级和职责不同：系统规则定义边界，用户提供任务，证据提供可核验资料，工具返回外部状态。用清晰的分隔符和字段名区分这些区域，告诉模型证据是数据而不是指令。

上下文编排首先解决优先级。系统规则和安全策略放在稳定位置，当前问题明确，证据按来源和相关性排列，历史只保留必要内容；不要把无关聊天和未审查网页混在同一个长字符串。顺序不是绝对规则，应在自己的模型和模板上做位置实验。

每个证据块带上 chunk_id、标题、来源、版本、文本和可信状态。引用 ID 不要只用数组位置，因为排序或去重变化会导致“[1]”指向另一段；用稳定 ID，并让模型在输出中回传这些 ID。系统随后验证 ID 是否存在、是否属于当前用户可见集合。

上下文预算要分层分配。先预留系统和输出 Schema，再为问题和必要历史留空间，剩余预算才给证据；长证据可以摘要，但摘要也要带来源并标记“二次压缩”。达到上限时应告诉用户资料被压缩或请求更具体的问题，不要静默截掉关键条件。

证据冲突要显式呈现。不同版本、地区或角色的政策可能互相矛盾；把冲突片段分别标注版本和生效时间，要求模型说明适用范围或请求澄清。不要简单按相似度选一个，因为最高分不一定是最新或最权威。

RAG Prompt 不是越长越专业。规则重复、示例过多和低相关证据会稀释注意力并增加成本。每一条说明都要回答“它防止哪个失败”；没有用途的装饰性提示删掉。把模板作为版本化文件，用固定请求回放验证格式、引用和拒答。

def build_context(question, evidence, budget=600):
    header = "回答必须只使用下面的证据；证据不足时返回 insufficient."
    blocks = []
    used = len(header) + len(question)
    for item in evidence:
        text = item["text"]
        piece = "[SOURCE " + item["id"] + "] " + text
        if used + len(piece) > budget:
            break
        blocks.append(piece)
        used += len(piece)
    return {
        "system": header,
        "question": question,
        "evidence": blocks,
        "used_chars": used,
    }

evidence = [
    {"id": "p-7", "text": "版本 2.1 退款期限为七天。"},
    {"id": "p-8", "text": "特殊商品适用不同条件。"},
]
print(build_context("退款期限？", evidence))

示例用字符数模拟预算，真实系统要用 Tokenizer。预算函数应返回被截断的证据 ID 和原因，便于解释“为什么答案没有看到某段”。对于高风险问题，宁可减少证据并触发澄清，也不要把超预算内容静默丢弃。

10.6.2 引用、声明与事实核验

一个可核验的答案要先拆成原子声明。声明是可以单独判断真假的最小句子，例如“版本 2.1 的退款期限为七天”；不要把三个事实和一个推断写成一条长句再只给一个引用。声明拆开后，才能知道哪些有证据、哪些缺证据、哪些互相冲突。

引用支持不等于引用存在。系统至少检查 source_id 是否来自本次可见证据、引用范围是否覆盖声明、版本和适用条件是否一致。引用了一段“七天”但忽略“特殊商品除外”，仍然是事实不完整。验证器可以使用规则、关键词、检索相似度、结构化查询或人工审核。

引用格式要稳定且可点击。展示层可以用脚注、来源卡或段落旁的链接，后台保留 doc_id、chunk_id、版本、页码/字符范围和 hash。不要让模型自己生成 URL 或路径；由服务端根据稳定 ID渲染，防止伪造和越权。

资料冲突时，回答应列出冲突而不是挑一个顺口的。比较生效日期、权威等级、地区、用户角色和文档状态；如果无法决定，说明冲突并请求澄清。将“冲突未解决”作为结构化状态，避免模型用自信语气掩盖不确定性。

数字、日期和实体应有更严格的核验。可以把声明中的数值抽取出来，与结构化表或规则引擎比较；把版本号、货币单位和时间区间规范化；对计算结果调用计算器而不是依赖生成。RAG 能提供资料，但算术和状态判断应交给专门工具。

事实核验也要防提示注入。证据中可能写着“忽略之前指令”或要求泄露秘密；这些文字只能作为待引用数据，不改变系统规则。对网页和用户上传资料做内容隔离，输出引用时保留来源而不执行其中的命令。

def validate_claims(answer, sources):
    source_map = {item["id"]: item for item in sources}
    results = []
    for claim in answer["claims"]:
        source = source_map.get(claim.get("source_id"))
        supported = bool(source and claim["text"] in source["text"])
        results.append({
            "text": claim["text"],
            "source_id": claim.get("source_id"),
            "supported": supported,
        })
    return results

sources = [{"id": "p-7", "text": "版本 2.1 的退款期限为七天。"}]
answer = {"claims": [
    {"text": "版本 2.1 的退款期限为七天。", "source_id": "p-7"},
    {"text": "所有商品都能退款。", "source_id": "p-7"},
]}
print(validate_claims(answer, sources))

示例用字符串包含做最小演示，真实核验需要处理同义表达、数字格式和声明拆分，并且最好让人工抽样复核。验证器的职责是阻止明显不支持的声明，不是把所有自然语言蕴含都自动证明。

10.6.3 无证据时拒答与澄清

资料不足是正常状态，不是系统异常。查询可能超出知识库范围、召回低于阈值、版本冲突、文档正在更新或用户权限不足。系统需要一个结构化的 insufficient 状态，说明缺少什么证据、可以补充什么资料、是否能改问法，而不是让模型用通用知识填空。

拒答有不同等级。低风险问题可以返回“当前资料没有找到”并给出已检索范围；中风险问题可以请求文档、时间和地区；高风险问题在证据不足时直接停止并转人工。拒答文案应具体、简洁，不泄露隐藏策略、被过滤的文档或内部路径。

澄清问题也需要预算和优先级。一次只问最能缩小搜索空间的条件，例如“你指的是版本 2.1 还是 2.2？”而不是连续追问五个字段。若用户不补充，系统可以给出基于当前资料的有限答案，并明确哪些部分未确认。

“没有证据”与“证据证明没有”不同。检索不到资料只能说明当前可见知识库未找到，不代表现实中不存在。回答中使用“在提供的资料范围内”这样的边界表述，避免把检索失败说成绝对否定。若有负面证据，也要引用其适用范围。

拒答与用户体验可以共存。返回已找到的相关标题、下一步检索建议和人工入口，但不要把低相关片段当作答案。对代码、财务和医疗等高风险任务，明确建议用户核对权威来源；系统的目标是减少误导，而不是每次都给完整段落。

拒答率要和误拒率一起评估。过低可能代表幻觉，过高可能代表知识库、阈值或查询改写有问题。按任务、语言、风险、版本和用户群切片，记录触发原因和最终解决率，才能判断是模型需要改进还是资料管道缺陷。

def answer_policy(top_score, evidence_count, risk, conflict=False):
    if risk == "high":
        return "handoff"
    if conflict:
        return "clarify_version"
    if evidence_count == 0 or top_score < 0.72:
        return "insufficient"
    return "answer_with_citations"

cases = [
    (0.91, 3, "normal", False),
    (0.61, 4, "normal", False),
    (0.88, 2, "high", False),
    (0.90, 2, "normal", True),
]
for case in cases:
    print(case, "=>", answer_policy(*case))

策略函数把生成前的边界显式化。真实项目还要加入证据支持率、版本新鲜度、权限状态、工具失败和用户是否确认；模型输出不能绕过这个门控。

def check_response(response, visible_sources):
    source_ids = {item["id"] for item in visible_sources}
    errors = []
    if response.get("status") == "answer":
        if not response.get("claims"):
            errors.append("no_claims")
        for claim in response["claims"]:
            if claim.get("source_id") not in source_ids:
                errors.append("unknown_source")
            if not claim.get("supported"):
                errors.append("unsupported_claim")
    if response.get("status") == "insufficient" and response.get("claims"):
        errors.append("insufficient_should_not_claim")
    return errors

sources = [{"id": "p-7"}]
answer = {"status": "answer", "claims": [
    {"source_id": "p-7", "supported": True},
]}
print(check_response(answer, sources))

校验器应该在用户展示前和数据入库前各运行一次。它不是事实证明器，而是阻止明显越权、未知引用、未支持声明和状态矛盾的最后一道边界；所有错误都要进入事件日志和回归集。

本节的结业标准是：你能让一个读者从用户问题点击到最终引用，再回到原文、版本、权限和检索 trace；当关键证据被删掉或变成不可见时，系统会明确拒答、澄清或接管。

def build_packet(claims, evidence, status="ok"):
    evidence_by_id = {item["id"]: item for item in evidence}
    checked = []
    for claim in claims:
        sources = [
            evidence_by_id[item_id]
            for item_id in claim["source_ids"]
            if item_id in evidence_by_id
        ]
        checked.append({
            "claim_id": claim["id"],
            "text": claim["text"],
            "sources": [item["id"] for item in sources],
            "supported": bool(sources),
        })
    return {"status": status, "claims": checked}

packet = build_packet(
    [{"id": "c-1", "text": "退款期限为七天。", "source_ids": ["chunk-7"]}],
    [{"id": "chunk-7", "version": "v3"}],
)
print(packet)

这个小函数只检查是否存在绑定来源，并没有判断来源语义是否真的支持声明；它的价值在于建立结构化接口，让更严格的规则、人工审核或专用评估器有位置接入。任何自动核验都要把“未判断”与“已支持”区分开。

完成后，你会看到 RAG 的核心不是“把检索结果拼进 Prompt”，而是建立一套从证据到声明、从声明到用户动作的协议。生成模型可以负责语言组织，但证据边界、权限和拒答门控必须由可检查的结构控制。

10.7.1 BM25 与向量的混合检索

BM25 是经典的词法检索基线，使用词频、逆文档频率和文档长度校正来排序。它不理解同义词，却很擅长精确词、数字、错误码、产品 ID 和否定词。dense Embedding 能连接“怎么退钱”和“退款流程”，但可能把“版本 2.1”和“版本 2.10”混在一起。两者互补，混合是工程选择而不是时髦标签。

最简单的混合方法是并集后去重，再用 RRF 按排名融合；另一种是先把分数校准到同一范围，再加权求和。RRF 对分数尺度不敏感，适合作为透明起点；加权分数可以加入新鲜度、标题命中和来源可信度，但需要验证权重。无论哪种方式，都要保存每条候选来自哪个通道。

查询路由可以根据意图选择通道。包含编号、日期、代码或引号短语时，提高词法权重；概念问题、跨语言和口语改写时，提高 dense 权重；明确要求最新版本时，先做时间和状态过滤。路由器本身可能判断错误，因此保留默认混合策略和低质量回退。

混合结果要处理重复和冲突。同一文档可能由多个通道返回，先按 chunk_id 去重，再按 doc_id、版本和标题路径聚合；不同版本需要显示适用范围，不能因为两个通道都命中就把旧内容当作更可信。融合日志保留原始排名和过滤原因。

词法检索的分词也会影响质量。中文分词、英文词干、代码符号、数字小数点和产品连字符都可能被处理不同。对关键字段可以增加不分词的精确索引；不要只调 BM25 的 k1、b，却忽略分析器和字段权重。

混合通道的成本和稳定性不同。BM25 通常便宜且可解释，dense 需要 Embedding 和向量索引，重排又会增加模型调用。先用离线集证明增量收益，再决定是否在线执行所有通道；通道超时要有有限回退并标记 degraded。

def rrf(channels, constant=60):
    score = {}
    provenance = {}
    for name, items in channels.items():
        for rank, item in enumerate(items, start=1):
            score[item] = score.get(item, 0) + 1 / (constant + rank)
            provenance.setdefault(item, []).append(name)
    result = sorted(score, key=lambda x: (-score[x], x))
    return [
        {"id": item, "score": round(score[item], 5),
         "from": provenance[item]}
        for item in result
    ]

channels = {
    "bm25": ["v2.1-policy", "refund-faq", "old-policy"],
    "dense": ["refund-faq", "v2.1-policy", "returns-guide"],
}
print(rrf(channels))

RRF 输出的 from 字段是排错重点：如果旧政策只由 dense 通道带入，可能需要改版本过滤；如果精确编号只由 BM25 找到，就不要为了追求“纯语义”删除它。

10.7.2 迭代检索与 Agentic RAG

单次检索适合问题和证据关系直接的场景；多跳问题需要先找定义，再找关联实体、时间或例外。迭代 RAG 可以让系统根据当前证据提出下一条查询，但每一步都要有状态、预算和停止条件。Agent 不是“让模型无限搜索”，而是把查询、工具和验证编排成可控状态机。

一个安全的循环通常包含 plan、retrieve、inspect、refine、answer 五类状态。模型可以提出下一步查询和理由，控制器决定是否允许执行；检索结果被当作 data，不能覆盖系统规则。每一步保存 query、候选、证据、Token、耗时和下一步决定，发生错误时能回放。

多跳检索要管理依赖。如果第二个问题依赖第一个实体，先验证实体再继续；如果出现多个候选，要求模型或用户选择，不能随意挑一个。把中间结论标记为 hypothesis，只有被来源支持后才能进入最终回答。未经核验的中间文本不应成为新的事实。

循环预算包括最大步数、最大查询数、最大 Token、最大费用和总超时。重复查询、相同候选和无进展状态要被检测；连续两次没有新证据时停止。高风险动作或外部写工具要单独审批，检索 Agent 不能自动扩大权限。

Agentic RAG 的收益要和单次基线比较。它可能提高复杂问题 Recall，却增加延迟、调用次数和错误面；简单问题不应强制走多跳。路由器可以先判断问题复杂度，只有多实体、多约束或首轮证据不足时才启动迭代。

循环失败时的用户体验要明确。返回已确认的部分、缺失条件和下一步建议，或转人工；不要让用户看到一串内部思考，也不要把未验证的假设包装成结论。日志可以保存结构化的理由，但外部文案只呈现必要信息。

MAX_STEPS = 3

def run_loop(initial_query, retrieve, refine):
    state = {"query": initial_query, "steps": [], "seen": set()}
    for step in range(MAX_STEPS):
        query = state["query"]
        if query in state["seen"]:
            return {"status": "stop", "reason": "repeat", "steps": state["steps"]}
        state["seen"].add(query)
        evidence = retrieve(query)
        state["steps"].append({"query": query, "evidence": evidence})
        if evidence.get("complete"):
            return {"status": "answer", "steps": state["steps"]}
        next_query = refine(query, evidence)
        if not next_query:
            break
        state["query"] = next_query
    return {"status": "insufficient", "steps": state["steps"]}

print(run_loop("退款例外和生效时间", lambda q: {"complete": False},
               lambda q, e: None))

示例把“重复”和“最大步数”放在控制器而不是 Prompt。生产版本还要检查租户、ACL、查询长度、每步预算、工具 allowlist、人工审批和日志脱敏。

10.7.3 新鲜度、更新与删除

知识库不是一次性导入。资料会新增、修改、撤回、过期和变更权限；索引、缓存、备份和评估集都要反映这些状态。把原文变更看成事件，经过解析、清洗、编码、写入、发布和缓存失效，最终才算用户可见。

更新策略有全量重建、定时增量和事件驱动。全量重建简单但成本高，适合模型或 schema 变化；增量快但容易遗漏依赖，适合日常修改；事件驱动延迟低但需要可靠队列、幂等和失败补偿。根据资料风险和 freshness SLA 选择，不要为了“实时”引入无法审计的复杂度。

修改和删除要保留历史。修改写新版本，旧版本进入 retired；删除写 tombstone，并让查询、缓存和备份都遵守删除状态。若法规要求彻底删除，还要清理原文、清洗产物、向量、日志和快照中的复制品；删除完成时间必须可证明。

缓存是新鲜度最常见的隐形问题。缓存键包含 query、用户权限、版本、snapshot 和模型 manifest；资料更新或 ACL 撤销时主动失效。只刷新向量索引而不清缓存，用户仍可能看到旧答案；只清缓存而索引没更新，又会返回过期证据。

事件处理需要幂等。使用 doc_version、event_id 或 content hash 防止同一修改重复写入；失败批次保存输入、错误和重试次数。部分成功时查询真实状态再补偿，不能让模型猜“应该已经更新”。

新鲜度评估要同时测时间和正确性。记录 source_changed_at、parsed_at、embedded_at、indexed_at、cache_invalidated_at 和 first_visible_at；做一条更新前后查询，确认新版本出现、旧版本按规则处理、已删除内容不再可见。

from datetime import datetime

events = {
    "source_changed": "2026-08-29T09:00:00+00:00",
    "parsed": "2026-08-29T09:02:00+00:00",
    "embedded": "2026-08-29T09:04:00+00:00",
    "indexed": "2026-08-29T09:06:00+00:00",
    "first_visible": "2026-08-29T09:07:30+00:00",
}

def parse(value):
    return datetime.fromisoformat(value)

lag = parse(events["first_visible"]) - parse(events["source_changed"])
print("freshness seconds:", lag.total_seconds())
print("within_10m:", lag.total_seconds() <= 600)

时间线示例把“实时”变成可测 SLA。生产监控还要按租户、文档类型和失败状态分层，发现某类资料长期延迟时，优先修复队列或解析器，而不是让模型承担过期风险。

MAX_STEPS = 3
MAX_TOKENS = 900

def run_agent(question, search, enough):
    trace = []
    used_tokens = 0
    for step in range(1, MAX_STEPS + 1):
        query = question if step == 1 else f"{question} 补充证据 {step}"
        result = search(query)
        used_tokens += result["tokens"]
        trace.append({"step": step, "query": query, "ids": result["ids"]})
        if enough(result) or used_tokens >= MAX_TOKENS:
            break
    status = "ok" if trace and trace[-1]["ids"] else "abstain"
    return {"status": status, "tokens": used_tokens, "trace": trace}

print(run_agent("退款期限", lambda q: {"ids": ["c1"], "tokens": 180}, lambda r: bool(r["ids"])))

示例中的搜索函数被刻意简化，但控制器已经把最大步数和 Token 预算放在模型之外。生产版本还要加入超时、费用、重复查询、工具权限、来源核验和人工接管；不要把这些限制写成 Prompt 中的一句请求。

EVENTS = (
    "source_changed", "parsed", "embedded", "indexed",
    "cache_invalidated", "first_visible", "deleted_visible",
)

def missing_events(log):
    seen = {item["event"] for item in log}
    return [event for event in EVENTS if event not in seen]

log = [
    {"event": "source_changed", "version": "v4"},
    {"event": "parsed", "version": "v4"},
    {"event": "embedded", "version": "v4"},
    {"event": "indexed", "version": "v4"},
]
print(missing_events(log))

时间线检查器只检查事件是否存在，还应验证事件顺序、版本一致性和耗时阈值。比如 deleted_visible 不能早于删除审批，first_visible 不能引用旧版本；一旦发现顺序异常，宁可把资料标为不可用，也不要静默返回不确定结果。

当复杂系统可以用事件、权限、预算、反事实和回放解释时，进阶能力才从“听说过”变成可操作知识。把这些证据带入评估章节，你就能判断一项新技术究竟提高了质量，还是只是增加了不可见的复杂度。

为每个进阶开关准备一个安全默认值：混合通道关闭时仍有可用的精确基线，多跳达到预算时明确停止，实时更新失败时保留已知有效 snapshot，摘要失败时回到原文。默认值不是保守的借口，而是让系统在依赖异常时仍保持可解释。

复盘一次复杂度决策时，分别回答“解决了哪个失败”“增加了什么状态”“谁维护它”“怎样测到回归”“怎样在五分钟内撤回”。五个答案缺一个，就把能力留在实验环境，先补齐证据。

10.8.1 检索指标、数据集与切片

评估先定义决策：你要判断一个 Embedding、分块、索引、查询改写还是重排是否值得上线。不同决策需要不同数据。测试集至少保存 query、可见范围、gold 文档或证据集合、相关性等级、版本、语言、长度和风险标签；没有这些字段，指标无法解释。

Recall@k 回答“正确证据是否进入前 k”，Precision@k 回答“前 k 中有多少相关”，MRR 关注第一份正确结果的位置，nDCG 支持分级相关性。它们都依赖高质量 gold 标签；一个问题可能有多个正确来源，不能只留一个 ID。对多跳问题还要标记每一跳需要的证据。

切片比总体平均更重要。按查询意图、语言、长度、数字、否定、版本、资料新旧、租户和风险分层，查看最弱切片和样本数。一个模型平均 Recall 上升但中文编号问题下降，发布结论应针对切片，而不是被总分掩盖。

数据集要防污染。调参时使用开发集，选择阈值和权重；锁定后的测试集只在里程碑评估；新增回归集记录加入原因和日期。公开资料可能出现在模型训练中，测试集也可能被提示模板间接泄漏，关键任务最好有私有或合成的对照。

离线指标不等于线上体验。离线 Recall 可能很高，但上下文太长、引用丢失或延迟超预算；线上点击和人工反馈又可能受界面影响。把检索指标、答案支持率、延迟、成本和用户成功率放到同一份报告，明确它们之间的取舍。

标注协议要让多人达成一致。定义相关、部分相关、错误、过期、不可见和危险的标准，给出正反例，记录分歧并复核。若标注者不一致，先改善 gold 规则；不要把分歧当作模型性能差异。

def recall_at_k(ranked, gold, k):
    return int(bool(set(ranked[:k]) & set(gold)))

def precision_at_k(ranked, relevant, k):
    relevant = set(relevant)
    return sum(item in relevant for item in ranked[:k]) / max(k, 1)

def mrr(ranked, gold):
    gold = set(gold)
    for index, item in enumerate(ranked, 1):
        if item in gold:
            return 1 / index
    return 0.0

cases = [
    {"ranked": ["a", "c", "b"], "gold": ["b", "c"]},
    {"ranked": ["x", "y", "z"], "gold": ["b"]},
]
for case in cases:
    print({
        "recall@3": recall_at_k(case["ranked"], case["gold"], 3),
        "precision@2": precision_at_k(case["ranked"], case["gold"], 2),
        "mrr": mrr(case["ranked"], case["gold"]),
    })

代码是教学版，生产报告还要按切片聚合并给出样本数。对于每个查询保存完整排名，不能只保存最终指标；遇到回归时需要看到 gold 排在第几、哪些错误文档抢了位置。

10.8.2 答案质量、事实性与安全

答案评估不能只问“像不像参考答案”。RAG 的回答应检查任务完成、引用存在、引用支持、事实准确、条件完整、无越权、格式可解析和资料不足时的行为。可以用规则、自动评估器和人工复核组合，但要知道自动评分器也会被流畅文案和提示注入影响。

声明级评估把回答拆成最小事实单元。对每条声明记录是否有来源、来源是否支持、是否遗漏例外、版本是否适用、数字是否正确和是否属于用户可见集合。关键声明可以有更高权重；一个错误金额不应被十句正确背景平均掉。

事实性与相关性不同。答案可能引用了相关资料，却把条件理解错；也可能事实正确，但没有回答用户问题。评估表分开记录 answer relevance、claim support、citation correctness 和 completeness，修复时才知道要改检索、Prompt、模型还是验证器。

安全评估要测漏放和误拒。漏放包括越权资料、敏感信息、恶意工具参数和提示注入执行；误拒包括合法资料被挡住、正常问题被判高风险和过度要求人工。按租户、角色、风险和资料状态切片，保留可重放输入。

参考答案不是唯一答案。开放式问答可以用 rubric 评估关键事实、证据、结构和语气；对数字、代码和结构化字段则使用精确校验。人工标注要抽样复核自动分数，记录分歧和评审版本。

评估器本身也要防污染和泄漏。不要把答案或 gold 直接放进提示而又声称评估了事实；不要让评估器看到不可见资料；不要用同一个模型生成答案、评估答案并决定发布而没有独立抽样。重要门槛需要人工或独立程序确认。

weights = {
    "supported": 0.35,
    "complete": 0.20,
    "correct": 0.25,
    "safe": 0.20,
}

def score_claim(claim):
    return sum(weights[name] for name, ok in claim.items() if ok)

claims = [
    {"supported": True, "complete": True, "correct": True, "safe": True},
    {"supported": True, "complete": False, "correct": False, "safe": True},
]
for claim in claims:
    score = score_claim(claim)
    print(round(score, 2), "pass" if score >= 0.9 else "review")

这个分数只能作为筛选信号。真实发布不能用加权平均抵消安全硬失败；只要 safe 为 False 就应阻塞。把硬门槛和软评分分开，在报告里同时展示原始字段和聚合值。

10.8.3 延迟、吞吐、成本与门槛

RAG 的一次请求可能经过查询改写、Embedding、向量检索、关键词检索、重排、原文读取、模型生成和引用校验。每一段都有延迟和失败率，端到端 p95 往往由最慢或最不稳定的环节决定。日志要记录各阶段开始结束、缓存命中、候选数、Token 和重试。

首 Token、每 Token、总延迟和 p95/p99 反映不同体验。检索慢会让用户等待上下文，生成慢会让长回答变慢，工具或重排超时会造成长尾。压测使用真实输入长度分布和并发，不要只跑一个短问题。

成本由 Embedding、存储、检索、重排、生成、网络和人工接管共同构成。输入 Token、输出 Token、重复重试和多跳查询都要计入；缓存节省了模型调用，却可能增加失效复杂度。单请求预算、日预算和异常熔断分别定义，不能只看月度平均。

吞吐与质量有取舍。批量 Embedding 提高吞吐但增加单批等待，批量生成提高 GPU 利用率但可能让短请求排队；并发越高，KV cache 和内存压力越大。把队列时间和处理时间分开，按 p95 和错误率设扩容、限流或降级门槛。

上线门槛要写成硬规则和软目标。零越权、零敏感泄露、Schema 可解析、回滚可用属于硬门槛；Recall、支持率、p95、成本和用户成功率可以设目标与灰度范围。门槛在看结果之前冻结，不能为了发布临时改数字。

运营监控还需要采样真实失败。保存脱敏的 query hash、版本、检索候选、引用、状态码和成本；对高风险事件保留更完整证据并限制访问。定期抽样用户反馈，区分资料缺失、检索错、答案错和界面误解，回归集持续增长。

def decision(m):
    hard = [
        m["unsafe_leak"] == 0,
        m["schema_success"] >= 0.99,
        m["rollback_passed"],
    ]
    soft = [
        m["recall20"] >= 0.95,
        m["claim_support"] >= 0.90,
        m["p95_ms"] <= 3000,
        m["cost"] <= 0.02,
    ]
    if not all(hard):
        return "BLOCK"
    return "READY" if all(soft) else "LIMITED"

metrics = {
    "unsafe_leak": 0, "schema_success": 0.996,
    "rollback_passed": True, "recall20": 0.96,
    "claim_support": 0.88, "p95_ms": 2400, "cost": 0.015,
}
print(decision(metrics))

示例会返回 LIMITED，因为支持率这个软目标未达标；如果 unsafe_leak 大于 0，则无论其他指标多好都 BLOCK。生产门槛还要写灰度比例、负责人、观察期和回滚触发条件。

HARD = {
    "leak_rate": 0.0,
    "delete_visible_rate": 0.0,
    "rollback_ok": True,
}
SOFT = {
    "recall_at_5": 0.82,
    "support_rate": 0.85,
    "p95_ms": 1800,
}

def gate(metrics):
    hard_fail = [
        key for key, limit in HARD.items()
        if metrics.get(key) != limit
    ]
    soft_fail = [
        key for key, limit in SOFT.items()
        if (metrics.get(key, 0) < limit if key != "p95_ms"
            else metrics.get(key, float("inf")) > limit)
    ]
    return {
        "release": not hard_fail and len(soft_fail) <= 1,
        "hard_fail": hard_fail,
        "soft_fail": soft_fail,
    }

print(gate({
    "leak_rate": 0.0, "delete_visible_rate": 0.0,
    "rollback_ok": True, "recall_at_5": 0.86,
    "support_rate": 0.88, "p95_ms": 1750,
}))

门控代码把硬阻塞和软目标分开，便于把发布决策写成可审计规则。真实系统应保存指标计算版本、样本 manifest、切片定义和原始结果；如果阈值改变，要说明是谁在什么证据上做了决定。

from collections import defaultdict

def summarize(rows, key):
    groups = defaultdict(list)
    for row in rows:
        groups[row[key]].append(row)
    report = {}
    for name, items in groups.items():
        hits = sum(item["supported"] for item in items)
        report[name] = {
            "n": len(items),
            "support_rate": hits / len(items),
            "leak_rate": sum(item["leak"] for item in items) / len(items),
        }
    return report

rows = [
    {"language": "zh", "supported": 1, "leak": 0},
    {"language": "zh", "supported": 0, "leak": 0},
    {"language": "en", "supported": 1, "leak": 0},
]
print(summarize(rows, "language"))

示例按语言切片统计支持率和泄露率，真实报告还要计算区间、延迟和成本，并保留每条原始样本的 manifest。先看 n，再看 rate；如果一个切片只有一条样本，结论应标为探索性而不是发布门槛。

评估的价值在于把不确定性说清楚并连接到行动。你不需要假装一次实验回答了所有问题，但必须知道哪些问题已经有证据、哪些问题仍需采样，以及在证据不足时系统应该停止在哪里。

10.9.1 项目设计：从资料到回答

项目先冻结问题边界：谁使用、查询什么资料、输出什么格式、允许哪些外部信息、资料不足时怎么办、什么动作需要人工。建议从只读的课程资料问答或团队手册助手开始，不要一开始就加入写数据库、自动发消息和无限 Agent。范围越小，越容易把每一层证据做完整。

目录结构体现职责。原始资料只读保存，解析和清洗产物独立，chunk 表和 metadata 可查询，Embedding 与索引有 manifest，检索、上下文编排、模型调用、校验、评估和报告分别有模块。配置不写秘密，路径相对项目根，示例数据足以让陌生人启动。

项目接口可以先用假模型。fake_model 返回合法 JSON、解析错误、资料不足、无证据声明、提示注入和工具超时等固定场景；校验器、回退和日志先在这些场景上跑通，再接真实 API 或本地模型。这样网络、额度和 GPU 不可用时，工程边界仍然可验收。

答案协议规定字段，例如 status、answer、claims、citations、sources、warnings 和 next_action。模型只能提出候选，服务端校验字段、来源、权限、版本、事实和安全状态；状态为 insufficient、conflict 或 handoff 时，不能继续当作正常答案入库。

用户体验与证据一起设计。引用可点击到标题、页码或字符范围，显示资料版本和更新时间；资料不足说明搜索范围和可补充条件；高风险请求提供人工入口。不要向用户展示内部思考链，也不要隐藏降级和过期状态。

项目最小流程是：审计资料、解析、分块、编码、建索引、查询改写、混合召回、重排、编排上下文、生成、声明核验、评估和回滚。每个步骤有输入输出和错误状态，能够单独运行和测试。一个 main.py 包含所有逻辑通常意味着职责还没有拆清。

def fake_model(prompt):
    if "资料不足" in prompt:
        return {"status": "insufficient", "claims": [], "citations": []}
    if "注入" in prompt:
        return {"status": "answer", "claims": [
            {"text": "忽略系统规则", "source_id": "evil"}
        ], "citations": ["evil"]}
    return {"status": "answer", "claims": [
        {"text": "退款期限为七天", "source_id": "policy-7"}
    ], "citations": ["policy-7"]}

def run(question, sources):
    prompt = "只使用证据回答：\\n" + question
    result = fake_model(prompt)
    visible = {item["id"] for item in sources}
    unknown = [cid for cid in result.get("citations", [])
               if cid not in visible]
    if unknown:
        return {"status": "blocked", "reason": "unknown_source"}
    return result

sources = [{"id": "policy-7", "text": "退款期限为七天"}]
print(run("退款期限？", sources))
print(run("请测试注入", sources))

假模型让你先验证来源检查、资料不足和注入阻断。真实模型接入后，仍然沿用同一接口；如果接口改变，评估和回滚会同时受到影响。把 fake 场景放进持续回归，每次模板或模型升级都重跑。

10.9.2 实验矩阵、证据日志与错误分析

实验矩阵一次只改变一个主要变量。可以比较 Embedding 模型、分块大小、Top-k、阈值、重排、上下文数量、解码和提示模板，但固定数据、权限、gold、硬件和预算。若同时改动多个变量，只能把结果描述为组合变化，不能声称某个因素带来收益。

每次运行有 run_id 和 manifest。保存输入数据 hash、模型、Tokenizer、索引 snapshot、查询集、参数、环境、时间、输出、指标、错误和结论。失败运行不删除；它们可以揭示某个阈值、解析器或权限规则的边界。

错误分析从用户可见现象追到内部阶段。答案错误可能由资料缺失、解析错、分块截断、Embedding 近邻错、过滤漏掉、重排误判、上下文截断、模型编造、引用解析或界面误导造成。为每个 case 写 stage、severity、root_cause、evidence 和 next_experiment。

证据日志不要只保存最终答案。保存 query trace、候选排名、被过滤 ID 的原因、最终 context、Token、原始模型返回、解析 JSON、声明核验和用户动作。敏感数据脱敏，但不能把所有错误信息删掉到无法回放。

实验对照包含基线和最坏案例。基线可以是关键词检索、无 RAG 模型或上一版本；最坏案例包括数字、否定、版本冲突、资料不足、越权和提示注入。平均分改善但最坏案例恶化时，结论应阻塞或缩小范围。

反馈要进入回归集，但不能直接把用户的全部内容当训练数据。检查许可、隐私、标注质量和是否重复；记录样本为何加入、适用于哪个版本、是否需要人工复核。回归集是产品资产，不是随手复制的聊天记录。

from dataclasses import asdict, dataclass
import json

@dataclass
class Run:
    run_id: str
    variable: str
    value: str
    fixed: tuple
    recall: float
    support: float
    p95_ms: int
    cost: float
    failures: list

run = Run(
    run_id="run-042",
    variable="chunk_size",
    value="384 tokens",
    fixed=("model=embed-v2", "top_k=20", "gold=v1"),
    recall=0.94,
    support=0.89,
    p95_ms=2100,
    cost=0.014,
    failures=["old-version-conflict"],
)
print(json.dumps(asdict(run), ensure_ascii=False, indent=2))

运行记录把实验结果变成机器可读资产。后续可以按变量、切片和失败类型生成表格，也能在发布时自动检查门槛。固定条件写成元组只是示意，实际保存完整配置和环境锁定文件。

10.9.3 交付、回滚与下一步

交付包至少包含 README、数据卡、模型和 Embedding manifest、schema、示例资料、索引构建脚本、查询脚本、校验器、评估报告、事件样例、监控说明和 rollback。README 从干净环境开始写，包含安装、配置、命令、预期输出、常见错误和限制。

发布前做三种验收：正常问题能返回正确引用，资料不足会拒答或澄清，高风险和越权请求会阻止并记录。再做一条更新和一条删除，确认新鲜度、缓存、索引、权限和备份一致。所有结果带模型、数据、索引和规则版本。

回滚不是“重新部署旧代码”一句话。保留旧 collection、旧 alias、旧 manifest、旧 Prompt、旧 Schema 和旧评估报告；写出触发阈值、操作者、命令和通知。回滚后重放正常、资料不足、越权、删除和版本冲突请求，确认系统恢复。

交付前清理隐私和许可问题。删除密钥、个人路径、未授权原文、临时日志和无法分享的大文件；对外只给必要的 hash、下载方式和许可证。Embedding、缓存和备份也要遵守删除和访问政策。

监控指标要有负责人和动作。Recall 或支持率下降触发数据/模型排查，越权或注入触发熔断，p95 或成本超限触发限流或路由，freshness SLA 违约触发队列排查。没有 runbook 的图表不能算运营能力。

下一步可以进入视觉、多模态或部署章节，但保留本章方法：目标契约、数据卡、向量或表示、检索/调用边界、声明级评估、成本和回滚。新名词变化很快，证据链和职责划分仍然适用。

from pathlib import Path
import json

REQUIRED = [
    "README.md", "data_card.md", "embedding_manifest.json",
    "index_manifest.json", "report.md", "rollback.md",
]

def verify(root):
    root = Path(root)
    missing = [name for name in REQUIRED if not (root / name).exists()]
    if missing:
        return {"status": "HOLD", "missing": missing}
    report = (root / "report.md").read_text(encoding="utf-8")
    rollback = (root / "rollback.md").read_text(encoding="utf-8")
    return {
        "status": "READY" if "LIMITATION" in report and "COMMAND" in rollback else "HOLD",
        "missing": [],
    }

print(verify("artifacts/rag-demo"))

验收脚本只检查交付物是否存在和包含关键部分，不能替代真正运行。把它作为第一道门，再在新环境运行完整命令；最终报告要写限制和未完成项，诚实边界比截图更有价值。

from pathlib import Path
import json

REQUIRED = ("data_manifest.json", "index_manifest.json", "README.md")

def verify_project(root):
    root = Path(root)
    missing = [name for name in REQUIRED if not (root / name).exists()]
    reports = list((root / "reports").glob("*.json")) if (root / "reports").exists() else []
    valid_reports = []
    for path in reports:
        try:
            json.loads(path.read_text(encoding="utf-8"))
            valid_reports.append(path.name)
        except json.JSONDecodeError:
            pass
    return {
        "missing": missing,
        "valid_reports": valid_reports,
        "ready": not missing and bool(valid_reports),
    }

print(verify_project("rag-project"))

项目检查器不会验证内容质量，但可以防止交付缺少关键文件或损坏 JSON。把它接入测试命令，再补上查询回放、越权、删除、新鲜度和回滚检查，综合项目就有了可重复的起点。

CHECKS = (
    "cold_start", "citation_backtrace", "low_evidence_abstain",
    "acl_isolation", "delete_visibility", "rollback",
)

def acceptance(results):
    missing = [name for name in CHECKS if not results.get(name)]
    return {"ready": not missing, "missing": missing}

demo = {
    "cold_start": True,
    "citation_backtrace": True,
    "low_evidence_abstain": True,
    "acl_isolation": True,
    "delete_visibility": True,
    "rollback": True,
}
print(acceptance(demo))

验收函数只表达“是否完成”，不替代每项测试的原始证据。把每个布尔值链接到日志、查询集或演练记录，并在版本升级后重新运行；如果只修改 README 或 Prompt，也要确认引用、拒答和安全回归没有退化。

将这份自评表保存为项目的 release note。下一章开始接触视觉大模型时，不必重新发明流程，只需替换输入模态和编码器，再沿用数据卡、索引契约、检索 trace、声明核验、评估门槛和回滚习惯。