Graph Neural Networks

import torch
import torch.nn.functional as F
from torch_geometric.nn import GCNConv
from torch_geometric.datasets import Planetoid

# Cargar dataset Cora (2708 nodos, 10556 aristas, 7 clases)
dataset = Planetoid(root='/tmp/Cora', name='Cora')
data = dataset[0]

class GCN(torch.nn.Module):
    def __init__(self, in_channels, hidden_channels, out_channels):
        super().__init__()
        self.conv1 = GCNConv(in_channels, hidden_channels)
        self.conv2 = GCNConv(hidden_channels, out_channels)

    def forward(self, x, edge_index):
        # Capa 1: features → hidden
        x = self.conv1(x, edge_index)
        x = F.relu(x)
        x = F.dropout(x, p=0.5, training=self.training)
        # Capa 2: hidden → clases
        x = self.conv2(x, edge_index)
        return x

# Inicializar
model = GCN(dataset.num_features, 16, dataset.num_classes)
optimizer = torch.optim.Adam(model.parameters(), lr=0.01, weight_decay=5e-4)

# Entrenamiento
def train():
    model.train()
    optimizer.zero_grad()
    out = model(data.x, data.edge_index)
    # Loss SOLO en nodos de entrenamiento
    loss = F.cross_entropy(out[data.train_mask], data.y[data.train_mask])
    loss.backward()
    optimizer.step()
    return loss.item()

# Evaluación
@torch.no_grad()
def test():
    model.eval()
    out = model(data.x, data.edge_index)
    pred = out.argmax(dim=1)
    acc = (pred[data.test_mask] == data.y[data.test_mask]).float().mean()
    return acc.item()

for epoch in range(200):
    loss = train()
    if epoch % 50 == 0:
        acc = test()
        print(f'Epoch {epoch:3d} | Loss: {loss:.4f} | Test Acc: {acc:.4f}')

# Resultado típico: ~81% accuracy con solo 2 capas GCN

from torch_geometric.nn import GATConv

class GAT(torch.nn.Module):
    def __init__(self, in_channels, hidden_channels, out_channels, heads=8):
        super().__init__()
        # Multi-head attention: 8 cabezas × hidden → 8·hidden
        self.conv1 = GATConv(in_channels, hidden_channels, heads=heads,
                             dropout=0.6)
        # Última capa: 1 cabeza (output)
        self.conv2 = GATConv(hidden_channels * heads, out_channels,
                             heads=1, concat=False, dropout=0.6)

    def forward(self, x, edge_index):
        x = F.dropout(x, p=0.6, training=self.training)
        x = F.elu(self.conv1(x, edge_index))
        x = F.dropout(x, p=0.6, training=self.training)
        x = self.conv2(x, edge_index)
        return x

model = GAT(dataset.num_features, 8, dataset.num_classes, heads=8)
# GAT alcanza ~83% en Cora, ligeramente mejor que GCN

from torch_geometric.nn import GINConv, global_add_pool
from torch_geometric.datasets import TUDataset
from torch_geometric.loader import DataLoader
import torch.nn as nn

# Dataset molecular MUTAG (188 moléculas, 2 clases)
dataset = TUDataset(root='/tmp/MUTAG', name='MUTAG')
loader = DataLoader(dataset, batch_size=32, shuffle=True)

class GIN(torch.nn.Module):
    def __init__(self, in_dim, hidden_dim, out_dim):
        super().__init__()
        # GIN usa MLPs como función de actualización
        nn1 = nn.Sequential(nn.Linear(in_dim, hidden_dim), nn.ReLU(),
                            nn.Linear(hidden_dim, hidden_dim))
        nn2 = nn.Sequential(nn.Linear(hidden_dim, hidden_dim), nn.ReLU(),
                            nn.Linear(hidden_dim, hidden_dim))
        self.conv1 = GINConv(nn1)
        self.conv2 = GINConv(nn2)
        self.classifier = nn.Linear(hidden_dim, out_dim)

    def forward(self, x, edge_index, batch):
        # Message passing
        x = F.relu(self.conv1(x, edge_index))
        x = F.relu(self.conv2(x, edge_index))
        # Readout: sum pooling (GIN recomienda sum)
        x = global_add_pool(x, batch)
        # Clasificación del grafo
        return self.classifier(x)

model = GIN(dataset.num_features, 64, dataset.num_classes)
optimizer = torch.optim.Adam(model.parameters(), lr=0.01)

for epoch in range(50):
    model.train()
    for batch in loader:
        optimizer.zero_grad()
        out = model(batch.x, batch.edge_index, batch.batch)
        loss = F.cross_entropy(out, batch.y)
        loss.backward()
        optimizer.step()

import numpy as np
import tensorflow as tf
from spektral.datasets import Citation
from spektral.layers import GCNConv
from spektral.transforms import AdjToSpTensor, LayerPreprocess

# Cargar Cora
dataset = Citation('cora', transforms=[LayerPreprocess(GCNConv),
                                        AdjToSpTensor()])
graph = dataset[0]

# graph.x: (2708, 1433) features
# graph.a: (2708, 2708) adjacency (sparse)
# graph.y: (2708, 7) one-hot labels

class GCNModel(tf.keras.Model):
    def __init__(self, n_hidden, n_classes):
        super().__init__()
        self.conv1 = GCNConv(n_hidden, activation='relu')
        self.dropout = tf.keras.layers.Dropout(0.5)
        self.conv2 = GCNConv(n_classes, activation='softmax')

    def call(self, inputs, training=False):
        x, a = inputs
        x = self.conv1([x, a])
        x = self.dropout(x, training=training)
        x = self.conv2([x, a])
        return x

model = GCNModel(16, dataset.n_labels)
model.compile(
    optimizer=tf.keras.optimizers.Adam(learning_rate=0.01),
    loss='categorical_crossentropy',
    weighted_metrics=['accuracy']
)

# Crear máscaras de train/val/test
# (Cora split estándar: 140 train, 500 val, 1000 test)
mask_tr = dataset.mask_tr
mask_va = dataset.mask_va
mask_te = dataset.mask_te

# Entrenamiento
# Spektral soporta modo "single" (grafo completo en memoria)
model.fit(
    [graph.x, graph.a],
    graph.y,
    sample_weight=mask_tr.astype(float),
    validation_data=([graph.x, graph.a], graph.y, mask_va.astype(float)),
    epochs=200,
    batch_size=2708,  # grafo completo
    verbose=0
)

# Evaluación
loss, acc = model.evaluate(
    [graph.x, graph.a], graph.y,
    sample_weight=mask_te.astype(float)
)
print(f"Test accuracy: {acc:.4f}")

import tensorflow as tf
import numpy as np

class GraphConvLayer(tf.keras.layers.Layer):
    """Capa GCN implementada desde cero."""

    def __init__(self, output_dim, activation='relu', **kwargs):
        super().__init__(**kwargs)
        self.output_dim = output_dim
        self.activation = tf.keras.activations.get(activation)

    def build(self, input_shape):
        # input_shape[0] = (N, F_in) features
        f_in = input_shape[0][-1]
        self.W = self.add_weight('W', shape=(f_in, self.output_dim),
                                 initializer='glorot_uniform')
        self.b = self.add_weight('b', shape=(self.output_dim,),
                                 initializer='zeros')

    def call(self, inputs):
        x, adj_norm = inputs  # x: (N, F), adj_norm: (N, N)
        # 1. Transformación lineal
        h = tf.matmul(x, self.W)
        # 2. Agregación (multiplicación por adyacencia normalizada)
        h = tf.matmul(adj_norm, h)
        # 3. Bias + activación
        return self.activation(h + self.b)

# Crear grafo ejemplo (5 nodos)
N, F_in, F_hidden, n_classes = 5, 3, 8, 2
A = np.array([[0,1,0,1,0],[1,0,1,0,0],[0,1,0,1,1],
              [1,0,1,0,0],[0,0,1,0,0]], dtype=np.float32)
X = np.random.randn(N, F_in).astype(np.float32)

# Normalización: D^{-1/2} (A+I) D^{-1/2}
A_hat = A + np.eye(N, dtype=np.float32)
D_hat = np.diag(A_hat.sum(axis=1) ** -0.5)
A_norm = D_hat @ A_hat @ D_hat

# Modelo
x_in = tf.keras.Input(shape=(F_in,))
a_in = tf.keras.Input(shape=(N,))
h = GraphConvLayer(F_hidden)([x_in, a_in])
h = GraphConvLayer(n_classes, activation='softmax')([h, a_in])
model = tf.keras.Model(inputs=[x_in, a_in], outputs=h)

out = model([X, A_norm])
print(f"Output shape: {out.shape}")  # (5, 2) — 1 predicción por nodo