158 lines
7.0 KiB
Python
158 lines
7.0 KiB
Python
"""Text/markdown renderers + unicode bar & radar charts (zero dependencies)."""
|
|
|
|
import math
|
|
from typing import Dict, List, Union
|
|
|
|
from ...eval.record import EvalReport
|
|
from .. import register_renderer
|
|
|
|
|
|
def _bars(value: float, width: int = 30, char: str = '█') -> str:
|
|
filled = int(round(max(0.0, min(1.0, value)) * width))
|
|
return char * filled + '·' * (width - filled)
|
|
|
|
|
|
def _pct(value: float) -> str:
|
|
return f'{value * 100:.1f}%'
|
|
|
|
|
|
@register_renderer('text')
|
|
def text_table(target: Union[EvalReport, List[EvalReport]], opts: Dict) -> str:
|
|
reports = target if isinstance(target, list) else [target]
|
|
out: List[str] = []
|
|
for rep in reports:
|
|
head = f'{rep.dataset} [{rep.recipe}] model={rep.model or "?"} n={rep.num_samples}'
|
|
out.append('=' * max(len(head), 40))
|
|
out.append(head)
|
|
out.append('=' * max(len(head), 40))
|
|
# run stats: duration, tokens, cost (from run_info + usage aggregates)
|
|
info = rep.metric_groups.get('run_info', {}) or {}
|
|
total_tokens = info.get('gen_total_tokens', 0)
|
|
tok_s = ''
|
|
if total_tokens:
|
|
tok_s = f' tokens={total_tokens}'
|
|
secs = 0.0
|
|
for s in rep.samples:
|
|
secs += float((s.usage or {}).get('latency_s', 0) or 0)
|
|
dur = f' time={secs / 3600:.2f}h' if secs >= 3600 else (f' time={secs:.0f}s' if secs else '')
|
|
if dur or tok_s:
|
|
out.append(f'n_samples={rep.num_samples}{dur}{tok_s}')
|
|
if rep.num_failed_extractions:
|
|
warn = (f'!! {rep.num_failed_extractions}/{rep.num_samples} extractions failed '
|
|
f'({_pct(rep.metrics.get("extraction_failure_rate", 0))}) -- check recipe/model fit')
|
|
out.append(warn)
|
|
for metric, value in rep.metrics.items():
|
|
if metric == 'extraction_failure_rate':
|
|
continue
|
|
out.append(f'{metric:<16} {_pct(value):>7} {_bars(value)}')
|
|
for group_name, groups in rep.metric_groups.items():
|
|
if group_name == 'run_info' or group_name.startswith('agg_error'):
|
|
continue
|
|
out.append(f'-- {group_name} ' + '-' * max(0, 30 - len(group_name)))
|
|
for g, v in groups.items():
|
|
if isinstance(v, (int, float)):
|
|
out.append(f' {g:<28} {_pct(v):>7} {_bars(v, 20)}')
|
|
out.append('')
|
|
return '\n'.join(out)
|
|
|
|
|
|
@register_renderer('md')
|
|
def markdown(target: Union[EvalReport, List[EvalReport]], opts: Dict) -> str:
|
|
reports = target if isinstance(target, list) else [target]
|
|
out = ['# Eval Report', '']
|
|
for rep in reports:
|
|
out += [f'## {rep.dataset} (`{rep.recipe}`)', '',
|
|
f'- model: `{rep.model or "?"}` samples: {rep.num_samples} '
|
|
f'created: {rep.created_at}', '']
|
|
rows = ['| metric | value |', '|---|---|']
|
|
for metric, value in rep.metrics.items():
|
|
rows.append(f'| {metric} | {_pct(value) if metric != "extraction_failure_rate" else _pct(value)} |')
|
|
out += rows + ['']
|
|
for gname, groups in rep.metric_groups.items():
|
|
if gname in ('run_info',) or gname.startswith('agg_error') or not isinstance(groups, dict):
|
|
continue
|
|
out += [f'### {gname}', '', '| group | value |', '|---|---|']
|
|
out += [f'| {g} | {_pct(v) if isinstance(v, (int, float)) else v} |' for g, v in groups.items()]
|
|
out.append('')
|
|
return '\n'.join(out)
|
|
|
|
|
|
@register_renderer('md_compare')
|
|
def md_compare(target: List[EvalReport], opts: Dict) -> str:
|
|
"""Side-by-side metric table for N reports (e.g. two models on one bench)."""
|
|
if not isinstance(target, list) or len(target) < 1:
|
|
raise ValueError('md_compare needs a list of reports')
|
|
metrics: List[str] = []
|
|
for rep in target:
|
|
for m in rep.metrics:
|
|
if m not in metrics and m != 'extraction_failure_rate':
|
|
metrics.append(m)
|
|
cols = [f'{rep.dataset}/{rep.recipe}[{rep.model or "?"}]' for rep in target]
|
|
out = ['# Comparison', '', '| metric | ' + ' | '.join(cols) + ' |',
|
|
'|---' * (len(cols) + 1) + '|']
|
|
for m in metrics:
|
|
cells = [_pct(rep.metrics.get(m, 0.0)) for rep in target]
|
|
best = max(rep.metrics.get(m, 0.0) for rep in target)
|
|
cells = [f'**{c}**' if rep.metrics.get(m, 0.0) == best and len(target) > 1 else c
|
|
for c, rep in zip(cells, target)]
|
|
out.append(f'| {m} | ' + ' | '.join(cells) + ' |')
|
|
out.append('')
|
|
return '\n'.join(out)
|
|
|
|
|
|
@register_renderer('radar')
|
|
def radar(target: Union[EvalReport, List[EvalReport]], opts: Dict) -> str:
|
|
"""Unicode radar chart over each report's metric_groups entries.
|
|
|
|
opts: group (default: first non-run_info group), top (default 10 axes).
|
|
"""
|
|
reports = target if isinstance(target, list) else [target]
|
|
group = opts.get('group')
|
|
axes: List[str] = []
|
|
series: List[Dict[str, float]] = []
|
|
for rep in reports:
|
|
gname = group or next((k for k in rep.metric_groups
|
|
if k != 'run_info' and not k.startswith('agg_error')
|
|
and isinstance(rep.metric_groups[k], dict)), None)
|
|
data = rep.metric_groups.get(gname, {}) if gname else {}
|
|
data = {k: v for k, v in data.items() if isinstance(v, (int, float))}
|
|
if not data:
|
|
return f'(no grouped metrics to chart for {rep.dataset})'
|
|
top = opts.get('top', 10)
|
|
picked = sorted(data.items(), key=lambda kv: -kv[1])[:top]
|
|
axes = [k for k, _ in picked]
|
|
series.append(dict(picked))
|
|
|
|
# simple ascii radar: axis list + per-report bars side by side
|
|
W = 24
|
|
header = 'axis'.ljust(26) + ''.join((rep.dataset[:12] or '?').rjust(14) for rep in reports)
|
|
lines = [header, '-' * len(header)]
|
|
for ax in axes:
|
|
row = ax[:24].ljust(26)
|
|
for rep, s in zip(reports, series):
|
|
v = s.get(ax, 0.0)
|
|
row += (_bars(v, W // 2)[:W // 2] + f'{v * 100:5.1f}%').rjust(14)
|
|
lines.append(row)
|
|
return '\n'.join(lines)
|
|
|
|
|
|
@register_renderer('errors')
|
|
def errors(target: Union[EvalReport, List[EvalReport]], opts: Dict) -> str:
|
|
"""Failed samples browser: worst-N samples with extraction + target."""
|
|
rep = target[0] if isinstance(target, list) else target
|
|
n = opts.get('n', 10)
|
|
only_failed = opts.get('only_failed', True)
|
|
rows = [r for r in rep.samples if r.error or (not r.extraction_ok if only_failed else False)]
|
|
rows = sorted(rows, key=lambda r: sum(r.scores.values()))[:n]
|
|
out = [f'# Errors / failed extractions: {rep.dataset} ({len(rows)} shown)', '']
|
|
for r in rows:
|
|
out.append(f'## sample {r.sample_id} scores={r.scores}')
|
|
out.append(f'- extraction_ok={r.extraction_ok} note={r.extraction_note!r}')
|
|
out.append(f'- target: {str(r.target)[:120]!r}')
|
|
out.append(f'- extracted: {r.extracted_prediction[:120]!r}')
|
|
if r.error:
|
|
out.append(f'- error: {r.error[:300]}')
|
|
out.append(f'- raw: {r.raw_prediction[:200]!r}')
|
|
out.append('')
|
|
return '\n'.join(out)
|