graph TD
A[训练任务启动] --> B{运行中}
B -->|GPU 故障| C[自动检测]
B -->|网络异常| C
B -->|存储故障| C
B -->|正常完成| Z[训练完成]
C --> D{能否恢复?}
D -->|是| E[加载最近 Checkpoint]
D -->|否| F[标记节点故障]
E --> G[排除故障节点]
F --> G
G --> H[重新配置并行组]
H --> B
style C fill:#fff3e0
style E fill:#e8f5e9
style F fill:#ffebee
第21章 可靠性与灾备
第21章 可靠性与灾备
万卡集群的平均无故障时间（MTBF）以小时计。这不是危言耸听——这是工程现实。当你运行一个需要持续数周甚至数月的训练任务时，可靠性不是一个”锦上添花”的特性，而是一个”没有它就无法完成训练”的基础设施。
21.1 大规模训练中的稳定性设计
故障是常态，不是异常
一个直观的计算：
假设单 GPU 的 MTBF = 5 年
万卡集群的期望故障间隔 = 5 年 / 10000 = 4.38 小时
也就是说，万卡集群平均每 4-5 小时就有一次 GPU 故障。
如果训练需要 30 天 = 720 小时，
预期故障次数 ≈ 720 / 4.38 ≈ 164 次这意味着不处理故障就不可能完成大规模训练。
故障预测
# GPU 健康度监控与故障预测
import subprocess
import json
from datetime import datetime, timedelta
class GPUHealthMonitor:
"""GPU 健康度监控器"""
def __init__(self):
self.health_history = {} # rank -> health records
self.alert_thresholds = {
"ecc_error_count": 100, # ECC 错误计数
"temperature_critical": 90, # 温度阈值 (°C)
"power_limit_hit_ratio": 0.8, # 功率限制触发频率
"xid_error_count": 10, # XID 错误计数
"nvlink_error": 1, # NVLink 错误（零容忍）
}
def check_gpu_health(self, rank, node):
"""
检查单个 GPU 的健康状况
"""
# 获取 GPU 状态
nvidia_smi = self._run_nvidia_smi(node)
# 检查关键指标
health_score = 100 # 满分
issues = []
# 1. ECC 错误
ecc_errors = nvidia_smi.get("ecc_errors", {}).get("volatile_total", 0)
if ecc_errors > 0:
health_score -= min(50, ecc_errors * 5)
issues.append(f"ECC errors: {ecc_errors}")
# 2. 温度
temp = nvidia_smi.get("temperature", 0)
if temp > 85:
health_score -= 20
issues.append(f"High temperature: {temp}°C")
# 3. XID 错误（GPU 硬件异常）
xid_errors = self._check_dmesg_xid(node)
if xid_errors > 0:
health_score -= min(30, xid_errors * 10)
issues.append(f"XID errors: {xid_errors}")
# 4. NVLink 状态
nvlink_errors = self._check_nvlink_errors(node, rank)
if nvlink_errors > 0:
health_score -= 40
issues.append(f"NVLink errors: {nvlink_errors}")
# 5. 内存使用异常（可能的内存泄漏）
mem_util = nvidia_smi.get("memory_used", 0) / nvidia_smi.get("memory_total", 1)
if mem_util > 0.95 and not self._is_training_peak():
health_score -= 15
issues.append(f"High memory usage: {mem_util:.1%}")
status = "healthy" if health_score >= 70 else "degraded" if health_score >= 40 else "critical"
return {
"rank": rank,
"node": node,
"timestamp": datetime.now().isoformat(),
"health_score": health_score,
"status": status,
"issues": issues,
}
def _run_nvidia_smi(self, node):
"""远程执行 nvidia-smi 并解析结果"""
cmd = f"ssh {node} nvidia-smi --query-gpu=index," \
f"temperature.gpu,utilization.gpu,memory.used,memory.total," \
f"ecc.errors.uncorrected.volatile.total," \
f"power.draw,clocks.current.sm " \
f"--format=csv,noheader,nounits"
result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
# ... 解析逻辑
return {}
def _check_dmesg_xid(self, node):
"""检查内核日志中的 XID 错误"""
cmd = f"ssh {node} dmesg | grep -i 'Xid' | wc -l"
result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
return int(result.stdout.strip())
# 冗余设计：多余的 GPU 用于故障替换
# 典型配置：每 32 台服务器配备 1 台备机
redundancy_config = {
"nodes_per_group": 32,
"standby_nodes": 1, # 3.1% 冗余
"auto_failover": True,
"max_concurrent_failures": 2, # 同时容忍的故障数
"replacement_strategy": "warm_standby", # 热备
}
Tip
冗余经济学：一个 1000 卡集群，GPU 故障率假设每年 5%，平均维修时间 48 小时。无冗余 = 每年损失约 5000 GPU·小时。增加 3% 的冗余卡（30 张），成本约 30 万美元，但节省的训练时间价值远超此数。
21.2 Checkpoint 与状态持久化
为什么 Checkpoint 是核心问题
训练一个万亿参数模型可能需要数周。如果没有 Checkpoint，一次故障就意味着从零开始。但 Checkpoint 本身也是巨大的工程挑战：
# Checkpoint 大小估算
def estimate_checkpoint_size(model_params, optimizer="AdamW",
precision="bf16"):
"""
估算 Checkpoint 大小
AdamW 优化器需要存储:
- 模型权重 (fp32 copy + bf16)
- 梯度 (bf16)
- 一阶动量 (fp32)
- 二阶动量 (fp32)
"""
bytes_per_param = {
"weights_fp32": 4, # FP32 权重副本
"weights_bf16": 2, # BF16 权重
"gradients_bf16": 2, # BF16 梯度
"momentum_fp32": 4, # FP32 一阶动量
"variance_fp32": 4, # FP32 二阶动量
}
total_bytes_per_param = sum(bytes_per_param.values()) # 16 bytes
checkpoint_size = model_params * total_bytes_per_param
print(f"模型参数量: {model_params/1e9:.1f}B")
print(f"每参数占用: {total_bytes_per_param} bytes")
print(f"Checkpoint 大小: {checkpoint_size / 1e12:.2f} TB")
return checkpoint_size
estimate_checkpoint_size(175e9) # 175B 参数 (GPT-3 级别)
# 输出: 2.80 TB
estimate_checkpoint_size(540e9) # 540B 参数
# 输出: 8.64 TB
estimate_checkpoint_size(1e12) # 1T 参数
# 输出: 16.00 TB保存一个 2.8 TB 的 Checkpoint 到存储系统，即使以 10 GB/s 的带宽也需要 280 秒。如果频繁保存，会显著降低训练效率。
分布式 Checkpoint 策略
# 分布式 Checkpoint 保存与加载
import torch
import torch.distributed as dist
import os
import time
from pathlib import Path
class DistributedCheckpoint:
"""
分布式 Checkpoint 管理器
核心策略：
1. 每个 GPU 只保存自己的分片（避免聚合开销）
2. 并行写入（利用所有 GPU 的带宽）
3. 异步保存（与训练重叠）
"""
def __init__(self, save_dir, max_to_keep=3):
self.save_dir = Path(save_dir)
self.max_to_keep = max_to_keep
self.rank = dist.get_rank()
self.world_size = dist.get_world_size()
def save(self, model, optimizer, lr_scheduler, epoch, step,
async_save=True):
"""
保存 Checkpoint
策略：每个 rank 保存自己的模型/优化器分片
"""
ckpt_dir = self.save_dir / f"step_{step}"
ckpt_dir.mkdir(parents=True, exist_ok=True)
# 每个 rank 只保存自己的状态
state = {
"model": self._get_model_state_shard(model),
"optimizer": optimizer.state_dict(),
"lr_scheduler": lr_scheduler.state_dict() if lr_scheduler else None,
"epoch": epoch,
"step": step,
"rank": self.rank,
"world_size": self.world_size,
}
save_path = ckpt_dir / f"rank_{self.rank:05d}.pt"
if async_save:
# 异步保存：在后台线程中执行 I/O
# 注意：需要确保 tensor 不被修改（deep copy）
state_copy = self._deep_copy_state(state)
torch.save(state_copy, save_path) # 简化版，实际应在单独线程
else:
torch.save(state, save_path)
# 同步所有 rank 完成保存
dist.barrier()
if self.rank == 0:
# 保存元数据
self._save_metadata(ckpt_dir, model, epoch, step)
# 清理旧 checkpoint
self._cleanup_old()
return ckpt_dir
def load(self, model, optimizer=None, lr_scheduler=None,
ckpt_dir=None, load_optimizer=True):
"""
加载 Checkpoint
关键：从正确的分片恢复
"""
if ckpt_dir is None:
ckpt_dir = self._find_latest_checkpoint()
ckpt_dir = Path(ckpt_dir)
save_path = ckpt_dir / f"rank_{self.rank:05d}.pt"
if not save_path.exists():
# 可能是重映射场景（rank 变化）
save_path = self._remap_shard(ckpt_dir, model)
state = torch.load(save_path, map_location="cpu", weights_only=False)
# 加载模型状态
self._load_model_state(model, state["model"])
if load_optimizer and optimizer:
optimizer.load_state_dict(state["optimizer"])
if lr_scheduler and state.get("lr_scheduler"):
lr_scheduler.load_state_dict(state["lr_scheduler"])
return state["epoch"], state["step"]
def _deep_copy_state(self, state):
"""深拷贝状态（用于异步保存）"""
import copy
return copy.deepcopy(state)
# 高级：Checkpoint 压缩
class CompressedCheckpoint(DistributedCheckpoint):
"""带压缩的 Checkpoint"""
def save(self, model, optimizer, **kwargs):
# 方案 1: 仅保存 FP16 权重（推理级 checkpoint）
# 大小减少 75%，但不包含优化器状态
# 方案 2: 量化优化器状态
# Adam 的方差项可以用 INT8 存储，精度损失可接受
# 方案 3: 增量保存（只保存与上一个 checkpoint 的差异）
passCheckpoint 频率的优化
# Checkpoint 策略优化：在可靠性和效率之间取得平衡
class CheckpointScheduler:
"""
智能 Checkpoint 调度器
策略：
- 定期全量 Checkpoint（如每 1000 步）
- 高频轻量 Checkpoint（如每 100 步，只保存权重）
- 故障触发 Checkpoint（检测到异常时立即保存）
"""
def __init__(self, config):
self.full_interval = config.get("full_interval", 1000)
self.light_interval = config.get("light_interval", 100)
self.last_full_step = 0
self.last_light_step = 0
def should_save_full(self, step):
return (step - self.last_full_step) >= self.full_interval
def should_save_light(self, step):
return (step - self.last_light_step) >= self.light_interval
def get_save_config(self, step):
if self.should_save_full(step):
self.last_full_step = step
return {
"save_optimizer": True,
"save_scheduler": True,
"compression": None, # 全量不压缩
"estimated_time": "300s",
"estimated_size": "2.8 TB",
}
elif self.should_save_light(step):
self.last_light_step = step
return {
"save_optimizer": False, # 不保存优化器
"save_scheduler": False,
"compression": "fp16", # FP16 压缩
"estimated_time": "60s",
"estimated_size": "350 GB",
}
return None
# Checkpoint 与训练重叠
# 关键技术：在 GPU 计算的同时进行 I/O
async def async_checkpoint_with_overlap(
model, optimizer, step, storage_path
):
"""
将 Checkpoint 保存与训练计算重叠
实现：
1. 将 GPU 上的 tensor 复制到 CPU（GPU→CPU 传输）
2. 在 CPU 上序列化
3. CPU 线程写入存储
"""
# Step 1: 异步 GPU→CPU 传输
cpu_state = {}
for name, param in model.named_parameters():
cpu_state[name] = param.data.cpu() # 异步传输
# Step 2: 训练继续执行（GPU 不阻塞）
# ... training step ...
# Step 3: CPU 线程写入存储（与下一步训练并行）
import threading
def write_to_storage():
torch.save({
"model": cpu_state,
"optimizer": optimizer.state_dict(),
"step": step,
}, storage_path)
writer = threading.Thread(target=write_to_storage)
writer.start()
# 不等待 writer 完成，继续训练
# 下一次 Checkpoint 前检查 writer 是否完成
Warning
异步 Checkpoint 的陷阱：异步保存时，如果下一步训练修改了 tensor，会导致保存的数据不一致。必须确保在异步保存完成前，对应的 tensor 不会被原地修改。安全做法是保存 tensor 的副本（.clone().cpu()），而不是引用。
21.3 跨机房与跨地域容灾
多活架构
graph TB
subgraph "Region A (主训练集群)"
RA_LB[负载均衡 / 调度器]
RA_C[训练集群<br/>1024 GPUs]
RA_S[存储<br/>并行文件系统]
RA_LB --> RA_C
RA_C --> RA_S
end
subgraph "Region B (容灾集群)"
RB_LB[负载均衡 / 调度器]
RB_C[训练集群<br/>1024 GPUs]
RB_S[存储<br/>并行文件系统]
RB_LB --> RB_C
RB_C --> RB_S
end
subgraph "Region C (备份)"
RC_S[冷存储<br/>Checkpoint 归档]
end
RA_S <-->|Checkpoint 同步<br/>高速专线| RB_S
RA_S -->|定期归档| RC_S
RB_S -.->|故障接管| RB_C
style RA_C fill:#e1f5fe
style RB_C fill:#e8f5e9
style RC_S fill:#fff3e0
数据同步策略
# 跨机房 Checkpoint 同步
class CrossRegionCheckpointSync:
"""
跨机房 Checkpoint 同步管理器
"""
def __init__(self, primary_region, standby_region, sync_config):
self.primary = primary_region # "us-west"
self.standby = standby_region # "us-east"
self.bandwidth_gbps = sync_config.get("bandwidth_gbps", 10) # 跨机房带宽
self.latency_ms = sync_config.get("latency_ms", 30) # RTT
def estimate_sync_time(self, checkpoint_size_tb):
"""估算同步时间"""
size_gb = checkpoint_size_tb * 1024
time_seconds = size_gb / self.bandwidth_gbps
print(f"Checkpoint 大小: {checkpoint_size_tb:.1f} TB")
print(f"跨机房带宽: {self.bandwidth_gbps} Gbps")
print(f"预计同步时间: {time_seconds:.0f} 秒 ({time_seconds/60:.1f} 分钟)")
return time_seconds
def sync_strategy(self, checkpoint_size_tb):
"""
选择同步策略
全量同步 vs 增量同步
"""
full_sync_time = self.estimate_sync_time(checkpoint_size_tb)
if full_sync_time < 300: # 5 分钟内
return "full_sync"
else:
# 增量同步：只传输变化的部分
# 利用 rsync --partial 或自定义差异算法
return "incremental_sync"
def sync_checkpoint(self, ckpt_path):
"""
执行同步
"""
strategy = self.sync_strategy(self._get_size(ckpt_path))
if strategy == "full_sync":
# 全量同步（使用 rclone 或自定义工具）
cmd = f"rclone copy {ckpt_path} remote-{self.standby}:{ckpt_path} " \
f"--transfers 32 --checkers 16 " \
f"--bandwidth {self.bandwidth_gbps}M"
else:
# 增量同步
cmd = f"rsync -avz --partial --inplace " \
f"--bwlimit={self.bandwidth_gbps * 1000} " \
f"{ckpt_path}/ standby-host:{ckpt_path}/"
return cmd
# 实际配置示例
sync_manager = CrossRegionCheckpointSync(
primary_region="dc-beijing",
standby_region="dc-shanghai",
sync_config={
"bandwidth_gbps": 25, # 25 Gbps 专线
"latency_ms": 15, # 15ms RTT
}
)
# 2.8 TB Checkpoint 同步
sync_manager.estimate_sync_time(2.8)
# 输出: 预计同步时间: 1147 秒 (19.1 分钟)21.4 模型权重安全与知识产权保护
威胁模型
大模型训练成本动辄数百万美元，模型权重是核心知识产权。主要威胁包括：
| 威胁类型 | 风险等级 | 典型场景 |
|---|---|---|
| 内部人员窃取 | ⭐⭐⭐⭐⭐ | 研究人员下载权重到个人设备 |
| 权限滥用 | ⭐⭐⭐⭐ | 过高的访问权限被利用 |
| 供应链攻击 | ⭐⭐⭐ | 恶意的第三方库 |
| 推理 API 提取 | ⭐⭐⭐ | 通过大量查询重建模型 |
| 存储系统入侵 | ⭐⭐⭐ | Checkpoint 存储被攻破 |
加密存储方案
# 模型权重的加密存储
import torch
from cryptography.fernet import Fernet
import hashlib
import hmac
import os
from pathlib import Path
class SecureCheckpointManager:
"""
安全 Checkpoint 管理器
特性：
- Checkpoint 加密存储（AES-256）
- 完整性校验（HMAC-SHA256）
- 访问审计日志
- 密钥轮换支持
"""
def __init__(self, storage_path, key_manager):
self.storage_path = Path(storage_path)
self.key_manager = key_manager # 密钥管理服务（如 HashiCorp Vault）
def save_encrypted(self, model_state, ckpt_name):
"""
加密保存 Checkpoint
"""
# 1. 序列化模型状态
buffer = torch.save(model_state, format="binary")
# 2. 获取加密密钥
key = self.key_manager.get_current_key()
cipher = Fernet(key)
# 3. 计算完整性校验码
h = hmac.new(key, buffer, hashlib.sha256)
hmac_digest = h.digest()
# 4. 加密
encrypted_data = cipher.encrypt(buffer)
# 5. 写入文件（HMAC + 加密数据）
ckpt_path = self.storage_path / f"{ckpt_name}.enc"
with open(ckpt_path, "wb") as f:
f.write(hmac_digest) # 32 bytes
f.write(encrypted_data)
# 6. 记录审计日志
self._log_access("save", ckpt_name, user=os.getuser())
return ckpt_path
def load_decrypted(self, ckpt_path, verify=True):
"""
解密加载 Checkpoint
"""
key = self.key_manager.get_current_key()
with open(ckpt_path, "rb") as f:
stored_hmac = f.read(32)
encrypted_data = f.read()
if verify:
# 验证完整性
h = hmac.new(key, encrypted_data, hashlib.sha256)
if not hmac.compare_digest(h.digest(), stored_hmac):
raise SecurityError("Checkpoint 完整性校验失败！可能被篡改。")
# 解密
cipher = Fernet(key)
buffer = cipher.decrypt(encrypted_data)
# 记录审计日志
self._log_access("load", ckpt_path, user=os.getuser())
# 反序列化
return torch.load(buffer)
def _log_access(self, operation, target, user):
"""记录访问日志到不可篡改的审计系统"""
log_entry = {
"timestamp": datetime.now().isoformat(),
"operation": operation, # save / load / delete
"target": str(target),
"user": user,
"host": os.uname().nodename,
"pid": os.getpid(),
}
# 发送到集中式审计系统（如 ELK / Splunk）
# log_entry 应该写入只追加（append-only）的日志存储
Tip
密钥管理最佳实践：(1) 使用专用的密钥管理服务（HashiCorp Vault、AWS KMS），不要硬编码密钥；(2) 定期轮换密钥（如每 90 天）；(3) 分离加密密钥和数据存储的管理权限；(4) 对所有 Checkpoint 访问（包括读取）记录审计日志。
21.5 伦理与安全基础设施
内容安全管道
# 训练数据与推理输出的安全过滤管道
class ContentSafetyPipeline:
"""
多层内容安全管道
层级：
1. 训练前：数据过滤（移除有害内容）
2. 训练中：安全约束（RLHF / DPO 对齐）
3. 推理时：输入/输出过滤
"""
def __init__(self):
self.filters = [
self.pii_filter, # 个人信息过滤
self.toxicity_filter, # 毒性检测
self.copyright_filter, # 版权检测
self.csfam_filter, # 儿童安全
]
def filter_training_data(self, dataset):
"""
训练数据过滤管道
"""
filtered_count = 0
for sample in dataset:
should_filter = False
filter_reasons = []
for filter_fn in self.filters:
result = filter_fn(sample)
if not result["safe"]:
should_filter = True
filter_reasons.append(result["reason"])
if not should_filter:
yield sample
else:
filtered_count += 1
self._log_filter(sample, filter_reasons)
print(f"过滤完成: {filtered_count} 条数据被移除")
def pii_filter(self, sample):
"""检测并移除 PII（个人身份信息）"""
import re
patterns = {
"ssn": r"\b\d{3}-\d{2}-\d{4}\b",
"email": r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b",
"phone": r"\b\d{3}[-.]?\d{3}[-.]?\d{4}\b",
"credit_card": r"\b\d{4}[- ]?\d{4}[- ]?\d{4}[- ]?\d{4}\b",
}
for pii_type, pattern in patterns.items():
if re.search(pattern, sample.get("text", "")):
return {"safe": False, "reason": f"Contains PII: {pii_type}"}
return {"safe": True, "reason": None}
def toxicity_filter(self, sample):
"""使用分类器检测有害内容"""
# 实际使用中，这里调用预训练的毒性检测模型
# 如 Perspective API 或自训练的分类器
toxicity_score = 0 # self.toxicity_model.predict(sample["text"])
if toxicity_score > 0.7:
return {"safe": False, "reason": f"Toxicity score: {toxicity_score}"}
return {"safe": True, "reason": None}模型水印
# 模型水印：在模型中嵌入可追溯的标记
import torch.nn as nn
class ModelWatermark:
"""
模型水印技术
方法：
1. 后训练水印：在特定输入上嵌入特定输出
2. 参数水印：在权重中嵌入统计特征
3. 输出水印：推理结果中嵌入隐藏信号
"""
@staticmethod
def embed_trigger_watermark(model, trigger_inputs, target_outputs,
lambda_w=0.01):
"""
通过微调嵌入触发式水印
当输入包含特定的 trigger 时，模型输出预设的响应。
这种水印不影响正常使用，但可以用来证明模型归属。
"""
optimizer = torch.optim.AdamW(model.parameters(), lr=1e-5)
for trigger, target in zip(trigger_inputs, target_outputs):
output = model(trigger)
# 正常训练损失
normal_loss = model.compute_loss(trigger, target)
# 水印损失
watermark_loss = nn.CrossEntropyLoss()(
output, target
)
total_loss = normal_loss + lambda_w * watermark_loss
total_loss.backward()
optimizer.step()
return model
@staticmethod
def verify_watermark(model, trigger_inputs, expected_outputs,
threshold=0.95):
"""
验证模型是否包含特定水印
"""
match_count = 0
for trigger, expected in zip(trigger_inputs, expected_outputs):
output = model(trigger)
if torch.allclose(output.argmax(dim=-1), expected, atol=1):
match_count += 1
match_rate = match_count / len(trigger_inputs)
return match_rate >= threshold
Warning
水印不是万能的：确定性的水印可以被微调或量化擦除。对于核心 IP 保护，建议组合使用多种水印方案，并结合法律手段。水印的目的是提供”数字证据”，而非完全防止盗用。
审计日志基础设施
# 审计日志系统配置（ELK Stack）
audit_logging:
# 数据采集
inputs:
- type: filestream
id: model-access-logs
paths:
- /var/log/ai-infra/model_access.log
- /var/log/ai-infra/checkpoint_access.log
parsers:
- json:
target: model_access
- type: filestream
id: training-logs
paths:
- /var/log/ai-infra/training_events.log
parsers:
- json:
target: training_events
# 处理管道
pipeline:
- process_fields:
- timestamp: "@timestamp"
- user: "user.name"
- action: "model_access.operation"
- resource: "model_access.target"
- result: "model_access.status"
# 敏感字段脱敏
- redact:
fields: ["api_key", "ssh_key", "model_weights_hash"]
redact_value: "[REDACTED]"
# 异常检测规则
- rule_alerting:
rules:
- name: "bulk_model_download"
condition: "count(model_access.operation='download') > 5 in 1h"
severity: high
action: "block_user + notify_admin"
- name: "off_hours_access"
condition: "timestamp.hour NOT IN [9..18] AND user.role != 'oncall'"
severity: medium
action: "log + require_mfa"
# 存储（不可篡改）
storage:
type: "s3"
bucket: "ai-infra-audit-logs"
immutable: true # WORM (Write Once Read Many)
retention_days: 365 # 保留 1 年
encryption: "aws:kms"小结
可靠性和灾备是”直到你需要它们时才发现它们重要”的系统属性。本章覆盖了五个关键维度：
- 稳定性设计：假设故障是常态。万卡集群每 4-5 小时就有一次故障，必须设计自动检测、自动恢复的训练框架
- Checkpoint 系统：这不是”保存文件”——这是一个涉及 2.8 TB 数据、I/O 带宽、存储系统和训练效率的复杂工程问题
- 跨地域容灾：在专线带宽约束下，设计合理的全量/增量同步策略
- IP 保护：模型权重是数百万美元的资产，需要加密存储、访问控制和审计日志
- 安全基础设施：内容安全、模型水印和审计日志构成了模型安全的最后一道防线
一个核心原则：可靠性的投入应该与训练成本成正比。一个 3 天就能完成的训练任务不需要复杂容灾；但一个运行 2 个月、成本数百万美元的训练任务，每一分钟的停机都是真金白银的损失。
延伸阅读
- Megatron Checkpoint 论文：在 Megatron-LM 的文档中有详细的分布式 Checkpoint 设计
- PyTorch Distributed Checkpoint：PyTorch 2.x 的新分布式 Checkpoint API
- Google’s Pathways：Barham et al., “Pathways: Asynchronous Distributed Dataflow for ML” — 设计了从故障中快速恢复的机制
- Zero-Trust Security for ML：关于 ML 系统零信任架构的工业实践
- Model Watering：Kirchenbauer et al., “A Watermark for Large Language Models” — LLM 水印的经典论文
- OWASP ML Security：OWASP Machine Learning Security Top 10