PyTorch 分布式訓(xùn)練排障先對齊單卡數(shù)值分布式訓(xùn)練一出問題先別讓產(chǎn)品、算法和平臺團(tuán)隊(duì)各盯一張監(jiān)控圖。單卡數(shù)值、數(shù)據(jù)供給、顯存和通信要分層驗(yàn)證否則吞吐下降和模型不收斂很容易被混成同一件事。1. 先證明單卡路徑正確訓(xùn)練問題應(yīng)拆成數(shù)值正確性、數(shù)據(jù)供給、顯存使用和通信行為四部分。先以小規(guī)模、固定輸入驗(yàn)證前向和反向結(jié)果再觀察多進(jìn)程路徑避免把單一監(jiān)控值當(dāng)成整體結(jié)論。單卡結(jié)果是多卡診斷的對照組。模型、數(shù)據(jù)預(yù)處理或損失實(shí)現(xiàn)變化后都應(yīng)重建這條基線。2. 再加入進(jìn)程和通信變量每次試驗(yàn)都應(yīng)寫清框架版本、設(shè)備類型、批量形狀、隨機(jī)種子和啟動方式。發(fā)生偏差時(shí)優(yōu)先比較中間張量與梯度而不是直接調(diào)整并行參數(shù)。每輪只改變一個(gè)變量并保存張量摘要、配置與錯(cuò)誤類型。原始訓(xùn)練數(shù)據(jù)不進(jìn)入排障包必要時(shí)用公開或合成樣本復(fù)現(xiàn)。3. 數(shù)值一致性檢查示例以下片段保留原有技術(shù)結(jié)構(gòu)。運(yùn)行前請?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ù)代理管理器 負(fù)責(zé)隔離上層任務(wù)請求與底層的 PyTorch DDP 進(jìn)程組調(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é)點(diǎn) DDP 初始化與 Epoch 循環(huán) for epoch in range(1, 4): await asyncio.sleep(1.0) # 模擬耗時(shí) # 更新實(shí)驗(yàn)過程指標(biāo) 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) # 將耗時(shí)的分布式訓(xùn)練掛到后臺異步任務(wù)隊(duì)列 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)化器步進(jìn)是否穩(wěn)定。rank、world size、后端和啟動命令是否記錄。各進(jìn)程拿到的數(shù)據(jù)分片是否符合預(yù)期。超時(shí)、進(jìn)程退出和 checkpoint 恢復(fù)是否有測試?？偨Y(jié)先把單卡數(shù)值對齊再碰通信參數(shù)。這個(gè)順序看起來慢實(shí)際能少掉大量無效調(diào)參。