目录

小样本学习（Few-Shot Learning）

概念介绍

核心思想

多模态学习（Multimodal Learning）

概念介绍

融合策略

代码示例

应用场景

优势与挑战

优势

挑战

---

小样本学习（Few-Shot Learning）

概念介绍

小样本学习旨在让模型能够从极少量样本中学习新概念。主要方法包括：

• 元学习（Meta-Learning）：学习如何学习
• 度量学习（Metric Learning）：学习样本间的相似度度量
• 数据增强：生成更多训练样本
• 迁移学习：利用预训练模型的知识

核心思想

• 支持集（Support Set）：少量标注样本
• 查询集（Query Set）：需要分类的测试样本
• N-way K-shot：N个类别，每个类别K个样本

多模态学习（Multimodal Learning）

概念介绍

多模态学习处理来自不同模态（如图像、文本、音频）的数据，通过融合多源信息提升模型性能。

融合策略

• 早期融合：在输入层融合不同模态
• 中期融合：在中间层融合特征
• 晚期融合：在决策层融合结果

代码示例

import torch
import torch.nn as nn
import torch.optim as optim
import torchvision
import torchvision.transforms as transforms
import numpy as np
from torch.utils.data import DataLoader
from PIL import Image
import requests
from io import BytesIO

# 设置设备
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print(f"使用设备: {device}")


class SimpleCNN(nn.Module):
    """CNN用于图像特征提取"""

    def __init__(self):
        super(SimpleCNN, self).__init__()
        self.features = nn.Sequential(
            nn.Conv2d(3, 64, 3, padding=1),
            nn.ReLU(),
            nn.MaxPool2d(2),
            nn.Conv2d(64, 128, 3, padding=1),
            nn.ReLU(),
            nn.MaxPool2d(2),
            nn.Conv2d(128, 256, 3, padding=1),
            nn.ReLU(),
            nn.AdaptiveAvgPool2d((1, 1))
        )
        self.image_proj = nn.Linear(256, 128)

    def forward(self, x):
        x = self.features(x)
        x = x.view(x.size(0), -1)
        return self.image_proj(x)


class TextEncoder(nn.Module):
    """文本编码器"""

    def __init__(self, vocab_size=1000, embed_dim=128, hidden_dim=128):
        super(TextEncoder, self).__init__()
        self.embedding = nn.Embedding(vocab_size, embed_dim)
        self.encoder = nn.Sequential(
            nn.Linear(embed_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, 128)
        )

    def forward(self, text_indices):
        embedded = self.embedding(text_indices)
        # 简单平均池化
        if embedded.dim() == 3:  # 如果有序列长度维度
            embedded = embedded.mean(dim=1)
        return self.encoder(embedded)


class SimpleMultimodalModel(nn.Module):
    """多模态小样本学习模型"""

    def __init__(self):
        super(SimpleMultimodalModel, self).__init__()
        self.image_encoder = SimpleCNN()
        self.text_encoder = TextEncoder()

        # 多模态融合层
        self.fusion = nn.Sequential(
            nn.Linear(256, 128),  # 图像128维 + 文本128维 = 256维
            nn.ReLU(),
            nn.Dropout(0.3)
        )

    def encode_multimodal(self, images, text_indices):
        image_features = self.image_encoder(images)
        text_features = self.text_encoder(text_indices)

        # 拼接多模态特征
        multimodal_features = torch.cat([image_features, text_features], dim=1)
        return self.fusion(multimodal_features)

    def forward(self, support_images, support_texts, support_labels, query_images, query_texts):
        # 支持集特征
        support_features = self.encode_multimodal(support_images, support_texts)

        # 计算原型（每个类别的中心）
        prototypes = []
        unique_labels = torch.unique(support_labels)
        for label in unique_labels:
            mask = (support_labels == label)
            class_features = support_features[mask]
            prototype = class_features.mean(dim=0)
            prototypes.append(prototype)
        prototypes = torch.stack(prototypes)

        # 查询集特征
        query_features = self.encode_multimodal(query_images, query_texts)

        # 计算相似度（欧氏距离的负数）
        distances = torch.cdist(query_features.unsqueeze(0), prototypes.unsqueeze(0)).squeeze(0)
        return -distances  # 距离越小，相似度越高


# CIFAR-10数据预处理
transform = transforms.Compose([
    transforms.Resize((32, 32)),
    transforms.ToTensor(),
    transforms.Normalize((0.5, 0.5, 0.5), (0.5, 0.5, 0.5))
])

# CIFAR-10类别和对应的文本描述
cifar10_classes = ['airplane', 'automobile', 'bird', 'cat', 'deer',
                   'dog', 'frog', 'horse', 'ship', 'truck']


class SimpleTokenizer:
    """文本tokenizer"""

    def __init__(self, class_names):
        self.vocab = {name: i + 1 for i, name in enumerate(class_names)}  # 0留给padding
        self.vocab['photo'] = len(class_names) + 1
        self.vocab['of'] = len(class_names) + 2
        self.vocab['a'] = len(class_names) + 3
        self.vocab_size = len(self.vocab) + 1  # +1 for padding

    def encode(self, text):
        words = text.lower().split()
        indices = [self.vocab.get(word, 0) for word in words]  # 0 for unknown
        return torch.tensor(indices, dtype=torch.long)


def create_few_shot_episode(dataset, n_way=5, k_shot=1, query_size=5):
    """创建一个小样本学习episode"""
    # 按类别组织数据
    class_data = {}
    for img, label in dataset:
        if label not in class_data:
            class_data[label] = []
        class_data[label].append(img)

    # 随机选择N个类别
    selected_classes = np.random.choice(list(class_data.keys()), n_way, replace=False)

    support_images, support_texts, support_labels = [], [], []
    query_images, query_texts, query_labels = [], [], []

    tokenizer = SimpleTokenizer(cifar10_classes)

    for new_label, class_id in enumerate(selected_classes):
        class_name = cifar10_classes[class_id]
        class_samples = class_data[class_id]

        # 随机选择样本
        selected_indices = np.random.choice(len(class_samples), k_shot + query_size, replace=False)

        # 支持集
        for i in range(k_shot):
            support_images.append(class_samples[selected_indices[i]])
            text = f"a photo of a {class_name}"
            support_texts.append(tokenizer.encode(text))
            support_labels.append(new_label)

        # 查询集
        for i in range(k_shot, k_shot + query_size):
            query_images.append(class_samples[selected_indices[i]])
            text = f"a photo of a {class_name}"
            query_texts.append(tokenizer.encode(text))
            query_labels.append(new_label)

    # 处理文本长度不一致的问题（填充到相同长度）
    max_len = max(len(text) for text in support_texts + query_texts)

    def pad_text(text, max_len):
        if len(text) < max_len:
            return torch.cat([text, torch.zeros(max_len - len(text), dtype=torch.long)])
        return text[:max_len]

    support_texts = torch.stack([pad_text(text, max_len) for text in support_texts])
    query_texts = torch.stack([pad_text(text, max_len) for text in query_texts])

    return {
        'support_images': torch.stack(support_images),
        'support_texts': support_texts,
        'support_labels': torch.tensor(support_labels),
        'query_images': torch.stack(query_images),
        'query_texts': query_texts,
        'query_labels': torch.tensor(query_labels)
    }


def train_model():
    """训练模型"""
    print("加载CIFAR-10数据集...")
    trainset = torchvision.datasets.CIFAR10(root='./data', train=True, download=True, transform=transform)
    testset = torchvision.datasets.CIFAR10(root='./data', train=False, download=True, transform=transform)

    print("创建模型...")
    model = SimpleMultimodalModel().to(device)
    optimizer = optim.Adam(model.parameters(), lr=0.001)
    criterion = nn.CrossEntropyLoss()

    print("开始训练...")
    model.train()

    for epoch in range(10):
        total_loss = 0
        correct = 0
        total = 0

        # 创建多个episode进行训练
        for episode in range(50):
            # 创建训练episode
            batch = create_few_shot_episode(trainset, n_way=5, k_shot=1, query_size=5)

            # 移动到设备
            batch = {k: v.to(device) if isinstance(v, torch.Tensor) else v for k, v in batch.items()}

            optimizer.zero_grad()

            # 前向传播
            logits = model(
                batch['support_images'],
                batch['support_texts'],
                batch['support_labels'],
                batch['query_images'],
                batch['query_texts']
            )

            loss = criterion(logits, batch['query_labels'])
            loss.backward()
            optimizer.step()

            total_loss += loss.item()

            # 计算准确率
            preds = logits.argmax(dim=1)
            correct += (preds == batch['query_labels']).sum().item()
            total += batch['query_labels'].size(0)

        accuracy = correct / total if total > 0 else 0
        print(f'Epoch {epoch + 1}/10: Loss = {total_loss / 50:.4f}, Accuracy = {accuracy:.4f}')
    # Epoch
    # 1 / 10: Loss = 1.0424, Accuracy = 0.8112
    # Epoch
    # 2 / 10: Loss = 0.2722, Accuracy = 0.9184
    # Epoch
    # 3 / 10: Loss = 0.1104, Accuracy = 0.9808
    # Epoch
    # 4 / 10: Loss = 0.0469, Accuracy = 0.9896
    # Epoch
    # 5 / 10: Loss = 0.0232, Accuracy = 0.9984
    # Epoch
    # 6 / 10: Loss = 0.0178, Accuracy = 0.9984
    # Epoch
    # 7 / 10: Loss = 0.0109, Accuracy = 0.9984
    # Epoch
    # 8 / 10: Loss = 0.0165, Accuracy = 0.9952
    # Epoch
    # 9 / 10: Loss = 0.0087, Accuracy = 0.9984
    # Epoch
    # 10 / 10: Loss = 0.0101, Accuracy = 0.9976

    # 测试
    print("\n开始测试...")
    model.eval()
    test_correct = 0
    test_total = 0

    with torch.no_grad():
        for episode in range(20):
            batch = create_few_shot_episode(testset, n_way=5, k_shot=1, query_size=5)
            batch = {k: v.to(device) if isinstance(v, torch.Tensor) else v for k, v in batch.items()}

            logits = model(
                batch['support_images'],
                batch['support_texts'],
                batch['support_labels'],
                batch['query_images'],
                batch['query_texts']
            )

            preds = logits.argmax(dim=1)
            test_correct += (preds == batch['query_labels']).sum().item()
            test_total += batch['query_labels'].size(0)

    test_accuracy = test_correct / test_total if test_total > 0 else 0
    print(f'测试准确率: {test_accuracy:.4f}')  # 1.0000

    return model


def demo_model(model):

    testset = torchvision.datasets.CIFAR10(root='./data', train=False, download=True, transform=transform)

    batch = create_few_shot_episode(testset, n_way=3, k_shot=1, query_size=3)

    print("支持集类别:")
    tokenizer = SimpleTokenizer(cifar10_classes)
    for i in range(3):
        class_id = batch['support_labels'][i].item()
        print(f"  类别 {class_id}")
        # 类别 0
        # 类别 1
        # 类别 2

    model.eval()
    with torch.no_grad():
        batch = {k: v.to(device) if isinstance(v, torch.Tensor) else v for k, v in batch.items()}

        logits = model(
            batch['support_images'],
            batch['support_texts'],
            batch['support_labels'],
            batch['query_images'],
            batch['query_texts']
        )

        preds = logits.argmax(dim=1)
        accuracy = (preds == batch['query_labels']).float().mean().item()

        print(f"查询集准确率: {accuracy:.4f}")  # 1.0000
        print(f"预测: {preds.cpu().numpy()}")  # [0 0 0 1 1 1 2 2 2]
        print(f"真实: {batch['query_labels'].cpu().numpy()}")  # [0 0 0 1 1 1 2 2 2]


if __name__ == "__main__":
    model = train_model()
    demo_model(model)

应用场景

• 医疗影像诊断：结合医学图像和报告文本
• 商品识别：结合商品图片和描述文本
• 安防监控：结合视频帧和音频信息
• 教育领域：结合题目图片和文字描述

优势与挑战

优势：

• 充分利用多模态信息的互补性
• 减少对大量标注数据的依赖
• 提高模型的泛化能力

挑战：

• 模态间对齐困难
• 模态缺失问题处理
• 计算复杂度较高