| |
| from __future__ import annotations |
|
|
| import argparse |
| import os |
| import sys |
| from pathlib import Path |
| from types import SimpleNamespace |
| from typing import Any |
|
|
| ROOT = Path(__file__).resolve().parents[2] |
| if str(ROOT) not in sys.path: |
| sys.path.insert(0, str(ROOT)) |
|
|
| from scripts.analysis.pcd_artifact_core import ( |
| COUNT_ALIASES, |
| bin_training_rows, |
| collect_run_data, |
| count_value as _count_value, |
| fmt_rate as _fmt_rate, |
| mean_or_none as _mean, |
| read_manifest, |
| safe_float as _safe_float, |
| write_csv as _write_csv, |
| ) |
| from scripts.analysis.pcd_artifact_framework import ( |
| ArtifactRegistry, |
| ArtifactSpec, |
| write_artifacts_manifest, |
| write_dashboard_html, |
| ) |
|
|
|
|
| CANONICAL_VARIANTS = ( |
| "deplot_no_vs_opd", |
| "deplot_no_vs_opd_va", |
| "deplot_no_vs_opd_pcd", |
| "deplot_no_vs_opd_va_pcd", |
| ) |
|
|
| VARIANT_LABELS = { |
| "deplot_no_vs_opd": "anchor", |
| "deplot_no_vs_opd_va": "VA", |
| "deplot_no_vs_opd_pcd": "PCD", |
| "deplot_no_vs_opd_va_pcd": "VA+PCD", |
| } |
|
|
| VARIANT_COLORS = { |
| "deplot_no_vs_opd": "#565656", |
| "deplot_no_vs_opd_va": "#b48a00", |
| "deplot_no_vs_opd_pcd": "#2f6f73", |
| "deplot_no_vs_opd_va_pcd": "#7a4e9d", |
| } |
|
|
| PAPER_COLORS = { |
| "anchor": "#4d4d4d", |
| "pcd": "#2f6f73", |
| "probe": "#9b9b9b", |
| "correct": "#2f6f73", |
| "risk": "#b5483c", |
| "grid": "#d8d8d8", |
| } |
|
|
| METRIC_SHORT_LABELS = { |
| "grpo_route_rate": "GRPO", |
| "opd_route_rate": "OPD", |
| "sft_route_rate": "SFT", |
| } |
|
|
|
|
| def _candidate_group_type(record: dict[str, Any]) -> str: |
| if record.get("group_all_wrong") is True or record.get("is_all_wrong_probe_candidate") is True: |
| return "all_wrong" |
| if record.get("is_mixed_wrong_probe_candidate") is True or record.get("group_has_correct") is True: |
| return "mixed_wrong" |
| return "unknown" |
|
|
|
|
| def summarize_candidates(records: list[dict[str, Any]]) -> list[dict[str, Any]]: |
| grouped: dict[str, list[dict[str, Any]]] = {} |
| for record in records: |
| grouped.setdefault(_candidate_group_type(record), []).append(record) |
| rows: list[dict[str, Any]] = [] |
| for group_type, group in sorted(grouped.items()): |
| n = len(group) |
| teacher_correct = sum(1 for row in group if row.get("teacher_correct") is True) |
| parse_failed = sum(1 for row in group if row.get("parse_failed") is True) |
| placeholders = sum(1 for row in group if row.get("teacher_output_is_placeholder") is True) |
| rows.append( |
| { |
| "group_type": group_type, |
| "n": n, |
| "teacher_correct_rate": teacher_correct / n if n else None, |
| "parse_fail_rate": parse_failed / n if n else None, |
| "placeholder_rate": placeholders / n if n else None, |
| } |
| ) |
| return rows |
|
|
|
|
| def _setup_matplotlib(): |
| os.environ.setdefault("MPLCONFIGDIR", "/tmp/matplotlib") |
| import matplotlib |
|
|
| matplotlib.use("Agg") |
| import matplotlib.pyplot as plt |
|
|
| plt.rcParams.update( |
| { |
| "font.family": "serif", |
| "font.size": 8, |
| "axes.titlesize": 9, |
| "axes.labelsize": 8, |
| "xtick.labelsize": 7, |
| "ytick.labelsize": 7, |
| "legend.fontsize": 7, |
| "figure.titlesize": 10, |
| "mathtext.fontset": "stix", |
| "axes.linewidth": 0.6, |
| "xtick.major.width": 0.6, |
| "ytick.major.width": 0.6, |
| "savefig.bbox": "tight", |
| } |
| ) |
| return plt |
|
|
|
|
| def _save_figure(fig: Any, out_dir: Path, stem: str, *, svg: bool = False) -> None: |
| out_dir.mkdir(parents=True, exist_ok=True) |
| fig.savefig(out_dir / f"{stem}.png", dpi=220) |
| fig.savefig(out_dir / f"{stem}.pdf") |
| if svg: |
| fig.savefig(out_dir / f"{stem}.svg") |
|
|
|
|
| def _fmt_count(value: float | int | None) -> str: |
| return "" if value is None else str(int(value)) |
|
|
|
|
| def _compact_count(value: float | int | None) -> str: |
| if value is None: |
| return "" |
| value = float(value) |
| if value >= 1_000_000: |
| return f"{value / 1_000_000:.1f}M" |
| if value >= 1_000: |
| return f"{value / 1_000:.1f}K" |
| return str(int(value)) |
|
|
|
|
| def _variant_label_list(variants: tuple[str, ...] | list[str]) -> list[str]: |
| return [VARIANT_LABELS.get(variant, variant) for variant in variants] |
|
|
|
|
| def _variant_color(variant: str) -> str: |
| return VARIANT_COLORS.get(variant, "#565656") |
|
|
|
|
| def _candidate_counts(payload: dict[str, Any]) -> tuple[int, int]: |
| records = payload.get("candidates", []) |
| return len(records), sum(1 for record in records if record.get("teacher_correct") is True) |
|
|
|
|
| def _paper_axes(ax: Any, *, y_grid: bool = True) -> None: |
| ax.spines["top"].set_visible(False) |
| ax.spines["right"].set_visible(False) |
| ax.spines["left"].set_linewidth(0.6) |
| ax.spines["bottom"].set_linewidth(0.6) |
| if y_grid: |
| ax.grid(True, axis="y", color=PAPER_COLORS["grid"], alpha=0.45, linewidth=0.45) |
| ax.set_axisbelow(True) |
|
|
|
|
| def _panel_label(ax: Any, label: str, title: str) -> None: |
| ax.text( |
| 0.0, |
| 1.04, |
| f"({label}) {title}", |
| transform=ax.transAxes, |
| ha="left", |
| va="bottom", |
| fontsize=9, |
| fontweight="bold", |
| ) |
|
|
|
|
| def _moving_average(values: list[float], window: int = 5) -> list[float]: |
| out: list[float] = [] |
| for index in range(len(values)): |
| chunk = values[max(0, index - window + 1) : index + 1] |
| out.append(sum(chunk) / len(chunk)) |
| return out |
|
|
|
|
| def _safe_ratio(numerator: float | None, denominator: float | None) -> float | None: |
| if numerator is None or denominator is None or denominator == 0: |
| return None |
| return numerator / denominator |
|
|
|
|
| def _latex_escape(value: Any) -> str: |
| text = "" if value is None else str(value) |
| return text.replace("\\", "\\textbackslash{}").replace("&", "\\&").replace("%", "\\%").replace("_", "\\_") |
|
|
|
|
| def _write_booktabs_table(path: Path, headers: list[str], rows: list[list[Any]], *, highlight_first_cell: str | None = None) -> None: |
| lines = [ |
| "```latex", |
| "\\begin{tabular}{" + "l" + "c" * (len(headers) - 1) + "}", |
| "\\toprule", |
| " & ".join(_latex_escape(header) for header in headers) + " \\\\", |
| "\\midrule", |
| ] |
| for row in rows: |
| escaped = [_latex_escape(cell) for cell in row] |
| if highlight_first_cell and str(row[0]) == highlight_first_cell: |
| escaped[0] = f"\\rowcolor{{gray!10}} {escaped[0]}" |
| lines.append(" & ".join(escaped) + " \\\\") |
| lines.extend(["\\bottomrule", "\\end{tabular}", "```", ""]) |
| path.write_text("\n".join(lines), encoding="utf-8") |
|
|
|
|
| def _last_binned_value(data: dict[str, dict[str, Any]], variant: str, metric: str) -> float | None: |
| rows = [row for row in bin_training_rows(data.get(variant, {}).get("train", [])) if row.get(metric) is not None] |
| return float(rows[-1][metric]) if rows else None |
|
|
|
|
| def make_fig0(data: dict[str, dict[str, Any]], out_dir: Path) -> None: |
| fields = [ |
| "variant", |
| "step_bin", |
| "loss_mean", |
| "opsd_loss_mean", |
| "reward_mean", |
| "accuracy_reward_mean", |
| "format_reward_mean", |
| "reward_std_mean", |
| "grpo_zero_loss_rate", |
| "advantage_abs_mean", |
| "opsd_effective_weight", |
| "opsd_adaptive_multiplier", |
| "completion_clipped_rate", |
| "completion_eos_rate", |
| "degenerate_rate", |
| ] |
| csv_rows: list[dict[str, Any]] = [] |
| for variant in CANONICAL_VARIANTS: |
| for row in bin_training_rows(data.get(variant, {}).get("train", [])): |
| csv_rows.append({"variant": variant, **{field: _fmt_rate(row.get(field)) for field in fields if field not in {"variant", "step_bin"}}, "step_bin": row["step_bin"]}) |
| _write_csv(out_dir / "fig0_training_basics.csv", csv_rows, fields) |
|
|
| plt = _setup_matplotlib() |
| fig, axes = plt.subplots(2, 2, figsize=(9.6, 5.6), squeeze=False) |
| panels = [ |
| ("Training loss", "loss_mean"), |
| ("Accuracy reward", "accuracy_reward_mean"), |
| ("Reward std", "reward_std_mean"), |
| ("Degenerate completion rate", "degenerate_rate"), |
| ] |
| for ax, (title, metric) in zip([ax for row in axes for ax in row], panels): |
| for variant in CANONICAL_VARIANTS: |
| rows = bin_training_rows(data.get(variant, {}).get("train", [])) |
| x = [row["step_bin"] for row in rows if row.get(metric) is not None] |
| y = [row[metric] for row in rows if row.get(metric) is not None] |
| if x: |
| ax.plot(x, y, linewidth=1.4, color=_variant_color(variant), label=VARIANT_LABELS.get(variant, variant)) |
| ax.set_title(title) |
| ax.set_xlabel("step bin") |
| _paper_axes(ax) |
| handles, labels = ax.get_legend_handles_labels() |
| if handles: |
| ax.legend(handles, labels, frameon=False, ncol=2) |
| fig.tight_layout() |
| _save_figure(fig, out_dir, "fig0_training_basics") |
| plt.close(fig) |
|
|
|
|
| def make_fig1(data: dict[str, dict[str, Any]], out_dir: Path) -> None: |
| csv_rows: list[dict[str, Any]] = [] |
| for variant in ("deplot_no_vs_opd", "deplot_no_vs_opd_pcd"): |
| for row in bin_training_rows(data.get(variant, {}).get("train", [])): |
| for metric in ("group_all_wrong_rate", "group_mixed_rate", "reward_std_lt_0_05_rate"): |
| csv_rows.append( |
| { |
| "panel": "reward_sparsity", |
| "variant": variant, |
| "step_bin": row["step_bin"], |
| "group_type": "", |
| "metric": metric, |
| "value": _fmt_rate(row.get(metric)), |
| "n": "", |
| "data_quality": "exact" if row.get(metric) is not None else "missing", |
| } |
| ) |
| for variant, payload in data.items(): |
| candidates = payload.get("candidates", []) |
| candidate_quality = "exact" if candidates and "group_all_wrong" in candidates[0] else "proxy" |
| probe_count, correct_count = _candidate_counts(payload) |
| for metric, value in (("probe_candidate_count", probe_count), ("teacher_correct_count", correct_count)): |
| csv_rows.append({"panel": "candidate_volume", "variant": variant, "step_bin": "", "group_type": "", "metric": metric, "value": str(value), "n": probe_count, "data_quality": candidate_quality}) |
| for row in summarize_candidates(candidates): |
| for metric in ("teacher_correct_rate", "parse_fail_rate", "placeholder_rate"): |
| csv_rows.append({"panel": "teacher_rescue", "variant": variant, "step_bin": "", "group_type": row["group_type"], "metric": metric, "value": _fmt_rate(row.get(metric)), "n": row["n"], "data_quality": candidate_quality}) |
| _write_csv(out_dir / "fig1_motivation.csv", csv_rows, ["panel", "variant", "step_bin", "group_type", "metric", "value", "n", "data_quality"]) |
| _write_fig1_cases(data, out_dir / "fig1_cases.md") |
|
|
| plt = _setup_matplotlib() |
| fig, axes = plt.subplots(1, 3, figsize=(10.6, 3.1), gridspec_kw={"width_ratios": [1.15, 1.0, 1.2]}) |
| ax_volume, ax_rescue, ax_cases = axes |
| variants = [variant for variant in CANONICAL_VARIANTS if variant in data] |
| x_positions = list(range(len(variants))) |
| width = 0.34 |
| probe_counts = [_candidate_counts(data.get(variant, {}))[0] for variant in variants] |
| correct_counts = [_candidate_counts(data.get(variant, {}))[1] for variant in variants] |
| ax_volume.bar([x - width / 2 for x in x_positions], probe_counts, width=width, color=PAPER_COLORS["probe"], label="probed") |
| ax_volume.bar([x + width / 2 for x in x_positions], correct_counts, width=width, color=PAPER_COLORS["correct"], label="teacher-correct") |
| if any(value > 0 for value in probe_counts + correct_counts): |
| ax_volume.set_yscale("log") |
| _panel_label(ax_volume, "a", "candidate volume") |
| ax_volume.set_ylabel("count (log)") |
| ax_volume.set_xticks(x_positions) |
| ax_volume.set_xticklabels(_variant_label_list(variants), rotation=20, ha="right") |
| ax_volume.legend(frameon=False, loc="upper left", handlelength=1.0) |
| _paper_axes(ax_volume) |
| if "deplot_no_vs_opd_pcd" in variants: |
| pcd_index = variants.index("deplot_no_vs_opd_pcd") |
| ax_volume.text(pcd_index + width / 2, max(correct_counts[pcd_index], 1) * 1.35, f"{correct_counts[pcd_index]:,}", ha="center", va="bottom", fontsize=7, fontweight="bold", color=PAPER_COLORS["pcd"]) |
|
|
| rates = [correct / probe if probe else 0.0 for probe, correct in zip(probe_counts, correct_counts)] |
| bars = ax_rescue.bar(x_positions, rates, color=[_variant_color(variant) for variant in variants], width=0.62) |
| _panel_label(ax_rescue, "b", "recoverability rate") |
| ax_rescue.set_ylabel("teacher-correct / probed") |
| ax_rescue.set_ylim(0, max(0.35, max(rates or [0]) * 1.2)) |
| ax_rescue.set_xticks(x_positions) |
| ax_rescue.set_xticklabels(_variant_label_list(variants), rotation=20, ha="right") |
| _paper_axes(ax_rescue) |
| for bar, rate, variant in zip(bars, rates, variants): |
| if variant in {"deplot_no_vs_opd", "deplot_no_vs_opd_pcd"}: |
| ax_rescue.text(bar.get_x() + bar.get_width() / 2, bar.get_height() + 0.008, f"{rate:.3f}", ha="center", va="bottom", fontsize=7) |
| if "deplot_no_vs_opd" in variants and "deplot_no_vs_opd_pcd" in variants: |
| anchor_i = variants.index("deplot_no_vs_opd") |
| pcd_i = variants.index("deplot_no_vs_opd_pcd") |
| probe_mult = _safe_ratio(float(probe_counts[pcd_i]), float(probe_counts[anchor_i])) |
| correct_mult = _safe_ratio(float(correct_counts[pcd_i]), float(correct_counts[anchor_i])) |
| if probe_mult is not None and correct_mult is not None: |
| ax_rescue.text(0.03, 0.93, f"PCD: {probe_mult:.1f}x probes\n{correct_mult:.1f}x rescued", transform=ax_rescue.transAxes, ha="left", va="top", fontsize=7, color=PAPER_COLORS["pcd"]) |
|
|
| _panel_label(ax_cases, "c", "qualitative routing") |
| ax_cases.axis("off") |
| archetypes = [("mixed wrong", "teacher-correct", "OPD"), ("all-wrong", "teacher-correct", "OPD"), ("wrong", "teacher-wrong", "SFT")] |
| seen_records: dict[tuple[str, str, str], bool] = {} |
| for payload in data.values(): |
| for record in payload.get("candidates", []): |
| group = "all-wrong" if record.get("group_all_wrong") is True or record.get("is_all_wrong_probe_candidate") is True else "mixed wrong" |
| teacher = "teacher-correct" if record.get("teacher_correct") is True else "teacher-wrong" |
| route = "OPD" if record.get("final_route") == "opd" else "SFT" |
| seen_records[(group, teacher, route)] = True |
| seen_records[("wrong", teacher, route)] = True |
| table_data = [["completion", "teacher probe", "route"]] |
| for archetype in archetypes: |
| route_text = archetype[2] if seen_records.get(archetype) else archetype[2] + "*" |
| table_data.append([archetype[0], archetype[1], route_text]) |
| table = ax_cases.table(cellText=table_data, loc="center", cellLoc="left", colWidths=[0.36, 0.42, 0.22]) |
| table.auto_set_font_size(False) |
| table.set_fontsize(6.2) |
| table.scale(1.0, 1.35) |
| for (row, col), cell in table.get_celld().items(): |
| cell.set_linewidth(0.4) |
| cell.set_edgecolor("#bbbbbb") |
| if row == 0: |
| cell.set_facecolor("#f0f0f0") |
| cell.get_text().set_fontweight("bold") |
| elif col == 2 and "OPD" in cell.get_text().get_text(): |
| cell.get_text().set_color(PAPER_COLORS["pcd"]) |
| cell.get_text().set_fontweight("bold") |
| elif col == 2 and "SFT" in cell.get_text().get_text(): |
| cell.get_text().set_color(PAPER_COLORS["risk"]) |
| ax_cases.text(0.0, -0.06, "* archetype shown when exact route example is absent", transform=ax_cases.transAxes, fontsize=6.5, color="#666666") |
| fig.tight_layout(w_pad=1.0) |
| _save_figure(fig, out_dir, "fig1_motivation") |
| plt.close(fig) |
|
|
|
|
| def _write_fig1_cases(data: dict[str, dict[str, Any]], path: Path) -> None: |
| cases: list[tuple[str, dict[str, Any]]] = [] |
| for payload in data.values(): |
| for record in payload.get("candidates", []): |
| if len(cases) >= 3: |
| break |
| label = "" |
| if record.get("student_correct") is False and record.get("teacher_correct") is True: |
| label = "student_wrong_teacher_correct" |
| elif record.get("student_correct") is False and record.get("teacher_correct") is False: |
| label = "student_wrong_teacher_wrong" |
| if record.get("group_all_wrong") is True and record.get("teacher_correct") is True: |
| label = "all_wrong_teacher_rescued" |
| if label and all(existing != label for existing, _ in cases): |
| cases.append((label, record)) |
| lines = ["# Figure 1 Qualitative Cases", ""] |
| for label, record in cases: |
| lines.extend([f"## {label}", f"- image: {record.get('image', '')}", f"- question: {record.get('question', '')}", f"- reference: {record.get('reference', '')}", f"- student_output: {record.get('student_output', '')}", f"- teacher_output: {record.get('teacher_output', '')}", f"- final_route: {record.get('final_route', '')}", ""]) |
| path.parent.mkdir(parents=True, exist_ok=True) |
| path.write_text("\n".join(lines), encoding="utf-8") |
|
|
|
|
| def make_fig4(data: dict[str, dict[str, Any]], out_dir: Path) -> None: |
| variants = ("deplot_no_vs_opd", "deplot_no_vs_opd_pcd", "deplot_no_vs_opd_va_pcd") |
| fields = ["variant", "step_bin", "reward_mean", "accuracy_reward_mean", "format_reward_mean", "reward_std_mean", "group_all_wrong_rate", "grpo_route_rate", "opd_route_rate", "sft_route_rate", "teacher_probe_candidate_rate", "teacher_correct_rate", "opsd_effective_weight", "opsd_adaptive_multiplier", "degenerate_rate"] |
| csv_rows: list[dict[str, Any]] = [] |
| for variant in variants: |
| for row in bin_training_rows(data.get(variant, {}).get("train", [])): |
| csv_rows.append({"variant": variant, **{field: _fmt_rate(row.get(field)) for field in fields if field not in {"variant", "step_bin"}}, "step_bin": row["step_bin"]}) |
| _write_csv(out_dir / "fig4_training_dynamics.csv", csv_rows, fields) |
|
|
| plt = _setup_matplotlib() |
| fig, axes = plt.subplots(1, 3, figsize=(10.2, 2.85), sharex=True) |
| panels = [("a", "accuracy reward", "accuracy_reward_mean", "reward"), ("b", "reward std", "reward_std_mean", "std"), ("c", "degenerate rate", "degenerate_rate", "rate")] |
| highlight = {"deplot_no_vs_opd", "deplot_no_vs_opd_pcd"} |
| pcd_rows = bin_training_rows(data.get("deplot_no_vs_opd_pcd", {}).get("train", [])) |
| active_steps = [row["step_bin"] for row in pcd_rows if (row.get("teacher_probe_candidate_rate") or 0) > 0.01 or (row.get("opd_route_rate") or 0) > 0.01] |
| active_step = active_steps[0] if active_steps else None |
| for ax, (panel, title, metric, ylabel) in zip(axes, panels): |
| for variant in CANONICAL_VARIANTS: |
| rows_for_variant = bin_training_rows(data.get(variant, {}).get("train", [])) |
| points = [(row["step_bin"], row[metric]) for row in rows_for_variant if row.get(metric) is not None] |
| if not points: |
| continue |
| x = [point[0] for point in points] |
| y = [float(point[1]) for point in points] |
| color = _variant_color(variant) |
| if variant in highlight: |
| ax.plot(x, y, color=color, alpha=0.16, linewidth=0.8) |
| ax.plot(x, _moving_average(y), color=color, linewidth=1.7, label=VARIANT_LABELS.get(variant, variant)) |
| else: |
| ax.plot(x, _moving_average(y), color=color, alpha=0.32, linewidth=1.0, linestyle="--") |
| if active_step is not None: |
| ax.axvline(active_step, color="#888888", linewidth=0.6, linestyle=":", alpha=0.8) |
| _panel_label(ax, panel, title) |
| ax.set_xlabel("step bin") |
| ax.set_ylabel(ylabel) |
| _paper_axes(ax) |
| if active_step is not None: |
| axes[0].text(active_step, axes[0].get_ylim()[1], "probe active", rotation=90, va="top", ha="right", fontsize=6.5, color="#666666") |
| handles, labels = axes[0].get_legend_handles_labels() |
| if handles: |
| axes[0].legend(handles, labels, frameon=False, loc="best", handlelength=1.4) |
| fig.tight_layout(w_pad=1.0) |
| _save_figure(fig, out_dir, "fig4_training_dynamics") |
| plt.close(fig) |
|
|
|
|
| def make_fig5(data: dict[str, dict[str, Any]], out_dir: Path) -> None: |
| rows: list[dict[str, Any]] = [] |
| for variant, payload in data.items(): |
| totals = {key: 0.0 for key in COUNT_ALIASES} |
| for train_row in payload.get("train", []): |
| for key in totals: |
| value = _count_value(train_row, key) |
| if value is not None: |
| totals[key] += value |
| candidates = payload.get("candidates", []) |
| if totals["probe_candidate_count"] == 0 and candidates: |
| totals["probe_candidate_count"] = len(candidates) |
| totals["teacher_correct_count"] = sum(1 for rec in candidates if rec.get("teacher_correct") is True) |
| totals["opd_route_count"] = sum(1 for rec in candidates if rec.get("final_route") == "opd") |
| totals["sft_route_count"] = sum(1 for rec in candidates if str(rec.get("final_route", "")).startswith("sft")) |
| has_exact_totals = totals["total_completion_count"] > 0 and totals["wrong_completion_count"] > 0 |
| probe = totals["probe_candidate_count"] |
| correct = totals["teacher_correct_count"] |
| sft_fallback = max(probe - correct, 0.0) |
| rows.append({"variant": variant, "funnel_scope": "exact_completion_counts" if has_exact_totals else "candidate_proxy", "total_completion_count": _fmt_count(totals["total_completion_count"] if has_exact_totals else None), "wrong_completion_count": _fmt_count(totals["wrong_completion_count"] if has_exact_totals else None), "probe_candidate_count": _fmt_count(probe), "teacher_correct_count": _fmt_count(correct), "opd_route_count": _fmt_count(totals["opd_route_count"]), "sft_fallback_count": _fmt_count(sft_fallback), "probe_candidate_rate": _fmt_rate(probe / totals["wrong_completion_count"] if totals["wrong_completion_count"] else None), "teacher_correct_given_probe_rate": _fmt_rate(correct / probe if probe else None), "opd_given_teacher_correct_rate": _fmt_rate(totals["opd_route_count"] / correct if correct else None), "sft_fallback_given_probe_rate": _fmt_rate(sft_fallback / probe if probe else None)}) |
| fields = ["variant", "funnel_scope", "total_completion_count", "wrong_completion_count", "probe_candidate_count", "teacher_correct_count", "opd_route_count", "sft_fallback_count", "probe_candidate_rate", "teacher_correct_given_probe_rate", "opd_given_teacher_correct_rate", "sft_fallback_given_probe_rate"] |
| _write_csv(out_dir / "fig5_teacher_rescue_funnel.csv", rows, fields) |
|
|
| main = next((row for row in rows if row["variant"] == "deplot_no_vs_opd_pcd"), rows[0] if rows else None) |
| plt = _setup_matplotlib() |
| fig, axes = plt.subplots(1, 2, figsize=(10.0, 3.25), gridspec_kw={"width_ratios": [1.55, 1.0]}) |
| if main: |
| labels = ["Probed", "Teacher-correct", "OPD", "SFT fallback"] |
| values = [int(main["probe_candidate_count"] or 0), int(main["teacher_correct_count"] or 0), int(main["opd_route_count"] or 0), int(main["sft_fallback_count"] or 0)] |
| axes[0].barh(labels, values, color=["#b7b7b7", "#6aa59d", PAPER_COLORS["pcd"], "#c7b9d8"], height=0.58) |
| axes[0].invert_yaxis() |
| _panel_label(axes[0], "a", "teacher-rescue funnel") |
| axes[0].set_xlabel("candidate count") |
| _paper_axes(axes[0]) |
| for index, value in enumerate(values): |
| axes[0].text(value, index, f" {_compact_count(value)}", va="center", fontsize=7) |
| if main.get("funnel_scope") != "exact_completion_counts": |
| axes[0].text(0.02, -0.16, "candidate-log proxy", transform=axes[0].transAxes, fontsize=6.5, color="#666666") |
|
|
| compare_rows = [row for row in rows if row["variant"] in {"deplot_no_vs_opd", "deplot_no_vs_opd_pcd"}] |
| compare_labels = [VARIANT_LABELS.get(row["variant"], row["variant"]) for row in compare_rows] |
| x_positions = list(range(len(compare_labels))) |
| width = 0.32 |
| probe_counts = [int(row["probe_candidate_count"] or 0) for row in compare_rows] |
| correct_counts = [int(row["teacher_correct_count"] or 0) for row in compare_rows] |
| axes[1].bar([x - width / 2 for x in x_positions], probe_counts, width=width, color=PAPER_COLORS["probe"], label="probed") |
| axes[1].bar([x + width / 2 for x in x_positions], correct_counts, width=width, color=PAPER_COLORS["correct"], label="teacher-correct") |
| if any(value > 0 for value in probe_counts + correct_counts): |
| axes[1].set_yscale("log") |
| _panel_label(axes[1], "b", "anchor vs PCD") |
| axes[1].set_ylabel("count (log)") |
| axes[1].set_xticks(x_positions) |
| axes[1].set_xticklabels(compare_labels, rotation=0) |
| _paper_axes(axes[1]) |
| axes[1].legend(frameon=False, loc="upper left", handlelength=1.0) |
| if len(probe_counts) == 2 and probe_counts[0] and correct_counts[0]: |
| axes[1].text(0.98, 0.94, f"{probe_counts[1] / probe_counts[0]:.1f}x probes\n{correct_counts[1] / correct_counts[0]:.1f}x rescued", transform=axes[1].transAxes, ha="right", va="top", fontsize=7, fontweight="bold", color=PAPER_COLORS["pcd"]) |
| fig.tight_layout(w_pad=1.4) |
| _save_figure(fig, out_dir, "fig5_teacher_rescue_funnel") |
| plt.close(fig) |
|
|
|
|
| def make_fig6(data: dict[str, dict[str, Any]], out_dir: Path) -> None: |
| fields = ["variant", "step_bin", "reward_std_mean", "opsd_adaptive_multiplier", "opsd_effective_weight", "opd_route_rate", "sft_route_rate", "teacher_correct_rate", "final_accuracy", "final_full_cot_rate", "final_other_rate"] |
| rows: list[dict[str, Any]] = [] |
| for variant in CANONICAL_VARIANTS: |
| final_eval = data.get(variant, {}).get("eval", {}) |
| binned_rows = bin_training_rows(data.get(variant, {}).get("train", [])) |
| if not binned_rows and final_eval: |
| rows.append({"variant": variant, "step_bin": "", "reward_std_mean": "", "opsd_adaptive_multiplier": "", "opsd_effective_weight": "", "opd_route_rate": "", "sft_route_rate": "", "teacher_correct_rate": "", "final_accuracy": _fmt_rate(final_eval.get("accuracy")), "final_full_cot_rate": _fmt_rate(final_eval.get("full_cot_rate")), "final_other_rate": _fmt_rate(final_eval.get("other_rate"))}) |
| continue |
| for row in binned_rows: |
| rows.append({"variant": variant, "step_bin": row["step_bin"], "reward_std_mean": _fmt_rate(row.get("reward_std_mean")), "opsd_adaptive_multiplier": _fmt_rate(row.get("opsd_adaptive_multiplier")), "opsd_effective_weight": _fmt_rate(row.get("opsd_effective_weight")), "opd_route_rate": _fmt_rate(row.get("opd_route_rate")), "sft_route_rate": _fmt_rate(row.get("sft_route_rate")), "teacher_correct_rate": _fmt_rate(row.get("teacher_correct_rate")), "final_accuracy": _fmt_rate(final_eval.get("accuracy")), "final_full_cot_rate": _fmt_rate(final_eval.get("full_cot_rate")), "final_other_rate": _fmt_rate(final_eval.get("other_rate"))}) |
| _write_csv(out_dir / "fig6_va_vs_pcd_diagnosis.csv", rows, fields) |
|
|
| plt = _setup_matplotlib() |
| fig, axes = plt.subplots(1, 3, figsize=(10.6, 3.05)) |
| for variant in ("deplot_no_vs_opd_va", "deplot_no_vs_opd_va_pcd"): |
| vrows = [row for row in rows if row["variant"] == variant and row["step_bin"] != ""] |
| x = [int(row["step_bin"]) for row in vrows] |
| y = [float(row["opsd_adaptive_multiplier"] or 0.0) for row in vrows] |
| if x: |
| axes[0].plot(x, y, color=VARIANT_COLORS.get(variant), alpha=0.18, linewidth=0.8) |
| axes[0].plot(x, _moving_average(y), linewidth=1.5, color=VARIANT_COLORS.get(variant), label=VARIANT_LABELS.get(variant, variant)) |
| _panel_label(axes[0], "a", "VA weight response") |
| axes[0].set_xlabel("step bin") |
| axes[0].set_ylabel("adaptive multiplier") |
| _paper_axes(axes[0]) |
| handles, labels = axes[0].get_legend_handles_labels() |
| if handles: |
| axes[0].legend(handles, labels, frameon=False, loc="best", handlelength=1.4) |
|
|
| last_by_variant = {variant: next((row for row in reversed(rows) if row["variant"] == variant), {}) for variant in CANONICAL_VARIANTS} |
| x = list(range(len(CANONICAL_VARIANTS))) |
| route_values = [float(last_by_variant[v].get("opd_route_rate") or 0.0) for v in CANONICAL_VARIANTS] |
| correct_counts = [_candidate_counts(data.get(v, {}))[1] for v in CANONICAL_VARIANTS] |
| width = 0.34 |
| axes[1].bar([i - width / 2 for i in x], route_values, width=width, color=[VARIANT_COLORS.get(v, "#2f6f73") for v in CANONICAL_VARIANTS], label="OPD route") |
| ax1b = axes[1].twinx() |
| ax1b.plot(x, correct_counts, color=PAPER_COLORS["correct"], marker="o", linewidth=1.3, label="teacher-correct") |
| _panel_label(axes[1], "b", "PCD route increase") |
| axes[1].set_ylabel("OPD route rate") |
| ax1b.set_ylabel("teacher-correct count") |
| axes[1].set_xticks(x) |
| axes[1].set_xticklabels([VARIANT_LABELS.get(v, v) for v in CANONICAL_VARIANTS], rotation=20, ha="right") |
| _paper_axes(axes[1]) |
| ax1b.spines["top"].set_visible(False) |
| ax1b.spines["left"].set_visible(False) |
| ax1b.spines["right"].set_linewidth(0.6) |
| lines, labels = axes[1].get_legend_handles_labels() |
| lines2, labels2 = ax1b.get_legend_handles_labels() |
| axes[1].legend(lines + lines2, labels + labels2, frameon=False, loc="upper left", handlelength=1.2) |
| pcd_index = list(CANONICAL_VARIANTS).index("deplot_no_vs_opd_pcd") |
| if route_values[pcd_index] > 0: |
| axes[1].text(pcd_index - width / 2, route_values[pcd_index] * 0.55, f"{route_values[pcd_index]:.3f}", ha="center", va="center", fontsize=7, fontweight="bold", color="white") |
|
|
| acc_values = [float(last_by_variant[v].get("final_accuracy") or 0.0) for v in CANONICAL_VARIANTS] |
| other_values = [float(last_by_variant[v].get("final_other_rate") or 0.0) for v in CANONICAL_VARIANTS] |
| axes[2].bar([i - width / 2 for i in x], acc_values, width=width, color=[VARIANT_COLORS.get(v, "#6f55a0") for v in CANONICAL_VARIANTS], label="Accuracy") |
| axes[2].bar([i + width / 2 for i in x], other_values, width=width, color="#b85c5c", label="Other output") |
| _panel_label(axes[2], "c", "final behavior") |
| axes[2].set_ylim(0, 1.05) |
| axes[2].set_xticks(x) |
| axes[2].set_xticklabels([VARIANT_LABELS.get(v, v) for v in CANONICAL_VARIANTS], rotation=20, ha="right") |
| axes[2].set_ylabel("rate") |
| _paper_axes(axes[2]) |
| axes[2].legend(frameon=False, loc="upper left", handlelength=1.2) |
| for index, value in enumerate(acc_values): |
| if CANONICAL_VARIANTS[index] in {"deplot_no_vs_opd", "deplot_no_vs_opd_pcd"}: |
| axes[2].text(index - width / 2, value + 0.015, f"{value:.3f}", ha="center", va="bottom", fontsize=7) |
| for index, value in enumerate(other_values): |
| if CANONICAL_VARIANTS[index] == "deplot_no_vs_opd_va_pcd": |
| axes[2].text(index + width / 2, value + 0.015, f"{value:.3f}", ha="center", va="bottom", fontsize=7, color=PAPER_COLORS["risk"], fontweight="bold") |
| fig.tight_layout(w_pad=1.2) |
| _save_figure(fig, out_dir, "fig6_va_vs_pcd_diagnosis") |
| plt.close(fig) |
|
|
|
|
| def make_table1(out_dir: Path) -> None: |
| rows = [ |
| ("SFT", "No", "Only as targets", "No", "No", "No", "Optional", "Cannot exploit online correctness signal"), |
| ("GRPO/RLVR", "Yes", "Discarded or negative", "Group-level", "No", "Yes", "Optional", "Sparse rewards leave all-wrong groups uninformative"), |
| ("GRPO + filtering", "Yes", "Filtered", "Partial", "No", "Yes", "Optional", "May throw away recoverable wrong completions"), |
| ("Distillation", "Teacher target", "Teacher-forced", "No", "No", "Depends", "Optional", "Risk of copying teacher errors or privileged traces"), |
| ("OPD-style correction", "Yes", "Yes", "Often prompt-level", "Weak", "Yes", "Optional", "Can over-correct unrecoverable wrong outputs"), |
| ("PCD-OPD", "Yes", "Routed", "Yes", "Yes", "Yes", "Yes", "Requires teacher-probe budget"), |
| ] |
| header = ["Method", "Learns from correct completions", "Uses wrong completions", "Completion-level routing", "Recoverability gate", "Avoids gold CoT", "Uses visual evidence", "Main limitation"] |
| lines = ["| " + " | ".join(header) + " |", "| " + " | ".join(["---"] * len(header)) + " |"] |
| lines.extend("| " + " | ".join(row) + " |" for row in rows) |
| (out_dir / "table1_method_positioning.md").write_text("\n".join(lines) + "\n", encoding="utf-8") |
|
|
|
|
| def _deplot_real_rate(records: list[dict[str, Any]]) -> float | None: |
| if not records: |
| return None |
| real = 0 |
| for record in records: |
| privileged = record.get("privileged") if isinstance(record.get("privileged"), dict) else {} |
| if privileged.get("visual_fact_deplot_status") == "real": |
| real += 1 |
| return real / len(records) |
|
|
|
|
| def make_table3(data: dict[str, dict[str, Any]], out_dir: Path) -> None: |
| rows: list[dict[str, Any]] = [] |
| for variant, payload in data.items(): |
| records = payload.get("candidates", []) |
| summary = summarize_candidates(records) |
| n = sum(int(row["n"]) for row in summary) |
| teacher_correct = sum(int(row["n"]) * float(row["teacher_correct_rate"] or 0.0) for row in summary) |
| parse_fail = sum(int(row["n"]) * float(row["parse_fail_rate"] or 0.0) for row in summary) |
| placeholder = sum(int(row["n"]) * float(row["placeholder_rate"] or 0.0) for row in summary) |
| tokens = [float(rec.get("teacher_output_word_count", 0) or 0) for rec in records] |
| rows.append({"section": "by_variant", "control": payload.get("manifest", {}).get("role") or variant, "variant": variant, "n": n, "teacher_correct_rate": _fmt_rate(teacher_correct / n if n else None), "parse_fail_rate": _fmt_rate(parse_fail / n if n else None), "placeholder_rate": _fmt_rate(placeholder / n if n else None), "generated_tokens_mean": _fmt_rate(_mean(tokens)), "deplot_real_rate": _fmt_rate(_deplot_real_rate(records)), "status": "from_candidate_log" if records else "missing"}) |
| fields = ["section", "control", "variant", "n", "teacher_correct_rate", "parse_fail_rate", "placeholder_rate", "generated_tokens_mean", "deplot_real_rate", "status"] |
| _write_csv(out_dir / "table3_recoverability_controls.csv", rows, fields) |
| table_rows = [] |
| for row in rows: |
| label = "PCD" if row["variant"] == "deplot_no_vs_opd_pcd" else VARIANT_LABELS.get(row["variant"], row["control"]) |
| table_rows.append([label, row["n"], row["teacher_correct_rate"], row["parse_fail_rate"], row["placeholder_rate"], row["deplot_real_rate"]]) |
| _write_booktabs_table(out_dir / "table3_recoverability_controls.md", ["Method", "Probe Cand. ↑", "Recover ↑", "Parse Fail ↓", "Placeholder ↓", "DePlot Real ↑"], table_rows, highlight_first_cell="PCD") |
|
|
|
|
| def make_table4(data: dict[str, dict[str, Any]], out_dir: Path) -> None: |
| fields = ["variant", "visual_supervision_on", "gold_cot_used", "deplot_evidence_on", "teacher_probe_gold_suffix_rate", "teacher_probe_deplot_real_rate", "teacher_probe_skipped_no_evidence_rate", "group_all_wrong_rate", "wrong_completion_rate", "teacher_probe_candidate_rate", "teacher_correct_rate", "grpo_route_rate", "opd_route_rate", "sft_route_rate", "generated_tokens_mean", "generated_tokens_p95", "data_quality"] |
| aliases = { |
| "teacher_probe_gold_suffix_rate": ("routing/teacher_probe_gold_suffix_rate",), |
| "teacher_probe_deplot_real_rate": ("routing/teacher_probe_deplot_real_rate",), |
| "teacher_probe_skipped_no_evidence_rate": ("routing/teacher_probe_skipped_no_evidence_rate",), |
| "group_all_wrong_rate": ("signal/group_all_wrong_rate",), |
| "wrong_completion_rate": ("routing/wrong_completion_rate",), |
| "teacher_probe_candidate_rate": ("routing/teacher_probe_candidate_rate",), |
| "teacher_correct_rate": ("routing/teacher_probe_correct_rate",), |
| "grpo_route_rate": ("routing/grpo_route_rate", "routing/grpo_on_correct_rate"), |
| "opd_route_rate": ("routing/opd_route_rate", "routing/opd_teacher_call_rate"), |
| "sft_route_rate": ("routing/sft_route_rate", "routing/sft_replaced_ratio"), |
| "generated_tokens_mean": ("teacher_probe/generated_tokens_mean",), |
| "generated_tokens_p95": ("teacher_probe/generated_tokens_p95",), |
| } |
| rows: list[dict[str, Any]] = [] |
| for variant, payload in data.items(): |
| train_rows = payload.get("train", []) |
| out: dict[str, Any] = {"variant": variant} |
| for key, key_aliases in aliases.items(): |
| values = [] |
| for row in train_rows: |
| for alias in key_aliases: |
| value = _safe_float(row.get(alias)) |
| if value is not None: |
| values.append(value) |
| break |
| out[key] = _fmt_rate(_mean(values)) |
| config_path = payload.get("manifest", {}).get("config_path") |
| config_text = config_path.read_text(encoding="utf-8", errors="replace").lower() if config_path and hasattr(config_path, "exists") and config_path.exists() else "" |
| out["visual_supervision_on"] = "false" if "visual_supervision=false" in config_text else "unknown" |
| out["gold_cot_used"] = "true" if '"text_include_gold": true' in config_text or "'text_include_gold': true" in config_text else "false" |
| out["deplot_evidence_on"] = "true" if "deplot" in config_text or (_deplot_real_rate(payload.get("candidates", [])) or 0) > 0 else "unknown" |
| exact_train = bool(train_rows and all(_count_value(train_rows[0], key) is not None for key in COUNT_ALIASES)) |
| exact_candidate = bool(payload.get("candidates") and "group_all_wrong" in payload["candidates"][0]) |
| out["data_quality"] = "exact" if exact_train and exact_candidate else "proxy" |
| rows.append(out) |
| _write_csv(out_dir / "table4_routing_antileakage.csv", rows, fields) |
| table_rows = [] |
| for row in rows: |
| variant = row["variant"] |
| eval_summary = data.get(variant, {}).get("eval", {}) |
| label = "PCD" if variant == "deplot_no_vs_opd_pcd" else VARIANT_LABELS.get(variant, variant) |
| table_rows.append([label, _fmt_rate(eval_summary.get("accuracy")), _fmt_rate(eval_summary.get("other_rate")), row["gold_cot_used"], row["teacher_probe_gold_suffix_rate"], row["teacher_probe_candidate_rate"], row["teacher_correct_rate"], row["opd_route_rate"], row["data_quality"]]) |
| _write_booktabs_table(out_dir / "table4_routing_antileakage.md", ["Method", "Acc ↑", "Other ↓", "Gold CoT", "Gold Suffix ↓", "Probe Cand. ↑", "Teacher Correct ↑", "OPD Route ↑", "Quality"], table_rows, highlight_first_cell="PCD") |
|
|
|
|
| def _candidate_summary(payload: dict[str, Any]) -> dict[str, float | None]: |
| records = payload.get("candidates", []) |
| if not records: |
| return {"probe_count": None, "teacher_correct_count": None, "teacher_correct_rate": None, "opd_count": None, "sft_fallback_count": None} |
| probe_count = float(len(records)) |
| teacher_correct_count = float(sum(1 for rec in records if rec.get("teacher_correct") is True)) |
| return {"probe_count": probe_count, "teacher_correct_count": teacher_correct_count, "teacher_correct_rate": _safe_ratio(teacher_correct_count, probe_count), "opd_count": float(sum(1 for rec in records if rec.get("final_route") == "opd")), "sft_fallback_count": float(sum(1 for rec in records if str(rec.get("final_route", "")).startswith("sft")))} |
|
|
|
|
| def _final_eval(payload: dict[str, Any], key: str) -> float | None: |
| return payload.get("eval", {}).get(key) |
|
|
|
|
| def make_argument_report(data: dict[str, dict[str, Any]], out_dir: Path) -> None: |
| anchor = data.get("deplot_no_vs_opd", {}) |
| pcd = data.get("deplot_no_vs_opd_pcd", {}) |
| va = data.get("deplot_no_vs_opd_va", {}) |
| va_pcd = data.get("deplot_no_vs_opd_va_pcd", {}) |
| anchor_summary = _candidate_summary(anchor) |
| pcd_summary = _candidate_summary(pcd) |
| va_summary = _candidate_summary(va) |
| va_pcd_summary = _candidate_summary(va_pcd) |
| probe_mult = _safe_ratio(pcd_summary["probe_count"], anchor_summary["probe_count"]) |
| correct_mult = _safe_ratio(pcd_summary["teacher_correct_count"], anchor_summary["teacher_correct_count"]) |
| anchor_acc = _final_eval(anchor, "accuracy") |
| pcd_acc = _final_eval(pcd, "accuracy") |
| va_acc = _final_eval(va, "accuracy") |
| va_pcd_acc = _final_eval(va_pcd, "accuracy") |
| pcd_gain = pcd_acc - anchor_acc if pcd_acc is not None and anchor_acc is not None else None |
| va_gain = va_acc - anchor_acc if va_acc is not None and anchor_acc is not None else None |
| va_pcd_gain = va_pcd_acc - anchor_acc if va_pcd_acc is not None and anchor_acc is not None else None |
| lines = [ |
| "# PCD-OPD Paper Argument Report", |
| "", |
| "## Recommended paper claim", |
| "", |
| "PCD-OPD should be framed as a completion-level recoverability gate: it expands the set of wrong completions that can be checked by a teacher, converts the teacher-correct subset into OPD trajectory supervision, and leaves unrecoverable cases on the SFT fallback path. PCD is the main method; VA should be presented as an auxiliary ablation with visible side effects.", |
| "", |
| "## Evidence from current artifacts", |
| "", |
| "| Claim | Evidence | Figure/Table | Caveat |", |
| "| --- | --- | --- | --- |", |
| f"| PCD expands recoverable wrong-completion supervision | Probe candidates increase from {_fmt_count(anchor_summary['probe_count'])} to {_fmt_count(pcd_summary['probe_count'])}{f' ({probe_mult:.1f}x)' if probe_mult is not None else ''}; teacher-correct candidates increase from {_fmt_count(anchor_summary['teacher_correct_count'])} to {_fmt_count(pcd_summary['teacher_correct_count'])}{f' ({correct_mult:.1f}x)' if correct_mult is not None else ''}. | Figure 5, Table 3 | Current funnel uses candidate-log proxy because exact total/wrong counts are missing in old logs. |", |
| f"| PCD is the main useful mechanism among the tested variants | Final accuracy: anchor={_fmt_rate(anchor_acc)}, PCD={_fmt_rate(pcd_acc)}, VA={_fmt_rate(va_acc)}, VA+PCD={_fmt_rate(va_pcd_acc)}; PCD gain over anchor={_fmt_rate(pcd_gain)}, VA-only gain={_fmt_rate(va_gain)}, VA+PCD gain={_fmt_rate(va_pcd_gain)}. | Figure 6 | Treat as mechanism diagnosis unless promoted into the main-result table. |", |
| f"| VA changes weighting but is not sufficient as the main story | VA-only teacher-correct candidates={_fmt_count(va_summary['teacher_correct_count'])}; VA+PCD teacher-correct candidates={_fmt_count(va_pcd_summary['teacher_correct_count'])}; VA-only final accuracy={_fmt_rate(va_acc)}. | Figure 6 | Keep VA as auxiliary ablation, not the headline method. |", |
| "| Anti-leakage needs a stricter exact-log rerun before final claims | Table 4 shows gold_cot_used=false and DePlot evidence on, but old logs report nonzero teacher_probe_gold_suffix_rate under proxy quality. | Table 4, data_quality_report | Treat anti-leakage as an audit target, not a concluded claim, until rerun with exact no-gold teacher-probe logging. |", |
| "", |
| "## Recommended figure changes already applied", |
| "", |
| "- Figure 1/4/5/6 now use paper-style panels with serif fonts, small multiples, light grids, and fixed semantic colors.", |
| "- Figure 5 now falls back to a candidate-level funnel when exact generated/wrong completion counts are absent.", |
| "- Figure 2 is intentionally removed from this artifact pipeline; method schematics should be handled outside this dashboard.", |
| "- The dashboard includes this report so figure inspection and paper claims stay connected.", |
| ] |
| (out_dir / "paper_argument_report.md").write_text("\n".join(lines) + "\n", encoding="utf-8") |
|
|
|
|
| def make_chart_review_report(data: dict[str, dict[str, Any]], out_dir: Path) -> None: |
| anchor = data.get("deplot_no_vs_opd", {}) |
| pcd = data.get("deplot_no_vs_opd_pcd", {}) |
| va = data.get("deplot_no_vs_opd_va", {}) |
| va_pcd = data.get("deplot_no_vs_opd_va_pcd", {}) |
| anchor_probe, anchor_correct = _candidate_counts(anchor) |
| pcd_probe, pcd_correct = _candidate_counts(pcd) |
| rows = [ |
| ("fig0_training_basics", "Appendix diagnostic; acceptable but secondary to the paper-style review figures.", "Use in paper: appendix or training sanity check.", "Shows loss, accuracy reward, reward std, and degeneration trends without overclaiming performance.", "Not a main-method proof; keep it as diagnostics."), |
| ("fig1_motivation", "Paper-style review: three compact panels separate candidate volume, recoverability rate, and qualitative routing.", "Use in paper: motivation or appendix until exact all-wrong logging is rerun.", f"PCD probes {pcd_probe:,} candidates and finds {pcd_correct:,} teacher-correct cases, versus anchor {anchor_probe:,}/{anchor_correct:,}.", "Old logs lack exact group_all_wrong_rate; exact rerun should replace proxy wording."), |
| ("fig4_training_dynamics", "Paper-style review: small multiples highlight anchor vs PCD trends while de-emphasizing auxiliary variants.", "Use in paper: mechanism diagnosis, likely appendix if main table already carries performance.", f"PCD final accuracy {_fmt_rate(pcd.get('eval', {}).get('accuracy'))} beats anchor {_fmt_rate(anchor.get('eval', {}).get('accuracy'))} while greatly increasing probe volume.", "Route/count exactness still depends on rerun quality report."), |
| ("fig5_teacher_rescue_funnel", "Paper-style review: strongest mechanism figure; funnel plus anchor/PCD inset directly communicates rescue amplification.", "Use in paper: main mechanism figure after exact-count rerun; current version can support draft.", f"Teacher-correct candidates grow from {anchor_correct:,} to {pcd_correct:,}; PCD fallback remains explicit.", "Current funnel scope is candidate_proxy because old logs miss generated/wrong counts."), |
| ("fig6_va_vs_pcd_diagnosis", "Paper-style review: separates VA weighting, PCD route increase, and final behavior without mixing mechanisms.", "Use in paper: ablation/mechanism figure.", f"VA-only accuracy {_fmt_rate(va.get('eval', {}).get('accuracy'))} underperforms; PCD reaches {_fmt_rate(pcd.get('eval', {}).get('accuracy'))}; VA+PCD other-output rate is {_fmt_rate(va_pcd.get('eval', {}).get('other_rate'))}.", "Use this to argue VA is auxiliary, not the main method."), |
| ("table4_routing_antileakage", "Useful but not yet final: it exposes that the current logs are proxy-quality for leakage-related routing fields.", "Use in paper: anti-leakage audit table only after exact rerun.", "gold_cot_used=false is encouraging, but teacher_probe_gold_suffix_rate is nonzero in old proxy logs.", "Do not claim anti-leakage is fully proven until exact no-gold teacher-probe logs show zero gold suffix rate."), |
| ] |
| lines = ["# Chart Review Report", "", "This paper-style review records the manual dashboard review and the plotting changes applied to make each artifact usable for the paper argument.", "", "| Artifact | Visual assessment | Use in paper | Evidence | Caveat / next setting |", "| --- | --- | --- | --- | --- |"] |
| lines.extend("| " + " | ".join(row) + " |" for row in rows) |
| (out_dir / "chart_review_report.md").write_text("\n".join(lines) + "\n", encoding="utf-8") |
|
|
|
|
| def write_quality_report(data: dict[str, dict[str, Any]], out_dir: Path) -> None: |
| lines = ["# PCD Paper Data Quality", "", "| variant | train_counts | candidate_group_fields | eval_log |", "| --- | --- | --- | --- |"] |
| for variant, payload in data.items(): |
| train_rows = payload.get("train", []) |
| candidates = payload.get("candidates", []) |
| train_quality = "exact" if train_rows and all(_count_value(train_rows[0], key) is not None for key in COUNT_ALIASES) else "missing" |
| candidate_quality = "exact" if candidates and "group_all_wrong" in candidates[0] else ("proxy" if candidates else "missing") |
| eval_quality = "exact" if payload.get("eval", {}).get("accuracy") is not None else "missing" |
| lines.append(f"| {variant} | {train_quality} | {candidate_quality} | {eval_quality} |") |
| (out_dir / "data_quality_report.md").write_text("\n".join(lines) + "\n", encoding="utf-8") |
|
|
|
|
| def build_registry() -> ArtifactRegistry: |
| registry = ArtifactRegistry() |
| registry.register(ArtifactSpec(id="fig0_training_basics", title="Figure 0: Training Basics", kind="figure", description="Loss, reward, reward-std, OPD/VA weights, and generation-health curves.", outputs=("fig0_training_basics.png", "fig0_training_basics.pdf", "fig0_training_basics.csv"), tags=("training", "loss", "reward", "diagnostic"), producer=lambda ctx: make_fig0(ctx.data, ctx.out_dir), aliases=("fig0", "training_basics", "basics"))) |
| registry.register(ArtifactSpec(id="fig1_motivation", title="Figure 1: Motivation", kind="figure", description="Wrong-completion heterogeneity and teacher-rescue evidence.", outputs=("fig1_motivation.png", "fig1_motivation.pdf", "fig1_motivation.csv", "fig1_cases.md"), tags=("motivation", "teacher-rescue"), producer=lambda ctx: make_fig1(ctx.data, ctx.out_dir), aliases=("fig1", "motivation"))) |
| registry.register(ArtifactSpec(id="fig4_training_dynamics", title="Figure 4: Training Dynamics", kind="figure", description="Reward, routing, and teacher-probe dynamics over training.", outputs=("fig4_training_dynamics.png", "fig4_training_dynamics.pdf", "fig4_training_dynamics.csv"), tags=("training", "routing", "teacher-probe"), producer=lambda ctx: make_fig4(ctx.data, ctx.out_dir), aliases=("fig4", "dynamics"))) |
| registry.register(ArtifactSpec(id="fig5_teacher_rescue_funnel", title="Figure 5: Teacher-Rescue Funnel", kind="figure", description="Generated-to-wrong-to-probed-to-rescued completion funnel.", outputs=("fig5_teacher_rescue_funnel.png", "fig5_teacher_rescue_funnel.pdf", "fig5_teacher_rescue_funnel.csv"), tags=("funnel", "teacher-rescue", "routing"), producer=lambda ctx: make_fig5(ctx.data, ctx.out_dir), aliases=("fig5", "funnel"))) |
| registry.register(ArtifactSpec(id="fig6_va_vs_pcd_diagnosis", title="Figure 6: VA vs PCD Diagnosis", kind="figure", description="Separates adaptive OPD weighting from PCD route effects.", outputs=("fig6_va_vs_pcd_diagnosis.png", "fig6_va_vs_pcd_diagnosis.pdf", "fig6_va_vs_pcd_diagnosis.csv"), tags=("ablation", "variance-adaptive", "pcd"), producer=lambda ctx: make_fig6(ctx.data, ctx.out_dir), aliases=("fig6", "va_diagnosis"))) |
| registry.register(ArtifactSpec(id="table1_method_positioning", title="Table 1: Method Positioning", kind="table", description="Static positioning table for SFT/RLVR/filtering/distillation/OPD/PCD-OPD.", outputs=("table1_method_positioning.md",), tags=("table", "related-work"), producer=lambda ctx: make_table1(ctx.out_dir), aliases=("table1",))) |
| registry.register(ArtifactSpec(id="table3_recoverability_controls", title="Table 3: Recoverability Controls", kind="table", description="Compatible schema combining by-variant recoverability rows and offline probe-control rows.", outputs=("table3_recoverability_controls.csv", "table3_recoverability_controls.md"), tags=("table", "recoverability", "controls"), producer=lambda ctx: make_table3(ctx.data, ctx.out_dir), aliases=("table3",))) |
| registry.register(ArtifactSpec(id="table4_routing_antileakage", title="Table 4: Routing and Anti-Leakage", kind="table", description="Routing statistics, evidence usage, and no-gold-CoT checks.", outputs=("table4_routing_antileakage.csv", "table4_routing_antileakage.md"), tags=("table", "routing", "anti-leakage"), producer=lambda ctx: make_table4(ctx.data, ctx.out_dir), aliases=("table4",))) |
| registry.register(ArtifactSpec(id="paper_argument_report", title="Paper Argument Report", kind="report", description="Concise claims, supporting numbers, caveats, and next experiment setting for the paper.", outputs=("paper_argument_report.md",), tags=("paper", "claims", "diagnosis"), producer=lambda ctx: make_argument_report(ctx.data, ctx.out_dir), aliases=("argument", "claims"))) |
| registry.register(ArtifactSpec(id="chart_review_report", title="Chart Review Report", kind="report", description="Manual dashboard review: visual quality, paper use, evidence, and caveats for each figure.", outputs=("chart_review_report.md",), tags=("paper", "review", "figures"), producer=lambda ctx: make_chart_review_report(ctx.data, ctx.out_dir), aliases=("chart_review", "figure_review"))) |
| registry.register(ArtifactSpec(id="data_quality_report", title="Data Quality Report", kind="report", description="Marks exact/proxy/missing fields for each run.", outputs=("data_quality_report.md",), tags=("quality", "report"), producer=lambda ctx: write_quality_report(ctx.data, ctx.out_dir), aliases=("quality",))) |
| registry.register(ArtifactSpec(id="dashboard", title="HTML Dashboard", kind="dashboard", description="Dynamic local HTML browser for all registered figures and tables.", outputs=("index.html", "artifacts_manifest.json"), tags=("dashboard", "html"), producer=None, aliases=("html", "index"))) |
| return registry |
|
|
|
|
| def run(makes: set[str], manifest: Path, out_dir: Path) -> int: |
| out_dir.mkdir(parents=True, exist_ok=True) |
| manifest_rows = read_manifest(manifest) |
| data = collect_run_data(manifest_rows) |
| registry = build_registry() |
| ctx = SimpleNamespace(data=data, out_dir=out_dir, manifest_rows=manifest_rows, manifest_path=manifest) |
| selected = registry.run(makes, ctx) |
| selected_ids = {spec.id for spec in selected} |
| if selected_ids and "data_quality_report" not in selected_ids: |
| write_quality_report(data, out_dir) |
| write_artifacts_manifest(registry=registry, out_dir=out_dir) |
| write_dashboard_html(out_dir=out_dir) |
| return 0 |
|
|
|
|
| def main(argv: list[str] | None = None) -> int: |
| parser = argparse.ArgumentParser(description="Build PCD-OPD paper non-main-result artifacts.") |
| parser.add_argument("--manifest", type=Path, required=True) |
| parser.add_argument("--out-dir", type=Path, default=Path("docs/figures/pcd_paper")) |
| parser.add_argument("--make", action="append", default=None) |
| args = parser.parse_args(argv) |
| requested: set[str] = set() |
| for item in args.make or ["all"]: |
| requested.update(part.strip() for part in str(item).split(",") if part.strip()) |
| return run(requested, args.manifest, args.out_dir) |
|
|
|
|
| if __name__ == "__main__": |
| raise SystemExit(main()) |
|
|
