# mypy: 允许未类型化定义
导入
内置函数
导入
集合
导入 contextlib
导入
复制
导入 functools
导入
检查
导入
数学
导入
操作系统
导入
系统
导入
警告
from itertools 导入
连接
from 类型
导入
代码类型,
函数类型,
模块类型
from 打字
导入
任何,
可调用,
命名元组,
可选,
联合
导入
火炬
导入 torch.utils._pytree
是 pytree
from torch._C 导入
脚本对象
# 类型：忽略[已定义]
from torch._library.fake_class_registry 导入
模拟脚本对象
from 兼容性
导入
兼容性
from _lazy_graph_module
导入 _make_graph_module
from .graph 导入
_PyTree 代码生成器, _PyTreeInfo,
图
from .图模块
导入
图模块
from .节点
导入
参数,
基础类型,
地图聚合
from 代理
导入
参数代理,
代理,
范围,
范围上下文管理器,
跟踪器基类
有变量内容 =
检查.CO_VARARGS |
检查.CO_VARKEYWORDS
这些需要在全局范围内运行以正确处理嵌套调用
_orig_module_call: 可调用 =
火炬.
神经网络.
模块.__call__
_orig_module_getattr: 可调用 =
火炬.
神经网络.
模块.__getattr__
_proxyable_classes: 字典[
类型,
无] = {}
_is_fx_tracing_flag = 假
def is_fx_tracing():
返回 _is_fx_tracing_flag
@compatibility(兼容旧版本=
是)
类
可代理类元数据(
类型):
""
可代理类元数据允许您构建给定的 Python 类
符号可追踪。例如：
导入 torch
导入 torch.fx
class TensorPair(metaclass=torch.fx.ProxyableClassMeta):
def __init__(self, left, right):
self.left, self.right = left, right
def add(self, other):
l = self.left + other.left
r = self.right + other.right
return TensorPair(l, r)
def mul(self, other):
l = self.left * other.left
r = self.right * other.right
return TensorPair(l, r)
def use_tensor_pair_ctor(x: TensorPair, y: torch.Tensor):
s = x.add(TensorPair(y, y))
return s.mul(x)
x = TensorPair(torch.randn(5, 3), torch.randn(5, 3))
y = torch.randn(5, 3)
ref_out = 使用 tensor_pair_ctor(x, y)
traced = torch.fx.symbolic_trace(use_tensor_pair_ctor)
print(traced.code)
'''
def forward(self, x : __main___TensorPair, y : torch.Tensor):
tensor_pair = __main___TensorPair(y, y); y = None
add = x.add(tensor_pair); tensor_pair = None
mul = add.mul(x); add = x = None
return mul
'''
从这个例子中，我们可以看到使用（`TensorPair`）作为元类的类（``ProxyableClassMeta``）可以被记录在符号追踪中。
定义为具有元类（`ProxyableClassMeta`）的类可以记录在符号追踪中。
。
""
def __init__(类,
名称, bases,
属性):
_proxyable_classes.setdefault(类)
超级().__init__(
名称, bases,
属性)
def __调用__(
类, *
参数, **kwargs):
实例 =
类.__new__(
类) # type: ignore[call-overload]
如果
不是 is_fx_tracing():
类.__init__(
实例, *
参数, **kwargs)
# 类型：忽略[杂项]
返回
实例
发现代理 =
输入文本为空，请提供需要翻译的文本
def 检查代理(a):
如果 isinstance(a,
代理):
发现代理.append(a)
map_aggregate(参数,
检查代理)
map_aggregate(kwargs, 检查代理)
如果
长度(
找到代理) != 0:
追踪器 =
找到代理[0].
追踪器
返回
追踪器.
创建代理(
调用函数,
类,
参数, kwargs)
else:
类.__init__(
实例, *
参数, **kwargs)
# 类型：忽略[杂项]
返回
实例
def _修补函数(
函数:
函数类型,
参数个数: int)
翻译
函数类型:
co = 函数.__code__
co_flags = 参数列表.co_flags & ~
包含 VARSTUFF
命名参数:
元组
如果
有属性(
参数列表,
"合格名称"):
Python-3.11+ 代码签名
参数 = (
参数个数,
0,
0,
参数列表.
参数局部变量个数,
co.co_stacksize,
co_flags,
co.代码,
公司.
公司常量,
公司.
公司名称,
公司.
公司变量名,
公司.co_filename,
公司.
公司名称,
公司.
公司全称,
# 类型：忽略[已定义]
co.co_firstlineno,
co.co_lnotab,
co.可诉的,
# 类型：忽略[已定义]
co.co_freevars,
co.co_cellvars,
)
如果...否则
有属性(co,
co_posonlyargcount):
co_args = (
nargs,
0,
0,
co.co_nlocals,
co.co_stacksize,
co_flags,
co.co_code,
co.co_consts,
co.co_names,
co.co_varnames,
co.co_filename,
co.公司名称,
公司.
公司第一行号,
公司.co_lnotab,
co.co_freevars,
co.co_cellvars,
)
else:
参数 = (
参数个数,
0,
参数列表.
参数局部变量个数,
co.co_stacksize,
co_flags,
co.代码,
公司.
公司常量,
公司.
公司名称,
公司.
公司变量名,
公司.co_filename,
公司.
公司名称,
公司.
公司第一行号,
co.co_lnotab,
co.co_freevars,
co.co_cellvars,
)
new_code = CodeType(*co_args) # type: ignore[arg-type]
返回
函数类型(
新代码,
函数.
__全局变量__,
函数.__name__,
函数.
默认值,
函数.
闭包
)
需要插入用于 *args 和 **kwargs 的占位符节点
我们不能正常调用这个函数，否则它将尝试解包它们
取而代之，让我们让 Python 认为 args 和 kwargs 是正常变量
@compatibility(兼容旧版本=
错误)
类 PHBase:
""
代表输入占位符的 `concrete_args` 对象
""
def __repr__(self):
返回 "PH"
PH = PHBase()
@compatibility(兼容旧版本=
错误)
类 PHWithMeta(PHBase):
""
表示输入占位符的对象，用于`concrete_args`
""
def __init__(self, ph_key: 可选[
字符串] =
无):
超级().__init__()
# 请用户提供一个标识符以供用户在分析期间识别占位符节点
self.ph_key = ph_key
def _transfer_attrs(fr, 到):
为
属性名称
在
目录(
法语):
属性值 = getattr(
法语,
属性名称)
如果 (
不是
可调用(
属性值)
和
不是
属性名称.
以...开头("__")
和
不是
有属性(
到,
属性名称)
):
setattr(到,
属性名称, attr_val)
[文档]@compatibility(
兼容旧版本=
是)
类
追踪器(TracerBase):
# 参考：https://github.com/pytorch/pytorch/issues/54354
# 该文档字符串的第一行覆盖了 Sphinx 生成的文档字符串
# 文档。我们需要它，这样 Sphinx 就不会泄露 `math`s 路径（例如 `）。
# 构建环境（例如 `）。
"""Tracer(autowrap_modules=(math,), autowrap_functions=())
``Tracer`` 是实现符号跟踪功能的类
``torch.fx.symbolic_trace`` 的调用与 ``Tracer().trace(m)`` 等价
。
可以通过子类化 ``Tracer`` 来覆盖跟踪的各种行为
进程。可以覆盖的不同行为被描述
在类中方法的文档字符串中。
""
此 API 未检查 BC，因为`autowrap_modules`的默认值
包含指向 `math` 模块的本地文件路径，该路径会抖动
跨机器
@compatibility(兼容旧版本=
是)
def __init__(
self,
自动包装模块:
元组[
模块类型] = (
数学,),
自动包装函数:
元组[
可调用, ...] = (),
参数形状常量:
布尔值 =
错误,
) 翻译
无:
此方法签名被本类的第一行覆盖
如果此方法的签名被修改，覆盖它的签名也应相应修改。
如果此方法的签名被修改，覆盖它的签名也应相应修改。
""
构建一个 Tracer 对象。
参数：
autowrap_modules (模块类型元组): 默认为`(math, )`，
Python 模块中应自动包装其函数的模块
无需使用 fx.wrap()。向后兼容性
此参数有保证。
自动换行函数（Tuple[Callable, ...]）：默认为 `()`
Python 应自动包装的功能
需要使用 fx.wrap()。对此的向后兼容性
参数有保证。
param_shapes_constant (bool): 当此标志被设置时，对 shape 的调用
模块参数的大小和其他一些形状属性将被直接评估，而不是返回一个新的代理值
用于属性访问。此参数向后兼容性得到保证。
对于属性访问。此参数向后兼容性得到保证。
此保证得到保证。
""
超级().__init__()
# 我们在追踪时将积极包装的函数
# 这将自动捕获 `math.sqrt()` 和 `from math import sqrt`
self._autowrap_function_ids: 设置[int] = {
id(值)
为
名称,
值
在 chain.from_iterable(
m.字典.
项目()
为 m
在
自动包装模块
)
如果
不是
名称.
以...开头(
“_”)
和
可调用(
值)
}
self._自动包装函数 ID.
更新({id(f)
为 f
在
自动包装函数})
# 在开始时应用自动包装的 Python 模块，除以下模块外
追踪时看到的模块
self._autowrap_search: 列表[
模块类型] =
列表(autowrap_modules)
self.param_shapes_constant = 常量参数形状
self.子模块路径:
可选[
字典[
火炬.
神经网络.
模块,
字符串]] =
无
self.根模块名称:
字符串 =
请提供需要翻译的文本
将包含模块的名称映射到操作符名称
self.范围 =
范围(
输入文本翻译为简体中文为："",
无)
记录模块调用栈
self.模块堆栈 =
集合.
有序字典()
self.调用次数:
字典[
字符串, int] = {}
节点名称到模块作用域的映射
self.节点名称到作用域映射:
字典[
字符串,
元组[
字符串,
类型]] = {}
_qualname 计数器:
字典[
字符串, int] =
集合.
默认字典(int)
[文档] @兼容性(向后兼容=True)
def get_fresh_qualname(self, prefix: str) -> str:
"""
获取一个前缀的新名称并返回它。此函数确保
它不会与图上的现有属性冲突。
"""
# 这里指的是如果模块根本没有任何前缀，那么
# 应该重置计数器从头开始
# 这有点儿像...一种小聪明的做法（不能涵盖所有情况），但精确
# 地命名前缀并不是一个正确性问题，而是一个美观问题
# 问题
qualname = f"{prefix}0"
if not hasattr(self.root, qualname):
self._qualname_counter[prefix] = 0
返回全名
i = self._qualname_counter[prefix]
while True:
qualname = f"{prefix}{i}"
i += 1
if not hasattr(self.root, qualname):
break
self._qualname_counter[prefix] = i
返回 qualname
[文档] @compatibility(
兼容旧版本=
是)
def create_arg(self, a: 任何)
翻译
"论点":
""
一种指定在准备值时跟踪行为的方法
用作 ``Graph`` 节点的参数。
默认情况下，行为包括：
#. 遍历集合类型（例如元组、列表、字典）并递归
在元素上调用 `create_args`。
#. 给定一个代理对象，返回底层 IR ``Node`` 的引用
#. 给定一个非代理张量对象，针对各种情况发出 IR：
* 对于一个参数，发出一个指向该参数的 `get_attr` 节点
对于非参数张量，将张量存储在特殊属性中。
该方法可以被重写以支持更多类型。
此方法可以被重写以支持更多类型。
参数：
a（任何类型）：要作为“图”中的“参数”发出的值。
返回：
``a`` 转换为适当的 ``参数``
""
# 基础追踪器用于在没有关联的情况下构建图
模块层次结构，因此它永远无法创建参数引用。
默认追踪器添加了引用参数的能力
追踪模块。
如果 isinstance(a,
火炬.
神经网络.
参数):
为 n, p
在 self.
根.
命名参数。():
如果 a is p:
返回 self.
创建节点(
获取属性, n, (), {})
提升
名字错误(
"该参数不是此模块的成员")
如果...否则 isinstance(a,
火炬.
张量):
为 n_, p_
在 self.
根.
命名缓冲区():
如果 a is p_:
返回 self.
创建节点(
获取属性, n_, (), {})
如果...否则 isinstance(a,
火炬.
神经网络.
模块):
为 n_, p_
在 self.
根.
命名模块():
如果 a is p_:
返回 self.
创建节点(
获取属性, n_, (), {})
对于直接作为参数出现的 NamedTuple 实例，我们发出一个节点来构建 NamedTuple，并使用该节点作为参数。
使用该节点作为构建 NamedTuple 的参数。
如果 isinstance(a,
元组)
和
有属性(a,
"_字段"):
args = 元组(self.create_arg(
元素)
为
元素
在 a)
返回 self.
创建节点(
调用函数, a.
类,
参数, {})
张量没有可靠的字符串 repr()，无法从中构建（我们可能也不想依赖它），因此对于遇到的任何常量张量值，首先搜索它们是否是某个模块层次结构中模块的属性。如果是，则发出
（并且我们可能不想依赖它），所以对于遇到的任何常量张量值，首先搜索它们是否是某个模块层次结构中模块的属性。如果是，则发出
（并且我们可能不想依赖它），所以对于遇到的任何常量张量值，首先搜索它们是否是某个模块层次结构中模块的属性。如果是，则发出
（并且我们可能不想依赖它），所以对于遇到的任何常量张量值，首先搜索它们是否是某个模块层次结构中模块的属性。如果是，则发出
# 一个 get_attr 来检索该张量。否则，我们将将其存储在 Module 的特殊属性中，以便我们可以
# 将张量值存储到 Module 的特殊属性中，以便我们可以
# 使用 get_attr 来检索它。
如果 isinstance(a, (
火炬.
张量,
脚本对象,
假脚本对象)):
qualname: 可选[
字符串] = self.
张量属性.
获取(a)
# 在 Module 层次结构中未找到 Tensor，将其存储起来。
特殊属性并设置 qualname 以引用该属性
如果
不是 qualname:
基础名称 = (
"_tensor_constant"
如果 isinstance(a,
火炬.
张量)
否则 "_torchbind_obj"
)
qualname = self.获取新鲜的全名(
基础名称)
断言 isinstance(qualname,
字符串)
self.张量属性[a] = qualname
setattr(self.根, qualname, a)
返回 self.
创建节点(
获取属性, qualname, (), {})
如果
类型(a)
在
_可代理类:
# 这是一个我们没有见证其构造的代理类实例。
# 将其作为常量属性内部化。
# TODO: 二分查找
qualname = self.获取新鲜的全名(f
"_"{a.
类.__name__}
_常量_")
断言 isinstance(qualname,
字符串)
setattr(self.根, qualname, a)
返回 self.
创建节点(
获取属性, qualname, (), {})
返回
超级().create_arg(a)
[文档] @兼容性(向后兼容=True)
def is_leaf_module(self, m: torch.nn.Module, module_qualified_name: str) -> bool:
"""
指定给定 ``nn.Module`` 是否为“叶子”模块的方法。
叶子模块是 IR 中出现的原子单元，通过`call_module`调用进行引用。
默认情况下，PyTorch 标准库命名空间（torch.nn）中的模块是叶子模块。所有其他模块都通过它进行追踪。
PyTorch 标准库命名空间（torch.nn）中的模块是叶子模块。所有其他模块都通过它进行追踪。
所有其他模块都通过它进行追踪。
他们的操作记录被记录，除非另有说明
通过此参数。
Args:
m (模块): 正在被查询的模块
模块限定名称（字符串）：此模块根路径。例如，
如果你有模块层次结构，其中子模块 ``foo`` 包含
子模块 ``bar``，而 ``bar`` 包含子模块 ``baz``，那么该模块将
在此处显示为限定名称 ``foo.bar.baz``。
"""
返回 (
m.__module__.startswith("torch.nn")
或者 m.__module__.startswith("torch.ao.nn")
) 且不是 torch.nn.Sequential 的实例
[文档] @兼容性(向后兼容=True)
def path_of_module(self, mod: torch.nn.Module) -> str:
"""
查找模块层次结构中 ``mod`` 的限定名称的辅助方法
的 ``root``。例如，如果 ``root`` 有一个名为 ``foo`` 的子模块，该子模块
又有一个名为 ``bar`` 的子模块，将 ``bar`` 传递给此函数将返回
字符串 "foo.bar"。
Args:
模块 (str): 获取指定模块的合格名称。
"""
# 优先选择 O(1) 算法
如果 self.submodule_paths 不为空：
path = self.submodule_paths.get(mod)
如果 path 为 None：
抛出 NameError 异常("模块未作为子模块安装")
assert isinstance(path, str)
return path
# O(N^2) fallback in the case that we didn't store the submodule
# paths.
else:
for n, p in self.root.named_modules():
if mod is p:
return n
引发 NameError("模块未作为子模块安装")
[文档] @compatibility(
兼容旧版本=
是)
def 调用模块(
self,
m: 火炬.
神经网络.
模块,
前向:
可调用[...,
任何
]
参数:
元组[
任何, ...
]
kwargs: 字典[
字符串,
任何
]
) 翻译
任何:
""
指定此 `Tracer` 遇到 `nn.Module` 实例调用时的行为方法。
默认情况下，行为是检查被调用的模块是否是叶子模块。
通过 `is_leaf_module` 来检查。如果是，则发出一个指向 `call_module` 节点的引用。
通过 `is_leaf_module` 来检查。如果是，则发出一个指向 `call_module` 节点的引用。
在 ``Graph`` 中使用 ``m``。否则，正常调用 ``Module``，通过其 ``forward`` 函数进行跟踪。
跟踪该方法的操作。
此方法可以被重写，例如创建嵌套的跟踪 ``GraphModules``，或者您在跟踪过程中想要的其他任何行为。
任何其他行为你想要在跟踪过程中实现。
模块边界。
参数：
m（模块）：正在发出调用的模块
forward（可调用）：要调用的 ``Module`` 的 forward() 方法
args（元组）：模块调用位置的参数
kwargs（字典）：模块调用位置的 kwargs
返回：
模块调用的返回值。在“call_module”的情况下。
node 被发出，这是一个“代理”值。否则，它是任何值。
从“模块”调用中返回了值。
""
模块限定名 = self.
模块路径(m)
与
范围上下文管理器(
self.范围,
范围(
模块限定名称,
类型(m))
) 是
范围:
# 模块栈是一个有序字典，因此写入然后删除相当于在列表上的 push/pop 操作
# 条目写入和删除等同于列表的 push/pop 操作
num_calls = self.调用次数.
获取(
模块限定名称, 0)
模块键 = (
f"{范围.
模块路径}@{
呼叫次数}"
如果 num_calls > 0
否则
范围.
模块路径
)
self.模块栈[
模块键] = (
模块限定名称,
范围.
模块类型)
self.呼叫次数[
模块限定名称] = num_calls + 1
如果
不是 self.
叶子模块(m,
模块限定名称):
返回值 =
前向(*
参数, **kwargs)
else:
返回值 = self.
创建代理(
调用模块,
模块限定名,
参数, kwargs
)
键, _ = self.
模块栈.popitem(
最后=
是)
断言 key ==
模块键, f
"意外键"{
键}"
返回
返回值
[文档] @compatibility(
兼容旧版本=
错误)
def getattr(self, 属性:
字符串,
属性值:
任何,
参数代理缓存:
字典[
字符串,
任何
)]
""
指定此 `Tracer` 在调用 getattr 时行为的方法。
在对 `nn.Module` 实例的调用上。
默认情况下，行为是返回属性的代理值。
同时也会将代理值存储在 `parameter_proxy_cache` 中，以便未来的调用可以重用代理而不是创建一个新的。
此方法可以被覆盖，例如，不返回代理。
这种方法可以被覆盖，例如，不返回代理。
查询参数。
参数：
attr（字符串）：要查询的属性名称
attr_val（任何类型）：属性的值
parameter_proxy_cache（字典[字符串，任何类型]）：属性名称到代理的缓存
返回值：
getattr 调用返回的值。
""
def maybe_get_proxy_for_attr(
attr_val, 搜索集合,
参数代理缓存
):
为 n, p
在
搜索集合:
如果
属性值 is p:
如果 n
不是
在
参数代理缓存:
kwargs = {}
如果 (
"proxy_factory_fn"
在
检查.
签名(self.
创建代理).
参数
):
kwargs["proxy_factory_fn"] = (
无
如果
不是 self.param_shapes_constant
否则 lambda
节点:
参数代理(
self, 节点, n,
属性值
)
)
代理值 = self.
创建代理(
获取属性, n, (), {}, **kwargs) # type: ignore[arg-type]
参数代理缓存[n] =
val 代理
返回
参数代理缓存[n]
返回
无
如果 isinstance(
属性值,
火炬.
神经网络.
参数):
可能的参数代理 =
可能获取代理以属性(
属性值, self.
根.
命名参数。(),
参数代理缓存
)
如果
可能的参数代理 is
不是
无:
返回
可能的参数代理
如果 self.
代理缓冲区属性
和 isinstance(
属性值,
火炬.
张量):
可能的缓冲代理 =
可能获取属性代理(
属性值, self.
根.
命名缓冲区(),
参数代理缓存
)
如果
可能的缓冲代理 is
不是
无:
返回
可能的缓冲代理
返回
属性值
此方法将被重构
[文档] @compatibility(
兼容旧版本=
错误)
def 为 root 创建参数(self,
root 函数,
是否是模块,
具体参数=
无):
""
创建与“root”签名对应的“placeholder”节点
模块。此方法检查根的签名并发出这些
节点相应地，也支持 `*args` 和 `**kwargs`。
""
在某些情况下，一个函数或方法被装饰器包装
通过 `functools.wraps` 定义。在这种情况下，外部的代码对象
很可能不包含我们关心的实际参数，因此需要解包
函数以到达最内层的可调用对象。
分析用函数 =
检查.
解包(
根函数)
co = 分析用函数.
__代码__
参数总数 =
参数列表.
参数计数 +
参数列表.
仅参数计数
原始参数 =
列表(
参数列表.
变量名)
迭代器名称 =
迭代(
参数列表.
变量名)
参数:
列表[
任何] =
输入文本为空，请提供需要翻译的文本
跳过参数索引 = 0
如果
是否是模块:
如果
总参数数 == 0:
提升
运行时错误(
"``self`` 参数不能是 *args 展开的一部分！"
)
跳过参数索引 = 1
下一(
名称迭代器)
# 跳过自身
参数.append(self.
根)
sig = 检查.
签名(fn_for_analysis)
这涵盖了非常具体的情况，即我们正在传递平面
# concrete_args 作为元组，但我们的跟踪函数接受(*args, **kwargs)。
# 在这种情况下，只需取 concrete_args 并通过它们传递。
name_idx = 0
如果 (
isinstance(concrete_args, 元组)
和
长度(concrete_args) > 0
和 (
参数列表.co_flags & HAS_VARSTUFF)
和
总参数 == 1
):
为
具体参数
在 concrete_args:
out = self.创建代理(
占位符, f
输入_{
名称索引}", (), {})
如果 isinstance(concrete_arg, PHBase):
如果
混凝土参数 != PH:
# 在使用其他占位符的情况下传输属性
# 除了单例 PH（PH 没有可传输的属性）。
代理是从占位符创建的。
# 将任何元数据（以占位符形式放入）传输
用户设置的属性从占位符到
# 基础节点（代理由用户解包）
# 元数据应包含）。
_transfer_attrs(法语=
具体参数,
到=
输出.
节点)
参数.append(
输出)
name 索引 += 1
返回
根函数, args
参数名 = [
下一(
名称迭代器)
为
索引
在
范围(
跳过参数索引,
总参数)]
如果 isinstance(
具体参数,
元组):
如果
长度(
参数名) !=
长度(
具体参数):
提升
运行时错误(
f跟踪预期{
长度(
参数名)}
参数但得到{
长度(
具体参数)}
具体参数
)
concrete_args = 字典(zip(
参数名,
concrete 参数))
def 代理占位符(
名称):
返回 self.
_代理占位符(
名称,
concrete 参数, sig,
分析用函数)
参数.
扩展(
代理占位符(names)
为
名称
在
参数名)
如果 co.
仅参数计数协变量 > 0
或者 co.co_flags & HAS_VARSTUFF:
# TODO: 为 *args 和 **kwargs 添加类型注解
如果 co.co_flags &
检查.CO_VARARGS:
参数.append(
代理占位符("*" +
下一(
名字迭代器)))
如果 co.co_flags &
检查.CO_VARKEYWORDS:
参数.append(proxy_placeholder(
** +
下一(
名称迭代)))
根函数 =
_补丁函数(
根函数,
长度(
参数))
平铺参数,
在规范中 =
py 树.
树形展平(
元组(
参数))
如果
不是
所有(
儿童.
是否为叶子节点()
为
儿童
在
在规范中.
儿童规格):
# 如果我们有扁平化的 pytree 输入
# 对于`具体参数`，生成一个扁平化包装器
# 原始根函数及其返回值
self.graph.代码生成器 =
_PyTree 代码生成器(
_PyTreeInfo(orig_args[
总参数数
]
在规范中,
无)
)
def 展平函数(*
参数):
树形参数 =
py 树.
树形展开(
列表(
参数),
在规范中)
树输出 =
根函数(*
树形参数)
输出参数,
输出规范 =
py 树.
树形展平(
树出)
断言 isinstance(self.graph.
代码生成器,
_PyTree 代码生成器)
self.graph.代码生成器.
Py 树信息 = (
self.graph.代码生成器.
pytree 信息.
替换(
输出规范=
输出规范)
)
返回
输出参数
返回
展平函数,
平铺参数
返回
根函数,
参数
[文档] @compatibility(
兼容旧版本=
是)
def 跟踪(
self,
根:
联合[
火炬.
神经网络.
模块,
可调用[...,
任何]],
concrete_args: 可选[
字典[
字符串,
任何]] =
无,
) 翻译
图:
""
跟踪 ``root`` 并返回相应的 FX ``Graph`` 表示形式。 ``root``
可以是一个 ``nn.Module`` 实例或 Python 可调用对象。
注意，在此调用之后，``self.root`` 可能与传入的 ``root`` 不同。
在这里。例如，当将一个自由函数传递给 `trace()` 时，我们将
创建一个 `nn.Module` 实例作为根，并添加嵌入常量
to.
参数：
root (Union[Module, Callable]): 要么是 ``Module``，要么是一个函数，可以通过它进行追踪。此参数向后兼容性得到保证。
traced through. Backwards-compatibility for this parameter is guaranteed.
guaranteed.
concrete_args (可选[Dict[str, any]]): 应当被视为非代理的具体参数。
此参数为实验性参数，其向后兼容性**不**受保证。
表示传入的 `root` 的语义的 ``Graph``。
返回：
表示传入的 `root` 的语义的 ``Graph``。
""
全局 _is_fx_tracing_flag
旧_is_fx_tracing_flag = _is_fx_tracing_flag
_is_fx_tracing_flag = 真实
尝试:
如果 isinstance(
根,
火炬.
神经网络.
模块):
在重新跟踪之前，请先对_LazyGraphModule 进行真正的重新编译，因为跟踪方法无法跟踪 LazyForward 方法。错误信息：
请访问以下链接获取更多信息：
https://gist.github.com/shunting314/75549c2e82ae07ac1139c94a3583d259
否则无法进行。
from torch.fx._lazy_graph_module 导入 _LazyGraphModule
_LazyGraphModule.force_recompile(根)
self.根 =
根
断言
有属性(
类型(
根), self.
跟踪的函数名
), f"跟踪的函数名="{self.
跟踪的函数名}
不存在于{
类型(
根).__name__}"
fn = getattr(类型(
根), self.
跟踪函数名)
self.根模块名 =
根.
_获取名称()
self.子模块路径 = {
模块:
名称
为
名称,
修饰
在
根.
命名模块()}
else:
self.根 =
火炬.
神经网络.
模块()
fn = 根
tracer_cls: 可选[
类型[
追踪器]] = getattr(self,
"__类__",
无)
self.图 =
图(tracer_cls=tracer_cls)
如果
有属性(
函数,
"__代码__"):
代码 =
函数.
__代码__
self.graph._公司字段_ = {
"公司名称":
代码.
公司名称,
"源文件名":
代码.co_filename,
"源第一行号":
代码.
源第一行号,
}
当我们遇到一个不是参数的 Tensor 值时，我们会检查它
# 是模型上的其他属性。构建一个将张量值映射到合格名称的字典以提高效率。这用于下游的 create_arg 中
# 这用于在 create_arg 中创建参数
# 这用于在 create_arg 中创建参数
self.张量属性:
字典[
联合[
火炬.
张量,
脚本对象,
假脚本对象
]
字符串
] = {}
def 收集张量属性(m:
火炬.
神经网络.
模块,
前缀原子:
列表[
字符串
)]
为 k, v
在 m.
字典.
项目():
如果 isinstance(v, (
火炬.
张量,
脚本对象,
假脚本对象)):
self.张量属性[v] =
“。”.
连接(
前缀原子 + [k])
为 k, v
在 m.
命名子项():
收集张量属性(v,
前缀原子 + [k])
收集张量属性(self.
根,
[]
断言 isinstance(
函数,
函数类型)
全局变量 =
函数.
__全局__
# 在修补之前运行
函数, args = self.
创建根目录的参数(
函数, isinstance(
根,
火炬.
神经网络.
模块),
concrete 参数
)
参数代理缓存:
字典[
字符串,
代理
] = {} 减少 get_attr 调用次数
方法参数的派发调用不记录，除非它被直接使用。
因此，当 __getattr__ 请求参数时，我们需要插入一个代理。
@functools.包装(_orig_module_getattr)
def module_getattr_wrapper(模块,
属性):
属性值 = _orig_module_getattr(
模块,
属性)
返回 self.getattr(
属性,
属性值,
参数代理缓存)
@functools.包装(
原模块调用)
def 模块调用包装器(
模块, *
参数, **kwargs):
def 前向(*
参数, **kwargs):
返回
原模块调用(
模块, *
参数, **kwargs)
自动换行检查(
补丁,
# 类型：忽略[has-type]
getattr(getattr(模块,
前进,
模块),
全局变量, {}),
self._自动包装函数 ID,
)
返回 self.
调用模块(
模块,
前向,
参数, kwargs)
与
新增补丁器()
是
补丁器:
允许重复补丁以支持嵌套调用的场景
补丁器.
补丁方法(
火炬.
神经网络.
模块,
"__getattr__",
模块 getattr 包装器,
去重=
错误,
)
修补器.
修补方法(
火炬.
神经网络.
模块,
__call__,
模块调用包装器,
去重=
错误,
)
_包装函数补丁(
补丁程序)
自动包装检查(
补丁程序,
函数全局变量, self.
_自动包装函数 ID)
为
模块
在 self._autowrap_search:
自动包装检查(
热补丁,
模块.
字典, self.
_自动包装函数 ID
)
self.创建节点(
输出,
输出,
(self.create_arg(函数(*
参数)),),
{},
type_expr=函数.
__注解__.
获取(
返回,
无),
)
self.模块路径 =
无
除了
运行时错误
是 e:
如果 isinstance(e.
参数[0
]
字符串)
和
数据依赖的
在 e.
参数[0]:
打印(
"输入文本翻译为简体中文为：\n"
+ self.graph.Python 代码(
根模块=
self,
详细模式=
是,
).源,
文件=
系统模块.
标准错误输出,
)
提升
最后:
_is_fx_tracing_flag = old_is_fx_tracing_flag
返回 self.graph
def 深拷贝(self,
描述):
# _autowrap_search 包含无法深度复制的模块。
new_tracer = 追踪器.__new__(
追踪器)
为 k, v
在 self.
字典.
项目():
如果 k
在 {
自动换行搜索}:
新对象 =
复制.
复制(v)
else:
新对象 =
复制.
深拷贝(v,
描述)
新的追踪器.
字典[k] =
新对象
返回
新的追踪器
def 代理占位符(self,
名称, concrete_args,
签名, fn_for_analysis):
如果
具体参数 is
不
无
和
名称
在
具体参数:
计数 = 0
def 替换符(x):
非局部
计数
计数 += 1
参数 =
签名.
参数[
名称]
默认:
元组[
任何, ...] = (
() 如果
参数.
默认 is
检查.
参数.
空的
否则 (
参数.
默认,)
)
out = self.创建代理(
占位符, f"{
名称}_{
字符串(
计数)}",
默认, {}
)
如果 isinstance(x, PHBase):
如果 x !=
菲律宾:
# 在使用占位符的情况下，将属性转移
# 当占位符不是单例 PH（PH 没有属性可以转移）时。
# 代理是从占位符中创建的。
# 将任何元数据（以占位符形式放入）传输
用户设置的属性从占位符到
# 基础节点（代理由用户解包）
# 元数据应包含）。
_transfer_attrs(法语=x,
到=
输出.
节点)
返回 out
Python 3.6 及以下版本中，联合类型 int 和 bool 等同于 bool。
如果
类型(x) ==
布尔值
或者
类型(x)
在
基础类型
和
类型(x) !=
火炬.
张量:
火炬._assert(
out == x,
f"{名称}
已专门化为具有值{x}
但得到了另一个值",
)
如果...否则 x is
无:
args = (
输出,
f"{名称}
已专门化为具有值 None 但得到了另一个值",
)
self.创建代理(
调用函数,
_断言为空,
参数, {})
else:
警告.
警告(
f"无法添加断言来保证正确的输入"{
名称}
到
f"专门的功能。确保您的输入与您专门化该函数的输入相匹配的责任在于用户。"
f"您专门化函数的输入相匹配的输入。"
)
返回 x
返回
py 树.tree_map(
替换_ph,
具体参数[
名称])
如果
名称[0] == "*":
默认:
元组[
任何, ...] = ()
else:
参数 = sig.
参数[
名称]
默认 = (
# 类型：忽略[赋值]
() 如果
参数.
默认 is
检查.
参数.
空的
否则 (
参数.
默认,)
)
返回 self.
创建代理(
占位符,
名称,
默认,
{},
type_expr=fn_for_analysis.__annotations__.获取(
名称,
无),
)
# 字典：(全局字典 id, 函数名) => 要修补的全局字典，用于 wrap() API 的目的。
# 的用途。
通过全局字典的 id 和函数名来键入，以确保我们只包装给定的函数一次。
只包装一次函数。
_wrapped_fns_to_patch: 字典[
元组[int,
字符串
]
字典] = {}
要包装的类的方法列表（类类型，函数名）
这目前仅适用于未正确追踪的 Tensor.*方法
需要修复的包装方法:
列表[
元组[
类型,
字符串]] =
输入文本为空，请提供需要翻译的文本
如果 os.
环境.
获取("FX_PATCH_GETITEM") == "1":
此更改是为了追踪 BERT 中的 PositionalEmbedding 等模型：
# https://github.com/pytorch/benchmark/blob/master/torchbenchmark/models/BERT_pytorch/bert_pytorch/model/embedding/position.py
# 但在此处记录了它会导致问题：
# https://github.com/pytorch/pytorch/issues/50710
# 一旦修复，我们就可以将其设置为默认行为。
被包装的方法要修复.append((
火炬.
张量, "__getitem__"))
def _find_proxy(*objects_to_search):
""
递归搜索数据结构以查找 Proxy()并返回它，
未找到则返回 None。
"文档"
代理 =
无
def 查找代理(x):
非局部
代理
如果 isinstance(x,
代理):
代理 = x
map_aggregate(要搜索的对象,
查找代理)
返回
代理
def 创建包装函数(
原函数):
@functools.包装(
原始文件名)
def 包装(*
参数, **kwargs):
""
给定一个要调用的闭包 `orig_function`，搜索参数和关键字参数
一个代理对象。如果有，则发出一个 `call_function` 节点以保留
直接调用此叶子函数。否则，只需返回结果即可。
这个函数调用，因为这个函数没有被追踪。
"文档"
代理 = _find_proxy(
参数, kwargs)
如果
代理 is
不
无:
return_proxy = 代理.
追踪器.
创建代理(
调用函数, orig_fn,
参数, kwargs
)
返回代理.
节点.
元数据[
is_wrapped] =
真实
返回
返回代理
返回
原始函数(*
参数, **kwargs)
返回
包装
def 创建包装方法(
类,
名称):
原始文件名 = getattr(
类,
名称)
@functools.包装(orig_fn)
def 包装(*
参数, **kwargs):
""
在参数和关键字参数中搜索代理对象。如果存在一个，则
发射一个 `call_method` 节点以保留对该方法的调用
直接。否则，只需返回此函数的结果
调用，因为这个函数没有被追踪。
"文档"
代理 = _find_proxy(
参数, kwargs)
如果
代理 is
不
无:
返回
代理.
追踪器.
创建代理(
"调用方法",
名称,
参数, kwargs)
返回 orig_fn(*
参数, **kwargs)
返回
包装
类 _PatchedFn(
命名元组):
帧字典:
任何
函数名:
字符串
原始文件名:
任何
新函数:
任何
def 撤销(self):
提升
未实现异常
def 补丁(self):
提升
未实现异常
类 _PatchedFnSetItem(_PatchedFn):
def 撤销(self):
self.框架字典[self.
函数名] = self.
原函数名
def 补丁(self):
self.frame 字典[self.
函数名] = self.
新函数
类
_修复函数删除(
_修复函数):
def 撤销(self):
删除 self.
帧字典[self.
函数名]
def 补丁(self):
self.帧字典[self.
函数名] = self.
新函数
类
_修复函数设置属性(
_修复函数):
def 撤销(self):
setattr(self.frame_dict, self.函数名, self.
原函数)
def 补丁(self):
setattr(self.帧字典, self.
函数名, self.
新函数)
类
_修补器:
def __init__(self) 翻译
无:
超级().__init__()
self.修补过的:
列表[
_已修补函数] =
输入文本为空，请提供需要翻译的文本
self.访问过的:
设置[int] =
设置()
def 补丁(
self,
框架字典:
字典[
字符串,
任何
]
名称:
字符串,
新函数:
可调用,
去重:
布尔值 =
是,
):
""
在退出上下文管理器之前，将 frame_dict[name] 替换为 new_fn。
"文档"
新函数.
已修复的 fx =
去重
# 类型：忽略[已定义]
如果
名称
不
在
帧字典
和
有属性(builtins,
名称):
self.已做的补丁.append(_PatchedFnDel(frame_dict,
名称,
无,
新函数))
self.patches_made[-1].patch()
如果...否则 getattr(
frame 字典[
名称
]
"__fx 已经修补",
错误):
返回
# 已经修补，无需再次执行
else:
self.已修补的补丁.append(
修正后的函数集项(
框架字典,
名称,
框架字典[
名称
]
新函数)
)
self.修补过的[-1].
补丁()
def 补丁方法(
self, 类:
类型,
名称:
字符串,
新函数:
可调用,
去重:
布尔值 =
真实
):
""
在退出上下文管理器之前，将 object_or_dict.name 替换为 new_fn。
"文档"
新函数.__fx_already_patched =
去重
# 类型：忽略[已定义]
orig_fn = getattr(类,
名称)
如果 getattr(
原始函数名,
"__fx 已经修复，无需再次操作",
错误):
返回
# 已经修复，无需再次执行
self.已制作补丁.append(
_修正 FnSetAttr(
类,
名称,
原函数,
新函数))
self.修改过的补丁[-1].
补丁()
def 一游(self,
事物:
任何):
在第一次调用 with 事物时返回 True，否则返回 false
索引 = id(
事物)
如果
索引
在 self.
访问过:
返回
假
self.访问过.
添加(
索引)
返回
真实
def 撤销所有补丁(self):
""
删除所有存储的已修复补丁。它不会修改 patches_made。
"文档"
为
补丁
在 self.
修补过的:
补丁.
撤销()
返回 self.
修补过的
def 重新应用所有补丁(self):
""
补丁所有存储的补丁。它不会修改 patches_made。
"文档"
为
补丁
在 self.patches_made:
补丁.
补丁()
返回 self.
制作的补丁
def __进入__(self):
返回 self
def __退出__(self,
异常类型,
异常值,
异常堆栈):
""
撤销通过 self.patch()和 self.patch_method()所做的所有更改
"文档"
当 self.
修补过的:
反向取消修补以正确处理重复项
self.修补过的.
流行().
撤销()
self.访问过.
清晰()
CURRENT_PATCHER: 可选[_Patcher] =
无
@contextlib.contextmanager
def _new_patcher():
全局
当前补丁程序
早期补丁程序 =
当前补丁程序
尝试:
当前补丁程序 = _Patcher()
产生 CURRENT_PATCHER
最后:
# 使用当前修补器清除所有修补内容。
断言 CURRENT_PATCHER is
不
无
当前补丁器.
撤销所有补丁()
当前补丁器 =
早期补丁器
@contextlib.contextmanager
def 可能撤销所有补丁():
当前修补者 = CURRENT_PATCHER
已做的补丁 =
无
已移除补丁 =
无
尝试:
如果
当前修补者 is
不
无:
已移除补丁 =
当前修补者.
撤销所有补丁()
产生
最后:
如果
当前修补者 is
不
无:
已制作补丁 =
当前修补者.
重新应用所有补丁()
断言 (
已制作补丁 ==
已移除补丁
), "CURRENT_PATCHER 在执行 revert_all_patches 期间已更改"
def _包装函数(
修补器: _Patcher):
""
遍历 ``_wrapped_fn_patch_table``，并对每个帧对象进行包装
列出在`_create_wrapped_func`包装器中的全局函数。
"文档"
为 (_,
名称),
frame 字典
在
需要修补的包装函数.
复制().
项目():
如果
名称
不
在
frame 字典
和
有属性(builtins,
名称):
原始函数 = getattr(builtins,
名称)
else:
原函数 =
框架字典[
名称]
修补器.
修补(
frame 字典,
名称,
_创建包装函数(
原函数))
为
类,
名称
在
_要修补的包装方法:
修补器.
修补方法(
类,
名称,
_创建包装方法(
类,
名称))
def _自动包装检查(
修补器:
_修补器,
帧字典:
字典[
字符串,
任何
]
函数 ID:
设置[int]
):
""
一些方法，如 `math.sqrt` 非常常见，我们希望看到它们时自动将其包裹起来。
此方法在作用域中搜索它们，并在找到时进行修补。
"文档"
如果
修补器.visit_once(frame_dict):
为
名称,
值
在 frame_dict.
项目():
如果 (
不
名称.
以...开头(
“_”)
和
可调用(
值)
和 id(
值)
在 function_ids
):
patcher.补丁(
框架字典,
名称,
创建包装函数(
值))
[文档]@compatibility(
兼容旧版本=
是)
def 包裹(
函数或名称:
联合[
字符串,
可调用
)]
""
此函数可以在模块级作用域中调用，以将 fn_or_name 注册为“叶子函数”。
“叶子函数”将被保留为 FX 跟踪中的 CallFunction 节点，而不是通过它进行跟踪：
通过::
# foo/bar/baz.py
def my_custom_function(x, y):
return x * x + y * y
torch.fx.wrap("my_custom_function")
def fn_to_be_traced(x, y):
当进行符号跟踪时，下面的对 my_custom_function 的调用将被插入到图中，而不是跟踪它。
该函数也可以等价地用作装饰器：
返回 my_custom_function(x, y)
此函数也可以等价地用作装饰器
# foo/bar/baz.py
@torch.fx.wrap
def my_custom_function(x, y):
return x * x + y * y
一个包装过的函数可以被视为一个“叶子函数”，类似于该概念
叶模块，即它们是留在 FX 跟踪中的函数调用
而不是追踪。
参数：
fn_or_name (Union[str, Callable]): 要插入的全局函数或名称的函数或名称
图当它被调用时
"文档"
如果
不
可调用(
函数或名称)
和
不 isinstance(
函数名或名称,
字符串):
提升
运行时错误(
不支持全局函数的此类型！必须为可调用或
字符串名称
)
如果
可调用(
函数名或名称):
断言
不 isinstance(
函数或名称,
字符串)
为了让 mypy 高兴
函数名 =
函数或名称.__name__
else:
断言 isinstance(
函数或名称,
字符串
), "fn_or_name 必须是一个全局函数或字符串名称"
函数名 = fn_or_name
currentframe = 检查.currentframe()
断言 currentframe is
不
无
f = currentframe.f_back
断言 f is
不
无
如果 f.
代码文件.
公司名称 !=
《模块》:
提升
不支持的操作异常(
"wrap 必须在模块的最顶层调用")
# 考虑通过 _autowrap_function_ids / _autowrap_search 实现此 Callable 版本
# 语义略有不同，但将支持 `from x import wrapped_function`
_wrapped_fns_to_patch[id(f.
全局变量),
函数名)] = f.
全局变量
返回
函数名或名称
[文档]@compatibility(
兼容旧版本=
是)
def 符号追踪(
根:
联合[
火炬.
神经网络.
模块,
可调用[...,
任何]],
concrete_args: 可选[
字典[
字符串,
任何]] =
无,
) 翻译 GraphModule:
""
符号追踪 API
给定一个 `nn.Module` 或函数实例 `root`，此函数将返回一个 `GraphModule`
通过在 `root` 中追踪操作所构建的。
``concrete_args`` 允许您部分特化您的函数，无论是为了去除控制流还是数据结构。
例如：
def f(a, b):
if b == True:
return a
否则:
返回 a 的两倍
FX 通常无法追踪到这一点，因为存在控制
流程。然而，我们可以使用 `concrete_args` 来专门针对值的类型进行特殊化。
`b` 跟踪此::
f = fx 符号跟踪(f, concrete_args={"b": False})
assert f(3, False) == 6
注意，尽管您仍然可以传入不同的 `b` 值，但它们将被忽略。
我们还可以使用 `concrete_args` 来消除函数中的数据结构处理。这将使用 pytrees 来展平您的输入。为了避免
我们可以使用 `concrete_args` 来消除函数中的数据结构处理。这将使用 pytrees 来展平您的输入。为了避免
过度专业化，对于不应该使用 `fx.PH` 的值进行传递
专业化。例如：
def f(x):
out = 0
for v in x.values():
out += v
return out
f = fx.symbolic_trace(f, concrete_args={"x": {"a": fx.PH, "b": fx.PH, "c": fx.PH}})
assert f({"a": 1, "b": 2, "c": 4}) == 7
参数：
root (Union[torch.nn.Module, Callable]): 要追踪和转换的模块或函数
into a Graph representation.
concrete_args (Optional[Dict[str, any]]): 部分特化的输入
返回：
GraphModule：由“root”记录的操作创建的模块。
"文档"
追踪器 =
追踪器()
图 =
追踪器.
跟踪(
根, concrete_args)
名称 = (
根.
类.__name__
如果 isinstance(
根,
火炬.
神经网络.
模块)
否则
根.__name__
)
返回 _make_graph_module(
追踪器.
根, graph,
名称)
@wrap
def _assert_is_none(值,
信息):
断言
值 is
无, msg
