diff --git a/ml/eval/diag_tail.py b/ml/eval/diag_tail.py new file mode 100644 index 0000000..f532ad1 --- /dev/null +++ b/ml/eval/diag_tail.py @@ -0,0 +1,130 @@ +#!/usr/bin/env python +"""闭环末位误差诊断（只读，手动运行；不在流水线里）。 + +回答一个问题：策略"差一点没进 1 mm 门"还是"根本没朝目标走"。 +输出每题： + - 最终位置误差 err_mm / 最终姿态误差 oerr_deg + - 全程最小位置误差 min_err_mm、最小综合误差时是否进过 5 mm & 5° 窗口（DLS 接管门槛） + - done（是否到达）、碰撞步数、用时 +以及按 cat/mode 分组的汇总与分位数。 + +用法（仓库根目录）： + PYTHONUTF8=1 python ml/eval/diag_tail.py --ckpt ml/runs/base/last.pt --n 512 --out ml/runs/diag_tail.json +""" +import argparse, json, math, os, sys + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))) + +import torch + + +def pct(v, p): + if not v: + return None + v = sorted(v) + return v[min(len(v) - 1, int(len(v) * p))] + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument('--ckpt', required=True) + ap.add_argument('--samples', default='ml/data/samples') + ap.add_argument('--tasks', default=None, help='任务 jsonl（覆盖 --samples/--split）') + ap.add_argument('--split', default='val') + ap.add_argument('--n', type=int, default=512) + ap.add_argument('--bs', type=int, default=64, help='每批题数（显存小就调小）') + ap.add_argument('--K', type=int, default=8) + ap.add_argument('--device', default='cuda' if torch.cuda.is_available() else 'cpu') + ap.add_argument('--out', default=None) + a = ap.parse_args() + + from ml.common.kin_torch import Kin, D2R + from ml.data import load_tasks + from ml.model.policy import build + from ml.sim.closed_loop import rollout, goal_err + + ck = torch.load(a.ckpt, map_location=a.device) + model = build(ck['cfg']).to(a.device) + model.load_state_dict(ck['model']) + model.eval() + kin = Kin(device=a.device) + if ck.get('fp') and ck['fp'] != kin.fp: + sys.exit('checkpoint 的碰撞模型指纹 %s 与当前 kin.json %s 不符' % (ck['fp'], kin.fp)) + + if a.tasks: + tasks = [json.loads(l) for l in open(a.tasks, encoding='utf-8') if l.strip()] + else: + tasks = [t for t in load_tasks([a.samples]) if t['split'] == a.split] + tasks = tasks[:a.n] + + per_task = [] + with torch.no_grad(): + for i in range(0, len(tasks), a.bs): + chunk = tasks[i:i + a.bs] + res, per = rollout(model, kin, chunk, a.device, K=a.K, safety=False, seed=i, return_traj=True) + traj = per['traj'] # [B, T, 6] + B, Tn, _ = traj.shape + Tt = torch.tensor([t['T'] for t in chunk], dtype=torch.float32, device=a.device) + Rt = torch.tensor([t['R'] if t.get('R') else [0.] * 9 for t in chunk], dtype=torch.float32, device=a.device) + min_e = torch.full((B,), float('inf'), device=a.device) + min_o = torch.full((B,), float('inf'), device=a.device) + entered = torch.zeros(B, dtype=torch.bool, device=a.device) # 曾同时满足 e<5mm & (无姿态要求或 oe<5°) + step = 256 + for t0 in range(0, Tn, step): + q = traj[:, t0:t0 + step].reshape(-1, 6) + n = q.shape[0] // B + e, oe = goal_err(kin, q, Tt.repeat_interleave(n, 0), Rt.repeat_interleave(n, 0)) + e = e.view(B, n); oe = oe.view(B, n) + min_e = torch.minimum(min_e, e.min(1).values) + min_o = torch.minimum(min_o, oe.min(1).values) + pose = (Rt.abs().sum(-1) > 0)[:, None] + window = (e < 5) & (~pose | (oe < 5)) + entered |= window.any(1) + done = per['done'].to(a.device); col = per['col'].to(a.device); td = per['t_done'].to(a.device) + end_e = per['err'].to(a.device) + # 末态姿态误差：traj 最后一帧 + qend = traj[:, -1] + ee, eo = goal_err(kin, qend, Tt, Rt) + for j, t in enumerate(chunk): + per_task.append({ + 'tid': t.get('tid'), 'cat': t.get('cat'), 'mode': t.get('mode'), + 'err_mm': float(end_e[j]), 'oerr_deg': float(eo[j]), + 'min_err_mm': float(min_e[j]), 'min_oerr_deg': float(min_o[j]), + 'entered_5mm5deg': bool(entered[j]), + 'done': bool(done[j]), 'collision_steps': float(col[j]), + 't_done_s': None if math.isnan(float(td[j])) else float(td[j]), + }) + print('[diag] %d/%d done=%d entered=%d min_err<=5mm=%d' % ( + min(i + a.bs, len(tasks)), len(tasks), int(done.sum()), int(entered.sum()), + sum(1 for p in per_task if p['min_err_mm'] <= 5)), flush=True) + + def summ(rows): + if not rows: + return {'n': 0} + e = [r['err_mm'] for r in rows]; me = [r['min_err_mm'] for r in rows] + o = [r['oerr_deg'] for r in rows] + return {'n': len(rows), + 'done': sum(r['done'] for r in rows), + 'entered_5mm5deg': sum(r['entered_5mm5deg'] for r in rows), + 'min_err_mm<=5': sum(1 for v in me if v <= 5), + 'min_err_mm<=10': sum(1 for v in me if v <= 10), + 'err_mm_p10/p50/p90': [pct(e, .1), pct(e, .5), pct(e, .9)], + 'min_err_mm_p10/p50/p90': [pct(me, .1), pct(me, .5), pct(me, .9)], + 'oerr_deg_p50/p90': [pct(o, .5), pct(o, .9)], + 'collision_steps_mean': sum(r['collision_steps'] for r in rows) / len(rows)} + + out = {'ckpt': a.ckpt, 'step': ck.get('step'), 'fp': ck.get('fp'), 'n': len(per_task), + 'overall': summ(per_task), + 'by_cat_mode': {k: summ([r for r in per_task if '%s/%s' % (r['cat'], r['mode']) == k]) + for k in sorted({('%s/%s' % (r['cat'], r['mode'])) for r in per_task})}, + 'tasks': per_task} + s = json.dumps(out, indent=1, ensure_ascii=False) + print(s[:2000]) + if a.out: + open(a.out + '.tmp', 'w', encoding='utf-8').write(s) + os.replace(a.out + '.tmp', a.out) + print('[diag] 已写', a.out) + + +if __name__ == '__main__': + main() diff --git a/ml/monitor.html b/ml/monitor.html index f3bc05a..ce431a7 100644 --- a/ml/monitor.html +++ b/ml/monitor.html @@ -207,12 +207,13 @@

开始训练

后台执行 bash ml/pipeline.sh all，从对拍一路跑到验收。已完成的步骤自动跳过；训练每 10 分钟存档、生成按条续写，断了再点一次就从断点接着跑。网页和控制台窗口都可以关，训练不受影响。
- + + - +
@@ -460,6 +461,7 @@ function renderProgress() { function renderKpis() { const r = recs[recs.length - 1] || {}, e = evals[evals.length - 1] || {}, pe = evals[evals.length - 2] || {}, kind = cur.kind; const best = evals.length ? Math.max(...evals.map(v => v.success ?? -1)) : null; + const bestE = evals.length ? evals.reduce((m, v) => ((v.success ?? -1) > (m.success ?? -1) ? v : m)) : null; const dlt = (k, good = 1) => { if (!isNum(e[k]) || !isNum(pe[k])) return ''; const d = (e[k] - pe[k]) * 100; if (Math.abs(d) < 0.05) return ''; const up = d > 0; return `${up ? '▲' : '▼'} ${Math.abs(d).toFixed(1)}pp`; }; const last = k => recs.filter(v => isNum(v[k])).slice(-300).map(v => [xOf(v), v[k]]); const evs = k => evals.filter(v => isNum(v[k])).map(v => [v.x, v[k]]); @@ -475,7 +477,7 @@ function renderKpis() { const sps = r.samples_per_s; tile('总损失', fmt(r.loss, 4), `MF ${fmt(r.mf)} · FK ${fmt(r.fk_cm)} cm`, '', ema(last('loss'), 0.9), COLORS[0]); tile('吞吐', sps ? `${(sps / 1000).toFixed(2)}k 样本/s` : '—', sps ? (sps >= 5000 ? 'R10 达标（≥ 5k/s）' : 'R10 未达标（< 5k/s）') : 'R10 通过线 5k/s', sps ? (sps >= 5000 ? 'ok' : 'bad') : '', last('samples_per_s'), COLORS[5]); - tile('闭环到达率', pct(e.success) + dlt('success'), evals.length ? `最好 ${pct(best)} · @${e.x.toLocaleString()} 步` : '尚未评估', '', evs('success'), COLORS[2]); + tile('闭环到达率', pct(e.success) + dlt('success'), bestE ? `最好 ${pct(best)} · @${bestE.x.toLocaleString()} 步` : '尚未评估', '', evs('success'), COLORS[2]); tile('碰撞率', pct(e.collision_rate, 2) + dlt('collision_rate', -1), '关安全层，部署另有护栏', e.collision_rate > 0.03 ? 'warn' : '', evs('collision_rate'), COLORS[3]); tile('梯度范数', fmt(r.gnorm), '裁剪阈值 1.0', r.gnorm > 5 ? 'warn' : '', ema(last('gnorm'), 0.9), COLORS[4]); tile('显存峰值', isNum(r.vram) ? `${r.vram.toFixed(2)}GiB` : '—', '预算 19–20 GB（整卡 24，桌面占 4–5）', r.vram > 17 ? 'bad' : r.vram > 14 ? 'warn' : '', last('vram'), COLORS[1]); @@ -712,14 +714,17 @@ function openStart() { const w = $('fWarn'); const upd = () => { const msgs = []; const done = P ? (P.plan || []).filter(k => P.stages[k] && P.stages[k].rc === 0) : []; if ($('fMode').value === 'resume' && done.length) msgs.push(`将跳过已完成的 ${done.length} 步：${done.map(k => SM[k]?.n || k).join('、')}。改了参数会自动改为从头重跑。`); - if ($('fMode').value === 'fresh' && P) msgs.push('从头重跑：上次的数据、模型、评测结果整体移到 ml/archive/<时间>/（不删除）。'); + if ($('fMode').value === 'from' && P) { const nm = $('fFrom').selectedOptions[0].textContent; msgs.push(`从「${nm}」开始重训：该阶段及之后的旧产物移到 ml/archive/<时间>/（不删除），之前的阶段保持不动。${$('fFrom').value === 'gen' ? '注意：生成的数据会被归档，重跑约 7 小时。' : ''}`); } + if ($('fMode').value === 'fresh' && P) msgs.push('全部重跑：上次的数据、模型、评测结果整体移到 ml/archive/<时间>/（不删除）。生成数据要重跑（约 7 小时）。'); + $('fFrom').disabled = $('fMode').value !== 'from'; w.style.display = msgs.length ? 'block' : 'none'; w.innerHTML = msgs.join('
'); }; $('fMode').onchange = upd; upd(); $('modal').classList.add('show'); } async function startJob() { - const env = { CFG: $('fCfg').value, CONSENSUS: $('fCons').value, N_TASKS: $('fN').value.trim(), NPROC: $('fP').value.trim(), TARGET_SAMPLES: $('fT').value.trim(), ROUNDS: $('fR').value.trim(), MIN_GAP_MM: $('fG').value.trim() }; - const body = JSON.stringify({ fresh: $('fMode').value === 'fresh', env }); + const mode = $('fMode').value, from = mode === 'from' ? $('fFrom').value : 'kin'; + const env = { CFG: $('fCfg').value, CONSENSUS: $('fCons').value, N_TASKS: $('fN').value.trim(), NPROC: $('fP').value.trim(), TARGET_SAMPLES: $('fT').value.trim(), ROUNDS: $('fR').value.trim(), MIN_GAP_MM: $('fG').value.trim(), FRESH_FROM: mode === 'from' ? from : '' }; + const body = JSON.stringify({ fresh: mode === 'fresh', from, to: 'accept', env }); const r = DEMO ? await api('/api/start', body) : await post('/api/start', body); if (r.error) { toast('没能开始：' + r.error); return; } $('modal').classList.remove('show'); toast('已开始'); conOff = 0; userPicked = false; poll(); diff --git a/ml/monitor.py b/ml/monitor.py index d1b577c..850b373 100644 --- a/ml/monitor.py +++ b/ml/monitor.py @@ -363,7 +363,7 @@ def status(): STAGES = pstate.ORDER ENV_KEYS = {'N_TASKS': r'\d+', 'NPROC': r'\d+', 'TARGET_SAMPLES': r'\d+', 'CFG': r'ml/configs/(base|wta|small)\.json', 'CONSENSUS': r'[01]?', 'ROUNDS': r'\d+', 'APG_STEPS': r'\d+', 'MINE_N': r'\d+', 'MIN_GAP_MM': r'\d+(\.\d+)?', - 'FRESH': r'[01]', 'R2_STRICT': r'[01]', 'AUTO_WTA': r'[01]'} + 'FRESH': r'[01]', 'FRESH_FROM': r'(setup|kin|r4|gate1|gen|prep|train|m5|apg|mine|export|bench|accept)', 'R2_STRICT': r'[01]', 'AUTO_WTA': r'[01]'} CONSOLE = os.path.join(ROOT, RUNS, 'console.log') @@ -419,6 +419,29 @@ def lock_info(): return dict(lk, alive=pstate.lock_alive(lk), age_s=time.time() - lk.get('beat', 0)) +NT_NAMES = "'node.exe','xargs.exe','bash.exe','sh.exe','timeout.exe'" +NT_SWEEP = (r"pipeline\.sh|run_prep\.sh|run_parallel\.sh|build_dataset\.cjs|" + r"gen_dataset\.cjs|teacher\.cjs|policy_bench\.cjs") + + +def sweep_nt(keep_pid): + """taskkill /T 在 Git Bash 下只杀到 bash，xargs/node 孙进程会逃（10-18 实测：停止后还有 28 个 + gen_dataset 在跑；当晚二次踩到：两批 build_dataset 抢同一批 shard）。按命令行补扫整条流水线， + keep_pid 的祖先链（= 控制台自己拉起的 monitor bash）不动。先杀包装再杀 node，否则 node 一死 + xargs 会立刻补下一个 shard。返回仍存活个数，-1 = pwsh 没跑成。""" + q = ("$anc=@(); $c=%d; while ($c -gt 0 -and ($p=Get-CimInstance Win32_Process -Filter \"ProcessId=$c\" " + "-ErrorAction SilentlyContinue)) { $anc+=$p.ProcessId; $c=$p.ParentProcessId }; " + "$m={ $_.Name -in %s -and $_.CommandLine -match '%s' -and $anc -notcontains $_.ProcessId }; " + "foreach ($x in @(Get-CimInstance Win32_Process | Where-Object $m | " + "Sort-Object { if ($_.Name -eq 'node.exe') { 1 } else { 0 } })) " + "{ Stop-Process -Id $x.ProcessId -Force -ErrorAction SilentlyContinue }; " + "Start-Sleep -Seconds 2; " + "@(Get-CimInstance Win32_Process | Where-Object $m).Count" % (int(keep_pid), NT_NAMES, NT_SWEEP)) + r = subprocess.run(['pwsh', '-NoProfile', '-Command', q], capture_output=True, text=True, timeout=90) + lines = [s for s in (r.stdout or '').split() if s.isdigit()] + return int(lines[-1]) if lines else -1 + + def kill_session(sid): """POSIX：结束会话 sid 里的所有进程（含 timeout 另起的进程组）。""" pids = [] @@ -496,14 +519,14 @@ class Job: with self.lock: if os.name == 'nt': subprocess.run(['taskkill', '/T', '/F', '/PID', str(pid)], capture_output=True) - # Git Bash 下 taskkill /T 只杀到 bash，node 孙进程会逃掉（2026-10-18 实测：点了停止还有 28 个 - # gen_dataset 在跑）。按命令行补扫：只动本仓库 ml/ 下的脚本，绝不碰用户自己的 node 进程。 + # 补扫漏网的子孙；有残留就写进网页日志，别让用户以为停干净了 try: - q = ("Get-CimInstance Win32_Process -Filter \"Name='node.exe'\" | " - "Where-Object { $_.CommandLine -match 'gen_dataset\\.cjs|run_parallel\\.sh|teacher\\.cjs|policy_bench\\.cjs' } | " - "ForEach-Object { Stop-Process -Id $_.ProcessId -Force -ErrorAction SilentlyContinue }") - subprocess.run(['pwsh', '-NoProfile', '-Command', q], capture_output=True, timeout=60) - except Exception: pass + left = sweep_nt(os.getpid()) + except Exception: + left = -1 + if left != 0: + with open(CONSOLE, 'a', encoding='utf-8') as f: + f.write(f'\n[停止补扫：仍有 {left} 个流水线子进程存活（-1=补扫没跑成），把这条发我]\n') else: if ours: kill_session(pid) diff --git a/ml/pipeline.sh b/ml/pipeline.sh index 4cd2c6f..89457d6 100755 --- a/ml/pipeline.sh +++ b/ml/pipeline.sh @@ -186,7 +186,7 @@ case $stage in # 启动前自检（审查 N-10：不能等到 train 才发现这个 Python 没有 torch） $PY -c "import torch" 2>/dev/null || { echo "当前 Python（$($PY -c 'import sys;print(sys.executable)')）没有 torch：请先激活训练环境或设 PYTHON=..."; exit 1; } command -v node >/dev/null || { echo "找不到 node"; exit 1; } - fl=""; [ "${FRESH:-0}" = 1 ] && fl="--fresh" + fl=""; [ "${FRESH:-0}" = 1 ] && fl="--fresh"; [ -n "${FRESH_FROM:-}" ] && fl="--fresh-from=$FRESH_FROM" $PY -m ml.pstate reset ${PLAN// /,} $fl # 续跑 / 参数变了作废受影响阶段（旧产物移到 ml/archive/<时间>/） TODO=$($PY -m ml.pstate todo) echo "计划：$PLAN"; echo "待跑：${TODO:-（全部已完成）}" diff --git a/ml/pstate.py b/ml/pstate.py index 548d0fa..61cce6d 100644 --- a/ml/pstate.py +++ b/ml/pstate.py @@ -1,6 +1,6 @@ """流水线状态与互斥锁（单一写者：所有对 ml/runs/pipeline.json / pipeline.lock 的写都经过本模块，原子替换）。 pipeline.sh 自动调用，一般不需要手动运行： - python -m ml.pstate reset <阶段1,...> [--fresh] # `all` 开始：保留 rc=0 的已完成阶段（续跑）；参数变了只作废受影响的阶段；--fresh 全部作废 + python -m ml.pstate reset <阶段1,...> [--fresh] # `all` 开始：保留 rc=0 的已完成阶段（续跑）；参数变了只作废受影响的阶段；--fresh 全部作废；--fresh-from=<阶段> 只作废该阶段及之后 python -m ml.pstate todo # 打印计划中尚未完成的阶段（空格分隔） python -m ml.pstate begin <阶段> [键=值 ...] # 阶段开始 python -m ml.pstate set <阶段> 键=值 ... # 阶段中途更新 @@ -194,11 +194,12 @@ def main(): if cmd == 'reset': plan = rest[0].split(',') if rest and rest[0] else [] pn = params_now(); prev = d.get('params') or {} - changed = [] + changed = [k for k in PARAM_KEYS if prev and prev.get(k) != pn[k]] + ff = next((r.split('=', 1)[1] for r in rest if r.startswith('--fresh-from=')), '') if '--fresh' in rest: bad = set(ORDER) else: - changed = [k for k in PARAM_KEYS if prev and prev.get(k) != pn[k]] bad = {st for k in changed for st in AFFECTS[k]} + if ff in ORDER: bad |= set(ORDER[ORDER.index(ff):]) if changed: print('参数与上次不同：' + json.dumps({k: (prev.get(k), pn[k]) for k in changed}, ensure_ascii=False) + '；受影响的阶段重跑：' + ' '.join(st for st in ORDER if st in bad), file=sys.stderr) keep = {k: v for k, v in d.get('stages', {}).items() if v.get('rc') == 0 and k not in bad} diff --git a/ml/train.py b/ml/train.py index 2706eab..17f3cda 100644 --- a/ml/train.py +++ b/ml/train.py @@ -117,7 +117,9 @@ def main(): print(json.dumps(rec), flush=True); log.write(json.dumps(rec) + '\n'); log.flush() if not a.no_eval and (step + 1) % tc['eval_every'] == 0: model.eval() - res, _ = evaluate(model, kin, val_tasks, dev) + # bs=64 + 归还缓存池：WDDM 卡满时 15.9 GiB 的评估峰值会被驱动换出到内存，eval 从 2 分钟变 1 小时 + res, _ = evaluate(model, kin, val_tasks, dev, bs=64) + torch.cuda.empty_cache() res['step'] = step + 1; res['t'] = time.time() print('EVAL', json.dumps(res), flush=True); log.write(json.dumps({'eval': res}) + '\n'); log.flush() save('last.pt', step + 1)