从文档、权限与索引到可以核查的证据
2026-10-21
企业知识库最容易展示的部分，是输入一个问题后出现一段流畅答案。真正决定它能否被信任的，却往往是演示之外的细节：制度昨天已经变更，为什么仍引用旧金额；员工刚被移出项目，为什么缓存里还有项目原文；同名文档来自另一家企业，为什么会进入当前回答。
这本书从这些问题出发，搭建一个可以运行、检查和推翻的检索系统。它接收 Markdown 和限定结构的 JSON，把文本投影成可定位的片段，写入真实的 SQLite FTS5 表，计算词项向量或可选的本地模型向量，再执行混合召回、规则重排和引用复核。更新、删除和撤权不是末尾的提醒，而是贯穿索引生命周期的行为约束。
随书核心实验只依赖 Python 标准库与支持 FTS5 的 SQLite。另有已经实际执行的 BGE 中文模型量化 ONNX 实验，用来观察词面匹配与模型表示的差别。核心的词项计数向量没有语义模型，书中不会把它包装成语义嵌入；回答函数直接摘录数据库原文，没有调用大语言模型，也不声称完成了任意语义蕴含判断。
全书使用八篇作者构造的示例文档与十三条标注查询。企业、员工、金额和秘密标记都是教学数据。小数据集可以清楚暴露因果关系，也限制了结论：命中一条同义改写，不能证明模型在真实业务中普遍更优；权限结果测试全部通过，不能证明不存在时间或统计侧信道。
建议按三遍阅读。第一遍顺序理解文档、索引、权限和证据的契约；第二遍运行第十三、十四章，查看原始 JSON 报告与失败测试；第三遍带着自己的文档格式和权限事件，阅读增量更新、墓碑与生产边界。不要先替换模型再寻找问题，应先确认究竟是哪一个契约被破坏。
正文十六章之外，附录提供来源、实验记录和迁移清单；构建时还会把实际实验源码和测试放入代码附录。在线页、EPUB、离线 HTML 与完整 Markdown 来自同一份书稿，实验包与附录代码同源。读者可以脱离在线服务检查本书的核心行为。
本书的目标不是交付一套现成企业产品，而是让你能回答四个具体问题：检索的是什么版本，当前是谁有权看，引用能否重建，以及系统在什么时候应该停止回答。一个允许读者验证这些答案的系统，才有继续扩展的基础。
假设差旅制度规定酒店每晚五百元，系统也准确回答了五百元。这个成功仍不完整。如果回答引用的是另一租户的制度，或者该制度已经废止，文字再准确也没有用。检索工程的第一项工作，是说明“正确”依赖哪些状态，而不是只看答案中的数字。
随书实现把链路拆为源文档、规范化投影、片段、召回结果和回答证据。源文档说明原始内容从哪里来；投影规定不同输入格式如何成为统一文本；片段保存投影中的起止位置与摘要；召回结果携带文档版本和权限版本；回答阶段重新读取当前数据库，确认这条链仍然成立。
例如 acme/travel 的第一版内容经过 NFC
与换行规范化后，产生一个文档摘要。片段记录 [start,end)
和自己的 SHA-256。查询中的 Context("acme", "alice")
决定允许参与检索的集合。摘录时，代码再次比对文档版本、文档摘要、ACL
版本和片段内容。任何一步失效，都不能继续沿用旧证据。
这些字段各自解决不同的问题。摘要标识内容，不能代表授权；文档版本说明更新顺序，不能替代内容校验；ACL 版本说明策略变化，不能替代用户身份认证。把三个概念揉成一个更新时间字段，会使重放事件、撤权与内容变更难以区分。
本书的核心承诺可以逐条运行：没有可信查询上下文时拒绝检索；不同租户的同名文档不能串用；更新退款政策后旧引用失效；Alice 被撤权后，旧命中不能继续组装回答；删除文档后，旧内容事件不能让它自动复活。
这些承诺比“安全可靠”更窄，也更有检查价值。tests/test_retrieval.py
中每条测试都构造一个具体状态变化，再观察真实 SQLite
数据或返回值。例如撤权测试先保留搜索结果，再修改
ACL，然后把保留结果交给回答函数。它覆盖了两次调用之间的状态变化，而非只检查一个静态权限函数。
工程上还应分开两类错误。相关性错误是可见文档中没有找到好证据；权限错误是不可见文档进入结果。扩大候选池可能改善前者，却不能接受后者。评估报告因此单列拒答与权限违规，不把所有失败压缩成一个平均分。
实验提供 Index 类和可信上下文对象，没有 HTTP
服务、登录系统或身份提供方。调用方负责把已经认证的身份转换为上下文，不能直接相信浏览器提交的租户和用户名。status()
是维护接口，会返回文档内部状态，也不能直接公开给终端用户。
向量检索扫描当前可见片段，适合观察行为，不是大规模近邻索引。全文检索使用共享 FTS5 表，虽然结果受到租户和 ACL 限制，底层词频统计仍然共享。书中验证结果隔离，没有审计严格的统计非干扰或时间侧信道。
读完这一章，应能为自己的系统画出一条证据链，并给每一段写一个能实际失败的断言。下一章先固定输入数据，因为没有稳定来源，后面的版本与评估都无法解释。
知识库常从一个共享目录起步。文件名被当成文档身份，修改时间被当成版本，上传者填写一组权限。等到同名文件覆盖、同步乱序和部门调整同时出现，这些隐含规则就会互相冲突。索引之前，应把来源契约写清楚。
本书用 (tenant, doc_id)
组成文档唯一键。数据集中，acme 和 beta 都有
travel，前者住宿上限五百元，后者九百元。两篇文档的编号相同不代表它们可以合并；同一个
alice 字符串出现在不同租户，也不代表拥有跨租户权限。
SQLite 的 docs
表对这两个字段建立联合唯一约束。查询时租户来自可信上下文，并进入每条权限筛选
SQL。这个约束要落实在存储与查询里，不能只靠拼接不同的文件路径。测试刻意保留同名文档，正是为了使遗漏租户条件的实现立即暴露。
文档使用递增正整数
revision。相同版本且规范化摘要相同，是可重复处理的同一事件；相同版本但摘要不同，是冲突；较小版本是过期事件。系统拒绝后两种情况，而不是猜测最后到达的消息一定最新。删除也占用一个新的内容版本，以便挡住延迟到达的旧内容。
lab/data/corpus.json
保存八篇文档，包含内容、格式、租户、版本和初始
ACL。queries.json
保存十三个查询及相关文档标签。manifest.json 保存两个文件的
SHA-256，加载器在读取时复核它们。这样，一份报告对应哪版输入就可以被检查。
这些指纹没有证明制度内容来自真实企业。文档和查询都由作者编写，只是可审查的教学样例。SECRET-ACME-BOB
与 SECRET-BETA
是用来观察权限边界的合成标记，不是实际机密。数据作者、标签作者与实现作者相同，也意味着这套评估没有独立盲测的性质。
真实接入时，应增加源系统地址、原文件指纹、连接器版本、采集时间与授权依据。不要把原始文件摘要和本书的投影摘要混淆：两个不同的 JSON 排版可能投影成同一正文，而一个原始文件经过不同解析器也可能产生不同投影。
第一次接入文档时，初始 ACL
是明确列表；空列表拒绝所有人，星号只表示该租户内的用户。后续内容更新保留已有
ACL，避免旧同步任务把已经撤销的用户重新加回去。权限更新走独立的
set_acl()，通过 expected_acl_revision
比较交换，拒绝过期策略写入。
这是一项有意的取舍：上传新正文时附带的 ACL 不会自动覆盖当前授权。如果源系统把正文和策略作为一个原子事件发送，连接器应建立更完整的事件协议，而不是依赖调用顺序碰运气。本书选择分离两个状态机，让撤权含义保持清晰。
接入下一批文档前，先回答身份是否稳定、版本能否比较、权限由谁授权、删除是否有墓碑。只有这些答案清楚，才值得讨论哪一种模型更能理解内容。
一段回答附带“来源：员工手册”只能说明候选出处。要核查金额是否确实存在，需要知道检索处理后的文本是什么，以及引用位置怎样映射回它。解析器的产出因此应包含稳定投影，而不是一串无法重建的清洗结果。
lab/text.py 的 normalize() 把 CRLF 和 CR
换为 LF，再执行 Unicode NFC 规范化。Markdown 输入保留正文标记，采用简单
ATX 标题表达式寻找标题，没有把 Markdown
渲染为纯文本。规范化后的字符串就是后续切分、摘要和引用共同使用的基准。
JSON 输入只接受一种明确结构：顶层恰好包含 title 与
sections，每个章节恰好包含 heading 与
text，相关值必须是规定类型。投影依次生成一级标题、二级标题和段落，章节之间用空行连接。不认识的格式和多余字段会报错，而不是被悄悄丢弃。
{"title":"远程办公申请","sections":[
{"heading":"审批流程","text":"申请居家办公须部门负责人同意，每周最多2天。"}
]}这份结构会成为一段 Markdown 风格的规范文本。引用坐标指向这段投影，不指向原始 JSON 的字节位置。接入 PDF、表格或网页时，也必须说明投影保留了什么结构、丢失了什么关系，以及原始页码如何另行保存。
本书的 [start,end) 使用 Python Unicode
码点索引，左闭右开。它不是 UTF-8
字节偏移，也不是手机屏幕上看到的字形数量。一个组合字符在 NFC
前后可能有不同长度；换行转换也会改变原始文件位置，因此坐标只能作用于同一版规范文本。
片段必须满足 body[start:end] == quote，并把片段 UTF-8
编码后的 SHA-256 记录下来。回答函数不是相信搜索结果中的
text
字段，而是重新从当前数据库读出片段，再复核它与文档切片的关系。测试把调用方命中的文本改成伪造内容，得到的仍是数据库里的真实摘录。
摘要也有边界。一个匹配的 SHA-256 说明投影内容没有意外变化，不说明内容事实正确，更不说明用户有权阅读。引用完整性、授权和业务真实性需要分别判断。
当前 Markdown 标题识别只是正则表达式，会把围栏代码块里形似标题的行也当成标题元数据。它没有实现完整 CommonMark 语法、表格关系、附件、PDF OCR 或网页正文抽取。示例文档使用简单标题与自然段，因此这个范围足以解释实验，但不能直接推广到任意企业文档。
面对复杂输入，应先保留原始对象与解析日志，选择适配格式的解析器，再建立可回放的投影版本。解析失败时应保留旧的有效索引，而不是先删除旧内容再尝试新解析。随书更新流程先在事务外完成解析，失败不会触碰已有数据库状态。
可以用一条简单验收检查解析接入：从最终引用反查规范文本，再回到带版本的源对象，是否有任何一步只能靠人工猜测。如果有，问题应在投影层解决，不能交给回答模型补全。
把整本手册作为一个向量，容易让短小规定淹没在大量背景里；把每句话拆开，又可能把“适用于正式员工”与后面的金额限制分离。切分不是寻找一个神奇长度，而是在召回粒度、上下文和引用稳定性之间做可检查的选择。
chunks()
接收规范文本、窗口大小和重叠长度。默认窗口二百二十个码点，重叠四十；随书小语料为了观察跨片段行为，使用八十与十六。窗口到达中间位置以后，如果后半段存在空行，就尽量在空行后结束；否则按窗口上限切断。
每个片段携带顺序号、起止位置、原文、最近的标题和摘要。最近标题不等于完整标题树，长文档中可能缺少更高层的语境。索引没有把标题额外拼接进片段正文，以免引用坐标与实际正文混淆；标题在规则重排中作为单独特征使用。
from lab.text import chunks
body = "# 制度\n\n" + "说明文字" * 80
pieces = list(chunks(body, size=80, overlap=16))
assert all(body[p["start"]:p["end"]] == p["text"] for p in pieces)下一片起点取
max(start + 1, end - overlap)，即使自然段很长或边界安排不理想，也会向前推进。参数要求窗口在一到四千零九十六之间，重叠不小于零且小于窗口。测试覆盖超长自然段的有限推进、正文覆盖和坐标一致性。
同一段制度可能出现在相邻两个片段里。它们分别命中检索，不能证明有两个来源支持金额。回答阶段会跳过同一文档同一版本中与已选证据区间相交的片段，避免直接重复摘录；不相交的片段即使都被保留，也仍来自同一篇文档。
评估排名同样先按文档编号去重，再取前三篇。否则一篇相关文档的三个重叠片段会占满榜单，令片段命中数看起来很好，实际上没有增加证据覆盖。报告保留
returned_chunks，使读者能区分返回片段量和评估文档量。
窗口变大可能减少截断，却也可能把多个制度条款混在一起。重叠变大可以补回边界语境，但增加存储、编码时间和重复候选。要调整它们，应固定文档与查询，逐项比较遗漏的答案位置、重复比例和引用长度，而不是只看一个总体召回分数。
同一正文在八十与二百二十窗口下，会得到不同片段编号、边界和向量。索引把编码器标识、窗口和重叠写入投影契约。用不同配置打开已有数据库会得到
index_projection_contract_changed_rebuild_required，要求重新建立索引。
教学实现选择整体拒绝混用，而没有实现双版本在线迁移。生产环境可以在新索引完整构建和验证后切换读指针，但必须让查询向量、片段与引用坐标属于一致版本。让旧片段和新编码器短暂混用，并不能通过最终一次全量重建来解释期间的错误回答。
切分验收的终点应是证据能否恢复，而不是片段数量是否整齐。下一章把这些片段放入真正的全文索引，观察词面召回究竟做了什么。
一个可解释的词面基线有两个作用：它可以直接解决编号、产品名与制度关键词查询，也可以帮助区分模型收益来自哪里。假如系统连“退款政策”这样的标题词都找不到，应先检查解析和索引，再考虑换模型。
SQLite FTS5 的默认 Unicode 分词器不会自动理解本书需要的中文词边界。随书代码在进入 FTS5 前先生成词项：ASCII 字母、数字与下划线组成连续词，并统一小写；连续中文生成相邻双字词项，只有一个汉字的连续段保留单字。比如“住宿发票”会生成“住宿、宿发、发票”。
这是一种便于观察的字符方案，没有领域词典，也不等同于中文词法分析。它可能召回共享局部字符的无关内容，对同义改写尤其脆弱。它的价值是规则透明、依赖少，可以作为后续表示学习的对照。
索引创建了真实虚拟表，而不是用 Python 列表模仿全文搜索：
CREATE VIRTUAL TABLE chunks_fts USING fts5(terms);每个片段的预处理词项用空格连接写入 terms 列，FTS
行号与片段主键一致。写入、替换和删除时维护这份对应关系。文档表保存投影正文，片段表保存坐标与向量，FTS
表负责倒排召回；三者各有职责。
搜索把去重后的前六十四个查询词项加引号，再用 OR 连接。整个 MATCH 表达式通过参数绑定传入 SQL；词项内部的引号也被转义。用户输入中的操作符不会直接成为任意全文查询语法，输入长度被限制为一千零二十四个码点。
参数绑定防止 SQL 结构被用户拼接，但不意味着任意 MATCH 表达式都安全。若直接把原始查询作为全文语法传入，仍可能得到不可控的运算或语法错误。本书明确选择有限的词项 OR 查询，测试包含引号和操作符输入，验证实际调用可执行且不会绕过过滤。
FTS5 使用 bm25(chunks_fts) 排序。按照 SQLite FTS5
文档，这个辅助函数把更好的匹配放在更小的数值方向，因此代码使用升序。不要把其他系统中“分数越大越好”的经验直接套进来。
词面 SQL 同时关联文档，并要求租户相同、文档未删除、ACL 包含当前用户或租户内星号，然后才排序并限制前一百条。先取全局前一百、再过滤权限，可能让大量不可见结果挤掉可见文档，也更容易使日志和调试接口暴露内容。
不过，共享 FTS 表的文档频率等统计仍然是全局的。过滤返回行解决结果可见性，没有自动提供严格统计隔离。若租户之间连统计影响都不可接受，应使用独立索引或经过专门设计与审计的隔离方案。
实际评估中，词面召回在十条正例上的 Recall@3 与 MRR@3 都为零点九，漏掉“旅馆住一夜能花多少钱”。报告中的前三文档已经去重；这个结果只描述八篇作者文档，不能代表企业搜索平均水平。下一章用相同语料对比词项向量与实际模型表示。
向量只是数值表示的形式，不保证有语义能力。把双字词项计数装进向量，再计算余弦相似度，仍然依赖词面重叠。理解这一点，可以避免把一段很小的数学代码误称为已经完成了语义检索。
默认 TermEncoder
用前一章的词项规则生成计数表。两个片段的余弦值等于词项计数点积除以各自长度。稀疏字典不需要预先固定词表，但只有共享词项才会产生非零点积；没有出现过的同义表达，不会因为采用余弦就自动接近。
词项模式的演示截断阈值为零点零五。这个值只控制是否把候选放进列表，没有经过业务相关性概率校准。词面和词项向量在当前语料里都得到零点九的 Recall@3，正好说明改变排序数学形式，并没有解决那条同义查询。
本书另行运行了 BAAI/bge-small-zh-v1.5
的中文表示模型，使用 Xenova/bge-small-zh-v1.5
仓库的社区量化 ONNX
导出。原模型和导出不是同一个分发物；书中记录了双方提交号、模型文件与
tokenizer 的 SHA-256，没有声称量化文件就是原始全精度权重。
编码器通过 ONNX Runtime 的 CPU 执行提供方直接推理，不依赖云
API。它读取 tokenizer，按图输入要求提供
input_ids、attention_mask 与可能需要的
token_type_ids，取末层输出第一个 token 的表示，再执行 L2
归一化，得到五百一十二维向量。
短查询前添加“为这个句子生成表示以用于检索相关文章：”，文档不添加该指令；输入最多五百一十二个 token。这与 BGE 原模型卡 中的 CLS、归一化与检索指令说明相对应。不同任务仍需比较是否加指令更合适。
在相同八篇文档与十三条查询中，真实模型向量找到了“旅馆住一夜能花多少钱”对应的差旅规定。十条正例的 Recall@3、MRR@3 都达到一点零，三个负例没有返回候选。词面召回在同一次模型实验中仍为零点九，便于检查输入没有随模型一起改变。
这个观察支持一个很窄的结论：该量化模型在这组构造数据里弥补了一条词面遗漏。它不支持中文检索普遍提升百分之十、适用于所有企业或胜过其他模型的结论。数据量小、标签由作者编写，阈值零点四也是演示参数，不是可靠相关性的通用分界。
代码把模型文件、tokenizer 指纹和编码方案放进编码器标识，因此替换权重需要重建索引。即使向量维度相同，不同模型或池化方案的坐标空间也不能混用。
本书没有引入近似最近邻库，而是对权限允许的片段逐一算余弦。这样可以看清授权先于评分的顺序，代价是线性扫描。实际规模增长时，需要重新评估过滤召回策略、向量存储和近邻索引，不能用这个小实验的毫秒数推断线上吞吐量。
词面检索擅长精确术语，模型向量可能补回同义改写。直接相加原始分数却很危险：FTS5 的 BM25 越小越靠前，余弦越大越靠前，两者数值范围与含义都不同。混合检索需要明确融合规则，而不是给两列数字随便配权重。
随书实现采用 Reciprocal Rank Fusion，简称
RRF。对于候选片段，分别查找它在词面和向量列表中的名次，每个出现的通道贡献
1 / (60 + rank)，再把贡献相加。名次从一开始，不在某个通道中出现就不获得该通道贡献。
如果一个片段在两个通道都排名第一，融合分数为
2 / 61；另一个片段只在一个通道排名第一，得到
1 / 61。这个例子解释了融合偏好多个通道支持的候选，也提醒我们：两个通道若高度相似，重复支持并不等于增加了独立证据。
常数六十沿用 Cormack 等人的 RRF 论文 中的参数。论文在自己的实验条件下选择并验证它，本书没有重新证明六十最适合中文企业知识库，也没有把论文的提升幅度移植到本书结果中。
每条命中保存
channels，其中包含通道名、原始名次与原始分数，还保留总
rrf。读者可以从 JSON 直接重算融合值。测试
test_hybrid_exposes_channels_and_rrf_arithmetic
就用实际搜索结果验证这个等式，而不只检查出现了名为 RRF 的函数。
这种可观察性对排错很关键。若退款文档被排到后面，先看它是否进入过候选池，再看各通道名次，最后检查重排特征。一个未召回的文档不可能被重排拯救；一个召回后被标题规则降下去的文档，也不能归咎于向量模型。
每个通道最多保留一百个候选，最终返回数量由调用者设置，范围一到一百。这个上限控制教学实现的候选规模，也形成召回边界。如果相关文档位于通道第一百零一名，融合看不到它。评估候选深度时应同时记录权限后的候选数量，而不只记录最终答案数。
使用词项向量时，词面、向量与混合三个模式在当前数据上都是零点九。使用真实 BGE 时，向量与混合都为一点零，混合没有超越表现最好的单一通道。三个负例的空结果行为也相同。
因此，本书没有“混合一定优于向量”的实验结论。融合提供结构上可解释的组合方式，实际效果取决于通道的互补性、候选深度与后续排序。若新通道不断引入噪声，RRF 也可能改变原本正确的顺序。
把词面、向量和混合三份报告同时保存，比只公布最好的数字更有用。下一章继续检查融合后的重排，并区分排序收益、证据重复和真正的相关性判断。
“加入重排模型”不是一个完整设计。需要先回答：候选是什么，重排使用哪些信息，输出如何影响证据，以及失败时怎么定位。本书采用一个故意简单的规则，让这些问题在没有额外模型的条件下仍能被看见。
实现先计算去重查询词项与文档标题词项的交集，除以查询词项数量，得到
title_overlap。开启重排时，按标题重合降序，再按 RRF
降序，最后用片段主键稳定打破平局。关闭时，标题特征不参与优先级，主要依据
RRF。
这不是把两个数加权平均，而是字典序排序。一个标题重合更高的候选，即使融合分数稍低，也会排到前面。这个选择简单直观，也可能过度偏好通用标题。例如“报销说明”标题很相关，但正文只有流程，真正的金额表可能在标题较弱的附件中。
因此，书中的 rerank=True
明确表示标题规则重排，不是交叉编码器、学习排序或 LLM
判断。代码没有加载重排权重，报告也没有把规则特征说成模型推理。将来引入实际重排模型，应分别版本化权重、输入模板、长度截断和排序策略。
评估一个重排器，至少需要看重排前的候选覆盖和重排后的首个相关位置。若相关文档已经在候选池里但顺序靠后，重排可能改善 MRR；若从未进入候选池，重排没有机会补救。只比较最终答案是否“看起来更好”，会把召回与排序混为一谈。
随书评估默认开启这条规则，JSON 明确记录
rerank: true。当前报告没有单独设计足够丰富的重排困难集，因此不能据此宣称标题规则获得了普遍提升。evaluate(index, mode, rerank=False)
保留了做对照的入口，读者扩充标签以后可以再比较，而不是从本书的小样例推导收益。
候选池以片段为单位；评估先按文档去重；摘录则跳过同文档同版本的重叠区间。这三个操作目的不同。候选去重避免完全重复对象，文档去重避免排名指标被相邻片段灌满，区间去重避免回答反复输出同一段文字。
单纯按 doc_id
只保留一个片段，可能丢掉一篇制度中分别解释申请资格和金额上限的两段必要证据。因此回答实现允许同篇文档中不重叠的片段进入结果，但仍不能把它们描述为两个独立来源。若业务要求多来源交叉核查，需要额外的源对象身份与独立性规则。
还要注意摘录数量上限。默认两条证据限制输出长度，不保证两条足够覆盖复杂问题。返回两段原文只是一个可核查的检索产物，距离自然语言的完整答案仍有业务判断距离。
好的重排设计会留下可反查的名次、特征和版本。发生错误时，工程师应能指出哪条规则把哪份文档提前了，而不是只说“系统认为它相关”。下一章把这些候选变成可以复核的引用。
搜索结果中的原文并不天然可信。它可能来自很早以前的缓存，也可能被调用者改写；即使原文没变，当前用户也可能已经失去权限。因此，引用的关键动作发生在组装回答时，而不是只在检索成功时加一个来源链接。
随书命中携带租户、用户、文档编号、内容版本、ACL
版本、文档摘要、片段主键、片段摘要与坐标。answer()
首先验证可信上下文，再要求命中的租户和用户与本次调用一致。随后重新读取文档与
ACL，拒绝删除、换版、撤权或摘要变化的对象。
代码继续从当前片段表读取内容，检查片段属于该文档、摘要符合命中记录，并确认片段文本与当前正文切片相同。最终使用的是数据库片段文本，不使用调用者传来的
hit["text"]。这避免把“搜索结果里有原文”误当成可以直接转发原文的授权。
hits = index.search("退款政策", Context("acme", "alice"))
# 两次调用之间，文档或权限都可能改变。
result = index.answer(hits, Context("acme", "alice"))这些检查在一个读事务中完成，使同一次摘录看到一致的数据库状态。它没有跨越应用外部的网络发送过程：读取完成以后发生的撤权，不能收回已经交付给用户的内容，也不能自动擦除日志或屏幕截图。
有可用证据时，函数返回
state="extractive"、non_llm=True，以及原文组成的答案和结构化证据。无可用证据时返回
state="abstain"，提示“没有当前权限下可核查的证据”。这里没有调用语言模型，也没有把不同片段改写成一个新结论。
这个实现核查的是证据完整性和当前可见性，不是任意语义蕴含。原文确实存在，也可能不回答用户的全部问题；标题碰巧相近的片段，也可能在业务上无关。如果要生成综合回答，仍需要单独检查每项断言由哪段证据支持，以及引用是否真的支撑它。
第一次演示检索住宿报销，得到文档 travel
第一个片段，坐标零到六十三。摘录同时包含住宿五百元与餐费八十元，其
SHA-256 为
6c0b230598d5e8b3bf87b66874730c931bf32b7584dff3f0d1a4d6590885b808。读者可以直接对报告中的
quote 重新编码计算，或在规范正文上重建相同切片。
回答函数会跳过失效命中，继续检查后面的可用候选。这意味着某篇文档删除后，若列表里还有其他当前可见证据，整体回答仍可能是摘录状态。不能把“目标文档不可引用”写成“任何查询都应完全拒答”。
删除演示因此先保留目标文档的命中，再单独验证这些命中全部失效，同时检查新搜索不再返回该文档。这样的断言精确对应删除承诺，避免因为另一个合法候选存在而误判系统错误。
引用对象应当是一张等待复核的凭据，不能成为绕过权限的通行证。接下来的三章分别改变正文、删除状态和 ACL，观察同一条凭据如何失效。
退款政策从三十天改为十四天，如果文档表已经更新而全文索引尚未替换，查询仍可能命中旧数字。向量表、片段坐标和引用摘要也可能暂时属于不同版本。增量索引的核心问题，是怎样让读者看到完整旧版或完整新版。
upsert()
先在事务外验证参数、解析输入、生成规范正文、切分并编码。解析器失败或编码器返回数量异常时，数据库尚未变化。这样既保留已有有效投影，也避免在模型计算期间长时间占有写事务。
准备完成后，代码进入
BEGIN IMMEDIATE，重新读取当前文档版本，再决定是否幂等返回、拒绝过期事件或写入新版本。这个重新读取很关键：准备工作可能耗时，不能根据编码前的旧状态假定更新仍然有效。
写入时移除旧片段与对应 FTS 行，更新正文、标题、版本和摘要，再插入新片段、向量与 FTS 词项，最后提交。任意一步抛出异常就回滚。向量存为 JSON，禁止 NaN 序列化；本书没有在事务内部访问外部向量服务。
相同版本、相同规范化摘要且文档仍存活时，返回
unchanged。相同版本却内容不同，或输入版本落后，返回
stale_or_conflicting_revision。这允许消息传递重复，但不把到达时间当成事实先后。
幂等比较针对规范投影，原始 JSON 空格变化不一定触发索引变化。若业务需要保存源对象每次修改，应另设源事件版本；不要要求一个内容摘要同时承担审计原文件与去重索引的职责。
已有文档的内容事件保留当前 ACL。即使旧连接器在正文更新中又带回 Alice，代码也不会因此重新授权。内容版本和权限版本独立，这使“先撤权，再收到新正文”不会悄悄恢复访问。
演示先缓存第一版退款政策命中，再写入第二版十四天政策。旧命中经
answer()
复核后拒绝，数字“30”的新搜索为空，新摘录包含“14天”。第二版投影摘要记录为
e74729dbab7284e5e7d8a9f4e5aa18882154b79cd3b07e667fdafa27f0bf0f94。
这些检查同时覆盖可检索内容与引用有效性。只断言数据库正文出现十四天，不足以证明旧 FTS 行被清除；只检查新搜索正确，也不足以证明缓存命中已经失效。两个方向都需要运行。
SQLite
事务文档
说明，同一数据库可以有多个读事务，但同时只有一个写事务。BEGIN IMMEDIATE
提前尝试取得写入能力，其他写者活动时可能返回忙。本书用进程内可重入锁串行使用单连接，没有构造多进程高并发服务，也没有设置自动重试队列。
扩展成服务时，需要限制写入批次、处理忙错误、观察写入延迟，并保证重试仍遵守版本规则。若采用外部全文与向量系统，单个 SQLite 事务便无法包住所有状态，应引入版本化投影、事件重放与切换协议。这个小实验提供的是一致性目标，而不是所有存储的通用事务方案。
“搜索不到了”只是删除的一个表面结果。若同步队列里还保留旧内容事件，系统可能在下一次重试时把文档重新写回来。若缓存命中仍能摘录，用户甚至不用再搜索。一个明确的删除协议，必须同时回答这两个问题。
随书 delete()
要求指定比当前内容版本更大的整数。它在写事务中清除片段、FTS 行和
ACL，把正文置空，标记
deleted=1，并推进内容与权限版本。文档身份和墓碑仍然保留，因此系统知道这个编号已经被主动删除。
保留墓碑意味着旧内容的 revision=1
再次到达时，不会被当成全新文档。即使使用更高版本，默认更新也会得到
deleted_document_requires_explicit_restore。恢复必须明确传入
restore=True，并提供新的显式
ACL，让“恢复内容”和“重新授权”成为一个有意识的动作。
这个教学接口没有实现删除事件的重复幂等确认；再次使用相同删除版本会被拒绝，因为它要求新版本。连接器可以记录事件处理结果来处理重复投递，或者设计完整的删除事件身份，但不能把接口报错随意当成删除成功。
第一步，在删除前保存目标文档命中；删除后交给
answer()，应该无法再形成证据。第二步，执行新搜索，确认返回列表没有目标编号；单查正文标记不够，因为重复的
FTS 行或旧向量仍可能存在。第三步，重放旧版本内容，确认它被墓碑挡住。
演示用年假文档完成这组操作：先检索并保留 leave
的命中，随后以第二版删除，旧证据拒绝、新搜索没有该文档，重放第一版得到
stale_or_conflicting_revision。测试还覆盖较高版本未明确恢复时仍被拒绝，以及明确恢复后使用新授权列表。
如果同一查询还匹配其他合法文档，检索可以继续返回那些文档。删除承诺针对目标身份与内容，不要求所有带有共同词语的查询都变成空结果。对实验断言保持这种精度，可以避免把合法检索行为误当成删除失败。
本书的保证是当前活动索引不再返回正文或引用。SQLite 数据页、日志、备份、源系统、观察日志和已经交付的回答不在这个保证内。墓碑还保留标题、身份和摘要等维护信息，不能声称删除后所有元数据消失。
若业务要求物理擦除，需要另外定义保留策略、备份淘汰、密钥生命周期、审计记录与验证方法。简单执行 SQL DELETE 不等于完成所有副本的销毁，本书也没有进行磁盘取证或备份清除实验。
只要旧事件仍可能出现，丢弃墓碑就可能失去防重放信息。实际系统应把源事件保留上限、同步水位、备份恢复窗口和墓碑生命周期联系起来。不能仅以“最近没人搜索”为依据删除墓碑。
删除因此属于索引生命周期的一等状态。它不是一次清理脚本，而是与版本顺序、恢复授权和引用复核共同工作的协议。下一章讨论内容没有删除、权限却改变时，应怎样执行相同的失效原则。
权限系统最危险的时刻，常常发生在合法访问之后。Alice 昨天有权读取项目，今天被移出；搜索缓存仍在，浏览器还有一个命中对象。若回答函数把“曾经检索到”理解为“现在仍可展示”，撤权就没有真正生效。
实验要求调用方传入
Context，租户和用户只能包含受限身份字符。没有该对象就拒绝查询，未知租户看不到内容。但这只是接口契约，不是认证方案；任何能任意创建
Python 对象的调用者，都可以构造不同身份。
上线时应由服务端认证层产生上下文，把令牌、组织成员关系和实际租户映射验证完毕，再调用检索。不能把用户提交的两个字符串直接包装成
Context，然后声称系统完成了企业授权。维护方法
status()
也没有终端用户权限检查，必须保持在受信管理边界内。
词面查询在 SQL 中先限制租户与 ACL，再排序限额；向量查询先从数据库读取允许的片段，再计算余弦。因此当前实现不会先把不可见原文交给向量评分循环，再事后隐藏结果。空 ACL 拒绝全部用户，星号只在同一租户条件下生效。
回答时再次检查 ACL 与权限版本，弥补搜索和输出组装之间的状态变化。搜索结果携带的用户也要与当前上下文相同，避免把 Bob 的命中对象交给 Alice 复用。测试同时覆盖跨租户同名文档、跨用户命中和空权限列表。
撤权演示先以 Alice 搜索远程办公，再调用 set_acl() 只保留
Bob。旧 Alice 命中不能形成摘录，新搜索不再出现远程办公文档；Bob
的新查询仍能访问。这既证明拒绝发生，也证明没有把撤权错误实现成对所有人删除文档。
set_acl() 要求调用者给出
expected_acl_revision，与数据库当前值一致才替换列表，随后版本加一。两个管理员并发修改时，较旧的写入会得到
stale_acl_revision，必须重新读取策略后决策。
正文更新保留当前 ACL，所以携带旧成员列表的新内容事件不会恢复已撤销权限。相反，显式恢复已删除文档会建立新的 ACL；调用者应先确认恢复依据，而不是自动沿用删除前的缓存策略。
共享 FTS 表的全局词频会影响分数，向量扫描耗时也可能随可见片段数变化。本书没有审计计时、统计或资源竞争侧信道；测试通过不等于严格租户非干扰。高隔离场景应考虑租户独立索引、独立资源和专门的威胁分析。
撤权以数据库当前读快照和回答组装边界为准，不能收回已经发出的原文。外部缓存、追踪日志与用户终端也需要自己的保留和清除协议。这里的复核使旧命中失去继续输出的资格，没有承诺网络传输与权限变更形成一个全局原子事务。
把权限写进每个阶段的契约，比给最终页面加一个“仅内部可见”标签更可靠。接下来运行完整实验，亲自查看这些状态变化与拒绝结果。
本章把前面的契约放进同一个可重复实验。核心链路没有模型下载、账号或服务密钥，只需要 Python 3.10 及以上和启用了 FTS5 的 SQLite。建议用 Python 3.12 建立独立环境，避免系统解释器的扩展缺失影响结果。
在仓库根目录执行：
python3 -m venv .venv
. .venv/bin/activate
python -c 'import sqlite3; c=sqlite3.connect(":memory:"); c.execute("CREATE VIRTUAL TABLE probe USING fts5(text)"); print(sqlite3.sqlite_version)'
python -m unittest discover -s tests -v探针实际创建一张 FTS5 表，比只打印 SQLite 版本更能验证所需能力。如果
sqlite3 无法导入，或报告没有 FTS5 模块，应换用包含该功能的
Python 发行版。不要通过跳过探针把故障掩盖成一次成功实验。
未设置模型目录时，核心二十二条测试运行，两条可选 ONNX 测试明确跳过。它们不是两个成功结果。CI 也采用这个无模型下载路径；实际本地模型验证见下一章，两类记录在交付信息中分别列出。
python -m lab.demo --output /tmp/knowledge-demo.json
python -m lab.evaluate --output /tmp/knowledge-term-evaluation.jsondemo 创建临时 SQLite
文件并载入清单校验后的文档。它先搜索住宿规定并输出摘录，随后完成退款版本更新、Alice
撤权、年假删除和旧内容重放。程序结束后临时数据库清除，JSON
报告保留在显式指定的位置。
报告中的 old_30_day_fact_retrievable
为假，新答案包含十四天；stale_revision_citation_rejected、revoked_alice_citation_rejected
与 deleted_citation_rejected
为真；bob_access_preserved
为真。三个私有、跨租户和不相关查询的返回数量为
[0,0,0]。
注意这些字段来自真实断言通过后的输出，不是配置文件写入的期望值。任何断言失败，实验都会异常退出，不会把后面的成功字段正常输出。对照源码，可以看到每项声明对应哪些搜索和数据库操作。
test_actual_fts5_table_returns_ranked_rows
检查实际虚拟表与命中；test_incremental_update_removes_old_fact_and_invalidates_citation
同时观察旧数字和缓存引用；test_new_content_event_cannot_regrant_old_acl
覆盖正文事件重新授予权限的反例。
还有一个容易忽略的测试，把候选的文本字段替换为伪造字串，再调用回答函数。系统仍从数据库读出规范片段。这验证了引用不是简单转发调用方内容，但没有把数据库本身当作不受信攻击者；存储篡改防护属于更大的完整性设计。
比较语料指纹、模式、文档排名、拒答和状态迁移，不要要求
wall_ms 完全相同。机器、Python 与 SQLite
版本会影响耗时，也可能影响同分排序的细节。当前实现用主键稳定打破同分，在同一语料顺序下应得到相同逻辑结果。
仓库的 lab/evidence/
保存本次真实运行记录，供离线检查；读者的新报告建议写到
/tmp，保留原始证据不变。若修改语料，必须更新清单并重新评估，不能继续把旧报告当成新系统的证据。
上一章证明的是检索与权限状态机可运行。本章单独加入一个真实本地表示模型，观察它能否补回词面基线遗漏的同义查询。模型是可选路径，下载和推理都不参与核心实验，也没有调用大语言模型生成答案。
原模型为 BAAI/bge-small-zh-v1.5，模型卡提交为
7999e1d3359715c523056ef9478215996d62a620；本书下载社区 ONNX
导出 Xenova/bge-small-zh-v1.5，提交为
75c43b069aac4d136ba6bc1122f995fedcfd2781。下载脚本使用后一个固定提交，避免默认分支更新后得到不同文件。
python -m pip install -r lab/requirements-embedding.txt
python -m lab.fetch_embedding脚本依赖 curl，下载后校验模型和 tokenizer 的完整
SHA-256，临时文件通过校验才替换目标。模型文件约二千四百万字节，保存在忽略提交的
.tools/models/bge-small-zh-v1.5/。仓库与实验 ZIP
不分发权重，使用者应同时查看原模型卡的 MIT 声明与社区导出说明。
文件校验不等于证明来源无误，但能固定本书实际使用的分发物。已下载且摘要符合的文件直接复用；不符合的文件不会被默默当成同一个模型。两份完整摘要见来源与实验附录，运行时也会写进报告。
BGE_MODEL_DIR=.tools/models/bge-small-zh-v1.5 \
python -m unittest discover -s tests -v
python -m lab.evaluate \
--onnx-model .tools/models/bge-small-zh-v1.5 \
--output /tmp/knowledge-bge-evaluation.json本次实际运行使用 Python 3.12.15、ONNX Runtime 1.31.0、NumPy 2.5.4、tokenizers 0.23.3。推理固定 CPU 执行提供方和单线程设置，采用 CLS 池化与 L2 归一化。二十四条测试全部通过，其中两条确实加载了 ONNX 文件，不是以跳过替代验证。
第一条可选测试检查向量维度五百一十二、数值有限且范数接近一；第二条实际用模型查询同义表达，并验证私有文档与其他租户仍被过滤。模型改善相关性不能成为放松权限顺序的理由。
观察 q07“旅馆住一夜能花多少钱”：词面结果为空，BGE 向量返回
travel。其余正例在前三篇去重文档中也都找到标注来源，因此当前十条正例的
Recall@3 与 MRR@3
均为一点零。混合模式同样为一点零，没有比向量模式更高。
模型也给部分查询返回更多非首位候选。排名指标达标不代表每个候选都相关，更不代表所有片段都能直接作为答案。演示阈值零点四控制候选进入，不能被解释成百分之四十的正确概率。
本机耗时只描述这次小语料线性扫描与 CPU 推理。报告没有并发压测、吞吐量、模型热启动分布或生产延迟保证。若在其他架构遇到依赖轮子缺失，可以使用兼容环境复现；不要改成词项向量后仍把报告标作真实 BGE 实验。
这一章的价值，是让“使用了嵌入模型”有可追查的权重、推理代码、输入契约和运行证据。接下来把单条成功放回整体评估边界，避免从小样例推出过大的承诺。
如果评估只收集系统能回答的问题，模型很容易显得无所不能。企业知识库还必须面对不存在的事实、无权访问的文档、旧版本和错误租户。它们不是正例召回的附属指标，而是系统是否会在不该输出时停下来的证据。
本书语料八篇，查询十三条，其中十条有相关文档、三条没有当前上下文可用的相关文档。评估固定为
acme/alice，每次搜索最多十个片段，按文档编号去重后只保留前三篇。当前每条正例只有一篇标注相关文档。
Recall@3 计算前三篇覆盖相关文档的比例，MRR@3 使用第一个相关文档名次的倒数，找不到记零，最后仅在十条正例上求平均。三条负例不被当成“零相关文档所以召回满分”，而是单列空候选次数。
| 表示方式 | 模式 | Recall@3 | MRR@3 | 负例空候选 |
|---|---|---|---|---|
| 词项计数 | 词面 | 0.9 | 0.9 | 3/3 |
| 词项计数 | 向量 | 0.9 | 0.9 | 3/3 |
| 词项计数 | 混合 | 0.9 | 0.9 | 3/3 |
| 量化 BGE | 词面 | 0.9 | 0.9 | 3/3 |
| 量化 BGE | 向量 | 1.0 | 1.0 | 3/3 |
| 量化 BGE | 混合 | 1.0 | 1.0 | 3/3 |
所有模式都开启标题规则重排。量化 BGE 模式下的词面通道仍然只使用 FTS5；表格把它保留下来，是为了让比较条件透明。
两个表示方案的差异集中在 q07，同义表达没有共享足够词项，模型向量却返回了住宿规定。只有十条正例，命中一条就改变零点一，所以不能把这个差值描述成稳定业务提升。报告逐条保存排名，让读者能发现总体变化究竟由几条样例贡献。
三条负例分别搜索当前用户无权读取的合成标记、其他租户标记和无关词串。权限断言还检查所有命中的租户及已知私有文档编号，违反时直接中止评估。报告中的
access_violations=0
表示这些断言全部通过，没有覆盖任意文档权限组合或未知攻击类型。
评估报告在初始语料上计算相关性。政策从三十天变十四天、撤权、删除和重放属于另一组状态变化，由演示和单元测试验证。不能因为静态 Recall@3 满分，就宣称新版本不会残留或权限缓存一定安全。
同样，负例空候选是本书的拒答代理：没有候选，摘录函数便无法形成证据。它没有评估生成模型是否会在空上下文中编造答案，因为本书根本没有生成模型。将来接入 LLM 后，需要另测回答层的事实支持与拒答行为。
真实业务应增加跨时间独立测试集、多个相关来源、难负例、同名制度、不同权限角色和撤权后的缓存查询。标签说明应写清文档相关还是片段足以回答，必要时由独立人员复核。调过阈值的查询不应继续冒充从未见过的验收样本。
当前数据全部由作者构造，没有做总体统计推断，也没有模型排行榜意义。把这些限制留在报告里，能够防止未来换了标题以后，教学样例被误用成商业效果证明。一个可解释的小结果，比一个没有边界的大数字更适合指导下一次实验。
实验结束时，系统已经能解析文档、建立索引、检索、摘录，并处理更新、删除与撤权。把它用于生产仍需要更多部件。扩展的正确顺序，应围绕已经识别的契约补齐能力，而不是先把 SQLite 换成多个服务，再重新寻找一致性定义。
连接器需要稳定文档身份、可比较版本、删除事件和明确策略来源。拉取失败不能自动解释成源文档已删除；短暂没有权限读取，也不等于应在其他系统中公开旧缓存。应保存同步水位和失败原因，让维护者区分未到达、解析失败、已撤权与已删除。
复杂格式解析应产生带版本的规范投影和原始位置映射。PDF 页码、表格单元格与网页段落无法凭一个字符坐标通吃。新增解析器时，先让引用可以回到源对象，再测召回；否则更好的排名只会更快找到无法核查的文字。
编码器、tokenizer、池化、查询指令和切分配置都属于索引契约。大规模迁移可以并行建立新索引，完成数据校验和权限同步，再切换读取版本；旧索引保留窗口必须服从删除与撤权要求。切换后回滚，也不能把已删除文档和旧 ACL 一起复活。
若使用近似近邻索引，要观察过滤后召回是否下降，以及权限变更多久影响候选。外部向量服务的过滤语法、候选深度和一致性不同，不能用“支持 metadata filter”替代真实撤权实验。每种部署形态都应重跑缓存命中再撤权的反例。
可记录投影版本、事件版本、各通道候选数量、过滤后数量、引用失效原因、拒答次数与写事务失败。原始查询和正文可能包含敏感信息，应按实际权限与保留策略处理；调试日志不应成为绕过检索 ACL 的另一套知识库。
引用失效应区分内容换版、策略换版、文档删除和片段不一致。把它们统一报成“没有答案”，会掩盖索引损坏；把完整原文放进错误日志，又可能扩大泄露面。维护接口与终端回答应具有不同的可观察细节。
一次可追查发布至少包含源码提交、语料清单、编码器指纹、评估报告与状态迁移测试结果。书的发布也遵循这个原则：EPUB、在线阅读和实验 ZIP 来自同一次构建，校验文件与 Release 使用同一份产物。只改交付记录不会自动重建电子书，避免内容相同却反复产生不同二进制文件。
生产服务还需要认证、访问审计、容量规划、备份恢复和物理删除策略。本书没有实现这些设施，也没有高可用或侧信道审计结论。将它们逐项加入验收，比在文档末尾笼统写“注意安全”更有行动价值。
最终可以用四个问题检查系统：这段证据来自哪个投影版本，当前用户为什么有权读，原文如何被重新定位，变化发生后旧证据如何失效。模型、数据库与部署平台都可以替换，但这些答案应当始终具体、可运行、可反驳。
以下一手资料于 2026-10-21 获取并阅读。源码仓库的
book/sources.json 记录完整
URL、快照摘要、使用章节与支撑范围；原站
HTML、论文和模型权重不随书再分发。引用用于解释机制，实验结果来自随书代码实际运行，两者不能互相替代。
SQLite FTS5 Extension
用于核对虚拟表、默认 Unicode 分词器、MATCH 与 BM25 辅助函数。快照
SHA-256：de3092630358518a6869bf047055b3938e18b476e3fbc54f4e2146d1e593497f。文档属于可更新页面，摘要固定本次阅读快照，不代表上游永久版本。
本书自己选择预生成中文双字词项、词项 OR 查询、候选上限和 ACL SQL。它们不是 SQLite 文档推荐的通用企业检索方案，也不是对 SQLite 隔离能力的完整安全审计。
Transaction
支撑读写事务、单写者与 BEGIN IMMEDIATE 的说明。快照
SHA-256：b65dc308fd9e0ce471844c97366a4f5ad3a1f42833a3b19486ad7d6555a8e24e。本书实际只测试单进程教学实现，没有把文档中的并发机制转换为自己的并发压测结果。
Gordon V. Cormack、Charles L. A. Clarke、Stefan Buettcher，2009，Reciprocal
Rank Fusion outperforms Condorcet and individual Rank Learning
Methods。PDF
SHA-256：5c1f9f1fe0d80f67d82041bbe151ea17437bc7a7d9a920564cf122d1b09eafb5。
使用其按名次倒数求和的公式，以及论文中固定六十的参数背景。论文数据集与本书的八篇构造文档不同，不移用其提升幅度、显著性或比较结论。
BAAI/bge-small-zh-v1.5
模型卡，提交
7999e1d3359715c523056ef9478215996d62a620，快照
SHA-256：c48a4eeea77f6b1d38b48ec1c5b8d4f86d5550cc43fa345a0db1b2ca1d082369。用于核对模型身份、MIT
声明、CLS 池化、归一化与中文检索指令。
Xenova/bge-small-zh-v1.5
导出说明，提交
75c43b069aac4d136ba6bc1122f995fedcfd2781，快照
SHA-256：cb22054df1a98f7ee9d5b281d023e67c27cc316e1733baf03fc994d95c3048ab。本书使用其量化
ONNX 文件，明确标注社区导出，不将其称为原始全精度分发物。
文档可以说明如何使用模型，不能代替本书的真实推理。运行依赖、权重摘要和实际逐条排名见附录B与保存的 JSON 报告。
本次实验日期为 2026-10-21，使用 Python 3.12.15 与 SQLite 3.53.1。所有制度、身份、秘密标记和查询均为作者构造。记录不是独立业务验收集、模型排行榜或生产性能基准。
语料版本
knowledge-fixture-v1，八篇文档，十三条查询，其中十条正例、三条负例。lab/data/manifest.json
保存 corpus.json 与 queries.json 的
SHA-256，加载时自动复核，避免报告与输入无声错位。
lab/evidence/demo.json 记录真实 SQLite
操作：退款三十天改十四天、旧引用失效；Alice
撤权后旧引用与新搜索拒绝，Bob
保持访问；年假删除后不再可检索，旧事件重放拒绝；私有、跨租户、无关查询计数均为零。回答标明
non_llm=True。
term-evaluation.json 保存词项基线逐条结果，三个模式
Recall@3 与 MRR@3 均为零点九。bge-evaluation.json
保存真实量化模型结果，词面零点九，向量与混合一点零；所有模式负例空候选三次、权限断言零违规。
量化 ONNX 文件 model_quantized.onnx：24,010,842
字节，SHA-256
15b717c382bcb518ba457b93ea6850ede7f4f1cd8937454aa06972366cd19bcc。
tokenizer.json：439,125 字节，SHA-256
48cea5d44424912a6fd1ea647bf4fe50b55ab8b1e5879c3275f80e339e8fae26。
运行库为 ONNX Runtime 1.31.0、NumPy 2.5.4、tokenizers 0.23.3，CPU 执行提供方，单线程，CLS、L2、五百一十二维、最长五百一十二 token。查询带中文检索指令，文档不带。演示最小余弦值零点四，没有进行相关性概率校准。
核心测试二十二条；可选模型测试两条。本地设置
BGE_MODEL_DIR 后二十四条全部实际通过；无模型的 CI
则运行核心二十二条并跳过两条。跳过不计为模型通过，CI
不重复下载权重。
出版验证检查书稿与在线页、EPUB 的内容映射，EPUBCheck 验证格式；浏览器检查浅色默认、深色切换、字号记忆、全书搜索、目录、无脚本阅读与存储禁用。它们能发现出版与交互缺陷，不能证明书稿事实或检索业务效果。
报告的 wall_ms
是本机单次查询观察值，允许其他机器不同；状态变化、语料指纹、授权结果与相关文档排名才是本实验的主要复现对象。公开构建的精确提交、EPUB
字节和 CI 地址以仓库 DELIVERY.md
为准，避免在正文中嵌入会自我改变构建的产物摘要。
下面的清单按证据流转顺序安排。它不是生产合规声明，而是一组可以转化为本地实验的问题。
如果只能先做一项，先做“旧命中经历状态变化后仍能否输出”的实验。它同时触及版本、授权、缓存和引用，通常比再调整一个相似度阈值更容易暴露系统结构中的问题。
完成清单之后，再决定是否需要近邻索引、独立重排模型或生成式回答。新增组件应能够解释它解决哪一种已经观察到的失败，并接受相同的生命周期检验。
以下文件在构建时从随书实验与测试目录读取，与下载区实验包同源。它们是本书教学实现；实际验证范围以实验章节和验证记录为准。
# 随书检索实验
在仓库根目录以 `python -m lab...` 运行。核心只依赖 Python 标准库与 FTS5；这是可审查的教学索引，不是部署完成的企业服务。
## 文件与契约
- `text.py`：NFC/换行投影、Markdown 与限定 JSON 解析、带码点区间的切分、中文双字词项和词项计数余弦。
- `index.py`：真实 SQLite 表/FTS5、可见片段先过滤、余弦扫描、RRF、标题规则重排、数据库重读引用，以及事务更新/删除/ACL。
- `fixtures.py` 与 `data/`：8 篇作者构造文档、13 条查询，manifest 校验原始 JSON 文件摘要。
- `demo.py`：退款更新、撤权、删除与重放，实际失败就退出。
- `evaluate.py`：每个查询的前三篇去重文档、正例 Recall/MRR、负例空候选和权限断言。
- `embedding.py` 与 `fetch_embedding.py`：可选真实量化 BGE 推理、固定提交下载与摘要校验。
- `evidence/`：本次实际运行的完整 JSON，不是未来运行的期望模板。
文档坐标是规范投影的 Python Unicode 码点 `[start,end)`，不是原始字节。Markdown 只识别简单 ATX 标题，未实现完整 CommonMark（包括围栏内标题区分）、PDF/OCR 或表格抽取。`Context` 必须由可信层建立；`status()` 属于维护接口，不能公开给普通用户。
## 核心与模型分别验证
```sh
python -m unittest discover -s tests -v
python -m lab.demo --output /tmp/knowledge-demo.json
python -m lab.evaluate --output /tmp/knowledge-term-evaluation.json
```
核心 22 条通过时，2 条可选模型测试应报告跳过。要验证真实模型：
```sh
python -m pip install -r lab/requirements-embedding.txt
python -m lab.fetch_embedding
BGE_MODEL_DIR=.tools/models/bge-small-zh-v1.5 python -m unittest discover -s tests -v
python -m lab.evaluate --onnx-model .tools/models/bge-small-zh-v1.5 --output /tmp/knowledge-bge-evaluation.json
```
下载需要 `curl`。原模型 BAAI/bge-small-zh-v1.5 卡提交 `7999e1d3359715c523056ef9478215996d62a620`；社区导出 Xenova/bge-small-zh-v1.5 提交 `75c43b069aac4d136ba6bc1122f995fedcfd2781`，不与原始全精度权重混称。文件摘要在附录与报告中，权重不进入 Git/实验包。
本机实际执行 24 条测试全部通过；模型和 tokenizer 摘要、CPU 提供方、依赖版本、CLS/L2、查询指令写入 `evidence/bge-evaluation.json`。词项/向量/混合在基础表示下均 0.9/0.9；真实模型的向量与混合均 1.0/1.0。三条负例各模式均空，当前断言无权限违规。
语料与标签都由作者构造，没有独立盲测；演示阈值未校准，耗时不是并发性能。所有回答为非 LLM 摘录，完整性验证不等于语义蕴含；删除不清备份；撤权不能召回已经交付内容；共享 FTS 统计与时间侧信道未审计。"""Executable retrieval experiments; standard-library core, optional local ONNX."""[
{
"tenant": "acme",
"doc_id": "travel",
"revision": 1,
"format": "markdown",
"acl": [
"alice",
"bob"
],
"source": "# 差旅费用标准\n\n## 住宿\n员工出差住宿每晚限额为500元，需提供酒店发票。\n\n## 餐费\n差旅餐费每日上限为80元。\n\n## 交通\n城市内交通凭有效票据报销。"
},
{
"tenant": "acme",
"doc_id": "refund",
"revision": 1,
"format": "markdown",
"acl": [
"alice",
"bob"
],
"source": "# 退款政策\n\n客户购买商品后30天内可申请退款，需保留订单编号和付款凭证。"
},
{
"tenant": "acme",
"doc_id": "remote",
"revision": 1,
"format": "json",
"acl": [
"alice",
"bob"
],
"source": "{\"title\": \"远程办公申请\", \"sections\": [{\"heading\": \"审批流程\", \"text\": \"申请居家办公须部门负责人同意，每周最多2天。\"}]}"
},
{
"tenant": "acme",
"doc_id": "board",
"revision": 1,
"format": "markdown",
"acl": [
"bob"
],
"source": "# 并购谈判计划\n\n仅董事会及法务可阅，代号猎户座，预算为900000元。SECRET-ACME-BOB"
},
{
"tenant": "beta",
"doc_id": "travel",
"revision": 1,
"format": "markdown",
"acl": [
"alice"
],
"source": "# 乙公司差旅费用标准\n\n乙公司酒店每晚限额900元。SECRET-BETA"
},
{
"tenant": "acme",
"doc_id": "support",
"revision": 1,
"format": "json",
"acl": [
"*"
],
"source": "{\"title\": \"售后服务\", \"sections\": [{\"heading\": \"保修\", \"text\": \"设备保修期间可携带购买凭证至服务中心，工作时间为周一至周五。\"}]}"
},
{
"tenant": "acme",
"doc_id": "leave",
"revision": 1,
"format": "markdown",
"acl": [
"alice",
"bob"
],
"source": "# 年假申请\n\n入职一年后员工有5天带薪年假，申请应提前3天提交。"
},
{
"tenant": "acme",
"doc_id": "health",
"revision": 1,
"format": "markdown",
"acl": [
"bob"
],
"source": "# 健康福利\n\n管理层年度体检额度为1000元。仅指定人员可见。"
}
]{
"version": "knowledge-fixture-v1",
"kind": "author-created Chinese retrieval and ACL examples, not business benchmark",
"documents": 8,
"queries": 13,
"positive_queries": 10,
"negative_queries": 3,
"sha256": {
"corpus.json": "e103bc584fd485a377ff31bbeb9fc6c08c71d9bfafaf81649c616234d608edbc",
"queries.json": "14f5e7a6f26248e30ed4483c30963551a7d10ad2e7ffacaf7edb84ada1dae6a8"
}
}[
{
"id": "q01",
"query": "住宿发票报销",
"relevant": [
"travel"
]
},
{
"id": "q02",
"query": "差旅餐费上限",
"relevant": [
"travel"
]
},
{
"id": "q03",
"query": "退款时间窗口",
"relevant": [
"refund"
]
},
{
"id": "q04",
"query": "远程办公每周几天",
"relevant": [
"remote"
]
},
{
"id": "q05",
"query": "设备保修服务时间",
"relevant": [
"support"
]
},
{
"id": "q06",
"query": "年假提前申请",
"relevant": [
"leave"
]
},
{
"id": "q07",
"query": "旅馆住一夜能花多少钱",
"relevant": [
"travel"
]
},
{
"id": "q08",
"query": "在家上班需要谁同意",
"relevant": [
"remote"
]
},
{
"id": "q09",
"query": "带薪休息有多少天",
"relevant": [
"leave"
]
},
{
"id": "q10",
"query": "商品不满意多久能退钱",
"relevant": [
"refund"
]
},
{
"id": "q11",
"query": "SECRET-ACME-BOB",
"relevant": []
},
{
"id": "q12",
"query": "SECRET-BETA",
"relevant": []
},
{
"id": "q13",
"query": "zxqv-unrelated-token",
"relevant": []
}
]"""Actual SQLite ingestion/search/update/revocation/deletion experiment."""
import argparse
import json
from pathlib import Path
import tempfile
from .fixtures import populate
from .index import Context, Index
from .text import digest
def run():
with tempfile.TemporaryDirectory(prefix="knowledge-demo-") as directory:
index = Index(Path(directory)/"index.sqlite", chunk_size=80, overlap=16)
populate(index)
context = Context("acme", "alice")
hits = index.search("住宿发票报销", context)
answer = index.answer(hits, context)
assert answer["state"] == "extractive" and any(e["doc_id"] == "travel" for e in answer["evidence"])
for evidence in answer["evidence"]:
assert digest(evidence["quote"]) == evidence["sha256"]
stale = index.search("退款政策", context)
update = index.upsert("acme", "refund", 2, "markdown", "# 退款政策\n\n客户购买商品后14天内可申请退款，需保留订单编号和付款凭证。", ["alice", "bob"])
assert index.answer(stale, context)["state"] == "abstain"
assert not index.search("30", context)
updated = index.answer(index.search("退款政策", context), context)
assert "14天" in updated["answer"]
before_revoke = index.search("远程办公", context)
acl_revision = index.status("acme", "remote")["acl_revision"]
index.set_acl("acme", "remote", ["bob"], acl_revision)
assert index.answer(before_revoke, context)["state"] == "abstain"
assert all(h["doc_id"] != "remote" for h in index.search("远程办公", context))
assert any(h["doc_id"] == "remote" for h in index.search("远程办公", Context("acme", "bob")))
before_delete = [hit for hit in index.search("年假申请", context) if hit["doc_id"] == "leave"]
index.delete("acme", "leave", 2)
assert index.answer(before_delete, context)["state"] == "abstain"
assert all(h["doc_id"] != "leave" for h in index.search("年假", context))
try:
index.upsert("acme", "leave", 1, "markdown", "# 年假\n旧内容", ["alice"])
except ValueError as error:
replay = str(error)
else:
raise AssertionError("old event resurrected deleted document")
denied = [index.search(q, context) for q in ["SECRET-ACME-BOB", "SECRET-BETA", "zxqv-unrelated-token"]]
assert all(not value for value in denied)
result = {"kind": "real_sqlite_fts5_and_term_vectors_non_llm",
"initial_answer": answer, "version_update": update,
"old_30_day_fact_retrievable": False, "new_answer": updated,
"stale_revision_citation_rejected": True,
"revoked_alice_citation_rejected": True, "bob_access_preserved": True,
"deleted_citation_rejected": True, "stale_replay_error": replay,
"cross_tenant_and_private_results": [len(v) for v in denied],
"limitations": ["extractive answers, no LLM", "term vectors are not semantic embeddings",
"deletion is a live-index visibility guarantee, not backup/physical erasure",
"revocation cannot recall information already delivered"]}
index.close()
return result
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--output", type=Path, default=Path("lab/evidence/demo.json"))
args = parser.parse_args()
result = run()
args.output.parent.mkdir(parents=True, exist_ok=True)
args.output.write_text(json.dumps(result, ensure_ascii=False, indent=2)+"\n")
print(json.dumps(result, ensure_ascii=False, indent=2))
if __name__ == "__main__":
main()"""Optional real local BGE embedding using a pinned community ONNX export.
Original model: BAAI/bge-small-zh-v1.5. Quantized export: Xenova repository.
No model files are redistributed by this book. No cloud inference is called.
"""
import hashlib
import importlib.metadata
from pathlib import Path
class BGEEncoder:
minimum_score = 0.40 # Demonstration cutoff, not a relevance probability.
def __init__(self, directory):
import numpy as np
import onnxruntime as ort
from tokenizers import Tokenizer
root = Path(directory)
model = root / "model_quantized.onnx"
tokenizer = root / "tokenizer.json"
hashes = {p.name: hashlib.sha256(p.read_bytes()).hexdigest() for p in [model, tokenizer]}
self.identifier = "bge-zh-v15-onnx-quantized-cls512-instruction-v1:" + hashes[model.name] + ":" + hashes[tokenizer.name]
self.metadata = {"original_model": "BAAI/bge-small-zh-v1.5",
"original_revision": "7999e1d3359715c523056ef9478215996d62a620",
"export": "Xenova/bge-small-zh-v1.5",
"export_revision": "75c43b069aac4d136ba6bc1122f995fedcfd2781",
"files_sha256": hashes, "pooling": "CLS", "normalization": "L2",
"max_tokens": 512, "query_instruction": "为这个句子生成表示以用于检索相关文章：",
"runtime": {name: importlib.metadata.version(name) for name in ["onnxruntime", "numpy", "tokenizers"]},
"provider": "CPUExecutionProvider", "dimension": 512}
self.np = np
self.tokenizer = Tokenizer.from_file(str(tokenizer))
self.tokenizer.enable_truncation(max_length=512)
self.tokenizer.enable_padding()
options = ort.SessionOptions()
options.intra_op_num_threads = 1
options.inter_op_num_threads = 1
self.session = ort.InferenceSession(str(model), sess_options=options, providers=["CPUExecutionProvider"])
def encode(self, texts, query=False):
if not texts:
return []
if query:
texts = [self.metadata["query_instruction"] + text for text in texts]
encoded = self.tokenizer.encode_batch(texts)
inputs = {"input_ids": self.np.array([e.ids for e in encoded], dtype=self.np.int64),
"attention_mask": self.np.array([e.attention_mask for e in encoded], dtype=self.np.int64),
"token_type_ids": self.np.array([e.type_ids for e in encoded], dtype=self.np.int64)}
required = {i.name for i in self.session.get_inputs()}
output = self.session.run(None, {k: v for k, v in inputs.items() if k in required})[0]
vectors = output[:, 0, :]
norms = self.np.linalg.norm(vectors, axis=1, keepdims=True)
if vectors.shape[1] != 512 or not self.np.isfinite(vectors).all() or (norms == 0).any():
raise ValueError("invalid_embedding_output")
return (vectors / norms).tolist()"""Per-query evaluation of a small authored fixture, not a model leaderboard."""
import argparse
import json
from pathlib import Path
import platform
import sqlite3
import tempfile
import time
from .fixtures import corpus, populate, queries
from .index import Context, Index
def evaluate(index, mode, rerank=True):
rows, recall, reciprocal, negatives, negative_abstain = [], [], [], 0, 0
for case in queries():
start = time.monotonic()
hits = index.search(case["query"], Context("acme", "alice"), limit=10, mode=mode, rerank=rerank)
doc_ids = list(dict.fromkeys(h["doc_id"] for h in hits))[:3]
rank = next((i for i, value in enumerate(doc_ids, 1) if value in case["relevant"]), None)
if case["relevant"]:
recall.append(len(set(doc_ids) & set(case["relevant"])) / len(case["relevant"]))
reciprocal.append(1 / rank if rank else 0)
else:
negatives += 1
negative_abstain += not hits
assert all(h["tenant"] == "acme" and h["doc_id"] not in {"board", "health"} for h in hits)
rows.append({**case, "top3_documents": doc_ids, "first_relevant_rank": rank,
"returned_chunks": len(hits), "wall_ms": round((time.monotonic()-start)*1000, 3)})
return {"mode": mode, "rerank": rerank, "queries": len(rows),
"positive_queries": len(recall), "recall_at_3": sum(recall)/len(recall),
"mrr_at_3": sum(reciprocal)/len(reciprocal), "negative_queries": negatives,
"negative_abstentions": negative_abstain, "access_violations": 0, "rows": rows}
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--onnx-model", type=Path)
parser.add_argument("--output", type=Path, default=Path("lab/evidence/evaluation.json"))
args = parser.parse_args()
encoder, metadata = None, {"encoder": "character/ASCII term counts, not semantic embedding"}
if args.onnx_model:
from .embedding import BGEEncoder
encoder = BGEEncoder(args.onnx_model)
metadata = encoder.metadata
with tempfile.TemporaryDirectory(prefix="knowledge-eval-") as directory:
index = Index(Path(directory) / "index.sqlite", encoder=encoder, chunk_size=80, overlap=16)
populate(index)
reports = [evaluate(index, mode) for mode in ["lexical", "vector", "hybrid"]]
index.close()
result = {"kind": "small_authored_retrieval_fixture_not_a_business_or_model_benchmark",
"python": platform.python_version(), "sqlite": sqlite3.sqlite_version,
"documents": len(corpus()), "metadata": metadata, "reports": reports,
"limitations": ["single authored corpus; no independent heldout population",
"ONNX export is community quantization, not original full-precision weights",
"embedding similarity is not a calibrated relevance probability",
"CPU timings are local execution observations, not throughput benchmarks",
"ACL result filtering is tested; timing/statistical side channels are not audited"]}
args.output.parent.mkdir(parents=True, exist_ok=True)
args.output.write_text(json.dumps(result, ensure_ascii=False, indent=2)+"\n")
print(json.dumps({"metadata": metadata, "results": [{k:v for k,v in r.items() if k != "rows"} for r in reports],
"output": str(args.output)}, ensure_ascii=False, indent=2))
if __name__ == "__main__":
main(){
"kind": "small_authored_retrieval_fixture_not_a_business_or_model_benchmark",
"python": "3.12.15",
"sqlite": "3.53.1",
"documents": 8,
"metadata": {
"original_model": "BAAI/bge-small-zh-v1.5",
"original_revision": "7999e1d3359715c523056ef9478215996d62a620",
"export": "Xenova/bge-small-zh-v1.5",
"export_revision": "75c43b069aac4d136ba6bc1122f995fedcfd2781",
"files_sha256": {
"model_quantized.onnx": "15b717c382bcb518ba457b93ea6850ede7f4f1cd8937454aa06972366cd19bcc",
"tokenizer.json": "48cea5d44424912a6fd1ea647bf4fe50b55ab8b1e5879c3275f80e339e8fae26"
},
"pooling": "CLS",
"normalization": "L2",
"max_tokens": 512,
"query_instruction": "为这个句子生成表示以用于检索相关文章：",
"runtime": {
"onnxruntime": "1.31.0",
"numpy": "2.5.4",
"tokenizers": "0.23.3"
},
"provider": "CPUExecutionProvider",
"dimension": 512
},
"reports": [
{
"mode": "lexical",
"rerank": true,
"queries": 13,
"positive_queries": 10,
"recall_at_3": 0.9,
"mrr_at_3": 0.9,
"negative_queries": 3,
"negative_abstentions": 3,
"access_violations": 0,
"rows": [
{
"id": "q01",
"query": "住宿发票报销",
"relevant": [
"travel"
],
"top3_documents": [
"travel"
],
"first_relevant_rank": 1,
"returned_chunks": 2,
"wall_ms": 0.341
},
{
"id": "q02",
"query": "差旅餐费上限",
"relevant": [
"travel"
],
"top3_documents": [
"travel"
],
"first_relevant_rank": 1,
"returned_chunks": 2,
"wall_ms": 0.116
},
{
"id": "q03",
"query": "退款时间窗口",
"relevant": [
"refund"
],
"top3_documents": [
"refund",
"support"
],
"first_relevant_rank": 1,
"returned_chunks": 2,
"wall_ms": 0.099
},
{
"id": "q04",
"query": "远程办公每周几天",
"relevant": [
"remote"
],
"top3_documents": [
"remote"
],
"first_relevant_rank": 1,
"returned_chunks": 1,
"wall_ms": 0.101
},
{
"id": "q05",
"query": "设备保修服务时间",
"relevant": [
"support"
],
"top3_documents": [
"support"
],
"first_relevant_rank": 1,
"returned_chunks": 1,
"wall_ms": 0.097
},
{
"id": "q06",
"query": "年假提前申请",
"relevant": [
"leave"
],
"top3_documents": [
"leave",
"remote",
"refund"
],
"first_relevant_rank": 1,
"returned_chunks": 3,
"wall_ms": 0.091
},
{
"id": "q07",
"query": "旅馆住一夜能花多少钱",
"relevant": [
"travel"
],
"top3_documents": [],
"first_relevant_rank": null,
"returned_chunks": 0,
"wall_ms": 0.075
},
{
"id": "q08",
"query": "在家上班需要谁同意",
"relevant": [
"remote"
],
"top3_documents": [
"remote"
],
"first_relevant_rank": 1,
"returned_chunks": 1,
"wall_ms": 0.186
},
{
"id": "q09",
"query": "带薪休息有多少天",
"relevant": [
"leave"
],
"top3_documents": [
"leave"
],
"first_relevant_rank": 1,
"returned_chunks": 1,
"wall_ms": 0.138
},
{
"id": "q10",
"query": "商品不满意多久能退钱",
"relevant": [
"refund"
],
"top3_documents": [
"refund"
],
"first_relevant_rank": 1,
"returned_chunks": 1,
"wall_ms": 0.299
},
{
"id": "q11",
"query": "SECRET-ACME-BOB",
"relevant": [],
"top3_documents": [],
"first_relevant_rank": null,
"returned_chunks": 0,
"wall_ms": 0.078
},
{
"id": "q12",
"query": "SECRET-BETA",
"relevant": [],
"top3_documents": [],
"first_relevant_rank": null,
"returned_chunks": 0,
"wall_ms": 0.059
},
{
"id": "q13",
"query": "zxqv-unrelated-token",
"relevant": [],
"top3_documents": [],
"first_relevant_rank": null,
"returned_chunks": 0,
"wall_ms": 0.055
}
]
},
{
"mode": "vector",
"rerank": true,
"queries": 13,
"positive_queries": 10,
"recall_at_3": 1.0,
"mrr_at_3": 1.0,
"negative_queries": 3,
"negative_abstentions": 3,
"access_violations": 0,
"rows": [
{
"id": "q01",
"query": "住宿发票报销",
"relevant": [
"travel"
],
"top3_documents": [
"travel",
"remote",
"refund"
],
"first_relevant_rank": 1,
"returned_chunks": 4,
"wall_ms": 3.048
},
{
"id": "q02",
"query": "差旅餐费上限",
"relevant": [
"travel"
],
"top3_documents": [
"travel"
],
"first_relevant_rank": 1,
"returned_chunks": 2,
"wall_ms": 2.773
},
{
"id": "q03",
"query": "退款时间窗口",
"relevant": [
"refund"
],
"top3_documents": [
"refund",
"travel",
"remote"
],
"first_relevant_rank": 1,
"returned_chunks": 4,
"wall_ms": 2.687
},
{
"id": "q04",
"query": "远程办公每周几天",
"relevant": [
"remote"
],
"top3_documents": [
"remote",
"leave"
],
"first_relevant_rank": 1,
"returned_chunks": 2,
"wall_ms": 2.836
},
{
"id": "q05",
"query": "设备保修服务时间",
"relevant": [
"support"
],
"top3_documents": [
"support",
"refund",
"remote"
],
"first_relevant_rank": 1,
"returned_chunks": 3,
"wall_ms": 3.041
},
{
"id": "q06",
"query": "年假提前申请",
"relevant": [
"leave"
],
"top3_documents": [
"leave",
"remote"
],
"first_relevant_rank": 1,
"returned_chunks": 2,
"wall_ms": 3.264
},
{
"id": "q07",
"query": "旅馆住一夜能花多少钱",
"relevant": [
"travel"
],
"top3_documents": [
"travel"
],
"first_relevant_rank": 1,
"returned_chunks": 2,
"wall_ms": 3.03
},
{
"id": "q08",
"query": "在家上班需要谁同意",
"relevant": [
"remote"
],
"top3_documents": [
"remote",
"leave"
],
"first_relevant_rank": 1,
"returned_chunks": 2,
"wall_ms": 2.99
},
{
"id": "q09",
"query": "带薪休息有多少天",
"relevant": [
"leave"
],
"top3_documents": [
"leave",
"remote",
"travel"
],
"first_relevant_rank": 1,
"returned_chunks": 3,
"wall_ms": 2.802
},
{
"id": "q10",
"query": "商品不满意多久能退钱",
"relevant": [
"refund"
],
"top3_documents": [
"refund",
"support",
"travel"
],
"first_relevant_rank": 1,
"returned_chunks": 6,
"wall_ms": 3.017
},
{
"id": "q11",
"query": "SECRET-ACME-BOB",
"relevant": [],
"top3_documents": [],
"first_relevant_rank": null,
"returned_chunks": 0,
"wall_ms": 3.51
},
{
"id": "q12",
"query": "SECRET-BETA",
"relevant": [],
"top3_documents": [],
"first_relevant_rank": null,
"returned_chunks": 0,
"wall_ms": 3.168
},
{
"id": "q13",
"query": "zxqv-unrelated-token",
"relevant": [],
"top3_documents": [],
"first_relevant_rank": null,
"returned_chunks": 0,
"wall_ms": 3.349
}
]
},
{
"mode": "hybrid",
"rerank": true,
"queries": 13,
"positive_queries": 10,
"recall_at_3": 1.0,
"mrr_at_3": 1.0,
"negative_queries": 3,
"negative_abstentions": 3,
"access_violations": 0,
"rows": [
{
"id": "q01",
"query": "住宿发票报销",
"relevant": [
"travel"
],
"top3_documents": [
"travel",
"remote",
"refund"
],
"first_relevant_rank": 1,
"returned_chunks": 4,
"wall_ms": 3.285
},
{
"id": "q02",
"query": "差旅餐费上限",
"relevant": [
"travel"
],
"top3_documents": [
"travel"
],
"first_relevant_rank": 1,
"returned_chunks": 2,
"wall_ms": 3.121
},
{
"id": "q03",
"query": "退款时间窗口",
"relevant": [
"refund"
],
"top3_documents": [
"refund",
"travel",
"support"
],
"first_relevant_rank": 1,
"returned_chunks": 5,
"wall_ms": 3.193
},
{
"id": "q04",
"query": "远程办公每周几天",
"relevant": [
"remote"
],
"top3_documents": [
"remote",
"leave"
],
"first_relevant_rank": 1,
"returned_chunks": 2,
"wall_ms": 2.935
},
{
"id": "q05",
"query": "设备保修服务时间",
"relevant": [
"support"
],
"top3_documents": [
"support",
"refund",
"remote"
],
"first_relevant_rank": 1,
"returned_chunks": 3,
"wall_ms": 3.15
},
{
"id": "q06",
"query": "年假提前申请",
"relevant": [
"leave"
],
"top3_documents": [
"leave",
"remote",
"refund"
],
"first_relevant_rank": 1,
"returned_chunks": 3,
"wall_ms": 2.916
},
{
"id": "q07",
"query": "旅馆住一夜能花多少钱",
"relevant": [
"travel"
],
"top3_documents": [
"travel"
],
"first_relevant_rank": 1,
"returned_chunks": 2,
"wall_ms": 2.996
},
{
"id": "q08",
"query": "在家上班需要谁同意",
"relevant": [
"remote"
],
"top3_documents": [
"remote",
"leave"
],
"first_relevant_rank": 1,
"returned_chunks": 2,
"wall_ms": 3.39
},
{
"id": "q09",
"query": "带薪休息有多少天",
"relevant": [
"leave"
],
"top3_documents": [
"leave",
"remote",
"travel"
],
"first_relevant_rank": 1,
"returned_chunks": 3,
"wall_ms": 2.938
},
{
"id": "q10",
"query": "商品不满意多久能退钱",
"relevant": [
"refund"
],
"top3_documents": [
"refund",
"support",
"travel"
],
"first_relevant_rank": 1,
"returned_chunks": 6,
"wall_ms": 3.109
},
{
"id": "q11",
"query": "SECRET-ACME-BOB",
"relevant": [],
"top3_documents": [],
"first_relevant_rank": null,
"returned_chunks": 0,
"wall_ms": 2.736
},
{
"id": "q12",
"query": "SECRET-BETA",
"relevant": [],
"top3_documents": [],
"first_relevant_rank": null,
"returned_chunks": 0,
"wall_ms": 2.377
},
{
"id": "q13",
"query": "zxqv-unrelated-token",
"relevant": [],
"top3_documents": [],
"first_relevant_rank": null,
"returned_chunks": 0,
"wall_ms": 3.223
}
]
}
],
"limitations": [
"single authored corpus; no independent heldout population",
"ONNX export is community quantization, not original full-precision weights",
"embedding similarity is not a calibrated relevance probability",
"CPU timings are local execution observations, not throughput benchmarks",
"ACL result filtering is tested; timing/statistical side channels are not audited"
]
}{
"kind": "real_sqlite_fts5_and_term_vectors_non_llm",
"initial_answer": {
"state": "extractive",
"non_llm": true,
"answer": "# 差旅费用标准\n\n## 住宿\n员工出差住宿每晚限额为500元，需提供酒店发票。\n\n## 餐费\n差旅餐费每日上限为80元。\n\n",
"evidence": [
{
"doc_id": "travel",
"revision": 1,
"start": 0,
"end": 63,
"sha256": "6c0b230598d5e8b3bf87b66874730c931bf32b7584dff3f0d1a4d6590885b808",
"quote": "# 差旅费用标准\n\n## 住宿\n员工出差住宿每晚限额为500元，需提供酒店发票。\n\n## 餐费\n差旅餐费每日上限为80元。\n\n"
}
]
},
"version_update": {
"state": "indexed",
"revision": 2,
"chunks": 1,
"sha256": "e74729dbab7284e5e7d8a9f4e5aa18882154b79cd3b07e667fdafa27f0bf0f94"
},
"old_30_day_fact_retrievable": false,
"new_answer": {
"state": "extractive",
"non_llm": true,
"answer": "# 退款政策\n\n客户购买商品后14天内可申请退款，需保留订单编号和付款凭证。",
"evidence": [
{
"doc_id": "refund",
"revision": 2,
"start": 0,
"end": 38,
"sha256": "e74729dbab7284e5e7d8a9f4e5aa18882154b79cd3b07e667fdafa27f0bf0f94",
"quote": "# 退款政策\n\n客户购买商品后14天内可申请退款，需保留订单编号和付款凭证。"
}
]
},
"stale_revision_citation_rejected": true,
"revoked_alice_citation_rejected": true,
"bob_access_preserved": true,
"deleted_citation_rejected": true,
"stale_replay_error": "stale_or_conflicting_revision",
"cross_tenant_and_private_results": [
0,
0,
0
],
"limitations": [
"extractive answers, no LLM",
"term vectors are not semantic embeddings",
"deletion is a live-index visibility guarantee, not backup/physical erasure",
"revocation cannot recall information already delivered"
]
}{
"kind": "small_authored_retrieval_fixture_not_a_business_or_model_benchmark",
"python": "3.12.15",
"sqlite": "3.53.1",
"documents": 8,
"metadata": {
"encoder": "character/ASCII term counts, not semantic embedding"
},
"reports": [
{
"mode": "lexical",
"rerank": true,
"queries": 13,
"positive_queries": 10,
"recall_at_3": 0.9,
"mrr_at_3": 0.9,
"negative_queries": 3,
"negative_abstentions": 3,
"access_violations": 0,
"rows": [
{
"id": "q01",
"query": "住宿发票报销",
"relevant": [
"travel"
],
"top3_documents": [
"travel"
],
"first_relevant_rank": 1,
"returned_chunks": 2,
"wall_ms": 0.277
},
{
"id": "q02",
"query": "差旅餐费上限",
"relevant": [
"travel"
],
"top3_documents": [
"travel"
],
"first_relevant_rank": 1,
"returned_chunks": 2,
"wall_ms": 0.103
},
{
"id": "q03",
"query": "退款时间窗口",
"relevant": [
"refund"
],
"top3_documents": [
"refund",
"support"
],
"first_relevant_rank": 1,
"returned_chunks": 2,
"wall_ms": 0.079
},
{
"id": "q04",
"query": "远程办公每周几天",
"relevant": [
"remote"
],
"top3_documents": [
"remote"
],
"first_relevant_rank": 1,
"returned_chunks": 1,
"wall_ms": 0.09
},
{
"id": "q05",
"query": "设备保修服务时间",
"relevant": [
"support"
],
"top3_documents": [
"support"
],
"first_relevant_rank": 1,
"returned_chunks": 1,
"wall_ms": 0.087
},
{
"id": "q06",
"query": "年假提前申请",
"relevant": [
"leave"
],
"top3_documents": [
"leave",
"remote",
"refund"
],
"first_relevant_rank": 1,
"returned_chunks": 3,
"wall_ms": 0.083
},
{
"id": "q07",
"query": "旅馆住一夜能花多少钱",
"relevant": [
"travel"
],
"top3_documents": [],
"first_relevant_rank": null,
"returned_chunks": 0,
"wall_ms": 0.068
},
{
"id": "q08",
"query": "在家上班需要谁同意",
"relevant": [
"remote"
],
"top3_documents": [
"remote"
],
"first_relevant_rank": 1,
"returned_chunks": 1,
"wall_ms": 0.098
},
{
"id": "q09",
"query": "带薪休息有多少天",
"relevant": [
"leave"
],
"top3_documents": [
"leave"
],
"first_relevant_rank": 1,
"returned_chunks": 1,
"wall_ms": 0.085
},
{
"id": "q10",
"query": "商品不满意多久能退钱",
"relevant": [
"refund"
],
"top3_documents": [
"refund"
],
"first_relevant_rank": 1,
"returned_chunks": 1,
"wall_ms": 0.101
},
{
"id": "q11",
"query": "SECRET-ACME-BOB",
"relevant": [],
"top3_documents": [],
"first_relevant_rank": null,
"returned_chunks": 0,
"wall_ms": 0.048
},
{
"id": "q12",
"query": "SECRET-BETA",
"relevant": [],
"top3_documents": [],
"first_relevant_rank": null,
"returned_chunks": 0,
"wall_ms": 0.04
},
{
"id": "q13",
"query": "zxqv-unrelated-token",
"relevant": [],
"top3_documents": [],
"first_relevant_rank": null,
"returned_chunks": 0,
"wall_ms": 0.042
}
]
},
{
"mode": "vector",
"rerank": true,
"queries": 13,
"positive_queries": 10,
"recall_at_3": 0.9,
"mrr_at_3": 0.9,
"negative_queries": 3,
"negative_abstentions": 3,
"access_violations": 0,
"rows": [
{
"id": "q01",
"query": "住宿发票报销",
"relevant": [
"travel"
],
"top3_documents": [
"travel"
],
"first_relevant_rank": 1,
"returned_chunks": 2,
"wall_ms": 0.092
},
{
"id": "q02",
"query": "差旅餐费上限",
"relevant": [
"travel"
],
"top3_documents": [
"travel"
],
"first_relevant_rank": 1,
"returned_chunks": 2,
"wall_ms": 0.066
},
{
"id": "q03",
"query": "退款时间窗口",
"relevant": [
"refund"
],
"top3_documents": [
"refund",
"support"
],
"first_relevant_rank": 1,
"returned_chunks": 2,
"wall_ms": 0.063
},
{
"id": "q04",
"query": "远程办公每周几天",
"relevant": [
"remote"
],
"top3_documents": [
"remote"
],
"first_relevant_rank": 1,
"returned_chunks": 1,
"wall_ms": 0.061
},
{
"id": "q05",
"query": "设备保修服务时间",
"relevant": [
"support"
],
"top3_documents": [
"support"
],
"first_relevant_rank": 1,
"returned_chunks": 1,
"wall_ms": 0.059
},
{
"id": "q06",
"query": "年假提前申请",
"relevant": [
"leave"
],
"top3_documents": [
"leave",
"remote",
"refund"
],
"first_relevant_rank": 1,
"returned_chunks": 3,
"wall_ms": 0.062
},
{
"id": "q07",
"query": "旅馆住一夜能花多少钱",
"relevant": [
"travel"
],
"top3_documents": [],
"first_relevant_rank": null,
"returned_chunks": 0,
"wall_ms": 0.056
},
{
"id": "q08",
"query": "在家上班需要谁同意",
"relevant": [
"remote"
],
"top3_documents": [
"remote"
],
"first_relevant_rank": 1,
"returned_chunks": 1,
"wall_ms": 0.059
},
{
"id": "q09",
"query": "带薪休息有多少天",
"relevant": [
"leave"
],
"top3_documents": [
"leave"
],
"first_relevant_rank": 1,
"returned_chunks": 1,
"wall_ms": 0.057
},
{
"id": "q10",
"query": "商品不满意多久能退钱",
"relevant": [
"refund"
],
"top3_documents": [
"refund"
],
"first_relevant_rank": 1,
"returned_chunks": 1,
"wall_ms": 0.058
},
{
"id": "q11",
"query": "SECRET-ACME-BOB",
"relevant": [],
"top3_documents": [],
"first_relevant_rank": null,
"returned_chunks": 0,
"wall_ms": 0.053
},
{
"id": "q12",
"query": "SECRET-BETA",
"relevant": [],
"top3_documents": [],
"first_relevant_rank": null,
"returned_chunks": 0,
"wall_ms": 0.052
},
{
"id": "q13",
"query": "zxqv-unrelated-token",
"relevant": [],
"top3_documents": [],
"first_relevant_rank": null,
"returned_chunks": 0,
"wall_ms": 0.052
}
]
},
{
"mode": "hybrid",
"rerank": true,
"queries": 13,
"positive_queries": 10,
"recall_at_3": 0.9,
"mrr_at_3": 0.9,
"negative_queries": 3,
"negative_abstentions": 3,
"access_violations": 0,
"rows": [
{
"id": "q01",
"query": "住宿发票报销",
"relevant": [
"travel"
],
"top3_documents": [
"travel"
],
"first_relevant_rank": 1,
"returned_chunks": 2,
"wall_ms": 0.113
},
{
"id": "q02",
"query": "差旅餐费上限",
"relevant": [
"travel"
],
"top3_documents": [
"travel"
],
"first_relevant_rank": 1,
"returned_chunks": 2,
"wall_ms": 0.108
},
{
"id": "q03",
"query": "退款时间窗口",
"relevant": [
"refund"
],
"top3_documents": [
"refund",
"support"
],
"first_relevant_rank": 1,
"returned_chunks": 2,
"wall_ms": 0.101
},
{
"id": "q04",
"query": "远程办公每周几天",
"relevant": [
"remote"
],
"top3_documents": [
"remote"
],
"first_relevant_rank": 1,
"returned_chunks": 1,
"wall_ms": 0.117
},
{
"id": "q05",
"query": "设备保修服务时间",
"relevant": [
"support"
],
"top3_documents": [
"support"
],
"first_relevant_rank": 1,
"returned_chunks": 1,
"wall_ms": 0.112
},
{
"id": "q06",
"query": "年假提前申请",
"relevant": [
"leave"
],
"top3_documents": [
"leave",
"remote",
"refund"
],
"first_relevant_rank": 1,
"returned_chunks": 3,
"wall_ms": 0.105
},
{
"id": "q07",
"query": "旅馆住一夜能花多少钱",
"relevant": [
"travel"
],
"top3_documents": [],
"first_relevant_rank": null,
"returned_chunks": 0,
"wall_ms": 0.092
},
{
"id": "q08",
"query": "在家上班需要谁同意",
"relevant": [
"remote"
],
"top3_documents": [
"remote"
],
"first_relevant_rank": 1,
"returned_chunks": 1,
"wall_ms": 0.118
},
{
"id": "q09",
"query": "带薪休息有多少天",
"relevant": [
"leave"
],
"top3_documents": [
"leave"
],
"first_relevant_rank": 1,
"returned_chunks": 1,
"wall_ms": 0.109
},
{
"id": "q10",
"query": "商品不满意多久能退钱",
"relevant": [
"refund"
],
"top3_documents": [
"refund"
],
"first_relevant_rank": 1,
"returned_chunks": 1,
"wall_ms": 0.124
},
{
"id": "q11",
"query": "SECRET-ACME-BOB",
"relevant": [],
"top3_documents": [],
"first_relevant_rank": null,
"returned_chunks": 0,
"wall_ms": 0.068
},
{
"id": "q12",
"query": "SECRET-BETA",
"relevant": [],
"top3_documents": [],
"first_relevant_rank": null,
"returned_chunks": 0,
"wall_ms": 0.063
},
{
"id": "q13",
"query": "zxqv-unrelated-token",
"relevant": [],
"top3_documents": [],
"first_relevant_rank": null,
"returned_chunks": 0,
"wall_ms": 0.065
}
]
}
],
"limitations": [
"single authored corpus; no independent heldout population",
"ONNX export is community quantization, not original full-precision weights",
"embedding similarity is not a calibrated relevance probability",
"CPU timings are local execution observations, not throughput benchmarks",
"ACL result filtering is tested; timing/statistical side channels are not audited"
]
}"""Explicit optional model download, pinned commit and expected file SHA-256."""
import argparse
import hashlib
from pathlib import Path
import subprocess
REVISION = "75c43b069aac4d136ba6bc1122f995fedcfd2781"
FILES = {
"model_quantized.onnx": ("onnx/model_quantized.onnx", "15b717c382bcb518ba457b93ea6850ede7f4f1cd8937454aa06972366cd19bcc"),
"tokenizer.json": ("tokenizer.json", "48cea5d44424912a6fd1ea647bf4fe50b55ab8b1e5879c3275f80e339e8fae26"),
}
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--directory", type=Path, default=Path(".tools/models/bge-small-zh-v1.5"))
args = parser.parse_args()
args.directory.mkdir(parents=True, exist_ok=True)
for name, (remote, expected) in FILES.items():
destination = args.directory / name
if not destination.is_file() or hashlib.sha256(destination.read_bytes()).hexdigest() != expected:
temporary = destination.with_suffix(destination.suffix + ".part")
url = f"https://huggingface.co/Xenova/bge-small-zh-v1.5/resolve/{REVISION}/{remote}"
subprocess.run(["curl", "--fail", "--location", "--silent", "--show-error", "--max-time", "180",
"--max-filesize", "200000000", url, "--output", str(temporary)], check=True)
if hashlib.sha256(temporary.read_bytes()).hexdigest() != expected:
raise ValueError("model_download_hash_mismatch:" + name)
temporary.replace(destination)
print(name, destination.stat().st_size, expected)
if __name__ == "__main__":
main()"""Versioned author-created documents and labels, not a business benchmark."""
import hashlib
import json
from pathlib import Path
ROOT = Path(__file__).parent / "data"
def read(name):
manifest = json.loads((ROOT / "manifest.json").read_text())
data = (ROOT / name).read_bytes()
if hashlib.sha256(data).hexdigest() != manifest["sha256"][name]:
raise ValueError("fixture_hash_mismatch:" + name)
return json.loads(data)
def corpus():
return read("corpus.json")
def queries():
return read("queries.json")
def populate(index):
return [index.upsert(**document) for document in corpus()]"""SQLite FTS5, local vectors, ACL prefilter and citation revalidation."""
from contextlib import contextmanager
from dataclasses import dataclass
import json
from pathlib import Path
import re
import sqlite3
import threading
from .text import TermEncoder, chunks, cosine, digest, parse_source, terms
@dataclass(frozen=True)
class Context:
tenant: str
principal: str
class Index:
def __init__(self, path, encoder=None, chunk_size=220, overlap=40):
list(chunks("", chunk_size, overlap))
self.encoder = encoder or TermEncoder()
self.chunk_size, self.overlap = chunk_size, overlap
self.lock = threading.RLock()
self.db = sqlite3.connect(str(path), check_same_thread=False)
self.db.row_factory = sqlite3.Row
self.db.execute("PRAGMA foreign_keys=ON")
self.db.executescript("""
CREATE TABLE IF NOT EXISTS meta(key TEXT PRIMARY KEY,value TEXT NOT NULL);
CREATE TABLE IF NOT EXISTS docs(
pk INTEGER PRIMARY KEY,tenant TEXT NOT NULL,doc_id TEXT NOT NULL,
revision INTEGER NOT NULL,acl_revision INTEGER NOT NULL,
title TEXT NOT NULL,body TEXT NOT NULL,sha256 TEXT NOT NULL,
deleted INTEGER NOT NULL DEFAULT 0,UNIQUE(tenant,doc_id));
CREATE TABLE IF NOT EXISTS acl(doc_pk INTEGER REFERENCES docs(pk),principal TEXT,
PRIMARY KEY(doc_pk,principal));
CREATE TABLE IF NOT EXISTS chunks(
pk INTEGER PRIMARY KEY,doc_pk INTEGER REFERENCES docs(pk),number INTEGER,
start INTEGER,end INTEGER,text TEXT,heading TEXT,sha256 TEXT,vector TEXT,
UNIQUE(doc_pk,number));
CREATE VIRTUAL TABLE IF NOT EXISTS chunks_fts USING fts5(terms);
""")
existing = self.db.execute("SELECT value FROM meta WHERE key='encoder'").fetchone()
contract = json.dumps({"encoder": self.encoder.identifier, "chunk_size": chunk_size,
"overlap": overlap}, sort_keys=True)
if existing and existing[0] != contract:
self.db.close()
raise ValueError("index_projection_contract_changed_rebuild_required")
self.db.execute("INSERT OR IGNORE INTO meta VALUES('encoder',?)", (contract,))
self.db.commit()
@contextmanager
def transaction(self, write=False):
with self.lock:
self.db.execute("BEGIN IMMEDIATE" if write else "BEGIN")
try:
yield
self.db.commit()
except BaseException:
self.db.rollback()
raise
@staticmethod
def _identity(value):
if not isinstance(value, str) or not re.fullmatch(r"[A-Za-z0-9_-]{1,80}", value):
raise ValueError("invalid_identity")
@staticmethod
def _principals(values):
if not isinstance(values, list) or any(not isinstance(v, str) or not v for v in values):
raise ValueError("explicit_acl_list_required")
return sorted(set(values))
def _doc(self, tenant, doc_id):
return self.db.execute("SELECT * FROM docs WHERE tenant=? AND doc_id=?", (tenant, doc_id)).fetchone()
def _remove_chunks(self, doc_pk):
ids = [r[0] for r in self.db.execute("SELECT pk FROM chunks WHERE doc_pk=?", (doc_pk,))]
self.db.executemany("DELETE FROM chunks_fts WHERE rowid=?", [(pk,) for pk in ids])
self.db.execute("DELETE FROM chunks WHERE doc_pk=?", (doc_pk,))
def upsert(self, tenant, doc_id, revision, format, source, acl, restore=False):
self._identity(tenant); self._identity(doc_id)
principals = self._principals(acl)
if type(restore) is not bool:
raise ValueError("explicit_restore_boolean_required")
if type(revision) is not int or revision < 1:
raise ValueError("positive_revision_required")
title, body = parse_source(format, source)
digest_value = digest(body)
pieces = list(chunks(body, self.chunk_size, self.overlap))
vectors = self.encoder.encode([p["text"] for p in pieces])
if len(vectors) != len(pieces):
raise ValueError("encoder_result_count_mismatch")
with self.transaction(write=True):
old = self._doc(tenant, doc_id)
if old:
if revision == old["revision"] and digest_value == old["sha256"] and not old["deleted"]:
return {"state": "unchanged", "revision": revision}
if revision <= old["revision"]:
raise ValueError("stale_or_conflicting_revision")
if old["deleted"] and not restore:
raise ValueError("deleted_document_requires_explicit_restore")
doc_pk = old["pk"]
self._remove_chunks(doc_pk)
self.db.execute("UPDATE docs SET revision=?,title=?,body=?,sha256=?,deleted=0 WHERE pk=?",
(revision, title, body, digest_value, doc_pk))
# Content events never regrant a revoked ACL. Only explicit
# restoration or the separate CAS policy API can change it.
if old["deleted"]:
self.db.executemany("INSERT INTO acl VALUES(?,?)", [(doc_pk, p) for p in principals])
self.db.execute("UPDATE docs SET acl_revision=acl_revision+1 WHERE pk=?", (doc_pk,))
else:
cursor = self.db.execute("INSERT INTO docs(tenant,doc_id,revision,acl_revision,title,body,sha256) VALUES(?,?,?,1,?,?,?)",
(tenant, doc_id, revision, title, body, digest_value))
doc_pk = cursor.lastrowid
self.db.executemany("INSERT INTO acl VALUES(?,?)", [(doc_pk, p) for p in principals])
for piece, vector in zip(pieces, vectors):
encoded = json.dumps(vector, ensure_ascii=False, allow_nan=False)
cursor = self.db.execute("INSERT INTO chunks(doc_pk,number,start,end,text,heading,sha256,vector) VALUES(?,?,?,?,?,?,?,?)",
(doc_pk, piece["index"], piece["start"], piece["end"], piece["text"], piece["heading"], piece["sha256"], encoded))
self.db.execute("INSERT INTO chunks_fts(rowid,terms) VALUES(?,?)", (cursor.lastrowid, " ".join(terms(piece["text"]))))
return {"state": "indexed", "revision": revision, "chunks": len(pieces), "sha256": digest_value}
def set_acl(self, tenant, doc_id, principals, expected_acl_revision):
principals = self._principals(principals)
if type(expected_acl_revision) is not int or expected_acl_revision < 1:
raise ValueError("positive_acl_revision_required")
with self.transaction(write=True):
old = self._doc(tenant, doc_id)
if not old or old["deleted"]:
raise ValueError("document_not_live")
if expected_acl_revision != old["acl_revision"]:
raise ValueError("stale_acl_revision")
self.db.execute("DELETE FROM acl WHERE doc_pk=?", (old["pk"],))
self.db.executemany("INSERT INTO acl VALUES(?,?)", [(old["pk"], p) for p in principals])
self.db.execute("UPDATE docs SET acl_revision=acl_revision+1 WHERE pk=?", (old["pk"],))
return old["acl_revision"] + 1
def delete(self, tenant, doc_id, revision):
with self.transaction(write=True):
old = self._doc(tenant, doc_id)
if not old or type(revision) is not int or revision <= old["revision"]:
raise ValueError("delete_requires_new_revision")
self._remove_chunks(old["pk"])
self.db.execute("DELETE FROM acl WHERE doc_pk=?", (old["pk"],))
self.db.execute("UPDATE docs SET revision=?,acl_revision=acl_revision+1,body='',deleted=1 WHERE pk=?", (revision, old["pk"]))
@staticmethod
def _context(context):
if not isinstance(context, Context):
raise PermissionError("trusted_context_required")
Index._identity(context.tenant); Index._identity(context.principal)
def search(self, query, context, limit=5, mode="hybrid", rerank=True):
self._context(context)
if not isinstance(query, str) or not 0 < len(query) <= 1024:
raise ValueError("bounded_query_required")
if mode not in {"lexical", "vector", "hybrid"} or not 1 <= limit <= 100:
raise ValueError("invalid_search_options")
qterms = list(dict.fromkeys(terms(query)))[:64]
if not qterms:
return []
qvector = self.encoder.encode([query], query=True)[0] if mode != "lexical" else None
with self.transaction():
allowed_sql = """FROM chunks c JOIN docs d ON c.doc_pk=d.pk
WHERE d.tenant=? AND d.deleted=0 AND EXISTS(
SELECT 1 FROM acl a WHERE a.doc_pk=d.pk AND a.principal IN(?,'*'))"""
params = (context.tenant, context.principal)
columns = "c.*,d.tenant,d.doc_id,d.revision,d.acl_revision,d.title,d.sha256 AS doc_sha256"
allowed = {r["pk"]: dict(r) for r in self.db.execute("SELECT " + columns + " " + allowed_sql, params)}
lexical, vector = [], []
if mode != "vector":
match = " OR ".join('"' + term.replace('"', '""') + '"' for term in qterms)
lexical_sql = """SELECT c.pk,bm25(chunks_fts) AS rank FROM chunks_fts
JOIN chunks c ON c.pk=chunks_fts.rowid JOIN docs d ON c.doc_pk=d.pk
WHERE chunks_fts MATCH ? AND d.tenant=? AND d.deleted=0
AND EXISTS(SELECT 1 FROM acl a WHERE a.doc_pk=d.pk AND a.principal IN(?,'*'))
ORDER BY rank,c.pk LIMIT 100"""
lexical = [(r["pk"], r["rank"]) for r in self.db.execute(lexical_sql, (match, *params))]
if mode != "lexical":
for pk, row in allowed.items():
similarity = cosine(qvector, json.loads(row["vector"]))
if similarity >= self.encoder.minimum_score:
vector.append((pk, similarity))
vector.sort(key=lambda x: (-x[1], x[0]))
vector = vector[:100]
fused = {}
for channel, ranked in [("lexical", lexical), ("vector", vector)]:
for rank, (pk, raw_score) in enumerate(ranked, 1):
value = fused.setdefault(pk, {"rrf": 0.0, "channels": {}})
value["rrf"] += 1 / (60 + rank)
value["channels"][channel] = {"rank": rank, "score": raw_score}
hits = []
for pk, extra in fused.items():
row = allowed[pk]
title_overlap = len(set(qterms) & set(terms(row["title"]))) / len(qterms)
hits.append({"chunk_pk": pk, "tenant": row["tenant"], "doc_id": row["doc_id"],
"revision": row["revision"], "acl_revision": row["acl_revision"],
"doc_sha256": row["doc_sha256"], "start": row["start"], "end": row["end"],
"text": row["text"], "sha256": row["sha256"], "heading": row["heading"],
"title": row["title"], "rrf": extra["rrf"], "channels": extra["channels"],
"title_overlap": title_overlap, "principal": context.principal})
hits.sort(key=lambda h: (-(h["title_overlap"] if rerank else 0), -h["rrf"], h["chunk_pk"]))
return hits[:limit]
def answer(self, hits, context, limit=2):
self._context(context)
if type(limit) is not int or not 1 <= limit <= 10:
raise ValueError("invalid_answer_limit")
evidence = []
with self.transaction():
for hit in hits:
if hit["tenant"] != context.tenant or hit["principal"] != context.principal:
continue
doc = self._doc(context.tenant, hit["doc_id"])
permitted = doc and self.db.execute("SELECT 1 FROM acl WHERE doc_pk=? AND principal IN(?,'*')", (doc["pk"], context.principal)).fetchone()
if (not permitted or doc["deleted"] or doc["revision"] != hit["revision"] or
doc["acl_revision"] != hit["acl_revision"] or doc["sha256"] != hit["doc_sha256"]):
continue
piece = self.db.execute("SELECT * FROM chunks WHERE pk=? AND doc_pk=?", (hit["chunk_pk"], doc["pk"])).fetchone()
if not piece or piece["sha256"] != hit["sha256"] or piece["text"] != doc["body"][piece["start"]:piece["end"]]:
continue
if any(e["doc_id"] == doc["doc_id"] and e["revision"] == doc["revision"] and
max(e["start"], piece["start"]) < min(e["end"], piece["end"]) for e in evidence):
continue
evidence.append({"doc_id": doc["doc_id"], "revision": doc["revision"],
"start": piece["start"], "end": piece["end"], "sha256": piece["sha256"],
"quote": piece["text"]})
if len(evidence) >= limit:
break
return {"state": "extractive" if evidence else "abstain", "non_llm": True,
"answer": "\n\n".join(e["quote"] for e in evidence) if evidence else "没有当前权限下可核查的证据。",
"evidence": evidence}
def status(self, tenant, doc_id):
with self.lock:
row = self._doc(tenant, doc_id)
return dict(row) if row else None
def close(self):
self.db.close()numpy==2.5.4
onnxruntime==1.31.0
tokenizers==0.23.3"""Explicit text projection, source coordinates and lexical representations."""
from collections import Counter
import hashlib
import json
import math
import re
import unicodedata
def digest(text):
return hashlib.sha256(text.encode("utf-8")).hexdigest()
def normalize(text):
if not isinstance(text, str):
raise ValueError("text_required")
return unicodedata.normalize("NFC", text.replace("\r\n", "\n").replace("\r", "\n"))
def parse_source(format, source):
if format == "markdown":
body = normalize(source)
match = re.search(r"^#\s+(.+)$", body, re.M)
return (match[1].strip() if match else "Untitled"), body
if format == "json":
value = json.loads(source)
if not isinstance(value, dict) or set(value) != {"title", "sections"}:
raise ValueError("unsupported_json_shape")
if not isinstance(value["title"], str) or not isinstance(value["sections"], list):
raise ValueError("unsupported_json_shape")
parts = ["# " + value["title"]]
for section in value["sections"]:
if not isinstance(section, dict) or set(section) != {"heading", "text"}:
raise ValueError("unsupported_json_section")
if not all(isinstance(section[k], str) for k in ["heading", "text"]):
raise ValueError("unsupported_json_section")
parts.append("## " + section["heading"] + "\n" + section["text"])
return value["title"], normalize("\n\n".join(parts))
raise ValueError("unsupported_format")
def chunks(body, size=220, overlap=40):
if not 1 <= size <= 4096 or not 0 <= overlap < size:
raise ValueError("invalid_chunk_window")
headings = [(m.start(), m[2].strip()) for m in re.finditer(r"^(#{1,6})\s+(.+)$", body, re.M)]
start, index = 0, 0
while start < len(body):
end = min(start + size, len(body))
if end < len(body):
boundary = body.rfind("\n\n", start + size // 2, end)
if boundary > start:
end = boundary + 2
quote = body[start:end]
heading = next((title for pos, title in reversed(headings) if pos <= start), "")
yield {"index": index, "start": start, "end": end, "text": quote,
"heading": heading, "sha256": digest(quote)}
if end == len(body):
break
start = max(start + 1, end - overlap)
index += 1
def terms(text):
result = []
for span in re.findall(r"[A-Za-z0-9_]+|[\u3400-\u9fff]+", normalize(text).lower()):
if re.fullmatch(r"[\u3400-\u9fff]+", span):
result.extend(span[i:i+2] for i in range(len(span)-1))
if len(span) == 1:
result.append(span)
else:
result.append(span)
return result
class TermEncoder:
identifier = "chinese-bigrams-ascii-terms-v1"
minimum_score = 0.05
def encode(self, texts, query=False):
return [dict(Counter(terms(text))) for text in texts]
def cosine(a, b):
if isinstance(a, dict) and isinstance(b, dict):
numerator = sum(value * b.get(key, 0) for key, value in a.items())
denominator = math.sqrt(sum(v*v for v in a.values()) * sum(v*v for v in b.values()))
elif isinstance(a, list) and isinstance(b, list) and len(a) == len(b):
numerator = sum(x*y for x, y in zip(a, b))
denominator = math.sqrt(sum(v*v for v in a) * sum(v*v for v in b))
else:
raise ValueError("incompatible_vector_contract")
return numerator / denominator if denominator else 0.0import math
import os
from pathlib import Path
import tempfile
import unittest
from lab.fixtures import populate
from lab.index import Context, Index
@unittest.skipUnless(os.environ.get("BGE_MODEL_DIR"), "optional pinned model not configured")
class RealLocalEmbeddingTests(unittest.TestCase):
@classmethod
def setUpClass(cls):
from lab.embedding import BGEEncoder
cls.encoder = BGEEncoder(os.environ["BGE_MODEL_DIR"])
def test_actual_onnx_vectors_have_expected_dimension_and_l2_norm(self):
vectors = self.encoder.encode(["旅馆住宿发票", "员工在家办公"])
self.assertEqual(len(vectors), 2)
for vector in vectors:
self.assertEqual(len(vector), 512)
self.assertTrue(all(math.isfinite(v) for v in vector))
self.assertAlmostEqual(sum(v*v for v in vector), 1, places=5)
def test_real_embedding_search_still_obeys_acl_and_tenant_filter(self):
with tempfile.TemporaryDirectory() as directory:
index = Index(Path(directory)/"bge.sqlite",encoder=self.encoder,chunk_size=80,overlap=16)
try:
populate(index)
hits=index.search("旅馆住一夜能花多少钱",Context("acme","alice"),mode="vector")
self.assertEqual(hits[0]["doc_id"],"travel")
private=index.search("并购董事会秘密预算",Context("acme","alice"),mode="vector",limit=100)
self.assertTrue(all(h["tenant"]=="acme" and h["doc_id"] not in {"board","health"} for h in private))
finally:index.close()
if __name__=="__main__":unittest.main()from pathlib import Path
import copy
import json
import tempfile
import unittest
from lab.fixtures import corpus, populate
from lab.index import Context, Index
from lab.text import TermEncoder, chunks, cosine, digest, normalize, parse_source, terms
class RetrievalTests(unittest.TestCase):
def setUp(self):
self.directory = tempfile.TemporaryDirectory()
self.addCleanup(self.directory.cleanup)
self.path = Path(self.directory.name)/"index.sqlite"
self.index = Index(self.path, chunk_size=80, overlap=16)
self.addCleanup(self.index.close)
populate(self.index)
self.alice = Context("acme", "alice")
self.bob = Context("acme", "bob")
def hits(self, doc_id, query, context=None):
return [h for h in self.index.search(query, context or self.alice, limit=100) if h["doc_id"] == doc_id]
def test_actual_fts5_table_returns_ranked_rows(self):
rows = self.index.db.execute("SELECT rowid,bm25(chunks_fts) FROM chunks_fts WHERE chunks_fts MATCH ?", ('"住宿"',)).fetchall()
self.assertGreater(len(rows), 0)
result = self.index.search("住宿发票", self.alice, mode="lexical")
self.assertEqual(result[0]["doc_id"], "travel")
self.assertIn("lexical", result[0]["channels"])
def test_markdown_and_json_projection_and_exact_coordinates(self):
title, body = parse_source("json", json.dumps({"title":"标题","sections":[{"heading":"小节","text":"第一行\r\n第二行"}]}))
self.assertEqual(title, "标题")
self.assertNotIn("\r", body)
for piece in chunks(body, 12, 3):
self.assertEqual(body[piece["start"]:piece["end"]], piece["text"])
self.assertEqual(digest(piece["text"]), piece["sha256"])
def test_chunk_overlap_is_bounded_and_covers_oversized_paragraph(self):
body = "一二三四五六七八九十"*15
pieces = list(chunks(body, 30, 5))
self.assertEqual(pieces[0]["start"], 0)
self.assertEqual(pieces[-1]["end"], len(body))
self.assertTrue(all(len(p["text"]) <= 30 for p in pieces))
self.assertTrue(all(a["end"] >= b["start"] for a,b in zip(pieces,pieces[1:])))
def test_chinese_terms_are_explicit_not_semantic_embedding(self):
self.assertIn("住宿", terms("住宿500元"))
self.assertIn("500", terms("住宿500元"))
a,b=TermEncoder().encode(["酒店住宿","旅馆睡觉"])
self.assertEqual(cosine(a,b), 0)
def test_hybrid_exposes_channels_and_rrf_arithmetic(self):
hits=self.index.search("住宿发票",self.alice)
top=hits[0]
self.assertEqual(top["doc_id"],"travel")
self.assertAlmostEqual(top["rrf"],sum(1/(60+v["rank"]) for v in top["channels"].values()))
def test_vector_search_prefilters_private_documents(self):
for mode in ["lexical","vector","hybrid"]:
with self.subTest(mode=mode):
hits=self.index.search("并购谈判猎户座健康福利",self.alice,mode=mode,limit=100)
self.assertTrue(all(h["doc_id"] not in {"board","health"} for h in hits))
self.assertTrue(all("SECRET" not in h["text"] for h in hits))
self.assertGreater(len(self.hits("board","并购谈判",self.bob)),0)
def test_same_document_id_in_other_tenant_cannot_cross_boundary(self):
hits=self.index.search("酒店限额",self.alice,limit=100)
self.assertTrue(all(h["tenant"]=="acme" for h in hits))
beta=self.index.search("酒店限额",Context("beta","alice"))
self.assertTrue(all(h["tenant"]=="beta" for h in beta))
self.assertTrue(any("900元" in h["text"] for h in beta))
def test_missing_context_fails_closed_and_unknown_tenant_sees_nothing(self):
with self.assertRaises(PermissionError):self.index.search("退款",None)
self.assertEqual(self.index.search("退款",Context("unknown","alice")),[])
def test_match_operators_and_quotes_are_not_executable_query_syntax(self):
self.index.search('" OR 住宿 NEAR(secret) -退款 *',self.alice)
self.assertEqual(self.index.db.execute("SELECT count(*) FROM docs").fetchone()[0],8)
def test_idempotent_ingestion_and_conflicting_revision(self):
document=corpus()[0]
count=self.index.db.execute("SELECT count(*) FROM chunks").fetchone()[0]
self.assertEqual(self.index.upsert(**document)["state"],"unchanged")
self.assertEqual(self.index.db.execute("SELECT count(*) FROM chunks").fetchone()[0],count)
with self.assertRaises(ValueError):self.index.upsert(**{**document,"source":"# changed"})
def test_incremental_update_removes_old_fact_and_invalidates_citation(self):
old=self.hits("refund","退款政策")
self.index.upsert("acme","refund",2,"markdown","# 退款政策\n\n购买后14天内申请退款。",["alice"])
self.assertEqual(self.index.answer(old,self.alice)["state"],"abstain")
self.assertEqual(self.index.search("30",self.alice),[])
fresh=self.index.answer(self.hits("refund","退款政策"),self.alice)
self.assertIn("14天",fresh["answer"])
def test_policy_revocation_rechecks_cached_hits_and_keeps_other_user(self):
old=self.hits("remote","远程办公")
version=self.index.status("acme","remote")["acl_revision"]
self.index.set_acl("acme","remote",["bob"],version)
self.assertEqual(self.index.answer(old,self.alice)["state"],"abstain")
self.assertEqual(self.hits("remote","远程办公"),[])
self.assertGreater(len(self.hits("remote","远程办公",self.bob)),0)
def test_new_content_event_cannot_regrant_old_acl(self):
version=self.index.status("acme","remote")["acl_revision"]
self.index.set_acl("acme","remote",["bob"],version)
doc=next(d for d in corpus() if d["doc_id"]=="remote")
self.index.upsert(**{**doc,"revision":2})
self.assertEqual(self.hits("remote","远程办公"),[])
def test_stale_acl_compare_and_swap_is_rejected(self):
self.index.set_acl("acme","refund",["bob"],1)
with self.assertRaises(ValueError):self.index.set_acl("acme","refund",["alice"],1)
self.assertEqual(self.hits("refund","退款政策"),[])
def test_empty_acl_denies_and_wildcard_is_tenant_scoped(self):
self.index.upsert("acme","empty",1,"markdown","# 独占口令\nSENSITIVE-EMPTY",[])
self.assertEqual(self.hits("empty","SENSITIVE-EMPTY"),[])
self.assertGreater(len(self.hits("support","售后保修",Context("acme","charlie"))),0)
self.assertEqual(self.index.search("售后保修",Context("beta","charlie")),[])
def test_delete_clears_live_index_and_rejects_old_or_implicit_restore(self):
old=self.hits("leave","年假申请")
self.index.delete("acme","leave",2)
self.assertEqual(self.index.answer(old,self.alice)["state"],"abstain")
self.assertEqual(self.hits("leave","年假申请"),[])
doc=next(d for d in corpus() if d["doc_id"]=="leave")
with self.assertRaises(ValueError):self.index.upsert(**doc)
with self.assertRaises(ValueError):self.index.upsert(**{**doc,"revision":3})
self.assertEqual(self.index.status("acme","leave")["body"],"")
self.index.upsert(**{**doc,"revision":3},restore=True)
self.assertGreater(len(self.hits("leave","年假申请")),0)
def test_citation_quote_is_read_from_current_database_not_caller_payload(self):
hits=self.hits("refund","退款政策")
hits[0]["text"]="FORGED ANSWER"
result=self.index.answer(hits,self.alice)
self.assertNotIn("FORGED",result["answer"])
evidence=result["evidence"][0]
self.assertEqual(digest(evidence["quote"]),evidence["sha256"])
def test_cross_principal_hit_reuse_is_rejected(self):
bob=self.hits("board","并购谈判",self.bob)
self.assertEqual(self.index.answer(bob,self.alice)["state"],"abstain")
def test_parse_failure_rolls_back_without_changing_existing_projection(self):
old=self.index.status("acme","refund")
with self.assertRaises((ValueError,json.JSONDecodeError)):
self.index.upsert("acme","refund",2,"json",'{bad json}',["alice"])
self.assertEqual(self.index.status("acme","refund"),old)
def test_empty_or_invalid_window_is_rejected(self):
with self.assertRaises(ValueError):list(chunks("abc",10,10))
with self.assertRaises(ValueError):self.index.search("",self.alice)
with self.assertRaises(ValueError):self.index.search("a"*1025,self.alice)
def test_encoder_or_chunk_contract_change_requires_rebuild(self):
with self.assertRaises(ValueError):Index(self.path,chunk_size=81,overlap=16)
def test_unknown_query_abstains_and_answer_is_explicitly_non_llm(self):
hits=self.index.search("zxqv-unrelated-token",self.alice)
self.assertEqual(hits,[])
self.assertEqual(self.index.answer(hits,self.alice),{"state":"abstain","non_llm":True,
"answer":"没有当前权限下可核查的证据。","evidence":[]})
if __name__=="__main__":unittest.main()