百度360必应搜狗淘宝本站头条
当前位置:网站首页 > 技术分类 > 正文

利用PyTorch的三元组损失Hard Triplet Loss进行嵌入模型微调

ztj100 2024-12-01 07:00 15 浏览 0 评论

本文介绍如何使用 PyTorch 和三元组边缘损失 (Triplet Margin Loss) 微调嵌入模型,并重点阐述实现细节和代码示例。三元组损失是一种对比损失函数,通过缩小锚点与正例间的距离,同时扩大锚点与负例间的距离来优化模型。

数据集准备与处理

一般的嵌入模型都会使用Sentence Transformer ,其中的 encode() 方法可以直接处理文本输入。但是为了进行微调,我们需要采用 Transformer 库,所以就要将文本转换为模型可接受的 token IDs 和 attention masks。Token IDs 代表模型词汇表中的词或字符,attention masks 用于防止模型关注填充 tokens。

本文使用 thenlper/gte-base 模型,需要对应的 tokenizer 对文本进行预处理。该模型基于 BertModel 架构:

BertModel(
(embeddings): BertEmbeddings(
(word_embeddings): Embedding(30522, 768, padding_idx=0)
(position_embeddings): Embedding(512, 768)
(token_type_embeddings): Embedding(2, 768)
(LayerNorm): LayerNorm((768,), eps=1e-12, elementwise_affine=True)
(dropout): Dropout(p=0.1, inplace=False)
)
(encoder): BertEncoder(
(layer): ModuleList(
(0-11): 12 x BertLayer(
(attention): BertAttention(
(self): BertSdpaSelfAttention(
(query): Linear(in_features=768, out_features=768, bias=True)
(key): Linear(in_features=768, out_features=768, bias=True)
(value): Linear(in_features=768, out_features=768, bias=True)
(dropout): Dropout(p=0.1, inplace=False)
)
(output): BertSelfOutput(
(dense): Linear(in_features=768, out_features=768, bias=True)
(LayerNorm): LayerNorm((768,), eps=1e-12, elementwise_affine=True)
(dropout): Dropout(p=0.1, inplace=False)
)
)
(intermediate): BertIntermediate(
(dense): Linear(in_features=768, out_features=3072, bias=True)
(intermediate_act_fn): GELUActivation()
)
(output): BertOutput(
(dense): Linear(in_features=3072, out_features=768, bias=True)
(LayerNorm): LayerNorm((768,), eps=1e-12, elementwise_affine=True)
(dropout): Dropout(p=0.1, inplace=False)
)
)
)
)
(pooler): BertPooler(
(dense): Linear(in_features=768, out_features=768, bias=True)
(activation): Tanh()
)
)

利用 Transformers 库的 AutoTokenizer 和 AutoModel 可以简化模型加载过程,无需手动处理底层架构和配置细节。

from transformers import AutoTokenizer, AutoModel 
from tqdm import tqdm 
tokenizer = AutoTokenizer.from_pretrained("thenlper/gte-base") 

# 获取文本并进行标记 
train_texts = [df_train.loc[i]['content'] for i in range(df_train.shape[0])] 
dev_texts = [df_dev.loc[i]['content'] for i in range(df_dev.shape[0])] 
test_texts = [df_test.loc[i]['content'] for i in range(df_test.shape[0])] 

train_tokens = [] 
train_attention_masks = [] 
dev_tokens = [] 
dev_attention_masks = [] 
test_tokens = [] 
test_attention_masks = [] 

for sent in tqdm(train_texts): 
encoding = tokenizer(sent, truncation=True, padding='max_length', return_tensors='pt') 
train_tokens.append(encoding['input_ids'].squeeze(0)) 
train_attention_masks.append(encoding['attention_mask'].squeeze(0)) 

for sent in tqdm(dev_texts): 
encoding = tokenizer(sent, truncation=True, padding='max_length', return_tensors='pt') 
dev_tokens.append(encoding['input_ids'].squeeze(0)) 
dev_attention_masks.append(encoding['attention_mask'].squeeze(0)) 

for sent in tqdm(test_texts): 
encoding = tokenizer(sent, truncation=True, padding='max_length', return_tensors='pt') 
test_tokens.append(encoding['input_ids'].squeeze(0)) 
test_attention_masks.append(encoding['attention_mask'].squeeze(0))

获取 token IDs 和 attention masks 后,需要将其存储并创建一个自定义的 PyTorch 数据集。

import random 
from collections import defaultdict 
import torch 
from torch.utils.data import Dataset, DataLoader, Sampler, SequentialSampler 

class CustomTripletDataset(Dataset): 
def __init__(self, tokens, attention_masks, labels): 
self.tokens = tokens 
self.attention_masks = attention_masks 
self.labels = torch.Tensor(labels) 
self.label_dict = defaultdict(list) 

for i in range(len(tokens)): 
self.label_dict[int(self.labels[i])].append(i) 
self.unique_classes = list(self.label_dict.keys()) 

def __len__(self): 
return len(self.tokens) 

def __getitem__(self, index): 
ids = self.tokens[index].to(device) 
ams = self.attention_masks[index].to(device) 
y = self.labels[index].to(device) 
return ids, ams, y

由于采用三元组损失,需要从数据集中采样正例和负例。label_dict 字典用于存储每个类别及其对应的数据索引,方便随机采样。DataLoader 用于加载数据集:

train_loader = DataLoader(train_dataset, batch_sampler=train_batch_sampler)

其中 train_batch_sampler 是自定义的批次采样器:

class CustomBatchSampler(SequentialSampler): 
def __init__(self, dataset, batch_size): 
self.dataset = dataset 
self.batch_size = batch_size 
self.unique_classes = sorted(dataset.unique_classes) 
self.label_dict = dataset.label_dict 
self.num_batches = len(self.dataset) // self.batch_size 
self.class_size = self.batch_size // 4 

def __iter__(self): 
total_samples_used = 0 
weights = np.repeat(1, len(self.unique_classes)) 

while total_samples_used < len(self.dataset): 
batch = [] 
classes = [] 
for _ in range(4): 
next_selected_class = self._select_class(weights) 
while next_selected_class in classes: 
next_selected_class = self._select_class(weights) 
weights[next_selected_class] += 1 
classes.append(next_selected_class) 
new_choices = self.label_dict[next_selected_class] 
remaining_samples = list(np.random.choice(new_choices, min(self.class_size, len(new_choices)), replace=False)) 
batch.extend(remaining_samples) 

total_samples_used += len(batch) 

yield batch 

def _select_class(self, weights): 
dist = 1/weights 
dist = dist/np.sum(dist) 
selected = int(np.random.choice(self.unique_classes, p=dist)) 
return selected 

def __len__(self): 
return self.num_batches

自定义批次采样器控制训练批次的构成,本文的实现确保每个批次包含 4 个类别,每个类别包含 8 个数据点。验证采样器则确保验证集批次在不同 epoch 间保持一致。

模型构建

嵌入模型通常基于 Transformer 架构,输出每个 token 的嵌入。为了获得句子嵌入,需要对 token 嵌入进行汇总。常用的方法包括 CLS 池化和平均池化。本文使用的 gte-base 模型采用平均池化,需要从模型输出中提取 token 嵌入并计算平均值。

import torch.nn.functional as F 
import torch.nn as nn 

class EmbeddingModel(nn.Module): 
def __init__(self, base_model): 
super().__init__() 
self.base_model = base_model 

def average_pool(self, last_hidden_states, attention_mask): 
# 平均 token 嵌入 
last_hidden = last_hidden_states.masked_fill(~attention_mask[..., None].bool(), 0.0) 
return last_hidden.sum(dim=1) / attention_mask.sum(dim=1)[..., None] 

def forward(self, input_ids, attention_mask): 
outputs = self.base_model(input_ids=input_ids, attention_mask=attention_mask) 
last_hidden_state = outputs.last_hidden_state 
pooled_output = self.average_pool(last_hidden_state, attention_mask) 
normalized_output = F.normalize(pooled_output, p=2, dim=1) 
return normalized_output 

base_model = AutoModel.from_pretrained("thenlper/gte-base") 
model = EmbeddingModel(base_model)

EmbeddingModel 类封装了 Hugging Face 模型,并实现了平均池化和嵌入归一化。

模型训练

训练循环中需要动态计算每个锚点的最难正例和最难负例。

import numpy as np 

def train(model, train_loader, criterion, optimizer, scheduler): 
model.train() 
epoch_train_losses = [] 

for idx, (ids, attention_masks, labels) in enumerate(train_loader): 
optimizer.zero_grad() 

embeddings = model(ids, attention_masks) 

distance_matrix = torch.cdist(embeddings, embeddings, p=2) # 创建方形距离矩阵 

anchors = [] 
positives = [] 
negatives = [] 

for i in range(len(labels)): 

anchor_label = labels[i].item() 
anchor_distance = distance_matrix[i] # 锚点与所有其他点之间的距离 

# 最难的正例(同一类别中最远的) 
hardest_positive_idx = (labels == anchor_label).nonzero(as_tuple=True)[0] # 所有同类索引 
hardest_positive_idx = hardest_positive_idx[hardest_positive_idx != i] # 排除自己的标签 
hardest_positive = hardest_positive_idx[anchor_distance[hardest_positive_idx].argmax()] # 最远同类的标签 

# 最难的负例(不同类别中最近的) 
hardest_negative_idx = (labels != anchor_label).nonzero(as_tuple=True)[0] # 所有不同类索引 
hardest_negative = hardest_negative_idx[anchor_distance[hardest_negative_idx].argmin()] # 最近不同类的标签 

# 加载选择的 
anchors.append(embeddings[i]) 
positives.append(embeddings[hardest_positive]) 
negatives.append(embeddings[hardest_negative]) 

# 将列表转换为张量 
anchors = torch.stack(anchors) 
positives = torch.stack(positives) 
negatives = torch.stack(negatives) 

# 计算损失 
loss = criterion(anchors, positives, negatives) 
epoch_train_losses.append(loss.item()) 

# 反向传播和优化 
loss.backward() 
optimizer.step() 

# 更新学习率 
scheduler.step() 

return np.mean(epoch_train_losses)

训练过程中使用 torch.cdist() 计算嵌入间的距离矩阵,并根据距离选择最难正例和最难负例。PyTorch 的 TripletMarginLoss 用于计算损失。

结论与讨论

实践表明,Batch Hard Triplet Loss 在某些情况下并非最优选择。例如,当正例样本内部差异较大时,强制其嵌入相似可能适得其反。

本文的重点在于 PyTorch 中自定义批次采样和动态距离计算的实现。

对于某些任务,直接在分类任务上微调嵌入模型可能比使用三元组损失更有效。

相关推荐

人生苦短,我要在VSCode里面用Python

轻沉发自浅度寺量子位出品|公众号QbitAI在程序员圈子里,VisualStudioCode(以下简称VSCode)可以说是目前最火的代码编辑器之一了。它是微软出品的一款可扩展的轻量...

亲测可用:Pycharm2019.3专业版永久激活教程

概述随着2020年的到来,又有一批Pycharm的激活码到期了,各位同仁估计也是在到处搜索激活方案,在这里,笔者为大家收录了一个永久激活的方案,亲测可用,欢迎下载尝试:免责声明本项目只做个人学习研究之...

Python新手入门很简单(python教程入门)

我之前学习python走过很多的歧途,自学永远都是瞎猫碰死耗子一样,毫无头绪。后来心里一直都有一个做头条知识分享的梦,希望自己能够帮助曾经类似自己的人,于是我来了,每天更新5篇Python文章,喜欢的...

Pycharm的设置和基本使用(pycharm运行设置)

这篇文章,主要是针对刚开始学习python语言,不怎么会使用pycharm的童鞋们;我来带领大家详细了解下pycharm页面及常用的一些功能,让大家能通过此篇文章能快速的开始编写python代码。一...

依旧是25年最拔尖的PyTorch实用教程!堪比付费级内容!

我真的想知道作者到底咋把PyTorch教程整得这么牛的啊?明明在内容上已经足以成为付费教材了,但作者偏要免费开源给大家学习!...

手把手教你 在Pytorch框架上部署和测试关键点人脸检测项目DBFace

这期教向大家介绍仅仅1.3M的轻量级高精度的关键点人脸检测模型DBFace,并手把手教你如何在自己的电脑端进行部署和测试运行,运行时bug解决。01.前言前段时间DBFace人脸检测库横空出世,...

进入Python的世界02外篇-Pycharm配置Pyqt6

为什么这样配置,要开发带UI的python也只能这样了,安装过程如下:一安装工具打开终端:pipinstallPyQt6PyQt6-tools二打开设置并汉化点击plugin,安装汉化插件,...

vs code如何配置使用Anaconda(vscode调用anaconda库)

上一篇文章中(Anaconda使用完全指南),我们能介绍了Anaconda的安装和使用,以及如何在pycharm中配置Anaconda。本篇,将继续介绍在vscode中配置conda...

pycharm中conda解释器无法配置(pycharm配置anaconda解释器)

之前用的好好的pycharm正常配置解释器突然不能用了?可以显示有这个环境然后确认后可以conda正在配置解释器,但是进度条结束后还是不成功!!试过了pycharm重启,pycharm重装,anaco...

Volta:跨平台开发者的福音,统一前端js工具链从未如此简单!

我们都知道现在已经进入了Rust时代,不仅很多终端常用的工具都被rust重写了,而且现在很多前端工具也开始被Rust接手了,这不,现在就出现了一款JS工具管理工具,有了它,你可以管理多版本的js工具,...

开发者的福音,ElectronEgg: 新一代桌面应用开发框架

今天给大家介绍一个开源项目electron-egg。如果你是一个JS的前端开发人员,以前面对这项任务桌面应用开发在时,可能会感到无从下手,甚至觉得这是一项困难的挑战。ElectronEgg的出现,它能...

超强经得起考验的低代码开发平台Frappe

#挑战30天在头条写日记#开始进行管理软件的开发来讲,如果从头做起不是不可以,但选择一款免费的且经得起时间考验的低代码开发平台是非常有必要的,将大幅提升代码的质量、加快开发的效率、以及提高程序的扩展性...

一文带你搞懂Vue3 底层源码(vue3核心源码解析)

作者:妹红大大转发链接:https://mp.weixin.qq.com/s/D_PRIMAD6i225Pn-a_lzPA前言vue3出来有一段时间了。今天正式开始记录一下梗vue3.0.0-be...

Windows 11 + WSL2 打造轻量级 Linux 本地开发环境实战教程

一、前言...

基于小程序 DSL(微信、支付宝)的,可扩展的多端研发框架

Mor(发音为/mr/,类似more),是饿了么开发的一款基于小程序DSL的,可扩展的多端研发框架,使用小程序原生DSL构建,使用者只需书写一套(微信或支付宝)小程序,就可以通过Mor...

取消回复欢迎 发表评论: