TGraphX › Insights › Building a Graph Pipeline with sklearn-like Estimators
← Back to Insights
Tutorial · · 8 min read

Building a Graph Pipeline with sklearn-like Estimators

Target keyword: graph machine learning pipeline sklearn

Building a Graph Pipeline with sklearn-like Estimators

Machine learning practitioners trained on tabular data have strong intuitions about pipelines: fit a scaler, transform features, pass them to a classifier, evaluate. scikit-learn made this pattern so ergonomic that it became the default mental model for supervised learning. TGraphX brings the same design pattern to graph neural networks, exposing an estimators module whose classes implement fit, transform, and predict interfaces that compose naturally into structured experiment workflows.

This article walks through the sklearn-style pipeline design in TGraphX, shows how to combine preprocessing and GNN layers in a reproducible sequence, and explains why the pipeline abstraction pays dividends when you need to iterate on experiments quickly.

What This Builds On

This tutorial assumes you have TGraphX installed and understand the basics of GNN training — forward passes, loss functions, and the tgraphx.Graph object. Familiarity with scikit-learn's Pipeline and BaseEstimator classes will help but is not required. The reproducibility article covers complementary ideas about setting seeds, tracking runs, and ensuring consistent results — all of which become easier when experiment logic lives inside estimators.

Why a Pipeline Abstraction for Graphs?

The core problem with ad-hoc GNN scripts is that preprocessing and model training become entangled. When you change a preprocessing step — say, switching from degree-normalized features to raw features — you need to trace through a script looking for every place that step affects downstream state. In a pipeline, the boundary between steps is explicit: each stage exposes fit (learns parameters from training data), transform (applies those parameters to new data), and optionally predict (produces task outputs).

For graph learning, the equivalent stages are:

  1. Feature normalization — zero-mean, unit-variance normalization of node features, fit on the training subgraph, applied to validation and test.
  2. Graph construction — building the edge_index from raw data, determining k for k-NN graphs.
  3. Embedding — running a GNN or non-parametric method to produce node embeddings.
  4. Classification or regression — a final linear head or separate classifier.

When these stages are encapsulated, swapping step 3 from a GraphSAGE encoder to a GIN encoder is a one-line change that does not affect any other stage.

TGraphX Estimators: The Base Interface

TGraphX's tgraphx.estimators module provides a set of classes following the sklearn convention. The base class exposes:

  • fit(graph, labels=None) — learn any parameters (normalizer statistics, model weights, etc.)
  • transform(graph) — apply learned parameters and return a modified graph or feature matrix
  • predict(graph) — for supervised estimators, run inference and return predictions
  • fit_transform(graph, labels=None) — convenience method combining fit and transform

Below is a minimal example of using the label propagation estimator from tgraphx.estimators:

python
import torch
        from tgraphx import Graph
        from tgraphx.estimators.label_propagation import LabelPropagationEstimator
        
        g = Graph(
            node_features=torch.randn(200, 32),
            edge_index=torch.randint(0, 200, (2, 800), dtype=torch.long),
        )
        
        labels = torch.full((200,), -1, dtype=torch.long)
        labels[:20] = torch.randint(0, 3, (20,))   # 20 labeled nodes
        
        model = LabelPropagationEstimator(alpha=0.9, num_layers=100)
        model.fit(g, labels)
        preds = model.predict(g)
        print(preds[:5])

The estimator stores no state outside of what fit computes. Calling fit twice resets the estimator. This stateless design makes it safe to use inside cross-validation loops.

Building a Manual Pipeline

TGraphX does not ship a GraphPipeline class that mirrors sklearn's Pipeline exactly. Instead, the pipeline pattern is achieved by chaining estimator calls. Here is a three-stage pipeline: normalize features, embed with a GNN, classify with a linear head.

python
import torch
        import torch.nn as nn
        import torch.nn.functional as F
        from tgraphx import Graph
        from tgraphx.layers.sage import TensorGraphSAGELayer
        
        # --- Stage 1: Feature normalization (sklearn-style, fit on train split) ---
        class NodeFeatureNormalizer:
            def __init__(self):
                self.mean = None
                self.std = None
        
            def fit(self, x_train):
                self.mean = x_train.mean(dim=0)
                self.std = x_train.std(dim=0).clamp(min=1e-6)
        
            def transform(self, x):
                return (x - self.mean) / self.std
        
        # --- Stage 2: GNN encoder ---
        class SAGEEncoder(nn.Module):
            def __init__(self, in_dim, hidden_dim, out_dim):
                super().__init__()
                self.layer1 = TensorGraphSAGELayer(in_dim, hidden_dim)
                self.layer2 = TensorGraphSAGELayer(hidden_dim, out_dim)
        
            def forward(self, x, edge_index):
                x = F.relu(self.layer1(x, edge_index))
                return self.layer2(x, edge_index)
        
        # --- Stage 3: Linear classifier ---
        class LinearClassifier(nn.Module):
            def __init__(self, in_dim, num_classes):
                super().__init__()
                self.fc = nn.Linear(in_dim, num_classes)
        
            def forward(self, h):
                return self.fc(h)
        
        # --- Wiring the pipeline ---
        N, D, H, C = 300, 32, 64, 4
        
        g = Graph(
            node_features=torch.randn(N, D),
            edge_index=torch.randint(0, N, (2, 1200), dtype=torch.long),
        )
        labels = torch.randint(0, C, (N,))
        train_mask = torch.zeros(N, dtype=torch.bool)
        train_mask[:200] = True
        
        # Stage 1: fit normalizer on training nodes only
        normalizer = NodeFeatureNormalizer()
        normalizer.fit(g.node_features[train_mask])
        x_norm = normalizer.transform(g.node_features)
        
        # Stage 2: GNN encoder
        encoder = SAGEEncoder(D, H, H)
        # Stage 3: Classifier
        classifier = LinearClassifier(H, C)
        
        # Combine for training
        optimizer = torch.optim.Adam(
            list(encoder.parameters()) + list(classifier.parameters()), lr=1e-3
        )
        
        for epoch in range(100):
            encoder.train(); classifier.train()
            h = encoder(x_norm, g.edge_index)
            logits = classifier(h)
            loss = F.cross_entropy(logits[train_mask], labels[train_mask])
            optimizer.zero_grad()
            loss.backward()
            optimizer.step()
        
        print(f"Final train loss: {loss.item():.4f}")

The pipeline structure makes it obvious where training-data leakage could occur (only normalizer.fit should see the training split) and easy to replace any stage independently.

Early Stopping as an Estimator

TGraphX's tgraphx.estimators.early_stopping provides an EarlyStopping class that wraps validation-loss tracking. Using it as a pipeline component clarifies the stopping logic without cluttering the training loop:

python
from tgraphx.estimators.early_stopping import EarlyStopping
        
        stopper = EarlyStopping(patience=15, min_delta=1e-4)
        
        for epoch in range(500):
            encoder.train(); classifier.train()
            h = encoder(x_norm, g.edge_index)
            logits = classifier(h)
            train_loss = F.cross_entropy(logits[train_mask], labels[train_mask])
            optimizer.zero_grad()
            train_loss.backward()
            optimizer.step()
        
            # Validation step
            encoder.eval(); classifier.eval()
            with torch.no_grad():
                val_mask = ~train_mask
                val_logits = classifier(encoder(x_norm, g.edge_index))
                val_loss = F.cross_entropy(val_logits[val_mask], labels[val_mask])
        
            if stopper.step(val_loss.item()):
                print(f"Early stopping at epoch {epoch}")
                break

The EarlyStopping object maintains its own state (best loss, patience counter) without polluting the training loop.

Train/Val/Test Splits with the Estimators Module

TGraphX provides a tgraphx.estimators.splits module for generating node-level train/val/test masks:

python
from tgraphx.estimators.splits import random_node_split
        
        train_mask, val_mask, test_mask = random_node_split(
            num_nodes=N,
            train_ratio=0.6,
            val_ratio=0.2,
            seed=42,
        )

Using this instead of writing ad-hoc masking logic means the split behavior is reproducible (the seed is explicit), reusable across scripts, and easy to swap for a different split strategy.

Benefits for Reproducibility

When experiment logic lives inside estimators with explicit fit / transform / predict boundaries, several reproducibility benefits follow automatically.

First, serialization is straightforward. An estimator that has been fit can be saved with torch.save and reloaded for inference without re-running the training loop. The fitted normalizer statistics, model weights, and stopping criteria are all encapsulated in the estimator's state dict.

Second, hyperparameter sweeps are cleaner. Each configuration produces an independent estimator instance. Fitting two instances with different alpha values for label propagation requires no shared mutable state.

Third, the pipeline structure makes it explicit what computation depends on training data and what does not. The normalizer's fit sees training nodes only. The GNN encoder is trained on training labels only. Test-set evaluation calls only transform and predict, never fit.

For more on reproducibility practices in TGraphX, see the GNN research reproducibility article.

Combining with the Reproducibility Context Manager

TGraphX's tgraphx.reproducibility module provides a context manager that sets all random seeds:

python
from tgraphx.reproducibility import set_reproducibility
        
        with set_reproducibility(seed=42):
            train_mask, val_mask, test_mask = random_node_split(N, 0.6, 0.2)
            encoder = SAGEEncoder(D, H, H)
            classifier = LinearClassifier(H, C)
            # ... training loop ...

Wrapping the entire pipeline setup in this context ensures that the split, weight initialization, and any dropout masks all use the same seed, making results fully reproducible across runs.

Comparing Estimator Approaches: TGraphX vs Full Training Scripts

To make the tradeoffs concrete, consider two implementations of the same experiment: a flat training script and a pipeline using TGraphX estimators.

Flat script approach:
- All preprocessing, training, and evaluation logic in one function
- Fast to write for a single experiment
- Difficult to swap preprocessing or model components
- Easy to accidentally use validation or test data during preprocessing

Estimator pipeline approach:
- Preprocessing and model training are separate objects with explicit fit/transform interfaces
- Slower to write initially, but each component is independently testable
- Swapping a normalizer or model requires changing one line
- The fit boundary makes data leakage visible and easy to avoid

For one-off explorations, a flat script is perfectly reasonable. The pipeline pattern pays off when you have multiple experiments, multiple team members, or when code will be revisited after a gap of weeks or months.

Connecting to Experiment Tracking

TGraphX's tgraphx.tracking module integrates with the estimator pattern by recording which estimators were used, their hyperparameters, and the results of each fit call:

python
from tgraphx.tracking import ExperimentTracker
        
        tracker = ExperimentTracker(
            experiment_name="sage_pipeline_v1",
            output_dir="./tracking_logs"
        )
        
        tracker.log_params({
            'model_type': 'sage',
            'hidden_dim': 64,
            'dropout': 0.5,
            'lr': 1e-3,
            'seed': 42,
        })
        
        # ... training loop ...
        
        tracker.log_metrics({'test_accuracy': test_acc, 'val_loss': best_val_loss})
        tracker.save()

Wrapping experiment execution in a tracker creates a log entry for each run. Over many experiments, the log becomes a record of what was tried and what performed best — invaluable when preparing results for a paper.

Limitations and Honest Notes

TGraphX's estimators module is designed for research use, not production deployment. A few important caveats:

The sklearn Pipeline object provides additional functionality — parameter grids for GridSearchCV, clone() for creating fresh copies, and set_params() for nested parameter updates. TGraphX estimators do not implement the full sklearn interface. You cannot pass them directly to sklearn.model_selection.cross_val_score without a compatibility wrapper.

The fit / transform boundary assumes that the graph structure does not change between training and inference. For dynamic graphs where new edges appear at inference time, this assumption breaks and the normalizer statistics may be stale.

Estimators that wrap neural networks store model state as PyTorch module parameters. Unlike sklearn estimators, they are not trivially picklable without handling CUDA tensors carefully. Use torch.save with map_location='cpu' for cross-device portability.

The pipeline pattern adds a small amount of code overhead relative to a flat training script. For one-off experiments, this overhead may not be worthwhile. The pattern pays off most when you have many experiments, many configurations, or when the code will be revisited by others.

Frequently Asked Questions

Can I use TGraphX estimators inside a scikit-learn Pipeline?
Not directly, because TGraphX estimators operate on Graph objects rather than numpy arrays. You would need a wrapper that extracts the relevant tensor and converts it to a numpy array at the boundary.

Does early stopping save the best model checkpoint automatically?
The EarlyStopping class in TGraphX tracks loss values and signals when to stop, but does not automatically save model weights. You need to add a checkpoint save inside the loop when the validation loss improves.

How do I handle inductive splits where test nodes are truly unseen?
The random_node_split function produces masks over all nodes in the graph. For truly inductive evaluation, you need to construct a separate test graph object and run inference on it, ensuring that the training graph's normalizer statistics and the trained encoder are applied to the new graph without refitting.