Skip to content

Method adapters

Adapters are the stable boundary around paper-specific implementations. Their class attributes declare feature level and encoder compatibility; inherited methods follow the lifecycle documented by BaseMethod.

CompositeMethod 

CompositeMethod(cfg, device='cuda')

Bases: BaseMethod

Adapt the configurable selector/prompt/aggregator composition model.

Source code in methods/composite/adapter.py 
30
31
32
def __init__(self, cfg, device="cuda"):
    super().__init__(cfg, device)
    self._recipe = None

get_backbone_contract classmethod 

get_backbone_contract() -> MethodBackboneContract

Return the encoder/feature contract declared by the adapter.

Raises:

Type Description
NotImplementedError

If the adapter omitted its required contract.
Source code in methods/base.py 
103
104
105
106
107
108
109
110
111
112
113
@classmethod
def get_backbone_contract(cls) -> MethodBackboneContract:
    """Return the encoder/feature contract declared by the adapter.

    Raises:
        NotImplementedError: If the adapter omitted its required contract.
    """
    if cls.backbone_contract is None:
        raise NotImplementedError(
            f"Method adapter {cls.__name__} has no backbone contract")
    return cls.backbone_contract

load_encoder 

load_encoder(
    *,
    weights_path: Optional[str] = None,
    **loader_options: Any,
) -> EncoderBundle

Build and validate the configured encoder in one operation.

Parameters:

Name Type Description Default
weights_path Optional[str]

Optional local checkpoint or model identifier. When omitted, cfg['backbone_weights'] is used.

 None
**loader_options Any

Family-specific options forwarded to the encoder registry, such as a pinned revision or offline-only loading.

 {}

Returns:

Type Description
EncoderBundle

A validated capability-aware encoder bundle.

Raises:

Type Description
RuntimeError

If the adapter has no runtime backbone contract.
BackboneCompatibilityError

If the bundle violates the method contract or declared feature provenance.
Source code in methods/base.py 
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
def load_encoder(self, *, weights_path: Optional[str] = None,
                 **loader_options: Any) -> EncoderBundle:
    """Build and validate the configured encoder in one operation.

    Args:
        weights_path: Optional local checkpoint or model identifier. When
            omitted, ``cfg['backbone_weights']`` is used.
        **loader_options: Family-specific options forwarded to the encoder
            registry, such as a pinned revision or offline-only loading.

    Returns:
        A validated capability-aware encoder bundle.

    Raises:
        RuntimeError: If the adapter has no runtime backbone contract.
        BackboneCompatibilityError: If the bundle violates the method
            contract or declared feature provenance.
    """
    if self.backbone_contract is None or self.backbone_name is None:
        raise RuntimeError(f"Method '{self.name}' has no runtime backbone")
    from common.backbones import build_encoder
    if weights_path is None:
        weights_path = self.cfg.get("backbone_weights")
    bundle = build_encoder(self.backbone_name, weights_path=weights_path,
                           device=self.device, **loader_options)
    return self.backbone_contract.validate_bundle(self.cfg, bundle)

validation_monitor 

validation_monitor(
    metrics: Dict[str, Any],
) -> tuple[str, float]

Return the minimized value used for best-checkpoint selection.

The unified loop cannot assume that every released method selects its checkpoint by cross-entropy. Several paper repositories use validation error, while MSCPT uses a task-dependent F1/AUC or F1/ACC composite. Adapters override this hook to preserve that executable recipe. Values are always expressed as a minimization objective so the checkpoint writer has one unambiguous contract.

Source code in methods/base.py 
190
191
192
193
194
195
196
197
198
199
200
201
202
def validation_monitor(
    self, metrics: Dict[str, Any],
) -> tuple[str, float]:
    """Return the minimized value used for best-checkpoint selection.

    The unified loop cannot assume that every released method selects its
    checkpoint by cross-entropy.  Several paper repositories use
    validation error, while MSCPT uses a task-dependent F1/AUC or F1/ACC
    composite.  Adapters override this hook to preserve that executable
    recipe.  Values are always expressed as a minimization objective so
    the checkpoint writer has one unambiguous contract.
    """
    return "val_loss", float(metrics["val_loss"])

prepare_fold 

prepare_fold(
    fold: int, model: Module, train_loader: Any
) -> None

Prepare training-split-only state after model/data construction.

Methods such as PathPT use this boundary to select prompts without seeing validation or test slides. The default is deliberately a no-op.

Source code in methods/base.py 
204
205
206
207
208
209
210
211
def prepare_fold(
    self, fold: int, model: nn.Module, train_loader: Any,
) -> None:
    """Prepare training-split-only state after model/data construction.

    Methods such as PathPT use this boundary to select prompts without
    seeing validation or test slides. The default is deliberately a no-op.
    """

on_train_epoch_start 

on_train_epoch_start(epoch: int, model: Module) -> None

Handle the boundary immediately before an epoch's first batch.

Source code in methods/base.py 
213
214
def on_train_epoch_start(self, epoch: int, model: nn.Module) -> None:
    """Handle the boundary immediately before an epoch's first batch."""

on_epoch_end 

on_epoch_end(epoch: int, metrics: Dict[str, float]) -> None

Handle an optional callback after a full train/validation epoch.

Source code in methods/base.py 
216
217
def on_epoch_end(self, epoch: int, metrics: Dict[str, float]) -> None:
    """Handle an optional callback after a full train/validation epoch."""

on_validation_end 

on_validation_end(
    epoch: int, metrics: Dict[str, float]
) -> None

Handle a completed validation pass, including epoch -1.

The default forwards to the historical on_epoch_end hook. Adapters which accumulate validation predictions therefore get a clear boundary for the optional initial evaluation as well as every trained epoch.

Source code in methods/base.py 
219
220
221
222
223
224
225
226
227
228
def on_validation_end(
    self, epoch: int, metrics: Dict[str, float],
) -> None:
    """Handle a completed validation pass, including epoch ``-1``.

    The default forwards to the historical ``on_epoch_end`` hook. Adapters
    which accumulate validation predictions therefore get a clear boundary
    for the optional initial evaluation as well as every trained epoch.
    """
    self.on_epoch_end(epoch, metrics)

on_fold_end 

on_fold_end(fold: int, metrics: Dict[str, float]) -> None

Handle an optional callback after a cross-validation fold.

Source code in methods/base.py 
230
231
def on_fold_end(self, fold: int, metrics: Dict[str, float]) -> None:
    """Handle an optional callback after a cross-validation fold."""

on_checkpoint_loaded 

on_checkpoint_loaded(
    model: Module, checkpoint_kind: str, fold: int
) -> None

Restore adapter state not represented by model.state_dict().

Source code in methods/base.py 
233
234
235
236
def on_checkpoint_loaded(
    self, model: nn.Module, checkpoint_kind: str, fold: int,
) -> None:
    """Restore adapter state not represented by ``model.state_dict()``."""

interpret_step 

interpret_step(
    batch: Any,
    model: Module,
    target_class: int | None = None,
) -> "InterpretabilityResult"

Return native, patch-aligned evidence for one evaluation slide.

Adapters must override this hook only when their score semantics and patch ordering are known. The default intentionally refuses to invent a generic "attention" map from hidden activations.

Parameters:

Name Type Description Default
batch Any

One loader batch for a single slide.

 required
model Module

Loaded fold checkpoint in evaluation mode.

 required
target_class int | None

Optional class index for class-specific evidence.

 None

Raises:

Type Description
NotImplementedError

If the method has no audited patch-evidence mapping in PGVL-Gym.
Source code in methods/base.py 
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
def interpret_step(
    self, batch: Any, model: nn.Module, target_class: int | None = None,
) -> "InterpretabilityResult":
    """Return native, patch-aligned evidence for one evaluation slide.

    Adapters must override this hook only when their score semantics and
    patch ordering are known. The default intentionally refuses to invent
    a generic "attention" map from hidden activations.

    Args:
        batch: One loader batch for a single slide.
        model: Loaded fold checkpoint in evaluation mode.
        target_class: Optional class index for class-specific evidence.

    Raises:
        NotImplementedError: If the method has no audited patch-evidence
            mapping in PGVL-Gym.
    """
    del batch, model, target_class
    raise NotImplementedError(
        f"Method '{self.name}' has no audited patch-level "
        "interpretability provider")

FOCUSMethod 

FOCUSMethod(cfg, device='cuda')

Bases: BaseMethod

Adapt FOCUS's single high-resolution bag and CONCH soft prompts.

Source code in methods/focus/adapter.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def __init__(self, cfg, device="cuda"):
    super().__init__(cfg, device)
    self.is_encoder_extension = self.cfg.get("encoder_extension") is not None
    if self.is_encoder_extension:
        validate_encoder_extension(self.name, self.cfg)
        if self.cfg.get("encoder_extension_strategy") != \
                "paired_feature_context_v1":
            raise ValueError(
                "FOCUS encoder extensions require "
                "encoder_extension_strategy=paired_feature_context_v1")
    elif self.backbone_name != "conch":
        raise ValueError(
            "Non-CONCH FOCUS requires serialized encoder_extension "
            "provenance")

interpret_step 

interpret_step(batch, model, target_class=None)

Expose FOCUS's native class-query compression attention.

Source code in methods/focus/adapter.py 
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
@torch.no_grad()
def interpret_step(self, batch, model, target_class=None):
    """Expose FOCUS's native class-query compression attention."""
    features, label = self._features_and_label(batch)
    features, label = features.to(self.device), label.to(self.device)
    if self.is_encoder_extension:
        features = project_paired_features(model, features)
    details = model(
        features, features, torch.zeros_like(label),
        return_details=True)
    logits = details["logits"]
    class_index = (
        int(logits.argmax(dim=1).item())
        if target_class is None else int(target_class))
    if class_index < 0 or class_index >= logits.shape[1]:
        raise ValueError("FOCUS target class is outside the classifier range")
    return InterpretabilityResult(
        logits=logits,
        evidence=(PatchEvidence(
            scores=details["patch_attention"][class_index],
            score_type="class_query_cross_attention",
            description=(
                "Native FOCUS class-query cross-attention after adaptive "
                "selection and spatial compression; unselected patches "
                "have zero weight."),
            class_index=class_index,
            feature_path_key=(
                "feature_path_column" if self.cfg.get(
                    "feature_path_column") else "feature_path_column_l"),
        ),),
    )

get_backbone_contract classmethod 

get_backbone_contract() -> MethodBackboneContract

Return the encoder/feature contract declared by the adapter.

Raises:

Type Description
NotImplementedError

If the adapter omitted its required contract.
Source code in methods/base.py 
103
104
105
106
107
108
109
110
111
112
113
@classmethod
def get_backbone_contract(cls) -> MethodBackboneContract:
    """Return the encoder/feature contract declared by the adapter.

    Raises:
        NotImplementedError: If the adapter omitted its required contract.
    """
    if cls.backbone_contract is None:
        raise NotImplementedError(
            f"Method adapter {cls.__name__} has no backbone contract")
    return cls.backbone_contract

load_encoder 

load_encoder(
    *,
    weights_path: Optional[str] = None,
    **loader_options: Any,
) -> EncoderBundle

Build and validate the configured encoder in one operation.

Parameters:

Name Type Description Default
weights_path Optional[str]

Optional local checkpoint or model identifier. When omitted, cfg['backbone_weights'] is used.

 None
**loader_options Any

Family-specific options forwarded to the encoder registry, such as a pinned revision or offline-only loading.

 {}

Returns:

Type Description
EncoderBundle

A validated capability-aware encoder bundle.

Raises:

Type Description
RuntimeError

If the adapter has no runtime backbone contract.
BackboneCompatibilityError

If the bundle violates the method contract or declared feature provenance.
Source code in methods/base.py 
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
def load_encoder(self, *, weights_path: Optional[str] = None,
                 **loader_options: Any) -> EncoderBundle:
    """Build and validate the configured encoder in one operation.

    Args:
        weights_path: Optional local checkpoint or model identifier. When
            omitted, ``cfg['backbone_weights']`` is used.
        **loader_options: Family-specific options forwarded to the encoder
            registry, such as a pinned revision or offline-only loading.

    Returns:
        A validated capability-aware encoder bundle.

    Raises:
        RuntimeError: If the adapter has no runtime backbone contract.
        BackboneCompatibilityError: If the bundle violates the method
            contract or declared feature provenance.
    """
    if self.backbone_contract is None or self.backbone_name is None:
        raise RuntimeError(f"Method '{self.name}' has no runtime backbone")
    from common.backbones import build_encoder
    if weights_path is None:
        weights_path = self.cfg.get("backbone_weights")
    bundle = build_encoder(self.backbone_name, weights_path=weights_path,
                           device=self.device, **loader_options)
    return self.backbone_contract.validate_bundle(self.cfg, bundle)

build_optimizer 

build_optimizer(model: Module) -> Optimizer

Build Adam over trainable parameters using the configured recipe.

Source code in methods/base.py 
172
173
174
175
176
177
178
179
def build_optimizer(self, model: nn.Module) -> torch.optim.Optimizer:
    """Build Adam over trainable parameters using the configured recipe."""
    params = filter(lambda p: p.requires_grad, model.parameters())
    return torch.optim.Adam(
        params,
        lr=self.cfg.get("lr", 1e-4),
        weight_decay=self.cfg.get("weight_decay", 1e-5),
    )

prepare_fold 

prepare_fold(
    fold: int, model: Module, train_loader: Any
) -> None

Prepare training-split-only state after model/data construction.

Methods such as PathPT use this boundary to select prompts without seeing validation or test slides. The default is deliberately a no-op.

Source code in methods/base.py 
204
205
206
207
208
209
210
211
def prepare_fold(
    self, fold: int, model: nn.Module, train_loader: Any,
) -> None:
    """Prepare training-split-only state after model/data construction.

    Methods such as PathPT use this boundary to select prompts without
    seeing validation or test slides. The default is deliberately a no-op.
    """

on_train_epoch_start 

on_train_epoch_start(epoch: int, model: Module) -> None

Handle the boundary immediately before an epoch's first batch.

Source code in methods/base.py 
213
214
def on_train_epoch_start(self, epoch: int, model: nn.Module) -> None:
    """Handle the boundary immediately before an epoch's first batch."""

on_epoch_end 

on_epoch_end(epoch: int, metrics: Dict[str, float]) -> None

Handle an optional callback after a full train/validation epoch.

Source code in methods/base.py 
216
217
def on_epoch_end(self, epoch: int, metrics: Dict[str, float]) -> None:
    """Handle an optional callback after a full train/validation epoch."""

on_validation_end 

on_validation_end(
    epoch: int, metrics: Dict[str, float]
) -> None

Handle a completed validation pass, including epoch -1.

The default forwards to the historical on_epoch_end hook. Adapters which accumulate validation predictions therefore get a clear boundary for the optional initial evaluation as well as every trained epoch.

Source code in methods/base.py 
219
220
221
222
223
224
225
226
227
228
def on_validation_end(
    self, epoch: int, metrics: Dict[str, float],
) -> None:
    """Handle a completed validation pass, including epoch ``-1``.

    The default forwards to the historical ``on_epoch_end`` hook. Adapters
    which accumulate validation predictions therefore get a clear boundary
    for the optional initial evaluation as well as every trained epoch.
    """
    self.on_epoch_end(epoch, metrics)

on_fold_end 

on_fold_end(fold: int, metrics: Dict[str, float]) -> None

Handle an optional callback after a cross-validation fold.

Source code in methods/base.py 
230
231
def on_fold_end(self, fold: int, metrics: Dict[str, float]) -> None:
    """Handle an optional callback after a cross-validation fold."""

on_checkpoint_loaded 

on_checkpoint_loaded(
    model: Module, checkpoint_kind: str, fold: int
) -> None

Restore adapter state not represented by model.state_dict().

Source code in methods/base.py 
233
234
235
236
def on_checkpoint_loaded(
    self, model: nn.Module, checkpoint_kind: str, fold: int,
) -> None:
    """Restore adapter state not represented by ``model.state_dict()``."""

ViLaMILMethod 

ViLaMILMethod(cfg, device='cuda')

Bases: BaseMethod

Adapt ViLa-MIL's CLIP-RN50 dual-scale patch-bag workflow.

Source code in methods/vila_mil/adapter.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
def __init__(self, cfg, device="cuda"):
    super().__init__(cfg, device)
    self.is_encoder_extension = self.cfg.get("encoder_extension") is not None
    if self.is_encoder_extension:
        validate_encoder_extension(self.name, self.cfg)
        if self.cfg.get("encoder_extension_strategy") != \
                "paired_feature_context_v1":
            raise ValueError(
                "ViLa-MIL encoder extensions require "
                "encoder_extension_strategy=paired_feature_context_v1")
    elif self.backbone_name != "clip-rn50":
        raise ValueError(
            "Non-RN50 ViLa-MIL requires serialized encoder_extension "
            "provenance")

interpret_step 

interpret_step(batch, model, target_class=None)

Expose ViLa-MIL's native class-to-patch context attention.

Source code in methods/vila_mil/adapter.py 
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
@torch.no_grad()
def interpret_step(self, batch, model, target_class=None):
    """Expose ViLa-MIL's native class-to-patch context attention."""
    x_s, x_l, label = batch[0], batch[1], batch[-1]
    x_s = self._slide_bag(x_s.to(self.device))
    x_l = self._slide_bag(x_l.to(self.device))
    if self.is_encoder_extension:
        x_s = project_paired_features(model, x_s)
        x_l = project_paired_features(model, x_l)
    label = label.to(self.device)
    coord_s = torch.zeros(x_s.shape[0], 2, device=self.device)
    coord_l = torch.zeros(x_l.shape[0], 2, device=self.device)
    details = model(
        x_s, coord_s, x_l, coord_l, torch.zeros_like(label),
        return_details=True)
    logits = details["logits"]
    class_index = (
        int(logits.argmax(dim=1).item())
        if target_class is None else int(target_class))
    if class_index < 0 or class_index >= logits.shape[1]:
        raise ValueError("ViLa-MIL target class is outside the classifier range")
    description = (
        "Native ViLa-MIL class-prompt cross-attention restricted to "
        "patch keys; learned prototype keys are not painted on the WSI.")
    return InterpretabilityResult(
        logits=logits,
        evidence=(
            PatchEvidence(
                scores=details["low_patch_attention"][class_index],
                score_type="class_context_cross_attention",
                description=description,
                scale="low",
                class_index=class_index,
                feature_path_key="feature_path_column_s",
            ),
            PatchEvidence(
                scores=details["high_patch_attention"][class_index],
                score_type="class_context_cross_attention",
                description=description,
                scale="high",
                class_index=class_index,
                feature_path_key="feature_path_column_l",
            ),
        ),
    )

get_backbone_contract classmethod 

get_backbone_contract() -> MethodBackboneContract

Return the encoder/feature contract declared by the adapter.

Raises:

Type Description
NotImplementedError

If the adapter omitted its required contract.
Source code in methods/base.py 
103
104
105
106
107
108
109
110
111
112
113
@classmethod
def get_backbone_contract(cls) -> MethodBackboneContract:
    """Return the encoder/feature contract declared by the adapter.

    Raises:
        NotImplementedError: If the adapter omitted its required contract.
    """
    if cls.backbone_contract is None:
        raise NotImplementedError(
            f"Method adapter {cls.__name__} has no backbone contract")
    return cls.backbone_contract

load_encoder 

load_encoder(
    *,
    weights_path: Optional[str] = None,
    **loader_options: Any,
) -> EncoderBundle

Build and validate the configured encoder in one operation.

Parameters:

Name Type Description Default
weights_path Optional[str]

Optional local checkpoint or model identifier. When omitted, cfg['backbone_weights'] is used.

 None
**loader_options Any

Family-specific options forwarded to the encoder registry, such as a pinned revision or offline-only loading.

 {}

Returns:

Type Description
EncoderBundle

A validated capability-aware encoder bundle.

Raises:

Type Description
RuntimeError

If the adapter has no runtime backbone contract.
BackboneCompatibilityError

If the bundle violates the method contract or declared feature provenance.
Source code in methods/base.py 
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
def load_encoder(self, *, weights_path: Optional[str] = None,
                 **loader_options: Any) -> EncoderBundle:
    """Build and validate the configured encoder in one operation.

    Args:
        weights_path: Optional local checkpoint or model identifier. When
            omitted, ``cfg['backbone_weights']`` is used.
        **loader_options: Family-specific options forwarded to the encoder
            registry, such as a pinned revision or offline-only loading.

    Returns:
        A validated capability-aware encoder bundle.

    Raises:
        RuntimeError: If the adapter has no runtime backbone contract.
        BackboneCompatibilityError: If the bundle violates the method
            contract or declared feature provenance.
    """
    if self.backbone_contract is None or self.backbone_name is None:
        raise RuntimeError(f"Method '{self.name}' has no runtime backbone")
    from common.backbones import build_encoder
    if weights_path is None:
        weights_path = self.cfg.get("backbone_weights")
    bundle = build_encoder(self.backbone_name, weights_path=weights_path,
                           device=self.device, **loader_options)
    return self.backbone_contract.validate_bundle(self.cfg, bundle)

build_optimizer 

build_optimizer(model: Module) -> Optimizer

Build Adam over trainable parameters using the configured recipe.

Source code in methods/base.py 
172
173
174
175
176
177
178
179
def build_optimizer(self, model: nn.Module) -> torch.optim.Optimizer:
    """Build Adam over trainable parameters using the configured recipe."""
    params = filter(lambda p: p.requires_grad, model.parameters())
    return torch.optim.Adam(
        params,
        lr=self.cfg.get("lr", 1e-4),
        weight_decay=self.cfg.get("weight_decay", 1e-5),
    )

prepare_fold 

prepare_fold(
    fold: int, model: Module, train_loader: Any
) -> None

Prepare training-split-only state after model/data construction.

Methods such as PathPT use this boundary to select prompts without seeing validation or test slides. The default is deliberately a no-op.

Source code in methods/base.py 
204
205
206
207
208
209
210
211
def prepare_fold(
    self, fold: int, model: nn.Module, train_loader: Any,
) -> None:
    """Prepare training-split-only state after model/data construction.

    Methods such as PathPT use this boundary to select prompts without
    seeing validation or test slides. The default is deliberately a no-op.
    """

on_train_epoch_start 

on_train_epoch_start(epoch: int, model: Module) -> None

Handle the boundary immediately before an epoch's first batch.

Source code in methods/base.py 
213
214
def on_train_epoch_start(self, epoch: int, model: nn.Module) -> None:
    """Handle the boundary immediately before an epoch's first batch."""

on_epoch_end 

on_epoch_end(epoch: int, metrics: Dict[str, float]) -> None

Handle an optional callback after a full train/validation epoch.

Source code in methods/base.py 
216
217
def on_epoch_end(self, epoch: int, metrics: Dict[str, float]) -> None:
    """Handle an optional callback after a full train/validation epoch."""

on_validation_end 

on_validation_end(
    epoch: int, metrics: Dict[str, float]
) -> None

Handle a completed validation pass, including epoch -1.

The default forwards to the historical on_epoch_end hook. Adapters which accumulate validation predictions therefore get a clear boundary for the optional initial evaluation as well as every trained epoch.

Source code in methods/base.py 
219
220
221
222
223
224
225
226
227
228
def on_validation_end(
    self, epoch: int, metrics: Dict[str, float],
) -> None:
    """Handle a completed validation pass, including epoch ``-1``.

    The default forwards to the historical ``on_epoch_end`` hook. Adapters
    which accumulate validation predictions therefore get a clear boundary
    for the optional initial evaluation as well as every trained epoch.
    """
    self.on_epoch_end(epoch, metrics)

on_fold_end 

on_fold_end(fold: int, metrics: Dict[str, float]) -> None

Handle an optional callback after a cross-validation fold.

Source code in methods/base.py 
230
231
def on_fold_end(self, fold: int, metrics: Dict[str, float]) -> None:
    """Handle an optional callback after a cross-validation fold."""

on_checkpoint_loaded 

on_checkpoint_loaded(
    model: Module, checkpoint_kind: str, fold: int
) -> None

Restore adapter state not represented by model.state_dict().

Source code in methods/base.py 
233
234
235
236
def on_checkpoint_loaded(
    self, model: nn.Module, checkpoint_kind: str, fold: int,
) -> None:
    """Restore adapter state not represented by ``model.state_dict()``."""

CoDMILMethod 

CoDMILMethod(cfg: Dict[str, Any], device: str = 'cuda')

Bases: BaseMethod

Adapt CoD-MIL's precomputed prompts and cross-scale correspondence.

Source code in methods/base.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def __init__(self, cfg: Dict[str, Any], device: str = "cuda"):
    # A few vendored adapters fill in derived defaults while constructing
    # their model.  Keeping those writes private is essential: train.py has
    # already snapshotted and fingerprinted the resolved run config, and a
    # later mutation would otherwise make an interrupted run impossible to
    # resume.  A deep copy also prevents nested recipe dictionaries from
    # carrying state between fold-specific adapter instances.
    self.cfg = deepcopy(cfg)
    self.device = device
    self.backbone_name = (self.backbone_contract.validate_config(self.cfg)
                          if self.backbone_contract is not None else None)

build_scheduler 

build_scheduler(optimizer)

Match the released loop, which constructs but never steps one.

Source code in methods/cod_mil/adapter.py 
58
59
60
def build_scheduler(self, optimizer):
    """Match the released loop, which constructs but never steps one."""
    return None

interpret_step 

interpret_step(batch, model, target_class=None)

Expose CoD-MIL's native predicted-class low-scale evidence.

Source code in methods/cod_mil/adapter.py 
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
@torch.no_grad()
def interpret_step(self, batch, model, target_class=None):
    """Expose CoD-MIL's native predicted-class low-scale evidence."""
    x_s, x_l, cross_map, label = batch[0], batch[1], batch[2], batch[-1]
    x_s, x_l = x_s.to(self.device), x_l.to(self.device)
    cross_map, label = cross_map.to(self.device), label.to(self.device)
    patch_label = torch.zeros(
        x_s.shape[0], dtype=torch.long, device=self.device)
    output = model(
        x_s, None, x_l, None, patch_label, torch.zeros_like(label),
        self._prepare_text_features(), cross_map)
    logits, attention = output[0], output[4]
    predicted = int(logits.argmax(dim=1).item())
    if target_class is not None and int(target_class) != predicted:
        raise ValueError(
            "CoD-MIL's released forward exposes evidence only for its "
            f"predicted class ({predicted}), not requested class "
            f"{int(target_class)}")
    return InterpretabilityResult(
        logits=logits,
        evidence=(PatchEvidence(
            scores=attention.reshape(-1),
            score_type="predicted_class_prompt_evidence",
            description=(
                "Native CoD-MIL low-scale prompt similarity used to "
                "select the linked high-scale diagnostic patches."),
            scale="low",
            class_index=predicted,
            feature_path_key="feature_path_column_s",
        ),),
        notes=(
            "The released CoD-MIL path emits the prompt map for the "
            "predicted class only.",
        ),
    )

get_backbone_contract classmethod 

get_backbone_contract() -> MethodBackboneContract

Return the encoder/feature contract declared by the adapter.

Raises:

Type Description
NotImplementedError

If the adapter omitted its required contract.
Source code in methods/base.py 
103
104
105
106
107
108
109
110
111
112
113
@classmethod
def get_backbone_contract(cls) -> MethodBackboneContract:
    """Return the encoder/feature contract declared by the adapter.

    Raises:
        NotImplementedError: If the adapter omitted its required contract.
    """
    if cls.backbone_contract is None:
        raise NotImplementedError(
            f"Method adapter {cls.__name__} has no backbone contract")
    return cls.backbone_contract

load_encoder 

load_encoder(
    *,
    weights_path: Optional[str] = None,
    **loader_options: Any,
) -> EncoderBundle

Build and validate the configured encoder in one operation.

Parameters:

Name Type Description Default
weights_path Optional[str]

Optional local checkpoint or model identifier. When omitted, cfg['backbone_weights'] is used.

 None
**loader_options Any

Family-specific options forwarded to the encoder registry, such as a pinned revision or offline-only loading.

 {}

Returns:

Type Description
EncoderBundle

A validated capability-aware encoder bundle.

Raises:

Type Description
RuntimeError

If the adapter has no runtime backbone contract.
BackboneCompatibilityError

If the bundle violates the method contract or declared feature provenance.
Source code in methods/base.py 
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
def load_encoder(self, *, weights_path: Optional[str] = None,
                 **loader_options: Any) -> EncoderBundle:
    """Build and validate the configured encoder in one operation.

    Args:
        weights_path: Optional local checkpoint or model identifier. When
            omitted, ``cfg['backbone_weights']`` is used.
        **loader_options: Family-specific options forwarded to the encoder
            registry, such as a pinned revision or offline-only loading.

    Returns:
        A validated capability-aware encoder bundle.

    Raises:
        RuntimeError: If the adapter has no runtime backbone contract.
        BackboneCompatibilityError: If the bundle violates the method
            contract or declared feature provenance.
    """
    if self.backbone_contract is None or self.backbone_name is None:
        raise RuntimeError(f"Method '{self.name}' has no runtime backbone")
    from common.backbones import build_encoder
    if weights_path is None:
        weights_path = self.cfg.get("backbone_weights")
    bundle = build_encoder(self.backbone_name, weights_path=weights_path,
                           device=self.device, **loader_options)
    return self.backbone_contract.validate_bundle(self.cfg, bundle)

build_optimizer 

build_optimizer(model: Module) -> Optimizer

Build Adam over trainable parameters using the configured recipe.

Source code in methods/base.py 
172
173
174
175
176
177
178
179
def build_optimizer(self, model: nn.Module) -> torch.optim.Optimizer:
    """Build Adam over trainable parameters using the configured recipe."""
    params = filter(lambda p: p.requires_grad, model.parameters())
    return torch.optim.Adam(
        params,
        lr=self.cfg.get("lr", 1e-4),
        weight_decay=self.cfg.get("weight_decay", 1e-5),
    )

validation_monitor 

validation_monitor(
    metrics: Dict[str, Any],
) -> tuple[str, float]

Return the minimized value used for best-checkpoint selection.

The unified loop cannot assume that every released method selects its checkpoint by cross-entropy. Several paper repositories use validation error, while MSCPT uses a task-dependent F1/AUC or F1/ACC composite. Adapters override this hook to preserve that executable recipe. Values are always expressed as a minimization objective so the checkpoint writer has one unambiguous contract.

Source code in methods/base.py 
190
191
192
193
194
195
196
197
198
199
200
201
202
def validation_monitor(
    self, metrics: Dict[str, Any],
) -> tuple[str, float]:
    """Return the minimized value used for best-checkpoint selection.

    The unified loop cannot assume that every released method selects its
    checkpoint by cross-entropy.  Several paper repositories use
    validation error, while MSCPT uses a task-dependent F1/AUC or F1/ACC
    composite.  Adapters override this hook to preserve that executable
    recipe.  Values are always expressed as a minimization objective so
    the checkpoint writer has one unambiguous contract.
    """
    return "val_loss", float(metrics["val_loss"])

prepare_fold 

prepare_fold(
    fold: int, model: Module, train_loader: Any
) -> None

Prepare training-split-only state after model/data construction.

Methods such as PathPT use this boundary to select prompts without seeing validation or test slides. The default is deliberately a no-op.

Source code in methods/base.py 
204
205
206
207
208
209
210
211
def prepare_fold(
    self, fold: int, model: nn.Module, train_loader: Any,
) -> None:
    """Prepare training-split-only state after model/data construction.

    Methods such as PathPT use this boundary to select prompts without
    seeing validation or test slides. The default is deliberately a no-op.
    """

on_train_epoch_start 

on_train_epoch_start(epoch: int, model: Module) -> None

Handle the boundary immediately before an epoch's first batch.

Source code in methods/base.py 
213
214
def on_train_epoch_start(self, epoch: int, model: nn.Module) -> None:
    """Handle the boundary immediately before an epoch's first batch."""

on_epoch_end 

on_epoch_end(epoch: int, metrics: Dict[str, float]) -> None

Handle an optional callback after a full train/validation epoch.

Source code in methods/base.py 
216
217
def on_epoch_end(self, epoch: int, metrics: Dict[str, float]) -> None:
    """Handle an optional callback after a full train/validation epoch."""

on_validation_end 

on_validation_end(
    epoch: int, metrics: Dict[str, float]
) -> None

Handle a completed validation pass, including epoch -1.

The default forwards to the historical on_epoch_end hook. Adapters which accumulate validation predictions therefore get a clear boundary for the optional initial evaluation as well as every trained epoch.

Source code in methods/base.py 
219
220
221
222
223
224
225
226
227
228
def on_validation_end(
    self, epoch: int, metrics: Dict[str, float],
) -> None:
    """Handle a completed validation pass, including epoch ``-1``.

    The default forwards to the historical ``on_epoch_end`` hook. Adapters
    which accumulate validation predictions therefore get a clear boundary
    for the optional initial evaluation as well as every trained epoch.
    """
    self.on_epoch_end(epoch, metrics)

on_fold_end 

on_fold_end(fold: int, metrics: Dict[str, float]) -> None

Handle an optional callback after a cross-validation fold.

Source code in methods/base.py 
230
231
def on_fold_end(self, fold: int, metrics: Dict[str, float]) -> None:
    """Handle an optional callback after a cross-validation fold."""

on_checkpoint_loaded 

on_checkpoint_loaded(
    model: Module, checkpoint_kind: str, fold: int
) -> None

Restore adapter state not represented by model.state_dict().

Source code in methods/base.py 
233
234
235
236
def on_checkpoint_loaded(
    self, model: nn.Module, checkpoint_kind: str, fold: int,
) -> None:
    """Restore adapter state not represented by ``model.state_dict()``."""

MAPLEMethod 

MAPLEMethod(cfg: Dict[str, Any], device: str = 'cuda')

Bases: BaseMethod

Adapt MAPLE's multiscale entity prompts and graph aggregation.

Source code in methods/base.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def __init__(self, cfg: Dict[str, Any], device: str = "cuda"):
    # A few vendored adapters fill in derived defaults while constructing
    # their model.  Keeping those writes private is essential: train.py has
    # already snapshotted and fingerprinted the resolved run config, and a
    # later mutation would otherwise make an interrupted run impossible to
    # resume.  A deep copy also prevents nested recipe dictionaries from
    # carrying state between fold-specific adapter instances.
    self.cfg = deepcopy(cfg)
    self.device = device
    self.backbone_name = (self.backbone_contract.validate_config(self.cfg)
                          if self.backbone_contract is not None else None)

build_optimizer 

build_optimizer(model: Module) -> Optimizer

Select the explicitly declared upstream MAPLE optimizer recipe.

The NeurIPS paper specifies AdamW, while the released main.py uses Adam. Keeping that choice in the run config makes the two upstream recipes reproducible without conflating their results.

Source code in methods/maple/adapter.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
def build_optimizer(self, model: nn.Module) -> torch.optim.Optimizer:
    """Select the explicitly declared upstream MAPLE optimizer recipe.

    The NeurIPS paper specifies AdamW, while the released ``main.py`` uses
    Adam.  Keeping that choice in the run config makes the two upstream
    recipes reproducible without conflating their results.
    """
    optimizer_name = str(self.cfg.get("optimizer", "adam")).lower()
    params = filter(lambda parameter: parameter.requires_grad,
                    model.parameters())
    kwargs = {
        "lr": float(self.cfg.get("lr", 2e-4)),
        "weight_decay": float(self.cfg.get("weight_decay", 1e-5)),
    }
    if optimizer_name == "adam":
        return torch.optim.Adam(params, **kwargs)
    if optimizer_name == "adamw":
        return torch.optim.AdamW(params, **kwargs)
    raise ValueError(
        "MAPLE optimizer must be 'adam' (released code) or 'adamw' "
        f"(paper), got {optimizer_name!r}")

get_backbone_contract classmethod 

get_backbone_contract() -> MethodBackboneContract

Return the encoder/feature contract declared by the adapter.

Raises:

Type Description
NotImplementedError

If the adapter omitted its required contract.
Source code in methods/base.py 
103
104
105
106
107
108
109
110
111
112
113
@classmethod
def get_backbone_contract(cls) -> MethodBackboneContract:
    """Return the encoder/feature contract declared by the adapter.

    Raises:
        NotImplementedError: If the adapter omitted its required contract.
    """
    if cls.backbone_contract is None:
        raise NotImplementedError(
            f"Method adapter {cls.__name__} has no backbone contract")
    return cls.backbone_contract

load_encoder 

load_encoder(
    *,
    weights_path: Optional[str] = None,
    **loader_options: Any,
) -> EncoderBundle

Build and validate the configured encoder in one operation.

Parameters:

Name Type Description Default
weights_path Optional[str]

Optional local checkpoint or model identifier. When omitted, cfg['backbone_weights'] is used.

 None
**loader_options Any

Family-specific options forwarded to the encoder registry, such as a pinned revision or offline-only loading.

 {}

Returns:

Type Description
EncoderBundle

A validated capability-aware encoder bundle.

Raises:

Type Description
RuntimeError

If the adapter has no runtime backbone contract.
BackboneCompatibilityError

If the bundle violates the method contract or declared feature provenance.
Source code in methods/base.py 
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
def load_encoder(self, *, weights_path: Optional[str] = None,
                 **loader_options: Any) -> EncoderBundle:
    """Build and validate the configured encoder in one operation.

    Args:
        weights_path: Optional local checkpoint or model identifier. When
            omitted, ``cfg['backbone_weights']`` is used.
        **loader_options: Family-specific options forwarded to the encoder
            registry, such as a pinned revision or offline-only loading.

    Returns:
        A validated capability-aware encoder bundle.

    Raises:
        RuntimeError: If the adapter has no runtime backbone contract.
        BackboneCompatibilityError: If the bundle violates the method
            contract or declared feature provenance.
    """
    if self.backbone_contract is None or self.backbone_name is None:
        raise RuntimeError(f"Method '{self.name}' has no runtime backbone")
    from common.backbones import build_encoder
    if weights_path is None:
        weights_path = self.cfg.get("backbone_weights")
    bundle = build_encoder(self.backbone_name, weights_path=weights_path,
                           device=self.device, **loader_options)
    return self.backbone_contract.validate_bundle(self.cfg, bundle)

prepare_fold 

prepare_fold(
    fold: int, model: Module, train_loader: Any
) -> None

Prepare training-split-only state after model/data construction.

Methods such as PathPT use this boundary to select prompts without seeing validation or test slides. The default is deliberately a no-op.

Source code in methods/base.py 
204
205
206
207
208
209
210
211
def prepare_fold(
    self, fold: int, model: nn.Module, train_loader: Any,
) -> None:
    """Prepare training-split-only state after model/data construction.

    Methods such as PathPT use this boundary to select prompts without
    seeing validation or test slides. The default is deliberately a no-op.
    """

on_train_epoch_start 

on_train_epoch_start(epoch: int, model: Module) -> None

Handle the boundary immediately before an epoch's first batch.

Source code in methods/base.py 
213
214
def on_train_epoch_start(self, epoch: int, model: nn.Module) -> None:
    """Handle the boundary immediately before an epoch's first batch."""

on_epoch_end 

on_epoch_end(epoch: int, metrics: Dict[str, float]) -> None

Handle an optional callback after a full train/validation epoch.

Source code in methods/base.py 
216
217
def on_epoch_end(self, epoch: int, metrics: Dict[str, float]) -> None:
    """Handle an optional callback after a full train/validation epoch."""

on_validation_end 

on_validation_end(
    epoch: int, metrics: Dict[str, float]
) -> None

Handle a completed validation pass, including epoch -1.

The default forwards to the historical on_epoch_end hook. Adapters which accumulate validation predictions therefore get a clear boundary for the optional initial evaluation as well as every trained epoch.

Source code in methods/base.py 
219
220
221
222
223
224
225
226
227
228
def on_validation_end(
    self, epoch: int, metrics: Dict[str, float],
) -> None:
    """Handle a completed validation pass, including epoch ``-1``.

    The default forwards to the historical ``on_epoch_end`` hook. Adapters
    which accumulate validation predictions therefore get a clear boundary
    for the optional initial evaluation as well as every trained epoch.
    """
    self.on_epoch_end(epoch, metrics)

on_fold_end 

on_fold_end(fold: int, metrics: Dict[str, float]) -> None

Handle an optional callback after a cross-validation fold.

Source code in methods/base.py 
230
231
def on_fold_end(self, fold: int, metrics: Dict[str, float]) -> None:
    """Handle an optional callback after a cross-validation fold."""

on_checkpoint_loaded 

on_checkpoint_loaded(
    model: Module, checkpoint_kind: str, fold: int
) -> None

Restore adapter state not represented by model.state_dict().

Source code in methods/base.py 
233
234
235
236
def on_checkpoint_loaded(
    self, model: nn.Module, checkpoint_kind: str, fold: int,
) -> None:
    """Restore adapter state not represented by ``model.state_dict()``."""

interpret_step 

interpret_step(
    batch: Any,
    model: Module,
    target_class: int | None = None,
) -> "InterpretabilityResult"

Return native, patch-aligned evidence for one evaluation slide.

Adapters must override this hook only when their score semantics and patch ordering are known. The default intentionally refuses to invent a generic "attention" map from hidden activations.

Parameters:

Name Type Description Default
batch Any

One loader batch for a single slide.

 required
model Module

Loaded fold checkpoint in evaluation mode.

 required
target_class int | None

Optional class index for class-specific evidence.

 None

Raises:

Type Description
NotImplementedError

If the method has no audited patch-evidence mapping in PGVL-Gym.
Source code in methods/base.py 
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
def interpret_step(
    self, batch: Any, model: nn.Module, target_class: int | None = None,
) -> "InterpretabilityResult":
    """Return native, patch-aligned evidence for one evaluation slide.

    Adapters must override this hook only when their score semantics and
    patch ordering are known. The default intentionally refuses to invent
    a generic "attention" map from hidden activations.

    Args:
        batch: One loader batch for a single slide.
        model: Loaded fold checkpoint in evaluation mode.
        target_class: Optional class index for class-specific evidence.

    Raises:
        NotImplementedError: If the method has no audited patch-evidence
            mapping in PGVL-Gym.
    """
    del batch, model, target_class
    raise NotImplementedError(
        f"Method '{self.name}' has no audited patch-level "
        "interpretability provider")

MSCPTMethod 

MSCPTMethod(cfg: Dict[str, Any], device: str = 'cuda')

Bases: BaseMethod

Adapt MSCPT's paired deep text/vision prompting implementation.

Source code in methods/base.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def __init__(self, cfg: Dict[str, Any], device: str = "cuda"):
    # A few vendored adapters fill in derived defaults while constructing
    # their model.  Keeping those writes private is essential: train.py has
    # already snapshotted and fingerprinted the resolved run config, and a
    # later mutation would otherwise make an interrupted run impossible to
    # resume.  A deep copy also prevents nested recipe dictionaries from
    # carrying state between fold-specific adapter instances.
    self.cfg = deepcopy(cfg)
    self.device = device
    self.backbone_name = (self.backbone_contract.validate_config(self.cfg)
                          if self.backbone_contract is not None else None)

validation_monitor 

validation_monitor(metrics)

Match MSCPT's task-dependent val_best_score checkpoint.

Source code in methods/mscpt/adapter.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
def validation_monitor(self, metrics):
    """Match MSCPT's task-dependent ``val_best_score`` checkpoint."""
    f1 = float(metrics["val_macro_f1"])
    dataset_name = str(self.cfg.get("dataset_name", ""))
    if dataset_name in {"RCC", "UBC-OCEAN"}:
        score = 0.5 * (f1 + float(metrics["val_accuracy"]))
        name = "negative_val_f1_acc"
    else:
        auc = metrics.get("val_auroc_ovr")
        if auc is None:
            raise ValueError(
                "MSCPT's upstream validation score requires AUROC for "
                f"dataset {dataset_name!r}")
        score = 0.5 * (f1 + float(auc))
        name = "negative_val_f1_auc"
    # BaseMethod's monitor contract is minimized; upstream maximizes score.
    return name, -score

get_backbone_contract classmethod 

get_backbone_contract() -> MethodBackboneContract

Return the encoder/feature contract declared by the adapter.

Raises:

Type Description
NotImplementedError

If the adapter omitted its required contract.
Source code in methods/base.py 
103
104
105
106
107
108
109
110
111
112
113
@classmethod
def get_backbone_contract(cls) -> MethodBackboneContract:
    """Return the encoder/feature contract declared by the adapter.

    Raises:
        NotImplementedError: If the adapter omitted its required contract.
    """
    if cls.backbone_contract is None:
        raise NotImplementedError(
            f"Method adapter {cls.__name__} has no backbone contract")
    return cls.backbone_contract

load_encoder 

load_encoder(
    *,
    weights_path: Optional[str] = None,
    **loader_options: Any,
) -> EncoderBundle

Build and validate the configured encoder in one operation.

Parameters:

Name Type Description Default
weights_path Optional[str]

Optional local checkpoint or model identifier. When omitted, cfg['backbone_weights'] is used.

 None
**loader_options Any

Family-specific options forwarded to the encoder registry, such as a pinned revision or offline-only loading.

 {}

Returns:

Type Description
EncoderBundle

A validated capability-aware encoder bundle.

Raises:

Type Description
RuntimeError

If the adapter has no runtime backbone contract.
BackboneCompatibilityError

If the bundle violates the method contract or declared feature provenance.
Source code in methods/base.py 
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
def load_encoder(self, *, weights_path: Optional[str] = None,
                 **loader_options: Any) -> EncoderBundle:
    """Build and validate the configured encoder in one operation.

    Args:
        weights_path: Optional local checkpoint or model identifier. When
            omitted, ``cfg['backbone_weights']`` is used.
        **loader_options: Family-specific options forwarded to the encoder
            registry, such as a pinned revision or offline-only loading.

    Returns:
        A validated capability-aware encoder bundle.

    Raises:
        RuntimeError: If the adapter has no runtime backbone contract.
        BackboneCompatibilityError: If the bundle violates the method
            contract or declared feature provenance.
    """
    if self.backbone_contract is None or self.backbone_name is None:
        raise RuntimeError(f"Method '{self.name}' has no runtime backbone")
    from common.backbones import build_encoder
    if weights_path is None:
        weights_path = self.cfg.get("backbone_weights")
    bundle = build_encoder(self.backbone_name, weights_path=weights_path,
                           device=self.device, **loader_options)
    return self.backbone_contract.validate_bundle(self.cfg, bundle)

build_optimizer 

build_optimizer(model: Module) -> Optimizer

Build Adam over trainable parameters using the configured recipe.

Source code in methods/base.py 
172
173
174
175
176
177
178
179
def build_optimizer(self, model: nn.Module) -> torch.optim.Optimizer:
    """Build Adam over trainable parameters using the configured recipe."""
    params = filter(lambda p: p.requires_grad, model.parameters())
    return torch.optim.Adam(
        params,
        lr=self.cfg.get("lr", 1e-4),
        weight_decay=self.cfg.get("weight_decay", 1e-5),
    )

prepare_fold 

prepare_fold(
    fold: int, model: Module, train_loader: Any
) -> None

Prepare training-split-only state after model/data construction.

Methods such as PathPT use this boundary to select prompts without seeing validation or test slides. The default is deliberately a no-op.

Source code in methods/base.py 
204
205
206
207
208
209
210
211
def prepare_fold(
    self, fold: int, model: nn.Module, train_loader: Any,
) -> None:
    """Prepare training-split-only state after model/data construction.

    Methods such as PathPT use this boundary to select prompts without
    seeing validation or test slides. The default is deliberately a no-op.
    """

on_train_epoch_start 

on_train_epoch_start(epoch: int, model: Module) -> None

Handle the boundary immediately before an epoch's first batch.

Source code in methods/base.py 
213
214
def on_train_epoch_start(self, epoch: int, model: nn.Module) -> None:
    """Handle the boundary immediately before an epoch's first batch."""

on_epoch_end 

on_epoch_end(epoch: int, metrics: Dict[str, float]) -> None

Handle an optional callback after a full train/validation epoch.

Source code in methods/base.py 
216
217
def on_epoch_end(self, epoch: int, metrics: Dict[str, float]) -> None:
    """Handle an optional callback after a full train/validation epoch."""

on_validation_end 

on_validation_end(
    epoch: int, metrics: Dict[str, float]
) -> None

Handle a completed validation pass, including epoch -1.

The default forwards to the historical on_epoch_end hook. Adapters which accumulate validation predictions therefore get a clear boundary for the optional initial evaluation as well as every trained epoch.

Source code in methods/base.py 
219
220
221
222
223
224
225
226
227
228
def on_validation_end(
    self, epoch: int, metrics: Dict[str, float],
) -> None:
    """Handle a completed validation pass, including epoch ``-1``.

    The default forwards to the historical ``on_epoch_end`` hook. Adapters
    which accumulate validation predictions therefore get a clear boundary
    for the optional initial evaluation as well as every trained epoch.
    """
    self.on_epoch_end(epoch, metrics)

on_fold_end 

on_fold_end(fold: int, metrics: Dict[str, float]) -> None

Handle an optional callback after a cross-validation fold.

Source code in methods/base.py 
230
231
def on_fold_end(self, fold: int, metrics: Dict[str, float]) -> None:
    """Handle an optional callback after a cross-validation fold."""

on_checkpoint_loaded 

on_checkpoint_loaded(
    model: Module, checkpoint_kind: str, fold: int
) -> None

Restore adapter state not represented by model.state_dict().

Source code in methods/base.py 
233
234
235
236
def on_checkpoint_loaded(
    self, model: nn.Module, checkpoint_kind: str, fold: int,
) -> None:
    """Restore adapter state not represented by ``model.state_dict()``."""

interpret_step 

interpret_step(
    batch: Any,
    model: Module,
    target_class: int | None = None,
) -> "InterpretabilityResult"

Return native, patch-aligned evidence for one evaluation slide.

Adapters must override this hook only when their score semantics and patch ordering are known. The default intentionally refuses to invent a generic "attention" map from hidden activations.

Parameters:

Name Type Description Default
batch Any

One loader batch for a single slide.

 required
model Module

Loaded fold checkpoint in evaluation mode.

 required
target_class int | None

Optional class index for class-specific evidence.

 None

Raises:

Type Description
NotImplementedError

If the method has no audited patch-evidence mapping in PGVL-Gym.
Source code in methods/base.py 
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
def interpret_step(
    self, batch: Any, model: nn.Module, target_class: int | None = None,
) -> "InterpretabilityResult":
    """Return native, patch-aligned evidence for one evaluation slide.

    Adapters must override this hook only when their score semantics and
    patch ordering are known. The default intentionally refuses to invent
    a generic "attention" map from hidden activations.

    Args:
        batch: One loader batch for a single slide.
        model: Loaded fold checkpoint in evaluation mode.
        target_class: Optional class index for class-specific evidence.

    Raises:
        NotImplementedError: If the method has no audited patch-evidence
            mapping in PGVL-Gym.
    """
    del batch, model, target_class
    raise NotImplementedError(
        f"Method '{self.name}' has no audited patch-level "
        "interpretability provider")

PathPTMethod 

PathPTMethod(cfg, device='cuda')

Bases: BaseMethod

Adapt PathPT while preserving its encoder-independent training recipe.

Source code in methods/pathpt/adapter.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
def __init__(self, cfg, device="cuda"):
    super().__init__(cfg, device)
    self.training_mode = str(
        self.cfg.get("training_mode", self.LEGACY_MODE)).strip().lower()
    if self.training_mode not in {self.LEGACY_MODE, self.NATIVE_MODE}:
        raise ValueError(
            "PathPT training_mode must be 'simplified_slide_ce' or "
            "'upstream_patch_ssl'")
    self.prompt_bank: PathPTPromptBank | None = None
    self._epoch = 0
    self._fold = 0
    self._encoder = None
    self._binary_validation_evidence: list[float] = []
    self._binary_validation_labels: list[int] = []
    self._binary_calibration_model = None

prepare_fold 

prepare_fold(fold, model, train_loader) -> None

Select the zero-shot patch classifier from training slides only.

Source code in methods/pathpt/adapter.py 
215
216
217
218
219
220
221
222
223
224
225
226
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
def prepare_fold(self, fold, model, train_loader) -> None:
    """Select the zero-shot patch classifier from training slides only."""
    if not self.native_mode:
        return
    if self.prompt_bank is None or self._encoder is None:
        raise RuntimeError("PathPT model must be built before prepare_fold")
    self._fold = int(fold)
    self._binary_validation_evidence.clear()
    self._binary_validation_labels.clear()
    self._binary_calibration_model = None
    prompt_select = bool(self.cfg.get("prompt_select", True))
    classifier_count = int(self.cfg.get("prompt_classifier_count", 200))
    select_count = int(self.cfg.get("prompt_select_count", 100))
    # Upstream uses a second, unsampled WSI loader for prompt selection.
    # Reuse the fold loader without duplicating feature files, temporarily
    # disabling only its per-epoch patch cap while selector workers exist.
    selection_seed = int(self.cfg.get("seed", 1)) + int(fold)
    if not prompt_select:
        selection = choose_prompt_embedding(
            self._encoder.encode_text, self.prompt_bank.prompts,
            device=torch.device(self.device), seed=selection_seed)
    else:
        dataset = getattr(train_loader, "dataset", None)
        saved_patch_num = getattr(dataset, "patch_num", None)
        if dataset is not None and hasattr(dataset, "patch_num"):
            dataset.patch_num = None
        try:
            selection = select_prompt_embedding(
                self._encoder.encode_text,
                train_loader,
                self.prompt_bank.prompts,
                n_slide_classes=int(self.cfg["n_classes"]),
                synthetic_normal=self.prompt_bank.synthetic_normal,
                device=torch.device(self.device),
                classifier_count=classifier_count,
                select_count=select_count,
                top_patches=int(self.cfg.get("prompt_top_patches", 100)),
                classifier_batch_size=int(
                    self.cfg.get("prompt_classifier_batch_size", 16)),
                text_batch_size=int(
                    self.cfg.get("prompt_text_batch_size", 128)),
            )
        finally:
            if dataset is not None and hasattr(dataset, "patch_num"):
                dataset.patch_num = saved_patch_num
    selected = selection.embedding.detach().to(
        device=model.pathpt_selected_prompt_embedding.device,
        dtype=model.pathpt_selected_prompt_embedding.dtype)
    if selected.shape != model.pathpt_selected_prompt_embedding.shape:
        raise ValueError(
            "PathPT selected prompt embedding shape changed from "
            f"{tuple(model.pathpt_selected_prompt_embedding.shape)} to "
            f"{tuple(selected.shape)}")
    model.pathpt_selected_prompt_embedding.copy_(selected)
    self._write_prompt_trace(selection)

interpret_step 

interpret_step(batch, model, target_class=None)

Expose native PathPT patch-class probabilities.

Source code in methods/pathpt/adapter.py 
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
@torch.no_grad()
def interpret_step(self, batch, model, target_class=None):
    """Expose native PathPT patch-class probabilities."""
    if not self.native_mode:
        raise NotImplementedError(
            "PathPT heatmaps require training_mode=upstream_patch_ssl; "
            "the legacy slide-CE condition has no audited patch score")
    features = batch[0].to(self.device)
    coordinates = batch[1]
    patch_scores = self._native_eval_patch_scores(model, features)
    logits = self._native_slide_logits(patch_scores, model)
    class_index = (
        int(logits.argmax(dim=1).item())
        if target_class is None else int(target_class))
    if class_index < 0 or class_index >= int(self.cfg["n_classes"]):
        raise ValueError("PathPT target class is outside the classifier range")
    assert self.prompt_bank is not None
    patch_class_index = (
        class_index + 1 if self.prompt_bank.synthetic_normal
        else class_index)
    return InterpretabilityResult(
        logits=logits,
        evidence=(PatchEvidence(
            scores=patch_scores[:, patch_class_index],
            score_type="patch_class_probability",
            description=(
                "Native PathPT per-patch probability for the selected "
                "slide class; synthetic Normal is excluded from the "
                "reported class index."),
            class_index=class_index,
            feature_path_key="feature_path_column",
            coordinates=coordinates,
        ),),
    )

get_backbone_contract classmethod 

get_backbone_contract() -> MethodBackboneContract

Return the encoder/feature contract declared by the adapter.

Raises:

Type Description
NotImplementedError

If the adapter omitted its required contract.
Source code in methods/base.py 
103
104
105
106
107
108
109
110
111
112
113
@classmethod
def get_backbone_contract(cls) -> MethodBackboneContract:
    """Return the encoder/feature contract declared by the adapter.

    Raises:
        NotImplementedError: If the adapter omitted its required contract.
    """
    if cls.backbone_contract is None:
        raise NotImplementedError(
            f"Method adapter {cls.__name__} has no backbone contract")
    return cls.backbone_contract

load_encoder 

load_encoder(
    *,
    weights_path: Optional[str] = None,
    **loader_options: Any,
) -> EncoderBundle

Build and validate the configured encoder in one operation.

Parameters:

Name Type Description Default
weights_path Optional[str]

Optional local checkpoint or model identifier. When omitted, cfg['backbone_weights'] is used.

 None
**loader_options Any

Family-specific options forwarded to the encoder registry, such as a pinned revision or offline-only loading.

 {}

Returns:

Type Description
EncoderBundle

A validated capability-aware encoder bundle.

Raises:

Type Description
RuntimeError

If the adapter has no runtime backbone contract.
BackboneCompatibilityError

If the bundle violates the method contract or declared feature provenance.
Source code in methods/base.py 
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
def load_encoder(self, *, weights_path: Optional[str] = None,
                 **loader_options: Any) -> EncoderBundle:
    """Build and validate the configured encoder in one operation.

    Args:
        weights_path: Optional local checkpoint or model identifier. When
            omitted, ``cfg['backbone_weights']`` is used.
        **loader_options: Family-specific options forwarded to the encoder
            registry, such as a pinned revision or offline-only loading.

    Returns:
        A validated capability-aware encoder bundle.

    Raises:
        RuntimeError: If the adapter has no runtime backbone contract.
        BackboneCompatibilityError: If the bundle violates the method
            contract or declared feature provenance.
    """
    if self.backbone_contract is None or self.backbone_name is None:
        raise RuntimeError(f"Method '{self.name}' has no runtime backbone")
    from common.backbones import build_encoder
    if weights_path is None:
        weights_path = self.cfg.get("backbone_weights")
    bundle = build_encoder(self.backbone_name, weights_path=weights_path,
                           device=self.device, **loader_options)
    return self.backbone_contract.validate_bundle(self.cfg, bundle)

validation_monitor 

validation_monitor(
    metrics: Dict[str, Any],
) -> tuple[str, float]

Return the minimized value used for best-checkpoint selection.

The unified loop cannot assume that every released method selects its checkpoint by cross-entropy. Several paper repositories use validation error, while MSCPT uses a task-dependent F1/AUC or F1/ACC composite. Adapters override this hook to preserve that executable recipe. Values are always expressed as a minimization objective so the checkpoint writer has one unambiguous contract.

Source code in methods/base.py 
190
191
192
193
194
195
196
197
198
199
200
201
202
def validation_monitor(
    self, metrics: Dict[str, Any],
) -> tuple[str, float]:
    """Return the minimized value used for best-checkpoint selection.

    The unified loop cannot assume that every released method selects its
    checkpoint by cross-entropy.  Several paper repositories use
    validation error, while MSCPT uses a task-dependent F1/AUC or F1/ACC
    composite.  Adapters override this hook to preserve that executable
    recipe.  Values are always expressed as a minimization objective so
    the checkpoint writer has one unambiguous contract.
    """
    return "val_loss", float(metrics["val_loss"])

on_epoch_end 

on_epoch_end(epoch: int, metrics: Dict[str, float]) -> None

Handle an optional callback after a full train/validation epoch.

Source code in methods/base.py 
216
217
def on_epoch_end(self, epoch: int, metrics: Dict[str, float]) -> None:
    """Handle an optional callback after a full train/validation epoch."""

on_fold_end 

on_fold_end(fold: int, metrics: Dict[str, float]) -> None

Handle an optional callback after a cross-validation fold.

Source code in methods/base.py 
230
231
def on_fold_end(self, fold: int, metrics: Dict[str, float]) -> None:
    """Handle an optional callback after a cross-validation fold."""

TOPMethod 

TOPMethod(cfg, device='cuda')

Bases: BaseMethod

Adapt TOP's two-level CLIP-RN50 prompt and pooling objective.

Source code in methods/top/adapter.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
def __init__(self, cfg, device="cuda"):
    super().__init__(cfg, device)
    self.is_encoder_extension = self.cfg.get("encoder_extension") is not None
    if self.is_encoder_extension:
        validate_encoder_extension(self.name, self.cfg)
        if self.cfg.get("encoder_extension_strategy") != \
                "paired_feature_context_v1":
            raise ValueError(
                "TOP encoder extensions require "
                "encoder_extension_strategy=paired_feature_context_v1")
    elif self.backbone_name != "clip-rn50":
        raise ValueError(
            "Non-RN50 TOP requires serialized encoder_extension provenance")
    if self.cfg.get("all_ctx_trainable", False) is not False:
        raise ValueError(
            "TOP requires all_ctx_trainable=false: the released trainer "
            "freezes description tokens and learns only '*' context slots")

on_train_epoch_start 

on_train_epoch_start(epoch: int, model: Module) -> None

Start an epoch-wide optimizer-liveness probe.

One saturated slide can legitimately round an otherwise finite FP32 SGD step to zero. The released TOP loop simply proceeds to the next slide, so treating the first such step as fatal was stricter than the upstream executable. We instead require at least one representable parameter update across the complete first epoch.

Source code in methods/top/adapter.py 
331
332
333
334
335
336
337
338
339
340
341
342
343
def on_train_epoch_start(self, epoch: int, model: nn.Module) -> None:
    """Start an epoch-wide optimizer-liveness probe.

    One saturated slide can legitimately round an otherwise finite FP32
    SGD step to zero.  The released TOP loop simply proceeds to the next
    slide, so treating the first such step as fatal was stricter than the
    upstream executable.  We instead require at least one representable
    parameter update across the complete first epoch.
    """
    if epoch == 0 and not getattr(
            self, "_optimizer_update_verified", False):
        self._optimizer_probe_steps = 0
        self._optimizer_probe_max_gradient = 0.0

on_epoch_end 

on_epoch_end(epoch: int, metrics) -> None

Report an inert first epoch without overriding upstream training.

Source code in methods/top/adapter.py 
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
def on_epoch_end(self, epoch: int, metrics) -> None:
    """Report an inert first epoch without overriding upstream training."""
    if epoch != 0 or getattr(self, "_optimizer_update_verified", False):
        return
    steps = getattr(self, "_optimizer_probe_steps", 0)
    max_gradient = getattr(self, "_optimizer_probe_max_gradient", 0.0)
    # The released TOP trainers do not abort on an inert first epoch and
    # run very long recipes (8,000 epochs for TCGA). The local fatal guard
    # prevented every registered run from reaching that recipe, even with
    # finite, substantial gradients. Preserve the diagnostic, but do not
    # turn it into a benchmark-specific stopping rule.
    print(
        "  ! TOP optimizer probe observed no representable parameter "
        f"update in epoch 0 ({steps} steps; maximum gradient "
        f"{max_gradient:.3e}); continuing the released training loop")
    self._optimizer_update_verified = True

interpret_step 

interpret_step(batch, model, target_class=None)

Expose TOP's native instance-prototype pooling distribution.

Source code in methods/top/adapter.py 
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
@torch.no_grad()
def interpret_step(self, batch, model, target_class=None):
    """Expose TOP's native instance-prototype pooling distribution."""
    del target_class  # TOP's instance prototypes are not task classes.
    supported = {
        "NoCoOp", "ABMIL", "learnablePrompt",
        "learnablePrompt_noCoOp", "learnablePrompt_argmax",
        "learnablePrompt_multi", "learnablePrompt_paper_mean",
        "learnablePrompt_multi_noCoOp",
    }
    if model.pooling_strategy not in supported:
        raise NotImplementedError(
            f"TOP pooling strategy {model.pooling_strategy!r} has no "
            "native patch-attention output")
    features = batch[0].to(self.device)
    if self.is_encoder_extension:
        features = project_paired_features(model, features)
    output = model(features)
    logits, _ = self._slide_logits(output)
    attention = (
        output[1] if isinstance(output, tuple) and len(output) > 1
        else None)
    if not torch.is_tensor(attention):
        raise NotImplementedError(
            f"TOP pooling strategy {model.pooling_strategy!r} does not "
            "emit patch-aligned scores")
    patch_count = features.shape[-2]
    if attention.ndim == 1:
        if attention.numel() != patch_count:
            raise ValueError("TOP attention does not align with its patch bag")
        scores = torch.softmax(attention.float(), dim=0)
    elif attention.ndim == 2 and attention.shape[0] == patch_count:
        scores = torch.softmax(attention.float(), dim=0).mean(dim=1)
    elif attention.ndim == 2 and attention.shape[1] == patch_count:
        scores = torch.softmax(attention.float(), dim=1).mean(dim=0)
    else:
        raise ValueError(
            "TOP attention must contain one patch axis, got "
            f"{tuple(attention.shape)} for {patch_count} patches")
    return InterpretabilityResult(
        logits=logits,
        evidence=(PatchEvidence(
            scores=scores,
            score_type="prototype_pooling_attention",
            description=(
                "Mean native TOP instance-prototype pooling attention; "
                "this score is class-independent."),
            feature_path_key="feature_path_column",
        ),),
    )

get_backbone_contract classmethod 

get_backbone_contract() -> MethodBackboneContract

Return the encoder/feature contract declared by the adapter.

Raises:

Type Description
NotImplementedError

If the adapter omitted its required contract.
Source code in methods/base.py 
103
104
105
106
107
108
109
110
111
112
113
@classmethod
def get_backbone_contract(cls) -> MethodBackboneContract:
    """Return the encoder/feature contract declared by the adapter.

    Raises:
        NotImplementedError: If the adapter omitted its required contract.
    """
    if cls.backbone_contract is None:
        raise NotImplementedError(
            f"Method adapter {cls.__name__} has no backbone contract")
    return cls.backbone_contract

load_encoder 

load_encoder(
    *,
    weights_path: Optional[str] = None,
    **loader_options: Any,
) -> EncoderBundle

Build and validate the configured encoder in one operation.

Parameters:

Name Type Description Default
weights_path Optional[str]

Optional local checkpoint or model identifier. When omitted, cfg['backbone_weights'] is used.

 None
**loader_options Any

Family-specific options forwarded to the encoder registry, such as a pinned revision or offline-only loading.

 {}

Returns:

Type Description
EncoderBundle

A validated capability-aware encoder bundle.

Raises:

Type Description
RuntimeError

If the adapter has no runtime backbone contract.
BackboneCompatibilityError

If the bundle violates the method contract or declared feature provenance.
Source code in methods/base.py 
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
def load_encoder(self, *, weights_path: Optional[str] = None,
                 **loader_options: Any) -> EncoderBundle:
    """Build and validate the configured encoder in one operation.

    Args:
        weights_path: Optional local checkpoint or model identifier. When
            omitted, ``cfg['backbone_weights']`` is used.
        **loader_options: Family-specific options forwarded to the encoder
            registry, such as a pinned revision or offline-only loading.

    Returns:
        A validated capability-aware encoder bundle.

    Raises:
        RuntimeError: If the adapter has no runtime backbone contract.
        BackboneCompatibilityError: If the bundle violates the method
            contract or declared feature provenance.
    """
    if self.backbone_contract is None or self.backbone_name is None:
        raise RuntimeError(f"Method '{self.name}' has no runtime backbone")
    from common.backbones import build_encoder
    if weights_path is None:
        weights_path = self.cfg.get("backbone_weights")
    bundle = build_encoder(self.backbone_name, weights_path=weights_path,
                           device=self.device, **loader_options)
    return self.backbone_contract.validate_bundle(self.cfg, bundle)

validation_monitor 

validation_monitor(
    metrics: Dict[str, Any],
) -> tuple[str, float]

Return the minimized value used for best-checkpoint selection.

The unified loop cannot assume that every released method selects its checkpoint by cross-entropy. Several paper repositories use validation error, while MSCPT uses a task-dependent F1/AUC or F1/ACC composite. Adapters override this hook to preserve that executable recipe. Values are always expressed as a minimization objective so the checkpoint writer has one unambiguous contract.

Source code in methods/base.py 
190
191
192
193
194
195
196
197
198
199
200
201
202
def validation_monitor(
    self, metrics: Dict[str, Any],
) -> tuple[str, float]:
    """Return the minimized value used for best-checkpoint selection.

    The unified loop cannot assume that every released method selects its
    checkpoint by cross-entropy.  Several paper repositories use
    validation error, while MSCPT uses a task-dependent F1/AUC or F1/ACC
    composite.  Adapters override this hook to preserve that executable
    recipe.  Values are always expressed as a minimization objective so
    the checkpoint writer has one unambiguous contract.
    """
    return "val_loss", float(metrics["val_loss"])

prepare_fold 

prepare_fold(
    fold: int, model: Module, train_loader: Any
) -> None

Prepare training-split-only state after model/data construction.

Methods such as PathPT use this boundary to select prompts without seeing validation or test slides. The default is deliberately a no-op.

Source code in methods/base.py 
204
205
206
207
208
209
210
211
def prepare_fold(
    self, fold: int, model: nn.Module, train_loader: Any,
) -> None:
    """Prepare training-split-only state after model/data construction.

    Methods such as PathPT use this boundary to select prompts without
    seeing validation or test slides. The default is deliberately a no-op.
    """

on_validation_end 

on_validation_end(
    epoch: int, metrics: Dict[str, float]
) -> None

Handle a completed validation pass, including epoch -1.

The default forwards to the historical on_epoch_end hook. Adapters which accumulate validation predictions therefore get a clear boundary for the optional initial evaluation as well as every trained epoch.

Source code in methods/base.py 
219
220
221
222
223
224
225
226
227
228
def on_validation_end(
    self, epoch: int, metrics: Dict[str, float],
) -> None:
    """Handle a completed validation pass, including epoch ``-1``.

    The default forwards to the historical ``on_epoch_end`` hook. Adapters
    which accumulate validation predictions therefore get a clear boundary
    for the optional initial evaluation as well as every trained epoch.
    """
    self.on_epoch_end(epoch, metrics)

on_fold_end 

on_fold_end(fold: int, metrics: Dict[str, float]) -> None

Handle an optional callback after a cross-validation fold.

Source code in methods/base.py 
230
231
def on_fold_end(self, fold: int, metrics: Dict[str, float]) -> None:
    """Handle an optional callback after a cross-validation fold."""

on_checkpoint_loaded 

on_checkpoint_loaded(
    model: Module, checkpoint_kind: str, fold: int
) -> None

Restore adapter state not represented by model.state_dict().

Source code in methods/base.py 
233
234
235
236
def on_checkpoint_loaded(
    self, model: nn.Module, checkpoint_kind: str, fold: int,
) -> None:
    """Restore adapter state not represented by ``model.state_dict()``."""

SLIPMethod 

SLIPMethod(cfg: Dict[str, Any], device: str = 'cuda')

Bases: BaseMethod

Adapt SLIP's tissue-routed prompt learner for supported CLIP families.

Source code in methods/base.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def __init__(self, cfg: Dict[str, Any], device: str = "cuda"):
    # A few vendored adapters fill in derived defaults while constructing
    # their model.  Keeping those writes private is essential: train.py has
    # already snapshotted and fingerprinted the resolved run config, and a
    # later mutation would otherwise make an interrupted run impossible to
    # resume.  A deep copy also prevents nested recipe dictionaries from
    # carrying state between fold-specific adapter instances.
    self.cfg = deepcopy(cfg)
    self.device = device
    self.backbone_name = (self.backbone_contract.validate_config(self.cfg)
                          if self.backbone_contract is not None else None)

interpret_step 

interpret_step(batch, model, target_class=None)

Expose SLIP's native class-specific tissue-routing evidence.

Source code in methods/slip/adapter.py 
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
@torch.no_grad()
def interpret_step(self, batch, model, target_class=None):
    """Expose SLIP's native class-specific tissue-routing evidence."""
    features = batch[0].to(self.device)
    if features.ndim == 3 and features.shape[0] == 1:
        features = features.squeeze(0)
    logits, _correlation, patch_attention, _routing = model(
        features, return_metadata=True)
    logits = logits.unsqueeze(0)
    class_index = (
        int(logits.argmax(dim=1).item())
        if target_class is None else int(target_class))
    if class_index < 0 or class_index >= logits.shape[1]:
        raise ValueError("SLIP target class is outside the classifier range")
    return InterpretabilityResult(
        logits=logits,
        evidence=(PatchEvidence(
            scores=patch_attention[:, class_index],
            score_type="class_tissue_routing_evidence",
            description=(
                "Native SLIP patch-to-tissue-to-slide routing evidence "
                "for the selected class."),
            class_index=class_index,
            feature_path_key="feature_path_column",
        ),),
    )

get_backbone_contract classmethod 

get_backbone_contract() -> MethodBackboneContract

Return the encoder/feature contract declared by the adapter.

Raises:

Type Description
NotImplementedError

If the adapter omitted its required contract.
Source code in methods/base.py 
103
104
105
106
107
108
109
110
111
112
113
@classmethod
def get_backbone_contract(cls) -> MethodBackboneContract:
    """Return the encoder/feature contract declared by the adapter.

    Raises:
        NotImplementedError: If the adapter omitted its required contract.
    """
    if cls.backbone_contract is None:
        raise NotImplementedError(
            f"Method adapter {cls.__name__} has no backbone contract")
    return cls.backbone_contract

load_encoder 

load_encoder(
    *,
    weights_path: Optional[str] = None,
    **loader_options: Any,
) -> EncoderBundle

Build and validate the configured encoder in one operation.

Parameters:

Name Type Description Default
weights_path Optional[str]

Optional local checkpoint or model identifier. When omitted, cfg['backbone_weights'] is used.

 None
**loader_options Any

Family-specific options forwarded to the encoder registry, such as a pinned revision or offline-only loading.

 {}

Returns:

Type Description
EncoderBundle

A validated capability-aware encoder bundle.

Raises:

Type Description
RuntimeError

If the adapter has no runtime backbone contract.
BackboneCompatibilityError

If the bundle violates the method contract or declared feature provenance.
Source code in methods/base.py 
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
def load_encoder(self, *, weights_path: Optional[str] = None,
                 **loader_options: Any) -> EncoderBundle:
    """Build and validate the configured encoder in one operation.

    Args:
        weights_path: Optional local checkpoint or model identifier. When
            omitted, ``cfg['backbone_weights']`` is used.
        **loader_options: Family-specific options forwarded to the encoder
            registry, such as a pinned revision or offline-only loading.

    Returns:
        A validated capability-aware encoder bundle.

    Raises:
        RuntimeError: If the adapter has no runtime backbone contract.
        BackboneCompatibilityError: If the bundle violates the method
            contract or declared feature provenance.
    """
    if self.backbone_contract is None or self.backbone_name is None:
        raise RuntimeError(f"Method '{self.name}' has no runtime backbone")
    from common.backbones import build_encoder
    if weights_path is None:
        weights_path = self.cfg.get("backbone_weights")
    bundle = build_encoder(self.backbone_name, weights_path=weights_path,
                           device=self.device, **loader_options)
    return self.backbone_contract.validate_bundle(self.cfg, bundle)

validation_monitor 

validation_monitor(
    metrics: Dict[str, Any],
) -> tuple[str, float]

Return the minimized value used for best-checkpoint selection.

The unified loop cannot assume that every released method selects its checkpoint by cross-entropy. Several paper repositories use validation error, while MSCPT uses a task-dependent F1/AUC or F1/ACC composite. Adapters override this hook to preserve that executable recipe. Values are always expressed as a minimization objective so the checkpoint writer has one unambiguous contract.

Source code in methods/base.py 
190
191
192
193
194
195
196
197
198
199
200
201
202
def validation_monitor(
    self, metrics: Dict[str, Any],
) -> tuple[str, float]:
    """Return the minimized value used for best-checkpoint selection.

    The unified loop cannot assume that every released method selects its
    checkpoint by cross-entropy.  Several paper repositories use
    validation error, while MSCPT uses a task-dependent F1/AUC or F1/ACC
    composite.  Adapters override this hook to preserve that executable
    recipe.  Values are always expressed as a minimization objective so
    the checkpoint writer has one unambiguous contract.
    """
    return "val_loss", float(metrics["val_loss"])

prepare_fold 

prepare_fold(
    fold: int, model: Module, train_loader: Any
) -> None

Prepare training-split-only state after model/data construction.

Methods such as PathPT use this boundary to select prompts without seeing validation or test slides. The default is deliberately a no-op.

Source code in methods/base.py 
204
205
206
207
208
209
210
211
def prepare_fold(
    self, fold: int, model: nn.Module, train_loader: Any,
) -> None:
    """Prepare training-split-only state after model/data construction.

    Methods such as PathPT use this boundary to select prompts without
    seeing validation or test slides. The default is deliberately a no-op.
    """

on_train_epoch_start 

on_train_epoch_start(epoch: int, model: Module) -> None

Handle the boundary immediately before an epoch's first batch.

Source code in methods/base.py 
213
214
def on_train_epoch_start(self, epoch: int, model: nn.Module) -> None:
    """Handle the boundary immediately before an epoch's first batch."""

on_epoch_end 

on_epoch_end(epoch: int, metrics: Dict[str, float]) -> None

Handle an optional callback after a full train/validation epoch.

Source code in methods/base.py 
216
217
def on_epoch_end(self, epoch: int, metrics: Dict[str, float]) -> None:
    """Handle an optional callback after a full train/validation epoch."""

on_validation_end 

on_validation_end(
    epoch: int, metrics: Dict[str, float]
) -> None

Handle a completed validation pass, including epoch -1.

The default forwards to the historical on_epoch_end hook. Adapters which accumulate validation predictions therefore get a clear boundary for the optional initial evaluation as well as every trained epoch.

Source code in methods/base.py 
219
220
221
222
223
224
225
226
227
228
def on_validation_end(
    self, epoch: int, metrics: Dict[str, float],
) -> None:
    """Handle a completed validation pass, including epoch ``-1``.

    The default forwards to the historical ``on_epoch_end`` hook. Adapters
    which accumulate validation predictions therefore get a clear boundary
    for the optional initial evaluation as well as every trained epoch.
    """
    self.on_epoch_end(epoch, metrics)

on_fold_end 

on_fold_end(fold: int, metrics: Dict[str, float]) -> None

Handle an optional callback after a cross-validation fold.

Source code in methods/base.py 
230
231
def on_fold_end(self, fold: int, metrics: Dict[str, float]) -> None:
    """Handle an optional callback after a cross-validation fold."""

on_checkpoint_loaded 

on_checkpoint_loaded(
    model: Module, checkpoint_kind: str, fold: int
) -> None

Restore adapter state not represented by model.state_dict().

Source code in methods/base.py 
233
234
235
236
def on_checkpoint_loaded(
    self, model: nn.Module, checkpoint_kind: str, fold: int,
) -> None:
    """Restore adapter state not represented by ``model.state_dict()``."""

WSIFiVEMethod 

WSIFiVEMethod(cfg, device='cuda')

Bases: BaseMethod

Adapt WSI-FiVE patch bags, questions, and answer-bank supervision.

Source code in methods/wsi_five/adapter.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 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
159
160
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
225
226
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
def __init__(self, cfg, device="cuda"):
    super().__init__(cfg, device)
    self.text_alignment_mode = str(self.cfg.get(
        "text_alignment_mode", self.BIOCLINICALBERT_ALIGNMENT
    )).strip().lower()
    if self.text_alignment_mode not in {
            self.BIOCLINICALBERT_ALIGNMENT, self.PAIRED_ALIGNMENT}:
        raise ValueError(
            "WSI-FiVE text_alignment_mode must be 'learned_projection' "
            "or 'paired_text_tower'")
    self.paired_text_tower = (
        self.text_alignment_mode == self.PAIRED_ALIGNMENT)
    self.paired_text_overlength_policy = None
    if self.paired_text_tower and self.cfg.get("backbone") == "clip-rn50":
        extension = self.cfg.get("encoder_extension")
        declared_policy = (
            extension.get("text_overlength_policy")
            if isinstance(extension, dict) else None)
        if declared_policy != PAIRED_CLIP_TEXT_OVERLENGTH_POLICY:
            raise ValueError(
                "Paired CLIP-RN50 WSI-FiVE requires disclosed "
                "text_overlength_policy="
                f"{PAIRED_CLIP_TEXT_OVERLENGTH_POLICY!r}")
        self.paired_text_overlength_policy = declared_policy
    self.training_mode = str(self.cfg.get(
        "training_mode", self.SIMPLIFIED_MODE)).strip().lower()
    if self.training_mode not in {
            self.NATIVE_MODE, self.UPSTREAM_CLASSNAME_MODE,
            self.SIMPLIFIED_MODE}:
        raise ValueError(
            "WSI-FiVE training_mode must be 'upstream_answer_bank', "
            "'upstream_classname_bank', or 'simplified_classnames'")
    self.few_shot_class_anchor_weight = float(
        self.cfg.get("few_shot_class_anchor_weight", 0.0))
    if (not math.isfinite(self.few_shot_class_anchor_weight)
            or self.few_shot_class_anchor_weight < 0.0):
        raise ValueError(
            "WSI-FiVE few_shot_class_anchor_weight must be finite and "
            "non-negative")
    if (self.training_mode != self.NATIVE_MODE
            and self.few_shot_class_anchor_weight != 0.0):
        raise ValueError(
            "WSI-FiVE few_shot_class_anchor_weight is only valid with "
            "upstream_answer_bank training")
    self.pretrained_checkpoint_report: dict | None = None
    if self.upstream_classname_mode:
        required = (
            "wsi_pretrained_checkpoint",
            "wsi_pretrained_checkpoint_sha256",
            "wsi_pretrained_checkpoint_bytes",
            "wsi_initialization",
            "wsi_checkpoint_position_transfer",
            "wsi_soft_prompt_pooling",
            "wsi_cam_question_augmentation",
        )
        missing = [key for key in required if not self.cfg.get(key)]
        if missing:
            raise ValueError(
                "WSI-FiVE upstream_classname_bank requires: "
                + ", ".join(missing))
        expected = {
            "wsi_pretrained_checkpoint_sha256": (
                RELEASED_CHECKPOINT_SHA256),
            "wsi_pretrained_checkpoint_bytes": (
                RELEASED_CHECKPOINT_BYTES),
            "wsi_initialization": RELEASED_INITIALIZATION,
            "wsi_checkpoint_position_transfer": (
                RELEASED_POSITION_TRANSFER),
            "wsi_soft_prompt_pooling": RELEASED_SOFT_PROMPT_POOLING,
            "wsi_cam_question_augmentation": (
                RELEASED_CAM_QUESTION_AUGMENTATION),
        }
        for key, value in expected.items():
            if self.cfg.get(key) != value:
                raise ValueError(
                    f"WSI-FiVE {key}={self.cfg.get(key)!r} does not "
                    f"match the released CAMELYON transfer condition "
                    f"({value!r})")
    elif self.cfg.get("wsi_pretrained_checkpoint"):
        raise ValueError(
            "WSI-FiVE TCGA-pretrained transfer checkpoint is only "
            "supported by upstream_classname_bank mode")
    if self.paired_text_tower:
        if self.upstream_classname_mode:
            raise ValueError(
                "Paired-tower WSI-FiVE cannot load the released "
                "BioClinicalBERT CAMELYON checkpoint")
        if self.cfg.get("backbone") not in {
                "clip-rn50", "quiltnet", "conch", "plip", "keep",
                "musk"}:
            raise ValueError(
                "Paired-tower WSI-FiVE currently supports clip-rn50, "
                "quiltnet, conch, plip, keep, and musk")
        prompt_encoder = self.cfg.get("prompt_encoder")
        if not isinstance(prompt_encoder, dict):
            raise ValueError(
                "Paired-tower WSI-FiVE requires a prompt_encoder mapping")
        from common.backbones import get_spec
        spec = get_spec(self.cfg["backbone"])
        expected_pairs = {
            "name": self.cfg.get("backbone"),
            "feature_space_id": spec.feature_space_id,
            "feature_dim": int(spec.shared_dim or 0),
        }
        mismatched = {
            key: (prompt_encoder.get(key), expected)
            for key, expected in expected_pairs.items()
            if prompt_encoder.get(key) != expected
        }
        if mismatched:
            raise ValueError(
                "Paired-tower WSI-FiVE prompt encoder does not match its "
                f"cached visual space: {mismatched}")
        if not prompt_encoder.get("weights"):
            raise ValueError(
                "Paired-tower WSI-FiVE requires exact text-tower weights")
    if self.cfg.get("wsi_prompt_format") != WSI_FIVE_PROMPT_FORMAT:
        raise ValueError(
            "WSI-FiVE requires wsi_prompt_format=" +
            WSI_FIVE_PROMPT_FORMAT)
    for key in (
        "clinical_questions", "wsi_question_file_sha256",
        "wsi_question_bank_sha256", "wsi_question_provenance",
        "prompt_provenance", "prompt_source",
    ):
        if not self.cfg.get(key):
            raise ValueError(f"WSI-FiVE requires {key}")
    self.question_bank = load_wsi_five_question_bank(
        self.cfg["clinical_questions"],
        expected_file_sha256=self.cfg["wsi_question_file_sha256"],
        expected_prompt_bank_sha256=self.cfg["wsi_question_bank_sha256"],
        expected_provenance=self.cfg["wsi_question_provenance"],
    )
    self.answer_bank = None
    evaluation = None
    if self.uses_upstream_evaluation_bank:
        for key in (
            "evaluation_prompt_path", "wsi_evaluation_file_sha256",
            "wsi_evaluation_bank_sha256", "wsi_evaluation_provenance",
        ):
            if not self.cfg.get(key):
                raise ValueError(
                    f"WSI-FiVE {self.training_mode} requires {key}")
        evaluation = load_wsi_five_evaluation_bank(
            self.cfg["evaluation_prompt_path"], self.cfg["label_dict"],
            expected_file_sha256=self.cfg["wsi_evaluation_file_sha256"],
            expected_prompt_bank_sha256=(
                self.cfg["wsi_evaluation_bank_sha256"]),
            expected_provenance=self.cfg["wsi_evaluation_provenance"],
        )
        self.evaluation_prompts = evaluation.prompts
    if self.native_mode:
        for key in (
            "report_csv", "wsi_answer_file_sha256",
            "wsi_answer_bank_sha256", "wsi_answer_provenance",
        ):
            if not self.cfg.get(key):
                raise ValueError(
                    f"WSI-FiVE upstream_answer_bank requires {key}")
        self.answer_bank = load_wsi_five_answer_bank(
            self.cfg["report_csv"],
            expected_file_sha256=self.cfg["wsi_answer_file_sha256"],
            expected_answer_bank_sha256=self.cfg["wsi_answer_bank_sha256"],
            expected_provenance=self.cfg["wsi_answer_provenance"],
        )
        assert evaluation is not None
        expected_prompt_provenance = (
            f"{self.question_bank.provenance}_questions_with_"
            f"{self.answer_bank.provenance}_answer_and_"
            f"{evaluation.provenance}_evaluation_banks")
        expected_prompt_source = "wsi_five_derived_upstream_text_assets"
    elif self.upstream_classname_mode:
        assert evaluation is not None
        expected_prompt_provenance = (
            f"{self.question_bank.provenance}_questions_with_"
            f"{evaluation.provenance}_evaluation_bank")
        expected_prompt_source = (
            "wsi_five_upstream_camelyon_classname_bank")
    else:
        self.evaluation_prompts = tuple(self.cfg.get("classnames", ()))
        expected_prompt_provenance = (
            f"{self.question_bank.provenance}_questions_with_"
            "classname_comparison")
        expected_prompt_source = "wsi_five_simplified_classname_baseline"
    if self.cfg["prompt_provenance"] != expected_prompt_provenance:
        raise ValueError(
            "WSI-FiVE prompt_provenance does not match active text roles")
    if self.cfg["prompt_source"] != expected_prompt_source:
        raise ValueError(
            "WSI-FiVE prompt_source does not match active text roles")
    if len(self.evaluation_prompts) != int(self.cfg["n_classes"]):
        raise ValueError(
            "WSI-FiVE evaluation prompt count must match n_classes")
    self._train_answer_bank: tuple[tuple[str, ...], ...] | None = None
    self._answer_to_index: dict[tuple[str, ...], int] = {}
    self._fold = 0
    self._epoch = 0
    self._batch_in_epoch = 0
    self._train_batches = 1

upstream_classname_mode property 

upstream_classname_mode: bool

Whether training uses an exact released task class-text bank.

build_optimizer 

build_optimizer(model: Module) -> Optimizer

Reproduce the released AdamW parameter grouping.

Upstream applies the configured learning rate to its text/other group, 10x to mit/message_/prompts parameters, and removes weight decay from vectors, biases, and positional embeddings.

Source code in methods/wsi_five/adapter.py 
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
def build_optimizer(self, model: nn.Module) -> torch.optim.Optimizer:
    """Reproduce the released AdamW parameter grouping.

    Upstream applies the configured learning rate to its text/other group,
    10x to ``mit``/``message_``/``prompts`` parameters, and removes weight
    decay from vectors, biases, and positional embeddings.
    """
    base_lr = float(self.cfg.get("lr", 3e-6))
    weight_decay = float(self.cfg.get("weight_decay", 1e-3))
    buckets: dict[tuple[str, bool], list[nn.Parameter]] = {}
    for name, parameter in model.named_parameters():
        if not parameter.requires_grad:
            continue
        if "message_" in name:
            family = "message"
        elif "mit" in name:
            family = "mit"
        elif "prompts" in name:
            family = "prompts"
        else:
            family = "base"
        no_decay = (
            parameter.ndim == 1 or name.endswith(".bias")
            or "positional_embedding" in name)
        buckets.setdefault((family, no_decay), []).append(parameter)
    if not buckets:
        raise RuntimeError("WSI-FiVE model has no trainable parameters")

    groups = []
    for (family, no_decay), parameters in buckets.items():
        maximum_lr = base_lr if family == "base" else base_lr * 10.0
        groups.append({
            "params": parameters,
            "lr": maximum_lr,
            "weight_decay": 0.0 if no_decay else weight_decay,
            "wsi_five_family": family,
            "wsi_five_max_lr": maximum_lr,
        })
    optimizer = torch.optim.AdamW(
        groups, betas=(0.9, 0.98), eps=1e-8)
    # timm's upstream CosineLRScheduler initializes every group at zero
    # when warmup is enabled. The adapter advances the schedule at the
    # same optimizer-update boundary because the unified trainer exposes
    # an epoch scheduler hook rather than a per-batch scheduler hook.
    if int(self.cfg.get("warmup_epochs", 5)) > 0:
        for group in optimizer.param_groups:
            group["lr"] = 0.0
    return optimizer

build_scheduler 

build_scheduler(optimizer)

The released per-update cosine schedule is stepped in train_step.

Source code in methods/wsi_five/adapter.py 
390
391
392
def build_scheduler(self, optimizer):
    """The released per-update cosine schedule is stepped in train_step."""
    return None

prepare_fold 

prepare_fold(fold, model, train_loader) -> None

Build the answer candidate bank from this training fold only.

Source code in methods/wsi_five/adapter.py 
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
def prepare_fold(self, fold, model, train_loader) -> None:
    """Build the answer candidate bank from this training fold only."""
    self._fold = int(fold)
    try:
        self._train_batches = max(int(len(train_loader)), 1)
    except TypeError:
        # Lightweight adapter unit tests use a loader-shaped namespace.
        self._train_batches = 1
    if not self.native_mode:
        return
    if len(model.prompt_list) != ANSWER_FIELD_COUNT:
        raise ValueError(
            "WSI-FiVE native mode requires the six upstream questions")
    dataset = getattr(train_loader, "dataset", None)
    if dataset is None or not hasattr(dataset, "native_answer_bank"):
        raise TypeError(
            "WSI-FiVE native mode requires its structured-answer dataset")
    self._train_answer_bank = dataset.native_answer_bank()
    self._answer_to_index = {
        fields: index for index, fields in enumerate(self._train_answer_bank)}
    self._write_answer_bank_trace()

get_backbone_contract classmethod 

get_backbone_contract() -> MethodBackboneContract

Return the encoder/feature contract declared by the adapter.

Raises:

Type Description
NotImplementedError

If the adapter omitted its required contract.
Source code in methods/base.py 
103
104
105
106
107
108
109
110
111
112
113
@classmethod
def get_backbone_contract(cls) -> MethodBackboneContract:
    """Return the encoder/feature contract declared by the adapter.

    Raises:
        NotImplementedError: If the adapter omitted its required contract.
    """
    if cls.backbone_contract is None:
        raise NotImplementedError(
            f"Method adapter {cls.__name__} has no backbone contract")
    return cls.backbone_contract

load_encoder 

load_encoder(
    *,
    weights_path: Optional[str] = None,
    **loader_options: Any,
) -> EncoderBundle

Build and validate the configured encoder in one operation.

Parameters:

Name Type Description Default
weights_path Optional[str]

Optional local checkpoint or model identifier. When omitted, cfg['backbone_weights'] is used.

 None
**loader_options Any

Family-specific options forwarded to the encoder registry, such as a pinned revision or offline-only loading.

 {}

Returns:

Type Description
EncoderBundle

A validated capability-aware encoder bundle.

Raises:

Type Description
RuntimeError

If the adapter has no runtime backbone contract.
BackboneCompatibilityError

If the bundle violates the method contract or declared feature provenance.
Source code in methods/base.py 
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
def load_encoder(self, *, weights_path: Optional[str] = None,
                 **loader_options: Any) -> EncoderBundle:
    """Build and validate the configured encoder in one operation.

    Args:
        weights_path: Optional local checkpoint or model identifier. When
            omitted, ``cfg['backbone_weights']`` is used.
        **loader_options: Family-specific options forwarded to the encoder
            registry, such as a pinned revision or offline-only loading.

    Returns:
        A validated capability-aware encoder bundle.

    Raises:
        RuntimeError: If the adapter has no runtime backbone contract.
        BackboneCompatibilityError: If the bundle violates the method
            contract or declared feature provenance.
    """
    if self.backbone_contract is None or self.backbone_name is None:
        raise RuntimeError(f"Method '{self.name}' has no runtime backbone")
    from common.backbones import build_encoder
    if weights_path is None:
        weights_path = self.cfg.get("backbone_weights")
    bundle = build_encoder(self.backbone_name, weights_path=weights_path,
                           device=self.device, **loader_options)
    return self.backbone_contract.validate_bundle(self.cfg, bundle)

validation_monitor 

validation_monitor(
    metrics: Dict[str, Any],
) -> tuple[str, float]

Return the minimized value used for best-checkpoint selection.

The unified loop cannot assume that every released method selects its checkpoint by cross-entropy. Several paper repositories use validation error, while MSCPT uses a task-dependent F1/AUC or F1/ACC composite. Adapters override this hook to preserve that executable recipe. Values are always expressed as a minimization objective so the checkpoint writer has one unambiguous contract.

Source code in methods/base.py 
190
191
192
193
194
195
196
197
198
199
200
201
202
def validation_monitor(
    self, metrics: Dict[str, Any],
) -> tuple[str, float]:
    """Return the minimized value used for best-checkpoint selection.

    The unified loop cannot assume that every released method selects its
    checkpoint by cross-entropy.  Several paper repositories use
    validation error, while MSCPT uses a task-dependent F1/AUC or F1/ACC
    composite.  Adapters override this hook to preserve that executable
    recipe.  Values are always expressed as a minimization objective so
    the checkpoint writer has one unambiguous contract.
    """
    return "val_loss", float(metrics["val_loss"])

on_epoch_end 

on_epoch_end(epoch: int, metrics: Dict[str, float]) -> None

Handle an optional callback after a full train/validation epoch.

Source code in methods/base.py 
216
217
def on_epoch_end(self, epoch: int, metrics: Dict[str, float]) -> None:
    """Handle an optional callback after a full train/validation epoch."""

on_validation_end 

on_validation_end(
    epoch: int, metrics: Dict[str, float]
) -> None

Handle a completed validation pass, including epoch -1.

The default forwards to the historical on_epoch_end hook. Adapters which accumulate validation predictions therefore get a clear boundary for the optional initial evaluation as well as every trained epoch.

Source code in methods/base.py 
219
220
221
222
223
224
225
226
227
228
def on_validation_end(
    self, epoch: int, metrics: Dict[str, float],
) -> None:
    """Handle a completed validation pass, including epoch ``-1``.

    The default forwards to the historical ``on_epoch_end`` hook. Adapters
    which accumulate validation predictions therefore get a clear boundary
    for the optional initial evaluation as well as every trained epoch.
    """
    self.on_epoch_end(epoch, metrics)

on_fold_end 

on_fold_end(fold: int, metrics: Dict[str, float]) -> None

Handle an optional callback after a cross-validation fold.

Source code in methods/base.py 
230
231
def on_fold_end(self, fold: int, metrics: Dict[str, float]) -> None:
    """Handle an optional callback after a cross-validation fold."""

on_checkpoint_loaded 

on_checkpoint_loaded(
    model: Module, checkpoint_kind: str, fold: int
) -> None

Restore adapter state not represented by model.state_dict().

Source code in methods/base.py 
233
234
235
236
def on_checkpoint_loaded(
    self, model: nn.Module, checkpoint_kind: str, fold: int,
) -> None:
    """Restore adapter state not represented by ``model.state_dict()``."""

interpret_step 

interpret_step(
    batch: Any,
    model: Module,
    target_class: int | None = None,
) -> "InterpretabilityResult"

Return native, patch-aligned evidence for one evaluation slide.

Adapters must override this hook only when their score semantics and patch ordering are known. The default intentionally refuses to invent a generic "attention" map from hidden activations.

Parameters:

Name Type Description Default
batch Any

One loader batch for a single slide.

 required
model Module

Loaded fold checkpoint in evaluation mode.

 required
target_class int | None

Optional class index for class-specific evidence.

 None

Raises:

Type Description
NotImplementedError

If the method has no audited patch-evidence mapping in PGVL-Gym.
Source code in methods/base.py 
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
def interpret_step(
    self, batch: Any, model: nn.Module, target_class: int | None = None,
) -> "InterpretabilityResult":
    """Return native, patch-aligned evidence for one evaluation slide.

    Adapters must override this hook only when their score semantics and
    patch ordering are known. The default intentionally refuses to invent
    a generic "attention" map from hidden activations.

    Args:
        batch: One loader batch for a single slide.
        model: Loaded fold checkpoint in evaluation mode.
        target_class: Optional class index for class-specific evidence.

    Raises:
        NotImplementedError: If the method has no audited patch-evidence
            mapping in PGVL-Gym.
    """
    del batch, model, target_class
    raise NotImplementedError(
        f"Method '{self.name}' has no audited patch-level "
        "interpretability provider")

MUSEMethod 

MUSEMethod(cfg: Dict[str, Any], device: str = 'cuda')

Bases: BaseMethod

Adapt MUSE with independent offline patch and runtime prompt encoders.

Source code in methods/base.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def __init__(self, cfg: Dict[str, Any], device: str = "cuda"):
    # A few vendored adapters fill in derived defaults while constructing
    # their model.  Keeping those writes private is essential: train.py has
    # already snapshotted and fingerprinted the resolved run config, and a
    # later mutation would otherwise make an interrupted run impossible to
    # resume.  A deep copy also prevents nested recipe dictionaries from
    # carrying state between fold-specific adapter instances.
    self.cfg = deepcopy(cfg)
    self.device = device
    self.backbone_name = (self.backbone_contract.validate_config(self.cfg)
                          if self.backbone_contract is not None else None)

build_scheduler 

build_scheduler(optimizer)

Return no scheduler; the released core constructs but never steps it.

Source code in methods/muse/adapter.py 
272
273
274
def build_scheduler(self, optimizer):
    """Return no scheduler; the released core constructs but never steps it."""
    return None

validation_monitor 

validation_monitor(metrics)

Select checkpoints by validation error as in the released trainer.

Source code in methods/muse/adapter.py 
276
277
278
def validation_monitor(self, metrics):
    """Select checkpoints by validation error as in the released trainer."""
    return "val_error", 1.0 - float(metrics["val_accuracy"])

interpret_step 

interpret_step(batch, model, target_class=None)

Expose MUSE's label-free class-semantic patch routing.

Source code in methods/muse/adapter.py 
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
@torch.no_grad()
def interpret_step(self, batch, model, target_class=None):
    """Expose MUSE's label-free class-semantic patch routing."""
    features = batch[0].to(self.device)
    details = model(features, return_details=True)
    logits = details["logits"]
    class_index = (
        int(logits.argmax(dim=1).item())
        if target_class is None else int(target_class))
    if class_index < 0 or class_index >= logits.shape[1]:
        raise ValueError("MUSE target class is outside the classifier range")
    return InterpretabilityResult(
        logits=logits,
        evidence=(PatchEvidence(
            scores=details["patch_attention"][class_index],
            score_type="class_semantic_routing_attention",
            description=(
                "Native MUSE evaluation-branch attention after sparse "
                "expert routing and top-patch filtering for the selected "
                "class semantic token."),
            class_index=class_index,
            feature_path_key="feature_path_column",
        ),),
        notes=(
            "This is the label-free MUSE evaluation branch; no "
            "ground-truth semantic retrieval is used.",
        ),
    )

get_backbone_contract classmethod 

get_backbone_contract() -> MethodBackboneContract

Return the encoder/feature contract declared by the adapter.

Raises:

Type Description
NotImplementedError

If the adapter omitted its required contract.
Source code in methods/base.py 
103
104
105
106
107
108
109
110
111
112
113
@classmethod
def get_backbone_contract(cls) -> MethodBackboneContract:
    """Return the encoder/feature contract declared by the adapter.

    Raises:
        NotImplementedError: If the adapter omitted its required contract.
    """
    if cls.backbone_contract is None:
        raise NotImplementedError(
            f"Method adapter {cls.__name__} has no backbone contract")
    return cls.backbone_contract

load_encoder 

load_encoder(
    *,
    weights_path: Optional[str] = None,
    **loader_options: Any,
) -> EncoderBundle

Build and validate the configured encoder in one operation.

Parameters:

Name Type Description Default
weights_path Optional[str]

Optional local checkpoint or model identifier. When omitted, cfg['backbone_weights'] is used.

 None
**loader_options Any

Family-specific options forwarded to the encoder registry, such as a pinned revision or offline-only loading.

 {}

Returns:

Type Description
EncoderBundle

A validated capability-aware encoder bundle.

Raises:

Type Description
RuntimeError

If the adapter has no runtime backbone contract.
BackboneCompatibilityError

If the bundle violates the method contract or declared feature provenance.
Source code in methods/base.py 
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
def load_encoder(self, *, weights_path: Optional[str] = None,
                 **loader_options: Any) -> EncoderBundle:
    """Build and validate the configured encoder in one operation.

    Args:
        weights_path: Optional local checkpoint or model identifier. When
            omitted, ``cfg['backbone_weights']`` is used.
        **loader_options: Family-specific options forwarded to the encoder
            registry, such as a pinned revision or offline-only loading.

    Returns:
        A validated capability-aware encoder bundle.

    Raises:
        RuntimeError: If the adapter has no runtime backbone contract.
        BackboneCompatibilityError: If the bundle violates the method
            contract or declared feature provenance.
    """
    if self.backbone_contract is None or self.backbone_name is None:
        raise RuntimeError(f"Method '{self.name}' has no runtime backbone")
    from common.backbones import build_encoder
    if weights_path is None:
        weights_path = self.cfg.get("backbone_weights")
    bundle = build_encoder(self.backbone_name, weights_path=weights_path,
                           device=self.device, **loader_options)
    return self.backbone_contract.validate_bundle(self.cfg, bundle)

build_optimizer 

build_optimizer(model: Module) -> Optimizer

Build Adam over trainable parameters using the configured recipe.

Source code in methods/base.py 
172
173
174
175
176
177
178
179
def build_optimizer(self, model: nn.Module) -> torch.optim.Optimizer:
    """Build Adam over trainable parameters using the configured recipe."""
    params = filter(lambda p: p.requires_grad, model.parameters())
    return torch.optim.Adam(
        params,
        lr=self.cfg.get("lr", 1e-4),
        weight_decay=self.cfg.get("weight_decay", 1e-5),
    )

prepare_fold 

prepare_fold(
    fold: int, model: Module, train_loader: Any
) -> None

Prepare training-split-only state after model/data construction.

Methods such as PathPT use this boundary to select prompts without seeing validation or test slides. The default is deliberately a no-op.

Source code in methods/base.py 
204
205
206
207
208
209
210
211
def prepare_fold(
    self, fold: int, model: nn.Module, train_loader: Any,
) -> None:
    """Prepare training-split-only state after model/data construction.

    Methods such as PathPT use this boundary to select prompts without
    seeing validation or test slides. The default is deliberately a no-op.
    """

on_train_epoch_start 

on_train_epoch_start(epoch: int, model: Module) -> None

Handle the boundary immediately before an epoch's first batch.

Source code in methods/base.py 
213
214
def on_train_epoch_start(self, epoch: int, model: nn.Module) -> None:
    """Handle the boundary immediately before an epoch's first batch."""

on_epoch_end 

on_epoch_end(epoch: int, metrics: Dict[str, float]) -> None

Handle an optional callback after a full train/validation epoch.

Source code in methods/base.py 
216
217
def on_epoch_end(self, epoch: int, metrics: Dict[str, float]) -> None:
    """Handle an optional callback after a full train/validation epoch."""

on_validation_end 

on_validation_end(
    epoch: int, metrics: Dict[str, float]
) -> None

Handle a completed validation pass, including epoch -1.

The default forwards to the historical on_epoch_end hook. Adapters which accumulate validation predictions therefore get a clear boundary for the optional initial evaluation as well as every trained epoch.

Source code in methods/base.py 
219
220
221
222
223
224
225
226
227
228
def on_validation_end(
    self, epoch: int, metrics: Dict[str, float],
) -> None:
    """Handle a completed validation pass, including epoch ``-1``.

    The default forwards to the historical ``on_epoch_end`` hook. Adapters
    which accumulate validation predictions therefore get a clear boundary
    for the optional initial evaluation as well as every trained epoch.
    """
    self.on_epoch_end(epoch, metrics)

on_fold_end 

on_fold_end(fold: int, metrics: Dict[str, float]) -> None

Handle an optional callback after a cross-validation fold.

Source code in methods/base.py 
230
231
def on_fold_end(self, fold: int, metrics: Dict[str, float]) -> None:
    """Handle an optional callback after a cross-validation fold."""

on_checkpoint_loaded 

on_checkpoint_loaded(
    model: Module, checkpoint_kind: str, fold: int
) -> None

Restore adapter state not represented by model.state_dict().

Source code in methods/base.py 
233
234
235
236
def on_checkpoint_loaded(
    self, model: nn.Module, checkpoint_kind: str, fold: int,
) -> None:
    """Restore adapter state not represented by ``model.state_dict()``."""

ConVLMMethod 

ConVLMMethod(cfg: Dict[str, Any], device: str = 'cuda')

Bases: BaseMethod

Run the local patch-bag, attribute-conditioned ConVLM reconstruction.

Source code in methods/base.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def __init__(self, cfg: Dict[str, Any], device: str = "cuda"):
    # A few vendored adapters fill in derived defaults while constructing
    # their model.  Keeping those writes private is essential: train.py has
    # already snapshotted and fingerprinted the resolved run config, and a
    # later mutation would otherwise make an interrupted run impossible to
    # resume.  A deep copy also prevents nested recipe dictionaries from
    # carrying state between fold-specific adapter instances.
    self.cfg = deepcopy(cfg)
    self.device = device
    self.backbone_name = (self.backbone_contract.validate_config(self.cfg)
                          if self.backbone_contract is not None else None)

get_backbone_contract classmethod 

get_backbone_contract() -> MethodBackboneContract

Return the encoder/feature contract declared by the adapter.

Raises:

Type Description
NotImplementedError

If the adapter omitted its required contract.
Source code in methods/base.py 
103
104
105
106
107
108
109
110
111
112
113
@classmethod
def get_backbone_contract(cls) -> MethodBackboneContract:
    """Return the encoder/feature contract declared by the adapter.

    Raises:
        NotImplementedError: If the adapter omitted its required contract.
    """
    if cls.backbone_contract is None:
        raise NotImplementedError(
            f"Method adapter {cls.__name__} has no backbone contract")
    return cls.backbone_contract

load_encoder 

load_encoder(
    *,
    weights_path: Optional[str] = None,
    **loader_options: Any,
) -> EncoderBundle

Build and validate the configured encoder in one operation.

Parameters:

Name Type Description Default
weights_path Optional[str]

Optional local checkpoint or model identifier. When omitted, cfg['backbone_weights'] is used.

 None
**loader_options Any

Family-specific options forwarded to the encoder registry, such as a pinned revision or offline-only loading.

 {}

Returns:

Type Description
EncoderBundle

A validated capability-aware encoder bundle.

Raises:

Type Description
RuntimeError

If the adapter has no runtime backbone contract.
BackboneCompatibilityError

If the bundle violates the method contract or declared feature provenance.
Source code in methods/base.py 
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
def load_encoder(self, *, weights_path: Optional[str] = None,
                 **loader_options: Any) -> EncoderBundle:
    """Build and validate the configured encoder in one operation.

    Args:
        weights_path: Optional local checkpoint or model identifier. When
            omitted, ``cfg['backbone_weights']`` is used.
        **loader_options: Family-specific options forwarded to the encoder
            registry, such as a pinned revision or offline-only loading.

    Returns:
        A validated capability-aware encoder bundle.

    Raises:
        RuntimeError: If the adapter has no runtime backbone contract.
        BackboneCompatibilityError: If the bundle violates the method
            contract or declared feature provenance.
    """
    if self.backbone_contract is None or self.backbone_name is None:
        raise RuntimeError(f"Method '{self.name}' has no runtime backbone")
    from common.backbones import build_encoder
    if weights_path is None:
        weights_path = self.cfg.get("backbone_weights")
    bundle = build_encoder(self.backbone_name, weights_path=weights_path,
                           device=self.device, **loader_options)
    return self.backbone_contract.validate_bundle(self.cfg, bundle)

validation_monitor 

validation_monitor(
    metrics: Dict[str, Any],
) -> tuple[str, float]

Return the minimized value used for best-checkpoint selection.

The unified loop cannot assume that every released method selects its checkpoint by cross-entropy. Several paper repositories use validation error, while MSCPT uses a task-dependent F1/AUC or F1/ACC composite. Adapters override this hook to preserve that executable recipe. Values are always expressed as a minimization objective so the checkpoint writer has one unambiguous contract.

Source code in methods/base.py 
190
191
192
193
194
195
196
197
198
199
200
201
202
def validation_monitor(
    self, metrics: Dict[str, Any],
) -> tuple[str, float]:
    """Return the minimized value used for best-checkpoint selection.

    The unified loop cannot assume that every released method selects its
    checkpoint by cross-entropy.  Several paper repositories use
    validation error, while MSCPT uses a task-dependent F1/AUC or F1/ACC
    composite.  Adapters override this hook to preserve that executable
    recipe.  Values are always expressed as a minimization objective so
    the checkpoint writer has one unambiguous contract.
    """
    return "val_loss", float(metrics["val_loss"])

prepare_fold 

prepare_fold(
    fold: int, model: Module, train_loader: Any
) -> None

Prepare training-split-only state after model/data construction.

Methods such as PathPT use this boundary to select prompts without seeing validation or test slides. The default is deliberately a no-op.

Source code in methods/base.py 
204
205
206
207
208
209
210
211
def prepare_fold(
    self, fold: int, model: nn.Module, train_loader: Any,
) -> None:
    """Prepare training-split-only state after model/data construction.

    Methods such as PathPT use this boundary to select prompts without
    seeing validation or test slides. The default is deliberately a no-op.
    """

on_train_epoch_start 

on_train_epoch_start(epoch: int, model: Module) -> None

Handle the boundary immediately before an epoch's first batch.

Source code in methods/base.py 
213
214
def on_train_epoch_start(self, epoch: int, model: nn.Module) -> None:
    """Handle the boundary immediately before an epoch's first batch."""

on_epoch_end 

on_epoch_end(epoch: int, metrics: Dict[str, float]) -> None

Handle an optional callback after a full train/validation epoch.

Source code in methods/base.py 
216
217
def on_epoch_end(self, epoch: int, metrics: Dict[str, float]) -> None:
    """Handle an optional callback after a full train/validation epoch."""

on_validation_end 

on_validation_end(
    epoch: int, metrics: Dict[str, float]
) -> None

Handle a completed validation pass, including epoch -1.

The default forwards to the historical on_epoch_end hook. Adapters which accumulate validation predictions therefore get a clear boundary for the optional initial evaluation as well as every trained epoch.

Source code in methods/base.py 
219
220
221
222
223
224
225
226
227
228
def on_validation_end(
    self, epoch: int, metrics: Dict[str, float],
) -> None:
    """Handle a completed validation pass, including epoch ``-1``.

    The default forwards to the historical ``on_epoch_end`` hook. Adapters
    which accumulate validation predictions therefore get a clear boundary
    for the optional initial evaluation as well as every trained epoch.
    """
    self.on_epoch_end(epoch, metrics)

on_fold_end 

on_fold_end(fold: int, metrics: Dict[str, float]) -> None

Handle an optional callback after a cross-validation fold.

Source code in methods/base.py 
230
231
def on_fold_end(self, fold: int, metrics: Dict[str, float]) -> None:
    """Handle an optional callback after a cross-validation fold."""

on_checkpoint_loaded 

on_checkpoint_loaded(
    model: Module, checkpoint_kind: str, fold: int
) -> None

Restore adapter state not represented by model.state_dict().

Source code in methods/base.py 
233
234
235
236
def on_checkpoint_loaded(
    self, model: nn.Module, checkpoint_kind: str, fold: int,
) -> None:
    """Restore adapter state not represented by ``model.state_dict()``."""

interpret_step 

interpret_step(
    batch: Any,
    model: Module,
    target_class: int | None = None,
) -> "InterpretabilityResult"

Return native, patch-aligned evidence for one evaluation slide.

Adapters must override this hook only when their score semantics and patch ordering are known. The default intentionally refuses to invent a generic "attention" map from hidden activations.

Parameters:

Name Type Description Default
batch Any

One loader batch for a single slide.

 required
model Module

Loaded fold checkpoint in evaluation mode.

 required
target_class int | None

Optional class index for class-specific evidence.

 None

Raises:

Type Description
NotImplementedError

If the method has no audited patch-evidence mapping in PGVL-Gym.
Source code in methods/base.py 
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
def interpret_step(
    self, batch: Any, model: nn.Module, target_class: int | None = None,
) -> "InterpretabilityResult":
    """Return native, patch-aligned evidence for one evaluation slide.

    Adapters must override this hook only when their score semantics and
    patch ordering are known. The default intentionally refuses to invent
    a generic "attention" map from hidden activations.

    Args:
        batch: One loader batch for a single slide.
        model: Loaded fold checkpoint in evaluation mode.
        target_class: Optional class index for class-specific evidence.

    Raises:
        NotImplementedError: If the method has no audited patch-evidence
            mapping in PGVL-Gym.
    """
    del batch, model, target_class
    raise NotImplementedError(
        f"Method '{self.name}' has no audited patch-level "
        "interpretability provider")

SLDPCMethod 

SLDPCMethod(cfg: Dict[str, Any], device: str = 'cuda')

Bases: BaseMethod

Adapt SLDPC's staged prompt refinement over registered slide vectors.

Source code in methods/base.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def __init__(self, cfg: Dict[str, Any], device: str = "cuda"):
    # A few vendored adapters fill in derived defaults while constructing
    # their model.  Keeping those writes private is essential: train.py has
    # already snapshotted and fingerprinted the resolved run config, and a
    # later mutation would otherwise make an interrupted run impossible to
    # resume.  A deep copy also prevents nested recipe dictionaries from
    # carrying state between fold-specific adapter instances.
    self.cfg = deepcopy(cfg)
    self.device = device
    self.backbone_name = (self.backbone_contract.validate_config(self.cfg)
                          if self.backbone_contract is not None else None)

get_backbone_contract classmethod 

get_backbone_contract() -> MethodBackboneContract

Return the encoder/feature contract declared by the adapter.

Raises:

Type Description
NotImplementedError

If the adapter omitted its required contract.
Source code in methods/base.py 
103
104
105
106
107
108
109
110
111
112
113
@classmethod
def get_backbone_contract(cls) -> MethodBackboneContract:
    """Return the encoder/feature contract declared by the adapter.

    Raises:
        NotImplementedError: If the adapter omitted its required contract.
    """
    if cls.backbone_contract is None:
        raise NotImplementedError(
            f"Method adapter {cls.__name__} has no backbone contract")
    return cls.backbone_contract

load_encoder 

load_encoder(
    *,
    weights_path: Optional[str] = None,
    **loader_options: Any,
) -> EncoderBundle

Build and validate the configured encoder in one operation.

Parameters:

Name Type Description Default
weights_path Optional[str]

Optional local checkpoint or model identifier. When omitted, cfg['backbone_weights'] is used.

 None
**loader_options Any

Family-specific options forwarded to the encoder registry, such as a pinned revision or offline-only loading.

 {}

Returns:

Type Description
EncoderBundle

A validated capability-aware encoder bundle.

Raises:

Type Description
RuntimeError

If the adapter has no runtime backbone contract.
BackboneCompatibilityError

If the bundle violates the method contract or declared feature provenance.
Source code in methods/base.py 
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
def load_encoder(self, *, weights_path: Optional[str] = None,
                 **loader_options: Any) -> EncoderBundle:
    """Build and validate the configured encoder in one operation.

    Args:
        weights_path: Optional local checkpoint or model identifier. When
            omitted, ``cfg['backbone_weights']`` is used.
        **loader_options: Family-specific options forwarded to the encoder
            registry, such as a pinned revision or offline-only loading.

    Returns:
        A validated capability-aware encoder bundle.

    Raises:
        RuntimeError: If the adapter has no runtime backbone contract.
        BackboneCompatibilityError: If the bundle violates the method
            contract or declared feature provenance.
    """
    if self.backbone_contract is None or self.backbone_name is None:
        raise RuntimeError(f"Method '{self.name}' has no runtime backbone")
    from common.backbones import build_encoder
    if weights_path is None:
        weights_path = self.cfg.get("backbone_weights")
    bundle = build_encoder(self.backbone_name, weights_path=weights_path,
                           device=self.device, **loader_options)
    return self.backbone_contract.validate_bundle(self.cfg, bundle)

validation_monitor 

validation_monitor(
    metrics: Dict[str, Any],
) -> tuple[str, float]

Return the minimized value used for best-checkpoint selection.

The unified loop cannot assume that every released method selects its checkpoint by cross-entropy. Several paper repositories use validation error, while MSCPT uses a task-dependent F1/AUC or F1/ACC composite. Adapters override this hook to preserve that executable recipe. Values are always expressed as a minimization objective so the checkpoint writer has one unambiguous contract.

Source code in methods/base.py 
190
191
192
193
194
195
196
197
198
199
200
201
202
def validation_monitor(
    self, metrics: Dict[str, Any],
) -> tuple[str, float]:
    """Return the minimized value used for best-checkpoint selection.

    The unified loop cannot assume that every released method selects its
    checkpoint by cross-entropy.  Several paper repositories use
    validation error, while MSCPT uses a task-dependent F1/AUC or F1/ACC
    composite.  Adapters override this hook to preserve that executable
    recipe.  Values are always expressed as a minimization objective so
    the checkpoint writer has one unambiguous contract.
    """
    return "val_loss", float(metrics["val_loss"])

prepare_fold 

prepare_fold(
    fold: int, model: Module, train_loader: Any
) -> None

Prepare training-split-only state after model/data construction.

Methods such as PathPT use this boundary to select prompts without seeing validation or test slides. The default is deliberately a no-op.

Source code in methods/base.py 
204
205
206
207
208
209
210
211
def prepare_fold(
    self, fold: int, model: nn.Module, train_loader: Any,
) -> None:
    """Prepare training-split-only state after model/data construction.

    Methods such as PathPT use this boundary to select prompts without
    seeing validation or test slides. The default is deliberately a no-op.
    """

on_train_epoch_start 

on_train_epoch_start(epoch: int, model: Module) -> None

Handle the boundary immediately before an epoch's first batch.

Source code in methods/base.py 
213
214
def on_train_epoch_start(self, epoch: int, model: nn.Module) -> None:
    """Handle the boundary immediately before an epoch's first batch."""

on_validation_end 

on_validation_end(
    epoch: int, metrics: Dict[str, float]
) -> None

Handle a completed validation pass, including epoch -1.

The default forwards to the historical on_epoch_end hook. Adapters which accumulate validation predictions therefore get a clear boundary for the optional initial evaluation as well as every trained epoch.

Source code in methods/base.py 
219
220
221
222
223
224
225
226
227
228
def on_validation_end(
    self, epoch: int, metrics: Dict[str, float],
) -> None:
    """Handle a completed validation pass, including epoch ``-1``.

    The default forwards to the historical ``on_epoch_end`` hook. Adapters
    which accumulate validation predictions therefore get a clear boundary
    for the optional initial evaluation as well as every trained epoch.
    """
    self.on_epoch_end(epoch, metrics)

interpret_step 

interpret_step(
    batch: Any,
    model: Module,
    target_class: int | None = None,
) -> "InterpretabilityResult"

Return native, patch-aligned evidence for one evaluation slide.

Adapters must override this hook only when their score semantics and patch ordering are known. The default intentionally refuses to invent a generic "attention" map from hidden activations.

Parameters:

Name Type Description Default
batch Any

One loader batch for a single slide.

 required
model Module

Loaded fold checkpoint in evaluation mode.

 required
target_class int | None

Optional class index for class-specific evidence.

 None

Raises:

Type Description
NotImplementedError

If the method has no audited patch-evidence mapping in PGVL-Gym.
Source code in methods/base.py 
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
def interpret_step(
    self, batch: Any, model: nn.Module, target_class: int | None = None,
) -> "InterpretabilityResult":
    """Return native, patch-aligned evidence for one evaluation slide.

    Adapters must override this hook only when their score semantics and
    patch ordering are known. The default intentionally refuses to invent
    a generic "attention" map from hidden activations.

    Args:
        batch: One loader batch for a single slide.
        model: Loaded fold checkpoint in evaluation mode.
        target_class: Optional class index for class-specific evidence.

    Raises:
        NotImplementedError: If the method has no audited patch-evidence
            mapping in PGVL-Gym.
    """
    del batch, model, target_class
    raise NotImplementedError(
        f"Method '{self.name}' has no audited patch-level "
        "interpretability provider")

HiVEMILMethod 

HiVEMILMethod(cfg, device='cuda')

Bases: BaseMethod

Run native CONCH HiVE-MIL on coordinate-aligned 5x/20x features.

Source code in methods/hive_mil/adapter.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def __init__(self, cfg, device="cuda"):
    super().__init__(cfg, device)
    required = {
        "feature_path_column_l", "feature_path_column_s",
        "hierarchy_geometry", "text_prompt_path",
    }
    missing = sorted(key for key in required if not self.cfg.get(key))
    if missing:
        raise ValueError(
            f"HiVE-MIL config is missing required keys: {missing}")
    if int(self.cfg.get("max_children", 16)) != 16:
        raise ValueError(
            "HiVE-MIL's released 5x-to-20x graph requires max_children=16")
    if self.cfg["hierarchy_geometry"] != "per_slide_hdf5":
        raise ValueError(
            "HiVE-MIL requires hierarchy_geometry='per_slide_hdf5'")
    low_mag = str(self.cfg.get("low_mag", "5x"))
    high_mag = str(self.cfg.get("high_mag", "20x"))
    if (low_mag, high_mag) != ("5x", "20x"):
        raise ValueError(
            "HiVE-MIL's released hierarchy requires low_mag=5x and high_mag=20x")
    declared = self.cfg.get("upstream_commit")
    if declared and declared != UPSTREAM_COMMIT:
        raise ValueError(
            "HiVE-MIL config upstream_commit does not match the vendored "
            f"implementation ({UPSTREAM_COMMIT})")

get_backbone_contract classmethod 

get_backbone_contract() -> MethodBackboneContract

Return the encoder/feature contract declared by the adapter.

Raises:

Type Description
NotImplementedError

If the adapter omitted its required contract.
Source code in methods/base.py 
103
104
105
106
107
108
109
110
111
112
113
@classmethod
def get_backbone_contract(cls) -> MethodBackboneContract:
    """Return the encoder/feature contract declared by the adapter.

    Raises:
        NotImplementedError: If the adapter omitted its required contract.
    """
    if cls.backbone_contract is None:
        raise NotImplementedError(
            f"Method adapter {cls.__name__} has no backbone contract")
    return cls.backbone_contract

load_encoder 

load_encoder(
    *,
    weights_path: Optional[str] = None,
    **loader_options: Any,
) -> EncoderBundle

Build and validate the configured encoder in one operation.

Parameters:

Name Type Description Default
weights_path Optional[str]

Optional local checkpoint or model identifier. When omitted, cfg['backbone_weights'] is used.

 None
**loader_options Any

Family-specific options forwarded to the encoder registry, such as a pinned revision or offline-only loading.

 {}

Returns:

Type Description
EncoderBundle

A validated capability-aware encoder bundle.

Raises:

Type Description
RuntimeError

If the adapter has no runtime backbone contract.
BackboneCompatibilityError

If the bundle violates the method contract or declared feature provenance.
Source code in methods/base.py 
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
def load_encoder(self, *, weights_path: Optional[str] = None,
                 **loader_options: Any) -> EncoderBundle:
    """Build and validate the configured encoder in one operation.

    Args:
        weights_path: Optional local checkpoint or model identifier. When
            omitted, ``cfg['backbone_weights']`` is used.
        **loader_options: Family-specific options forwarded to the encoder
            registry, such as a pinned revision or offline-only loading.

    Returns:
        A validated capability-aware encoder bundle.

    Raises:
        RuntimeError: If the adapter has no runtime backbone contract.
        BackboneCompatibilityError: If the bundle violates the method
            contract or declared feature provenance.
    """
    if self.backbone_contract is None or self.backbone_name is None:
        raise RuntimeError(f"Method '{self.name}' has no runtime backbone")
    from common.backbones import build_encoder
    if weights_path is None:
        weights_path = self.cfg.get("backbone_weights")
    bundle = build_encoder(self.backbone_name, weights_path=weights_path,
                           device=self.device, **loader_options)
    return self.backbone_contract.validate_bundle(self.cfg, bundle)

build_optimizer 

build_optimizer(model: Module) -> Optimizer

Build Adam over trainable parameters using the configured recipe.

Source code in methods/base.py 
172
173
174
175
176
177
178
179
def build_optimizer(self, model: nn.Module) -> torch.optim.Optimizer:
    """Build Adam over trainable parameters using the configured recipe."""
    params = filter(lambda p: p.requires_grad, model.parameters())
    return torch.optim.Adam(
        params,
        lr=self.cfg.get("lr", 1e-4),
        weight_decay=self.cfg.get("weight_decay", 1e-5),
    )

prepare_fold 

prepare_fold(
    fold: int, model: Module, train_loader: Any
) -> None

Prepare training-split-only state after model/data construction.

Methods such as PathPT use this boundary to select prompts without seeing validation or test slides. The default is deliberately a no-op.

Source code in methods/base.py 
204
205
206
207
208
209
210
211
def prepare_fold(
    self, fold: int, model: nn.Module, train_loader: Any,
) -> None:
    """Prepare training-split-only state after model/data construction.

    Methods such as PathPT use this boundary to select prompts without
    seeing validation or test slides. The default is deliberately a no-op.
    """

on_train_epoch_start 

on_train_epoch_start(epoch: int, model: Module) -> None

Handle the boundary immediately before an epoch's first batch.

Source code in methods/base.py 
213
214
def on_train_epoch_start(self, epoch: int, model: nn.Module) -> None:
    """Handle the boundary immediately before an epoch's first batch."""

on_epoch_end 

on_epoch_end(epoch: int, metrics: Dict[str, float]) -> None

Handle an optional callback after a full train/validation epoch.

Source code in methods/base.py 
216
217
def on_epoch_end(self, epoch: int, metrics: Dict[str, float]) -> None:
    """Handle an optional callback after a full train/validation epoch."""

on_validation_end 

on_validation_end(
    epoch: int, metrics: Dict[str, float]
) -> None

Handle a completed validation pass, including epoch -1.

The default forwards to the historical on_epoch_end hook. Adapters which accumulate validation predictions therefore get a clear boundary for the optional initial evaluation as well as every trained epoch.

Source code in methods/base.py 
219
220
221
222
223
224
225
226
227
228
def on_validation_end(
    self, epoch: int, metrics: Dict[str, float],
) -> None:
    """Handle a completed validation pass, including epoch ``-1``.

    The default forwards to the historical ``on_epoch_end`` hook. Adapters
    which accumulate validation predictions therefore get a clear boundary
    for the optional initial evaluation as well as every trained epoch.
    """
    self.on_epoch_end(epoch, metrics)

on_fold_end 

on_fold_end(fold: int, metrics: Dict[str, float]) -> None

Handle an optional callback after a cross-validation fold.

Source code in methods/base.py 
230
231
def on_fold_end(self, fold: int, metrics: Dict[str, float]) -> None:
    """Handle an optional callback after a cross-validation fold."""

on_checkpoint_loaded 

on_checkpoint_loaded(
    model: Module, checkpoint_kind: str, fold: int
) -> None

Restore adapter state not represented by model.state_dict().

Source code in methods/base.py 
233
234
235
236
def on_checkpoint_loaded(
    self, model: nn.Module, checkpoint_kind: str, fold: int,
) -> None:
    """Restore adapter state not represented by ``model.state_dict()``."""

interpret_step 

interpret_step(
    batch: Any,
    model: Module,
    target_class: int | None = None,
) -> "InterpretabilityResult"

Return native, patch-aligned evidence for one evaluation slide.

Adapters must override this hook only when their score semantics and patch ordering are known. The default intentionally refuses to invent a generic "attention" map from hidden activations.

Parameters:

Name Type Description Default
batch Any

One loader batch for a single slide.

 required
model Module

Loaded fold checkpoint in evaluation mode.

 required
target_class int | None

Optional class index for class-specific evidence.

 None

Raises:

Type Description
NotImplementedError

If the method has no audited patch-evidence mapping in PGVL-Gym.
Source code in methods/base.py 
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
def interpret_step(
    self, batch: Any, model: nn.Module, target_class: int | None = None,
) -> "InterpretabilityResult":
    """Return native, patch-aligned evidence for one evaluation slide.

    Adapters must override this hook only when their score semantics and
    patch ordering are known. The default intentionally refuses to invent
    a generic "attention" map from hidden activations.

    Args:
        batch: One loader batch for a single slide.
        model: Loaded fold checkpoint in evaluation mode.
        target_class: Optional class index for class-specific evidence.

    Raises:
        NotImplementedError: If the method has no audited patch-evidence
            mapping in PGVL-Gym.
    """
    del batch, model, target_class
    raise NotImplementedError(
        f"Method '{self.name}' has no audited patch-level "
        "interpretability provider")

MIVisionShotMethod 

MIVisionShotMethod(cfg, device='cuda')

Bases: BaseMethod

Native PLIP top-K support prototypes with label-free BGAP inference.

Source code in methods/mi_visionshot/adapter.py 
46
47
48
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
def __init__(self, cfg, device="cuda"):
    super().__init__(cfg, device)
    expected = {
        "training_mode": "nonparametric_prototypes",
        "support_aggregation": "label_guided_topk",
        "inference_aggregation": "normalized_bgap",
        "top_k": 200,
        "feature_resolutions": {"bag": "20x"},
        "upstream_commit": UPSTREAM_COMMIT,
    }
    drifted = {
        key: (self.cfg.get(key), value)
        for key, value in expected.items()
        if self.cfg.get(key) != value
    }
    if drifted:
        raise ValueError(
            f"MI-VisionShot config contradicts its published recipe: "
            f"{drifted}")
    if not self.cfg.get("text_prompt_path"):
        raise ValueError("MI-VisionShot requires text_prompt_path")
    if self.cfg.get("feature_projection") not in {
            "none", "native_visual_projection"}:
        raise ValueError(
            "MI-VisionShot feature_projection must be none or "
            "native_visual_projection")
    normalization_by_recipe = {
        "released_code_unnormalized": "none",
        "paper_eq2_2_l2": "l2",
    }
    recipe = self.cfg.get("mi_visionshot_recipe")
    if recipe not in normalization_by_recipe:
        raise ValueError(
            "MI-VisionShot mi_visionshot_recipe must be "
            "released_code_unnormalized or paper_eq2_2_l2")
    expected_normalization = normalization_by_recipe[recipe]
    if self.cfg.get(
            "patch_similarity_normalization") != expected_normalization:
        raise ValueError(
            "MI-VisionShot patch normalization contradicts its recipe: "
            f"expected {expected_normalization!r}")

get_backbone_contract classmethod 

get_backbone_contract() -> MethodBackboneContract

Return the encoder/feature contract declared by the adapter.

Raises:

Type Description
NotImplementedError

If the adapter omitted its required contract.
Source code in methods/base.py 
103
104
105
106
107
108
109
110
111
112
113
@classmethod
def get_backbone_contract(cls) -> MethodBackboneContract:
    """Return the encoder/feature contract declared by the adapter.

    Raises:
        NotImplementedError: If the adapter omitted its required contract.
    """
    if cls.backbone_contract is None:
        raise NotImplementedError(
            f"Method adapter {cls.__name__} has no backbone contract")
    return cls.backbone_contract

load_encoder 

load_encoder(
    *,
    weights_path: Optional[str] = None,
    **loader_options: Any,
) -> EncoderBundle

Build and validate the configured encoder in one operation.

Parameters:

Name Type Description Default
weights_path Optional[str]

Optional local checkpoint or model identifier. When omitted, cfg['backbone_weights'] is used.

 None
**loader_options Any

Family-specific options forwarded to the encoder registry, such as a pinned revision or offline-only loading.

 {}

Returns:

Type Description
EncoderBundle

A validated capability-aware encoder bundle.

Raises:

Type Description
RuntimeError

If the adapter has no runtime backbone contract.
BackboneCompatibilityError

If the bundle violates the method contract or declared feature provenance.
Source code in methods/base.py 
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
def load_encoder(self, *, weights_path: Optional[str] = None,
                 **loader_options: Any) -> EncoderBundle:
    """Build and validate the configured encoder in one operation.

    Args:
        weights_path: Optional local checkpoint or model identifier. When
            omitted, ``cfg['backbone_weights']`` is used.
        **loader_options: Family-specific options forwarded to the encoder
            registry, such as a pinned revision or offline-only loading.

    Returns:
        A validated capability-aware encoder bundle.

    Raises:
        RuntimeError: If the adapter has no runtime backbone contract.
        BackboneCompatibilityError: If the bundle violates the method
            contract or declared feature provenance.
    """
    if self.backbone_contract is None or self.backbone_name is None:
        raise RuntimeError(f"Method '{self.name}' has no runtime backbone")
    from common.backbones import build_encoder
    if weights_path is None:
        weights_path = self.cfg.get("backbone_weights")
    bundle = build_encoder(self.backbone_name, weights_path=weights_path,
                           device=self.device, **loader_options)
    return self.backbone_contract.validate_bundle(self.cfg, bundle)

on_train_epoch_start 

on_train_epoch_start(epoch: int, model: Module) -> None

Handle the boundary immediately before an epoch's first batch.

Source code in methods/base.py 
213
214
def on_train_epoch_start(self, epoch: int, model: nn.Module) -> None:
    """Handle the boundary immediately before an epoch's first batch."""

on_epoch_end 

on_epoch_end(epoch: int, metrics: Dict[str, float]) -> None

Handle an optional callback after a full train/validation epoch.

Source code in methods/base.py 
216
217
def on_epoch_end(self, epoch: int, metrics: Dict[str, float]) -> None:
    """Handle an optional callback after a full train/validation epoch."""

on_validation_end 

on_validation_end(
    epoch: int, metrics: Dict[str, float]
) -> None

Handle a completed validation pass, including epoch -1.

The default forwards to the historical on_epoch_end hook. Adapters which accumulate validation predictions therefore get a clear boundary for the optional initial evaluation as well as every trained epoch.

Source code in methods/base.py 
219
220
221
222
223
224
225
226
227
228
def on_validation_end(
    self, epoch: int, metrics: Dict[str, float],
) -> None:
    """Handle a completed validation pass, including epoch ``-1``.

    The default forwards to the historical ``on_epoch_end`` hook. Adapters
    which accumulate validation predictions therefore get a clear boundary
    for the optional initial evaluation as well as every trained epoch.
    """
    self.on_epoch_end(epoch, metrics)

on_fold_end 

on_fold_end(fold: int, metrics: Dict[str, float]) -> None

Handle an optional callback after a cross-validation fold.

Source code in methods/base.py 
230
231
def on_fold_end(self, fold: int, metrics: Dict[str, float]) -> None:
    """Handle an optional callback after a cross-validation fold."""

on_checkpoint_loaded 

on_checkpoint_loaded(
    model: Module, checkpoint_kind: str, fold: int
) -> None

Restore adapter state not represented by model.state_dict().

Source code in methods/base.py 
233
234
235
236
def on_checkpoint_loaded(
    self, model: nn.Module, checkpoint_kind: str, fold: int,
) -> None:
    """Restore adapter state not represented by ``model.state_dict()``."""

interpret_step 

interpret_step(
    batch: Any,
    model: Module,
    target_class: int | None = None,
) -> "InterpretabilityResult"

Return native, patch-aligned evidence for one evaluation slide.

Adapters must override this hook only when their score semantics and patch ordering are known. The default intentionally refuses to invent a generic "attention" map from hidden activations.

Parameters:

Name Type Description Default
batch Any

One loader batch for a single slide.

 required
model Module

Loaded fold checkpoint in evaluation mode.

 required
target_class int | None

Optional class index for class-specific evidence.

 None

Raises:

Type Description
NotImplementedError

If the method has no audited patch-evidence mapping in PGVL-Gym.
Source code in methods/base.py 
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
def interpret_step(
    self, batch: Any, model: nn.Module, target_class: int | None = None,
) -> "InterpretabilityResult":
    """Return native, patch-aligned evidence for one evaluation slide.

    Adapters must override this hook only when their score semantics and
    patch ordering are known. The default intentionally refuses to invent
    a generic "attention" map from hidden activations.

    Args:
        batch: One loader batch for a single slide.
        model: Loaded fold checkpoint in evaluation mode.
        target_class: Optional class index for class-specific evidence.

    Raises:
        NotImplementedError: If the method has no audited patch-evidence
            mapping in PGVL-Gym.
    """
    del batch, model, target_class
    raise NotImplementedError(
        f"Method '{self.name}' has no audited patch-level "
        "interpretability provider")

LibraMILMethod 

LibraMILMethod(cfg, device='cuda')

Bases: BaseMethod

Run Libra-MIL's CONCH dual-prototype optimal-transport classifier.

Source code in methods/libra_mil/adapter.py 
 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
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def __init__(self, cfg, device="cuda"):
    super().__init__(cfg, device)
    self.is_encoder_extension = self.cfg.get("encoder_extension") is not None
    if self.is_encoder_extension:
        validate_encoder_extension(self.name, self.cfg)
        if self.cfg.get("encoder_extension_strategy") != \
                "paired_text_reencoding_v1":
            raise ValueError(
                "Libra-MIL encoder extensions require "
                "encoder_extension_strategy=paired_text_reencoding_v1")
    elif self.backbone_name != "conch":
        raise ValueError(
            "Non-CONCH Libra-MIL is an encoder extension and requires "
            "serialized encoder_extension provenance")
    expected = {
        "num_instance_prompts": INSTANCE_PROMPT_COUNT,
        "num_visual_prototypes": RCC_VISUAL_PROTOTYPES,
        "num_heads": 8,
        "ot_epsilon": 0.05,
        "ot_iterations": 20,
        "attention_temperature": 0.2,
        "feature_resolutions": {"bag": "20x"},
        "upstream_commit": UPSTREAM_COMMIT,
    }
    if not self.is_encoder_extension:
        expected["patch_geometry"] = "20x_512px_0px_overlap"
    drifted = {
        key: (self.cfg.get(key), value)
        for key, value in expected.items()
        if self.cfg.get(key) != value
    }
    if drifted:
        raise ValueError(
            f"Libra-MIL config contradicts the registered RCC recipe: {drifted}")
    missing = [
        key for key in ("instance_prompt_path", "bag_prompt_path")
        if not self.cfg.get(key)
    ]
    if missing:
        raise ValueError(f"Libra-MIL config is missing prompt assets: {missing}")
    self._train_batches = 1
    self._update_index = 0

interpret_step 

interpret_step(batch, model, target_class=None)

Expose Libra-MIL's native multimodal SOT pooling attention.

Source code in methods/libra_mil/adapter.py 
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
@torch.no_grad()
def interpret_step(self, batch, model, target_class=None):
    """Expose Libra-MIL's native multimodal SOT pooling attention."""
    del target_class  # The SOT patch marginal is shared by bag queries.
    features, _labels = self._unpack(batch, self.device)
    if self.is_encoder_extension:
        features = project_paired_features(model, features)
    details = model(features, return_details=True)
    return InterpretabilityResult(
        logits=details["logits"],
        evidence=(PatchEvidence(
            scores=details["patch_attention"][0],
            score_type="multimodal_transport_attention",
            description=(
                "Native Libra-MIL patch attention derived from fused "
                "visual/text prototype similarity and the SOT plan."),
            feature_path_key="feature_path_column",
        ),),
    )

get_backbone_contract classmethod 

get_backbone_contract() -> MethodBackboneContract

Return the encoder/feature contract declared by the adapter.

Raises:

Type Description
NotImplementedError

If the adapter omitted its required contract.
Source code in methods/base.py 
103
104
105
106
107
108
109
110
111
112
113
@classmethod
def get_backbone_contract(cls) -> MethodBackboneContract:
    """Return the encoder/feature contract declared by the adapter.

    Raises:
        NotImplementedError: If the adapter omitted its required contract.
    """
    if cls.backbone_contract is None:
        raise NotImplementedError(
            f"Method adapter {cls.__name__} has no backbone contract")
    return cls.backbone_contract

load_encoder 

load_encoder(
    *,
    weights_path: Optional[str] = None,
    **loader_options: Any,
) -> EncoderBundle

Build and validate the configured encoder in one operation.

Parameters:

Name Type Description Default
weights_path Optional[str]

Optional local checkpoint or model identifier. When omitted, cfg['backbone_weights'] is used.

 None
**loader_options Any

Family-specific options forwarded to the encoder registry, such as a pinned revision or offline-only loading.

 {}

Returns:

Type Description
EncoderBundle

A validated capability-aware encoder bundle.

Raises:

Type Description
RuntimeError

If the adapter has no runtime backbone contract.
BackboneCompatibilityError

If the bundle violates the method contract or declared feature provenance.
Source code in methods/base.py 
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
def load_encoder(self, *, weights_path: Optional[str] = None,
                 **loader_options: Any) -> EncoderBundle:
    """Build and validate the configured encoder in one operation.

    Args:
        weights_path: Optional local checkpoint or model identifier. When
            omitted, ``cfg['backbone_weights']`` is used.
        **loader_options: Family-specific options forwarded to the encoder
            registry, such as a pinned revision or offline-only loading.

    Returns:
        A validated capability-aware encoder bundle.

    Raises:
        RuntimeError: If the adapter has no runtime backbone contract.
        BackboneCompatibilityError: If the bundle violates the method
            contract or declared feature provenance.
    """
    if self.backbone_contract is None or self.backbone_name is None:
        raise RuntimeError(f"Method '{self.name}' has no runtime backbone")
    from common.backbones import build_encoder
    if weights_path is None:
        weights_path = self.cfg.get("backbone_weights")
    bundle = build_encoder(self.backbone_name, weights_path=weights_path,
                           device=self.device, **loader_options)
    return self.backbone_contract.validate_bundle(self.cfg, bundle)

on_train_epoch_start 

on_train_epoch_start(epoch: int, model: Module) -> None

Handle the boundary immediately before an epoch's first batch.

Source code in methods/base.py 
213
214
def on_train_epoch_start(self, epoch: int, model: nn.Module) -> None:
    """Handle the boundary immediately before an epoch's first batch."""

on_epoch_end 

on_epoch_end(epoch: int, metrics: Dict[str, float]) -> None

Handle an optional callback after a full train/validation epoch.

Source code in methods/base.py 
216
217
def on_epoch_end(self, epoch: int, metrics: Dict[str, float]) -> None:
    """Handle an optional callback after a full train/validation epoch."""

on_validation_end 

on_validation_end(
    epoch: int, metrics: Dict[str, float]
) -> None

Handle a completed validation pass, including epoch -1.

The default forwards to the historical on_epoch_end hook. Adapters which accumulate validation predictions therefore get a clear boundary for the optional initial evaluation as well as every trained epoch.

Source code in methods/base.py 
219
220
221
222
223
224
225
226
227
228
def on_validation_end(
    self, epoch: int, metrics: Dict[str, float],
) -> None:
    """Handle a completed validation pass, including epoch ``-1``.

    The default forwards to the historical ``on_epoch_end`` hook. Adapters
    which accumulate validation predictions therefore get a clear boundary
    for the optional initial evaluation as well as every trained epoch.
    """
    self.on_epoch_end(epoch, metrics)

on_fold_end 

on_fold_end(fold: int, metrics: Dict[str, float]) -> None

Handle an optional callback after a cross-validation fold.

Source code in methods/base.py 
230
231
def on_fold_end(self, fold: int, metrics: Dict[str, float]) -> None:
    """Handle an optional callback after a cross-validation fold."""

on_checkpoint_loaded 

on_checkpoint_loaded(
    model: Module, checkpoint_kind: str, fold: int
) -> None

Restore adapter state not represented by model.state_dict().

Source code in methods/base.py 
233
234
235
236
def on_checkpoint_loaded(
    self, model: nn.Module, checkpoint_kind: str, fold: int,
) -> None:
    """Restore adapter state not represented by ``model.state_dict()``."""

DyKoMethod 

DyKoMethod(cfg, device='cuda')

Bases: BaseMethod

Run clean-room DyKo on TITAN/CONCH-v1.5 patch features.

Source code in methods/dyko/adapter.py 
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
def __init__(self, cfg, device="cuda"):
    super().__init__(cfg, device)
    self.is_encoder_extension = self.cfg.get("encoder_extension") is not None
    if self.is_encoder_extension:
        validate_encoder_extension(self.name, self.cfg)
        if self.cfg.get("encoder_extension_strategy") != \
                "paired_feature_context_concept_bridge_v1":
            raise ValueError(
                "DyKo encoder extensions require encoder_extension_strategy="
                "paired_feature_context_concept_bridge_v1")
    elif self.backbone_name != "titan":
        raise ValueError(
            "Non-TITAN DyKo requires serialized encoder_extension provenance")
    expected = {
        "upstream_commit": UPSTREAM_COMMIT,
        "n_ctx": 16,
        "visual_prototypes": 10,
        "concepts_per_prototype": 10,
        "num_heads": 8,
        "retrieval_temperature": 0.1,
        "structural_consistency_weight": 1.0,
        "kmeans_iterations": 20,
        "kmeans_seed": 42,
    }
    if not self.is_encoder_extension:
        expected["patch_geometry"] = "20x_512px_0px_overlap_extension"
    drifted = {
        key: (self.cfg.get(key), value)
        for key, value in expected.items() if self.cfg.get(key) != value
    }
    if drifted:
        raise ValueError(f"DyKo config contradicts its registered recipe: {drifted}")
    for key in (
            "text_prompt_path", "prompt_file_classnames",
            "prompt_class_bindings", "concept_feature_path",
            "concept_feature_sha256"):
        if not self.cfg.get(key):
            raise ValueError(f"DyKo config is missing {key}")
    self._train_batches = 1
    self._batch_index = 0

interpret_step 

interpret_step(batch, model, target_class=None)

Expose DyKo's native visual prompt-to-patch attention.

Source code in methods/dyko/adapter.py 
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
@torch.no_grad()
def interpret_step(self, batch, model, target_class=None):
    """Expose DyKo's native visual prompt-to-patch attention."""
    patches, _labels = self._unpack(batch, self.device)
    if self.is_encoder_extension:
        patches = project_paired_features(model, patches)
    output = model(patches)
    attention = output["visual_patch_attention"]
    if attention.ndim != 3 or attention.shape[0] != 1:
        raise ValueError("DyKo visual attention has an unexpected shape")
    logits = output["logits"]
    query_count = attention.shape[1]
    if query_count == logits.shape[1]:
        class_index = (
            int(logits.argmax(dim=1).item())
            if target_class is None else int(target_class))
        if class_index < 0 or class_index >= query_count:
            raise ValueError("DyKo target class is outside the prompt range")
        scores = attention[0, class_index]
    else:
        if target_class is not None:
            raise ValueError(
                "DyKo prompt queries are not class-aligned for this config")
        class_index = None
        scores = attention[0].mean(dim=0)
    return InterpretabilityResult(
        logits=logits,
        evidence=(PatchEvidence(
            scores=scores,
            score_type="visual_prompt_cross_attention",
            description=(
                "Native DyKo visual prompt-to-patch cross-attention "
                "before knowledge fusion."),
            class_index=class_index,
            feature_path_key="feature_path_column",
        ),),
    )

get_backbone_contract classmethod 

get_backbone_contract() -> MethodBackboneContract

Return the encoder/feature contract declared by the adapter.

Raises:

Type Description
NotImplementedError

If the adapter omitted its required contract.
Source code in methods/base.py 
103
104
105
106
107
108
109
110
111
112
113
@classmethod
def get_backbone_contract(cls) -> MethodBackboneContract:
    """Return the encoder/feature contract declared by the adapter.

    Raises:
        NotImplementedError: If the adapter omitted its required contract.
    """
    if cls.backbone_contract is None:
        raise NotImplementedError(
            f"Method adapter {cls.__name__} has no backbone contract")
    return cls.backbone_contract

load_encoder 

load_encoder(
    *,
    weights_path: Optional[str] = None,
    **loader_options: Any,
) -> EncoderBundle

Build and validate the configured encoder in one operation.

Parameters:

Name Type Description Default
weights_path Optional[str]

Optional local checkpoint or model identifier. When omitted, cfg['backbone_weights'] is used.

 None
**loader_options Any

Family-specific options forwarded to the encoder registry, such as a pinned revision or offline-only loading.

 {}

Returns:

Type Description
EncoderBundle

A validated capability-aware encoder bundle.

Raises:

Type Description
RuntimeError

If the adapter has no runtime backbone contract.
BackboneCompatibilityError

If the bundle violates the method contract or declared feature provenance.
Source code in methods/base.py 
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
def load_encoder(self, *, weights_path: Optional[str] = None,
                 **loader_options: Any) -> EncoderBundle:
    """Build and validate the configured encoder in one operation.

    Args:
        weights_path: Optional local checkpoint or model identifier. When
            omitted, ``cfg['backbone_weights']`` is used.
        **loader_options: Family-specific options forwarded to the encoder
            registry, such as a pinned revision or offline-only loading.

    Returns:
        A validated capability-aware encoder bundle.

    Raises:
        RuntimeError: If the adapter has no runtime backbone contract.
        BackboneCompatibilityError: If the bundle violates the method
            contract or declared feature provenance.
    """
    if self.backbone_contract is None or self.backbone_name is None:
        raise RuntimeError(f"Method '{self.name}' has no runtime backbone")
    from common.backbones import build_encoder
    if weights_path is None:
        weights_path = self.cfg.get("backbone_weights")
    bundle = build_encoder(self.backbone_name, weights_path=weights_path,
                           device=self.device, **loader_options)
    return self.backbone_contract.validate_bundle(self.cfg, bundle)

build_optimizer 

build_optimizer(model: Module) -> Optimizer

Build Adam over trainable parameters using the configured recipe.

Source code in methods/base.py 
172
173
174
175
176
177
178
179
def build_optimizer(self, model: nn.Module) -> torch.optim.Optimizer:
    """Build Adam over trainable parameters using the configured recipe."""
    params = filter(lambda p: p.requires_grad, model.parameters())
    return torch.optim.Adam(
        params,
        lr=self.cfg.get("lr", 1e-4),
        weight_decay=self.cfg.get("weight_decay", 1e-5),
    )

on_epoch_end 

on_epoch_end(epoch: int, metrics: Dict[str, float]) -> None

Handle an optional callback after a full train/validation epoch.

Source code in methods/base.py 
216
217
def on_epoch_end(self, epoch: int, metrics: Dict[str, float]) -> None:
    """Handle an optional callback after a full train/validation epoch."""

on_validation_end 

on_validation_end(
    epoch: int, metrics: Dict[str, float]
) -> None

Handle a completed validation pass, including epoch -1.

The default forwards to the historical on_epoch_end hook. Adapters which accumulate validation predictions therefore get a clear boundary for the optional initial evaluation as well as every trained epoch.

Source code in methods/base.py 
219
220
221
222
223
224
225
226
227
228
def on_validation_end(
    self, epoch: int, metrics: Dict[str, float],
) -> None:
    """Handle a completed validation pass, including epoch ``-1``.

    The default forwards to the historical ``on_epoch_end`` hook. Adapters
    which accumulate validation predictions therefore get a clear boundary
    for the optional initial evaluation as well as every trained epoch.
    """
    self.on_epoch_end(epoch, metrics)

on_fold_end 

on_fold_end(fold: int, metrics: Dict[str, float]) -> None

Handle an optional callback after a cross-validation fold.

Source code in methods/base.py 
230
231
def on_fold_end(self, fold: int, metrics: Dict[str, float]) -> None:
    """Handle an optional callback after a cross-validation fold."""

on_checkpoint_loaded 

on_checkpoint_loaded(
    model: Module, checkpoint_kind: str, fold: int
) -> None

Restore adapter state not represented by model.state_dict().

Source code in methods/base.py 
233
234
235
236
def on_checkpoint_loaded(
    self, model: nn.Module, checkpoint_kind: str, fold: int,
) -> None:
    """Restore adapter state not represented by ``model.state_dict()``."""

MGPathMethod 

MGPathMethod(cfg, device='cuda')

Bases: BaseMethod

Source code in methods/mgpath/adapter.py 
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
def __init__(self, cfg, device="cuda"):
    super().__init__(cfg, device)
    self.is_encoder_extension = self.cfg.get("encoder_extension") is not None
    if self.is_encoder_extension:
        validate_encoder_extension(self.name, self.cfg)
        if self.cfg.get("encoder_extension_strategy") != \
                "paired_feature_context_v1":
            raise ValueError(
                "MGPATH encoder extensions require "
                "encoder_extension_strategy=paired_feature_context_v1")
    elif self.backbone_name != "plip":
        raise ValueError(
            "Non-PLIP MGPATH requires serialized encoder_extension "
            "provenance")
    expected = {
        "upstream_commit": UPSTREAM_COMMIT,
        "mgpath_recipe": "paper_tmlr_2025",
        "n_ctx": 16, "prompt_views": 4, "image_centers": 64,
        "type_gnn": "gat_conv", "ratio_graph": 0.2,
        "ot_epsilon": 0.1, "ot_iterations": 100,
    }
    expected["mgpath_runtime"] = (
        "paired_feature_context_v1" if self.is_encoder_extension else
        "paper_plip_only_5x10x_no_augmentation")
    drifted = {key: (self.cfg.get(key), value)
               for key, value in expected.items()
               if self.cfg.get(key) != value}
    if drifted:
        raise ValueError(f"MGPATH config contradicts its recipe: {drifted}")

interpret_step 

interpret_step(batch, model, target_class=None)

Expose MGPATH's native learned-center aggregation weights.

Source code in methods/mgpath/adapter.py 
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
@torch.no_grad()
def interpret_step(self, batch, model, target_class=None):
    """Expose MGPATH's native learned-center aggregation weights."""
    del target_class  # Center pooling precedes class-specific OT scoring.
    low, low_edges, high, high_edges, _label = self._unpack(
        batch, self.device)
    details = model(
        low, low_edges, high, high_edges, return_details=True)
    description = (
        "Mean native MGPATH patch weight across the 64 learned image "
        "centers, combining raw and graph branches by ratio_graph.")
    return InterpretabilityResult(
        logits=details["logits"],
        evidence=(
            PatchEvidence(
                scores=details["low_patch_attention"],
                score_type="image_center_pooling_attention",
                description=description,
                scale="low",
                feature_path_key="feature_path_column_l",
            ),
            PatchEvidence(
                scores=details["high_patch_attention"],
                score_type="image_center_pooling_attention",
                description=description,
                scale="high",
                feature_path_key="feature_path_column_s",
            ),
        ),
    )

get_backbone_contract classmethod 

get_backbone_contract() -> MethodBackboneContract

Return the encoder/feature contract declared by the adapter.

Raises:

Type Description
NotImplementedError

If the adapter omitted its required contract.
Source code in methods/base.py 
103
104
105
106
107
108
109
110
111
112
113
@classmethod
def get_backbone_contract(cls) -> MethodBackboneContract:
    """Return the encoder/feature contract declared by the adapter.

    Raises:
        NotImplementedError: If the adapter omitted its required contract.
    """
    if cls.backbone_contract is None:
        raise NotImplementedError(
            f"Method adapter {cls.__name__} has no backbone contract")
    return cls.backbone_contract

load_encoder 

load_encoder(
    *,
    weights_path: Optional[str] = None,
    **loader_options: Any,
) -> EncoderBundle

Build and validate the configured encoder in one operation.

Parameters:

Name Type Description Default
weights_path Optional[str]

Optional local checkpoint or model identifier. When omitted, cfg['backbone_weights'] is used.

 None
**loader_options Any

Family-specific options forwarded to the encoder registry, such as a pinned revision or offline-only loading.

 {}

Returns:

Type Description
EncoderBundle

A validated capability-aware encoder bundle.

Raises:

Type Description
RuntimeError

If the adapter has no runtime backbone contract.
BackboneCompatibilityError

If the bundle violates the method contract or declared feature provenance.
Source code in methods/base.py 
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
def load_encoder(self, *, weights_path: Optional[str] = None,
                 **loader_options: Any) -> EncoderBundle:
    """Build and validate the configured encoder in one operation.

    Args:
        weights_path: Optional local checkpoint or model identifier. When
            omitted, ``cfg['backbone_weights']`` is used.
        **loader_options: Family-specific options forwarded to the encoder
            registry, such as a pinned revision or offline-only loading.

    Returns:
        A validated capability-aware encoder bundle.

    Raises:
        RuntimeError: If the adapter has no runtime backbone contract.
        BackboneCompatibilityError: If the bundle violates the method
            contract or declared feature provenance.
    """
    if self.backbone_contract is None or self.backbone_name is None:
        raise RuntimeError(f"Method '{self.name}' has no runtime backbone")
    from common.backbones import build_encoder
    if weights_path is None:
        weights_path = self.cfg.get("backbone_weights")
    bundle = build_encoder(self.backbone_name, weights_path=weights_path,
                           device=self.device, **loader_options)
    return self.backbone_contract.validate_bundle(self.cfg, bundle)

prepare_fold 

prepare_fold(
    fold: int, model: Module, train_loader: Any
) -> None

Prepare training-split-only state after model/data construction.

Methods such as PathPT use this boundary to select prompts without seeing validation or test slides. The default is deliberately a no-op.

Source code in methods/base.py 
204
205
206
207
208
209
210
211
def prepare_fold(
    self, fold: int, model: nn.Module, train_loader: Any,
) -> None:
    """Prepare training-split-only state after model/data construction.

    Methods such as PathPT use this boundary to select prompts without
    seeing validation or test slides. The default is deliberately a no-op.
    """

on_train_epoch_start 

on_train_epoch_start(epoch: int, model: Module) -> None

Handle the boundary immediately before an epoch's first batch.

Source code in methods/base.py 
213
214
def on_train_epoch_start(self, epoch: int, model: nn.Module) -> None:
    """Handle the boundary immediately before an epoch's first batch."""

on_epoch_end 

on_epoch_end(epoch: int, metrics: Dict[str, float]) -> None

Handle an optional callback after a full train/validation epoch.

Source code in methods/base.py 
216
217
def on_epoch_end(self, epoch: int, metrics: Dict[str, float]) -> None:
    """Handle an optional callback after a full train/validation epoch."""

on_validation_end 

on_validation_end(
    epoch: int, metrics: Dict[str, float]
) -> None

Handle a completed validation pass, including epoch -1.

The default forwards to the historical on_epoch_end hook. Adapters which accumulate validation predictions therefore get a clear boundary for the optional initial evaluation as well as every trained epoch.

Source code in methods/base.py 
219
220
221
222
223
224
225
226
227
228
def on_validation_end(
    self, epoch: int, metrics: Dict[str, float],
) -> None:
    """Handle a completed validation pass, including epoch ``-1``.

    The default forwards to the historical ``on_epoch_end`` hook. Adapters
    which accumulate validation predictions therefore get a clear boundary
    for the optional initial evaluation as well as every trained epoch.
    """
    self.on_epoch_end(epoch, metrics)

on_fold_end 

on_fold_end(fold: int, metrics: Dict[str, float]) -> None

Handle an optional callback after a cross-validation fold.

Source code in methods/base.py 
230
231
def on_fold_end(self, fold: int, metrics: Dict[str, float]) -> None:
    """Handle an optional callback after a cross-validation fold."""

on_checkpoint_loaded 

on_checkpoint_loaded(
    model: Module, checkpoint_kind: str, fold: int
) -> None

Restore adapter state not represented by model.state_dict().

Source code in methods/base.py 
233
234
235
236
def on_checkpoint_loaded(
    self, model: nn.Module, checkpoint_kind: str, fold: int,
) -> None:
    """Restore adapter state not represented by ``model.state_dict()``."""

HIPSSMethod 

HIPSSMethod(cfg, device='cuda')

Bases: BaseMethod

Source code in methods/hipss/adapter.py 
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
def __init__(self, cfg, device="cuda"):
    super().__init__(cfg, device)
    self.is_encoder_extension = self.cfg.get("encoder_extension") is not None
    if self.is_encoder_extension:
        validate_encoder_extension(self.name, self.cfg)
        if self.cfg.get("encoder_extension_strategy") != \
                "paired_feature_affine_ssf_v1":
            raise ValueError(
                "HIPSS encoder extensions require "
                "encoder_extension_strategy=paired_feature_affine_ssf_v1")
    elif self.backbone_name != "conch":
        raise ValueError(
            "Non-CONCH HIPSS requires serialized encoder_extension provenance")
    expected_layers = 8 if self.cfg.get("task") == "ubc_ocean" else 2
    expected = {
        "upstream_commit": UPSTREAM_COMMIT,
        "prompt_generator": GENERATOR,
        "ssf_layers": expected_layers,
        "ssf_mode": (
            "final_feature_affine_v1" if self.is_encoder_extension else
            "layerwise_text_transformer_reimplementation"),
        "text_guidance_mode": "paper_binary_class_1_multiclass_mean",
        "region_span_level0": 4096,
        "region_encoder_lambda": 10.0,
        "region_encoder_alpha": 0.2,
        "wsi_encoder_lambda": 10.0,
        "wsi_encoder_alpha": 0.2,
    }
    drifted = {key: (self.cfg.get(key), value)
               for key, value in expected.items()
               if self.cfg.get(key) != value}
    if drifted:
        raise ValueError(f"HIPSS config contradicts its recipe: {drifted}")

interpret_step 

interpret_step(batch, model, target_class=None)

Expose HIPSS's native hierarchical patch contribution weights.

Source code in methods/hipss/adapter.py 
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
@torch.no_grad()
def interpret_step(self, batch, model, target_class=None):
    """Expose HIPSS's native hierarchical patch contribution weights."""
    del target_class  # HIPSS guidance is fixed by its released task rule.
    regions, mask, _label = self._unpack(batch, self.device)
    if self.is_encoder_extension:
        regions = project_paired_features(model, regions)
    details = model(regions, mask, return_details=True)
    combined = (
        details["patch_attention"]
        * details["region_attention"].unsqueeze(-1))
    scores = combined[details["mask"]]
    return InterpretabilityResult(
        logits=details["logits"],
        evidence=(PatchEvidence(
            scores=scores,
            score_type="hierarchical_pooling_contribution",
            description=(
                "Product of native HIPSS within-region patch attention "
                "and region-to-slide attention."),
            feature_path_key="feature_path_column",
            index_order="hipss_regions",
        ),),
    )

get_backbone_contract classmethod 

get_backbone_contract() -> MethodBackboneContract

Return the encoder/feature contract declared by the adapter.

Raises:

Type Description
NotImplementedError

If the adapter omitted its required contract.
Source code in methods/base.py 
103
104
105
106
107
108
109
110
111
112
113
@classmethod
def get_backbone_contract(cls) -> MethodBackboneContract:
    """Return the encoder/feature contract declared by the adapter.

    Raises:
        NotImplementedError: If the adapter omitted its required contract.
    """
    if cls.backbone_contract is None:
        raise NotImplementedError(
            f"Method adapter {cls.__name__} has no backbone contract")
    return cls.backbone_contract

load_encoder 

load_encoder(
    *,
    weights_path: Optional[str] = None,
    **loader_options: Any,
) -> EncoderBundle

Build and validate the configured encoder in one operation.

Parameters:

Name Type Description Default
weights_path Optional[str]

Optional local checkpoint or model identifier. When omitted, cfg['backbone_weights'] is used.

 None
**loader_options Any

Family-specific options forwarded to the encoder registry, such as a pinned revision or offline-only loading.

 {}

Returns:

Type Description
EncoderBundle

A validated capability-aware encoder bundle.

Raises:

Type Description
RuntimeError

If the adapter has no runtime backbone contract.
BackboneCompatibilityError

If the bundle violates the method contract or declared feature provenance.
Source code in methods/base.py 
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
def load_encoder(self, *, weights_path: Optional[str] = None,
                 **loader_options: Any) -> EncoderBundle:
    """Build and validate the configured encoder in one operation.

    Args:
        weights_path: Optional local checkpoint or model identifier. When
            omitted, ``cfg['backbone_weights']`` is used.
        **loader_options: Family-specific options forwarded to the encoder
            registry, such as a pinned revision or offline-only loading.

    Returns:
        A validated capability-aware encoder bundle.

    Raises:
        RuntimeError: If the adapter has no runtime backbone contract.
        BackboneCompatibilityError: If the bundle violates the method
            contract or declared feature provenance.
    """
    if self.backbone_contract is None or self.backbone_name is None:
        raise RuntimeError(f"Method '{self.name}' has no runtime backbone")
    from common.backbones import build_encoder
    if weights_path is None:
        weights_path = self.cfg.get("backbone_weights")
    bundle = build_encoder(self.backbone_name, weights_path=weights_path,
                           device=self.device, **loader_options)
    return self.backbone_contract.validate_bundle(self.cfg, bundle)

build_optimizer 

build_optimizer(model: Module) -> Optimizer

Build Adam over trainable parameters using the configured recipe.

Source code in methods/base.py 
172
173
174
175
176
177
178
179
def build_optimizer(self, model: nn.Module) -> torch.optim.Optimizer:
    """Build Adam over trainable parameters using the configured recipe."""
    params = filter(lambda p: p.requires_grad, model.parameters())
    return torch.optim.Adam(
        params,
        lr=self.cfg.get("lr", 1e-4),
        weight_decay=self.cfg.get("weight_decay", 1e-5),
    )

prepare_fold 

prepare_fold(
    fold: int, model: Module, train_loader: Any
) -> None

Prepare training-split-only state after model/data construction.

Methods such as PathPT use this boundary to select prompts without seeing validation or test slides. The default is deliberately a no-op.

Source code in methods/base.py 
204
205
206
207
208
209
210
211
def prepare_fold(
    self, fold: int, model: nn.Module, train_loader: Any,
) -> None:
    """Prepare training-split-only state after model/data construction.

    Methods such as PathPT use this boundary to select prompts without
    seeing validation or test slides. The default is deliberately a no-op.
    """

on_train_epoch_start 

on_train_epoch_start(epoch: int, model: Module) -> None

Handle the boundary immediately before an epoch's first batch.

Source code in methods/base.py 
213
214
def on_train_epoch_start(self, epoch: int, model: nn.Module) -> None:
    """Handle the boundary immediately before an epoch's first batch."""

on_epoch_end 

on_epoch_end(epoch: int, metrics: Dict[str, float]) -> None

Handle an optional callback after a full train/validation epoch.

Source code in methods/base.py 
216
217
def on_epoch_end(self, epoch: int, metrics: Dict[str, float]) -> None:
    """Handle an optional callback after a full train/validation epoch."""

on_validation_end 

on_validation_end(
    epoch: int, metrics: Dict[str, float]
) -> None

Handle a completed validation pass, including epoch -1.

The default forwards to the historical on_epoch_end hook. Adapters which accumulate validation predictions therefore get a clear boundary for the optional initial evaluation as well as every trained epoch.

Source code in methods/base.py 
219
220
221
222
223
224
225
226
227
228
def on_validation_end(
    self, epoch: int, metrics: Dict[str, float],
) -> None:
    """Handle a completed validation pass, including epoch ``-1``.

    The default forwards to the historical ``on_epoch_end`` hook. Adapters
    which accumulate validation predictions therefore get a clear boundary
    for the optional initial evaluation as well as every trained epoch.
    """
    self.on_epoch_end(epoch, metrics)

on_fold_end 

on_fold_end(fold: int, metrics: Dict[str, float]) -> None

Handle an optional callback after a cross-validation fold.

Source code in methods/base.py 
230
231
def on_fold_end(self, fold: int, metrics: Dict[str, float]) -> None:
    """Handle an optional callback after a cross-validation fold."""

on_checkpoint_loaded 

on_checkpoint_loaded(
    model: Module, checkpoint_kind: str, fold: int
) -> None

Restore adapter state not represented by model.state_dict().

Source code in methods/base.py 
233
234
235
236
def on_checkpoint_loaded(
    self, model: nn.Module, checkpoint_kind: str, fold: int,
) -> None:
    """Restore adapter state not represented by ``model.state_dict()``."""