1 argsort
'''
目的是对一个列表 seq 中的子列表或数组按其长度进行排序,并返回排序后的索引。
具体来说,它会按长度的降序排列,这意味着最长的子列表/数组的索引会在前面。
'''
def argsort(seq):"""sort by length in reverse order---seq (list[array[int32]])"""return [x for x,y in sorted(enumerate(seq),key = lambda x: len(x[1]),reverse=True)]
- `enumerate(seq)` 返回一个枚举对象,其中包含 `seq` 中每个项的索引和值。
- `key = lambda x: len(x[1])` 是排序的关键。它告诉 `sorted` 函数按照长度 (`len(x[1])`) 进行排序。这里的 `x[1]` 表示 `enumerate` 枚举对象的值,即 `seq` 中的子列表/数组。
- `reverse=True` 指定排序应按降序进行。
- 最后,列表解析 `[x for x,y in ...]` 提取了排序后的索引值,这些索引值随后返回。
1.1 举例
seq = [[1, 2], [1, 2, 3, 4], [1], [1, 2, 3]]
result = argsort(seq)
print(result)
#[1, 3, 0, 2]
2 pad_array相关
2.1 pad_array
'''
对给定的数组 a 进行填充,使其长度达到指定的 max_length。填充的值是 PAD
'''def pad_array(a, max_length, PAD=constants.PAD):"""a (array[int32])"""return np.concatenate((a, [PAD]*(max_length - len(a))))
- `[PAD]*(max_length - len(a))`:这部分代码会创建一个新的列表,其长度为 `(max_length - len(a))`,并且所有的元素都是 `PAD`
- `np.concatenate((a, ...))`:这将原始数组 `a` 和新生成的 `PAD` 值列表连接起来,从而产生一个新的、长度为 `max_length` 的数组。
2.2 pad_arrays
'''
接受一个数组的列表(或序列的序列)a,并确保所有这些数组都填充到与列表中最长的数组相同的长度
'''
def pad_arrays(a):max_length = max(map(len, a)) #确定 a 中的所有数组中的最大长度a = [pad_array(a[i], max_length) for i in range(len(a))]#遍历 a 中的每个数组,然后使用先前定义的 pad_array 函数将每个数组填充到 max_lengtha = np.stack(a).astype(np.int)#将填充后的数组堆叠成一个二维数组,确保所有的值都是整数类型return torch.LongTensor(a)
2.3 pad_arrays_pair
处理源-目标(src-trg)序列对,使源/目标序列分别填充到相同的长度
def pad_arrays_pair(src, trg, keep_invp=False):"""输入:src (list[array[int32]]):一个包含源序列的列表。trg (list[array[int32]]):一个包含目标序列的列表。keep_invp:一个布尔值,用来决定是否保留反向排列的信息。输出:src (seq_len1, batch):填充后的源序列的张量,形状为 (序列长度, 批量大小)。trg (seq_len2, batch):填充后的目标序列的张量,形状为 (序列长度, 批量大小)。lengths (1, batch):一个张量,包含源序列的原始长度。invp (batch,):一个张量,包含反向排列的信息,只有在 keep_invp 为 True 时才返回。"""TD = namedtuple('TD', ['src', 'lengths', 'trg', 'invp'])assert len(src) == len(trg), "source and target should have the same length"#确保源序列和目标序列的长度是相同的。idx = argsort(src)src = list(np.array(src)[idx])trg = list(np.array(trg)[idx])'''使用 `argsort` 函数来获取按长度降序排序后的序列的索引。然后,使用这些索引来对 `src` 和 `trg` 进行重新排序。'''lengths = list(map(len, src))lengths = torch.LongTensor(lengths)#计算每个源序列的长度并将其转换为一个 PyTorch 张量。src = pad_arrays(src)trg = pad_arrays(trg)#使用之前定义的 `pad_arrays` 函数来对 `src` 和 `trg` 中的每个序列进行填充。#(batch,seq_len1) (batch,seq_len2)if keep_invp == True:invp = torch.LongTensor(invpermute(idx))# (batch, seq_len) => (seq_len, batch)return TD(src=src.t().contiguous(), lengths=lengths.view(1, -1), trg=trg.t().contiguous(), invp=invp)else:# (batch, seq_len) => (seq_len, batch)return TD(src=src.t().contiguous(), lengths=lengths.view(1, -1), trg=trg.t().contiguous(), invp=[])
2.4 pad_arrays_keep_invp
'''
对输入的序列src进行填充,以便它们在批处理中具有相同的长度。
同时,它还保存并返回一个逆排列,这样你可以使用这个逆排列来恢复原始的顺序
'''
def pad_arrays_keep_invp(src):"""Input:src (list[array[int32]])---Output:src (seq_len, batch)lengths (1, batch)invp (batch,): inverse permutation, src.t()[invp] gets original order"""idx = argsort(src)#对src中的序列按长度进行排序,返回一个排列索引src = list(np.array(src)[idx])#通过索引idx从src中提取排序后的序列lengths = list(map(len, src))lengths = torch.LongTensor(lengths)#获取每个序列的长度src = pad_arrays(src)#序列进行填充,使所有序列具有相同的长度invp = torch.LongTensor(invpermute(idx))#获得idx的逆排列,之后可以使用这个逆排列来恢复原始的顺序return src.t().contiguous(), lengths.view(1, -1), invp
3 invpermute
'''
计算一个排列的逆排列给定一个排列 p,它的逆排列是一个排列 invp,满足 invp[p[i]] = i 和 p[invp[i]] = i 对所有 i 成立
'''
def invpermute(p):"""inverse permutation"""p = np.asarray(p)invp = np.empty_like(p)for i in range(p.size):invp[p[i]] = i#对于每个 i,设置 invp 在 p[i] 的位置上的值为 ireturn invp
3.1 举例
p = [2, 0, 1]
invp = invpermute(p)
invp
#[1, 2, 0]
- invp的第p[0]个位置(第2个位置)值为0
- invp的第p[1]个位置(第0个位置)值为1
- invp的第p[2]个位置(第1个位置)值为2
4 random_subseq
'''
根据给定的 rate 从输入的数组 a 中随机丢弃一些元素,但始终保留首尾两个元素
'''
def random_subseq(a, rate):idx = np.random.rand(len(a)) < rate#生成一个与 a 同长度的随机数组,其值在 [0, 1) 之间。#然后,该随机数组中小于 rate 的值被标记为 True,其他的被标记为 Falseidx[0], idx[-1] = True, True'''无论先前的随机操作的结果如何,这行代码确保数组 idx 的首元素和尾元素始终为 True这意味着在最终结果中,a 的首尾元素永远不会被丢弃'''return a[idx]
5 DataLoader
5.1 init
class DataLoader():"""srcfile: source file nametrgfile: target file namebatch: batch sizevalidate: if validate = True return batch orderly otherwise returnbatch randomly"""def __init__(self, srcfile, trgfile, mtafile, batch, bucketsize, validate=False):self.srcfile = srcfileself.trgfile = trgfileself.mtafile = mtafile$源、目标、元文件文件名self.batch = batchself.validate = validate#如果为True,则按照顺序返回batch,否则随即返回#self.bucketsize = [(30, 30), (30, 50), (50, 50), (50, 70), (70, 70)]self.bucketsize = bucketsize
5.1.1 桶策略
- 桶是一种策略,用于在序列长度可变的情况下处理数据。
- 例如,假设我们有不同长度的句子,我们可能不希望为每个句子单独创建一个批处理,因为这会非常低效。
- 桶策略是将句子放入一个接近它们长度的桶中,以尽量减少填充。
5.2 insert
'''
将一个源序列(s)、一个目标序列(t)和一个元数据序列(m)插入到适当的桶中
'''
def insert(self, s, t, m):for i in range(len(self.bucketsize)):#遍历每个桶的大小if len(s) <= self.bucketsize[i][0] and len(t) <= self.bucketsize[i][1]:#检查序列 s 和 t 的长度是否匹配当前桶的大小self.srcdata[i].append(np.array(s, dtype=np.int32))self.trgdata[i].append(np.array(t, dtype=np.int32))self.mtadata[i].append(np.array(m, dtype=np.float32))return 1return 0
#如果循环结束后,数据没有被插入到任何桶中,则返回 0,表示数据没有被插入
5.3 load
注:self.srcdata = list(map(np.array, self.srcdata)) 这几行需要用老版本的numpy(我是用1.23.5,有warning不报错),新版本的numpy,需要srcdata长度相同
'''
从指定的源文件、目标文件和元数据文件中加载数据,并按照之前定义的桶大小进行组织
'''
def load(self, max_num_line=0):self.srcdata = [[] for _ in range(len(self.bucketsize))]self.trgdata = [[] for _ in range(len(self.bucketsize))]self.mtadata = [[] for _ in range(len(self.bucketsize))]#根据桶的数量为 self.srcdata, self.trgdata 和 self.mtadata 初始化空列表。srcstream, trgstream, mtastream = open(self.srcfile, 'r'), open(self.trgfile, 'r'), open(self.mtafile, 'r')#从源、目标和元数据文件中读取数据num_line = 0for (s, t, m) in zip(srcstream, trgstream, mtastream):#每次读取srcstream, trgstream, mtastream的各一行s = [int(x) for x in s.split()]#源数据直接转为整数列表t = [constants.BOS] + [int(x) for x in t.split()] + [constants.EOS]#目标数据前后添加起始(BOS)和结束(EOS)标记m = [float(x) for x in m.split()]num_line += self.insert(s, t, m)#使用 insert 方法将数据插入到适当的桶中#如果插入成功,那么num_line加一if num_line >= max_num_line and max_num_line > 0: break#如果达到最大行数 max_num_line(如果设置了的话),则停止读取if num_line % 500000 == 0:print("Read line {}".format(num_line))#每读取 500,000 行,打印一条消息表示进度## 如果处于验证模式,合并所有桶:if self.validate == True:self.srcdata = np.array(merge(*self.srcdata))self.trgdata = np.array(merge(*self.trgdata))self.mtadata = np.array(merge(*self.mtadata))self.start = 0self.size = len(self.srcdata)else:#否则,将数据转为 NumPy 数组,并计算桶的分配比例:self.srcdata = list(map(np.array, self.srcdata))self.trgdata = list(map(np.array, self.trgdata))self.mtadata = list(map(np.array, self.mtadata))self.allocation = list(map(len, self.srcdata))#计算每个桶中的数据量,并存储在 self.allocation 中。self.p = np.array(self.allocation) / sum(self.allocation)#计算每个桶的相对大小,并存储在 self.p 中。srcstream.close(), trgstream.close(), mtastream.close()#关闭源、目标和元数据文件的文件流。
5.4 getbatch_one
'''
获取一个批次的数据
'''
def getbatch_one(self):if self.validate == True:#验证模式src = self.srcdata[self.start:self.start+self.batch]trg = self.trgdata[self.start:self.start+self.batch]mta = self.mtadata[self.start:self.start+self.batch]#从当前的start位置开始,连续选取batch大小的数据self.start += self.batch#更新start以便下一次获取批次时从新的位置开始。if self.start >= self.size:self.start = 0#如果start加上batch的大小超过了数据的大小,那么start将被重置为0#这意味着验证数据是循环的return list(src), list(trg), list(mta)#返回这个批次的源数据、目标数据和元数据else:## 非验证模式sample = np.random.multinomial(1, self.p)#使用多项分布从self.p中抽取一个样本,表示抽取哪一个桶。#这里的self.p是一个数组,表示每个"bucket"被选中的概率'''如果self.p = [0.2, 0.5, 0.3],那么返回的数组可能是[0, 1, 0],表示第二个"bucket"被选中'''bucket = np.nonzero(sample)[0][0]'''np.nonzero(sample) 返回一个元组,其中包含了sample中所有非零元素的索引因为sample中只有一个元素是1(即被选中的"bucket"),所以这个函数会返回一个只有一个元素的数组[0][0] 提取这个数组的第一个元素,即被选中的"bucket"的索引'''idx = np.random.choice(len(self.srcdata[bucket]), self.batch)#从所选"bucket"中随机选择batch大小的数据。src = self.srcdata[bucket][idx]trg = self.trgdata[bucket][idx]mta = self.mtadata[bucket][idx]return list(src), list(trg), list(mta)#返回这个批次的源数据、目标数据和元数据。
5.4.1 选择桶的两行举例
import numpy as np
sample=np.random.multinomial(1, np.array([0.2,0.3,0.5]))
sample
#array([1, 0, 0])bucket = np.nonzero(sample)
bucket
#(array([0]),)bucket[0][0]
#0
5.5 getbatch_generative
'''
从数据加载器中获取一个批次的数据
'''
def getbatch_generative(self):src, trg, _ = self.getbatch_one()#调用getbatch_one函数,该函数返回一批数据。它返回三个值:源数据、目标数据和元数据return pad_arrays_pair(src, trg, keep_invp=False)'''填充src和trg,使得他们分别是相同长度返回值一个名为TD的namedtuple,其中包含填充和排序后的src和trg,以及源数据的长度'''
5.6 getbatch_discriminative_cross
'''
获取三个batch的数据,a、p和n[锚点(anchor)、正例(positive)和负例(negative)]
'''
def getbatch_discriminative_cross(self):def distance(x, y):return np.linalg.norm(x - y)#定义了一个简单的欧几里得距离函数来计算两个向量之间的距离。a_src, a_trg, a_mta = self.getbatch_one()p_src, p_trg, p_mta = self.getbatch_one()n_src, n_trg, n_mta = self.getbatch_one()'''连续三次调用getbatch_one方法,获取三个不同的批次数据:a、p和n代表了锚点(anchor)、正例(positive)和负例(negative)。'''for i in range(len(a_src)):if distance(a_mta[i], p_mta[i]) > distance(a_mta[i], n_mta[i]):p_src[i], n_src[i] = n_src[i], p_src[i]p_trg[i], n_trg[i] = n_trg[i], p_trg[i]p_mta[i], n_mta[i] = n_mta[i], p_mta[i]'''如果a和p之间的距离大于a和n之间的距离,我们交换p和n。这是为了确保p是与a更接近的点,而n是更远离的点'''a = pad_arrays_pair(a_src, a_trg, keep_invp=True)p = pad_arrays_pair(p_src, p_trg, keep_invp=True)n = pad_arrays_pair(n_src, n_trg, keep_invp=True)'''对于每个数据批次,调用pad_arrays_pair进行填充和排序。这里保持了逆排列信息,因为在后续的处理中需要它'''return a, p, n
5.7 getbatch_discriminative_inner
'''
从给定的输入数据中生成a(锚点)、p(正例)、和n(负例)的批次,类似于上一个函数。但在这里,所有这些批次数据都是从同一个初始数据的不同部分产生的。
'''
def getbatch_discriminative_inner(self):a_src, a_trg = [], []p_src, p_trg = [], []n_src, n_trg = [], []_, trgs, _ = self.getbatch_one()#从getbatch_one方法中仅获取一个batch的目标数据trgs#回忆:getbatch_one返回的东西还没有padfor i in range(len(trgs)):trg = trgs[i][1:-1]#目标轨迹掐头去尾(BOS、EOS)if len(trg) < 10: continue#如果目标轨迹长度小于10,那么不考虑这条轨迹a1, a3, a5 = 0, len(trg)//2, len(trg)a2, a4 = (a1 + a3)//2, (a3 + a5)//2#将trg数据(不考虑开头和结尾)分为三部分:a1到a5'''根据一个随机的rate,使用random_subseq函数来从这三部分中随机选取子序列'''rate = np.random.choice([0.5, 0.6, 0.8])if np.random.rand() > 0.5:a_src.append(random_subseq(trg[a1:a4], rate))a_trg.append(np.r_[constants.BOS, trg[a1:a4], constants.EOS])p_src.append(random_subseq(trg[a2:a5], rate))p_trg.append(np.r_[constants.BOS, trg[a2:a5], constants.EOS])n_src.append(random_subseq(trg[a3:a5], rate))n_trg.append(np.r_[constants.BOS, trg[a3:a5], constants.EOS])'''如果np.random.rand() > 0.5,那么:取a1到a4为锚点、a2到a5为正例、a3到a5为负例'''else:a_src.append(random_subseq(trg[a2:a5], rate))a_trg.append(np.r_[constants.BOS, trg[a2:a5], constants.EOS])p_src.append(random_subseq(trg[a1:a4], rate))p_trg.append(np.r_[constants.BOS, trg[a1:a4], constants.EOS])n_src.append(random_subseq(trg[a1:a3], rate))n_trg.append(np.r_[constants.BOS, trg[a1:a3], constants.EOS])a = pad_arrays_pair(a_src, a_trg, keep_invp=True)p = pad_arrays_pair(p_src, p_trg, keep_invp=True)n = pad_arrays_pair(n_src, n_trg, keep_invp=True)'''使用pad_arrays_pair函数,我们将a_src, a_trg, p_src, p_trg, n_src, 和n_trg填充并排序'''return a, p, n
6 DataOrderScaner
'''
从一个源文件中扫描和加载数据,并以批量方式返回这些数据
'''
class DataOrderScaner():def __init__(self, srcfile, batch):self.srcfile = srcfileself.batch = batchself.srcdata = []self.start = 0# 初始化函数,设置源文件、批处理大小,并初始化存放数据的列表'''从指定的源文件self.srcfile中加载数据,并将其存储到self.srcdata列表中'''def load(self, max_num_line=0):num_line = 0with open(self.srcfile, 'r') as srcstream:for s in srcstream:s = [int(x) for x in s.split()]self.srcdata.append(np.array(s, dtype=np.int32))#从指定的源文件self.srcfile中加载数据,并将其存储到self.srcdata列表中num_line += 1if max_num_line > 0 and num_line >= max_num_line:breakself.size = len(self.srcdata)self.start = 0'''从加载的数据中按批次获取数据,并进行适当的格式化处理'''def getbatch(self):"""Output:src (seq_len, batch)lengths (1, batch)invp (batch,): inverse permutation, src.t()[invp] gets original order"""if self.start >= self.size:return None, None, None'''首先,检查self.start(当前批次的开始位置)是否已经超过或等于self.size(总的数据大小)。如果是这样,表示所有数据都已经被提取过了,所以直接返回None'''src = self.srcdata[self.start:self.start+self.batch]#从self.srcdata中提取当前批次的数据
=self.start += self.batchreturn pad_arrays_keep_invp(src)#对当前批次的数据进行处理,以获得适当的格式和反向排列