# mypy: 允许未类型化定义
导入 contextlib
导入
复制
导入 itertools
导入
linecache 模块
导入
操作系统
导入
系统
导入
跟踪回溯
导入
警告
from pathlib 导入 Path
from 打字
导入
任何,
可调用,
可选,
联合
导入
火炬
导入 torch.nn
是
神经网络
导入 torch.overrides
from torch.nn.modules 下的 module 模块
导入
_addindent 函数
from torch.package 导入
导入器,
包导出器,
包导入器,
sys 导入器
from 兼容性
导入
兼容性
from .graph 导入
自定义内置函数, _is_from_torch,
_PyTree 代码生成器,
图,
Python 代码
全部 = [
简化图模块,
"减少包图模块",
"减少部署图模块",
"图模块",
]
_用户保留属性键 =
用户保留属性
# 正常执行会丢失源代码，但是我们可以使用
# linecache 模块来恢复它。
# 使用_exec_with_source 会将它添加到我们的本地缓存中
然后像 TorchScript 这样的工具将能够获取源信息。
类 _EvalCacheLoader:
def __init__(self):
self.评估缓存 = {}
self.下一个 ID = 0
def 缓存(self,
源:
字符串,
全局变量:
字典[
字符串,
任何
] co_fields=
无):
将源代码存储在私有缓存中，并在 linecache 中添加一个懒加载条目
允许通过'filename'检索源代码。
参数：
src (str): 要缓存的模块源代码
globals (dict): 模块的全局变量
返回：
为 src 生成的缓存键（以及虚拟文件名）
""
key = self._get_key()
如果 co_fields:
key += f"来自"{
字段[
文件名]}:{
字段[
第一行号]}
在{co_fields['co_name']}"
self.eval_cache[键] =
源
# 不要修改全局变量，这样这个加载器就只会被使用
用于填充行缓存，并且不与其他模块交互
可能会检查 `__loader__`
全局变量复制 =
全局变量.
复制()
全局变量复制[
"__文件__"] = key
全局变量副本[
"__名称__"] = key
全局变量副本[
"__加载器__"] = self
linecache.懒加载缓存(
键,
全局变量复制)
返回 key
# 加载协议的一部分（PEP 302）
当 linecache 尝试查找源代码时将使用此方法
def 获取源(self,
模块名称)
翻译
可选[
字符串
]
如果
模块名称
在 self.
评估缓存:
返回 self.
评估缓存[
模块名称]
返回
无
def _获取键(self):
key = f使用密钥进行评估。{self.
下一个_id}"
self.下一个 ID += 1
返回 key
_加载器 =
_评估缓存加载器()
def _带有源执行(
源:
字符串,
全局变量:
字典[
字符串,
任何
]
字段=
无):
key = 加载器.
缓存(
源,
全局变量,
字段)
执行(
编译(
源,
键,
执行),
全局变量)
def 来自源转发(
源:
字符串,
全局变量:
字典[
字符串,
任何
]
共享字段=
无):
返回
来自源的方法(
方法名=
前进,
源=
源,
全局变量=
全局变量,
共享字段=
共享字段
)
def _方法_from_src(
方法名称:
字符串,
源:
字符串,
全局变量:
字典[
字符串,
任何
]
co 字段=
无
) 翻译
可调用:
# 避免修改传入的字典
全局副本 =
全局变量.
复制()
带源代码执行(
源,
全局副本,
共享字段)
fn = 全局副本[
方法名称]
删除
全局副本[
方法名称]
返回 fn
def 格式化导入语句(
名称:
字符串,
对象:
任何,
导入器:
导入器)
翻译
字符串:
如果
名称
在
_自定义内置:
返回
_自定义内置[
名称].import_str
如果
来自 torch(
名称):
返回
导入 torch
模块名称,
属性名称 =
导入器.
获取名称(
对象)
返回 f
"从"{
模块名称}
导入{
属性名称}
作为{
名称}"
def _格式化导入块(
全局变量:
字典[
字符串,
任何
]
导入器:
导入器):
导入字符串:
设置[
字符串] = {
_格式化导入语句(
名称,
对象, importer)
为
名称,
对象
在
全局变量.
项目()
}
# 对导入进行排序，以便我们有一个稳定的导入块，允许我们
# 对图模块进行哈希处理，以获取用于缓存的连续键。
返回 "
输入文本翻译为简体中文为：\n".
连接(
排序(import_strs))
@compatibility(兼容旧版本=
是)
def 减少图模块(
主体:
字典[
任何,
任何
]
导入块:
字符串)
翻译
火炬.
神经网络.
模块:
# BC: 将属性名从 `code` 更改为 `_code` 以便于将 `code` 转换为属性并添加文档字符串
# 使 `code` 成为属性并添加文档字符串
fn_src = 主体.
获取(
_代码)
或者
主体[
代码]
前向传播 =
_从源转发(
导入块 +
函数源, {})
返回
_反序列化图模块(
前向,
主体)
@compatibility(兼容旧版本=
是)
def 减少包图模块(
导入器:
包导入器,
主体:
字典[
任何,
任何
]
生成的模块名称:
字符串
) 翻译
火炬.
神经网络.
模块:
前向传播 =
导入器.
导入模块(
生成模块名称).
前向传播
返回
_反序列化图模块(
前向,
主体)
@compatibility(兼容旧版本=
是)
def 减少部署图模块(
导入器:
包导入器,
主体:
字典[
任何,
任何
]
导入块:
字符串
) 翻译
火炬.
神经网络.
模块:
命名空间 = {}
ns["__内置__"] =
导入器.
补丁内置函数
函数源代码 =
主体.
获取(
_代码)
断言
函数源代码 is
不是
无
前向传播 =
从源处转发(
导入块 + fn_src, ns)
返回
反序列化图模块(
前向,
主体)
我们在这里创建一个虚拟类，因为 symbolic_trace 会调用 forward()
# 从类而非实例中调用的函数。此类用于
# 下面的 _deserialize_graph_module() 中。
类
_仅代码模块(
火炬.
神经网络.
模块):
def __init__(self, 主体):
超级().__init__()
self.字典 =
身体
def 反序列化图模块(
前向,
主体:
字典[
任何,
任何
]
图模块类=
无
) 翻译
火炬.
神经网络.
模块:
""
根据原始模块的字典反序列化 GraphModule，使用代码重构图。
在保存字典之前删除实际图，以确保内存中图格式的更改不会序列化。
在保存字典之前删除实际图，以确保内存中图格式的更改不会序列化。
在保存字典之前删除实际图，以确保内存中图格式的更改不会序列化。
""
尝试以向后兼容的方式检索正向源
仅代码模块.
前向传播 =
前向传播
tracer_cls = 主体.
获取(
_tracer_cls)
如果 tracer_cls is
无:
from ._symbolic_trace 导入
跟踪器
跟踪器类 =
跟踪器
图模块类名 =
主体.
获取("_graphmodule_cls_name",
"图模块")
# 这是一个针对 mypy linter 问题的解决方案，相关问题与
# 将基类作为参数传递有关 - https://github.com/python/mypy/issues/5865.
cls_tracer: 任何 = tracer_cls
类
保持模块(cls_tracer):
我们不应该追踪到任何子模块，
因为它们在原始 GraphModule 中没有被追踪
def 叶子模块(self, _:
火炬.
神经网络.
模块, __:
字符串)
翻译
布尔:
返回
真实
com = 仅代码模块(
主体)
跟踪器扩展 =
主体.
获取(
_tracer_extras, {})
图 =
保持模块().
跟踪(com, **tracer_extras)
在重构的 Graph 上手动设置 Tracer 类，以避免
引用私有本地子类 KeepModules。
graph._tracer_cls = tracer_cls
from _lazy_graph_module
导入 _make_graph_module
gm = _make_graph_module(
com, graph, 类名=
图模块类名, graph_module_cls=graph_module_cls
)
# The GraphModule constructor only retains attributes referenced by the graph.
# In this case, our goal is return a GraphModule as close to identical as the one
将内容放入包中。如果身体中存在任何其他属性，
我们应该保留它们。
为 k, v
在
主体.
项目():
如果
不是
有属性(gm, k):
setattr(gm, k, v)
返回 gm
从'from_module'模块复制具有限定名称'target'的属性值到'to_module'模块
此命令在目标路径下不存在时安装空的模块
def _copy_attr(from_module: 火炬.
神经网络.
模块,
到模块:
火炬.
神经网络.
模块,
目标:
字符串):
*前缀,
字段 =
目标.
分割(
“。”)
为
项目
在
前缀:
f = getattr(from_module, 项目)
t = getattr(到模块,
项目,
无)
如果 f is t:
# 我们已经安装了其父级之一
# (例如，target = root.linear.weight，但我们已经安装了 root.linear)
# 一次安装了父节点，就不再需要复制子节点
# 因为所有需要的属性已经存在
返回
如果 t is
无:
t = 火炬.
神经网络.
模块()
setattr(到模块,
项目, t)
来自模块,
到模块 = f, t
原始 = getattr(
来自模块,
字段)
如果它是一个张量而不是模块的参数属性，则它应该是一个命名缓冲区。
因此，我们将其注册为目标模块中的命名缓冲区。
如果 isinstance(
原始,
火炬.
张量)
和
不是 isinstance(
原始,
火炬.
神经网络.
参数):
到模块.
注册缓冲区(
字段,
原始)
else:
setattr(到模块,
字段,
原始)
将属性 'from_obj' 分配给 'to_module' 上的 'target' 限定名称
如果它们是目标路径的子路径，则此操作将安装不存在的新空模块
def _分配属性(from_obj:
任何,
到模块:
火炬.
神经网络.
模块,
目标:
字符串):
*前缀,
字段 =
目标.
分割(
“。”)
为
项目
在
前缀:
t = getattr(到模块,
项目,
无)
如果 t is
无:
t = 火炬.
神经网络.
模块()
setattr(到模块,
项目, t)
to_module = t
如果它是一个张量而不是模块的参数属性，则它应该是一个命名缓冲区。
因此，我们在目标模块中将它注册为一个命名缓冲区。
如果 isinstance(from_obj,
火炬.
张量)
和
不是 isinstance(
from_obj, 火炬.
神经网络.
参数
):
到模块.
注册缓冲区(
字段, from_obj)
else:
setattr(到模块,
字段, from_obj)
递归地从图模块中查找目标。
def _get_attr(模型:
火炬.
神经网络.
模块,
属性名称:
字符串):
返回
通过属性列表获取属性(
模型,
属性名称.
分割(
“。”))
def 删除属性(
模型:
火炬.
神经网络.
模块,
属性名称:
字符串):
属性名称 =
属性名称.
分割(
“。”)
t = 通过属性列表获取属性(
模型,
属性名称
[-1])
返回 delattr(t,
属性名称[-1])
def 通过属性列表获取属性(
模型:
火炬.
神经网络.
模块,
属性列表:
列表[
字符串
)]
如果
长度(
属性列表) == 0:
返回
模型
*前缀,
字段 =
属性列表
t = 模型
为
项目
在
前缀:
t = getattr(t, 项目,
无)
# 类型：忽略[赋值]
断言 t is
不是
无
返回 getattr(t,
字段)
def _有属性(
模型:
火炬.
神经网络.
模块,
属性名称:
字符串):
*前缀,
字段 =
属性名称.
分割(
“。”)
t = 模型
为
项目
在
前缀:
t = 有属性(t,
项目)
# 类型：忽略[赋值]
如果 t is
错误:
返回
假
返回
有属性(t,
字段)
def 打印可读的(
模块,
模块名称,
打印输出=
是,
包含步长=
错误,
包含设备=
错误,
彩色=
错误,
):
图 =
模块.
图
断言
图 is
不是
无
和 isinstance(
graph, 火炬.fx.
图
), "必须在具有图的模块上使用 print_readable"
verbose_python_code = graph.python_code(
根模块=
self,
详细模式=
是,
包含步长=
包含步长,
包含设备=
包含设备,
彩色=
彩色,
)
模块代码 =
详细的 Python 代码.
源
模块代码 =
模块代码.
去除字符串左侧空白字符("
输入文本翻译为简体中文为：\n")
模块代码 = f
类{
模块名称}(torch.nn.Module):
输入文本翻译为简体中文为：\n" +
模块代码
模块代码 = _addindent(
模块代码, 4)
子模块代码列表 = [
输入文本翻译为简体中文为：""]
为
子模块名称,
子模块
在
模块.
命名子项():
如果
有属性(
子模块,
"图"):
子模块代码列表.append(
打印可读的(
子模块,
子模块名称,
打印输出=
错误,
包含步长=
包含步长,
包含设备=
包含设备,
彩色=
彩色,
)
)
子模块代码 = "
输入文本翻译为简体中文为：\n".
连接(
子模块代码列表)
子模块代码 = _addindent(
子模块代码, 4)
输出 =
模块代码 +
子模块代码
如果
打印输出:
打印(
模块代码 +
子模块代码)
返回
输出
类
_包装调用:
def __init__(self, 类, cls_call):
self.类 =
类
self.cls_call = cls_call
#之前，如果发生错误时有效
# 符号追踪代码使用无效输入运行，用户会看到错误来源为
# `File ""`，其中 N 是某个数字。我们使用
# 此函数来生成更详细的错误信息。我们
# 使用此函数来生成更详细的错误信息。
# 返回跟踪堆栈本身，一个解释在跟踪模块生成的正向函数中发生错误的消息，以及围绕错误行的五行上下文
# 错误发生在跟踪模块生成的正向函数中
# 函数，以及错误行的五行上下文
#
@staticmethod
def 生成错误信息(
框架摘要:
跟踪回溯.
框架摘要)
翻译
字符串:
# 辅助变量（以提高可读性）
行号错误 =
调用帧摘要.
行号
断言
错误行号 is
不是
无
行 =
调用栈摘要.
行
断言
行 is
不是
无
错误行长度 =
长度(
行)
所有源代码行 =
行缓存.
获取行(
框架摘要.
文件名)
# 错误消息的构成子串
tb 表示 =
火炬._dynamo.
禁用(
跟踪回溯.format_exc)()
custom_msg = (
"使用 FX-traced 模块调用，"
f"行"{
错误行号}
跟踪模块的 " "
"生成的正向函数："
)
before_err = 输入文本翻译为简体中文为："".
连接(
所有源行[
错误行号 - 2 :
错误行号])
标记 =
波浪号 *
错误行长度 +
"~~~ <--- 这里"
错误及错误之后 = "
输入文本翻译为简体中文为：\n".
连接(
所有源代码行[
错误行号 :
错误行号 + 2])
# 合并消息
返回 "
输入文本翻译为简体中文为：\n".
连接
[
调试表示,
自定义消息,
错误之前,
标记,
错误及错误之后])
def __调用__(self,
对象, *
参数, **kwargs):
尝试:
如果 self.
cls 调用 is
不是
无:
返回 self.
cls 调用(
对象, *
参数, **kwargs)
else:
返回
超级(self.
类,
对象).
__调用__(*
参数, **kwargs)
# 类型：忽略[杂项]
除了
异常
是 e:
断言 e.
跟踪回溯
最顶层帧摘要:
跟踪回溯.
帧摘要 = (
跟踪回溯.
栈摘要.
提取(
跟踪回溯.
步进跟踪(e.
追踪回溯))[-1]
)
如果
使用密钥评估
在
最顶层帧摘要.
文件名:
打印(
包装调用.
生成错误信息(
最顶层帧摘要),
文件=
系统模块.
标准错误输出,
)
提升 e.
带有 traceback(
无) # noqa: B904
else:
提升 e
[文档]@compatibility(
兼容旧版本=
是)
类 GraphModule(
火炬.
神经网络.
模块):
""
GraphModule 是由 fx.Graph 生成的 nn.Module。Graphmodule 具有来自那个 ``graph`` 的 ``graph`` 属性，以及由 ``graph`` 生成的 ``code`` 和 ``forward`` 属性。
``graph`` 属性，以及 ``code`` 和 ``forward`` 属性生成
从那个 ``graph``。
.. 警告::
当重新分配 `graph` 时，`code` 和 `forward` 将会自动
重新生成。然而，如果您在未重新分配的情况下编辑了 ``graph`` 的内容
该 ``graph`` 属性本身，你必须调用 ``recompile()`` 来更新生成的
代码
""
def __new__(类:
"type[图模块]", *
参数, **kwargs):
# 每个图模块的实例都需要自己的前向方法
# 因此为每个实例创建一个新的单例类。
# 它是用户定义类的子类，唯一的区别是
安装前向方法的多余层
详细问题请见 https://github.com/pytorch/pytorch/issues/63883
换句话说，遍历类层次结构以修复冗余类定义问题
为 t
在
类.__mro__:
c = t.__qualname__.分割(
“。”
)-1]
如果 c !=
"图模块实现":
类 = t
断开
类
图模块实现(
类):
忽略[misc, valid-type]
通过
返回
超级().__new__(GraphModuleImpl)
[文档] @compatibility(
兼容旧版本=
是)
def __init__(
self,
根:
联合[
火炬.
神经网络.
模块,
字典[
字符串,
任何]],
graph: 图,
类名:
字符串 =
图模块,
):
""
构建一个图模块。
参数：
root (Union[torch.nn.Module, Dict[str, Any]]):
``root`` 可以是一个 nn.Module 实例，或者是一个将字符串映射到任何属性类型的 Dict。
在 ``root`` 是一个 Module 的情况下，Graph 的 Nodes 的 ``target`` 字段中任何对基于 Module 的对象的引用（通过限定名称）都将从相应位置复制过来。
如果 ``root`` 是一个模块，那么在 Graph 的节点 ``target`` 字段中，任何基于模块的对象引用（通过限定名称）都将从相应位置复制。
在`root`的模块层次结构中进入 GraphModule 的模块层次结构。
如果`root`是一个字典，则会在 Node 的`target`中找到的限定名将直接在字典的键中查找。
由 Dict 映射的对象将被复制到 GraphModule 的模块层次结构中的适当位置。
在 GraphModule 的模块层次结构中的适当位置。
graph (图): ``graph`` 包含此 GraphModule 应用于代码生成的节点
class_name (字符串): ``name`` 表示此 GraphModule 的名称，用于调试目的。如果未设置，所有
错误信息都将报告来自 ``GraphModule``。将其设置为 ``root`` 的原始名称或适合您转换上下文的名称可能很有帮助。
可能很有帮助将其设置为 ``root`` 的原始名称或适合您转换上下文的名称。
""
超级().__init__()
self.类.__name__ =
类名
如果 isinstance(
根,
火炬.
神经网络.
模块):
如果
有属性(
根,
"训练"):
self.训练 =
根.
训练
# 当我们序列化/反序列化图模块时，我们不想丢失任何模块或属性。
如果 isinstance(
根,
_仅代码模块):
为 k, _
在
根.
命名子项():
_复制属性(
根, self, k)
为 k, _
在
根.
命名缓冲区():
复制属性(
根, self, k)
为 k, _
在
根.
命名参数():
_复制属性(
根, self, k)
为
节点
在 graph.
节点:
如果
节点.
操作符
在 [
获取属性,
调用模块
]
断言 isinstance(
节点.
目标,
字符串)
_复制属性(
根, self,
节点.
目标)
如果...否则 isinstance(
根,
字典):
要复制的目标 =
输入文本为空，请提供需要翻译的文本
为
节点
在 graph.
节点:
如果
节点.
操作符
在 [
获取属性,
调用模块
]
断言 isinstance(
节点.
目标,
字符串)
如果
节点.
目标
不是
在
根:
提升
运行时错误(
"节点 "
+ 字符串(
节点)
+ "引用的目标 "
+ 节点.
目标
+ "但该目标未在 ``root`` 中提供！"
)
targets_to_copy.append(节点.
目标)
按原子数量升序排序目标。
这将确保将较浅层嵌套的属性分配给
在更深层嵌套的属性之前。例如，foo.bar
将在 foo.bar.baz 之前分配。否则，我们可能会分配
清除先前分配的 ``foo.bar``
``foo.bar.baz``
要复制的目标.
排序(
键=lambda t: t.
数量(
“。”))
为
要复制的目标
在
要复制的目标:
_分配属性(
根[
要复制的目标
] self,
要复制的目标)
else:
提升
运行时错误(
"不支持类型 " +
字符串(
根) +
"通过了 root！")
self.图 =
图
# 将负责创建 Graph 的 Tracer 类单独存储，作为 GraphModule 状态的一部分
# 当 Tracer 定义在局部命名空间中时，除外
# 局部定义的 Tracer 不可序列化。这是必要的，因为 torch.package 将
序列化 GraphModule 而不保留 Graph，并需要使用正确的 Tracer
在反序列化期间重新创建 Graph。
self._tracer_cls = 无
如果 (
self.graph._tracer_cls
和
本地
不是
在 self.graph._tracer_cls.__qualname__
):
self._tracer_cls = self.graph._tracer_cls
self._tracer_extras = {}
如果 self.graph._tracer_extras:
self._tracer_extras = self.graph._tracer_extras
# 存储元数据的字典
self.元数据:
字典[
字符串,
任何] = {}
self.替换钩子:
列表[
可调用] =
输入文本为空，请提供需要翻译的文本
self.创建节点钩子:
列表[
可调用] =
输入文本为空，请提供需要翻译的文本
self._erase_node_hooks: 列表[
可调用] =
输入文本为空，请提供需要翻译的文本
用于从上下文管理器中的深拷贝图模块中移除钩子。
self._deepcopy_hooks: 列表[
可调用] = []
# TorchScript 编译图设置器时出现错误，因为
# 字符串字面量继续。问题在这里：https://github.com/pytorch/pytorch/issues/44842
#
# 由于这些方法本来就不应该在 TorchScript 中使用，所以不应有问题
__jit_unused_properties__ = ["图"]
@property
def graph(self) 翻译
图:
""
返回此 `GraphModule` 的底层 `Graph`
""
返回 self._graph
@graph.setter
def graph(self, g: 图)
翻译
无:
""
设置此 `GraphModule` 的底层 `Graph`。这将内部
重新编译 `GraphModule` 以生成 `forward()` 函数
对应于 `g`
""
断言 isinstance(g,
图), f
预期得到一个图实例，但得到了{
类型(g)}"
self._graph = g
g.拥有模块 = self
self.重新编译()
[文档] @compatibility(
兼容旧版本=
错误)
def 到文件夹(self,
文件夹:
联合[
字符串, os.PathLike
]
模块名称:
字符串 =
"Fx 模块"):
"将模块输出到 ``folder`` 目录下，以 ``module_name`` 为名，以便可以"
使用 ``from import `` 导入
参数：
folder (Union[str, os.PathLike]): 将代码写出的目录（字符串或路径类型）
模块名称（字符串）：在输出代码时使用的 ``Module`` 的顶级名称
写入代码
""
文件夹 =
路径(
文件夹)
路径(
文件夹).
建立目录(exist_ok=
是)
火炬.
保存(self.
状态字典(),
文件夹 /
state_dict.pt)
标签 =
输入文本为空，请提供需要翻译的文本 * 4
自定义内置函数 = "
输入文本翻译为简体中文为：\n".
连接
[v.
导入字符串
为 v
在
_自定义内置函数.
值()])
模型字符串 = f
""
导入 torch
{自定义内置函数}
从 torch.nn 导入所有内容
类{
模块名称}
(torch.nn.Module)
def __init__(self):
super().__init__()
""
def _gen_model_repr(模块名称:
字符串,
模块:
火炬.
神经网络.
模块)
翻译
可选[
字符串
]
safe_reprs = [
神经网络.
线性,
神经网络.
卷积 1D,
神经网络.
卷积 2D,
神经网络.
卷积 3D,
神经网络.
批标准化 1d,
神经网络.
批标准化 2d,
神经网络.
批标准化 3d,
]
如果
类型(
模块)
在 safe_reprs:
返回 f"{
模块.__repr__()}"
else:
返回
无
模块化 =
输入文本为空，请提供需要翻译的文本
为
模块名称,
模块
在 self.
命名子项():
模块字符串 =
生成模型表示(
模块名称,
模块)
如果
模块字符串 is
无:
模块文件 =
文件夹 / f"{
模块名称}
.pt
火炬.
保存(
模块,
模块文件)
转换为 blob 的模块.append(
模块名称)
模块表示 =
模块.__repr__().
替换("
回车符",
输入文本为空，请提供需要翻译的文本).
替换("
输入文本翻译为简体中文为：\n",
输入文本为空，请提供需要翻译的文本)
# weights_only=False，因为这是保存模型的旧代码
模块字符串 = (
f"torch.load(r'"{
模块文件}
', 仅权重=False) # '{
模块表示}"
)
模型字符串 += f"{
标签 * 2}
我{
模块名称} = {
模块字符串}
输入文本翻译为简体中文为：\n"
为
缓冲区名称,
缓冲区
在 self.
_缓冲区.
项目():
如果
缓冲区 is
无:
continue
模型字符串 += f"{tab * 2}self.register_buffer('{buffer_name}', torch.empty({
列表(
缓冲区.
形状)}
, 数据类型={
缓冲区.
数据类型}))
输入文本翻译为简体中文为：\n"
# 无需注意：B950
为
` 的类型为 List[torch.Tensor],
参数
在 self.
参数.
项目():
如果
参数 is
无:
continue
模型字符串 += f"{
标签 * 2}self.{
` 的类型为 List[torch.Tensor]} = torch.nn.Parameter(torch.empty({
列表(
参数.
形状)}
, 数据类型={
参数.
数据类型}))
输入文本翻译为简体中文为：\n"
# 无需注意：B950
模型字符串 += (
f"{标签 * 2}self.load_state_dict(torch.load(r'{
文件夹}
/state_dict.pt
输入文本翻译为简体中文为：\n"
)
模型字符串 += f"{_addindent(self.
代码, 4)}
输入文本翻译为简体中文为：\n"
模块文件 =
文件夹 /
"模块.py"
模块文件.
编写文本(
模型字符串)
初始化文件 =
文件夹 / "__init__.py"
初始化文件.
写入文本(
"从 .module 导入 *")
如果
长度(blobified_modules) > 0:
警告.
警告(
无法保存以下子模块作为 reprs -
f"保存为 pickle 文件："{
blob 化模块}"
)
[文档] @兼容性(向后兼容=True)
def add_submodule(self, target: str, m: torch.nn.Module) -> bool:
"""
将给定的子模块添加到 ``self``。
这将安装空模块，如果它们不存在且是 ``target`` 的子路径。
参数：
Args:
target: 新子模块的完全限定字符串名称
（参见 `nn.Module.get_submodule` 中的示例，了解如何）
指定一个完全限定的字符串。
m：子模块本身；我们想要的实际对象
安装到当前模块
返回：
布尔值：子模块是否可以插入。
此方法返回 True，链中的每个对象
表示为 `target` 的目标必须要么 a) 尚未存在
或 b) 引用 ``nn.Module``（非参数或其他属性）
)
"""
*前缀，字段 = target.split(".")
mod: torch.nn.Module = self
for item in prefix:
submod = getattr(mod, item, None)
if submod is None:
submod = torch.nn.Module()
setattr(mod, item, submod)
if not isinstance(submod, torch.nn.Module):
return False
mod = 子模块
mod.add_module(字段, m)
返回 True
[文档] @兼容性(is_backward_compatible=True)
def delete_submodule(self, target: str) -> bool:
"""
从 ``self`` 中删除指定的子模块。
如果 ``target`` 不是一个有效的模块，则不会删除该模块。
目标.
参数:
目标：新子模块的完全限定字符串名称
（请参阅 ``nn.Module.get_submodule`` 中的示例了解如何
指定一个完全限定的字符串。)
返回：
bool：目标字符串是否引用了
我们想要删除的子模块。返回值如果为 ``False``
表示“目标”不是一个有效的引用
的子模块。
"""
原子 = 目标.split(".")
路径，目标子模块 = atoms[:-1]，atoms[-1]
模块：torch.nn.Module = self
# 获取父模块
for item in path:
如果没有 hasattr(mod, item):
返回 False
mod = getattr(mod, item)
如果 mod 不是 torch.nn.Module 类型:
返回 False
如果没有 mod 对象具有 target_submod 属性
返回 False
如果 getattr(mod, target_submod)不是 torch.nn.Module 类型
返回 False
删除对象的属性
返回 True
[文档] @compatibility(
兼容旧版本=
是)
def 删除所有未使用的子模块(self)
翻译
无:
""
从 ``self`` 中删除所有未使用的子模块。
如果以下任一条件为真，则认为模块是“已使用”的：
true:
1. 它有子节点被使用
2. 通过 `call_module` 节点直接调用其前向
3. 它有一个非 Module 属性，从
``get_attr`` 节点使用
可以调用此方法来清理 ``nn.Module`` 而无需
手动调用 ``delete_submodule`` 来删除每个未使用的子模块。
""
使用:
列表[
字符串] =
输入文本为空，请提供需要翻译的文本
为
节点
在 self.graph.
节点:
如果
节点.
操作符 ==
"调用模块"
或者
节点.
操作符 ==
获取属性:
列表字符串表示路径的不同部分
例如，`foo.bar.baz` 将给我们
# ["foo", "bar", "baz"]
完整路径 =
节点.
目标.
分割(
“。”)
如果我们正在查看路径的多个部分，则使用点号连接它们。
否则，返回该单个元素，不做任何处理。
join_fn
def 连接函数(x:
字符串, y:
字符串)
翻译
字符串:
返回
“。”.
连接
[x, y]
如果 y
否则 [x])
逐步收集所有中间名称
模块。例如，如果我们有目标
`foo.bar.baz`，我们将添加`foo`、`foo.bar`和
将 `foo.bar.baz` 添加到列表中。
使用.
扩展(itertools.
累加(
完整路径,
合并函数))
对于 `call_module` 节点，同时注册所有递归子模块
如使用
如果
节点.
操作符 ==
调用模块:
尝试:
子模块 = self.
获取子模块(
节点.
目标)
为
子模块名称, _
在
子模块.
命名模块():
如果
子模块名称 !=
输入文本翻译为简体中文为："":
使用.append(
“。”.
连接
[
节点.
目标,
子模块名称]))
除了
属性错误:
# 节点引用了不存在的子模块，无需担心垃圾回收
# 无需担心垃圾回收
通过
要删除 = [
名称
为
名称, _
在 self.
命名模块()
如果
名称
不是
在
使用]
为
名称
在
删除:
self.删除子模块(
名称)
@property
def 代码(self)
翻译
字符串:
""
返回由此 ``Graph`` 下的 Python 代码生成
``GraphModule``。
""
如果
不是
有属性(self,
"_代码"):
提升
运行时错误(
"代码尚未生成！请向 PyTorch 报告一个错误"
)
返回 self.
_代码
[文档] @兼容性(向后兼容=True)
def recompile(self) -> PythonCode:
"""
从其 ``graph`` 属性重新编译此 GraphModule。这应该
在编辑完包含的“图”之后调用，否则生成的
此 ``GraphModule`` 的代码将过时。
```python
# 输入文本
input_text = '"""'
# 翻译函数（此处为示例，实际翻译功能需调用真实的翻译 API）
def translate_to_simplified_chinese(text):
# 假设的翻译结果
return text
# 输出翻译结果
translated_text = translate_to_simplified_chinese(input_text)
print(translated_text)
```
如果 isinstance(self._graph._codegen, _PyTreeCodeGen):
self._in_spec = self._graph._codegen.pytree_info.in_spec
self._out_spec = self._graph._codegen.pytree_info.out_spec
python_code = self._graph.python_code(root_module="self")
self._code = python_code.src
self._lineno_map = python_code._lineno_map
cls = type(self)
co_fields = self._graph._co_fields if hasattr(self._graph, "_co_fields") else {}
cls.forward = _forward_from_src(self._code, python_code.globals, co_fields)
判断这个类是否显式定义了 __call__ 实现方法
如果它定义了，保存它，以便 wrapped_call 可以调用它。
如果它没有定义，wrapped_call 可以使用动态调用 super() 来代替。
在大多数情况下，super().__call__ 应该是 torch.nn.Module.__call__。
我们不想在这里引用 Module.__call__；这样做将
绕过符号跟踪期间对 torch.nn.Module.__call__ 执行的补丁。
cls_call = cls.__call__ if "__call__" in vars(cls) else None
如果 "_wrapped_call" 不在 cls 的 vars 中：
cls._wrapped_call = _WrappedCall(cls, cls_call) # 忽略未定义的属性
def call_wrapped(self, *args, **kwargs):
return self._wrapped_call(self, *args, **kwargs)
cls.__call__ = call_wrapped # 忽略方法分配
返回 python 代码
允许子类扩展 fx.GraphModule 时传递 Tracer 作为参数
定义它们自己的 Tracer（扩展 fx.Tracer）
def __reduce_deploy__(self, 导入器:
导入器):
无图字典 = self.
字典.
复制()
无图字典[
_图模块类名] = self.
类.__name__
删除
无图字典[
_图]
Python 代码 = self.
重新编译()
导入块 =
格式导入块(
Python 代码.
全局变量,
导入器)
返回 (
减少部署图模块, (
字典无图,
导入块))
def __reduce_package__(self, 导出器:
包导出器):
无图字典 = self.
字典.
复制()
无图字典[
"_graph 模块类名"] = self.
类.__name__
删除
无图字典[
_图]
生成的模块名 = f
fx-生成._{
出口商.
获取唯一 ID()}"
Python 代码 = self.
重新编译()
导入块 =
格式化导入块(
python 代码.
全局变量,
导出器.
导入器)
模块代码 =
导入块 + self.
代码
出口商.
保存源字符串(
生成模块名称,
模块代码)
返回 (
优化包图模块,
(不含图的字典,
生成模块名称),
)
def __reduce__(self):
""
图模块的序列化。我们只序列化生成的代码，而不是
底层的 ``Graph``。这是因为 ``Graph`` 没有磁盘上的
向后兼容保证，而 Python 源代码则不是。
在反序列化方面，我们通过生成的代码进行符号跟踪。
代码以重新生成底层的“图”。
""
无图字典 = self.
字典.
复制()
python 代码 = self.
重新编译()
导入块 =
格式化导入块(
python 代码.
全局变量,
sys 导入器)
删除
不含图的字典[
_图]
返回 (
减少图模块, (
不含图的字典,
导入块))
def _deepcopy_init(self):
返回 GraphModule.__init__
# 因为定义了 __reduce__ 用于序列化，
# 我们需要定义 deepcopy 否则它会调用 __reduce__
# 和每次尝试复制对象时都会触发符号跟踪
def 深拷贝(self,
描述):
res = 类型(self).__new__(
类型(self))
描述[id(self)] = res
模拟模块 =
仅代码模块(
复制.
深拷贝(self.
字典,
描述))
self.深拷贝初始化()(
资源,
模拟模块,
模拟模块.
字典["_graph"])
# 在 `GraphModule.__init__` 过程中钩子会丢失，因此我们需要显式地复制
# 它们，目前我们只复制与 state_dict 相关的钩子，以减少向后兼容性问题
# 我们可以复制前向/后向相关的钩子，以减少向后兼容性问题
未来如有需要也可挂钩
额外保留的属性 = [
"_状态字典挂钩",
"_加载状态字典前挂钩",
"_load_state_dict_post_hooks",
"_replace_hooks",
创建节点钩子,
删除节点钩子,
深拷贝钩子,
]
为
属性
在
额外保留属性:
如果
属性
在 self.
字典:
setattr(资源,
属性,
复制.
深拷贝(self.
字典[
属性
]
描述))
资源.
元数据 =
复制.
深拷贝(getattr(self,
元数据, {}),
描述)
如果
用户保留属性键
在
资源.
元数据:
为
属性名称,
属性
在
资源.
元数据[
用户保留属性键].
项目():
setattr(资源,
属性名称,
属性)
如果
有属性(self,
_deepcopy_hooks):
为
钩子
在 self._deepcopy_hooks:
hook(资源)
返回 res
def 复制(self):
from ._lazy_graph_module 导入
_创建图形模块
res = _创建图形模块(self, self.graph)
资源.
元数据 = getattr(self,
元数据, {})
返回 res
[文档] @兼容性(向后兼容=False)
def 打印可读(
self,
print_output=True,
include_stride=False,
include_device=False,
colored=False,
):
"""
返回当前 GraphModule 及其子 GraphModule 生成的 Python 代码
"""
返回 _print_readable(
self,
self._get_name(),
打印输出,
包含步长,
包含设备,
彩色,
)
def __str__(self) 翻译
字符串:
原始字符串 =
超级().__str__()
打印可读性提醒 = (
“要查看更多调试信息，请使用 `graph_module.print_readable()`”
)
返回 "
输入文本翻译为简体中文为：\n".
连接
[
原始字符串, self.
_代码,
打印可读性提醒])
def _为数据并行复制(self):
新 gm = self.
复制()
新 gm.
_是副本 =
真实
返回
新 gm
@contextlib.contextmanager
def _set_replace_hook(self, f):
""
接收一个可调用的函数，每次替换节点时都会调用该函数
将节点移动到新位置，或更改节点的名称。可调用函数需要三个参数：
旧节点以及新节点的名称，然后是
消费即将被替换的旧节点的用户节点。
""
断言
可调用(f),
"替换钩子必须是一个可调用对象。"
self.注册替换节点钩子(f)
尝试:
产生
最后:
self.取消注册替换节点钩子(f)
def 注册替换节点钩子(self, f):
""
接受一个可调用的函数，每次替换节点时都会调用该函数
将节点移动到新位置，或更改节点的名称。Callable 函数接受三个参数：
我们正在更改的旧节点，以及新节点的名称，后面跟着的
消费即将被替换的旧节点的用户节点。
""
断言
可调用(f),
"创建节点钩子必须是一个可调用对象。"
self.替换钩子.append(f)
def 取消注册替换节点钩子(self, f):
""
接收一个之前已注册的调用函数，每次替换节点时都会调用该函数。
此函数将取消注册该调用函数，使其在节点替换时不再被调用。
""
断言
可调用(f),
"创建节点钩子必须是一个可调用对象。"
self.替换钩子.
删除(f)
def 注册创建节点钩子(self, f):
""
接收一个可调用对象，该对象将在我们创建新节点后被调用。
可调用对象接收新创建的节点作为输入，并返回 None。
""
断言
可调用(f),
"创建节点钩子必须是一个可调用对象。"
self._create_node_hooks.append(f)
def _unregister_create_node_hook(self, f):
""
接收一个之前已注册的可调用对象，在创建节点后调用。
此函数将注销该可调用对象，使其在节点创建时不再被调用。
""
断言
可调用(f),
"创建节点钩子必须是一个可调用对象。"
self._create_node_hooks.删除(f)
def _register_erase_node_hook(self, f):
""
接收一个可调用对象，该对象将在我们删除节点后被调用。
可调用函数接收要删除的节点作为输入并返回 None。
""
断言
可调用(f),
"erase_node 钩子必须是可调用的。"
self._erase_node_hooks.append(f)
def _unregister_erase_node_hook(self, f):
""
接收一个之前注册用于在删除节点后调用的可调用对象。
此函数将取消注册该可调用对象，使其在节点删除时不再被调用。
""
断言
可调用(f),
"删除节点钩子必须是一个可调用对象。"
self._erase_node_hooks.删除(f)
def 注册深度复制钩子(self, f):
""
接收一个可调用对象，当深度复制此图模块时将被调用。该可调用对象接收深度复制的图模块作为结果。
可调用对象必须是一个可调用对象。
""
断言
可调用(f),
"深度复制钩子必须是一个可调用对象。"
self._deepcopy_hooks.append(f)
def _unregister_deepcopy_hook(self, f):
""
接收一个之前注册用于在 deepcopy 之后调用的可调用对象。
此函数将取消注册该可调用对象，使其在 deepcopy 时不再被调用。
""
断言
可调用(f),
深度复制钩子必须是一个可调用对象。
self._deepcopy_hooks.删除(f)
# __torch_function__ 中的问题解决方案
# 对于 __torch_function__ 无法处理张量列表的问题的应对措施
# 修复方案在 https://github.com/pytorch/pytorch/pull/34725
# 原始_cat = torch.cat
# 定义修补后的_cat(*args, **kwargs):
# tensors = args[0]
# for t in tensors:
# 如果 isinstance(t, Proxy):
# 返回 t.__torch_function__(patched_cat, (), args, kwargs)
# 返回 orig_cat(*args, **kwargs)
# patched_cat.__module__ = 'torch'
# patched_cat.__name__ = '猫'
# torch.cat = patched_cat
