当前位置: 首页 > article >正文

【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():打开文件,返回一个文件对象,用于读取或写入文件。

参数名类型描述
filestr文件路径。
modestr文件打开模式(如 "r" 读取,"w" 写入,"a" 追加,默认 "r")。
encodingstr文件编码(如 "utf8",默认 None)。
errorsstr指定编码错误的处理方式(如 "ignore",默认 None)。
newlinestr控制换行符的行为(如 "\n",默认 None)。
bufferingint设置缓冲策略(如 -1 系统默认,0 无缓冲,1 行缓冲,默认 -1)。
closefdbool是否关闭文件描述符(默认 True)。
openercallable自定义文件打开器(默认 None)。

enumerate():返回一个枚举对象,将可迭代对象(如列表、字符串)转换为索引和值的组合。

参数名类型描述
iterableiterable可迭代对象(如列表、字符串)。
startint索引的起始值(默认 0)。

strip():去除字符串首尾的空白字符(如空格、换行符)或指定字符。

参数名类型描述
charsstr指定要删除的字符(默认去除空白字符)。
#加载字表或词表
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():打开文件,返回一个文件对象,用于读取或写入文件。

参数名类型描述
filestr文件路径。
modestr文件打开模式(如 "r" 读取,"w" 写入,"a" 追加,默认 "r")。
encodingstr文件编码(如 "utf8",默认 None)。
errorsstr指定编码错误的处理方式(如 "ignore",默认 None)。
newlinestr控制换行符的行为(如 "\n",默认 None)。
bufferingint设置缓冲策略(如 -1 系统默认,0 无缓冲,1 行缓冲,默认 -1)。
closefdbool是否关闭文件描述符(默认 True)。
openercallable自定义文件打开器(默认 None)。

json.loads():将 JSON 格式的字符串解析为 Python 对象(如字典、列表)。

参数名类型描述
sstrJSON 格式的字符串。
clsclass自定义 JSON 解码器(默认 None)。
object_hookcallable自定义对象解码函数(默认 None)。
parse_floatcallable自定义浮点数解码函数(默认 None)。
parse_intcallable自定义整数解码函数(默认 None)。
parse_constantcallable自定义常量解码函数(默认 None)。
object_pairs_hookcallable自定义键值对解码函数(默认 None)。

文件对象.read():从文件对象中读取全部内容,返回一个字符串(文本文件)或字节对象(二进制文件)。

参数名类型描述
sizeint可选参数,指定读取的字节数(文本文件为字符数)。如果未指定,则读取全部内容。
#加载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,支持批量加载、打乱数据顺序等功能。

参数名类型描述
datasetDataset数据集对象(如 DataGenerator)。
batch_sizeint每个批次的大小(默认 1)。
shufflebool是否打乱数据顺序(默认 False)。
samplerSampler自定义采样器(默认 None)。
batch_samplerSampler自定义批次采样器(默认 None)。
num_workersint数据加载的线程数(默认 0,表示在主线程中加载)。
collate_fncallable自定义批次数据处理函数(默认 None)。
pin_memorybool是否将数据加载到 GPU 的固定内存中(默认 False)。
drop_lastbool是否丢弃最后一个不完整的批次(默认 False)。
timeoutint数据加载的超时时间(默认 0,表示不超时)。
worker_init_fncallable自定义工作线程初始化函数(默认 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 对象,包括字符串、列表、元组、字典、集合等。

参数名类型描述
objobject需要计算长度的对象,可以是字符串、列表、元组、字典、集合等。
    #补齐或截断输入的序列,使其可以在一个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]

Ⅲ、加载和处理训练样本和测试样本

  1. 加载数据:从指定路径的文件中逐行读取数据。
  2. 区分训练集和测试集
    • 训练集数据格式为字典(dict),包含 questions 和 target 字段。
    • 测试集数据格式为列表(list),包含 question 和 label
  3. 数据预处理
    • 将文本数据编码为索引序列。
    • 将标签映射为索引。
    • 将处理后的数据存储到 self.knwb(训练集)或 self.data(测试集)中。

data:用于存储测试集数据

knwb:知识库数据的存储对象,通常是一个字典,键是标准问题的索引,值是对应的问题 ID 列表。

参数名类型描述
default_factory可调用对象或类型默认工厂函数,用于生成默认值。如果未提供,默认为 None,此时行为与普通字典相同。
**kwargs关键字参数其他参数会传递给 dict 的构造函数,用于初始化字典内容。

defaultdict():Python 中 collections 模块提供的一个字典子类,它的主要作用是在访问不存在的键时返回一个默认值,而不是抛出 KeyError 异常。

open():打开文件,返回一个文件对象,用于读取或写入文件。

参数名类型描述
filestr文件路径。
modestr文件打开模式(如 "r" 读取,"w" 写入,"a" 追加,默认 "r")。
encodingstr文件编码(如 "utf8",默认 None)。
errorsstr指定编码错误的处理方式(如 "ignore",默认 None)。
newlinestr控制换行符的行为(如 "\n",默认 None)。
bufferingint设置缓冲策略(如 -1 系统默认,0 无缓冲,1 行缓冲,默认 -1)。
closefdbool是否关闭文件描述符(默认 True)。
openercallable自定义文件打开器(默认 None)。

json.loads():将 JSON 格式的字符串解析为 Python 对象(如字典、列表)。

参数名类型描述
sstrJSON 格式的字符串。
clsclass自定义 JSON 解码器(默认 None)。
object_hookcallable自定义对象解码函数(默认 None)。
parse_floatcallable自定义浮点数解码函数(默认 None)。
parse_intcallable自定义整数解码函数(默认 None)。
parse_constantcallable自定义常量解码函数(默认 None)。
object_pairs_hookcallable自定义键值对解码函数(默认 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)])

Ⅳ、前向传播

  1. 如果传入 3 个句子,则计算三元组损失(Triplet Loss)。
  2. 如果传入 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.统计模型效果并展示

Ⅰ、计算统计预测结果

  1. 使用 assert len(labels) == len(test_question_vectors) 确保输入的长度一致。
  2. 遍历 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():记录信息级别的日志,用于输出程序运行时的普通信息

参数名类型是否必选描述
msgstr要记录的日志信息。
*args任意类型用于格式化日志信息的参数。例如,msg 中包含 {} 时,*args 会填充这些占位符。
**kwargsdict可选参数,如 exc_infostack_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 模块中最常用的配置函数,通常用于简单的日志记录需求

参数名类型说明
filenamestr指定日志文件名,日志会被写入该文件。如果未指定,日志默认输出到控制台。
filemodestr文件打开模式,默认为 'a'(追加模式)。可设置为 'w'(覆盖模式)。
formatstr定义日志输出格式。默认格式为 '%(levelname)s:%(name)s:%(message)s'
datefmtstr定义日期时间格式。默认格式为 '%Y-%m-%d %H:%M:%S'
levelint设置日志级别,低于该级别的日志将被忽略。默认级别为 WARNING
streamIO指定日志输出流(如 sys.stderr 或 sys.stdout)。
handlerslist指定处理器列表。如果指定了 handlers,则 filename 和 stream 会被忽略。

logging.getLogger():返回一个 Logger 对象,用于记录日志。如果没有指定名称,则返回根日志器(root logger)。通过 Logger 对象,可以更灵活地控制日志的输出,如添加多个处理器、设置日志级别等。

参数名类型说明
namestr日志器的名称。如果未指定或为 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

参数名类型说明
pathstr需要检查的路径(文件或目录)。

os.mkdir():创建一个目录。如果目录已存在或路径无效,会抛出 OSError 异常

参数名类型说明
pathstr要创建的目录路径。
modeint目录权限模式,默认为 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():记录信息级别的日志,用于输出程序运行中的一般性信息

参数名类型说明
msgstr要记录的日志信息。
*argsAny格式化日志信息的参数。
**kwargsAny其他关键字参数(如 exc_info)。

cuda():将张量或模型移动到 GPU 上进行计算。如果没有可用的 GPU,会抛出异常

参数名类型说明
deviceint指定 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,会抛出异常。

参数名类型说明
deviceint指定 GPU 设备编号(如 0)。
            if cuda_flag:
                batch_data = [d.cuda() for d in batch_data]
④ 损失计算

列表.append():在列表末尾添加一个元素。

参数名类型说明
objAny要添加到列表末尾的元素。

item():从张量中提取标量值,并返回其高精度值(如 int 或 float

            input_id1, input_id2, labels = batch_data   #输入变化时这里需要修改,比如多输入,多输出的情况
            loss = model(input_id1, input_id2, labels)
            train_loss.append(loss.item())
⑤ 反向传播

backward():计算损失函数关于模型参数的梯度,用于反向传播。

参数名类型说明
gradientTensor用于链式求导的梯度张量(默认为 None
            loss.backward()
⑥ 模型参数更新

optimizer.step():优化器的方法,根据计算出的梯度更新模型参数

            optimizer.step()
⑦ 日志记录

logger.info():记录信息级别的日志,用于输出程序运行中的一般性信息。

参数名类型说明
msgstr要记录的日志信息。
*argsAny格式化日志信息的参数。
**kwargsAny其他关键字参数(如 exc_info)。

np.mean():计算数组的均值

参数名类型说明
aarray_like输入数组。
axisint计算均值的轴(默认为 None)。
dtypedtype输出数组的数据类型(默认为 None)。
outndarray输出数组(默认为 None)。

evaluator.eval():evaluate 模块中导入 Evaluator 类,用于模型评估

        logger.info("epoch average loss: %f" % np.mean(train_loss))
        evaluator.eval(epoch)

Ⅷ、保存模型

os.path.join():用于连接多个路径片段,生成一个完整的路径字符串。它会根据操作系统的不同自动处理路径分隔符(如 Windows 使用 \,Linux/Mac 使用 /),确保生成的路径是有效的

参数名类型说明
path1str第一个路径片段。
path2str第二个路径片段。
*pathsstr可选的更多路径片段。

torch.save():用于保存 PyTorch 对象(如模型、张量、字典等)到文件中。它使用 Python 的 pickle 进行序列化,方便后续加载和使用

参数名类型说明
objAny要保存的对象(如模型、张量、字典等)。
fstr 或 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 不可用,调用此方法会报错

参数名类型说明
deviceint 或 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():用于在列表末尾添加一个元素。它会直接修改原列表,而不会创建新的列表

参数名类型说明
objAny要添加到列表末尾的元素。

torch.no_grad():用于临时禁用梯度计算,通常在模型推理或评估时使用。它可以减少内存消耗并加速计算。

torch.stack():沿指定维度连接多个张量,生成一个新的张量。所有输入张量的形状必须相同。

参数名类型说明
tensorslist 或 tuple要连接的张量序列。
dimint指定连接的维度。

torch.cuda.is_available():检查当前环境是否支持 CUDA(即是否有可用的 GPU)。如果支持,返回 True;否则返回 False

cuda():将张量或模型移动到 GPU 上进行计算。如果没有指定设备编号,默认使用第一个 GPU(cuda:0)。

参数名类型说明
deviceint 或 torch.device指定 GPU 设备编号(如 0

torch.nn.functional.normalize():对输入张量在指定维度上进行归一化,使其 L-p 范数为 1

参数名类型说明
inputTensor输入张量。
pfloat范数的类型,默认为 2(L2 范数)。
dimint指定归一化的维度。
epsfloat防止除零的小值,默认为 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():将中文句子分割成独立的单词,支持精确模式、全模式和搜索引擎模式。

参数名类型说明
sentencestr需要分词的字符串。
cut_allbool是否使用全模式,默认为 False(精确模式)。
HMMbool是否使用隐马尔可夫模型(HMM),默认为 True

字典.get():安全地获取字典中指定键的值。如果键不存在,返回默认值(默认为 None),而不会引发 KeyError 异常。

参数名类型说明
keyAny要查找的键。
defaultAny键不存在时返回的默认值,默认为 None

列表.append():在列表的末尾添加一个元素,直接修改原列表。

参数名类型说明
objAny要添加到列表末尾的元素。
    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 来替代。

参数名类型说明
datalist 或 tuple用于初始化张量的数据。

torch.cuda.is_available():检查当前系统是否支持 CUDA(即是否有可用的 GPU),并返回布尔值

cuda():将张量或模型移动到 GPU 上进行计算。如果没有指定设备编号,默认使用第一个 GPU(cuda:0

参数名类型说明
deviceint 或 torch.device指定 GPU 设备编号(如 0)。

torch.no_grad():用于临时禁用梯度计算,通常在模型推理或评估时使用。它可以减少内存消耗并加速计算。

torch.mm():用于两个二维张量(矩阵)之间的矩阵乘法。仅支持二维张量,不支持高维张量或广播机制。

参数名类型说明
inputTensor第一个矩阵(二维张量)。
mat2Tensor第二个矩阵(二维张量)。

squeeze():从张量的形状中移除所有维度为 1 的维度,从而对张量进行降维。

参数名类型说明
dimint 或 None指定要移除的维度,默认为 None(移除所有维度为 1 的维度)。

unsqueeze():在指定维度上增加一个维度,维度大小为 1。

参数名类型说明
dimint

指定要增加维度的位置。

    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_dictdict包含模型参数的字典,键是参数名称,值是参数张量。
strictbool,默认为 True是否严格匹配状态字典的键。如果为 False,则允许部分匹配。

torch.load():加载由 torch.save() 保存的 PyTorch 对象,例如模型的状态字典(state_dict)、整个模型、张量等。

参数名类型/默认值描述
fstr 或 os.PathLike 或文件对象要加载的文件路径或文件对象。
map_locationcallable 或 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)


http://www.kler.cn/a/584292.html

相关文章:

  • 数字化新零售与 AI 大模型,如何重塑大健康赛道?​
  • IDE 使用技巧与插件推荐:全面提升开发效率
  • ESP32移植Openharmony外设篇(10)inmp441麦克风
  • 基于PyTorch通信算子的分布式训练阻塞定位方法
  • 算法手记3
  • 算法日记40:最长上升子序列LIS(单调栈优化)n*log^n
  • DeepSeek一键生成可视化看板
  • 3.12-1 html讲解
  • QQuick-Binding的介绍
  • e2studio开发RA4L1(1)---开发板测试
  • 【Linux】动/静态库
  • 重生之我在学Vue--第10天 Vue 3 项目收尾与部署
  • Unity Lerp和InverseLerp函数用处
  • 【C++】每日一练(用队列实现栈)
  • 【fnOS飞牛云NAS本地部署跨平台视频下载工具MediaGo与远程访问下载视频流程】
  • VS Code 配置优化指南
  • 【TES817】基于XCZU19EG FPGA的高性能实时信号处理平台
  • 【从零开始学习计算机科学】数据库系统(七)并发控制技术
  • 元宇宙与数字孪生
  • 如何查看mysql某个表占用的空间大小