从炼丹到实验管理：用软件工程方法论管理 AI 实验

从炼丹到实验管理：用软件工程方法论管理 AI 实验

一、你的实验记录还躺在聊天记录和终端历史里吗？

打开终端，按 50 次上箭头——你能找到上周三那次跑出最佳结果的训练命令吗？学习率是 3e-4 还是 5e-4？那轮用了几个 GPU？batch size 是 32 还是 64？

"炼丹"这个词精准概括了 AI 工程师的工作状态：半科学、半玄学、实验结果时好时坏。但问题不在于"玄学"，而在于实验过程没有被当作软件工程产物来管理。

软件工程有版本控制、CI/CD、代码审查、需求跟踪。AI 实验管理凭什么不能有？见证奇迹的时刻是当你发现"炼丹"和"软件工程"可以用同一套方法论管理。

二、AI 实验的数据模型与生命周期

一个 AI 实验不是"一个训练 run"这么简单。它的数据模型至少包含五个维度：

erDiagram EXPERIMENT { string id PK string hypothesis "实验假设" string status "planned/running/done/failed" datetime created_at } CONFIG { string id PK string experiment_id FK float learning_rate int batch_size string optimizer string model_arch } RUN { string id PK string experiment_id FK int step_count float best_loss string checkpoint_path } METRIC { string id PK string run_id FK int step float train_loss float val_loss } EXPERIMENT ||--o{ CONFIG : "has" EXPERIMENT ||--o{ RUN : "runs" RUN ||--o{ METRIC : "records"

实验的生命周期：假设阶段→配置阶段→执行阶段→分析阶段→归档阶段。

三、实验追踪系统的工程实现

import json import uuid import hashlib from pathlib import Path from datetime import datetime from typing import Dict, List, Optional, Any from dataclasses import dataclass, field, asdict from enum import Enum class ExperimentStatus(Enum): """ 实验状态枚举 设计理念：引入比"成功/失败"更丰富的状态语义。 INTERRUPTED 表示手动停止（可能是有价值的中间结果）， SUSPICIOUS 表示训练完成但怀疑数据有误。 不同状态在统计"最佳配置"时有不同的权重 """ PLANNED = "planned" # 已规划，尚未执行 RUNNING = "running" # 训练中 COMPLETED = "completed" # 正常完成 FAILED = "failed" # 运行时错误（OOM/代码bug） INTERRUPTED = "interrupted" # 被手动中断 SUSPICIOUS = "suspicious" # 结果看起来有问题，需要人工审查 @dataclass class HyperParams: """超参数配置 作为独立数据类而非嵌套字典的原因是： IDE 可以提供自动补全和类型检查，避免"lr"和"learning_rate"混用 导致的运行时 bug """ learning_rate: float = 1e-3 batch_size: int = 32 optimizer: str = "Adam" epochs: int = 100 warmup_steps: int = 0 weight_decay: float = 0.01 model_arch: str = "" extra: Dict[str, Any] = field(default_factory=dict) # 扩展槽位 @dataclass class ExperimentRun: """单次实验运行的完整记录""" experiment_id: str run_id: str = field(default_factory=lambda: str(uuid.uuid4())[:8]) status: ExperimentStatus = ExperimentStatus.PLANNED hypothesis: str = "" # 实验假设 hyperparams: HyperParams = field(default_factory=HyperParams) metrics: List[Dict[str, Any]] = field(default_factory=list) started_at: Optional[str] = None ended_at: Optional[str] = None notes: str = "" # 自由格式备注 git_commit: str = "" # 关联的代码版本 class ExperimentTracker: """ 实验追踪器 核心设计决策：JSON 文件存储而非数据库。 对于单机/小团队场景，数据库引入了运维成本而收益有限。 JSON 文件可以被 grep、被版本控制、被任意脚本处理。 当实验规模超过 1000 个/天时，再考虑迁移到数据库 """ def __init__(self, store_dir: str = "./experiments"): self.store_dir = Path(store_dir) self.store_dir.mkdir(parents=True, exist_ok=True) self._index_path = self.store_dir / "index.json" self._index: Dict[str, Dict] = self._load_index() def _load_index(self) -> Dict: """加载实验索引，用于快速查找""" if self._index_path.exists(): return json.loads(self._index_path.read_text()) return {} def _save_index(self): """持久化索引到磁盘""" self._index_path.write_text( json.dumps(self._index, indent=2, ensure_ascii=False) ) def create_experiment(self, hypothesis: str, **hyperparams) -> str: """ 创建新实验 为什么 experiment_id 使用 UUID 而非自增数字？ 分布式场景下多台机器同时创建实验时，自增 ID 会冲突。 UUID 的唯一性不需要中心化协调 """ exp_id = str(uuid.uuid4())[:12] run = ExperimentRun( experiment_id=exp_id, hypothesis=hypothesis, hyperparams=HyperParams(**{ k: v for k, v in hyperparams.items() if k in HyperParams.__dataclass_fields__ }), status=ExperimentStatus.PLANNED, ) # 捕获当前 git commit，确保实验与代码版本关联 try: import subprocess run.git_commit = subprocess.check_output( ["git", "rev-parse", "--short", "HEAD"], text=True ).strip() except Exception: pass self._save_run(exp_id, run) self._index[exp_id] = { "hypothesis": hypothesis, "status": run.status.value, "created_at": datetime.now().isoformat(), } self._save_index() return exp_id def log_metrics(self, experiment_id: str, step: int, metrics: Dict[str, float]): """ 记录训练指标 追加模式而非覆盖模式：指标列表随时间增长。 这对磁盘有影响（每个 step 约 200 bytes）， 但对 1000 epochs × 10 实验的场景约 2MB，完全可以接受 """ run = self._load_run(experiment_id) if run is None: raise ValueError(f"实验 {experiment_id} 不存在") entry = {"step": step, "timestamp": datetime.now().isoformat()} entry.update(metrics) run.metrics.append(entry) run.status = ExperimentStatus.RUNNING if run.started_at is None: run.started_at = datetime.now().isoformat() self._save_run(experiment_id, run) def complete_experiment(self, experiment_id: str, status: ExperimentStatus = ExperimentStatus.COMPLETED): """标记实验完成""" run = self._load_run(experiment_id) if run: run.status = status run.ended_at = datetime.now().isoformat() self._save_run(experiment_id, run) # 更新索引 if experiment_id in self._index: self._index[experiment_id]["status"] = status.value self._save_index() def compare(self, exp_id_a: str, exp_id_b: str) -> Dict: """对比两个实验的配置和结果差异""" run_a = self._load_run(exp_id_a) run_b = self._load_run(exp_id_b) if not run_a or not run_b: return {"error": "experiment not found"} # 对比最终指标 final_metrics_a = run_a.metrics[-1] if run_a.metrics else {} final_metrics_b = run_b.metrics[-1] if run_b.metrics else {} comparison = {} for key in set(list(final_metrics_a.keys()) + list(final_metrics_b.keys())): if key in ("step", "timestamp"): continue a_val = final_metrics_a.get(key, None) b_val = final_metrics_b.get(key, None) if a_val is not None and b_val is not None: comparison[key] = { "a": a_val, "b": b_val, "delta": round(b_val - a_val, 6), "better": "B" if b_val > a_val else "A", } return { "config_a": asdict(run_a.hyperparams) if run_a else None, "config_b": asdict(run_b.hyperparams) if run_b else None, "metrics_comparison": comparison, } def _run_path(self, experiment_id: str) -> Path: return self.store_dir / f"{experiment_id}.json" def _save_run(self, experiment_id: str, run: ExperimentRun): """保存实验到 JSON 文件""" data = asdict(run) data["status"] = run.status.value self._run_path(experiment_id).write_text( json.dumps(data, indent=2, ensure_ascii=False) ) def _load_run(self, experiment_id: str) -> Optional[ExperimentRun]: """从 JSON 文件加载实验""" path = self._run_path(experiment_id) if not path.exists(): return None data = json.loads(path.read_text()) data["status"] = ExperimentStatus(data["status"]) data["hyperparams"] = HyperParams(**data["hyperparams"]) return ExperimentRun(**data) # ============================================================ # 使用示例 # ============================================================ if __name__ == "__main__": tracker = ExperimentTracker() exp_1 = tracker.create_experiment( hypothesis="提高 batch_size 到 128 应该能加速收敛", learning_rate=1e-3, batch_size=128, optimizer="AdamW", ) # 模拟训练过程中记录指标 for step in range(10): tracker.log_metrics(exp_1, step, { "train_loss": 2.5 - step * 0.2, "val_accuracy": 0.1 + step * 0.05, }) tracker.complete_experiment(exp_1) print(f"实验 {exp_1} 完成") ## 四、实验管理的架构权衡与工程选型 **专用平台 vs 自建工具**。 MLflow、Weights & Biases 是成熟的实验管理平台，但它们都有锁定风险——实验数据格式一旦被平台绑定，迁移成本很高。建议的策略是两层架构：底层用通用数据格式（JSON）存储实验元数据，上层可挂接任意平台做可视化。 **实验状态的语义细化**。 真实训练中会出现介于"成功"和"失败"之间的状态：被手动中断的、OOM 后恢复了但 loss 不正常的、跑完了但怀疑数据有问题的。建议引入 `INTERRUPTED`、`SUSPICIOUS`、`NEEDS_REVIEW` 等中间态标签。一个被标为 `SUSPICIOUS` 的实验不应该自动参与"哪个配置效果最好"的统计。 **协作冲突的解决**。 多人同时修改实验参数是典型冲突场景。解决方案是**配置即代码（Config as Code）**：所有实验参数必须定义在版本控制追踪的配置文件中，不允许在启动命令中通过 argparse 覆盖关键参数。见证奇迹的时刻是你发现 git blame 可以告诉你"谁在上周三把 learning rate 改成了 0.01"。 ## 五、总结 AI 实验管理应该用软件工程的方法论来组织。三个核心实践： 1. **每个实验必须绑定假设**：记录"为什么做"比"用了什么参数"更重要——因为三个月后你只记得动机不记得配置。 2. **实验状态必须细化**：三个终态不够，需要中间态来区分"真的失败"和"可能有问题"。 3. **配置必须入版本控制**：一行命令行参数不是可复现的实验记录，一个 git 提交的 YAML 配置文件才是。