Run this notebook

Download it with the View page source link at the top right, or from the repository at docs/source/implementing-a-recommender.ipynb.

Implementing a Recommender

This tutorial implements a complete recommender for Compresso Recsys. The ranking rule is deliberately simple: recommend the most popular items in the training data. That leaves room for the parts every usable model needs: validation, fitted state, candidate filtering, stable item IDs, batched prediction, recommendation, persistence, evaluation, and contract checks.

The result is not a sketch. TutorialTopPopular is a fully functioning collaborative recommender. More sophisticated models change how scores are learned; the surrounding contract stays the same.

What the base class owns

BaseCollaborativeRecommender already implements two model-independent workflows:

• predict() splits a CSR source matrix into bounded batches and joins their ranked outputs.

• recommend() maps stable item-ID histories, allowlists, and blocklists to catalog rows and maps predictions back to IDs.

It also owns the versioned ZIP checkpoint workflow. Our model must provide the configuration, fitted-state properties, fit(), predict_on_batch(), and the hooks that reconstruct model-specific state. This division is why a new model can be small without being incomplete.

Data

We use the fixed MovieLens 1M user split from the dataset tutorial. Ratings of four or five become implicit positive interactions. Building the checkpoint is the only potentially slow step and is skipped when the artifact already exists.

from dataclasses import dataclass
from pathlib import Path
import shutil
import tempfile

import numpy as np
import torch
from scipy.sparse import csr_matrix

from compresso import SRPTensor
import compresso_recsys as cr
from compresso_recsys.evaluation import evaluate_recommender
from compresso_recsys.metrics import CalibratedRecall, MRR, NDCG
from compresso_recsys.models import BaseCollaborativeRecommender
from compresso_recsys.persistence import ModelCheckpointReader, ModelCheckpointWriter

project_root = next(
    (path for path in (Path.cwd(), *Path.cwd().parents) if (path / "pyproject.toml").exists()),
    Path.cwd(),
)
data_checkpoint = project_root / "artifacts/tutorials/ml1m-user-split.zip"
if not data_checkpoint.exists():
    cr.build_recsys_checkpoint(
        dataset="ml1m",
        data_dir=str(project_root / "data"),
        checkpoint_path=str(data_checkpoint),
        split_mode="user_split",
        eval_draws=1,
        seed=42,
        min_entity_text_words=0,
        annotation_source="none",
        show_progress=True,
    )

with cr.read_checkpoint(data_checkpoint) as root:
    split = cr.load_recsys_split(root)

x_train = split["x_train"]
test_source = split["test_source_matrix"]
test_targets = split["test_target_matrix"]
item_ids = split["item_ids"]
x_train.shape, test_source.shape, test_targets.nnz

((4534, 3501), (1000, 3501), 18850)

Configuration

Configuration records repeatable choices, not learned state. A frozen dataclass is directly serializable by the inherited checkpoint workflow. Validation belongs here so invalid experiments fail before touching data.

@dataclass(frozen=True)
class TutorialTopPopularConfig:
    use_values: bool = False

    def __post_init__(self):
        if not isinstance(self.use_values, (bool, np.bool_)):
            raise TypeError("use_values must be a boolean")

The complete model

Top Popular has no separately optimized Torch module, so it does not need an artificial trainer class. fit() is its training algorithm. The learned state is one popularity value per catalog item. A robust fit() prepares and validates the vocabulary and learned values in local variables, then publishes them together. If validation or computation raises, a previously fitted model therefore remains intact.

predict_on_batch() is where the low-level recommender contract becomes concrete. It must honor candidate selections, exclude seen items when asked, reject impossible k values, rank ties deterministically, and report global catalog columns in an SRPTensor.

class TutorialTopPopular(BaseCollaborativeRecommender):
    checkpoint_type = "tutorial_top_popular"

    def __init__(self, config=None):
        self.cfg = config or TutorialTopPopularConfig()
        self.popularity_ = None
        self.n_items_ = None

    @property
    def is_fitted(self):
        return self.popularity_ is not None

    @property
    def n_items(self):
        return self.n_items_

    def fit(self, interactions, *, item_ids=None):
        if not isinstance(interactions, csr_matrix):
            raise TypeError("interactions must be a scipy.sparse.csr_matrix")
        if interactions.shape[0] < 1 or interactions.shape[1] < 1:
            raise ValueError("interactions must contain at least one user and one item")

        matrix = interactions.astype(np.float64, copy=True)
        matrix.sum_duplicates()
        matrix.eliminate_zeros()
        matrix.sort_indices()
        if np.any(matrix.data < 0):
            raise ValueError("interactions must contain nonnegative values")

        n_items = int(matrix.shape[1])
        vocabulary = self._prepare_item_vocabulary(item_ids, n_items=n_items)

        if self.cfg.use_values:
            popularity = np.asarray(matrix.sum(axis=0)).ravel()
        else:
            popularity = np.asarray(matrix.getnnz(axis=0))
        popularity = popularity.astype(np.float64, copy=False)

        self.popularity_ = popularity
        self.n_items_ = n_items
        self._publish_item_vocabulary(vocabulary)
        return self

    def predict_on_batch(
        self,
        source,
        *,
        k,
        exclude_seen=True,
        candidate_ids=None,
    ):
        source = self._prepare_source(source)
        candidates = self._candidate_rows(candidate_ids)
        if isinstance(k, (bool, np.bool_)) or not isinstance(k, (int, np.integer)):
            raise TypeError("k must be an integer")
        if not 1 <= int(k) <= candidates.size:
            raise ValueError(f"k must be in [1, {candidates.size}], got {k}")

        scores = np.broadcast_to(
            self.popularity_[candidates],
            (source.shape[0], candidates.size),
        ).copy()

        candidate_to_local = np.full(source.shape[1], -1, dtype=np.int64)
        candidate_to_local[candidates] = np.arange(candidates.size)
        seen_counts = np.diff(source.indptr)
        seen_rows = np.repeat(np.arange(source.shape[0]), seen_counts)
        seen_local = candidate_to_local[source.indices]
        selected_seen = seen_local >= 0

        if exclude_seen:
            available = candidates.size - np.bincount(
                seen_rows[selected_seen], minlength=source.shape[0]
            )
            if np.any(available < k):
                row = int(np.flatnonzero(available < k)[0])
                raise ValueError(
                    f"source row {row} has only {available[row]} unseen "
                    f"candidates, fewer than k={k}"
                )
            scores[seen_rows[selected_seen], seen_local[selected_seen]] = -np.inf

        columns = np.empty((source.shape[0], int(k)), dtype=np.int64)
        values = np.empty((source.shape[0], int(k)), dtype=np.float64)
        for row in range(source.shape[0]):
            # Popularity descending, then catalog row ascending for stable ties.
            local = np.lexsort((candidates, -scores[row]))[:k]
            columns[row] = candidates[local]
            values[row] = scores[row, local]

        return SRPTensor(
            cols=torch.from_numpy(columns),
            vals=torch.from_numpy(values),
            shape=source.shape,
        )

    @classmethod
    def _from_checkpoint_config(cls, config, reader, *, device):
        del reader, device
        return cls(TutorialTopPopularConfig(**config))

    def _save_checkpoint_state(self, writer: ModelCheckpointWriter):
        writer.write_numpy("state/popularity.npy", self.popularity_)

    def _load_checkpoint_state(self, reader: ModelCheckpointReader):
        popularity = reader.read_numpy("state/popularity.npy")
        if popularity.ndim != 1 or popularity.dtype != np.float64:
            raise ValueError("popularity state must be a float64 vector")
        if not np.isfinite(popularity).all() or np.any(popularity < 0):
            raise ValueError("popularity state must be finite and nonnegative")
        self.popularity_ = popularity
        self.n_items_ = int(popularity.size)

The base class serializes the dataclass configuration and stable item IDs. The three persistence hooks above only describe what is unique to this model: constructing it from configuration and writing or reading the learned popularity vector.

tutorial_model = TutorialTopPopular().fit(x_train, item_ids=item_ids)

assert tutorial_model.is_fitted
assert tutorial_model.n_items == x_train.shape[1]
assert np.array_equal(tutorial_model.source_item_ids, item_ids)
tutorial_model.popularity_[:5]

array([1265.,  377.,   27.,    7.,    5.])

Low-level prediction

Evaluation code works with catalog-aligned matrices and calls predict_on_batch(). Candidate IDs can restrict ranking without changing the output vocabulary: returned columns always refer to the fitted global catalog.

popular_rows = np.lexsort(
    (np.arange(tutorial_model.n_items), -tutorial_model.popularity_)
)
candidate_ids = item_ids[popular_rows[:20]]

predictions = tutorial_model.predict_on_batch(
    test_source[:3],
    k=5,
    exclude_seen=True,
    candidate_ids=candidate_ids,
)
assert predictions.shape == (3, tutorial_model.n_items)
predictions.cols

tensor([[1820,  179, 3141,  973, 1530],
        [1820, 1556,  179,  181, 3141],
        [1820, 1556,  179,  181, 3141]])

Stable-ID recommendation

Production callers usually have item IDs rather than catalog-column numbers. The inherited recommend() method maps histories and filters, groups requests that can return the same number of results, calls our prediction method, and returns IDs with scores. exclude_seen=False is the serving default; here we turn it on explicitly.

history_ids = item_ids[test_source[0].indices].tolist()
allowlist = candidate_ids.tolist()
blocklist = allowlist[:1]

recommendations = tutorial_model.recommend(
    [history_ids],
    k=5,
    exclude_seen=True,
    allowlist=allowlist,
    blocklist=blocklist,
    on_insufficient="raise",
)
recommendations.to_dicts()

[{np.str_('1196'): 1904.0,
  np.str_('593'): 1714.0,
  np.str_('2028'): 1700.0,
  np.str_('2571'): 1645.0,
  np.str_('608'): 1562.0}]

Standalone persistence

save() creates a safe, versioned ZIP containing configuration, item IDs, and model state. load() defaults to CPU and returns a prediction-ready instance. Because Top Popular has no optimizer, there is no optimizer state to request. Torch trainers can opt into that part of the same workflow with include_optimizer=True and load_optimizer=True.

with tempfile.TemporaryDirectory() as directory:
    model_path = Path(directory) / "top-popular.zip"
    tutorial_model.save(model_path)
    restored = TutorialTopPopular.load(model_path)

    before = tutorial_model.predict_on_batch(test_source[:3], k=5)
    after = restored.predict_on_batch(test_source[:3], k=5)
    torch.testing.assert_close(after.cols, before.cols)
    torch.testing.assert_close(after.vals, before.vals)
    assert np.array_equal(restored.source_item_ids, tutorial_model.source_item_ids)

restored.recommend([history_ids], k=5, exclude_seen=True).to_dicts()

[{'2858': 2147.0,
  '1196': 1904.0,
  '593': 1714.0,
  '2028': 1700.0,
  '2571': 1645.0}]

Persistence inside a data checkpoint

The same model can live under models/<name>.zip inside a dataset checkpoint. The inherited methods update and validate the data-checkpoint manifest; the model-specific hooks remain unchanged.

with tempfile.TemporaryDirectory() as directory:
    combined_checkpoint = Path(directory) / "ml1m-with-model.zip"
    shutil.copy2(data_checkpoint, combined_checkpoint)
    tutorial_model.save_to_checkpoint(combined_checkpoint, "top-popular")
    embedded = TutorialTopPopular.load_from_checkpoint(
        combined_checkpoint, "top-popular"
    )

    embedded_predictions = embedded.predict_on_batch(test_source[:3], k=5)
    torch.testing.assert_close(embedded_predictions.cols, before.cols)
    torch.testing.assert_close(embedded_predictions.vals, before.vals)

Evaluation

Evaluation is deliberately brief: its purpose here is to show that a newly implemented model enters the same protocol as the model zoo. The dedicated statistical comparison tutorial covers repeated training, paired uncertainty, multiple-comparison correction, and reporting.

metrics = [CalibratedRecall([10, 20]), NDCG(20), MRR([10, 20])]
tutorial_result = evaluate_recommender(
    tutorial_model,
    source=test_source,
    targets=test_targets,
    metrics=metrics,
    batch_size=512,
    show_progress=True,
)
dict(tutorial_result)

{'calibrated_recall@10': 0.158926590166986,
 'calibrated_recall@20': 0.17829685132205486,
 'ndcg@20': 0.1683201789818704,
 'mrr@10': 0.3303797630444169,
 'mrr@20': 0.33909823079034684,
 'n_scored_rows': 1000,
 'n_units': 1000}

The package already contains PopularityBaseline. Matching its learned popularity vector and ranked score values is a useful end-to-end check. Item IDs tied at the same popularity may appear in a different, equally valid order.

from compresso_recsys.models import PopularityBaseline, PopularityBaselineConfig

reference = PopularityBaseline(
    PopularityBaselineConfig(use_values=False)
).fit(x_train, item_ids=item_ids)
np.testing.assert_array_equal(
    tutorial_model.popularity_, reference.popularity_
)

tutorial_ranking = tutorial_model.predict_on_batch(test_source, k=20)
reference_ranking = reference.predict_on_batch(test_source, k=20)
torch.testing.assert_close(tutorial_ranking.vals, reference_ranking.vals)

Executable contract checks

A real implementation should turn these examples into focused unit tests. The notebook keeps a few assertions executable so its central promises cannot drift silently.

try:
    TutorialTopPopularConfig(use_values="yes")
except TypeError:
    pass
else:
    raise AssertionError("invalid configuration was accepted")

short = tutorial_model.recommend(
    [history_ids],
    k=5,
    allowlist=item_ids[:2],
    on_insufficient="truncate",
)
assert short.valid_mask.sum() == 2

try:
    tutorial_model.recommend(
        [history_ids],
        k=5,
        allowlist=item_ids[:2],
        on_insufficient="raise",
    )
except ValueError:
    pass
else:
    raise AssertionError("an impossible recommendation request was accepted")

Extending the pattern

A learned Torch recommender usually separates an nn.Module from its trainer. The persistence base automatically stores the module returned by _checkpoint_module() and can optionally store the optimizer exposed by _checkpoint_optimizer(). _build_checkpoint_optimizer() reconstructs that optimizer before its state is loaded, while _move_checkpoint_state() handles non-module device caches.

Iterative trainers should also follow the library’s reporting pattern. Accept a duck-typed job logger (only info(str) is required), resolve logger and show_progress per call, and let a logger disable tqdm unconditionally. Use log_every_n_steps for intra-epoch records, and latch a failing logger off after one warning so reporting cannot destroy a long training run. For fixed-epoch training, the notebook display should have two bars: an outer epoch bar and one manually updated batch bar that is reset and reused for every epoch. Close both in finally; do not construct a new batch bar inside each epoch. Mult-DAE and Mult-VAE are the reference implementations.

A sequential model inherits BaseSequentialRecommender and consumes ItemSequences. A cold-start model inherits BaseColdStartRecommender and publishes an owned candidate catalog with item features. Those families change their source representation and fitted state, not the stable-ID recommendation, evaluation, or checkpoint workflows demonstrated here.

That is the complete implementation pattern: keep the scoring idea local and let the shared bases own the repetitive serving and persistence machinery.