小样本学习Few-Shot：TensorFlow原型网络
在医疗影像诊断系统中，一个新发现的罕见肿瘤类型可能仅有三五个标注病例。传统深度学习模型面对这种“数据荒漠”往往束手无策——训练不动、泛化不了。这正是小样本学习（Few-Shot Learning）要解决的核心难题。
现实世界中的AI落地场景，很少像ImageNet那样拥有百万级标注数据。更多时候，我们面对的是冷启动问题：新品上架、异常事件出现、个性化需求涌现……这些都要求模型具备“举一反三”的能力。原型网络（Prototypical Networks）正是这一思想的优雅体现：它不执着于记住每个类别，而是学会如何“看懂”图像的本质特征，并通过简单的均值原型完成快速分类。
而要把这套机制从论文搬进生产环境？TensorFlow几乎是必然选择。不仅是因其强大的计算图优化和部署工具链，更在于它能将研究级别的算法转化为可监控、可扩展、可维护的工业级服务。
我们不妨先看一个典型的工作流。假设你要为一家医疗器械公司开发肺部结节识别系统，已知常见结节数量充足，但某些亚型仅有个位数样本。你会怎么做？
首先不是堆数据增强，也不是换更大的网络，而是重构学习范式——从“批量监督训练”转向“元学习”（Meta-Learning）。具体来说，就是模拟真实使用时的少样本情境，在训练阶段就不断构造N-way K-shot任务：每轮随机抽取N个类别，每个类别只给K个样本作为支持集（support set），再用其他样本来测试模型能否正确匹配查询样本（query sample）到对应原型。
这个过程听起来复杂，但在TensorFlow中可以被清晰地拆解为几个核心模块：
import tensorflow as tf from tensorflow.keras import layers, Model def build_embedding_network(input_shape=(28, 28, 1), embedding_dim=64): inputs = layers.Input(shape=input_shape) x = layers.Conv2D(32, 3, activation='relu')(inputs) x = layers.MaxPooling2D()(x) x = layers.Conv2D(64, 3, activation='relu')(x) x = layers.MaxPooling2D()(x) x = layers.GlobalAveragePooling2D()(x) embeddings = layers.Dense(embedding_dim)(x) return Model(inputs, embeddings)这段代码定义了一个轻量CNN作为特征提取器。注意这里没有使用复杂的ResNet或Vision Transformer，因为在小样本场景下，过强的模型反而容易过拟合有限的支持样本。相反，一个结构简单、收敛稳定的骨干网络更能发挥度量学习的优势。
接下来是关键一步：原型计算。对于每个类别c，其原型$\mathbf{p}_c$是该类所有支持样本嵌入向量的均值：
$$
\mathbf{p}c = \frac{1}{|S_c|} \sum{(\mathbf{x}i, y_i=c) \in S} f\theta(\mathbf{x}_i)
$$
实现起来看似平凡，但其中隐藏着工程上的考量。比如是否使用tf.boolean_mask进行标签筛选？在大批量episode训练时，这种方式虽然直观，却可能导致内存碎片化。更好的做法是预处理标签为one-hot形式，利用矩阵乘法完成聚合：
def compute_prototypes(support_embeddings, support_onehot, num_classes): # support_onehot: [batch_size, way] # 转置后相乘实现按类求和 class_sums = tf.linalg.matmul(support_onehot, support_embeddings, transpose_a=True) class_counts = tf.reduce_sum(support_onehot, axis=0, keepdims=True) # [1, way] prototypes = class_sums / (class_counts + 1e-8) # 防除零 return prototypes # [way, embedding_dim]这种向量化操作不仅更高效，也更容易被XLA编译器优化，尤其适合在TPU上运行。
整个前向逻辑封装在一个自定义Model中：
class PrototypicalNetwork(Model): def __init__(self, embedding_model): super().__init__() self.embedding_model = embedding_model def call(self, support_set, query_set, support_labels, num_classes, way, shot): batch_size = tf.shape(query_set)[0] # 提取嵌入 support_embeddings = self.embedding_model(support_set) # [B*S*N, D] query_embeddings = self.embedding_model(query_set) # [B*Q*N, D] # 构造one-hot标签 [B*N, way] support_labels_flat = tf.reshape(support_labels, [-1]) support_onehot = tf.one_hot(support_labels_flat, depth=way) # 计算原型 [way, D] prototypes = compute_prototypes(support_embeddings, support_onehot, way) # 扩展维度以便广播计算距离 query_expanded = tf.expand_dims(query_embeddings, axis=1) # [Q, 1, D] proto_expanded = tf.expand_dims(prototypes, axis=0) # [1, way, D] distances = tf.norm(query_expanded - proto_expanded, axis=-1) # [Q, way] return -distances # 返回负距离作为logits你会发现，整个模型并没有传统的“全连接分类头”。它的分类能力完全依赖于动态生成的原型和度量空间的一致性。这意味着一旦训练完成，新增一个类别不需要重新训练任何参数——只需用新的支持样本计算出一个新的原型向量即可上线。
这带来了巨大的工程优势。设想电商平台每天都有成千上万的新商品上架，如果每次都要微调模型，运维成本将不可承受。而基于原型的方法只需在数据库中添加一条原型记录，就能立即支持识别，真正实现“零停机扩展”。
当然，理想很丰满，实际训练中仍有不少坑需要避开。
首先是数据管道的设计。小样本任务天然不适合标准的minibatch训练模式。你需要一个episode sampler，每次从中随机抽取N个类别、每个类别K个样本组成支持集，再搭配若干查询样本。TensorFlow的tf.data.Dataset提供了强大支持：
def create_episode_dataset(dataset_dict, way=5, shot=1, query_num=15, episodes_per_epoch=1000): def sample_episode(_): classes = tf.random.shuffle(list(dataset_dict.keys()))[:way] support_set, query_set = [], [] support_labels, query_labels = [], [] for i, c in enumerate(classes): imgs = dataset_dict[c] indices = tf.random.shuffle(tf.range(len(imgs)))[:shot + query_num] sup_idx = indices[:shot] qry_idx = indices[shot:shot+query_num] support_set.append(tf.gather(imgs, sup_idx)) query_set.append(tf.gather(imgs, qry_idx)) support_labels.append(tf.fill([shot], i)) query_labels.append(tf.fill([query_num], i)) return (tf.concat(support_set, axis=0), tf.concat(query_set, axis=0), tf.concat(support_labels, axis=0), tf.concat(query_labels, axis=0)) return tf.data.Dataset.from_tensor_slices([0]*episodes_per_epoch)\ .map(sample_episode, num_parallel_calls=tf.data.AUTOTUNE)\ .batch(1) # 每次一个episode这种动态采样方式确保了每一episode都是独立的任务分布，有效防止模型对特定类别产生偏好。
其次是训练稳定性问题。由于每个episode样本极少，梯度噪声较大。因此学习率不宜过高，通常设置在1e-3~1e-4之间，并配合余弦退火策略：
lr_schedule = tf.keras.optimizers.schedules.CosineDecay( initial_learning_rate=1e-3, decay_steps=10000, alpha=1e-2 ) optimizer = tf.optimizers.Adam(lr_schedule)同时，引入BatchNorm而非Dropout更为稳妥——后者在极小批量下统计特性不稳定，可能破坏嵌入空间的一致性。
真正的考验在部署环节。你当然可以把整个训练流程打包上线，但那显然过于笨重。实际上，推理阶段只需要两部分：固定的特征编码器和可更新的原型库。
TensorFlow的SavedModel格式完美支持这种分离：
# 导出仅包含特征提取器的模型 tf.saved_model.save(embedding_net, 'embedding_model/1/') # 原型可以单独存为NumPy数组或数据库表 prototypes_db = { 'class_0': np.array([...]), 'class_1': np.array([...]), # ... }前端服务接收图像后，先调用TF Serving执行特征提取，再与本地缓存的原型做距离比对。整个过程可在毫秒级完成，即便在边缘设备上也能流畅运行。
更进一步，借助TFLite还能实现端侧推理：
converter = tf.lite.TFLiteConverter.from_keras_model(embedding_net) tflite_model = converter.convert() open('embedding_model.tflite', 'wb').write(tflite_model)这对隐私敏感的应用尤为重要——医疗影像不必上传云端，直接在本地完成初步筛查。
回过头来看，原型网络的成功并非来自多么复杂的架构，而是一种思维方式的转变：把分类问题转化为检索问题。只要特征空间足够规整，哪怕只有一个样本，也能形成有意义的“锚点”。
这也解释了为什么在某些任务中，余弦相似度优于欧氏距离。当嵌入向量被L2归一化后，内积等价于夹角余弦，此时距离度量不再受幅值干扰，更适合跨类别比较。
你可以轻松切换：
# 使用余弦相似度 normed_query = tf.nn.l2_normalize(query_embeddings, axis=-1) normed_proto = tf.nn.l2_normalize(prototypes, axis=-1) cos_sim = tf.matmul(normed_query, normed_proto, transpose_b=True) logits = cos_sim * scale # 可学习的温度系数甚至可以让温度系数scale成为一个可训练参数，在训练中自动调整决策边界锐度。
最后值得一提的是，近年来自监督预训练正在重塑小样本学习格局。比如先在大规模无标签医学图像上用SimCLR或MAE方式进行预训练，再用少量标注数据微调原型网络，往往能取得远超随机初始化的效果。
TensorFlow Hub中已有大量此类模型可供迁移：
import tensorflow_hub as hub backbone = hub.KerasLayer("https://tfhub.dev/google/imagenet/mobilenet_v3_small_100_224/feature_vector/5", output_shape=[1024], trainable=True)这种“预训练+度量学习”的组合拳，正成为资源受限场景下的主流解决方案。
当你站在医院放射科的服务器前，看着新来的三个疑似病例被准确归类到某个罕见亚型时，会意识到：AI的价值不在跑赢基准测试，而在处理那些“不够格”被传统方法训练的数据。而原型网络与TensorFlow的结合，恰恰为这种能力提供了坚实的落脚点。
