[多模态项目实践] - 模态表示：基于对应表征的跨模态搜索（图形交互搜索）

【多模态项目实战】-模态表示：基于对应表示的跨模态检索

---

参考：飞桨AI Studio星河社区-【多模态】实战案例：基于对应表示的跨模态检索

完整代码

---

1.任务介绍

任务说明：基于一个模态的数据，去另一个模态的候选集中进行检索，得到对应的数据

2.跨模态检索技术简介

跨模态检索的关键就是建立不同模态数据之间的关联，更直接地，模型需要能够输出多个模态数据的匹配分数。

如图所示，现有的方法可以被分为两类：

• 对应表示方法：学习图文多模态对应表示，然后直接利用图像和文本的对应表示的距离计算匹配分数；
• 共享表示方法：学习图文多模态共享表示，然后在共享表示层上增加一个或多个网络层直接输出图像和文本的匹配分数。

一般而言，和对应表示方法相比，**共享表示方法因为充分融合了图文信息，可以获得更好的性能。**一个直观的理解是给定两个模态的数据，对应表示方法限定了两个模态的关联必须是在没有交互的前提下建立，而共享表示方法则没有该限制。因此，共享表示方法拥有更大的*度来拟合数据的分布。

然而，共享表示方法的检索非常耗时。例如，在执行以文检图任务中，需要将文本查询和候选集中的每一张图片都成对的输入到模型中，才能得到文本查询与候选集中所有图片的匹配分数。 而对应表示方法只需要提前离线计算好候选集中所有图片的表示，然后在检索时，只需要实时计算文本查询的表示，再利用最近邻检索算法搜索图像最近邻即可。 因此，对应表示方法检索速度快，在实际的跨模态检索中使用更为广泛。

接下来，我们将具体介绍使用对应表示方法的模型VSE++的实战案例，其官方代码见链接。

3.模型训练流程

模型训练的一般流程：

从现代的深度学习框架基础下，模型训练的一般流程，包括读取数据、前馈计算、计算损失、更新参数、选择模型五个步骤。每个步骤需要实现相应的模块。

• 在读取数据阶段，我们首先需要下载数据集，然后对整理数据集的格式，以方便接下来构造数据集类，最后在数据集类的基础上构建能够按批次产生训练、验证和测试数据的对象。
• 在前馈计算阶段，我们需要实现具体的模型，使得模型能够根据输入产生相应的输出。
• 在计算损失阶段，我们需要将模型输出和预期输出进行对比，实现损失函数。
• 在更新参数阶段，我们需要给出具体的参数更新方法，即优化方法；由于现代深度学习框架能够自动计算参数梯度，并实现了绝大多数优化方法，我们通常只需要从中进行选择即可。
• 在选择模型阶段，我们需要实现具体的评估指标，选出在验证集上表现最优的模型参数。

下面，我们将介绍VSE++模型的这些阶段的具体实现，并在最后将这些阶段串联起来，最终实现模型的训练。

3.1读取数据

1)下载数据集????

我们使用的数据集为flickr8k(下载地址)，下载解压后，我们将其图片放在指定目录(本节的代码中将该目录设置为data/flickr8k)下的images文件夹里。该数据集包括8000张图片，每张图片对应5个句子描述。数据集划分采用Karpathy提供的方法(下载地址)，下载解压后，将其中的dataset_flickr8k.json文件拷贝到指定目录下。该划分方法将数据集分成3个子集：6,000张图片和其对应句子描述组成训练集，1,000张图片和描述为验证集，剩余的1,000张图片和描述为测试集。

本项目中，我们已经提供了下载好的数据集，直接运行下面的命令即可。

# 解压数据
!unzip -q ./data/data243982/flickr8k.zip -d ./data/

2)整理数据集

数据集下载完成后，我们需要对其进行处理，以适合之后构造的Paddle数据集类读取。

• 对于文本描述，我们首先构建词典，然后根据词典将文本描述转化为向量。
• 对于图像，我们这里仅记录文件路径。如果机器的内存和硬盘空间就比较大，这里也可以将图片读取并处理成三维数组，这样在模型训练和测试的阶段，就不需要再直接读取图片。

下面是整理数据集的函数的代码。

%matplotlib inline
import os
from os.path import join as pjoin
import json
import random 
from collections import defaultdict, Counter
from PIL import Image
from matplotlib import pyplot as plt

def create_dataset(data_dir='./data',
                   dataset='flickr8k',
                   captions_per_image=5, 
                   min_word_count=5, 
                   max_len=30):
    """
    参数：
        data_dir: 数据存储目录
        dataset：数据集名称
        captions_per_image：每张图片对应的文本描述数
        min_word_count：仅考虑在数据集中（除测试集外）出现5次的词
        max_len：文本描述包含的最大单词数，如果文本描述超过该值，则截断
    输出：
        一个词典文件： vocab.json
        三个数据集文件： train_data.json、 val_data.json、 test_data.json
    """

    karpathy_json_path = pjoin(data_dir, '%s/dataset_flickr8k.json' % dataset)
    image_folder=pjoin(data_dir, '%s/images' % dataset)
    output_folder=pjoin(data_dir, '%s' % dataset)

    with open(karpathy_json_path, 'r') as j:
        data = json.load(j)
    
    image_paths = defaultdict(list)
    image_captions = defaultdict(list)
    vocab = Counter()

    for img in data['images']:
        split = img['split']
        captions = []
        for c in img['sentences']:
            # 更新词频，测试集在训练过程中是未见数据集，不能统计
            if split != 'test':
                vocab.update(c['tokens'])
            # 不统计超过最大长度限制的词
            if len(c['tokens']) <= max_len:
                captions.append(c['tokens'])
        if len(captions) == 0:
            continue

        path = os.path.join(image_folder, img['filename'])
        
        image_paths[split].append(path)
        image_captions[split].append(captions)

    # 创建词典，增加占位标识符<pad>、未登录词标识符<unk>、句子首尾标识符<start>和<end>
    words = [w for w in vocab.keys() if vocab[w] > min_word_count]
    vocab = {k: v + 1 for v, k in enumerate(words)}
    vocab['<pad>'] = 0
    vocab['<unk>'] = len(vocab)
    vocab['<start>'] = len(vocab)
    vocab['<end>'] = len(vocab)

    # 存储词典
    with open(os.path.join(output_folder, 'vocab.json'), 'w') as fw:
        json.dump(vocab, fw)

    # 整理数据集
    for split in image_paths:
        imgpaths = image_paths[split]
        imcaps = image_captions[split]
        
        enc_captions = []

        for i, path in enumerate(imgpaths):
            # 合法性检查，检查图像是否可以被解析
            img = Image.open(path) 
            # 如果该图片对应的描述数量不足，则补足
            if len(imcaps[i]) < captions_per_image:
                captions = imcaps[i] + \
                    [random.choice(imcaps[i]) for _ in range(captions_per_image - len(imcaps[i]))]
            # 如果该图片对应的描述数量超了，则随机采样
            else:
                captions = random.sample(imcaps[i], k=captions_per_image)
            assert len(captions) == captions_per_image
            
            for j, c in enumerate(captions):
                # 对文本描述进行编码
                enc_c = [vocab['<start>']] + [vocab.get(word, vocab['<unk>']) for word in c] + [vocab['<end>']] 
                enc_captions.append(enc_c)
        # 合法性检查
        assert len(imgpaths) * captions_per_image == len(enc_captions)
        
        # 存储数据
        data = {'IMAGES': imgpaths, 
                'CAPTIONS': enc_captions}
        with open(pjoin(output_folder, split + '_data.json'), 'w') as fw:
            json.dump(data, fw)

data_dir = './data'
if not os.path.exists(pjoin(data_dir, 'flickr8k', 'vocab.json')):
    create_dataset(data_dir)

这段代码的作用就是构建词典vocab.json和创建数据集文件 train_data.json、 val_data.json、 test_data.json，词典里面是对文本的编码，数据集文件里面有IMAGES和CAPTIONS，IMAGES存放的全是图片的路径，而CAPTIONS字段下存放的是与每张图片对应的文本描述的编码，一张图片有5段描述，所以0-5都是第一张图片的文本描述编码，6-10是第二张图片的文本描述编码，以此类推。

查看：

# 读取词典和验证集
with open(pjoin(data_dir, 'flickr8k/vocab.json'), 'r') as f:
    vocab = json.load(f)
vocab_idx2word = {idx:word for word,idx in vocab.items()}
with open(pjoin(data_dir, 'flickr8k/val_data.json'), 'r') as f:
    data = json.load(f)

# 展示第12张图片，其对应的文本描述序号是60到64
content_img = Image.open(data['IMAGES'][12])
plt.imshow(content_img)
for i in range(5):
    print(' '.join([vocab_idx2word[word_idx] for word_idx in data['CAPTIONS'][12*5+i]]))

输出>：

a dog on a leash shakes while in some water
a dog standing in shallow water on a red leash
a dog splashes in the murky water
a black dog is shaking water off his body
black dog in the water shaking the water off of him

3)定义数据集类

在准备好的数据集的基础上，我们需要进一步定义Paddle Dataset类，以使用Paddle DataLoader类按批次产生数据。Paddle中仅预先定义了图像、文本和语音的单模态任务中常见的数据集类。因此，我们需要定义自己的数据集类。

在Paddle中定义数据集类非常简单，仅需要继承paddle.io.Dataset类，并实现__getitem__和__len__两个函数即可。

from argparse import Namespace 
import numpy as np

import paddle
from paddle.io import Dataset, BatchSampler, DataLoader
from paddle.vision import transforms 

class ImageTextDataset(Dataset):
    """
    Paddle数据类，用于Paddle DataLoader来按批次产生数据
    """

    def __init__(self, dataset_path, vocab_path, split, captions_per_image=5, max_len=30, transform=None):
        """
        参数：
            dataset_path：json格式数据文件路径
            vocab_path：json格式词典文件路径
            split：train、val、test
            captions_per_image：每张图片对应的文本描述数
            max_len：文本描述包含的最大单词数
            transform: 图像预处理方法
        """
        self.split = split
        assert self.split in {'train', 'val', 'test'}
        self.cpi = captions_per_image
        self.max_len = max_len

        # 载入数据集
        with open(dataset_path, 'r') as f:
            self.data = json.load(f)
        # 载入词典
        with open(vocab_path, 'r') as f:
            self.vocab = json.load(f)

        # PyTorch图像预处理流程
        self.transform = transform

        # Total number of datapoints
        self.dataset_size = len(self.data['CAPTIONS'])

    def __getitem__(self, i):
        # 第i个文本描述对应第(i // captions_per_image)张图片
        img = Image.open(self.data['IMAGES'][i // self.cpi]).convert('RGB')
        if self.transform is not None:
            img = self.transform(img)

        caplen = len(self.data['CAPTIONS'][i])
        caption = paddle.to_tensor(self.data['CAPTIONS'][i]+ [self.vocab['<pad>']] * (self.max_len + 2 - caplen), dtype='int64')

        return img, caption, caplen
        

    def __len__(self):
        return self.dataset_size

4)批量读取数据

利用刚才构造的数据集类，借助DataLoader类构建能够按批次产生训练、验证和测试数据的对象。

def mktrainval(data_dir, vocab_path, batch_size, workers=0):
    train_tx = transforms.Compose([
        transforms.Resize(256),
        transforms.RandomCrop(224),
        transforms.ToTensor(),
        transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225])
    ])
    val_tx = transforms.Compose([
        transforms.Resize(256),
        transforms.CenterCrop(224),
        transforms.ToTensor(),
        transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225])
    ])
    
    train_set = ImageTextDataset(os.path.join(data_dir, 'train_data.json'), 
                                 vocab_path, 'train',  transform=train_tx)
    valid_set = ImageTextDataset(os.path.join(data_dir, 'val_data.json'), 
                                 vocab_path, 'val', transform=val_tx)
    test_set = ImageTextDataset(os.path.join(data_dir, 'test_data.json'), 
                                 vocab_path, 'test', transform=val_tx)

    train_loader = DataLoader(
        train_set, batch_size=batch_size, shuffle=True, num_workers=workers)

    valid_loader = DataLoader(
        valid_set, batch_size=batch_size, shuffle=False, num_workers=workers, drop_last=False)
    
    test_loader = DataLoader(
        test_set, batch_size=batch_size, shuffle=False, num_workers=workers, drop_last=False)

    return train_loader, valid_loader, test_loader 

3.2定义模型

如图所示，VSE++模型是由图像表示提取器和文本表示提取构成，二者将图像和文本映射到对应表示空间。其中，图像表示提取器为在ImageNet数据集上预训练的VGG19或ResNet-152，VGG19和ResNet-152分别输出4096维和2048维的图像特征；文本表示提取器为GRU模型。

1)图像表示提取器

这里使用在ImageNet数据集上预训练过的两个分类模型ResNet-152和VGG19作为图像表示提取器，二者都需要更改其最后一个全连接层（分类层），以输出符合对应表示空间维度的图像表示。需要注意的是，这里对图像表示进行了长度归一化。

from paddle.vision import models
import paddle.nn as nn

class ImageRepExtractor(nn.Layer):
    def __init__(self, embed_size, pretrained_model='resnet152', finetuned=True):
        """
        参数：
            embed_size：对应表示维度
            pretrained_model：图像表示提取器，resnet152或vgg19
            finetuned：是否微调图像表示提取器的参数
        """
        super(ImageRepExtractor, self).__init__()
        if pretrained_model == 'resnet152':
            net = models.resnet152(pretrained=True)
            net = nn.Sequential(*list(net.children())[:-1])
            for param in net.parameters():
                param.requires_grad = finetuned
            fc = nn.Linear(2048, embed_size)
            nn.initializer.XavierNormal(fc.weight)
        elif pretrained_model == 'vgg19':
            net = models.vgg19(pretrained=True)
            net.classifier = nn.Sequential(*list(net.classifier.children())[:-1])
            for param in net.parameters():
                param.requires_grad = finetuned
            # 更改最后一层（fc层）
            fc = nn.Linear(4096, embed_size)
            nn.initializer.XavierNormal(fc.weight)
        else:
            raise ValueError("Unknown image model " + pretrained_model)
        self.net = net
        self.fc = fc

    def forward(self, x):
        out = self.net(x).squeeze()
        out = self.fc(out)
        out = nn.functional.normalize(out)
        return out

2)文本表示提取器

这里使用GRU模型作为文本表示提取器，它的输入层为词嵌入形式，文本表示为最后一个词对应的隐藏层输出。文本表示的维度也和对应表示空间的维度相同且也进行了长度归一化。

class TextRepExtractor(nn.Layer):
    def __init__(self, vocab_size, word_dim, embed_size, num_layers):
        """
        参数：
            vocab_size：词典大小
            word_dim：词嵌入维度
            embed_size：对应表示维度，也是RNN隐藏层维度
            num_layers：RNN隐藏层数
        """
        super(TextRepExtractor, self).__init__()
        self.embed_size = embed_size
        self.embed = nn.Embedding(vocab_size, word_dim, weight_attr=nn.initializer.Uniform(low=-0.1, high=0.1))
        # RNN默认已初始化
        self.rnn = nn.GRU(word_dim, embed_size, num_layers)
        
    def forward(self, x, lengths):
        x = self.embed(x)
        # 执行GRU的前馈过程会返回两个变量，第二个变量hidden为最后一个词（由length决定）对应的所有隐藏层输出
        output, hidden = self.rnn(x, None, lengths)
        # 最后一个词的最后一个隐藏层输出为hidden[-1]
        out = nn.functional.normalize(hidden[-1])
        return out

3)VSE++模型

有了图像表示提取器和文本表示提取器，我们就很容易构建VSE++模型了。仅需要利用图像表示提取器和文本表示提取器对成对的图像和文本数据输出表示即可。

class VSEPP(nn.Layer):
    def __init__(self, vocab_size, word_dim, embed_size, num_layers, image_model, finetuned=True):
        """
        参数：
            vocab_size: 词表大小
            word_dim: 词嵌入维度
            embed_size: 对应表示维度，也是RNN隐藏层维度
            num_layers: RNN隐藏层数
            image_model: 图像表示提取器，resnet152或vgg19
            finetuned: 是否微调图像表示提取器的参数
        """
        super(VSEPP, self).__init__()
        self.image_extractor = ImageRepExtractor(embed_size, image_model, finetuned)
        self.text_extractor = TextRepExtractor(vocab_size, word_dim, embed_size, num_layers)

    def forward(self, images, captions, cap_lens):
        image_code = self.image_extractor(images)
        text_code = self.text_extractor(captions, cap_lens)
        return image_code, text_code

3.3定义损失函数????

对于对应表示法，按照优化目标的不同，可分为

• 基于重构损失的方法
• 基于排序损失的方法
• 基于对抗损失的方法

对于基于排序损失的方法，有两种，一种是Triplet排序损失，一种是N-pair损失。以图检文为例，Triplet排序损失将计算图像与匹配文本与不匹配文本之间的值，要使得图像与匹配文本之间的值大于图像不匹配文本之间的值；而N-pair损失计算的是图像与所有文本之间的值。

上一篇： 论文 | 通过提示进行模型调整使 NLP 模型具有逆向鲁棒性

下一篇： cherry-markdown 开源 markdown 组件详细教程 - 研究