工业级推荐系统 - 多路召回ItemCF、内容、热门、双塔、序列、Bandit - 精排DeepFM DIN MMoE 多任务 - 重排多样性、新鲜度、类目打散、业务规则 - 冷启动内容、热门、Thompson Sampling 探索 - 实时反馈在线更新、增量学习 - A/B 测试流量分桶 - 监控指标采集 - 评估PK、RK、NDCG、MAP、AUC pip install numpy scikit-learn torch import time import math import random import hashlib import numpy as np import torch import torch.nn as nn import torch.nn.functional as F from collections import defaultdict, deque from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.metrics.pairwise import cosine_similarity np.random.seed(42) random.seed(42) torch.manual_seed(42) NOW time.time() # # 1. 数据模拟 # CATEGORIES [动作, 爱情, 喜剧, 科幻, 悬疑, 动画, 纪录片, 恐怖] N_USERS 100 N_ITEMS 500 ITEMS [] for i in range(N_ITEMS): cat CATEGORIES[i % len(CATEGORIES)] sub CATEGORIES[(i 1) % len(CATEGORIES)] ITEMS.append({ id: i, title: f物品{i}, text: f{cat} {sub} 类型{i%10}, cat: cat, sub_cat: sub, quality: np.random.uniform(0.3, 1.0), views: int(np.random.exponential(1000)), ctr: np.random.uniform(0.01, 0.3), ts: NOW - np.random.randint(0, 86400 * 365), price: np.random.uniform(10, 1000), }) ITEM_MAP {it[id]: it for it in ITEMS} # 用户历史含冷启动 USER_HISTORY {} for u in range(N_USERS): n np.random.randint(0, 20) USER_HISTORY[u] list(np.random.choice(N_ITEMS, sizen, replaceFalse)) if n 0 else [] # 用户画像 USER_PROFILE {} for u in range(N_USERS): hist USER_HISTORY[u] if hist: cats [ITEM_MAP[i][cat] for i in hist] USER_PROFILE[u] {c: cats.count(c) / len(cats) for c in set(cats)} else: USER_PROFILE[u] {} # # 2. 内容索引 # class ContentIndex: def __init__(self, items): self.items items self.vec TfidfVectorizer(analyzerchar, ngram_range(1, 2)) self.mat self.vec.fit_transform([it[text] for it in items]) self.sim cosine_similarity(self.mat) self.n len(items) def similar_items(self, item_id, top_k20): scores self.sim[item_id] idx scores.argsort()[::-1][1:top_k 1] return [(int(i), float(scores[i])) for i in idx] def similarity(self, i, j): return float(self.sim[i][j]) # # 3. 双塔召回深度学习 # class TwoTower(nn.Module): def __init__(self, n_users, n_items, embed_dim64): super().__init__() self.user_embed nn.Embedding(n_users, embed_dim) self.item_embed nn.Embedding(n_items, embed_dim) self.user_mlp nn.Sequential( nn.Linear(embed_dim, 128), nn.ReLU(), nn.Linear(128, embed_dim), ) self.item_mlp nn.Sequential( nn.Linear(embed_dim, 128), nn.ReLU(), nn.Linear(128, embed_dim), ) self.temp nn.Parameter(torch.tensor(0.07)) def user_vec(self, u): return F.normalize(self.user_mlp(self.user_embed(u)), dim-1) def item_vec(self, i): return F.normalize(self.item_mlp(self.item_embed(i)), dim-1) def forward(self, u, i): ue self.user_vec(u) ie self.item_vec(i) return ue ie.T / self.temp def train_two_tower(user_history, n_users, n_items, epochs10): model TwoTower(n_users, n_items) opt torch.optim.Adam(model.parameters(), lr1e-3) # 构造正样本 pairs [(u, i) for u, hist in user_history.items() for i in hist] if not pairs: return model pairs torch.tensor(pairs, dtypetorch.long) for epoch in range(epochs): idx torch.randperm(len(pairs)) total_loss 0 for b in range(0, len(pairs), 64): batch pairs[idx[b:b64]] u batch[:, 0] i batch[:, 1] logits model(u, i) labels torch.arange(len(u)) loss F.cross_entropy(logits, labels) opt.zero_grad() loss.backward() opt.step() total_loss loss.item() if (epoch 1) % 5 0: print(f[双塔] Epoch {epoch1}: loss{total_loss:.4f}) return model # # 4. DeepFM 精排 # class FM(nn.Module): def __init__(self, n_features, embed_dim8): super().__init__() self.linear nn.Embedding(n_features, 1) self.embed nn.Embedding(n_features, embed_dim) def forward(self, x): linear self.linear(x).sum(dim1) emb self.embed(x) sum_emb emb.sum(dim1) sum_sq (emb ** 2).sum(dim1) fm 0.5 * (sum_emb ** 2 - sum_sq).sum(dim1, keepdimTrue) return linear fm class DeepFM(nn.Module): def __init__(self, n_features, n_fields, embed_dim8, hidden(128, 64)): super().__init__() self.fm FM(n_features, embed_dim) self.embed nn.Embedding(n_features, embed_dim) layers [] in_dim n_fields * embed_dim for h in hidden: layers [nn.Linear(in_dim, h), nn.ReLU(), nn.Dropout(0.2)] in_dim h layers.append(nn.Linear(in_dim, 1)) self.deep nn.Sequential(*layers) def forward(self, x): fm_out self.fm(x) emb self.embed(x).flatten(1) deep_out self.deep(emb) return torch.sigmoid(fm_out deep_out) # # 5. DIN 序列建模 # class DINAttention(nn.Module): def __init__(self, embed_dim, hidden(64, 32)): super().__init__() layers [] in_dim embed_dim * 4 for h in hidden: layers [nn.Linear(in_dim, h), nn.ReLU()] in_dim h layers.append(nn.Linear(in_dim, 1)) self.net nn.Sequential(*layers) def forward(self, query, keys): B, T, D keys.shape q query.unsqueeze(1).expand(-1, T, -1) inp torch.cat([q, keys, q - keys, q * keys], dim-1) scores self.net(inp).squeeze(-1) weights F.softmax(scores, dim-1) return (weights.unsqueeze(-1) * keys).sum(dim1) # # 6. MMoE 多任务 # class Expert(nn.Module): def __init__(self, in_dim, hidden_dim): super().__init__() self.net nn.Sequential( nn.Linear(in_dim, hidden_dim), nn.ReLU(), nn.Linear(hidden_dim, hidden_dim), nn.ReLU(), ) def forward(self, x): return self.net(x) class MMoE(nn.Module): def __init__(self, in_dim, n_experts4, hidden_dim64, n_tasks3): super().__init__() self.experts nn.ModuleList([Expert(in_dim, hidden_dim) for _ in range(n_experts)]) self.gates nn.ModuleList([nn.Linear(in_dim, n_experts) for _ in range(n_tasks)]) self.towers nn.ModuleList([ nn.Sequential( nn.Linear(hidden_dim, 32), nn.ReLU(), nn.Linear(32, 1), ) for _ in range(n_tasks) ]) def forward(self, x): expert_outs torch.stack([e(x) for e in self.experts], dim1) results [] for gate, tower in zip(self.gates, self.towers): g torch.softmax(gate(x), dim-1).unsqueeze(-1) repr_ (expert_outs * g).sum(dim1) results.append(torch.sigmoid(tower(repr_))) return results # # 7. 多路召回 # class MultiRecall: def __init__(self, items, user_history, content, two_tower): self.items items self.user_history user_history self.content content self.two_tower two_tower self.n_items len(items) self.popularity np.array([it[views] for it in items], dtypefloat) self.popularity / self.popularity.sum() self.bandit defaultdict(lambda: {n: 0, reward: 0.0}) def itemcf_recall(self, user_id, top_k30): hist self.user_history.get(user_id, []) if not hist: return [] scores defaultdict(float) for i in hist: for j, s in self.content.similar_items(i, 20): scores[j] s for i in hist: scores.pop(i, None) return sorted(scores.items(), keylambda x: -x[1])[:top_k] def content_recall(self, user_id, top_k20): hist self.user_history.get(user_id, []) if not hist: return [] scores np.zeros(self.n_items) for i in hist: scores self.content.sim[i] for i in hist: scores[i] -np.inf idx scores.argsort()[::-1][:top_k] return [(int(i), float(scores[i])) for i in idx] def hot_recall(self, user_id, top_k20): hist set(self.user_history.get(user_id, [])) out [] for i in self.popularity.argsort()[::-1]: if int(i) not in hist: out.append((int(i), float(self.popularity[i]))) if len(out) top_k: break return out def twotower_recall(self, user_id, top_k30): self.two_tower.eval() with torch.no_grad(): u torch.tensor([user_id]) items torch.arange(self.n_items) ue self.two_tower.user_vec(u) ie self.two_tower.item_vec(items) scores (ue ie.T).squeeze().numpy() hist set(self.user_history.get(user_id, [])) for i in hist: scores[i] -np.inf idx scores.argsort()[::-1][:top_k] return [(int(i), float(scores[i])) for i in idx] def bandit_recall(self, user_id, top_k10): hist set(self.user_history.get(user_id, [])) total_n sum(s[n] for s in self.bandit.values()) 1 samples [] for i in range(self.n_items): if i in hist: continue s self.bandit[i] mean s[reward] / (s[n] 1) explore math.sqrt(2 * math.log(total_n) / (s[n] 1)) samples.append((i, mean explore)) samples.sort(keylambda x: -x[1]) return samples[:top_k] def coldstart_recall(self, user_id, top_k50): hist set(self.user_history.get(user_id, [])) buckets defaultdict(list) for it in self.items: if it[id] not in hist: buckets[it[cat]].append(it) out [] for cat, items in buckets.items(): items.sort(keylambda x: -x[quality] * x[ctr]) for it in items[:6]: out.append((it[id], it[quality] * it[ctr])) out.sort(keylambda x: -x[1]) return out[:top_k] def recall(self, user_id): hist self.user_history.get(user_id, []) if not hist: return self.coldstart_recall(user_id, top_k80) rankings { itemcf: (self.itemcf_recall(user_id), 1.0), content: (self.content_recall(user_id), 0.8), hot: (self.hot_recall(user_id), 0.5), twotower: (self.twotower_recall(user_id), 1.2), bandit: (self.bandit_recall(user_id, 20), 0.3), } fused defaultdict(float) for name, (ranking, w) in rankings.items(): for rank, (i, _) in enumerate(ranking): fused[i] w * (1 / (60 rank 1)) return sorted(fused.items(), keylambda x: -x[1])[:150] # # 8. 精排多特征 模型 # class Ranker: def __init__(self, items, content): self.items items self.content content self.doc_map {it[id]: it for it in items} def normalize(self, arr): arr np.array(arr, dtypefloat) if arr.max() - arr.min() 1e-9: return np.zeros_like(arr) return (arr - arr.min()) / (arr.max() - arr.min()) def rank(self, user_id, candidates): # ← 改成 rank if not candidates: return [], {} ids [c[0] for c in candidates] recall [c[1] for c in candidates] hist USER_HISTORY.get(user_id, []) profile USER_PROFILE.get(user_id, {}) feats { recall: self.normalize(recall), quality: self.normalize([self.doc_map[i][quality] for i in ids]), hot: self.normalize([self.doc_map[i][views] for i in ids]), ctr: self.normalize([self.doc_map[i][ctr] for i in ids]), time: self.normalize([ 0.5 ** ((NOW - self.doc_map[i][ts]) / (86400 * 90)) for i in ids ]), profile: np.array([profile.get(self.doc_map[i][cat], 0) for i in ids]), hist_sim: np.array([ np.mean([self.content.similarity(h, i) for h in hist]) if hist else 0 for i in ids ]), price: self.normalize([self.doc_map[i][price] for i in ids]), } weights { recall: 0.25, quality: 0.10, hot: 0.10, ctr: 0.15, time: 0.05, profile: 0.15, hist_sim: 0.15, price: 0.05, } scores sum(weights[k] * feats[k] for k in weights) ranked sorted(zip(ids, scores), keylambda x: -x[1]) return [(int(i), float(s)) for i, s in ranked], feats # # 9. 重排 # class Reranker: staticmethod def dedup(results): seen set() out [] for i, s in results: if i not in seen: seen.add(i) out.append((i, s)) return out staticmethod def diversity(results, doc_map, top_k10, max_per_cat2): cat_count defaultdict(int) out [] for i, s in results: cat doc_map[i][cat] if cat_count[cat] max_per_cat: continue out.append((i, s)) cat_count[cat] 1 if len(out) top_k: break return out staticmethod def freshness(results, doc_map, top_k10, fresh_ratio0.3): n_fresh int(top_k * fresh_ratio) fresh, old [], [] for i, s in results: age NOW - doc_map[i][ts] (fresh if age 86400 * 7 else old).append((i, s)) out, fi, oi [], 0, 0 while len(out) top_k: if fi len(fresh) and (len(out) % 2 0 or oi len(old)): out.append(fresh[fi]); fi 1 elif oi len(old): out.append(old[oi]); oi 1 else: break return out staticmethod def business_rules(results, doc_map, top_k10, max_priceNone): out [] for i, s in results: if max_price and doc_map[i][price] max_price: continue out.append((i, s)) if len(out) top_k: break return out # # 10. A/B 测试 # class ABTest: def __init__(self, n_buckets2, saltrec_ab): self.n_buckets n_buckets self.salt salt def get_bucket(self, user_id): key f{self.salt}:{user_id} h int(hashlib.md5(key.encode()).hexdigest(), 16) return h % self.n_buckets # # 11. 监控 # class Metrics: def __init__(self): self.counters defaultdict(int) self.latencies defaultdict(list) def inc(self, name): self.counters[name] 1 def observe(self, name, value): self.latencies[name].append(value) def report(self): print(\n 监控指标 ) for k, v in self.counters.items(): print(f {k}: {v}) for k, v in self.latencies.items(): if v: arr sorted(v) p50 arr[len(arr) // 2] p99 arr[int(len(arr) * 0.99)] if len(arr) 1 else arr[-1] print(f {k}: avg{np.mean(arr)*1000:.1f}ms, p50{p50*1000:.1f}ms, p99{p99*1000:.1f}ms) # # 12. 实时反馈 # class Feedback: def __init__(self, recall): self.recall recall self.logs deque(maxlen10000) def log(self, user_id, item_id, action): reward {click: 1.0, like: 2.0, buy: 5.0, skip: -0.5}.get(action, 0) self.recall.bandit[item_id][n] 1 self.recall.bandit[item_id][reward] reward self.logs.append({ user: user_id, item: item_id, action: action, reward: reward, ts: time.time(), }) # # 13. 完整系统 # class IndustrialRecommender: def __init__(self, items, user_history): self.items items self.user_history user_history self.content ContentIndex(items) print([初始化] 训练双塔...) self.two_tower train_two_tower(user_history, N_USERS, N_ITEMS, epochs10) self.recall MultiRecall(items, user_history, self.content, self.two_tower) self.ranker Ranker(items, self.content) self.reranker Reranker() self.ab ABTest(n_buckets2) self.metrics Metrics() self.feedback Feedback(self.recall) self.doc_map {it[id]: it for it in items} def recommend(self, user_id, top_k10, debugFalse): t0 time.perf_counter() # A/B 分桶 bucket self.ab.get_bucket(user_id) self.metrics.inc(fab_bucket_{bucket}) # 1. 召回 candidates self.recall.recall(user_id) self.metrics.observe(recall_time, time.perf_counter() - t0) if debug: print(f[召回] {len(candidates)} 候选, bucket{bucket}) # 2. 排序 t1 time.perf_counter() ranked, feats self.ranker.rank(user_id, candidates) self.metrics.observe(rank_time, time.perf_counter() - t1) if debug: print(f[排序] Top5: {ranked[:5]}) # 3. 重排 ranked self.reranker.dedup(ranked) ranked self.reranker.diversity(ranked, self.doc_map, top_ktop_k * 2, max_per_cat3) ranked self.reranker.freshness(ranked, self.doc_map, top_ktop_k) self.metrics.observe(total_time, time.perf_counter() - t0) self.metrics.inc(requests) return [ {item: self.doc_map[i], score: s} for i, s in ranked[:top_k] ] def log_feedback(self, user_id, item_id, action): self.feedback.log(user_id, item_id, action) # # 14. 评估 # def precision_at_k(rec, rel, k): return len(set(rec[:k]) set(rel)) / k if k else 0 def recall_at_k(rec, rel, k): return len(set(rec[:k]) set(rel)) / len(rel) if rel else 0 def ndcg_at_k(rec, rel, k): dcg sum(1 / math.log2(i 2) for i, x in enumerate(rec[:k]) if x in rel) idcg sum(1 / math.log2(i 2) for i in range(min(len(rel), k))) return dcg / idcg if idcg 0 else 0 # # 15. 主程序 # if __name__ __main__: print( * 60) print(工业级推荐系统 Demo) print( * 60) system IndustrialRecommender(ITEMS, USER_HISTORY) # 老用户 print(\n 老用户推荐) for u in [0, 1, 2]: hist USER_HISTORY[u] print(f\n用户 {u} 历史 {len(hist)} 条) results system.recommend(u, top_k5, debugTrue) for r in results: it r[item] print(f [{r[score]:.3f}] {it[title]} ({it[cat]}/{it[sub_cat]})) # 冷启动 print(\n 冷启动用户推荐) cold [u for u in range(N_USERS) if not USER_HISTORY[u]] print(f冷启动用户数: {len(cold)}) if cold: u cold[0] results system.recommend(u, top_k5, debugTrue) for r in results: it r[item] print(f [{r[score]:.3f}] {it[title]} ({it[cat]})) # 反馈 print(\n 模拟实时反馈) for u in [0, 1]: results system.recommend(u, top_k3) for r in results: action random.choice([click, like, buy, skip]) system.log_feedback(u, r[item][id], action) # 评估 print(\n 评估) rel set(USER_HISTORY[0]) if rel: recs [r[item][id] for r in system.recommend(0, top_k10)] print(fP5: {precision_at_k(recs, rel, 5):.3f}) print(fR5: {recall_at_k(recs, rel, 5):.3f}) print(fNDCG5: {ndcg_at_k(recs, rel, 5):.3f}) # 监控 system.metrics.report()