• 文档 >
  • 模块代码 >
  • torch >
  • torch.fx 节点
快捷键

torch.fx.node 的源代码

节点代表我们运算符图中的值定义。
导入 内置函数
导入 检查
导入 记录日志
导入 操作符
导入 类型
from collections.abc 导入 映射, 序列
from 打字 导入 任何, 可调用, 可选, 类型检查, 类型变量, 联合

导入 火炬
from torch._C 导入 _fx_map_aggregate, _fx_map_arg, _NodeBase
from torch.fx.operator_schemas 导入 (
    ArgsKwargsPair,
    normalize_function,
    normalize_module,
)

from .._ops 导入 ops  操作
from _兼容性 导入 兼容性


如果 类型检查:
    from .graph 导入 

全部 = ["Node", map_arg, 地图聚合, 有副作用]

日志 = 记录.获取日志记录器(__name__)

基础参数类型 = 联合[
    字符串,
    int,
    float,
    布尔,
    复杂,
    火炬.数据类型,
    火炬.张量,
    火炬.设备,
    火炬.内存格式,
    火炬.布局,
    火炬.操作符.操作符重载,
    火炬.SymInt,
    火炬.符号布尔值,
    火炬.SymFloat,
]
基础类型 = 基础参数类型.__args__  # 类型:忽略[已定义]

目标 = 联合[可调用[..., 任何] 字符串]

参数 = 可选[
    联合[
        元组[论点, ...]
        序列[论点]
        映射[字符串, 论点]
        切片,  # Slice[Argument, Argument, Argument],但 slice 在 typing 中不是一个模板类型
        范围,
        "Node",
        基础参数类型,
    ]
]
论证 T = 类型变量("论证 T", 绑定=参数)

_法律操作 = 字典.fromkeys(
    [
        占位符,
        "调用方法",
        调用模块,
        调用函数,
        获取属性,
        输出,
        "root",
    ]
)

# 动态无法追踪全局 set[Callable].__contains__。
# 请参阅 https://github.com/pytorch/pytorch/issues/145761。由于我们只有
# 几个操作,因此切换到可调用对象的列表。
需要保留的副作用需要在预分发前处理: 列表[可调用] = [
    火炬._C._设置梯度启用,
    火炬.转换._自动播放入口,
    火炬.转换._自动播出版本退出,
]

# TODO:要么将此重构为 2 个函数,1 个用于功能图,1 个用于所有图,
# 或者添加逻辑以正确标记所有原地操作为副作用。
_副作用函数: 设置[可调用] = {
    火炬._assert,
    火炬.断言异步,
    操作符.aten.断言异步.信息,
    操作符.aten._assert_scalar.默认,
    操作符.aten._assert_tensor_metadata.默认,
    操作符.aten.对称约束范围.默认,
    操作符.aten.sym_constrain_range_for_size.默认,
    操作符.分析器.记录函数进入,
    操作符.分析器._record_function_enter_new,
    操作符.分析器.记录函数退出,
    操作符.电感器.累加梯度.默认,
    操作符.设置项,
} | 设置(需要保留的副作用_预分发)

如果 有属性(操作符.电感器, "调整存储字节数_"):
    副作用函数_.添加(操作符.电感器.调整存储字节数.默认)


@compatibility(兼容旧版本=错误)
def 有副作用(函数: 可调用) 翻译 可调用:
    _有副作用的函数.添加(函数)
    返回 fn


# 在主分支上已修复,1.5 版本的 WAR
def _查找方法所属模块(原始方法: 可调用[..., 任何]) 翻译 字符串:
    名称 = 原始方法.__name__
    模块 = 原始方法.__module__
    如果 模块 is 不是 :
        返回 模块
     猜测  [火炬, 火炬.神经网络.功能性]
        如果 getattr(猜测, 名称, ) is 原始方法:
            返回 猜测.__name__
    提升 运行时错误(f"无法找到模块"{原始方法}")


从 CPython 类型模块借用
https://github.com/python/cpython/blob/f90dc36c15d7fee0efaf6d39e97be0bdf2683e93/Lib/typing.py#L156
def 类型表示(对象: 对象) 翻译 字符串:
    返回对象的 repr(),针对类型进行特殊处理(内部辅助函数)。
如果 obj 是类型,则返回比默认版本更短的版本。
type.__repr__,基于模块和限定名称,它是
通常足以唯一标识一个类型。对于所有事物
否则,我们回退到 repr(obj)。
""
    # 扩展:如果我们不忽略 GenericAlias,那么 `list[int]` 将会打印
    列表
    如果 isinstance(对象, 类型)  不是 isinstance(对象, 类型.通用别名):
        如果 对象.__module__ == 内建函数:
            返回 对象.__qualname__
        返回 f"{对象.__模块__}.{对象.__qualname__}"
    如果 对象 is ...:
        返回 "..."
    如果 isinstance(对象, 类型.函数类型):
        返回 对象.__name__
    返回 表示(对象)


def 获取合格名称(函数: 可调用[..., 任何]) 翻译 字符串:
    getattr 等类似函数仅出现在内置模块中
    如果 getattr(builtins, 函数.__name__, ) is 函数:
        返回 函数.__name__
    torch.Tensor.{} 函数
    如果 isinstance(
        函数, (类型.方法描述符类型, 类型.包装描述符类型)
    )  函数 is getattr(火炬.张量, 函数.__name__, ):
        返回 ftorch.Tensor.{函数.__name__}"
    名称 = 函数.__name__
    如果 名称 == "<lambda>":
        对于 lambda,尝试获取其在模块中的定义名称
        尝试:
            名称 = 检查.获取源(函数).分割(等于)0].strip()
        除了 异常  e:
            提升 运行时错误(无法表示 lambda) from e
    模块 = _查找方法所属模块(函数)
    模块 = 模块.替换(
        "torch._ops", "torch.ops"
    )  # WAR 对于 torch.ops 分配模块时的 bug
    # 修复 segment_reduce 不匹配问题
    如果 模块 == torch  名称 == "segment_reduce":
        名称 = _ + 名称
    返回 f"{模块}.{名称}"


def 格式化参数(arg: 对象, 最大列表长度: 浮点数 = float("无穷大")) 翻译 字符串:
    如果 有属性(arg, _自定义_fx_表示函数):
        返回 arg._自定义_fx_表示函数()
    如果...否则 isinstance(arg, 列表):
        项目 = “,”.连接(
            _格式化参数(a)  索引, a  列举(arg) 如果 索引 < 最大列表长度
        )
        可能的长度 = (
            请提供需要翻译的文本 如果 长度(arg) < 最大列表长度 + 1 否则 f", ...[总长度={长度(arg)}]"
        )
        返回 f"[{项目}{可能的长度}]"
    如果...否则 isinstance(arg, 元组):
        项目 = “,”.连接(
            _格式化参数(a)  索引, a  列举(arg) 如果 索引 < 最大列表长度
        )
        可能的长度 = (
            请提供需要翻译的文本 如果 长度(arg) < 最大列表长度 + 1 否则 f", ...[总长度={长度(arg)}]"
        )
        可能的逗号 = 逗号 如果 长度(arg) == 1 否则 请提供需要翻译的文本
        返回 f“(”{项目}{可能的逗号}{可能的长度})"
    如果...否则 isinstance(arg, 字典):
        项目字符串 = “,”.连接(f"{k}: {_格式化参数(v)}"  k, v  arg.项目())
        返回 f"{{{items_str}}}"

    如果 isinstance(arg, 节点):
        返回 百分号 + 字符串(arg)
    else:
        返回 字符串(arg)


[文档]@compatibility(兼容旧版本=) 节点(_NodeBase): "" ``节点``是表示单个操作的数据结构。在大多数情况下,节点代表对各种实体的调用位置, 对于大部分来说,节点代表对 ``图`` 的操作。 例如操作符、方法和模块(一些例外包括节点等) 指定函数的输入和输出。每个 ``节点`` 都指定了一个函数 通过其 ``op`` 属性。每个 ``op`` 值的 ``Node`` 语义如下: 占位符代表一个函数输入。`name` 属性指定这个值将采用的名称。 ``target``是参数的名称。``args``包含以下两种情况之一:1)没有内容,或 2)一个单独的参数 表示函数输入的默认参数。``kwargs``不予考虑。占位符对应于 函数参数(例如``x``)在图形输出中的参数。 - ``get_attr``从模块层次结构中检索一个参数。``name``是结果的名称 fetch 被分配到。 ``target`` 是参数在模块层次结构中的全限定名称。 ``args`` 和 ``kwargs`` 是无关紧要的。 - ``call_function`` 将一个自由函数应用于一些值。 ``name`` 同样是分配给值的名称。 到。 ``target`` 是要应用的功能。 ``args`` 和 ``kwargs`` 代表传递给函数的参数。 遵循 Python 调用约定 - ``call_module`` 将模块层次结构中的 ``forward()`` 方法应用于给定的参数。 ``name`` 是 如前所述。 ``target`` 是要调用的模块的完全限定名称。 ``args`` 和 ``kwargs`` 代表调用模块时要传递的参数,*不包括 self 参数*。 ``call_method`` 调用一个值上的方法。``name`` 与之相似。``target`` 是应用于 ``self`` 参数的字符串方法名 ``args`` 和 ``kwargs`` 表示调用模块时传递的参数,包括 ``self`` 参数 *包括 ``self`` 参数* ``output`` 包含被跟踪函数的输出,位于其 ``args[0]`` 属性中。这对应于 "return" 语句 在图打印输出中。 "" _参数: 元组[论点, ...] _关键字参数: 字典[字符串, 论点] graph: "图" 创建值的唯一名称 名称: 字符串 操作类型 = 占位符|调用方法|调用模块|调用函数|获取属性 操作: 字符串 方法/模块/函数/属性的名称 被调用的方法/模块/函数/属性,例如 add、layer1 或 torch.add 目标: "目标" # 所有 `Node`-值输入。键是节点,值是不关心的。 # 公共 API 为 `all_input_nodes`,此私有属性 # 不应直接访问。 输入节点: 字典["Node", ] # 所有使用此节点产生的值的节点 # 注意一个用户可能对应多个使用,例如节点“x + x” # 将在这里出现一次,但代表两个使用。 # 是一个字典,用作“有序集合”。键是重要的,值不重要 用户们: 字典["Node", ] 表示此节点输出值的表达式类型。 这应包含与函数输入/输出类型注解相同的类型对象类。 这将作为占位符节点的类型注解使用。 # 对于占位符节点,此值将用于类型注解。 生成函数参数。 返回节点,此值将用于类型注解。 生成函数返回类型。(注意这是一个特殊情况。``return`` 不会产生值,它更像是一种标记。因此,此值 返回节点中 args[0]的类型描述 类型: 可选[任何] 排序键: 任何 如果设置,使用此函数打印此节点 _repr_fn: 可选[可调用[["Node"] 字符串]] # 需要存储的元数据传递,需要进行转换 # 此元数据在节点副本间保留 元数据: 字典[字符串, 任何] @compatibility(兼容旧版本=) def __init__( self, graph: , 名称: 字符串, 操作: 字符串, 目标: 目标, 参数: 元组[论点, ...] kwargs: 字典[字符串, 论点] 返回类型: 可选[任何] = , ) 翻译 : "" 实例化一个 ``Node`` 对象。注意:通常情况下,您希望使用 图 API,例如 ``Graph.call_module``、``Graph.call_method`` 等,而不是直接实例化一个 ``Node``。 而是使用 ``Node``。 参数: graph (Graph):该 ``Node`` 应该属于的 ``Graph``。 name (str):该 ``Node`` 输出的输出应分配的名称。 op (str): 该 ``Node`` 的操作码。可以是以下之一 'placeholder', 'call_method', 'call_module', 'call_function', 'get_attr', 'output' target ('目标'): 该操作应调用的目标。参见更广泛的 ``Node``文档字符串,详情请参阅。 args (元组['参数']):要传递给 ``target`` 的参数 kwargs (字典[str, '参数']):要传递给 ``target`` 的关键字参数 return_type (可选[任何类型]):表示 Python 类型表达式的 Python 类型 该节点输出类型。此字段可用于 对生成代码中的值或其它分析类型的注释 的。 "" 如果 操作符 == 调用函数: 如果 不是 可调用(目标): 提升 ValueError( f"节点[图= "{graph}name = '{名称}'] 目标{目标}类型{火炬.类型名(目标)} " 但期望一个 Callable ) else: 断言 操作符 _法律操作 如果 不是 isinstance(目标, 字符串): 提升 ValueError( f节点 [图 ={graph}, 名称 ={名称}'] 目标{目标}类型{火炬.类型名(目标)} " 但期望一个字符串 ) 超级().__init__(graph, 名称, 操作, 目标, 返回类型) self._更新参数和关键字(参数, kwargs) def __getstate__(self) 翻译 字典[字符串, 任何] 返回 { **self.字典, "图": self.graph, 名称: self.名称, "操作": self.操作, "目标": self.目标, 类型: self.目标, _sort_key: self._sort_key, _args: self._args, _kwargs: self._kwargs, _erased: self._erased, "_前一个": self._prev, 下一个: self._next, 输入节点: self.输入节点, 用户: self.用户们, _repr_fn: self._repr_fn, 元数据: self.元数据, } def __setstate__(self, 状态: 字典[字符串, 任何]) 翻译 : k, v 状态.项目(): setattr(self, k, v) @property def 下一(self) 翻译 "Node": "" 返回节点链表中的下一个“Node”。 返回: 下一个节点在节点链表中。 "" 返回 self.下一行文本: _next 翻译为简体中文: _next @property def 上一页(self) 翻译 "Node": "" 返回节点链表中的前一个 ``Node``。 返回: 节点链表中的前一个 ``Node``。 "" 返回 self._prev
[文档] @兼容性(向后兼容=True) def prepend(self, x: "节点") -> None: """ 在图中的节点列表中在此节点之前插入 x。示例:: Before: p -> self bx -> x -> ax After: p -> x -> self bx -> ax Args: x (节点): 要放置在此节点之前。必须是同一图中的成员。 """ 断言 self.graph == x.graph, "尝试将节点移动到不同的图" if self == x: log.debug( "尝试将自己节点添加到自身之前。这种行为对图没有影响。" ) 返回 x._remove_from_list() p = self._prev p._next, x._prev = x, p x._next, self._prev = self, x # 计算 x._sort_key psk = x._prev._sort_key nsk = x._next._sort_key 如果 psk 的长度大于 nsk 的长度: idx: int *prefix, idx = psk[: len(nsk) + 1] x._sort_key = (*prefix, idx + 1) elif len(psk) < len(nsk): *prefix, idx = nsk[: len(psk) + 1] x._sort_key = (*prefix, idx - 1) else: # same length, increase length by 1 x._sort_key = (*psk, 0)
def __gt__(self, 其他
: "Node") 翻译 布尔: 返回 self.排序键 > 其他.排序键 def __lt__(self, 其他: "Node") 翻译 布尔: 返回 self.排序键 < 其他.排序键 def __ge__(self, 其他: "Node") 翻译 布尔: 返回 self > 其他 或者 self == 其他 def __le__(self, 其他: "Node") 翻译 布尔: 返回 self < 其他 或者 self == 其他
[文档] @兼容性(向后兼容=True) def append(self, x: "节点") -> None: """ 在图中节点列表中在此节点后插入 ``x``。 等价于 `self.next.prepend(x)` 参数: x (节点): 要放在此节点之后的节点。必须是同一图中的成员。 """ self._next.prepend(x)
def _remove_from_list(self) 翻译
: p, n = self._prev, self._next p.下一个, n.上一个 = n, p @property def 参数(self) 翻译 元组[参数, ...] "" 该 ``Node`` 的参数元组。参数的解释取决于节点的操作码。请参阅 :class:`Node` 的文档字符串以获取更多信息。 依赖于节点的操作码。请参阅 :class:`Node` 的文档字符串以获取更多信息。 有关信息。 允许对此属性进行赋值。所有关于使用者和用户的使用记录。 自动更新于分配。 "" 返回 self._args @args.setter def 参数(self, a: 元组[参数, ...]) 翻译 : "" 设置此节点的参数元组。参数的解释取决于节点的操作码。请参阅 `fx.Graph` 的文档字符串以获取更多信息。 依赖于节点的操作码。请参阅 `fx.Graph` 的文档字符串以获取更多信息。 信息。 "" 不要直接调用 `_update_args_kwargs`。正确的方式是... 通过直接赋值设置 `args`,即 `node.args = new_args` self._update_args_kwargs(a, self._kwargs) @property def kwargs(self) 翻译 字典[字符串, 参数] "" 该 ``Node`` 的关键字参数字典。参数的解释为 依赖于节点的操作码。请参阅:class:`Node` 的文档字符串以获取更多信息。 有关信息。 允许对此属性进行赋值。所有使用和用户的会计信息在赋值时自动更新。 会计信息在赋值时自动更新。 "" 返回 self._kwargs @kwargs.setter def kwargs(self, k: 字典[字符串, 参数]) 翻译 : "" 设置 kwargs 字典到该节点。参数的解释取决于节点的操作码。请参阅`fx.Graph`的文档字符串以获取更多信息。 有关详细信息,请参阅`fx.Graph`文档字符串。 请勿直接调用`_update_args_kwargs`。正确的方式是... "" # 不要直接调用`_update_args_kwargs`。正确的方式是... 通过直接赋值设置 `args`,即 `node.kwargs = new_kwargs` self._更新 args 和 kwargs(self._args, k) @property def 所有输入节点(self) 翻译 列表["Node"] "" 返回所有是该节点输入的节点。这相当于 遍历 `args` 和 `kwargs` 并仅收集值为节点的值 的节点。 返回: 出现在该节点的 `args` 和 `kwargs` 中的 `Nodes` 列表 ``节点`,按此顺序。 "" 返回 列表(self.输入节点.())
[文档] @兼容性(向后兼容=True) def update_arg(self, idx: int, arg: Argument) -> None: """ 更新现有的位置参数以包含新值 ``arg``。调用后,``self.args[idx] == arg``。 参数: idx (int): 要更新的 ``self.args`` 元素的索引 arg(参数):要将新参数值写入 `args` """ args = list(self.args) args[idx] = arg self.args = tuple(args)
[文档] @兼容性(向后兼容=True) def insert_arg(self, idx: int, arg: Argument) -> None: """ 在给定索引的位置向参数列表中插入一个参数。 Args: idx (int): 要插入到 ``self.args`` 中的元素的索引。 arg (Argument): 要插入到 ``args`` 中的新参数值。 """ assert ( 0 <= idx <= len(self.args) ), "插入参数索引必须在 0 和 len(self.args)之间" args_left = self.args[:idx] args_right = self.args[idx:] self._args = args_left + (arg,) + args_right _new_input_nodes: dict[Node, None] = {} _fx_map_arg(arg, _new_input_nodes.setdefault) for new_use in _new_input_nodes.keys(): if new_use not in self._input_nodes: self._input_nodes.setdefault(new_use) new_use.users.setdefault(self)
[文档] @兼容性(向后兼容=True) def update_kwarg(self, key: str, arg: Argument) -> None: """ 更新现有的关键字参数以包含新值 ``arg``. 调用后,``self.kwargs[key] == arg``。 Args: key (str): 要更新的元素在 ``self.kwargs`` 中的键 arg (Argument): 要写入 ``kwargs`` 的新参数值 """ self.kwargs = {**self.kwargs, key: arg}
@property def 堆栈跟踪
(self) 翻译 可选[字符串] "" 返回在跟踪期间记录的 Python 调用栈,如果有。 当使用 fx.Tracer 跟踪时,此属性通常会被填充。 `Tracer.create_proxy`。为了在调试过程中记录堆栈跟踪, 将 `record_stack_traces = True` 设置在 `Tracer` 实例上。 当使用 dynamo 追踪时,此属性将默认填充 创建输出图代理 堆栈跟踪的内部最深层将在字符串末尾。 "" 返回 self.元数据.获取(堆栈跟踪, ) @stack_trace.setter def 堆栈跟踪(self, 跟踪: 可选[字符串]) 翻译 : self.元数据[堆栈跟踪] = 跟踪 def __repr__(self) 翻译 字符串: 如果 self._repr_fn: 返回 self._repr_fn(self) 返回 self.名称 @staticmethod def _美化打印目标(目标: 对象) 翻译 字符串: "" 使打印输出更易于用户使用。 1) 内置函数将打印为 `builtins.xyz` 2) 运算符将打印为 `operator.xyz` 3) 其他可调用对象将以限定名称打印,例如 torch.add "" 如果 isinstance(目标, 字符串): 返回 目标 如果 有属性(目标, "__module__"): 名称 = getattr(目标, "__名称__", ) 如果 名称 is : 为了防御性考虑,如果我们没有 `__name__`,则获取 # qualname。不清楚这会发生在 `operator` 的任何成员上 # 或 `builtins`。这个回退路径并不太好,因为例如 # `operator` 中的事物它们的 `__module__` 是 `_operator`。 # TODO:这个有问题:_get_qualified_name 调用了 `__name__` 返回 获取合格名称(目标) # type: ignore[arg-type] 如果 目标.__module__ == 内建函数: 返回 f内置函数{名称}" 如果...否则 目标.__module__ == "_操作符": 返回 f操作符。{名称}" 返回 获取合格名称(目标) # type: ignore[arg-type]
[文档] @compatibility(兼容旧版本=) def 格式节点( self, 占位符名称: 可选[列表[字符串]] = , 可能返回类型名: 可选[列表[字符串]] = , ) 翻译 可选[字符串] "" 返回 ``self`` 的描述性字符串表示。 此方法可以不带参数使用,作为调试 工具。 此函数也用于内部在 ``__str__`` 方法中 ``placeholder_names`` 和 ``maybe_return_typename`` 中的字符串共同构成了 自动生成的 ``forward`` 函数的签名 在此 Graph 的周围环境中 GraphModule. ``placeholder_names`` 和 ``maybe_return_typename`` 不应使用。 否则不应使用。 参数: placeholder_names:一个将存储格式化字符串的列表 表示生成的占位符。 ``forward``函数。仅内部使用。 maybe_return_typename:一个存储单个元素的列表 表示生成的``forward``函数输出的格式化字符串 generated ``forward`` function. Internal use only. 返回: 如果 1)我们在``format_node``作为内部辅助函数 在``Graph``的``__str__``方法中,并且 2)``self`` 是一个占位符节点,则返回``None``。否则, 返回一个描述性的字符串表示形式。 当前节点 "" 如果 self.操作符 == 占位符: 断言 isinstance(self.目标, 字符串) arg_str = self.目标 arg_str += arg_str + f":"{类型表示(self.类型)}" 如果 self.类型 否则 请提供需要翻译的文本 如果 占位符名称: 占位符名称.append(arg_str) 返回 maybe_typename = f"{类型表示(self.类型)} " 如果 self.类型 否则 请提供需要翻译的文本 默认值 = "(默认=" + 字符串(self.参数[0]) + )" 如果 self.args 否则 请提供需要翻译的文本 返回 f"%"{self.名称} : {maybe_typename}[num_users=]{长度(self.用户们)}] = {self.操作}[target=]{self.目标}]{default_val}" 如果...否则 self.操作符 == 获取属性: 可能类型名 = ( f"{类型表示(self.类型)} " 如果 self.类型 is 不是 否则 请提供需要翻译的文本 ) 返回 ( f百分比{self.名称} : {maybe_typename}[num_users={长度(self.用户们)}] = " f"{self.操作}[目标={self._美化打印目标(self.目标)}]" ) 如果...否则 self.操作符 == 输出: 如果 self.类型 可能返回类型名: 可能返回类型名[0] = f" -> "{类型表示(self.类型)}" 返回 f"返回"{self.参数[0]}" else: 可能类型名 = ( f"{类型表示(self.类型)} " 如果 self.类型 is 不是 否则 请提供需要翻译的文本 ) 返回 ( f百分比{self.名称} : {maybe_typename}[num_users={长度(self.用户们)}] = " f"{self.操作}[目标={self._美化打印目标(self.目标)}] (" f"参数 ={格式化参数(self.参数)}, 参数 ={格式化参数(self.kwargs)})" )
[文档] @compatibility(兼容旧版本=) def 替换所有引用( self, 替换为: "Node", 删除用户回调: 可调用[["Node"] 布尔] = lambda 用户: , *, 传播元数据: 布尔值 = 错误, ) 翻译 列表["Node"] "" 将 Graph 中所有对`self`的使用替换为 Node `replace_with`。 参数: replace_with (Node):替换所有`self`使用的节点。 delete_user_cb (Callable):用于确定是否应删除 self 节点给定用户的回调。 是否应删除 self 节点给定用户的回调。 propagate_meta (bool): 是否复制所有属性 在原始节点上的 .meta 字段替换到目标节点上。 为确保安全,仅在替换节点时有效 该文件尚未存在一个现有的 .meta 字段。 返回: 已修改的节点列表。 "" 如果 传播元数据: 断言 长度(替换为.元数据) == 0, ( "调用 node.replace_all_uses_with(替换为, propagate_meta=True), " "但替换为已经存在 .meta 键" ) k, v self.元数据.项目(): 替换为.元数据[k] = v 待处理 = 列表(self.用户们) 跳过 = 输入文本为空,请提供需要翻译的文本 m = self.graph.所属模块 使用节点 处理中: 如果 不是 删除用户回调(使用节点): 跳过.append(使用节点) continue def 可能替换节点(n: 节点) 翻译 节点: 如果 n == self: 返回 替换为 else: 返回 n 如果 getattr(m, _替换钩子, ): 替换钩子 m.替换钩子: 替换钩子(旧的=self, =替换为.名称, 用户=使用节点) new_args = _fx_map 参数(使用节点.参数, 可能替换节点) 新的参数 = _fx_map_arg(使用节点.kwargs, 可能替换节点) 断言 isinstance(新参数, 元组) 断言 isinstance(新的参数, 字典) 使用节点._更新参数和关键字段(新参数, new_kwargs) 断言 长度(self.用户们) - 长度(跳过) == 0 返回 [n n 待处理 如果 n 不是 跳过]
[文档] @兼容性(向后不兼容=False) def is_impure(self) -> bool: """ 返回此操作是否为不纯操作,即其操作是否为占位符或 输出,或者如果调用函数或模块是纯的。 返回: 布尔值:如果操作是不纯的。 """ if self.op in {"placeholder", "output"}: return True if self.op == "call_function": schema = getattr(self.target, "_schema", None) 如果 schema 不为空且 schema.is_mutable: # 不纯,因为它会修改输入 返回 True 如果 getattr(self.target, "_nondeterministic_seeded", False): # 不纯,因为它会改变随机数生成器的状态 # 返回 True # 检查是否为不纯模块。 # 检查是否为不纯模块。 if self.op == "call_module": assert ( self.graph.owning_module is not None ), "self.graph.owning_module not set for purity check" target_mod = self.graph.owning_module.get_submodule(self.target) assert ( target_mod is not None ), f"未找到预期的子模块目标 {self.target}" return getattr(target_mod, "_is_impure", False) return False
[文档] @兼容性(向后不兼容=False) def normalized_arguments( self, root: torch.nn.Module, arg_types: Optional[tuple[Any]] = None, kwarg_types: Optional[dict[str, Any]] = None, normalize_to_only_use_kwargs: bool = False, ) -> Optional[ArgsKwargsPair]: """ 返回规范化后的参数给 Python 目标。这意味着`args/kwargs`将与模块/函数的签名匹配,并仅按位置顺序返回`kwargs` `args/kwargs`将与模块/函数的签名匹配,并按位置顺序返回 `kwargs`,并仅按位置顺序返回`kwargs` 如果 `normalize_to_only_use_kwargs` 为真。 也填充默认值。不支持仅位置参数 参数或可变参数。 支持模块调用。 需要使用 `arg_types` 和 `kwarg_types` 来区分重载。 Args: root (torch.nn.Module): 要解析模块目标的模块。 arg_types (Optional[Tuple[Any]]): 参数 arg 的类型元组 kwarg_types (Optional[Dict[str, Any]]): 参数类型字典,用于 kwargs normalize_to_only_use_kwargs (bool): 是否规范化以仅使用 kwargs。 返回: 返回 NamedTuple ArgsKwargsPair,或未成功时返回 `None`。 """ 如果 self.op 等于 "call_function": 断言 self.target 是可调用的 返回 normalize_function( self.target, self.args, # type: ignore[arg-type] self.kwargs, arg_types, kwarg_types, ) elif self.op == "call_module": assert isinstance(self.target, str) return normalize_module(root, self.target, self.args, self.kwargs) # 忽略[arg-type] return None
[文档] @兼容性(is_backward_compatible=True) def replace_input_with(self, old_input: "Node", new_input: "Node") -> None: ```python # 输入文本 input_text = '"""' # 翻译函数(此处为示例,实际翻译功能需调用真实的翻译 API) def translate_to_simplified_chinese(text): # 假设的翻译结果 return text # 输出翻译结果 translated_text = translate_to_simplified_chinese(input_text) print(translated_text) ``` 遍历`self`的输入节点,并替换所有实例为 ``old_input`` 与 ``new_input``。 参数: 旧输入(节点):要替换的旧输入节点。 新输入(节点):替换 ``旧输入`` 的新输入节点。 """ def maybe_replace_node(n: 节点) -> 节点: 返回 new_input 如果 n 等于 old_input 否则返回 n m = self.graph.owning_module 如果 getattr(m, "_replace_hooks", None): for replace_hook in m._replace_hooks: replace_hook(old=old_input, new=new_input.name, user=self) new_args = _fx_map_arg(self.args, maybe_replace_node) new_kwargs = _fx_map_arg(self.kwargs, maybe_replace_node) assert isinstance(new_args, tuple) assert isinstance(new_kwargs, dict) self._update_args_kwargs(new_args, new_kwargs)
def _rename(self, 候选人
: 字符串) 翻译 : 如果 候选人 == self.名称: 返回 名称 = self.graph._graph_namespace.创建名称(候选人, ) self.名称 = 名称 self.graph._graph_namespace.重命名对象(self, 名称) def __setattr__(self, 名称: 字符串, : 任何) 翻译 : 如果 名称 == "名称" 有属性(self, 名称): m = self.graph.所属模块 如果 getattr(m, "_替换钩子", ): 断言 isinstance(, 字符串) 用户 self.用户们: 替换钩子 m.替换钩子: 替换钩子(旧的=self, =, 用户=用户) 更新 = 如果 ( 有属性(self, 名称) 有属性(self.graph, _查找节点查找表) self self.graph._查找节点查找表 ): 更新 = 真实 self.graph._查找节点表.删除(self) 对象.__setattr__(self, 名称, ) 如果 更新: self.graph._查找节点表.插入(self)
@compatibility(兼容旧版本=) def map_arg(a: 论证 T, 函数: 可调用[[节点] 参数]) 翻译 论证 T: "" 将函数递归地应用于 arg 中出现的每个节点。 arg 可能是一个列表、元组、切片或具有字符串键的字典:返回值将是 类型和结构相同。 "" 断言 可调用(函数), torch.fx.map_arg(a, fn): fn 必须是一个可调用对象。 返回 _fx_map_arg(a, 函数) @compatibility(兼容旧版本=) def map_aggregate(a: 参数 T, 函数: 可调用[[参数] 参数]) 翻译 论点 T: "" 递归地将函数应用于 arg 中出现的每个对象。 arg 可以是列表、元组、切片或具有字符串键的字典:返回值将 具有相同的类型和结构。 "" 返回 _fx_map_aggregate(a, 函数)

© 版权所有 PyTorch 贡献者。

使用 Sphinx 构建,并使用 Read the Docs 提供的主题。

文档

查看 PyTorch 的全面开发者文档

查看文档

教程

深入了解初学者和高级开发者的教程

查看教程

资源

查找开发资源,获取您的疑问解答

查看资源