GCN 코드를 통해 이해

그래프는 노드와 간선으로 구성됨.

 

 

import torch
from torch_geometric.data import Data

edge_index = torch.tensor([[0, 1, 1, 2], [1, 0, 2, 1]], dtype=torch.long)
x = torch.tensor([[-1], [0], [1]], dtype=torch.float)

data = Data(x=x, edge_index=edge_index)

 

edge_index

각 열은 하나의 엣지를 나타내며,

 

첫 번째 행은 엣지의 출발 노드

두 번째 행은 엣지의 도착 노드를 나타냄

 

노드 0에서 노드 1로 엣지

노드 1에서 노드 0으로 엣지(양방향)

노드 1에서 노드 2로 엣지

노드 2에서 노드 1로 엣지 (양방향)

 

두 배열을 좌 우로 보면 무슨말인지 알 수 있을 거임.

 

x

각 노드는 하나의 피처(값)을 가짐

노드 0 : -1

노드 1 : 0

노드 2 : 1

 

Data 객체

Data 객체는 그래프 데이터를 캡슐화를 한다. 

여기에는 노드 피처 x 와 엣지 연결 정보 edge_index가 포함

이 객체는 이후 그래프 신경망 모델에 입력으로 사용될 수 있음

 

그래프로 뽑아봤다.

시각화까지 포함된 코드는 다음과 같다.

 

import torch
import networkx as nx
import matplotlib.pyplot as plt
from torch_geometric.data import Data
from torch_geometric.utils import to_networkx

edge_index = torch.tensor([[0, 1, 1, 2], [1, 0, 2, 1]], dtype=torch.long)
x = torch.tensor([[-1], [0], [1]], dtype=torch.float)

data = Data(x=x, edge_index=edge_index)

# PyTorch Geometric의 그래프를 NetworkX로 변환
G = to_networkx(data, to_undirected=True)

# 노드의 특징 (x 값을 노드의 색으로 표현)
node_colors = [data.x[i].item() for i in range(data.num_nodes)]

# 그래프 시각화
plt.figure(figsize=(8, 6))
nx.draw(G, with_labels=True, node_color=node_colors, cmap=plt.cm.Blues, node_size=500, font_size=16)
plt.show()

 

 

Cora 데이터셋을 활용하여 GNN 을 학습시켜 링크 예측 작업을 수행

그리고 최종적으로 t-SNE를 이용해 노드 임베딩을 시각화하는 전체 파이프라인

 

일반 Cora 데이터셋이 뭐냐?

논문이 노드로, 논문 간의 인용 관계가 에지로 표현된 그래프 구조이다.

즉, 하나의 논문이 다른 논문을 인용할 때 두 논문 간에 에지가 형성된다.

 

각 노드는 학술 논문에 해당, 각 논문에는 특징 벡터가 연결되어 있음

이 특징 벡터는 해당 논문에 등장한 단어들로 표현되며, 보통 Bag of Words 방식으로 인코딩됨

총 2708개의 노드(논문)이 있음.

 

에지는 논문 간의 인용 관계를 나타냄 총 5429개의 에지가 있음

 

<활용>

Cora 데이터셋의 활용은

- 노드 분류

주어진 노드의 특징 벡터와 그래프 구조를 바탕으로, 노드가 어떤 클래스로 분류될지 예측하는 작업

각 논문이 특정 연구 주제에 속하는지 예측하는 것이 목표

 

- 링크 예측

그래프에서 연결되지 않은 두 노드 사이에 에지가 존재할 가능성을 예측하는 작업

Cora 데이터셋의 인용 네트워크에서, 한 논문이 다른 논문을 인용할 가능성을 예측하는 방식으로 활용

 

 

- 그래프 표현 학습

노드 또는 그래프를 벡터 공간에 매핑하여, 이를 통해 다양한 그래프 관련 작업을 수행하는 방법

GCN과 같은 모델을 학습하여 노드 임베딩을 생성하고, 이를 다양한 다운스트림 작업에 사용할 수 있다.

 

 

 

데이터 로드

dataset_name = 'Cora'
path = osp.join('../', 'data', dataset_name)
dataset = Planetoid(path, dataset_name, transform=T.NormalizeFeatures())
data = dataset[0]

 

 

 

모델 정의

class Net(torch.nn.Module):
    def __init__(self):
        super(Net, self).__init__()
        self.conv1 = GCNConv(dataset.num_node_features, 16)
        self.conv2 = GCNConv(16, dataset.num_classes)

    def encode(self, x, edge_index):
        x = self.conv1(x, edge_index)  # 첫 번째 GCN 레이어
        x = x.relu()  # 활성화 함수 ReLU 적용
        return self.conv2(x, edge_index)  # 두 번째 GCN 레이어

    def decode(self, z, pos_edge_index, neg_edge_index):
        edge_index = torch.cat([pos_edge_index, neg_edge_index], dim=-1)
        logits = (z[edge_index[0]] * z[edge_index[1]]).sum(dim=-1)
        return logits

    def decode_all(self, z):
        prob_adj = z @ z.t()  # 인접 행렬 계산
        return (prob_adj > 0).nonzero(as_tuple=False).t()

 

모델 클래스는 두 개의 GCN 레이어로 구성되어 있으며,

그래프 데이터를 인코인하고 디코딩하는 기능이 들어가 있다.

 

encode : 노드 특성 x와 에지 정보 edge_index를 받아 노드 임베딩을 생성

decode : 양수 및 음수 에지 인덱스를 사용하여 두 노드 간의 연관성을 계산 (내적사용함)

decode_all : 전체 노드 간의 연결 가능성을 계산

 

 

 

유틸리티 함수 정의

def get_link_logits(model, x, edge_index, neg_edge_index):
    z = model.encode(x, edge_index)
    link_logits = model.decode(z, edge_index, neg_edge_index)
    return link_logits

def get_link_labels(pos_edge_index, neg_edge_index):
    E = pos_edge_index.size(1) + neg_edge_index.size(1)
    link_labels = torch.zeros(E, dtype=torch.float, device=device)
    link_labels[:pos_edge_index.size(1)] = 1.
    return link_labels

 

get_link_logits : 인코딩된 노드 임베딩을 사용해 에지의 연결 확률을 예측하는 함수

get_link_labels : 양수 및 음수 에지를 기반으로 라벨(1또는 0)을 생성

 

여기서 설명

 

1. 양수 에지(Positive Edges)

- 정의 : 양수 에지는 실제로 존재하는 에지를 의미한다.

- 즉, 두 노드 간에 실제로 연결(연결성 또는 관계)이 있는 경우를 나타냄

- 예시 : cora데이터셋의 경우, 양수 에지는 한 논문이 다른 논문을 인용하는 경우(실제)

그래서 링크 예측 모델은 이 양수 에지를 사용해 두 노드가 연결될 가능성을 학습함

나중에 새로운 노드 쌍에 대해 연결될 가능성을 예측할 수 있게 됨

 

2. 음수 에지(Negative Edges)

- 음수 에지는 존재하지 않는 에지를 의미한다. 

- 두 노드 간에 실제로는 연결이 없지만, 모델을 학습할 때는 학습 데이터를 보강하기 위해 인위적으로 생성된 에지

음수 샘플링 : 두 노드 간에 임의로 에지를 추가하고, 이 에지를 "연결되지 않음"으로 레이블링하는 과정

- 예시 : Cora 데이터셋에서 연결되지 않은 논문 쌍을 선택하여 음수 에지를 생성할 수 있다.

이때 이 에지는 학습 중에 모델이 "이 노드 쌍은 연결되지 않는다" 는 것을 학습하도록 돕는다.

즉, 음수 에지는 모델이 두 노드 간의 연결이 없을 가능성을 학습하는 데 사용된다. 

모델이 단순히 모든 노드를 연결된다고 예측하는 것을 방지하고, 실제로 연결되지 않은 노드 쌍으로 올바르게 예측할 수 있도록 돕는다.

양수 에지

실제로 존재하는 노드 간의 연결

• (1, 2)
• (2, 3)
• (3, 4)
• (4, 5)

음수 에지

존재하지 않는, 인위적으로 생성된 노드 간의 연결

• (1, 3)
• (2, 4)
• (3, 5)

 

 

 

모델 초기화 및 옵티마이저 설정

model = Net().to(device)
optimizer = torch.optim.Adam(model.parameters(), lr=0.01)

 

 

 

 

훈련 함수

def train():
    model.train()
    neg_edge_index = negative_sampling(
        edge_index=data.train_pos_edge_index,
        num_nodes=data.num_nodes,
        num_neg_samples=data.train_pos_edge_index.size(1)
    )

    link_logits = get_link_logits(model, data.x, data.train_pos_edge_index, neg_edge_index)
    optimizer.zero_grad()
    link_labels = get_link_labels(data.train_pos_edge_index, neg_edge_index)
    loss = F.binary_cross_entropy_with_logits(link_logits, link_labels)
    loss.backward()
    optimizer.step()

    return loss

 

음수 샘플링을 통해 연결되지 않은 에지를 생성

링크 예측을 위한 로짓을 계산하고, 이진 교차 엔트로피 손실을 통해 학습

옵티마이저를 통해 손실을 역전파하여 모델의 파라미터 갱신

 

 

 

테스트 함수

@torch.no_grad()
def test():
    model.eval()
    perfs = []
    for prefix in ["val", "test"]:
        pos_edge_index = data[f'{prefix}_pos_edge_index']
        neg_edge_index = data[f'{prefix}_neg_edge_index']

        link_logits = get_link_logits(model, data.x, pos_edge_index, neg_edge_index)
        link_probs = link_logits.sigmoid()  # 시그모이드 함수로 확률 계산
        link_labels = get_link_labels(pos_edge_index, neg_edge_index)

        perfs.append(roc_auc_score(link_labels.cpu(), link_probs.cpu()))  # ROC AUC 스코어 계산
    return perfs

 

모델을 평가 모드로 설정한 후, 검증 및 데스트를 통해 예측을 수행하고 ROC AUC 점수 계산

Final Val AUC: 0.8619, Final Test AUC: 0.8703

 

 

노드 임베딩 시각화

z = model.encode(data.x, data.train_pos_edge_index)
emb = TSNE(n_components=2, learning_rate='auto').fit_transform(z.detach().numpy())
labels = dataset[0].y.detach().numpy()

fig, ax = plt.subplots()
number_of_colors = len(np.unique(labels))
color = ["#"+''.join([random.choice('0123456789ABCDEF') for j in range(6)])
         for i in range(number_of_colors)]

for idx, i in enumerate(np.unique(labels)):
    emb_ = emb[np.where(labels == i), :].squeeze()
    ax.scatter(x=emb_[:, 0], y=emb_[:, 1], c=color[idx], label=i, alpha=0.2)

ax.legend()
plt.show()

 

t-SNE 시각화 : 학습된 노드 임베딩을 2차원으로 축소한 후, 노드의 레이블에 따라 색상으로 구분하여 시각화함

 

 

 

pip install torch torch-geometric scikit-learn matplotlib numpy torch-scatter torch-sparse torch-cluster torch-spline-conv

 

전체 코드 

import os.path as osp
import torch
import torch.nn.functional as F
from sklearn.metrics import roc_auc_score
from sklearn.manifold import TSNE
import matplotlib.pyplot as plt
import numpy as np
import random

from torch_geometric.utils import negative_sampling
from torch_geometric.datasets import Planetoid
import torch_geometric.transforms as T
from torch_geometric.nn import GCNConv
from torch_geometric.utils import train_test_split_edges

# Set the device
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')

# Load Data
dataset_name = 'Cora'
path = osp.join('../', 'data', dataset_name)
dataset = Planetoid(path, dataset_name, transform=T.NormalizeFeatures())
data = dataset[0]

# Prepare Data: Split edges into training, validation, and test sets
data.train_mask = data.val_mask = data.test_mask = data.y = None
data = train_test_split_edges(data)

# Define Model
class Net(torch.nn.Module):
    def __init__(self):
        super(Net, self).__init__()
        self.conv1 = GCNConv(dataset.num_node_features, 16)
        self.conv2 = GCNConv(16, dataset.num_classes)

    def encode(self, x, edge_index):
        x = self.conv1(x, edge_index)  # convolution 1
        x = x.relu()
        return self.conv2(x, edge_index)  # convolution 2

    def decode(self, z, pos_edge_index, neg_edge_index):
        edge_index = torch.cat([pos_edge_index, neg_edge_index], dim=-1)  # concatenate pos and neg edges
        logits = (z[edge_index[0]] * z[edge_index[1]]).sum(dim=-1)  # dot product
        return logits

    def decode_all(self, z):
        prob_adj = z @ z.t()  # get adjacency matrix NxN
        return (prob_adj > 0).nonzero(as_tuple=False).t()  # get predicted edge_list

# Utility Functions
def get_link_logits(model, x, edge_index, neg_edge_index):
    z = model.encode(x, edge_index)  # encode
    link_logits = model.decode(z, edge_index, neg_edge_index)  # decode
    return link_logits

def get_link_labels(pos_edge_index, neg_edge_index):
    E = pos_edge_index.size(1) + neg_edge_index.size(1)
    link_labels = torch.zeros(E, dtype=torch.float, device=device)
    link_labels[:pos_edge_index.size(1)] = 1.
    return link_labels

# Initialize Model and Optimizer
model = Net().to(device)
optimizer = torch.optim.Adam(model.parameters(), lr=0.01)

# Training Function
def train():
    model.train()
    neg_edge_index = negative_sampling(
        edge_index=data.train_pos_edge_index,
        num_nodes=data.num_nodes,
        num_neg_samples=data.train_pos_edge_index.size(1)
    )

    link_logits = get_link_logits(model, data.x, data.train_pos_edge_index, neg_edge_index)
    optimizer.zero_grad()
    link_labels = get_link_labels(data.train_pos_edge_index, neg_edge_index)
    loss = F.binary_cross_entropy_with_logits(link_logits, link_labels)
    loss.backward()
    optimizer.step()

    return loss

# Test Function
@torch.no_grad()
def test():
    model.eval()
    perfs = []
    for prefix in ["val", "test"]:
        pos_edge_index = data[f'{prefix}_pos_edge_index']
        neg_edge_index = data[f'{prefix}_neg_edge_index']

        link_logits = get_link_logits(model, data.x, pos_edge_index, neg_edge_index)
        link_probs = link_logits.sigmoid()  # apply sigmoid
        link_labels = get_link_labels(pos_edge_index, neg_edge_index)

        perfs.append(roc_auc_score(link_labels.cpu(), link_probs.cpu()))  # compute roc_auc score
    return perfs

# Training Loop
for epoch in range(1, 201):
    loss = train()
    if epoch % 10 == 0:
        val_auc, test_auc = test()
        print(f'Epoch: {epoch:03d}, Loss: {loss:.4f}, Val AUC: {val_auc:.4f}, Test AUC: {test_auc:.4f}')

# 최종 검증 및 테스트 AUC 점수 출력
val_auc, test_auc = test()
print(f'Final Val AUC: {val_auc:.4f}, Final Test AUC: {test_auc:.4f}')

# Node Embedding Visualization using t-SNE
z = model.encode(data.x, data.train_pos_edge_index)
emb = TSNE(n_components=2, learning_rate='auto').fit_transform(z.detach().numpy())
labels = dataset[0].y.detach().numpy()

fig, ax = plt.subplots()
number_of_colors = len(np.unique(labels))
color = ["#"+''.join([random.choice('0123456789ABCDEF') for j in range(6)])
         for i in range(number_of_colors)]

for idx, i in enumerate(np.unique(labels)):
    emb_ = emb[np.where(labels == i), :].squeeze()
    ax.scatter(x=emb_[:, 0], y=emb_[:, 1], c=color[idx], label=i, alpha=0.2)

ax.legend()
plt.show()