如何基于TensorFlow实现MAML元学习解决少样本多标签回归问题

核心思路适配你的场景 #

• 将feature1的每个类别（A/B/C/D/E/F）视为独立元任务，训练阶段仅用A/B/C/D的任务优化模型初始化参数
• 内循环：针对单个任务用少量支持样本快速微调临时参数
• 外循环：跨所有训练任务优化初始化参数，让模型能快速适配未见过的E/F类任务
• 多标签回归任务采用MSE损失，输出层用线性激活（回归无需非线性映射）

代码实现（TensorFlow 2.x） #

1. 导入依赖与基础模型定义 #

import tensorflow as tf
import numpy as np
from sklearn.model_selection import train_test_split

# 定义共享基础回归网络
def build_base_model(input_dim=152, output_dim=18211):
    inputs = tf.keras.Input(shape=(input_dim,))
    x = tf.keras.layers.Dense(256, activation='relu')(inputs)
    x = tf.keras.layers.Dense(128, activation='relu')(x)
    outputs = tf.keras.layers.Dense(output_dim, activation='linear')(x)
    model = tf.keras.Model(inputs=inputs, outputs=outputs)
    return model

2. MAML核心逻辑实现 #

class MAML:
    def __init__(self, base_model, inner_lr=0.01, outer_lr=0.001):
        self.base_model = base_model
        self.inner_lr = inner_lr
        self.outer_optimizer = tf.keras.optimizers.Adam(learning_rate=outer_lr)
        self.loss_fn = tf.keras.losses.MeanSquaredError()

    # 内循环：单任务参数微调
    def inner_update(self, support_x, support_y):
        params = self.base_model.trainable_variables
        with tf.GradientTape() as tape:
            preds = self.base_model(support_x, training=True)
            loss = self.loss_fn(support_y, preds)
        grads = tape.gradient(loss, params)
        return [p - self.inner_lr * g for p, g in zip(params, grads)]

    # 外循环：跨任务优化初始化参数
    def outer_update(self, tasks):
        total_loss = 0.0
        with tf.GradientTape() as outer_tape:
            for task in tasks:
                support_x, support_y, query_x, query_y = task
                # 内循环得到临时参数
                updated_params = self.inner_update(support_x, support_y)
                # 用临时参数计算查询集损失
                with tf.GradientTape(persistent=True):
                    # 替换模型参数为临时参数
                    for orig_p, new_p in zip(self.base_model.trainable_variables, updated_params):
                        orig_p.assign(new_p)
                    preds = self.base_model(query_x, training=True)
                    task_loss = self.loss_fn(query_y, preds)
                    total_loss += task_loss
                    # 恢复原初始化参数
                    for orig_p, new_p in zip(self.base_model.trainable_variables, updated_params):
                        orig_p.assign(tf.Variable(orig_p.numpy()))
            avg_loss = total_loss / len(tasks)
        # 更新全局初始化参数
        outer_grads = outer_tape.gradient(avg_loss, self.base_model.trainable_variables)
        self.outer_optimizer.apply_gradients(zip(outer_grads, self.base_model.trainable_variables))
        return avg_loss.numpy()

    # 测试阶段：用少量样本快速适配后预测
    def adapt_and_predict(self, support_x, support_y, test_x, num_inner_steps=3):
        params = self.base_model.trainable_variables
        # 多步内循环微调
        for _ in range(num_inner_steps):
            with tf.GradientTape() as tape:
                preds = self.base_model(support_x, training=True)
                loss = self.loss_fn(support_y, preds)
            grads = tape.gradient(loss, params)
            params = [p - self.inner_lr * g for p, g in zip(params, grads)]
        # 用微调后的参数预测
        for orig_p, new_p in zip(self.base_model.trainable_variables, params):
            orig_p.assign(new_p)
        preds = self.base_model(test_x, training=False)
        # 恢复原参数
        for orig_p, new_p in zip(self.base_model.trainable_variables, params):
            orig_p.assign(tf.Variable(orig_p.numpy()))
        return preds.numpy()

3. 数据适配与任务生成 #

# 按feature1类别拆分数据（假设feature1的one-hot编码占前6列，对应A-F）
def split_data_by_feature1(features, labels):
    feature1_onehot = features[:, :6]
    class_indices = {}
    for i in range(6):
        mask = feature1_onehot[:, i] == 1
        class_indices[chr(ord('A')+i)] = np.where(mask)[0]
    class_data = {}
    for cls, idx in class_indices.items():
        class_data[cls] = (features[idx], labels[idx])
    return class_data

# 假设你已有的数据集变量：train_features, train_labels, val_features, val_labels
train_class_data = split_data_by_feature1(train_features, train_labels)
val_class_data = split_data_by_feature1(val_features, val_labels)

# 生成元任务（每个任务包含支持集和查询集）
def generate_tasks(class_data, task_classes, support_size=10, query_size=20):
    tasks = []
    for cls in task_classes:
        x, y = class_data[cls]
        support_x, query_x, support_y, query_y = train_test_split(x, y, train_size=support_size, test_size=query_size, random_state=42)
        tasks.append((support_x, support_y, query_x, query_y))
    return tasks

# 训练任务：仅用A/B/C/D类
train_tasks = generate_tasks(train_class_data, task_classes=['A','B','C','D'], support_size=20, query_size=30)
# 验证任务：用E/F类（少量样本）
val_tasks = generate_tasks(val_class_data, task_classes=['E','F'], support_size=5, query_size=10)

4. 训练与验证循环 #

# 初始化模型与MAML
base_model = build_base_model()
maml = MAML(base_model, inner_lr=0.005, outer_lr=0.0005)

# 训练参数
num_epochs = 50
num_tasks_per_epoch = 4

for epoch in range(num_epochs):
    epoch_loss = 0.0
    np.random.shuffle(train_tasks)
    batch_tasks = train_tasks[:num_tasks_per_epoch]
    loss = maml.outer_update(batch_tasks)
    epoch_loss += loss

    # 每5轮验证一次
    if (epoch + 1) % 5 == 0:
        val_loss = 0.0
        for task in val_tasks:
            support_x, support_y, query_x, query_y = task
            preds = maml.adapt_and_predict(support_x, support_y, query_x)
            val_loss += tf.keras.losses.MSE(query_y, preds).numpy().mean()
        val_loss /= len(val_tasks)
        print(f"Epoch {epoch+1}, Train Loss: {epoch_loss:.4f}, Val Loss: {val_loss:.4f}")
    else:
        print(f"Epoch {epoch+1}, Train Loss: {epoch_loss:.4f}")

5. 测试阶段（适配E/F类） #

# 假设你有E/F类的测试数据：test_features_E, test_features_F
# 用验证集中的E类少量样本做支持集，快速适配后预测
support_x_E, support_y_E = val_class_data['E']
test_preds_E = maml.adapt_and_predict(support_x_E, support_y_E, test_features_E, num_inner_steps=5)

# 适配F类并预测
support_x_F, support_y_F = val_class_data['F']
test_preds_F = maml.adapt_and_predict(support_x_F, support_y_F, test_features_F, num_inner_steps=5)

关键调参提示 #

• inner_lr：内循环学习率，建议0.001-0.01，太小微调效率低，太大易过拟合支持集
• outer_lr：外循环学习率，建议比内循环小一个数量级（如0.0001-0.001）
• support_size：根据E/F类样本数调整，最少5-10个样本即可完成适配
• 网络结构：可根据数据复杂度调整隐藏层大小（如增加到512或减少到64）

内容的提问来源于stack exchange，提问作者Pranav Belhekar