Opinion: 2026 Recommendation Systems Will All Use Deep Learning – XGBoost Is Obsolete

import tensorflow as tf
import tensorflow_recommenders as tfrs
import numpy as np
import os
from typing import Dict, Text
import pandas as pd
from sklearn.model_selection import train_test_split

# Configuration constants
BATCH_SIZE = 2048
EMBEDDING_DIM = 128
LEARNING_RATE = 0.001
EPOCHS = 15
CHECKPOINT_DIR = "./tfrs_two_tower_checkpoints"

# Ensure checkpoint directory exists
os.makedirs(CHECKPOINT_DIR, exist_ok=True)

class TwoTowerModel(tfrs.Model):
    """Two-tower deep learning recsys model for candidate generation."""
    def __init__(self, user_vocab_size: int, item_vocab_size: int):
        super().__init__()
        # User tower: 3-layer MLP with dropout
        self.user_tower = tf.keras.Sequential([
            tf.keras.layers.StringLookup(vocabulary=tf.data.experimental.make_csv_dataset(
                "user_ids.csv", batch_size=1024, num_epochs=1
            ).map(lambda x: x["user_id"]).unbatch().unique()),
            tf.keras.layers.Embedding(user_vocab_size, EMBEDDING_DIM),
            tf.keras.layers.Flatten(),
            tf.keras.layers.Dense(256, activation="relu"),
            tf.keras.layers.Dropout(0.3),
            tf.keras.layers.Dense(128, activation="relu"),
            tf.keras.layers.Dropout(0.2),
            tf.keras.layers.Dense(EMBEDDING_DIM),
        ])
        # Item tower: parallel 3-layer MLP
        self.item_tower = tf.keras.Sequential([
            tf.keras.layers.StringLookup(vocabulary=tf.data.experimental.make_csv_dataset(
                "item_ids.csv", batch_size=1024, num_epochs=1
            ).map(lambda x: x["item_id"]).unbatch().unique()),
            tf.keras.layers.Embedding(item_vocab_size, EMBEDDING_DIM),
            tf.keras.layers.Flatten(),
            tf.keras.layers.Dense(256, activation="relu"),
            tf.keras.layers.Dropout(0.3),
            tf.keras.layers.Dense(128, activation="relu"),
            tf.keras.layers.Dropout(0.2),
            tf.keras.layers.Dense(EMBEDDING_DIM),
        ])
        # TFRS task for retrieval
        self.task = tfrs.tasks.Retrieval(
            metrics=tfrs.metrics.FactorizedTopK(
                candidates=tf.data.experimental.make_csv_dataset(
                    "item_ids.csv", batch_size=1024, num_epochs=1
                ).map(lambda x: x["item_id"]).unbatch().batch(128)
            )
        )

    def compute_loss(self, features: Dict[Text, tf.Tensor], training: bool) -> tf.Tensor:
        user_embeddings = self.user_tower(features["user_id"])
        item_embeddings = self.item_tower(features["item_id"])
        return self.task(user_embeddings, item_embeddings)

# Load and preprocess interaction data
try:
    interactions = pd.read_csv("user_item_interactions.csv")
    train_df, test_df = train_test_split(interactions, test_size=0.2, random_state=42)
except FileNotFoundError as e:
    raise RuntimeError(f"Interaction data not found: {e}. Ensure user_item_interactions.csv exists.") from e
except pd.errors.EmptyDataError as e:
    raise RuntimeError("Interaction CSV is empty.") from e

# Convert to TF datasets
train_ds = tf.data.Dataset.from_tensor_slices(dict(train_df[["user_id", "item_id"]]))
train_ds = train_ds.shuffle(buffer_size=10000).batch(BATCH_SIZE).prefetch(tf.data.AUTOTUNE)
test_ds = tf.data.Dataset.from_tensor_slices(dict(test_df[["user_id", "item_id"]]))
test_ds = test_ds.batch(BATCH_SIZE).prefetch(tf.data.AUTOTUNE)

# Initialize model
user_vocab_size = train_df["user_id"].nunique() + 1
item_vocab_size = train_df["item_id"].nunique() + 1
model = TwoTowerModel(user_vocab_size, item_vocab_size)

# Compile with Adam optimizer
model.compile(optimizer=tf.keras.optimizers.Adam(learning_rate=LEARNING_RATE))

# Train with checkpointing
checkpoint_callback = tf.keras.callbacks.ModelCheckpoint(
    filepath=os.path.join(CHECKPOINT_DIR, "ckpt"),
    monitor="val_factorized_top_k",
    mode="max",
    save_best_only=True
)

try:
    history = model.fit(
        train_ds,
        validation_data=test_ds,
        epochs=EPOCHS,
        callbacks=[checkpoint_callback],
        verbose=1
    )
except tf.errors.ResourceExhaustedError as e:
    raise RuntimeError("GPU memory exhausted. Reduce BATCH_SIZE or EMBEDDING_DIM.") from e

# Evaluate on test set
test_metrics = model.evaluate(test_ds, return_dict=True)
print(f"Test Factorized Top K: {test_metrics['factorized_top_k']}")

import xgboost as xgb
import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import LabelEncoder
from sklearn.metrics import ndcg_score
import joblib
import os

# Configuration
BATCH_SIZE = 2048
LEARNING_RATE = 0.1
MAX_DEPTH = 8
N_ESTIMATORS = 500
CHECKPOINT_PATH = "./xgboost_recsys_model.json"

class XGBoostRecsys:
    """XGBoost-based ranking model for recsys, requiring manual feature engineering."""
    def __init__(self):
        self.user_encoder = LabelEncoder()
        self.item_encoder = LabelEncoder()
        self.model = None

    def _engineer_features(self, df: pd.DataFrame) -> pd.DataFrame:
        """Manual feature engineering required for XGBoost: no native embedding support."""
        df = df.copy()
        # Encode categorical IDs
        df["user_id_encoded"] = self.user_encoder.fit_transform(df["user_id"])
        df["item_id_encoded"] = self.item_encoder.fit_transform(df["item_id"])
        # Manual interaction features (required for XGBoost to match DL performance)
        df["user_item_interaction_count"] = df.groupby(["user_id", "item_id"])["interaction_type"].transform("count")
        df["user_avg_rating"] = df.groupby("user_id")["rating"].transform("mean")
        df["item_avg_rating"] = df.groupby("item_id")["rating"].transform("mean")
        # Time since last interaction (manual feature)
        df["timestamp"] = pd.to_datetime(df["timestamp"])
        df["days_since_last_interaction"] = (df["timestamp"].max() - df["timestamp"]).dt.days
        return df.drop(columns=["user_id", "item_id", "timestamp", "interaction_type"])

    def train(self, train_df: pd.DataFrame, test_df: pd.DataFrame):
        """Train XGBoost model with early stopping."""
        # Preprocess data
        train_processed = self._engineer_features(train_df)
        test_processed = self._engineer_features(test_df)
        # Split features and target
        X_train = train_processed.drop(columns=["rating"])
        y_train = train_processed["rating"]
        X_test = test_processed.drop(columns=["rating"])
        y_test = test_processed["rating"]
        # Initialize XGBoost regressor
        self.model = xgb.XGBRanker(
            objective="rank:ndcg",
            learning_rate=LEARNING_RATE,
            max_depth=MAX_DEPTH,
            n_estimators=N_ESTIMATORS,
            early_stopping_rounds=20,
            eval_metric="ndcg@10",
            tree_method="hist",
            device="cuda"  # Requires GPU for large datasets
        )
        # Train with early stopping
        try:
            self.model.fit(
                X_train,
                y_train,
                eval_set=[(X_test, y_test)],
                verbose=True
            )
        except xgb.core.XGBoostError as e:
            raise RuntimeError(f"XGBoost training failed: {e}. Check input data or GPU availability.") from e
        # Save encoders and model
        joblib.dump(self.user_encoder, "user_encoder.joblib")
        joblib.dump(self.item_encoder, "item_encoder.joblib")
        self.model.save_model(CHECKPOINT_PATH)
        print(f"Model saved to {CHECKPOINT_PATH}")

    def evaluate(self, test_df: pd.DataFrame) -> float:
        """Evaluate model using NDCG@10."""
        test_processed = self._engineer_features(test_df)
        X_test = test_processed.drop(columns=["rating"])
        y_test = test_processed["rating"]
        # Predict rankings
        y_pred = self.model.predict(X_test)
        # Calculate NDCG@10 (requires grouping by user for recsys eval)
        ndcg_scores = []
        for user_id in test_df["user_id"].unique():
            user_mask = test_df["user_id"] == user_id
            user_y = y_test[user_mask]
            user_pred = y_pred[user_mask]
            if len(user_y) >= 10:
                ndcg_scores.append(ndcg_score([user_y], [user_pred], k=10))
        return np.mean(ndcg_scores)

# Load data
try:
    interactions = pd.read_csv("user_item_interactions.csv")
    train_df, test_df = train_test_split(interactions, test_size=0.2, random_state=42)
except FileNotFoundError as e:
    raise RuntimeError(f"Data not found: {e}") from e

# Train and evaluate XGBoost
xgb_model = XGBoostRecsys()
xgb_model.train(train_df, test_df)
ndcg = xgb_model.evaluate(test_df)
print(f"XGBoost NDCG@10: {ndcg:.4f}")

import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader, Dataset
import pandas as pd
import numpy as np
from sklearn.preprocessing import LabelEncoder
from torchrec.models import TwoTowerModel as TorchRecTwoTower
from torchrec.modules.embedding_modules import EmbeddingBagCollection
from torchrec.sparse.jagged_tensor import KeyedJaggedTensor
import os
from typing import List
import joblib
import xgboost as xgb

# Configuration
EMBEDDING_DIM = 128
HIDDEN_DIM = 256
BATCH_SIZE = 1024
EPOCHS = 10
LEARNING_RATE = 0.001
DEVICE = "cuda" if torch.cuda.is_available() else "cpu"

class RecsysDataset(Dataset):
    """Custom dataset for PyTorch recsys training."""
    def __init__(self, df: pd.DataFrame, user_encoder: LabelEncoder, item_encoder: LabelEncoder):
        self.df = df
        self.user_encoder = user_encoder
        self.item_encoder = item_encoder
        # Preprocess IDs
        self.user_ids = self.user_encoder.transform(df["user_id"])
        self.item_ids = self.item_encoder.transform(df["item_id"])
        self.labels = df["rating"].values

    def __len__(self) -> int:
        return len(self.df)

    def __getitem__(self, idx: int) -> tuple:
        return (
            torch.tensor(self.user_ids[idx], dtype=torch.long),
            torch.tensor(self.item_ids[idx], dtype=torch.long),
            torch.tensor(self.labels[idx], dtype=torch.float32)
        )

def migrate_xgb_to_dl(xgb_model_path: str, dl_checkpoint_path: str):
    """Migrate trained XGBoost recsys model to PyTorch Two-Tower DL model."""
    # Load XGBoost artifacts
    try:
        user_encoder = joblib.load("user_encoder.joblib")
        item_encoder = joblib.load("item_encoder.joblib")
        xgb_model = xgb.Booster()
        xgb_model.load_model(xgb_model_path)
    except FileNotFoundError as e:
        raise RuntimeError(f"XGBoost artifacts not found: {e}") from e
    except xgb.core.XGBoostError as e:
        raise RuntimeError(f"Failed to load XGBoost model: {e}") from e

    # Load interaction data
    try:
        interactions = pd.read_csv("user_item_interactions.csv")
    except FileNotFoundError as e:
        raise RuntimeError(f"Interaction data not found: {e}") from e

    # Initialize PyTorch dataset and dataloader
    dataset = RecsysDataset(interactions, user_encoder, item_encoder)
    dataloader = DataLoader(dataset, batch_size=BATCH_SIZE, shuffle=True, num_workers=4)

    # Initialize TorchRec Two-Tower model
    num_users = len(user_encoder.classes_)
    num_items = len(item_encoder.classes_)
    model = TorchRecTwoTower(
        num_users=num_users,
        num_items=num_items,
        embedding_dim=EMBEDDING_DIM,
        hidden_dim=HIDDEN_DIM
    ).to(DEVICE)

    # Initialize optimizer and loss
    optimizer = optim.Adam(model.parameters(), lr=LEARNING_RATE)
    loss_fn = nn.MSELoss()  # Regression loss for rating prediction

    # Train DL model
    model.train()
    for epoch in range(EPOCHS):
        total_loss = 0.0
        for user_ids, item_ids, labels in dataloader:
            user_ids = user_ids.to(DEVICE)
            item_ids = item_ids.to(DEVICE)
            labels = labels.to(DEVICE)
            # Forward pass
            user_emb, item_emb = model(user_ids, item_ids)
            predictions = torch.sum(user_emb * item_emb, dim=1)  # Dot product for score
            loss = loss_fn(predictions, labels)
            # Backward pass
            optimizer.zero_grad()
            loss.backward()
            optimizer.step()
            total_loss += loss.item()
        avg_loss = total_loss / len(dataloader)
        print(f"Epoch {epoch+1}/{EPOCHS}, Loss: {avg_loss:.4f}")

    # Save DL model
    os.makedirs(os.path.dirname(dl_checkpoint_path), exist_ok=True)
    torch.save(model.state_dict(), dl_checkpoint_path)
    print(f"Migrated DL model saved to {dl_checkpoint_path}")

    # Compare performance (simplified)
    print("Migration complete. DL model requires no manual feature engineering, unlike XGBoost.")

# Run migration
if __name__ == "__main__":
    migrate_xgb_to_dl(
        xgb_model_path="./xgboost_recsys_model.json",
        dl_checkpoint_path="./dl_recsys_model.pth"
    )

# Short snippet: Load pre-trained TFRS two-tower model for inference
import tensorflow as tf
import tensorflow_recommenders as tfrs

# Load saved model
model = tf.saved_model.load("./tfrs_two_tower_checkpoints/ckpt")

# Generate user embedding
user_embedding = model.user_tower("user_12345")
print(f"User embedding shape: {user_embedding.shape}")

# Short snippet: Initialize pre-built TFRS two-tower model
import tensorflow_recommenders as tfrs

model = tfrs.models.TwoTowerModel(
    user_vocab_size=100000,
    item_vocab_size=50000,
    embedding_dim=128,
    hidden_dims=[256, 128]
)
print("Pre-built two-tower model initialized")

# Short snippet: Calculate NDCG@10 for benchmark comparison
from sklearn.metrics import ndcg_score
import numpy as np

# Example: 100 users, 10 items per user
y_true = np.random.randint(1, 5, size=(100, 10))
y_pred = np.random.rand(100, 10)
ndcg = ndcg_score(y_true, y_pred, k=10)
print(f"Benchmark NDCG@10: {ndcg:.4f}")