File size: 4,343 Bytes
d13981b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 | #!/usr/bin/env python3
"""Analyze RLSD training log health."""
from __future__ import annotations
import argparse
import ast
import re
import sys
from collections import Counter
from pathlib import Path
def parse_metrics(text: str) -> list[dict]:
metrics = []
for m in re.finditer(r"\{'loss':[^\n]+\}", text):
try:
metrics.append(ast.literal_eval(m.group()))
except (SyntaxError, ValueError):
pass
return metrics
def alert_counts(text: str) -> Counter:
return Counter(re.findall(r"\[ALERT\] (\w+)", text))
def health_snapshots(text: str) -> list[str]:
return re.findall(r"\[health\] step snapshot \| ([^\n]+)", text)
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("log_path")
args = parser.parse_args()
text = Path(args.log_path).read_text(encoding="utf-8", errors="replace")
alerts = alert_counts(text)
metrics = parse_metrics(text)
health = health_snapshots(text)
print("# RLSD 训练健康分析\n")
print("## 进度")
prog = re.findall(r"(\d+)%\|[^|]+\| (\d+)/(\d+)", text)
if prog:
pct, cur, total = prog[-1]
print(f"- 当前: step {cur}/{total} ({pct}%), epoch≈{metrics[-1].get('epoch', '?') if metrics else '?'}")
print(f"- 已记录 metrics 行数: {len(metrics)}")
print("\n## 防泄露指标 (RLSD 核心验收)")
if metrics:
last = metrics[-1]
checks = [
("opsd_on_correct_rate == 0", last.get("routing/opsd_on_correct_rate", -1) == 0.0),
("privileged_suffix_has_gold == 0", last.get("teacher/privileged_suffix_has_gold_rate", -1) == 0.0),
("leakage skipped only", last.get("routing/opsd_skipped_leakage", 0) == 0.0),
]
for name, ok in checks:
print(f"- [{'PASS' if ok else 'FAIL'}] {name}")
print(f"- routing/opsd_on_correct_rate: {last.get('routing/opsd_on_correct_rate', 'N/A')}")
print(f"- teacher/privileged_suffix_has_gold_rate: {last.get('teacher/privileged_suffix_has_gold_rate', 'N/A')}")
print(f"- routing/opd_teacher_call_rate: {last.get('routing/opd_teacher_call_rate', 'N/A')}")
print("\n## 学习信号 (是否真正在学)")
if metrics:
for idx in [0, 50, 100, 200, 500, 1000, len(metrics) - 1]:
if idx >= len(metrics):
continue
m = metrics[idx]
print(
f"- step {idx:4d}: acc={m.get('rewards/accuracy/mean', 0):.3f} "
f"fmt={m.get('rewards/format/mean', 0):.3f} "
f"len={m.get('completions/mean_length', 0):.1f} "
f"degen={m.get('completions/degenerate_rate', 0):.3f} "
f"grpo_zero={m.get('signal/grpo_zero_loss_rate', 0):.3f} "
f"loss={m.get('loss', 0):.4f}"
)
print("\n## ALERT 分布 (Top 10)")
for k, v in alerts.most_common(10):
print(f"- {k}: {v}")
print("\n## 最新 batch 生成样例 (step 1099 附近)")
samples = re.findall(r"decode_skip_special='([^']{1,80})'", text[-8000:])
if samples:
print(f"- 典型输出: {samples[:4]}")
print("\n## 综合判断")
if metrics:
last = metrics[-1]
degen = last.get("completions/degenerate_rate", 0)
acc = last.get("rewards/accuracy/mean", 0)
fmt = last.get("rewards/format/mean", 0)
grpo_zero = last.get("signal/grpo_zero_loss_rate", 1)
anti_leak_ok = (
last.get("routing/opsd_on_correct_rate", 1) == 0
and last.get("teacher/privileged_suffix_has_gold_rate", 1) == 0
)
learning_ok = acc > 0.05 and fmt > 0.3 and degen < 0.5 and grpo_zero < 0.9
print(f"- 防泄露路由: {'健康' if anti_leak_ok else '异常'}")
print(f"- 学习效果: {'不健康 (退化/零信号)' if not learning_ok else '尚可'}")
if degen >= 0.9:
print(" → 模型已塌缩为极短重复输出 (如 ' Sund\\n')，GRPO 无法介入")
if grpo_zero >= 0.99:
print(" → GRPO 全程零梯度，仅靠 self-OPSD 在更新")
if last.get("routing/sft_replaced_ratio", 0) == 0:
print(" → 在线 SFT 冷启动未触发 (可能组内总有部分样本非全错)")
return 0
if __name__ == "__main__":
sys.exit(main())
|