evaluation

agentic_ml.core.evaluation

Generic trial evaluation.

This module has two roles:

• Parent side (:func:evaluate_trial): serialize an evaluation request, run the child in an isolated subprocess via :mod:agentic_ml.core.sandbox, and parse the result.
• Child side (python -m agentic_ml.core.evaluation <in.json> <out.json>): the actual, task-agnostic scoring loop that imports the trial's Model, runs cross-validated fit/predict over the given partitions, computes metrics and checks the save/load roundtrip acceptance gate.

The parts that are task-specific (which metrics, how partitions are built, the roundtrip tolerance) are provided by the caller, so the runner itself is reused by every task.

EvaluationResult dataclass

EvaluationResult(
    status: str,
    metrics: dict[str, float | None],
    runtime: float | None,
    roundtrip_ok: bool,
    error: str | None = None,
    per_fold: list[dict[str, float]] = list(),
)

Aggregated outcome of evaluating a single trial.

evaluate_trial

evaluate_trial(
    trial_dir: str | Path,
    dataset_path: str | Path,
    metrics: list[str],
    partitions: list[Partition],
    tolerance: float,
    timeout: float,
) -> EvaluationResult

Evaluate the trial in trial_dir and return an :class:EvaluationResult.

Source code in src/agentic_ml/core/evaluation.py

def evaluate_trial(
    trial_dir: str | Path,
    dataset_path: str | Path,
    metrics: list[str],
    partitions: list[Partition],
    tolerance: float,
    timeout: float,
) -> EvaluationResult:
    """Evaluate the trial in ``trial_dir`` and return an :class:`EvaluationResult`."""
    failed_metrics: dict[str, float | None] = {name: None for name in metrics}
    with tempfile.TemporaryDirectory() as tmp:
        tmp_path = Path(tmp)
        in_path = tmp_path / "in.json"
        out_path = tmp_path / "out.json"
        # The child runs with tmp_path as cwd, so paths must be absolute to resolve correctly.
        payload = {
            "trial_dir": str(Path(trial_dir).resolve()),
            "dataset_path": str(Path(dataset_path).resolve()),
            "metrics": list(metrics),
            "partitions": [[list(map(int, tr)), list(map(int, va))] for tr, va in partitions],
            "tolerance": tolerance,
        }
        in_path.write_text(json.dumps(payload))

        result = sandbox.run(
            [sys.executable, "-m", "agentic_ml.core.evaluation", str(in_path), str(out_path)],
            cwd=tmp_path,
            timeout=timeout,
        )

        if result.timed_out:
            return EvaluationResult(
                "failed", failed_metrics, None, False, error="evaluation timed out"
            )
        if not out_path.exists():
            detail = result.stderr.strip()[-2000:] or "no output produced by evaluation subprocess"
            return EvaluationResult("failed", failed_metrics, None, False, error=detail)

        data = json.loads(out_path.read_text())
        return EvaluationResult(
            status=data["status"],
            metrics=data.get("metrics", failed_metrics),
            runtime=data.get("runtime"),
            roundtrip_ok=data.get("roundtrip_ok", False),
            error=data.get("error"),
            per_fold=data.get("per_fold", []),
        )