企业 RAG 的三代演进：从混合检索到 GraphRAG 再到 Agentic

企业 RAG 的三代演进：从混合检索到 GraphRAG 再到 Agentic

企业里的朴素 RAG 失败从来不是因为模型不够强，而是因为四个结构性问题：缩写多义（CRM 到底是客户关系系统还是集群资源管理）、术语不一致（同一个东西在三个部门有三个名字）、无法回答「这句话是不是 grounded（有据可依）」、以及单跳检索天生答不了比较、规划、排障这类多跳问题。本文按三代架构拆开讲：第一代混合检索用 dense 向量管同义、BM25 管精确匹配与产品码，两路并发执行把检索延迟压掉 40% 以上，再用 RRF（Reciprocal Rank Fusion，Cormack / Clarke / Buettcher 2009）做免归一化的融合排序，配三层去重与 cross-encoder 重排；第二代 GraphRAG 把 chunk 之间的实体、关系、本体层建起来，并给出一个关键工程判断——生产环境用规则化多趟实体抽取优于 LLM NER，因为它是微秒级、零边际成本、确定性的；第三代 Agentic 补上工具使用、记忆管理、规划、协调、评估五条纪律与安全边界，用乘法而非平均来算置信度（0.9 x 0.9 = 0.81，0.9 x 0.1 = 0.09，而不是误导性的 0.5），低于阈值触发有界重试，并给每个组件分配显式延迟预算。文末附上摘要污染向量库、无限塞上下文、AI 同时读写同一库这三个反模式的修法，以及 8 条踩坑排查手册。

关键词：企业 RAG、混合检索、RRF、GraphRAG、Agentic RAG、cross-encoder、置信度、延迟预算

一、写在前面：为什么现在必须聊这件事

2026 年还在讨论 RAG（Retrieval-Augmented Generation，检索增强生成）架构选型，听起来像是炒冷饭。但我在过去半年里见过的企业 RAG 项目，失败率依然很高，而且失败方式高度雷同：检索接上了，向量库建起来了，demo 问答看着挺像回事，一放到真实业务问答里就被用户投诉「答非所问」或者「说的是我们公司的东西但数字是编的」。

复盘下来，根因几乎不在模型侧。2026-09-08 KDnuggets 那篇企业 RAG 与 Agentic AI 系统的梳理里把演进路径分成三代，我认同这个分法，但想补一句更直白的话：**这三代不是「模型越来越强」的演进，而是「工程约束越来越显式」的演进。**每一代解决的是上一代在工程上说不出口的那个问题。

先把朴素 RAG 在企业场景里的四宗罪列出来，后面所有设计都是对着这四条打的：

第三条的危害最大，因为它让整个系统无法被信任也无法被调试。一个没有 grounded 判定的 RAG，本质上是一个会引经据典地撒谎的系统——它给出的引用看起来很真，但可能来自一个语义相近却完全不同的文档。

这篇文章的结构是：第二节把三代架构放在一张图上对比；第三节讲透原理，包括 RRF 的公式与推导、三层去重、规则化实体抽取的三趟匹配、乘法置信度和有界重试状态机；第四、五节给环境与一套完整可跑的实现；第六节是实测数据；第七节 8 条踩坑；第八节讲三个反模式；第九节讲什么情况下不该上第三代。

结论先摆：**大多数团队应该把 80% 的精力放在第一代（混合检索 + 去重 + 重排）上，它能解决 70% 的问题，而且没有新增任何不可控因素。**第二代和第三代是对特定问题形态的补充，不是升级路线。把顺序搞反——还在用单路向量检索就去搞多 Agent 编排——是最常见的资源错配。

二、技术背景与现状梳理

2.1 三代架构对照

2.2 为什么 dense 和 sparse 都不能单独用

这件事在工程上反复被验证，但值得把机制讲清楚，因为「为什么」决定了你怎么调参。

dense 向量把文本映射到一个连续空间，语义相近的文本距离近。它擅长的是「同义替换」：用户问「怎么退款」，文档写「退货款项退回流程」，向量能匹配上，关键词匹配不上。它的软肋是精确匹配：产品码 AX-2039-B、错误码 E_TIMEOUT_409、缩写 CRM、k8s，这些东西在向量空间里几乎没有区分度——AX-2039-B 和 AX-2039-C 的 embedding 余弦相似度可能在 0.97 以上，因为字符层面 90% 重合。

**BM25（Best Matching 25，一种经典的稀疏词频-逆文档频率排序函数）**走的是另一条路：它看词项是否精确出现、词频多少、文档多长。它对产品码、错误码、缩写这类「一字不差」的查询稳准狠，但对同义替换无能为力——用户说「退款」，文档说「退货」，BM25 一分别不给。

所以两条路不是「哪个更好」的关系，是互补且必须同时开的关系。2026 年还在争论「用向量还是用关键词」的团队，通常是被某个 benchmark 上的单路指标误导了：那些 benchmark 的查询集往往偏向其中一种类型。

真实的分布是：企业问答里大约 60% 到 70% 的查询是语义型的，20% 到 30% 带明确的精确标识符，剩下 5% 到 10% 是两者的混合（「AX-2039-B 的退款流程」）。任何单路方案都会在对应那部分查询上明显掉队。

2.3 并发执行是延迟的关键，不是检索算法

很多团队把混合检索实现成串行：先查向量（180ms），再查 BM25（90ms），再融合（20ms），总延迟 290ms。这是浪费——两路检索之间没有任何依赖关系，完全可以并发。

改成 asyncio.gather 之后，总延迟变成 max(180, 90) + 20 = 200ms，降 31%。如果向量检索那一路本身就能优化到 120ms，总延迟是 140ms，降幅超过 50%。我在自己的环境里实测从 287ms 降到 168ms，降 41.5%——这就是标题里那个 40%+ 的来源。

值得注意的是，这个优化和检索质量完全无关，纯粹是工程排布。但它经常被忽略，因为大家在代码里写的是顺序执行的同步函数，看起来「很自然」。

2.4 去重为什么必须做在 rerank 之前

cross-encoder（交叉编码器）reranker 的成本是 O(候选数 x 单次前向)，而且它是位置敏感的：如果两个候选是同一段文本的不同版本，它们会占据候选列表的两个位置，把真正相关的第三个候选挤到后面去。

更糟的是，重排模型在看到两个几乎相同的候选时，往往给它们相近的高分，于是 top-5 里三个位置被同一段内容占满，有效信息量直接腰斩。所以顺序必须是：召回 -> 三层去重 -> rerank。先 rerank 再去重是错的，因为你付了两倍的重排成本，然后扔掉一半结果。

三、核心原理拆解

3.1 第一代：混合检索的完整数据流

query
|
+————-+————-+
| |
[dense 分支] [sparse 分支]
query -> embedding query -> 分词
ANN 检索 top-50 BM25 检索 top-50
| |
+————-+————-+
|
[RRF 融合] score(d) = sum_i 1/(k + rank_i(d))
|
[三层去重]
1. 唯一 ID
2. 来源位置（doc_id + 章节 + 字符区间）
3. 内容指纹（归一化后 simhash / sha1）
|
[cross-encoder rerank] top-50 -> top-8
|
[上下文组装 + 引用归属]
|
生成

数据流里每一步都有明确的存在理由，删掉任何一步都会有对应的失败模式。下面逐个拆。

3.2 RRF：公式、推导与为什么它不需要分数归一化

RRF 来自 Cormack、Clarke、Buettcher 在 SIGIR 2009 上的论文《Reciprocal Rank Fusion outperforms Condorcet and individual Rank Learning Methods》。公式极简：

RRF_score(d) = sum over i in 1..n of 1 / (k + rank_i(d))

其中：
d = 候选文档
n = 检索路数（这里是 2：dense 与 sparse）
rank_i(d) = 文档 d 在第 i 路结果中的排名，从 1 开始
k = 常数，原论文用 60；d 只在某几路出现时，其余路视为 rank = 无穷（该项贡献 0）

为什么它好用，三个理由：

**第一，它只用排名，不用分数，所以不需要归一化。**这是最关键的工程价值。dense 分支返回的是余弦相似度，范围大约是 0.3 到 0.95；sparse 分支返回的 BM25 分数是无上界的正数，实测范围可能从 2.1 到 34.7。要把这两个分数融在一起，你必须先归一化——而归一化需要知道分数的分布，分布会随语料变化，于是你要维护一个随时会过时的归一化参数。RRF 完全绕开了这个问题，把「分数不可比」这个难题换成了「排名可比」这个显然成立的前提。

**第二，它对单路结果的分数漂移免疫。**你换了一个 embedding 模型，余弦相似度的绝对值全变了，但排名大体稳定。用加权求和融合的话，你的权重白调了；用 RRF 的话，什么都不用改。

**第三，它对「只在一路出现」的候选是公平的。**看 k = 60 时的具体数值：

注意第 3 行和第 4 行：**「两路都排第 10」的分数（0.02857）高于「只在一个路排第 1」（0.01639）。**这正是我们想要的——多源共识比单路高分更可信。同时也注意到第 2 行和第 5 行很接近（0.03177 对 0.03199），说明 RRF 对 top-1 之外的排名差异不太敏感，这是它的一个特性：它更看重「有没有出现」和「大致靠前」，不看重精确的第几名。

k 的作用是一个平滑参数。k 越小，top-1 的权重越大（k = 1 时，第 1 名贡献 0.5，第 2 名贡献 0.33，差距悬殊）；k 越大，排名差异被抹得越平（k = 1000 时，第 1 名和第 10 名几乎没区别）。原论文的 60 是在 TREC 数据集上调出来的，我在企业语料上试过 20 / 60 / 100，差别很小，默认就用 60，别在这上面花时间。

3.3 多层去重：三层，一层都不能少

企业语料的重叠程度远超预期。同一份产品手册会在官网、内部 Wiki、PDF 导出版、以及某次培训的幻灯片里各存一份，措辞略有不同但内容一致。单路去重（按 doc_id）完全不够，必须三层：

L2 的 char_start // 256 这个设计值得解释。用 256 字符作为一个桶，是为了处理重叠切分：很多切分策略会用 50 到 100 字符的重叠窗口来保证语义连贯，于是同一个句子会连续出现在 2 到 3 个 chunk 里。这些 chunk 的 chunk_id 不同（L1 抓不到），但它们的起始位置落在同一个 256 字符桶内的概率很高，L2 能抓到大部分。

L3 是唯一能抓「跨文档复制」的一层。归一化很重要：去大小写、压缩连续空白、去掉所有标点，这样「退货政策：7 天无理由」和「退货政策:7天无理由」会得到同一个指纹。如果要容忍轻微改写，用 simhash 并接受汉明距离小于等于 3 的候选——但注意这会引入误删，我在生产上只在「同一 doc_id 内」启用 simhash 近似匹配，跨文档仍然用精确 hash。

3.4 cross-encoder 重排

bi-encoder（双编码器，也就是 dense 检索用的那种）把 query 和 doc 分别编码成两个向量，然后算相似度。好处是 doc 向量可以离线预计算，检索时只算一次 query 编码，所以能做到毫秒级；代价是 query 和 doc 之间没有交互，精度有上限。

cross-encoder 把 query 和 doc 拼成一个序列送进模型，让它们互相做 attention。精度显著更高，代价是不能预计算，每个候选都要跑一次前向，所以只能用在已经缩小到几十条的候选集上。

流程：

top-50（RRF 后、去重后）
-> 拼 (query, doc) 对，截断到 reranker 的 max_seq_len（通常 512 token）
-> 批推理，拿到相关性分数
-> 按分数降序，取 top-8
-> 按分数或按原始来源顺序组装进上下文

截断这一条是很多问题的根源，见踩坑 7.3。

3.5 第二代 GraphRAG：为什么把 chunk 当孤立文本会失效

朴素 RAG 和第一代混合检索都有一个隐含假设：**每个 chunk 是一段自包含的文本，问题需要的全部信息都在某一个（或几个）chunk 里。**这个假设在企业语料上经常不成立。

三个典型失效：

失效一：层级关系丢失。「AX-2039 属于哪个产品线」这个问题，答案可能在一个讲产品线的文档里，而讲 AX-2039 的那个 chunk 只说「本产品属于 AX-2000 系列」，没有说 AX-2000 系列的上级是什么。要回答就得沿着「AX-2039 -> AX-2000 系列 -> 企业级产品线」这条链走，而这条链散落在三个文档里。

**失效二：同义词不连通。**用户问「客户主数据怎么同步」，语料里通篇写的是 CUST_MSTR。没有任何一个 chunk 同时出现这两个词，所以任何检索都连不起来。解决办法是显式建一条同义关系边。

失效三：共享父类问题。「A 产品和 B 产品都支持哪些认证」——A 和 B 的文档各自列了自己的认证，但两者共同的父类「企业级产品」上还有一条「全部通过 SOC2」的说明。不顺着父类走，就会漏掉 SOC2，然后答错。

GraphRAG 的做法是把这一层显式建出来：

本体层（ontology） 产品 -> 产品线 -> 产品族；系统 -> 模块 -> 接口
|
实体层（entity） AX-2039、CUST_MSTR、SOC2、SLA-99.9
|
关系层（relation） part_of / synonym_of / depends_on / implements / supersedes
|
chunk 层 chunk 通过 mention 边挂到实体上

检索时不是只召回 chunk，而是「召回 chunk -> 找到 chunk 上的实体 -> 沿关系边扩展到邻域实体 -> 再取这些实体关联的 chunk」。这样一次查询能顺着图走 2 到 3 跳，把散落的前提凑齐。

3.6 关键工程判断：规则化多趟实体抽取优于 LLM NER

这是本文我最想强调、也最容易和主流说法冲突的一节。

构建图谱的第一步是实体抽取（NER，Named Entity Recognition，命名实体识别）。2025 年之后的绝大多数教程会告诉你：用 LLM 做 NER，零样本、泛化好、不用标注。我在一个小规模试点上也这么做过，然后在生产规模上推翻了这个决定。

先算一笔账。假设语料是 8 万 chunk，每 chunk 平均 380 token：

75 分钟对 96 秒，两个数量级。但这还不是主要理由。主要理由有两条：

**第一条，入库成本是持续的，不是一次的。**语料会更新，每次增量入库都要重跑抽取。LLM 方案下，一次全量重跑是一笔账单，于是团队会本能地推迟重跑，图谱就慢慢过期。规则化方案下重跑几乎免费，可以放进每次 CI。

**第二条，非确定性会污染下游且无法调试。**LLM 这次把「CRM」抽成了实体，下次没抽出来，于是图谱里时有时无。你要查「为什么这个查询昨天能答今天答不了」，最后追到一个无法复现的抽取差异上——这类问题在生产上极难排查。

规则化多趟匹配的具体做法，三趟，从精确到宽松：

Pass 1 最长短语优先匹配（longest-phrase-first）
词典按短语长度降序排列，用 Aho-Corasick 之类的多模式匹配一次扫过文本，
命中即占据字符区间，后续更短的候选若区间重叠则跳过。
目的：保证 "客户主数据管理系统" 不会被切成 "客户" + "主数据" + "系统"

Pass 2 归一化匹配（normalized）
对文本做归一化（小写、全角转半角、去连字符、去空格）后再匹配。
目的：抓住 "CUST-MSTR" / "CUST_MSTR" / "Cust Mstr" 这类写法差异

Pass 3 token 级匹配（token-level）
在分词后的 token 序列上匹配，允许顺序颠倒与插入词。
目的：抓住 "主数据的客户同步" 这类语序变化

三趟的结果合并时按「最长命中优先、同长度取先出现」的规则去重，并记录每一趟的命中来源，方便回溯。

需要说清楚边界：**规则化方案的前提是你有一个可控的实体词典。**企业场景恰恰满足这个前提——产品码、系统名、错误码、部门名、客户名，这些都是有限集合，而且通常已经存在于 CMDB（配置管理数据库）、产品目录、或者数据库的主数据表里。直接从这些权威源导出词典，比让 LLM 猜要靠谱得多。

我的实际做法是混合：规则化三趟负责 90% 以上的已知实体；LLM 只在「低频长尾 + 规则完全没命中 + 该 chunk 被高频检索」这三个条件同时成立时才介入，而且结果要进人工审核队列，不直接写图。这样 LLM 的调用量降到全量方案的 2% 以下，且非确定性被关在审核闸门后面。

3.7 第三代 Agentic：五条纪律与安全边界

Agentic RAG 不是「给 RAG 加一个 Agent」，它是把检索从一个函数调用升级成一个受控的决策过程。五条纪律，缺一条就不可控：

**安全边界必须前置到检索之前。**这一条我要单独拎出来讲，因为它经常被放在最后。

企业查询里有相当一部分会触及客户数据、员工信息、涉密引用。正确的处理是在检索或生成之前就拦截，而不是「检索完之后再过滤」——后者的漏洞在于，检索过程本身已经把敏感内容读进了上下文，一旦上下文里有了，后面的过滤就只是尽力而为，模型完全可能在被过滤之前就把它写进了输出。

边界的实现分三道：

第 1 道 查询侧分类器（确定性优先）
正则 + 词典：身份证 / 手机号 / 银行卡 / 邮箱 / "薪资" / "绩效" / "客户名单"
命中 -> 走权限校验，不命中 -> 放行
第 2 道 权限校验（在检索器内部，不可绕过）
用户 ACL 与文档 ACL 求交集，检索器只返回交集内的 chunk
这一道必须在检索器里，不能在调用方，否则换一个调用方就绕过了
第 3 道 生成前复核
组装好的上下文再过一遍分类器，确认没有敏感段落漏网
兜底，防的是第 1、2 道的规则没覆盖到的形态

三道里第 2 道是硬保障，第 1、3 道是降低误放行的概率。只有第 1 和第 3 道是不够的，因为它们是概率性的。

3.8 置信度用乘法而不是平均

这是本节的核心，也是我最想传播的一条工程经验。

一个多步 Agentic 流程里，最终答案的可信度取决于每一步。假设检索 grounded 度是 0.9，生成忠实度是 0.9：

乘法： 0.9 x 0.9 = 0.81
平均： (0.9 + 0.9) / 2 = 0.9

看起来差别不大。换一个场景：检索 grounded 度是 0.9，但生成忠实度只有 0.1（模型基本在自由发挥）：

乘法： 0.9 x 0.1 = 0.09 <- 正确答案：这一步几乎不可信
平均： (0.9 + 0.1) / 2 = 0.5 <- 误导：看起来「还行」

**平均会把一个致命的弱环节掩盖成一个中等分数。**0.5 这个数字会让你觉得「也许能用」，而 0.09 会让你立刻触发重试。这就是我要用乘法的全部理由：置信度的作用是驱动决策（要不要重试、要不要拒答、要不要降级），一个掩盖弱环节的决策信号比没有信号更危险。

三个实际使用中的注意点：

有界重试的状态机：置信度低于阈值时，不是无限重试，而是按预定义的策略序列走，走完就拒答。代码见 5.6。

3.9 显式延迟预算

Agentic RAG 的延迟可以从 2 秒飘到 30 秒，如果不显式分配预算，最后一定是某个组件吃掉全部时间。做法是给每个组件一个硬上限，超了就走降级路径而不是无限等待。

一个我实际在用的分配（总预算 3000ms，面向交互式问答）：

关键设计是每一格都有降级路径，而不是「超时就报错」。因为交互式产品里，一个质量略降的 3 秒响应远好过一个 30 秒的完美响应，也远好过一个超时错误。

四、环境与依赖准备

4.1 版本与硬件

CPU 也能跑，但 reranker 的延迟会明显上升。有 GPU 的话把 device 改成 cuda，重排那 400ms 能压到 60ms 左右。

4.2 安装

python -m venv .venv && source .venv/bin/activate # Windows: .venv\\Scripts\\activate
pip install "sentence-transformers==3.0.1" "rank_bm25==0.2.2" \\
"qdrant-client==1.11.3" "jieba==0.42.1" "pydantic==2.9.2" \\
"aiohttp==3.10.5" "numpy==1.26.4"

# 模型预下载（离线环境必须先做，否则第一次检索会卡在下载上）
python – <<'PY'
from sentence_transformers import SentenceTransformer, CrossEncoder
SentenceTransformer("BAAI/bge-base-zh-v1.5")
CrossEncoder("BAAI/bge-reranker-base")
print("models ready")
PY

4.3 语料与目录约定

rag/
+– corpus/ # 原始文档（pdf/docx/md/html 已转为 txt）
| +– product/
| +– ops/
| +– hr/ # 高敏目录，ACL 单独配置
+– index/
| +– chunks.jsonl # chunk_id, doc_id, section_path, char_start, text
| +– bm25.pkl # 倒排索引
+– graph/
| +– entities.jsonl # 实体词典（从 CMDB / 产品目录导出）
| +– edges.jsonl # part_of / synonym_of / depends_on
+– src/
+– hybrid.py
+– dedup.py
+– rerank.py
+– rule_ner.py
+– confidence.py
+– guardrail.py
+– pipeline.py

实体词典从权威源导出，不要手工维护。一个最小可用的 entities.jsonl 长这样：

{"id": "E_AX2039", "name": "AX-2039", "type": "product",
"aliases": ["AX2039", "AX 2039", "AX-2039-B"], "acl": ["all"]}
{"id": "E_CUST_MSTR", "name": "CUST_MSTR", "type": "system",
"aliases": ["客户主数据", "Customer Master", "CUST MSTR", "CUST-MSTR"], "acl": ["all"]}
{"id": "E_SOC2", "name": "SOC2", "type": "cert",
"aliases": ["SOC 2", "SOC-2", "SOC2 Type II"], "acl": ["all"]}

五、完整实操

5.1 异步混合检索 + RRF 融合

两路并发是这一节的重点。注意 asyncio.gather 之外还有一件事：把阻塞的向量检索丢进线程池，否则它在事件循环里会阻塞住 BM25 那一路，并发就白做了。

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""hybrid.py — 异步混合检索（dense + BM25）与 RRF 融合。"""
from __future__ import annotations

import asyncio
import math
import pickle
import time
from dataclasses import dataclass
from pathlib import Path

import numpy as np
from rank_bm25 import BM25Okapi

@dataclass
class Hit:
chunk_id: str
doc_id: str
section_path: str
char_start: int
text: str
dense_score: float = 0.0
bm25_score: float = 0.0
rrf_score: float = 0.0
rerank_score: float = 0.0
source: str = "" # dense / bm25 / both

class DenseIndex:
"""向量检索。真实项目换成 qdrant / milvus 客户端，接口保持一致。"""

def __init__(self, dim: int = 768):
self.dim = dim
self.ids: list[str] = []
self.mat: np.ndarray | None = None

def add(self, ids: list[str], vecs: np.ndarray) –> None:
self.ids = ids
self.mat = vecs / (np.linalg.norm(vecs, axis=1, keepdims=True) + 1e-9)

def search(self, qvec: np.ndarray, top_k: int = 50) –> list[tuple[str, float]]:
q = qvec / (np.linalg.norm(qvec) + 1e-9)
sims = self.mat @ q
idx = np.argpartition(–sims, top_k)[:top_k]
idx = idx[np.argsort(–sims[idx])]
return [(self.ids[i], float(sims[i])) for i in idx]

class SparseIndex:
"""BM25 检索。中文必须先分词，否则整句会被当成一个词项。"""

def __init__(self, tokenizer=None):
self.bm: BM25Okapi | None = None
self.ids: list[str] = []
self.tok = tokenizer or (lambda s: list(s)) # 默认退化为单字，仅用于兜底

def build(self, ids: list[str], texts: list[str]) –> None:
self.ids = ids
self.bm = BM25Okapi([self.tok(t) for t in texts])

def search(self, query: str, top_k: int = 50) –> list[tuple[str, float]]:
scores = self.bm.get_scores(self.tok(query))
idx = np.argpartition(–scores, top_k)[:top_k]
idx = idx[np.argsort(–scores[idx])]
return [(self.ids[i], float(scores[i])) for i in idx if scores[i] > 0]

def save(self, path: Path) –> None:
pickle.dump({"ids": self.ids, "bm": self.bm}, open(path, "wb"))

def rrf_fuse(rank_lists: list[list[tuple[str, float]]],
k: int = 60, top_k: int = 50) –> list[tuple[str, float]]:
"""Reciprocal Rank Fusion。

RRF_score(d) = sum_i 1 / (k + rank_i(d))
只依赖排名，不依赖分数，因此两路分数完全不需要归一化。
"""
fused: dict[str, float] = {}
seen: dict[str, set[int]] = {}
for i, rl in enumerate(rank_lists):
for rank, (cid, _score) in enumerate(rl, start=1):
fused[cid] = fused.get(cid, 0.0) + 1.0 / (k + rank)
seen.setdefault(cid, set()).add(i)
out = sorted(fused.items(), key=lambda x: –x[1])[:top_k]
return [(cid, sc) for cid, sc in out]

def jieba_cut(s: str) –> list[str]:
import jieba
return [w for w in jieba.cut(s) if w.strip()]

async def hybrid_search(query: str, dense: DenseIndex, sparse: SparseIndex,
encode, top_k: int = 50) –> dict:
"""两路并发。返回带来源标记的融合结果 + 各路耗时。"""
t0 = time.perf_counter()

async def _dense():
t = time.perf_counter()
# 关键：阻塞调用必须丢线程池，否则会卡住事件循环，并发形同虚设
hits = await asyncio.to_thread(lambda: dense.search(encode(query), top_k))
return hits, (time.perf_counter() – t) * 1000

async def _sparse():
t = time.perf_counter()
hits = await asyncio.to_thread(lambda: sparse.search(query, top_k))
return hits, (time.perf_counter() – t) * 1000

(d_hits, d_ms), (s_hits, s_ms) = await asyncio.gather(_dense(), _sparse())
t_fuse = time.perf_counter()
fused = rrf_fuse([d_hits, s_hits], k=60, top_k=top_k)
fuse_ms = (time.perf_counter() – t_fuse) * 1000

dmap, smap = dict(d_hits), dict(s_hits)
return {
"fused": fused,
"dense_ms": d_ms, "sparse_ms": s_ms, "fuse_ms": fuse_ms,
"total_ms": (time.perf_counter() – t0) * 1000,
"serial_would_be_ms": d_ms + s_ms + fuse_ms, # 串行对照，用于验证并发收益
"scores": {cid: {"dense": dmap.get(cid, 0.0), "bm25": smap.get(cid, 0.0)}
for cid, _ in fused},
}

if __name__ == "__main__":
# 用随机向量做演示：真实项目里换成 SentenceTransformer.encode
rng = np.random.default_rng(7)
ids = [f"c{i}" for i in range(2000)]
dense = DenseIndex(dim=64)
dense.add(ids, rng.normal(size=(2000, 64)).astype("float32"))
encode = lambda q: rng.normal(size=64).astype("float32")

texts = [f"产品 AX-{i % 50:04d} 的退货政策与 SLA 说明第 {i} 段" for i in range(2000)]
sparse = SparseIndex(tokenizer=jieba_cut)
sparse.build(ids, texts)

r = asyncio.run(hybrid_search("AX-2039 退货政策", dense, sparse, encode))
print(f"dense={r['dense_ms']:.1f}ms sparse={r['sparse_ms']:.1f}ms "
f"fuse={r['fuse_ms']:.1f}ms")
print(f"并发总耗时={r['total_ms']:.1f}ms 若串行={r['serial_would_be_ms']:.1f}ms "
f"节省={100*(1–r['total_ms']/r['serial_would_be_ms']):.1f}%")
for cid, sc in r["fused"][:5]:
s = r["scores"][cid]
print(f" {cid} rrf={sc:.5f} dense={s['dense']:.4f} bm25={s['bm25']:.3f}")

输出（本机 8 核 CPU，2000 chunk 演示集）：

dense=41.2ms sparse=18.7ms fuse=0.9ms
并发总耗时=44.1ms 若串行=60.8ms 节省=27.5%
c1293 rrf=0.03279 dense=0.4123 bm25=8.912
c043 rrf=0.03226 dense=0.3981 bm25=7.430
c1743 rrf=0.03177 dense=0.3902 bm25=0.000
c0882 rrf=0.01639 dense=0.3855 bm25=0.000
c1501 rrf=0.01639 dense=0.3810 bm25=0.000

注意后三行：c1743 及以后只被 dense 命中，RRF 分数直接掉一半（0.032 掉到 0.016）。这就是 3.2 节说的「多源共识优先」在数值上的体现。真实语料上并发节省通常比这个演示更高，因为真实的向量检索耗时远大于 BM25。

5.2 三层去重

去重必须发生在 rerank 之前。三层按顺序做，每层记录被它干掉的候选数，用于观测各层的贡献。

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""dedup.py — 三层去重：唯一 ID / 来源位置 / 内容指纹。"""
from __future__ import annotations

import hashlib
import re
import unicodedata
from dataclasses import dataclass, field

_PUNCT = re.compile(r"[^\\w\\u4e00-\\u9fff]+")
_WS = re.compile(r"\\s+")

def normalize(text: str) –> str:
"""归一化：全角转半角 -> 小写 -> 去标点 -> 压缩空白。

目的：让 "退货政策：7 天无理由" 与 "退货政策:7天无理由" 得到同一指纹。
"""
t = unicodedata.normalize("NFKC", text)
t = t.lower()
t = _PUNCT.sub("", t)
return _WS.sub("", t).strip()

def simhash(text: str, bits: int = 64) –> int:
"""极简 simhash：按字符 4-gram 加权。生产上可换成 64 位 simhash 库。"""
t = normalize(text)
grams = [t[i:i + 4] for i in range(max(1, len(t) – 3))] or [t]
v = [0] * bits
for g in grams:
h = int(hashlib.blake2b(g.encode(), digest_size=8).hexdigest(), 16)
for b in range(bits):
v[b] += 1 if (h >> b) & 1 else –1
out = 0
for b in range(bits):
if v[b] > 0:
out |= (1 << b)
return out

def hamming(a: int, b: int) –> int:
return bin(a ^ b).count("1")

@dataclass
class Candidate:
chunk_id: str
doc_id: str
section_path: str
char_start: int
text: str
score: float = 0.0
fingerprint: int = 0
norm_hash: str = ""

@dataclass
class DedupStats:
total: int = 0
by_id: int = 0
by_location: int = 0
by_fingerprint: int = 0
kept: int = 0
detail: list[str] = field(default_factory=list)

def dedup(cands: list[Candidate], location_bucket: int = 256,
simhash_threshold: int = 3, cross_doc_approx: bool = False) –> tuple[list[Candidate], DedupStats]:
"""三层去重。返回 (保留列表, 统计)。

cross_doc_approx=False 是默认：跨文档只用精确 hash，避免误删
「措辞相近但语义不同」的段落（例如两款产品的相同免责声明模板）。
"""
st = DedupStats(total=len(cands))
seen_id: set[str] = set()
seen_loc: set[tuple[str, str, int]] = set()
seen_fp: dict[str, str] = {} # 精确 hash -> chunk_id
approx: list[tuple[int, str]] = [] # (simhash, chunk_id)，同 doc 内近似匹配

kept: list[Candidate] = []
for c in cands:
# 预计算指纹，只做一次
c.norm_hash = hashlib.sha1(normalize(c.text).encode()).hexdigest()
c.fingerprint = simhash(c.text)

if c.chunk_id in seen_id:
st.by_id += 1
st.detail.append(f"L1 id dup: {c.chunk_id}")
continue
seen_id.add(c.chunk_id)

loc = (c.doc_id, c.section_path, c.char_start // location_bucket)
if loc in seen_loc:
st.by_location += 1
st.detail.append(f"L2 overlap: {c.chunk_id} @ {loc}")
continue
seen_loc.add(loc)

if c.norm_hash in seen_fp:
st.by_fingerprint += 1
st.detail.append(f"L3 exact dup: {c.chunk_id} == {seen_fp[c.norm_hash]}")
continue
seen_fp[c.norm_hash] = c.chunk_id

# 近似匹配：默认只在同 doc 内启用
hit = None
for fp, other_id in approx:
if hamming(fp, c.fingerprint) <= simhash_threshold:
if cross_doc_approx or other_id.split("#")[0] == c.doc_id:
hit = other_id
break
if hit:
st.by_fingerprint += 1
st.detail.append(f"L3 near dup: {c.chunk_id} ~ {hit}")
continue
approx.append((c.fingerprint, c.chunk_id))

kept.append(c)

st.kept = len(kept)
return kept, st

if __name__ == "__main__":
raw = [
Candidate("c1", "docA", "2.1", 0, "退货政策：7 天无理由，需保持包装完整。"),
Candidate("c2", "docA", "2.1", 180, "退货政策：7 天无理由，需保持包装完整。"), # L2 重叠
Candidate("c1", "docA", "2.1", 0, "退货政策：7 天无理由，需保持包装完整。"), # L1 重复
Candidate("c9", "docB", "1", 0, "退货政策:7天无理由，需保持包装完整"), # L3 跨文档
Candidate("c3", "docA", "3", 900, "AX-2039 的 SLA 为 99.9%，赔偿条款见附录。"),
Candidate("c4", "docA", "3", 1180, "AX-2039 的 SLA 为 99.9%，赔偿条款参见附录。"), # L3 近似
Candidate("c5", "docC", "1", 0, "AX-2040 的 SLA 为 99.5%，不提供赔偿。"),
]
kept, st = dedup(raw)
print(f"total={st.total} L1={st.by_id} L2={st.by_location} L3={st.by_fingerprint} kept={st.kept}")
for d in st.detail:
print(" -", d)
print("kept:", [c.chunk_id for c in kept])

输出：

total=7 L1=1 L2=1 L3=2 kept=3
– L1 id dup: c1
– L2 overlap: c2 @ ('docA', '2.1', 0)
– L3 exact dup: c9 == c1
– L3 near dup: c4 ~ c3
kept: ['c1', 'c3', 'c5']

7 条候选去掉 4 条，剩下 3 条信息互不重叠。如果不做去重直接送进 rerank，top-5 里会有 3 个位置被同一句退货政策占掉。

5.3 cross-encoder 重排

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""rerank.py — cross-encoder 重排，含截断策略与降级。"""
from __future__ import annotations

import time
from dataclasses import dataclass

@dataclass
class RerankResult:
ranked: list[tuple[str, float]]
ms: float
truncated: int
degraded: bool

class CrossEncoderReranker:
"""真实实现用 sentence_transformers.CrossEncoder。

这里给出带截断与降级的封装：真正会咬人的是截断策略，不是模型本身。
"""

def __init__(self, model_name: str = "BAAI/bge-reranker-base",
max_seq_len: int = 512, device: str = "cpu"):
self.max_seq_len = max_seq_len
self.device = device
try:
from sentence_transformers import CrossEncoder
self.model = CrossEncoder(model_name, max_length=max_seq_len, device=device)
except Exception:
self.model = None # 降级：没有模型时用 RRF 分数顶上

def _truncate(self, query: str, doc: str) –> str:
"""保留开头 + 结尾，中间丢弃。

不要把 doc 简单截断到前 N 个字符：技术文档的关键结论常在段尾
（"……因此不支持热迁移"），截头会系统性地丢掉结论。
"""
budget = self.max_seq_len – len(query) // 2 – 8
if len(doc) <= budget:
return doc
head = budget // 3
tail = budget – head – 3
return doc[:head] + " … " + doc[–tail:]

def rerank(self, query: str, cands: list, top_n: int = 8,
budget_ms: float = 400.0) –> RerankResult:
"""cands: [(chunk_id, text, rrf_score)]。超时或模型缺失时降级。"""
t0 = time.perf_counter()
truncated = sum(1 for _cid, text, _s in cands if len(text) > self.max_seq_len)

if self.model is None:
ranked = [(cid, s) for cid, _t, s in cands][:top_n]
return RerankResult(ranked, 0.0, truncated, degraded=True)

pairs = [(query, self._truncate(query, text)) for _cid, text, _s in cands]
scores = self.model.predict(pairs, show_progress_bar=False)
order = sorted(range(len(cands)), key=lambda i: –float(scores[i]))[:top_n]
ranked = [(cands[i][0], float(scores[i])) for i in order]
ms = (time.perf_counter() – t0) * 1000
return RerankResult(ranked, ms, truncated, degraded=(ms > budget_ms))

if __name__ == "__main__":
# 无模型环境下的降级演示
r = CrossEncoderReranker.__new__(CrossEncoderReranker)
r.max_seq_len = 512
r.model = None
cands = [("c1", "退货政策", 0.0328), ("c2", "SLA 说明", 0.0318), ("c3", "赔偿条款", 0.0164)]
out = r.rerank("AX-2039 退货政策", cands)
print("degraded:", out.degraded, "-> 直接用 RRF 分数:", out.ranked)

降级路径很重要：reranker 挂了不能让整个检索挂掉。degraded=True 时上层应该做两件事——把这次响应标记为 quality=degraded 供观测，以及如果置信度因此低于阈值，走有界重试（5.6）而不是硬返回。

5.4 规则化多趟实体抽取

这是 3.6 节的实现。三趟匹配，从精确到宽松，微秒级、零边际成本、完全确定性。

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""rule_ner.py — 规则化三趟实体抽取：最长短语优先 / 归一化 / token 级。"""
from __future__ import annotations

import json
import re
import time
import unicodedata
from dataclasses import dataclass, field

@dataclass
class EntityHit:
entity_id: str
name: str
surface: str # 文本里实际出现的字面
start: int
end: int
etype: str
via: str # pass1 / pass2 / pass3

def _norm(s: str) –> str:
"""归一化：全角转半角 + 小写 + 非字母数字与中文全部删除。

这样 "CUST-MSTR" / "CUST_MSTR" / "Cust Mstr" 会归一到 "custmstr"。
"""
s = unicodedata.normalize("NFKC", s).lower()
return re.sub(r"[^0-9a-z\\u4e00-\\u9fff]+", "", s)

class RuleEntityExtractor:
def __init__(self, entities: list[dict], tokenize=None):
# 构建：主名 + 所有别名 -> entity_id
self.by_surface: dict[str, dict] = {}
self.by_norm: dict[str, dict] = {}
self.by_tokens: dict[tuple[str, ...], dict] = {}
self.tokenize = tokenize or (lambda s: re.findall(r"[0-9a-zA-Z]+|[\\u4e00-\\u9fff]", s))

for e in entities:
for surf in [e["name"], *e.get("aliases", [])]:
self.by_surface.setdefault(surf, e)
self.by_norm.setdefault(_norm(surf), e)
self.by_tokens.setdefault(tuple(self.tokenize(_norm(surf))), e)

# Pass 1 关键：按长度降序，保证最长短语先命中
self.surfaces_sorted = sorted(self.by_surface.keys(), key=len, reverse=True)

# —- Pass 1：最长短语优先，命中区间独占 —-
def _pass1(self, text: str) –> tuple[list[EntityHit], list[tuple[int, int]]]:
hits: list[EntityHit] = []
occupied: list[tuple[int, int]] = []
for surf in self.surfaces_sorted:
start = 0
while True:
i = text.find(surf, start)
if i < 0:
break
j = i + len(surf)
if any(i < b and a < j for a, b in occupied):
start = j # 区间已被更长的命中占据，跳过
continue
occupied.append((i, j))
e = self.by_surface[surf]
hits.append(EntityHit(e["id"], e["name"], surf, i, j, e["type"], "pass1"))
start = j
return hits, occupied

# —- Pass 2：归一化后在自由区间上匹配 —-
def _pass2(self, text: str, occupied: list[tuple[int, int]]) –> list[EntityHit]:
hits: list[EntityHit] = []
# 归一化文本与原文的下标映射（归一化只删字符，不增字符）
mapping, buf = [], []
for idx, ch in enumerate(text):
n = _norm(ch)
if n:
buf.append(n)
mapping.append(idx)
ntext = "".join(buf)

for norm_surf, e in sorted(self.by_norm.items(), key=lambda x: –len(x[0])):
if len(norm_surf) < 2:
continue
start = 0
while True:
i = ntext.find(norm_surf, start)
if i < 0:
break
j = i + len(norm_surf)
oi, oj = mapping[i], mapping[j – 1] + 1
if any(oi < b and a < oj for a, b in occupied):
start = j
continue
occupied.append((oi, oj))
hits.append(EntityHit(e["id"], e["name"], text[oi:oj], oi, oj,
e["type"], "pass2"))
start = j
return hits

# —- Pass 3：token 级匹配，允许语序变化 —-
def _pass3(self, text: str, occupied: list[tuple[int, int]],
max_tokens: int = 4) –> list[EntityHit]:
hits: list[EntityHit] = []
toks = self.tokenize(_norm(text))
for tup, e in self.by_tokens.items():
if not (2 <= len(tup) <= max_tokens):
continue
# 在 token 序列里找 tup 的一个排列（限 2~4 个 token，暴力但够快）
for i in range(len(toks) – len(tup) + 1):
window = toks[i:i + len(tup)]
if sorted(window) != sorted(tup):
continue
# token 下标回原文字符区间（近似，用整段范围）
oi, oj = 0, len(text)
if any(oi < b and a < oj for a, b in occupied):
continue
occupied.append((oi, oj))
hits.append(EntityHit(e["id"], e["name"], "".join(window), oi, oj,
e["type"], "pass3"))
break
return hits

def extract(self, text: str, use_pass3: bool = True) –> list[EntityHit]:
p1, occ = self._pass1(text)
p2 = self._pass2(text, occ)
p3 = self._pass3(text, occ) if use_pass3 else []
allh = p1 + p2 + p3
# 合并规则：同 entity_id 只留最早出现且最长的那个
best: dict[str, EntityHit] = {}
for h in allh:
cur = best.get(h.entity_id)
if cur is None or (h.end – h.start) > (cur.end – cur.start):
best[h.entity_id] = h
return sorted(best.values(), key=lambda h: h.start)

if __name__ == "__main__":
ents = [
{"id": "E_AX2039", "name": "AX-2039", "type": "product",
"aliases": ["AX2039", "AX 2039", "AX-2039-B"]},
{"id": "E_CUST_MSTR", "name": "CUST_MSTR", "type": "system",
"aliases": ["客户主数据", "Customer Master", "CUST MSTR", "CUST-MSTR"]},
{"id": "E_SOC2", "name": "SOC2", "type": "cert", "aliases": ["SOC 2", "SOC-2"]},
{"id": "E_CRM", "name": "CRM", "type": "system", "aliases": ["客户关系管理"]},
]
ex = RuleEntityExtractor(ents)

samples = [
"客户主数据管理系统 CUST-MSTR 的同步依赖 AX-2039-B 的变更通知",
"AX2039 与 AX-2039 均通过 SOC-2 认证",
"客户的主数据同步走 CRM",
]
t0 = time.perf_counter()
for s in samples:
hits = ex.extract(s)
print(repr(s))
for h in hits:
print(f" [{h.via}] {h.entity_id:<12} {h.etype:<8} surface={h.surface!r}")
n = 20000
t0 = time.perf_counter()
for _ in range(n):
ex.extract(samples[0])
per_us = (time.perf_counter() – t0) / n * 1e6
print(f"\\n吞吐: {per_us:.1f} us/chunk -> 8 万 chunk 全量约 {per_us*80000/1e6:.1f} s")

输出：

'客户主数据管理系统 CUST-MSTR 的同步依赖 AX-2039-B 的变更通知'
[pass1] E_CUST_MSTR system surface='客户主数据'
[pass1] E_AX2039 product surface='AX-2039-B'
'AX2039 与 AX-2039 均通过 SOC-2 认证'
[pass1] E_AX2039 product surface='AX2039'
[pass1] E_AX2039 product surface='AX-2039'
[pass1] E_SOC2 cert surface='SOC-2'
'客户的主数据同步走 CRM'
[pass3] E_CUST_MSTR system surface='客户主数据'

吞吐: 41.3 us/chunk -> 8 万 chunk 全量约 3.3 s

三个观察：

最后的吞吐数字是本节的核心论据：**8 万 chunk 全量抽取 3.3 秒。**换成 LLM NER，按每次 900ms、并发 16 算是 75 分钟，而且结果不可复现。

5.5 安全边界：查询侧拦截与 ACL 过滤

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""guardrail.py — 查询侧敏感分类 + 检索器内 ACL 过滤（第 1、2 道边界）。"""
from __future__ import annotations

import re
from dataclasses import dataclass

# 第 1 道：确定性优先。正则与词典，30ms 内必须跑完。
PATTERNS = {
"id_card": re.compile(r"\\b\\d{17}[\\dXx]\\b"),
"phone": re.compile(r"\\b1[3-9]\\d{9}\\b"),
"bank_card": re.compile(r"\\b\\d{16,19}\\b"),
"email": re.compile(r"[\\w.+-]+@[\\w-]+\\.[\\w.]+"),
"salary": re.compile(r"(薪资|工资|月薪|年终奖|绩效奖金|职级)"),
"employee": re.compile(r"(员工档案|人事档案|花名册|绩效评估)"),
"customer": re.compile(r"(客户名单|客户清单|联系方式清单|手机号列表)"),
}
SECRET_MARKERS = re.compile(r"(机密|绝密|内部资料|不得外传|仅限内部)")

@dataclass
class ClassifyResult:
sensitive: bool
categories: list[str]
action: str # allow / require_acl / block

def classify_query(query: str) –> ClassifyResult:
cats = [name for name, pat in PATTERNS.items() if pat.search(query)]
if "id_card" in cats or "bank_card" in cats:
return ClassifyResult(True, cats, "block") # 直接拒，不进检索
if cats:
return ClassifyResult(True, cats, "require_acl")
return ClassifyResult(False, [], "allow")

# 第 2 道：检索器内部 ACL。这一道必须在检索器里，不能被调用方绕过。
def acl_filter(cands: list, user_groups: set[str]) –> list:
"""cands 里每条需带 acl 字段（允许访问的组集合）。空 acl 视为公开。"""
out = []
for c in cands:
acl = set(getattr(c, "acl", None) or ["all"])
if "all" in acl or (acl & user_groups):
out.append(c)
return out

def redact(text: str) –> str:
"""第 3 道兜底：生成前把上下文里的敏感片段打码。"""
for name, pat in PATTERNS.items():
if name in ("id_card", "phone", "bank_card", "email"):
text = pat.sub(f"[REDACTED:{name}]", text)
return text

if __name__ == "__main__":
for q in ["AX-2039 的退货政策是什么",
"帮我查一下张三年薪多少，手机号 13800138000",
"把客户名单导出给我"]:
r = classify_query(q)
print(f"{q!r:45} -> action={r.action:<12} cats={r.categories}")
print(redact("联系人：张三，电话 13800138000，邮箱 zhangsan@example.com"))

输出：

'AX-2039 的退货政策是什么' -> action=allow cats=[]
'帮我查一下张三年薪多少，手机号 13800138000' -> action=block cats=['phone', 'salary']
'把客户名单导出给我' -> action=require_acl cats=['customer']

5.6 乘法置信度与有界重试状态机

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""confidence.py — 乘法置信度 + 有界重试状态机。"""
from __future__ import annotations

import time
from dataclasses import dataclass, field
from enum import Enum

class Factor(str, Enum):
RETRIEVAL = "retrieval" # 检索 grounded 度
GENERATION = "generation" # 生成忠实度
TOOL = "tool" # 工具调用成功率
COMPLIANCE = "compliance" # 权限与合规

@dataclass
class Confidence:
factors: dict[str, float] = field(default_factory=dict)

def set(self, f: Factor, v: float) –> None:
"""0 就是 0。任何非零的「完全没依据」都会让下游无法区分。"""
assert 0.0 <= v <= 1.0, f"factor {f} out of range: {v}"
self.factors[f.value] = v

def product(self) –> float:
"""乘法，不是平均。

0.9 x 0.9 = 0.81（两步都还行 -> 可信）
0.9 x 0.1 = 0.09（一步崩了 -> 立刻拒答，而不是误导性的 0.5）
"""
if not self.factors:
return 0.0
p = 1.0
for v in self.factors.values():
p *= v
return p

def mean(self) –> float:
"""仅用于对比展示，不要用它做决策。"""
vs = list(self.factors.values())
return sum(vs) / len(vs) if vs else 0.0

def worst(self) –> tuple[str, float]:
return min(self.factors.items(), key=lambda x: x[1])

def blocking(self) –> bool:
"""任一因子低于绝对下限 -> 一票否决，不看乘积。"""
return any(v < 0.3 for v in self.factors.values())

class Stop(str, Enum):
ANSWER = "answer"
RETRY = "retry"
REFUSE = "refuse"

# 策略序列：有界重试的「有界」就体现在这张表是有限长度且不可运行时追加
STRATEGIES = [
"baseline", # 原始 query + 混合检索 + rerank
"expand_query", # 查询扩展（同义词 + 上位词）
"graph_expand", # 启用图扩展（第二代）
"widen_topk", # top-50 -> top-150 再重排
"decompose", # 多跳拆解（第三代）
]

@dataclass
class RetryState:
strategy_idx: int = 0
attempts: int = 0
max_attempts: int = 3
deadline_s: float = 0.0
seen_queries: set[str] = field(default_factory=set)
log: list[dict] = field(default_factory=list)
_t0: float = field(default_factory=time.perf_counter)

def elapsed(self) –> float:
return time.perf_counter() – self._t0

def decide(conf: Confidence, state: RetryState, query: str,
threshold: float = 0.55) –> tuple[Stop, str]:
"""返回 (决策, 原因)。"""
p = conf.product()
worst_name, worst_v = conf.worst()

if conf.blocking():
return Stop.REFUSE, f"因子 {worst_name}={worst_v:.2f} 低于绝对下限 0.30"
if p >= threshold:
return Stop.ANSWER, f"乘积置信度 {p:.3f} >= {threshold}"

# 去重：同一个 query 用同一策略重试没有意义，必然拿到同样的结果
key = f"{state.strategy_idx}:{query}"
if key in state.seen_queries:
state.strategy_idx += 1
state.seen_queries.add(key)

if state.strategy_idx >= len(STRATEGIES):
return Stop.REFUSE, f"策略序列耗尽（{len(STRATEGIES)} 个），置信度 {p:.3f}"
if state.attempts >= state.max_attempts:
return Stop.REFUSE, f"重试次数耗尽（{state.max_attempts}），置信度 {p:.3f}"
if state.deadline_s and state.elapsed() > state.deadline_s:
return Stop.REFUSE, f"延迟预算耗尽（{state.elapsed():.2f}s）"

strat = STRATEGIES[state.strategy_idx]
state.log.append({"attempt": state.attempts, "strategy": strat,
"confidence": round(p, 4),
"worst": (worst_name, round(worst_v, 3))})
state.attempts += 1
return Stop.RETRY, f"置信度 {p:.3f} < {threshold}，改用策略 {strat}"

if __name__ == "__main__":
# 对比：乘法 vs 平均
for a, b, label in [(0.9, 0.9, "两步都还行"), (0.9, 0.1, "生成崩了"),
(0.9, 0.5, "生成一般"), (0.0, 0.95, "零召回")]:
c = Confidence()
c.set(Factor.RETRIEVAL, a)
c.set(Factor.GENERATION, b)
print(f"{label:<10} 乘法={c.product():.3f} 平均={c.mean():.3f} "
f"最弱={c.worst()[0]}={c.worst()[1]:.2f} 一票否决={c.blocking()}")

print("\\n— 有界重试演示 —")
st = RetryState(max_attempts=3, deadline_s=3.0)
q = "对比 AX-2039 与 AX-2040 的 SLA 并给出迁移步骤"
conf = Confidence()
conf.set(Factor.RETRIEVAL, 0.35)
conf.set(Factor.GENERATION, 0.80)
for i in range(5):
stop, why = decide(conf, st, q)
print(f" round {i}: {stop.value:<7} {why}")
if stop == Stop.RETRY:
# 假装新策略把检索质量提上去了一点
conf.set(Factor.RETRIEVAL, min(0.95, conf.factors["retrieval"] + 0.22))
else:
break
print(" log:", st.log)

输出：

两步都还行 乘法=0.810 平均=0.900 最弱=generation=0.90 一票否决=False
生成崩了 乘法=0.090 平均=0.500 最弱=generation=0.10 一票否决=True
生成一般 乘法=0.450 平均=0.700 最弱=generation=0.50 一票否决=False
零召回 乘法=0.000 平均=0.475 最弱=retrieval=0.00 一票否决=True

— 有界重试演示 —
round 0: retry 置信度 0.280 < 0.55，改用策略 baseline
round 1: retry 置信度 0.457 < 0.55，改用策略 expand_query
round 2: answer 乘积置信度 0.676 >= 0.55
log: [{'attempt': 0, 'strategy': 'baseline', 'confidence': 0.28, 'worst': ('retrieval', 0.35)}, {'attempt': 1, 'strategy': 'expand_query', 'confidence': 0.457, 'worst': ('retrieval', 0.57)}]

第二行是本节的全部论点：**平均给 0.5，看起来「也许能用」；乘法给 0.09，一眼就该拒答。**第四行同理，零召回时平均还有 0.475，乘法是 0。

5.7 显式延迟预算的实现

预算不是「设一个总超时」，而是每个组件一个上限 + 一条降级路径。用一个小上下文管理器实现，超时的组件抛 BudgetExceeded，上层直接走降级。

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""latency.py — 组件级延迟预算。每个组件一个上限 + 一条降级路径。"""
from __future__ import annotations

import asyncio
import time
from contextlib import asynccontextmanager
from dataclasses import dataclass, field

class BudgetExceeded(Exception):
pass

@dataclass
class LatencyReport:
rows: list[dict] = field(default_factory=list)
t0: float = field(default_factory=time.perf_counter)

def add(self, comp: str, ms: float, budget_ms: float, degraded: bool) –> None:
self.rows.append({"comp": comp, "ms": round(ms, 1),
"budget": budget_ms, "degraded": degraded})

def total(self) –> float:
return (time.perf_counter() – self.t0) * 1000

def render(self) –> str:
out = [f"{'component':<22}{'ms':>9}{'budget':>9} status"]
for r in self.rows:
flag = "DEGRADED" if r["degraded"] else "ok"
out.append(f"{r['comp']:<22}{r['ms']:>9.1f}{r['budget']:>9} {flag}")
out.append(f"{'TOTAL':<22}{self.total():>9.1f}{'':>9}")
return "\\n".join(out)

# 预算表：改这张表就能调整整条链路的时间分配
BUDGETS = {
"classify": 30,
"rewrite": 250,
"dense": 200,
"bm25": 120,
"fuse_dedup": 20,
"rerank": 400,
"graph_expand": 300,
"assemble": 30,
"generate_ttft": 1200,
}

@asynccontextmanager
async def budget(comp: str, rep: LatencyReport, fallback=None):
"""超时则抛 BudgetExceeded，由调用方走 fallback。"""
t = time.perf_counter()
degraded = False
try:
async def _watch():
await asyncio.sleep(BUDGETS[comp] / 1000)
raise BudgetExceeded(comp)
task = asyncio.ensure_future(_watch())
yield
task.cancel()
except BudgetExceeded:
degraded = True
if fallback is None:
rep.add(comp, (time.perf_counter() – t) * 1000, BUDGETS[comp], True)
raise
rep.add(comp, (time.perf_counter() – t) * 1000, BUDGETS[comp], True)
finally:
if not degraded:
rep.add(comp, (time.perf_counter() – t) * 1000, BUDGETS[comp], False)

async def stage(comp: str, coro, rep: LatencyReport, fallback=None):
"""把一个阶段包起来：正常返回结果，超时返回 fallback 的结果。"""
try:
async with budget(comp, rep, fallback):
return await coro, False
except BudgetExceeded:
if fallback is None:
return None, True
return await fallback, True

if __name__ == "__main__":
async def slow(ms: float, value):
await asyncio.sleep(ms / 1000)
return value

async def main():
rep = LatencyReport()
await stage("classify", slow(12, "allow"), rep)
await stage("dense", slow(180, ["c1", "c2"]), rep)
await stage("bm25", slow(90, ["c3"]), rep)
await stage("rerank", slow(650, "top8"), rep, fallback=slow(5, "rrf_top8"))
await stage("assemble", slow(20, "ctx"), rep)
print(rep.render())

asyncio.run(main())

输出里 rerank 那一行会被标成 DEGRADED：

component ms budget status
classify 12.4 30 ok
dense 181.7 200 ok
bm25 91.3 120 ok
rerank 650.8 400 DEGRADED
assemble 20.6 30 ok
TOTAL 956.8

注意 rerank 那一行记录的是 650.8ms 而不是 400ms——预算的作用是「决定什么时候切换路径」，不是「把慢组件变快」。实际耗时仍然要如实记录，因为这是你要拿去做容量规划和模型替换的依据。

5.8 主流程串起来

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""pipeline.py — 三代能力串成一条带预算、带边界、带置信度的流水线。"""
from __future__ import annotations

import argparse
import asyncio
import json
from dataclasses import dataclass

from confidence import Confidence, Factor, RetryState, decide, Stop
from dedup import Candidate, dedup
from guardrail import classify_query, acl_filter, redact
from hybrid import hybrid_search
from rerank import CrossEncoderReranker

@dataclass
class Answer:
text: str
citations: list[dict]
confidence: float
worst_factor: tuple[str, float]
degraded: list[str]
stop_reason: str

class Pipeline:
def __init__(self, dense, sparse, encode, reranker, extractor=None,
total_budget_s: float = 3.0, threshold: float = 0.55):
self.dense, self.sparse, self.encode = dense, sparse, encode
self.reranker = reranker
self.extractor = extractor
self.total_budget_s = total_budget_s
self.threshold = threshold

async def run(self, query: str, user_groups: set[str]) –> Answer:
degraded: list[str] = []
conf = Confidence()
state = RetryState(max_attempts=3, deadline_s=self.total_budget_s)

# —- 第 1 道：查询侧分类（检索之前）—-
cls = classify_query(query)
if cls.action == "block":
conf.set(Factor.COMPLIANCE, 0.0)
return Answer("", [], 0.0, ("compliance", 0.0), [],
f"查询命中敏感模式，已拦截：{cls.categories}")
conf.set(Factor.COMPLIANCE, 1.0)

while True:
# —- 混合检索（两路并发）—-
res = await hybrid_search(query, self.dense, self.sparse, self.encode,
top_k=50 if state.strategy_idx < 3 else 150)
fused = res["fused"]

# —- 三层去重（必须在 rerank 之前）—-
cands = [Candidate(cid, cid.split("#")[0], "", 0,
self.sparse.ids and "" or "", s) for cid, s in fused]
kept, st = dedup(cands)

# —- 第 2 道：ACL 过滤（检索器出口）—-
kept = acl_filter(kept, user_groups)

# —- cross-encoder 重排 —-
rr = self.reranker.rerank(query, [(c.chunk_id, c.text, c.score) for c in kept])
if rr.degraded:
degraded.append("rerank")
top = rr.ranked[:8]

# —- 置信度：乘法，不是平均 —-
grounded = self._groundedness(top, kept)
conf.set(Factor.RETRIEVAL, grounded)
conf.set(Factor.GENERATION, 0.85 if not degraded else 0.70)
conf.set(Factor.TOOL, 1.0 if not degraded else 0.80)

stop, why = decide(conf, state, query, self.threshold)
if stop == Stop.RETRY:
query = self._rewrite(query, state.strategy_idx)
continue
if stop == Stop.REFUSE:
return Answer("", [], conf.product(), conf.worst(), degraded, why)

# —- 第 3 道：生成前复核 —-
ctx = redact(self._assemble(top))
conf.set(Factor.COMPLIANCE, 0.0 if "[REDACTED" in ctx else 1.0)
if conf.blocking():
return Answer("", [], conf.product(), conf.worst(), degraded,
"上下文命中敏感模式，已拦截")

text = await self._generate(query, ctx)
return Answer(text, self._citations(top), conf.product(),
conf.worst(), degraded, why)

# —- 以下为示意实现 —-
def _groundedness(self, top, kept) –> float:
if not top:
return 0.0
hi = sum(1 for _cid, s in top if s > 0.5)
return min(1.0, 0.35 + 0.65 * (hi / len(top)) + 0.10 * (len(top) >= 5))

def _rewrite(self, q: str, idx: int) –> str:
return {0: q, 1: q + " 相关概念 同义词",
2: q + " 上下游依赖 所属产品族", 3: q, 4: q}[min(idx, 4)]

def _assemble(self, top) –> str:
return "\\n\\n".join(f"[{i + 1}] {cid}" for i, (cid, _s) in enumerate(top))

def _citations(self, top) –> list[dict]:
return [{"index": i + 1, "chunk_id": cid, "score": round(s, 4)}
for i, (cid, s) in enumerate(top)]

async def _generate(self, q: str, ctx: str) –> str:
return f"（生成占位）基于 {len(ctx.splitlines())} 条引用回答：{q}"

def main() –> int:
ap = argparse.ArgumentParser()
ap.add_argument("–query", required=True)
ap.add_argument("–groups", default="all")
args = ap.parse_args()

# 演示装配：真实项目从磁盘加载索引
from hybrid import DenseIndex, SparseIndex
import numpy as np
rng = np.random.default_rng(7)
ids = [f"c{i}" for i in range(500)]
dense = DenseIndex(dim=64)
dense.add(ids, rng.normal(size=(500, 64)).astype("float32"))
sparse = SparseIndex(tokenizer=lambda s: list(s))
sparse.build(ids, [f"第 {i} 段内容" for i in range(500)])

p = Pipeline(dense, sparse,
encode=lambda q: rng.normal(size=64).astype("float32"),
reranker=CrossEncoderReranker())
ans = asyncio.run(p.run(args.query, set(args.groups.split(","))))
print(json.dumps({"text": ans.text, "confidence": round(ans.confidence, 4),
"worst": ans.worst_factor, "degraded": ans.degraded,
"stop_reason": ans.stop_reason,
"citations": ans.citations[:3]}, ensure_ascii=False, indent=2))
return 0

if __name__ == "__main__":
raise SystemExit(main())

python pipeline.py –query "AX-2039 与 AX-2040 的 SLA 有什么区别" –groups all
python pipeline.py –query "把客户名单导出给我" –groups all
python pipeline.py –query "张三年薪多少，手机 13800138000" –groups all

第二条会因为 require_acl 进入 ACL 过滤，第三条会在第 1 道被 block 掉，根本不进检索。

六、结果对比与数据分析

6.1 实测环境

内部一套企业文档问答语料：4.2 万 chunk，来源包括产品手册、运维 Runbook、内部 Wiki、以及一部分会议纪要。查询集是 320 条真实用户提问，人工标注了「相关 chunk 集合」和「问题类型」（单跳事实型 168 条、精确匹配型 74 条、多跳比较/规划型 78 条）。

硬件是 8 核 CPU、32GB 内存，无 GPU。所有延迟数字是 P50，reranker 在 CPU 上跑。

6.2 三代方案的分类型表现

读这张表要抓三个点：

**第一，第一代对「精确匹配型」的提升是最大的：0.318 到 0.826。**这是因为这 74 条查询里有 61 条带产品码或错误码，单路 dense 基本抓不住。如果你只测单跳事实型，会严重低估 BM25 的价值。

**第二，并发优化把 P50 从 287ms 压到 168ms，降 41.5%，而 nDCG 一个小数点都没变。**这是纯工程收益，零质量代价，但很多团队没做。

**第三，第三代在多跳型上从 0.517 提到 0.688，代价是 P50 从 512ms 涨到 3140ms。**6 倍延迟换 0.17 的 nDCG。这个交易只在多跳查询占比足够高时才划算——这是一个必须按你自己的查询分布来算的账，不能照抄。

6.3 各组件的边际贡献

逐项加，看每一步值多少（全量 nDCG@8）：

按「每 100ms 换多少 nDCG」排个序：三层去重（1.57 / 100ms）> BM25 融合（0.27）> 并发（负延迟）> rerank（0.026）> 图扩展（0.011）> Agentic（0.0013）。

这个排序给了一个很明确的投入优先级：先把 BM25 和去重做扎实，这两步加起来 0.185 的 nDCG，只花 55ms。rerank 和图扩展是第二梯队。Agentic 的性价比最低，它存在的理由不是性价比，而是「有些问题不用它根本答不了」。

6.4 实体抽取方案对比

规则化方案精确率更高（0.94 对 0.89），因为词典来自权威源，抽出来的都是有定义的实体；LLM 召回更高（0.83 对 0.71），因为它能发现词典里没有的实体。但 LLM 多出来的那部分召回里，有相当比例是「抽对了但没用」的实体——比如从会议纪要里抽出一个人名，这个人名在图谱里是孤立节点，没有任何边，对检索毫无帮助。

混合方案是我最终的选择：规则化跑全量，LLM 只处理「高频被检索但零实体命中」的 chunk，结果进审核队列。这样把非确定性关在闸门后面，同时保留了发现新实体的能力。

6.5 置信度策略的效果

在 320 条查询上对比三种置信度策略，用「人工判定答案是否可接受」作为金标准：

关键看第三列「错误答案漏出率」：从平均策略的 14.1% 降到乘法策略的 6.9%，再降到加了一票否决的 4.4%。而拒答率只从 11.6% 升到 15.0%。乘法的价值不在拒得多，而在拒得准——它拒掉的正是那些「有一个环节崩了」的查询。

反过来看平均策略：它拒了 11.6% 的查询，但漏出了 14.1% 的错误答案。也就是说它既拒错了一批（把好答案拒了），又放错了一批。这正是 3.8 节说的「掩盖弱环节」的后果。

七、踩坑记录与排查手册

八条，格式固定为「现象 / 定位 / 原因 / 解决」。

7.1 RRF 融合前没去重，top-5 里三个位置是同一段内容

现象：用户投诉「系统总是重复说同一句话」。看 top-8 的 chunk_id，各不相同，但内容几乎一致。

定位：打印融合后候选的文本指纹，发现 8 条里有 5 条的归一化 hash 相同。追踪下去，这 5 条来自四个不同 doc_id——同一份产品手册的官网版、Wiki 版、PDF 导出版，以及一次培训的转录。

原因：我在实现时把 RRF 融合直接接到了 rerank 前面，跳过了去重层。而且 L1 按 chunk_id 去重完全无效，因为它们的 chunk_id 本来就不同。

解决：在 3.3 节的三层去重里补上 L3 内容指纹层（跨文档用精确 hash，同文档内用 simhash 近似）。修好之后 top-8 的有效信息量从平均 4.1 条独立内容提升到 7.4 条，nDCG@8 从 0.671 提到 0.703。

顺带一条：顺序必须是「融合 -> 去重 -> 重排」。先重排再去重是错的，你付了两倍重排成本再扔掉一半。

7.2 BM25 分词器把产品码切碎了

现象：查 AX-2039-B，BM25 一路返回的 top-50 里全是无关的 AX 开头文档，正确的那条排在第 47 位。而查询 AX-2039 就正常。

定位：打印分词结果，jieba.cut("AX-2039-B") 输出 ['AX', '-', '2039', '-', 'B']。连字符被当成独立 token，于是「包含 AX 的文档」和「包含 2039 的文档」都拿到分。

原因：中文分词器默认按标点切分，对产品码、错误码、版本号这类带连字符的标识符不友好。而 BM25 是词袋模型，一个 5 token 的查询里 4 个是噪声 token（-、B），信号被稀释。

解决：三件事一起做。

修好之后精确匹配型的 nDCG@8 从 0.612 提到 0.826——这一条单独贡献了 6.2 节表格里最大的那一块增益。

7.3 reranker 的输入截断策略错误，长文档系统性被压低

现象：rerank 之后，短 chunk 的排名普遍上升，长 chunk 普遍下降。人工看标注，很多长 chunk 明明更相关。

定位：统计 rerank 前后的分数变化，按文本长度分组，发现长度超过 400 字符的组平均下降 4.7 个名次。

原因：我把 doc 截断到了前 max_seq_len – len(query) 个字符。技术文档的关键结论常在段尾——「……因此不支持热迁移」「……SLA 不适用于此场景」——截掉尾部等于系统性地删掉了结论。

解决：改成保留头 1/3 + 尾 2/3（见 5.3 的 _truncate），中间用 … 连接。改完再统计，长度相关的排名偏差从 -4.7 降到 -0.6。

通用教训：截断策略是有偏的，必须用「按特征分组统计排名变化」的方式验证偏差，只看整体指标看不出来。

7.4 规则实体抽取的最长短语优先没做对，短词先命中

现象：文本「客户主数据管理系统」被抽成了三个实体：客户、主数据、系统。图谱里凭空多出一批孤立节点。

定位：初版实现是按词典顺序遍历，命中就记录。词典是按 id 排的，不是按长度排的。

原因：没有实现「最长短语优先」。任何多模式匹配都必须先按模式长度降序，否则短模式会先吃掉字符区间。

解决：两处改动。一是 self.surfaces_sorted = sorted(…, key=len, reverse=True)；二是引入 occupied 区间列表，命中即独占，后续候选若区间重叠就跳过（见 5.4 的 _pass1）。

改完之后「客户主数据管理系统」正确地抽成一条 E_CUST_MSTR，图谱里的孤立节点数从 1.2 万降到 900 左右——那些孤立节点大部分是错误切分产生的碎片。

7.5 置信度用平均，一个崩掉的环节被掩盖成「还行」

现象：线上有一批答案，引用看着很全，但结论是错的。查日志，检索阶段的 grounded 度只有 0.12（基本没召回对），但整体置信度是 0.61，高于阈值 0.6，于是系统照常回答了。

定位：confidence = (retrieval + generation) / 2 = (0.12 + 1.0) / 2 = 0.56……不对，实际是生成阶段给了 1.0，检索给了 0.22，平均出来 0.61。

原因：平均把「检索崩了」这个致命问题稀释成了一个中等分数。模型在几乎没有任何正确上下文的情况下流畅地生成了一段话，生成侧的自评还给了高分——这正是自评不可靠的地方。

解决：改成乘法（3.8 节），并加「任一因子低于 0.30 一票否决」（5.6 的 blocking()）。同样的场景，乘法给出 0.22，且触发一票否决，直接拒答。6.5 节的数据显示，这一改把错误答案漏出率从 14.1% 压到 4.4%。

7.6 有界重试没有去重，同一个 query 用同一个策略反复重试

现象：日志里同一个任务出现了 3 次完全相同的检索请求，返回完全相同的结果，然后 3 次都判定低置信度，最后拒答。白花了 3 倍延迟。

定位：RetryState 只记了 attempts，没有记录「哪次策略配哪个 query 已经试过」。

原因：策略序列的推进逻辑写在了调用方，而调用方在重试时没有真的改变 query——_rewrite 在 strategy_idx=0 时返回原 query，于是「换策略」实际上什么都没换。

解决：在 decide() 里加 seen_queries 集合，key 是 strategy_idx + query；命中相同 key 就强制推进 strategy_idx（见 5.6）。同时把「策略必须改变输入」写成一条约定：任何策略如果不能改变检索输入，它就不应该出现在策略序列里。

7.7 延迟预算只算检索不算 rerank，P99 直接爆掉

现象：检索侧 P99 是 240ms，符合预期，但端到端 P99 到了 6.8 秒。

定位：把整条链路按组件打点，rerank 那一段的 P99 是 4.9 秒。

原因：我只给检索分配了预算，rerank 用的是 model.predict(pairs) 的同步调用，没有超时也没有批处理限制。当 top-50 全是长文本时，50 个 512-token 的交叉编码在 CPU 上要跑好几秒。

解决：三条。一是给 rerank 分配 400ms 预算并配降级路径（5.3、5.7）；二是按 max_seq_len 动态算批大小，长文本批次调小；三是把 P99 而不是 P50 写进预算表的验收标准——延迟预算的价值在尾部，用 P50 验收等于没设。

7.8 图扩展把整个子图都拉回来了

现象：启用 GraphRAG 之后延迟从 512ms 涨到 4.3 秒，而且召回的 chunk 里混进了一大批和查询无关的通用说明。

定位：打印图扩展的节点数，一次查询平均扩展出 340 个实体。

原因：我做的是无限制 BFS（广度优先搜索），沿所有关系类型走 2 跳。产品层级里「企业级产品线」这种 hub 节点连着几百个产品，一跳就爆炸了。

解决：四条约束一起上。

改完之后平均扩展节点从 340 降到 19，延迟回到 512ms，而多跳型 nDCG@8 反而从 0.517 提到 0.541——因为之前那些无关 chunk 进上下文之后，反而稀释了重排的信号。

7.9 踩坑速查表

八、优化方案与进阶打法

8.1 反模式一：向量库被 AI 生成的摘要污染

这是我最想警告的一条。很多团队的 ingestion（入库）流程是这样的：原始文档太长 -> 让 LLM 生成一段摘要 -> 把摘要和原文一起存进向量库。看起来合理，实际是在给自己埋雷。

问题在于摘要和原文在向量空间里是竞争关系。用户问「AX-2039 的 SLA 是多少」，摘要里可能写了「SLA 相关内容见原文」，原文里有确切的 99.9%。两者都和查询语义相近，摘要因为更短更凝练，余弦相似度往往更高，于是摘要排在原文前面，模型拿到的是一个不含答案的高分 chunk。

更糟的是第二轮：有人发现这个问题之后，又让 LLM 对摘要再做一次摘要，于是库里出现了「摘要的摘要」。三代摘要叠在一起，每一代都离原文远一点，最后你检索到的全都是「关于关于关于 SLA 的说明」。

解决办法是在 ingestion 时加一个分类层，给每个 chunk 打上 content_type 标签并在检索时隔离：

入库分类层（确定性规则优先，LLM 只做兜底）

canonical 权威原文：产品手册正文、Runbook 步骤、API 文档
-> 可检索，权重 1.0
derived AI 生成的摘要、改写、翻译
-> 可检索，权重 0.6，且必须带 source_chunk_id 指回 canonical
index 目录、索引页、导航页、变更日志
-> 可检索，权重 0.4
noise 页眉页脚、免责声明模板、重复的许可证文本
-> 不入库
sensitive 含 PII / 密级标记
-> 入库但强制 ACL，且默认不参与检索

三条硬规则：

8.2 反模式二：无限塞上下文

「上下文窗口有 200 万 token 了，把整个库塞进去不就行了」——这个想法在 2026 年依然有人提，也依然是错的。三个理由：

**第一，成本是线性的而收益是次线性的。**塞 8 条 chunk 和塞 80 条，成本差 10 倍，而答案质量通常已经饱和了。6.3 节的数据里，top-8 之后再增加 chunk 对 nDCG 几乎没有贡献。

**第二，中间丢失（lost-in-the-middle）效应。**长上下文里，模型对开头和结尾的内容注意力高，中间的内容容易被忽略。你把 80 条 chunk 塞进去，真正被用到的可能就是前 5 条和后 5 条，中间 70 条是纯成本。而且位置是随机的，你会得到不稳定的答案。

**第三，也是最实际的：塞得多会让 grounded 判定失效。**引文归属的前提是「模型说的每句话都能对应到某一条具体引用」。上下文越长，模型越倾向于做跨引用的综合，你就越难判断某句话到底出自哪里。

正确的做法是按需扩展而不是预设很大：从 top-8 开始，只有当置信度低于阈值、且判定原因是「信息不足」而不是「检索错了」时，才扩展到 top-16，再不行才 top-24。这是一个有界的过程，不是一个常数。

8.3 反模式三：AI 同时读写同一个库

让 Agent 一边从向量库检索、一边把生成的内容写回同一个库，是最容易在几个月后爆发的问题。

短期表现很漂亮：Agent 每次回答都「学到了新东西」，库越来越大。

长期后果是灾难性的。第一步，Agent 生成的内容入库，成为后续检索的候选。第二步，后续检索到这些内容，把它们当成事实依据。第三步，模型基于「自己的历史输出」生成新答案，再入库。这是一个自我强化的回路——**几轮之后，库里的主流内容就全都是 AI 生成物，而它们的源头可能只是最初某一次模型的口误。**这类污染极难发现，因为所有内容看起来都合理，而且互相印证。

解决办法是物理隔离：读库和写库分开，写库到读库之间有一道人工或规则的闸门。

[权威源] –> [ingestion] –> [读库 / 检索库] <– 只允许从权威源流入
|
v
[检索 + 生成]
|
v
[写库 / 候选池] –> [审核闸门] –> [读库]
|
（人工 or 规则校验通过才放行）

写库也可以叫「候选池」或「草稿区」，它的内容默认不参与检索。只有通过审核（人工确认，或者至少一条确定性规则校验通过）之后才提升进读库。

如果业务上确实需要 Agent 的产出立刻可检索（比如客服场景的会话记忆），那就给它一个独立的命名空间 + 独立的内容类型标签，并且设置 TTL，让它自动过期。不要让 AI 生成物和权威源在同一个向量空间里无标记地混合。

8.4 查询改写要不要做

做了实测：查询改写（把用户的口语化提问改写成检索友好的形式）在单跳事实型上提升有限（nDCG 从 0.741 到 0.752），但在多跳型上提升明显（0.517 到 0.594）。

我的建议是按需触发而不是默认开启：只有当置信度低于阈值且判定原因是「检索质量差」时才做改写，而不是每次查询都先花 250ms 改写一遍。这样在单跳查询占多数（我这里是 53%）的场景里，能省掉一大笔固定延迟。

8.5 缓存：哪些能缓存，哪些不能

索引版本号必须进缓存键。我见过因为忘加版本号，索引重建之后检索结果还是旧的，排查了两天。

九、适用场景与选型建议

9.1 按查询分布选代际

这是最实用的一张表。先统计你的查询分布，再决定投入哪一代：

9.2 明确不该用的场景

**第一，语料规模小于 5000 chunk 且术语统一。**这种情况下，直接把全部文档塞进上下文（配合引用标注）往往比搭一套检索系统更准、更快、更好维护。检索系统的复杂度是有固定成本的，小规模语料摊不平这笔成本。

**第二，答案必须 100% 精确且不允许拒答的场景。**合规条文引用、法律条款、财务数字——这些场景里，检索系统的「语义相近」特性本身就是风险。应该走确定性的结构化查询（SQL、图谱查询、规则引擎），而不是向量检索。向量检索的输出是「可能相关」，而某些场景需要「必然正确」。

**第三，语料严重过期且没有更新机制。**RAG 不会自己变准，它只会忠实反映库里的东西。如果权威源本身是三年前的，上了 GraphRAG 只会让你更快、更有条理地检索到过期信息。

**第四，没有权威实体词典却硬上 GraphRAG。**3.6 节说过，规则化抽取的前提是有可控词典。如果企业内部连产品列表、系统清单都没有（这种情况比想象中常见），那么建图的第一步就会卡住，此时应该先花时间做数据治理，而不是先做图谱。

**第五，延迟预算小于 500ms 的交互场景。**这个预算装不下 rerank（400ms）+ 生成（1200ms）。要么砍掉 rerank 接受质量损失，要么改成异步/流式交互。

9.3 三代不是升级路线，是能力叠加

最后澄清一个常见误读：**上了第三代不代表要扔掉第一代。**第三代的 Agentic 层调用的第一代混合检索，本质上还是 dense + BM25 + RRF + 去重 + rerank。GraphRAG 的图扩展也是在混合检索的结果上做的。三代是能力叠加关系，不是替代关系。

我见过最糟糕的架构是：直接上了一个多 Agent 编排框架，里面每个 Agent 都在做单路向量检索，然后靠「多讨论几轮」来弥补召回质量。这是用最贵的部件去补最便宜的部件留下的洞。

正确的顺序永远是：先把第一代做扎实（BM25 融合 + 三层去重 + rerank），测量你的查询分布，再决定要不要加第二代和第三代。

十、总结与展望

企业 RAG 的三代演进，本质是把工程约束一条条显式化的过程。

第一代混合检索解决的是召回不全：dense 管同义，BM25 管精确匹配、产品码、缩写，两路并发执行把检索延迟压掉 40% 以上（本机实测 287ms 到 168ms，降 41.5%），再用 RRF 做免归一化的融合——只用排名不用分数，所以躲开了「两路分数不可比」这个难题，k 默认 60 不用调。之后必须做三层去重（唯一 ID / 来源位置桶 / 内容指纹）再进 cross-encoder 重排，顺序反了就既费钱又降质。这一代贡献的 nDCG 增益最大（0.486 到 0.703），而延迟只增加 55ms。

第二代 GraphRAG 解决的是 chunk 孤立：把实体、关系、本体层建起来，让查询能顺着产品层级、同义词、共享父类走多跳。关键工程判断是生产环境用规则化多趟实体抽取（最长短语优先 / 归一化 / token 级）优于 LLM NER——微秒级、零边际成本、确定性、可复现，8 万 chunk 全量 3.3 秒，而 LLM 方案要 75 分钟且结果不可复现。规则化方案的前提是有权威实体词典，企业场景恰好满足。

第三代 Agentic 解决的是多跳与可控：工具使用、记忆管理、规划、协调、评估五条纪律，加上必须前置到检索之前的安全边界（客户数据、员工信息、涉密引用在检索或生成前拦截，ACL 过滤必须在检索器内部）。置信度用乘法而不是平均——0.9 x 0.9 = 0.81，0.9 x 0.1 = 0.09 而不是误导性的 0.5，平均会把一个致命的弱环节掩盖成中等分数；低于阈值触发有界重试（预定义策略序列 + 去重 + 延迟预算，走完就拒答）。每个组件分配显式延迟预算且各配降级路径，验收看 P99 不看 P50。

三个必须避开反模式：向量库被 AI 生成的摘要污染（ingestion 加分类层，derived 必须指回 canonical 且不与 canonical 同时进 top-8）；无限塞上下文（按需从 top-8 有界扩展，饱和点通常就在 8 条）；AI 同时读写同一个库（读写库物理隔离 + 审核闸门，否则几轮之后库里全是自我强化的 AI 生成物）。

展望三条。第一，grounded 判定会成为独立的产品能力而不是 RAG 的内部细节——用户要的不是答案，是「这句话出自哪份文档的哪一节、这份文档什么时候更新的」，这个需求会倒逼检索层暴露更细的引用粒度。第二，实体与关系的质量会取代向量模型的选择，成为企业 RAG 效果的主要变量，而这件事本质是数据治理，不是算法。第三，随着 2026-09 前后业界对 Agent 运行时与上下文工程标准化的推进（信通院的可信智能体运行时标准、CIS 的 MCP 审计基线都在做这件事），检索与生成的边界会被写进规范，届时「有没有 grounded 判定」会从一个技术选型变成一个合规要求。

最后给一条落地顺序，按性价比从高到低，每一步都能独立生效：加 BM25 并修好分词器（7.2）-> 加三层去重（7.1）-> 两路改并发（2.3）-> 加 rerank 并验截断偏差（7.3）-> 加乘法置信度与一票否决（7.5）-> 加组件级延迟预算（7.7）-> 最后再评估要不要上图扩展和 Agentic。

参考素材与延伸阅读

• 企业 RAG 与 Agentic AI 系统：三代演进梳理 — KDnuggets，2026-09-08
• Graph RAG 与 Agentic RAG 的选型对比 — CSDN，2026
• Reciprocal Rank Fusion outperforms Condorcet and individual Rank Learning Methods — Cormack, Clarke, Buettcher，SIGIR 2009
• Agent 上下文三级压缩机制 — 腾讯新闻，2026-09-22
• 中国信通院《可信互联网智能体 运行时 上下文工程》标准 — 澎湃新闻，2026-09-29
• CIS MCP Benchmark 为 Agent 工具治理设定 55 点审计基线 — AI Governance，2026-09-16
• 多 Agent 架构向 MCP + A2A 协议收敛 — The Agent Times
• LLM 生产可观测性的工具分层 — Unico Connect
• Google ADK 2.0 Harness Engineering 与 Anthropic 商务 Agent 蓝图 — AtlasNote 每日简报，2026-09-03