| |
| """Offline degeneration attribution report from TriMode training logs.""" |
| from __future__ import annotations |
|
|
| import argparse |
| import ast |
| import json |
| import re |
| import sys |
| from collections import Counter, defaultdict |
| from pathlib import Path |
|
|
|
|
| def parse_kv_line(line: str) -> dict: |
| out: dict = {} |
| for m in re.finditer(r"(\w+)=([^|]+?)(?=\s*\||\s*$)", line): |
| k, v = m.group(1).strip(), m.group(2).strip() |
| try: |
| if v.startswith("[") or v.startswith("{"): |
| out[k] = json.loads(v.replace("'", '"')) |
| elif re.match(r"^-?\d+\.\d+([eE][+-]?\d+)?$", v): |
| out[k] = float(v) |
| elif re.match(r"^-?\d+$", v): |
| out[k] = int(v) |
| else: |
| out[k] = v |
| except (json.JSONDecodeError, ValueError): |
| out[k] = v |
| return out |
|
|
|
|
| def load_log(path: str) -> str: |
| return Path(path).read_text(encoding="utf-8", errors="replace") |
|
|
|
|
| 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 parse_health_generate(text: str) -> list[tuple[int, dict]]: |
| rows = [] |
| pat = re.compile( |
| r"\[OPSD-HEALTH\][^\n]*\[global_step=(\d+)\]\[generate\] batch health \| (.+)" |
| ) |
| for m in pat.finditer(text): |
| rows.append((int(m.group(1)), parse_kv_line(m.group(2)))) |
| return rows |
|
|
|
|
| def parse_health_step(text: str) -> list[tuple[int, dict]]: |
| rows = [] |
| pat = re.compile( |
| r"\[OPSD-HEALTH\][^\n]*\[global_step=(\d+)\]\[step\] step summary \| (.+)" |
| ) |
| for m in pat.finditer(text): |
| rows.append((int(m.group(1)), parse_kv_line(m.group(2)))) |
| return rows |
|
|
|
|
| def parse_health_alerts(text: str) -> list[tuple[int, str, dict]]: |
| rows = [] |
| pat = re.compile( |
| r"\[OPSD-HEALTH\][^\n]*\[global_step=(\d+)\]\[ALERT\] (\w+) \| (.+)" |
| ) |
| for m in pat.finditer(text): |
| rows.append((int(m.group(1)), m.group(2), parse_kv_line(m.group(3)))) |
| return rows |
|
|
|
|
| def parse_detail_health(text: str) -> list[tuple[int, str, dict]]: |
| rows = [] |
| pat = re.compile( |
| r"\[OPSD-DETAIL\][^\n]*\[step=(\d+)\]\[every=\d+\]\[health\] ([^|]+) \| (.+)" |
| ) |
| for m in pat.finditer(text): |
| rows.append((int(m.group(1)), m.group(2).strip(), parse_kv_line(m.group(3)))) |
| return rows |
|
|
|
|
| def detect_step1_collapse(metrics: list[dict]) -> str | None: |
| if len(metrics) < 2: |
| return None |
| d0, d1 = metrics[0], metrics[1] |
| clip0 = d0.get("completions/clipped_ratio", 0) |
| clip1 = d1.get("completions/clipped_ratio", 0) |
| if clip0 < 0.2 and clip1 > 0.8: |
| return ( |
| f"step 1 collapse: clipped {clip0:.2f} -> {clip1:.2f}; " |
| f"format {d0.get('rewards/format/mean', 'NA')} -> {d1.get('rewards/format/mean', 'NA')}" |
| ) |
| return None |
|
|
|
|
| def summarize_alerts(alerts: list[tuple[int, str, dict]]) -> Counter: |
| return Counter(code for _, code, _ in alerts) |
|
|
|
|
| def build_report(text: str, baseline_text: str | None = None) -> str: |
| metrics = parse_metrics(text) |
| health_gen = parse_health_generate(text) |
| health_step = parse_health_step(text) |
| alerts = parse_health_alerts(text) |
| detail = parse_detail_health(text) |
|
|
| lines = ["# TriMode Degeneration Report", ""] |
| lines.append(f"- Metric steps parsed: {len(metrics)}") |
| lines.append(f"- Health generate lines: {len(health_gen)}") |
| lines.append(f"- Health step summaries: {len(health_step)}") |
| lines.append(f"- Health alerts: {len(alerts)}") |
| lines.append(f"- Detail health bundles: {len(detail)}") |
| lines.append("") |
|
|
| collapse = detect_step1_collapse(metrics) |
| if collapse: |
| lines.append("## Step 1 collapse") |
| lines.append(f"- {collapse}") |
| lines.append("") |
|
|
| if alerts: |
| lines.append("## Alert summary") |
| for code, count in summarize_alerts(alerts).most_common(): |
| first_step = min(s for s, c, _ in alerts if c == code) |
| lines.append(f"- `{code}`: {count}x (first at step {first_step})") |
| lines.append("") |
|
|
| if health_gen: |
| lines.append("## Generation health timeline (sampled)") |
| for step, fields in health_gen: |
| if step <= 20 or step % 25 == 0: |
| lines.append( |
| f"- step {step}: degenerate={fields.get('degenerate_rate', 'NA')} " |
| f"clip={fields.get('clipped_rate', 'NA')} eos={fields.get('eos_rate', 'NA')} " |
| f"alerts={fields.get('alerts', 'none')}" |
| ) |
| lines.append("") |
|
|
| qi_count = len(re.findall(r"其其其", text)) |
| lines.append("## CJK repeat in log") |
| lines.append(f"- `其其其` occurrences: {qi_count}") |
| lines.append("") |
|
|
| if baseline_text: |
| base_metrics = parse_metrics(baseline_text) |
| base_alerts = len(parse_health_alerts(baseline_text)) |
| lines.append("## Baseline comparison") |
| lines.append(f"- Baseline metric steps: {len(base_metrics)}") |
| lines.append(f"- Baseline alerts: {base_alerts} vs current: {len(alerts)}") |
| if metrics and base_metrics: |
| m_cur = metrics[min(10, len(metrics) - 1)] |
| m_base = base_metrics[min(10, len(base_metrics) - 1)] |
| lines.append( |
| f"- At ~step 10: current clip={m_cur.get('completions/clipped_ratio', 'NA')} " |
| f"vs baseline {m_base.get('completions/clipped_ratio', 'NA')}" |
| ) |
| lines.append("") |
|
|
| hints: list[str] = [] |
| for _, msg, fields in detail: |
| if "correlation" in msg and "root_cause_hints" in fields: |
| h = fields["root_cause_hints"] |
| if isinstance(h, list): |
| hints.extend(str(x) for x in h if x != "none") |
| if hints: |
| lines.append("## Root cause hints (from DETAIL health)") |
| for h in dict.fromkeys(hints): |
| lines.append(f"- {h}") |
| lines.append("") |
|
|
| return "\n".join(lines) |
|
|
|
|
| def main() -> None: |
| parser = argparse.ArgumentParser(description="Degeneration report from training log") |
| parser.add_argument("log", nargs="?", default=None, help="Training log path") |
| parser.add_argument("--baseline", default=None, help="Optional baseline log for comparison") |
| parser.add_argument("--json", action="store_true", help="Emit JSON summary") |
| args = parser.parse_args() |
|
|
| log_path = args.log or (sys.argv[1] if len(sys.argv) > 1 else "train_trimode.log") |
| text = load_log(log_path) |
| baseline_text = load_log(args.baseline) if args.baseline else None |
|
|
| report = build_report(text, baseline_text) |
| if args.json: |
| payload = { |
| "log": log_path, |
| "metrics_count": len(parse_metrics(text)), |
| "alerts": parse_health_alerts(text), |
| "report_md": report, |
| } |
| print(json.dumps(payload, ensure_ascii=False, indent=2)) |
| else: |
| print(report) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
