#!/usr/bin/env python3 """01_hybrid_vs_single_retrieval.py — 无需 API key 的混合检索演示。 对比三种检索策略在同一合成语料 + 金标查询集上的表现: A) BM25 词法检索 (Lexical / sparse) B) 向量语义检索 (用「概念空间」玩具嵌入模拟真实 embedding model 的行为: 常见词映射到概念维度; 罕见 token(如错误码 TS-999)哈希到弱维度 → 会被稀释, 这正是官方文档说 embedding 会漏精确匹配的原因) C) 混合检索 (Hybrid: A + B 双路召回 + RRF 排名融合) 输出 recall@k / precision@k / MRR 与逐查询对比, 佐证: "Embeddings+BM25 is better than embeddings on their own" (来源: https://www.anthropic.com/engineering/contextual-retrieval, 访问于 2026-08-30) 运行: python3 01_hybrid_vs_single_retrieval.py 依赖: 仅 Python 标准库 """ from __future__ import annotations import math import re from collections import Counter # ---------------------------------------------------------------- 语料 (合成) -- # 每块: (chunk_id, text, metadata) metadata 用于演示 pre-filter 思想 CHUNKS = [ ("c01", "Our refund policy allows customers to request a full refund within 30 days of purchase.", {"doc_type": "policy", "product": "billing"}), ("c02", "To return a product, log in to your account, open the order page and choose return or exchange.", {"doc_type": "howto", "product": "billing"}), ("c03", "Subscription invoices are issued monthly and can be downloaded from the billing portal.", {"doc_type": "howto", "product": "billing"}), ("c04", "Error code TS-999 indicates a failed session handshake. Restart the gateway and check the clock skew.", {"doc_type": "troubleshooting", "product": "edge"}), ("c05", "Error code TS-421 means the license key expired. Renew the license in the admin console.", {"doc_type": "troubleshooting", "product": "edge"}), ("c06", "Error code TS-777 appears when the config file is malformed. Validate the YAML syntax first.", {"doc_type": "troubleshooting", "product": "edge"}), ("c07", "General guidance for all error codes: capture the log bundle before contacting support.", {"doc_type": "troubleshooting", "product": "edge"}), ("c08", "Rate limits on the public API are 60 requests per minute and 50,000 requests per day.", {"doc_type": "reference", "product": "api"}), ("c09", "Authentication uses short-lived bearer tokens. Refresh them with the /oauth/token endpoint.", {"doc_type": "reference", "product": "api"}), ("c10", "Data residency options let you pin storage to EU or US regions at the workspace level.", {"doc_type": "policy", "product": "security"}), ("c11", "SSO with SAML 2.0 is available on the enterprise plan. Ask your admin to enable it.", {"doc_type": "reference", "product": "security"}), ("c12", "The free tier includes 3 projects and community support only.", {"doc_type": "reference", "product": "billing"}), ("c13", "The pro plan costs 42 dollars per seat per month and adds priority support.", {"doc_type": "reference", "product": "billing"}), ("c14", "Scheduled downtime windows are announced 7 days in advance on the status page.", {"doc_type": "policy", "product": "api"}), ("c15", "How to get your money back if you were charged twice: contact billing and we reverse the duplicate charge.", {"doc_type": "howto", "product": "billing"}), ("c16", "SLA: monthly uptime of 99.9 percent for enterprise customers, with credits for breaches.", {"doc_type": "policy", "product": "api"}), ] # ------------------------------------------------------- 金标评测集 (golden set) -- # 意图: semantic = 换说法、少共享词 (向量强); exact = 精确代码/型号 (BM25 强); mixed = 两者 QUERIES = [ {"q": "how do I get my money back", "golden": ["c15", "c01"], "intent": "semantic"}, {"q": "I want to send the item back and swap it", "golden": ["c02"], "intent": "semantic"}, {"q": "what happens if I was billed two times", "golden": ["c15"], "intent": "semantic"}, {"q": "Error code TS-999", "golden": ["c04"], "intent": "exact"}, {"q": "TS-421 how to fix", "golden": ["c05"], "intent": "exact"}, {"q": "malformed config triggers TS-777", "golden": ["c06"], "intent": "exact"}, {"q": "how many API calls am I allowed each minute", "golden": ["c08"], "intent": "semantic"}, {"q": "where do I download my monthly invoice", "golden": ["c03"], "intent": "semantic"}, {"q": "uptime guarantee for enterprise customers", "golden": ["c16"], "intent": "mixed"}, {"q": "renew an expired license key TS-421", "golden": ["c05"], "intent": "mixed"}, ] # ------------------------------------------------------------------- 工具函数 -- _TOKEN_RE = re.compile(r"[a-z0-9]+(?:-[a-z0-9]+)*") def tokenize(text: str) -> list[str]: return _TOKEN_RE.findall(text.lower()) # ------------------------------------------------------------------- BM25 ------ class BM25: """经典 Okapi BM25 (k1=1.5, b=0.75), 演示实现。""" def __init__(self, docs: list[list[str]], k1: float = 1.5, b: float = 0.75): self.k1, self.b = k1, b self.doc_len = [len(d) for d in docs] self.avgdl = sum(self.doc_len) / len(docs) self.tf = [Counter(d) for d in docs] df: Counter = Counter() for d in docs: df.update(set(d)) n = len(docs) self.idf = {t: math.log(1 + (n - c + 0.5) / (c + 0.5)) for t, c in df.items()} def score(self, query: list[str], idx: int) -> float: s, dl = 0.0, self.doc_len[idx] for t in query: if t not in self.tf[idx]: continue f = self.tf[idx][t] idf = self.idf.get(t, 0.0) s += idf * f * (self.k1 + 1) / (f + self.k1 * (1 - self.b + self.b * dl / self.avgdl)) return s def rank(self, query: str, ids: list[str]) -> list[str]: q = tokenize(query) return sorted(ids, key=lambda cid: -self.score(q, int(cid[1:]) - 1)) # --------------------------------------------- 玩具「语义嵌入」(concept space) -- # 设计: 已知概念词 → 概念维度(权重 1.0); 未知 token → 稳定哈希维度(权重 0.3, 模拟真实 # embedding 对罕见字母数字 token 的「稀释」效应)。文档向量 = 词向量求和后归一化。 CONCEPTS = { "money": "finance", "back": "finance", "refund": "finance", "charged": "finance", "dollars": "finance", "invoice": "finance", "costs": "finance", "plan": "finance", "return": "returns", "swap": "returns", "exchange": "returns", "item": "returns", "back": "returns", "send": "shipping", "error": "errors", "fix": "repair", "restart": "repair", "renew": "repair", "license": "license", "expired": "license", "config": "config", "malformed": "config", "validate": "repair", "yaml": "config", "api": "api", "requests": "api", "calls": "api", "endpoint": "api", "minute": "limits", "allowed": "limits", "limits": "limits", "daily": "limits", "uptime": "sla", "guarantee": "sla", "sla": "sla", "credits": "sla", "download": "billing_portal", "monthly": "billing_portal", "portal": "billing_portal", "enterprise": "enterprise", "customers": "enterprise", "support": "enterprise", "auth": "security", "tokens": "security", "sso": "security", "security": "security", } CONCEPT_LIST = sorted(set(CONCEPTS.values())) CONCEPT_IDX = {c: i for i, c in enumerate(CONCEPT_LIST)} HASH_BASE = len(CONCEPT_LIST) DIM = HASH_BASE + 512 # 概念维度 + 哈希维度 RARE_WEIGHT = 0.3 # 罕见 token(错误码等)的稀释权重 def _stable_hash(token: str) -> int: h = 2166136261 for ch in token: h = ((h ^ ord(ch)) * 16777619) & 0xFFFFFFFF return h % 512 def embed(tokens: list[str]) -> list[float]: vec = [0.0] * DIM for t in tokens: concept = CONCEPTS.get(t) if concept is not None: vec[CONCEPT_IDX[concept]] += 1.0 else: # 罕见/未知 token 走哈希降权 → 被概念信号稀释 (模拟真实 embedding 行为) vec[HASH_BASE + _stable_hash(t)] += RARE_WEIGHT norm = math.sqrt(sum(v * v for v in vec)) or 1.0 return [v / norm for v in vec] def cosine(a: list[float], b: list[float]) -> float: return sum(x * y for x, y in zip(a, b)) # ------------------------------------------------------------------ RRF 融合 -- def rrf_fusion(rankings: list[list[str]], k: int = 60) -> list[str]: """Reciprocal Rank Fusion: score = Σ 1/(k + rank_i)。""" score: dict[str, float] = {} for ranking in rankings: for r, cid in enumerate(ranking, start=1): score[cid] = score.get(cid, 0.0) + 1.0 / (k + r) return sorted(score, key=lambda c: -score[c]) # ------------------------------------------------------------------- 指标 ----- def evaluate(rank_fn, queries, k: int = 5) -> dict[str, float]: recalls, precisions, rrs = [], [], [] for item in queries: ranked = rank_fn(item["q"])[:k] hits = len(set(ranked) & set(item["golden"])) recalls.append(hits / len(item["golden"])) precisions.append(hits / k) rr = 0.0 for r, cid in enumerate(rank_fn(item["q"]), start=1): if cid in item["golden"]: rr = 1.0 / r break rrs.append(rr) return {"recall@5": sum(recalls) / len(recalls), "precision@5": sum(precisions) / len(precisions), "MRR": sum(rrs) / len(rrs)} # ------------------------------------------------------------------- 主流程 --- def main() -> None: ids = [c[0] for c in CHUNKS] texts = [c[1] for c in CHUNKS] bm25 = BM25([tokenize(t) for t in texts]) doc_vecs = [embed(tokenize(t)) for t in texts] def bm25_rank(q: str) -> list[str]: return bm25.rank(q, ids) def vector_rank(q: str) -> list[str]: qv = embed(tokenize(q)) return sorted(ids, key=lambda cid: -cosine(qv, doc_vecs[int(cid[1:]) - 1])) def hybrid_rank(q: str) -> list[str]: return rrf_fusion([bm25_rank(q), vector_rank(q)]) print("=" * 78) print("合成语料: 16 块 | 金标查询: 10 条 (semantic/exact/mixed 三种意图)") print("=" * 78) # 逐查询: 三种方法的第一命中 print(f"\n{'query':44s} | {'BM25 首中':10s} | {'Vector 首中':10s} | {'Hybrid 首中':10s}") print("-" * 90) for item in QUERIES: first = lambda r: "HIT " + r[0] if r and r[0] in item["golden"] else "miss " + (r[0] if r else "-") print(f"{item['q'][:44]:44s} | {first(bm25_rank(item['q'])):10s} | " f"{first(vector_rank(item['q'])):10s} | {first(hybrid_rank(item['q'])):10s}") # 聚合指标 results = { "BM25 (lexical only)": evaluate(bm25_rank, QUERIES), "Vector (semantic only)": evaluate(vector_rank, QUERIES), "Hybrid (BM25 + Vector + RRF)": evaluate(hybrid_rank, QUERIES), } print("\n" + "=" * 78) print("聚合指标 (k=5)") print("=" * 78) print(f"{'策略':32s} | {'recall@5':9s} | {'precision@5':12s} | {'MRR':6s}") print("-" * 70) for name, m in results.items(): print(f"{name:32s} | {m['recall@5']:<9.2f} | {m['precision@5']:<12.2f} | {m['MRR']:<6.2f}") # 分意图统计: 展示两种单路的系统性盲区 print("\n分意图 recall@5 (按golden是否在Top-5):") for intent in ("semantic", "exact", "mixed"): sub = [q for q in QUERIES if q["intent"] == intent] row = [] for label, fn in (("BM25", bm25_rank), ("Vector", vector_rank), ("Hybrid", hybrid_rank)): hits = sum(1 for it in sub if set(fn(it["q"])[:5]) & set(it["golden"])) row.append(f"{label}: {hits}/{len(sub)}") print(f" {intent:9s} → " + " | ".join(row)) print(""" 结论 (与本语料/玩具嵌入的设计意图一致): 1. 精确代码查询 (TS-999 / TS-421 / TS-777): BM25 全中; 玩具向量把错误码哈希成弱维度, 被 error/license/config 等概念信号稀释 → 命中同主题的兄弟文档而非目标文档。 官方对照: "An embedding model might find content about error codes in general, but could miss the exact TS-999 match. BM25 looks for this specific text string." 2. 换说法查询 (money back / send the item back): 向量检索靠概念维度跨词面命中; BM25 因共享词少而失手。 3. Hybrid (RRF) 两路互补, 任一单路的系统性盲区都被另一路补上 —— 与 Anthropic 官方实验结论一致: Embeddings+BM25 好于单独 embeddings。 来源: https://www.anthropic.com/engineering/contextual-retrieval (访问于 2026-08-30) """) if __name__ == "__main__": main()