Case Study 1: StreamRec GNN Collaborative Filtering — From Interaction Matrix to Interaction Graph

Context

StreamRec's recommendation system has evolved through this book's progressive project. In Chapter 1, we decomposed the user-item interaction matrix with SVD, discovering latent factors that explain co-consumption patterns. In Chapter 13, we trained separate encoder towers for users and items, learning embeddings in a shared space via contrastive learning. Both approaches treat user-item interactions as entries in a matrix — independent observations to be predicted or factored.

But interactions are not independent. They form a graph. User A watches Documentary X. User B also watches Documentary X and Jazz Concert Y. User C watches Jazz Concert Y and Literary Reading Z. The chain A-X-B-Y-C-Z connects User A to Literary Reading Z through a path of shared tastes, even though User A has never encountered literary content. Matrix factorization can capture some of this through the latent factor structure (if documentary fans and jazz fans overlap in factor space), but it does not explicitly model the multi-hop paths that carry collaborative signal.

This case study builds a GNN-based collaborative filter on the StreamRec user-item bipartite graph and compares it against the matrix factorization and two-tower baselines established in earlier chapters.

The Bipartite Interaction Graph

The StreamRec interaction data is naturally a bipartite graph: users on one side, items on the other, edges representing engagement. We construct this graph from the same data used in Chapter 1.

import torch
import torch.nn as nn
import torch.nn.functional as F
import numpy as np
from torch_geometric.data import Data
from torch_geometric.nn import MessagePassing
from torch_geometric.utils import degree
from typing import Dict, Tuple, List


def build_interaction_graph(
    n_users: int = 10000,
    n_items: int = 5000,
    n_latent_true: int = 15,
    density: float = 0.008,
    noise_std: float = 0.3,
    seed: int = 42,
    val_ratio: float = 0.1,
    test_ratio: float = 0.1,
) -> Dict[str, torch.Tensor]:
    """Build the StreamRec bipartite interaction graph.

    Generates user-item interactions from a low-rank model (matching
    the Chapter 1 setup) and constructs a PyG-compatible graph with
    train/val/test edge splits.

    Args:
        n_users: Number of users.
        n_items: Number of items.
        n_latent_true: Number of true latent factors.
        density: Fraction of observed interactions.
        noise_std: Observation noise.
        seed: Random seed.
        val_ratio: Fraction of edges for validation.
        test_ratio: Fraction of edges for testing.

    Returns:
        Dictionary with edge indices, node counts, and split masks.
    """
    rng = np.random.default_rng(seed)

    # Generate low-rank interaction structure (same as Chapter 1)
    singular_values = np.array([
        100 * np.exp(-0.3 * i) for i in range(n_latent_true)
    ])
    U_true = rng.standard_normal((n_users, n_latent_true))
    V_true = rng.standard_normal((n_items, n_latent_true))
    U_true, _ = np.linalg.qr(U_true)
    V_true, _ = np.linalg.qr(V_true)
    U_true = U_true[:, :n_latent_true]
    V_true = V_true[:, :n_latent_true]
    R_true = U_true @ np.diag(singular_values) @ V_true.T

    # Scale to [1, 5]
    R_true = 1 + 4 * (R_true - R_true.min()) / (R_true.max() - R_true.min())

    # Sample observed interactions
    mask = rng.random((n_users, n_items)) < density
    observed_users, observed_items = np.where(mask)

    # Binarize: interaction exists or not (for implicit feedback)
    n_interactions = len(observed_users)
    print(f"Graph: {n_users} users, {n_items} items, {n_interactions} interactions")
    print(f"Density: {density:.4f}, Avg interactions/user: {n_interactions/n_users:.1f}")

    # Shuffle and split edges
    perm = rng.permutation(n_interactions)
    n_test = int(n_interactions * test_ratio)
    n_val = int(n_interactions * val_ratio)
    n_train = n_interactions - n_val - n_test

    train_idx = perm[:n_train]
    val_idx = perm[n_train:n_train + n_val]
    test_idx = perm[n_train + n_val:]

    # Build edge indices (item indices offset by n_users)
    def make_edges(idx: np.ndarray) -> torch.Tensor:
        u = torch.tensor(observed_users[idx], dtype=torch.long)
        i = torch.tensor(observed_items[idx] + n_users, dtype=torch.long)
        # Undirected: add both directions
        src = torch.cat([u, i])
        dst = torch.cat([i, u])
        return torch.stack([src, dst])

    return {
        "n_users": n_users,
        "n_items": n_items,
        "n_nodes": n_users + n_items,
        "train_edges": make_edges(train_idx),
        "val_edges": make_edges(val_idx),
        "test_edges": make_edges(test_idx),
        "val_user_item": (
            torch.tensor(observed_users[val_idx], dtype=torch.long),
            torch.tensor(observed_items[val_idx], dtype=torch.long),
        ),
        "test_user_item": (
            torch.tensor(observed_users[test_idx], dtype=torch.long),
            torch.tensor(observed_items[test_idx], dtype=torch.long),
        ),
    }

The Model: LightGCN

For collaborative filtering on bipartite graphs, the most effective architecture is not the standard GCN but LightGCN (He et al., 2020) — a simplified variant that removes feature transformations and nonlinearities entirely. The rationale: in pure collaborative filtering (no side features), the only input to the GNN is learnable embedding vectors. Applying $\mathbf{W}$ and $\sigma$ to learnable embeddings is redundant — the embeddings can absorb the transformation directly. LightGCN keeps only the neighborhood aggregation.

class LightGCNConv(MessagePassing):
    """LightGCN convolution: normalized neighborhood aggregation only.

    No feature transformation, no nonlinearity, no self-loops.
    The layer simply averages neighbors' embeddings with symmetric
    degree normalization.

    This simplification is optimal for collaborative filtering where
    the only node features are learnable embeddings.
    """

    def __init__(self) -> None:
        super().__init__(aggr="add")

    def forward(
        self, x: torch.Tensor, edge_index: torch.Tensor
    ) -> torch.Tensor:
        """Propagate embeddings through the graph.

        Args:
            x: Node embeddings, shape (N, d).
            edge_index: Edge indices, shape (2, E).

        Returns:
            Aggregated embeddings, shape (N, d).
        """
        row, col = edge_index
        deg = degree(col, x.size(0), dtype=x.dtype)
        deg_inv_sqrt = deg.pow(-0.5)
        deg_inv_sqrt[deg_inv_sqrt == float("inf")] = 0.0
        norm = deg_inv_sqrt[row] * deg_inv_sqrt[col]
        return self.propagate(edge_index, x=x, norm=norm)

    def message(
        self, x_j: torch.Tensor, norm: torch.Tensor
    ) -> torch.Tensor:
        return norm.view(-1, 1) * x_j


class LightGCN(nn.Module):
    """LightGCN for collaborative filtering.

    Learns user and item embeddings, propagates them through the
    bipartite interaction graph, and averages across layers to
    produce final representations.

    The final embedding is the layer-wise mean:
        e_final = (1 / (L + 1)) * sum_{l=0}^{L} e^{(l)}

    This multi-scale combination allows the model to capture both
    local (1-hop) and broader (K-hop) collaborative signals.

    Args:
        n_users: Number of users.
        n_items: Number of items.
        embedding_dim: Embedding dimensionality.
        num_layers: Number of LightGCN propagation layers.
    """

    def __init__(
        self,
        n_users: int,
        n_items: int,
        embedding_dim: int = 64,
        num_layers: int = 3,
    ) -> None:
        super().__init__()
        self.n_users = n_users
        self.n_items = n_items
        self.num_layers = num_layers

        # Learnable embeddings (layer 0)
        self.user_embedding = nn.Embedding(n_users, embedding_dim)
        self.item_embedding = nn.Embedding(n_items, embedding_dim)

        # LightGCN layers (no learnable parameters)
        self.convs = nn.ModuleList([LightGCNConv() for _ in range(num_layers)])

        self._init_weights()

    def _init_weights(self) -> None:
        nn.init.xavier_uniform_(self.user_embedding.weight)
        nn.init.xavier_uniform_(self.item_embedding.weight)

    def get_embeddings(
        self, edge_index: torch.Tensor
    ) -> Tuple[torch.Tensor, torch.Tensor]:
        """Compute user and item embeddings via LightGCN propagation.

        Args:
            edge_index: Training edges, shape (2, E).

        Returns:
            Tuple of (user_embeddings, item_embeddings).
        """
        # Concatenate user and item embeddings
        x = torch.cat([
            self.user_embedding.weight,
            self.item_embedding.weight,
        ])

        # Collect all layers' embeddings
        all_embeddings = [x]
        for conv in self.convs:
            x = conv(x, edge_index)
            all_embeddings.append(x)

        # Layer-wise mean
        all_embeddings = torch.stack(all_embeddings, dim=0)
        final = all_embeddings.mean(dim=0)

        user_emb = final[:self.n_users]
        item_emb = final[self.n_users:]
        return user_emb, item_emb

    def forward(
        self,
        edge_index: torch.Tensor,
        users: torch.Tensor,
        pos_items: torch.Tensor,
        neg_items: torch.Tensor,
    ) -> Tuple[torch.Tensor, torch.Tensor, torch.Tensor]:
        """Compute BPR loss components.

        Args:
            edge_index: Training graph edges.
            users: User indices for this batch, shape (B,).
            pos_items: Positive item indices, shape (B,).
            neg_items: Negative item indices, shape (B,).

        Returns:
            Tuple of (user_emb, pos_item_emb, neg_item_emb).
        """
        user_emb, item_emb = self.get_embeddings(edge_index)
        return user_emb[users], item_emb[pos_items], item_emb[neg_items]

Training with BPR Loss

Bayesian Personalized Ranking (Rendle et al., 2009) optimizes the pairwise ranking: for each user, a positive (interacted) item should be scored higher than a negative (non-interacted) item.

def bpr_loss(
    user_emb: torch.Tensor,
    pos_emb: torch.Tensor,
    neg_emb: torch.Tensor,
    reg_weight: float = 1e-4,
) -> torch.Tensor:
    """Compute BPR loss with L2 regularization.

    BPR loss = -log(sigma(pos_score - neg_score)) + reg * ||embeddings||^2

    Args:
        user_emb: User embeddings, shape (B, d).
        pos_emb: Positive item embeddings, shape (B, d).
        neg_emb: Negative item embeddings, shape (B, d).
        reg_weight: L2 regularization weight.

    Returns:
        Scalar BPR loss.
    """
    pos_score = (user_emb * pos_emb).sum(dim=1)
    neg_score = (user_emb * neg_emb).sum(dim=1)
    bpr = -F.logsigmoid(pos_score - neg_score).mean()

    reg = reg_weight * (
        user_emb.norm(dim=1).pow(2).mean()
        + pos_emb.norm(dim=1).pow(2).mean()
        + neg_emb.norm(dim=1).pow(2).mean()
    )
    return bpr + reg


def train_lightgcn(
    graph_data: Dict[str, torch.Tensor],
    embedding_dim: int = 64,
    num_layers: int = 3,
    epochs: int = 100,
    batch_size: int = 2048,
    lr: float = 1e-3,
    reg_weight: float = 1e-4,
) -> Dict[str, List[float]]:
    """Train LightGCN on the StreamRec interaction graph.

    Args:
        graph_data: Output of build_interaction_graph().
        embedding_dim: Embedding size.
        num_layers: Number of propagation layers.
        epochs: Number of training epochs.
        batch_size: Batch size for BPR sampling.
        lr: Learning rate.
        reg_weight: L2 regularization weight.

    Returns:
        Dictionary with training loss history and evaluation metrics.
    """
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")

    model = LightGCN(
        n_users=graph_data["n_users"],
        n_items=graph_data["n_items"],
        embedding_dim=embedding_dim,
        num_layers=num_layers,
    ).to(device)
    optimizer = torch.optim.Adam(model.parameters(), lr=lr)
    train_edges = graph_data["train_edges"].to(device)

    # Build training interaction set for negative sampling
    train_user_item = set()
    src, dst = graph_data["train_edges"]
    n_users = graph_data["n_users"]
    for s, d in zip(src.tolist(), dst.tolist()):
        if s < n_users:  # user -> item edge
            train_user_item.add((s, d - n_users))

    loss_history = []

    for epoch in range(epochs):
        model.train()

        # Sample batch: users, positive items, negative items
        users_list = []
        pos_items_list = []
        neg_items_list = []

        interactions = list(train_user_item)
        perm = np.random.permutation(len(interactions))[:batch_size]
        for idx in perm:
            u, i = interactions[idx]
            users_list.append(u)
            pos_items_list.append(i)
            # Random negative item (not interacted by this user)
            while True:
                neg_i = np.random.randint(graph_data["n_items"])
                if (u, neg_i) not in train_user_item:
                    break
            neg_items_list.append(neg_i)

        users = torch.tensor(users_list, dtype=torch.long, device=device)
        pos_items = torch.tensor(pos_items_list, dtype=torch.long, device=device)
        neg_items = torch.tensor(neg_items_list, dtype=torch.long, device=device)

        optimizer.zero_grad()
        u_emb, pi_emb, ni_emb = model(train_edges, users, pos_items, neg_items)
        loss = bpr_loss(u_emb, pi_emb, ni_emb, reg_weight=reg_weight)
        loss.backward()
        optimizer.step()

        loss_history.append(loss.item())

        if (epoch + 1) % 20 == 0:
            metrics = evaluate_recall_ndcg(
                model, train_edges, graph_data, device, k=20
            )
            print(
                f"Epoch {epoch+1:3d} | Loss: {loss.item():.4f} | "
                f"Recall@20: {metrics['recall@20']:.4f} | "
                f"NDCG@20: {metrics['ndcg@20']:.4f}"
            )

    return {"loss_history": loss_history}

Evaluation: Recall@K and NDCG@K

def evaluate_recall_ndcg(
    model: LightGCN,
    train_edges: torch.Tensor,
    graph_data: Dict[str, torch.Tensor],
    device: torch.device,
    k: int = 20,
) -> Dict[str, float]:
    """Evaluate recommendation quality with Recall@K and NDCG@K.

    For each test user, rank all non-interacted items by predicted
    score and check if the held-out test items appear in the top K.

    Args:
        model: Trained LightGCN model.
        train_edges: Training edge indices.
        graph_data: Full graph data dictionary.
        device: Computation device.
        k: Number of top recommendations to consider.

    Returns:
        Dictionary with recall@k and ndcg@k.
    """
    model.eval()
    with torch.no_grad():
        user_emb, item_emb = model.get_embeddings(train_edges)

    test_users, test_items = graph_data["test_user_item"]

    # Group test items by user
    user_test_items: Dict[int, List[int]] = {}
    for u, i in zip(test_users.tolist(), test_items.tolist()):
        user_test_items.setdefault(u, []).append(i)

    recalls = []
    ndcgs = []

    for user_id, true_items in user_test_items.items():
        # Score all items for this user
        scores = (user_emb[user_id] @ item_emb.T).cpu()

        # Mask out training items
        # (In production, you would maintain per-user interaction sets)
        _, top_k = torch.topk(scores, k)
        top_k_set = set(top_k.tolist())
        true_set = set(true_items)

        # Recall@K: fraction of true items in top K
        hits = len(top_k_set & true_set)
        recalls.append(hits / min(len(true_set), k))

        # NDCG@K: discounted cumulative gain
        dcg = 0.0
        for rank, item_id in enumerate(top_k.tolist()):
            if item_id in true_set:
                dcg += 1.0 / np.log2(rank + 2)
        idcg = sum(1.0 / np.log2(r + 2) for r in range(min(len(true_set), k)))
        ndcgs.append(dcg / idcg if idcg > 0 else 0.0)

    return {"recall@20": np.mean(recalls), "ndcg@20": np.mean(ndcgs)}

Results and Comparison

Running the pipeline:

graph_data = build_interaction_graph(n_users=10000, n_items=5000, density=0.008)
train_lightgcn(graph_data, embedding_dim=64, num_layers=3, epochs=100)

Expected approximate results (10K users, 5K items, density 0.008):

The GNN consistently outperforms both baselines. The improvement comes from explicit modeling of multi-hop collaborative paths: LightGCN's 3-layer propagation captures 3-hop user-item-user-item chains that matrix factorization and two-tower models must learn implicitly through their latent factor structure.

Why the Graph Matters

Three insights emerge from the comparison:

1. Multi-hop signal. The 2-hop path (user $\to$ item $\to$ user) captures taste similarity between users who share items. The 3-hop path (user $\to$ item $\to$ user $\to$ item) discovers items liked by taste-similar users. LightGCN explicitly propagates along these paths; matrix factorization captures the same signal only indirectly through latent factors.

2. Layer-wise aggregation is critical. The mean over layers $\frac{1}{L+1}\sum_\ell \mathbf{e}^{(\ell)}$ is not an afterthought — it is the key design choice. Layer 0 is the raw embedding (item identity). Layer 1 captures direct interactions (who rated this item). Layer 2 captures collaborative similarity (what other items did those users rate). Averaging combines all scales, giving each user's embedding contributions from both local and global collaborative structure.

3. Simplicity wins for collaborative filtering. LightGCN removes transformations and nonlinearities from GCN — and outperforms it. The lesson for production systems: when the only input features are learnable embeddings (no side information), the GNN's role is purely structural (propagate collaborative signal through the graph), and adding parameters to the propagation step hurts by introducing optimization difficulty without adding representational capacity.

Limitations

1. Cold start. LightGCN requires at least a few interactions to embed a user or item. New users with zero interactions have random embeddings. Chapter 13's two-tower model, which can use content features, handles cold start better. A hybrid approach — LightGCN for warm users, content-based for cold — is the production solution.

2. Scalability. Full-batch propagation on the entire graph limits scalability. For StreamRec's full 5M-user graph, mini-batch training with neighborhood sampling (GraphSAGE-style) is required.

3. Temporal dynamics. The interaction graph is static — it does not capture the temporal ordering of interactions. A user's recent behavior is more predictive than their behavior from two years ago, but LightGCN treats all interactions equally. Temporal GNNs (e.g., TGAT, TGN) address this by incorporating timestamps into message passing.