Skip to content

Evaluation

Zero-shot / recovery evaluation, baselines, and ensembling.

polyweave.evaluation

Evaluation: zero-shot accuracy, recovery curves, baselines.

The read-side companion to :mod:polyweave.training. Given a trained teacher, these measure the weights it generates — without fine-tuning (zero-shot) and as a fine-tuning initialisation (recovery) — and provide the non-hypernetwork baselines (random, nearest-class-centroid) the teachers are compared against.

centroids_to_fc

centroids_to_fc(centroids: Tensor) -> Dict[str, torch.Tensor]

Turn class centroids into an equivalent linear head {"weight","bias"}.

For centroid c_k, the score -½‖x - c_k‖² ranks classes identically to nearest-centroid; dropping the shared -½‖x‖² term leaves the linear form wₖ·x + bₖ with wₖ = c_k and bₖ = -½‖c_k‖².

Source code in polyweave/evaluation/baselines.py
def centroids_to_fc(centroids: torch.Tensor) -> Dict[str, torch.Tensor]:
    """Turn class centroids into an equivalent linear head ``{"weight","bias"}``.

    For centroid ``c_k``, the score ``-½‖x - c_k‖²`` ranks classes identically to
    nearest-centroid; dropping the shared ``-½‖x‖²`` term leaves the linear form
    ``wₖ·x + bₖ`` with ``wₖ = c_k`` and ``bₖ = -½‖c_k‖²``.
    """
    weight = centroids.clone()
    bias = -0.5 * (centroids ** 2).sum(dim=1)
    return {"weight": weight, "bias": bias}

class_centroids

class_centroids(feats: Tensor, y: Tensor, num_classes: int) -> torch.Tensor

Mean feature vector per class, [num_classes, feature_dim].

Empty classes get a zero centroid (they contribute nothing discriminative).

Source code in polyweave/evaluation/baselines.py
def class_centroids(feats: torch.Tensor, y: torch.Tensor, num_classes: int) -> torch.Tensor:
    """Mean feature vector per class, ``[num_classes, feature_dim]``.

    Empty classes get a zero centroid (they contribute nothing discriminative).
    """
    F = feats.shape[1]
    centroids = torch.zeros(num_classes, F, device=feats.device, dtype=feats.dtype)
    counts = torch.zeros(num_classes, device=feats.device)
    centroids.index_add_(0, y, feats)
    counts.index_add_(0, y, torch.ones_like(y, dtype=feats.dtype))
    nonempty = counts > 0
    centroids[nonempty] /= counts[nonempty].unsqueeze(1)
    return centroids

ncc_fc

ncc_fc(feats: Tensor, y: Tensor, num_classes: int) -> Dict[str, torch.Tensor]

Convenience: build the NCC linear head directly from features and labels.

Source code in polyweave/evaluation/baselines.py
def ncc_fc(feats: torch.Tensor, y: torch.Tensor, num_classes: int) -> Dict[str, torch.Tensor]:
    """Convenience: build the NCC linear head directly from features and labels."""
    return centroids_to_fc(class_centroids(feats, y, num_classes))

random_like

random_like(reference: WeightStruct, *, scale: float = 1.0, zero_bias: bool = True, fan_in_scale: bool = True) -> WeightStruct

Random weights matching the shapes of a generated structure.

Weight-like tensors (ndim >= 2) are filled with random normal values; 1-D tensors are treated as biases and zeroed when zero_bias (the experiments' random baseline uses random weights with zero bias). Recurses over a list of dicts for the per-layer Q/K structure.

By default (fan_in_scale=True) each weight tensor is scaled to a Kaiming-linear standard deviation 1/sqrt(fan_in) — matching nn.init.kaiming_normal_(..., nonlinearity="linear") — so the random baseline is a well-conditioned initialisation rather than a saturating one. fan_in is prod(shape[1:]) (input features, times kernel area for convolutions). scale multiplies on top of this. Passing fan_in_scale=False recovers the raw scale * randn behaviour.

Note: a unit-variance random head (fan_in_scale=False, scale=1) produces logits ~sqrt(fan_in) too large, saturating the softmax and stalling fine-tuning — which is why fan-in scaling is the default for the baseline.

Source code in polyweave/evaluation/baselines.py
def random_like(
    reference: WeightStruct,
    *,
    scale: float = 1.0,
    zero_bias: bool = True,
    fan_in_scale: bool = True,
) -> WeightStruct:
    """Random weights matching the shapes of a generated structure.

    Weight-like tensors (``ndim >= 2``) are filled with random normal values; 1-D
    tensors are treated as biases and zeroed when ``zero_bias`` (the experiments'
    random baseline uses random weights with zero bias). Recurses over a list of
    dicts for the per-layer Q/K structure.

    By default (``fan_in_scale=True``) each weight tensor is scaled to a
    Kaiming-linear standard deviation ``1/sqrt(fan_in)`` — matching
    ``nn.init.kaiming_normal_(..., nonlinearity="linear")`` — so the random
    baseline is a *well-conditioned* initialisation rather than a saturating one.
    ``fan_in`` is ``prod(shape[1:])`` (input features, times kernel area for
    convolutions). ``scale`` multiplies on top of this. Passing
    ``fan_in_scale=False`` recovers the raw ``scale * randn`` behaviour.

    Note: a unit-variance random head (``fan_in_scale=False, scale=1``) produces
    logits ~``sqrt(fan_in)`` too large, saturating the softmax and stalling
    fine-tuning — which is why fan-in scaling is the default for the baseline.
    """
    if isinstance(reference, dict):
        out: Dict[str, torch.Tensor] = {}
        for k, v in reference.items():
            if v.ndim >= 2:
                std = scale
                if fan_in_scale:
                    fan_in = int(v[0].numel())  # prod(shape[1:])
                    std = scale / (fan_in ** 0.5)
                out[k] = std * torch.randn_like(v)
            else:
                out[k] = torch.zeros_like(v) if zero_bias else scale * torch.randn_like(v)
        return out
    if isinstance(reference, list):
        return [
            random_like(r, scale=scale, zero_bias=zero_bias, fan_in_scale=fan_in_scale)
            for r in reference
        ]
    raise TypeError(f"cannot build random_like for type {type(reference).__name__}")

accuracy_from_probs

accuracy_from_probs(probs: Tensor, labels: Tensor) -> float

Top-1 accuracy of a [N, C] probability/logit tensor against [N] labels.

Source code in polyweave/evaluation/ensemble.py
def accuracy_from_probs(probs: torch.Tensor, labels: torch.Tensor) -> float:
    """Top-1 accuracy of a ``[N, C]`` probability/logit tensor against ``[N]`` labels."""
    pred = probs.argmax(dim=-1).reshape(-1)
    labels = labels.reshape(-1)
    return (pred == labels).float().mean().item()

disagreement_matrix

disagreement_matrix(probs: Tensor) -> torch.Tensor

Symmetric [M, M] matrix of pairwise prediction-disagreement rates.

Entry (i, j) is the fraction of examples where members i and j differ; the diagonal is 0. Useful for a heatmap of which members are redundant vs complementary.

Source code in polyweave/evaluation/ensemble.py
def disagreement_matrix(probs: torch.Tensor) -> torch.Tensor:
    """Symmetric ``[M, M]`` matrix of pairwise prediction-disagreement rates.

    Entry ``(i, j)`` is the fraction of examples where members ``i`` and ``j``
    differ; the diagonal is 0. Useful for a heatmap of which members are
    redundant vs complementary.
    """
    preds = probs.argmax(dim=-1)  # [M, N]
    M = preds.shape[0]
    out = probs.new_zeros(M, M)
    for i in range(M):
        for j in range(i + 1, M):
            d = (preds[i] != preds[j]).float().mean()
            out[i, j] = d
            out[j, i] = d
    return out

ensemble_accuracy

ensemble_accuracy(probs: Tensor, labels: Tensor) -> float

Top-1 accuracy of the mean soft vote over members.

Source code in polyweave/evaluation/ensemble.py
def ensemble_accuracy(probs: torch.Tensor, labels: torch.Tensor) -> float:
    """Top-1 accuracy of the mean soft vote over members."""
    return accuracy_from_probs(ensemble_probs(probs), labels)

ensemble_gain

ensemble_gain(probs: Tensor, labels: Tensor) -> float

Ensemble accuracy minus mean single-member accuracy (the diversity payoff).

Source code in polyweave/evaluation/ensemble.py
def ensemble_gain(probs: torch.Tensor, labels: torch.Tensor) -> float:
    """Ensemble accuracy minus mean single-member accuracy (the diversity payoff)."""
    return ensemble_accuracy(probs, labels) - member_accuracies(probs, labels).mean().item()

ensemble_probs

ensemble_probs(probs: Tensor) -> torch.Tensor

Mean soft vote over members: [M, N, C] -> [N, C].

Source code in polyweave/evaluation/ensemble.py
def ensemble_probs(probs: torch.Tensor) -> torch.Tensor:
    """Mean soft vote over members: ``[M, N, C] -> [N, C]``."""
    if probs.ndim != 3:
        raise ValueError(f"expected [M, N, C], got shape {tuple(probs.shape)}")
    return probs.mean(dim=0)

member_accuracies

member_accuracies(probs: Tensor, labels: Tensor) -> torch.Tensor

Per-member top-1 accuracy: [M, N, C], [N] -> [M].

Source code in polyweave/evaluation/ensemble.py
def member_accuracies(probs: torch.Tensor, labels: torch.Tensor) -> torch.Tensor:
    """Per-member top-1 accuracy: ``[M, N, C], [N] -> [M]``."""
    preds = probs.argmax(dim=-1)  # [M, N]
    labels = labels.reshape(1, -1)
    return (preds == labels).float().mean(dim=1)

pairwise_disagreement

pairwise_disagreement(probs: Tensor) -> float

Mean over distinct member pairs of the fraction of examples they disagree on.

probs is [M, N, C]; predictions are taken as the per-member argmax. Returns a scalar in [0, 1]: 0 when every member predicts identically, higher when members make different predictions (error diversity). Undefined for a single member (returns 0.0).

Source code in polyweave/evaluation/ensemble.py
def pairwise_disagreement(probs: torch.Tensor) -> float:
    """Mean over distinct member pairs of the fraction of examples they disagree on.

    ``probs`` is ``[M, N, C]``; predictions are taken as the per-member argmax.
    Returns a scalar in ``[0, 1]``: 0 when every member predicts identically,
    higher when members make different predictions (error diversity). Undefined
    for a single member (returns 0.0).
    """
    if probs.ndim != 3:
        raise ValueError(f"expected [M, N, C], got shape {tuple(probs.shape)}")
    preds = probs.argmax(dim=-1)  # [M, N]
    M = preds.shape[0]
    if M < 2:
        return 0.0
    total = 0.0
    pairs = 0
    for i in range(M):
        for j in range(i + 1, M):
            total += (preds[i] != preds[j]).float().mean().item()
            pairs += 1
    return total / pairs

average_weights

average_weights(structures: Sequence[WeightStruct]) -> WeightStruct

Element-wise mean of a list of identically-structured weight pytrees.

Handles a dict of tensors and a list of such dicts (the two shapes the teachers emit). Raises on anything else.

Source code in polyweave/evaluation/loops.py
def average_weights(structures: Sequence[WeightStruct]) -> WeightStruct:
    """Element-wise mean of a list of identically-structured weight pytrees.

    Handles a ``dict`` of tensors and a ``list`` of such dicts (the two shapes
    the teachers emit). Raises on anything else.
    """
    if not structures:
        raise ValueError("need at least one structure to average")
    first = structures[0]
    if isinstance(first, dict):
        return {k: torch.stack([s[k] for s in structures], 0).mean(0) for k in first}
    if isinstance(first, list):
        return [average_weights([s[i] for s in structures]) for i in range(len(first))]
    raise TypeError(f"cannot average weight structure of type {type(first).__name__}")

evaluate_accuracy

evaluate_accuracy(student: Module, eval_batches: Iterable[Batch], gen: Any, forward: Forward) -> float

Top-1 accuracy of student over eval_batches using weights gen.

gen is passed straight to forward; pass the generated structure for zero-shot evaluation, or None to evaluate the student's own (e.g. already installed / recovered) weights.

Source code in polyweave/evaluation/loops.py
@torch.no_grad()
def evaluate_accuracy(
    student: nn.Module,
    eval_batches: Iterable[Batch],
    gen: Any,
    forward: Forward,
) -> float:
    """Top-1 accuracy of ``student`` over ``eval_batches`` using weights ``gen``.

    ``gen`` is passed straight to ``forward``; pass the generated structure for
    zero-shot evaluation, or ``None`` to evaluate the student's own (e.g. already
    installed / recovered) weights.
    """
    student.eval()
    correct = total = 0
    for batch in eval_batches:
        logits, target = forward(student, batch, gen)
        correct += (logits.argmax(1) == target).sum().item()
        total += target.numel()
    return correct / max(total, 1)

evaluate_macro_f1

evaluate_macro_f1(student: Module, eval_batches: Iterable[Batch], gen: Any, forward: Forward, num_classes: int) -> float

Macro-averaged F1 of student over eval_batches using weights gen.

The balanced complement to :func:evaluate_accuracy: per-class F1 averaged with equal weight, so a method cannot score well by ignoring minority classes. On the (near-)balanced tasks in this paper macro-F1 tracks accuracy closely; it is provided so a re-run can report both at no extra forward-pass cost. Classes that are neither present nor predicted are excluded from the mean.

Source code in polyweave/evaluation/loops.py
@torch.no_grad()
def evaluate_macro_f1(
    student: nn.Module,
    eval_batches: Iterable[Batch],
    gen: Any,
    forward: Forward,
    num_classes: int,
) -> float:
    """Macro-averaged F1 of ``student`` over ``eval_batches`` using weights ``gen``.

    The balanced complement to :func:`evaluate_accuracy`: per-class F1 averaged
    with equal weight, so a method cannot score well by ignoring minority classes.
    On the (near-)balanced tasks in this paper macro-F1 tracks accuracy closely;
    it is provided so a re-run can report both at no extra forward-pass cost.
    Classes that are neither present nor predicted are excluded from the mean.
    """
    student.eval()
    tp = fp = fn = None
    for batch in eval_batches:
        logits, target = forward(student, batch, gen)
        if tp is None:
            tp = torch.zeros(num_classes, device=logits.device)
            fp = torch.zeros(num_classes, device=logits.device)
            fn = torch.zeros(num_classes, device=logits.device)
        pred = logits.argmax(1).reshape(-1)
        target = target.reshape(-1)
        for c in range(num_classes):
            pc = pred == c
            tc = target == c
            tp[c] += (pc & tc).sum()
            fp[c] += (pc & ~tc).sum()
            fn[c] += (~pc & tc).sum()
    if tp is None:
        return 0.0
    precision = tp / (tp + fp).clamp(min=1)
    recall = tp / (tp + fn).clamp(min=1)
    f1 = 2 * precision * recall / (precision + recall).clamp(min=1e-12)
    present = (tp + fn) > 0
    return f1[present].mean().item() if bool(present.any()) else 0.0

generate_averaged

generate_averaged(teacher: Module, student: Module, support_batches: Iterable[Batch], build_prototype: Callable[[Module, Batch], Tensor]) -> WeightStruct

Generate weights for student from each support batch and average them.

Averaging over a handful of support batches reduces the variance of the prototype statistics, matching the experiments' eval_support_batches.

Source code in polyweave/evaluation/loops.py
@torch.no_grad()
def generate_averaged(
    teacher: nn.Module,
    student: nn.Module,
    support_batches: Iterable[Batch],
    build_prototype: Callable[[nn.Module, Batch], torch.Tensor],
) -> WeightStruct:
    """Generate weights for ``student`` from each support batch and average them.

    Averaging over a handful of support batches reduces the variance of the
    prototype statistics, matching the experiments' ``eval_support_batches``.
    """
    teacher.eval()
    structures = [teacher(build_prototype(student, b)) for b in support_batches]
    return average_weights(structures)

mean_curves

mean_curves(curves: Sequence[Sequence[Tuple[int, float]]]) -> List[Tuple[int, float]]

Average several recovery curves that share the same step grid.

Source code in polyweave/evaluation/loops.py
def mean_curves(curves: Sequence[Sequence[Tuple[int, float]]]) -> List[Tuple[int, float]]:
    """Average several recovery curves that share the same step grid."""
    if not curves:
        raise ValueError("no curves to average")
    steps = [s for s, _ in curves[0]]
    n = len(curves)
    means = [sum(c[i][1] for c in curves) / n for i in range(len(steps))]
    return list(zip(steps, means))

recovery_curve

recovery_curve(student: Module, *, init: RecoveryInit, sample_batch: Callable[[], Batch], forward: Forward, eval_fn: Callable[[Module], float], steps: int, lr: float = 0.001, eval_every: int = 20, grad_mask: Optional[Callable[[Module], None]] = None) -> List[Tuple[int, float]]

Fine-tune a generated initialisation and record the accuracy curve.

Parameters:

Name Type Description Default
student Module

the student to recover (modified in place).

required
init RecoveryInit

installs the initial weights into student and returns the list of parameters to optimise (e.g. just the target layer). Any BN reset belongs here too.

required
sample_batch Callable[[], Batch]

fresh training batch each step.

required
forward Forward

(student, batch, None) -> (logits, target) — called with gen=None so the student uses its own (installed) weights.

required
eval_fn Callable[[Module], float]

maps student to a scalar accuracy (evaluated at step 0 and every eval_every steps, and always at the final step).

required
steps int

fine-tuning budget.

required
lr float

Adam learning rate.

0.001
eval_every int

evaluation cadence in steps.

20
grad_mask Optional[Callable[[Module], None]]

optional hook run after backward and before opt.step to zero gradients on parameters that must stay frozen (e.g. the value projection when only Q/K is being recovered).

None

Returns:

Type Description
List[Tuple[int, float]]

A list of (step, accuracy) pairs, starting at (0, zero-shot acc).

Source code in polyweave/evaluation/loops.py
def recovery_curve(
    student: nn.Module,
    *,
    init: RecoveryInit,
    sample_batch: Callable[[], Batch],
    forward: Forward,
    eval_fn: Callable[[nn.Module], float],
    steps: int,
    lr: float = 1e-3,
    eval_every: int = 20,
    grad_mask: Optional[Callable[[nn.Module], None]] = None,
) -> List[Tuple[int, float]]:
    """Fine-tune a generated initialisation and record the accuracy curve.

    Args:
        student: the student to recover (modified in place).
        init: installs the initial weights into ``student`` and returns the list
            of parameters to optimise (e.g. just the target layer). Any BN reset
            belongs here too.
        sample_batch: fresh training batch each step.
        forward: ``(student, batch, None) -> (logits, target)`` — called with
            ``gen=None`` so the student uses its own (installed) weights.
        eval_fn: maps ``student`` to a scalar accuracy (evaluated at step 0 and
            every ``eval_every`` steps, and always at the final step).
        steps: fine-tuning budget.
        lr: Adam learning rate.
        eval_every: evaluation cadence in steps.
        grad_mask: optional hook run after ``backward`` and before ``opt.step``
            to zero gradients on parameters that must stay frozen (e.g. the value
            projection when only Q/K is being recovered).

    Returns:
        A list of ``(step, accuracy)`` pairs, starting at ``(0, zero-shot acc)``.
    """
    params = list(init(student))
    opt = torch.optim.Adam(params, lr=lr)

    curve: List[Tuple[int, float]] = [(0, eval_fn(student))]
    for step in range(1, steps + 1):
        student.train()
        batch = sample_batch()
        logits, target = forward(student, batch, None)
        loss = F.cross_entropy(logits, target)
        opt.zero_grad()
        loss.backward()
        if grad_mask is not None:
            grad_mask(student)
        opt.step()

        if step % eval_every == 0 or step == steps:
            curve.append((step, eval_fn(student)))
    return curve

reset_bn_stats

reset_bn_stats(model: Module, batches: Iterable[Batch], run: Callable[[Module, Batch], Any], *, max_batches: int = 10) -> None

Re-estimate BatchNorm running statistics after a weight swap.

Resets the running mean/var of every _BatchNorm in model and re-runs run(model, batch) (a forward pass) in train mode over up to max_batches batches. Restores eval mode afterwards.

Source code in polyweave/evaluation/loops.py
@torch.no_grad()
def reset_bn_stats(
    model: nn.Module,
    batches: Iterable[Batch],
    run: Callable[[nn.Module, Batch], Any],
    *,
    max_batches: int = 10,
) -> None:
    """Re-estimate BatchNorm running statistics after a weight swap.

    Resets the running mean/var of every ``_BatchNorm`` in ``model`` and re-runs
    ``run(model, batch)`` (a forward pass) in train mode over up to
    ``max_batches`` batches. Restores ``eval`` mode afterwards.
    """
    for m in model.modules():
        if isinstance(m, _BatchNorm):
            m.reset_running_stats()
    model.train()
    for i, batch in enumerate(batches):
        if i >= max_batches:
            break
        run(model, batch)
    model.eval()