4. GraphSAGE 模型架构：聚合函数、层级传播与损失函数设计

GraphSAGE（Graph Sample and Aggregation）的核心思想是通过采样邻居和聚合邻居特征来生成节点的嵌入表示，从而避免对全图进行拉普拉斯分解，使其能够高效处理大规模工业级反欺诈图谱。本节将深入剖析其三大核心组件：聚合函数、层级传播机制以及损失函数设计。

4.1 聚合函数（Aggregator Functions）

聚合函数决定了如何将采样到的邻居节点特征融合到目标节点上。GraphSAGE 提出了三种可选的聚合器，每种在反欺诈场景下有不同的适用性。

4.2 层级传播（Layer-wise Propagation）

GraphSAGE 通过多层聚合实现层级传播，每一层都扩展了节点的感受野。在反欺诈场景中，层级传播使得模型能够从“直接交易对手”逐步扩展到“二度关系”甚至“三度关系”，从而发现隐蔽的团伙结构。

层级传播的数学递推公式：

h_v^(k) = σ( W^(k) · CONCAT( h_v^(k-1), AGG^(k)( { h_u^(k-1), ∀u ∈ N(v) } ) ) )

其中：

• h_v^(k)：节点 v 在第 k 层的嵌入表示。
• h_v^(k-1)：节点 v 在上一层的嵌入。
• AGG^(k)：第 k 层使用的聚合函数（Mean/LSTM/Pooling）。
• N(v)：节点 v 的采样邻居集合。
• CONCAT：将自身特征与聚合后的邻居特征拼接，保留自身信息。

4.3 损失函数设计（Loss Function Design）

GraphSAGE 支持无监督和有监督两种训练范式。在反欺诈场景中，通常采用有监督方式，但无监督预训练也可用于冷启动。

4.3.1 无监督损失（基于图结构）

无监督损失鼓励相邻节点具有相似的嵌入，而远离的节点嵌入不同。公式如下：

J_G(z_u) = -log( σ(z_u^T z_v) ) - Q · E_{v_n ~ P_n(v)} [ log( σ(-z_u^T z_{v_n}) ) ]

• z_u, z_v：正样本对（图中相邻节点）的嵌入。
• z_{v_n}：负样本节点嵌入，从噪声分布 P_n 中采样。
• Q：负采样数量，通常设为 5~20。
• σ：sigmoid 函数。

4.3.2 有监督损失（反欺诈场景）

在反欺诈标注数据充足时，使用交叉熵损失或 Focal Loss 来优化模型：

L = - Σ [ y · log(ŷ) + (1 - y) · log(1 - ŷ) ]

针对欺诈样本极度不平衡的问题（欺诈率通常 < 1%），推荐使用 Focal Loss：

L_focal = - α_t · (1 - p_t)^γ · log(p_t)

• α_t：类别权重，用于平衡正负样本。
• γ：聚焦参数（通常设为 2），降低易分类样本的损失权重，使模型更关注难分的欺诈样本。

4.4 完整前向传播流程

以下代码展示了 GraphSAGE 单层前向传播的 PyTorch 风格伪代码，整合了采样、聚合和层级传播：

class GraphSAGELayer(nn.Module):
    def __init__(self, in_dim, out_dim, aggregator_type='mean'):
        super().__init__()
        self.aggregator_type = aggregator_type
        self.W = nn.Linear(in_dim * 2, out_dim)  # 拼接自身与邻居

        if aggregator_type == 'mean':
            self.agg = MeanAggregator()
        elif aggregator_type == 'pool':
            self.agg = PoolingAggregator(in_dim, out_dim)
        elif aggregator_type == 'lstm':
            self.agg = LSTMAggregator(in_dim, out_dim)

    def forward(self, h_self, h_neighbors):
        # h_self: [batch, in_dim]
        # h_neighbors: [batch, num_samples, in_dim]
        h_neigh_agg = self.agg(h_neighbors)          # [batch, in_dim]
        h_concat = torch.cat([h_self, h_neigh_agg], dim=1)  # [batch, 2*in_dim]
        h_out = torch.relu(self.W(h_concat))         # [batch, out_dim]
        return h_out