【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 dl2.处理数据
Ⅰ、补齐或截断
截断序列：如果输入序列的长度超过
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 x2.计算句子间相似度 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)
return3.统计模型效果并展示
Ⅰ、计算统计预测结果
- 使用
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("--------------------")
return4.模型表现评估函数
评估模型在验证集上的表现
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()
return5.模型效果测试
# -*- 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_data2、日志配置
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)
return3.句子编码
将输入的文本
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_id4.根据相似度预测
根据输入的句子
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)
