""" LSH Banding：碰撞概率公式、S 型曲线与候选集（第18章 §18.3.3） 正文的推导（签名长度 n = b·r）： 单个 Band 匹配 P = t^r b 个 Band 全不匹配 P = (1 − t^r)^b 至少一个 Band 匹配 P(candidate) = 1 − (1 − t^r)^b ← 经典公式 其中 t = P(h(x) = h(y))：MinHash 时是 Jaccard，SimHash 时是 1 − θ/π。 拐点（碰撞概率达到 1/2 的位置）满足 t_0^r = 1 − 2^(−1/b) ⟹ t_0 ≈ (ln2 / b)^(1/r) ≈ (1/b)^(1/r) 本脚本分四段，每一段都对应正文里的一个论断： 1) 曲线本身：固定 (r, b) 打印 P(t)，并验证「r 增大曲线右移、b 增大左移」； 2) 用随机 bit 实测碰撞率，确认 1−(1−t^r)^b 不是近似而是精确概率； 3) 拐点：精确解 1−2^(−1/b) 与工程近似 (1/b)^(1/r) 的差距； 4) 真实检索：MinHash 签名 + Banding 建桶，扫 (b, r)， 统计 Recall@10 / 候选数 / 耗时，展示 b 与 r 的取舍。 运行：python3 ch03_lsh/lsh_banding.py """ from __future__ import annotations import pathlib import sys import time import numpy as np sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent)) sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent)) from common import table # noqa: E402 from minhash import jaccard, make_sets # noqa: E402 SEED = 2026 P = 2 ** 31 - 1 def collide_prob(t: float, r: int, b: int) -> float: """正文的碰撞概率公式 1 − (1 − t^r)^b。""" return 1.0 - (1.0 - t ** r) ** b def hinge_exact(b: int, r: int) -> float: """P(t_0)=1/2 的精确解：t_0^r = 1 − 2^(−1/b)。""" return float((1.0 - 2.0 ** (-1.0 / b)) ** (1.0 / r)) def hinge_approx(b: int, r: int) -> float: """正文的工程近似 t_0 ≈ (1/b)^(1/r)。""" return float((1.0 / b) ** (1.0 / r)) # -------------------------------------------------------------------------- # MinHash 签名（线性哈希，与 §18.3.1 的实现一致） # -------------------------------------------------------------------------- class MinHashSigner: def __init__(self, num_hash: int, seed: int = SEED) -> None: rng = np.random.default_rng(seed) self.L = num_hash self.a = rng.integers(1, P, size=num_hash, dtype=np.int64) self.b = rng.integers(0, P, size=num_hash, dtype=np.int64) def sign(self, s: set) -> np.ndarray: items = np.fromiter(s, dtype=np.int64, count=len(s)) h = (self.a[:, None] * items[None, :] + self.b[:, None]) % P return h.min(axis=1) def band_keys(sig: np.ndarray, r: int) -> list[tuple]: """把签名切成 b 个连续 Band，每个 Band 的 r 位作为倒排表的 Key。""" out = [] for s in range(0, len(sig) - r + 1, r): out.append(tuple(int(x) for x in sig[s:s + r])) return out def main() -> None: print("LSH Banding：碰撞概率公式、S 型曲线与候选集（§18.3.3）\n") # ---------------- 1) S 型曲线 ---------------- print("① 碰撞概率 P(t) = 1 − (1 − t^r)^b：") ts = [0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9] rows = [] for t in ts: rows.append([f"{t:.1f}"] + [f"{collide_prob(t, r, 16):.3f}" for r in (2, 4, 8)]) table(["t", "r=2, b=16", "r=4, b=16", "r=8, b=16"], rows) print("\n 固定 b=16，只增大 r：同一条 t 上的碰撞概率单调下降，曲线整体右移。") print(" 含义是「筛选更严格」——只有更像的对象才能进候选集。") rows = [] for t in ts: rows.append([f"{t:.1f}"] + [f"{collide_prob(t, 4, b):.3f}" for b in (4, 16, 64)]) table(["t", "r=4, b=4", "r=4, b=16", "r=4, b=64"], rows) print("\n 固定 r=4，只增大 b：碰撞概率单调上升，曲线整体左移。") print(" b 是「独立抽签次数」，次数越多，只要中一次就进候选，所以召回更高、") print(" 候选也更多。两者合起来构成一条可控的 S 型判别曲线。\n") # ---------------- 2) 实测碰撞率 ---------------- print("② 实测：从随机 bit 签名统计碰撞率，验证公式是精确恒等式") print(" （每条 2000 对、签名长度 n=256，t 位相同的概率由独立翻转控制）") n_bits = 256 n_pairs = 2000 rng = np.random.default_rng(SEED) base = rng.integers(0, 2, size=(n_pairs, n_bits)).astype(np.int8) rows = [] for t in (0.3, 0.5, 0.7, 0.9): flips = rng.random((n_pairs, n_bits)) > t # 以 1−t 的概率翻转 other = base ^ flips.astype(np.int8) for r, b in ((4, 16), (8, 16)): hit = np.zeros(n_pairs, dtype=bool) for s in range(0, r * b, r): same_band = (base[:, s:s + r] == other[:, s:s + r]).all(axis=1) hit |= same_band rows.append([f"{t:.1f}", r, b, f"{hit.mean():.4f}", f"{collide_prob(t, r, b):.4f}"]) table(["t", "r", "b", "实测碰撞率", "公式值"], rows) print("\n 逐行相等（误差在 2000 次抽样的 ±0.01 量级）。") print(" 注意公式里唯一的假设是「各位独立」；MinHash 的 L 个哈希如果用的是") print(" 同一组随机数，独立性会打折扣，工程上通常直接用 L 个独立种子。\n") # ---------------- 3) 拐点 ---------------- print("③ 拐点位置：精确解 vs 正文的 (1/b)^(1/r) 近似") rows = [] for b, r in ((16, 4), (16, 8), (32, 4), (64, 4), (64, 8), (256, 4)): te = hinge_exact(b, r) ta = hinge_approx(b, r) rows.append([b, r, f"{te:.4f}", f"{ta:.4f}", f"{(ta - te) / te:+.1%}", f"1−2^(−1/{b}) = {1 - 2 ** (-1 / b):.5f}"]) table(["b", "r", "精确 t_0", "近似 t_0", "相对误差", "t_0^r 的精确值"], rows) print("\n 近似把 t_0^r 从 1−2^(−1/b) 换成了 1/b；b ≥ 16 时两者的差不到 4%，") print(" b 越大越准（1−2^(−1/b) ≈ ln2/b）。所以正文说它「较准确」，") print(" 做参数初筛足够，做严格的召回率承诺时应该用精确解。\n") # ---------------- 4) 真实检索：MinHash + Banding ---------------- print("④ 真实检索：2000 个集合、40 个簇（同簇 J≈0.47）、Top-10") print(" （签名长度 L 固定，只改 Band 的切法；Recall@10 用真值 Jaccard 排序做基准）") sets, labels = make_sets(n_sets=2000, universe=20000, size=40, n_groups=40, keep=0.8) # 全量精确 Jaccard，作为真值 truth = [] for i in range(len(sets)): js = [] for j in range(len(sets)): if i == j: js.append(-1.0) continue inter = len(sets[i] & sets[j]) js.append(inter / (len(sets[i]) + len(sets[j]) - inter)) truth.append(np.argsort(-np.asarray(js))[:10]) L_FIX = 512 signer = MinHashSigner(L_FIX, seed=SEED) sigs = np.stack([signer.sign(s) for s in sets]) # [2000, 512] rows = [] for r in (2, 4, 8, 16): b = L_FIX // r t0 = time.perf_counter() tables = [dict() for _ in range(b)] for doc in range(len(sets)): for bi, key in enumerate(band_keys(sigs[doc], r)): tables[bi].setdefault(key, []).append(doc) t_build = time.perf_counter() - t0 rec = 0.0 cand_tot = 0 t0_ = time.perf_counter() for q in range(len(sets)): cand: set = set() for bi, key in enumerate(band_keys(sigs[q], r)): cand.update(tables[bi].get(key, ())) cand.discard(q) cand_arr = np.fromiter(cand, dtype=np.int64, count=len(cand)) if len(cand_arr) == 0: continue # 候选内用签名估计相似度排序（真实系统里这一步是「粗排」） sim = (sigs[cand_arr] == sigs[q][None, :]).mean(axis=1) top = cand_arr[np.argsort(-sim)[:10]] rec += len(set(top.tolist()) & set(truth[q].tolist())) / 10 cand_tot += len(cand_arr) dt = (time.perf_counter() - t0_) / len(sets) p_theory = collide_prob(0.47, r, b) rows.append([r, b, f"{p_theory:.3f}", f"{rec / len(sets):.3f}", f"{cand_tot / len(sets):.0f}", f"{t_build * 1e3:.0f} ms", f"{dt * 1e3:.2f} ms"]) table(["r", "b", "理论碰撞率(t=0.47)", "Recall@10", "平均候选数", "建表耗时", "单查询"], rows) print("\n 三列对照着读：r 从 2 增到 16，理论碰撞率按 1−(1−t^r)^b 剧烈下降，") print(" 实测 Recall 同步下降，候选数也从 61 条收敛到 0 条。") print(" r=2、r=4 两档都一样，是因为 61 和 50 个候选里已经包含了几乎全部") print(" 同簇对象，差别只剩粗排能否排进 Top-10——这正是「候选数到瓶颈后，") print(" 召回由精排决定」的现象。") print(" r=8 时理论碰撞率只剩 0.14，候选数掉到 8 条，召回随之掉到 0.26：") print(" Banding 一旦成了瓶颈，后面的粗排再准也救不回来。\n") print(" 固定 r=2、只加长签名（同时让 b = L/2），看召回的天花板在哪里：") rows = [] for L in (64, 128, 256, 512, 1024): mh = MinHashSigner(L, seed=SEED) sg = np.stack([mh.sign(s) for s in sets]) b2 = L // 2 tabs = [dict() for _ in range(b2)] for doc in range(len(sets)): for bi in range(b2): tabs[bi].setdefault(tuple(int(x) for x in sg[doc, 2 * bi:2 * bi + 2]), []).append(doc) rec = 0.0 cand_tot = 0 for q in range(len(sets)): cand: set = set() for bi in range(b2): cand.update(tabs[bi].get(tuple(int(x) for x in sg[q, 2 * bi:2 * bi + 2]), ())) cand.discard(q) ca = np.fromiter(cand, dtype=np.int64, count=len(cand)) if len(ca) == 0: continue sim = (sg[ca] == sg[q][None, :]).mean(axis=1) top = ca[np.argsort(-sim, kind="stable")[:10]] rec += len(set(top.tolist()) & set(truth[q].tolist())) / 10 cand_tot += len(ca) rows.append([L, b2, f"{rec / len(sets):.3f}", f"{cand_tot / len(sets):.1f}"]) table(["签名长度 L", "b（r=2）", "Recall@10", "平均候选数"], rows) print("\n 候选数几乎不变（50 → 62），召回却从 0.45 一路升到 0.76：") print(" 瓶颈已经不在 Banding，而在「用签名估计相似度来排序」这一步——") print(" L=512 时同簇对象两两估计值的标准差约 0.02，几十个对象挤在同一个") print(" 分数上，Top-10 里谁进谁不进基本靠运气（L=64 时更糟）。") print(" 工业系统的做法是在候选内回原始表算精确相似度，而不是继续加长签名。\n") print(" 同一份签名只改 b（固定 r=2，签名长度 512 不变）：") rows = [] for b in (8, 16, 32, 64, 128, 256): tables = [dict() for _ in range(b)] for doc in range(len(sets)): for bi in range(b): key = tuple(int(x) for x in sigs[doc, 2 * bi:2 * bi + 2]) tables[bi].setdefault(key, []).append(doc) rec = 0.0 cand_tot = 0 for q in range(len(sets)): cand: set = set() for bi in range(b): key = tuple(int(x) for x in sigs[q, 2 * bi:2 * bi + 2]) cand.update(tables[bi].get(key, ())) cand.discard(q) cand_arr = np.fromiter(cand, dtype=np.int64, count=len(cand)) if len(cand_arr) == 0: continue sim = (sigs[cand_arr] == sigs[q][None, :]).mean(axis=1) top = cand_arr[np.argsort(-sim)[:10]] rec += len(set(top.tolist()) & set(truth[q].tolist())) / 10 cand_tot += len(cand_arr) rows.append([b, f"{collide_prob(0.47, 2, b):.4f}", f"{rec / len(sets):.3f}", f"{cand_tot / len(sets):.0f}"]) table(["b（r=2）", "理论碰撞率", "Recall@10", "平均候选数"], rows) print("\n b=8 时碰撞率 ≈ 0.86，约有 14% 的近邻根本进不了候选，召回卡在 0.67；") print(" b 增大后碰撞率趋于 1，召回升到 0.70 附近就不再涨——剩下的缺口来自") print(" 签名排序的分辨率，不是候选覆盖。候选数从 44 缓升到 61。") print(" 所以「加 b」是用候选数换召回，「加 r」是用召回换精度，二者要联调。\n") # 对照：随机采样同样候选数能做到多少 print(" 对照：随机抽相同数量的候选（200 条查询，签名粗排 Top-10）：") rows = [] rng2 = np.random.default_rng(SEED) for m in (50, 200, 800, 2000): rec = 0.0 for q in range(200): cand = rng2.choice(2000, m, replace=False) sim = (sigs[cand] == sigs[q][None, :]).mean(axis=1) top = cand[np.argsort(-sim)[:10]] rec += len(set(top.tolist()) & set(truth[q].tolist())) / 10 rows.append([f"{m}", f"{rec / 200:.3f}"]) table(["随机候选数", "Recall@10"], rows) print("\n 同样几百个候选，随机采样只有 10%–38% 的召回，而 Banding 的同桶候选") print(" 在同等规模下能到 70%——LSH 的价值不在于「少扫」，") print(" 而在于「扫的那一小撮是对的」。\n") print("⑤ 一句话：P(candidate)=1−(1−t^r)^b 把签名变成判别器，") print(" r 管每个 Band 的严格程度，b 管独立机会数，拐点在 t_0≈(1/b)^(1/r)。") if __name__ == "__main__": main()