PyTorch中高效查找张量B元素在张量A中的所有索引位置

1. 问题背景与挑战

在数据处理和机器学习任务中，我们经常需要在一个大型张量（例如张量a）中查找另一个张量（例如张量b）中所有元素的出现位置。具体来说，给定张量a（[1,2,3,3,2,1,4,5,9]）和张量b（[1,2,3,9]），我们的目标是为b中的每个值，找到它在a中出现的所有索引。理想的输出形式类似于 [[0,5], [1,4], [2,3], [8]]，其中每个子列表对应b中一个值的索引。

直接使用PyTorch的广播机制，例如通过扩展维度创建布尔掩码 (B == A_expanded)，虽然能够实现功能，但对于非常大的张量A和B，这种操作会消耗巨大的内存，导致程序崩溃或运行效率低下。因此，我们需要寻找内存效率更高、同时保持合理计算性能的解决方案。

2. 传统广播方法的局限性

最初的尝试往往会利用PyTorch强大的广播能力。例如，以下代码片段展示了通过扩展维度进行广播的方法：

import torch

def vectorized_find_indices_broadcasting(A, B):
    # 扩展A的维度以与B进行广播比较
    # A_expanded 的形状将是 (A.size(0), 1, 1)
    A_expanded = A[:, None, None]

    # 创建布尔掩码，形状为 (A.size(0), B.size(0), 1)
    # mask[i, j, k] 为 True 表示 A[i] == B[j]
    mask = (B == A_expanded)

    # 获取匹配的索引。这里会生成一个形状为 (A.size(0), B.size(0), 1) 的张量
    # 其中对应 True 的位置是A的索引，False 的位置是 -1
    indices = torch.where(mask, torch.arange(A.size(0), device=A.device)[:, None, None], torch.tensor(-1, device=A.device))

    # 调整结果形状，使其更符合期望的输出结构
    # 最终形状可能需要进一步处理以得到 [[idx1, idx2], ...] 形式
    result = indices.permute(1, 2, 0)
    return result

# 示例
A = torch.tensor([1,2,3,3,2,1,4,5,9])
B = torch.tensor([1,2,3,9])
# result_broadcasting = vectorized_find_indices_broadcasting(A, B)
# print(result_broadcasting)

尽管上述方法在逻辑上是“完全向量化”的，但其核心问题在于 mask 张量和 indices 张量的大小会急剧增加，其维度通常是 (len(A), len(B), ...)。当 len(A) 和 len(B) 都非常大时，即使是中间结果也会轻易耗尽可用内存，使得这种方法不适用于大规模张量。

3. 优化方案一：混合广播与Python循环

为了克服纯广播的内存限制，我们可以采用一种混合方法：首先利用有限的广播操作找出所有匹配的索引对，然后通过Python循环将这些索引对归类到对应的张量B元素下。这种方法在内存和计算效率之间找到了一个较好的平衡点。

3.1 实现原理

1. 找出所有匹配对： 使用 a.unsqueeze(1) == b 进行比较。a.unsqueeze(1) 将张量A的维度从 (N,) 变为 (N, 1)，使其可以与 b (形状 (M,)) 进行广播比较，生成一个形状为 (N, M) 的布尔张量。True 表示 A[i] == B[j]。
2. 获取匹配索引： 对布尔张量调用 .nonzero() 方法，将返回一个 (K, 2) 的张量，其中 K 是匹配的总数。每一行 (a_idx, b_idx) 表示 A[a_idx] 与 B[b_idx] 相匹配。
3. 归类索引： 初始化一个与张量B长度相同的空列表的列表。遍历 (a_idx, b_idx) 对，将 a_idx 添加到 output[b_idx] 中。

3.2 代码示例

import torch

def find_indices_hybrid(a, b):
    # 1. 找出所有匹配的 (A_index, B_index) 对
    # a.unsqueeze(1) 将 a 变为 (len(a), 1)
    # (a.unsqueeze(1) == b) 广播为 (len(a), len(b)) 的布尔张量
    # .nonzero() 返回所有 True 值的坐标，形状为 (K, 2)，其中 K 是匹配总数
    # 每行 (a_idx, b_idx) 表示 a[a_idx] == b[b_idx]
    overlap_idxs = (a.unsqueeze(1) == b).nonzero()

    # 2. 初始化结果列表，为B中每个元素准备一个空列表
    output = [[] for _ in b]

    # 3. 遍历匹配对，将A的索引归类到B的对应元素下
    for a_idx, b_idx in overlap_idxs:
        output[b_idx.item()].append(a_idx.item())

    return output

# 示例使用
A = torch.tensor([1,2,3,3,2,1,4,5,9])
B = torch.tensor([1,2,3,9])
result_hybrid = find_indices_hybrid(A, B)
print(f"混合方法结果: {result_hybrid}") # 预期: [[0, 5], [1, 4], [2, 3], [8]]

A_large = torch.arange(100000) # 模拟大张量A
B_large = torch.tensor([100, 50000, 99999, 100001]) # B中可能包含A中不存在的值
result_large_hybrid = find_indices_hybrid(A_large, B_large)
print(f"大型张量混合方法结果 (部分): {result_large_hybrid[:2]}...")

3.3 优缺点分析

• 优点：
  • 相比纯广播方法，overlap_idxs 的内存占用大大降低，它只存储实际匹配的索引对，而不是整个 (len(A), len(B)) 大小的布尔矩阵。
  • nonzero() 操作是高度优化的C++实现，效率较高。
  • 在 len(A) * len(B) 比较大但匹配数量 K 相对较小的情况下表现良好。
• 缺点：
  • 仍然需要创建一个 (len(A), len(B)) 大小的布尔张量作为中间结果（尽管 nonzero() 可以在某些情况下避免完全实例化）。
  • 最后的归类步骤是一个Python级别的循环，对于 K 非常大（即匹配非常多）的情况，可能会成为性能瓶颈。

4. 优化方案二：纯Python循环遍历张量B

当张量B的长度相对较小，或者希望将内存使用降到最低时，可以采用纯Python循环遍历张量B的每个元素，并在张量A中独立查找其索引。

商汤商量

商汤科技研发的AI对话工具，商量商量，都能解决。

36

查看详情

4.1 实现原理

1. 遍历B的每个元素： 使用Python for 循环迭代张量B中的每一个值 _b。
2. 在A中查找： 对于每个 _b，使用 (a == _b).nonzero() 在张量A中查找所有匹配的索引。nonzero() 返回的张量通常是 (num_matches, 1) 的形状。
3. 处理结果： 使用 .squeeze().tolist() 将结果转换为Python列表。如果 _b 在A中没有匹配项，nonzero() 将返回空张量，squeeze() 后会得到空列表。如果只有一个匹配项，squeeze() 会将其降为标量，需要特殊处理以确保其始终为列表。
4. 收集结果： 将每个 _b 对应的索引列表添加到最终结果列表中。

4.2 代码示例

import torch

def find_indices_pure_python_loop(a, b):
    output = []
    for _b in b:
        # 查找当前 _b 在 a 中的所有索引
        idxs_tensor = (a == _b).nonzero().squeeze()

        # 将张量转换为Python列表
        # 注意处理只有单个匹配项时 squeeze() 会将张量变为标量的情况
        if idxs_tensor.dim() == 0: # 如果是标量（只有一个匹配项）
            idxs = [idxs_tensor.item()]
        elif idxs_tensor.numel() == 0: # 如果没有匹配项
            idxs = []
        else: # 多个匹配项
            idxs = idxs_tensor.tolist()

        output.append(idxs)
    return output

# 示例使用
A = torch.tensor([1,2,3,3,2,1,4,5,9])
B = torch.tensor([1,2,3,9, 10]) # 添加一个不存在的值
result_pure_loop = find_indices_pure_python_loop(A, B)
print(f"纯Python循环方法结果: {result_pure_loop}") # 预期: [[0, 5], [1, 4], [2, 3], [8], []]

A_large = torch.arange(100000) # 模拟大张量A
B_small = torch.tensor([100, 50000, 99999, 100001]) # B的长度较小
result_large_A_small_B_loop = find_indices_pure_python_loop(A_large, B_small)
print(f"大型A小型B纯循环方法结果: {result_large_A_small_B_loop}")

4.3 优缺点分析

• 优点：
  • 内存使用效率最高，每次只处理 B 中的一个元素，不会产生大的中间张量。
  • 对于 len(B) 较小而 len(A) 很大的情况，这种方法可能比混合方法更优，因为它避免了 (len(A), len(B)) 大小的布尔张量创建。
• 缺点：
  • 完全依赖Python循环，相比于PyTorch的向量化操作，计算速度可能较慢，尤其当 len(B) 非常大时。
  • 每次迭代都需要在GPU/CPU之间进行数据传输（如果 A 在GPU上），这会增加开销。

5. 选择策略与注意事项

在选择上述两种优化方案时，需要根据实际场景中的张量大小、内存限制和性能要求进行权衡：

• *当 len(A) 和 len(B) 都非常大，但预期匹配的数量 K 相对较小（即 `K << len(A) len(B)）时，推荐使用“混合广播与Python循环”方案。** 这种方案利用了nonzero()` 的高效性，并避免了创建巨大的布尔矩阵。
• 当 len(B) 相对较小，而 len(A) 非常大，且内存是主要限制因素时，推荐使用“纯Python循环遍历张量B”方案。 这种方案的内存占用最小，但可能会牺牲一些计算速度。
• 如果 len(A) 和 len(B) 都不是特别大，或者内存不是瓶颈，可以考虑最初的完全广播方法（但需要确保内存能够承受）。

注意事项：

• 未找到的元素： 两种优化方案都能自然地处理张量B中的元素在张量A中不存在的情况。此时，该元素对应的输出列表中将是空列表 []。
• 数据类型与设备： 确保张量A和B具有兼容的数据类型，并且它们位于相同的设备（CPU或GPU）上，以避免不必要的数据传输开销。
• 性能分析： 对于生产环境，建议使用 torch.cuda.synchronize() 和 time.time() 或 torch.benchmark 等工具对不同方法进行性能测试，以选择最适合具体工作负载的方案。

6. 总结

在PyTorch中高效地查找一个张量中另一个张量元素的索引是一个常见的需求，尤其是在处理大规模数据时，内存效率至关重要。本文介绍了两种优化的方法：结合部分广播和Python循环的混合方案，以及纯Python循环遍历张量B的方案。理解它们的实现原理、优缺点和适用场景，可以帮助开发者根据具体需求选择最合适的策略，从而在内存使用和计算性能之间取得最佳平衡。

以上就是PyTorch中高效查找张量B元素在张量A中的所有索引位置的详细内容，更多请关注php中文网其它相关文章！