4. 漂移检测指标：嵌入空间距离、分布差异与统计检验方法

在 GraphSAGE 漂移监控中，选择合适的检测指标是准确识别数据分布变化的核心。本章系统梳理三类主流漂移检测指标：嵌入空间距离、分布差异度量和统计检验方法，并给出在 GraphSAGE 场景下的适用性对比。

4.1 嵌入空间距离

嵌入空间距离直接度量节点或子图在 GraphSAGE 输出向量空间中的位置变化。当模型参数或输入图结构发生漂移时，同一节点在不同时间窗口的嵌入向量会偏离原始位置。

4.1.1 余弦距离

余弦距离衡量两个嵌入向量在方向上的差异，忽略向量模长的影响。适用于关注节点语义角色是否改变的场景，如社区归属、功能类别等。

4.1.2 欧氏距离

欧氏距离同时考虑嵌入向量的方向和模长差异，对向量整体偏移更敏感。适用于需要捕捉特征强度变化的场景，如节点属性值整体增大或缩小。

4.2 分布差异度量

分布差异度量从概率分布层面比较两个时间窗口的嵌入向量整体分布。这类指标不关注单个节点的变化，而是捕捉整体分布形态的漂移。

4.2.1 KL 散度（Kullback-Leibler Divergence）

KL 散度衡量两个概率分布 P 和 Q 之间的非对称性差异。在 GraphSAGE 场景中，通常将参考窗口的嵌入分布作为 P，当前窗口的嵌入分布作为 Q。

4.2.2 Wasserstein 距离（Earth Mover's Distance）

Wasserstein 距离度量将一个概率分布“搬运”到另一个概率分布所需的最小代价。相比 KL 散度，它对分布重叠区域不敏感，能更稳定地反映分布的整体位移。

4.3 统计检验方法

统计检验方法通过假设检验判断两个样本集是否来自同一分布。这类方法提供显著性水平，帮助确定漂移是否具有统计意义，而非仅凭距离阈值判断。

4.3.1 两样本 Kolmogorov-Smirnov 检验（KS 检验）

KS 检验比较两个经验分布函数的最大垂直差异。适用于 GraphSAGE 嵌入向量每个维度的独立检验。

4.3.2 Maximum Mean Discrepancy（MMD）

MMD 通过将分布映射到再生核希尔伯特空间（RKHS）来比较两个分布。它天然支持高维嵌入向量的整体分布比较，是 GraphSAGE 漂移检测中最常用的统计检验方法。

4.3.3 卡方检验（Chi-Square Test）

卡方检验适用于离散化后的嵌入向量分布比较。将嵌入空间划分为多个区域后，统计每个区域内的节点频数变化。

4.4 指标选择策略

在实际 GraphSAGE 漂移监控系统中，应根据监控目标和资源约束选择合适的指标组合。下表总结了各指标的推荐使用场景：

4.5 指标计算代码示例

以下 Python 代码展示了如何在 GraphSAGE 嵌入向量上计算上述漂移检测指标：

import numpy as np
from scipy.spatial.distance import cosine, euclidean
from scipy.stats import ks_2samp, chi2_contingency
import ot  # Python Optimal Transport 库

def compute_cosine_distance(embed_ref, embed_curr):
    """计算两个嵌入向量的余弦距离"""
    return cosine(embed_ref, embed_curr)

def compute_euclidean_distance(embed_ref, embed_curr):
    """计算两个嵌入向量的欧氏距离"""
    return euclidean(embed_ref, embed_curr)

def compute_kl_divergence(p, q, eps=1e-10):
    """计算 KL 散度，p 和 q 为概率分布数组"""
    p = np.clip(p, eps, 1.0)
    q = np.clip(q, eps, 1.0)
    return np.sum(p * np.log(p / q))

def compute_wasserstein_distance(samples_ref, samples_curr):
    """计算 Wasserstein 距离"""
    # 将样本排序后计算一维 Wasserstein 距离
    ref_sorted = np.sort(samples_ref)
    curr_sorted = np.sort(samples_curr)
    return np.mean(np.abs(ref_sorted - curr_sorted))

def compute_mmd(samples_ref, samples_curr, kernel='rbf', sigma=1.0):
    """计算 Maximum Mean Discrepancy"""
    def rbf_kernel(x, y):
        return np.exp(-np.linalg.norm(x - y)**2 / (2 * sigma**2))
    
    n = len(samples_ref)
    m = len(samples_curr)
    
    # 计算三项核均值
    xx = np.mean([rbf_kernel(samples_ref[i], samples_ref[j]) 
                  for i in range(n) for j in range(n)])
    yy = np.mean([rbf_kernel(samples_curr[i], samples_curr[j]) 
                  for i in range(m) for j in range(m)])
    xy = np.mean([rbf_kernel(samples_ref[i], samples_curr[j]) 
                  for i in range(n) for j in range(m)])
    
    return xx + yy - 2 * xy

def compute_ks_statistic(samples_ref, samples_curr):
    """计算 KS 检验统计量"""
    statistic, p_value = ks_2samp(samples_ref, samples_curr)
    return statistic, p_value

# 使用示例
if __name__ == "__main__":
    # 模拟 GraphSAGE 嵌入向量
    np.random.seed(42)
    embed_ref = np.random.randn(128)  # 参考窗口嵌入
    embed_curr = np.random.randn(128) + 0.1  # 当前窗口嵌入（轻微漂移）
    
    # 计算距离指标
    cos_dist = compute_cosine_distance(embed_ref, embed_curr)
    euc_dist = compute_euclidean_distance(embed_ref, embed_curr)
    
    print(f"余弦距离: {cos_dist:.4f}")
    print(f"欧氏距离: {euc_dist:.4f}")
    
    # 模拟分布样本
    samples_ref = np.random.randn(1000)
    samples_curr = np.random.randn(1000) + 0.2
    
    # 计算分布差异
    wass_dist = compute_wasserstein_distance(samples_ref, samples_curr)
    ks_stat, ks_p = compute_ks_statistic(samples_ref, samples_curr)
    
    print(f"Wasserstein 距离: {wass_dist:.4f}")
    print(f"KS 统计量: {ks_stat:.4f}, p-value: {ks_p:.4f}")