YOLOv11【第八章：遥感与无人机航拍篇·第14节-下篇】FP16 半精度训练在遥感大数据集中的加速效果分析！

🏆本文收录于专栏 《YOLOv11实战：从入门到深度优化》。 本专栏围绕 YOLOv11 的改进、训练、部署与工程优化 展开，系统梳理并复现当前主流的 YOLOv11 实战案例与优化方案，内容目前已覆盖 分类、检测、分割、追踪、关键点、OBB 检测 等多个方向。 整体坚持 持续更新 + 深度解析 + 工程导向 的写作思路，不仅关注模型结构本身，也关注训练策略、损失函数设计、推理加速、部署适配以及真实项目中的问题排查。部分章节还会结合国内外前沿论文与 AIGC 大模型技术，对主流改进方案进行重构与再设计。

🎯当前专栏限时优惠中：一次订阅，终身有效，后续更新内容均可免费解锁 👉 点此查看专栏详情 👈️   🎉本专栏还不够过瘾？别急，好戏才刚刚开始！我已经为你准备了一整套 YOLO 进阶实战大礼包🎁：

👉《YOLOv8实战》 👉《YOLOv9实战》 👉《YOLOv10实战》 👉《YOLOv11实战》 👉《YOLOv12实战》 👉以及最新上线的 《YOLOv26实战》

想一次搞定所有版本？直接冲 《YOLO全栈实战合集》，一站式涵盖 YOLO 各版本实战教学！

🚀想学哪个版本？直接找 bug 菌“许愿”，安排！必须安排！🚀

🎯 本文定位：目标检测 × YOLOv11 遥感与无人机航拍篇 📅 预计阅读时间：约60～90分钟 ⭐ 难度等级：⭐⭐⭐⭐☆（高级） 🔧 技术栈：Ultralytics YOLO11 | Python v3.9+ | PyTorch v2.0+ | torchvision v0.9+ | Ultralytics v8.x | CUDA v11.8+

全文目录：

• • 5️⃣续接上篇
  • 6️⃣ 端到端工程化实践
  • • 6.1 分布式训练中的混合精度
    • 6.2 TensorRT FP16 推理部署
    • 6.3 完整的训练监控与日志系统
  • 7️⃣ 性能瓶颈分析与优化
  • • 7.1 Profiling工具的使用
    • 7.2 数据加载与预处理的并行化
  • 8️⃣ 案例研究：超大规模卫星影像训练
  • • 8.1 100万张遥感图像的训练实战
  • 📝 本章总结
  • • 核心知识点
    • 关键性能指标
    • 最佳实践建议
  • 9️⃣ 常见问题与解决方案
  • • 9.1 训练不收敛问题排查
    • 9.2 FP16训练常见错误速查表
  • 🔟 未来展望与进阶方向
  • • 10.1 BF16与FP8的新兴技术
    • 10.2 自适应精度训练
  • 📚 参考资料与延伸阅读
  • • 官方文档
    • 经典论文
    • 遥感数据集
    • 工具与框架
  • 💬 本章结语
  • 🧧🧧 文末福利，等你来拿！🧧🧧
  • 🫵 Who am I?

5️⃣续接上篇

由于平台有字数限制，本期内容分为上下两篇，具体请看如下：

• 上篇：YOLOv11【第八章：遥感与无人机航拍篇·第14节-上篇】FP16 半精度训练在遥感大数据集中的加速效果分析！
• 下篇：YOLOv11【第八章：遥感与无人机航拍篇·第14节-下篇】FP16 半精度训练在遥感大数据集中的加速效果分析！

6️⃣ 端到端工程化实践

6.1 分布式训练中的混合精度

在多GPU环境下实现高效的混合精度分布式训练 🌐

import torch.distributed as dist
import torch.multiprocessing as mp
from torch.nn.parallel import DistributedDataParallel as DDP
from torch.utils.data.distributed import DistributedSampler

class DistributedAMPTrainer:
"""
分布式混合精度训练器
"""

def __init__(self, rank, world_size, args):
"""
初始化分布式训练器

参数:
rank: 当前进程的rank
world_size: 总进程数
args: 训练参数
"""
self.rank = rank
self.world_size = world_size
self.args = args

# 设置设备
self.device = torch.device(f'cuda:{rank}')
torch.cuda.set_device(self.device)

# 初始化进程组
self.setup_distributed()

# 创建模型
self.model = self.build_model()
self.model = DDP(self.model, device_ids=[rank])

# 优化器和缩放器
self.optimizer = torch.optim.AdamW(self.model.parameters(), lr=args.lr)
self.scaler = GradScaler()

if rank == 0:
print(f"\\n✅ 分布式训练初始化完成")
print(f" 进程数: {world_size}")
print(f" 混合精度: 启用")

def setup_distributed(self):
"""
设置分布式环境
"""
# 初始化进程组
dist.init_process_group(
backend='nccl',
init_method='env://',
world_size=self.world_size,
rank=self.rank
)

# 设置随机种子
torch.manual_seed(42 + self.rank)
np.random.seed(42 + self.rank)

def build_model(self):
"""
构建模型
"""
model = nn.Sequential(
nn.Conv2d(3, 64, 3, padding=1),
nn.BatchNorm2d(64),
nn.SiLU(),
nn.Conv2d(64, 128, 3, stride=2, padding=1),
nn.BatchNorm2d(128),
nn.SiLU(),
nn.Conv2d(128, 256, 3, stride=2, padding=1),
nn.BatchNorm2d(256),
nn.SiLU(),
nn.AdaptiveAvgPool2d(1),
nn.Flatten(),
nn.Linear(256, 15)
).to(self.device)

return model

def train_epoch(self, dataloader, epoch):
"""
训练一个epoch
"""
self.model.train()
dataloader.sampler.set_epoch(epoch) # 确保每个epoch的数据打乱不同

total_loss = 0.0
num_batches = 0

for batch_idx, (images, targets) in enumerate(dataloader):
images = images.to(self.device)
targets = targets.to(self.device)

# 混合精度训练
with autocast():
outputs = self.model(images)
loss = outputs.abs().mean() # 简化损失

self.optimizer.zero_grad()
self.scaler.scale(loss).backward()
self.scaler.step(self.optimizer)
self.scaler.update()

total_loss += loss.item()
num_batches += 1

# 只在主进程打印
if self.rank == 0 and (batch_idx + 1) % 10 == 0:
avg_loss = total_loss / num_batches
print(f" Epoch {epoch} [{batch_idx+1}/{len(dataloader)}] "
f"Loss: {avg_loss:.4f}")

# 同步所有进程的损失
avg_loss = total_loss / num_batches
avg_loss_tensor = torch.tensor(avg_loss, device=self.device)
dist.all_reduce(avg_loss_tensor, op=dist.ReduceOp.SUM)
avg_loss = avg_loss_tensor.item() / self.world_size

return avg_loss

def cleanup(self):
"""
清理分布式环境
"""
dist.destroy_process_group()

def distributed_training_worker(rank, world_size, args):
"""
分布式训练工作进程
"""
# 创建训练器
trainer = DistributedAMPTrainer(rank, world_size, args)

# 创建数据集和数据加载器
dataset = RemoteSensingDataset(num_samples=1000, img_size=640)
sampler = DistributedSampler(
dataset,
num_replicas=world_size,
rank=rank,
shuffle=True
)

dataloader = DataLoader(
dataset,
batch_size=args.batch_size,
sampler=sampler,
num_workers=4,
pin_memory=True
)

# 训练
for epoch in range(args.epochs):
avg_loss = trainer.train_epoch(dataloader, epoch)

if rank == 0:
print(f"\\n✅ Epoch {epoch} 完成 | 平均损失: {avg_loss:.4f}")

# 清理
trainer.cleanup()

def launch_distributed_training():
"""
启动分布式训练
"""
print("\\n" + "="*70)
print("分布式混合精度训练")
print("="*70)

# 检查GPU数量
if not torch.cuda.is_available():
print("⚠️ 未检测到CUDA设备")
return

world_size = torch.cuda.device_count()

if world_size < 2:
print(f"⚠️ 只检测到{world_size}个GPU,分布式训练需要至少2个GPU")
print(" 将使用单GPU演示")
world_size = 1

print(f"\\n🖥️ 检测到{world_size}个GPU")

# 训练参数
class Args:
epochs = 3
batch_size = 8
lr = 0.001

args = Args()

if world_size > 1:
# 设置环境变量
os.environ['MASTER_ADDR'] = 'localhost'
os.environ['MASTER_PORT'] = '12355'

# 启动多进程
mp.spawn(
distributed_training_worker,
args=(world_size, args),
nprocs=world_size,
join=True
)
else:
print(" 使用单GPU模式")
# 单GPU训练
distributed_training_worker(0, 1, args)

# 注意: 分布式训练需要在主程序中调用
# 这里仅作演示,实际使用时需要在单独的脚本中运行
print("\\n💡 分布式训练示例代码已准备就绪")
print(" 实际使用时,请在命令行运行:")
print(" python -m torch.distributed.launch –nproc_per_node=4 train.py")

6.2 TensorRT FP16 推理部署

将训练好的FP16模型部署到TensorRT进行高效推理 🚀

class TensorRTDeployment:
"""
TensorRT FP16推理部署
"""

def __init__(self, onnx_path, engine_path=None):
"""
初始化TensorRT部署

参数:
onnx_path: ONNX模型路径
engine_path: TensorRT引擎保存路径
"""
self.onnx_path = onnx_path
self.engine_path = engine_path or onnx_path.replace('.onnx', '_fp16.engine')

try:
import tensorrt as trt
self.trt = trt
print("✅ TensorRT已安装")
except ImportError:
print("⚠️ TensorRT未安装,请先安装TensorRT")
self.trt = None

def export_to_onnx(self, model, input_shape=(1, 3, 640, 640)):
"""
导出PyTorch模型为ONNX格式
"""
print("\\n" + "="*70)
print("导出ONNX模型")
print("="*70)

model.eval()
device = next(model.parameters()).device

# 创建示例输入
dummy_input = torch.randn(*input_shape, device=device)

# 导出ONNX
torch.onnx.export(
model,
dummy_input,
self.onnx_path,
export_params=True,
opset_version=13,
do_constant_folding=True,
input_names=['input'],
output_names=['output'],
dynamic_axes={
'input': {0: 'batch_size'},
'output': {0: 'batch_size'}
}
)

print(f"✅ ONNX模型已保存: {self.onnx_path}")

# 验证ONNX模型
import onnx
onnx_model = onnx.load(self.onnx_path)
onnx.checker.check_model(onnx_model)
print("✅ ONNX模型验证通过")

def build_tensorrt_engine(self, fp16=True, workspace_size=4):
"""
构建TensorRT引擎

参数:
fp16: 是否使用FP16精度
workspace_size: 工作空间大小(GB)
"""
if self.trt is None:
print("⚠️ TensorRT未安装,跳过引擎构建")
return None

print("\\n" + "="*70)
print(f"构建TensorRT引擎 ({'FP16' if fp16 else 'FP32'})")
print("="*70)

# 创建builder
logger = self.trt.Logger(self.trt.Logger.INFO)
builder = self.trt.Builder(logger)
network = builder.create_network(
1 << int(self.trt.NetworkDefinitionCreationFlag.EXPLICIT_BATCH)
)
parser = self.trt.OnnxParser(network, logger)

# 解析ONNX模型
print(f"📂 加载ONNX模型: {self.onnx_path}")
with open(self.onnx_path, 'rb') as f:
if not parser.parse(f.read()):
print("❌ ONNX解析失败:")
for error in range(parser.num_errors):
print(f" {parser.get_error(error)}")
return None

print("✅ ONNX模型解析成功")

# 配置builder
config = builder.create_builder_config()
config.max_workspace_size = workspace_size * (1 << 30) # GB转字节

if fp16:
config.set_flag(self.trt.BuilderFlag.FP16)
print("✅ FP16模式已启用")

# 构建引擎
print("🔨 开始构建引擎(可能需要几分钟)…")
engine = builder.build_engine(network, config)

if engine is None:
print("❌ 引擎构建失败")
return None

# 保存引擎
with open(self.engine_path, 'wb') as f:
f.write(engine.serialize())

print(f"✅ TensorRT引擎已保存: {self.engine_path}")

return engine

def benchmark_inference(self, engine_path=None, num_iterations=100):
"""
基准测试推理性能
"""
if self.trt is None:
print("⚠️ TensorRT未安装,跳过基准测试")
return

engine_path = engine_path or self.engine_path

print("\\n" + "="*70)
print("TensorRT推理基准测试")
print("="*70)

# 加载引擎
logger = self.trt.Logger(self.trt.Logger.WARNING)
with open(engine_path, 'rb') as f:
runtime = self.trt.Runtime(logger)
engine = runtime.deserialize_cuda_engine(f.read())

# 创建执行上下文
context = engine.create_execution_context()

# 准备输入输出缓冲区
import pycuda.driver as cuda
import pycuda.autoinit

# 获取输入输出形状
input_shape = engine.get_binding_shape(0)
output_shape = engine.get_binding_shape(1)

# 分配内存
h_input = cuda.pagelocked_empty(
np.prod(input_shape), dtype=np.float32
)
h_output = cuda.pagelocked_empty(
np.prod(output_shape), dtype=np.float32
)

d_input = cuda.mem_alloc(h_input.nbytes)
d_output = cuda.mem_alloc(h_output.nbytes)

stream = cuda.Stream()

# 预热
for _ in range(10):
cuda.memcpy_htod_async(d_input, h_input, stream)
context.execute_async_v2(
bindings=[int(d_input), int(d_output)],
stream_handle=stream.handle
)
cuda.memcpy_dtoh_async(h_output, d_output, stream)
stream.synchronize()

# 基准测试
times = []
for _ in range(num_iterations):
start = time.time()

cuda.memcpy_htod_async(d_input, h_input, stream)
context.execute_async_v2(
bindings=[int(d_input), int(d_output)],
stream_handle=stream.handle
)
cuda.memcpy_dtoh_async(h_output, d_output, stream)
stream.synchronize()

times.append(time.time() – start)

# 统计结果
avg_time = np.mean(times) * 1000 # 转换为毫秒
std_time = np.std(times) * 1000
fps = 1000 / avg_time

print(f"\\n📊 推理性能:")
print(f" 平均延迟: {avg_time:.2f} ± {std_time:.2f} ms")
print(f" FPS: {fps:.2f}")
print(f" 最小延迟: {min(times)*1000:.2f} ms")
print(f" 最大延迟: {max(times)*1000:.2f} ms")

return {
'avg_time': avg_time,
'std_time': std_time,
'fps': fps
}

# TensorRT部署演示
def demo_tensorrt_deployment():
"""
演示TensorRT FP16部署
"""
print("\\n" + "="*70)
print("TensorRT FP16部署演示")
print("="*70)

# 创建简单模型
model = nn.Sequential(
nn.Conv2d(3, 64, 3, padding=1),
nn.BatchNorm2d(64),
nn.ReLU(),
nn.Conv2d(64, 128, 3, stride=2, padding=1),
nn.BatchNorm2d(128),
nn.ReLU(),
nn.AdaptiveAvgPool2d(1),
nn.Flatten(),
nn.Linear(128, 15)
)

if torch.cuda.is_available():
model = model.cuda()

# 创建部署器
deployer = TensorRTDeployment('yolov11_remote_sensing.onnx')

# 导出ONNX
deployer.export_to_onnx(model, input_shape=(1, 3, 640, 640))

print("\\n💡 TensorRT部署流程:")
print(" 1. 导出ONNX模型 ✅")
print(" 2. 构建TensorRT引擎 (需要安装TensorRT)")
print(" 3. 运行推理基准测试")
print("\\n 完整部署请参考TensorRT官方文档")

# 执行演示
demo_tensorrt_deployment()

6.3 完整的训练监控与日志系统

建立完善的训练监控和日志记录系统 📈

import json
from datetime import datetime
import matplotlib.pyplot as plt
from pathlib import Path

class TrainingMonitor:
"""
训练监控与日志系统
"""

def __init__(self, log_dir='runs/monitor', experiment_name=None):
"""
初始化监控系统

参数:
log_dir: 日志保存目录
experiment_name: 实验名称
"""
self.log_dir = Path(log_dir)
self.experiment_name = experiment_name or datetime.now().strftime('%Y%m%d_%H%M%S')
self.exp_dir = self.log_dir / self.experiment_name
self.exp_dir.mkdir(parents=True, exist_ok=True)

# 初始化日志文件
self.log_file = self.exp_dir / 'training.log'
self.metrics_file = self.exp_dir / 'metrics.json'

# 训练指标
self.metrics = {
'train_loss': [],
'val_loss': [],
'learning_rate': [],
'gpu_memory': [],
'batch_time': [],
'scale_factor': [],
'overflow_count': [],
'epoch_time': []
}

# 系统信息
self.system_info = self.collect_system_info()
self.save_system_info()

print(f"\\n📊 训练监控系统已初始化")
print(f" 实验名称: {self.experiment_name}")
print(f" 日志目录: {self.exp_dir}")

def collect_system_info(self):
"""
收集系统信息
"""
info = {
'timestamp': datetime.now().isoformat(),
'pytorch_version': torch.__version__,
'cuda_available': torch.cuda.is_available(),
}

if torch.cuda.is_available():
info.update({
'cuda_version': torch.version.cuda,
'cudnn_version': torch.backends.cudnn.version(),
'gpu_count': torch.cuda.device_count(),
'gpu_names': [torch.cuda.get_device_name(i)
for i in range(torch.cuda.device_count())],
'gpu_memory': [torch.cuda.get_device_properties(i).total_memory / 1024**3
for i in range(torch.cuda.device_count())]
})

return info

def save_system_info(self):
"""
保存系统信息
"""
info_file = self.exp_dir / 'system_info.json'
with open(info_file, 'w', encoding='utf-8') as f:
json.dump(self.system_info, f, indent=2, ensure_ascii=False)

def log(self, message, level='INFO'):
"""
记录日志

参数:
message: 日志消息
level: 日志级别
"""
timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
log_message = f"[{timestamp}] [{level}] {message}"

# 打印到控制台
print(log_message)

# 写入文件
with open(self.log_file, 'a', encoding='utf-8') as f:
f.write(log_message + '\\n')

def update_metrics(self, **kwargs):
"""
更新训练指标

参数:
**kwargs: 指标名称和值的键值对
"""
for key, value in kwargs.items():
if key in self.metrics:
self.metrics[key].append(value)
else:
self.metrics[key] = [value]

# 实时保存指标
self.save_metrics()

def save_metrics(self):
"""
保存指标到JSON文件
"""
with open(self.metrics_file, 'w', encoding='utf-8') as f:
json.dump(self.metrics, f, indent=2)

def plot_metrics(self, save_path=None):
"""
绘制训练指标图表
"""
save_path = save_path or self.exp_dir / 'training_metrics.png'

# 创建子图
fig = plt.figure(figsize=(16, 12))
gs = fig.add_gridspec(3, 3, hspace=0.3, wspace=0.3)

# 1. 损失曲线
ax1 = fig.add_subplot(gs[0, :2])
if self.metrics['train_loss']:
ax1.plot(self.metrics['train_loss'], label='训练损失',
linewidth=2, alpha=0.8)
if self.metrics['val_loss']:
ax1.plot(self.metrics['val_loss'], label='验证损失',
linewidth=2, alpha=0.8)
ax1.set_xlabel('Epoch', fontsize=11)
ax1.set_ylabel('损失值', fontsize=11)
ax1.set_title('损失曲线', fontsize=13, fontweight='bold')
ax1.legend(fontsize=10)
ax1.grid(True, alpha=0.3)

# 2. 学习率变化
ax2 = fig.add_subplot(gs[0, 2])
if self.metrics['learning_rate']:
ax2.plot(self.metrics['learning_rate'],
color='green', linewidth=2)
ax2.set_xlabel('Epoch', fontsize=11)
ax2.set_ylabel('学习率', fontsize=11)
ax2.set_title('学习率变化', fontsize=13, fontweight='bold')
ax2.grid(True, alpha=0.3)

# 3. GPU显存占用
ax3 = fig.add_subplot(gs[1, 0])
if self.metrics['gpu_memory']:
ax3.plot(self.metrics['gpu_memory'],
color='orange', linewidth=2)
ax3.set_xlabel('Batch', fontsize=11)
ax3.set_ylabel('显存 (GB)', fontsize=11)
ax3.set_title('GPU显存占用', fontsize=13, fontweight='bold')
ax3.grid(True, alpha=0.3)

# 4. 批次处理时间
ax4 = fig.add_subplot(gs[1, 1])
if self.metrics['batch_time']:
ax4.plot(self.metrics['batch_time'],
color='purple', linewidth=1, alpha=0.6)
# 添加移动平均线
if len(self.metrics['batch_time']) > 10:
window = 10
moving_avg = np.convolve(
self.metrics['batch_time'],
np.ones(window)/window,
mode='valid'
)
ax4.plot(range(window–1, len(self.metrics['batch_time'])),
moving_avg, color='red', linewidth=2,
label=f'{window}步移动平均')
ax4.legend(fontsize=9)
ax4.set_xlabel('Batch', fontsize=11)
ax4.set_ylabel('时间 (秒)', fontsize=11)
ax4.set_title('批次处理时间', fontsize=13, fontweight='bold')
ax4.grid(True, alpha=0.3)

# 5. 损失缩放因子
ax5 = fig.add_subplot(gs[1, 2])
if self.metrics['scale_factor']:
ax5.plot(self.metrics['scale_factor'],
color='brown', linewidth=2)
ax5.set_xlabel('Batch', fontsize=11)
ax5.set_ylabel('缩放因子', fontsize=11)
ax5.set_title('损失缩放因子', fontsize=13, fontweight='bold')
ax5.set_yscale('log')
ax5.grid(True, alpha=0.3)

# 6. 溢出统计
ax6 = fig.add_subplot(gs[2, 0])
if self.metrics['overflow_count']:
cumulative_overflow = np.cumsum(self.metrics['overflow_count'])
ax6.plot(cumulative_overflow, color='red', linewidth=2)
ax6.set_xlabel('Batch', fontsize=11)
ax6.set_ylabel('累积溢出次数', fontsize=11)
ax6.set_title('梯度溢出统计', fontsize=13, fontweight='bold')
ax6.grid(True, alpha=0.3)

# 7. Epoch耗时
ax7 = fig.add_subplot(gs[2, 1])
if self.metrics['epoch_time']:
epochs = range(1, len(self.metrics['epoch_time']) + 1)
ax7.bar(epochs, self.metrics['epoch_time'],
color='teal', alpha=0.7, edgecolor='black')
ax7.set_xlabel('Epoch', fontsize=11)
ax7.set_ylabel('时间 (分钟)', fontsize=11)
ax7.set_title('每Epoch耗时', fontsize=13, fontweight='bold')
ax7.grid(True, alpha=0.3, axis='y')

# 8. 训练统计摘要
ax8 = fig.add_subplot(gs[2, 2])
ax8.axis('off')

summary_text = "训练统计摘要\\n" + "="*30 + "\\n\\n"

if self.metrics['train_loss']:
summary_text += f"最终训练损失: {self.metrics['train_loss'][–1]:.4f}\\n"
if self.metrics['val_loss']:
summary_text += f"最终验证损失: {self.metrics['val_loss'][–1]:.4f}\\n"
if self.metrics['batch_time']:
avg_batch_time = np.mean(self.metrics['batch_time'])
summary_text += f"平均批次时间: {avg_batch_time:.3f}s\\n"
if self.metrics['overflow_count']:
total_overflow = sum(self.metrics['overflow_count'])
summary_text += f"总溢出次数: {total_overflow}\\n"
if self.metrics['epoch_time']:
total_time = sum(self.metrics['epoch_time'])
summary_text += f"总训练时间: {total_time:.1f}分钟\\n"

ax8.text(0.1, 0.5, summary_text, fontsize=10,
verticalalignment='center', family='monospace',
bbox=dict(boxstyle='round', facecolor='wheat', alpha=0.3))

plt.savefig(save_path, dpi=300, bbox_inches='tight')
self.log(f"训练指标图表已保存: {save_path}")

return fig

def generate_report(self):
"""
生成训练报告
"""
report_path = self.exp_dir / 'training_report.txt'

with open(report_path, 'w', encoding='utf-8') as f:
f.write("="*70 + "\\n")
f.write("YOLOv11 遥感数据集 FP16 训练报告\\n")
f.write("="*70 + "\\n\\n")

# 实验信息
f.write("实验信息\\n")
f.write("-"*70 + "\\n")
f.write(f"实验名称: {self.experiment_name}\\n")
f.write(f"开始时间: {self.system_info['timestamp']}\\n")
f.write(f"PyTorch版本: {self.system_info['pytorch_version']}\\n")

if self.system_info['cuda_available']:
f.write(f"CUDA版本: {self.system_info['cuda_version']}\\n")
f.write(f"GPU数量: {self.system_info['gpu_count']}\\n")
for i, name in enumerate(self.system_info['gpu_names']):
memory = self.system_info['gpu_memory'][i]
f.write(f"GPU {i}: {name} ({memory:.1f} GB)\\n")

f.write("\\n")

# 训练统计
f.write("训练统计\\n")
f.write("-"*70 + "\\n")

if self.metrics['train_loss']:
f.write(f"训练轮数: {len(self.metrics['train_loss'])}\\n")
f.write(f"初始训练损失: {self.metrics['train_loss'][0]:.6f}\\n")
f.write(f"最终训练损失: {self.metrics['train_loss'][–1]:.6f}\\n")
improvement = (self.metrics['train_loss'][0] –
self.metrics['train_loss'][–1]) / self.metrics['train_loss'][0] * 100
f.write(f"损失改善: {improvement:.2f}%\\n")

if self.metrics['val_loss']:
f.write(f"最终验证损失: {self.metrics['val_loss'][–1]:.6f}\\n")
best_val_loss = min(self.metrics['val_loss'])
best_epoch = self.metrics['val_loss'].index(best_val_loss) + 1
f.write(f"最佳验证损失: {best_val_loss:.6f} (Epoch {best_epoch})\\n")

f.write("\\n")

# 性能统计
f.write("性能统计\\n")
f.write("-"*70 + "\\n")

if self.metrics['batch_time']:
avg_time = np.mean(self.metrics['batch_time'])
std_time = np.std(self.metrics['batch_time'])
f.write(f"平均批次时间: {avg_time:.3f} ± {std_time:.3f} 秒\\n")
f.write(f"最快批次: {min(self.metrics['batch_time']):.3f} 秒\\n")
f.write(f"最慢批次: {max(self.metrics['batch_time']):.3f} 秒\\n")

if self.metrics['epoch_time']:
total_time = sum(self.metrics['epoch_time'])
avg_epoch_time = np.mean(self.metrics['epoch_time'])
f.write(f"总训练时间: {total_time:.1f} 分钟 ({total_time/60:.2f} 小时)\\n")
f.write(f"平均每轮时间: {avg_epoch_time:.1f} 分钟\\n")

if self.metrics['gpu_memory']:
avg_memory = np.mean(self.metrics['gpu_memory'])
max_memory = max(self.metrics['gpu_memory'])
f.write(f"平均显存占用: {avg_memory:.2f} GB\\n")
f.write(f"峰值显存占用: {max_memory:.2f} GB\\n")

f.write("\\n")

# 混合精度统计
f.write("混合精度统计\\n")
f.write("-"*70 + "\\n")

if self.metrics['scale_factor']:
initial_scale = self.metrics['scale_factor'][0]
final_scale = self.metrics['scale_factor'][–1]
f.write(f"初始缩放因子: {initial_scale:.0f}\\n")
f.write(f"最终缩放因子: {final_scale:.0f}\\n")

if self.metrics['overflow_count']:
total_overflow = sum(self.metrics['overflow_count'])
total_batches = len(self.metrics['overflow_count'])
overflow_rate = total_overflow / total_batches * 100
f.write(f"总溢出次数: {total_overflow}\\n")
f.write(f"溢出率: {overflow_rate:.2f}%\\n")

f.write("\\n")
f.write("="*70 + "\\n")

self.log(f"训练报告已生成: {report_path}")

return report_path

# 完整训练监控演示
def demo_training_monitor():
"""
演示完整的训练监控系统
"""
print("\\n" + "="*70)
print("训练监控系统演示")
print("="*70)

# 创建监控器
monitor = TrainingMonitor(
log_dir='runs/monitor_demo',
experiment_name='yolov11_dota_fp16'
)

# 模拟训练过程
num_epochs = 5
batches_per_epoch = 50

monitor.log("开始训练", level='INFO')

for epoch in range(num_epochs):
epoch_start = time.time()
monitor.log(f"Epoch {epoch+1}/{num_epochs} 开始", level='INFO')

epoch_loss = 0.0

for batch in range(batches_per_epoch):
batch_start = time.time()

# 模拟训练
loss = 2.0 * np.exp(–epoch * 0.3) + np.random.rand() * 0.1
lr = 0.01 * (0.95 ** epoch)

if torch.cuda.is_available():
memory = torch.cuda.max_memory_allocated() / 1024**3
torch.cuda.reset_peak_memory_stats()
else:
memory = 0

scale_factor = 65536 * (0.9 ** (epoch * batches_per_epoch + batch) // 100)
overflow = 1 if np.random.rand() < 0.01 else 0

batch_time = time.time() – batch_start + np.random.rand() * 0.01

# 更新指标
monitor.update_metrics(
train_loss=loss,
learning_rate=lr,
gpu_memory=memory,
batch_time=batch_time,
scale_factor=scale_factor,
overflow_count=overflow
)

epoch_loss += loss

# Epoch结束
avg_loss = epoch_loss / batches_per_epoch
epoch_time = (time.time() – epoch_start) / 60 # 转换为分钟

# 验证
val_loss = avg_loss * 0.9 + np.random.rand() * 0.05

monitor.update_metrics(
val_loss=val_loss,
epoch_time=epoch_time
)

monitor.log(
f"Epoch {epoch+1} 完成 | "
f"训练损失: {avg_loss:.4f} | "
f"验证损失: {val_loss:.4f} | "
f"耗时: {epoch_time:.1f}分钟",
level='INFO'
)

monitor.log("训练完成", level='INFO')

# 生成可视化和报告
monitor.plot_metrics()
monitor.generate_report()

print(f"\\n✅ 监控演示完成,结果保存在: {monitor.exp_dir}")

# 执行演示
demo_training_monitor()

7️⃣ 性能瓶颈分析与优化

7.1 Profiling工具的使用

使用PyTorch Profiler深入分析性能瓶颈 🔬

from torch.profiler import profile, record_function, ProfilerActivity

class PerformanceProfiler:
"""
性能分析工具
"""

def __init__(self, model, dataloader):
self.model = model
self.dataloader = dataloader
self.device = next(model.parameters()).device

def profile_training_step(self, use_amp=True, num_steps=10):
"""
分析训练步骤的性能
"""
print("\\n" + "="*70)
print(f"训练步骤性能分析 ({'FP16' if use_amp else 'FP32'})")
print("="*70)

self.model.train()
optimizer = torch.optim.Adam(self.model.parameters(), lr=0.001)
scaler = GradScaler() if use_amp else None

# 使用profiler
with profile(
activities=[ProfilerActivity.CPU, ProfilerActivity.CUDA],
record_shapes=True,
profile_memory=True,
with_stack=True
) as prof:
for step, (images, targets) in enumerate(self.dataloader):
if step >= num_steps:
break

images = images.to(self.device)
targets = targets.to(self.device)

with record_function("forward"):
if use_amp:
with autocast():
outputs = self.model(images)
loss = outputs.abs().mean()
else:
outputs = self.model(images)
loss = outputs.abs().mean()

with record_function("backward"):
optimizer.zero_grad()
if use_amp:
scaler.scale(loss).backward()
else:
loss.backward()

with record_function("optimizer_step"):
if use_amp:
scaler.step(optimizer)
scaler.update()
else:
optimizer.step()

# 打印统计信息
print("\\n📊 按操作类型统计:")
print(prof.key_averages().table(
sort_by="cuda_time_total",
row_limit=20
))

# 按函数统计
print("\\n📊 按函数统计:")
print(prof.key_averages(group_by_input_shape=True).table(
sort_by="cuda_time_total",
row_limit=15
))

# 导出Chrome trace
trace_path = f"trace_{'fp16' if use_amp else 'fp32'}.json"
prof.export_chrome_trace(trace_path)
print(f"\\n✅ Chrome trace已导出: {trace_path}")
print(f" 在Chrome浏览器中打开 chrome://tracing 加载该文件查看详细时间线")

return prof

def analyze_memory_usage(self, use_amp=True):
"""
分析显存使用情况
"""
print("\\n" + "="*70)
print(f"显存使用分析 ({'FP16' if use_amp else 'FP32'})")
print("="*70)

if not torch.cuda.is_available():
print("⚠️ 需要CUDA设备")
return

self.model.train()
optimizer = torch.optim.Adam(self.model.parameters(), lr=0.001)
scaler = GradScaler() if use_amp else None

# 重置显存统计
torch.cuda.reset_peak_memory_stats()
torch.cuda.empty_cache()

memory_stats = {
'before_forward': [],
'after_forward': [],
'after_backward': [],
'after_step': []
}

for step, (images, targets) in enumerate(self.dataloader):
if step >= 5:
break

images = images.to(self.device)
targets = targets.to(self.device)

# 前向传播前
memory_stats['before_forward'].append(
torch.cuda.memory_allocated() / 1024**2
)

# 前向传播
if use_amp:
with autocast():
outputs = self.model(images)
loss = outputs.abs().mean()
else:
outputs = self.model(images)
loss = outputs.abs().mean()

memory_stats['after_forward'].append(
torch.cuda.memory_allocated() / 1024**2
)

# 反向传播
optimizer.zero_grad()
if use_amp:
scaler.scale(loss).backward()
else:
loss.backward()

memory_stats['after_backward'].append(
torch.cuda.memory_allocated() / 1024**2
)

# 参数更新
if use_amp:
scaler.step(optimizer)
scaler.update()
else:
optimizer.step()

memory_stats['after_step'].append(
torch.cuda.memory_allocated() / 1024**2
)

# 统计分析
print("\\n📊 显存使用统计 (MB):")
print(f"{'阶段':<20} {'平均':<12} {'最大':<12} {'最小':<12}")
print("-" * 60)

for stage, values in memory_stats.items():
avg = np.mean(values)
max_val = np.max(values)
min_val = np.min(values)
print(f"{stage:<20} {avg:<12.2f} {max_val:<12.2f} {min_val:<12.2f}")

# 峰值显存
peak_memory = torch.cuda.max_memory_allocated() / 1024**2
print(f"\\n峰值显存: {peak_memory:.2f} MB")

# 可视化
self.visualize_memory_usage(memory_stats)

return memory_stats

def visualize_memory_usage(self, memory_stats):
"""
可视化显存使用
"""
fig, ax = plt.subplots(figsize=(12, 6))

stages = list(memory_stats.keys())
num_steps = len(memory_stats[stages[0]])

x = np.arange(num_steps)
width = 0.2

colors = ['#3498db', '#e74c3c', '#2ecc71', '#f39c12']

for i, (stage, values) in enumerate(memory_stats.items()):
offset = (i – len(stages)/2) * width
ax.bar(x + offset, values, width, label=stage,
color=colors[i], alpha=0.8, edgecolor='black')

ax.set_xlabel('训练步骤', fontsize=12)
ax.set_ylabel('显存占用 (MB)', fontsize=12)
ax.set_title('训练过程显存使用情况', fontsize=14, fontweight='bold')
ax.set_xticks(x)
ax.set_xticklabels([f'Step {i+1}' for i in range(num_steps)])
ax.legend(fontsize=10)
ax.grid(True, alpha=0.3, axis='y')

plt.tight_layout()
plt.savefig('memory_usage_analysis.png', dpi=300, bbox_inches='tight')
print("\\n✅ 显存使用图已保存为 memory_usage_analysis.png")

def compare_fp32_fp16_performance(self):
"""
对比FP32和FP16的性能
"""
print("\\n" + "="*70)
print("FP32 vs FP16 性能对比")
print("="*70)

# FP32分析
print("\\n🔵 FP32模式:")
fp32_prof = self.profile_training_step(use_amp=False, num_steps=10)
fp32_memory = self.analyze_memory_usage(use_amp=False)

# FP16分析
print("\\n🟢 FP16模式:")
fp16_prof = self.profile_training_step(use_amp=True, num_steps=10)
fp16_memory = self.analyze_memory_usage(use_amp=True)

# 对比总结
print("\\n" + "="*70)
print("性能对比总结")
print("="*70)

# 计算加速比
fp32_time = sum([event.cuda_time_total for event in fp32_prof.key_averages()])
fp16_time = sum([event.cuda_time_total for event in fp16_prof.key_averages()])
speedup = fp32_time / fp16_time

print(f"\\n⚡ 速度提升:")
print(f" FP32总时间: {fp32_time/1000:.2f} ms")
print(f" FP16总时间: {fp16_time/1000:.2f} ms")
print(f" 加速比: {speedup:.2f}x")

# 显存节省
fp32_peak = np.max([v for values in fp32_memory.values() for v in values])
fp16_peak = np.max([v for values in fp16_memory.values() for v in values])
memory_saving = (fp32_peak – fp16_peak) / fp32_peak * 100

print(f"\\n💾 显存节省:")
print(f" FP32峰值: {fp32_peak:.2f} MB")
print(f" FP16峰值: {fp16_peak:.2f} MB")
print(f" 节省: {memory_saving:.2f}%")

# 执行性能分析
def demo_performance_profiling():
"""
演示性能分析
"""
if not torch.cuda.is_available():
print("⚠️ 未检测到CUDA设备,跳过性能分析")
return

# 创建模型
model = nn.Sequential(
nn.Conv2d(3, 64, 3, padding=1),
nn.BatchNorm2d(64),
nn.SiLU(),
nn.Conv2d(64, 128, 3, stride=2, padding=1),
nn.BatchNorm2d(128),
nn.SiLU(),
nn.Conv2d(128, 256, 3, stride=2, padding=1),
nn.BatchNorm2d(256),
nn.SiLU(),
nn.AdaptiveAvgPool2d(1),
nn.Flatten(),
nn.Linear(256, 15)
).cuda()

# 创建数据加载器
dataset = RemoteSensingDataset(num_samples=50, img_size=256)
dataloader = DataLoader(dataset, batch_size=4, shuffle=False)

# 创建性能分析器
profiler = PerformanceProfiler(model, dataloader)

# 执行对比分析
profiler.compare_fp32_fp16_performance()

# 执行演示
demo_performance_profiling()

7.2 数据加载与预处理的并行化

优化数据加载流程,消除I/O瓶颈 📦

import multiprocessing as mp
from torch.utils.data import Dataset, DataLoader
import cv2

class OptimizedRemoteSensingDataset(Dataset):
"""
优化的遥感数据集加载器
"""

def __init__(self, image_paths, labels, img_size=640, augment=True, cache_images=False):
"""
初始化数据集

参数:
image_paths: 图像路径列表
labels: 标签列表
img_size: 图像尺寸
augment: 是否使用数据增强
cache_images: 是否缓存图像到内存
"""
self.image_paths = image_paths
self.labels = labels
self.img_size = img_size
self.augment = augment
self.cache_images = cache_images

# 图像缓存
self.cached_images = {}
if cache_images:
print(f"🔄 缓存{len(image_paths)}张图像到内存…")
for idx, path in enumerate(image_paths):
self.cached_images[idx] = self.load_image(path)
if (idx + 1) % 100 == 0:
print(f" 已缓存 {idx+1}/{len(image_paths)} 张图像")
print("✅ 图像缓存完成")

def __len__(self):
return len(self.image_paths)

def load_image(self, path):
"""
加载图像(模拟)
"""
# 实际使用时应该读取真实图像
# img = cv2.imread(path)
# img = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)

# 这里使用模拟数据
img = np.random.randint(0, 255, (self.img_size, self.img_size, 3), dtype=np.uint8)
return img

def augment_image(self, img):
"""
数据增强
"""
# 随机水平翻转
if np.random.rand() > 0.5:
img = np.fliplr(img)

# 随机垂直翻转
if np.random.rand() > 0.5:
img = np.flipud(img)

# 随机旋转
if np.random.rand() > 0.5:
angle = np.random.choice([90, 180, 270])
img = np.rot90(img, k=angle//90)

# 随机亮度调整
if np.random.rand() > 0.5:
factor = np.random.uniform(0.8, 1.2)
img = np.clip(img * factor, 0, 255).astype(np.uint8)

return img

def __getitem__(self, idx):
"""
获取数据项
"""
# 加载图像
if self.cache_images and idx in self.cached_images:
img = self.cached_images[idx].copy()
else:
img = self.load_image(self.image_paths[idx])

# 数据增强
if self.augment:
img = self.augment_image(img)

# 转换为tensor
img = torch.from_numpy(img).permute(2, 0, 1).float() / 255.0

# 获取标签
label = self.labels[idx]

return img, label

class DataLoadingOptimizer:
"""
数据加载优化器
"""

@staticmethod
def benchmark_dataloader_configs(dataset, configs):
"""
测试不同数据加载配置的性能

参数:
dataset: 数据集
configs: 配置列表
"""
print("\\n" + "="*70)
print("数据加载配置性能测试")
print("="*70)

results = {}

for config in configs:
print(f"\\n🔄 测试配置: {config['name']}")
print(f" num_workers: {config['num_workers']}")
print(f" batch_size: {config['batch_size']}")
print(f" pin_memory: {config['pin_memory']}")
print(f" prefetch_factor: {config.get('prefetch_factor', 2)}")

dataloader = DataLoader(
dataset,
batch_size=config['batch_size'],
num_workers=config['num_workers'],
pin_memory=config['pin_memory'],
prefetch_factor=config.get('prefetch_factor', 2),
persistent_workers=config['num_workers'] > 0
)

# 预热
for _ in range(5):
next(iter(dataloader))

# 测试
times = []
for _ in range(20):
start = time.time()
for batch in dataloader:
pass
times.append(time.time() – start)

avg_time = np.mean(times)
std_time = np.std(times)
throughput = len(dataset) / avg_time

results[config['name']] = {
'avg_time': avg_time,
'std_time': std_time,
'throughput': throughput
}

print(f" 平均耗时: {avg_time:.3f} ± {std_time:.3f} 秒")
print(f" 吞吐量: {throughput:.1f} 样本/秒")

# 可视化对比
DataLoadingOptimizer.visualize_dataloader_comparison(results)

return results

@staticmethod
def visualize_dataloader_comparison(results):
"""
可视化数据加载器对比
"""
fig, axes = plt.subplots(1, 2, figsize=(14, 5))

names = list(results.keys())
avg_times = [results[n]['avg_time'] for n in names]
throughputs = [results[n]['throughput'] for n in names]

colors = plt.cm.Set3(np.linspace(0, 1, len(names)))

# 加载时间对比
axes[0].bar(range(len(names)), avg_times, color=colors,
alpha=0.7, edgecolor='black')
axes[0].set_xticks(range(len(names)))
axes[0].set_xticklabels(names, rotation=15, ha='right')
axes[0].set_ylabel('加载时间 (秒)', fontsize=11)
axes[0].set_title('数据加载时间对比', fontsize=13, fontweight='bold')
axes[0].grid(True, alpha=0.3, axis='y')

# 吞吐量对比
axes[1].bar(range(len(names)), throughputs, color=colors,
alpha=0.7, edgecolor='black')
axes[1].set_xticks(range(len(names)))
axes[1].set_xticklabels(names, rotation=15, ha='right')
axes[1].set_ylabel('吞吐量 (样本/秒)', fontsize=11)
axes[1].set_title('数据加载吞吐量对比', fontsize=13, fontweight='bold')
axes[1].grid(True, alpha=0.3, axis='y')

plt.tight_layout()
plt.savefig('dataloader_comparison.png', dpi=300, bbox_inches='tight')
print("\\n✅ 数据加载对比图已保存为 dataloader_comparison.png")

@staticmethod
def find_optimal_num_workers():
"""
自动寻找最优的num_workers数量
"""
print("\\n" + "="*70)
print("寻找最优num_workers")
print("="*70)

# 创建测试数据集
image_paths = [f'image_{i}.jpg' for i in range(200)]
labels = [torch.randint(0, 15, (85,)) for _ in range(200)]
dataset = OptimizedRemoteSensingDataset(
image_paths, labels, img_size=256, cache_images=False
)

# 测试不同的num_workers
max_workers = mp.cpu_count()
test_workers = [0, 2, 4, 8, max_workers]
test_workers = [w for w in test_workers if w <= max_workers]

print(f"\\n🔍 CPU核心数: {max_workers}")
print(f" 测试范围: {test_workers}")

results = {}

for num_workers in test_workers:
print(f"\\n测试 num_workers={num_workers}…")

dataloader = DataLoader(
dataset,
batch_size=8,
num_workers=num_workers,
pin_memory=True,
persistent_workers=num_workers > 0
)

# 预热
for _ in range(3):
next(iter(dataloader))

# 测试
times = []
for _ in range(10):
start = time.time()
for batch in dataloader:
pass
times.append(time.time() – start)

avg_time = np.mean(times)
results[num_workers] = avg_time

print(f" 平均耗时: {avg_time:.3f} 秒")

# 找到最优值
optimal_workers = min(results, key=results.get)

print(f"\\n✅ 推荐num_workers: {optimal_workers}")
print(f" 最快耗时: {results[optimal_workers]:.3f} 秒")

# 可视化
fig, ax = plt.subplots(figsize=(10, 6))

workers = list(results.keys())
times = list(results.values())

ax.plot(workers, times, 'o-', linewidth=2, markersize=10, color='blue')
ax.axvline(x=optimal_workers, color='red', linestyle='–',
label=f'最优值: {optimal_workers}', linewidth=2)
ax.set_xlabel('num_workers', fontsize=12)
ax.set_ylabel('加载时间 (秒)', fontsize=12)
ax.set_title('num_workers对数据加载性能的影响', fontsize=14, fontweight='bold')
ax.legend(fontsize=11)
ax.grid(True, alpha=0.3)

plt.tight_layout()
plt.savefig('optimal_num_workers.png', dpi=300, bbox_inches='tight')
print("✅ 最优num_workers图已保存为 optimal_num_workers.png")

return optimal_workers

# 执行数据加载优化演示
def demo_data_loading_optimization():
"""
演示数据加载优化
"""
print("\\n" + "="*70)
print("数据加载优化演示")
print("="*70)

# 创建数据集
image_paths = [f'image_{i}.jpg' for i in range(500)]
labels = [torch.randint(0, 15, (85,)) for _ in range(500)]

# 测试不同配置
configs = [
{
'name': '基准配置',
'num_workers': 0,
'batch_size': 8,
'pin_memory': False
},
{
'name': '启用pin_memory',
'num_workers': 0,
'batch_size': 8,
'pin_memory': True
},
{
'name': '多线程加载',
'num_workers': 4,
'batch_size': 8,
'pin_memory': True
},
{
'name': '大批量+多线程',
'num_workers': 4,
'batch_size': 16,
'pin_memory': True
}
]

# 不缓存图像的测试
print("\\n📊 测试1: 不缓存图像")
dataset = OptimizedRemoteSensingDataset(
image_paths, labels, img_size=256, cache_images=False
)
DataLoadingOptimizer.benchmark_dataloader_configs(dataset, configs)

# 缓存图像的测试
print("\\n📊 测试2: 缓存图像到内存")
dataset_cached = OptimizedRemoteSensingDataset(
image_paths, labels, img_size=256, cache_images=True
)
DataLoadingOptimizer.benchmark_dataloader_configs(dataset_cached, configs)

# 寻找最优num_workers
DataLoadingOptimizer.find_optimal_num_workers()

# 执行演示
demo_data_loading_optimization()

8️⃣ 案例研究：超大规模卫星影像训练

8.1 100万张遥感图像的训练实战

真实场景下的大规模训练案例分析 🌍

class LargeScaleTrainingCase:
"""
大规模遥感数据集训练案例
"""

def __init__(self):
self.dataset_info = {
'数据集名称': 'Global Satellite Dataset',
'图像数量': 1000000,
'总数据量': '5 TB',
'平均图像尺寸': '2048×2048',
'类别数': 20,
'训练集': 800000,
'验证集': 150000,
'测试集': 50000
}

def analyze_training_requirements(self):
"""
分析训练需求
"""
print("\\n" + "="*70)
print("大规模训练需求分析")
print("="*70)

print("\\n📊 数据集信息:")
for key, value in self.dataset_info.items():
print(f" {key}: {value}")

# 计算训练时间估算
print("\\n⏱️ 训练时间估算:")

# 假设参数
batch_size = 16
img_size = 640
samples_per_epoch = self.dataset_info['训练集']
batches_per_epoch = samples_per_epoch // batch_size

# FP32训练
fp32_time_per_batch = 0.5 # 秒
fp32_epoch_time = batches_per_epoch * fp32_time_per_batch / 3600 # 小时
fp32_total_time = fp32_epoch_time * 300 # 300个epoch

# FP16训练
fp16_time_per_batch = 0.2 # 秒
fp16_epoch_time = batches_per_epoch * fp16_time_per_batch / 3600 # 小时
fp16_total_time = fp16_epoch_time * 300

print(f"\\n 批次大小: {batch_size}")
print(f" 每epoch批次数: {batches_per_epoch:,}")
print(f" 训练轮数: 300")

print(f"\\n FP32训练:")
print(f" 每批次耗时: {fp32_time_per_batch:.2f} 秒")
print(f" 每epoch耗时: {fp32_epoch_time:.1f} 小时")
print(f" 总训练时间: {fp32_total_time:.1f} 小时 ({fp32_total_time/24:.1f} 天)")

print(f"\\n FP16训练:")
print(f" 每批次耗时: {fp16_time_per_batch:.2f} 秒")
print(f" 每epoch耗时: {fp16_epoch_time:.1f} 小时")
print(f" 总训练时间: {fp16_total_time:.1f} 小时 ({fp16_total_time/24:.1f} 天)")

time_saved = fp32_total_time – fp16_total_time
print(f"\\n ⚡ 使用FP16节省时间: {time_saved:.1f} 小时 ({time_saved/24:.1f} 天)")

# 硬件需求
print("\\n🖥️ 硬件需求:")

# 显存需求估算
model_size = 100 # MB
batch_memory = batch_size * 3 * img_size * img_size * 4 / 1024**2 # MB
gradient_memory = model_size * 2
optimizer_memory = model_size * 2

fp32_memory = model_size + batch_memory + gradient_memory + optimizer_memory
fp16_memory = fp32_memory * 0.6 # FP16约节省40%显存

print(f" FP32显存需求: {fp32_memory:.0f} MB ({fp32_memory/1024:.1f} GB)")
print(f" FP16显存需求: {fp16_memory:.0f} MB ({fp16_memory/1024:.1f} GB)")

print(f"\\n 推荐GPU配置:")
if fp16_memory < 12000:
print(f" 单卡训练: RTX 3090 (24GB) 或 A10 (24GB)")
else:
print(f" 多卡训练: 4×A100 (40GB) 或 8×V100 (32GB)")

# 存储需求
print(f"\\n💾 存储需求:")
print(f" 原始数据: {self.dataset_info['总数据量']}")
print(f" 预处理缓存: ~1 TB")
print(f" 检查点存储: ~50 GB")
print(f" 日志和可视化: ~10 GB")
print(f" 总计: ~6 TB")

return {
'fp32_time': fp32_total_time,
'fp16_time': fp16_total_time,
'time_saved': time_saved,
'fp32_memory': fp32_memory,
'fp16_memory': fp16_memory
}

def training_strategy(self):
"""
训练策略规划
"""
print("\\n" + "="*70)
print("训练策略规划")
print("="*70)

strategies = {
'阶段1: 数据准备': [
'使用滑窗裁剪将大图切分为640×640小图',
'构建LMDB数据库加速随机访问',
'预计算数据集统计信息(均值、标准差)',
'生成训练/验证/测试集划分索引'
],
'阶段2: 模型预训练': [
'使用ImageNet预训练权重初始化骨干网络',
'在10%数据子集上快速验证配置',
'调整学习率和批量大小',
'确定最优的混合精度配置'
],
'阶段3: 全量训练': [
'启用FP16混合精度训练',
'使用余弦退火学习率调度',
'每10个epoch保存检查点',
'实时监控训练指标和系统资源'
],
'阶段4: 模型优化': [
'在验证集上评估不同epoch的模型',
'选择最佳模型进行知识蒸馏',
'导出ONNX和TensorRT格式',
'在测试集上进行最终评估'
]
}

for stage, tasks in strategies.items():
print(f"\\n{stage}:")
for task in tasks:
print(f" • {task}")