2 · 同一表示，改变标签预算或读出 · 实现代码

def make_source_cnn():
    return nn.Sequential(OrderedDict([
      ('conv1',nn.Conv2d(1,8,3,padding=1)),('relu1',nn.ReLU()),('pool1',nn.MaxPool2d(2)),
      ('conv2',nn.Conv2d(8,16,3,padding=1)),('relu2',nn.ReLU()),('pool2',nn.MaxPool2d(2)),
      ('flatten',nn.Flatten()),('hidden',nn.Linear(784,32)),('relu3',nn.ReLU()),('logits',nn.Linear(32,10))]))
model=make_source_cnn()
model.load_state_dict(torch.load(ROOT/'assets/cnn-plain-state.pt',map_location='cpu',weights_only=True))
assert sum(p.numel() for p in model.parameters())==26698
model.eval()
for param in model.parameters(): param.requires_grad_(False)
before={k:v.detach().clone() for k,v in model.state_dict().items()}
encoder=nn.Sequential(OrderedDict(list(model.named_children())[:-1]))
with torch.no_grad(): raw_features=encoder(x).numpy()
features=raw_features/np.maximum(np.linalg.norm(raw_features,axis=1,keepdims=True),1e-12)
assert features.shape==(640,32)
assert all(torch.equal(v,before[k]) for k,v in model.state_dict().items())
np.savez_compressed(OUT/'frozen-features.npz',features=features,raw_features=raw_features,sample_ids=sample_ids,split=split)
print('Frozen features:',features.shape,'; all parameters unchanged.')

LABEL_BUDGETS=[32,64,128,256]
K=5
C=10.0
records=[]; fitted={}
for budget in LABEL_BUDGETS:
    labeled=train[:budget]
    # Identical labeled sample identities for both readouts at this budget.
    methods={'kNN':KNeighborsClassifier(n_neighbors=K,weights='uniform',metric='euclidean'),
             'linear':LogisticRegression(C=C,max_iter=2000,solver='lbfgs',random_state=20260918)}
    for method,readout in methods.items():
        readout.fit(features[labeled],y[labeled])
        fitted[(method,budget)]=readout
        records.append({'method':method,'budget':budget,'labeled_ids':sample_ids[labeled].tolist(),
                        'train_accuracy':float(readout.score(features[labeled],y[labeled])),
                        'validation_accuracy':float(readout.score(features[val],y[val]))})
# The methods and hyperparameters are now locked. Held-out labels enter evaluation only.
for record in records:
    pred=fitted[(record['method'],record['budget'])].predict(features[heldout])
    record['heldout_accuracy']=float(accuracy_score(y[heldout],pred))
    record['heldout_predictions']=pred.tolist()
    record['heldout_errors']=int((pred!=y[heldout]).sum())
assert all(torch.equal(v,before[k]) for k,v in model.state_dict().items())
readout_results={'evidence':'actual local CPU run','protocol':{'target_task':'digit >= 5','shift_right':2,
 'feature':'last 32-dimensional ReLU, then per-sample L2 normalization','label_budgets':LABEL_BUDGETS,
 'k':K,'C':C,'tuning':'none; fixed before evaluation','seed':20260918,'encoder_sha256':provenance['source_checkpoint_sha256']},
 'records':records,'limitations':['one fixed split; no seed-sweep uncertainty','not a supervised-vs-SSL comparison','MNIST source training differs from natural-image pretraining']}
(OUT/'readout-results.json').write_text(json.dumps(readout_results,indent=2))
print('method | labels | train | validation | heldout | heldout errors / 256')
for r in records: print(r['method'],r['budget'],*[round(r[k],4) for k in ['train_accuracy','validation_accuracy','heldout_accuracy']],r['heldout_errors'])
fig,axes=plt.subplots(1,2,figsize=(10,3.8),sharey=True)
for method,color in [('kNN',BLUE),('linear',GREEN)]:
    rows=[r for r in records if r['method']==method]
    for ax,metric in zip(axes,['validation_accuracy','heldout_accuracy']):
        ax.plot([r['budget'] for r in rows],[r[metric] for r in rows],marker='o',color=color,label=method)
for ax,title in zip(axes,['Validation (128 images)','Held-out evaluation (256 images)']):
    ax.set(xlabel='Labeled training images',ylabel='Binary accuracy',ylim=(0,1.01),title=title,xticks=LABEL_BUDGETS);ax.legend();ax.grid(alpha=.18)
savefig('03-readout-budget.png')