torch.distributions.transforms 的源代码

# mypy: 允许未类型化定义
导入 functools
导入 数学
导入 操作符
导入 弱引用
from 打字 导入 可选

导入 火炬
导入 torch.nn.functional 是 F
from torch.distributions 导入 约束
from torch.distributions.utils 导入 (
    _sum_rightmost,
    广播全部,
    懒属性,
    tril_matrix_to_vec,
    向量到三对角矩阵的转换,
)
from torch.nn.functional 导入 填充, softplus
from torch.types 导入 数值


全部 = [
    "Abs 变换",
    "仿射变换",
    "猫变换",
    "组合变换",
    "CorrCholesky 变换",
    "累积分布变换",
    "指数变换",
    "独立变换",
    "下三角 Cholesky 变换",
    "正定变换",
    "幂变换",
    "重塑变换",
    "Sigmoid 变换",
    "Softplus 变换",
    "Tanh 变换",
    "Softmax 变换",
    "Stack 变换",
    "粘性断裂变换",
    "转换",
    "恒等变换",
]


类 变换:
    ""
抽象类，用于可逆变换的可计算对数
计算雅可比矩阵。它们主要用于
`:class:`torch.distributions.TransformedDistribution`。

缓存对于逆变换昂贵或不可行的转换很有用
数值不稳定。请注意，由于自动微分图可能被反转，因此对缓存值必须小心处理
因为自动微分图可能会被反转。例如，以下代码：
无论是否缓存都可以工作：

        y = t(x)
t.log_abs_det_jacobian(x, y).backward()  # x 将接收梯度。

然而，由于依赖反转，以下内容在缓存时将出错：

        y = t(x)
        z = t.inv(y)
grad(z.sum(), [y])  # 错误，因为 z 是 x

派生类应实现一个或两个：:meth:`_call` 或
meth:`_inverse`。设置 `bijective=True` 的派生类还应实现
meth:`log_abs_det_jacobian`。

参数：
缓存大小（int）：缓存的大小。如果为零，则不进行缓存。如果为 1，
最新单值已缓存。仅支持 0 和 1。

属性：
区域：（:类`~torch.distributions.constraints.Constraint`）：
表示此转换有效输入的约束。
定义域 (:class:`~torch.distributions.constraints.Constraint`)：
表示此转换有效输出的约束
这些是逆转换的输入。
双射（布尔值）：此变换是否为双射。一个变换
`t` 是双射当且仅当 `t.inv(t(x)) == x` 且
对于域中的每个 x 和 y，t(t.inv(y))等于 y
目标域。非双射的变换至少
维持较弱的伪逆性质
``t(t.inv(t(x)) == t(x)`` 和 ``t.inv(t(t.inv(y))) == t.inv(y)``。
符号（整数或张量）：对于双射单变量变换，这应该是 +1 或 -1，具体取决于变换是否单调
应该是 +1 或 -1，具体取决于变换是否单调
增加或减少。
""

    双射 = 假
    域名: 约束.约束
    定义域: 约束.约束

    def __init__(self, 缓存大小=0):
        self._cache_size = 缓存大小
        self._inv: 可选[弱引用.参考类型[变换]] = 无
        如果 缓存大小 == 0:
            通过  # 默认行为
        如果...否则 缓存大小 == 1:
            self._缓存的_x_y = 无, 无
        else:
            提升 ValueError("缓存大小必须是 0 或 1")
        超级().__init__()

    def __getstate__(self):
        状态 = self.字典.复制()
        状态["_inv"] = 无
        返回 状态

    @property
    def 事件维度(self) 翻译 int:
        如果 self.域名.事件维度 == self.定义域.事件维度:
            返回 self.域名.事件维度
        提升 ValueError("请使用 .domain.event_dim 或 .codomain.event_dim")

    @property
    def 涉及(self) 翻译 "转换":
        ""
返回此变换的逆变换 :class:`Transform`。
这应该满足 ``t.inv.inv 是 t``。
""
        求逆 = 无
        如果 self._逆 is 不是 无:
            求逆 = self._inv()
        如果 求逆 is 无:
            求逆 = 逆变换(self)
            self._逆 = 弱引用.ref(涉及)
        返回 求逆

    @property
    def 符号(self) 翻译 int:
        ""
返回雅可比行列式的符号，如果适用的话。
通常情况下，这仅适用于双射变换。
""
        提升 未实现异常

    def 带缓存(self, 缓存大小=1):
        如果 self._cache_size == 缓存大小:
            返回 self
        如果 类型(self).__init__ is 变换.__init__:
            返回 类型(self)(缓存大小=缓存大小)
        提升 不支持的操作异常(f"{类型(self)}.with_cache 未实现")

    def __等于__(self, 其他):
        返回 self is 其他

    def __ne__(self, 其他):
        # 需要 Python2
        返回 不是 self.__等于__(其他)

    def __调用__(self, x):
        ""
计算变换 `x => y`。
""
        如果 self._cache_size == 0:
            返回 self._调用(x)
        x_旧, y_旧 = self._缓存的_x_y
        如果 x is x_旧:
            返回 y_旧
        y = self._调用(x)
        self._缓存的_x_y = x, y
        返回 y

    def _inv_call(self, y):
        ""
反转变换 `y => x`。
""
        如果 self._cache_size == 0:
            返回 self.反向(y)
        x_旧, y_旧 = self._缓存的_x_y
        如果 y is y_old:
            返回 x 旧
        x = self.反向(y)
        self._缓存的_x_y = x, y
        返回 x

    def _调用(self, x):
        ""
计算前向变换的抽象方法。
""
        提升 未实现异常

    def 反向(self, y):
        ""
计算逆变换的抽象方法。
""
        提升 未实现异常

[文档]    def log_abs_det_jacobian(self, x, y):
"``"
计算给定输入和输出时的对数雅可比行列式 `log |dy/dx|`。
"``"
引发未实现异常

    def __repr__(self):
        返回 self.类.__name__ + "()"

[文档]    def forward_shape(self, shape):
"""
根据输入形状推断前向计算的形状。
默认保留形状。
"""
返回形状

[文档]    def inverse_shape(self, shape):
""
根据输出形状推断逆计算的形状。
默认保留形状。
""
返回形状



类 逆变换(变换):
    ""
反转单个 :class:`变换`。
这个类是私有的；请改用 `Transform.inv` 属性。
""

    def __init__(self, 转换: 变换):
        超级().__init__(缓存大小=转换._缓存大小)
        self._inv: 变换 = 转换  # 类型：忽略[赋值]

    @constraints.依赖属性(是否离散=错误)
    def 域名(self):
        断言 self._逆 is 不是 无
        返回 self._inv.目标域

    @constraints.依赖属性(是否离散=错误)
    def 定义域(self):
        断言 self._逆 is 不是 无
        返回 self._inv.域名

    @property
    def 双射(self) 翻译 布尔:  # 类型：忽略[覆盖]
        断言 self._逆 is 不是 无
        返回 self._inv.双射

    @property
    def 符号(self) 翻译 int:
        断言 self._逆 is 不是 无
        返回 self._inv.符号

    @property
    def 涉及(self) 翻译 变换:
        返回 self._逆

    def 带缓存(self, 缓存大小=1):
        断言 self._逆 is 不是 无
        返回 self.涉及.带缓存(缓存大小).求逆

    def __等于__(self, 其他):
        如果 不是 isinstance(其他, 逆变换):
            返回 假
        断言 self._逆 is 不是 无
        返回 self._逆 == 其他._逆

    def __repr__(self):
        返回 f"{self.类.__name__}({表示(self._inv)})"

    def __调用__(self, x):
        断言 self._逆 is 不是 无
        返回 self._inv._inv_call(x)

    def 对数绝对雅可比行列式(self, x, y):
        断言 self._逆 is 不是 无
        返回 -self._inv.对数绝对雅可比行列式(y, x)

    def 前向形状(self, 形状):
        返回 self._inv.逆向形状(形状)

    def 逆向形状(self, 形状):
        返回 self._inv.前向形状(形状)


类 组合变换(变换):
    ""
将多个转换组合成链式结构。
负责缓存的变换正在被组合。

参数：
部分（:class:`Transform` 列表）：要组合的变换列表。
缓存大小（int）：缓存的大小。如果为零，则不进行缓存。如果为 1，
最新单值已缓存。仅支持 0 和 1。
""

    def __init__(self, parts: 列表[变换] 缓存大小=0):
        如果 缓存大小:
            部分 = [部分.带缓存(缓存大小) 为 部分 在 parts]
        超级().__init__(缓存大小=缓存大小)
        self.部分 = 部分

    def __等于__(self, 其他):
        如果 不是 isinstance(其他, 组合变换):
            返回 假
        返回 self.部分 == 其他.部分

    @constraints.依赖属性(是否离散=错误)
    def 域名(self):
        如果 不是 self.parts:
            返回 约束.真实
        域名 = self.parts[0].域名
        # 将 event_dim 调整为所有部分中的最大值。
        事件维度 = self.parts[-1].定义域.事件维度
        为 部分 在 反转(self.parts):
            事件维度 += 部分.域名.事件维度 - 部分.定义域.事件维度
            事件维度 = 最大值(事件维度, 部分.域名.事件维度)
        断言 事件维度 ≥ 域名.事件维度
        如果 事件维度 > 域名.事件维度:
            域名 = 约束.独立(域名, 事件维度 - 域名.事件维度)
        返回 域名

    @constraints.依赖属性(是否离散=错误)
    def 定义域(self):
        如果 不是 self.parts:
            返回 约束.真实
        目标域 = self.parts[-1].目标域
        # 将 event_dim 调整为所有部分中的最大值。
        事件维度 = self.parts[0].域名.事件维度
        为 部分 在 self.parts:
            事件维度 += 部分.定义域.事件维度 - 部分.域名.事件维度
            事件维度 = 最大值(事件维度, 部分.定义域.事件维度)
        断言 事件维度 ≥ 定义域.事件维度
        如果 事件维度 > 定义域.事件维度:
            目标域 = 约束.独立(定义域, 事件维度 - 定义域.事件维度)
        返回 目标域

    @lazy_property
    def 双射(self) 翻译 布尔:  # 类型：忽略[覆盖]
        返回 所有(p.双射 为 p 在 self.parts)

    @lazy_property
    def 符号(self) 翻译 int:  # 类型：忽略[覆盖]
        符号 = 1
        为 p 在 self.parts:
            符号 = 符号 * p.符号
        返回 符号

    @property
    def 涉及(self) 翻译 变换:
        求逆 = 无
        如果 self._逆 is 不是 无:
            求逆 = self._inv()
        如果 求逆 is 无:
            求逆 = 组合变换[p.求逆 为 p 在 反转(self.parts)]])
            self._逆 = 弱引用.ref(涉及)
            涉及._逆 = 弱引用.ref(self)
        返回 求逆

    def 带缓存(self, 缓存大小=1):
        如果 self._cache_size == 缓存大小:
            返回 self
        返回 组合变换(self.parts, 缓存大小=缓存大小)

    def __调用__(self, x):
        为 部分 在 self.parts:
            x = 部分(x)
        返回 x

    def 对数绝对雅可比行列式(self, x, y):
        如果 不是 self.parts:
            返回 火炬.等于零的(x)

        # 计算中间结果。如果 parts[:-1] 都已缓存，这将免费。
        xs = [x]
        为 部分 在 self.parts[-1]
            xs.append(部分(xs[-1]))
        xs.append(y)

        术语 = 输入文本为空，请提供需要翻译的文本
        事件维度 = self.域名.事件维度
        为 部分, x, y 在 zip(self.parts, xs[-1] xs[1(]:)
            术语.append(
                _sum_rightmost(
                    部分.对数绝对雅可比行列式(x, y), 事件维度 - 部分.域名.事件维度
                )
            )
            事件维度 += 部分.定义域.事件维度 - 部分.域名.事件维度
        返回 functools.归约(操作符.添加, 术语)

    def 前向形状(self, 形状):
        为 部分 在 self.parts:
            形状 = 部分.前向形状(形状)
        返回 形状

    def 逆向形状(self, 形状):
        为 部分 在 反转(self.parts):
            形状 = 部分.逆向形状(形状)
        返回 形状

    def __repr__(self):
        格式化字符串 = self.类.__name__ + “（”输入文本翻译为简体中文为：\n    "
        格式化字符串 += “输入文本翻译为简体中文为：\n    ".连接[p.__repr__() 为 p 在 self.parts])
        格式化字符串 += "输入文本翻译为简体中文为：\n)"
        返回 格式化字符串



身份转换 = 组合变换([])


类 独立转换(变换):
    ""
另一个转换的包装，用于处理
``reinterpreted_batch_ndims``-许多额外的最右侧维度作为
依赖项。这对正向或反向转换没有影响，
对应的 `reinterpreted_batch_ndims` 维度进行求和
在 :meth:`log_abs_det_jacobian` 方法中。

参数：
base_transform (:class:`Transform`): 一个基础变换。
reinterpreted_batch_ndims (int): 额外右端维度的数量
被视为依赖的维度。
""

    def __init__(self, 基础转换, 重新解释的批量维度, 缓存大小=0):
        超级().__init__(缓存大小=缓存大小)
        self.基础转换 = 基础转换.带缓存(缓存大小)
        self.重新解释的批量维度数 = 重新解释的批量维度数

    def 带缓存(self, 缓存大小=1):
        如果 self._cache_size == 缓存大小:
            返回 self
        返回 独立转换(
            self.基础转换, self.重新解释的批量维度, 缓存大小=缓存大小
        )

    @constraints.依赖属性(是否离散=错误)
    def 域名(self):
        返回 约束.独立(
            self.基础转换.域名, self.重新解释的批量维度数
        )

    @constraints.依赖属性(是否离散=错误)
    def 定义域(self):
        返回 约束.独立(
            self.基础转换.定义域, self.重新解释的批量维度数
        )

    @property
    def 双射(self) 翻译 布尔:  # 类型：忽略[覆盖]
        返回 self.基础转换.双射

    @property
    def 符号(self) 翻译 int:  # 类型：忽略[覆盖]
        返回 self.基础转换.符号

    def _调用(self, x):
        如果 x.暗() < self.域名.事件维度:
            提升 ValueError(输入维度太少)
        返回 self.基础转换(x)

    def 反向(self, y):
        如果 y.暗() < self.定义域.事件维度:
            提升 ValueError(输入维度太少)
        返回 self.基础转换.涉及(y)

    def 对数绝对雅可比行列式(self, x, y):
        结果 = self.基础转换.对数绝对雅可比行列式(x, y)
        结果 = _sum_rightmost(结果, self.重新解释的批量维度)
        返回 结果

    def __repr__(self):
        返回 f"{self.类.__name__}({表示(self.基础转换)}, {self.重新解释的批量维度})"

    def 前向形状(self, 形状):
        返回 self.基础转换.前向形状(形状)

    def 逆向形状(self, 形状):
        返回 self.基础转换.逆向形状(形状)



类 重新塑形转换(变换):
    ""
将单元雅可比变换应用于张量最右侧部分以重塑。

注意，与 :meth:`torch.Tensor.reshape` 相同，`in_shape` 和 `out_shape` 必须具有相同数量的元素。
注意，与 :meth:`torch.Tensor.reshape` 相同，`in_shape` 和 `out_shape` 必须具有相同数量的元素。

参数：
in_shape (torch.Size): 输入事件形状。
out_shape (torch.Size): 输出事件形状。
缓存大小（int）：缓存的大小。如果为零，则不进行缓存。如果为 1，
缓存最新的单个值。仅支持 0 和 1。 (默认 0。)
""

    双射 = 真实

    def __init__(self, 形状输入, 输出形状, 缓存大小=0):
        self.输入形状 = 火炬.尺寸(形状输入)
        self.输出形状 = 火炬.尺寸(输出形状)
        如果 self.形状输入.元素数量() != self.输出形状.元素数量():
            提升 ValueError("in_shape, out_shape 元素数量不同")
        超级().__init__(缓存大小=缓存大小)

    @constraints.依赖属性
    def 域名(self):
        返回 约束.独立(约束.真实, 长度(self.形状输入))

    @constraints.依赖属性
    def 定义域(self):
        返回 约束.独立(约束.真实, 长度(self.输出形状))

    def 带缓存(self, 缓存大小=1):
        如果 self._cache_size == 缓存大小:
            返回 self
        返回 重新塑形转换(self.形状输入, self.输出形状, 缓存大小=缓存大小)

    def _调用(self, x):
        批量形状 = x.形状[ x.暗() - 长度(self.形状输入)]
        返回 x.重塑(批量形状 + self.输出形状)

    def 反向(self, y):
        批量形状 = y.形状[ y.暗() - 长度(self.输出形状)]
        返回 y.重塑(批量形状 + self.形状输入)

    def 对数绝对雅可比行列式(self, x, y):
        批量形状 = x.形状[ x.暗() - 长度(self.形状输入)]
        返回 x.新零(批量形状)

    def 前向形状(self, 形状):
        如果 长度(形状) < 长度(self.形状输入):
            提升 ValueError(输入维度太少)
        切割 = 长度(形状) - 长度(self.形状输入)
        如果 形状[cut] != self.形状输入:
            提升 ValueError(
                f"形状不匹配：期望"{形状[cut]}但是得到了{self.形状输入}"
            )
        返回 形状[cut] + self.输出形状

    def 逆向形状(self, 形状):
        如果 长度(形状) < 长度(self.输出形状):
            提升 ValueError(输入维度太少)
        切割 = 长度(形状) - 长度(self.输出形状)
        如果 形状[cut] != self.输出形状:
            提升 ValueError(
                f"形状不匹配：期望"{形状[cut]}但是得到了{self.输出形状}"
            )
        返回 形状[cut] + self.形状输入



[文档]class ExpTransform(Transform):
r"""
通过映射 :math:`y = \exp(x)` 进行变换。
"""

domain = 实数约束
codomain = 正数约束
双射 = True
符号 = +1

def __eq__(self, other):
return isinstance(other, ExpTransform)

def _call(self, x):
return x.exp()

def _inverse(self, y):
return y.log()

def log_abs_det_jacobian(self, x, y):
return x


[文档]类 PowerTransform(Transform):
r"""
通过映射 :math:`y = x^{\text{指数}}` 进行转换。
"""

domain = 正向约束
codomain = 正向约束
bijective = True

def __init__(self, exponent, cache_size=0):
super().__init__(cache_size=cache_size)
(self.exponent,) = broadcast_all(exponent)

def with_cache(self, cache_size=1):
if self._cache_size == cache_size:
返回自身
返回 PowerTransform(self.exponent, cache_size=cache_size)

    @lazy_property
def sign(self) -> int:  # type: ignore[override]
return self.exponent.sign()

def __eq__(self, other):
if not isinstance(other, PowerTransform):
return False
return self.exponent.eq(other.exponent).all().item()

def _call(self, x):
return x.pow(self.exponent)

def _inverse(self, y):
return y.pow(1 / self.exponent)

def log_abs_det_jacobian(self, x, y):
return (self.exponent * y / x).abs().log()

def forward_shape(self, shape):
return torch.broadcast_shapes(shape, getattr(self.exponent, "shape", ()))

def inverse_shape(self, shape):
return torch.broadcast_shapes(shape, getattr(self.exponent, "shape", ()))


def _clipped_sigmoid(x):
    信息 = 火炬.finfo(x.数据类型)
    返回 火炬.卡钳(火炬.Sigmoid 函数(x), 最小值=finfo.微小。, 最大值=1.0 - finfo.eps)


[文档]class SigmoidTransform(Transform):
r"""
通过映射 :math:`y = \frac{1}{1 + \exp(-x)}` 和 :math:`x = \text{logit}(y)` 进行转换。
"""

domain = 实数约束
codomain = 单位区间约束
双射 = True
符号 = +1

def __eq__(self, other):
return isinstance(other, SigmoidTransform)

def _call(self, x):
return _clipped_sigmoid(x)

def _inverse(self, y):
finfo = torch.finfo(y.dtype)
y = y.clamp(min=finfo.tiny, max=1.0 - finfo.eps)
return y.log() - (-y).log1p()

def log_abs_det_jacobian(self, x, y):
return -F.softplus(-x) - F.softplus(x)


[文档]class SoftplusTransform(变换):
r"""
通过映射：math:`\text{Softplus}(x) = \log(1 + \exp(x))`进行转换。
当：math:`x > 20`时，实现将回退到线性函数。
"""

定义域 = 约束实数
目标域 = 约束的正数
双射 = True
符号 = +1

def __eq__(self, other):
return isinstance(other, SoftplusTransform)

def _call(self, x):
return softplus(x)

def _inverse(self, y):
return (-y).expm1().neg().log() + y

def log_abs_det_jacobian(self, x, y):
return -softplus(-x)


[文档]类 TanhTransform(Transform):
r"""
通过映射 :math:`y = \tanh(x)` 进行转换。

它等价于

.. 代码块 :: python

ComposeTransform(
            [
AffineTransform(0.0, 2.0),
Sigmoid 变换()
平移变换(-1.0, 2.0),
            ]
        )

然而，这可能在数值上不稳定，因此建议使用 `TanhTransform`
取而代之。

注意，当涉及到 `NaN/Inf` 值时，应使用 `cache_size=1`。

"""

domain = 实数约束
codomain = 区间约束(-1.0, 1.0)
bijective = True
sign = +1

def __eq__(self, other):
return isinstance(other, TanhTransform)

def _call(self, x):
return x.tanh()

def _inverse(self, y):
# 我们在这里不限制边界，因为这可能会降低某些算法的性能。
# 应该使用 `cache_size=1` 而不是。
return torch.atanh(y)

def log_abs_det_jacobian(self, x, y):
# 我们使用一个更数值稳定的公式，详细信息请见以下链接
# https://github.com/tensorflow/probability/blob/master/tensorflow_probability/python/bijectors/tanh.py#L69-L80
return 2.0 * (math.log(2.0) - x - softplus(-2.0 * x))


[文档]类 AbsTransform(Transform):
r"""通过映射 :math:`y = |x|` 进行变换。”

定义域 = 约束.实数
值域 = 约束.正数

def __eq__(self, other):
return isinstance(other, AbsTransform)

def _call(self, x):
return x.abs()

def _inverse(self, y):
return y


类 变换矩阵(变换):
    r""
通过点对齐仿射映射进行转换：\(y = \text{loc} + \text{scale} \times x\)。

参数：
loc (张量或浮点数)：位置参数。
缩放（张量或浮点数）：缩放参数。
event_dim (int): `event_shape`的可选大小。这应该是零
对于一元随机变量，1 表示向量的分布
2 用于矩阵等分布
""

    双射 = 真实

    def __init__(self, 位置, 比例, 事件维度=0, 缓存大小=0):
        超级().__init__(缓存大小=缓存大小)
        self.定位 = 定位
        self.缩放 = 缩放
        self._事件维度 = 事件维度

    @property
    def 事件维度(self) 翻译 int:
        返回 self._事件维度

    @constraints.依赖属性(是否离散=错误)
    def 域名(self):
        如果 self.事件维度 == 0:
            返回 约束.真实
        返回 约束.独立(约束.真实, self.事件维度)

    @constraints.依赖属性(是否离散=错误)
    def 定义域(self):
        如果 self.事件维度 == 0:
            返回 约束.真实
        返回 约束.独立(约束.真实, self.事件维度)

    def 带缓存(self, 缓存大小=1):
        如果 self._cache_size == 缓存大小:
            返回 self
        返回 变换矩阵(
            self.位置, self.比例, self.事件维度, 缓存大小=缓存大小
        )

    def __等于__(self, 其他):
        如果 不是 isinstance(其他, 变换矩阵):
            返回 假

        如果 isinstance(self.位置, 数) 和 isinstance(其他.位置, 数):
            如果 self.定位 != 其他.位置:
                返回 假
        else:
            如果 不是 (self.定位 == 其他.位置).所有().项目():
                返回 假

        如果 isinstance(self.比例, 数) 和 isinstance(其他.比例, 数):
            如果 self.缩放 != 其他.比例:
                返回 假
        else:
            如果 不是 (self.缩放 == 其他.比例).所有().项目():
                返回 假

        返回 真实

    @property
    def 符号(self) 翻译 int:
        如果 isinstance(self.比例, 数):
            返回 1 如果 float(self.比例) > 0 否则 -1 如果 float(self.比例) < 0 否则 0
        返回 self.比例.符号()

    def _调用(self, x):
        返回 self.定位 + self.缩放 * x

    def 反向(self, y):
        返回 (y - self.位置) / self.缩放

    def 对数绝对雅可比行列式(self, x, y):
        形状 = x.形状
        缩放 = self.缩放
        如果 isinstance(比例, 数):
            结果 = 火炬.完全一样(x, 数学.日志(绝对值(比例)))
        else:
            结果 = 火炬.绝对值(比例).日志()
        如果 self.事件维度:
            结果大小 = 结果.尺寸()[:] -self.事件维度] + (-1,)
            结果 = 结果.视图(结果大小).总和(-1)
            形状 = 形状[ -self.事件维度]
        返回 结果.展开(形状)

    def 前向形状(self, 形状):
        返回 火炬.广播形状(
            形状, getattr(self.位置, "形状", ()), getattr(self.比例, "形状", ())
        )

    def 逆向形状(self, 形状):
        返回 火炬.广播形状(
            形状, getattr(self.位置, "形状", ()), getattr(self.比例, "形状", ())
        )



类 CorrCholeskyTransform(变换):
    r""
将无约束的实向量 :math:`x`（长度为 :math:`D*(D-1)/2`）转换为
D-维相关矩阵的 Cholesky 因子。这个 Cholesky 因子是一个下三角矩阵
三角矩阵，对角线为正且每行的欧几里得范数为单位。
变换处理如下：

1. 首先将 x 转换为按行顺序排列的下三角矩阵。
2. 对于下三角部分的每一行 :math:`X_i`，我们应用一个 *带符号* 的版本
class :class:`StickBreakingTransform` 来将 :math:`X_i` 转换为
单位欧几里得长度向量，具体步骤如下：
- 尺度扩展到区间 :math:`(-1, 1)` 的域：:math:`r_i = \tanh(X_i)`。
- 转换为无符号域：:math:`z_i = r_i^2`。
- 应用 :math:`s_i = StickBreakingTransform(z_i)`。
- 反转换回有符号域：:math:`y_i = sign(r_i) * \sqrt{s_i}`。
""

    域名 = 约束.实向量
    目标域 = 约束.相关 Cholesky
    双射 = 真实

    def _调用(self, x):
        x = 火炬.双曲正切(x)
        eps = 火炬.finfo(x.数据类型).eps
        x = x.卡钳(最小值=-1 + eps, 最大值=1 - eps)
        r = 向量到三对角矩阵的转换(x, diag=-1)
        # 对平方值应用 stick-breaking
        # 注意 y = sign(r) * sqrt(z * z1m_cumprod)
        (r 的符号 * z 的平方根) * z1m 累积乘积的平方根 = r * z1m 累积乘积的平方根
        z = r**2
        z1m 累积乘积的平方根 = (1 - z).平方根().cumprod(-1)
        对角线元素必须为 1。
        r = r + 火炬.眼睛(r.形状[-1] 数据类型=r.数据类型, 设备=r.设备)
        y = r * 填充(z1m_cumprod_sqrt[..., :-1] [1, 0] 值=1)
        返回 y

    def 反向(self, y):
        # 逆 stick-breaking
        # 请参阅：https://mc-stan.org/docs/2_18/reference-manual/cholesky-factors-of-correlation-matrices-1.html
        累计和 = 1 - 火炬.累加和(y * y, 暗=-1)
        累计和偏移 = 填充(累计和[..., :-1] [1, 0] 值=1)
        y_vec = tril_matrix_to_vec(y, diag=-1)
        y 向累积和 = tril_matrix_to_vec(y 向累积和偏移, diag=-1)
        t = y_vec / (y_cumsum_vec).平方根()
        tanh 的逆
        x = (t.log1p() - t.否定().log1p()) / 2
        返回 x

    def 对数绝对雅可比行列式(self, x, y, 中间变量=无):
        因为定义域和值域是两个不同维度的空间，雅可比矩阵未定义。我们返回 `x` 的 `log_abs_det_jacobian`。
        # 雅可比矩阵未定义。我们返回 `x` 的 `log_abs_det_jacobian` 和 `y` 的展开下三角部分。
        # 请参阅：https://mc-stan.org/docs/2_18/reference-manual/cholesky-factors-of-correlation-matrices-1.html

        # 请参阅：https://mc-stan.org/docs/2_18/reference-manual/cholesky-factors-of-correlation-matrices-1.html
        y1m_cumsum = 1 - (y * y).累加和(暗=-1)
        通过取对角线=-2，我们不需要将 z_cumprod 向右移动
        同样适用于 2x2 矩阵
        y1m_cumsum_tril = tril_matrix_to_vec(y1m_cumsum, diag=-2)
        撑杆断裂对数行列式 = 0.5 * (y1m_cumsum_tril).日志().总和(-1)
        双曲正切对数行列式 = -2 * (x + 软性加(-2 * x) - 数学.日志(2.0)).总和(暗=-1)
        返回 撑杆断裂对数行列式 + 双曲正切对数行列式

    def 前向形状(self, 形状):
        # 将(..., N)重塑为(..., D, D)。
        如果 长度(形状) < 1:
            提升 ValueError(输入维度太少)
        N = 形状[-1]
        D = 四舍五入((0.25 + 2 * N) ** 0.5 + 0.5)
        如果 D * (D - 1) // 2 != N:
            提升 ValueError(输入不是一个扁平的下三角数)
        返回 形状[-1] + (D, D)

    def 逆向形状(self, 形状):
        将形状从(..., D, D)重塑为(..., N)。
        如果 长度(形状) < 2:
            提升 ValueError(输入维度太少)
        如果 形状[-2] != 形状[-1]
            提升 ValueError(输入不是正方形)
        D = 形状[-1]
        N = D * (D - 1) // 2
        返回 形状[-2] + (N,)



[文档]class SoftmaxTransform(Transform):
r"""
将非约束空间通过 :math:`y = \exp(x)` 转换到单纯形，然后
进行归一化。

这不是双射的，不能用于 HMC。然而这主要按坐标进行（除了最后的归一化），因此适用于坐标优化算法。
的（除了最后的归一化），因此适用于坐标优化算法。
的（除了最后的归一化），因此适用于坐标优化算法。
"""

domain = 约束的实向量
codomain = 约束的单形

def __eq__(self, other):
return isinstance(other, SoftmaxTransform)

def _call(self, x):
logprobs = x
probs = (logprobs - logprobs.max(-1, True)[0]).exp()
return probs / probs.sum(-1, True)

def _inverse(self, y):
probs = y
return probs.log()

def forward_shape(self, shape):
如果 shape 的长度小于 1：
抛出 ValueError 异常("输入维度太少")
返回 shape

def inverse_shape(self, shape):
如果 shape 的长度小于 1：
抛出 ValueError 异常("输入维度太少")
返回 shape


类 StickBreakingTransform(变换):
    ""
将转换从无约束空间到增加一维的单纯形通过断棒过程。
此转换在断棒过程中表现为迭代 sigmoid 转换。

该转换在断棒过程中表现为迭代 sigmoid 转换。
`Dirichlet`分布的构建：第一个对数几率
通过 sigmoid 函数转换为第一个概率和概率
其余一切，然后过程递归。

这是双射且适用于 HMC；然而它混合
坐标一起且不太适合优化。
""

    域名 = 约束.实向量
    目标域 = 约束.单纯形
    双射 = 真实

    def __等于__(self, 其他):
        返回 isinstance(其他, StickBreakingTransform)

    def _调用(self, x):
        偏移 = x.形状[-1] + 1 - x.新一(x.形状[-1]).累加和(-1)
        z = _clipped_sigmoid(x - 偏移量.日志())
        z 累积乘积 = (1 - z).cumprod(-1)
        y = 填充(z, [0, 1] 值=1) * 填充(z 累积乘积, [1, 0] 值=1)
        返回 y

    def 反向(self, y):
        y 裁剪 = y[..., :-1]
        偏移 = y.形状[-1] - y.新一(y_crop.形状[-1]).累加和(-1)
        sf = 1 - y_crop.累加和(-1)
        我们将值限制在正数，因为有时 sf 可能不是正数
        发生当 y[-1] ~ 0 或 y[:-1].sum() ~ 1 时
        sf = 火炬.卡钳(sf, 最小值=火炬.finfo(y.数据类型).微小。)
        x = y_crop.日志() - sf.日志() + 偏移量.日志()
        返回 x

    def 对数绝对雅可比行列式(self, x, y):
        偏移 = x.形状[-1] + 1 - x.新一(x.形状[-1]).累加和(-1)
        x = x - 偏移量.日志()
        # 使用身份 1 - sigmoid(x) = exp(-x) * sigmoid(x)
        detJ = (-x + F.对数 Sigmoid(x) + y[..., :-1].日志()).总和(-1)
        返回 detJ

    def 前向形状(self, 形状):
        如果 长度(形状) < 1:
            提升 ValueError(输入维度太少)
        返回 形状[-1] + (形状[-1] + 1,)

    def 逆向形状(self, 形状):
        如果 长度(形状) < 1:
            提升 ValueError(输入维度太少)
        返回 形状[-1] + (形状[-1] - 1,)



[文档]类 LowerCholeskyTransform(Transform):
"""
将非约束矩阵转换为具有非负对角线元素的下三角矩阵的转换。
对角线元素非负。

这对于用 Cholesky 分解来参数化正定矩阵非常有用
它们的 Cholesky 分解。
"""

约束域 = 约束.独立约束(constraints.real, 2)
目标域 = 约束的下三角 Cholesky 分解

def __eq__(self, other):
返回 isinstance(other, LowerCholeskyTransform)

def _call(self, x):
return x.tril(-1) + x.diagonal(dim1=-2, dim2=-1).exp().diag_embed()

def _inverse(self, y):
return y.tril(-1) + y.diagonal(dim1=-2, dim2=-1).log().diag_embed()


[文档]class PositiveDefiniteTransform(Transform):
"""
将无约束矩阵转换为正定矩阵的转换。
"""

domain = 独立约束(constraints.real, 2)
codomain = 紧定约束(constraints.positive_definite)  # 忽略赋值类型

def __eq__(self, other):
return isinstance(other, PositiveDefiniteTransform)

def _call(self, x):
x = LowerCholeskyTransform()(x)
return x @ x.mT

def _inverse(self, y):
y = torch.linalg.cholesky(y)
return LowerCholeskyTransform().inv(y)


类 猫转换(变换):
    ""
变换算子，将一系列变换 `tseq` 应用到 `dim` 的每个子矩阵上
，其长度为 `lengths[dim]`
与 :func:`torch.cat` 兼容的方式。

示例::

       x0 = torch.cat([torch.range(1, 10), torch.range(1, 10)], dim=0)
       x = torch.cat([x0, x0], dim=0)
       t0 = CatTransform([ExpTransform(), identity_transform], dim=0, lengths=[10, 10])
       t = CatTransform([t0, t0], dim=0, lengths=[20, 20])
       y = t(x)
""

    转换: 列表[变换]

    def __init__(self, tseq, 暗=0, 长度=无, 缓存大小=0):
        断言 所有(isinstance(t, 变换) 为 t 在 tseq)
        如果 缓存大小:
            tseq = [t.带缓存(缓存大小) 为 t 在 tseq]
        超级().__init__(缓存大小=缓存大小)
        self.转换 = 列表(tseq)
        如果 lengths is 无:
            lengths = [1] * 长度(self.转换)
        self.lengths = 列表(长度)
        断言 长度(self.长度) == 长度(self.转换)
        self.维度 = 维度

    @lazy_property
    def 事件维度(self) 翻译 int:  # 类型：忽略[覆盖]
        返回 最大值(t.事件维度 为 t 在 self.转换)

    @lazy_property
    def 长度(self) 翻译 int:
        返回 总和(self.长度)

    def 带缓存(self, 缓存大小=1):
        如果 self._cache_size == 缓存大小:
            返回 self
        返回 猫转换(self.转换, self.暗, self.长度, 缓存大小)

    def _调用(self, x):
        断言 -x.暗() <= self.维度 < x.暗()
        断言 x.尺寸(self.暗) == self.长度
        yslices = 输入文本为空，请提供需要翻译的文本
        开始 = 0
        为 trans, 长度 在 zip(self.转换, self.长度):
            xslice = x.狭窄(self.暗, 开始, 长度)
            yslices.append(trans(xslice))
            开始 = 开始 + 长度  避免使用 += 以兼容 JIT
        返回 火炬.猫(yslices, 暗=self.暗)

    def 反向(self, y):
        断言 -y.暗() <= self.维度 < y.暗()
        断言 y.尺寸(self.暗) == self.长度
        xslices = 输入文本为空，请提供需要翻译的文本
        开始 = 0
        为 trans, 长度 在 zip(self.转换, self.长度):
            纵切片 = y.狭窄(self.暗, 开始, 长度)
            xslices.append(trans.涉及(yslice))
            开始 = 开始 + 长度  避免使用 += 以兼容 JIT
        返回 火炬.猫(xslices, 暗=self.暗)

    def 对数绝对雅可比行列式(self, x, y):
        断言 -x.暗() <= self.维度 < x.暗()
        断言 x.尺寸(self.暗) == self.长度
        断言 -y.暗() <= self.维度 < y.暗()
        断言 y.尺寸(self.暗) == self.长度
        logdetjacs = 输入文本为空，请提供需要翻译的文本
        开始 = 0
        为 trans, 长度 在 zip(self.转换, self.长度):
            xslice = x.狭窄(self.暗, 开始, 长度)
            纵切片 = y.狭窄(self.暗, 开始, 长度)
            对数雅可比 = trans.对数绝对雅可比行列式(xslice, yslice)
            如果 trans.事件维度 < self.事件维度:
                对数雅可比 = _sum_rightmost(logdetjac, self.事件维度 - trans.事件维度)
            logdetjacs.append(logdetjac)
            开始 = 开始 + 长度  避免使用 += 以兼容 JIT
        决定是否连接或求和。
        维度 = self.维度
        如果 维度 ≥ 0:
            维度 = 维度 - x.暗()
        维度 = 维度 + self.事件维度
        如果 维度 < 0:
            返回 火炬.猫(logdetjacs, 暗=暗)
        else:
            返回 总和(logdetjacs)

    @property
    def 双射(self) 翻译 布尔:  # 类型：忽略[覆盖]
        返回 所有(t.双射 为 t 在 self.转换)

    @constraints.依赖属性
    def 域名(self):
        返回 约束.猫(
            [t.域名 为 t 在 self.转换] self.暗, self.lengths
        )

    @constraints.依赖属性
    def 定义域(self):
        返回 约束.猫(
            [t.目标域 为 t 在 self.转换] self.暗, self.lengths
        )



类 栈转换(变换):
    ""
变换算子，将一系列变换 `tseq` 应用到 `dim` 的每个子矩阵上
对 `dim` 中的每个子矩阵进行分量变换
与 :func:`torch.stack` 兼容的方式。

示例::

       x = torch.stack([torch.range(1, 10), torch.range(1, 10)], dim=1)
       t = StackTransform([ExpTransform(), identity_transform], dim=1)
       y = t(x)
""

    转换: 列表[变换]

    def __init__(self, tseq, 暗=0, 缓存大小=0):
        断言 所有(isinstance(t, 变换) 为 t 在 tseq)
        如果 缓存大小:
            tseq = [t.带缓存(缓存大小) 为 t 在 tseq]
        超级().__init__(缓存大小=缓存大小)
        self.转换 = 列表(tseq)
        self.维度 = 维度

    def 带缓存(self, 缓存大小=1):
        如果 self._cache_size == 缓存大小:
            返回 self
        返回 栈转换(self.转换, self.暗, 缓存大小)

    def _切片(self, z):
        返回 [z.选择(self.暗, i) 为 i 在 范围(z.尺寸(self.暗))]

    def _调用(self, x):
        断言 -x.暗() <= self.维度 < x.暗()
        断言 x.尺寸(self.暗) == 长度(self.转换)
        yslices = 输入文本为空，请提供需要翻译的文本
        为 xslice, 转换 在 zip(self._切片(x), self.转换):
            yslices.append(trans(xslice))
        返回 火炬.栈(yslices, 暗=self.暗)

    def 反向(self, y):
        断言 -y.暗() <= self.维度 < y.暗()
        断言 y.尺寸(self.暗) == 长度(self.转换)
        xslices = 输入文本为空，请提供需要翻译的文本
        为 yslice, 转换 在 zip(self._切片(y), self.转换):
            xslices.append(trans.涉及(yslice))
        返回 火炬.栈(xslices, 暗=self.暗)

    def 对数绝对雅可比行列式(self, x, y):
        断言 -x.暗() <= self.维度 < x.暗()
        断言 x.尺寸(self.暗) == 长度(self.转换)
        断言 -y.暗() <= self.维度 < y.暗()
        断言 y.尺寸(self.暗) == 长度(self.转换)
        logdetjacs = 输入文本为空，请提供需要翻译的文本
        yslices = self._切片(y)
        xslices = self._切片(x)
        为 xslice, yslice, 转换 在 zip(xslices, yslices, self.转换):
            logdetjacs.append(trans.对数绝对雅可比行列式(xslice, yslice))
        返回 火炬.栈(logdetjacs, 暗=self.暗)

    @property
    def 双射(self) 翻译 布尔:  # 类型：忽略[覆盖]
        返回 所有(t.双射 为 t 在 self.转换)

    @constraints.依赖属性
    def 域名(self):
        返回 约束.栈[t.域名 为 t 在 self.转换] self.暗)

    @constraints.依赖属性
    def 定义域(self):
        返回 约束.栈[t.目标域 为 t 在 self.转换] self.暗)



[文档]类 CumulativeDistributionTransform(Transform):
"""
通过概率分布的累积分布函数进行转换。

Args:
distribution (Distribution): 要用于转换的累积分布函数的分布。
the transformation.

示例::

从多元正态分布构建高斯 Copula。
base_dist = MultivariateNormal(
loc=torch.zeros(2),
scale_tril = LKJCholesky(2).sample(),
        )
transform = CumulativeDistributionTransform(Normal(0, 1))
copula = TransformedDistribution(base_dist, [transform])
"空行"

双射 = True
值域 = 约束的单位区间
符号 = +1

def __init__(self, distribution, cache_size=0):
super().__init__(cache_size=cache_size)
self.distribution = distribution

    @property
def domain(self) -> constraints.Constraint:  # type: ignore[override]
return self.distribution.support

def _call(self, x):
return self.distribution.cdf(x)

def _inverse(self, y):
return self.distribution.icdf(y)

def log_abs_det_jacobian(self, x, y):
return self.distribution.log_prob(x)

def with_cache(self, cache_size=1):
if self._cache_size == cache_size:
return self
return CumulativeDistributionTransform(self.distribution, cache_size=cache_size)