PyTorch 分布式訓(xùn)練排障先對齊單卡數(shù)值分布式訓(xùn)練一出問題先別讓產(chǎn)品、算法和平臺團隊各盯一張監(jiān)控圖。單卡數(shù)值、數(shù)據(jù)供給、顯存和通信要分層驗證否則吞吐下降和模型不收斂很容易被混成同一件事。1. 先證明單卡路徑正確訓(xùn)練問題應(yīng)拆成數(shù)值正確性、數(shù)據(jù)供給、顯存使用和通信行為四部分。先以小規(guī)模、固定輸入驗證前向和反向結(jié)果再觀察多進程路徑避免把單一監(jiān)控值當(dāng)成整體結(jié)論。單卡結(jié)果是多卡診斷的對照組。模型、數(shù)據(jù)預(yù)處理或損失實現(xiàn)變化后都應(yīng)重建這條基線。2. 再加入進程和通信變量每次試驗都應(yīng)寫清框架版本、設(shè)備類型、批量形狀、隨機種子和啟動方式。發(fā)生偏差時優(yōu)先比較中間張量與梯度而不是直接調(diào)整并行參數(shù)。每輪只改變一個變量并保存張量摘要、配置與錯誤類型。原始訓(xùn)練數(shù)據(jù)不進入排障包必要時用公開或合成樣本復(fù)現(xiàn)。3. 數(shù)值一致性檢查示例以下片段保留原有技術(shù)結(jié)構(gòu)。運行前請?zhí)鎿Q為本地的非敏感示例并根據(jù)依賴版本核對接口。單卡 Batch Size 32, 顯存占用 65% ── 穩(wěn)定訓(xùn)練 單卡 Batch Size 64, 顯存占用 98% ── 極易發(fā)生 OOM (Out Of Memory)import time import asyncio from typing import Dict, Any, Optional from dataclasses import dataclass, field from fastapi import FastAPI, HTTPException, BackgroundTasks dataclass class DistributedJobConfig: job_id: str world_size: int batch_size_per_gpu: int learning_rate: float status: str PENDING error_msg: Optional[str] None metrics: Dict[str, float] field(default_factorydict) class DDPJobManager: 分布式訓(xùn)練任務(wù)代理管理器 負責(zé)隔離上層任務(wù)請求與底層的 PyTorch DDP 進程組調(diào)度 def __init__(self): self.jobs: Dict[str, DistributedJobConfig] {} def submit_job(self, job_id: str, world_size: int, batch_size: int) - DistributedJobConfig: if job_id in self.jobs: raise ValueError(f任務(wù) ID {job_id} 已存在拒絕重復(fù)提交) config DistributedJobConfig( job_idjob_id, world_sizeworld_size, batch_size_per_gpubatch_size ) self.jobs[job_id] config return config async def execute_ddp_training_mock(self, job_id: str): 模擬后端拉起 PyTorch torchrun 分布式訓(xùn)練任務(wù)的過程 job self.jobs.get(job_id) if not job: return job.status RUNNING try: # 模擬多節(jié)點 DDP 初始化與 Epoch 循環(huán) for epoch in range(1, 4): await asyncio.sleep(1.0) # 模擬耗時 # 更新實驗過程指標 job.metrics[fepoch_{epoch}_loss] round(1.0 / epoch, 4) # 訓(xùn)練成功收尾 job.status COMPLETED job.metrics[final_accuracy] 0.945 except Exception as e: job.status FAILED job.error_msg str(e) # 初始化 API 應(yīng)用 app FastAPI(titleDDP Training Service Isolation Gateway) manager DDPJobManager() app.post(/api/v1/jobs/submit) async def submit_training_job(job_id: str, world_size: int, batch_size: int, background_tasks: BackgroundTasks): try: config manager.submit_job(job_id, world_size, batch_size) # 將耗時的分布式訓(xùn)練掛到后臺異步任務(wù)隊列 background_tasks.add_task(manager.execute_ddp_training_mock, job_id) return {code: 200, message: 訓(xùn)練任務(wù)已成功提交, job_id: job_id, status: config.status} except ValueError as ve: raise HTTPException(status_code400, detailstr(ve)) except Exception as e: raise HTTPException(status_code500, detailf服務(wù)器內(nèi)部異常: {str(e)}) app.get(/api/v1/jobs/status/{job_id}) async def get_job_status(job_id: str): job manager.jobs.get(job_id) if not job: raise HTTPException(status_code404, detail未找到指定任務(wù)) return { job_id: job.job_id, status: job.status, metrics: job.metrics, error: job.error_msg }4. 分布式啟動前復(fù)核單卡前向、反向和優(yōu)化器步進是否穩(wěn)定。rank、world size、后端和啟動命令是否記錄。各進程拿到的數(shù)據(jù)分片是否符合預(yù)期。超時、進程退出和 checkpoint 恢復(fù)是否有測試?？偨Y(jié)先把單卡數(shù)值對齊再碰通信參數(shù)。這個順序看起來慢實際能少掉大量無效調(diào)參。