retrieval_fusion_pipeline.py

1"""2对应正文第 4 章 §4.4「召回结果融合」—— 去重、Quota 配额控制、截断、兜底的完整链路。34正文把融合模块的流程定义为:去重 -> 配额控制 -> 截断 -> 兜底,核心目标是5「保证候选集规模满足排序要求 / 保证各召回源合理流量分配 / 提升覆盖率与多样性 /6提高系统稳定性与鲁棒性」。本脚本按这个顺序逐步实现并打印每一步的候选规模变化。78实现要点(全部对齐正文):9  1) 去重(§4.4 召回结果去重)10     - 按 Item ID 唯一化;11     - 为每个物品保留一个「召回来源标识 retrieval reason」。正文明确指出 reason12       只记录其中一路,工业界两种做法:13         (a) 按召回源优先级选择(热门 > CF > 双塔 ...);14         (b) 按召回源内部得分选择(但不同源打分不可比)。15       脚本两种都实现,并打印二者给出的 reason 分布差异;16     - 跨路同物品的分数归并:正文指出各路分数不可比,所以归并前先做17       「组内排名归一化」 norm_k(i) = 1 - rank_k(i) / |R_k| ,再用18         merged(i) = max_k  w_k · norm_k(i)              (取最大,保守)19         merged(i) = 1 - Π_k (1 - w_k · norm_k(i))        (概率或,多路命中加分)20       两种口径都算出来并对比,w_k 为召回源权重。21  2) Quota 配额控制(§4.4 召回源配额控制)22     - 复刻正文表 4.x 的例子:热门 10000 -> 500,Item-CF 3000 -> 1000,23       双塔 5000 -> 1500,图召回 1000 -> 1000,目标总量 <= 5000;24     - 静态 Quota 之外实现「动态配额调整」:按各路近期下游通过率(粗排通过率)25       与独占贡献率重新分配配额,quota_k ∝ (base_k)^(1-β) · (gain_k)^β 后归一化。26  3) 候选集截断(§4.4 候选集截断)27     - 四种口径都实现:按召回源 Quota 截断、按归并分数全局截断、28       按业务优先级截断、按动态流量分配截断,并对比截断后各路占比。29  4) 兜底机制(§4.4 召回源兜底机制)30     - 注入「双塔索引异常返回空 / 图召回 RPC 超时」故障,触发多层兜底:31       类目热门 -> 全局热门 -> Redis 缓存上次结果 -> 历史推荐结果;32     - 保证最终候选数不低于 min_candidates。33  5) 召回监控与质量评估(§4.4 召回监控与质量评估)34     - 正文列出的监控指标:各路召回数量、覆盖率、去重率、CTR、观看时长、35       超时率、空结果率,全部统计并打印成监控表;36     - 另加两个工业界常用的质量指标:37         独占贡献率 unique_k = |只被 k 召回且进入最终候选| / |最终候选|38         下游通过率 pass_k   = |k 的候选进入粗排 Top N| / |k 的候选数|39         两路重合度 jaccard(k, l) = |R_k ∩ R_l| / |R_k ∪ R_l|4041[MOCK] MockRetrievalChannel 用合成分布生成各路候选(可注入超时/空结果/延迟),42       MockRedisCache 与 MockHistoryStore 是内存假存储,MockPreRanker 是一个43       线性打分器充当下游粗排,用于计算下游通过率。全部不访问真实服务。44"""4546import random47from collections import Counter4849N_ITEMS = 60000            # 教学规模的物品库(真实场景为亿级)50TOP_N_PRERANK = 500        # 粗排输出规模(正文:精排输入 500)51MAX_CANDIDATES = 5000      # 粗排输入规模上限(正文:粗排输入 5000)52MIN_CANDIDATES = 3000      # 兜底阈值535455# ============================================================ [MOCK] 召回通道56class MockRetrievalChannel:57    """[MOCK] 教学用召回源:按配置好的「物品池 + 偏好」生成候选与内部得分。5859    fail_mode: None / "empty"(索引失效返回空) / "timeout"(RPC 超时)60    内部得分刻意使用不同量纲(例如热门用播放量、双塔用 0~1 余弦),以复现61    正文指出的「不同召回源打分不太可比」问题。62    """6364    def __init__(self, name, reason_id, priority, raw_size, quota, pool,65                 score_fn, weight=1.0, fail_mode=None, latency_ms=0.0):66        self.name = name67        self.reason_id = reason_id68        self.priority = priority          # 数字越小优先级越高69        self.raw_size = raw_size70        self.quota = quota71        self.pool = pool72        self.score_fn = score_fn73        self.weight = weight74        self.fail_mode = fail_mode75        self.latency_ms = latency_ms7677    def retrieve(self, rng):78        """返回 (items, scores, status)。status ∈ ok / empty / timeout。"""79        if self.fail_mode == "timeout":80            return [], [], "timeout"81        if self.fail_mode == "empty":82            return [], [], "empty"83        items = rng.sample(self.pool, min(self.raw_size, len(self.pool)))84        scores = [self.score_fn(rng, k) for k in range(len(items))]85        order = sorted(range(len(items)), key=lambda k: -scores[k])86        return [items[k] for k in order], [scores[k] for k in order], "ok"878889class MockRedisCache:90    """[MOCK] 兜底用的 Redis 缓存:存上一次请求的融合结果。"""9192    def __init__(self):93        self.store = {}9495    def put(self, uid, items):96        self.store[uid] = list(items)9798    def get(self, uid):99        return list(self.store.get(uid, []))100101102class MockHistoryStore:103    """[MOCK] 兜底用的历史推荐结果存储。"""104105    def __init__(self, rng, pool, n=1500):106        self.items = rng.sample(pool, n)107108    def get(self, n):109        return self.items[:n]110111112class MockPreRanker:113    """[MOCK] 下游粗排:对候选做一次线性打分,用于统计各路「下游通过率」。"""114115    def __init__(self, rng, n_items):116        self.q = [rng.random() for _ in range(n_items)]     # 物品内在质量117118    def rank(self, cand_scores, top_n):119        s = {i: 0.7 * self.q[i] + 0.3 * sc for i, sc in cand_scores.items()}120        return [i for i, _ in sorted(s.items(), key=lambda x: -x[1])[:top_n]]121122123# ============================================================ 1. 去重 + 分数归并124def rank_normalize(items):125    """组内排名归一化:norm(i) = 1 - rank(i)/|R|,把不可比的原始分拉到同一量纲。"""126    n = max(len(items), 1)127    return {it: 1.0 - k / n for k, it in enumerate(items)}128129130def dedup_and_merge(results):131    """results: {name: (channel, items, scores)}。132133    返回 merged = {item: {"sources": [...], "norm": {name: v},134                          "score_max": float, "score_or": float,135                          "reason_prio": name, "reason_score": name}}136    """137    merged = {}138    for name, (ch, items, _scores) in results.items():139        norm = rank_normalize(items)140        for it in items:141            e = merged.setdefault(it, {"sources": [], "norm": {}})142            e["sources"].append(name)143            e["norm"][name] = ch.weight * norm[it]144    return merged145146147def finalize_scores(merged, channels):148    prio = {c.name: c.priority for c in channels}149    for it, e in merged.items():150        e["score_max"] = max(e["norm"].values())151        p = 1.0152        for v in e["norm"].values():153            p *= (1.0 - min(v, 0.999))154        e["score_or"] = 1.0 - p155        # reason 口径 (a):按召回源优先级156        e["reason_prio"] = min(e["sources"], key=lambda n: prio[n])157        # reason 口径 (b):按召回源内部归一化得分158        e["reason_score"] = max(e["norm"].items(), key=lambda kv: kv[1])[0]159    return merged160161162# ============================================================ 2. Quota 配额控制163def apply_quota(results, quotas):164    """对召回源 k 保留 min(|R_k|, quota_k) 个候选(正文表:10000->500 等)。"""165    kept = {}166    for name, (ch, items, scores) in results.items():167        q = quotas[name]168        kept[name] = (ch, items[:q], scores[:q])169    return kept170171172def dynamic_quota(base_quota, gain, beta=0.6, total=MAX_CANDIDATES):173    """动态配额:quota_k ∝ base_k^(1-β) · gain_k^β,再归一化到 total。174175    gain_k 取「独占贡献率 × 下游通过率」,即这一路带来的、别人带不来的有效候选。176    β=0 退化为静态配额,β=1 完全由效果驱动。177    """178    raw = {k: (base_quota[k] ** (1 - beta)) * (max(gain.get(k, 0.0), 1e-4) ** beta)179           for k in base_quota}180    s = sum(raw.values())181    out = {k: max(int(total * v / s), 50) for k, v in raw.items()}182    return out183184185# ============================================================ 3. 截断186def truncate_by_quota(kept, limit):187    """按召回源 Quota 截断:各路轮转取(round-robin),保证小召回源不被挤死。"""188    pools = {n: list(v[1]) for n, v in kept.items()}189    out, seen = [], set()190    while len(out) < limit and any(pools.values()):191        for n in list(pools):192            if not pools[n]:193                continue194            it = pools[n].pop(0)195            if it not in seen:196                seen.add(it)197                out.append((it, n))198            if len(out) >= limit:199                break200    return out201202203def truncate_by_score(merged, limit, key="score_or"):204    """按归并后的全局分数截断。"""205    order = sorted(merged.items(), key=lambda kv: -kv[1][key])[:limit]206    return [(it, e["reason_prio"]) for it, e in order]207208209def truncate_by_priority(kept, channels, limit):210    """按业务优先级截断:高优先级召回源先灌满,剩余额度给低优先级。"""211    out, seen = [], set()212    for ch in sorted(channels, key=lambda c: c.priority):213        if ch.name not in kept:214            continue215        for it in kept[ch.name][1]:216            if it in seen:217                continue218            seen.add(it)219            out.append((it, ch.name))220            if len(out) >= limit:221                return out222    return out223224225def truncate_by_traffic(kept, merged, weights, limit):226    """按动态流量分配截断:先按权重把 limit 拆成各路名额,各路内部按分数取。"""227    s = sum(weights.values())228    out, seen = [], set()229    for name, (ch, items, _) in kept.items():230        n_k = max(int(limit * weights[name] / s), 1)231        picked = 0232        for it in sorted(items, key=lambda x: -merged[x]["norm"].get(name, 0.0)):233            if it in seen:234                continue235            seen.add(it)236            out.append((it, name))237            picked += 1238            if picked >= n_k:239                break240    return out[:limit]241242243# ============================================================ 4. 兜底244def backup_fill(cand, min_n, cat_hot, global_hot, cache, history, uid):245    """多层兜底:类目热门 -> 全局热门 -> Redis 缓存 -> 历史推荐结果。"""246    seen = {it for it, _ in cand}247    out = list(cand)248    layers = [("backup_cat_hot", cat_hot),249              ("backup_global_hot", global_hot),250              ("backup_redis_cache", cache.get(uid)),251              ("backup_history", history.get(2000))]252    trace = []253    for name, pool in layers:254        if len(out) >= min_n:255            break256        added = 0257        for it in pool:258            if len(out) >= min_n:259                break260            if it not in seen:261                seen.add(it)262                out.append((it, name))263                added += 1264        trace.append((name, added, len(out)))265    return out, trace266267268# ============================================================ 5. 监控与质量评估269def jaccard(a, b):270    sa, sb = set(a), set(b)271    return len(sa & sb) / max(len(sa | sb), 1)272273274def monitor_table(results, merged, final, prerank_top, ctr_prior, dur_prior):275    """正文列出的监控指标 + 独占贡献率 / 下游通过率。"""276    rows = []277    for name, (ch, items, _) in results.items():278        n = len(items)279        dup = sum(1 for it in items if len(merged[it]["sources"]) > 1)280        uniq_final = sum(1 for it, r in final if r == name281                         and len(merged[it]["sources"]) == 1)282        in_final = sum(1 for it, r in final if r == name)283        passed = sum(1 for it in items if it in prerank_top)284        rows.append({285            "name": name,286            "n": n,287            "latency": ch.latency_ms,288            "coverage": n / N_ITEMS,289            "dup_rate": dup / max(n, 1),290            "ctr": ctr_prior[name],291            "duration": dur_prior[name],292            "timeout": 1.0 if ch.fail_mode == "timeout" else 0.0,293            "empty": 1.0 if ch.fail_mode == "empty" or n == 0 else 0.0,294            "in_final": in_final,295            "unique_rate": uniq_final / max(len(final), 1),296            "pass_rate": passed / max(n, 1),297        })298    return rows299300301# ============================================================ 主流程302def build_channels(rng):303    """构造正文表「召回源 Quota 示例」中的 4 路 + 2 路垂类/实时召回。"""304    hot_pool = list(range(0, 15000))                # 热门池偏头部305    cf_pool = list(range(5000, 35000))306    dssm_pool = list(range(10000, 55000))307    graph_pool = list(range(20000, 40000))308    tag_pool = list(range(30000, 50000))309    rt_pool = list(range(45000, 60000))310    return [311        MockRetrievalChannel("热门召回", 5501, 1, 10000, 500, hot_pool,312                             lambda r, k: 1e6 * (0.999 ** k), weight=0.6,313                             latency_ms=4.0),314        MockRetrievalChannel("ItemCF召回", 5512, 2, 3000, 1000, cf_pool,315                             lambda r, k: 0.8 * (0.998 ** k), weight=1.0,316                             latency_ms=12.0),317        MockRetrievalChannel("双塔召回", 5536, 3, 5000, 1500, dssm_pool,318                             lambda r, k: 0.95 - 0.00005 * k, weight=1.2,319                             latency_ms=18.0),320        MockRetrievalChannel("图召回", 5548, 4, 1000, 1000, graph_pool,321                             lambda r, k: 10.0 / (k + 3), weight=1.1,322                             latency_ms=25.0),323        MockRetrievalChannel("垂类Tag召回", 5560, 5, 2000, 600, tag_pool,324                             lambda r, k: 1.0 - k / 2000, weight=0.8,325                             latency_ms=6.0),326        MockRetrievalChannel("实时行为召回", 5571, 6, 800, 400, rt_pool,327                             lambda r, k: 2.0 * (0.99 ** k), weight=1.3,328                             latency_ms=9.0),329    ]330331332def main():333    random.seed(42)334    rng = random.Random(42)335336    channels = build_channels(rng)337    ctr_prior = {"热门召回": 0.0312, "ItemCF召回": 0.0487, "双塔召回": 0.0561,338                 "图召回": 0.0529, "垂类Tag召回": 0.0405, "实时行为召回": 0.0683}339    dur_prior = {"热门召回": 23.4, "ItemCF召回": 31.8, "双塔召回": 36.2,340                 "图召回": 34.5, "垂类Tag召回": 27.1, "实时行为召回": 41.6}341342    print("=" * 86)343    print("§4.4 召回结果融合:去重 -> Quota 配额控制 -> 截断 -> 兜底 -> 监控评估")344    print("=" * 86)345346    # ---------------- 多路召回执行(正常场景)347    results = {}348    for ch in channels:349        items, scores, status = ch.retrieve(rng)350        results[ch.name] = (ch, items, scores)351    raw_total = sum(len(v[1]) for v in results.values())352353    print("\n[Step 0] 多路召回原始结果")354    print(f"{'召回源':<16}{'reason_id':>10}{'原始候选数':>12}{'静态Quota':>11}{'源权重':>9}")355    print("-" * 60)356    for ch in channels:357        print(f"{ch.name:<16}{ch.reason_id:>10d}{len(results[ch.name][1]):>12d}"358              f"{ch.quota:>11d}{ch.weight:>9.1f}")359    print(f"{'合计':<16}{'':>10}{raw_total:>12d}{sum(c.quota for c in channels):>11d}")360361    # ---------------- Step 1 去重 + 分数归并362    merged_all = finalize_scores(dedup_and_merge(results), channels)363    dedup_total = len(merged_all)364    hit_dist = Counter(len(e["sources"]) for e in merged_all.values())365    print(f"\n[Step 1] 去重:{raw_total} -> {dedup_total} "366          f"(整体去重率 {1 - dedup_total / raw_total:.2%})")367    print("  被命中召回源个数分布:", dict(sorted(hit_dist.items())))368    pa = Counter(e["reason_prio"] for e in merged_all.values())369    pb = Counter(e["reason_score"] for e in merged_all.values())370    print("\n  召回 reason 两种设置口径的分布差异(正文:只记录其中一路)")371    print(f"{'召回源':<16}{'按优先级(a)':>14}{'按内部得分(b)':>16}{'差值':>10}")372    print("-" * 58)373    for ch in channels:374        print(f"{ch.name:<16}{pa[ch.name]:>14d}{pb[ch.name]:>16d}"375              f"{pb[ch.name] - pa[ch.name]:>+10d}")376377    multi = [it for it, e in merged_all.items() if len(e["sources"]) >= 3][:5]378    if multi:379        print("\n  跨路同物品的分数归并示例(多路命中 >= 3 的物品):")380        print(f"{'item':>8}{'命中源数':>9}{'max归并':>10}{'概率或归并':>12}   命中召回源")381        for it in multi:382            e = merged_all[it]383            print(f"{it:>8d}{len(e['sources']):>9d}{e['score_max']:>10.4f}"384                  f"{e['score_or']:>12.4f}   {'+'.join(e['sources'])}")385386    # ---------------- Step 2 Quota 控制387    base_quota = {c.name: c.quota for c in channels}388    kept = apply_quota(results, base_quota)389    after_quota = len({it for v in kept.values() for it in v[1]})390    print("\n[Step 2] 静态 Quota 控制(正文表 4.x 口径)")391    print(f"{'召回源':<16}{'原始候选数':>12}{'保留数量':>10}{'保留比例':>10}")392    print("-" * 50)393    for ch in channels:394        n0, n1 = len(results[ch.name][1]), len(kept[ch.name][1])395        print(f"{ch.name:<16}{n0:>12d}{n1:>10d}{(n1 / max(n0, 1)):>10.1%}")396    print(f"{'合计(去重后)':<16}{dedup_total:>12d}{after_quota:>10d}")397398    # ---------------- Step 3 四种截断口径对比399    prerank = MockPreRanker(rng, N_ITEMS)400    cand_scores = {it: e["score_or"] for it, e in merged_all.items()}401    prerank_top = set(prerank.rank(cand_scores, TOP_N_PRERANK))402403    weights = {c.name: c.weight for c in channels}404    strategies = {405        "按召回源Quota轮转": truncate_by_quota(kept, MAX_CANDIDATES),406        "按归并得分全局": truncate_by_score(merged_all, MAX_CANDIDATES),407        "按业务优先级": truncate_by_priority(kept, channels, MAX_CANDIDATES),408        "按动态流量分配": truncate_by_traffic(kept, merged_all, weights, MAX_CANDIDATES),409    }410    print(f"\n[Step 3] 候选集截断到 {MAX_CANDIDATES}:四种口径下各召回源占比")411    hdr = f"{'截断策略':<20}{'候选数':>8}"412    for ch in channels:413        hdr += f"{ch.name:>14}"414    print(hdr)415    print("-" * (28 + 14 * len(channels)))416    for sname, out in strategies.items():417        cnt = Counter(r for _, r in out)418        line = f"{sname:<20}{len(out):>8d}"419        for ch in channels:420            line += f"{cnt[ch.name] / max(len(out), 1):>14.1%}"421        print(line)422423    final = strategies["按动态流量分配"]424425    # ---------------- Step 4 质量评估与监控(正常场景)426    rows = monitor_table(results, merged_all, final, prerank_top,427                         ctr_prior, dur_prior)428    print("\n[Step 4] 召回监控与质量评估(正文列出的监控指标)")429    print(f"{'召回源':<16}{'召回数':>8}{'耗时/ms':>9}{'覆盖率':>9}{'去重率':>9}"430          f"{'CTR':>8}{'时长/s':>8}{'超时率':>8}{'空率':>7}{'独占贡献':>10}{'下游通过':>10}")431    print("-" * 104)432    for r in rows:433        print(f"{r['name']:<16}{r['n']:>8d}{r['latency']:>9.1f}{r['coverage']:>9.2%}"434              f"{r['dup_rate']:>9.1%}{r['ctr']:>8.4f}{r['duration']:>8.1f}"435              f"{r['timeout']:>8.0%}{r['empty']:>7.0%}{r['unique_rate']:>10.2%}"436              f"{r['pass_rate']:>10.2%}")437438    print("\n  两两召回源重合度 Jaccard(|R_k ∩ R_l| / |R_k ∪ R_l|):")439    names = [c.name for c in channels]440    print(" " * 16 + "".join(f"{n:>14}" for n in names))441    for a in names:442        line = f"{a:<16}"443        for b in names:444            line += f"{jaccard(results[a][1], results[b][1]):>14.3f}"445        print(line)446447    # ---------------- Step 5 动态配额调整448    gain = {r["name"]: max(r["unique_rate"], 1e-4) * max(r["pass_rate"], 1e-4)449            for r in rows}450    dq = dynamic_quota(base_quota, gain, beta=0.6)451    print("\n[Step 5] 动态配额调整  quota_k ∝ base_k^(1-β)·gain_k^β  "452          "(β=0.6, gain = 独占贡献率 × 下游通过率)")453    print(f"{'召回源':<16}{'静态Quota':>11}{'gain':>11}{'动态Quota':>11}{'变化':>10}")454    print("-" * 60)455    for ch in channels:456        b, d = base_quota[ch.name], dq[ch.name]457        print(f"{ch.name:<16}{b:>11d}{gain[ch.name]:>11.5f}{d:>11d}"458              f"{(d - b) / b:>+10.1%}")459460    kept2 = apply_quota(results, dq)461    final2 = truncate_by_traffic(kept2, merged_all, weights, MAX_CANDIDATES)462    top2 = set(prerank.rank({it: merged_all[it]["score_or"] for it, _ in final2},463                            TOP_N_PRERANK))464    base_top = set(prerank.rank({it: merged_all[it]["score_or"] for it, _ in final},465                                TOP_N_PRERANK))466    good = {i for i in range(N_ITEMS) if prerank.q[i] > 0.9}467    print(f"\n  静态配额最终候选 {len(final)} 个,其中优质物品(内在质量>0.9) "468          f"{len({it for it, _ in final} & good)} 个;")469    print(f"  动态配额最终候选 {len(final2)} 个,其中优质物品 "470          f"{len({it for it, _ in final2} & good)} 个;")471    print(f"  两者粗排 Top{TOP_N_PRERANK} 的重合度 "472          f"{len(top2 & base_top) / TOP_N_PRERANK:.1%}")473474    # ---------------- Step 6 故障注入 + 兜底475    print("\n" + "=" * 86)476    print("[Step 6] 故障注入与兜底机制(正文:索引失效 / RPC 超时 -> 多层兜底)")477    print("=" * 86)478    cache = MockRedisCache()479    cache.put("u_demo", [it for it, _ in final][:2500])480    history = MockHistoryStore(rng, list(range(N_ITEMS)))481    cat_hot = list(range(0, 1200))482    global_hot = list(range(0, 2500))483484    for ch in channels:485        if ch.name == "双塔召回":486            ch.fail_mode = "empty"        # 向量索引服务异常487        if ch.name == "图召回":488            ch.fail_mode = "timeout"      # 图召回 RPC 超时489490    results_f = {}491    for ch in channels:492        items, scores, status = ch.retrieve(rng)493        results_f[ch.name] = (ch, items, scores)494    merged_f = finalize_scores(dedup_and_merge(results_f), channels)495    kept_f = apply_quota(results_f, base_quota)496    final_f = truncate_by_traffic(kept_f, merged_f, weights, MAX_CANDIDATES)497    print(f"\n  故障后融合候选数 {len(final_f)} "498          f"(正常场景 {len(final)}),低于兜底阈值 {MIN_CANDIDATES}:"499          f" {len(final_f) < MIN_CANDIDATES}")500    filled, trace = backup_fill(final_f, MIN_CANDIDATES, cat_hot, global_hot,501                                cache, history, "u_demo")502    print(f"{'兜底层':<22}{'补充数量':>10}{'补充后候选数':>14}")503    print("-" * 48)504    for name, added, tot in trace:505        print(f"{name:<22}{added:>10d}{tot:>14d}")506    print(f"  兜底后最终候选数 {len(filled)},达到阈值: {len(filled) >= MIN_CANDIDATES}")507508    rows_f = monitor_table(results_f, merged_f, filled, prerank_top,509                           ctr_prior, dur_prior)510    print("\n  故障场景监控表(超时率/空结果率告警)")511    print(f"{'召回源':<16}{'召回数':>8}{'超时率':>9}{'空结果率':>10}{'进入最终候选':>14}")512    print("-" * 58)513    for r in rows_f:514        print(f"{r['name']:<16}{r['n']:>8d}{r['timeout']:>9.0%}{r['empty']:>10.0%}"515              f"{r['in_final']:>14d}")516    bk = Counter(r for _, r in filled if r.startswith("backup_"))517    print("  兜底来源构成:", dict(bk))518519    print("\n结论(正文 §4.4 要点回顾):")520    print("  1) 去重必须保留 reason,但 reason 只记一路;按优先级与按内部得分两种口径")521    print("     会给出明显不同的来源分布(见 Step 1 表),而正文指出 reason 不带来 AB 收益,")522    print("     因此工业界通常选实现最简单的优先级口径;")523    print("  2) 各路内部得分量纲不同(播放量 vs 余弦),直接比大小没有意义,")524    print("     融合前必须做组内排名归一化;")525    print("  3) Quota 的作用是防止热门召回这类大候选源独占候选集,从而维持多样性;")526    print("  4) 截断口径的选择直接改变候选集的来源构成,需要与业务目标对齐;")527    print("  5) 任何一路召回都会挂,兜底必须分层且保证最低候选规模,")528    print("     监控表的超时率/空结果率是发现这类故障的第一道信号。")529530531if __name__ == "__main__":532    main()