企业官网建设流程全解析

【Bug已解决】How to solve the run time error "Only Tensors created explicitly by the user (graph leaves) support the deepcopy protocol at the moment" 解决方案

问题描述

在 PyTorch 中使用copy.deepcopy()复制 tensor 或模型时，开发者可能会遇到以下错误：

RuntimeError: Only Tensors created explicitly by the user (graph leaves) support the deepcopy protocol at the moment

这个错误通常发生在以下场景中：

• 使用copy.deepcopy()复制模型或 tensor
• 在多进程训练中使用 DataLoader（num_workers > 0）时
• 使用torch.multiprocessing进行分布式训练时
• 在自定义数据集或数据加载逻辑中使用deepcopy
• 尝试复制包含计算图的 tensor

该错误的根本原因是 PyTorch 的计算图机制与 Python 的deepcopy协议之间的冲突。只有"图叶节点"（graph leaves，即由用户显式创建的 tensor，如torch.randn()、torch.tensor()创建的 tensor）才支持deepcopy，而计算图中的中间结果（非叶节点）不支持。

错误复现

场景一：复制计算图中的中间 tensor

import torch import copy # 创建一个需要梯度的tensor（叶节点） x = torch.randn(10, 10, requires_grad=True) # 前向传播产生中间结果（非叶节点） y = x * 2 z = y.sum() z.backward() # 尝试deepcopy中间结果 try: y_copy = copy.deepcopy(y) except RuntimeError as e: print(f"错误: {e}") # 输出: Only Tensors created explicitly by the user (graph leaves) # support the deepcopy protocol at the moment

场景二：复制包含计算图的模型输出

import torch import torch.nn as nn import copy model = nn.Linear(10, 5) x = torch.randn(3, 10, requires_grad=True) # 前向传播 output = model(x) # 尝试deepcopy输出（非叶节点） try: output_copy = copy.deepcopy(output) except RuntimeError as e: print(f"错误: {e}")

场景三：DataLoader 多进程中的错误

import torch from torch.utils.data import Dataset, DataLoader import copy class MyDataset(Dataset): def __init__(self, data): self.data = data def __len__(self): return len(self.data) def __getitem__(self, idx): item = self.data[idx] # 如果item是计算图中的tensor，deepcopy会失败 return copy.deepcopy(item) # 可能触发错误 # 创建包含计算图tensor的数据 x = torch.randn(100, 10, requires_grad=True) y = x * 2 # 非叶节点 dataset = MyDataset(y) # 使用多进程DataLoader dataloader = DataLoader(dataset, batch_size=16, num_workers=2) try: for batch in dataloader: print(batch.shape) break except RuntimeError as e: print(f"错误: {e}")

场景四：复制整个模型时

import torch import torch.nn as nn import copy class MyModel(nn.Module): def __init__(self): super().__init__() self.linear = nn.Linear(10, 5) self.cache = None # 缓存前向传播结果 def forward(self, x): out = self.linear(x) self.cache = out # 缓存中间结果 return out model = MyModel() x = torch.randn(3, 10, requires_grad=True) output = model(x) # 此时model.cache是非叶节点tensor # 尝试deepcopy模型（模型内部缓存了非叶节点tensor） try: model_copy = copy.deepcopy(model) except RuntimeError as e: print(f"错误: {e}") # 因为model.cache是非叶节点tensor，deepcopy失败

根因分析

1. PyTorch 计算图与 Autograd 机制

PyTorch 使用动态计算图（Dynamic Computation Graph）来跟踪 tensor 上的操作，以支持自动微分。计算图由节点（tensor）和边（操作）组成：

• 叶节点（Leaf Node）：由用户显式创建的 tensor，且requires_grad=True。例如x = torch.randn(10, requires_grad=True)中的x是叶节点。
• 非叶节点（Non-Leaf Node）：通过对叶节点执行操作产生的中间结果。例如y = x * 2中的y是非叶节点。

PyTorch 的deepcopy协议只支持叶节点 tensor，因为非叶节点 tensor 包含了对计算图的引用，复制这些引用会导致计算图结构混乱。

2.deepcopy与计算图的冲突

当对非叶节点 tensor 执行deepcopy时，PyTorch 需要复制以下内容：

• tensor 的数据
• tensor 的梯度信息
• tensor 在计算图中的位置（grad_fn）
• 与其他 tensor 的依赖关系

复制grad_fn和依赖关系是非常复杂的，因为计算图是一个有向无环图（DAG），复制一个节点需要复制整个子图。PyTorch 目前不支持这种操作，因此直接报错。

3. DataLoader 多进程序列化

当DataLoader的num_workers > 0时，数据需要通过pickle序列化后在工作进程之间传输。pickle内部使用类似deepcopy的机制来复制对象。如果数据中包含非叶节点 tensor，序列化过程会触发相同的错误。

4. 模型内部缓存中间结果

如果模型在forward方法中将中间结果存储在实例变量中（如self.cache = output），当使用deepcopy复制模型时，这些中间结果也会被复制，从而触发错误。

5.retain_graph=True的影响

使用retain_graph=True进行多次反向传播时，计算图会被保留。此时尝试deepcopy相关 tensor 更容易触发错误，因为计算图更加复杂。

解决方案

方案一：使用detach()分离计算图

最直接的解决方案是在deepcopy之前使用detach()将 tensor 从计算图中分离：

import torch import copy x = torch.randn(10, 10, requires_grad=True) y = x * 2 # 非叶节点 # 方法1: detach后再deepcopy y_detached = y.detach() y_copy = copy.deepcopy(y_detached) print(f"复制成功: {y_copy.shape}") # 方法2: 如果不需要梯度，直接detach y_copy = copy.deepcopy(y.detach().clone()) print(f"复制成功: {y_copy.shape}")

方案二：使用clone()代替deepcopy

对于 tensor，clone()是更好的复制方式：

import torch x = torch.randn(10, 10, requires_grad=True) y = x * 2 # 非叶节点 # 使用clone复制tensor y_clone = y.clone() print(f"克隆成功: {y_clone.shape}") print(f"是否在计算图中: {y_clone.requires_grad}") # True，但有自己的计算图 # 如果不需要梯度 y_clone_no_grad = y.detach().clone() print(f"无梯度克隆: {y_clone_no_grad.requires_grad}") # False

方案三：复制模型时清理中间状态

import torch import torch.nn as nn import copy class SafeModel(nn.Module): def __init__(self): super().__init__() self.linear = nn.Linear(10, 5) self._cache = None def forward(self, x): out = self.linear(x) # 使用私有属性存储缓存，并在复制前清理 self._cache = out return out def __deepcopy__(self, memo): # 自定义deepcopy行为 # 先清理缓存 old_cache = self._cache self._cache = None # 创建新实例 cls = self.__class__ result = cls.__new__(cls) memo[id(self)] = result # 复制所有属性 for k, v in self.__dict__.items(): if k == '_cache': # 不复制缓存或复制detach版本 setattr(result, k, None) else: setattr(result, k, copy.deepcopy(v, memo)) # 恢复原模型的缓存 self._cache = old_cache return result # 使用 model = SafeModel() x = torch.randn(3, 10, requires_grad=True) output = model(x) # 现在可以安全deepcopy model_copy = copy.deepcopy(model) print("模型复制成功!")

方案四：使用torch.save和torch.load复制模型

import torch import torch.nn as nn import io class MyModel(nn.Module): def __init__(self): super().__init__() self.linear = nn.Linear(10, 5) def forward(self, x): return self.linear(x) model = MyModel() x = torch.randn(3, 10, requires_grad=True) output = model(x) # 使用save/load复制模型（不复制计算图） buffer = io.BytesIO() torch.save(model.state_dict(), buffer) buffer.seek(0) model_copy = MyModel() model_copy.load_state_dict(torch.load(buffer)) print("模型复制成功!") # 验证参数相同 for (n1, p1), (n2, p2) in zip(model.named_parameters(), model_copy.named_parameters()): assert n1 == n2 assert torch.equal(p1, p2) print("参数验证通过!")

方案五：修复 DataLoader 中的问题

import torch from torch.utils.data import Dataset, DataLoader class SafeDataset(Dataset): def __init__(self, data): # 在初始化时就detach所有tensor if isinstance(data, torch.Tensor): self.data = data.detach().clone() else: self.data = data def __len__(self): return len(self.data) def __getitem__(self, idx): item = self.data[idx] # 返回detach的tensor，避免计算图传播 if isinstance(item, torch.Tensor): return item.detach().clone() return item # 创建数据 x = torch.randn(100, 10, requires_grad=True) y = x * 2 # 非叶节点 # 使用安全的数据集 dataset = SafeDataset(y) dataloader = DataLoader(dataset, batch_size=16, num_workers=2) # 现在可以正常使用 for batch in dataloader: print(f"Batch shape: {batch.shape}, requires_grad: {batch.requires_grad}") break

完整修复代码

以下是一个完整的解决方案，包含安全的 tensor 复制、模型复制和数据加载功能：

""" 解决 PyTorch deepcopy 错误的完整工具集 RuntimeError: Only Tensors created explicitly by the user (graph leaves) support the deepcopy protocol at the moment """ import torch import torch.nn as nn import copy import io import warnings from typing import Any, Dict, List, Optional class SafeDeepCopy: """安全的PyTorch对象深拷贝工具""" @staticmethod def copy_tensor(tensor: torch.Tensor, detach: bool = True, clone: bool = True) -> torch.Tensor: """ 安全复制tensor Args: tensor: 要复制的tensor detach: 是否分离计算图 clone: 是否克隆数据 Returns: 复制后的tensor """ if detach: tensor = tensor.detach() if clone: tensor = tensor.clone() return tensor @staticmethod def copy_model(model: nn.Module, mode: str = 'state_dict') -> nn.Module: """ 安全复制模型 Args: model: 要复制的模型 mode: 复制模式 - 'state_dict': 使用state_dict复制（推荐，不复制计算图） - 'deepcopy': 使用deepcopy（需要先清理中间状态） - 'cpu_clone': 转移到CPU后克隆 Returns: 复制后的模型 """ if mode == 'state_dict': # 方法1: 使用state_dict（最安全） buffer = io.BytesIO() torch.save(model.state_dict(), buffer) buffer.seek(0) model_copy = type(model)() model_copy.load_state_dict(torch.load(buffer)) # 复制设备位置 device = next(model.parameters()).device model_copy = model_copy.to(device) # 复制训练/评估模式 model_copy.train(model.training) return model_copy elif mode == 'deepcopy': # 方法2: 清理中间状态后deepcopy # 保存并清理所有非参数/非缓冲区的tensor属性 saved_attrs = {} for name, attr in model.__dict__.items(): if isinstance(attr, torch.Tensor) and not isinstance(attr, nn.Parameter): # 检查是否是计算图中的tensor if attr.requires_grad and attr.grad_fn is not None: saved_attrs[name] = attr setattr(model, name, attr.detach()) try: model_copy = copy.deepcopy(model) finally: # 恢复原模型的属性 for name, attr in saved_attrs.items(): setattr(model, name, attr) return model_copy elif mode == 'cpu_clone': # 方法3: 转移到CPU后复制 original_device = next(model.parameters()).device model_cpu = model.cpu() model_copy = copy.deepcopy(model_cpu) model.to(original_device) model_copy.to(original_device) return model_copy else: raise ValueError(f"不支持的复制模式: {mode}") @staticmethod def copy_any(obj: Any) -> Any: """ 安全复制任意对象（自动处理tensor） Args: obj: 要复制的对象 Returns: 复制后的对象 """ if isinstance(obj, torch.Tensor): return SafeDeepCopy.copy_tensor(obj) elif isinstance(obj, nn.Module): return SafeDeepCopy.copy_model(obj) elif isinstance(obj, dict): return {k: SafeDeepCopy.copy_any(v) for k, v in obj.items()} elif isinstance(obj, (list, tuple)): return type(obj)(SafeDeepCopy.copy_any(item) for item in obj) else: try: return copy.deepcopy(obj) except RuntimeError as e: if "graph leaves" in str(e): warnings.warn(f"无法deepcopy对象，尝试detach: {e}") if hasattr(obj, 'detach'): return obj.detach().clone() raise @staticmethod def clean_model_internals(model: nn.Module) -> Dict[str, Any]: """ 清理模型内部的非叶节点tensor 返回被清理的属性，以便后续恢复 Args: model: 要清理的模型 Returns: 被清理的属性字典 {name: value} """ cleaned = {} for name, attr in list(model.__dict__.items()): if isinstance(attr, torch.Tensor): # 检查是否是非叶节点tensor if attr.grad_fn is not None or (attr.requires_grad and not attr.is_leaf): cleaned[name] = attr # 替换为detach版本或None model.__dict__[name] = attr.detach() # 递归清理子模块 for name, module in model.named_children(): sub_cleaned = SafeDeepCopy.clean_model_internals(module) if sub_cleaned: cleaned[f"{name}.{list(sub_cleaned.keys())[0]}"] = list(sub_cleaned.values())[0] return cleaned @staticmethod def restore_model_internals(model: nn.Module, cleaned: Dict[str, Any]): """恢复被清理的模型属性""" for name, value in cleaned.items(): if '.' in name: # 处理嵌套属性 parts = name.split('.') obj = model for part in parts[:-1]: obj = getattr(obj, part) setattr(obj, parts[-1], value) else: model.__dict__[name] = value class SafeDataLoaderWrapper: """安全的DataLoader包装器，自动处理计算图tensor""" def __init__(self, dataset, batch_size=32, num_workers=0, **kwargs): """ Args: dataset: 数据集，可能包含计算图tensor batch_size: 批量大小 num_workers: 工作进程数 """ # 预处理数据集，detach所有tensor self._preprocess_dataset(dataset) self.dataloader = DataLoader( dataset, batch_size=batch_size, num_workers=num_workers, **kwargs ) def _preprocess_dataset(self, dataset): """预处理数据集，detach所有tensor""" if hasattr(dataset, 'data') and isinstance(dataset.data, torch.Tensor): dataset.data = dataset.data.detach().clone() if hasattr(dataset, 'targets') and isinstance(dataset.targets, torch.Tensor): dataset.targets = dataset.targets.detach().clone() def __iter__(self): return iter(self.dataloader) def __len__(self): return len(self.dataloader) # ============ 使用示例 ============ def demo_tensor_copy(): """演示安全tensor复制""" print("=" * 60) print("演示: 安全tensor复制") print("=" * 60) # 创建计算图中的tensor x = torch.randn(10, 10, requires_grad=True) y = x * 2 + 1 # 非叶节点 print(f"原始tensor: shape={y.shape}, is_leaf={y.is_leaf}, grad_fn={y.grad_fn}") # 错误的deepcopy try: y_copy = copy.deepcopy(y) except RuntimeError as e: print(f"deepcopy失败: {e}") # 安全复制 y_copy = SafeDeepCopy.copy_tensor(y) print(f"安全复制: shape={y_copy.shape}, is_leaf={y_copy.is_leaf}, grad_fn={y_copy.grad_fn}") # 验证数据相同 assert torch.equal(y, y_copy) print("数据验证通过!") def demo_model_copy(): """演示安全模型复制""" print("\n" + "=" * 60) print("演示: 安全模型复制") print("=" * 60) class TestModel(nn.Module): def __init__(self): super().__init__() self.linear1 = nn.Linear(10, 20) self.linear2 = nn.Linear(20, 5) self.cache = None def forward(self, x): h = self.linear1(x) self.cache = h # 缓存中间结果 return self.linear2(h) model = TestModel() x = torch.randn(3, 10, requires_grad=True) output = model(x) # 此时model.cache是非叶节点 print(f"模型cache: is_leaf={model.cache.is_leaf}, grad_fn={model.cache.grad_fn is not None}") # 错误的deepcopy try: model_copy = copy.deepcopy(model) except RuntimeError as e: print(f"deepcopy失败: {e}") # 安全复制 - state_dict模式 model_copy = SafeDeepCopy.copy_model(model, mode='state_dict') print(f"state_dict复制成功! 参数数量: {sum(p.numel() for p in model_copy.parameters())}") # 安全复制 - deepcopy模式（自动清理中间状态） model_copy2 = SafeDeepCopy.copy_model(model, mode='deepcopy') print(f"deepcopy复制成功! 参数数量: {sum(p.numel() for p in model_copy2.parameters())}") # 验证参数相同 for (n1, p1), (n2, p2) in zip(model.named_parameters(), model_copy.named_parameters()): assert torch.equal(p1, p2), f"参数 {n1} 不匹配" print("参数验证通过!") def demo_dataloader(): """演示安全DataLoader""" print("\n" + "=" * 60) print("演示: 安全DataLoader") print("=" * 60) from torch.utils.data import Dataset, DataLoader class TensorDataset(Dataset): def __init__(self, data, labels): self.data = data self.labels = labels def __len__(self): return len(self.data) def __getitem__(self, idx): return self.data[idx], self.labels[idx] # 创建包含计算图的数据 x = torch.randn(100, 10, requires_grad=True) y = x * 2 # 非叶节点tensor labels = torch.randint(0, 5, (100,)) dataset = TensorDataset(y, labels) # 使用安全包装器 safe_loader = SafeDataLoaderWrapper(dataset, batch_size=16, num_workers=0) for batch_x, batch_y in safe_loader: print(f"Batch: x={batch_x.shape}, y={batch_y.shape}") print(f" x.requires_grad={batch_x.requires_grad}") break print("DataLoader测试通过!") if __name__ == "__main__": demo_tensor_copy() demo_model_copy() demo_dataloader() print("\n" + "=" * 60) print("所有演示完成!") print("=" * 60)

常见陷阱与注意事项

1.detach()与clone()的区别

x = torch.randn(10, requires_grad=True) y = x * 2 # detach(): 分离计算图，但共享数据存储 y_detached = y.detach() y_detached[0] = 100 # 会修改y的数据！ # clone(): 创建新数据，但保留计算图 y_cloned = y.clone() y_cloned[0] = 100 # 不会修改y # detach().clone(): 分离计算图 + 创建新数据（最安全） y_safe = y.detach().clone() y_safe[0] = 100 # 不会修改y，且不在计算图中

2.copy.copy()vscopy.deepcopy()

import copy x = torch.randn(10, requires_grad=True) y = x * 2 # copy.copy() (浅拷贝): 对tensor返回同一个对象 y_shallow = copy.copy(y) print(y_shallow is y) # True，浅拷贝对tensor没有效果 # copy.deepcopy() (深拷贝): 尝试复制所有内容，但对非叶节点报错 # y_deep = copy.deepcopy(y) # RuntimeError!

3. 模型的__getstate__和__setstate__

如果自定义模型类实现了__getstate__和__setstate__，需要确保不序列化非叶节点 tensor：

class SafeModel(nn.Module): def __init__(self): super().__init__() self.linear = nn.Linear(10, 5) self._cache = None def __getstate__(self): state = self.__dict__.copy() # 移除非叶节点tensor if state.get('_cache') is not None: if isinstance(state['_cache'], torch.Tensor) and state['_cache'].grad_fn is not None: state['_cache'] = state['_cache'].detach() return state def __setstate__(self, state): self.__dict__.update(state)

4. 多进程中的 pickle 问题

import torch import pickle x = torch.randn(10, requires_grad=True) y = x * 2 # 非叶节点 # pickle非叶节点tensor会失败 try: data = pickle.dumps(y) except RuntimeError as e: print(f"pickle失败: {e}") # 解决方案：先detach data = pickle.dumps(y.detach()) y_loaded = pickle.loads(data) print(f"pickle成功: {y_loaded.shape}")

5. EMA（指数移动平均）模型更新

在实现 EMA 时，需要安全地复制模型参数：

import copy class EMA: def __init__(self, model, decay=0.999): self.decay = decay # 使用state_dict安全复制 self.shadow = {} for name, param in model.named_parameters(): self.shadow[name] = param.detach().clone() def update(self, model): for name, param in model.named_parameters(): self.shadow[name] = ( self.decay * self.shadow[name] + (1 - self.decay) * param.detach() ) def apply_to(self, model): for name, param in model.named_parameters(): param.data.copy_(self.shadow[name])

6. 不要在forward中存储非叶节点 tensor

# 错误做法：在forward中存储中间结果 class BadModel(nn.Module): def __init__(self): super().__init__() self.linear = nn.Linear(10, 5) self.intermediate = None def forward(self, x): out = self.linear(x) self.intermediate = out # 非叶节点，会导致deepcopy失败 return out # 正确做法：存储detach的版本 class GoodModel(nn.Module): def __init__(self): super().__init__() self.linear = nn.Linear(10, 5) self.intermediate = None def forward(self, x): out = self.linear(x) # 存储detach版本，不影响deepcopy self.intermediate = out.detach() return out

7. 使用register_buffer存储持久状态

class ModelWithBuffer(nn.Module): def __init__(self): super().__init__() self.linear = nn.Linear(10, 5) # 使用register_buffer存储不需要梯度的状态 self.register_buffer('running_mean', torch.zeros(10)) def forward(self, x): # 更新buffer（不会产生计算图问题） self.running_mean = 0.9 * self.running_mean + 0.1 * x.mean(dim=0).detach() return self.linear(x - self.running_mean) # register_buffer的tensor会被state_dict正确处理 model = ModelWithBuffer() model_copy = copy.deepcopy(model) # 不会报错

总结

RuntimeError: Only Tensors created explicitly by the user (graph leaves) support the deepcopy protocol at the moment错误的核心原因是 PyTorch 的deepcopy协议只支持计算图的叶节点 tensor，不支持包含grad_fn的非叶节点 tensor。解决此问题的核心策略包括：

1. 使用detach()分离计算图：在deepcopy之前调用tensor.detach()将 tensor 从计算图中分离
2. 使用clone()代替deepcopy：对于 tensor，detach().clone()是更安全的复制方式
3. 使用state_dict复制模型：通过torch.save/torch.load复制模型参数，避免计算图问题
4. 清理模型中间状态：在deepcopy模型之前，清理模型内部缓存的非叶节点 tensor
5. 避免在forward中存储非叶节点：如果需要缓存中间结果，存储detach()版本

关键要点：

• 只有叶节点 tensor（用户显式创建且requires_grad=True）支持deepcopy
• detach()分离计算图但共享数据，clone()创建新数据但保留计算图，detach().clone()两者兼得
• 使用state_dict是复制模型最安全的方式，不会涉及计算图
• DataLoader 多进程中的 pickle 操作也会触发此错误，需要确保数据是 detach 的
• 使用register_buffer存储不需要梯度的持久状态，可以避免deepcopy问题

通过理解 PyTorch 计算图机制和正确使用detach、clone、state_dict等工具，可以有效避免deepcopy相关的错误。