# 实验管理：版本控制、实验跟踪与可复现性——MLflow 与 W&B 实践 **TL;DR：** ML实验的可复现性不仅是学术要求，更是工程必需。本文讲解实验管理的三大支柱——代码版本化、数据版本化、实验跟踪，以及MLflow和W&B两个主流工具的实践用法。 ## 为什么这很重要 一个真实的场景： 三个月前，你的团队训练了一个模型，在A/B测试中表现很好，于是上线了。现在，模型的表现开始下降，你需要重新训练。但是： - 当时用了哪个版本的代码？你不确定，因为git commit信息写的是"fix bug" - 当时用了哪个版本的数据？不知道，数据已经更新了好几次 - 当时的超参数是什么？记在一个.txt文件里，但是那个文件已经找不到了 - 当时的训练环境是什么？Python版本、CUDA版本、库版本全都不知道 这个模型成了"不可复现的黑箱"。你无法重建三个月前的训练条件，也就无法判断性能下降是因为数据分布变化、还是代码变更引入了bug。 这不是假设场景——在工业界，这种事情每天都在发生。一项2023年的调查显示，超过60%的ML从业者表示他们无法复现自己6个月前的实验结果。 实验管理的核心目标就是**让每一个实验都完全可复现**： ``` 实验 = f(代码, 数据, 超参数, 环境, 随机种子) ``` 如果以上五个要素都被完整记录和版本化，那么任何实验都可以在任何时间被精确复现。 ## 核心概念 ### 实验管理的三个层次 ``` ┌────────────────────────────────────────────────────┐ │ 第一层：代码版本化（Git） │ │ 记录每次实验使用的代码版本 │ │ 不仅是模型代码，还包括数据处理、训练脚本 │ │ │ │ 第二层：数据版本化（DVC） │ │ 记录每次实验使用的数据集版本 │ │ 大文件用DVC管理，元数据用Git管理 │ │ │ │ 第三层：实验跟踪（MLflow / W&B） │ │ 记录超参数、指标、模型权重、环境信息 │ │ 提供可视化、比较、搜索、协作功能 │ └────────────────────────────────────────────────────┘ ``` ### 实验跟踪的核心要素 一个完整的实验记录应该包含： | 要素 | 内容 | 为什么重要 | |------|------|-----------| | 代码版本 | Git commit hash | 知道用了什么代码 | | 数据版本 | 数据集hash或DVC版本 | 知道用了什么数据 | | 超参数 | 学习率、batch size、模型配置 | 知道训练配置 | | 指标 | Loss、accuracy、F1等 | 知道模型表现 | | 模型权重 | 训练产出的模型文件 | 可以直接使用 | | 环境信息 | Python版本、库版本、CUDA版本 | 知道运行环境 | | 随机种子 | 所有随机种子 | 确保完全复现 | | 训练日志 | Loss曲线、梯度统计 | 诊断训练过程 | | 备注 | 实验假设、观察、结论 | 知道为什么做这个实验 | ## 工作原理（简化的心智模型） ### 用实验室记录本来理解 把实验管理想象成化学实验室的记录本： **代码版本化** = 在记录本上写清楚你用了哪个实验步骤。如果别人想重复你的实验，他可以精确地按照你的步骤来做。Git就是ML实验的"步骤记录器"。 **数据版本化** = 在记录本上写清楚你用了哪批化学试剂（批号、供应商、生产日期）。不同的试剂批次可能导致不同的实验结果。DVC就是ML实验的"试剂追踪器"。 **实验跟踪** = 在记录本上详细记录温度、压力、反应时间、产率、观察到的现象。MLflow/W&B就是ML实验的"自动化记录本"——你不需要手动写，系统自动帮你记录一切。 一个没有实验管理的ML项目，就像一个没有记录本的实验室——三个月后，你完全不知道当时做了什么。 ## 工作原理（详细机制） ### 一、MLflow 实践 MLflow是Databricks开发的开源实验管理平台。它由四个组件组成： - **MLflow Tracking**：记录和查询实验（超参数、指标、产物） - **MLflow Projects**：打包和可复现地运行ML代码 - **MLflow Models**：管理模型的生命周期 - **MLflow Registry**：模型的版本管理和阶段管理 ```python """ MLflow 实验跟踪完整示例 展示如何使用MLflow记录训练实验 """ import mlflow import mlflow.pytorch import torch import torch.nn as nn import torch.optim as optim from torch.utils.data import DataLoader, TensorDataset import numpy as np import json import os from datetime import datetime # ============================================================ # 配置MLflow # ============================================================ # 设置跟踪URI（本地文件存储或远程服务器） mlflow.set_tracking_uri("mlruns") # 本地存储 mlflow.set_experiment("sentiment-analysis-experiment") # ============================================================ # 模型定义 # ============================================================ class SentimentModel(nn.Module): def __init__(self, vocab_size=10000, embed_dim=128, hidden_dim=256, n_classes=3, n_layers=2, dropout=0.3): super().__init__() self.embedding = nn.Embedding(vocab_size, embed_dim) self.lstm = nn.LSTM(embed_dim, hidden_dim, n_layers, batch_first=True, dropout=dropout if n_layers > 1 else 0) self.dropout = nn.Dropout(dropout) self.fc = nn.Linear(hidden_dim, n_classes) def forward(self, x): embedded = self.embedding(x) _, (hidden, _) = self.lstm(embedded) output = self.dropout(hidden[-1]) return self.fc(output) # ============================================================ # 训练函数（带MLflow跟踪） # ============================================================ def train_with_mlflow( # 超参数 learning_rate: float = 1e-3, batch_size: int = 64, n_epochs: int = 10, hidden_dim: int = 256, n_layers: int = 2, dropout: float = 0.3, optimizer_name: str = 'AdamW', weight_decay: float = 1e-4, # MLflow配置 run_name: str = None, tags: dict = None, ): """ 使用MLflow跟踪的训练函数 MLflow会自动记录： - 所有传入的参数（超参数） - 训练过程中记录的指标（loss, accuracy） - 训练结束时的模型 - 环境信息（Python版本、库版本） """ # 生成实验名称 if run_name is None: run_name = f"lr{learning_rate}_hd{hidden_dim}_nl{n_layers}" with mlflow.start_run(run_name=run_name) as run: run_id = run.info.run_id print(f"MLflow Run ID: {run_id}") # === 记录超参数 === params = { 'learning_rate': learning_rate, 'batch_size': batch_size, 'n_epochs': n_epochs, 'hidden_dim': hidden_dim, 'n_layers': n_layers, 'dropout': dropout, 'optimizer': optimizer_name, 'weight_decay': weight_decay, 'vocab_size': 10000, 'embed_dim': 128, } mlflow.log_params(params) # === 记录标签 === if tags: mlflow.set_tags(tags) mlflow.set_tag('model_type', 'LSTM') mlflow.set_tag('task', 'sentiment_analysis') mlflow.set_tag('framework', 'PyTorch') # === 记录代码版本 === import subprocess try: git_hash = subprocess.check_output(['git', 'rev-parse', 'HEAD']).decode().strip() mlflow.set_tag('git_hash', git_hash) except: mlflow.set_tag('git_hash', 'unknown') # === 创建模型和数据 === device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') model = SentimentModel( hidden_dim=hidden_dim, n_layers=n_layers, dropout=dropout ).to(device) # 模拟数据 n_samples = 5000 seq_length = 50 X = torch.randint(0, 10000, (n_samples, seq_length)) y = torch.randint(0, 3, (n_samples,)) dataset = TensorDataset(X, y) train_size = int(0.8 * len(dataset)) val_size = len(dataset) - train_size train_dataset, val_dataset = torch.utils.data.random_split(dataset, [train_size, val_size]) train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True) val_loader = DataLoader(val_dataset, batch_size=batch_size) # === 优化器和损失函数 === if optimizer_name == 'AdamW': optimizer = optim.AdamW(model.parameters(), lr=learning_rate, weight_decay=weight_decay) elif optimizer_name == 'SGD': optimizer = optim.SGD(model.parameters(), lr=learning_rate, weight_decay=weight_decay, momentum=0.9) criterion = nn.CrossEntropyLoss() # === 训练循环 === best_val_acc = 0 for epoch in range(n_epochs): # 训练 model.train() train_loss = 0 train_correct = 0 train_total = 0 for batch_x, batch_y in train_loader: batch_x, batch_y = batch_x.to(device), batch_y.to(device) optimizer.zero_grad() output = model(batch_x) loss = criterion(output, batch_y) loss.backward() optimizer.step() train_loss += loss.item() * len(batch_x) train_correct += (output.argmax(1) == batch_y).sum().item() train_total += len(batch_y) train_loss /= train_total train_acc = train_correct / train_total # 验证 model.eval() val_loss = 0 val_correct = 0 val_total = 0 with torch.no_grad(): for batch_x, batch_y in val_loader: batch_x, batch_y = batch_x.to(device), batch_y.to(device) output = model(batch_x) loss = criterion(output, batch_y) val_loss += loss.item() * len(batch_x) val_correct += (output.argmax(1) == batch_y).sum().item() val_total += len(batch_y) val_loss /= val_total val_acc = val_correct / val_total # === 记录每个epoch的指标 === mlflow.log_metrics({ 'train_loss': train_loss, 'train_accuracy': train_acc, 'val_loss': val_loss, 'val_accuracy': val_acc, }, step=epoch) if val_acc > best_val_acc: best_val_acc = val_acc # 记录最佳模型 mlflow.log_metric('best_val_accuracy', best_val_acc) if epoch % 2 == 0 or epoch == n_epochs - 1: print(f" Epoch {epoch:2d}: train_loss={train_loss:.4f}, " f"train_acc={train_acc:.4f}, val_loss={val_loss:.4f}, " f"val_acc={val_acc:.4f}") # === 记录模型 === mlflow.pytorch.log_model(model, "model") # === 记录训练摘要 === mlflow.log_metrics({ 'final_train_loss': train_loss, 'final_val_loss': val_loss, 'final_train_accuracy': train_acc, 'final_val_accuracy': val_acc, 'best_val_accuracy': best_val_acc, 'total_epochs': n_epochs, }) # === 记录环境信息 === mlflow.set_tag('python_version', f"{os.sys.version}") mlflow.set_tag('torch_version', torch.__version__) mlflow.set_tag('device', str(device)) print(f"\n实验完成: run_id={run_id}") print(f"最佳验证准确率: {best_val_acc:.4f}") print(f"查看结果: mlflow ui") return run_id, best_val_acc # ============================================================ # 运行多个实验 # ============================================================ print("="*60) print("MLflow 实验管理演示") print("="*60) # 实验1：基线 run_id_1, acc_1 = train_with_mlflow( learning_rate=1e-3, hidden_dim=256, n_layers=2, dropout=0.3, run_name="baseline_lr1e-3" ) # 实验2：更大模型 run_id_2, acc_2 = train_with_mlflow( learning_rate=1e-3, hidden_dim=512, n_layers=3, dropout=0.4, run_name="larger_model" ) # 实验3：更小学习率 run_id_3, acc_3 = train_with_mlflow( learning_rate=1e-4, hidden_dim=256, n_layers=2, dropout=0.3, run_name="smaller_lr" ) # 对比结果 print(f"\n{'='*60}") print("实验对比:") print(f" 实验1 (基线): val_acc = {acc_1:.4f}") print(f" 实验2 (大模型): val_acc = {acc_2:.4f}") print(f" 实验3 (小学习率): val_acc = {acc_3:.4f}") # 查询最佳实验 best_run = max([(run_id_1, acc_1), (run_id_2, acc_2), (run_id_3, acc_3)], key=lambda x: x[1]) print(f"\n最佳实验: {best_run[0][:12]}... (val_acc = {best_run[1]:.4f})") ``` ### 二、Weights & Biases 实践 ```python """ Weights & Biases (W&B) 实验跟踪示例 展示W&B的核心功能：实验跟踪、可视化、超参数扫描 """ # 注意：运行此代码需要 wandb login # 以下展示W&B的使用方式 import numpy as np from datetime import datetime class MockWandBRun: """ 模拟W&B的核心功能（不需要实际安装） 实际使用时替换为 import wandb """ def __init__(self, project: str, config: dict, name: str = None): self.project = project self.config = config self.name = name or f"run-{datetime.now().strftime('%Y%m%d_%H%M%S')}" self.metrics_history = [] self.step = 0 self.summary = {} print(f"\n[W&B] 项目: {project}") print(f"[W&B] 实验名: {self.name}") print(f"[W&B] 配置:") for k, v in config.items(): print(f" {k}: {v}") def log(self, metrics: dict, step: int = None): """记录指标""" if step is not None: self.step = step else: self.step += 1 self.metrics_history.append({'step': self.step, **metrics}) self.summary.update(metrics) def finish(self): """结束实验""" print(f"\n[W&B] 实验完成: {self.name}") print(f"[W&B] 最终指标:") for k, v in self.summary.items(): if isinstance(v, float): print(f" {k}: {v:.4f}") def wandb_train_example(): """使用W&B跟踪训练""" # === W&B配置 === config = { 'learning_rate': 1e-3, 'batch_size': 64, 'n_epochs': 10, 'hidden_dim': 256, 'n_layers': 2, 'dropout': 0.3, 'architecture': 'LSTM', 'dataset': 'sentiment-v2', } # 初始化W&B实验 # 实际代码: wandb.init(project="sentiment-analysis", config=config) run = MockWandBRun(project="sentiment-analysis", config=config, name="baseline-lstm") # 模拟训练 for epoch in range(config['n_epochs']): train_loss = 1.5 * np.exp(-0.15 * epoch) + np.random.normal(0, 0.02) train_acc = min(0.95, 0.4 + 0.05 * epoch + np.random.normal(0, 0.01)) val_loss = train_loss + 0.1 + np.random.normal(0, 0.03) val_acc = max(0, train_acc - 0.05 + np.random.normal(0, 0.01)) # 记录指标 run.log({ 'train/loss': train_loss, 'train/accuracy': train_acc, 'val/loss': val_loss, 'val/accuracy': val_acc, 'epoch': epoch, }) if epoch % 3 == 0: print(f" Epoch {epoch}: train_loss={train_loss:.4f}, val_acc={val_acc:.4f}") run.finish() # 运行W&B示例 wandb_train_example() # === W&B超参数扫描 === print(f"\n{'='*60}") print("W&B Sweep (超参数扫描)") print("="*60) sweep_config = { 'method': 'bayes', # 'grid', 'random', 'bayes' 'metric': { 'name': 'val/accuracy', 'goal': 'maximize' }, 'parameters': { 'learning_rate': { 'distribution': 'log_uniform', 'min': 1e-5, 'max': 1e-2, }, 'hidden_dim': { 'values': [128, 256, 512], }, 'n_layers': { 'values': [1, 2, 3], }, 'dropout': { 'distribution': 'uniform', 'min': 0.1, 'max': 0.5, }, }, } print("超参数搜索空间:") print(json.dumps(sweep_config, indent=2)) print("\n实际使用:") print(" 1. wandb.sweep(sweep_config, project='sentiment-analysis')") print(" 2. wandb agent(sweep_id, function=train_func)") print(" 3. W&B自动运行多次实验，贝叶斯优化选择超参数") ``` ### 三、DVC 数据版本管理 ```python """ DVC (Data Version Control) 数据版本管理示例 展示如何将数据版本与代码版本绑定 """ print("DVC 数据版本管理") print("=" * 60) dvc_workflow = """ 完整的数据+代码版本管理工作流: 1. 初始化 git init dvc init 2. 添加数据文件（大文件不进入Git） dvc add data/train.csv # 这会创建 data/train.csv.dvc (元数据) 和 .gitignore git add data/train.csv.dvc data/.gitignore git commit -m "Add training data v1" 3. 修改数据后，更新版本 dvc add data/train.csv git add data/train.csv.dvc git commit -m "Update training data v2 (add 5000 samples)" 4. 切换到历史版本的数据 git checkout v1.0 dvc checkout # 自动恢复v1.0版本的数据 5. 远程存储（S3, GCS, Azure, etc.） dvc remote add -d myremote s3://my-bucket/dvc-storage dvc push # 上传数据到远程存储 dvc pull # 从远程存储下载数据 6. 完整的实验复现 git checkout experiment-42 dvc pull pip install -r requirements.txt python train.py # 精确复现实验 核心原则： - Git管理代码和.dvc元数据文件 - DVC管理实际的大数据文件 - 两者配合使用，确保代码和数据版本的精确对应 """ print(dvc_workflow) # 实验记录模板 experiment_template = """ 实验记录模板 ============ 基本信息: - 实验ID: EXP-2026-001 - 日期: 2026-05-27 - 研究者: [名字] - 目标: [一句话描述实验目的] 版本信息: - 代码: git:abc1234 (分支: feature/new-model) - 数据: dvc:v3.2 (描述: 增加了5000条标注数据) - 环境: requirements.txt (或 environment.yml) 超参数: - learning_rate: 1e-4 - batch_size: 32 - model: LSTM(hidden=256, layers=2, dropout=0.3) - optimizer: AdamW(weight_decay=1e-4) - epochs: 20 - random_seed: 42 结果: - train_accuracy: 0.9234 - val_accuracy: 0.8912 - test_accuracy: 0.8876 - best_epoch: 15 - training_time: 2.5 hours 观察和结论: - [你注意到的现象和分析] 下一步: - [基于这个实验的发现，下一步要做什么] """ print(experiment_template) ``` ### 四、MLflow vs W&B 对比 ```python """ MLflow 和 W&B 的详细对比 帮助选择适合你团队的工具 """ comparison = """ ┌─────────────────┬─────────────────────┬─────────────────────┐ │ 特性 │ MLflow │ Weights & Biases │ ├─────────────────┼─────────────────────┼─────────────────────┤ │ 开源 │ ✓ 完全开源 │ 部分开源（核心开源） │ │ 自托管 │ ✓ 完全支持 │ 有限支持 │ │ SaaS │ ✓ (Databricks) │ ✓ (主要模式) │ │ 实验跟踪 │ ✓ │ ✓ (更强) │ │ 可视化 │ 基本 │ 强大（交互式图表） │ │ 超参数扫描 │ 基本支持 │ ✓ (Bayes/Grid/Rand) │ │ 模型注册表 │ ✓ │ ✓ │ │ 模型部署 │ ✓ (内置) │ 有限 │ │ 协作 │ 基本 │ ✓ (强项) │ │ 团队管理 │ 有限 │ ✓ (完善) │ │ 报告生成 │ 有限 │ ✓ (Reports) │ │ 数据版本管理 │ 有限 │ ✓ (Artifacts) │ │ Alerting │ 无 │ ✓ │ │ 定价 │ 免费（自托管） │ 免费层+付费 │ │ 学习曲线 │ 中等 │ 低 │ │ 生态集成 │ Spark, Databricks │ PyTorch, TF, HF │ │ 离线使用 │ ✓ │ 有限 │ └─────────────────┴─────────────────────┴─────────────────────┘ 选择建议: 选择MLflow，如果： - 需要完全自托管（数据敏感行业） - 需要完整的模型生命周期管理（训练→部署） - 已在使用Databricks生态 - 团队预算有限 - 需要离线使用 选择W&B，如果： - 团队协作是核心需求 - 需要强大的可视化和报告 - 需要高级的超参数扫描（贝叶斯优化） - 愿意使用SaaS - 研究团队（实验多，需要频繁比较） 最佳实践：两者都用 - 用W&B做实验跟踪和可视化 - 用MLflow做模型注册和部署 - 用DVC做数据版本管理 """ print(comparison) ``` ## 代码示例（完整可运行的 Python） ```python """ 完整的实验管理系统 将Git、DVC概念、MLflow整合在一起的实验管理框架 """ import json import hashlib import os import subprocess from datetime import datetime from typing import Dict, List, Optional, Any from dataclasses import dataclass, field, asdict @dataclass class ExperimentConfig: """实验配置""" experiment_id: str = "" name: str = "" description: str = "" # 模型配置 model_type: str = "" model_params: Dict[str, Any] = field(default_factory=dict) # 训练配置 hyperparams: Dict[str, Any] = field(default_factory=dict) # 数据配置 dataset_name: str = "" dataset_version: str = "" dataset_hash: str = "" # 环境配置 python_version: str = "" packages: Dict[str, str] = field(default_factory=dict) cuda_version: str = "" gpu_info: str = "" # 随机种子 random_seed: int = 42 # 元信息 created_at: str = field(default_factory=lambda: datetime.now().isoformat()) created_by: str = os.environ.get('USER', 'unknown') git_hash: str = "" git_branch: str = "" @dataclass class ExperimentResult: """实验结果""" experiment_id: str = "" metrics: Dict[str, float] = field(default_factory=dict) best_metrics: Dict[str, float] = field(default_factory=dict) training_log: List[Dict] = field(default_factory=list) notes: str = "" conclusion: str = "" next_steps: str = "" completed_at: str = "" class ExperimentManager: """ 实验管理器 功能： 1. 自动记录实验的所有上下文信息 2. 提供实验对比功能 3. 支持实验搜索和过滤 4. 生成实验报告 """ def __init__(self, storage_dir: str = "experiments"): self.storage_dir = storage_dir os.makedirs(storage_dir, exist_ok=True) self.current_experiment = None def create_experiment(self, name: str, description: str = "", config: Dict = None) -> str: """创建新实验""" # 生成唯一ID timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') exp_id = f"EXP-{timestamp}-{hashlib.md5(name.encode()).hexdigest()[:6]}" exp_config = ExperimentConfig( experiment_id=exp_id, name=name, description=description, ) # 自动收集环境信息 exp_config.python_version = f"{os.sys.version_info.major}.{os.sys.version_info.minor}.{os.sys.version_info.minor}" # 获取Git信息 try: exp_config.git_hash = subprocess.check_output( ['git', 'rev-parse', 'HEAD'], stderr=subprocess.DEVNULL ).decode().strip()[:12] exp_config.git_branch = subprocess.check_output( ['git', 'rev-parse', '--abbrev-ref', 'HEAD'], stderr=subprocess.DEVNULL ).decode().strip() except: exp_config.git_hash = "unknown" exp_config.git_branch = "unknown" # 应用用户配置 if config: for k, v in config.items(): if hasattr(exp_config, k): setattr(exp_config, k, v) self.current_experiment = exp_config # 创建实验目录 exp_dir = os.path.join(self.storage_dir, exp_id) os.makedirs(exp_dir, exist_ok=True) # 保存配置 config_path = os.path.join(exp_dir, 'config.json') with open(config_path, 'w') as f: json.dump(asdict(exp_config), f, indent=2, ensure_ascii=False) print(f"实验已创建: {exp_id}") print(f" 名称: {name}") print(f" Git: {exp_config.git_hash} ({exp_config.git_branch})") return exp_id def log_metrics(self, metrics: Dict[str, float], step: int = None): """记录训练指标""" if self.current_experiment is None: print("警告: 没有活跃的实验") return log_entry = { 'timestamp': datetime.now().isoformat(), 'step': step, **metrics, } exp_dir = os.path.join(self.storage_dir, self.current_experiment.experiment_id) log_path = os.path.join(exp_dir, 'metrics.jsonl') with open(log_path, 'a') as f: f.write(json.dumps(log_entry) + '\n') def log_result(self, result: ExperimentResult): """记录实验结果""" if self.current_experiment is None: return result.experiment_id = self.current_experiment.experiment_id result.completed_at = datetime.now().isoformat() exp_dir = os.path.join(self.storage_dir, self.current_experiment.experiment_id) result_path = os.path.join(exp_dir, 'result.json') with open(result_path, 'w') as f: json.dump(asdict(result), f, indent=2, ensure_ascii=False) print(f"结果已保存: {result.best_metrics}") def compare_experiments(self, exp_ids: List[str] = None) -> str: """对比多个实验""" if exp_ids is None: # 列出所有实验 exp_ids = [d for d in os.listdir(self.storage_dir) if os.path.isdir(os.path.join(self.storage_dir, d))] if not exp_ids: return "没有找到实验" experiments = [] for exp_id in exp_ids: config_path = os.path.join(self.storage_dir, exp_id, 'config.json') result_path = os.path.join(self.storage_dir, exp_id, 'result.json') if os.path.exists(config_path): with open(config_path) as f: config = json.load(f) else: config = {} if os.path.exists(result_path): with open(result_path) as f: result = json.load(f) else: result = {} experiments.append({'config': config, 'result': result}) # 生成对比报告 report = f"实验对比报告 (共 {len(experiments)} 个实验)\n" report += "=" * 80 + "\n" for exp in experiments: cfg = exp['config'] res = exp['result'] report += f"\nID: {cfg.get('experiment_id', 'N/A')}\n" report += f" 名称: {cfg.get('name', 'N/A')}\n" report += f" Git: {cfg.get('git_hash', 'N/A')}\n" if 'hyperparams' in cfg: report += f" 超参数: {json.dumps(cfg['hyperparams'])}\n" if 'best_metrics' in res: report += f" 最佳指标: {json.dumps(res['best_metrics'])}\n" if 'conclusion' in res: report += f" 结论: {res['conclusion']}\n" print(report) return report def search_experiments(self, query: str) -> List[Dict]: """搜索实验""" results = [] query = query.lower() for exp_id in os.listdir(self.storage_dir): exp_dir = os.path.join(self.storage_dir, exp_id) if not os.path.isdir(exp_dir): continue config_path = os.path.join(exp_dir, 'config.json') if os.path.exists(config_path): with open(config_path) as f: config_str = f.read() if query in config_str.lower(): results.append(json.loads(config_str)) return results # 使用示例 manager = ExperimentManager() # 创建并运行实验1 exp1_id = manager.create_experiment( name="LSTM基线模型", description="使用标准LSTM架构的基线实验", config={ 'model_type': 'LSTM', 'model_params': {'hidden_dim': 256, 'n_layers': 2}, 'hyperparams': {'learning_rate': 1e-3, 'batch_size': 64, 'n_epochs': 20}, 'dataset_name': 'sentiment-v2', } ) # 模拟训练过程 for epoch in range(20): train_loss = 1.5 * np.exp(-0.12 * epoch) + np.random.normal(0, 0.02) val_acc = min(0.92, 0.5 + 0.02 * epoch + np.random.normal(0, 0.01)) manager.log_metrics({'train_loss': train_loss, 'val_accuracy': val_acc}, step=epoch) # 记录结果 result1 = ExperimentResult( best_metrics={'val_accuracy': 0.89, 'train_loss': 0.12}, conclusion='基线模型达到89%验证准确率', next_steps='尝试增大隐藏维度' ) manager.log_result(result1) # 创建实验2 exp2_id = manager.create_experiment( name="更大LSTM模型", description="增大隐藏维度和层数", config={ 'model_type': 'LSTM', 'model_params': {'hidden_dim': 512, 'n_layers': 3}, 'hyperparams': {'learning_rate': 5e-4, 'batch_size': 32, 'n_epochs': 20}, 'dataset_name': 'sentiment-v2', } ) for epoch in range(20): train_loss = 1.5 * np.exp(-0.15 * epoch) + np.random.normal(0, 0.02) val_acc = min(0.94, 0.5 + 0.025 * epoch + np.random.normal(0, 0.01)) manager.log_metrics({'train_loss': train_loss, 'val_accuracy': val_acc}, step=epoch) result2 = ExperimentResult( best_metrics={'val_accuracy': 0.92, 'train_loss': 0.08}, conclusion='更大模型提升了3%准确率', next_steps='尝试dropout和正则化' ) manager.log_result(result2) # 对比实验 manager.compare_experiments([exp1_id, exp2_id]) ``` ## 真实案例 ### 案例1：OpenAI的实验管理 OpenAI在训练GPT-4时使用了内部开发的实验管理系统（代号" internally developed system"）。该系统： - 自动记录所有超参数和训练配置 - 每个实验分配唯一的run ID - 训练曲线实时可视化 - 支持跨团队共享实验结果 - 与内部代码审查系统整合 ### 案例2：Google的ML实验文化 Google的ML团队有一个不成文的规定：**每个实验都必须有明确的假设和预期的结果**。如果实验结果与预期不符，必须解释原因，然后决定下一步。W&B的报告功能很好地支持了这种文化——每个实验的报告中都包含假设、结果、分析和下一步计划。 ### 案例3：小型团队的实用方案 一个3-5人的ML团队不需要复杂的工具链。最小可行的实验管理方案： 1. **Git** 管理所有代码 2. **MLflow** 记录实验（本地部署） 3. **Git LFS或DVC** 管理数据 4. **一个共享的实验表格**（Google Sheets或Notion）追踪实验状态 ## 权衡取舍以及何时不该使用 ### MLflow vs W&B 的取舍 - **数据敏感**：不能将实验数据发送到第三方服务器 → MLflow自托管 - **团队协作**：需要频繁共享实验结果 → W&B的协作功能更强 - **预算有限**：MLflow完全免费，W&B的免费层有限制 - **简单场景**：只需要基本的实验跟踪 → MLflow更简单 ### 何时不该过度投资实验管理 1. **探索阶段**：还在快速试错，不需要严格的实验管理 2. **单次实验**：只需要跑一次训练，不需要对比 3. **小团队**：2-3人的团队，一个共享的Google Sheet可能比MLflow更实用 4. **学习阶段**：刚开始学ML，先理解概念，再学习工具 ## 关键要点 1. **可复现性是ML工程的基础**。每一个实验都应该能被精确复现。这需要记录五个要素：代码版本、数据版本、超参数、环境信息、随机种子。 2. **MLflow适合端到端的ML生命周期管理**，特别是需要自托管或与Databricks生态集成的场景。MLflow Tracking记录实验，MLflow Models管理模型，MLflow Registry管理模型版本。 3. **W&B适合实验可视化和团队协作**，特别是研究团队。强大的图表、报告、超参数扫描功能让实验比较和分享变得容易。 4. **DVC是数据版本管理的事实标准**，通过将大文件的元数据保存在Git中、实际数据保存在外部存储中，解决了Git不能管理大文件的问题。 5. **不要等到需要时才开始使用实验管理**。在项目的第一天就设置好实验管理工具，比在三个月后试图重建实验历史要容易得多。 6. **实验记录不只是数据，还包括上下文**。你为什么做这个实验？你观察到了什么？你的假设是什么？这些"人类知识"和数值指标一样重要。 7. **最小可行的实验管理只需要三样东西**：Git（代码）、DVC（数据）、一个实验表格（超参数和结果的映射）。 ## 延伸阅读 **工具文档**： - MLflow官方文档 (mlflow.org) — MLflow的完整使用指南 - Weights & Biases文档 (docs.wandb.ai) — W&B的完整使用指南 - DVC官方文档 (dvc.org) — 数据版本管理指南 - LakeFS — 数据版本管理的替代方案 **文章**： - "Versioning Machine Learning Experiments" (Towards Data Science) — 实验版本化的方法论 - "MLOps: Why experiment tracking is important" (LittleBigCode) — 实验跟踪的重要性 - "Managing Large-Scale ML Experiments" (Enhanced MLOps) — 大规模实验管理 **课程**： - DeepLearning.AI MLOps Specialization — MLOps的完整课程 - Full Stack Deep Learning (Stanford) — ML工程实践课程