node2vec.py

1"""2Node2Vec 图嵌入召回（第10章 §10.2.1 「Node2Vec召回」）34核心思想：5    在 DeepWalk 均匀随机游走的基础上引入「带偏置的二阶随机游走」（Biased Random Walk）。6    游走时不仅看当前节点 v，还要看上一跳节点 t，对候选下一跳 x 赋非归一化权重：7        pi_vx = alpha_pq(t, x) * w_vx,   P(x | v, t) = pi_vx / sum_{z in N(v)} pi_vz8        alpha_pq(t, x) = 1/p   若 d(t, x) = 0（x 就是上一跳 t，即「返回」）9                       = 1     若 d(t, x) = 1（x 与 t 直接相邻，即「同一局部邻域」）10                       = 1/q   若 d(t, x) = 2（x 离 t 更远，即「向外探索」）11    p 为返回参数（Return Parameter），越小越容易回头、越强化局部结构；q 为进出参数12    （In-Out Parameter），q > 1 偏 BFS（局部同质邻域），q < 1 偏 DFS（高阶/跨社区结构）。13    游走序列再喂给 Skip-Gram + 负采样训练 Embedding，线上用 ANN 做 Top-K 近邻召回。14论文出处：Grover A., Leskovec J. "node2vec: Scalable Feature Learning for Networks". KDD 2016.15仅使用 Python 标准库（math / random / collections / typing），便于原理演示。16"""1718import math19import random20from collections import defaultdict21from typing import Dict, List, Optional, Tuple2223Graph = Dict[str, Dict[str, float]]242526# ── 1. 构图：用户行为序列 → 带权物品图 ──────────────────────────────────────────27def build_item_graph(sessions: List[List[str]]) -> Graph:28    """由用户 Session 行为序列构建无向带权物品图：边权为共现次数 w_ij = Count(i -> j)。"""29    graph: Graph = defaultdict(dict)30    for seq in sessions:31        for a, b in zip(seq, seq[1:]):32            if a == b:33                continue34            graph[a][b] = graph[a].get(b, 0.0) + 1.035            graph[b][a] = graph[b].get(a, 0.0) + 1.036    return dict(graph)373839# ── 2. 别名采样表（Alias Table）：O(N) 预处理，O(1) 采样 ─────────────────────────40class AliasTable:41    """别名采样表，把任意离散分布转成两张查找表，单次采样复杂度 O(1)。"""42    def __init__(self, weights: List[float]) -> None:43        n = len(weights)44        total = sum(weights) or 1.045        scaled = [w * n / total for w in weights]  # 归一化到均值为 146        self.prob: List[float] = [0.0] * n47        self.alias: List[int] = [0] * n48        small, large = [], []49        for i, s in enumerate(scaled):50            (small if s < 1.0 else large).append(i)51        while small and large:52            si, li = small.pop(), large.pop()53            self.prob[si] = scaled[si]54            self.alias[si] = li55            scaled[li] -= 1.0 - scaled[si]56            (small if scaled[li] < 1.0 else large).append(li)57        for i in large + small:  # 残留项概率视为 158            self.prob[i], self.alias[i] = 1.0, i5960    def sample(self) -> int:61        i = random.randrange(len(self.prob))62        return i if random.random() < self.prob[i] else self.alias[i]636465# ── 3. Node2Vec 模型 ─────────────────────────────────────────────────────────66class Node2Vec:67    """Node2Vec：带偏置二阶随机游走 + Skip-Gram 负采样。"""68    def __init__(self, graph: Graph, dim: int = 16, p: float = 1.0,69                 q: float = 1.0, lr: float = 0.05) -> None:70        self.graph = graph71        self.p, self.q, self.dim, self.lr = p, q, dim, lr72        self.nodes: List[str] = sorted(graph.keys())73        self.node2id = {n: i for i, n in enumerate(self.nodes)}74        # 邻居列表与边权列表固定顺序，便于与 alias 下标对应75        self.nbrs: Dict[str, List[str]] = {v: sorted(graph[v]) for v in self.nodes}76        self._node_alias: Dict[str, AliasTable] = {77            v: AliasTable([graph[v][x] for x in self.nbrs[v]])78            for v in self.nodes if self.nbrs[v]79        }80        self._edge_alias: Dict[Tuple[str, str], AliasTable] = {}  # (t, v) → alias 表81        scale = 1.0 / math.sqrt(dim)82        n = len(self.nodes)83        self.emb_in = [[random.uniform(-scale, scale) for _ in range(dim)] for _ in range(n)]84        self.emb_out = [[0.0] * dim for _ in range(n)]8586    # ---- 3.1 带偏置的二阶游走 ----87    def _build_edge_alias(self, prev: str, cur: str) -> AliasTable:88        """按 pi_vx = alpha_pq(t,x) * w_vx 计算非归一化权重，并构建 alias 表（带缓存）。"""89        key = (prev, cur)90        if key in self._edge_alias:91            return self._edge_alias[key]92        prev_nbrs, weights = self.graph[prev], []93        for x in self.nbrs[cur]:94            w = self.graph[cur][x]95            if x == prev:            # d(t, x) = 0 → 返回上一跳96                weights.append(w / self.p)97            elif x in prev_nbrs:     # d(t, x) = 1 → 停留在 t 的邻域（BFS 式）98                weights.append(w)99            else:                    # d(t, x) = 2 → 走向更远的节点（DFS 式）100                weights.append(w / self.q)101        self._edge_alias[key] = AliasTable(weights)102        return self._edge_alias[key]103104    def walk(self, start: str, walk_len: int) -> List[str]:105        """从 start 出发做一次带偏置的二阶随机游走。"""106        path = [start]107        while len(path) < walk_len:108            cur = path[-1]109            if not self.nbrs.get(cur):110                break111            if len(path) == 1:  # 第一步没有上一跳，退化为按边权的一阶采样112                idx = self._node_alias[cur].sample()113            else:114                idx = self._build_edge_alias(path[-2], cur).sample()115            path.append(self.nbrs[cur][idx])116        return path117118    def generate_walks(self, num_walks: int = 5, walk_len: int = 10) -> List[List[str]]:119        """每个节点出发 num_walks 次游走，得到训练语料。"""120        walks, nodes = [], list(self.nodes)121        for _ in range(num_walks):122            random.shuffle(nodes)123            walks.extend(self.walk(v, walk_len) for v in nodes)124        return walks125126    # ---- 3.2 Skip-Gram + 负采样 ----127    @staticmethod128    def _sigmoid(x: float) -> float:129        return 1.0 / (1.0 + math.exp(-max(-30.0, min(30.0, x))))130131    def _sgd_step(self, center: int, ctx: int, label: float) -> float:132        """单个 (center, ctx, label) 样本的梯度更新，返回 logistic 损失。"""133        u, v = self.emb_in[center], self.emb_out[ctx]134        score = sum(a * b for a, b in zip(u, v))135        pred = self._sigmoid(score)136        grad = pred - label  # dL/dscore137        for k in range(self.dim):138            uk, vk = u[k], v[k]139            self.emb_out[ctx][k] = vk - self.lr * grad * uk140            self.emb_in[center][k] = uk - self.lr * grad * vk141        return -math.log(pred + 1e-10) if label > 0.5 else -math.log(1.0 - pred + 1e-10)142143    def train(self, walks: List[List[str]], epochs: int = 3, window: int = 2,144              neg_k: int = 3, verbose: bool = True) -> None:145        """Skip-Gram 负采样训练，按轮打印平均损失。"""146        n = len(self.nodes)147        for ep in range(1, epochs + 1):148            total, cnt = 0.0, 0149            for path in walks:150                ids = [self.node2id[v] for v in path]151                for i, center in enumerate(ids):152                    ctxs = ids[max(0, i - window):i] + ids[i + 1:i + window + 1]153                    for ctx in ctxs:154                        total += self._sgd_step(center, ctx, 1.0)155                        cnt += 1156                        for _ in range(neg_k):  # 负样本：全局均匀采样（简化版）157                            neg = random.randrange(n)158                            if neg == ctx:159                                continue160                            total += self._sgd_step(center, neg, 0.0)161                            cnt += 1162            if verbose:163                print(f"    epoch {ep}/{epochs}  平均损失 = {total / max(cnt, 1):.4f}")164165    # ---- 3.3 召回 ----166    def embedding(self, node: str) -> Optional[List[float]]:167        """取节点的 Embedding（emb_in 为节点表示，emb_out 为上下文表示）。"""168        idx = self.node2id.get(node)169        return self.emb_in[idx] if idx is not None else None170171    def recommend(self, node: str, top_k: int = 3) -> List[Tuple[str, float]]:172        """以 node 为 Trigger，用 Embedding 余弦相似度检索 Top-K 近邻物品。"""173        src = self.embedding(node)174        if src is None:175            return []176        scores = [(v, cosine_sim(src, self.emb_in[i]))177                  for i, v in enumerate(self.nodes) if v != node]178        scores.sort(key=lambda x: x[1], reverse=True)179        return scores[:top_k]180181182def cosine_sim(a: List[float], b: List[float]) -> float:183    dot = sum(x * y for x, y in zip(a, b))184    na = math.sqrt(sum(x * x for x in a))185    nb = math.sqrt(sum(x * x for x in b))186    return dot / (na * nb + 1e-9)187188189# ── 4. 演示 ─────────────────────────────────────────────────────────────────190def make_sessions(groups: Dict[str, List[str]], bridges: List[Tuple[str, str]],191                  num: int = 60) -> List[List[str]]:192    """构造合成行为序列：组内高频共现，少量跨组桥接行为。"""193    names = list(groups)194    sessions = []195    for _ in range(num):  # 每个 Session 只在同一个类目内部产生连续点击196        items = groups[random.choice(names)]197        sessions.append([random.choice(items) for _ in range(4)])198    for a, b in bridges:  # 桥接边，使不同社区之间存在少量通路199        sessions.extend([[a, b]] * 2)200    return sessions201202203def same_group_ratio(walks: List[List[str]], item2group: Dict[str, str]) -> float:204    """统计游走序列中「与起点同社区」的节点占比，用于观察 BFS / DFS 倾向。"""205    hit = tot = 0206    for path in walks:207        g0 = item2group[path[0]]208        for v in path[1:]:209            tot += 1210            hit += int(item2group[v] == g0)211    return hit / max(tot, 1)212213214if __name__ == "__main__":215    random.seed(42)216    groups = {217        "数码": [f"i{j}" for j in range(1, 9)],218        "服饰": [f"i{j}" for j in range(9, 17)],219        "食品": [f"i{j}" for j in range(17, 25)],220    }221    item2group = {it: g for g, its in groups.items() for it in its}222    sessions = make_sessions(groups, bridges=[("i8", "i9"), ("i16", "i17"), ("i24", "i1")])223    graph = build_item_graph(sessions)224    print(f"物品图：{len(graph)} 个节点，{sum(len(v) for v in graph.values()) // 2} 条无向边\n")225226    # (p, q) 的两种典型取值：q > 1 偏 BFS（局部同质），q < 1 偏 DFS（跨社区探索）227    configs = [("BFS 式 (p=1, q=4)", 1.0, 4.0), ("DFS 式 (p=1, q=0.25)", 1.0, 0.25)]228    trigger = "i8"  # 桥接物品，位于「数码」与「服饰」两个社区之间229    for name, p, q in configs:230        random.seed(42)  # 固定种子，仅 (p, q) 不同，便于对比231        model = Node2Vec(graph, dim=16, p=p, q=q, lr=0.05)232        walks = model.generate_walks(num_walks=5, walk_len=10)233        print(f"[{name}] 游走 {len(walks)} 条，同社区节点占比 = "234              f"{same_group_ratio(walks, item2group):.3f}")235        model.train(walks, epochs=3, window=2, neg_k=3)236237        ranking = model.recommend(trigger, top_k=len(graph))238        same = [s for it, s in ranking if item2group[it] == item2group[trigger]]239        cross = [(r, it, s) for r, (it, s) in enumerate(ranking, 1)240                 if item2group[it] != item2group[trigger]]241        avg_same, avg_cross = sum(same) / len(same), sum(s for _, _, s in cross) / len(cross)242        print(f"    Trigger={trigger} 的 Top-5 召回："243              + "  ".join(f"{it}({item2group[it]}):{s:.3f}" for it, s in ranking[:5]))244        print(f"    最高分跨类物品 {cross[0][1]}({item2group[cross[0][1]]}) 排名第 {cross[0][0]}；"245              f"同类均值 {avg_same:.3f} / 跨类均值 {avg_cross:.3f}，"246              f"类间差距 = {avg_same - avg_cross:.3f}")247248    print("\n结论：q 较大（BFS 式）时游走几乎不出起点社区，Embedding 把同类物品挤在一起，同类"249          "\n      与跨类的相似度差距很大；q 较小（DFS 式）时游走频繁跨社区，类间差距明显缩小，"250          "\n      跨类物品更容易被召回，因此更适合兴趣扩散、长尾发现类任务。")