附录：随书实验完整代码
以下文件在构建时从随书实验与测试目录读取，与下载区实验包同源。它们是本书教学实现；实际验证范围以实验章节和验证记录为准。
lab/README.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 统计与时间侧信道未审计。lab/init.py
"""Executable retrieval experiments; standard-library core, optional local ONNX."""lab/data/corpus.json
[
{
"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元。仅指定人员可见。"
}
]lab/data/manifest.json
{
"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"
}
}lab/data/queries.json
[
{
"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": []
}
]lab/demo.py
"""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()lab/embedding.py
"""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()lab/evaluate.py
"""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()lab/evidence/bge-evaluation.json
{
"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"
]
}lab/evidence/demo.json
{
"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"
]
}lab/evidence/term-evaluation.json
{
"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"
]
}lab/fetch_embedding.py
"""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()lab/fixtures.py
"""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()]lab/index.py
"""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()lab/requirements-embedding.txt
numpy==2.5.4
onnxruntime==1.31.0
tokenizers==0.23.3lab/text.py
"""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.0tests/test_embedding.py
import 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()tests/test_retrieval.py
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()