1. GCN 기반 추천 시스템 진화 과정
1-1. MF (Matrix Factorization)
- 기존 협업 필터링으로 정적 임베딩 후 사후 연산 방식
- 순전파 단계에서 유저 벡터와 아이템 벡터가 서로 독립적
- 예측 점수를 계산할 때 단순 내적을 수행하므로 그래프 구조나 이웃 정보가 임베딩 변화에 관여하지 않음
class MatrixFactorization(nn.Module):
def __init__(self, num_users, num_items, embed_dim):
super().__init__()
# Lookup Table 생성
self.user_embedding = nn.Embedding(num_users, embed_dim)
self.item_embedding = nn.Embedding(num_items, embed_dim)
def forward(self, user_idx, item_idx):
# 1. 단순히 ID에 해당하는 벡터를 꺼내옴 (Lookup)
# 이 시점에서 u_emb는 어떤 그래프 정보도 포함하지 않음.
u_emb = self.user_embedding(user_idx)
i_emb = self.item_embedding(item_idx)
# 2. 내적 수행 (Linear Interaction)
prediction = (u_emb * i_emb).sum(dim=1)
return prediction- 단순히 lookup(특정 벡터 꺼내옴) 후 내적을 수행하므로 이웃의 정보를 학습하지 못함
1-2. NGCF
- 추천 시스템에 GCN(Graph Convolution Network)를 처음 제대로 도입
- MF는 자신의 정보(user와 item)만 학습하여 이웃의 정보를 고려하지 않았지만, NGCF의 경우 layer의 개수만큼(여러 hop만큼) 이웃의 정보를 Aggregate(집계)해서 update하여 propagate(전파)할 수 있음
def forward(self, users, pos_items, neg_items, drop_flag=True):
# 노드 드롭아웃 적용 (Graph Structure Learning 효과)
# 인접 행렬의 일부 연결을 무작위로 제거하여 과적합 방지
# drop_flag가 True일 때만 노드 드롭아웃 적용 -> 학습 시에만 적용하고 테스트 시에는 적용하지 않음
A_hat = self.sparse_dropout(self.sparse_norm_adj,
self.node_dropout,
self.sparse_norm_adj._nnz()) if drop_flag else self.sparse_norm_adj
# 초기 임베딩 (Layer 0) - 사용자와 아이템 임베딩을 결합
# 학습 가능한 파라미터인 유저와 아이템의 ID 임베딩을 하나로 합침
# ego_embeddings shape: [유저수 + 아이템수, 임베딩 차원]
ego_embeddings = torch.cat([self.embedding_dict['user_emb'],
self.embedding_dict['item_emb']], 0)
# 각 레이어(Layer 0, 1, 2...)에서 나온 임베딩을 모두 저장할 리스트
# 나중에 이들을 모두 이어 붙여(Concat) 최종 임베딩으로 씀
all_embeddings = [ego_embeddings]
# 레이어 수만큼 임베딩 전파
for k in range(len(self.layers)):
# --- [A] 정보 집계 (Aggregation) ---
# 인접 행렬과 현재 임베딩을 곱하여 이웃 노드의 정보를 가져옴
# side_embeddings: 이웃 노드들의 임베딩 가중합
side_embeddings = torch.sparse.mm(A_hat, ego_embeddings)
# --- [B] Feature Transformation (가중치 변환) ---
# 이웃에서 가져온 정보를 단순히 더하는 게 아니라, 학습 가능한 가중치(W1)를 곱해 유의미한 특징을 추출
sum_embeddings = torch.matmul(side_embeddings, self.weight_dict['W_gc_%d' % k]) \
+ self.weight_dict['b_gc_%d' % k]
# --- [C] Bi-Interaction (상호작용 항) ---
# NGCF의 핵심 특징. 나와 이웃의 특징을 Element-wise Product(⊙)로 곱함
# "나와 비슷한 특징을 가진 이웃의 정보는 더 강하게 받아들이겠다"는 의도
bi_embeddings = torch.mul(ego_embeddings, side_embeddings)
bi_embeddings = torch.matmul(bi_embeddings, self.weight_dict['W_bi_%d' % k]) \
+ self.weight_dict['b_bi_%d' % k]
# --- [D] 비선형 활성화 (Non-linearity) ---
# Sum 정보와 Bi 정보(상호작용)를 합쳐서 비선형 변환
ego_embeddings = nn.LeakyReLU(negative_slope=0.2)(sum_embeddings + bi_embeddings)
# --- [E] 메시지 드롭아웃 (Message Dropout) ---
# 계산된 임베딩 값의 일부를 랜덤하게 0으로 만듦 (Feature level dropout)
ego_embeddings = nn.Dropout(self.mess_dropout[k])(ego_embeddings)
# --- [F] 정규화 (Normalization) ---
# 임베딩 벡터의 크기(Norm)를 1로 맞춰 학습 발산을 막습니다.
norm_embeddings = F.normalize(ego_embeddings, p=2, dim=1)
# 현재 레이어의 결과 임베딩을 리스트에 저장
all_embeddings += [norm_embeddings]
# 모든 레이어의 임베딩을 결합 (Concatenation)
all_embeddings = torch.cat(all_embeddings, 1)
# 결합된 전체 임베딩을 다시 사용자와 아이템 부분으로 분리
u_g_embeddings = all_embeddings[:self.n_user, :]
i_g_embeddings = all_embeddings[self.n_user:, :]
u_g_embeddings = u_g_embeddings[users, :]
pos_i_g_embeddings = i_g_embeddings[pos_items, :]
neg_i_g_embeddings = i_g_embeddings[neg_items, :]
return u_g_embeddings, pos_i_g_embeddings, neg_i_g_embeddings1-3. LightGCN
- NGCF는 가중치(W)와 비선형성(σ) 계산을 수행하느라 무겁고 학습 오래 걸림
- LightGCN은 가중치와 비선형성 계산을 제거하여 성능과 학습 속도 향상함
- 이미지의 경우 노드 즉, 픽셀 값 자체에 의미가 없으므로 특징을 만들어내기 위해 가중치(W)와 비선형성(σ) 필수
- 추천 시스템의 경우 각 노드는 그 자체로 의미가 있는 특징이므로, 가중치(W)와 비선형성(σ)은 과적합 가능성을 향상 시키고 학습이 오래걸리고 무거움
def computer(self):
"""
LightGCN의 핵심: Embedding Propagation (임베딩 전파)
NGCF와 달리 가중치 행렬(W)이나 활성화 함수(ReLU) 없이
오직 그래프 구조(인접 행렬)만을 이용해 임베딩을 확산
"""
# 1. 초기 임베딩 (Layer 0) 준비
# 학습 가능한 유일한 파라미터인 유저/아이템 ID 임베딩을 가져옵니다.
users_emb = self.embedding_user.weight
items_emb = self.embedding_item.weight
# 유저와 아이템을 하나의 행렬로 합침 (Shape: [N+M, Embedding_Dim])
all_emb = torch.cat([users_emb, items_emb])
# 각 레이어별 임베딩을 저장할 리스트 (Layer 0 저장)
embs = [all_emb]
# 2. Graph Dropout
# 과적합 방지를 위해 엣지(Edge)를 랜덤하게 끊어버린 그래프를 사용할지 결정
if self.config['dropout']:
if self.training:
# 학습 중일 때만 드롭아웃 적용
g_droped = self.__dropout(self.keep_prob)
else:
# 테스트 중일 때는 원본 그래프 사용
g_droped = self.Graph
else:
# 드롭아웃 미사용 시 미리 계산된 정규화 인접 행렬(Normalized Adj) 사용
g_droped = self.Graph
# 3. Light Graph Convolution (레이어 통과)
# 설정한 레이어 수(n_layers)만큼 반복하며 정보를 멀리까지 전파
for layer in range(self.n_layers):
if self.A_split:
# (대용량 데이터 처리를 위해 그래프를 쪼개서 연산하는 경우)
temp_emb = []
for f in range(len(g_droped)):
temp_emb.append(torch.sparse.mm(g_droped[f], all_emb))
side_emb = torch.cat(temp_emb, dim=0)
all_emb = side_emb
else:
# [핵심 로직] Linear Propagation
# - sparse.mm: 행렬 곱을 수행하여 "이웃의 정보를 가중 합(Weighted Sum)"함.
# - 중요: NGCF와 달리 W(가중치)를 곱하거나 ReLU를 씌우지 않음!
all_emb = torch.sparse.mm(g_droped, all_emb)
# 이번 레이어에서 확산된 임베딩을 리스트에 저장
embs.append(all_emb)
# 4. Layer Combination (최종 임베딩 생성)
# embs 리스트 구조: [E^0, E^1, E^2, ...]
# stack: 텐서를 쌓아서 차원 추가 (N+M, Layer+1, Dim)
embs = torch.stack(embs, dim=1)
# [LightGCN의 특징] 가중 합 대신 단순 평균(Mean) 사용
# 이유:
# - Layer 0 (본연의 특성)과 Layer K (이웃의 특성)를 골고루 반영하기 위함.
# - 학습 파라미터(Attention 등)를 줄여서 일반화 성능을 높임.
light_out = torch.mean(embs, dim=1)
# 최종 계산된 임베딩을 다시 유저용과 아이템용으로 분리하여 반환
users, items = torch.split(light_out, [self.num_users, self.num_items])
return users, items1-4. SGL
-
LightGCN 구조를 사용하면서 데이터를 스스로 증강하여 희소성 문제를 해결함
-
매 에폭마다 원본 그래프를 변형(Edge Dropout)하여 서로 다른 2개의 그래프 뷰(View 1, View 2)를 생성 (model.graph_reconstruction)
-
두 뷰(View 1, View 2)를 이용해 다음과 같은 대조 학습(Contrastive Learning)을 수행(model.cal_cl_loss)
-
임베딩 추출: 두 뷰를 LightGCN에 각각 통과시켜, 동일한 유저에 대한 두 개의 임베딩 벡터(e1,e2)를 얻음 (model.forward)
-
Positive Pair 학습 (Alignment): e1과 e2는 비록 다른 그래프에서 나왔지만, 태생이 같은 유저이므로 임베딩 공간에서 서로 가까워지도록(Pull) 학습 (InfoNCE - 분자 부분)
-
Negative Pair 학습 (Uniformity): 반면, 다른 유저들의 임베딩과는 서로 멀어지도록(Push) 학습 (InfoNCE - 분모 부분)
-
-
위 과정을 통해 모델은 특정 엣지(구매 이력)가 우연히 없더라도 유저 고유의 특성을 파악할 수 있음 -> 하지만 매 에폭마다 그래프를 생성하므로 연산 비용 up
(1) 그래프 증강
def random_graph_augment(self):
"""
[SGL의 핵심 1] 그래프 증강(Augmentation) 함수
SGL은 학습 시 원본 그래프를 그대로 쓰지 않고,
노드나 엣지를 무작위로 제거하여 '어려운 문제'를 만듭니다.
이를 통해 모델이 노이즈에 강해지고(Robust), 데이터 희소성을 극복하게 합니다.
"""
dropped_mat = None
# 1. 노드 드롭아웃 (Node Dropout)
# 특정 유저나 아이템 노드를 아예 삭제해버림. (가장 강력한 변형)
if self.aug_type == 0:
dropped_mat = GraphAugmentor.node_dropout(self.data.interaction_mat, self.drop_rate)
# 2. 엣지 드롭아웃 (Edge Dropout) - 논문에서 가장 성능이 좋았던 방식
# 유저-아이템 간의 연결 선만 끊어냄.
elif self.aug_type == 1 or self.aug_type == 2:
dropped_mat = GraphAugmentor.edge_dropout(self.data.interaction_mat, self.drop_rate)
# 3. 정규화 (Normalization)
# 변형된 그래프도 GCN 연산을 위해 라플라시안 행렬(D^-0.5 A D^-0.5)로 변환
dropped_mat = self.data.convert_to_laplacian_mat(dropped_mat)
# PyTorch 텐서로 변환하여 GPU로 올림
return TorchGraphInterface.convert_sparse_mat_to_tensor(dropped_mat).cuda()(2) 대조 학습 손실 계산(Contrastive Loss)
def forward(self, perturbed_adj=None):
ego_embeddings = torch.cat([self.embedding_dict['user_emb'], self.embedding_dict['item_emb']], 0)
all_embeddings = [ego_embeddings]
for k in range(self.n_layers):
if perturbed_adj is not None:
# 증강된 그래프 사용 (대조 학습용)
if isinstance(perturbed_adj, list):
# 레이어별 다른 증강 그래프 사용 (aug_type=2)
ego_embeddings = torch.sparse.mm(perturbed_adj[k], ego_embeddings)
else:
# 모든 레이어에서 같은 증강 그래프 사용 (aug_type=0,1)
ego_embeddings = torch.sparse.mm(perturbed_adj, ego_embeddings)
else:
# 원본 그래프 사용 (추천 손실 계산용)
ego_embeddings = torch.sparse.mm(self.sparse_norm_adj, ego_embeddings)
all_embeddings.append(ego_embeddings)
# 모든 레이어 출력을 스택하고 평균
all_embeddings = torch.stack(all_embeddings, dim=1)
all_embeddings = torch.mean(all_embeddings, dim=1)
user_all_embeddings, item_all_embeddings = torch.split(all_embeddings, [self.data.user_num, self.data.item_num])
return user_all_embeddings, item_all_embeddings
def cal_cl_loss(self, idx, perturbed_mat1, perturbed_mat2):
"""
[SGL의 핵심 2] InfoNCE Loss 계산 (대조 학습)
서로 다르게 망가뜨린 두 그래프(View 1, View 2)에서
동일한 유저(u)의 임베딩은 가까워져야 하고(Positive Pair),
다른 유저들과는 멀어져야 한다(Negative Pair)는 법칙을 수식화한 것입니다.
"""
# 현재 배치의 유저와 아이템 인덱스
u_idx = torch.unique(torch.Tensor(idx[0]).type(torch.long)).cuda()
i_idx = torch.unique(torch.Tensor(idx[1]).type(torch.long)).cuda()
# View 1 그래프를 통과시켜 얻은 임베딩
user_view_1, item_view_1 = self.forward(perturbed_mat1)
# View 2 그래프를 통과시켜 얻은 임베딩
user_view_2, item_view_2 = self.forward(perturbed_mat2)
# 계산 효율성을 위해 유저와 아이템 임베딩을 하나로 합침
# view1: [유저_임베딩_1; 아이템_임베딩_1]
# view2: [유저_임베딩_2; 아이템_임베딩_2]
view1 = torch.cat((user_view_1[u_idx], item_view_1[i_idx]), 0)
view2 = torch.cat((user_view_2[u_idx], item_view_2[i_idx]), 0)
# InfoNCE Loss 계산
# - 분자: 내 분신(view1의 나, view2의 나) 간의 유사도 (높여야 함)
# - 분모: 나랑 다른 모든 노드들과의 유사도 합 (낮춰야 함)
return InfoNCE(view1, view2, self.temp)(3) 핵심 함수 InfoNCE
def InfoNCE(view1, view2, temperature: float, b_cos: bool = True):
"""
InfoNCE (Information Noise Contrastive Estimation) 손실 함수
대조 학습(Contrastive Learning)의 핵심 손실 함수로,
같은 샘플의 서로 다른 뷰(양성 쌍)는 가깝게,
다른 샘플의 뷰(음성 쌍)는 멀게 학습시킵니다.
Args:
view1: (torch.Tensor - N x D)
view2: (torch.Tensor - N x D)
temperature: float
b_cos (bool)
Return: Average InfoNCE Loss
"""
if b_cos:
view1, view2 = F.normalize(view1, dim=1), F.normalize(view2, dim=1)
pos_score = (view1 @ view2.T) / temperature
score = torch.diag(F.log_softmax(pos_score, dim=1))
return -score.mean()(4) 학습 루프
def train(self):
"""
모델 학습 메서드
학습 흐름:
1. 각 에폭 시작 시 두 개의 증강된 그래프 생성
2. 미니배치 단위로:
- 원본 그래프로 추천 손실 계산
- 증강된 그래프들로 대조 학습 손실 계산
- 총 손실로 역전파 및 가중치 업데이트
3. 에폭 5 이후부터 검증 평가 수행
"""
# 모델을 GPU로 이동
model = self.model.cuda()
optimizer = torch.optim.Adam(model.parameters(), lr=self.lRate)
for epoch in range(self.maxEpoch):
# ==========================================================
# [Step 1] 매 에폭마다 새로운 뷰(View) 생성
# SGL이 느린 이유: 학습 할 때마다 그래프를 새로 그려야 함 (Reconstruction)
# ==========================================================
dropped_adj1 = model.graph_reconstruction() # 뷰 1 생성
dropped_adj2 = model.graph_reconstruction() # 뷰 2 생성
for n, batch in enumerate(next_batch_pairwise(self.data, self.batch_size)):
user_idx, pos_idx, neg_idx = batch
# [Step 2] Main Task: 추천 성능 학습 (BPR Loss)
# 원본 그래프를 사용하여 "유저가 좋아한 아이템"을 맞추도록 학습
rec_user_emb, rec_item_emb = model()
user_emb, pos_item_emb, neg_item_emb = rec_user_emb[user_idx], rec_item_emb[pos_idx], rec_item_emb[neg_idx]
rec_loss = bpr_loss(user_emb, pos_item_emb, neg_item_emb)
# [Step 3] Self-supervised Task: 대조 학습 (Contrastive Learning)
# 두 개의 증강된 그래프를 사용하여 노드끼리 가깝게, 다른 노드와는 멀게 학습
cl_loss = self.cl_rate * model.cal_cl_loss([user_idx,pos_idx],dropped_adj1,dropped_adj2)
# [Step 4] Joint Learning (최종 학습)
# 추천도 잘하고(rec_loss) + 본질도 잘 파악해라(cl_loss)
batch_loss = rec_loss + l2_reg_loss(self.reg, user_emb, pos_item_emb,neg_item_emb) + cl_loss
# 역전파 (Backpropagation)
optimizer.zero_grad()
batch_loss.backward()
optimizer.step()
if n % 100==0 and n>0:
print('training:', epoch + 1, 'batch', n, 'rec_loss:', rec_loss.item(), 'cl_loss', cl_loss.item())
with torch.no_grad():
self.user_emb, self.item_emb = self.model()
if epoch>=5:
self.fast_evaluation(epoch)
self.user_emb, self.item_emb = self.best_user_emb, self.best_item_emb1-5. SimGCL
- SGL은 매 에폭마다 그래프를 다시 만드는 과정이 느림
- SimGCL은 그래프 변형 대신, 임베딩 벡터에 노이즈를 더하는 방식 사용
- SGL은 구조적 증강 -> 인접행렬을 변형
- SimGCL은 특징 증강 -> user임베딩과 item임베딩에 변형
- 증강 방식: 임베딩 벡터 자체에 미세한 랜덤 노이즈를 추가하여 서로 다른 2개의 뷰(View 1, View 2)를 생성 (model.forward(perturbed=True))
- 대조학습은 SGL과 동일
- SGL 대비 매 에폭마다 그래프 생성 과정이 없으므로 학습 속도가 빠르고 메모리 효율적임
(1) 노이즈 주입
def forward(self, perturbed=False):
"""
[SimGCL의 핵심 1] 노이즈 주입을 통한 임베딩 증강
Args:
perturbed (bool): True일 경우 대조 학습을 위해 노이즈를 추가함
"""
# 1. 초기 임베딩 가져오기
ego_embeddings = torch.cat([self.embedding_dict['user_emb'],
self.embedding_dict['item_emb']], 0)
all_embeddings = []
# 2. 레이어별 전파 (Graph Propagation)
for k in range(self.n_layers):
# 기본 LightGCN 전파: 이웃 정보 집계
ego_embeddings = torch.sparse.mm(self.sparse_norm_adj, ego_embeddings)
# -----------------------------------------------------------
# [여기가 SGL과 다른 점!]
# 그래프 구조를 바꾸는 대신, 전파된 임베딩에 '노이즈'를 섞습니다.
# -----------------------------------------------------------
if perturbed:
# (1) 랜덤 노이즈 생성 (0~1 사이 값)
random_noise = torch.rand_like(ego_embeddings).cuda()
# (2) 노이즈 스케일링 및 주입
# torch.sign(ego): 원본 임베딩의 부호(방향)는 유지하되,
# F.normalize(noise): 노이즈의 크기를 정규화하고,
# self.eps: 아주 작은 값(예: 0.1)만큼만 흔들어줍니다.
# 결과: "원래 위치에서 아주 살짝만 벗어난 임베딩" 생성
ego_embeddings += torch.sign(ego_embeddings) * F.normalize(random_noise, dim=-1) * self.eps
all_embeddings.append(ego_embeddings)
# 3. 레이어 합치기 (Mean Aggregation)
all_embeddings = torch.stack(all_embeddings, dim=1)
all_embeddings = torch.mean(all_embeddings, dim=1)
user_all_embeddings, item_all_embeddings = torch.split(all_embeddings, [self.data.user_num, self.data.item_num])
return user_all_embeddings, item_all_embeddings(2) 대조 학습 수행
def cal_cl_loss(self, idx):
"""
[SimGCL의 핵심 2] 노이즈 뷰를 이용한 InfoNCE Loss 계산
SGL은 그래프를 다시 만들어서 forward를 돌려야 했지만,
SimGCL은 단순히 forward(perturbed=True)를 두 번 호출하면 끝입니다.
"""
# 현재 배치의 유저/아이템 인덱스
u_idx = torch.unique(torch.Tensor(idx[0]).type(torch.long)).cuda()
i_idx = torch.unique(torch.Tensor(idx[1]).type(torch.long)).cuda()
# [View 1 생성] 첫 번째 랜덤 노이즈가 섞인 임베딩
user_view_1, item_view_1 = self.model(perturbed=True)
# [View 2 생성] 두 번째 랜덤 노이즈가 섞인 임베딩 (노이즈 값이 다름)
user_view_2, item_view_2 = self.model(perturbed=True)
# [InfoNCE Loss]
# 같은 유저(u_idx)의 view1과 view2는 서로 당기고(Pull),
# 다른 유저들과는 서로 밀어냄(Push).
# 이를 통해 임베딩 공간이 고르게 퍼지고(Uniformity), 표현력이 강해짐.
user_cl_loss = InfoNCE(user_view_1[u_idx], user_view_2[u_idx], 0.2)
item_cl_loss = InfoNCE(item_view_1[i_idx], item_view_2[i_idx], 0.2)
return user_cl_loss + item_cl_loss(3) 핵심 함수 InfoNCE
def InfoNCE(view1, view2, temperature: float, b_cos: bool = True):
"""
InfoNCE (Information Noise Contrastive Estimation) 손실 함수
대조 학습(Contrastive Learning)의 핵심 손실 함수로,
같은 샘플의 서로 다른 뷰(양성 쌍)는 가깝게,
다른 샘플의 뷰(음성 쌍)는 멀게 학습시킵니다.
Args:
view1: (torch.Tensor - N x D)
view2: (torch.Tensor - N x D)
temperature: float
b_cos (bool)
Return: Average InfoNCE Loss
"""
if b_cos:
view1, view2 = F.normalize(view1, dim=1), F.normalize(view2, dim=1)
pos_score = (view1 @ view2.T) / temperature
score = torch.diag(F.log_softmax(pos_score, dim=1))
return -score.mean()** github 링크
-
https://github.com/Coder-Yu/SELFRec/tree/main (SGL, SimGCL)
하루살이