ranking_pipeline.py

1"""2精排模块端到端链路（第 6 章 §6.1 精排模块架构）34对应正文概念：5正文把精排模块内部拆成三个串行阶段（图 fig_fr_module）：6    (1) 精排主模型 + LTR 模型 → 输出各类 pxtr；7    (2) 多目标融合（Multi-objective Fusion）→ 统一排序分；8    (3) 流量调控与业务约束（Traffic Control）→ 最终 Top-K。9输入是粗排给出的 100~500 个候选，输出 50~100 个物品送往重排。1011本实现做了什么：12用纯标准库把上述四步（打分 → 融合 → 调控 → 输出）串成一条可运行链路，13并在每一步打印候选集规模、分数分布与业务指标的变化，让读者看清14「模型分」如何一步步被业务约束改写成「最终曝光顺序」。1516关键公式：17    多目标融合（正文 §6.4.1 乘法公式）：18        Score_i = prod_k ( beta_k + alpha_k * p_{i,k} ) ^ gamma_k19    流量调控（正文 §6.5.1 级联式伪代码）：20        Score_i <- Score_i * prod_j boost_j(item_i)21    保量约束（正文 §6.5.5）：22        Exposure_target >= Q_target2324[MOCK] 组件说明：25    - MockRankingModelService：假的精排主模型 RPC 服务，用确定性哈希+特征线性组合26      伪造 pctr / pevtr / pltr / pcmtr / pcvr / pwatch_time，不训练、不联网。27    - MockLTRService：假的 LTR 模型服务，输出一个业务 Reward 排序分。28    - MockFeatureStore：假的特征存储，按 item_id 返回物品画像。29    - MockExposureLog：假的曝光流水日志，用于统计调控效果。30"""3132import math33import random3435# --------------------------------------------------------------------------36# [MOCK] 特征存储37# --------------------------------------------------------------------------383940class MockFeatureStore:41    """[MOCK] 假的物品特征存储。线上应为 KV 存储（Redis / 自研 Feature Store）。"""4243    CATEGORIES = ["搞笑", "娱乐", "知识", "生活", "体育"]4445    def __init__(self, num_items: int, rng: random.Random):46        self.table: dict[int, dict] = {}47        for item_id in range(num_items):48            cat = self.CATEGORIES[item_id % len(self.CATEGORIES)]49            # 搞笑/娱乐天然高点击，知识/体育偏长尾 —— 复现正文说的生态失衡成因50            hot = cat in ("搞笑", "娱乐")51            self.table[item_id] = {52                "item_id": item_id,53                "category": cat,54                "duration": rng.choice([15.0, 30.0, 60.0, 120.0]),55                "age_hours": rng.choice([0.5, 2.0, 8.0, 48.0, 240.0]),56                "author_fans": int(10 ** rng.uniform(1.5, 6.0)),57                "in_campaign": rng.random() < 0.12,58                "quality": rng.betavariate(5, 2) if not hot else rng.betavariate(3, 3),59                "hot_prior": 0.55 if hot else 0.25,60            }6162    def get(self, item_id: int) -> dict:63        return self.table[item_id]6465    def is_new_video(self, item_id: int) -> bool:66        return self.table[item_id]["age_hours"] < 6.0676869# --------------------------------------------------------------------------70# [MOCK] 精排主模型与 LTR 模型服务71# --------------------------------------------------------------------------727374def _sigmoid(x: float) -> float:75    return 1.0 / (1.0 + math.exp(-max(-30.0, min(30.0, x))))767778class MockRankingModelService:79    """[MOCK] 假的精排主模型推理服务。8081    正文 §6.2：精排主模型把用户/物品/上下文特征统一拼接后送入 DNN，82    用多任务学习（MTL）框架联合预估多个行为目标。这里不训练网络，83    直接用「用户兴趣向量 · 物品属性向量」的确定性线性打分来伪造 pxtr，84    保证同一 (uid, item_id) 多次调用结果一致（模拟线上幂等推理）。85    """8687    # 正文 §6.3 列出的典型稀疏度量级：CTR 3%~10%，LTR <1%，CMTR <1‰，FTR <1‱88    TASK_BASE = {89        "pctr": 0.06,90        "pevtr": 0.35,91        "pltr": 0.008,92        "pcmtr": 0.0008,93        "pcvr": 0.004,94    }9596    def __init__(self, store: MockFeatureStore):97        self.store = store98        self.call_count = 099100    def _affinity(self, user: dict, feat: dict) -> float:101        """用户兴趣与物品类目的匹配度，取值约 [-1, 1]。"""102        pref = user["cat_pref"].get(feat["category"], 0.0)103        return 2.0 * pref - 1.0 + 0.6 * (feat["quality"] - 0.5)104105    def predict(self, user: dict, item_id: int) -> dict:106        """返回该候选物品的全部 pxtr 预估值。"""107        self.call_count += 1108        feat = self.store.get(item_id)109        a = self._affinity(user, feat)110        out = {}111        for task, base in self.TASK_BASE.items():112            logit = math.log(base / (1.0 - base)) + 2.2 * a + 1.4 * feat["hot_prior"]113            out[task] = _sigmoid(logit)114        # 时长预估是回归目标（正文 §6.3.1），单位秒，受视频时长上限约束115        out["pwatch_time"] = feat["duration"] * _sigmoid(1.8 * a + 0.3)116        return out117118119class MockLTRService:120    """[MOCK] 假的 LTR 模型服务。121122    正文 §6.2：LTR 模型不直接拟合单个行为标签，而是基于业务经验构造排序123    Reward（综合满意度 / 长周期价值），直接优化排序效果。这里用一个124    人工 Reward 公式充当「已训练好的 LTR 模型输出」。125    """126127    def __init__(self, store: MockFeatureStore):128        self.store = store129130    def predict(self, user: dict, item_id: int, pxtr: dict) -> float:131        feat = self.store.get(item_id)132        # 归一化时长 + 互动加权 + 内容质量，映射到 (0,1)133        norm_wt = pxtr["pwatch_time"] / max(feat["duration"], 1e-6)134        raw = (135            0.5 * norm_wt136            + 20.0 * pxtr["pltr"]137            + 120.0 * pxtr["pcmtr"]138            + 0.4 * feat["quality"]139        )140        return _sigmoid(2.5 * (raw - 0.6))141142143class MockExposureLog:144    """[MOCK] 假的曝光流水日志，仅在内存里累加统计量。"""145146    def __init__(self):147        self.records: list[dict] = []148149    def write(self, stage: str, items: list[int], store: MockFeatureStore) -> None:150        for pos, item_id in enumerate(items):151            self.records.append({"stage": stage, "pos": pos, "item_id": item_id})152153    def category_share(self, stage: str, store: MockFeatureStore) -> dict[str, float]:154        rows = [r for r in self.records if r["stage"] == stage]155        total = len(rows) or 1156        share: dict[str, float] = {}157        for r in rows:158            cat = store.get(r["item_id"])["category"]159            share[cat] = share.get(cat, 0.0) + 1.0 / total160        return share161162163# --------------------------------------------------------------------------164# Stage 2：多目标融合（乘法公式，正文 §6.4.1）165# --------------------------------------------------------------------------166167168class MultiplicativeFusion:169    """Score_i = prod_k (beta_k + alpha_k * p_{i,k}) ^ gamma_k"""170171    def __init__(self, params: dict[str, tuple[float, float, float]]):172        self.params = params  # factor -> (alpha, beta, gamma)173174    def fuse(self, factors: dict[str, float]) -> float:175        score = 1.0176        for name, (alpha, beta, gamma) in self.params.items():177            base = beta + alpha * factors[name]178            score *= max(base, 1e-9) ** gamma179        return score180181182# --------------------------------------------------------------------------183# Stage 3：流量调控（级联式，正文 §6.5.1）184# --------------------------------------------------------------------------185186187class CascadeTrafficController:188    """正文 §6.5.1 的级联式调控：融合 → 初排 → boost → 再排 → Top-K。"""189190    def __init__(self, store: MockFeatureStore, quota: dict[str, int],191                 cat_floor: dict[str, int] | None = None):192        self.store = store193        self.quota = quota  # 保量目标，正文 §6.5.5 Exposure_target >= Q_target194        self.cat_floor = cat_floor or {}  # 垂类保底坑位，正文 §6.5.3195196    def boost(self, item_id: int) -> tuple[float, list[str]]:197        feat = self.store.get(item_id)198        mul, hits = 1.0, []199        if self.store.is_new_video(item_id):      # 新内容冷启动扶持200            mul *= 1.10201            hits.append("new_video×1.10")202        if feat["author_fans"] < 1000:            # 新作者/低粉创作者保护203            mul *= 1.05204            hits.append("new_author×1.05")205        if feat["in_campaign"]:                   # 电商活动流量倾斜206            mul *= 1.15207            hits.append("campaign×1.15")208        return mul, hits209210    def force_insert(self, ranked: list[tuple[int, float]], top_k: int):211        """保量强插（正文 §6.5.5）：配额未达成则从池内未入选物品按分数补位。212213        同时实现正文 §6.5.3 的「各垂类内容分发均衡管控」：为每个类目设置214        最低曝光坑位 cat_floor，避免搞笑/娱乐等高点击类目垄断全部曝光。215        """216        selected = [i for i, _ in ranked[:top_k]]217        pool_of = {218            "new_video": lambda i: self.store.is_new_video(i),219            "campaign": lambda i: self.store.get(i)["in_campaign"],220        }221        inserted: list[tuple[str, int]] = []222223        def evict_one(protected) -> int | None:224            """从队尾找一个可被替换的坑位（不属于任何保量池、且其类目仍超额）。"""225            cat_cnt: dict[str, int] = {}226            for i in selected:227                c = self.store.get(i)["category"]228                cat_cnt[c] = cat_cnt.get(c, 0) + 1229            for pos in range(len(selected) - 1, -1, -1):230                cur = selected[pos]231                if any(p(cur) for p in pool_of.values()):232                    continue233                cur_cat = self.store.get(cur)["category"]234                if cat_cnt.get(cur_cat, 0) <= self.cat_floor.get(cur_cat, 0):235                    continue236                if cur in protected:237                    continue238                return pos239            return None240241        # ---- (a) 类目保底 ----242        for cat, floor in self.cat_floor.items():243            have = sum(1 for i in selected if self.store.get(i)["category"] == cat)244            spare = [i for i, _ in ranked[top_k:]245                     if self.store.get(i)["category"] == cat]246            for item_id in spare[: max(0, floor - have)]:247                pos = evict_one(protected=set())248                if pos is None:249                    break250                selected[pos] = item_id251                inserted.append((f"cat:{cat}", item_id))252253        # ---- (b) 业务池保量 ----254        floor_ids = {i for _, i in inserted}255        for pool, pred in pool_of.items():256            need = self.quota.get(pool, 0) - sum(1 for i in selected if pred(i))257            if need <= 0:258                continue259            spare = [i for i, _ in ranked[top_k:] if pred(i)]260            for item_id in spare[:need]:261                pos = evict_one(protected=floor_ids)262                if pos is None:263                    break264                selected[pos] = item_id265                inserted.append((pool, item_id))266        return selected, inserted267268269# --------------------------------------------------------------------------270# 完整链路271# --------------------------------------------------------------------------272273274def gini(values: list[float]) -> float:275    """基尼系数，用于衡量流量集中度（0 完全均匀，1 完全垄断）。"""276    xs = sorted(values)277    n = len(xs)278    if n == 0 or sum(xs) == 0:279        return 0.0280    cum = sum((2 * (i + 1) - n - 1) * x for i, x in enumerate(xs))281    return cum / (n * sum(xs))282283284class RankingModule:285    """精排模块：打分 → 多目标融合 → 流量调控 → 输出 Top-K。"""286287    def __init__(self, store, model, ltr, fusion, controller, log):288        self.store, self.model, self.ltr = store, model, ltr289        self.fusion, self.controller, self.log = fusion, controller, log290291    def rank(self, user: dict, candidates: list[int], top_k: int, verbose=True):292        # ---- Step 1 主模型 + LTR 模型打分 ----293        scored = []294        for item_id in candidates:295            pxtr = self.model.predict(user, item_id)296            pxtr["pltr_reward"] = self.ltr.predict(user, item_id, pxtr)297            scored.append((item_id, pxtr))298        if verbose:299            print(f"[Step1 打分] 候选 {len(candidates)} 个，"300                  f"主模型 RPC {self.model.call_count} 次")301            keys = ["pctr", "pevtr", "pltr", "pcmtr", "pcvr", "pwatch_time",302                    "pltr_reward"]303            print(f"  {'因子':<12}{'min':>10}{'mean':>10}{'max':>10}")304            for k in keys:305                vs = [p[k] for _, p in scored]306                print(f"  {k:<12}{min(vs):>10.5f}{sum(vs)/len(vs):>10.5f}"307                      f"{max(vs):>10.5f}")308309        # ---- Step 2 多目标融合 ----310        fused = []311        for item_id, pxtr in scored:312            factors = {313                "pctr": pxtr["pctr"],314                "pevtr": pxtr["pevtr"],315                "pltr": pxtr["pltr"],316                "pwatch_time_norm": pxtr["pwatch_time"]317                / self.store.get(item_id)["duration"],318                "ltr_reward": pxtr["pltr_reward"],319            }320            fused.append((item_id, self.fusion.fuse(factors)))321        fused.sort(key=lambda t: -t[1])322        if verbose:323            print(f"\n[Step2 多目标融合] Top-5 融合分："324                  + ", ".join(f"item{i}:{s:.4f}" for i, s in fused[:5]))325326        base_topk = [i for i, _ in fused[:top_k]]327        self.log.write("before_control", base_topk, self.store)328329        # ---- Step 3 流量调控（boost + 再排） ----330        boosted = []331        boost_hits = 0332        for item_id, s in fused:333            mul, hits = self.controller.boost(item_id)334            boost_hits += 1 if hits else 0335            boosted.append((item_id, s * mul))336        boosted.sort(key=lambda t: -t[1])337        if verbose:338            print(f"[Step3 流量调控] {boost_hits}/{len(fused)} 个候选命中 boost 规则")339340        # ---- Step 4 保量强插 + 输出 Top-K ----341        final, inserted = self.controller.force_insert(boosted, top_k)342        self.log.write("after_control", final, self.store)343        if verbose:344            if inserted:345                print(f"[Step4 保量强插] 强插 {len(inserted)} 个："346                      + ", ".join(f"{p}->item{i}" for p, i in inserted))347            else:348                print("[Step4 保量强插] 配额已自然满足，无需强插")349        return final350351352if __name__ == "__main__":353    random.seed(42)354    rng = random.Random(42)355356    NUM_ITEMS, NUM_CANDIDATES, TOP_K = 400, 300, 60357358    store = MockFeatureStore(NUM_ITEMS, rng)359    model = MockRankingModelService(store)360    ltr = MockLTRService(store)361    log = MockExposureLog()362363    # 乘法公式超参：(alpha, beta, gamma)，gamma 控制头部/中腰部流量形状364    fusion = MultiplicativeFusion({365        "pctr": (1.0, 0.05, 1.0),366        "pevtr": (1.0, 0.20, 0.8),367        "pltr": (10.0, 0.50, 0.6),368        "pwatch_time_norm": (1.0, 0.10, 1.2),369        "ltr_reward": (1.0, 0.30, 1.0),370    })371    controller = CascadeTrafficController(372        store,373        quota={"new_video": 8, "campaign": 6},374        cat_floor={"知识": 6, "生活": 5, "体育": 4},375    )376    user = {377        "uid": 1001,378        "cat_pref": {"搞笑": 0.70, "娱乐": 0.55, "知识": 0.45,379                     "生活": 0.35, "体育": 0.20},380    }381    candidates = rng.sample(range(NUM_ITEMS), NUM_CANDIDATES)382383    print("=" * 74)384    print("精排模块链路演示：粗排候选 300 → 精排输出 Top-60")385    print("=" * 74)386    module = RankingModule(store, model, ltr, fusion, controller, log)387    final = module.rank(user, candidates, TOP_K)388389    print("\n" + "=" * 74)390    print("调控前后对比")391    print("=" * 74)392    before = log.category_share("before_control", store)393    after = log.category_share("after_control", store)394    print(f"{'类目':<8}{'调控前占比':>12}{'调控后占比':>12}{'变化':>10}")395    for cat in MockFeatureStore.CATEGORIES:396        b, a = before.get(cat, 0.0), after.get(cat, 0.0)397        print(f"{cat:<8}{b:>12.3f}{a:>12.3f}{a - b:>+10.3f}")398399    before_ids = [r["item_id"] for r in log.records if r["stage"] == "before_control"]400    after_ids = [r["item_id"] for r in log.records if r["stage"] == "after_control"]401    for tag, ids, share in (("调控前", before_ids, before),402                            ("调控后", after_ids, after)):403        n_new = sum(1 for i in ids if store.is_new_video(i))404        n_camp = sum(1 for i in ids if store.get(i)["in_campaign"])405        n_small = sum(1 for i in ids if store.get(i)["author_fans"] < 1000)406        g = gini([share.get(c, 0.0) for c in MockFeatureStore.CATEGORIES])407        print(f"\n{tag}：新内容 {n_new}/{TOP_K}（配额 8），"408              f"活动商品 {n_camp}/{TOP_K}（配额 6），低粉作者 {n_small}/{TOP_K}")409        print(f"       覆盖类目数 = "410              f"{sum(1 for c in MockFeatureStore.CATEGORIES if share.get(c, 0) > 0)}/5"411              f"，类目分发基尼系数 = {g:.4f}")412413    overlap = len(set(before_ids) & set(after_ids))414    print(f"\nTop-{TOP_K} 集合重合度：{overlap}/{TOP_K} = {overlap / TOP_K:.1%}"415          f"（调控换掉 {TOP_K - overlap} 个坑位换来生态约束达成，"416          f"其余坑位仍由模型收益决定）")417    print(f"全链路主模型调用次数：{model.call_count}")