lda.chat/lda-wf
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

docs: publish thesis evaluation bundle

Browse Source
This commit is contained in:
lda
2026-06-30 22:46:17 +07:00 Verified
parent 5e9b76eb94
commit e12ccdf18a
28 changed files with 11555 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files
examples/agent_challenges/evaluation.py
+293
View File
@@ -0,0 +1,293 @@
from __future__ import annotations
import hashlib
import json
from collections import Counter
from dataclasses import dataclass
from pathlib import Path
from typing import Any
from .names import short_challenge_name, short_model_name
FIGURE_STEMS = (
"agent-challenge-audited-outcomes-by-cell",
"agent-challenge-automatic-vs-manual-outcomes",
"agent-challenge-longitudinal-outcomes",
"agent-challenge-duration-and-tokens",
)
_MANUAL_OUTCOMES = frozenset({"pass", "invalid", "fail"})
_CHALLENGE_ORDER = {"browser": 0, "report": 1}
_MODEL_ORDER = {"deepseek": 0, "mimo": 1}
_PROFILE_ORDER = {"none": 0, "skills": 1, "all": 2}
@dataclass(frozen=True, slots=True)
class EvaluationTrial:
"""One manually audited trial selected into an evaluation cohort."""
report_path: Path
wave: int
challenge: str
model: str
profile: str
trial_index: int
repository_commit: str
base_prompt_hash: str
manual_outcome: str
task_outcome: str
duration_seconds: float
tokens_total: int
audit_notes: str
@dataclass(frozen=True, slots=True)
class EvaluationCohort:
"""Immutable cohort metadata and its validated report projections."""
cohort_id: str
title: str
selection_rule: str
limitations: tuple[str, ...]
trials: tuple[EvaluationTrial, ...]
def _object(value: object, *, field: str) -> dict[str, Any]:
if not isinstance(value, dict):
raise ValueError(f"{field} must be an object")
return value
def _string(value: object, *, field: str) -> str:
if not isinstance(value, str) or not value:
raise ValueError(f"{field} must be a non-empty string")
return value
def _integer(value: object, *, field: str) -> int:
if not isinstance(value, int):
raise ValueError(f"{field} must be an integer")
return value
def _report_path(repository_root: Path, value: object) -> Path:
relative = Path(_string(value, field="runs[].report"))
if relative.is_absolute():
raise ValueError("runs[].report must be relative to the repository root")
resolved = (repository_root / relative).resolve()
if not resolved.is_relative_to(repository_root.resolve()):
raise ValueError(f"report path escapes repository root: {relative}")
return resolved
def _report_sha256(value: object, *, report_path: Path) -> str:
"""Validate snapshot provenance and, when available, its local report."""
expected = _string(value, field="runs[].report_sha256")
if len(expected) != 64 or any(
character not in "0123456789abcdef" for character in expected
):
raise ValueError(f"invalid report SHA-256 for {report_path}")
if report_path.is_file():
actual = hashlib.sha256(report_path.read_bytes()).hexdigest()
if actual != expected:
raise ValueError(
f"local report does not match cohort snapshot: {report_path}"
)
return expected
def _load_trial(
run: dict[str, Any], report_path: Path, *, wave: int
) -> EvaluationTrial:
_report_sha256(run.get("report_sha256"), report_path=report_path)
manual_outcome = _string(run.get("manual_outcome"), field="runs[].manual_outcome")
if manual_outcome not in _MANUAL_OUTCOMES:
raise ValueError(
f"unsupported manual outcome {manual_outcome!r}: {report_path}"
)
duration = run.get("duration_seconds")
if not isinstance(duration, int | float):
raise ValueError(f"runs[].duration_seconds must be numeric: {report_path}")
return EvaluationTrial(
report_path=report_path,
wave=wave,
challenge=short_challenge_name(
_string(run.get("challenge"), field="runs[].challenge")
),
model=short_model_name(_string(run.get("model"), field="runs[].model")),
profile=_string(run.get("profile"), field="runs[].profile"),
trial_index=_integer(run.get("trial_index"), field="runs[].trial_index"),
repository_commit=_string(
run.get("repository_commit"), field="runs[].repository_commit"
),
base_prompt_hash=_string(
run.get("base_prompt_hash"), field="runs[].base_prompt_hash"
),
manual_outcome=manual_outcome,
task_outcome=_string(run.get("task_outcome"), field="runs[].task_outcome"),
duration_seconds=float(duration),
tokens_total=_integer(run.get("tokens_total"), field="runs[].tokens_total"),
audit_notes=str(run.get("audit_notes") or ""),
)
def load_evaluation_cohort(
manifest_path: Path, *, repository_root: Path
) -> EvaluationCohort:
"""Load an explicit cohort manifest and validate every report projection."""
manifest = _object(
json.loads(manifest_path.read_text(encoding="utf-8")), field=str(manifest_path)
)
if manifest.get("schema_version") != 1:
raise ValueError("agent challenge cohort schema_version must be 1")
raw_runs = manifest.get("runs")
if not isinstance(raw_runs, list) or not raw_runs:
raise ValueError("agent challenge cohort runs must be a non-empty list")
trials: list[EvaluationTrial] = []
seen: set[Path] = set()
for index, raw_run in enumerate(raw_runs):
run = _object(raw_run, field=f"runs[{index}]")
wave = _integer(run.get("wave"), field=f"runs[{index}].wave")
if wave < 1:
raise ValueError(f"runs[{index}].wave must be positive")
report_path = _report_path(repository_root, run.get("report"))
if report_path in seen:
raise ValueError(f"duplicate report in cohort: {report_path}")
seen.add(report_path)
trials.append(_load_trial(run, report_path, wave=wave))
limitations = manifest.get("limitations")
if not isinstance(limitations, list) or not all(
isinstance(item, str) and item for item in limitations
):
raise ValueError("limitations must be a list of non-empty strings")
return EvaluationCohort(
cohort_id=_string(manifest.get("cohort_id"), field="cohort_id"),
title=_string(manifest.get("title"), field="title"),
selection_rule=_string(manifest.get("selection_rule"), field="selection_rule"),
limitations=tuple(limitations),
trials=tuple(trials),
)
def render_evaluation_figures(
cohort: EvaluationCohort, output_dir: Path
) -> tuple[Path, ...]:
"""Render the cohort through the optional Matplotlib figure layer."""
# Keeping plotting imports out of this data module lets summary tooling run
# in minimal environments that do not install thesis build dependencies.
from .evaluation_figures import render_evaluation_figures as render
return render(cohort, output_dir)
def _cell_rows(cohort: EvaluationCohort) -> list[tuple[str, Counter[str]]]:
grouped: dict[tuple[str, str, str], Counter[str]] = {}
for trial in cohort.trials:
key = (trial.challenge, trial.model, trial.profile)
grouped.setdefault(key, Counter())[trial.manual_outcome] += 1
keys = sorted(
grouped,
key=lambda key: (
_CHALLENGE_ORDER.get(key[0], 99),
_MODEL_ORDER.get(key[1], 99),
_PROFILE_ORDER.get(key[2], 99),
),
)
return [
(f"{challenge} / {model} / {profile}", grouped[(challenge, model, profile)])
for challenge, model, profile in keys
]
def render_evaluation_markdown(cohort: EvaluationCohort) -> str:
"""Render the checked cohort as a compact, auditable Markdown appendix."""
outcomes = Counter(trial.manual_outcome for trial in cohort.trials)
lines = [
"## Audited Agent Challenge Campaign",
"",
(
f"The primary campaign contains {len(cohort.trials)} audited trials: "
f"{outcomes['pass']} passes, {outcomes['invalid']} invalid samples, "
f"and {outcomes['fail']} failure."
),
"",
(
"The campaign crosses two challenges, two hosted models, three instruction "
"profiles (`none`, `skills`, and `all`), and three repetitions per cell. "
"The checked cohort snapshot records report hashes, prompt hashes, the "
"repository commit, automatic metrics, and manual-audit outcomes; local "
"raw report files are verified against those hashes when present."
),
"",
(
"Because repository snapshots and one prompt rule changed between waves, "
"this is longitudinal engineering evidence, not a controlled model comparison."
),
"",
f"Selection rule: {cohort.selection_rule}",
"",
"| Challenge / model / profile | Pass | Invalid | Fail |",
"| --- | ---: | ---: | ---: |",
]
for label, counts in _cell_rows(cohort):
lines.append(
f"| {label} | {counts['pass']} | {counts['invalid']} | {counts['fail']} |"
)
lines.extend(
[
"",
": Audited outcomes by challenge, model, and instruction profile. {#tbl:agent-challenge-outcomes}",
"",
(
"A manual `pass` requires both successful product-path evidence and an "
"acceptable audit trail. `Invalid` means the sample cannot support the "
"clean benchmark claim, commonly because the agent read repository or "
"example material outside its supplied workspace. `Fail` means the "
"challenge contract itself was not established."
),
"",
"![Audited outcomes by evaluation cell.](figures/agent-challenge-audited-outcomes-by-cell.svg){#fig:agent-challenge-audited-outcomes-by-cell width=95%}",
"",
(
"[@fig:agent-challenge-audited-outcomes-by-cell] reports all three "
"repetitions rather than hiding invalid samples. The profile labels are "
"descriptive; this campaign does not isolate instruction-profile effects."
),
"",
"![Automatic task outcomes compared with manual outcomes.](figures/agent-challenge-automatic-vs-manual-outcomes.svg){#fig:agent-challenge-automatic-vs-manual-outcomes width=75%}",
"",
(
"[@fig:agent-challenge-automatic-vs-manual-outcomes] shows why the "
"manual layer matters. Seven automatically successful trials were invalid "
"as clean evidence, while three automatically failed reports were accepted "
"after their saved run evidence and report artifacts were manually audited."
),
"",
"![Audited outcomes across the three longitudinal waves.](figures/agent-challenge-longitudinal-outcomes.svg){#fig:agent-challenge-longitudinal-outcomes width=75%}",
"",
(
"The waves in [@fig:agent-challenge-longitudinal-outcomes] are not "
"an improvement curve: product commits, prompt wording, and enforcement "
"changed. They preserve the chronology needed to study those changes."
),
"",
"![Duration and recorded token totals grouped by challenge, instruction profile, model, and wave.](figures/agent-challenge-duration-and-tokens.svg){#fig:agent-challenge-duration-and-tokens width=95%}",
"",
(
"[@fig:agent-challenge-duration-and-tokens] separates each challenge and "
"metric into its own panel. Circle and square markers redundantly identify "
"the models without relying on color. Wall-clock duration includes hosted-service "
"latency, and OpenCode token totals include cache-read accounting, so neither "
"axis is a normalized model-efficiency metric."
),
"",
"### Campaign Limitations",
"",
]
)
lines.extend(f"- {limitation}" for limitation in cohort.limitations)
return "\n".join(lines) + "\n"
examples/agent_challenges/evaluation_figures.py
+386
View File
@@ -0,0 +1,386 @@
from __future__ import annotations
from collections import Counter
from pathlib import Path
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from matplotlib.axes import Axes
from matplotlib.figure import Figure
from .evaluation import EvaluationCohort, EvaluationTrial
_OUTCOMES = ("pass", "invalid", "fail")
_OUTCOME_COLORS = {
"pass": "#00796B",
"invalid": "#E07A1F",
"fail": "#9E2A2B",
}
_OUTCOME_HATCHES = {"pass": "", "invalid": "///", "fail": "xx"}
_MODEL_STYLES = {
"deepseek": {"color": "#0067A5", "marker": "o", "offset": -0.13},
"mimo": {"color": "#D55E00", "marker": "s", "offset": 0.13},
}
_PROFILE_ORDER = {"none": 0, "skills": 1, "all": 2}
_CHALLENGE_ORDER = {"browser": 0, "report": 1}
_MODEL_ORDER = {"deepseek": 0, "mimo": 1}
def _configure_matplotlib() -> Any:
"""Import Matplotlib with a headless backend suitable for docs builds."""
import matplotlib
matplotlib.use("Agg")
from matplotlib import pyplot as plt
plt.rcParams.update(
{
"font.family": "DejaVu Sans",
"font.size": 9,
"svg.hashsalt": "lda-chat-agent-challenge-evaluation",
"axes.titleweight": "bold",
"axes.titlesize": 12,
"axes.labelcolor": "#28323c",
"axes.edgecolor": "#8b949e",
"axes.spines.top": False,
"axes.spines.right": False,
"figure.facecolor": "white",
"axes.facecolor": "#F3F6F7",
"grid.color": "#C9D1D5",
"grid.linewidth": 0.6,
}
)
return plt
def _trial_sort_key(trial: EvaluationTrial) -> tuple[int, int, int, int]:
return (
_CHALLENGE_ORDER.get(trial.challenge, 99),
_MODEL_ORDER.get(trial.model, 99),
_PROFILE_ORDER.get(trial.profile, 99),
trial.wave,
)
def _cell_key(trial: EvaluationTrial) -> tuple[str, str, str]:
return trial.challenge, trial.model, trial.profile
def _ordered_cells(cohort: EvaluationCohort) -> list[tuple[str, str, str]]:
return sorted(
{_cell_key(trial) for trial in cohort.trials},
key=lambda cell: (
_CHALLENGE_ORDER.get(cell[0], 99),
_MODEL_ORDER.get(cell[1], 99),
_PROFILE_ORDER.get(cell[2], 99),
),
)
def _save_figure(
figure: Figure, output_dir: Path, stem: str, plt: Any
) -> tuple[Path, Path]:
"""Save one source figure in web-native SVG and print-native PDF."""
output_dir.mkdir(parents=True, exist_ok=True)
written: list[Path] = []
for suffix in ("svg", "pdf"):
path = output_dir / f"{stem}.{suffix}"
# Matplotlib otherwise embeds current timestamps, and SVG element IDs
# use a random salt. Stable metadata makes checked-in figures reproducible.
metadata = (
{"Creator": "lda.chat thesis evaluation", "Date": None}
if suffix == "svg"
else {
"Creator": "lda.chat thesis evaluation",
"CreationDate": None,
"ModDate": None,
}
)
figure.savefig(
path,
bbox_inches="tight",
pad_inches=0.12,
metadata=metadata,
)
if suffix == "svg":
# Matplotlib writes SVG path data across lines with trailing spaces.
# Normalize it so generated figures can pass git whitespace checks.
lines = path.read_text(encoding="utf-8").splitlines()
path.write_text(
"\n".join(line.rstrip() for line in lines) + "\n",
encoding="utf-8",
newline="\n",
)
written.append(path)
plt.close(figure)
return written[0], written[1]
def _outcomes_by_cell(cohort: EvaluationCohort, plt: Any) -> Figure:
cells = _ordered_cells(cohort)
counts = {
cell: Counter(
trial.manual_outcome for trial in cohort.trials if _cell_key(trial) == cell
)
for cell in cells
}
labels = [
f"{challenge} | {model} | {profile}" for challenge, model, profile in cells
]
figure, axis = plt.subplots(figsize=(9.2, 6.4))
left = [0] * len(cells)
for outcome in _OUTCOMES:
values = [counts[cell][outcome] for cell in cells]
bars = axis.barh(
range(len(cells)),
values,
left=left,
label=outcome.capitalize(),
color=_OUTCOME_COLORS[outcome],
edgecolor="#263238",
linewidth=0.55,
hatch=_OUTCOME_HATCHES[outcome],
height=0.68,
)
for bar, value in zip(bars, values, strict=True):
if value:
axis.text(
bar.get_x() + bar.get_width() / 2,
bar.get_y() + bar.get_height() / 2,
str(value),
ha="center",
va="center",
color="#17212B" if outcome == "invalid" else "white",
fontweight="bold",
)
left = [current + value for current, value in zip(left, values, strict=True)]
axis.set_yticks(range(len(cells)), labels)
axis.invert_yaxis()
axis.set_xticks((0, 1, 2, 3))
axis.set_xlim(0, 3)
axis.set_xlabel("Manually audited trials (n=3 per cell)")
axis.set_title("Audited outcomes by challenge, model, and instruction profile")
axis.grid(axis="x")
axis.legend(loc="lower right", frameon=False, ncol=3)
figure.tight_layout()
return figure
def _automatic_vs_manual(cohort: EvaluationCohort, plt: Any) -> Figure:
from matplotlib.colors import LinearSegmentedColormap
task_labels = ("success", "failed")
manual_labels = ("pass", "invalid", "fail")
matrix = [
[
sum(
trial.task_outcome == task and trial.manual_outcome == manual
for trial in cohort.trials
)
for manual in manual_labels
]
for task in task_labels
]
figure, axis = plt.subplots(figsize=(6.8, 3.8))
count_cmap = LinearSegmentedColormap.from_list(
"lda_count", ("#F3F6F7", "#79B8B3", "#005F73")
)
maximum = max(map(max, matrix))
image = axis.imshow(matrix, cmap=count_cmap, vmin=0, vmax=maximum)
for row, values in enumerate(matrix):
for column, value in enumerate(values):
axis.text(
column,
row,
str(value),
ha="center",
va="center",
color="white" if value > maximum / 2 else "#17212B",
fontsize=13,
fontweight="bold",
)
axis.set_xticks(
range(len(manual_labels)), [label.capitalize() for label in manual_labels]
)
axis.set_yticks(
range(len(task_labels)), [label.capitalize() for label in task_labels]
)
axis.set_xlabel("Manual official outcome")
axis.set_ylabel("Automatic task outcome")
axis.set_title("Automatic completion does not imply clean evaluation evidence")
axis.set_xticks([value - 0.5 for value in range(1, len(manual_labels))], minor=True)
axis.set_yticks([0.5], minor=True)
axis.grid(which="minor", color="white", linewidth=2)
axis.tick_params(which="minor", bottom=False, left=False)
figure.colorbar(image, ax=axis, label="Trial count", shrink=0.82)
figure.tight_layout()
return figure
def _longitudinal_outcomes(cohort: EvaluationCohort, plt: Any) -> Figure:
waves = sorted({trial.wave for trial in cohort.trials})
counts = {
wave: Counter(
trial.manual_outcome for trial in cohort.trials if trial.wave == wave
)
for wave in waves
}
figure, axis = plt.subplots(figsize=(6.8, 4.0))
bottom = [0] * len(waves)
for outcome in _OUTCOMES:
values = [counts[wave][outcome] for wave in waves]
bars = axis.bar(
waves,
values,
bottom=bottom,
label=outcome.capitalize(),
color=_OUTCOME_COLORS[outcome],
edgecolor="#263238",
linewidth=0.55,
hatch=_OUTCOME_HATCHES[outcome],
width=0.62,
)
for bar, value in zip(bars, values, strict=True):
if value:
axis.text(
bar.get_x() + bar.get_width() / 2,
bar.get_y() + bar.get_height() / 2,
str(value),
ha="center",
va="center",
color="#17212B" if outcome == "invalid" else "white",
fontweight="bold",
)
bottom = [
current + value for current, value in zip(bottom, values, strict=True)
]
axis.set_xticks(waves, [f"Wave {wave}" for wave in waves])
axis.set_ylim(0, max(bottom) + 1)
axis.set_ylabel("Manually audited trials")
axis.set_title("Outcomes across three evolving product and prompt snapshots")
axis.grid(axis="y")
axis.legend(frameon=False, ncol=3, loc="upper center")
figure.tight_layout()
return figure
def _scatter_metric(
axis: Axes,
trials: list[EvaluationTrial],
*,
metric: str,
) -> None:
profiles = ("none", "skills", "all")
wave_offsets = {1: -0.055, 2: 0.0, 3: 0.055}
for trial in sorted(trials, key=_trial_sort_key):
style = _MODEL_STYLES[trial.model]
x = (
profiles.index(trial.profile)
+ float(style["offset"])
+ wave_offsets.get(trial.wave, 0.0)
)
if metric == "duration":
value = trial.duration_seconds / 60
else:
value = trial.tokens_total / 1_000_000
axis.scatter(
x,
value,
color=str(style["color"]),
marker=str(style["marker"]),
edgecolor="#17212B",
linewidth=0.65,
s=86,
zorder=3,
)
axis.annotate(
str(trial.wave),
(x, value),
ha="center",
va="center",
color="white",
fontsize=6,
fontweight="bold",
zorder=4,
)
axis.set_xticks(
range(len(profiles)), [profile.capitalize() for profile in profiles]
)
axis.set_xlim(-0.42, 2.42)
axis.grid(axis="y")
def _duration_and_tokens(cohort: EvaluationCohort, plt: Any) -> Figure:
from matplotlib.lines import Line2D
figure, axes = plt.subplots(2, 2, figsize=(9.6, 6.8), sharex=True)
challenges = (("browser", "Browser click"), ("report", "Report workflow"))
for row, (challenge, challenge_label) in enumerate(challenges):
trials = [trial for trial in cohort.trials if trial.challenge == challenge]
duration_axis, token_axis = axes[row]
_scatter_metric(duration_axis, trials, metric="duration")
_scatter_metric(token_axis, trials, metric="tokens")
duration_axis.set_ylabel(f"{challenge_label}\nMinutes")
token_axis.set_ylabel(f"{challenge_label}\nMillion tokens")
axes[0, 0].set_title("Wall-clock duration")
axes[0, 1].set_title("Recorded token volume")
axes[1, 0].set_xlabel("Instruction profile")
axes[1, 1].set_xlabel("Instruction profile")
legend_handles = [
Line2D(
[],
[],
color=str(style["color"]),
marker=str(style["marker"]),
linestyle="None",
markeredgecolor="#17212B",
markersize=8,
label=f"{model.capitalize()} ({'circle' if model == 'deepseek' else 'square'})",
)
for model, style in _MODEL_STYLES.items()
]
figure.suptitle(
"Runtime evidence by challenge, profile, model, and wave",
fontsize=13,
fontweight="bold",
)
figure.legend(
handles=legend_handles,
loc="upper center",
bbox_to_anchor=(0.5, 0.94),
frameon=False,
ncol=2,
)
figure.text(
0.5,
0.015,
"Point labels 1–3 identify waves; token totals include OpenCode cache-read accounting.",
ha="center",
color="#4C5961",
fontsize=8,
)
figure.tight_layout(rect=(0.0, 0.05, 1.0, 0.88))
return figure
def render_evaluation_figures(
cohort: EvaluationCohort, output_dir: Path
) -> tuple[Path, ...]:
"""Write all named evaluation figures as SVG and PDF pairs."""
plt = _configure_matplotlib()
figures = (
("agent-challenge-audited-outcomes-by-cell", _outcomes_by_cell(cohort, plt)),
(
"agent-challenge-automatic-vs-manual-outcomes",
_automatic_vs_manual(cohort, plt),
),
("agent-challenge-longitudinal-outcomes", _longitudinal_outcomes(cohort, plt)),
("agent-challenge-duration-and-tokens", _duration_and_tokens(cohort, plt)),
)
written: list[Path] = []
for stem, figure in figures:
written.extend(_save_figure(figure, output_dir, stem, plt))
return tuple(written)