Skip to content

Adapters

Summary

effector engines take a plain numpy-in / numpy-out callable. The adapters are facilitation only: each one returns such a callable for a common model object — you call the adapter, look at what it gave you, and make the final pass into the constructor yourself. Nothing in effector auto-detects or silently wraps a model.

import effector

# sklearn regressor / pipeline
model = effector.adapters.from_sklearn(estimator)

# classifier: explain P(class = k), one explanation per class
model = effector.adapters.classifier_proba(clf, class_="yes")

# torch module (jacobian=True also builds the autograd jacobian for RHALE/DerPDP)
model = effector.adapters.from_torch(net)
model, model_jac = effector.adapters.from_torch(net, jacobian=True)

# the explicit handshake: probe the wrapper on two rows of your data
effector.adapters.check(model, X)

pdp = effector.PDP(X, model, schema=schema)   # the final pass is yours

Every returned callable also validates its own output on each call (shape (N,), numeric) and raises a named, actionable error instead of letting a wrong shape travel into a kernel.


API

effector.adapters

Model adapters — wrappers that turn common model objects into the plain callable effector needs.

Every engine takes a numpy-in / numpy-out callable f(X: (N, D)) -> (N,). Each adapter here returns one; check probes it on two rows of your data before you build an engine.

model = effector.adapters.from_sklearn(est)
effector.adapters.check(model, X)
pdp = effector.PDP(X, model, schema=schema)   # the final pass is yours

Facilitation only

Adapters only return callables — nothing in effector auto-detects or silently wraps a model. You call the adapter, look at what it gave you, and pass it to the constructor yourself.

Every returned callable validates its own output on each call (shape (N,), numeric) and raises a named, actionable error instead of letting a wrong shape travel into a kernel. sklearn / torch are never imported at module import time — only inside the adapter that needs them.

Functions:

Name Description
check

Probe a model wrapper on two rows of your data — the explicit handshake.

classifier_proba

Wrap a classifier into a numpy->numpy callable for one class' probability.

from_sklearn

Wrap an sklearn-style regressor into a numpy->numpy callable.

from_torch

Wrap a torch module into numpy->numpy callable(s).

check(model, X, model_jac=None)

Probe a model wrapper on two rows of your data — the explicit handshake.

model = effector.adapters.from_sklearn(est)
effector.adapters.check(model, X)             # raises loudly, or is silent
pdp = effector.PDP(X, model, schema=schema)

Call it right before constructing an engine; it is the only place a model call happens outside the engines, and you trigger it. Only X[:2] is evaluated, so the probe is instant even for slow models.

Parameters:

Name Type Description Default
model

the callable you are about to pass to an engine.

required
X

your data (2-D numpy array); only X[:2] is evaluated.

required
model_jac

optional jacobian callable; probed for shape (2, D).

None

Returns:

Type Description
None

None — silence means the wrapper is numpy-in / numpy-out with the

None

shapes effector expects.

Raises:

Type Description
TypeError

model (or model_jac) is not callable.

ValueError

X is not 2-D, or an output has the wrong shape or a non-numeric dtype. A model exception on the probe is re-raised with context.

Source code in effector/adapters.py
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
def check(model, X, model_jac=None) -> None:
    """Probe a model wrapper on two rows of your data — the explicit handshake.

    ```python
    model = effector.adapters.from_sklearn(est)
    effector.adapters.check(model, X)             # raises loudly, or is silent
    pdp = effector.PDP(X, model, schema=schema)
    ```

    Call it right before constructing an engine; it is the only place a model
    call happens outside the engines, and *you* trigger it. Only ``X[:2]`` is
    evaluated, so the probe is instant even for slow models.

    Args:
        model: the callable you are about to pass to an engine.
        X: your data (2-D numpy array); only ``X[:2]`` is evaluated.
        model_jac: optional jacobian callable; probed for shape ``(2, D)``.

    Returns:
        None — silence means the wrapper is numpy-in / numpy-out with the
        shapes effector expects.

    Raises:
        TypeError: `model` (or `model_jac`) is not callable.
        ValueError: `X` is not 2-D, or an output has the wrong shape or a
            non-numeric dtype. A model exception on the probe is re-raised
            with context.
    """
    if not callable(model):
        raise TypeError(
            f"adapters.check: model is not callable (got "
            f"{type(model).__name__}); wrap it first — see "
            f"effector.adapters.from_sklearn / classifier_proba / from_torch."
        )
    X = np.asarray(X)
    if X.ndim != 2:
        raise ValueError(f"adapters.check: X must be 2-D, got {X.ndim}-D.")
    probe = X[:2]
    try:
        y = model(probe)
    except Exception as e:
        raise type(e)(
            f"adapters.check: model raised on a 2-row numpy probe — it is "
            f"not numpy-in/numpy-out as effector requires. Original error: {e}"
        ) from e
    _validate_output(y, len(probe), "adapters.check")
    if model_jac is not None:
        if not callable(model_jac):
            raise TypeError(
                f"adapters.check: model_jac is not callable (got "
                f"{type(model_jac).__name__})."
            )
        jac = np.asarray(model_jac(probe))
        if jac.shape != probe.shape:
            raise ValueError(
                f"adapters.check: model_jac output has shape {jac.shape}, "
                f"expected {probe.shape} (one gradient row per input row)."
            )

classifier_proba(estimator, class_=1)

Wrap a classifier into a numpy->numpy callable for one class' probability.

model = effector.adapters.classifier_proba(clf, class_="yes")
pdp = effector.PDP(X, model, schema=schema)

The per-class probability is effector's classification story: the returned callable computes predict_proba(X)[:, k] — the surface P(class = k) — which every engine explains like any regression output. One explanation per class; loop over classes yourself if you want them all.

How class_ is resolved

Against estimator.classes_ at wrap time: label match first, positional column index as fallback for plain ints.

Parameters:

Name Type Description Default
estimator

a fitted object with .predict_proba(X) -> (N, C) and .classes_.

required
class_

the class to explain — a label from classes_ or a positional column index (default 1, the positive class of a binary classifier).

1

Returns:

Type Description
Callable

a plain callable f(X) -> (N,) of probabilities in [0, 1].

Raises:

Type Description
TypeError

no predict_proba, or no .classes_ (is it fitted?).

ValueError

class_ matches neither a label nor a valid column index.

Source code in effector/adapters.py
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
def classifier_proba(estimator, class_=1) -> typing.Callable:
    """Wrap a classifier into a numpy->numpy callable for one class' probability.

    ```python
    model = effector.adapters.classifier_proba(clf, class_="yes")
    pdp = effector.PDP(X, model, schema=schema)
    ```

    The per-class probability is effector's classification story: the returned
    callable computes ``predict_proba(X)[:, k]`` — the surface P(class = k) —
    which every engine explains like any regression output. One explanation
    per class; loop over classes yourself if you want them all.

    !!! note "How `class_` is resolved"
        Against ``estimator.classes_`` at wrap time: label match first,
        positional column index as fallback for plain ints.

    Args:
        estimator: a fitted object with ``.predict_proba(X) -> (N, C)`` and
            ``.classes_``.
        class_: the class to explain — a label from ``classes_`` or a
            positional column index (default `1`, the positive class of a
            binary classifier).

    Returns:
        a plain callable ``f(X) -> (N,)`` of probabilities in [0, 1].

    Raises:
        TypeError: no ``predict_proba``, or no ``.classes_`` (is it fitted?).
        ValueError: `class_` matches neither a label nor a valid column index.
    """
    if not hasattr(estimator, "predict_proba"):
        raise TypeError(
            "adapters.classifier_proba: object has no .predict_proba method; "
            "for a regressor use effector.adapters.from_sklearn."
        )
    classes = np.asarray(getattr(estimator, "classes_", []))
    if classes.size == 0:
        raise TypeError(
            "adapters.classifier_proba: estimator has no .classes_; is it fitted?"
        )
    matches = np.flatnonzero(classes == class_)
    if matches.size == 1:
        col = int(matches[0])
    elif isinstance(class_, (int, np.integer)) and 0 <= class_ < classes.size:
        col = int(class_)
    else:
        raise ValueError(
            f"adapters.classifier_proba: class_={class_!r} is neither a label "
            f"in classes_={classes.tolist()} nor a valid column index."
        )

    def model(X: np.ndarray) -> np.ndarray:
        proba = np.asarray(estimator.predict_proba(X))
        if proba.ndim != 2 or proba.shape[0] != len(X):
            raise ValueError(
                f"adapters.classifier_proba: predict_proba returned shape "
                f"{proba.shape}, expected ({len(X)}, n_classes)."
            )
        return _validate_output(proba[:, col], len(X), "adapters.classifier_proba")

    return model

from_sklearn(estimator)

Wrap an sklearn-style regressor into a numpy->numpy callable.

model = effector.adapters.from_sklearn(est)
pdp = effector.PDP(X, model, schema=schema)

The returned function calls estimator.predict and validates the output (shape (N,), numeric) on every call.

Classifiers are rejected

Anything with predict_proba raises — class labels are not a regression surface. Explain a per-class probability instead: effector.adapters.classifier_proba(clf, class_=1).

Parameters:

Name Type Description Default
estimator

a fitted object with .predict(X) -> (N,) (an sklearn regressor or pipeline; anything duck-typing it works).

required

Returns:

Type Description
Callable

a plain callable f(X: (N, D)) -> (N,) — pass it to an engine

Callable

yourself.

Raises:

Type Description
TypeError

the object has no .predict method.

ValueError

the object is a classifier (has predict_proba).

Source code in effector/adapters.py
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
def from_sklearn(estimator) -> typing.Callable:
    """Wrap an sklearn-style regressor into a numpy->numpy callable.

    ```python
    model = effector.adapters.from_sklearn(est)
    pdp = effector.PDP(X, model, schema=schema)
    ```

    The returned function calls ``estimator.predict`` and validates the
    output (shape ``(N,)``, numeric) on every call.

    !!! warning "Classifiers are rejected"
        Anything with `predict_proba` raises — class labels are not a
        regression surface. Explain a per-class probability instead:
        ``effector.adapters.classifier_proba(clf, class_=1)``.

    Args:
        estimator: a fitted object with ``.predict(X) -> (N,)`` (an sklearn
            regressor or pipeline; anything duck-typing it works).

    Returns:
        a plain callable ``f(X: (N, D)) -> (N,)`` — pass it to an engine
        yourself.

    Raises:
        TypeError: the object has no ``.predict`` method.
        ValueError: the object is a classifier (has ``predict_proba``).
    """
    if not hasattr(estimator, "predict"):
        raise TypeError(
            "adapters.from_sklearn: object has no .predict method; "
            "pass a fitted sklearn-style estimator (or wrap your model as a "
            "numpy->numpy callable yourself)."
        )
    if hasattr(estimator, "predict_proba"):
        raise ValueError(
            "adapters.from_sklearn: this estimator is a classifier "
            "(has predict_proba). Explaining predicted labels is not "
            "meaningful; explain a per-class probability instead:\n"
            "    model = effector.adapters.classifier_proba(estimator, class_=...)"
        )

    def model(X: np.ndarray) -> np.ndarray:
        return _validate_output(estimator.predict(X), len(X), "adapters.from_sklearn")

    return model

from_torch(module, device=None, jacobian=False)

Wrap a torch module into numpy->numpy callable(s).

model = effector.adapters.from_torch(net)
model, model_jac = effector.adapters.from_torch(net, jacobian=True)
rhale = effector.RHALE(X, model, model_jac, schema=schema)

The forward wrapper puts the module in eval mode, runs under no_grad, and moves tensors to/from device. torch is imported lazily — effector gains no torch dependency.

jacobian=True returns a tuple

You get (model, model_jac), not a single callable. The jacobian is built on autograd via the sum-backward trick — valid because each row's output depends only on that row's input — and is exactly what RHALE / DerPDP want.

Parameters:

Name Type Description Default
module

a torch.nn.Module whose forward maps (N, D) to (N,) or (N, 1).

required
device

torch device for inference; default = the module's own.

None
jacobian bool

also build jac(X) -> (N, D) via autograd.

False

Returns:

Type Description

model — or the tuple (model, model_jac) when

jacobian=True. Pass them to an engine yourself.

Source code in effector/adapters.py
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
def from_torch(module, device=None, jacobian: bool = False):
    """Wrap a torch module into numpy->numpy callable(s).

    ```python
    model = effector.adapters.from_torch(net)
    model, model_jac = effector.adapters.from_torch(net, jacobian=True)
    rhale = effector.RHALE(X, model, model_jac, schema=schema)
    ```

    The forward wrapper puts the module in eval mode, runs under ``no_grad``,
    and moves tensors to/from `device`. torch is imported lazily — effector
    gains no torch dependency.

    !!! note "`jacobian=True` returns a *tuple*"
        You get ``(model, model_jac)``, not a single callable. The jacobian
        is built on autograd via the sum-backward trick — valid because each
        row's output depends only on that row's input — and is exactly what
        `RHALE` / `DerPDP` want.

    Args:
        module: a ``torch.nn.Module`` whose forward maps ``(N, D)`` to
            ``(N,)`` or ``(N, 1)``.
        device: torch device for inference; default = the module's own.
        jacobian: also build ``jac(X) -> (N, D)`` via autograd.

    Returns:
        ``model`` — or the tuple ``(model, model_jac)`` when
        ``jacobian=True``. Pass them to an engine yourself.
    """
    import torch

    module.eval()
    if device is None:
        try:
            device = next(module.parameters()).device
        except StopIteration:
            device = torch.device("cpu")

    def model(X: np.ndarray) -> np.ndarray:
        with torch.no_grad():
            t = torch.as_tensor(np.asarray(X), dtype=torch.float32, device=device)
            out = module(t)
        return _validate_output(
            out.detach().cpu().numpy(), len(X), "adapters.from_torch"
        )

    if not jacobian:
        return model

    def model_jac(X: np.ndarray) -> np.ndarray:
        t = torch.as_tensor(np.asarray(X), dtype=torch.float32, device=device)
        t.requires_grad_(True)
        out = module(t)
        if out.ndim == 2 and out.shape[1] == 1:
            out = out.ravel()
        if out.shape != (len(X),):
            raise ValueError(
                f"adapters.from_torch: module output has shape "
                f"{tuple(out.shape)}, expected ({len(X)},)."
            )
        out.sum().backward()
        return t.grad.detach().cpu().numpy().astype(float)

    return model, model_jac