简介:面向小样本学习方向的研究者与 PyTorch 开发者,这份实现依据论文 Prototypical Networks(1703.05175)复现了原型网络的核心思路,可用于 Omniglot 等数据集的少样本分类实验、算法对照与课程作业参考。代码围绕网络定义、episode 批采样、原型距离度量与损失计算、训练入口及数据集加载等模块组织,另含参数解析工具,便于按实验需求调整 N-way K-shot 设置。资源包共 12 个文件,以 7 个 py 源码为主体,辅以 2 张 png 示意图、1 份 README 说明、1 份 LICENSE 和 1 个 .gitignore,整体约 135KB,体积轻量而结构完整。已有 920 人学习下载,读者可据此快速搭建小样本实验基线,理解 episode 训练机制与度量分类链路,并在其上替换数据集或骨干网络开展自己的验证。
1. 为什么这个 PyTorch 原型网络值得先跑一遍
很多人第一次下载 PyTorch 版原型网络(Prototypical Networks,以下简称 ProtoNet)的仓库时,会误以为它是“先在大数据集上训练一个分类器,再拿少量新类别样本微调”,可跑完一遍才发现完全不是这么回事:训练的基本单位是一个个 episode,每个 episode 现场抽 N 个类别、每类 K 张 support 图外加若干 query 图,模型要在这一个 episode 内部学会“把 query 判给离它最近的那个类原型”。这是小样本学习(few-shot learning)里最经典的一套范式,代码量不大,却把 episode 采样、嵌入网络、原型计算、度量损失四件事串成了一条完整链路。
这个项目适合两类人:一类是想搞懂 few-shot 到底怎么训练、不想只看公式的开发者;另一类是手里有一批“每类只有几张标注样本”的图,想找个能跑通、能改的 PyTorch 基线。下面按“它是什么、怎么跑、怎么改”的顺序拆。
2. 项目骨架与技术选型:episodic 训练到底怎么落到代码里
ProtoNet 的核心命题可以用一句话概括:既然新类别只有 K 张样本,那就别学分类器,改学一个嵌入空间;在这个空间里,每个类的 K 个样本求个均值当作该类的“原型”,query 距离谁的原型近就判给谁。整篇论文的数学只有三行,工程实现的难点全在“怎么把数据集切成一个个 episode”上。
所以拆这个仓库,顺序不能乱:先看目录怎么划分模块,再看一个 episode 从采样到出 loss 走了哪条链路,最后才是网络和损失那几十行。
2.1 目录结构与模块边界
这类仓库的目录划分通常很稳定,看一眼就知道该改哪个文件:
| 路径 | 职责 | 改动的频率 |
|---|---|---|
data/ | 数据集封装(Omniglot、miniImageNet 等)和 episode 采样器 | 换数据集时必须改 |
models/ | 主干网络定义、ProtoNet 前向逻辑、距离与损失 | 换 backbone / 改度量时改 |
experiment.py | 把采样器、模型、优化器、记录器组装成一次实验 | 加训练策略时改 |
train.py/eval.py | 命令行入口,读参数、建实验、跑循环 | 基本不动 |
configs/ | JSON 或 YAML 的超参配置 | 每次实验都改 |
utils/ | 日志、模型保存、t-SNE 可视化等杂项 | 很少动 |
把“数据怎么切”和“模型怎么算”彻底分开,是这个结构最值得学的一点。很多自己写的 few-shot 代码把采样逻辑塞进Dataset.__getitem__里,结果换数据集时要重写整个训练循环,得不偿失。
2.2 技术栈与依赖取舍
技术栈本身没什么花哨:PyTorch 做计算,torchvision做基础图像变换,numpy处理索引,tensorboardX或 TensorBoard 做曲线记录,matplotlib画 t-SNE 散点图,命令行参数用argparse或 JSON 配置驱动。常见做法是把超参全部外置到配置文件,代码里不写死任何 5-way 5-shot 之类的数字,这样同一份代码能直接跑 20-way 1-shot 的消融实验。
有两个依赖细节容易被忽略。第一,Omniglot 是 28×28 灰度图,不需要 ImageNet 那套归一化参数,用(x - 0.5) / 0.5或者直接除以 255 就够;第二,如果打算跑 miniImageNet,一定要提前确认有没有缓存解码后的张量,否则每个 epoch 都从 JPEG 现解码,数据加载会成为瓶颈,GPU 长时间空转。
提示:先用 Omniglot 把流程跑通再换大数据集,Omniglot 只有几万张 28×28 的小图,CPU 都能在几十分钟内跑出可用结果,是最省时间的验证路径。
2.3 一个 episode 的完整数据流
整条链路可以按张量形状来追。假设 N=5、K=5、Q=15:
- 采样器从训练类别池里随机抽 5 个类,每个类随机抽 5 张作 support、再抽 15 张作 query(同一个类内不能重叠);
- 拼成一个 batch,共
5×5 + 5×15 = 100张图,张量形状[100, 1, 28, 28]; - 主干网络把每张图压成一个 64 维向量,得到
[100, 64]; - 前 25 行 reshape 成
[5, 5, 64],沿 K 维求均值得到 5 个原型[5, 64]; - 后 75 行
[75, 64]与原型两两算距离,得到[75, 5]的距离矩阵; - 取负号做
log_softmax,得到每个 query 属于 5 个类的对数概率; - 标签是“第 i 段 query 属于第 i 个类”的构造标签,算 NLL 损失,反向传播。
这里有个容易出错的地方:support 和 query 必须来自同一个类别集合,但样本不能重复。如果采样时允许重叠,模型会在训练集上看到“答案”,准确率虚高,验证时直接大幅下降。
3. 核心模块拆解:从卷积块到原型损失的那几十行
网络和损失加起来通常不到 100 行,但每一行都影响结果。我把它们拆成四段看:主干怎么压维度、原型怎么求、距离怎么算、训练循环怎么组织。
3.1 主干网络:四个 conv-bn-relu-maxpool 压到 1×1
Omniglot 版的主干非常朴素,四个卷积块,每块后接一次 2 倍下采样,28×28 的输入经过 14→7→3→1 正好压成 1×1,通道数固定 64,所以嵌入维度就是 64:
import torch.nn as nn def conv_block(in_ch, out_ch): """conv -> bn -> relu -> maxpool,这个顺序不要随意调换""" return nn.Sequential( nn.Conv2d(in_ch, out_ch, kernel_size=3, padding=1, bias=False), nn.BatchNorm2d(out_ch), nn.ReLU(inplace=True), nn.MaxPool2d(2), ) class OmniglotEncoder(nn.Module): def __init__(self, hid_dim=64): super().__init__() self.blocks = nn.Sequential( conv_block(1, hid_dim), # 28 -> 14 conv_block(hid_dim, hid_dim), # 14 -> 7 conv_block(hid_dim, hid_dim), # 7 -> 3 conv_block(hid_dim, hid_dim), # 3 -> 1 ) self.out_dim = hid_dim # 64 * 1 * 1 def forward(self, x): x = self.blocks(x) return x.view(x.size(0), -1) # 展平成 [B, 64]bias=False是因为后面紧跟 BN,偏置会被减均值抵消,省掉它不影响精度还能少一点参数。padding=1配kernel_size=3保证分辨率不变,下采样完全交给MaxPool2d(2),这样每层的输出尺寸可以手算出来,改输入分辨率的时候不容易出错。
换 84×84 的 miniImageNet 时,通常会换成 4 个 block 各自带残差的结构,out_dim变成 160 或 640,同时最前面加一次Conv2d(3, 64, 3, padding=1)把三通道升上去。这时候out_dim一定要跟着改,因为后面的原型计算和距离计算都是拿这个维度当特征长度用的,维度写错会在 reshape 时报错,错误信息还不在真正出错的那一行。
3.2 原型计算:一个 reshape 加一个 mean
原型就是类内 support 特征的均值,实现只有两行,但 reshape 的顺序决定了它算得对不对:
import torch def compute_prototypes(support_emb, n_way, k_shot): """ support_emb: [N*K, D],必须是按类别连续排列的 返回: [N, D] """ # 关键:先按 [N, K, D] 拆,再沿 K 求均值 support_emb = support_emb.view(n_way, k_shot, -1) return support_emb.mean(dim=1)这里的前提是采样器输出的样本必须按类别连续排列,即前 K 个属于第 0 类、接下来 K 个属于第 1 类。如果采样器为了打乱顺序把类别交错排列,这个view就会把不同类的样本平均到一起,模型仍然能训,但效果会变差很多,而且不会报错,属于最难排查的那类问题。我的做法是让采样器直接返回[N, K, ...]形状的张量,在拼 batch 之前显式reshape,而不是依赖 sampler 的隐式顺序。
K=1的时候均值退化成恒等映射,原型就是那张唯一的 support 图,这也是 1-shot 场景下 ProtoNet 与“最近邻分类器”等价的原因。想验证实现是否正确,把 K 设成 1 跑一下,结果应该和手写的最近邻完全一致。
3.3 距离与损失:欧氏距离还是平方欧氏距离
论文里写的是平方欧氏距离,但代码里很多人直接写欧氏距离。两者在排序上等价,但送进 softmax 之后尺度不同,等于隐式改了一个温度系数:
import torch import torch.nn.functional as F def prototypical_loss(support_emb, query_emb, n_way, k_shot, q_queries, squared=True): prototypes = compute_prototypes(support_emb, n_way, k_shot) # [N, D] # cdist 给的是欧氏距离;想复现论文就再平方一次 dists = torch.cdist(query_emb, prototypes, p=2) # [N*Q, N] if squared: dists = dists.pow(2) log_p_y = F.log_softmax(-dists, dim=1) # 构造标签:每 Q 个 query 属于同一个类 target = torch.arange(n_way, device=query_emb.device) target = target.repeat_interleave(q_queries) # [N*Q] loss = F.nll_loss(log_p_y, target) preds = log_p_y.argmax(dim=1) acc = (preds == target).float().mean() return loss, acctorch.cdist的好处是不用自己写广播,[N*Q, D]和[N, D]直接出[N*Q, N]的矩阵。用平方距离时,特征维度的绝对值会被放大(64 维下距离量级明显更大),softmax 更尖锐,前期梯度也更激进;用非平方距离相当于给 logits 降了温。两种写法都能收敛,但如果拿论文数字做对比却没对齐这一点,0.5 到 1 个百分点的差异是解释不清的。
target的构造是最容易错的地方。如果 query 的组织顺序是“按类连续”,就用repeat_interleave;如果采样器输出的是“按 query 序号交错”,就得用tile。写错了 loss 仍然会下降,只是模型学的是错的对应关系,表现为准确率一直在 20% 附近(5-way 的随机水平),这时候第一反应应该是去核对标签构造,而不是去调学习率。
3.4 训练循环:episode 为单位,而不是 epoch
训练循环和常规分类任务结构相似,但计数单位换了:
optimizer = torch.optim.Adam(model.parameters(), lr=1e-3) scheduler = torch.optim.lr_scheduler.StepLR(optimizer, step_size=2000, gamma=0.5) for ep in range(args.train_episodes): model.train() support_x, support_y, query_x, query_y = sampler.sample_episode() support_emb = model(support_x) query_emb = model(query_x) loss, acc = prototypical_loss(support_emb, query_emb, n_way=args.n_way, k_shot=args.k_shot, q_queries=args.q_queries) optimizer.zero_grad() loss.backward() optimizer.step() scheduler.step() if ep % args.log_interval == 0: writer.add_scalar("train/loss", loss.item(), ep) writer.add_scalar("train/acc", acc.item(), ep)三个参数值得单独说清楚。train_episodes是总 episode 数,量级通常在一万到几万,跟“数据集大小除以 batch size”完全无关,因为它每步都是现场采样;step_size=2000控制学习率衰减的节奏,衰减太早模型欠拟合,太晚 loss 会震荡;q_queries决定每步的梯度质量,5-way 15-shot 的 query 数能给出比较稳的估计,太小则 loss 噪声大。
另外注意model.train()和model.eval()的切换不只是开关 dropout。BN 在 train 模式下用当前 episode 的统计量、并更新 running stats,在 eval 模式下用 running stats。episode 之间类别完全不同,running stats 会被来回拉扯,这是 ProtoNet 类实现里比较典型的一个问题,下一章会专门讲。
4. 跑起来:环境、数据准备与训练验证的完整流程
代码看明白之后,真正的门槛在数据组织方式上。很多人卡在“我把图片放对了,为什么一跑就 index out of range”,问题多半出在目录层级和类别索引的映射上。
4.1 环境安装与版本取舍
先把基础环境装上:
# 建议单独建虚拟环境,别和系统 Python 混在一起 python -m venv venv source venv/bin/activate # 先装 CPU 版跑通,再考虑 GPU pip install torch torchvision --index-url https://download.pytorch.org/whl/cpu pip install numpy matplotlib tensorboardX tqdm pyyaml我一般先用 CPU 版跑 Omniglot,原因很实际:28×28 的小图在 CPU 上跑几百个 episode 也就几分钟,足够判断代码有没有问题;等确认 loss 在降,再切 GPU 拉长训练。反过来先上 GPU,一旦卡住,你还得先排查是环境问题、显存问题还是数据问题,多花很多时间。
PyTorch 版本上不用太纠结,1.8 以后的版本 API 基本没变,torch.cdist、repeat_interleave、log_softmax这些都在。唯一要注意的是torch.load的weights_only参数在新版本里默认值变过,加载旧 checkpoint 时显式传参比较省事。
4.2 Omniglot 的目录结构与类别索引
Omniglot 的原始结构是“按字母表分目录,每个字母表下若干字符,每个字符若干张 20 张左右的写法图”:
omniglot/ └── images_background/ ├── Alphabet_of_the_Magi/ │ ├── character01/ │ │ ├── 0709_01.png │ │ └── ... │ └── character02/ └── ...数据加载模块要做的事是:遍历所有character目录,给每个目录分配一个从 0 开始的整数标签,把所有路径和标签存成两个列表。训练用images_background,测试用images_evaluation,这两个集合的字符完全不重合,正是 few-shot 评估需要的“未见类别”。
import os from PIL import Image from torch.utils.data import Dataset class OmniglotDataset(Dataset): def __init__(self, root, transform=None): self.transform = transform self.paths, self.labels = [], [] # 按字母表 -> 字符两级遍历,字符目录即一个类别 for alpha in sorted(os.listdir(root)): alpha_dir = os.path.join(root, alpha) if not os.path.isdir(alpha_dir): continue for char in sorted(os.listdir(alpha_dir)): char_dir = os.path.join(alpha_dir, char) if not os.path.isdir(char_dir): continue label = len(set(self.labels)) # 新字符分配新标签 for fn in sorted(os.listdir(char_dir)): if fn.endswith(".png"): self.paths.append(os.path.join(char_dir, fn)) self.labels.append(label) def __len__(self): return len(self.paths) def __getitem__(self, idx): img = Image.open(self.paths[idx]).convert("L") if self.transform: img = self.transform(img) return img, self.labels[idx]label = len(set(self.labels))这种写法在类别多的时候是 O(n²),几十万张图会明显拖慢启动。实际工程里我一般先建一个char_dir -> label的字典,再统一映射,启动时间能从几十秒降到一秒以内。这是个小细节,但每次启动都等半分钟的时候就很烦。
4.3 训练参数:一份能直接用的配置
超参外置成 JSON,命令行走--config指定,是我比较推荐的组织方式:
{ "exp": { "name": "omniglot_5way_5shot", "seed": 42 }, "data": { "dataset": "omniglot", "datadir": "./data/omniglot/images_background", "eval_datadir": "./data/omniglot/images_evaluation" }, "train": { "n_way": 5, "k_shot": 5, "q_queries": 15, "episodes": 20000, "log_interval": 50, "eval_interval": 500, "eval_episodes": 200 }, "model": { "backbone": "omniglot_encoder", "hid_dim": 64 }, "optim": { "lr": 0.001, "step_size": 2000, "gamma": 0.5 } }跑起来就一行:
python train.py --config configs/omniglot_5way_5shot.json参数含义和调整建议:
| 参数 | 含义 | 调整经验 |
|---|---|---|
n_way | 每个 episode 的类别数 | 越大越难,Omniglot 上 5 和 20 都常见 |
k_shot | 每类 support 样本数 | 1 和 5 是最常对比的两个设置 |
q_queries | 每类 query 样本数 | 影响梯度稳定性,建议 ≥ 10 |
episodes | 总训练 episode 数 | 小数据集 1~2 万足够 |
eval_episodes | 评估时的 episode 数 | 少于 100 时置信区间太宽,看不出差异 |
lr | 初始学习率 | Adam 下 1e-3 是安全起点 |
eval_episodes这个参数容易被忽视。评估时是随机采 episode,单次结果波动很大,跑 100 次和跑 600 次得到的均值可能差 1~2 个百分点。我一般至少跑 600 个 episode,再按1.96 × std / sqrt(n)算 95% 置信区间,不然两个配置之间的差异根本说不清是真实的还是噪声。
4.4 评估与结果解读
评估流程和训练几乎一样,只是不反传,并且要在model.eval()下跑:
model.eval() accs = [] with torch.no_grad(): for _ in range(args.eval_episodes): sx, sy, qx, qy = eval_sampler.sample_episode() _, acc = prototypical_loss(model(sx), model(qx), args.n_way, args.k_shot, args.q_queries) accs.append(acc.item()) mean = sum(accs) / len(accs) std = (sum((a - mean) ** 2 for a in accs) / len(accs)) ** 0.5 ci95 = 1.96 * std / (len(accs) ** 0.5) print(f"acc = {mean*100:.2f}% ± {ci95*100:.2f}%")5-way 5-shot 在 Omniglot 上正常收敛后准确率应该在 95% 以上,5-way 1-shot 在 90% 上下。如果你跑出来 5-way 5-shot 只有 60%,先别怀疑网络容量,八成是数据或标签的问题。判断方法很直接:把训练 episode 里的 query 换成 support 本身再评估一次,如果准确率接近 100%,说明前向链路没问题,问题出在泛化或者评估采样上;如果这一步都不到 90%,那就是实现有 bug。
t-SNE 可视化是这类仓库常见的附带功能,把 query 特征降到二维画散点,看类簇分不分得开。它的用途是演示,不是指标,别拿图好看当作模型好的证据,因为 t-SNE 的困惑度参数能显著改变视觉效果。
5. 避坑与排查:几个让人白跑一晚上的问题
下面这几条都是我在跑这类实现时真实翻过车的,按“现象 → 原因 → 解决”写,遇到了可以对照着查。
5.1 现象:loss 一直卡在 ln(N) 附近不动
原因基本是模型对每个类输出等概率,也就是距离矩阵几乎常数。常见触发点有三个:学习率太大导致第一轮就把 BN 的 running stats 打乱;特征全部塌缩到同一个点(BN 之前的卷积输出恒为 0);距离计算时用了cdist但特征没归一化,数值尺度大到 softmax 直接饱和。
解决路径按成本从低到高:先把lr降到 1e-4 跑 200 个 episode 看 loss 有没有细微下降;再打印query_emb.std()和prototypes.std(),如果接近 0 说明特征塌缩,检查卷积层有没有漏掉padding导致尺寸算错;最后检查输入归一化,Omniglot 用(x-0.5)/0.5就够了,别硬套 ImageNet 的三通道均值方差。
5.2 现象:训练准确率 99%,评估只有 60%
这不是过拟合,而是 BN 的 running stats 被 episode 之间的类别切换搞坏了。训练时每个 episode 的类别都不一样,BN 的 running mean/var 在两类完全不同的分布之间来回平均,eval()模式下用这套统计量做归一化,输出自然是偏的。
解决办法有两个。一是把主干里的BatchNorm2d换成nn.GroupNorm或者干脆去掉,代价是收敛稍慢;二是在评估时对每个 episode 单独跑一次前向、保持model.train()的统计方式但不反传,等价于用 episode 内统计量。我一般在改动之前先跑一次train()模式下的评估,如果准确率立刻回到 95%,就可以确认是这个原因。
5.3 现象:DataLoader 开了多进程之后,每个 epoch 采到的 episode 一模一样
原因是 episode 采样器用的是 Python 的random或numpy.random,fork出来的子进程继承了父进程相同的随机种子状态,多个 worker 采出完全相同的类别组合。表面上训练在跑,实际每个 batch 都是重复数据。
解决方式是给DataLoader传worker_init_fn,在函数里用torch.initial_seed() % 2**32重置每个 worker 的种子;或者干脆把 episode 采样放在主进程里做完,DataLoader只负责按索引读图。后者代码更清晰,我一般用这个。
5.4 现象:换了自己的数据集,报 reshape 维度不匹配
compute_prototypes里的view(n_way, k_shot, -1)隐含了一个前提:support 部分的样本数是n_way * k_shot。如果某个类实际可用的图片数少于k_shot,采样器要么抛异常,要么做了补齐,补齐之后的张量形状对不上。
换数据集时先统计每个类的样本数分布,把min(counts)打出来。如果最小值是 3,那k_shot就不能超过 3,或者采样时排除掉样本太少的类。别指望采样器帮你处理,很多实现的采样代码里根本没有这个检查。
5.5 现象:加载保存的模型时报 key 不匹配
state_dict的 key 多了module.前缀,说明保存时用了nn.DataParallel,加载时没有。解决就是加载前做一次字符串替换:
state = torch.load(ckpt_path, map_location="cpu") state = {k.replace("module.", "", 1): v for k, v in state.items()} model.load_state_dict(state)反过来(保存时单卡、加载时多卡)也要处理,但更常见的做法是统一保存model.module.state_dict(),从源头上避免这个问题。这个坑不大,但每次换机器都要踩一次,属于典型的后悔药场景。
6. 二次开发:把 ProtoNet 接到自己的数据和任务上
想在现有代码上做改造,改动量从小到大有三个方向,我按经验排了序。
第一层,换数据集。只要实现一个返回(image, label)的Dataset,再准备一个按类别分组的索引表给采样器就够了。关键在于采样器要能快速回答“第 c 类有哪些样本”,我一般预处理成dict[label] -> List[idx],采样时直接random.sample,避免每次都遍历全表。
第二层,换主干。把OmniglotEncoder换成任意输出[B, D]的模块即可,唯一要同步改的是out_dim,以及配置文件里的hid_dim。如果换成带残差的网络,注意别在最后加全局池化后忘记展平,view(x.size(0), -1)这一步漏掉会在 cdist 处报维度错误。
第三层,改度量方式。把欧氏距离换成余弦相似度,只需要把cdist那行替换成归一化后的矩阵乘,此时log_softmax前的符号要跟着改(相似度越大越好,所以是+sim而不是-dist)。改动只有几行,但足够做一组对照实验,看看在你的数据上哪种度量更合适。
# 余弦相似度版本,替换 prototypical_loss 里的距离计算部分 q = F.normalize(query_emb, dim=1) p = F.normalize(prototypes, dim=1) sim = q @ p.t() # [N*Q, N],越大越相似 log_p_y = F.log_softmax(sim * 10.0, dim=1) # 温度系数需要单独调那个10.0的温度是必须的。余弦相似度的值域只有[-1, 1],直接送进 softmax 会非常平,梯度接近 0,模型基本学不动。改度量的时候一定要同步检查 logits 的尺度,这是最容易翻车的地方。
验证改造是否有效,方法比改造本身更重要。我的习惯是固定随机种子,在同一批评估 episode 上比较改动前后的准确率和置信区间,只有当区间不重叠时才认为改动有效。评估 episode 要提前采好存下来,不能每次现采,否则两边比的根本不是同一批题。
| 改造点 | 涉及文件 | 主要风险 |
|---|---|---|
| 自定义数据集 | data/下新增文件 | 类别样本数不足导致采样失败 |
| 更换主干 | models/与配置 | 输出维度、输入通道数不一致 |
| 更换度量 | models/损失函数 | logits 尺度失衡,梯度消失 |
| 加验证早停 | experiment.py | 验证集类别与训练集重叠造成泄漏 |
最后说个血泪经验:这个仓库里最不能省的步骤,是每次改动前先跑一遍原始配置存下基线数字。有次我为了省事直接改了损失函数,跑出来掉了三个点,回头想确认是新度量的问题还是随机波动,基线已经没了,只能重跑一整晚。从那以后我每次动核心模块之前,都强制先跑一遍原始配置把mean ± ci95记到表格里再动手。希望帮到你。
本文还有配套的精品资源,点击获取