Spaces:
Sleeping
Sleeping
File size: 3,348 Bytes
be9f416 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 | """Benchmark evaluation helpers for Ontology Architect agents and baselines."""
from __future__ import annotations
from dataclasses import replace
import json
from pathlib import Path
from statistics import mean
try:
from .baselines import get_baseline
from .config import ExperimentConfig
from .models import OntologyArchitectAction
from .server.ontology_architect_environment import OntologyArchitectEnvironment
except ImportError: # pragma: no cover
from baselines import get_baseline
from config import ExperimentConfig
from models import OntologyArchitectAction
from server.ontology_architect_environment import OntologyArchitectEnvironment
def evaluate_baselines(
config: ExperimentConfig,
baseline_names: list[str] | None = None,
seeds: list[int] | None = None,
) -> dict:
baseline_names = baseline_names or ["static", "linear", "teacher"]
seeds = seeds or [config.universe.seed, config.universe.seed + 1]
report = {
"baselines": {},
"seeds": seeds,
"universe_family": config.universe.family,
"split": config.universe.split,
}
for baseline in baseline_names:
module = get_baseline(baseline)
runs = []
for seed in seeds:
run_config = replace(config, universe=replace(config.universe, seed=seed))
runs.append(_run_module(run_config, module, baseline))
rewards = [run["total_reward"] for run in runs]
log_likelihoods = [
metric["log_likelihood"]
for run in runs
for metric in run["step_metrics"]
if metric.get("execution_ok")
]
report["baselines"][baseline] = {
"runs": runs,
"mean_total_reward": mean(rewards) if rewards else 0.0,
"mean_log_likelihood": mean(log_likelihoods) if log_likelihoods else float("-inf"),
"execution_failure_rate": _failure_rate(runs),
"discovery_score": mean(rewards) if rewards else 0.0,
}
return report
def write_report(report: dict, output_path: str | Path) -> None:
output = Path(output_path)
output.parent.mkdir(parents=True, exist_ok=True)
with output.open("w", encoding="utf-8") as handle:
json.dump(report, handle, indent=2, sort_keys=True)
def _run_module(config: ExperimentConfig, module: str, label: str) -> dict:
env = OntologyArchitectEnvironment(config)
observation = env.reset()
total_reward = 0.0
step_metrics = []
for step in range(config.universe.max_steps):
observation = env.step(
OntologyArchitectAction(
theory_module=module,
revision_note=f"benchmark baseline={label} step={step}",
)
)
total_reward += float(observation.reward or 0.0)
step_metrics.append(observation.metadata.get("last_metrics", {}))
if observation.done:
break
return {
"steps": len(step_metrics),
"total_reward": total_reward,
"step_metrics": step_metrics,
}
def _failure_rate(runs: list[dict]) -> float:
metrics = [metric for run in runs for metric in run["step_metrics"]]
if not metrics:
return 1.0
failures = [metric for metric in metrics if not metric.get("execution_ok")]
return len(failures) / len(metrics)
|