diff --git a/docs/ML运行手册审查.md b/docs/ML运行手册审查.md index e05d1ea..1774431 100644 --- a/docs/ML运行手册审查.md +++ b/docs/ML运行手册审查.md @@ -40,12 +40,19 @@ 4. APG 收敛与机时（batch 48 下首测）→ APG 日志。 5. 真模型 R9 → `r9_browser.json`。 +## 三′、首次正式预训练失败（2026-10-18，21850 步）与修正 + +现象：闭环到达 0–0.2%、穿柱 77–97%；MF 残差 2 万步停在 7–8、17k–19k 尖峰到 96；梯度 100% 被裁剪。用 last.pt 实测定位： +1. **多峰头塌成单峰（主因）**：同一状态 8 个候选间标准差 0.030、动作幅度 0.151。FK / 平滑损失加在 `one_step(随机噪声)` 上去对齐这一条老师 Y，最优解是条件均值 E[Y|X]，左绕 / 右绕取平均正好穿柱。→ 改加在 r = t、t ≤ 0.3 的配对去噪估计 x̂ = z − t·u 上。 +2. **MeanFlow 目标失控**：∂u/∂t 均方根 6–12，而 v 约 1.1，目标被自激的 JVP 项主导。时间嵌入 sin(1000·t) 太陡，JVP 又在 bf16 下算。→ 最高角频率 50；MF 分支 fp32；前 5000 步 p_neq = 0。 +3. **闭环分布偏移被低估（推断）**：增广只有 1°。→ 增广 40% / 3° / 12 步收回；新增 DAgger 阶段（策略走到的状态请老师重规划，2 轮）；开训前加老师轨迹回放自检；每次评估加开环误差与候选多样性。 + ## 四、正式训练（三段，每段跑完回传） | 段 | 命令 | 耗时 | 回传 | |---|---|---|---| | 数据 | `FROM=kin TO=prep bash ml/pipeline.sh all` | 数小时 | `ml/data/stats.json`、`ml/data/r2.json`、`ml/runs/r4.json` | -| 训练 | `FROM=train TO=m5 bash ml/pipeline.sh all` | 约 3–6 小时 | `ml/runs/m5_gate_*.json` | +| 训练（含 DAgger） | `FRESH_FROM=prep FROM=prep TO=m5 bash ml/pipeline.sh all`（增广改了，样本要重打包；gen 不重跑） | 约 9–11 小时（未实测：prep 0.5 + 预训练约 5（fp32 变慢）+ DAgger 约 3.5 + m5 0.3） | `ml/runs/replay_teacher.json`、`ml/runs/base/log.jsonl`、`ml/runs/m5_gate_*.json` | | 后训练与评测 | `FROM=apg TO=accept bash ml/pipeline.sh all` | 约 10–20 小时 | `ml/runs/acceptance.json`、`policy_bench.json`、`r9_browser.json` | 中途出错：把出错信息发我，修完用 `FROM=<出错阶段>` 接着跑，前面的不重做。 diff --git a/ml/README.md b/ml/README.md index 7ca36ea..ff04097 100644 --- a/ml/README.md +++ b/ml/README.md @@ -25,6 +25,7 @@ | 4 生成数据 | `bash ml/pipeline.sh gen` | 不共识约 1 小时；共识约 4–5 小时 | `ml/data/shard_*.jsonl`、`stats.json` | | 5 打包样本 | `bash ml/pipeline.sh prep` | 15–30 分钟 | `ml/data/samples/`、`ml/data/r2.json` | | 6 预训练 | `bash ml/pipeline.sh train` | 测速 2000 步约 6 分钟，正式约 3 小时 | `ml/runs/base/best.pt` | +| 6′ DAgger 纠偏 | `bash ml/pipeline.sh dagger` | 未实测，估 3–4 小时：2 轮 ×（采起点 5 分钟 + 老师重规划约 30 分钟 + 打包 + 微调 15000 步） | `ml/runs/base_dagger.ckpt`、`ml/runs/base_dag2/best.pt` | | 7 M5 门禁 | `bash ml/pipeline.sh m5` | 估算 10–20 分钟 | `ml/runs/m5_gate_base.json` | | 8 APG 微调 | `bash ml/pipeline.sh apg` | 未实测：展开长度 60–150 次推理随批变化，以监控页前 50 步换算的 ETA 为准 | `ml/runs/apg/best.pt` | | 9 挖错循环 | `bash ml/pipeline.sh mine` | 未实测：每轮 = 5 万题作答 + APG，以第 1 轮实际耗时 × 轮数为准 | `ml/runs/current.pt` | @@ -127,10 +128,22 @@ bash ml/pipeline.sh prep bash ml/pipeline.sh train # MeanFlow 头 → ml/runs/bench_base/、ml/runs/base/ ``` +开训前先跑老师轨迹回放自检（`ml/runs/replay_teacher.json`，约 1 分钟）：老师轨迹放进 PyTorch 仿真口径，到达 ≥ 95%、碰撞 ≤ 1% 才开训，否则 rc 5 停——口径对不上时闭环指标不可信。 + 先跑 2000 步测吞吐（`ml/runs/bench_base/log.jsonl` 的 `samples_per_s`，`eta_h` 是剩余小时数），再正式训练到 60000 步或闭环到达率连续 5 次不升为止。 - **通过线（R10）：`samples_per_s` ≥ 5000，即 60000 步 ≤ 3.5 小时**。参考：3090、batch 1024 实测 6.2 k 样本/s ≈ 2.8 小时。MeanFlow 的 JVP 双前向使训练受算力限制，加大 batch 不会提高样本/s；低于通过线时把 `configs/base.json` 的 `fk_caps` 从 96 降到 48，仍不够再把 `steps` 降到 40000 -- 训练中每 2000 步打印一行 `EVAL {...}`：val 任务闭环（部署节奏、不开安全层）`success`、`collision_rate`、`time_ok_1p3`、`start_fail`。时限 = 老师时长 × 2（至少 3 s） +- 训练中每 2000 步打印一行 `EVAL {...}`：val 任务闭环（部署节奏、不开安全层）`success`、`collision_rate`、`time_ok_1p3`、`start_fail`。时限 = 老师时长 × 2（至少 3 s）；另有开环三项（val 样本、老师状态上）：`ol_fk_mm` 候选 0 的 16 步末端误差、`ol_best_mm` 8 个候选里最好的一条、`ol_div` 候选多样性（候选间标准差 / 动作幅度，塌成单峰时接近 0）。best.pt 与早停按「到达率 − 0.001 × ol_fk_mm」 +- 损失（2026-10-18 首次正式预训练后改）：FK / 平滑只加在 MF 分支 r = t 且 t ≤ 0.3 的配对去噪估计 x̂ = z − t·u 上（原来加在随机噪声的一步采样上，等于回归条件均值，多峰塌成单峰）；时间嵌入最高角频率 50（原 1000）；MF 分支与 JVP 走 fp32（TF32 矩阵乘）；前 5000 步 p_neq = 0。fp32 会让吞吐低于原 6.2 k/s，R10 以新实测为准 +- 数据增广：40% 样本加 σ = 3° 噪声、12 步收回（原 30% / 1° / 8 步），收回途中须无碰 + +## 6′. DAgger 纠偏 + +```bash +bash ml/pipeline.sh dagger +``` + +用预训练 best.pt 在 train 任务上闭环跑（关安全层），每题取「首次碰撞前 0.2 s」和一个随机时刻的状态做起点（须无碰、间隙 > `MIN_GAP_MM`、离目标 > 5 mm），老师从这些起点重新规划（`gen_dataset.cjs --tasks`，与主数据同认证、同时间律），打包后与主数据一起微调。默认 2 轮（`DAG_ROUNDS`），每轮 8000 题（`DAG_N`）、微调 15000 步（`DAG_STEPS`，lr 3e-4）。老师只用于训练标注，部署仍无老师。M5 门禁读 `ml/runs/<头>_dagger.ckpt` 指向的模型；APG / 挖错的 BC 项也用 DAgger 样本。可断点续跑（采集、老师 shard、打包、微调各有完成标记）。 - 显存预算**浮动**：整卡留 1 GB 给系统（`VRAM_RESERVE_GIB`，默认 1），训练与 APG 进程启动时按"整卡 − nvidia-smi 已用 − 余量"和"CUDA 空闲 − 余量"两者取小自设本进程上限（`ml/common/vram.py`）。占显存的应用多，上限就自动变小，不用改配置；想按定值对拍给 `VRAM_FRAC`（旧口径 0.78）。日志 `vram` 字段记峰值；预训练 `batch=1024` 实测峰值 2.76 GiB、APG `batch=48` 13.08 GiB。显存不够把 `batch` 改成 512。中断后再跑 `bash ml/pipeline.sh train`（或 `all`）会自动从 `ml/runs/<头>/last.pt` 续训（每 10 分钟存一次），测速那 2000 步做过就跳过。训练发散时不会用发散的权重覆盖 `last.pt` / `best.pt`。 ## 7. M5 门禁：选定动作头 diff --git a/ml/configs/base.json b/ml/configs/base.json index b3695d7..20269b3 100644 --- a/ml/configs/base.json +++ b/ml/configs/base.json @@ -4,7 +4,7 @@ "batch": 1024, "lr": 1e-3, "warmup": 1000, "steps": 60000, "wd": 0.01, "betas": [0.9, 0.95], "amp": "bf16", "grad_clip": 1.0, "w_mf": 1.0, "w_fk": 0.5, "w_smooth": 0.05, "w_value": 0.1, "w_sub": 0.1, - "p_neq": 0.25, "fk_caps": 96, + "p_neq": 0.25, "p_neq_warm": 5000, "fk_tmax": 0.3, "fk_caps": 96, "ckpt_min": 10, "eval_every": 2000, "eval_tasks": 512, "log_every": 50, "patience_evals": 5 } } diff --git a/ml/configs/small.json b/ml/configs/small.json index 5e7c200..2c6c3b0 100644 --- a/ml/configs/small.json +++ b/ml/configs/small.json @@ -4,7 +4,7 @@ "batch": 1024, "lr": 1e-3, "warmup": 1000, "steps": 60000, "wd": 0.01, "betas": [0.9, 0.95], "amp": "bf16", "grad_clip": 1.0, "w_mf": 1.0, "w_fk": 0.5, "w_smooth": 0.05, "w_value": 0.1, "w_sub": 0.1, - "p_neq": 0.25, "fk_caps": 96, + "p_neq": 0.25, "p_neq_warm": 5000, "fk_tmax": 0.3, "fk_caps": 96, "ckpt_min": 10, "eval_every": 2000, "eval_tasks": 512, "log_every": 50, "patience_evals": 5 } } diff --git a/ml/configs/smoke.json b/ml/configs/smoke.json index 7fea377..e7d0ecb 100644 --- a/ml/configs/smoke.json +++ b/ml/configs/smoke.json @@ -29,15 +29,36 @@ "eval_every": 100, "eval_tasks": 16, "log_every": 10, - "patience_evals": 5 + "patience_evals": 5, + "p_neq_warm": 100, + "fk_tmax": 0.3 }, "apg": { - "batch": 8, "lr": 0.0001, "steps": 10, "grad_clip": 1.0, "amp": "bf16", - "plans": 20, "plans_max": 30, "seg_plans": 5, + "batch": 8, + "lr": 0.0001, + "steps": 10, + "grad_clip": 1.0, + "amp": "bf16", + "plans": 20, + "plans_max": 30, + "seg_plans": 5, "margin_train_mm": 3.0, - "w_tgt": 1.0, "w_rot": 0.3, "w_col": 2.0, "w_smooth": 0.02, "w_len": 0.001, "w_bc": 0.5, - "cat_weights": {"A": 0.4, "B": 1.0, "C": 1.6, "D": 0.6, "Dr": 0.6}, - "ckpt_min": 10, "eval_every": 1000, "eval_tasks": 4 + "w_tgt": 1.0, + "w_rot": 0.3, + "w_col": 2.0, + "w_smooth": 0.02, + "w_len": 0.001, + "w_bc": 0.5, + "cat_weights": { + "A": 0.4, + "B": 1.0, + "C": 1.6, + "D": 0.6, + "Dr": 0.6 + }, + "ckpt_min": 10, + "eval_every": 1000, + "eval_tasks": 4 }, "_note": "冒烟与自检专用（bash ml/pipeline.sh selftest / smoke）：只验证流程能跑通，数值无意义。apg 段必须有：selftest.sh 第 6 项原先不带 --config，走的是正式 apg.json 的 batch=128，本机实测第一次反向就 OOM；改成正式值 48 也要约 13 GiB，自检不必吃这么多。" } \ No newline at end of file diff --git a/ml/configs/wta.json b/ml/configs/wta.json index 407e9a5..91c8897 100644 --- a/ml/configs/wta.json +++ b/ml/configs/wta.json @@ -4,7 +4,7 @@ "batch": 1024, "lr": 1e-3, "warmup": 1000, "steps": 60000, "wd": 0.01, "betas": [0.9, 0.95], "amp": "bf16", "grad_clip": 1.0, "w_mf": 1.0, "w_fk": 0.5, "w_smooth": 0.05, "w_value": 0.1, "w_sub": 0.1, - "p_neq": 0.25, "fk_caps": 96, + "p_neq": 0.25, "p_neq_warm": 5000, "fk_tmax": 0.3, "fk_caps": 96, "ckpt_min": 10, "eval_every": 2000, "eval_tasks": 512, "log_every": 50, "patience_evals": 5 } } diff --git a/ml/dagger/__init__.py b/ml/dagger/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/ml/dagger/collect.py b/ml/dagger/collect.py new file mode 100644 index 0000000..49e8779 --- /dev/null +++ b/ml/dagger/collect.py @@ -0,0 +1,75 @@ +"""DAgger 状态采集（2026-10-18 首次正式预训练后新增）：用当前策略在 train 任务上闭环跑（关安全层）， +从它自己走到的状态里挑起点，交给老师重新规划 → 这些"策略会去、老师数据里没有"的状态拿到正确标签。 + python -m ml.dagger.collect --ckpt ml/runs/base/best.pt --samples ml/data/samples --n 8000 --seed 1 --parts 28 --out ml/data/dag1 +每题最多取 2 个起点： + - 撞了的题：首次碰撞前 0.2 s 的状态（策略"快要撞上"的地方，最有用）； + - 另取一个 [0.3 s, 首撞前或结束] 内均匀随机时刻的状态（偏离老师的一般状态）。 +起点必须无碰且间隙 > MIN_GAP_MM（老师从碰撞态出发规划不出来，与 build_dataset 的增广接受同一地板），且离目标 > 5 mm（否则 DLS 接管，不需要老师）。 +输出 tasks_XX.jsonl（按 --parts 均分，供 gen_dataset.cjs --tasks 并行），每行 {q0, T, R, cat, src}。 +为什么需要：纯行为克隆在闭环里会偏出老师分布（首次 21850 步闭环到达 0%），原方案指望 APG/挖错修，但它们没法从 0% 起步。 +老师只用于训练标注，部署仍然无老师。""" +import argparse, json, os, random +import torch + +from ml.common.kin_torch import Kin +from ml.data import load_tasks +from ml.model.policy import build +from ml.sim.closed_loop import rollout, goal_err, tasks_to_tensors, DT + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument('--ckpt', required=True); ap.add_argument('--samples', nargs='+', default=['ml/data/samples']) + ap.add_argument('--n', type=int, default=8000); ap.add_argument('--seed', type=int, default=1) + ap.add_argument('--parts', type=int, default=8); ap.add_argument('--out', required=True) + ap.add_argument('--bs', type=int, default=64) + ap.add_argument('--device', default='cuda' if torch.cuda.is_available() else 'cpu') + a = ap.parse_args() + gap_min = float(os.environ.get('MIN_GAP_MM') or 3) + os.makedirs(a.out, exist_ok=True) + kin = Kin(device=a.device) + ck = torch.load(a.ckpt, map_location=a.device) + assert ck.get('fp') in (None, kin.fp), '碰撞模型指纹与 checkpoint 不符' + model = build(ck['cfg']).to(a.device); model.load_state_dict(ck['model']); model.eval() + pool = [t for t in load_tasks(a.samples[:1]) if t['split'] == 'train' and t.get('cat') not in ('D',)] + rng = random.Random(a.seed); rng.shuffle(pool); tasks = pool[:a.n] + out, stat = [], {'tasks': len(tasks), 'collided': 0, 'states': 0, 'rej_gap': 0, 'rej_near': 0} + for i in range(0, len(tasks), a.bs): + tb = tasks[i:i + a.bs] + with torch.no_grad(): + _, per = rollout(model, kin, tb, a.device, K=8, safety=False, seed=a.seed * 100003 + i, return_traj=True) + tr = per['traj'].to(a.device) # [B,T,6] + B, Tn, _ = tr.shape + _, T, R, _, _ = tasks_to_tensors(tb, a.device) + with torch.no_grad(): + g = torch.cat([kin.gaps(tr[:, s:s + 64].reshape(-1, 6), 40.0)['gm'].min(-1).values.view(B, -1) for s in range(0, Tn, 64)], 1) + for b, t in enumerate(tb): + hit = (g[b] <= 0).nonzero() + c = int(hit[0]) if hit.numel() else Tn + if hit.numel(): stat['collided'] += 1 + cand = [] + if hit.numel(): cand.append(max(0, c - int(round(0.2 / DT)))) + lo = int(round(0.3 / DT)) + if c - 1 > lo: cand.append(rng.randint(lo, c - 1)) + for k in cand: + if g[b, k] <= gap_min: stat['rej_gap'] += 1; continue + q = tr[b, k:k + 1] + e, oe = goal_err(kin, q, T[b:b + 1], R[b:b + 1]) + if e.item() < 5 and (not t.get('R') or oe.item() < 5): stat['rej_near'] += 1; continue + out.append({'q0': [round(v, 3) for v in q[0].tolist()], 'T': t['T'], 'R': t.get('R'), 'cat': t.get('cat', 'A'), + 'src': {'tid': t['tid'], 'k': k, 'col': bool(hit.numel())}}) + print(f'[dagger] {min(i + a.bs, len(tasks))}/{len(tasks)} 题，起点 {len(out)}', flush=True) + stat['states'] = len(out) + rng.shuffle(out) + per_part = (len(out) + a.parts - 1) // max(1, a.parts) + for p in range(a.parts): + chunk = out[p * per_part:(p + 1) * per_part] + if not chunk: continue + fn = os.path.join(a.out, 'tasks_%02d.jsonl' % (p + 1)) + open(fn + '.tmp', 'w', encoding='utf-8').write(''.join(json.dumps(x) + '\n' for x in chunk)); os.replace(fn + '.tmp', fn) + json.dump(stat, open(os.path.join(a.out, 'collect.json'), 'w', encoding='utf-8'), indent=1) # 最后写：它存在 = 采集完成 + print(json.dumps(stat, ensure_ascii=False), flush=True) + + +if __name__ == '__main__': + main() diff --git a/ml/dagger/run.sh b/ml/dagger/run.sh new file mode 100644 index 0000000..080d4fe --- /dev/null +++ b/ml/dagger/run.sh @@ -0,0 +1,61 @@ +#!/usr/bin/env bash +# DAgger 纠偏（2026-10-18 首次正式预训练后新增）：bash ml/dagger/run.sh <头标签> <起点 ckpt> +# 每轮：策略闭环采起点（ml.dagger.collect）→ 老师从这些起点重新规划（gen_dataset.cjs --tasks，NPROC 并行） +# → 打包成样本（build_dataset.cjs，与主数据同口径）→ 主数据 + 历轮 DAgger 样本一起微调 DAG_STEPS 步。 +# 环境变量：DAG_ROUNDS（默认 2）、DAG_N（每轮采样题数，默认 8000，每题最多 2 个起点）、DAG_STEPS（每轮微调步数，默认 15000）、 +# DAG_LR（默认 3e-4）、NPROC、MIN_GAP_MM。 +# 产物：ml/data/dag__r/（起点与老师 shard）、ml/data/samples_dag__r/、ml/runs/_dag/； +# 最终模型路径写进 ml/runs/_dagger.ckpt（m5 读它）。 +# 断点续跑：采集以 collect.json 为完成标记；老师 shard 先写 .part 再改名、未写完的从断点续写；打包以 .tasks.jsonl 为完成标记； +# 微调用该轮 last.pt 续训；整轮完成打 ml/runs/_dag/done。 +set -e -o pipefail +cd "$(dirname "$0")/../.." +PY=${PYTHON:-python} +TAG=$1; CK=$2 +[ -n "$TAG" ] && [ -f "$CK" ] || { echo "usage: run.sh （$CK 不存在）"; exit 1; } +R=${DAG_ROUNDS:-2}; N=${DAG_N:-8000}; STEPS=${DAG_STEPS:-15000}; LR=${DAG_LR:-3e-4}; P=${NPROC:-$(nproc)} +CFGF=ml/configs/$TAG.json +SAMPLES="ml/data/samples" +for ((r=1; r<=R; r++)); do + D=ml/data/dag_${TAG}_r$r; SD=ml/data/samples_dag_${TAG}_r$r; OUT=ml/runs/${TAG}_dag$r + SAMPLES="$SAMPLES $SD" + if [ -f $OUT/done ]; then echo "[dagger] 第 $r 轮已完成，跳过"; CK=$OUT/best.pt; continue; fi + if [ -f $D/collect.json ]; then echo "[dagger] 第 $r 轮起点已采集，跳过" + else echo "[dagger] 第 $r 轮：用 $CK 闭环采起点（$N 题）"; $PY -m ml.dagger.collect --ckpt $CK --samples ml/data/samples --n $N --seed $r --parts $P --out $D; fi + cat $D/collect.json + pids=() + for f in $D/tasks_*.jsonl; do + [ -e "$f" ] || continue + k=$(basename $f .jsonl); k=${k#tasks_}; sh=$D/shard_dag_${TAG}_r${r}_$k.jsonl + [ -f $sh ] && continue + ( node ml/gen/gen_dataset.cjs --tasks $f --seed $((9000 + 100 * r + 10#$k)) --out $sh.part 2>> $D/log_$k.txt && mv $sh.part $sh ) & + pids+=($!) + done + if [ ${#pids[@]} -gt 0 ]; then + echo "[dagger] 第 $r 轮：老师重新规划（${#pids[@]} 进程）" + t0=$(date +%s) + while :; do + alive=0; for p in "${pids[@]}"; do kill -0 $p 2>/dev/null && alive=$((alive+1)); done + [ $alive = 0 ] && break; sleep 30 + echo "[dagger 节拍] $(( ($(date +%s) - t0) / 60 )) 分钟 运行中 $alive/${#pids[@]} 已写 $(cat $D/*.part $D/shard_*.jsonl 2>/dev/null | wc -l) 条" + done + for p in "${pids[@]}"; do wait $p || true; done # 判定按产物（同 run_parallel.sh：Git Bash 长跑后 wait 返回码不可靠） + for f in $D/tasks_*.jsonl; do + k=$(basename $f .jsonl); k=${k#tasks_}; sh=$D/shard_dag_${TAG}_r${r}_$k.jsonl + if [ ! -f $sh ]; then + n=0; [ -f $sh.part ] && n=$(wc -l < $sh.part); need=$(wc -l < $f) + if [ "$n" -ge "$need" ]; then mv $sh.part $sh; else echo "[dagger] $sh 未写完（$n/$need），见 $D/log_$k.txt；重跑本阶段从断点续写"; exit 6; fi + fi + done + fi + mkdir -p $SD + todo=(); for sh in $D/shard_dag_*.jsonl; do [ -e "$sh" ] || continue; b=$(basename $sh .jsonl); [ -f $SD/$b.tasks.jsonl ] || todo+=("$sh"); done + [ ${#todo[@]} = 0 ] || printf '%s\n' "${todo[@]}" | xargs -P "$P" -I{} node ml/prep/build_dataset.cjs {} $SD + $PY -m ml.data --index $SD | tail -n 3 + res=""; [ -f $OUT/last.pt ] && res="--resume $OUT/last.pt" + echo "[dagger] 第 $r 轮：微调 $STEPS 步（样本：$SAMPLES）" + $PY -m ml.train --samples $SAMPLES --out $OUT --config $CFGF --init $CK --steps $STEPS --lr $LR --warmup 500 $res + touch $OUT/done; CK=$OUT/best.pt +done +echo "$CK" > ml/runs/${TAG}_dagger.ckpt +echo "[dagger] 完成：后续阶段使用 $CK" diff --git a/ml/eval/replay_teacher.py b/ml/eval/replay_teacher.py new file mode 100644 index 0000000..5448414 --- /dev/null +++ b/ml/eval/replay_teacher.py @@ -0,0 +1,49 @@ +"""老师轨迹回放自检（2026-10-18 首次正式预训练后新增；train 阶段开训前自动跑，约 1 分钟）： +把 gen 写出的老师 50 Hz 轨迹（keep=true）直接放进闭环仿真用的 PyTorch 运动学 / 碰撞 / 到达判据里检查。 +老师轨迹在 JS 侧已认证无碰、已到达；如果这里判出碰撞或未到达，说明仿真口径（kin.json、间隙、到达门槛）和数据对不上， +那么闭环指标本身不可信，训练多少步都没用 —— 直接停下（rc 5）。 + python -m ml.eval.replay_teacher --n 512 --out ml/runs/replay_teacher.json +判据：到达率（cat≠D）≥ 0.95，碰撞率 ≤ 0.01。""" +import argparse, glob, json, os, sys +import torch + +from ml.common.kin_torch import Kin +from ml.sim.closed_loop import goal_err + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument('--shards', default='ml/data/shard_*.jsonl'); ap.add_argument('--n', type=int, default=512) + ap.add_argument('--out', default='ml/runs/replay_teacher.json') + ap.add_argument('--device', default='cuda' if torch.cuda.is_available() else 'cpu') + a = ap.parse_args() + kin = Kin(device=a.device) + recs = [] + for f in sorted(glob.glob(a.shards)): + for line in open(f, encoding='utf-8'): + r = json.loads(line) + if r.get('keep') and r.get('traj') and r.get('cat') != 'D': recs.append(r) + if len(recs) >= a.n: break + if len(recs) >= a.n: break + if not recs: sys.exit('没有可回放的老师轨迹（先跑 gen）') + reach = col = 0; worst_gap = [] + for r in recs: + q = torch.tensor(r['traj'], dtype=torch.float32, device=a.device) + mid = 0.5 * (q[1:] + q[:-1]) + g = kin.gaps(torch.cat([q, mid], 0), 0.5)['gm'].min(-1).values + hit = bool((g <= 0).any()); col += hit; worst_gap.append(g.min().item()) + T = torch.tensor([r['T']], dtype=torch.float32, device=a.device) + R = torch.tensor([r['R'] if r.get('R') else [0.] * 9], dtype=torch.float32, device=a.device) + e, oe = goal_err(kin, q[-1:], T, R) + reach += bool((e < 1).item() and (oe < 0.5).item()) + n = len(recs) + res = {'n': n, 'reach': reach / n, 'collision': col / n, 'min_gap_mm_p10': sorted(worst_gap)[n // 10], + 'pass': reach / n >= 0.95 and col / n <= 0.01, 'fp': kin.fp} + os.makedirs(os.path.dirname(a.out) or '.', exist_ok=True) + json.dump(res, open(a.out, 'w', encoding='utf-8'), indent=1) + print('老师轨迹回放：到达 %.1f%%、碰撞 %.1f%%（%d 条）' % (100 * res['reach'], 100 * res['collision'], n), flush=True) + sys.exit(0 if res['pass'] else 5) + + +if __name__ == '__main__': + main() diff --git a/ml/mining/loop.sh b/ml/mining/loop.sh index 93d61fa..e280053 100755 --- a/ml/mining/loop.sh +++ b/ml/mining/loop.sh @@ -12,13 +12,14 @@ ROUNDS=${1:-6}; STEPS=${2:-1000}; N=${3:-50000} mkdir -p ml/runs/mine [ -f ml/runs/gate_tasks.jsonl ] || $PY -m ml.mining.mine_round --make-gate ${MINE_GATE_N:-2000} --seed 777 --out ml/runs [ -f ml/runs/current.pt ] || cp ml/runs/apg/best.pt ml/runs/current.pt -SAMPLES="ml/data/samples"; FIXED=""; small=0 +BASE_S="ml/data/samples $(ls -d ml/data/samples_dag_* 2>/dev/null | tr '\n' ' ')" # 2026-10-18：BC 正则也用 DAgger 样本 +SAMPLES="$BASE_S"; FIXED=""; small=0 for ((r=1; r<=ROUNDS; r++)); do D=ml/runs/mine/r$r; mkdir -p $D if [ -f $D/gate.json ]; then gain=$($PY -c "import json;print(json.load(open('$D/gate.json'))['gain_pp'])") echo "round $r 已完成（gain ${gain} pp），跳过"; FIXED="--fixed $D/gate_fixed.jsonl" - [ -s ml/data/e2/shard_e2_r$r.jsonl ] && SAMPLES="ml/data/samples ml/data/samples_e2" + [ -s ml/data/e2/shard_e2_r$r.jsonl ] && SAMPLES="$BASE_S ml/data/samples_e2" if $PY -c "import sys;sys.exit(0 if $gain < 0.5 else 1)"; then small=$((small+1)); else small=0; fi if [ $small -ge 2 ]; then echo "连续 2 轮提升 < 0.5 pp，停止"; break; fi continue @@ -35,7 +36,7 @@ for ((r=1; r<=ROUNDS; r++)); do if [ -s $D/e2_tasks.jsonl ]; then [ -f $E2F ] || { node ml/gen/gen_dataset.cjs --tasks $D/e2_tasks.jsonl --seed $((5000+r)) --out $E2F.part && mv $E2F.part $E2F; } node ml/prep/build_dataset.cjs ml/data/e2/shard_e2_r$r.jsonl ml/data/samples_e2 - SAMPLES="ml/data/samples ml/data/samples_e2" + SAMPLES="$BASE_S ml/data/samples_e2" fi res=""; [ -f ml/runs/apg_r$r/last.pt ] && res="--resume ml/runs/apg_r$r/last.pt" $PY -m ml.apg.apg_finetune --init ml/runs/current.pt --samples $SAMPLES --pool $D/pool.jsonl --out ml/runs/apg_r$r --steps $STEPS $res diff --git a/ml/model/meanflow.py b/ml/model/meanflow.py index 71c31e5..3b0324f 100644 --- a/ml/model/meanflow.py +++ b/ml/model/meanflow.py @@ -1,6 +1,12 @@ """MeanFlow 损失（Geng et al. 2025；MP1 配方：25% 样本 r≠t，自适应损失加权）。 约定：t=0 为数据、t=1 为噪声；z_t = (1−t)·x + t·ε，瞬时速度 v = ε − x；网络输出平均速度 u(z,r,t)。 -目标 u_tgt = v − (t − r)·du/dt，du/dt = v·∂_z u + ∂_t u 用 torch.func.jvp 一次前向得到（切向量 (v, 0, 1)）。""" +目标 u_tgt = v − (t − r)·du/dt，du/dt = v·∂_z u + ∂_t u 用 torch.func.jvp 一次前向得到（切向量 (v, 0, 1)）。 +2026-10-18 首次正式预训练后改： + - 整个 MF 分支（解码器前向 + JVP）在 fp32 下算：目标里含网络自己的导数，bf16 下 JVP 误差直接变成目标噪声； + - return_aux=True 时返回 (z, t, r, u)，训练用 r==t 且 t 小的样本做"配对去噪估计" x̂ = z − t·u 来加 FK / 平滑损失。 + 原来把 FK 加在 one_step(随机噪声) 上、对齐这一条样本的老师 Y：对任意噪声都回归同一个 Y，最优解是条件均值 E[Y|X]， + 多峰头被训成单峰（实测 last.pt 8 个候选间标准差 0.030，动作幅度 0.151），左绕/右绕取平均正好穿柱。 + x̂ 由 z_t 里带着的那一条 x 决定（t 小时 z_t 几乎就是 x），不会把不同模式拉到一起。""" import torch from torch.func import jvp @@ -14,18 +20,23 @@ def sample_rt(B, device, dtype, p_neq=0.25, mu=-0.4, sigma=1.0): return r, t -def meanflow_loss(model, mem, x, p_neq=0.25, ada_p=1.0, ada_c=1e-3): - B = x.shape[0] - r, t = sample_rt(B, x.device, x.dtype, p_neq) - e = torch.randn_like(x) - z = (1 - t)[:, None, None] * x + t[:, None, None] * e - v = e - x - fn = lambda z_, r_, t_: model.decode(mem, z_, r_, t_) - u, dudt = jvp(fn, (z, r, t), (v, torch.zeros_like(r), torch.ones_like(t))) - u_tgt = (v - (t - r)[:, None, None] * dudt).detach() - err = (u - u_tgt).pow(2).mean(dim=(1, 2)) - w = 1.0 / (err.detach() + ada_c).pow(ada_p) - return (w * err).mean(), err.mean().detach() +def meanflow_loss(model, mem, x, p_neq=0.25, ada_p=1.0, ada_c=1e-3, return_aux=False): + mem, x = mem.float(), x.float() + with torch.autocast('cuda', enabled=False): + B = x.shape[0] + r, t = sample_rt(B, x.device, x.dtype, p_neq) + e = torch.randn_like(x) + z = (1 - t)[:, None, None] * x + t[:, None, None] * e + v = e - x + fn = lambda z_, r_, t_: model.decode(mem, z_, r_, t_) + u, dudt = jvp(fn, (z, r, t), (v, torch.zeros_like(r), torch.ones_like(t))) + u_tgt = (v - (t - r)[:, None, None] * dudt).detach() + err = (u - u_tgt).pow(2).mean(dim=(1, 2)) + w = 1.0 / (err.detach() + ada_c).pow(ada_p) + loss = (w * err).mean() + if return_aux: + return loss, err.mean().detach(), {'z': z, 't': t, 'r': r, 'u': u} + return loss, err.mean().detach() def one_step(model, mem, e): diff --git a/ml/model/policy.py b/ml/model/policy.py index f41e6f3..3166f01 100644 --- a/ml/model/policy.py +++ b/ml/model/policy.py @@ -55,10 +55,17 @@ class DecBlock(nn.Module): return x + self.ff(self.n3(x)) +TEMB_WMAX = 50.0 # 时间嵌入最高角频率（rad / 单位 t） + + def time_emb(t, dim): + """2026-10-18 首次正式预训练后改（step 21850 闭环到达 0%）：原实现照扩散模型写成 sin(1000·t·f)，f∈[1e-3, 1]， + 最高角频率 1000 —— t∈[0,1] 上网络对 t 极其敏感，实测 last.pt 的 ∂u/∂t 均方根 6–12，而 v 只有约 1.1， + MeanFlow 目标 u_tgt = v − (t−r)·du/dt 被自激的 JVP 项主导：MF 残差 2 万步一直停在 7–8、17k–19k 尖峰到 96、梯度 100% 被裁剪。 + 改为角频率在 [0.5, 50] 几何分布（最快一维在 [0,1] 上约转 8 圈，足够分辨 t，导数量级降 20 倍）。""" half = dim // 2 - fr = torch.exp(-math.log(1000.0) * torch.arange(half, device=t.device, dtype=t.dtype) / half) - a = t[:, None] * 1000.0 * fr[None] + w = torch.exp(torch.linspace(math.log(0.5), math.log(TEMB_WMAX), half, device=t.device, dtype=t.dtype)) + a = t[:, None] * w[None] return torch.cat([torch.sin(a), torch.cos(a)], -1) diff --git a/ml/monitor.html b/ml/monitor.html index ce431a7..de0cf9e 100644 --- a/ml/monitor.html +++ b/ml/monitor.html @@ -208,7 +208,7 @@
后台执行 bash ml/pipeline.sh all，从对拍一路跑到验收。已完成的步骤自动跳过；训练每 10 分钟存档、生成按条续写，断了再点一次就从断点接着跑。网页和控制台窗口都可以关，训练不受影响。
- + @@ -378,7 +378,7 @@ let logyOn = false, job = null, userPicked = false, conOff = 0; 下面这份只给示例数据用，数值与 pstate.py 一致。 */ let STAGE_META = [ ['setup', '安装依赖', 0.2], ['kin', '对拍', 0.2], ['r4', '推理延迟', 0.2], ['gate1', '一致性门禁', 0.5], ['gen', '生成老师轨迹', 5.0], - ['prep', '打包样本', 0.5], ['train', '预训练', 3.0], ['m5', '头门禁', 0.5], ['apg', 'APG 微调', 4.0], ['mine', '挖错循环', 9.0], + ['prep', '打包样本', 0.5], ['train', '预训练', 4.5], ['dagger', 'DAgger 纠偏', 3.5], ['m5', '头门禁', 0.5], ['apg', 'APG 微调', 4.0], ['mine', '挖错循环', 9.0], ['export', '导出', 0.1], ['bench', '部署评测', 1.0], ['accept', '验收汇总', 0.5]]; let SM = Object.fromEntries(STAGE_META.map(([k, n, w]) => [k, { n, w, to: Math.max(1, 3 * w) }])); function applyStageMeta(m) { @@ -788,7 +788,7 @@ function demoApi() { const mkPipe = (done, cur, curStartH) => { const t = now(); const st = {}, el = {}; let tt = t - curStartH * 3600; for (let i = done.length - 1; i >= 0; i--) { const [k, dur] = done[i]; st[k] = { start: tt - dur * 3600, end: tt, rc: 0, info: {} }; el[k] = dur * 3600; tt -= dur * 3600; } if (cur) { st[cur] = { start: t - curStartH * 3600, end: null, rc: null, info: {} }; el[cur] = curStartH * 3600; } - return { started: tt, plan: ['kin', 'r4', 'gate1', 'gen', 'prep', 'train', 'm5', 'apg', 'mine', 'export', 'bench', 'accept'], stages: st, elapsed: el, current: cur }; }; + return { started: tt, plan: ['kin', 'r4', 'gate1', 'gen', 'prep', 'train', 'dagger', 'm5', 'apg', 'mine', 'export', 'bench', 'accept'], stages: st, elapsed: el, current: cur }; }; const CON = { train: '$ FROM=kin TO=accept bash ml/pipeline.sh all\n计划：kin r4 gate1 gen prep train m5 apg mine export bench accept\n== node ml/prep/export_kin.cjs\n== node ml/eval/dump_parity.cjs 2000\n== python -m ml.eval.check_parity\n{"PASS": true, "fk_mm": 0.0004, "gap_mm": 0.0011}\n== node ml/gen/consistency.cjs 300 4242\n{"consistentByGateRule": 0.86}\n一致性门禁结果 → CONSENSUS=1\n== bash ml/gen/run_parallel.sh 1786 28 1 --consensus 4\n样本数 1702000 ≥ 目标，完成\n== bash ml/prep/run_prep.sh 28\n== python -m ml.eval.check_r2 --samples ml/data/samples --out ml/data/r2.json\n{"pairs": 250, "label_std_mean": 0.071, "pass": true}\n== python -m ml.train --samples ml/data/samples --out ml/runs/bench_base --steps 2000 --no-eval\nstep 2000 loss 1.912 6150 samples/s\n== python -m ml.train --samples ml/data/samples --out ml/runs/base\n', gen: '$ FROM=kin TO=accept bash ml/pipeline.sh all\n计划：kin r4 gate1 gen prep train m5 apg mine export bench accept\n== node ml/prep/export_kin.cjs\n== python -m ml.eval.check_parity\n{"PASS": true}\n== node ml/gen/consistency.cjs 300 4242\n{"consistentByGateRule": 0.86}\n一致性门禁结果 → CONSENSUS=1\n== bash ml/gen/run_parallel.sh 1786 28 1 --consensus 4\n', diff --git a/ml/monitor.py b/ml/monitor.py index 850b373..c0b8401 100644 --- a/ml/monitor.py +++ b/ml/monitor.py @@ -265,6 +265,21 @@ def stage_progress(st, s): 'eta_s': m['eta_h'] * 3600 if m.get('eta_h') else None} # 训练脚本自己按实测样本/s 算的 if b: return {'frac': b['step'] / (bs + 60000), 'label': f'测速 {b["step"]:,} / {bs:,} 步', 'run': f'bench_{tag}'} return {'frac': 0, 'label': '加载数据中'} + if st == 'dagger': # 2026-10-18 新增：每轮 = 采起点 → 老师重规划 → 打包 → 微调（微调占大头，进度按 0.1/0.25/0.05/0.6 分） + tag, R = info.get('tag', 'base'), int(info.get('rounds', 2)) + k = len(newer(f'{RUNS}/{tag}_dag*/done', t0)); k = min(k, R) + base = f'第 {min(k + 1, R)} / {R} 轮' + if k >= R: return {'frac': 1, 'label': 'DAgger 完成'} + D, run = f'{DATA}/dag_{tag}_r{k + 1}', f'{tag}_dag{k + 1}' + m = last_rec(run) if newer(f'{RUNS}/{run}/log.jsonl', t0) else None + if m and m.get('total'): + return {'frac': (k + 0.4 + 0.6 * m['step'] / m['total']) / R, 'label': f'{base}：微调 {m["step"]:,} / {m["total"]:,} 步', 'run': run, + 'eta_s': m['eta_h'] * 3600 if m.get('eta_h') else None} + if os.path.exists(os.path.join(ROOT, D, 'collect.json')): + nt = sum(LC.count(f) for f in glob.glob(os.path.join(ROOT, D, 'tasks_*.jsonl'))) + nd = sum(LC.count(f) for f in glob.glob(os.path.join(ROOT, D, 'shard_dag_*.jsonl')) + glob.glob(os.path.join(ROOT, D, 'shard_dag_*.jsonl.part'))) + return {'frac': (k + 0.1 + 0.25 * (nd / nt if nt else 0)) / R, 'label': f'{base}：老师重新规划 {nd:,} / {nt:,} 个起点'} + return {'frac': k / R, 'label': f'{base}：闭环采起点'} if st == 'apg': r = last_rec('apg') if newer(f'{RUNS}/apg/log.jsonl', t0) else None eta = None @@ -509,7 +524,7 @@ class Job: pid = self.p.pid if ours else int(lk.get('pid') or 0) if not pid: return {'error': '锁里没有 pid，无法停止'} cur = (pstate.load().get('current') or '') - if cur in ('train', 'm5', 'apg', 'mine'): + if cur in ('train', 'dagger', 'm5', 'apg', 'mine'): open(STOPF, 'w').write(time.strftime('%H:%M:%S')) with open(CONSOLE, 'a', encoding='utf-8') as f: f.write(f'\n[停止请求已发出，等待 {cur} 存档 last.pt…]\n') for _ in range(180): diff --git a/ml/pipeline.sh b/ml/pipeline.sh index 89457d6..123de82 100755 --- a/ml/pipeline.sh +++ b/ml/pipeline.sh @@ -7,6 +7,7 @@ # gen 生成老师轨迹（NPROC 进程；CONSENSUS=1 时加共识过滤；样本不足 TARGET_SAMPLES 自动补生成） # prep 打包训练样本 + R2 标签一致性检查 # train 预训练（先 2000 步摸吞吐，再正式训练）；输出目录随头分开：ml/runs/ +# dagger DAgger 纠偏（2026-10-18 新增）：策略闭环走到的状态请老师重新规划、并入数据微调 DAG_ROUNDS 轮（ml/dagger/run.sh） # m5 头门禁：贴立柱集穿柱率 ≤ 2%（纯策略口径），同时报仿真安全层口径；通过则记为选定模型，不过自动训 WTA 头再判 # apg APG 首轮微调（从选定模型开始） # mine 挖错循环（3–6 轮，含门禁回滚） @@ -45,10 +46,11 @@ SEL=ml/runs/selected_ckpt ckpt() { if [ -n "${CKPT:-}" ]; then echo "$CKPT"; elif [ -f $SEL ]; then cat $SEL; else echo ml/runs/$TAG/best.pt; fi; } stage=${1:-help} run() { echo "== $*" >&2; "$@"; } # 回显走 stderr：gate1 用 tee 把 stdout 存成 consistency.json，回显混进去会让后面读它时 JSONDecodeError -ALL_STAGES="setup kin r4 gate1 gen prep train m5 apg mine export bench accept" +ALL_STAGES="setup kin r4 gate1 gen prep train dagger m5 apg mine export bench accept" stage_info() { case $1 in gen) echo "target_tasks=$(( (N_TASKS + NPROC - 1) / NPROC * NPROC ))" ;; train) echo "tag=$TAG bench_steps=2000" ;; + dagger) echo "tag=$TAG rounds=${DAG_ROUNDS:-2}" ;; m5) echo "tag=$TAG" ;; mine) echo "rounds=$ROUNDS apg_steps=$APG_STEPS" ;; esac; } @@ -78,13 +80,21 @@ trap 'exit 143' TERM INT HUP # 被停止 / 超时：先走 on_exit 记返 # ---- 训练与门禁的单元（m5 自动分支要对两种头各调一次） train_one() { # $1 = 配置文件 local cfg=$1 tag; tag=$(basename "$1" .json) + # 2026-10-18：开训前先把老师轨迹放进 PyTorch 仿真口径里回放——口径对不上时闭环指标不可信，训多少步都白训（rc 5 停） + [ -f ml/runs/replay_teacher.json ] || run $PY -m ml.eval.replay_teacher --n 512 --out ml/runs/replay_teacher.json if [ -f ml/runs/bench_$tag/last.pt ]; then echo "吞吐摸底 ml/runs/bench_$tag 已有，跳过" else run $PY -m ml.train --samples ml/data/samples --out ml/runs/bench_$tag --steps ${BENCH_STEPS:-2000} --no-eval --config $cfg; fi local res=""; [ -f ml/runs/$tag/last.pt ] && { res="--resume ml/runs/$tag/last.pt"; echo "发现 ml/runs/$tag/last.pt → 续训"; } run $PY -m ml.train --samples ml/data/samples --out ml/runs/$tag --config $cfg $res } -m5_one() { # $1 = 头标签；返回 0 通过 / 3 不通过 - local tag=$1 ck=${CKPT:-ml/runs/$1/best.pt} o +dagger_one() { # $1 = 头标签：从该头的预训练 best.pt 起做 DAgger，最终模型路径写进 ml/runs/_dagger.ckpt + local tag=$1 + [ -f ml/runs/$tag/best.pt ] || { echo "DAgger：ml/runs/$tag/best.pt 不存在（预训练没完成）"; return 3; } + run bash ml/dagger/run.sh $tag ml/runs/$tag/best.pt +} +m5_one() { # $1 = 头标签；返回 0 通过 / 3 不通过。起点模型：CKPT > DAgger 结果 > 预训练 best.pt + local tag=$1 ck o + if [ -n "${CKPT:-}" ]; then ck=$CKPT; elif [ -f ml/runs/${1}_dagger.ckpt ]; then ck=$(cat ml/runs/${1}_dagger.ckpt); else ck=ml/runs/$1/best.pt; fi [ -f "$ck" ] || { echo "M5：$ck 不存在（训练发散或没训完）→ 按不通过处理"; return 3; } # 断点续跑：结果文件比检查点新 → 这一半已经评过，跳过 o=ml/runs/m5_${tag}_policy.json; [ -f $o ] && [ $o -nt "$ck" ] && echo "M5：$o 已有，跳过" || run $PY -m ml.sim.closed_loop --ckpt $ck --tasks ml/runs/nc_gate.jsonl --n ${M5_N:-1000} --out $o @@ -154,6 +164,7 @@ case $stage in echo "警告：R2 不通过（近邻标签不自洽，见 ml/data/r2.json）。按默认继续训练，验收汇总会列出" fi ;; train) train_one $CFG ;; + dagger) dagger_one $TAG ;; m5) [ -f ml/runs/nc_gate.jsonl ] || run $PY -m ml.mining.mine_round --make-gate ${M5_N:-1000} --seed 4321 --mix 0,0,1,0 --name nc_gate.jsonl --out ml/runs if m5_one $TAG; then : elif [ "$TAG" != wta ] && [ "${AUTO_WTA:-1}" = 1 ] && [ -z "${CKPT:-}" ]; then @@ -161,11 +172,12 @@ case $stage in echo "M5：$TAG 头不通过 → 自动训练 WTA 回退头（约 3 小时）" $PY -m ml.pstate set m5 branch=wta train_one ml/configs/wta.json + dagger_one wta m5_one wta || { echo "M5：两种头都未通过，把 ml/runs/m5_gate_*.json 发我"; exit 3; } elif [ "${M5_STRICT:-1}" = 0 ]; then echo "M5：不通过，M5_STRICT=0（冒烟）→ 照常继续"; echo ml/runs/$TAG/best.pt > $SEL else echo "M5：不通过，把 ml/runs/m5_gate_*.json 发我"; exit 3; fi ;; apg) ck=$(ckpt); res=""; [ -f ml/runs/apg/last.pt ] && { res="--resume ml/runs/apg/last.pt"; echo "发现 ml/runs/apg/last.pt → 续跑"; } - run $PY -m ml.apg.apg_finetune --init $ck --samples ml/data/samples --out ml/runs/apg ${APG_FIRST_STEPS:+--steps $APG_FIRST_STEPS} $res; cp ml/runs/apg/best.pt ml/runs/current.pt ;; + run $PY -m ml.apg.apg_finetune --init $ck --samples ml/data/samples $(ls -d ml/data/samples_dag_* 2>/dev/null | tr '\n' ' ') --out ml/runs/apg ${APG_FIRST_STEPS:+--steps $APG_FIRST_STEPS} $res; cp ml/runs/apg/best.pt ml/runs/current.pt ;; mine) run bash ml/mining/loop.sh $ROUNDS $APG_STEPS $MINE_N ;; monitor) run $PY -m ml.monitor ${MONITOR_ARGS:-} ;; selftest) run bash ml/selftest.sh ;; diff --git a/ml/prep/build_dataset.cjs b/ml/prep/build_dataset.cjs index 6dfe610..85146e2 100644 --- a/ml/prep/build_dataset.cjs +++ b/ml/prep/build_dataset.cjs @@ -6,7 +6,9 @@ * V [N] 剩余时间（s）/ 5 * SG [N,13] 子目标标签：剩余两个途经点相对当前 q 的差 / 180（不足用终点补）+ 是否需要 * TID[N] 任务编号（tasks.jsonl 行号所在的全局 id，用于划分与闭环验证） - * 增广：30% 样本对 q_t 加 σ=1° 噪声，标签改为 8 步内线性收回老师轨迹（噪声态须无干涉且间隙 > 1 mm）； + * 增广（2026-10-18 首次正式预训练后加大）：40% 样本对 q_t 加 σ=AUG_SIGMA（默认 3°）噪声，标签改为 AUG_BACK（默认 12）步内线性收回老师轨迹； + * 噪声态须无干涉且间隙 > MIN_GAP_MM，收回途中第 3/6/9/12 步也须无干涉（σ 加大后直接线性拉回可能穿过障碍）。 + * 原为 30%、σ=1°、8 步：闭环一偏出 1° 就是没见过的状态，首次 21850 步闭环到达 0%。更大的偏离由 DAgger 阶段补（ml/dagger）。 * 已提交前缀长度 r ∈ {0..3} 均匀随机，前缀取老师接下来 r 步的增量。 */ 'use strict'; const fs = require('fs'), path = require('path'); @@ -28,6 +30,7 @@ const lines = fs.readFileSync(IN, 'utf8').split('\n').filter(Boolean); let nAug = 0, nAugRej = 0, nGapDrop = 0; /* 2026-10-18：认证最小间隙 < 3 mm 的老师轨迹不进训练集；噪声增广接受地板同步为 3 mm（原 1 mm）。 */ const MIN_GAP_MM = +(process.env.MIN_GAP_MM || 3); +const AUG_P = +(process.env.AUG_P || 0.4), AUG_SIGMA = +(process.env.AUG_SIGMA || 3), AUG_BACK = +(process.env.AUG_BACK || 12); for (const line of lines) { const r = JSON.parse(line); if (!r.keep) continue; @@ -43,13 +46,16 @@ for (const line of lines) { for (let t = 0; t <= N; t += 3) { let q = traj[t].slice(), qPrev = (t > 0 ? traj[t - 1] : traj[0]).slice(); let off = [0, 0, 0, 0, 0, 0], aug = false; - if (rnd() < 0.3) { - const qn = q.map((v, j) => { let x = v + gauss(); const t2 = L[O[j]]; if (!t2.cont) x = Math.max(t2.lo, Math.min(t2.hi, x)); return x; }); + if (rnd() < AUG_P) { + const qn = q.map((v, j) => { let x = v + AUG_SIGMA * gauss(); const t2 = L[O[j]]; if (!t2.cont) x = Math.max(t2.lo, Math.min(t2.hi, x)); return x; }); nAug++; - if (S.collideFree(qn) && S.clearance(qn, 5) > MIN_GAP_MM) { off = qn.map((v, j) => v - q[j]); q = qn; qPrev = qPrev.map((v, j) => v + off[j]); aug = true; } - else nAugRej++; + const o2 = qn.map((v, j) => v - q[j]); + const back = k => traj[Math.min(N, t + k)].map((v, j) => v + o2[j] * Math.max(0, 1 - k / AUG_BACK)); + if (S.collideFree(qn) && S.clearance(qn, 5) > MIN_GAP_MM && [3, 6, 9, 12].every(k => k >= AUG_BACK || S.collideFree(back(k)))) { + off = o2; q = qn; qPrev = qPrev.map((v, j) => v + off[j]); aug = true; + } else nAugRej++; } - const pos = k => { const i = Math.min(N, t + k); const w = aug ? Math.max(0, 1 - k / 8) : 0; return traj[i].map((v, j) => v + off[j] * w); }; + const pos = k => { const i = Math.min(N, t + k); const w = aug ? Math.max(0, 1 - k / AUG_BACK) : 0; return traj[i].map((v, j) => v + off[j] * w); }; const y = new Float32Array(H * 6); let prevP = q; for (let k = 1; k <= H; k++) { const pk = pos(k); for (let j = 0; j < 6; j++) y[(k - 1) * 6 + j] = Math.max(-1, Math.min(1, (pk[j] - prevP[j]) / STEP[j])); prevP = pk; } diff --git a/ml/pstate.py b/ml/pstate.py index 61cce6d..c6b88bc 100644 --- a/ml/pstate.py +++ b/ml/pstate.py @@ -26,27 +26,29 @@ LOCK_STALE_S = 300 # 心跳每 60 s 一次；5 分钟没刷新 = 持锁 # 阶段顺序、名称、估时（小时，按 README §耗时表的共识分支上界；网页进度权重即此值） STAGES = [ ('setup', '安装依赖', 0.2), ('kin', '对拍', 0.2), ('r4', '推理延迟', 0.2), ('gate1', '一致性门禁', 0.5), - ('gen', '生成老师轨迹', 5.0), ('prep', '打包样本', 0.5), ('train', '预训练', 3.0), ('m5', '头门禁', 0.5), + ('gen', '生成老师轨迹', 5.0), ('prep', '打包样本', 0.5), ('train', '预训练', 4.5), ('dagger', 'DAgger 纠偏', 3.5), ('m5', '头门禁', 0.5), ('apg', 'APG 微调', 4.0), ('mine', '挖错循环', 9.0), ('export', '导出', 0.1), ('bench', '部署评测', 1.0), ('accept', '验收汇总', 0.5)] ORDER = [s[0] for s in STAGES] EST_H = {s[0]: s[2] for s in STAGES} # m5 不过时流程内自动训 WTA 头（训练 3 h + 门禁），超时按分支后的总量给 TIMEOUT_H = {k: max(1.0, 3 * v) for k, v in EST_H.items()} -TIMEOUT_H['m5'] = 3 * (EST_H['train'] + 2 * EST_H['m5']) +TIMEOUT_H['m5'] = 3 * (EST_H['train'] + EST_H['dagger'] + 2 * EST_H['m5']) # 参数 → 受它影响的阶段（这些阶段即使上次完成也要重跑，其产物先挪进 ml/archive/）。只改挖错轮数不会让 5 小时的生成作废。 _from = lambda st: ORDER[ORDER.index(st):] AFFECTS = {**{k: _from('gen') for k in ('N_TASKS', 'NPROC', 'CONSENSUS', 'MIN_GAP_MM')}, # TARGET_SAMPLES 只决定"补量到多少"：不作废 gen 的已有分片（重跑 gen 会跳过写完的 shard、按差额补量）， # 但样本集会变，所以 prep 及之后作废。2026-10-18 实测：把它当"从 gen 作废"会让 5 小时的 5 万条分片白跑。 'TARGET_SAMPLES': ['gen'] + _from('prep'), - 'CFG': ['r4'] + _from('train'), **{k: _from('mine') for k in ('ROUNDS', 'APG_STEPS', 'MINE_N')}} + 'CFG': ['r4'] + _from('train'), **{k: _from('dagger') for k in ('DAG_ROUNDS', 'DAG_N', 'DAG_STEPS')}, + **{k: _from('mine') for k in ('ROUNDS', 'APG_STEPS', 'MINE_N')}} PARAM_KEYS = list(AFFECTS) # 各阶段产物（相对仓库根，可含通配）；作废 / 从头重跑时整体移到 ml/archive/<时间>/，不删除 PRODUCTS = { 'r4': ['ml/runs/r4', 'ml/runs/r4.json'], 'gate1': ['ml/data/consistency.json'], 'gen': ['ml/data/shard_*', 'ml/data/log_*', 'ml/data/stats.json'], 'prep': ['ml/data/samples', 'ml/data/r2.json'], - 'train': ['ml/runs/base', 'ml/runs/wta', 'ml/runs/small', 'ml/runs/bench_*'], + 'train': ['ml/runs/base', 'ml/runs/wta', 'ml/runs/small', 'ml/runs/bench_*', 'ml/runs/replay_teacher.json'], + 'dagger': ['ml/runs/*_dag[0-9]*', 'ml/runs/*_dagger.ckpt', 'ml/data/dag_*', 'ml/data/samples_dag_*'], 'm5': ['ml/runs/m5_*', 'ml/runs/selected_ckpt'], 'apg': ['ml/runs/apg', 'ml/runs/current.pt'], 'mine': ['ml/runs/mine', 'ml/runs/apg_r*', 'ml/data/e2', 'ml/data/samples_e2'], 'export': ['webapp/models/traj_policy.onnx', 'webapp/models/traj_policy.meta.json'], diff --git a/ml/train.py b/ml/train.py index 17f3cda..d6e1ada 100644 --- a/ml/train.py +++ b/ml/train.py @@ -2,7 +2,13 @@ python -m ml.train --samples ml/data/samples --out ml/runs/base [--config ml/configs/base.json] 吞吐摸底：python -m ml.train --samples ml/data/samples --out ml/runs/bench_base --steps 2000 --no-eval 通过线（方案 R10）：全程 60000 步 ≤ 3.5 h，即 samples_per_s ≥ 5000（3090 实测 6.2 k/s ≈ 2.8 h）；日志 eta_h 为剩余小时数。 -每 eval_every 步做一次闭环验证（ml/sim/closed_loop.py，部署节奏），按 success 存 best.pt；连续 patience_evals 次不升即停。""" +每 eval_every 步做一次闭环验证（ml/sim/closed_loop.py，部署节奏），按 success 存 best.pt；连续 patience_evals 次不升即停。 +2026-10-18 首次正式预训练（21850 步，闭环到达 0–0.2%、穿柱 77–97%）后改： + - FK / 平滑损失不再加在 one_step(随机噪声) 上（那样等于对任意噪声回归同一个 Y → 条件均值，多峰头塌成单峰）， + 改加在 MF 分支里 r==t 且 t ≤ fk_tmax 的样本的配对去噪估计 x̂ = z − t·u 上（见 ml/model/meanflow.py）； + - p_neq 前 p_neq_warm 步为 0（纯 flow matching 先把瞬时速度场学稳，再开 r≠t 的自举目标）； + - 每次评估加开环指标（val 样本上一步采样的 16 步末端 FK 误差 mm、K 个候选的多样性）， + 塌缩或学偏当场可见；best / 早停按 success 为主、开环误差为次（success 同为 0 时也能分出好坏，不会 10000 步就误停）。""" import argparse, json, math, os, sys, time import torch @@ -12,7 +18,7 @@ from ml.common.ckpt import Saver, atomic_save, rng_state, set_rng_state from ml.common import vram from ml.data import Samples, load_tasks from ml.model.policy import build, wta_loss -from ml.model.meanflow import meanflow_loss, one_step +from ml.model.meanflow import meanflow_loss from ml.sim.closed_loop import evaluate @@ -28,28 +34,61 @@ def q_from_feat(X): return torch.atan2(s, c) / D2R +@torch.no_grad() +def open_loop(model, kin, vdata, stepv, fk_idx, n=1024, K=8): + """开环指标（val 样本，老师状态上）： + ol_fk_mm 候选 0 的 16 步末端与老师 Y 末端的胶囊点平均距离（mm）； + ol_best_mm K 个候选里最接近老师的那条的同一误差（多峰时它应明显小于 ol_fk_mm）； + ol_div K 个候选间标准差 / 动作平均幅度（塌成单峰时接近 0；首次正式预训练 last.pt 实测 0.20）。""" + g = torch.Generator(device=vdata.X.device); g.manual_seed(123) + X, Y, _, _ = vdata.batch(min(n, vdata.n), gen=g) + act, _, _ = model.sample_k(X, K) + act = act.float().clamp(-1, 1) + q = q_from_feat(X) + qt = q + (Y * stepv).sum(1) + qh = q[:, None] + (act * stepv).sum(2) # [B,K,6] + pt = fk_points(kin, qt, fk_idx)[:, None] + ph = fk_points(kin, qh.reshape(-1, 6), fk_idx).view(X.shape[0], K, -1, 3) + err = (ph - pt).norm(dim=-1).mean(-1) # [B,K] mm + div = (act.std(1).mean() / act.abs().mean().clamp(min=1e-6)).item() + return {'ol_fk_mm': err[:, 0].mean().item(), 'ol_best_mm': err.min(1).values.mean().item(), 'ol_div': div} + + def main(): ap = argparse.ArgumentParser() ap.add_argument('--samples', nargs='+', default=['ml/data/samples']); ap.add_argument('--out', default='ml/runs/base') ap.add_argument('--config', default='ml/configs/base.json'); ap.add_argument('--steps', type=int, default=None) ap.add_argument('--resume', default=None); ap.add_argument('--no-eval', action='store_true') + ap.add_argument('--init', default=None, help='只载入权重、优化器与步数从 0 开始（DAgger 微调用）') + ap.add_argument('--lr', type=float, default=None); ap.add_argument('--warmup', type=int, default=None) ap.add_argument('--stop-at', type=int, default=0, help='自检用：跑到第 N 步按停止请求处理（存 last.pt、rc 143）') ap.add_argument('--device', default='cuda' if torch.cuda.is_available() else 'cpu'); ap.add_argument('--seed', type=int, default=0) a = ap.parse_args() cfg = json.load(open(a.config)); tc = cfg['train'] if a.steps: tc['steps'] = a.steps + if a.lr: tc['lr'] = a.lr + if a.warmup is not None: tc['warmup'] = a.warmup + if a.init: tc['p_neq_warm'] = 0 # 微调起点已训过 r≠t，不再预热 os.makedirs(a.out, exist_ok=True) torch.manual_seed(a.seed) dev = a.device vram.limit() + if dev.startswith('cuda'): + torch.backends.cuda.matmul.allow_tf32 = True; torch.backends.cudnn.allow_tf32 = True # MF 分支改 fp32 后用 TF32 保吞吐 kin = Kin(device=dev) data = Samples(a.samples, device=dev, split='train') val_tasks = [t for t in load_tasks(a.samples) if t['split'] == 'val'][:tc['eval_tasks']] + try: vdata = Samples(a.samples, device=dev, split='val') + except SystemExit: vdata = None + if vdata is not None and vdata.n == 0: vdata = None print(f'train samples {data.n}, val tasks {len(val_tasks)}, device {dev}', flush=True) model = build(cfg).to(dev) print(f'params {sum(p.numel() for p in model.parameters()) / 1e6:.2f} M', flush=True) opt = torch.optim.AdamW(model.parameters(), lr=tc['lr'], betas=tuple(tc['betas']), weight_decay=tc['wd']) step0, best, bad_evals = 0, -1.0, 0 + if a.init and not a.resume: + ck = torch.load(a.init, map_location=dev); model.load_state_dict(ck['model']) + print(f'微调：权重取自 {a.init}（第 {ck.get("step")} 步），优化器重置', flush=True) if a.resume: ck = torch.load(a.resume, map_location=dev); model.load_state_dict(ck['model']) if 'opt' in ck: opt.load_state_dict(ck['opt']) @@ -82,19 +121,23 @@ def main(): if model.head == 'wta': l_mf, xh = wta_loss(model, mem, Y); mf_raw = l_mf.detach(); xh = xh.float() else: - l_mf, mf_raw = meanflow_loss(model, mem, Y, p_neq=tc['p_neq']) - e = torch.randn_like(Y) - xh = one_step(model, mem, e).float() + p_neq = tc['p_neq'] if step >= tc.get('p_neq_warm', 5000) else 0.0 + l_mf, mf_raw, aux = meanflow_loss(model, mem, Y, p_neq=p_neq, return_aux=True) + sel = ((aux['r'] == aux['t']) & (aux['t'] <= tc.get('fk_tmax', 0.3))).nonzero(as_tuple=True)[0] + tt = aux['t'][sel][:, None, None] + xh = aux['z'][sel] - tt * aux['u'][sel] # 配对去噪估计（fp32） v, s = model.heads(mem) l_val = (v.float() - V).abs().mean(); l_sub = (s.float() - SG).abs().mean() - d2 = xh[:, 2:] - 2 * xh[:, 1:-1] + xh[:, :-2] - l_sm = d2.pow(2).mean() - l_fk = torch.zeros((), device=dev) - if tc['w_fk'] > 0: - q = q_from_feat(X) + Xf, Yf = (X, Y) if model.head == 'wta' else (X[sel], Y[sel]) + l_sm = torch.zeros((), device=dev); l_fk = torch.zeros((), device=dev) + if xh.shape[0] > 0: + d2 = xh[:, 2:] - 2 * xh[:, 1:-1] + xh[:, :-2] + l_sm = d2.pow(2).mean() + if tc['w_fk'] > 0 and xh.shape[0] > 0: + q = q_from_feat(Xf) ks = torch.tensor([3, 7, 11, 15], device=dev) qh = q[:, None] + torch.cumsum(xh.clamp(-1, 1) * stepv, 1)[:, ks] - qt = q[:, None] + torch.cumsum(Y * stepv, 1)[:, ks] + qt = q[:, None] + torch.cumsum(Yf * stepv, 1)[:, ks] B4 = qh.shape[0] * 4 ph = fk_points(kin, qh.reshape(B4, 6), fk_idx); pt = fk_points(kin, qt.reshape(B4, 6), fk_idx) l_fk = ((ph - pt).norm(dim=-1) / 10).mean() @@ -119,12 +162,14 @@ def main(): model.eval() # bs=64 + 归还缓存池：WDDM 卡满时 15.9 GiB 的评估峰值会被驱动换出到内存，eval 从 2 分钟变 1 小时 res, _ = evaluate(model, kin, val_tasks, dev, bs=64) + if vdata is not None: res.update(open_loop(model, kin, vdata, stepv, fk_idx)) 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) - if res['success'] > best + 1e-4: - best = res['success']; bad_evals = 0; save('best.pt', step + 1, {'eval': res}) + sc = res['success'] - 1e-3 * min(res.get('ol_fk_mm', 0.0), 500.0) # 到达率为主，开环误差（mm）只在到达率打平时起作用 + if sc > best + 1e-5: + best = sc; bad_evals = 0; save('best.pt', step + 1, {'eval': res}) else: bad_evals += 1 if bad_evals >= tc['patience_evals']: print('闭环到达率不再上升，停止', flush=True); break @@ -132,7 +177,7 @@ def main(): print('训练发散，last.pt / best.pt 保留发散前的存档', flush=True); return save('last.pt', step + 1) if a.no_eval: save('best.pt', step + 1) - print(f'done, best success {best:.4f}, {(time.time() - t0) / 3600:.2f} h', flush=True) + print(f'done, best score（到达率 − 0.001×开环 mm）{best:.4f}, {(time.time() - t0) / 3600:.2f} h', flush=True) if __name__ == '__main__':