Conceptual Framework Implementation (Python)

The implementation below illustrates the core mechanisms of ACCF: adversarial perturbation generation, adversarial contrastive learning, instance-sensitive weight generation, and the bi-level optimization structure. This educational code demonstrates the key architectural concepts described in the paper.

# ─────────────────────────────────────────────────────────────────────────────
# ACCF: Adversarial Contrastive Collaborative Filtering
# Wu, Zhang, Fan, Ye · Knowledge-Based Systems 2026
# Conceptual implementation of adversarial CL and instance-sensitive BPR
# ─────────────────────────────────────────────────────────────────────────────

import torch
import torch.nn as nn
import torch.nn.functional as F
import numpy as np
from typing import Tuple, Optional
from dataclasses import dataclass


# ─── Section 1: Adversarial Perturbation Generator ───────────────────────────

class AdversarialPerturbation:
    """
    Generates adversarial perturbations using fast gradient method.
    Perturbations maximize instance-sensitive BPR loss within epsilon ball.
    """
    
    def __init__(self, epsilon: float = 0.5):
        self.epsilon = epsilon
        
    def generate(self, 
                 embeddings: torch.Tensor,
                 loss_fn,
                 *loss_args) -> torch.Tensor:
        """
        Compute Delta_adv = epsilon * Gamma / ||Gamma||
        where Gamma = dL/dDelta
        
        Args:
            embeddings: Initial embeddings E^(0) to perturb
            loss_fn: Function computing L_ibpr given embeddings
            loss_args: Additional arguments for loss computation
            
        Returns:
            perturbation: Adversarial perturbation Delta_adv
        """
        # Enable gradient computation for embeddings
        embeddings_pert = embeddings.clone().detach().requires_grad_(True)
        
        # Compute loss with perturbed embeddings
        loss = loss_fn(embeddings_pert, *loss_args)
        
        # Compute gradient
        loss.backward()
        grad = embeddings_pert.grad
        
        # Normalize and scale by epsilon
        grad_norm = torch.norm(grad, p=2, dim=-1, keepdim=True)
        perturbation = self.epsilon * grad / (grad_norm + 1e-8)
        
        return perturbation.detach()


# ─── Section 2: Instance-Sensitive Weight Generator ──────────────────────────

class InstanceWeightGenerator(nn.Module):
    """
    Two-layer MLP that generates importance weights for BPR triplets.
    Inputs: user, pos_item, neg_item embeddings plus their interactions.
    """
    
    def __init__(self, embed_dim: int, hidden_dim: int = 64):
        super(InstanceWeightGenerator, self).__init__()
        # Input: 5 * embed_dim (user, pos, neg, user*pos, user*neg)
        self.fc1 = nn.Linear(5 * embed_dim, hidden_dim)
        self.fc2 = nn.Linear(hidden_dim, 1)
        
    def forward(self,
                user_emb: torch.Tensor,
                pos_emb: torch.Tensor,
                neg_emb: torch.Tensor) -> torch.Tensor:
        """
        Compute delta_uij = softplus(w · relu(W · concat(features)))
        
        Returns:
            weights: Positive weights for each triplet
        """
        # Element-wise interactions
        user_pos_interact = user_emb * pos_emb
        user_neg_interact = user_emb * neg_emb
        
        # Concatenate all features
        features = torch.cat([
            user_emb, pos_emb, neg_emb,
            user_pos_interact, user_neg_interact
        ], dim=-1)
        
        # MLP forward
        h = F.relu(self.fc1(features))
        logits = self.fc2(h).squeeze(-1)
        
        # Softplus ensures positive weights
        weights = F.softplus(logits)
        return weights


# ─── Section 3: Adversarial Contrastive Learning Module ──────────────────────

class AdversarialContrastiveLearning:
    """
    Implements adversarial contrastive learning for user and item representations.
    Maximizes agreement between original and adversarially perturbed views.
    """
    
    def __init__(self, temperature: float = 0.5):
        self.temperature = temperature
        
    def contrastive_loss(self,
                         original_view: torch.Tensor,
                         perturbed_view: torch.Tensor,
                         all_negatives: torch.Tensor) -> torch.Tensor:
        """
        Compute InfoNCE-style loss for adversarial contrastive learning.
        
        L_ACL = -log(exp(sim(z, z_adv)/tau) / sum(exp(sim(z, z_adv)/tau) + negatives))
        
        Args:
            original_view: Original embeddings [batch_size, dim]
            perturbed_view: Adversarially perturbed embeddings [batch_size, dim]
            all_negatives: All other nodes as negatives [num_negatives, dim]
            
        Returns:
            loss: Scalar contrastive loss
        """
        # Positive similarities
        pos_sim = F.cosine_similarity(original_view, perturbed_view, dim=-1) / self.temperature
        
        # Negative similarities with all other nodes
        # [batch_size, 1, dim] vs [1, num_negatives, dim] -> [batch_size, num_negatives]
        orig_expanded = original_view.unsqueeze(1)
        neg_expanded = all_negatives.unsqueeze(0)
        
        neg_sim = F.cosine_similarity(
            orig_expanded.expand(-1, all_negatives.size(0), -1),
            neg_expanded.expand(original_view.size(0), -1, -1),
            dim=-1
        ) / self.temperature
        
        # Numerical stability: subtract max
        logits = torch.cat([pos_sim.unsqueeze(1), neg_sim], dim=1)
        logits_max, _ = torch.max(logits, dim=1, keepdim=True)
        logits = logits - logits_max.detach()
        
        # Softmax and negative log likelihood
        exp_logits = torch.exp(logits)
        log_prob = logits[:, 0] - torch.log(exp_logits.sum(dim=1) + 1e-10)
        
        loss = -log_prob.mean()
        return loss


# ─── Section 4: LightGCN Backbone with Adversarial Training ──────────────────

class LightGCNLayer(nn.Module):
    ">Single LightGCN propagation layer: symmetric normalization aggregation"""
    
    def __init__(self):
        super(LightGCNLayer, self).__init__()
        
    def forward(self, x: torch.Tensor, adj_norm: torch.Tensor) -> torch.Tensor:
        ">Propagate: X^(l+1) = D^-1/2 A D^-1/2 X^(l)"""
        return torch.sparse.mm(adj_norm, x)


class ACCFModel(nn.Module):
    """
    Complete ACCF model integrating:
    - LightGCN backbone for collaborative filtering
    - Adversarial contrastive learning module
    - Instance-sensitive BPR loss with bi-level optimization
    """
    
    def __init__(self,
                 num_users: int,
                 num_items: int,
                 embed_dim: int = 64,
                 num_layers: int = 3,
                 epsilon: float = 0.5,
                 lambda_adv: float = 0.1,
                 lambda_cl: float = 0.05):
        super(ACCFModel, self).__init__()
        
        self.num_users = num_users
        self.num_items = num_items
        self.embed_dim = embed_dim
        self.num_layers = num_layers
        self.lambda_adv = lambda_adv
        self.lambda_cl = lambda_cl
        
        # Initial embeddings (trainable parameters Theta)
        self.user_embedding = nn.Embedding(num_users, embed_dim)
        self.item_embedding = nn.Embedding(num_items, embed_dim)
        nn.init.xavier_uniform_(self.user_embedding.weight)
        nn.init.xavier_uniform_(self.item_embedding.weight)
        
        # LightGCN propagation layers
        self.gcn_layers = nn.ModuleList([LightGCNLayer() for _ in range(num_layers)])
        
        # Instance-sensitive weight generator (parameters Lambda)
        self.weight_generator = InstanceWeightGenerator(embed_dim)
        
        # Adversarial components
        self.adv_perturbation = AdversarialPerturbation(epsilon)
        self.adv_cl = AdversarialContrastiveLearning()
        
    def propagate(self, 
                  user_emb: torch.Tensor,
                  item_emb: torch.Tensor,
                  adj_norm: torch.Tensor) -> Tuple[torch.Tensor, torch.Tensor]:
        ">Propagate embeddings through LightGCN layers"""
        # Concatenate user and item embeddings
        all_emb = torch.cat([user_emb, item_emb], dim=0)
        embs = [all_emb]
        
        # Multi-layer propagation
        for layer in self.gcn_layers:
            all_emb = layer(all_emb, adj_norm)
            embs.append(all_emb)
        
        # Mean pooling of all layers
        final_emb = torch.stack(embs, dim=0).mean(dim=0)
        users_final, items_final = torch.split(final_emb, [self.num_users, self.num_items])
        
        return users_final, items_final
    
    def bpr_loss(self,
                 users: torch.Tensor,
                 pos_items: torch.Tensor,
                 neg_items: torch.Tensor,
                 user_emb: torch.Tensor,
                 item_emb: torch.Tensor,
                 use_instance_weights: bool = True) -> Tuple[torch.Tensor, torch.Tensor]:
        """
        Compute BPR loss with optional instance-sensitive weighting.
        
        Returns:
            loss: Scalar BPR loss
            weights: Instance weights (for monitoring)
        """
        # Look up embeddings
        user_e = user_emb[users]
        pos_e = item_emb[pos_items]
        neg_e = item_emb[neg_items]
        
        # BPR scores
        pos_scores = (user_e * pos_e).sum(dim=-1)
        neg_scores = (user_e * neg_e).sum(dim=-1)
        
        # Base BPR loss
        bpr_losses = -F.logsigmoid(pos_scores - neg_scores)
        
        if use_instance_weights:
            # Generate instance-sensitive weights
            weights = self.weight_generator(user_e, pos_e, neg_e)
            loss = (weights * bpr_losses).sum() / weights.sum()
        else:
            weights = torch.ones_like(bpr_losses)
            loss = bpr_losses.mean()
            
        return loss, weights
    
    def forward(self,
                users: torch.Tensor,
                pos_items: torch.Tensor,
                neg_items: torch.Tensor,
                adj_norm: torch.Tensor,
                compute_adversarial: bool = True) -> dict:
        ">
        Full forward pass with adversarial training and contrastive learning.
        
        Returns dict with losses for bi-level optimization:
        - L_inner: For embedding parameters Theta (includes adversarial)
        - L_outer: For weight generator parameters Lambda
        """
        # Get initial embeddings
        user_init = self.user_embedding.weight
        item_init = self.item_embedding.weight
        
        # === Standard forward pass ===
        user_emb, item_emb = self.propagate(user_init, item_init, adj_norm)
        
        # Standard BPR loss (for outer level validation)
        loss_bpr_clean, weights = self.bpr_loss(
            users, pos_items, neg_items, user_emb, item_emb, 
            use_instance_weights=True
        )
        
        # === Adversarial forward pass ===
        if compute_adversarial:
            # Generate adversarial perturbations on initial embeddings
            # (In practice, compute gradient of L_bpr w.r.t. initial embeddings)
            user_init_adv = user_init + self.adv_perturbation.generate(
                user_init, 
                lambda x: self.bpr_loss(users, pos_items, neg_items,
                    self.propagate(x, item_init, adj_norm)[0],
                    item_emb, False)[0].mean()
            )
            
            item_init_adv = item_init + self.adv_perturbation.generate(
                item_init,
                lambda x: self.bpr_loss(users, pos_items, neg_items,
                    user_emb,
                    self.propagate(user_init, x, adj_norm)[1],
                    False)[0].mean()
            )
            
            # Propagate adversarial embeddings
            user_emb_adv, item_emb_adv = self.propagate(
                user_init + (user_init_adv - user_init).detach(),
                item_init + (item_init_adv - item_init).detach(),
                adj_norm
            )
            
            # Adversarial BPR loss
            loss_bpr_adv, _ = self.bpr_loss(
                users, pos_items, neg_items, user_emb_adv, item_emb_adv,
                use_instance_weights=True
            )
            
            # Adversarial contrastive learning loss
            loss_acl_user = self.adv_cl.contrastive_loss(
                user_emb, user_emb_adv, user_emb
            )
            loss_acl_item = self.adv_cl.contrastive_loss(
                item_emb, item_emb_adv, item_emb
            )
            loss_acl = loss_acl_user + loss_acl_item
            
            # Combined adversarial loss (Eq. 3 in paper)
            loss_apr = loss_bpr_clean + self.lambda_adv * loss_bpr_adv
            loss_inner = loss_apr + self.lambda_cl * loss_acl
        else:
            loss_inner = loss_bpr_clean
            loss_apr = loss_bpr_clean
            loss_acl = torch.tensor(0.0)
        
        # === Bi-level optimization structure ===
        # Outer level: weight generator only
        # Inner level: embeddings with adversarial training
        
        return {
            'L_inner': loss_inner,           # For Theta (embeddings)
            'L_outer': loss_apr,            # For Lambda (weight gen)
            'L_bpr': loss_bpr_clean,
            'L_bpr_adv': loss_bpr_adv if compute_adversarial else None,
            'L_acl': loss_acl,
            'weights': weights
        }


# ─── Section 5: Bi-Level Optimizer ───────────────────────────────────────────

class BiLevelOptimizer:
    ">
    Implements alternating optimization for bi-level problem:
    - Inner: optimize Theta (embeddings) with fixed Lambda
    - Outer: optimize Lambda (weight generator) with fixed Theta
    """
    
    def __init__(self,
                 model: ACCFModel,
                 lr_inner: float = 0.001,
                 lr_outer: float = 0.001,
                 weight_decay: float = 1e-4):
        self.model = model
        
        # Separate parameter groups
        self.theta_params = [
            model.user_embedding.weight,
            model.item_embedding.weight
        ]
        self.lambda_params = list(model.weight_generator.parameters())
        
        # Separate optimizers
        self.opt_inner = torch.optim.Adam(
            self.theta_params, lr=lr_inner, weight_decay=weight_decay
        )
        self.opt_outer = torch.optim.Adam(
            self.lambda_params, lr=lr_outer, weight_decay=weight_decay
        )
        
    def step(self, batch: dict, adj_norm: torch.Tensor):
        ">
        Single bi-level optimization step:
        1. Update Theta (embeddings) using L_inner
        2. Update Lambda (weight generator) using L_outer
        """
        users = batch['users']
        pos_items = batch['pos_items']
        neg_items = batch['neg_items']
        
        # === Inner level: Update embeddings ===
        self.opt_inner.zero_grad()
        
        # Forward with adversarial training
        outputs = self.model(users, pos_items, neg_items, adj_norm, 
                              compute_adversarial=True)
        
        loss_inner = outputs['L_inner']
        loss_inner.backward()
        self.opt_inner.step()
        
        # === Outer level: Update weight generator ===
        # Detach embeddings to prevent gradients flowing back
        self.opt_outer.zero_grad()
        
        # Recompute with detached embeddings
        with torch.no_grad():
            user_emb, item_emb = self.model.propagate(
                self.model.user_embedding.weight,
                self.model.item_embedding.weight,
                adj_norm
            )
        
        # Compute L_outer for weight generator
        loss_bpr, _ = self.model.bpr_loss(
            users, pos_items, neg_items, user_emb, item_emb,
            use_instance_weights=True
        )
        
        loss_outer = loss_bpr
        loss_outer.backward()
        self.opt_outer.step()
        
        return {
            'loss_inner': loss_inner.item(),
            'loss_outer': loss_outer.item(),
            'mean_weight': outputs['weights'].mean().item()
        }


# ─── Section 6: Demonstration ────────────────────────────────────────────────

if __name__ == "__main__":
    """
    Demonstration of ACCF training pipeline.
    Shows initialization, bi-level optimization, and evaluation.
    """
    
    print("=" * 70)
    print("ACCF: Adversarial Contrastive Collaborative Filtering")
    print("Knowledge-Based Systems 2026 - Wu, Zhang, Fan, Ye")
    print("=" * 70)
    
    # Configuration
    NUM_USERS = 31668   # Yelp dataset scale
    NUM_ITEMS = 38048
    EMBED_DIM = 64
    NUM_LAYERS = 3
    
    # Initialize model
    print("\n[1] Initializing ACCF model...")
    model = ACCFModel(
        num_users=NUM_USERS,
        num_items=NUM_ITEMS,
        embed_dim=EMBED_DIM,
        num_layers=NUM_LAYERS,
        epsilon=0.5,      # Adversarial perturbation magnitude
        lambda_adv=0.1,  # Adversarial loss weight
        lambda_cl=0.05   # Contrastive loss weight
    )
    
    total_params = sum(p.numel() for p in model.parameters())
    print(f"    Total parameters: {total_params:,}")
    print(f"    Embedding params: {NUM_USERS * EMBED_DIM + NUM_ITEMS * EMBED_DIM:,}")
    print(f"    Weight generator params: {sum(p.numel() for p in model.weight_generator.parameters()):,}")
    
    # Initialize bi-level optimizer
    print("\n[2] Initializing bi-level optimizer...")
    optimizer = BiLevelOptimizer(model, lr_inner=0.001, lr_outer=0.001)
    
    # Simulate training batch
    print("\n[3] Simulating training step...")
    batch_size = 1024
    batch = {
        'users': torch.randint(0, NUM_USERS, (batch_size,)),
        'pos_items': torch.randint(0, NUM_ITEMS, (batch_size,)),
        'neg_items': torch.randint(0, NUM_ITEMS, (batch_size,))
    }
    
    # Create dummy normalized adjacency (in practice, from interaction graph)
    # Shape: [NUM_USERS + NUM_ITEMS, NUM_USERS + NUM_ITEMS]
    adj_indices = torch.randint(0, NUM_USERS + NUM_ITEMS, (2, 100000))
    adj_values = torch.ones(adj_indices.size(1))
    adj_norm = torch.sparse_coo_tensor(
        adj_indices, adj_values, 
        size=(NUM_USERS + NUM_ITEMS, NUM_USERS + NUM_ITEMS)
    ).coalesce()
    
    # Training step
    metrics = optimizer.step(batch, adj_norm)
    print(f"    Inner loss (embeddings): {metrics['loss_inner']:.4f}")
    print(f"    Outer loss (weight gen): {metrics['loss_outer']:.4f}")
    print(f"    Mean instance weight: {metrics['mean_weight']:.4f}")
    
    # Architecture summary
    print("\n[4] ACCF Architecture Summary:")
    print("    Backbone: LightGCN with {NUM_LAYERS} layers")
    print("    Adversarial Component:")
    print("      - Perturbation: Fast gradient method, epsilon={epsilon}")
    print("      - Loss: L_apr = L_bpr + lambda_adv * L_bpr(adv)")
    print("      - Contrastive: L_acl for user/item views")
    print("    Instance-Sensitive Component:")
    print("      - Input: [u, i, j, u*i, u*j] embeddings")
    print("      - Architecture: 2-layer MLP with softplus output")
    print("      - Loss: L_ibpr = sum(delta_uij * log(sigmoid(pos-neg)))")
    print(    Optimization: Bi-level alternating updates")
    print("      - Inner: Theta (embeddings) via L_inner")
    print("      - Outer: Lambda (weight gen) via L_outer")
    
    print("\n" + "=" * 70)
    print("ACCF enhances GNN-based recommenders through adversarial")
    print("robustness and dynamic instance weighting.")
    print("=" * 70)