RAG 检索评测与回归门禁：hit@k、MRR 与逐题保护实测

"""Synthetic support knowledge base and evaluation questions (fictional product, demo data only)."""
 
DOCS = {
    "d01-refund": "购买后七天内且云端存储使用量不超过 5GB 的个人套餐，可以在“订单”页面申请全额退款。退款会原路退回支付账户，通常在三个工作日内到账。企业套餐不适用七天无理由退款。",
    "d02-invoice": "发票可在“订单详情”中申请，支持电子普通发票和增值税专用发票。开票信息提交后一至三个工作日内开具，电子发票会发送到账号绑定的邮箱。已开具的发票不能修改抬头，需要红冲后重新开具。",
    "d03-upgrade": "个人套餐可随时升级到更大的容量，升级后立即生效，差价按剩余天数折算。容量降级只能在当前订阅到期后进行，且已用空间超过新容量时需要先删除多余文件。",
    "d04-share-link": "分享链接默认有效期为七天，可以设置为一天、七天、三十天或永久，也可以加提取码。链接创建者随时可以在“我的分享”里取消分享，取消后已经发出的链接立即失效。",
    "d05-trash": "删除的文件会先进入回收站并保留三十天，期间可以一键还原。超过三十天后文件被永久清除且无法恢复。清空回收站会立即永久删除其中所有文件。",
    "d06-login": "忘记密码时，可以在登录页点击“找回密码”，通过绑定的手机号或邮箱接收验证码重置。连续输入错误密码五次后，账号会被锁定三十分钟。",
    "d07-sync": "桌面客户端会监听同步文件夹的变动，修改后在几秒内上传到云端，并同步到登录同一账号的其他设备。两台设备同时修改同一个文件时，会保留两个版本并在文件名后加上设备名。",
    "d08-security": "所有文件在传输过程中使用 TLS 加密，存储时使用服务端加密。账号支持开启两步验证，开启后新设备登录需要额外输入动态验证码。",
    "d09-team-roles": "团队空间的成员分为所有者、管理员、编辑者和只读成员四种角色。只有所有者和管理员可以邀请新成员或修改他人角色，只读成员只能查看和下载文件。",
    "d10-export": "需要迁出全部数据时，可以在“设置”里申请数据导出，系统会打包所有文件并生成下载链接，下载链接七天内有效。体积较大时会拆分成多个压缩包。",
    "d11-close-account": "账号注销前需要先清空或转移团队空间的所有权，并确认没有未完成的订单。注销提交后有七天冷静期，期间重新登录即可取消注销，冷静期结束后账号和数据将被永久删除。",
    "d12-versions": "每个文件会保留最近三十个历史版本，可以在文件详情页的“版本历史”里预览并恢复到任意版本。企业套餐的版本保留数量可以由管理员调整。",
}
 
# kind: "direct" reuses the key terms of the answer passage; "paraphrase" avoids them.
QUESTIONS = [
    ("个人套餐七天内可以申请全额退款吗", "d01-refund", "direct"),
    ("增值税专用发票在哪里申请", "d02-invoice", "direct"),
    ("升级容量后差价怎么算", "d03-upgrade", "direct"),
    ("分享链接的默认有效期是多久", "d04-share-link", "direct"),
    ("回收站里的文件保留多少天", "d05-trash", "direct"),
    ("连续输错密码会被锁定多久", "d06-login", "direct"),
    ("桌面客户端修改文件后多久同步到云端", "d07-sync", "direct"),
    ("开启两步验证后新设备登录要做什么", "d08-security", "direct"),
    ("团队空间有哪几种成员角色", "d09-team-roles", "direct"),
    ("数据导出的下载链接多久有效", "d10-export", "direct"),
    ("买了没用上想把钱要回来怎么办", "d01-refund", "paraphrase"),
    ("公司报销需要带税号的单据", "d02-invoice", "paraphrase"),
    ("硬盘空间不够了想买大一点的", "d03-upgrade", "paraphrase"),
    ("不想让别人继续访问我发出去的地址", "d04-share-link", "paraphrase"),
    ("误删了东西还能找回来吗", "d05-trash", "paraphrase"),
    ("我忘了账号口令该怎么办", "d06-login", "paraphrase"),
    ("两个电脑同时改了一份文档会怎样", "d07-sync", "paraphrase"),
    ("想把我所有的资料搬到别的平台", "d10-export", "paraphrase"),
    ("我不想再用这个服务了怎么删号", "d11-close-account", "paraphrase"),
    ("文件被改坏了想退回之前的样子", "d12-versions", "paraphrase"),
]

"""Deterministic local embeddings: hashed character 1-gram and 2-gram counts.
 
This is a lexical stand-in for a real embedding model. It needs no network or API key and
returns identical vectors on every machine, which is what a regression experiment requires.
"""
import zlib
 
import numpy as np
from langchain_core.embeddings import Embeddings
 
 
class HashingEmbeddings(Embeddings):
    def __init__(self, dim: int = 512):
        if dim < 8:
            raise ValueError("dim must be at least 8")
        self.dim = dim
 
    def _embed(self, text: str) -> list[float]:
        vec = np.zeros(self.dim, dtype=np.float64)
        chars = [c for c in text if not c.isspace()]
        grams = chars + ["".join(chars[i : i + 2]) for i in range(len(chars) - 1)]
        for gram in grams:
            h = zlib.crc32(gram.encode("utf-8"))
            vec[h % self.dim] += 1.0 if (h >> 16) & 1 else -1.0
        norm = np.linalg.norm(vec)
        return (vec / norm).tolist() if norm else vec.tolist()
 
    def embed_documents(self, texts: list[str]) -> list[list[float]]:
        return [self._embed(t) for t in texts]
 
    def embed_query(self, text: str) -> list[float]:
        return self._embed(text)

"""Build a retriever over DOCS and evaluate it with hit@k and MRR."""
from dataclasses import dataclass
 
from langchain_core.documents import Document
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_text_splitters import RecursiveCharacterTextSplitter
 
from dataset import DOCS, QUESTIONS
from hashing_embeddings import HashingEmbeddings
 
 
@dataclass(frozen=True)
class Config:
    chunk_size: int = 200
    chunk_overlap: int = 0
    dim: int = 512
    k: int = 3
 
 
def build_search(cfg: Config):
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=cfg.chunk_size,
        chunk_overlap=cfg.chunk_overlap,
        separators=["\n\n", "\n", "。", "，", ""],
    )
    chunks = []
    for doc_id, text in DOCS.items():
        for piece in splitter.split_text(text):
            chunks.append(Document(page_content=piece, metadata={"doc_id": doc_id}))
    store = InMemoryVectorStore(HashingEmbeddings(cfg.dim))
    store.add_documents(chunks)
 
    def search(question: str) -> list[str]:
        """Return distinct doc ids in rank order, using up to 4*k chunks."""
        hits = store.similarity_search(question, k=cfg.k * 4)
        ranked: list[str] = []
        for hit in hits:
            doc_id = hit.metadata["doc_id"]
            if doc_id not in ranked:
                ranked.append(doc_id)
        return ranked[: cfg.k]
 
    search.n_chunks = len(chunks)
    return search
 
 
def evaluate(cfg: Config, questions=QUESTIONS) -> list[dict]:
    search = build_search(cfg)
    rows = []
    for question, gold, kind in questions:
        ranked = search(question)
        rank = ranked.index(gold) + 1 if gold in ranked else None
        rows.append({"question": question, "gold": gold, "kind": kind, "rank": rank})
    return rows
 
 
def hit_at(rows: list[dict], k: int) -> float:
    return sum(1 for r in rows if r["rank"] is not None and r["rank"] <= k) / len(rows)
 
 
def mrr(rows: list[dict]) -> float:
    return sum(1 / r["rank"] for r in rows if r["rank"]) / len(rows)

"""Evaluate the baseline retriever: overall metrics, split by question kind, and the misses."""
from retrieval import Config, build_search, evaluate, hit_at, mrr
 
cfg = Config()
rows = evaluate(cfg)
print(f"config: {cfg}")
print(f"chunks: {build_search(cfg).n_chunks}, questions: {len(rows)}")
 
 
def summary(label: str, subset: list[dict]) -> None:
    print(f"{label:<11} n={len(subset):>2}  hit@1={hit_at(subset, 1):.2f}  hit@3={hit_at(subset, 3):.2f}  MRR={mrr(subset):.3f}")
 
 
summary("all", rows)
summary("direct", [r for r in rows if r["kind"] == "direct"])
summary("paraphrase", [r for r in rows if r["kind"] == "paraphrase"])
 
print("misses at k=3:")
for r in rows:
    if r["rank"] is None or r["rank"] > 3:
        print(f"  [{r['kind']}] {r['question']} -> expected {r['gold']}")

config: Config(chunk_size=200, chunk_overlap=0, dim=512, k=3)
chunks: 12, questions: 20
all         n=20  hit@1=0.60  hit@3=0.80  MRR=0.683
direct      n=10  hit@1=1.00  hit@3=1.00  MRR=1.000
paraphrase  n=10  hit@1=0.20  hit@3=0.60  MRR=0.367
misses at k=3:
  [paraphrase] 公司报销需要带税号的单据 -> expected d02-invoice
  [paraphrase] 不想让别人继续访问我发出去的地址 -> expected d04-share-link
  [paraphrase] 误删了东西还能找回来吗 -> expected d05-trash
  [paraphrase] 想把我所有的资料搬到别的平台 -> expected d10-export

"""Record the baseline results that regression_gate.py compares against (writes baseline.json)."""
import json
from pathlib import Path
 
from regression_gate import dataset_fingerprint
from retrieval import Config, evaluate, hit_at
 
cfg = Config()
rows = evaluate(cfg)
payload = {
    "config": cfg.__dict__,
    "dataset": dataset_fingerprint(),
    "hit_at_3": hit_at(rows, 3),
    "questions": [{"question": r["question"], "rank": r["rank"]} for r in rows],
}
Path(__file__).with_name("baseline.json").write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
print(f"recorded baseline: hit@3={payload['hit_at_3']:.2f} over {len(rows)} questions")

"""Compare a candidate retriever configuration with the recorded baseline.
 
usage: regression_gate.py <candidate>
Two rules decide the verdict:
  1. aggregate: hit@3 must not drop below the baseline;
  2. protected questions: every question the baseline answered within the top 3 must still be.
Exit code 0 = pass, 1 = fail, 2 = the recorded baseline does not match the current dataset.
"""
import hashlib
import json
import sys
from pathlib import Path
 
from dataset import DOCS, QUESTIONS
from retrieval import Config, evaluate, hit_at, mrr
 
CANDIDATES = {
    "same": Config(),
    "chunk40": Config(chunk_size=40),
    "dim64": Config(dim=64),
    "dim1024": Config(dim=1024),
}
BASELINE_FILE = Path(__file__).with_name("baseline.json")
K = 3
 
 
def dataset_fingerprint() -> str:
    payload = json.dumps([DOCS, QUESTIONS], ensure_ascii=False, sort_keys=True)
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:12]
 
 
def main(name: str) -> int:
    if name not in CANDIDATES:
        print(f"unknown candidate {name!r}; choose from {sorted(CANDIDATES)}")
        return 2
    baseline = json.loads(BASELINE_FILE.read_text(encoding="utf-8"))
    if baseline["dataset"] != dataset_fingerprint():
        print("baseline is stale: the dataset changed; re-record it with record_baseline.py")
        return 2
 
    cand_rows = evaluate(CANDIDATES[name])
    base_rank = {q["question"]: q["rank"] for q in baseline["questions"]}
    base_hit3 = baseline["hit_at_3"]
    cand_hit3 = hit_at(cand_rows, K)
 
    print(f"candidate {name}: {CANDIDATES[name]}")
    print(f"baseline hit@3={base_hit3:.2f}  candidate hit@3={cand_hit3:.2f}  candidate MRR={mrr(cand_rows):.3f}")
 
    broken = [r for r in cand_rows if (base_rank[r["question"]] or 99) <= K and (r["rank"] or 99) > K]
    fixed = [r for r in cand_rows if (base_rank[r["question"]] or 99) > K and (r["rank"] or 99) <= K]
    for r in broken:
        print(f"  regressed: {r['question']} (rank {base_rank[r['question']]} -> {r['rank']})")
    for r in fixed:
        print(f"  improved:  {r['question']} (rank {base_rank[r['question']]} -> {r['rank']})")
 
    aggregate_ok = cand_hit3 >= base_hit3
    protected_ok = not broken
    print(f"aggregate gate: {'PASS' if aggregate_ok else 'FAIL'}")
    print(f"protected-question gate: {'PASS' if protected_ok else 'FAIL'}")
    verdict = aggregate_ok and protected_ok
    print(f"verdict: {'PASS' if verdict else 'FAIL'}")
    return 0 if verdict else 1
 
 
if __name__ == "__main__":
    sys.exit(main(sys.argv[1] if len(sys.argv) > 1 else "same"))

candidate same: Config(chunk_size=200, chunk_overlap=0, dim=512, k=3)
baseline hit@3=0.80  candidate hit@3=0.80  candidate MRR=0.683
aggregate gate: PASS
protected-question gate: PASS
verdict: PASS

candidate chunk40: Config(chunk_size=40, chunk_overlap=0, dim=512, k=3)
baseline hit@3=0.80  candidate hit@3=0.80  candidate MRR=0.708
  regressed: 硬盘空间不够了想买大一点的 (rank 3 -> None)
  regressed: 我忘了账号口令该怎么办 (rank 1 -> None)
  improved:  不想让别人继续访问我发出去的地址 (rank None -> 1)
  improved:  误删了东西还能找回来吗 (rank None -> 3)
aggregate gate: PASS
protected-question gate: FAIL
verdict: FAIL

candidate dim64: Config(chunk_size=200, chunk_overlap=0, dim=64, k=3)
baseline hit@3=0.80  candidate hit@3=0.70  candidate MRR=0.625
  regressed: 买了没用上想把钱要回来怎么办 (rank 2 -> None)
  regressed: 硬盘空间不够了想买大一点的 (rank 3 -> None)
  regressed: 我忘了账号口令该怎么办 (rank 1 -> None)
  improved:  公司报销需要带税号的单据 (rank None -> 2)
aggregate gate: FAIL
protected-question gate: FAIL
verdict: FAIL

candidate dim1024: Config(chunk_size=200, chunk_overlap=0, dim=1024, k=3)
baseline hit@3=0.80  candidate hit@3=0.85  candidate MRR=0.725
  regressed: 硬盘空间不够了想买大一点的 (rank 3 -> None)
  improved:  不想让别人继续访问我发出去的地址 (rank None -> 1)
  improved:  误删了东西还能找回来吗 (rank None -> 3)
aggregate gate: PASS
protected-question gate: FAIL
verdict: FAIL

"""How much can a hit@3 difference be trusted with only 20 questions? Paired bootstrap, fixed seed."""
import random
 
from retrieval import Config, evaluate
 
N_RESAMPLES = 5000
rng = random.Random(2026)
 
 
def hits(cfg: Config) -> list[int]:
    return [1 if r["rank"] is not None and r["rank"] <= 3 else 0 for r in evaluate(cfg)]
 
 
def interval(values: list[float]) -> tuple[float, float]:
    values = sorted(values)
    return values[int(0.025 * len(values))], values[int(0.975 * len(values)) - 1]
 
 
base = hits(Config())
n = len(base)
print(f"questions: {n}; one question flips hit@3 by {100 / n:.1f} points")
 
boot = [sum(base[rng.randrange(n)] for _ in range(n)) / n for _ in range(N_RESAMPLES)]
low, high = interval(boot)
print(f"baseline hit@3 = {sum(base) / n:.2f}, 95% bootstrap interval [{low:.2f}, {high:.2f}]")
 
for label, cfg in [("dim1024", Config(dim=1024)), ("dim64", Config(dim=64))]:
    cand = hits(cfg)
    diffs = []
    for _ in range(N_RESAMPLES):
        idx = [rng.randrange(n) for _ in range(n)]
        diffs.append(sum(cand[i] - base[i] for i in idx) / n)
    low, high = interval(diffs)
    print(f"{label}: hit@3 {sum(cand) / n:.2f} vs {sum(base) / n:.2f}, paired difference interval [{low:+.2f}, {high:+.2f}]")

questions: 20; one question flips hit@3 by 5.0 points
baseline hit@3 = 0.80, 95% bootstrap interval [0.60, 0.95]
dim1024: hit@3 0.85 vs 0.80, paired difference interval [-0.10, +0.20]
dim64: hit@3 0.70 vs 0.80, paired difference interval [-0.30, +0.10]

python regression_gate.py candidate   # exit 0 = pass, 1 = regression, 2 = stale baseline