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), invp3 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 invp3.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 = bucketsize5.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]
#05.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, n5.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, n6 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)#对当前批次的数据进行处理，以获得适当的格式和反向排列
)
之rename)
 | 传递临时对象作为线程参数的一些问题Ⅲ)
