filter_positions_pipeline.py

1"""2对应正文第 4 章 §4.5.1「过滤模块的位置」。34正文口径(逐条对齐):5  正文明确反驳「过滤模块只存在于召回之后」的误解,指出过滤逻辑贯穿整个推荐流程,6  在不同阶段承担不同职责:7    1) 召回前过滤(Pre-retrieval Filtering):剔除已下架、审核未通过、版权违规、8       已删除状态的内容,「避免无效内容进入后续推荐链路,从而减少系统计算资源浪费」;9    2) 召回后过滤(Post-retrieval Filtering):用户已消费内容过滤、用户屏蔽内容过滤、10       黑名单作者过滤、地域限制过滤,「进一步提升候选集质量」;11    3) 排序后过滤(Post-ranking Filtering):广告比例控制、风险内容降权、12       同作者内容数量限制,「避免排序模型过度追求点击率而损害用户体验」。1314本脚本做了什么:15  用一个进程内的 [MOCK] 推荐链路(内容库 -> 倒排索引 -> 召回 -> 粗排 -> 精排 -> 曝光)16  把这三个位置的过滤都实现出来,然后做三组对照实验:17    配置 A 全不过滤        —— 看会出现正文 §4.5 开头列出的哪些问题;18    配置 B 只在召回后过滤  —— 常见的初学者实现:无效内容仍然进了索引与召回打分,19                              算力被浪费,且候选经过滤后不足,要靠兜底补齐;20    配置 C 三段式过滤      —— 正文推荐的形态:越早过滤越省算力,只把必须依赖21                              最终排序结果的策略(广告占比/同作者打散)留到排序后。22  打印的指标:索引条目数、单请求召回扫描量、召回打分浪费率、候选不足触发兜底的次数、23  最终曝光列表中的无效内容率/已读率/屏蔽作者率/地域违规率/广告占比/同作者最大条数。2425关键公式(都是比例口径,方便与线上监控对齐):26  召回算力浪费率 = 进入召回打分但最终必被过滤的物品数 / 召回打分总数27  候选利用率     = 过滤后剩余候选数 / 召回候选数28  广告占比       = 曝光列表中的广告条数 / 曝光条数29  同作者约束     = max_author(曝光列表中同一作者的条数) <= L3031[MOCK] 内容库、倒排索引、用户画像、粗排/精排打分全部为合成数据与玩具打分函数,32不加载任何真实数据集,也不起真实服务。33"""3435import random36from collections import Counter3738N_ITEMS = 200_000        # 内容库规模39N_AUTHORS = 4_00040N_REGIONS = 641N_REQUESTS = 200         # 模拟请求数42RECALL_SIZE = 2_000      # 单请求召回候选规模43PRERANK_IN = 2_000       # 粗排输入上限44RANK_IN = 200            # 精排输入上限45EXPOSE_K = 20            # 单屏曝光条数46MIN_CANDIDATES = 1_900   # 候选不足则触发兜底47AD_RATIO_CAP = 0.20      # 排序后过滤:广告占比上限48SAME_AUTHOR_CAP = 2      # 排序后过滤:单屏同作者条数上限49RISK_DEMOTE = 0.3        # 排序后过滤:风险内容降权系数5051# 正文 §4.5.1 列出的「召回前过滤」要剔除的状态52INVALID_STATUS = ("offline", "audit_pending", "copyright", "deleted")53STATUS_DIST = [54    ("online", 0.82),55    ("offline", 0.06),56    ("audit_pending", 0.05),57    ("copyright", 0.04),58    ("deleted", 0.03),59]606162def gen_library(rng):63    """[MOCK] 生成内容库:状态 / 作者 / 地域许可 / 是否广告 / 风险等级 / 热度。"""64    items = []65    cum = []66    acc = 0.067    for name, p in STATUS_DIST:68        acc += p69        cum.append((acc, name))70    for iid in range(N_ITEMS):71        u = rng.random()72        status = next(name for bound, name in cum if u <= bound)73        items.append({74            "id": iid,75            "status": status,76            "author": rng.randrange(N_AUTHORS),77            # 地域限制:8% 的内容只在部分地域可分发78            "regions": (None if rng.random() > 0.0879                        else frozenset(rng.sample(range(N_REGIONS), 2))),80            "is_ad": rng.random() < 0.12,81            "risk": 1 if rng.random() < 0.05 else 0,82            "hot": rng.random() ** 2,          # 热度,决定被召回的概率83            "ctr": round(rng.uniform(0.01, 0.12), 4),84        })85    return items868788def build_index(items, pre_filter):89    """[MOCK] 倒排/正排索引。pre_filter=True 时执行正文的「召回前过滤」。9091    工业界的召回前过滤通常发生在索引构建/更新链路上:无效内容根本不入库,92    因此它省掉的不只是过滤本身的开销,而是整条下游链路对这些物品的全部计算。93    """94    if pre_filter:95        return [it for it in items if it["status"] not in INVALID_STATUS]96    return list(items)979899def retrieve(index, rng, size=RECALL_SIZE):100    """[MOCK] 多路召回的合并结果:按热度加权采样 size 个候选(含打分开销)。"""101    n = len(index)102    picked = {}103    tries = 0104    while len(picked) < size and tries < size * 6:105        tries += 1106        it = index[rng.randrange(n)]107        if rng.random() < 0.25 + 0.75 * it["hot"]:   # 热度越高越容易被召回108            picked[it["id"]] = it109    return list(picked.values()), tries110111112def post_retrieval_filter(cands, user, enable):113    """正文的「召回后过滤」:已读 / 用户屏蔽 / 黑名单作者 / 地域限制。"""114    if not enable:115        return cands, Counter()116    kept, dropped = [], Counter()117    for it in cands:118        if it["id"] in user["read"]:119            dropped["已读过滤"] += 1120        elif it["id"] in user["blocked_items"]:121            dropped["用户屏蔽内容"] += 1122        elif it["author"] in user["blocked_authors"]:123            dropped["黑名单作者"] += 1124        elif it["regions"] is not None and user["region"] not in it["regions"]:125            dropped["地域限制"] += 1126        else:127            kept.append(it)128    return kept, dropped129130131def rank_score(it, user, demote_risk):132    """[MOCK] 玩具排序分:CTR + 热度 + 地域/作者亲和 + 噪声;可叠加风险降权。"""133    s = (it["ctr"] * 6.0 + it["hot"] * 1.2134         + (0.35 if it["author"] in user["fav_authors"] else 0.0)135         + user["noise"][it["id"] % len(user["noise"])])136    if demote_risk and it["risk"]:137        s *= RISK_DEMOTE          # 正文:风险内容降权(而非直接过滤)138    return s139140141def post_ranking_filter(ranked, enable, k=EXPOSE_K):142    """正文的「排序后过滤」:广告比例控制 + 同作者内容数量限制。143144    这两条策略必须放在排序之后:它们约束的是「最终曝光列表的构成」,145    在召回阶段还不知道哪些物品会进入最终列表,因此无法提前执行。146    """147    if not enable:148        return ranked[:k], Counter()149    out, dropped = [], Counter()150    ad_cap = int(k * AD_RATIO_CAP)151    n_ad = 0152    per_author = Counter()153    for it in ranked:154        if len(out) >= k:155            break156        if it["is_ad"] and n_ad >= ad_cap:157            dropped["广告占比超限"] += 1158            continue159        if per_author[it["author"]] >= SAME_AUTHOR_CAP:160            dropped["同作者条数超限"] += 1161            continue162        out.append(it)163        n_ad += it["is_ad"]164        per_author[it["author"]] += 1165    return out, dropped166167168def gen_user(rng, items):169    """[MOCK] 用户画像:已读历史 / 屏蔽内容 / 黑名单作者 / 地域 / 偏好作者。"""170    read = {rng.randrange(N_ITEMS) for _ in range(1200)}171    return {172        "read": read,173        "blocked_items": {rng.randrange(N_ITEMS) for _ in range(300)},174        "blocked_authors": {rng.randrange(N_AUTHORS) for _ in range(60)},175        "region": rng.randrange(N_REGIONS),176        "fav_authors": {rng.randrange(N_AUTHORS) for _ in range(20)},177        "noise": [rng.uniform(0, 0.5) for _ in range(97)],178    }179180181def run_config(name, items, backup_pool, rng_seed,182               pre_filter, post_retrieval, post_ranking):183    """跑一组配置,返回链路各级的统计指标。"""184    rng = random.Random(rng_seed)185    index = build_index(items, pre_filter)186    st = {187        "name": name,188        "index_size": len(index),189        "scan": 0, "recall": 0, "waste": 0, "kept": 0,190        "backup_hit": 0, "backup_add": 0,191        "prerank_calls": 0,192        "expose": 0, "bad_status": 0, "read": 0, "blocked": 0,193        "region_bad": 0, "ad": 0, "risk": 0, "max_author": 0,194        "dropped": Counter(),195    }196    for _ in range(N_REQUESTS):197        user = gen_user(rng, items)198        cands, tries = retrieve(index, rng)199        st["scan"] += tries200        st["recall"] += len(cands)201        # 「浪费」= 进入召回打分但本就应该在召回前被剔除的物品202        st["waste"] += sum(1 for it in cands if it["status"] in INVALID_STATUS)203204        cands, dropped = post_retrieval_filter(cands, user, post_retrieval)205        st["dropped"] += dropped206        if len(cands) < MIN_CANDIDATES:207            st["backup_hit"] += 1208            need = MIN_CANDIDATES - len(cands)209            have = {it["id"] for it in cands}210            pool = [it for it in backup_pool if it["id"] not in have]211            # 兜底补入的内容同样要过一遍召回后过滤,否则「兜底」会变成过滤绕过口子212            pool, dropped_bk = post_retrieval_filter(pool, user, post_retrieval)213            st["dropped"] += dropped_bk214            add = pool[:need]215            st["backup_add"] += len(add)216            cands = cands + add217        st["kept"] += len(cands)218219        # 粗排 -> 精排:打分次数就是算力账单220        prerank = cands[:PRERANK_IN]221        st["prerank_calls"] += len(prerank)222        prerank.sort(key=lambda it: rank_score(it, user, False), reverse=True)223        rank_pool = prerank[:RANK_IN]224        ranked = sorted(rank_pool,225                        key=lambda it: rank_score(it, user, post_ranking),226                        reverse=True)227        final, dropped2 = post_ranking_filter(ranked, post_ranking)228        st["dropped"] += dropped2229230        st["expose"] += len(final)231        st["bad_status"] += sum(1 for it in final232                                if it["status"] in INVALID_STATUS)233        st["read"] += sum(1 for it in final if it["id"] in user["read"])234        st["blocked"] += sum(1 for it in final235                             if it["id"] in user["blocked_items"]236                             or it["author"] in user["blocked_authors"])237        st["region_bad"] += sum(1 for it in final238                                if it["regions"] is not None239                                and user["region"] not in it["regions"])240        st["ad"] += sum(1 for it in final if it["is_ad"])241        st["risk"] += sum(1 for it in final if it["risk"])242        if final:243            st["max_author"] = max(st["max_author"],244                                   Counter(it["author"]245                                           for it in final).most_common(1)[0][1])246    return st247248249def main():250    random.seed(42)251    rng = random.Random(42)252    items = gen_library(rng)253    # [MOCK] 兜底池:热门且状态有效的内容(正文:热门内容召回兜底)254    backup_pool = sorted((it for it in items255                          if it["status"] not in INVALID_STATUS),256                         key=lambda it: -it["hot"])[:4000]257258    print("=" * 94)259    print("§4.5.1 过滤模块的位置:召回前过滤 / 召回后过滤 / 排序后过滤 三段式链路对照")260    print("=" * 94)261    cnt = Counter(it["status"] for it in items)262    print(f"\n[MOCK] 内容库 {N_ITEMS:,} 条 / 作者 {N_AUTHORS:,} 个 / 地域 {N_REGIONS} 个;"263          f"模拟 {N_REQUESTS} 次请求,单请求召回 {RECALL_SIZE} -> 曝光 {EXPOSE_K}")264    print(f"{'内容状态':<16}{'条数':>10}{'占比':>9}{'召回前是否剔除':>16}")265    print("-" * 54)266    for name, _ in STATUS_DIST:267        flag = "是" if name in INVALID_STATUS else "否(可分发)"268        print(f"{name:<16}{cnt[name]:>10,}{cnt[name] / N_ITEMS:>9.2%}{flag:>16}")269    invalid = sum(cnt[s] for s in INVALID_STATUS)270    print(f"{'小计(无效内容)':<16}{invalid:>10,}{invalid / N_ITEMS:>9.2%}")271272    configs = [273        ("A 全不过滤", False, False, False),274        ("B 只在召回后过滤", False, True, True),275        ("C 三段式(正文)", True, True, True),276    ]277    stats = [run_config(n, items, backup_pool, 1234, pre, post, rank)278             for n, pre, post, rank in configs]279280    print("\n[链路算力账单] 过滤发生得越早,下游被浪费的计算越少")281    print(f"{'配置':<20}{'索引条目数':>12}{'召回扫描次数':>14}"282          f"{'召回打分浪费率':>16}{'候选利用率':>12}{'兜底触发请求数':>16}")283    print("-" * 92)284    for s in stats:285        waste = s["waste"] / max(s["recall"], 1)286        util = s["kept"] / max(s["recall"], 1)287        print(f"{s['name']:<20}{s['index_size']:>12,}{s['scan']:>14,}"288              f"{waste:>16.2%}{util:>12.2%}{s['backup_hit']:>16}")289    print(f"  (候选利用率 = 兜底补量后的候选数 / 召回候选数;兜底阈值 {MIN_CANDIDATES},"290          f"配置 B/C 共补入 {stats[1]['backup_add']:,}/{stats[2]['backup_add']:,} 个候选,")291    print("   因此两者过滤后虽各少了约 5% 候选,最终仍被补回到阈值线上)")292293    print("\n[曝光质量] 正文 §4.5 开头列出的问题是否真的出现")294    print(f"{'配置':<20}{'曝光条数':>10}{'无效内容率':>12}{'重复推荐率':>12}"295          f"{'屏蔽命中率':>12}{'地域违规率':>12}{'广告占比':>10}{'风险内容率':>12}"296          f"{'单屏同作者上限':>16}")297    print("-" * 120)298    for s in stats:299        e = max(s["expose"], 1)300        print(f"{s['name']:<20}{s['expose']:>10,}{s['bad_status'] / e:>12.2%}"301              f"{s['read'] / e:>12.2%}{s['blocked'] / e:>12.2%}"302              f"{s['region_bad'] / e:>12.2%}{s['ad'] / e:>10.2%}"303              f"{s['risk'] / e:>12.2%}{s['max_author']:>16}")304    print(f"  (广告占比上限 {AD_RATIO_CAP:.0%}、单屏同作者上限 {SAME_AUTHOR_CAP} 条、"305          f"风险内容降权系数 {RISK_DEMOTE})")306307    print("\n[各位置的拦截构成] 三段式配置 C 的过滤明细")308    print(f"{'过滤策略':<18}{'所属位置':<16}{'拦截次数':>12}{'占总拦截':>12}")309    print("-" * 60)310    pos = {311        "已读过滤": "召回后过滤", "用户屏蔽内容": "召回后过滤",312        "黑名单作者": "召回后过滤", "地域限制": "召回后过滤",313        "广告占比超限": "排序后过滤", "同作者条数超限": "排序后过滤",314    }315    total = sum(stats[2]["dropped"].values())316    for key, num in stats[2]["dropped"].most_common():317        print(f"{key:<18}{pos[key]:<16}{num:>12,}{num / max(total, 1):>12.2%}")318    pre_cut = N_ITEMS - stats[2]["index_size"]319    print(f"{'无效状态剔除':<18}{'召回前过滤':<16}{pre_cut:>12,}"320          f"{'(按物品计,不进索引)':>12}")321322    b, c = stats[1], stats[2]323    print("\n结论(正文 §4.5.1 的工程含义):")324    print("  1) 召回前过滤省的是「整条下游链路的算力」:配置 B 的索引里有 "325          f"{b['index_size'] - c['index_size']:,} 条无效内容,")326    print(f"     导致召回打分中有 {b['waste'] / max(b['recall'], 1):.2%} 的无效计算;"327          "配置 C 把这部分成本在建索引时就消掉了;")328    print("  2) 召回后过滤才是「个性化」的过滤:已读 / 屏蔽 / 黑名单 / 地域都依赖用户态,")329    print("     无法在离线索引侧完成,只能在拿到候选后逐个核对;")330    print("  3) 排序后过滤必须留在最后:广告占比与同作者打散约束的是最终列表的构成,")331    print("     在召回阶段根本不知道哪些物品会进 Top-K,所以提前做没有意义;")332    print("  4) 过滤会显著削减候选规模,因此必须配合兜底阈值 "333          f"(本次配置 C 有 {c['backup_hit']}/{N_REQUESTS} 次请求触发兜底),")334    print("     否则「过滤得干净」会变成「候选不够、推荐变空」;")335    print("  5) 对照配置 A 可以看到,缺了过滤模块就会同时出现正文开头列出的")336    print("     重复推荐、无效内容曝光、广告挤占、同作者刷屏等全部问题。")337338339if __name__ == "__main__":340    main()