torch.nested 的源代码

# mypy: 允许未类型化定义
from 打字 导入 可选, 联合

导入 火炬
导入 torch.nn.functional as F
from 火炬 导入 SymInt, 张量
from torch._C 导入 _添加文档字符串, _nested  # 类型：忽略[已定义]
from torch 的类型 导入 _device as 设备, _dtype as 数据类型


全部 = [
    "转换为填充张量",
    "转换为嵌套张量",
    "嵌套张量",
    "从锯齿状嵌套张量",
    "窄",
    "掩码选择",
]

仅允许这些用于 NJT 的 weights_only 加载
from ._internal.nested_tensor 导入 _rebuild_njt, NestedTensor as _NestedTensor


火炬.序列化.添加安全全局变量[_嵌套张量, _重建_njt])


定义 嵌套张量(
    ts: 联合[张量, 列表[张量] 元组[张量, ...]],
    数据类型: 可选[数据类型] = 无,
    设备: 可选[设备] = 无,
    布局=无,
) -> 张量:
    r""
构建一个保留自动微分历史的嵌套张量，从张量或列表/元组中
张量

如果传递了嵌套张量，它将直接返回，除非设备/数据类型/布局
不同。请注意，转换设备/数据类型将导致复制，而转换布局
此功能目前不支持。

如果传入非嵌套张量，则将其视为大小一致的成分批次。
如果传入的设备/数据类型与输入不同，或者输入是非连续的，则会发生复制。否则，将直接使用输入的存储。
如果输入是非连续的，则会发生复制。否则，将直接使用输入的存储。

如果提供了张量列表，则在构建嵌套张量时，列表中的张量总是被复制
的。

参数：
ts（张量或 List[Tensor]或 Tuple[Tensor]）：要作为嵌套张量处理的张量
或具有相同 ndim 的张量列表/元组

关键字参数：
dtype (:class:`torch.dtype`, 可选): 返回嵌套张量的期望类型。
默认：如果为 None，则与列表中最左侧张量的 :class:`torch.dtype` 相同。
device (:class:`torch.device`, 可选): 返回嵌套张量的期望设备。
默认：如果为 None，则与列表中最左侧张量相同的 :class:`torch.device`
layout (:class:`torch.layout`，可选)：返回嵌套张量的期望布局。
仅支持 Strided 和 Jagged 布局。默认：如果为 None，则使用 Strided 布局。

示例::

        >>> a = torch.arange(3, dtype=torch.float, requires_grad=True)
        >>> b = torch.arange(5, dtype=torch.float, requires_grad=True)
        >>> nt = torch.nested.as_nested_tensor([a, b])
        >>> nt.is_leaf
        False
        >>> fake_grad = torch.nested.nested_tensor([torch.ones_like(a), torch.zeros_like(b)])
        >>> nt.backward(fake_grad)
        >>> a.grad
        tensor([1., 1., 1.])
        >>> b.grad
        tensor([0., 0., 0., 0., 0.])
        >>> c = torch.randn(3, 5, requires_grad=True)
        >>> nt2 = torch.nested.as_nested_tensor(c)
    """
    张量列表 = isinstance(ts, (列表, 元组)) 以及 所有(
        isinstance(t, 张量) 对于 t 进入 ts
    )
    如果 not isinstance(ts, 张量) 以及 not is_tensor_list:
        抛出 类型错误(
            "as_nested_tensor()：期望第一个参数为张量或张量列表/元组"
        )
    # 将元组转换为列表（如果需要）
    如果 is_tensor_list 以及 not isinstance(简体中文, 列表):
        简体中文 = 列表(简体中文)

    如果 isinstance(简体中文, 张量) 以及 ts.dim() < 2:
        抛出 RuntimeError(
            "as_nested_tensor() 函数期望输入的 tensor 参数维度大于 1"
        )

    如果 isinstance(ts, 张量) 以及 ts.是嵌套的:
        如果 布局 == ts.布局:
            # 返回输入直接或输入复制到设备 / 数据类型
            返回 ts.到(设备=设备, 数据类型=数据类型)
        否则:
            # TODO: 当存在时，只需使用 nt.to(layout=layout)。
            抛出 RuntimeError(
                "as_nested_tensor(): 不支持转换嵌套张量布局"
            )

    如果 布局 是 无:
        布局 = 火炬.稀疏的
    如果 布局 == 火炬.步进:
        如果 isinstance(ts, 张量):
            # 可能需要连续()来获取展平视图。
            # 我们可能需要更精确地确定何时进行此操作以作为优化。
            缓冲区 = ts.连续的().视图(-1).到(设备=设备, 数据类型=数据类型)
            嵌套尺寸 = 火炬.张量[t.形状 对于 t 进入 ts])
            返回 火炬._从缓冲区获取嵌套视图(
                缓冲区,
                嵌套尺寸,
                *火炬._嵌套连续步长偏移量计算(嵌套尺寸),
            )
        否则:
            断言 isinstance(ts, 列表)
            返回 火炬._从张量列表创建嵌套张量(ts, 数据类型, 无, 设备, 无)
    elif 布局 == 火炬.jagged:
        如果 isinstance(ts, 张量):
            如果 设备 是 无:
                设备 = ts.设备

            可能需要连续()来获取扁平视图。
            我们可能更精确地确定何时进行此操作作为优化。
            values = ts.连续().展平(0, 1).到(设备=设备, 数据类型=数据类型)
            批处理大小 = 时间戳.形状[0]
            序列长度 = 时间序列.形状[1]
            偏移量 = 火炬.等差数列(
                0, 批处理大小 * 序列长度 + 1, 序列长度, 设备=设备, 数据类型=火炬.int64 类型
            )

            from torch.nested._internal.nested_tensor 导入 (
                嵌套视图从值偏移,
            )

            返回 嵌套视图从值偏移(
                值, 偏移量, 最小序列长度=序列长度, 最大序列长度=序列长度
            )
        否则:
            from torch.nested._internal.nested_tensor 导入 从列表生成不规则序列

            断言 isinstance(ts, 列表)
            nt, _ = jagged_from_list(ts, 偏移量=无, 设备=设备, 数据类型=数据类型)
            返回 nt
    否则:
        抛出 RuntimeError(
            f"指定的布局不支持嵌套张量："{布局}"
        )



# 不仅为嵌套操作添加文档字符串，而且
# 还将 torch.nested Python 命名空间连接到 torch._C._nested 内置函数。

to_padded_tensor = _添加文档字符串(
    _nested.nested_to_padded_tensor,
    r""
to_padded_tensor(input, padding, output_size=None, out=None) -> Tensor

返回一个新的（非嵌套）张量，通过填充 :attr:`input` 嵌套张量。
前导条目将填充嵌套数据，
而后续的条目将被填充。

.. 警告::

func:`to_padded_tensor` 总是复制底层数据，
因为嵌套的和非嵌套的张量在内存布局上不同。

参数：
填充（浮点数）：尾随条目的填充值。

关键字参数:
输出大小（元组[int]）：输出张量的大小。
如果提供，必须足够大，可以容纳所有嵌套数据；
否则，将通过取每个嵌套子张量在每个维度上的最大大小来推断。
输出（张量，可选）：输出张量。

示例::

    >>> nt = torch.nested.nested_tensor([torch.randn((2, 5)), torch.randn((3, 4))])
    nested_tensor([
张量([[ 1.6862, -1.1282,  1.1031,  0.0464, -1.3276],
[ -1.9967, -1.0054,  1.8972,  0.9174, -1.4995 ]
tensor([[-1.8546, -0.7194, -0.2918, -0.1846]])
[ 0.2773,  0.8793, -0.5183, -0.6447 ]
[ 1.8009,  1.8468, -0.9832, -1.5272 ]
    ])
    >>> pt_infer = torch.nested.to_padded_tensor(nt, 0.0)
    tensor([[[ 1.6862, -1.1282,  1.1031,  0.0464, -1.3276],
             [-1.9967, -1.0054,  1.8972,  0.9174, -1.4995],
[[0.0000, 0.0000, 0.0000, 0.0000, 0.0000]]
[[-1.8546, -0.7194, -0.2918, -0.1846, 0.0000]]
[[0.2773, 0.8793, -0.5183, -0.6447, 0.0000]]
[[1.8009, 1.8468, -0.9832, -1.5272, 0.0000]]
    >>> pt_large = torch.nested.to_padded_tensor(nt, 1.0, (2, 4, 6))
    tensor([[[ 1.6862, -1.1282,  1.1031,  0.0464, -1.3276,  1.0000],
             [-1.9967, -1.0054,  1.8972,  0.9174, -1.4995,  1.0000],
             [ 1.0000,  1.0000,  1.0000,  1.0000,  1.0000,  1.0000],
[[1.0000, 1.0000, 1.0000, 1.0000, 1.0000, 1.0000]]
[[-1.8546, -0.7194, -0.2918, -0.1846, 1.0000, 1.0000]]
[[0.2773, 0.8793, -0.5183, -0.6447, 1.0000, 1.0000]]
[[1.8009, 1.8468, -0.9832, -1.5272, 1.0000, 1.0000]]
[ 1.0000,  1.0000,  1.0000,  1.0000,  1.0000,  1.0000]]
    >>> pt_small = torch.nested.to_padded_tensor(nt, 2.0, (2, 2, 2))
运行时错误：输出大小中的值小于 NestedTensor 填充大小。不支持截断。

""",
)


定义 嵌套张量(
    张量列表,
    *,
    数据类型=无,
    布局=无,
    设备=无,
    需要梯度=False,
    锚定内存=False,
) -> 张量:
    r""
构建一个没有自动微分历史记录的嵌套张量（也称为“叶张量”，参见
ref:`自动微分机制 》）从 :attr:`tensor_list` 张量列表中。

参数：
tensor_list（列表[array_like]）：一个张量列表，或者任何可以传递给 torch.tensor 的对象，
其中列表的每个元素都具有相同的维度。

关键字参数：
dtype (torch.dtype 类型，可选): 返回嵌套张量的期望类型。
默认：如果为 None，则与列表中最左侧张量的同一 torch.dtype 类型。
layout (torch.layout 类型，可选): 返回嵌套张量的期望布局。
仅支持步进和交错布局。默认：如果为 None，则为步进布局。
device (:class:`torch.device`，可选)：返回嵌套张量的期望设备。
默认：如果为 None，则与列表中最左侧张量相同的 :class:`torch.device`。
requires_grad (bool，可选)：如果 autograd 应记录操作。
返回嵌套张量。默认：``False``。
pin_memory（布尔值，可选）：如果设置，返回的嵌套张量将在
锁定内存中分配。仅适用于 CPU 张量。默认：``False``。

示例::

        >>> a = torch.arange(3, dtype=torch.float, requires_grad=True)
        >>> b = torch.arange(5, dtype=torch.float, requires_grad=True)
        >>> nt = torch.nested.nested_tensor([a, b], requires_grad=True)
        >>> nt.is_leaf
        True
    """
    如果 布局 是 无:
        布局 = 火炬.稀疏的
    如果 布局 == 火炬.strided:
        返回 _嵌套.嵌套张量(
            张量列表,
            数据类型=数据类型,
            设备=设备,
            需要梯度=需要梯度,
            锁定内存=持久化内存,
        )
    elif 布局 == 火炬.错落有致:
        # 需要将标量列表包装为张量
        张量列表 = [
            t 如果 isinstance(t, 张量) 否则 火炬.as_tensor(t) 对于 t 进入 张量列表
        ]

        from torch.nested._internal.nested_tensor 导入 从列表中创建不规则的

        替换为 火炬.不梯度():
            nt, _ = 从列表中创建不规则的(
                张量列表, 偏移量=无, 设备=设备, 数据类型=dtype
            )

        nt.需要梯度_(需要梯度)
        如果 显存:
            nt = nt.显存()  # 类型：忽略[赋值]

        返回 nt
    否则:
        抛出 RuntimeError(
            f"指定的布局不支持嵌套张量："{布局}"
        )



定义 狭窄(
    张量: 张量,
    维度: int,
    开始: 联合[int, 张量]
    长度: 联合[int, 张量]
    布局=火炬.步长,
) -> 张量:
    r""
从 :attr:`tensor` 构建一个嵌套张量（可能是视图）。这遵循与 torch.Tensor.narrow 相似的语义，在 :attr:`dim`-th 维度中，新的嵌套张量
与 torch.Tensor.narrow 相似的语义，在 :attr:`dim`-th 维度中，新的嵌套张量
仅显示区间 `[start, start+length)` 中的元素。
允许在该维度的每一行使用不同的 `start` 和 `length`，:attr:`start` 和 :attr:`length`。
也可以是形状为 `tensor.shape[0]` 的张量。

根据您使用的嵌套张量布局，可能会有一些差异。如果使用对齐布局，
torch.narrow 将将缩小后的数据复制到一个连续的 NT 中，具有步长布局，
带有锯齿布局的 narrow()将创建一个非连续的原始步长张量的视图。这种特定的
表示方式对于在 Transformer 模型中表示 kv 缓存非常有用，因为专门的
SDPA 核可以轻松处理这种格式，从而提高性能。


参数：
tensor（:class:`torch.Tensor`）：一个带偏移的 tensor，将用作底层数据
对于使用交错布局的嵌套张量或将被复制用于步进布局。
dim (int)：应用狭窄的维度。仅支持 `dim=1`。
错落布局，而步进支持所有维度
start（Union[int, :class:`torch.Tensor`]）：narrow 操作的开始元素
length（Union[int, :class:`torch.Tensor`]）：narrow 操作中取出的元素数量

关键字参数：
layout（:class:`torch.layout`，可选）：返回嵌套张量的期望布局。
仅支持步进和交错布局。默认：如果为 None，则为步进布局。

示例::

        >>> starts = torch.tensor([0, 1, 2, 3, 4], dtype=torch.int64)
        >>> lengths = torch.tensor([3, 2, 2, 1, 5], dtype=torch.int64)
        >>> narrow_base = torch.randn(5, 10, 20)
        >>> nt_narrowed = torch.nested.narrow(narrow_base, 1, starts, lengths, layout=torch.jagged)
        >>> nt_narrowed.is_contiguous()
        False
    """
    如果 not isinstance(开始, (int, SymInt, 张量)):
        抛出 RuntimeError("start 必须是一个整数或一个张量")

    如果 not isinstance(长度, (int, SymInt, 张量)):
        抛出 RuntimeError("长度必须是整数或张量")

    如果 布局 == 火炬.步进:
        如果 isinstance(开始, 张量) 或者 isinstance(长度, 张量):
            抛出 RuntimeError(
                "对于 NT 布局的步进实现，起始位置和长度必须是整数"
            )
        # TODO: 当它可用时，切换到 as_nested_tensor(tensor)
        模块 = 嵌套张量(火炬.解绑(张量), 布局=火炬.步长).狭窄(
            dim, 开始, 长度
        )
    elif 布局 == 火炬.锯齿状:
        如果 dim != 1:
            抛出 RuntimeError(锯齿状布局仅支持 dim=1)

        from torch.nested._internal.nested_tensor 导入 从张量和长度生成锯齿状

        如果 isinstance(开始, (int, SymInt)):
            开始 = 火炬.张量[开始] 设备=张量.设备, 数据类型=火炬.int64)

        如果 isinstance(长度, (int, SymInt)):
            长度 = 火炬.张量[长度] 设备=张量.设备, 数据类型=火炬.int64)

        nt, _, _ = 从张量和长度中生成锯齿状(张量, 开始, 长度)
    否则:
        抛出 RuntimeError(
            f"指定的布局不支持嵌套窄布局："{布局}"
        )

    返回 nt



定义 从交错结构创建嵌套张量(
    值: 张量,
    偏移量: 可选[张量] = 无,
    长度: 可选[张量] = 无,
    锯齿状维度: 可选[int] = 无,
    min_seqlen: 可选[int] = 无,
    max_seqlen: 可选[int] = 无,
) -> 张量:
    r""
从给定的锯齿状组件构建一个嵌套张量的锯齿状布局。
它包含一个必需的值缓冲区，其中锯齿维度打包到一个单一维度中。
偏移量/长度元数据决定了如何将此维度分割成批处理元素
并且预期它们将分配在值缓冲区相同的设备上。

预期元数据格式：
* 偏移量：在打包维度内的索引，将其分割成不同大小的
批量元素。例如：[0, 2, 3, 6] 表示一个大小为 6 的打包交错维度
应概念上分为长度为[2, 1, 3]的批次元素。请注意，两者都
开始和结束偏移量对内核便利性是必需的（即形状 batch_size + 1）。
* 长度：单个批次元素的长度；形状等于批次大小。例如：[2, 1, 3]
指示一个大小为 6 的打包交错维度在概念上应拆分为批次
元素长度为[2, 1, 3]。

请注意，提供偏移量和长度可能很有用。这描述了一个嵌套张量
带有“holes”，其中偏移量指示每个批次项的起始位置和长度
指定元素总数（见以下示例）。

返回的交错布局嵌套张量将是输入值张量的一个视图。

参数：
values（:class:`torch.Tensor`）：底层缓冲区形状为
（sum_B(*), D_1, ..., D_N）。交错维度被压缩到一个单独的维度中，
使用偏移/长度元数据来区分批处理元素。
偏移（可选：:class:`torch.Tensor`）：形状为 B + 1 的交错维度的偏移量。
长度（可选：:class:`torch.Tensor`）：形状为 B 的批处理元素长度。
jagged_dim（可选 int）：指示 values 中的哪个维度是打包的交错维度。
维度。如果为 None，则设置为 dim=1（即紧随批处理维度的维度）。默认：None
min_seqlen（可选整数）：如果设置，则使用指定的值作为返回嵌套张量的缓存最小序列长度。这可以是一个有用的替代方案，用于计算
min_seqlen（可选整数）：如果设置，则使用指定的值作为返回嵌套张量的缓存最小序列长度。这可以是一个有用的替代方案，用于计算
min_seqlen（可选整数）：如果设置，则使用指定的值作为返回嵌套张量的缓存最小序列长度。这可以是一个有用的替代方案，用于计算
需要时按需获取此值，可能避免 GPU 到 CPU 的同步。默认：无
max_seqlen（可选整数）：如果设置，则使用指定的值作为缓存的序列最大长度
返回嵌套张量的长度。这可以是一个有用的替代方案，以按需计算此值
需要时按需获取此值，可能避免 GPU 到 CPU 的同步。默认：无

示例::

        >>> values = torch.randn(12, 5)
        >>> offsets = torch.tensor([0, 3, 5, 6, 10, 12])
        >>> nt = nested_tensor_from_jagged(values, offsets)
>>> # 3D 形状中间维度为交错
        >>> nt.shape
        torch.Size([5, j2, 5])
>>> # 每个批次中每个项目的长度：
        >>> offsets.diff()
        tensor([3, 2, 1, 4, 2])

        >>> values = torch.randn(6, 5)
        >>> offsets = torch.tensor([0, 2, 3, 6])
        >>> lengths = torch.tensor([1, 1, 2])
>>> # 空洞的 NT
        >>> nt = nested_tensor_from_jagged(values, offsets, lengths)
        >>> a, b, c = nt.unbind()
>>> # 批量项 1 包含索引[0, 1)
        >>> torch.equal(a, values[0:1, :])
        True
>>> # 批量项目 2 包含索引[2, 3)
        >>> torch.equal(b, values[2:3, :])
        True
>>> # 批量项目 3 包含索引[3, 5)
        >>> torch.equal(c, values[3:5, :])
        True
    """
    from torch.fx._symbolic_trace 导入 is_fx_tracing

    如果 is_fx_tracing():
        抛出 RuntimeError(
            "torch.nested.nested_tensor_from_jagged 不支持与 fx.symbolic_trace 跟踪。"
            "使用 fx.wrap 将调用 nested_tensor_from_jagged 的函数进行包装。"
        )

    如果 偏移量 是 无:
        如果 长度 是 无:
            抛出 RuntimeError(
                "nested_tensor_from_jagged()：至少需要 offsets 或 lengths 中的一个。"
            )
        否则:
            # TODO：将来真正支持 offsets=None？
            # 目前，为了内核方便，仅将 lengths 转换为 offsets
            偏移量 = F.填充(长度.累加和(0), (1, 0))
            长度 = 无

    如果 锯齿状维度 是 无:
        锯齿状维度 = 1

    from torch.nested._internal.nested_tensor 导入 (
        从值、偏移量、长度嵌套视图,
    )

    返回 嵌套视图从值偏移量长度(
        值,
        偏移量,
        长度,
        Ragged 索引=锯齿状维度,
        最小序列长度=最小序列长度,
        最大序列长度=max_seqlen,
    )



[docs]def masked_select(tensor: Tensor, mask: Tensor) -> Tensor:
r"""
构建一个嵌套张量，给定一个具有步进输入和步进掩码的张量，结果为具有交错布局的嵌套张量
将掩码等于 True 的位置保留值。掩码的维度被保留。
以偏移量表示，这与：func:`masked_select`不同，其输出被折叠成一个 1D 张量。

参数:
tensor（:class:`torch.Tensor`）：从其中构建嵌套张量的斜行张量。
mask (:class:`torch.Tensor`): 一个应用于张量输入的带偏移的掩码张量

示例::

>>> tensor = torch.randn(3, 3)
>>> mask = torch.tensor([[False, False, True], [True, False, True], [False, False, True]])
>>> nt = torch.nested.masked_select(tensor, mask)
>>> nt.shape
torch.Size([3, j4])
>>> # 每个批次中每个项目的长度：
>>> nt.offsets().diff()
tensor([1, 2, 1])

>>> tensor = torch.randn(6, 5)
>>> mask = torch.tensor([False])
>>> nt = torch.nested.masked_select(tensor, mask)
>>> nt.shape
torch.Size([6, j5])
>>> # 每个批次中每个项目的长度：
>>> nt.offsets().diff()
tensor([0, 0, 0, 0, 0, 0])
"""
if tensor.layout != torch.strided:
raise RuntimeError(
f"torch.nested.masked_select 需要一个带偏移的 tensor，给定 {tensor.layout}"
        )

if mask.layout != torch.strided:
raise RuntimeError(
f"torch.nested.masked_select 需要一个 strided 掩码，给定：{mask.layout}"
        )
res_values = tensor.masked_select(mask)
expanded_mask = mask.expand(tensor.shape)
res_lengths = expanded_mask.sum(dim=tensor.ndim - 1).view(-1)

from torch.nested._internal.nested_tensor import nested_view_from_values_offsets

return nested_view_from_values_offsets(
values=res_values,
offsets=F.pad(res_lengths.cumsum(dim=0), (1, 0)),
    )