ranklens.explain

ranklens.explain

Explanations of a ranking model: why one document is above another, what moves the metric.

EXACT_MAX_FEATURES module-attribute

EXACT_MAX_FEATURES = 12

2^12 = 4096 scored points: still one cheap call; beyond it orders are sampled.

CallableAdapter

Any scoring function features -> scores: the escape hatch for every other model.

Source code in src/ranklens/explain/adapters/base.py

class CallableAdapter:
    """Any scoring function ``features -> scores``: the escape hatch for every other model."""

    def __init__(
        self,
        function: Callable[[Features], Sequence[float] | Scores],
        feature_names: Sequence[str],
        importance: Mapping[str, float] | None = None,
    ) -> None:
        self._function = function
        self._names = _names(feature_names)
        self._importance = dict(importance) if importance is not None else None

    @property
    def feature_names(self) -> tuple[str, ...]:
        return self._names

    def score(self, features: Features) -> Scores:
        return _scores(self._function(as_features(features, self._names)), features)

    def native_importance(self) -> Mapping[str, float] | None:
        return self._importance

ScorerAdapter

Bases: Protocol

A ranking model seen through the three things explanation needs.

Source code in src/ranklens/explain/adapters/base.py

@runtime_checkable
class ScorerAdapter(Protocol):
    """A ranking model seen through the three things explanation needs."""

    @property
    def feature_names(self) -> tuple[str, ...]: ...

    def score(self, features: Features) -> Scores: ...

    def native_importance(self) -> Mapping[str, float] | None:
        """The model's own importance per feature, or None when it has none."""
        ...

native_importance

native_importance() -> Mapping[str, float] | None

The model's own importance per feature, or None when it has none.

Source code in src/ranklens/explain/adapters/base.py

def native_importance(self) -> Mapping[str, float] | None:
    """The model's own importance per feature, or None when it has none."""
    ...

QueryFeatures dataclass

features[i] belongs to docs[i]; judgements may cover only some of them.

The matrix is copied and made read-only: explanations shuffle copies of it, never it.

Source code in src/ranklens/explain/data.py

@dataclass(frozen=True)
class QueryFeatures:
    """``features[i]`` belongs to ``docs[i]``; ``judgements`` may cover only some of them.

    The matrix is copied and made read-only: explanations shuffle copies of it, never it.
    """

    query_id: QueryId
    docs: tuple[DocId, ...]
    features: Features
    judgements: Mapping[DocId, float] = field(default_factory=lambda: MappingProxyType({}))

    def __post_init__(self) -> None:
        matrix = np.array(self.features, dtype=np.float64)  # a copy
        if matrix.ndim != 2 or matrix.shape[0] != len(self.docs):
            raise ValueError(
                f"query {self.query_id!r}: {len(self.docs)} docs, features of shape {matrix.shape}"
            )
        matrix.setflags(write=False)
        object.__setattr__(self, "features", matrix)
        object.__setattr__(self, "docs", tuple(self.docs))
        object.__setattr__(self, "judgements", MappingProxyType(dict(self.judgements)))

    @classmethod
    def of(
        cls,
        query_id: str,
        docs: Sequence[str],
        features: object,
        judgements: Mapping[str, float] | None = None,
    ) -> "QueryFeatures":
        return cls(
            QueryId(query_id),
            tuple(DocId(d) for d in docs),
            np.asarray(features, dtype=np.float64),
            {DocId(d): float(v) for d, v in (judgements or {}).items()},
        )

    @classmethod
    def from_rows(
        cls,
        query_id: QueryId,
        docs: Sequence[DocId],
        rows: Sequence[Sequence[float]],
        judgements: Mapping[DocId, float],
        columns: Sequence[int] | None = None,
    ) -> "QueryFeatures":
        """From plain rows (as a feature table reader gives them), keeping ``columns``."""
        matrix = np.asarray(rows, dtype=np.float64).reshape(len(docs), -1)
        return cls(
            query_id, tuple(docs), matrix if columns is None else matrix[:, columns], judgements
        )

from_rows classmethod

from_rows(
    query_id: QueryId,
    docs: Sequence[DocId],
    rows: Sequence[Sequence[float]],
    judgements: Mapping[DocId, float],
    columns: Sequence[int] | None = None,
) -> QueryFeatures

From plain rows (as a feature table reader gives them), keeping columns.

Source code in src/ranklens/explain/data.py

@classmethod
def from_rows(
    cls,
    query_id: QueryId,
    docs: Sequence[DocId],
    rows: Sequence[Sequence[float]],
    judgements: Mapping[DocId, float],
    columns: Sequence[int] | None = None,
) -> "QueryFeatures":
    """From plain rows (as a feature table reader gives them), keeping ``columns``."""
    matrix = np.asarray(rows, dtype=np.float64).reshape(len(docs), -1)
    return cls(
        query_id, tuple(docs), matrix if columns is None else matrix[:, columns], judgements
    )

adapt

adapt(
    model: Any, feature_names: Sequence[str] | None = None
) -> ScorerAdapter

The adapter for model, found by the packages of the classes in its MRO.

Walking the MRO recognizes subclasses defined elsewhere (a user's class derived from LGBMRanker) and needs none of the libraries imported. An adapter is returned as is; anything unknown is an UnsupportedModelError that points to CallableAdapter.

Source code in src/ranklens/explain/adapters/__init__.py

def adapt(model: Any, feature_names: Sequence[str] | None = None) -> ScorerAdapter:
    """The adapter for ``model``, found by the packages of the classes in its MRO.

    Walking the MRO recognizes subclasses defined elsewhere (a user's class derived from
    ``LGBMRanker``) and needs none of the libraries imported. An adapter is returned
    as is; anything unknown is an `UnsupportedModelError` that points to `CallableAdapter`.
    """
    if isinstance(model, ScorerAdapter):
        return model
    for cls in type(model).__mro__:
        adapter = ADAPTERS.get(cls.__module__.split(".")[0])
        if adapter is not None:
            return adapter(model, feature_names)
    kind = type(model)
    raise UnsupportedModelError(f"{kind.__module__}.{kind.__qualname__}", sorted(ADAPTERS))

correlated_groups

correlated_groups(
    queries: Iterable[QueryFeatures],
    feature_names: Sequence[str],
    *,
    threshold: float = 0.8,
) -> tuple[tuple[str, ...], ...]

Groups of two or more features linked by |correlation| >= threshold.

Correlation is computed over all documents of all queries, pairwise over the documents where both values are present (nan is a missing value). Features in no group are not listed; a constant feature correlates with nothing.

Source code in src/ranklens/explain/grouping.py

def correlated_groups(
    queries: Iterable[QueryFeatures], feature_names: Sequence[str], *, threshold: float = 0.8
) -> tuple[tuple[str, ...], ...]:
    """Groups of two or more features linked by |correlation| >= ``threshold``.

    Correlation is computed over all documents of all queries, pairwise over the documents
    where both values are present (``nan`` is a missing value). Features in no group are
    not listed; a constant feature correlates with nothing.
    """
    if not 0.0 < threshold <= 1.0:
        raise ValueError(f"threshold must be in (0, 1], got {threshold}")
    matrices = [query.features for query in queries]
    if not matrices:
        return ()
    pooled = np.vstack(matrices)
    if pooled.shape[1] != len(feature_names):
        raise ValueError(f"{len(feature_names)} names for {pooled.shape[1]} feature columns")
    with np.errstate(invalid="ignore", divide="ignore"):
        # pairwise over the documents where both values are present: one missing value
        # must not cut a feature out of every group
        pairwise = np.ma.corrcoef(np.ma.masked_invalid(pooled), rowvar=False)  # type: ignore[no-untyped-call]
    correlation = np.nan_to_num(pairwise.filled(np.nan), nan=0.0)

    parent = list(range(len(feature_names)))

    def root(i: int) -> int:
        while parent[i] != i:
            parent[i] = parent[parent[i]]  # path halving
            i = parent[i]
        return i

    for i, j in zip(*np.nonzero(np.abs(np.atleast_2d(correlation)) >= threshold), strict=True):
        if i < j:
            parent[root(int(i))] = root(int(j))

    components: dict[int, list[str]] = {}
    for i, name in enumerate(feature_names):
        components.setdefault(root(i), []).append(name)
    return tuple(tuple(group) for group in components.values() if len(group) > 1)

load_model

load_model(
    path: str | Path,
    *,
    kind: str | None = None,
    allow_pickle: bool = False,
) -> Any

The model saved at path, ready for ranklens.explain.adapt.

Source code in src/ranklens/explain/loading.py

def load_model(path: str | Path, *, kind: str | None = None, allow_pickle: bool = False) -> Any:
    """The model saved at ``path``, ready for `ranklens.explain.adapt`."""
    path = Path(path)
    kind = kind or MODEL_FORMATS.get(path.suffix.lower())
    if kind is None:
        known = ", ".join(sorted(MODEL_FORMATS))
        raise ConfigError(f"{path}: unknown model format; known extensions: {known}")
    if not path.exists():
        raise FileNotFoundError(f"{path}: no such model file")
    if kind == "pickle":
        if not allow_pickle:
            raise ConfigError(
                f"{path}: loading a pickle runs code from the file; pass --allow-pickle only "
                "for a model you trust (see SECURITY.md)"
            )
        import pickle

        with path.open("rb") as fh:
            return pickle.load(fh)
    loader = _LOADERS.get(kind)
    if loader is None:
        raise ConfigError(f"unknown model kind {kind!r}; known: {', '.join(sorted(_LOADERS))}")
    return loader(path)

ranklens.explain.adapters

Model adapters: one protocol for every model the explanations work with.

Features module-attribute

Features = npt.NDArray[np.float64]

Documents by features.

Scores module-attribute

Scores = npt.NDArray[np.float64]

One score per document; higher ranks first.

ADAPTERS module-attribute

ADAPTERS: dict[str, Callable[..., ScorerAdapter]] = {
    "catboost": CatBoostAdapter,
    "lightgbm": LightGBMAdapter,
    "xgboost": XGBoostAdapter,
    "sklearn": SklearnAdapter,
    "torch": TorchAdapter,
    "onnxruntime": ONNXAdapter,
}

Top-level package of a model's class -> its adapter. A new library is one line here.

CallableAdapter

Any scoring function features -> scores: the escape hatch for every other model.

Source code in src/ranklens/explain/adapters/base.py

class CallableAdapter:
    """Any scoring function ``features -> scores``: the escape hatch for every other model."""

    def __init__(
        self,
        function: Callable[[Features], Sequence[float] | Scores],
        feature_names: Sequence[str],
        importance: Mapping[str, float] | None = None,
    ) -> None:
        self._function = function
        self._names = _names(feature_names)
        self._importance = dict(importance) if importance is not None else None

    @property
    def feature_names(self) -> tuple[str, ...]:
        return self._names

    def score(self, features: Features) -> Scores:
        return _scores(self._function(as_features(features, self._names)), features)

    def native_importance(self) -> Mapping[str, float] | None:
        return self._importance

ScorerAdapter

Bases: Protocol

A ranking model seen through the three things explanation needs.

Source code in src/ranklens/explain/adapters/base.py

@runtime_checkable
class ScorerAdapter(Protocol):
    """A ranking model seen through the three things explanation needs."""

    @property
    def feature_names(self) -> tuple[str, ...]: ...

    def score(self, features: Features) -> Scores: ...

    def native_importance(self) -> Mapping[str, float] | None:
        """The model's own importance per feature, or None when it has none."""
        ...

native_importance

native_importance() -> Mapping[str, float] | None

The model's own importance per feature, or None when it has none.

Source code in src/ranklens/explain/adapters/base.py

def native_importance(self) -> Mapping[str, float] | None:
    """The model's own importance per feature, or None when it has none."""
    ...

CatBoostAdapter

Bases: _Adapter

CatBoost ranker, regressor or classifier: raw formula value.

Source code in src/ranklens/explain/adapters/libraries.py

class CatBoostAdapter(_Adapter):
    """CatBoost ranker, regressor or classifier: raw formula value."""

    def __init__(self, model: Any, feature_names: Sequence[str] | None = None) -> None:
        known = [str(n) for n in model.feature_names_]
        if known == [str(i) for i in range(len(known))]:
            known = list(default_names(len(known)))  # trained without names: "0", "1", ...
        super().__init__(model, known, feature_names)

    def _raw(self, matrix: Features) -> object:
        # the base class: CatBoostRanker.predict has no prediction_type, the classifier's
        # predict would return labels; the model exists, so the import cannot fail
        from catboost import CatBoost

        return CatBoost.predict(self.model, matrix, prediction_type="RawFormulaVal")

    def native_importance(self) -> Mapping[str, float] | None:
        # rankers default to LossFunctionChange, which needs the training data
        return self._aligned(self.model.get_feature_importance(type="PredictionValuesChange"))

LightGBMAdapter

Bases: _Adapter

LightGBM Booster or its scikit-learn wrappers: raw score, gain importance.

Source code in src/ranklens/explain/adapters/libraries.py

class LightGBMAdapter(_Adapter):
    """LightGBM ``Booster`` or its scikit-learn wrappers: raw score, gain importance."""

    def __init__(self, model: Any, feature_names: Sequence[str] | None = None) -> None:
        self._booster = getattr(model, "booster_", model)
        known = list(self._booster.feature_name())
        if known == [f"Column_{i}" for i in range(len(known))]:
            known = list(default_names(len(known)))  # trained without names: Column_0, ...
        super().__init__(model, known, feature_names)

    def _raw(self, matrix: Features) -> object:
        return self._booster.predict(matrix, raw_score=True)

    def native_importance(self) -> Mapping[str, float] | None:
        return self._aligned(self._booster.feature_importance(importance_type="gain"))

SklearnAdapter

Bases: _Adapter

A fitted scikit-learn estimator; importance if it has one.

Regressors score with predict; binary classifiers with decision_function or, without it, the probability of the positive class.

Source code in src/ranklens/explain/adapters/libraries.py

class SklearnAdapter(_Adapter):
    """A fitted scikit-learn estimator; importance if it has one.

    Regressors score with ``predict``; binary classifiers with ``decision_function``
    or, without it, the probability of the positive class.
    """

    def __init__(self, model: Any, feature_names: Sequence[str] | None = None) -> None:
        if hasattr(model, "feature_names_in_"):
            known: Sequence[str] = [str(n) for n in model.feature_names_in_]
        else:
            known = default_names(int(model.n_features_in_))
        super().__init__(model, known, feature_names)

    def _raw(self, matrix: Features) -> object:
        if not hasattr(self.model, "classes_"):
            return self.model.predict(matrix)
        # a classifier's predict returns labels; the margin (or P(positive)) orders documents
        if hasattr(self.model, "decision_function"):
            return self.model.decision_function(matrix)
        return self.model.predict_proba(matrix)[:, 1:]

    def native_importance(self) -> Mapping[str, float] | None:
        importances = getattr(self.model, "feature_importances_", None)
        return None if importances is None else self._aligned(importances)

XGBoostAdapter

Bases: _Adapter

XGBoost Booster or its scikit-learn wrappers: margin, gain importance.

Source code in src/ranklens/explain/adapters/libraries.py

class XGBoostAdapter(_Adapter):
    """XGBoost ``Booster`` or its scikit-learn wrappers: margin, gain importance."""

    def __init__(self, model: Any, feature_names: Sequence[str] | None = None) -> None:
        self._booster = model.get_booster() if hasattr(model, "get_booster") else model
        known = self._booster.feature_names or default_names(self._booster.num_features())
        super().__init__(model, list(known), feature_names)

    def _raw(self, matrix: Features) -> object:
        return self._booster.inplace_predict(matrix, predict_type="margin")

    def native_importance(self) -> Mapping[str, float] | None:
        # get_score keys are the booster's own names; a feature never used for a split
        # is missing from it and gets 0
        own = self._booster.feature_names or default_names(self._booster.num_features())
        gains = self._booster.get_score(importance_type="gain")
        return self._aligned([gains.get(name, 0.0) for name in own])

ONNXAdapter

An onnxruntime.InferenceSession with one input of documents by features.

The first output is the score. The number of features comes from the input shape; names default to f0, f1, … as for other models trained without names.

Source code in src/ranklens/explain/adapters/neural.py

class ONNXAdapter:
    """An ``onnxruntime.InferenceSession`` with one input of documents by features.

    The first output is the score. The number of features comes from the input shape;
    names default to f0, f1, … as for other models trained without names.
    """

    def __init__(self, model: Any, feature_names: Sequence[str] | None = None) -> None:
        inputs = model.get_inputs()
        if len(inputs) != 1:
            raise ValueError(f"expected an ONNX model with one input, got {len(inputs)}")
        self.model = model
        self._input = inputs[0].name
        self._dtype = np.float64 if inputs[0].type == "tensor(double)" else np.float32
        width = inputs[0].shape[-1]
        if feature_names is None:
            if not isinstance(width, int):
                raise ValueError("the ONNX input has no fixed width: pass feature_names")
            feature_names = [f"f{i}" for i in range(width)]
        elif isinstance(width, int) and width != len(feature_names):
            raise ValueError(f"model has {width} features, got {len(feature_names)} names")
        self._names = _names(feature_names)

    @property
    def feature_names(self) -> tuple[str, ...]:
        return self._names

    def score(self, features: Features) -> Scores:
        matrix = as_features(features, self._names)
        (output, *_) = self.model.run(None, {self._input: matrix.astype(self._dtype)})
        return _scores(output, matrix)

    def native_importance(self) -> Mapping[str, float] | None:
        return None

TorchAdapter

A torch.nn.Module mapping documents by features to one score per document.

Scored in inference mode without changing the module: eval() is applied for the call (dropout off, batch norm frozen) and every submodule gets its mode back after it. A module knows no feature names, so they are required.

Source code in src/ranklens/explain/adapters/neural.py

class TorchAdapter:
    """A ``torch.nn.Module`` mapping documents by features to one score per document.

    Scored in inference mode without changing the module: ``eval()`` is applied for the
    call (dropout off, batch norm frozen) and every submodule gets its mode back after it.
    A module knows no feature names, so they are required.
    """

    def __init__(self, model: Any, feature_names: Sequence[str] | None = None) -> None:
        if feature_names is None:
            raise ValueError("a torch model has no feature names: pass feature_names")
        self.model = model
        self._names = _names(feature_names)

    @property
    def feature_names(self) -> tuple[str, ...]:
        return self._names

    def score(self, features: Features) -> Scores:
        import torch  # the model exists, so torch is installed

        matrix = as_features(features, self._names)
        parameter = next(self.model.parameters(), None)
        device = parameter.device if parameter is not None else "cpu"
        dtype = parameter.dtype if parameter is not None else torch.float32
        # per submodule: a frozen encoder may sit in eval inside a model in train mode
        modes = [(module, module.training) for module in self.model.modules()]
        self.model.eval()
        try:
            with torch.inference_mode():
                output = self.model(torch.as_tensor(matrix, dtype=dtype, device=device))
        finally:
            for module, training in modes:
                module.training = training
        return _scores(output.detach().to("cpu", torch.float64).numpy(), matrix)

    def native_importance(self) -> Mapping[str, float] | None:
        return None

adapt

adapt(
    model: Any, feature_names: Sequence[str] | None = None
) -> ScorerAdapter

The adapter for model, found by the packages of the classes in its MRO.

Walking the MRO recognizes subclasses defined elsewhere (a user's class derived from LGBMRanker) and needs none of the libraries imported. An adapter is returned as is; anything unknown is an UnsupportedModelError that points to CallableAdapter.

Source code in src/ranklens/explain/adapters/__init__.py

def adapt(model: Any, feature_names: Sequence[str] | None = None) -> ScorerAdapter:
    """The adapter for ``model``, found by the packages of the classes in its MRO.

    Walking the MRO recognizes subclasses defined elsewhere (a user's class derived from
    ``LGBMRanker``) and needs none of the libraries imported. An adapter is returned
    as is; anything unknown is an `UnsupportedModelError` that points to `CallableAdapter`.
    """
    if isinstance(model, ScorerAdapter):
        return model
    for cls in type(model).__mro__:
        adapter = ADAPTERS.get(cls.__module__.split(".")[0])
        if adapter is not None:
            return adapter(model, feature_names)
    kind = type(model)
    raise UnsupportedModelError(f"{kind.__module__}.{kind.__qualname__}", sorted(ADAPTERS))