【NLP 33、实践 ⑦ 基于Triple Loss作表示型文本匹配】
我咽下春天的花种,于是心里繁华连绵
—— 25.3.6
🚀 随机三个样本,两个相似和一个不相似的样本,基于Triple Loss的训练方式
源代码及数据:通过网盘分享的文件:使用tripletloss训练表示型文本匹配模型
链接: https://pan.baidu.com/s/1Jga9MGgio3rhCZVyaPI0GA?pwd=8i88 提取码: 8i88
--来自百度网盘超级会员v3的分享
一、配置文件 config.py
model_path:模型保存或加载的路径,训练时,模型会保存到该路径;推理或继续训练时,模型会从该路径加载
schema_path:定义标签(类别)与索引之间的映射关系,通常用于分类任务中,将文本数据对应的类别标签转换为模型可以处理的数值形式(如索引)
train_data_path:训练数据文件的路径
valid_data_path:验证数据文件的路径,指定验证数据的来源,用于评估模型在训练过程中的性能
vocab_path:词汇表文件的路径,词汇表通常包含模型使用的所有单词或字符及其对应的索引
max_length:输入序列的最大长度,对输入文本进行截断或填充,使其长度统一,确保模型输入的一致性
hidden_size:模型隐藏层的大小(神经元数量),控制模型的容量和复杂度;隐藏层越大,模型表达能力越强,但计算成本也越高
epoch:训练的轮数,定义模型在整个训练数据集上迭代的次数;增加 epoch 可以提高模型性能,但可能导致过拟合
batch_size:每次训练时输入模型的样本数量,控制训练过程中的内存使用和计算效率;较大的 batch_size 可以加速训练,但需要更多内存
epoch_data_size:每轮训练中使用的数据量,数据集较大,可以限制每轮训练使用的数据量,以加速训练
positive_sample_rate:正样本在训练数据中的比例,分类任务中,控制正样本和负样本的比例,避免类别不平衡问题
optimizer:优化器的类型,定义模型如何更新参数以最小化损失函数
learning_rate:学习率,控制模型参数更新的步长;学习率过大会导致训练不稳定,过小会导致训练缓慢
# -*- coding: utf-8 -*-
"""
配置参数信息
"""
Config = {
"model_path": "model_output",
"schema_path": r"F:\人工智能NLP\NLP\HomeWork\demo7.1_使用tripletloss训练表示型文本匹配模型\data\schema.json",
"train_data_path": r"F:\人工智能NLP\NLP\HomeWork\demo7.1_使用tripletloss训练表示型文本匹配模型\data\train.json",
"valid_data_path": r"F:\人工智能NLP\NLP\HomeWork\demo7.1_使用tripletloss训练表示型文本匹配模型\data\valid.json",
"vocab_path": r"F:\人工智能NLP\NLP\HomeWork\demo7.1_使用tripletloss训练表示型文本匹配模型\chars.txt",
"max_length": 20,
"hidden_size": 128,
"epoch": 10,
"batch_size": 32,
"epoch_data_size": 200, #每轮训练中采样数量
"positive_sample_rate": 0.5, #正样本比例
"optimizer": "adam",
"learning_rate": 1e-3,
}
二、 数据加载文件 loader.py
1.加载数据
Ⅰ、加载字表或词表
加载词汇表,将每个词或字符映射为唯一的索引。
open():打开文件,返回一个文件对象,用于读取或写入文件。
参数名 | 类型 | 描述 |
---|---|---|
file | str | 文件路径。 |
mode | str | 文件打开模式(如 "r" 读取,"w" 写入,"a" 追加,默认 "r" )。 |
encoding | str | 文件编码(如 "utf8" ,默认 None )。 |
errors | str | 指定编码错误的处理方式(如 "ignore" ,默认 None )。 |
newline | str | 控制换行符的行为(如 "\n" ,默认 None )。 |
buffering | int | 设置缓冲策略(如 -1 系统默认,0 无缓冲,1 行缓冲,默认 -1 )。 |
closefd | bool | 是否关闭文件描述符(默认 True )。 |
opener | callable | 自定义文件打开器(默认 None )。 |
enumerate():返回一个枚举对象,将可迭代对象(如列表、字符串)转换为索引和值的组合。
参数名 | 类型 | 描述 |
---|---|---|
iterable | iterable | 可迭代对象(如列表、字符串)。 |
start | int | 索引的起始值(默认 0 )。 |
strip():去除字符串首尾的空白字符(如空格、换行符)或指定字符。
参数名 | 类型 | 描述 |
---|---|---|
chars | str | 指定要删除的字符(默认去除空白字符)。 |
#加载字表或词表
def load_vocab(vocab_path):
token_dict = {}
with open(vocab_path, encoding="utf8") as f:
for index, line in enumerate(f):
token = line.strip()
token_dict[token] = index + 1 #0留给padding位置,所以从1开始
return token_dict
Ⅱ、加载标签映射表
加载标签映射表(schema),将标签映射为索引。
open():打开文件,返回一个文件对象,用于读取或写入文件。
参数名 | 类型 | 描述 |
---|---|---|
file | str | 文件路径。 |
mode | str | 文件打开模式(如 "r" 读取,"w" 写入,"a" 追加,默认 "r" )。 |
encoding | str | 文件编码(如 "utf8" ,默认 None )。 |
errors | str | 指定编码错误的处理方式(如 "ignore" ,默认 None )。 |
newline | str | 控制换行符的行为(如 "\n" ,默认 None )。 |
buffering | int | 设置缓冲策略(如 -1 系统默认,0 无缓冲,1 行缓冲,默认 -1 )。 |
closefd | bool | 是否关闭文件描述符(默认 True )。 |
opener | callable | 自定义文件打开器(默认 None )。 |
json.loads():将 JSON 格式的字符串解析为 Python 对象(如字典、列表)。
参数名 | 类型 | 描述 |
---|---|---|
s | str | JSON 格式的字符串。 |
cls | class | 自定义 JSON 解码器(默认 None )。 |
object_hook | callable | 自定义对象解码函数(默认 None )。 |
parse_float | callable | 自定义浮点数解码函数(默认 None )。 |
parse_int | callable | 自定义整数解码函数(默认 None )。 |
parse_constant | callable | 自定义常量解码函数(默认 None )。 |
object_pairs_hook | callable | 自定义键值对解码函数(默认 None )。 |
文件对象.read():从文件对象中读取全部内容,返回一个字符串(文本文件)或字节对象(二进制文件)。
参数名 | 类型 | 描述 |
---|---|---|
size | int | 可选参数,指定读取的字节数(文本文件为字符数)。如果未指定,则读取全部内容。 |
#加载schema
def load_schema(schema_path):
with open(schema_path, encoding="utf8") as f:
return json.loads(f.read())
Ⅲ、封装数据
使用 PyTorch 的
DataLoader
封装数据,方便批量加载
DataGenerator():调用 DataGenerator(data_path, config)
,传入数据路径和配置,创建数据生成器对象 dg
。
DataLoader():将数据集封装为 PyTorch 的 DataLoader
,支持批量加载、打乱数据顺序等功能。
参数名 | 类型 | 描述 |
---|---|---|
dataset | Dataset | 数据集对象(如 DataGenerator )。 |
batch_size | int | 每个批次的大小(默认 1 )。 |
shuffle | bool | 是否打乱数据顺序(默认 False )。 |
sampler | Sampler | 自定义采样器(默认 None )。 |
batch_sampler | Sampler | 自定义批次采样器(默认 None )。 |
num_workers | int | 数据加载的线程数(默认 0 ,表示在主线程中加载)。 |
collate_fn | callable | 自定义批次数据处理函数(默认 None )。 |
pin_memory | bool | 是否将数据加载到 GPU 的固定内存中(默认 False )。 |
drop_last | bool | 是否丢弃最后一个不完整的批次(默认 False )。 |
timeout | int | 数据加载的超时时间(默认 0 ,表示不超时)。 |
worker_init_fn | callable | 自定义工作线程初始化函数(默认 None )。 |
#用torch自带的DataLoader类封装数据
def load_data(data_path, config, shuffle=True):
dg = DataGenerator(data_path, config)
dl = DataLoader(dg, batch_size=config["batch_size"], shuffle=shuffle)
return dl
2.处理数据
Ⅰ、补齐或截断
截断序列:如果输入序列的长度超过
max_length
,则截断超出部分。补齐序列:如果输入序列的长度小于
max_length
,则在末尾填充0
,使其长度达到max_length
。
len():用于返回对象的长度(元素的数量)。它适用于多种 Python 对象,包括字符串、列表、元组、字典、集合等。
参数名 | 类型 | 描述 |
---|---|---|
obj | object | 需要计算长度的对象,可以是字符串、列表、元组、字典、集合等。 |
#补齐或截断输入的序列,使其可以在一个batch内运算
def padding(self, input_id):
input_id = input_id[:self.config["max_length"]]
input_id += [0] * (self.config["max_length"] - len(input_id))
return input_id
Ⅱ、定义类的特殊方法
① 返回数据集大小
返回数据集的大小(即数据集中样本的数量)
assert: Python 中的一个关键字,用于断言某个条件是否为真。如果条件为真,程序继续执行;如果条件为假,则抛出 AssertionError
异常,并可选地输出一条错误信息。
参数名 | 类型 | 是否必选 | 描述 |
---|---|---|---|
condition | 布尔表达式 | 是 | 需要检查的条件。如果为 False ,则触发 AssertionError 异常。 |
message | 字符串 | 否 | 可选参数,当条件为 False 时,输出的错误信息。 |
def __len__(self):
if self.data_type == "train":
return self.config["epoch_data_size"]
else:
assert self.data_type == "test", self.data_type
return len(self.data)
② 生成随机训练样本
standard_question_index:所有意图的列表
random.sample():从指定的序列中随机选择 不重复 的多个元素,返回一个列表。
参数名 | 类型 | 是否必选 | 描述 |
---|---|---|---|
population | 序列(列表、元组、集合等) | 是 | 从中随机选择元素的序列。 |
k | 整数 | 是 | 需要选择的元素数量。必须小于或等于 population 的长度。 |
random.choice():从指定的序列中随机选择 一个 元素。
参数名 | 类型 | 是否必选 | 描述 |
---|---|---|---|
seq | 序列(列表、元组、字符串等) | 是 | 从中随机选择元素的序列。 |
#随机生成3元组样本,2正1负
def random_train_sample(self):
standard_question_index = list(self.knwb.keys())
# 先选定两个意图,之后从第一个意图中取2个问题,第二个意图中取一个问题
p, n = random.sample(standard_question_index, 2)
# 如果某个意图下刚好只有一条问题,那只能两个正样本用一样的;
# 这种对训练没帮助,因为相同的样本距离肯定是0,但是数据充分的情况下这种情况很少
if len(self.knwb[p]) == 1:
s1 = s2 = self.knwb[p][0]
#这应当是一般情况
else:
s1, s2 = random.sample(self.knwb[p], 2)
# 随机一个负样本
s3 = random.choice(self.knwb[n])
# 前2个相似,后1个不相似,不需要额外在输入一个0或1的label,这与一般的loss计算不同
return [s1, s2, s3]
③ 根据索引返回样本
根据给定的索引
index
返回数据集中的一个样本
self.random_train_sample():生成随机训练样本
def __getitem__(self, index):
if self.data_type == "train":
return self.random_train_sample() #随机生成一个训练样本
else:
return self.data[index]
Ⅲ、加载和处理训练样本和测试样本
- 加载数据:从指定路径的文件中逐行读取数据。
- 区分训练集和测试集:
- 训练集数据格式为字典(
dict
),包含questions
和target
字段。- 测试集数据格式为列表(
list
),包含question
和label
。- 数据预处理:
- 将文本数据编码为索引序列。
- 将标签映射为索引。
- 将处理后的数据存储到
self.knwb
(训练集)或self.data
(测试集)中。
data:用于存储测试集数据
knwb:知识库数据的存储对象,通常是一个字典,键是标准问题的索引,值是对应的问题 ID 列表。
参数名 | 类型 | 描述 |
---|---|---|
default_factory | 可调用对象或类型 | 默认工厂函数,用于生成默认值。如果未提供,默认为 None ,此时行为与普通字典相同。 |
**kwargs | 关键字参数 | 其他参数会传递给 dict 的构造函数,用于初始化字典内容。 |
defaultdict():Python 中 collections
模块提供的一个字典子类,它的主要作用是在访问不存在的键时返回一个默认值,而不是抛出 KeyError
异常。
open():打开文件,返回一个文件对象,用于读取或写入文件。
参数名 | 类型 | 描述 |
---|---|---|
file | str | 文件路径。 |
mode | str | 文件打开模式(如 "r" 读取,"w" 写入,"a" 追加,默认 "r" )。 |
encoding | str | 文件编码(如 "utf8" ,默认 None )。 |
errors | str | 指定编码错误的处理方式(如 "ignore" ,默认 None )。 |
newline | str | 控制换行符的行为(如 "\n" ,默认 None )。 |
buffering | int | 设置缓冲策略(如 -1 系统默认,0 无缓冲,1 行缓冲,默认 -1 )。 |
closefd | bool | 是否关闭文件描述符(默认 True )。 |
opener | callable | 自定义文件打开器(默认 None )。 |
json.loads():将 JSON 格式的字符串解析为 Python 对象(如字典、列表)。
参数名 | 类型 | 描述 |
---|---|---|
s | str | JSON 格式的字符串。 |
cls | class | 自定义 JSON 解码器(默认 None )。 |
object_hook | callable | 自定义对象解码函数(默认 None )。 |
parse_float | callable | 自定义浮点数解码函数(默认 None )。 |
parse_int | callable | 自定义整数解码函数(默认 None )。 |
parse_constant | callable | 自定义常量解码函数(默认 None )。 |
object_pairs_hook | callable | 自定义键值对解码函数(默认 None )。 |
def load(self):
self.data = []
self.knwb = defaultdict(list)
with open(self.path, encoding="utf8") as f:
for line in f:
line = json.loads(line)
#加载训练集
if isinstance(line, dict):
self.data_type = "train"
questions = line["questions"]
label = line["target"]
for question in questions:
input_id = self.encode_sentence(question)
input_id = torch.LongTensor(input_id)
self.knwb[self.schema[label]].append(input_id)
#加载测试集
else:
self.data_type = "test"
assert isinstance(line, list)
question, label = line
input_id = self.encode_sentence(question)
input_id = torch.LongTensor(input_id)
label_index = torch.LongTensor([self.schema[label]])
self.data.append([input_id, label_index])
return
Ⅳ、初始化数据加载器
self.config:将传入的配置字典 config
保存到实例变量中
self.path:将数据路径 data_path
保存到实例变量中
self.vocab:从配置文件中指定的路径加载词汇表,并将其保存到实例变量 self.vocab
中。
self.config:计算词汇表的大小,并将其更新到配置字典中。
self.schema:从配置文件中指定的标签映射模式(Schema),并将其保存到实例变量 self.schema
中
self.train_data_size:从配置文件中获取每个训练周期(epoch)的样本数量,并将其保存到实例变量中
self.data_type:初始化一个变量 self.data_type
,用于标识当前加载的是训练集还是测试集。它的值可以是 "train"
或 "test"
self.load():调用 load
方法,加载数据(训练集或测试集数据)。
def __init__(self, data_path, config):
self.config = config
self.path = data_path
self.vocab = load_vocab(config["vocab_path"])
self.config["vocab_size"] = len(self.vocab)
self.schema = load_schema(config["schema_path"])
self.train_data_size = config["epoch_data_size"] #由于采取随机采样,所以需要设定一个采样数量,否则可以一直采
self.data_type = None #用来标识加载的是训练集还是测试集 "train" or "test"
self.load()
完整代码
# -*- coding: utf-8 -*-
import json
import re
import os
import torch
import random
import jieba
import numpy as np
from torch.utils.data import Dataset, DataLoader
from collections import defaultdict
"""
数据加载
"""
class DataGenerator:
def __init__(self, data_path, config):
self.config = config
self.path = data_path
self.vocab = load_vocab(config["vocab_path"])
self.config["vocab_size"] = len(self.vocab)
self.schema = load_schema(config["schema_path"])
self.train_data_size = config["epoch_data_size"] #由于采取随机采样,所以需要设定一个采样数量,否则可以一直采
self.data_type = None #用来标识加载的是训练集还是测试集 "train" or "test"
self.load()
def load(self):
self.data = []
self.knwb = defaultdict(list)
with open(self.path, encoding="utf8") as f:
for line in f:
line = json.loads(line)
#加载训练集
if isinstance(line, dict):
self.data_type = "train"
questions = line["questions"]
label = line["target"]
for question in questions:
input_id = self.encode_sentence(question)
input_id = torch.LongTensor(input_id)
self.knwb[self.schema[label]].append(input_id)
#加载测试集
else:
self.data_type = "test"
assert isinstance(line, list)
question, label = line
input_id = self.encode_sentence(question)
input_id = torch.LongTensor(input_id)
label_index = torch.LongTensor([self.schema[label]])
self.data.append([input_id, label_index])
return
def encode_sentence(self, text):
input_id = []
if self.config["vocab_path"] == "words.txt":
for word in jieba.cut(text):
input_id.append(self.vocab.get(word, self.vocab["[UNK]"]))
else:
for char in text:
input_id.append(self.vocab.get(char, self.vocab["[UNK]"]))
input_id = self.padding(input_id)
return input_id
#补齐或截断输入的序列,使其可以在一个batch内运算
def padding(self, input_id):
input_id = input_id[:self.config["max_length"]]
input_id += [0] * (self.config["max_length"] - len(input_id))
return input_id
def __len__(self):
if self.data_type == "train":
return self.config["epoch_data_size"]
else:
assert self.data_type == "test", self.data_type
return len(self.data)
def __getitem__(self, index):
if self.data_type == "train":
return self.random_train_sample() #随机生成一个训练样本
else:
return self.data[index]
#随机生成3元组样本,2正1负
def random_train_sample(self):
standard_question_index = list(self.knwb.keys())
# 先选定两个意图,之后从第一个意图中取2个问题,第二个意图中取一个问题
p, n = random.sample(standard_question_index, 2)
# 如果某个意图下刚好只有一条问题,那只能两个正样本用一样的;
# 这种对训练没帮助,因为相同的样本距离肯定是0,但是数据充分的情况下这种情况很少
if len(self.knwb[p]) == 1:
s1 = s2 = self.knwb[p][0]
#这应当是一般情况
else:
s1, s2 = random.sample(self.knwb[p], 2)
# 随机一个负样本
s3 = random.choice(self.knwb[n])
# 前2个相似,后1个不相似,不需要额外在输入一个0或1的label,这与一般的loss计算不同
return [s1, s2, s3]
#加载字表或词表
def load_vocab(vocab_path):
token_dict = {}
with open(vocab_path, encoding="utf8") as f:
for index, line in enumerate(f):
token = line.strip()
token_dict[token] = index + 1 #0留给padding位置,所以从1开始
return token_dict
#加载schema
def load_schema(schema_path):
with open(schema_path, encoding="utf8") as f:
return json.loads(f.read())
#用torch自带的DataLoader类封装数据
def load_data(data_path, config, shuffle=True):
dg = DataGenerator(data_path, config)
dl = DataLoader(dg, batch_size=config["batch_size"], shuffle=shuffle)
return dl
if __name__ == "__main__":
from config import Config
dg = DataGenerator("valid_tag_news.json", Config)
print(dg[1])
三、 模型定义文件 model.py
1.句子编码 SentenceEncoder
Ⅰ、模型初始化
config:配置字典
hidden_size:隐藏层大小
vocab_size:词汇表大小
max_length:句子的最大长度
nn.Embedding():将离散的索引(如字符编码或单词索引)映射为连续的向量表示。
参数名 | 类型 | 是否必选 | 描述 |
---|---|---|---|
num_embeddings | 整数 | 是 | 词汇表的大小(即索引的最大值 + 1)。 |
embedding_dim | 整数 | 是 | 嵌入向量的维度。 |
padding_idx | 整数 | 否 | 指定填充索引(如 0),该索引对应的向量会被固定为全 0。 |
max_norm | 浮点数 | 否 | 如果指定,嵌入向量会被归一化,使其范数不超过此值。 |
norm_type | 浮点数 | 否 | 归一化的范数类型(默认为 2.0,即 L2 范数)。 |
scale_grad_by_freq | 布尔值 | 否 | 是否根据频率缩放梯度(默认为 False )。 |
sparse | 布尔值 | 否 | 是否使用稀疏梯度(默认为 False )。 |
nn.Linear():实现线性变换:y = x * weight.T + bias
。常用于神经网络的全连接层。
参数名 | 类型 | 是否必选 | 描述 |
---|---|---|---|
in_features | 整数 | 是 | 输入特征的数量。 |
out_features | 整数 | 是 | 输出特征的数量。 |
bias | 布尔值 | 否 | 是否使用偏置项(默认为 True )。 |
nn.Dropout():在训练过程中随机将部分输入元素置为 0,以防止过拟合。常用于正则化。
参数名 | 类型 | 是否必选 | 描述 |
---|---|---|---|
p | 浮点数 | 是 | 元素被置为 0 的概率(默认为 0.5)。 |
def __init__(self, config):
super(SentenceEncoder, self).__init__()
hidden_size = config["hidden_size"]
vocab_size = config["vocab_size"] + 1
max_length = config["max_length"]
self.embedding = nn.Embedding(vocab_size, hidden_size, padding_idx=0)
# self.layer = nn.LSTM(hidden_size, hidden_size, batch_first=True, bidirectional=True)
self.layer = nn.Linear(hidden_size, hidden_size)
self.dropout = nn.Dropout(0.5)
Ⅱ、前向传播
x.gt():比较张量 x
中的元素是否大于给定值,返回布尔张量
参数名 | 类型 | 是否必选 | 描述 |
---|---|---|---|
value | 标量或张量 | 是 | 比较的阈值。 |
torch.sum():计算张量中元素的和
参数名 | 类型 | 是否必选 | 描述 |
---|---|---|---|
input | 张量 | 是 | 输入张量。 |
dim | 整数或元组 | 否 | 沿指定维度求和(默认为对所有元素求和)。 |
keepdim | 布尔值 | 否 | 是否保留求和后的维度(默认为 False ) |
nn.functional.max_pool1d():对输入张量进行 1D 最大池化操作,返回池化后的结果。常用于卷积神经网络中,降低特征图的维度。
参数名 | 类型 | 是否必选 | 描述 |
---|---|---|---|
input | 张量 | 是 | 输入张量,形状为 (batch_size, channels, length) 。 |
kernel_size | 整数或元组 | 是 | 池化窗口的大小。 |
stride | 整数或元组 | 否 | 池化窗口的步长(默认为 kernel_size )。 |
padding | 整数或元组 | 否 | 输入张量的填充大小(默认为 0)。 |
#输入为问题字符编码
def forward(self, x):
sentence_length = torch.sum(x.gt(0), dim=-1)
x = self.embedding(x)
#使用lstm
# x, _ = self.layer(x)
#使用线性层
x = self.layer(x)
x = nn.functional.max_pool1d(x.transpose(1, 2), x.shape[1]).squeeze()
return x
2.计算句子间相似度 SiameseNetwork
Ⅰ、模型初始化
初始化 Siamese Network 的结构
SentenceEncoder():初始化一个句子编码器,用于将句子编码为向量。
nn.CosineEmbeddingLoss():计算两个输入之间的 余弦相似度损失,用于训练模型学习相似性
参数名 | 类型 | 是否必选 | 描述 |
---|---|---|---|
margin | 浮点数 | 否 | 间隔参数(默认为 0.0 )。 |
reduction | 字符串 | 否 | 指定损失计算方式(默认为 'mean' ,可选 'sum' 或 'none' )。 |
def __init__(self, config):
super(SiameseNetwork, self).__init__()
self.sentence_encoder = SentenceEncoder(config)
self.loss = nn.CosineEmbeddingLoss()
Ⅱ、计算余弦距离
计算两个向量之间的余弦距离
cos=1时,两个向量相同,余弦距离为0;cos=0时,两个向量正交,余弦距离为1
torch.nn.functional.normalize():对输入张量沿指定维度进行归一化(默认使用 L2 归一化)
参数名 | 类型 | 是否必选 | 描述 |
---|---|---|---|
input | 张量 | 是 | 输入张量。 |
p | 浮点数 | 否 | 归一化的范数类型(默认为 2 ,即 L2 范数)。 |
dim | 整数 | 否 | 沿指定维度归一化(默认为 1 )。 |
eps | 浮点数 | 否 | 防止除以零的小值(默认为 1e-12 )。 |
torch.sum():计算张量中元素的和
参数名 | 类型 | 是否必选 | 描述 |
---|---|---|---|
input | 张量 | 是 | 输入张量。 |
dim | 整数或元组 | 否 | 沿指定维度求和。默认对所有元素求和,返回标量。 |
keepdim | 布尔值 | 否 | 是否保留求和后的维度(默认为 False )。 |
dtype | 数据类型 | 否 | 指定输出的数据类型(默认为 input.dtype )。 |
# 计算余弦距离 1-cos(a,b)
# cos=1时两个向量相同,余弦距离为0;cos=0时,两个向量正交,余弦距离为1
def cosine_distance(self, tensor1, tensor2):
tensor1 = torch.nn.functional.normalize(tensor1, dim=-1)
tensor2 = torch.nn.functional.normalize(tensor2, dim=-1)
cosine = torch.sum(torch.mul(tensor1, tensor2), axis=-1)
return 1 - cosine
Ⅲ、计算三元组损失
ap:锚点(anchor)和正样本(positive)之间的余弦距离
an:锚点(anchor)和负样本(negative)之间的余弦距离
margin:间隔参数,惩罚项 / 正则项
squeeze():移除张量中维度大小为 1 的轴。例如,将形状为 (3, 1, 4)
的张量压缩为 (3, 4)
参数名 | 类型 | 是否必选 | 描述 |
---|---|---|---|
dim | 整数 | 否 | 指定要移除的维度(必须为单维度轴)。默认移除所有单维度轴。 |
torch.mean():计算张量中元素的平均值
参数名 | 类型 | 是否必选 | 描述 |
---|---|---|---|
input | 张量 | 是 | 输入张量。 |
dim | 整数或元组 | 否 | 沿指定维度求平均值。默认对所有元素求平均,返回标量。 |
keepdim | 布尔值 | 否 | 是否保留求平均后的维度(默认为 False )。 |
dtype | 数据类型 | 否 | 指定输出的数据类型(默认为 input.dtype )。 |
x.gt():比较张量 x
中的元素是否大于给定值,返回布尔张量
参数名 | 类型 | 是否必选 | 描述 |
---|---|---|---|
value | 标量或张量 | 是 | 比较的阈值。 |
def cosine_triplet_loss(self, a, p, n, margin=None):
ap = self.cosine_distance(a, p)
an = self.cosine_distance(a, n)
if margin is None:
diff = ap - an + 0.1
else:
diff = ap - an + margin.squeeze()
return torch.mean(diff[diff.gt(0)])
Ⅳ、前向传播
- 如果传入 3 个句子,则计算三元组损失(Triplet Loss)。
- 如果传入 1 个句子,则返回该句子的编码向量。
#sentence : (batch_size, max_length)
def forward(self, sentence1, sentence2=None, sentence3=None):
#同时传入3个句子,则做tripletloss的loss计算
if sentence2 is not None and sentence3 is not None:
vector1 = self.sentence_encoder(sentence1)
vector2 = self.sentence_encoder(sentence2)
vector3 = self.sentence_encoder(sentence3)
return self.cosine_triplet_loss(vector1, vector2, vector3)
#单独传入一个句子时,认为正在使用向量化能力
else:
return self.sentence_encoder(sentence1)
3.选择优化器
Adam():是一种自适应学习率的优化算法,结合了动量法和自适应学习率的优点。它通过计算梯度的一阶矩(均值)和二阶矩(未中心化的方差)来动态调整每个参数的学习率,从而加速收敛并提高稳定性。
参数名 | 类型 | 是否必选 | 描述 |
---|---|---|---|
params | 可迭代对象 | 是 | 需要优化的模型参数(通常通过 model.parameters() 获取)。 |
lr | 浮点数 | 否 | 学习率(默认值:0.001 )。 |
betas | 元组 | 否 | 用于计算梯度一阶矩和二阶矩的衰减率(默认值:(0.9, 0.999) )。 |
eps | 浮点数 | 否 | 防止除零的小常数(默认值:1e-8 )。 |
weight_decay | 浮点数 | 否 | L2 正则化系数(默认值:0 )。 |
amsgrad | 布尔值 | 否 | 是否使用 AMSGrad 变体(默认值:False )。 |
SGD():是一种随机梯度下降优化算法。它通过计算损失函数关于模型参数的梯度来更新参数,从而最小化损失函数。SGD 是深度学习中最基础的优化算法之一。
参数名 | 类型 | 是否必选 | 描述 |
---|---|---|---|
params | 可迭代对象 | 是 | 需要优化的模型参数(通常通过 model.parameters() 获取)。 |
lr | 浮点数 | 是 | 学习率。 |
momentum | 浮点数 | 否 | 动量因子,用于加速收敛(默认值:0 )。 |
dampening | 浮点数 | 否 | 动量的阻尼因子(默认值:0 )。 |
weight_decay | 浮点数 | 否 | L2 正则化系数(默认值:0 )。 |
nesterov | 布尔值 | 否 | 是否使用 Nesterov 动量(默认值:False )。 |
model.parameters():PyTorch 中用于获取模型所有可学习参数的方法。它返回一个生成器,包含模型中所有需要优化的参数(如权重和偏置)。
def choose_optimizer(config, model):
optimizer = config["optimizer"]
learning_rate = config["learning_rate"]
if optimizer == "adam":
return Adam(model.parameters(), lr=learning_rate)
elif optimizer == "sgd":
return SGD(model.parameters(), lr=learning_rate)
4.建立网络模型结构
# -*- coding: utf-8 -*-
import torch
import torch.nn as nn
from torch.optim import Adam, SGD
from torch.nn.utils.rnn import pack_padded_sequence, pad_packed_sequence
"""
建立网络模型结构
"""
class SentenceEncoder(nn.Module):
def __init__(self, config):
super(SentenceEncoder, self).__init__()
hidden_size = config["hidden_size"]
vocab_size = config["vocab_size"] + 1
max_length = config["max_length"]
self.embedding = nn.Embedding(vocab_size, hidden_size, padding_idx=0)
# self.layer = nn.LSTM(hidden_size, hidden_size, batch_first=True, bidirectional=True)
self.layer = nn.Linear(hidden_size, hidden_size)
self.dropout = nn.Dropout(0.5)
#输入为问题字符编码
def forward(self, x):
sentence_length = torch.sum(x.gt(0), dim=-1)
x = self.embedding(x)
#使用lstm
# x, _ = self.layer(x)
#使用线性层
x = self.layer(x)
x = nn.functional.max_pool1d(x.transpose(1, 2), x.shape[1]).squeeze()
return x
class SiameseNetwork(nn.Module):
def __init__(self, config):
super(SiameseNetwork, self).__init__()
self.sentence_encoder = SentenceEncoder(config)
self.loss = nn.CosineEmbeddingLoss()
# 计算余弦距离 1-cos(a,b)
# cos=1时两个向量相同,余弦距离为0;cos=0时,两个向量正交,余弦距离为1
def cosine_distance(self, tensor1, tensor2):
tensor1 = torch.nn.functional.normalize(tensor1, dim=-1)
tensor2 = torch.nn.functional.normalize(tensor2, dim=-1)
cosine = torch.sum(torch.mul(tensor1, tensor2), axis=-1)
return 1 - cosine
def cosine_triplet_loss(self, a, p, n, margin=None):
ap = self.cosine_distance(a, p)
an = self.cosine_distance(a, n)
if margin is None:
diff = ap - an + 0.1
else:
diff = ap - an + margin.squeeze()
return torch.mean(diff[diff.gt(0)])
#sentence : (batch_size, max_length)
def forward(self, sentence1, sentence2=None, sentence3=None):
#同时传入3个句子,则做tripletloss的loss计算
if sentence2 is not None and sentence3 is not None:
vector1 = self.sentence_encoder(sentence1)
vector2 = self.sentence_encoder(sentence2)
vector3 = self.sentence_encoder(sentence3)
return self.cosine_triplet_loss(vector1, vector2, vector3)
#单独传入一个句子时,认为正在使用向量化能力
else:
return self.sentence_encoder(sentence1)
def choose_optimizer(config, model):
optimizer = config["optimizer"]
learning_rate = config["learning_rate"]
if optimizer == "adam":
return Adam(model.parameters(), lr=learning_rate)
elif optimizer == "sgd":
return SGD(model.parameters(), lr=learning_rate)
if __name__ == "__main__":
from config import Config
Config["vocab_size"] = 10
Config["max_length"] = 4
model = SiameseNetwork(Config)
s1 = torch.LongTensor([[1,2,3,0], [2,2,0,0]])
s2 = torch.LongTensor([[1,2,3,4], [3,2,3,4]])
l = torch.LongTensor([[1],[0]])
y = model(s1, s2, l)
print(y)
print(model.state_dict())
四、模型效果评估 evaluate.py
1.初始化测试类
初始化
Evaluator
类的实例,加载验证集和训练集,并初始化统计字典
self.config:配置字典,包含模型和数据路径等信息
self.model:待评估的模型
self.logger:日志记录器
self.valid_data:加载验证集
self.train_data:加载训练集
self.stats_dict:初始化统计字典,用于记录预测结果的正确和错误数量
def __init__(self, config, model, logger):
self.config = config
self.model = model
self.logger = logger
self.valid_data = load_data(config["valid_data_path"], config, shuffle=False)
# 由于效果测试需要训练集当做知识库,再次加载训练集。
# 事实上可以通过传参把前面加载的训练集传进来更合理,但是为了主流程代码改动量小,在这里重新加载一遍
self.train_data = load_data(config["train_data_path"], config)
self.stats_dict = {"correct":0, "wrong":0} #用于存储测试结果
2.问题编码转向量
将训练集中的问题编码为向量,并归一化这些向量,为后续的相似度计算做准备。
① 遍历训练集中的知识库 (
self.train_data.dataset.knwb
),记录问题编号到标准问题编号的映射。② 将所有问题编码为向量,并堆叠成一个矩阵 (
question_matrixs
)。③ 使用模型将问题矩阵编码为向量 (
self.knwb_vectors
)。④ 对向量进行归一化处理,使其模长为 1。
self.question_index_to_standard_question_index:: 一个字典,用于记录问题编号到标准问题编号的映射
self.question_ids:一个列表,用于存储所有标准问题的编号
standard_question_index:标准问题的编号
train_data.dataset.knwb:知识库(Knowledge Base, KB)的缩写,用于存储标准问题及其对应的问题 ID。
question_id:单个问题的编号
question_matrixs:将所有问题编号堆叠成的矩阵,形状为 (n, d)
,其中 n
是问题数量,d
是问题维度
self.knwb_vectors:知识库中所有问题的编码向量,形状为 (n, d)
,其中 d
是向量维度。
items():返回字典中所有键值对的可遍历对象,每个键值对以元组形式返回。
len():返回对象的长度或元素个数
参数名 | 类型 | 是否必选 | 描述 |
---|---|---|---|
obj | 对象 | 是 | 需要计算长度的对象。 |
列表.append():在列表末尾添加一个元素
参数名 | 类型 | 是否必选 | 描述 |
---|---|---|---|
obj | 对象 | 是 | 要添加到列表末尾的元素。 |
torch_no_grad():禁用梯度计算,用于推理阶段以减少内存消耗。
torch.stack():沿新维度连接张量序列
参数名 | 类型 | 是否必选 | 描述 |
---|---|---|---|
tensors | 张量序列 | 是 | 要连接的张量序列。 |
dim | 整数 | 是 | 新维度的索引。 |
torch.cuda.is_available():检查当前系统是否支持 CUDA。
cuda():将张量移动到 GPU 上
参数名 | 类型 | 是否必选 | 描述 |
---|---|---|---|
device | 整数/设备 | 否 | 目标设备,默认为当前设备。 |
torch.nn.functional.normalize():对输入张量沿指定维度进行归一化
参数名 | 类型 | 是否必选 | 描述 |
---|---|---|---|
input | 张量 | 是 | 输入张量。 |
p | 浮点数 | 否 | 归一化的范数类型,默认为 2。 |
dim | 整数 | 否 | 沿指定维度归一化,默认为 1。 |
eps | 浮点数 | 否 | 防止除零的小值,默认为 1e-12。 |
#将知识库中的问题向量化,为匹配做准备
#每轮训练的模型参数不一样,生成的向量也不一样,所以需要每轮测试都重新进行向量化
def knwb_to_vector(self):
self.question_index_to_standard_question_index = {}
self.question_ids = []
for standard_question_index, question_ids in self.train_data.dataset.knwb.items():
for question_id in question_ids:
#记录问题编号到标准问题标号的映射,用来确认答案是否正确
self.question_index_to_standard_question_index[len(self.question_ids)] = standard_question_index
self.question_ids.append(question_id)
with torch.no_grad():
question_matrixs = torch.stack(self.question_ids, dim=0)
if torch.cuda.is_available():
question_matrixs = question_matrixs.cuda()
self.knwb_vectors = self.model(question_matrixs)
#将所有向量都作归一化 v / |v|
self.knwb_vectors = torch.nn.functional.normalize(self.knwb_vectors, dim=-1)
return
3.统计模型效果并展示
Ⅰ、计算统计预测结果
- 使用
assert len(labels) == len(test_question_vectors)
确保输入的长度一致。- 遍历
test_question_vectors
和labels
,对每个问题向量和标签进行处理:
- 如果
hit_index
与label
一致,则增加self.stats_dict["correct"]
,否则增加self.stats_dict["wrong"]。
- 将
hit_index
转换为标准问题编号。- 使用
torch.argmax
找到相似度最高的索引hit_index
,即命中的问题编号。- 使用
torch.mm
计算当前问题向量与知识库中所有问题向量的相似度。test_question_vector.unsqueeze(0)
将向量扩展为[1, vec_size]
,self.knwb_vectors.T
是知识库向量的转置,形状为[vec_size, n]
,结果res
的形状为[1, n]
。
test_question_vectors
: 验证集中问题的编码向量,形状为 [batch_size, vec_size]
labels
: 验证集中问题的真实标签,形状为 [batch_size]
len():返回容器(如字符串、列表、元组、字典等)中元素的数量。
参数名 | 类型 | 是否必选 | 描述 |
---|---|---|---|
obj | 任意容器类型 | 是 | 需要计算长度的对象。 |
zip():将多个可迭代对象(如列表、元组等)中对应位置的元素配对,生成一个元组的迭代器。
参数名 | 类型 | 是否必选 | 描述 |
---|---|---|---|
*iterables | 可迭代对象 | 是 | 需要配对的多个可迭代对象。 |
unsqueeze():在指定维度上插入一个大小为 1 的维度,用于改变张量的形状。
参数名 | 类型 | 是否必选 | 描述 |
---|---|---|---|
input | 张量 | 是 | 需要操作的输入张量。 |
dim | 整数 | 是 | 插入新维度的位置索引。 |
torch.mm():执行两个二维张量(矩阵)的矩阵乘法
参数名 | 类型 | 是否必选 | 描述 |
---|---|---|---|
input | 二维张量 | 是 | 第一个矩阵。 |
mat2 | 二维张量 | 是 | 第二个矩阵。 |
int(): 将数字或字符串转换为整数,或对浮点数进行向下取整。
参数名 | 类型 | 是否必选 | 描述 |
---|---|---|---|
x | 数字或字符串 | 是 | 需要转换的对象。 |
base | 整数 | 否 | 进制基数(默认为 10)。 |
def write_stats(self, test_question_vectors, labels):
assert len(labels) == len(test_question_vectors)
for test_question_vector, label in zip(test_question_vectors, labels):
#通过一次矩阵乘法,计算输入问题和知识库中所有问题的相似度
#test_question_vector shape [vec_size] knwb_vectors shape = [n, vec_size]
res = torch.mm(test_question_vector.unsqueeze(0), self.knwb_vectors.T)
hit_index = int(torch.argmax(res.squeeze())) #命中问题标号
hit_index = self.question_index_to_standard_question_index[hit_index] #转化成标准问编号
if int(hit_index) == int(label):
self.stats_dict["correct"] += 1
else:
self.stats_dict["wrong"] += 1
return
Ⅱ、展示预测结果和准确率
输出模型在验证集上的预测结果和准确率
logger.info():记录信息级别的日志,用于输出程序运行时的普通信息
参数名 | 类型 | 是否必选 | 描述 |
---|---|---|---|
msg | str | 是 | 要记录的日志信息。 |
*args | 任意类型 | 否 | 用于格式化日志信息的参数。例如,msg 中包含 {} 时,*args 会填充这些占位符。 |
**kwargs | dict | 否 | 可选参数,如 exc_info 、stack_info 等,用于附加异常或堆栈信息。 |
def show_stats(self):
correct = self.stats_dict["correct"]
wrong = self.stats_dict["wrong"]
self.logger.info("预测集合条目总量:%d" % (correct +wrong))
self.logger.info("预测正确条目:%d,预测错误条目:%d" % (correct, wrong))
self.logger.info("预测准确率:%f" % (correct / (correct + wrong)))
self.logger.info("--------------------")
return
4.模型表现评估函数
评估模型在验证集上的表现
self.logger.info:记录当前测试的轮次
self.stats_dict:定义一个统计字典,清空前一轮的测试结果,初始化正确和错误的统计值。
self.knwb_to_vector():将训练集中的问题编码为向量,并进行归一化处理,为后续的相似度计算做准备。
self.wirte_states():计算统计预测结果
self.show_states():输出模型在验证集上的预测结果和准确率
model.eval():将模型设置为评估模式。在评估模式下,模型会关闭一些在训练过程中使用的特性,如 Dropout 和 Batch Normalization 层的训练模式,以确保模型在推理时表现一致。
enumerate():将一个可迭代对象(如列表、元组、字符串)组合为一个索引序列,返回一个枚举对象,每次迭代返回一个包含索引和对应元素的元组。
参数名 | 类型 | 是否必选 | 描述 |
---|---|---|---|
iterable | 可迭代对象 | 是 | 需要枚举的对象。 |
start | 整数 | 否 | 索引的起始值,默认为 0。 |
torch.cuda.is_available():检查当前系统是否支持 CUDA(即是否有可用的 GPU)。如果支持,返回 True
,否则返回 False
。
cuda():将张量或模型移动到 GPU 上。如果没有指定设备,默认使用当前 GPU。
参数名 | 类型 | 是否必选 | 描述 |
---|---|---|---|
device | 整数或字符串 | 否 | 目标 GPU 设备,如 0 或 'cuda:0' 。 |
torch.no_grad():禁用梯度计算,通常用于模型推理或评估阶段,以节省内存和计算资源。
def eval(self, epoch):
self.logger.info("开始测试第%d轮模型效果:" % epoch)
self.stats_dict = {"correct":0, "wrong":0} #清空前一轮的测试结果
self.model.eval()
self.knwb_to_vector()
for index, batch_data in enumerate(self.valid_data):
if torch.cuda.is_available():
batch_data = [d.cuda() for d in batch_data]
input_id, labels = batch_data #输入变化时这里需要修改,比如多输入,多输出的情况
with torch.no_grad():
test_question_vectors = self.model(input_id) #不输入labels,使用模型当前参数进行预测
self.write_stats(test_question_vectors, labels)
self.show_stats()
return
5.模型效果测试
# -*- coding: utf-8 -*-
import torch
from loader import load_data
"""
模型效果测试
"""
class Evaluator:
def __init__(self, config, model, logger):
self.config = config
self.model = model
self.logger = logger
self.valid_data = load_data(config["valid_data_path"], config, shuffle=False)
# 由于效果测试需要训练集当做知识库,再次加载训练集。
# 事实上可以通过传参把前面加载的训练集传进来更合理,但是为了主流程代码改动量小,在这里重新加载一遍
self.train_data = load_data(config["train_data_path"], config)
self.stats_dict = {"correct":0, "wrong":0} #用于存储测试结果
#将知识库中的问题向量化,为匹配做准备
#每轮训练的模型参数不一样,生成的向量也不一样,所以需要每轮测试都重新进行向量化
def knwb_to_vector(self):
self.question_index_to_standard_question_index = {}
self.question_ids = []
for standard_question_index, question_ids in self.train_data.dataset.knwb.items():
for question_id in question_ids:
#记录问题编号到标准问题标号的映射,用来确认答案是否正确
self.question_index_to_standard_question_index[len(self.question_ids)] = standard_question_index
self.question_ids.append(question_id)
with torch.no_grad():
question_matrixs = torch.stack(self.question_ids, dim=0)
if torch.cuda.is_available():
question_matrixs = question_matrixs.cuda()
self.knwb_vectors = self.model(question_matrixs)
#将所有向量都作归一化 v / |v|
self.knwb_vectors = torch.nn.functional.normalize(self.knwb_vectors, dim=-1)
return
def eval(self, epoch):
self.logger.info("开始测试第%d轮模型效果:" % epoch)
self.stats_dict = {"correct":0, "wrong":0} #清空前一轮的测试结果
self.model.eval()
self.knwb_to_vector()
for index, batch_data in enumerate(self.valid_data):
if torch.cuda.is_available():
batch_data = [d.cuda() for d in batch_data]
input_id, labels = batch_data #输入变化时这里需要修改,比如多输入,多输出的情况
with torch.no_grad():
test_question_vectors = self.model(input_id) #不输入labels,使用模型当前参数进行预测
self.write_stats(test_question_vectors, labels)
self.show_stats()
return
def write_stats(self, test_question_vectors, labels):
assert len(labels) == len(test_question_vectors)
for test_question_vector, label in zip(test_question_vectors, labels):
#通过一次矩阵乘法,计算输入问题和知识库中所有问题的相似度
#test_question_vector shape [vec_size] knwb_vectors shape = [n, vec_size]
res = torch.mm(test_question_vector.unsqueeze(0), self.knwb_vectors.T)
hit_index = int(torch.argmax(res.squeeze())) #命中问题标号
hit_index = self.question_index_to_standard_question_index[hit_index] #转化成标准问编号
if int(hit_index) == int(label):
self.stats_dict["correct"] += 1
else:
self.stats_dict["wrong"] += 1
return
def show_stats(self):
correct = self.stats_dict["correct"]
wrong = self.stats_dict["wrong"]
self.logger.info("预测集合条目总量:%d" % (correct +wrong))
self.logger.info("预测正确条目:%d,预测错误条目:%d" % (correct, wrong))
self.logger.info("预测准确率:%f" % (correct / (correct + wrong)))
self.logger.info("--------------------")
return
五、模型训练文件 main.py
1、导入文件
import torch:导入 PyTorch 库,用于构建和训练神经网络模型
import os:导入操作系统相关的模块,用于处理文件路径、环境变量等
import random:导入随机数生成模块,用于设置随机种子或进行随机操作
import numpy as np:导入 NumPy 库,用于进行高效的数值计算
import logging:导入日志模块,用于记录程序运行时的信息
from config import Config:从 config
模块中导入 Config
类,通常用于管理项目的配置参数
from model import SiameseNetwork, choose_optimizer:从 model
模块中导入 SiameseNetwork
类和 choose_optimizer
函数,SiameseNetwork
是一个孪生网络模型,choose_optimizer
用于选择优化器
from evaluate import Evaluator:从 evaluate
模块中导入 Evaluator
类,用于模型评估
from loader import load_data:从 loader
模块中导入 load_data
函数,用于加载数据集
import torch
import os
import random
import os
import numpy as np
import logging
from config import Config
from model import SiameseNetwork, choose_optimizer
from evaluate import Evaluator
from loader import load_data
2、日志配置
logging.basicConfig():用于对日志系统进行一次性配置,设置日志的默认行为,如日志级别、输出格式、输出位置等。它是 logging
模块中最常用的配置函数,通常用于简单的日志记录需求
参数名 | 类型 | 说明 |
---|---|---|
filename | str | 指定日志文件名,日志会被写入该文件。如果未指定,日志默认输出到控制台。 |
filemode | str | 文件打开模式,默认为 'a' (追加模式)。可设置为 'w' (覆盖模式)。 |
format | str | 定义日志输出格式。默认格式为 '%(levelname)s:%(name)s:%(message)s' 。 |
datefmt | str | 定义日期时间格式。默认格式为 '%Y-%m-%d %H:%M:%S' 。 |
level | int | 设置日志级别,低于该级别的日志将被忽略。默认级别为 WARNING 。 |
stream | IO | 指定日志输出流(如 sys.stderr 或 sys.stdout )。 |
handlers | list | 指定处理器列表。如果指定了 handlers ,则 filename 和 stream 会被忽略。 |
logging.getLogger():返回一个 Logger
对象,用于记录日志。如果没有指定名称,则返回根日志器(root logger
)。通过 Logger
对象,可以更灵活地控制日志的输出,如添加多个处理器、设置日志级别等。
参数名 | 类型 | 说明 |
---|---|---|
name | str | 日志器的名称。如果未指定或为 None ,则返回根日志器。 |
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
3、主函数 main
Ⅰ、创建模型保存目录
os.path.isdir():检查指定路径是否为目录。如果是目录,返回 True
;否则返回 False
参数名 | 类型 | 说明 |
---|---|---|
path | str | 需要检查的路径(文件或目录)。 |
os.mkdir():创建一个目录。如果目录已存在或路径无效,会抛出 OSError
异常
参数名 | 类型 | 说明 |
---|---|---|
path | str | 要创建的目录路径。 |
mode | int | 目录权限模式,默认为 0o777 |
#创建保存模型的目录
if not os.path.isdir(config["model_path"]):
os.mkdir(config["model_path"])
Ⅱ、加载训练数据
load_data():从 loader
模块中导入 load_data
函数,用于加载数据集
train_data:从配置文件中加载训练数据
#加载训练数据
train_data = load_data(config["train_data_path"], config)
Ⅲ、加载模型
SiameseNetwork():从 model
模块中导入孪生网络模型 SiameseNetwork
类
model(): 加载 SiameseNetwork
孪生网络模型
#加载模型
model = SiameseNetwork(config)
Ⅳ、检查GPU并迁移模型
torch.cuda.is_available():检查当前环境是否支持 CUDA(即是否有可用的 GPU)。如果支持,返回 True
;否则返回 False
。
logger.info():记录信息级别的日志,用于输出程序运行中的一般性信息
参数名 | 类型 | 说明 |
---|---|---|
msg | str | 要记录的日志信息。 |
*args | Any | 格式化日志信息的参数。 |
**kwargs | Any | 其他关键字参数(如 exc_info )。 |
cuda():将张量或模型移动到 GPU 上进行计算。如果没有可用的 GPU,会抛出异常
参数名 | 类型 | 说明 |
---|---|---|
device | int | 指定 GPU 设备编号(如 0 )。 |
# 标识是否使用gpu
cuda_flag = torch.cuda.is_available()
if cuda_flag:
logger.info("gpu可以使用,迁移模型至gpu")
model = model.cuda()
Ⅴ、加载优化器
choose_optimizer:从 model
模块中导入 choose_optimizer
函数,根据配置文件选择优化器
#加载优化器
optimizer = choose_optimizer(config, model)
Ⅵ、加载评估器
Evaluator():从 evaluate
模块中导入 Evaluator
类,用于模型评估
#加载效果测试类
evaluator = Evaluator(config, model, logger)
Ⅶ、训练循环 🚀
- 遍历每个 epoch,训练模型。
- 在每个 epoch 中,遍历训练数据(
train_data
),计算损失并更新模型参数。- 记录每个 batch 的损失,并在 epoch 结束时计算平均损失。
- 调用评估器(
evaluator
)对模型进行评估。
① 模型训练模式
model.train():将模型设置为训练模式,启用 Batch Normalization
和 Dropout
层。在训练模式下,Batch Normalization
会使用每一批数据的均值和方差,而 Dropout
会随机丢弃部分神经元。
model.train()
② 梯度清零
optimizer.zero_grad():优化器的方法,将模型参数的梯度清零。在每次反向传播之前调用,防止梯度累积。
optimizer.zero_grad()
③ GPU支持
cuda():将张量或模型移动到 GPU 上进行计算。如果没有可用的 GPU,会抛出异常。
参数名 | 类型 | 说明 |
---|---|---|
device | int | 指定 GPU 设备编号(如 0 )。 |
if cuda_flag:
batch_data = [d.cuda() for d in batch_data]
④ 损失计算
列表.append():在列表末尾添加一个元素。
参数名 | 类型 | 说明 |
---|---|---|
obj | Any | 要添加到列表末尾的元素。 |
item():从张量中提取标量值,并返回其高精度值(如 int
或 float
)
input_id1, input_id2, labels = batch_data #输入变化时这里需要修改,比如多输入,多输出的情况
loss = model(input_id1, input_id2, labels)
train_loss.append(loss.item())
⑤ 反向传播
backward():计算损失函数关于模型参数的梯度,用于反向传播。
参数名 | 类型 | 说明 |
---|---|---|
gradient | Tensor | 用于链式求导的梯度张量(默认为 None ) |
loss.backward()
⑥ 模型参数更新
optimizer.step():优化器的方法,根据计算出的梯度更新模型参数
optimizer.step()
⑦ 日志记录
logger.info():记录信息级别的日志,用于输出程序运行中的一般性信息。
参数名 | 类型 | 说明 |
---|---|---|
msg | str | 要记录的日志信息。 |
*args | Any | 格式化日志信息的参数。 |
**kwargs | Any | 其他关键字参数(如 exc_info )。 |
np.mean():计算数组的均值
参数名 | 类型 | 说明 |
---|---|---|
a | array_like | 输入数组。 |
axis | int | 计算均值的轴(默认为 None )。 |
dtype | dtype | 输出数组的数据类型(默认为 None )。 |
out | ndarray | 输出数组(默认为 None )。 |
evaluator.eval():evaluate
模块中导入 Evaluator
类,用于模型评估
logger.info("epoch average loss: %f" % np.mean(train_loss))
evaluator.eval(epoch)
Ⅷ、保存模型
os.path.join():用于连接多个路径片段,生成一个完整的路径字符串。它会根据操作系统的不同自动处理路径分隔符(如 Windows 使用 \
,Linux/Mac 使用 /
),确保生成的路径是有效的
参数名 | 类型 | 说明 |
---|---|---|
path1 | str | 第一个路径片段。 |
path2 | str | 第二个路径片段。 |
*paths | str | 可选的更多路径片段。 |
torch.save():用于保存 PyTorch 对象(如模型、张量、字典等)到文件中。它使用 Python 的 pickle
进行序列化,方便后续加载和使用
参数名 | 类型 | 说明 |
---|---|---|
obj | Any | 要保存的对象(如模型、张量、字典等)。 |
f | str 或 IO | 保存的目标文件路径或文件对象。 |
model.state_dict():返回一个包含模型所有参数的字典对象。字典的键是参数的名称,值是对应的张量。它通常用于保存和加载模型的参数,而不保存整个模型结构
model_path = os.path.join(config["model_path"], "epoch_%d.pth" % epoch)
torch.save(model.state_dict(), model_path)
4. 模型训练
# -*- coding: utf-8 -*-
import torch
import os
import random
import os
import numpy as np
import logging
from config import Config
from model import SiameseNetwork, choose_optimizer
from evaluate import Evaluator
from loader import load_data
logging.basicConfig(level = logging.INFO,format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
"""
模型训练主程序
"""
def main(config):
#创建保存模型的目录
if not os.path.isdir(config["model_path"]):
os.mkdir(config["model_path"])
#加载训练数据
train_data = load_data(config["train_data_path"], config)
#加载模型
model = SiameseNetwork(config)
# 标识是否使用gpu
cuda_flag = torch.cuda.is_available()
if cuda_flag:
logger.info("gpu可以使用,迁移模型至gpu")
model = model.cuda()
#加载优化器
optimizer = choose_optimizer(config, model)
#加载效果测试类
evaluator = Evaluator(config, model, logger)
#训练
for epoch in range(config["epoch"]):
epoch += 1
model.train()
logger.info("epoch %d begin" % epoch)
train_loss = []
for index, batch_data in enumerate(train_data):
optimizer.zero_grad()
if cuda_flag:
batch_data = [d.cuda() for d in batch_data]
input_id1, input_id2, labels = batch_data #输入变化时这里需要修改,比如多输入,多输出的情况
loss = model(input_id1, input_id2, labels)
train_loss.append(loss.item())
# if index % int(len(train_data) / 2) == 0:
# logger.info("batch loss %f" % loss)
loss.backward()
optimizer.step()
logger.info("epoch average loss: %f" % np.mean(train_loss))
evaluator.eval(epoch)
model_path = os.path.join(config["model_path"], "epoch_%d.pth" % epoch)
torch.save(model.state_dict(), model_path)
return
if __name__ == "__main__":
main(Config)
六、模型预测文件 predict.py
- 加载训练数据和预训练模型。
- 将知识库中的问题向量化,为匹配做准备。
- 对用户输入的问题进行编码,并计算其与知识库中问题的相似度。
- 返回最匹配的标准问题
1.初始化预测类
self.config:配置对象
self.model:模型
self.train_data:知识库数据
self.knwb_to_vector():将知识库中的问题向量化,为后续的匹配做准备。
torch.cuda,is_available():检查当前环境是否支持 CUDA(即是否有可用的 GPU)。如果支持,返回 True
;否则返回 False
。它通常用于判断是否可以使用 GPU 加速计算。
model.cuda():将模型的所有参数和缓冲区移动到 GPU 上。如果没有指定设备编号,默认使用第一个 GPU(cuda:0
)。如果 GPU 不可用,调用此方法会报错
参数名 | 类型 | 说明 |
---|---|---|
device | int 或 torch.device | 指定 GPU 设备编号(如 0 )。如果未指定,默认使用 cuda:0 。 |
model.cpu():将模型的所有参数和缓冲区移动回 CPU 上。通常在需要将模型从 GPU 移回 CPU 时使用。
model.eval():模型设置为评估模式,这会禁用某些特定于训练的操作(如 dropout 和 batch normalization 的更新)。
def __init__(self, config, model, knwb_data):
self.config = config
self.model = model
self.train_data = knwb_data
if torch.cuda.is_available():
self.model = model.cuda()
else:
self.model = model.cpu()
self.model.eval()
self.knwb_to_vector()
2.问题转向量
将知识库中的问题编号转换为向量,并进行归一化处理。通过禁用梯度计算和使用 GPU 加速,函数在保证高效性的同时,生成了可用于匹配的向量表示
① 初始化变量 ② 遍历知识库并记录问题 ③ 将问题编号转换为张量
④ 通过模型生成向量 ⑤ 向量归一化 ⑥ 返回结果
self.question_index_to_standard_question_index:创建一个字典,用于记录问题编号到标准问题编号的映射,方便后续确认答案是否正确。
self.vocab:创建一个字典,从训练数据中加载词汇表
self.schema:创建一个字典,从训练数据中加载标签映射表
self.train_data.dataset.knwb:训练数据,存储标准问题及其对应的问题 ID。
self.train_data.dataset.schema:知识库训练数据,存储问题和标签之间的映射关系
self.index_to_standard_question:创建一个字典,将模式中的键值对反转,方便通过索引查找标准问题。
standard_question_index:标准问题的索引。例如,知识库中有多个标准问题,每个标准问题都有一个唯一的索引
self.question_ids:创建一个列表,存储知识库中所有问题的编号。
question_id:包含问题 ID 的列表
question_matrix:问题张量
self.knwb_vectors:输入模型后,生成的对应向量
items():返回字典中所有键值对的可遍历对象,每个键值对以元组形式返回。
append():用于在列表末尾添加一个元素。它会直接修改原列表,而不会创建新的列表
参数名 | 类型 | 说明 |
---|---|---|
obj | Any | 要添加到列表末尾的元素。 |
torch.no_grad():用于临时禁用梯度计算,通常在模型推理或评估时使用。它可以减少内存消耗并加速计算。
torch.stack():沿指定维度连接多个张量,生成一个新的张量。所有输入张量的形状必须相同。
参数名 | 类型 | 说明 |
---|---|---|
tensors | list 或 tuple | 要连接的张量序列。 |
dim | int | 指定连接的维度。 |
torch.cuda.is_available():检查当前环境是否支持 CUDA(即是否有可用的 GPU)。如果支持,返回 True
;否则返回 False
。
cuda():将张量或模型移动到 GPU 上进行计算。如果没有指定设备编号,默认使用第一个 GPU(cuda:0
)。
参数名 | 类型 | 说明 |
---|---|---|
device | int 或 torch.device | 指定 GPU 设备编号(如 0 ) |
torch.nn.functional.normalize():对输入张量在指定维度上进行归一化,使其 L-p 范数为 1
参数名 | 类型 | 说明 |
---|---|---|
input | Tensor | 输入张量。 |
p | float | 范数的类型,默认为 2(L2 范数)。 |
dim | int | 指定归一化的维度。 |
eps | float | 防止除零的小值,默认为 1e-12 。 |
#将知识库中的问题向量化,为匹配做准备
#每轮训练的模型参数不一样,生成的向量也不一样,所以需要每轮测试都重新进行向量化
def knwb_to_vector(self):
self.question_index_to_standard_question_index = {}
self.question_ids = []
self.vocab = self.train_data.dataset.vocab
self.schema = self.train_data.dataset.schema
self.index_to_standard_question = dict((y, x) for x, y in self.schema.items())
for standard_question_index, question_ids in self.train_data.dataset.knwb.items():
for question_id in question_ids:
#记录问题编号到标准问题标号的映射,用来确认答案是否正确
self.question_index_to_standard_question_index[len(self.question_ids)] = standard_question_index
self.question_ids.append(question_id)
with torch.no_grad():
question_matrixs = torch.stack(self.question_ids, dim=0)
if torch.cuda.is_available():
question_matrixs = question_matrixs.cuda()
self.knwb_vectors = self.model(question_matrixs)
#将所有向量都作归一化 v / |v|
self.knwb_vectors = torch.nn.functional.normalize(self.knwb_vectors, dim=-1)
return
3.句子编码
将输入的文本
text
编码为一个整数索引列表input_id
,其中每个整数代表词汇表self.vocab
中对应单词或字符的索引。初始化空列表:创建一个空列表
input_id
,用于存储编码后的索引。判断词汇表类型:检查配置中的
vocab_path
是否为"words.txt"
。如果是,则使用jieba
对文本进行分词;否则,将文本按字符处理。分词处理:① 如果
vocab_path
是"words.txt"
,使用jieba.cut(text)
对文本进行分词,遍历每个分词结果。② 将每个分词在词汇表self.vocab
中查找对应的索引。如果分词不在词汇表中,则使用[UNK]
(未知词)的索引。字符处理:① 如果
vocab_path
不是"words.txt"
,则遍历文本中的每个字符。② 将每个字符在词汇表中查找对应的索引。如果字符不在词汇表中,则使用[UNK]
的索引。返回结果:返回编码后的索引列表
input_id
。
input_id:存储编码后的整数列表
jieba.cut():将中文句子分割成独立的单词,支持精确模式、全模式和搜索引擎模式。
参数名 | 类型 | 说明 |
---|---|---|
sentence | str | 需要分词的字符串。 |
cut_all | bool | 是否使用全模式,默认为 False (精确模式)。 |
HMM | bool | 是否使用隐马尔可夫模型(HMM),默认为 True 。 |
字典.get():安全地获取字典中指定键的值。如果键不存在,返回默认值(默认为 None
),而不会引发 KeyError
异常。
参数名 | 类型 | 说明 |
---|---|---|
key | Any | 要查找的键。 |
default | Any | 键不存在时返回的默认值,默认为 None 。 |
列表.append():在列表的末尾添加一个元素,直接修改原列表。
参数名 | 类型 | 说明 |
---|---|---|
obj | Any | 要添加到列表末尾的元素。 |
def encode_sentence(self, text):
input_id = []
if self.config["vocab_path"] == "words.txt":
for word in jieba.cut(text):
input_id.append(self.vocab.get(word, self.vocab["[UNK]"]))
else:
for char in text:
input_id.append(self.vocab.get(char, self.vocab["[UNK]"]))
return input_id
4.根据相似度预测
根据输入的句子
sentence
,通过模型预测并找到知识库中最匹配的标准问题① 将句子编码为张量 ② 将张量移动到GPU(如果可用)
③ 禁用梯度计算并生成问题向量 ④ 计算与知识库中问题的相似度
⑤ 找到最匹配的问题编号 ⑥ 返回标准问题
input_id:输入句子经过编码后的整数列表,每个整数代表词汇表中对应单词或字符的索引。它是模型的输入数据
test_question_vector:模型对输入句子 input_id
进行编码后生成的向量表示。它捕捉了句子的语义信息。
self.encode_sentence:是一个方法,将输入的句子 sentence
编码为一个整数列表 input_id
,每个整数代表词汇表中对应单词或字符的索引。
self.question_index_to_standard_question_index:是一个字典,用于记录知识库中每个问题编号到标准问题编号的映射关系
res:当前问题与知识库中问题的相似度
hit_index:当前问题在知识库中最相似问题的编号
torch.LongTensor():用于创建一个包含整数的张量(tensor),数据类型为 64 位整数(torch.long
)。在 PyTorch 1.6 版本之后,推荐使用 torch.tensor()
并指定 dtype=torch.long
来替代。
参数名 | 类型 | 说明 |
---|---|---|
data | list 或 tuple | 用于初始化张量的数据。 |
torch.cuda.is_available():检查当前系统是否支持 CUDA(即是否有可用的 GPU),并返回布尔值
cuda():将张量或模型移动到 GPU 上进行计算。如果没有指定设备编号,默认使用第一个 GPU(cuda:0
)
参数名 | 类型 | 说明 |
---|---|---|
device | int 或 torch.device | 指定 GPU 设备编号(如 0 )。 |
torch.no_grad():用于临时禁用梯度计算,通常在模型推理或评估时使用。它可以减少内存消耗并加速计算。
torch.mm():用于两个二维张量(矩阵)之间的矩阵乘法。仅支持二维张量,不支持高维张量或广播机制。
参数名 | 类型 | 说明 |
---|---|---|
input | Tensor | 第一个矩阵(二维张量)。 |
mat2 | Tensor | 第二个矩阵(二维张量)。 |
squeeze():从张量的形状中移除所有维度为 1 的维度,从而对张量进行降维。
参数名 | 类型 | 说明 |
---|---|---|
dim | int 或 None | 指定要移除的维度,默认为 None (移除所有维度为 1 的维度)。 |
unsqueeze():在指定维度上增加一个维度,维度大小为 1。
参数名 | 类型 | 说明 |
---|---|---|
dim | int | 指定要增加维度的位置。 |
def predict(self, sentence):
input_id = self.encode_sentence(sentence)
input_id = torch.LongTensor([input_id])
if torch.cuda.is_available():
input_id = input_id.cuda()
with torch.no_grad():
test_question_vector = self.model(input_id) #不输入labels,使用模型当前参数进行预测
res = torch.mm(test_question_vector.unsqueeze(0), self.knwb_vectors.T)
hit_index = int(torch.argmax(res.squeeze())) #命中问题标号
hit_index = self.question_index_to_standard_question_index[hit_index] #转化成标准问编号
return self.index_to_standard_question[hit_index]
5.加载数据并预测
knwb_data:训练数据集
model:根据配置文件加载一个孪生网络模型
model.load_state_dict():PyTorch 中用于加载模型参数的方法。它接受一个包含模型参数的状态字典(state_dict
),并将这些参数加载到模型中。
参数名 | 类型/默认值 | 描述 |
---|---|---|
state_dict | dict | 包含模型参数的字典,键是参数名称,值是参数张量。 |
strict | bool ,默认为 True | 是否严格匹配状态字典的键。如果为 False ,则允许部分匹配。 |
torch.load():加载由 torch.save()
保存的 PyTorch 对象,例如模型的状态字典(state_dict
)、整个模型、张量等。
参数名 | 类型/默认值 | 描述 |
---|---|---|
f | str 或 os.PathLike 或文件对象 | 要加载的文件路径或文件对象。 |
map_location | callable 或 torch.device 或 str 或 dict ,默认为 None | 指定如何重新映射存储位置(例如,从 GPU 到 CPU)。 |
pickle_module | 模块,默认为 pickle | 用于反序列化的模块。 |
**pickle_load_args | 额外参数 | 传递给 pickle.load 的额外参数。 |
# -*- coding: utf-8 -*-
import jieba
import torch
from loader import load_data
from config import Config
from model import SiameseNetwork, choose_optimizer
"""
模型效果测试
"""
class Predictor:
def __init__(self, config, model, knwb_data):
self.config = config
self.model = model
self.train_data = knwb_data
if torch.cuda.is_available():
self.model = model.cuda()
else:
self.model = model.cpu()
self.model.eval()
self.knwb_to_vector()
#将知识库中的问题向量化,为匹配做准备
#每轮训练的模型参数不一样,生成的向量也不一样,所以需要每轮测试都重新进行向量化
def knwb_to_vector(self):
self.question_index_to_standard_question_index = {}
self.question_ids = []
self.vocab = self.train_data.dataset.vocab
self.schema = self.train_data.dataset.schema
self.index_to_standard_question = dict((y, x) for x, y in self.schema.items())
for standard_question_index, question_ids in self.train_data.dataset.knwb.items():
for question_id in question_ids:
#记录问题编号到标准问题标号的映射,用来确认答案是否正确
self.question_index_to_standard_question_index[len(self.question_ids)] = standard_question_index
self.question_ids.append(question_id)
with torch.no_grad():
question_matrixs = torch.stack(self.question_ids, dim=0)
if torch.cuda.is_available():
question_matrixs = question_matrixs.cuda()
self.knwb_vectors = self.model(question_matrixs)
#将所有向量都作归一化 v / |v|
self.knwb_vectors = torch.nn.functional.normalize(self.knwb_vectors, dim=-1)
return
def encode_sentence(self, text):
input_id = []
if self.config["vocab_path"] == "words.txt":
for word in jieba.cut(text):
input_id.append(self.vocab.get(word, self.vocab["[UNK]"]))
else:
for char in text:
input_id.append(self.vocab.get(char, self.vocab["[UNK]"]))
return input_id
def predict(self, sentence):
input_id = self.encode_sentence(sentence)
input_id = torch.LongTensor([input_id])
if torch.cuda.is_available():
input_id = input_id.cuda()
with torch.no_grad():
test_question_vector = self.model(input_id) #不输入labels,使用模型当前参数进行预测
res = torch.mm(test_question_vector.unsqueeze(0), self.knwb_vectors.T)
hit_index = int(torch.argmax(res.squeeze())) #命中问题标号
hit_index = self.question_index_to_standard_question_index[hit_index] #转化成标准问编号
return self.index_to_standard_question[hit_index]
if __name__ == "__main__":
knwb_data = load_data(Config["train_data_path"], Config)
model = SiameseNetwork(Config)
model.load_state_dict(torch.load("model_output/epoch_10.pth"))
pd = Predictor(Config, model, knwb_data)
sentence = "如何修改手机号码"
res = pd.predict(sentence)
print(res)