Improving Prediction Reliability in Music Models

# Illustrative sketch (not from the paper)
import torch
import torch.nn as nn
import torch.optim as optim

# 1. Freeze a pre‑trained base classifier
base_model = ...  # loaded elsewhere
for p in base_model.parameters():
    p.requires_grad = False

# 2. Attach a lightweight confidence head to the penultimate layer
class ConfHead(nn.Module):
    def __init__(self, in_dim):
        super().__init__()
        self.fc = nn.Linear(in_dim, 1)  # scalar confidence output
    def forward(self, x):
        return torch.sigmoid(self.fc(x))

penultimate = base_model.penultimate  # placeholder for the penultimate module
conf_head = ConfHead(penultimate.out_features)
optimizer = optim.Adam(conf_head.parameters(), lr=1e-3)

# 3. TCP‑α target: high for correct, margin‑controlled lower for errors
def tcp_alpha_target(logits, labels, margin=0.5):
    correct = logits.argmax(dim=1) == labels
    target = torch.ones(logits.size(0), device=logits.device)
    # penalise mis‑classifications with a margin
    target[~correct] = torch.clamp(logits.max(dim=1)[0][~correct] - margin, min=0.0)
    return target

# 4. Imbalance‑aware training helpers
error_buffer = []
MAX_BUF = 1024

for epoch in range(num_epochs):
    for inputs, labels in stratified_loader:  # stratified mini‑batches
        # forward through frozen base (no grad)
        with torch.no_grad():
            pen_feat = base_model.penultimate(inputs)
            logits = base_model.classifier(pen_feat)
        # confidence prediction
        conf = conf_head(pen_feat).squeeze()
        # compute TCP‑α targets
        target = tcp_alpha_target(logits, labels)
        loss = nn.MSELoss()(conf, target)
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()
        # store error samples in a circular buffer
        err_idx = (logits.argmax(dim=1) != labels).nonzero(as_tuple=True)[0]
        error_buffer.extend(pen_feat[err_idx].cpu().tolist())
        if len(error_buffer) > MAX_BUF:
            error_buffer = error_buffer[-MAX_BUF:]

// Illustrative sketch (not from the paper)
const tf = require('@tensorflow/tfjs-node');

// 1. Freeze a pre‑trained base classifier (weights are not trainable)
const baseModel = loadBaseModel(); // placeholder loader
baseModel.trainable = false;

// 2. Attach a lightweight confidence head to the penultimate layer
function createConfHead(penultimateSize) {
  const input = tf.input({shape: [penultimateSize]});
  const out = tf.layers.dense({units: 1, activation: 'sigmoid'}).apply(input);
  return tf.model({inputs: input, outputs: out});
}
const penultimate = baseModel.getLayer('penultimate'); // placeholder name
const confHead = createConfHead(penultimate.outputShape[1]);
const optimizer = tf.train.adam(1e-3);

// 3. TCP‑α target computation (margin‑controlled for errors)
function tcpAlphaTarget(logits, labels, margin = 0.5) {
  const preds = logits.argMax(-1);
  const correctMask = preds.equal(labels);
  const high = tf.onesLike(preds, 'float32');
  const low = logits.max(-1).sub(margin).relu();
  return tf.where(correctMask, high, low);
}

// 4. Imbalance‑aware training helpers
let errorBuffer = [];
const MAX_BUF = 1024;

async function train(numEpochs, stratifiedDataset) {
  for (let epoch = 0; epoch < numEpochs; epoch++) {
    for await (const batch of stratifiedDataset) { // yields {xs, ys}
      const {xs, ys} = batch;
      // forward through frozen base (no gradient tracking)
      const penFeat = tf.tidy(() => penultimate.apply(baseModel.apply(xs, {training: false})));
      const logits = tf.tidy(() => baseModel.getLayer('classifier').apply(penFeat));
      // confidence prediction
      const conf = confHead.predict(penFeat).squeeze();
      // compute TCP‑α targets
      const target = tcpAlphaTarget(logits, ys);
      // loss = MSE(conf, target)
      const lossFn = () => tf.losses.meanSquaredError(target, conf);
      // back‑prop on confidence head only
      optimizer.minimize(lossFn, false, confHead.trainableWeights);
      // update circular error buffer
      const errMask = logits.argMax(-1).notEqual(ys);
      const errSamples = tf.booleanMaskAsync(penFeat, errMask);
      errSamples.then(buf => {
        errorBuffer.push(...buf.arraySync());
        if (errorBuffer.length > MAX_BUF) errorBuffer = errorBuffer.slice(-MAX_BUF);
      });
    }
  }
}