gen_rec_pipeline.py

1"""对应正文第 9 章 §9.2「生成式推荐架构」中生成式推荐模块的内部结构。23正文把生成式推荐模块拆成三个核心组成部分（图「生成式推荐模块内部结构」）：4    语义ID生成(Semantic ID Construction) -> 生成式推荐模型(Generative Recommendation Model)5    -> 物品ID检索(Item Retrieval)6而在整条链路上，正文还强调「大多数工业系统仍然会保留重排(Re-ranking)模块作为最后一道7决策环节」，因为生成式模型难以兼顾广告收益、多样性约束、生态治理等复杂业务目标。89本实现用纯标准库在进程内串起这 5 个环节，并逐级统计候选规模与真实耗时：10    1) 语义ID化：对 item embedding 做残差量化(RQ)，得到 L 层语义ID11           e_i -> (s_i^1, s_i^2, ..., s_i^L)12       正文四层 codebook、每层 512 的组合空间规模为 512^4 ~ 6.87e10。13    2) 序列建模：用户行为序列展开为语义 token 序列14           [s_1^1..s_1^L, s_2^1..s_2^L, ...]15    3) 自回归生成：按 Score(SID) = sum_{i=1}^{L} log P(s_i | s_{<i}, u) 做 beam search。16    4) 物品ID检索：SID -> item，多对一去重、无效 SID 丢弃并从其他路径补充。17    5) 过滤重排：去重/已看过滤 + 业务重排分 = alpha*生成分 + beta*质量分 + gamma*商业分。1819[MOCK] 本文件中的 `MockItemContentModel`（内容理解打 embedding）、`MockSeqGenModel`20（用共现统计代替 Transformer 的 next-token 分布）、`MockItemStore`（倒排索引/物品属性21存储）都是教学用的进程内假组件，不起真实服务、不读取任何真实数据。22"""2324import math25import random26import time27from collections import Counter, defaultdict2829L_LAYERS = 3          # 语义ID层数30CODEBOOK_SIZE = 8     # 每层码本大小 K31EMB_DIM = 832N_ITEMS = 24033N_CATEGORIES = 6343536# ----------------------------------------------------------------------------37# [MOCK] 内容理解模型：真实系统里是多模态大模型，这里只用类别中心 + 噪声合成 embedding38# ----------------------------------------------------------------------------39class MockItemContentModel:40    def __init__(self, n_items, n_cat, dim):41        self.centers = [[random.gauss(0, 1) for _ in range(dim)] for _ in range(n_cat)]42        self.items = {}43        for i in range(n_items):44            cat = i % n_cat45            c = self.centers[cat]46            self.items[i] = {47                "cat": cat,48                "emb": [c[d] + random.gauss(0, 0.45) for d in range(dim)],49                "quality": round(random.uniform(0.2, 1.0), 3),50                "biz": round(random.uniform(0.0, 1.0), 3),   # 商业化价值（广告/电商）51            }5253    def embedding(self, item_id):54        return self.items[item_id]["emb"]555657# ----------------------------------------------------------------------------58# 1) 语义ID生成：残差量化（这里用逐层 KMeans，对应 §9.3.3 RQ-KMeans）59# ----------------------------------------------------------------------------60def _l2(a, b):61    return sum((x - y) ** 2 for x, y in zip(a, b))626364def _kmeans(vectors, k, iters=12):65    centers = [list(v) for v in random.sample(vectors, k)]66    assign = [0] * len(vectors)67    for _ in range(iters):68        changed = False69        for i, v in enumerate(vectors):70            best = min(range(k), key=lambda j: _l2(v, centers[j]))71            if best != assign[i]:72                assign[i], changed = best, True73        buckets = defaultdict(list)74        for i, a in enumerate(assign):75            buckets[a].append(vectors[i])76        for j in range(k):77            if buckets[j]:78                centers[j] = [sum(col) / len(buckets[j]) for col in zip(*buckets[j])]79        if not changed:80            break81    return centers, assign828384def build_semantic_ids(content_model, n_layers=L_LAYERS, k=CODEBOOK_SIZE):85    ids = sorted(content_model.items)86    residual = [list(content_model.embedding(i)) for i in ids]87    codes = [[] for _ in ids]88    codebooks = []89    for _ in range(n_layers):90        centers, assign = _kmeans(residual, k)91        codebooks.append(centers)92        for idx, a in enumerate(assign):93            codes[idx].append(a)94            residual[idx] = [x - c for x, c in zip(residual[idx], centers[a])]95    return {ids[i]: tuple(codes[i]) for i in range(len(ids))}, codebooks969798# ----------------------------------------------------------------------------99# [MOCK] 生成式推荐模型：真实系统是 Transformer Decoder-only，这里用100#        (上文 token, 下一 token) 的平滑共现频率近似 P(s_t | s_<t, u)101# ----------------------------------------------------------------------------102class MockSeqGenModel:103    def __init__(self, vocab_per_layer, n_layers):104        self.K = vocab_per_layer105        self.L = n_layers106        self.counts = [defaultdict(Counter) for _ in range(n_layers)]107108    def fit(self, token_sequences):109        """token_sequences: 每条是展开后的语义 token 序列 [(layer, code), ...]"""110        for seq in token_sequences:111            for t in range(1, len(seq)):112                layer, code = seq[t]113                ctx = seq[t - 1]                     # 1 阶上下文，教学简化114                self.counts[layer][ctx][code] += 1115116    def log_probs(self, ctx, layer):117        """返回第 layer 层码本上的 log P(code | ctx)，Laplace 平滑。"""118        cnt = self.counts[layer][ctx]119        total = sum(cnt.values()) + self.K120        return [math.log((cnt[c] + 1) / total) for c in range(self.K)]121122123# ----------------------------------------------------------------------------124# [MOCK] 物品存储 / 倒排索引125# ----------------------------------------------------------------------------126class MockItemStore:127    def __init__(self, sid_map, content_model):128        self.sid_to_items = defaultdict(list)129        for item, sid in sid_map.items():130            self.sid_to_items[sid].append(item)131        self.meta = content_model.items132133    def lookup(self, sid):134        return self.sid_to_items.get(sid, [])135136137# ----------------------------------------------------------------------------138# 3) beam search：Score(SID) = sum_i log P(s_i | s_<i, u)139# ----------------------------------------------------------------------------140def beam_search(model, start_ctx, n_layers, beam=12):141    beams = [([], 0.0, start_ctx)]142    for layer in range(n_layers):143        nxt = []144        for codes, score, ctx in beams:145            lps = model.log_probs(ctx, layer)146            for c in range(model.K):147                nxt.append((codes + [c], score + lps[c], (layer, c)))148        nxt.sort(key=lambda x: -x[1])149        beams = nxt[:beam]150    return [(tuple(c), s) for c, s, _ in beams]151152153def main():154    random.seed(42)155    t_all = time.perf_counter()156    stages = []157158    # ---- 阶段 0：内容理解（离线） ----159    t0 = time.perf_counter()160    content = MockItemContentModel(N_ITEMS, N_CATEGORIES, EMB_DIM)161    t_content = (time.perf_counter() - t0) * 1e3162163    # ---- 阶段 1：语义ID生成（离线） ----164    t0 = time.perf_counter()165    sid_map, codebooks = build_semantic_ids(content)166    t_sid = (time.perf_counter() - t0) * 1e3167    uniq_sid = len(set(sid_map.values()))168    stages.append(("1 语义ID生成(离线)", N_ITEMS, uniq_sid, t_sid))169170    # ---- 构造用户行为序列（同类别内游走，制造可学习结构） ----171    sequences, user_hist = [], {}172    for u in range(300):173        cat = random.randrange(N_CATEGORIES)174        pool = [i for i in sid_map if content.items[i]["cat"] == cat]175        hist = random.sample(pool, 6)176        user_hist[u] = hist177        toks = []178        for it in hist:179            for layer, code in enumerate(sid_map[it]):180                toks.append((layer, code))181        sequences.append(toks)182183    # ---- 阶段 2：序列建模（训练） ----184    t0 = time.perf_counter()185    model = MockSeqGenModel(CODEBOOK_SIZE, L_LAYERS)186    model.fit(sequences)187    t_fit = (time.perf_counter() - t0) * 1e3188189    # ---- 在线链路：以 user 0 为例 ----190    uid = 0191    hist = user_hist[uid]192    ctx_tokens = []193    for it in hist[:-1]:194        for layer, code in enumerate(sid_map[it]):195            ctx_tokens.append((layer, code))196    stages.append(("2 序列建模(token化)", len(hist) - 1, len(ctx_tokens), t_fit))197198    # ---- 阶段 3：自回归生成 ----199    t0 = time.perf_counter()200    beams = beam_search(model, ctx_tokens[-1], L_LAYERS, beam=24)201    t_gen = (time.perf_counter() - t0) * 1e3202    stages.append(("3 自回归生成(beam=24)", CODEBOOK_SIZE ** L_LAYERS, len(beams), t_gen))203204    # ---- 阶段 4：物品ID检索（多对一去重 + 无效SID丢弃） ----205    t0 = time.perf_counter()206    store = MockItemStore(sid_map, content)207    cand, invalid, seen = [], 0, set()208    for sid, score in beams:209        items = store.lookup(sid)210        if not items:211            invalid += 1212            continue213        for it in items:214            if it in seen:        # 多个语义ID映射到同一物品 -> 仅保留得分最高者215                continue216            seen.add(it)217            cand.append((it, score))218    t_ret = (time.perf_counter() - t0) * 1e3219    stages.append(("4 物品ID检索", len(beams), len(cand), t_ret))220221    # ---- 阶段 5：过滤 + 重排 ----222    t0 = time.perf_counter()223    watched = set(hist)224    filtered = [(i, s) for i, s in cand if i not in watched]225    alpha, beta, gamma = 1.0, 2.0, 0.5226    reranked = sorted(227        (228            (229                i,230                alpha * s + beta * content.items[i]["quality"] + gamma * content.items[i]["biz"],231                s,232            )233            for i, s in filtered234        ),235        key=lambda x: -x[1],236    )[:10]237    t_rr = (time.perf_counter() - t0) * 1e3238    stages.append(("5 过滤+重排", len(cand), len(reranked), t_rr))239240    print("=" * 78)241    print("§9.2 生成式推荐模块链路仿真（语义ID化 -> 序列建模 -> 生成 -> 检索 -> 重排）")242    print("=" * 78)243    print(f"内容理解([MOCK] MLLM) 耗时 {t_content:.2f} ms；物品数 {N_ITEMS}，类别 {N_CATEGORIES}")244    print(f"语义ID配置：L={L_LAYERS} 层，每层 K={CODEBOOK_SIZE}，"245          f"理论组合空间 {CODEBOOK_SIZE}^{L_LAYERS}={CODEBOOK_SIZE ** L_LAYERS}")246    print(f"正文四层 512 码本的组合空间：512^4 = {512 ** 4:.4g}")247    print()248    print(f"{'阶段':<24}{'输入规模':>10}{'输出规模':>10}{'耗时(ms)':>12}")249    print("-" * 78)250    for name, nin, nout, ms in stages:251        print(f"{name:<24}{nin:>10}{nout:>10}{ms:>12.3f}")252    print("-" * 78)253    print(f"{'合计':<24}{'':>10}{'':>10}{sum(s[3] for s in stages):>12.3f}")254255    print()256    print(f"beam 候选中无效语义ID(Invalid SID) {invalid}/{len(beams)} "257          f"= {invalid / len(beams):.1%}（被直接丢弃，由其他路径补充）")258    collide = Counter(len(v) for v in store.sid_to_items.values())259    print("语义ID冲突分布（一个SID命中的物品数 -> SID个数）：",260          dict(sorted(collide.items())))261262    print()263    print("Top-10 重排结果：")264    print(f"{'rank':>5}{'item':>7}{'cat':>5}{'生成分':>10}{'质量':>8}{'商业':>8}{'重排分':>10}")265    for r, (it, final, gen) in enumerate(reranked, 1):266        m = content.items[it]267        print(f"{r:>5}{it:>7}{m['cat']:>5}{gen:>10.3f}{m['quality']:>8.3f}"268              f"{m['biz']:>8.3f}{final:>10.3f}")269270    cats = Counter(content.items[i]["cat"] for i, _, _ in reranked)271    print()272    print(f"Top-10 类别分布 {dict(sorted(cats.items()))}；用户历史类别 "273          f"{content.items[hist[0]]['cat']}")274    print(f"链路总墙钟耗时 {(time.perf_counter() - t_all) * 1e3:.1f} ms")275276277if __name__ == "__main__":278    main()