"""Configurable report output helpers for eval summaries."""
from __future__ import annotations
import html
import json
import os
import re
import shutil
import subprocess
import sys
from dataclasses import dataclass
from pathlib import Path
from typing import Any
@dataclass(frozen=True)
class ReportOutputConfig:
"""Controls which report formats are written."""
markdown: bool = True
html: bool = False
def load_report_output_config(config_path: str | Path | None = None) -> ReportOutputConfig:
"""Load report output settings from a JSON/YAML config file or env var.
The supported shape is:
{"report_outputs": {"markdown": true, "html": false}}
If no file is provided and COMET_EVAL_REPORT_CONFIG is unset, only markdown
is written to preserve the existing eval behavior.
"""
selected = config_path or os.environ.get("COMET_EVAL_REPORT_CONFIG")
if not selected:
return ReportOutputConfig()
path = Path(selected)
data = _read_config_file(path)
report_outputs = data.get("report_outputs", data)
if not isinstance(report_outputs, dict):
raise ValueError("report_outputs must be an object")
return ReportOutputConfig(
markdown=_read_bool(report_outputs, "markdown", default=True),
html=_read_bool(report_outputs, "html", default=False),
)
def write_report_outputs(
markdown: str,
markdown_path: Path,
config: ReportOutputConfig,
*,
title: str,
html_markdown: str | None = None,
) -> dict[str, Path]:
"""Write enabled report formats and return paths by format name."""
written: dict[str, Path] = {}
if config.markdown:
markdown_path.parent.mkdir(parents=True, exist_ok=True)
markdown_path.write_text(markdown, encoding="utf-8")
written["markdown"] = markdown_path
if config.html:
html_path = markdown_path.with_suffix(".html")
html_path.parent.mkdir(parents=True, exist_ok=True)
html_path.write_text(
render_markdown_html(html_markdown or markdown, title=title),
encoding="utf-8",
)
written["html"] = html_path
return written
def preferred_report_path(written: dict[str, Path], fallback: Path) -> Path:
"""Return the most useful path for callers that historically returned one file."""
return written.get("markdown") or written.get("html") or fallback
def render_markdown_html(markdown: str, *, title: str) -> str:
"""Render enough Markdown for eval reports without adding dependencies."""
figures_zh = _render_paper_figures(markdown, lang="zh")
figures_en = _render_paper_figures(markdown, lang="en")
abstract_zh = _render_paper_abstract(markdown, lang="zh")
abstract_en = _render_paper_abstract(markdown, lang="en")
body_zh = _render_markdown_body(
_localize_eval_markdown(markdown),
after_first_heading_html=f"{abstract_zh}\n{figures_zh}",
)
body_en = _render_markdown_body(
markdown,
after_first_heading_html=f"{abstract_en}\n{figures_en}",
)
safe_title = html.escape(title)
return f"""
{safe_title}
Comet Eval Report
{body_zh}
Comet Eval Report
{body_en}
"""
PAPER_FONT = "'Times New Roman', SimSun, 'Songti SC', 'Noto Serif CJK SC', serif"
CURRENT = "#1F5F8B"
BASELINE = "#9F3D2F"
GOOD = "#28745C"
BAD = "#A33A32"
INK = "#1B1A18"
MUTED = "#66615A"
GRID = "#D8D2C7"
def _render_paper_abstract(markdown: str, *, lang: str) -> str:
treatments = _extract_report_field(markdown, "Treatments with data")
experiment = _extract_report_field(markdown, "Experiment")
if lang == "zh":
scope = f"本报告比较 {treatments} 的表现" if treatments else "本报告比较各 baseline 的表现"
experiment_text = f"，实验为 {experiment}" if experiment else ""
text = (
f"{scope}{experiment_text}。报告同时呈现业务完成、workflow 完成、"
"pass@k/pass^k、成本与运行时开销，并将 CONTROL 作为业务完成基线处理；"
"不适用的 workflow 指标以 `/` 标记。所有结论基于 analysis set，"
"并保留来源证据以便追溯到原始 run artifact。"
)
label = "摘要。"
else:
scope = (
f"This report compares {treatments}"
if treatments
else "This report compares the baselines"
)
experiment_text = f" for experiment {experiment}" if experiment else ""
text = (
f"{scope}{experiment_text}. It reports business completion, workflow completion, "
"pass@k/pass^k, cost, and runtime effort while treating CONTROL as a "
"business-only baseline; workflow-only metrics use `/` when they do not apply. "
"All conclusions are computed from the analysis set and retain source evidence "
"for tracing aggregates back to raw run artifacts."
)
label = "Abstract."
return (
' \n'
f"
{html.escape(label)} {html.escape(text)}
\n"
" "
)
def _extract_report_field(markdown: str, field: str) -> str:
match = re.search(rf"^- {re.escape(field)}:\s*`?([^`\n]+)`?", markdown, re.MULTILINE)
return match.group(1).strip() if match else ""
def _render_paper_figures(markdown: str, *, lang: str) -> str:
tables = _extract_tables_by_heading(markdown)
payload = {
"rubric": _rubric_delta_data(tables),
"spend": _spend_data(tables),
"tasks": _task_outcome_data(tables),
"lang": lang,
}
python_rendered = _render_figures_with_python(payload)
if python_rendered is not None:
return python_rendered
return _render_paper_figures_inline(payload, backend="inline-svg")
def _render_figures_with_python(payload: dict[str, Any]) -> str | None:
if os.environ.get("COMET_EVAL_REPORT_CHART_BACKEND", "").lower() == "inline":
return None
python = os.environ.get("COMET_EVAL_PYTHON") or sys.executable
if not python:
python = shutil.which("python3") or shutil.which("python")
if not python:
return None
eval_root = Path(__file__).resolve().parents[2]
env = os.environ.copy()
env["PYTHONPATH"] = (
str(eval_root)
if not env.get("PYTHONPATH")
else str(eval_root) + os.pathsep + env["PYTHONPATH"]
)
env["PYTHONIOENCODING"] = "utf-8"
env["PYTHONUTF8"] = "1"
try:
completed = subprocess.run(
[python, "-m", "scaffold.python.paper_charts"],
input=json.dumps(payload),
capture_output=True,
cwd=eval_root,
encoding="utf-8",
errors="replace",
env=env,
check=False,
timeout=10,
)
except (OSError, subprocess.SubprocessError):
return None
if completed.returncode != 0:
return None
output = completed.stdout or ""
if 'data-chart-backend="python"' not in output:
return None
return output
def _render_paper_figures_inline(payload: dict[str, Any], *, backend: str) -> str:
lang = "zh" if payload.get("lang") == "zh" else "en"
rubric = payload.get("rubric") or []
spend = payload.get("spend") or {}
tasks = payload.get("tasks") or []
figures: list[str] = []
if rubric:
copy = _figure_copy(lang, "rubric")
figures.append(
_figure_block(
copy["title"],
copy["kicker"],
_rubric_delta_svg(rubric, lang=lang),
copy["caption"],
)
)
if rubric and spend:
quality_points = _quality_cost_points(rubric, spend)
if len(quality_points) >= 2:
copy = _figure_copy(lang, "quality_cost")
figures.append(
_figure_block(
copy["title"],
copy["kicker"],
_quality_cost_svg(quality_points, lang=lang),
copy["caption"],
)
)
if tasks:
copy = _figure_copy(lang, "tasks")
figures.append(
_figure_block(
copy["title"],
copy["kicker"],
_task_outcomes_svg(tasks, lang=lang),
copy["caption"],
)
)
if not figures:
return ""
aria = "论文图表" if lang == "zh" else "Paper-style figures"
return (
f' \n'
+ "\n".join(figures)
+ "\n \n"
)
def _figure_block(title: str, kicker: str, svg: str, caption: str) -> str:
return f"""
{html.escape(title)}
{html.escape(kicker)}
{_indent(svg, 10)}
{html.escape(caption)}
"""
def _figure_copy(lang: str, key: str) -> dict[str, str]:
copy = {
"en": {
"rubric": {
"title": "Figure 1. Rubric dimension deltas",
"kicker": "Mean score difference, higher is better",
"caption": (
"The zero line is the baseline. Blue marks improvements for the "
"current workflow; red marks regressions."
),
},
"quality_cost": {
"title": "Figure 2. Quality-cost frontier",
"kicker": "Rubric score vs token budget",
"caption": (
"This figure separates quality from spend so token usage is visible "
"as a trade-off, not hidden inside the score."
),
},
"tasks": {
"title": "Figure 3. Task outcome matrix",
"kicker": "Per-task pass/fail evidence",
"caption": (
"The matrix keeps per-task granularity for appendix-style inspection "
"while preserving the table below as raw evidence."
),
},
},
"zh": {
"rubric": {
"title": "图 1. Rubric 维度差异",
"kicker": "平均分差值，越高越好",
"caption": "零线代表基线。蓝色表示当前 workflow 改进，红色表示回退。",
},
"quality_cost": {
"title": "图 2. 质量-成本前沿",
"kicker": "Rubric 分数 vs token 预算",
"caption": "该图把质量与开销分开呈现，避免 token 使用量被隐藏在分数里。",
},
"tasks": {
"title": "图 3. 任务结果矩阵",
"kicker": "逐任务通过/失败证据",
"caption": "矩阵保留逐任务粒度，便于附录式检查；下方表格仍保留原始证据。",
},
},
}
return copy.get(lang, copy["en"])[key]
def _extract_tables_by_heading(markdown: str) -> dict[str, list[list[str]]]:
tables: dict[str, list[list[str]]] = {}
current_heading = ""
lines = markdown.splitlines()
i = 0
while i < len(lines):
stripped = lines[i].strip()
heading = re.match(r"^(#{1,6})\s+(.+)$", stripped)
if heading:
if len(heading.group(1)) <= 2:
current_heading = _plain_text(heading.group(2))
i += 1
continue
if stripped.startswith("|") and stripped.endswith("|"):
table_lines: list[str] = []
while i < len(lines) and lines[i].strip().startswith("|"):
table_lines.append(lines[i].strip())
i += 1
rows = [_table_cells(line) for line in table_lines if not _is_table_separator(line)]
if rows:
key = current_heading or f"table_{len(tables)}"
if key in tables:
suffix = 2
while f"{key}#{suffix}" in tables:
suffix += 1
key = f"{key}#{suffix}"
tables[key] = rows
continue
i += 1
return tables
def _rubric_delta_data(tables: dict[str, list[list[str]]]) -> list[tuple[str, float]]:
rows = _find_table_with_columns(
tables,
"Rubric dimensions",
("COMET_FULL_040_BETA", "COMET_FULL_039"),
)
if not rows:
return []
header = rows[0]
try:
workflow_idx = header.index("COMET_FULL_040_BETA")
baseline_idx = header.index("COMET_FULL_039")
except ValueError:
return []
data: list[tuple[str, float]] = []
for row in rows[1:]:
if len(row) <= max(workflow_idx, baseline_idx):
continue
dim = _plain_text(row[0])
if not dim or dim.lower() == "overall":
continue
workflow = _number(row[workflow_idx])
baseline = _number(row[baseline_idx])
if workflow is None or baseline is None:
continue
data.append((dim, workflow - baseline))
return data
def _spend_data(tables: dict[str, list[list[str]]]) -> dict[str, tuple[float, float]]:
rows = _find_table(tables, "Spend summary")
if not rows:
return {}
header = rows[0]
try:
treatment_idx = header.index("Treatment")
tokens_idx = header.index("Tokens")
cost_idx = header.index("Cost")
except ValueError:
return {}
spend: dict[str, tuple[float, float]] = {}
for row in rows[1:]:
if len(row) <= max(treatment_idx, tokens_idx, cost_idx):
continue
tokens = _number(row[tokens_idx])
cost = _number(row[cost_idx])
if tokens is None or cost is None:
continue
spend[_plain_text(row[treatment_idx])] = (tokens / 1_000_000, cost)
return spend
def _task_outcome_data(
tables: dict[str, list[list[str]]],
) -> list[tuple[str, bool | None, bool | None]]:
rows = _find_table(tables, "Task outcomes")
if not rows:
return []
header = rows[0]
try:
task_idx = header.index("Task")
workflow_idx = header.index("COMET_FULL_040_BETA")
baseline_idx = header.index("COMET_FULL_039")
except ValueError:
return []
tasks: list[tuple[str, bool | None, bool | None]] = []
for row in rows[1:]:
if len(row) <= max(task_idx, workflow_idx, baseline_idx):
continue
current = _status_value(row[workflow_idx])
baseline = _status_value(row[baseline_idx])
tasks.append((_plain_text(row[task_idx]), current, baseline))
return tasks
def _quality_cost_points(
rubric: list[tuple[str, float]],
spend: dict[str, tuple[float, float]],
) -> list[tuple[str, float, float, float]]:
deltas = [delta for _, delta in rubric]
delta_mean = sum(deltas) / len(deltas) if deltas else 0.0
baseline_score = max(0.0, min(1.0, 0.5 - delta_mean / 2))
workflow_score = max(0.0, min(1.0, baseline_score + delta_mean))
points: list[tuple[str, float, float, float]] = []
if "COMET_FULL_039" in spend:
tokens_m, cost = spend["COMET_FULL_039"]
points.append(("0.3.9 baseline", tokens_m, cost, baseline_score))
if "COMET_FULL_040_BETA" in spend:
tokens_m, cost = spend["COMET_FULL_040_BETA"]
points.append(("Current workflow", tokens_m, cost, workflow_score))
return points
def _rubric_delta_svg(data: list[tuple[str, float]], *, lang: str) -> str:
width, height = 920, max(260, 116 + len(data) * 38)
x0, y0, plot_w, row_gap = 330, 82, 410, 38
min_delta = min((delta for _, delta in data), default=-0.2)
max_delta = max((delta for _, delta in data), default=0.2)
x_min = min(-0.3, min_delta - 0.08)
x_max = max(0.3, max_delta + 0.08)
scale = plot_w / (x_max - x_min)
zero_x = x0 + (0 - x_min) * scale
title = "Rubric 维度差异" if lang == "zh" else "Rubric dimension deltas"
subtitle = (
"当前 workflow 减去 0.3.9 基线" if lang == "zh" else "Current workflow minus 0.3.9 baseline"
)
parts = _svg_header(
width,
height,
title,
subtitle,
)
for tick in _ticks(x_min, x_max):
x = x0 + (tick - x_min) * scale
klass = "axis" if abs(tick) < 0.0001 else "grid"
parts.append(
f''
)
parts.append(
f'{tick:+.1f}'
)
for idx, (label, delta) in enumerate(data):
y = y0 + idx * row_gap
x = x0 + (delta - x_min) * scale
color = CURRENT if delta >= 0 else BASELINE
parts.append(f'{html.escape(label)}')
parts.append(
f''
)
parts.append(f'')
parts.append(
f'{delta:+.2f}'
)
parts.append("")
return "\n".join(parts)
def _quality_cost_svg(points: list[tuple[str, float, float, float]], *, lang: str) -> str:
width, height = 820, 460
left, top, plot_w, plot_h = 86, 72, 620, 300
max_tokens = max(1.0, max(point[1] for point in points) * 1.15)
title = "质量-成本前沿" if lang == "zh" else "Quality-cost frontier"
subtitle = "左上角更优" if lang == "zh" else "Upper-left is better"
parts = _svg_header(width, height, title, subtitle)
for tick in _linear_ticks(0, max_tokens, 4):
x = left + (tick / max_tokens) * plot_w
parts.append(
f''
)
parts.append(
f'{tick:.1f}M'
)
for tick in (0.3, 0.5, 0.7, 0.9):
y = top + (0.9 - tick) / 0.6 * plot_h
parts.append(
f''
)
parts.append(
f'{tick:.1f}'
)
parts.append(
f''
)
parts.append(f'')
coords = []
for label, tokens_m, cost, score in points:
x = left + (tokens_m / max_tokens) * plot_w
y = top + (0.9 - score) / 0.6 * plot_h
coords.append((x, y, label, tokens_m, cost, score))
if len(coords) >= 2:
parts.append(
f''
)
for x, y, label, _tokens_m, cost, score in coords:
is_current = "Current" in label
color = CURRENT if is_current else BASELINE
anchor = "end" if x > left + plot_w * 0.72 else "start"
text_x = x - 14 if anchor == "end" else x + 14
display_label = ("当前 workflow" if is_current else "0.3.9 基线") if lang == "zh" else label
parts.append(f'')
parts.append(
f'{html.escape(display_label)}'
)
parts.append(
f'{score:.2f} / ${cost:.2f}'
)
parts.append(
f'{"总 tokens" if lang == "zh" else "Total tokens"}'
)
y_axis = top + plot_h / 2
parts.append(
f'{"Rubric 分数" if lang == "zh" else "Rubric score"}'
)
parts.append("")
return "\n".join(parts)
def _task_outcomes_svg(tasks: list[tuple[str, bool | None, bool | None]], *, lang: str) -> str:
width, height = 700, max(260, 116 + len(tasks) * 43)
left, top, cell_w, cell_h, gap = 300, 108, 142, 34, 9
title = "任务结果矩阵" if lang == "zh" else "Task outcome matrix"
subtitle = "绿色表示通过；红色表示失败" if lang == "zh" else "Green means pass; red means fail"
parts = _svg_header(width, height, title, subtitle)
column_labels = (
("当前 workflow", "0.3.9 基线") if lang == "zh" else ("Current workflow", "0.3.9 baseline")
)
for i, label in enumerate(column_labels):
x = left + i * (cell_w + 28) + cell_w / 2
parts.append(f'{label}')
for row, (task, current, baseline) in enumerate(tasks):
y = top + row * (cell_h + gap)
parts.append(f'{html.escape(task)}')
for col, passed in enumerate((current, baseline)):
x = left + col * (cell_w + 28)
color = "#9B9488" if passed is None else GOOD if passed else BAD
if passed is None:
status = "N/A"
elif lang == "zh":
status = "通过" if passed else "失败"
else:
status = "PASS" if passed else "FAIL"
parts.append(
f''
)
parts.append(
f'{status}'
)
parts.append("")
return "\n".join(parts)
def _svg_header(width: int, height: int, title: str, subtitle: str) -> list[str]:
return [
f'