EvalServer

gepa.optimize_anything.EvalServer(task: Task, evaluate: Callable[..., Any] | None, budget: BudgetTracker, *, batch_evaluate: Callable[..., Any] | None = None, max_concurrency: int = DEFAULT_MAX_CONCURRENCY, output_dir: str | Path | None = None)

Eval server with budget enforcement — the single source of truth.

In-process engines call :meth:evaluate directly. External engines POST to :attr:url. Both go through the same budget counter.

Parameters:

Name	Type	Description	Default
task	Task	Task definition (name, datasets, objective/background).	required
evaluate	Callable[..., Any] | None	Per-pair scoring function. Optional when batch_evaluate is provided (at least one is required); single-pair evaluations then route through it as singleton batches.	required
budget	BudgetTracker	Budget tracker for limiting evaluations.	required
batch_evaluate	Callable[..., Any] | None	Optional grouped scoring function taking a list of (candidate, example) pairs and returning one result per pair. Multi-pair evaluations prefer it over per-pair fan-out.	None
max_concurrency	int	Max parallel evaluations. Defaults to 8.	DEFAULT_MAX_CONCURRENCY
output_dir	str | Path | None	If set, each eval is persisted as <dir>/evals/<i>.json and a rolling summary.json is written.	None

Source code in gepa/oa/eval_server.py

def __init__(
    self,
    task: Task,
    evaluate: Callable[..., Any] | None,
    budget: BudgetTracker,
    *,
    batch_evaluate: Callable[..., Any] | None = None,
    max_concurrency: int = DEFAULT_MAX_CONCURRENCY,
    output_dir: str | Path | None = None,
) -> None:
    self.task = task
    self.batch_fn = _normalize_batch_fn(batch_evaluate) if batch_evaluate is not None else None
    if evaluate is None:
        if self.batch_fn is None:
            raise ValueError("Provide evaluator=, batch_evaluator=, or both.")
        batch_fn = self.batch_fn

        def _singleton(
            candidate: Any, example: Any | None = None, opt_state: Any | None = None
        ) -> tuple[float, dict[str, Any]]:
            # Launcher parity: with only a batch_evaluator, single-pair
            # evaluations route through it as singleton batches.
            return batch_fn([(candidate, example)], opt_states=[opt_state] if opt_state is not None else None)[0]

        evaluate = _singleton

    self.eval_fn: Callable[..., tuple[float, dict[str, Any]]] = _normalize_eval_fn(evaluate)
    self.budget = budget
    self.max_concurrency = max_concurrency
    self.best_score: float = float("-inf")
    # A dict seed is a legal multi-component candidate (see ``_track``,
    # which already reads/stores dict candidates), so the tracked best is
    # ``str | dict``, not ``str``.
    self.best_candidate: str | dict[str, str] = task.seed_candidate or ""
    self.total_cost: float = 0.0
    self.eval_log: list[dict[str, Any]] = []
    self._start_time: float = time.time()
    self._best_val_score: float = float("-inf")
    self._progress_log: list[dict[str, Any]] = []
    self._candidate_registry: dict[str, int] = {}
    self._next_candidate_id: int = 0
    self._lock = threading.Lock()
    self._io_lock = threading.Lock()
    self.output_dir: Path | None = None
    self._evals_dir: Path | None = None
    if output_dir is not None:
        self.output_dir = Path(output_dir)
        self.output_dir.mkdir(parents=True, exist_ok=True)
        self._evals_dir = self.output_dir / "evals"
        self._evals_dir.mkdir(exist_ok=True)
    self._eval_semaphore = threading.Semaphore(max_concurrency)

    # The eval server holds only the agent-visible pool (train + val). The
    # held-out test_set is never registered here: the optimize_anything
    # runner scores it directly via ``eval_fn``, so test data is unreachable
    # from any engine, agent, or HTTP endpoint by construction.
    self._examples: dict[str, Any] = {}
    self._split_ids: dict[str, list[str]] = {"train": [], "val": []}
    for split_name, dataset in (
        ("train", task.train_set),
        ("val", task.val_set),
    ):
        if not dataset:
            continue
        for i, item in enumerate(dataset):
            eid = _resolve_id(item, f"{split_name}_{i}")
            if eid in self._examples and self._examples[eid] is not item:
                eid = f"{split_name}_{i}"
            self._examples[eid] = item
            self._split_ids[split_name].append(eid)

    self._pool = ThreadPoolExecutor(max_workers=max_concurrency)
    self._server: HTTPServer | None = None
    self._thread: threading.Thread | None = None

Attributes

task = task instance-attribute

batch_fn = _normalize_batch_fn(batch_evaluate) if batch_evaluate is not None else None instance-attribute

eval_fn: Callable[..., tuple[float, dict[str, Any]]] = _normalize_eval_fn(evaluate) instance-attribute

budget = budget instance-attribute

max_concurrency = max_concurrency instance-attribute

best_score: float = float('-inf') instance-attribute

best_candidate: str | dict[str, str] = task.seed_candidate or '' instance-attribute

total_cost: float = 0.0 instance-attribute

eval_log: list[dict[str, Any]] = [] instance-attribute

output_dir: Path | None = None instance-attribute

progress_log: list[dict[str, Any]] property

port: int property

url: str property

Methods:

iter_split(split: str) -> Iterator[tuple[str, Any]]

Yield (id, item) pairs for the given split (train/val).

The server holds no test split, so iter_split("test") yields nothing; test is scored by the optimize_anything runner directly.

Source code in gepa/oa/eval_server.py

def iter_split(self, split: str) -> Iterator[tuple[str, Any]]:
    """Yield ``(id, item)`` pairs for the given split (``train``/``val``).

    The server holds no ``test`` split, so ``iter_split("test")`` yields
    nothing; test is scored by the optimize_anything runner directly.
    """
    for eid in self._split_ids.get(split, []):
        yield eid, self._examples[eid]

evaluate(candidate: str, example: Any | None = None, **kwargs: Any) -> tuple[float, dict[str, Any]]

Evaluate a candidate with budget enforcement.

Failed attempts consume budget (launcher parity). Extra kwargs (e.g. opt_state) are forwarded to the evaluator only if its signature accepts them.

Raises:

Type	Description
BudgetExhausted	When the budget has been used up.

Source code in gepa/oa/eval_server.py

def evaluate(self, candidate: str, example: Any | None = None, **kwargs: Any) -> tuple[float, dict[str, Any]]:
    """Evaluate a candidate with budget enforcement.

    Failed attempts consume budget (launcher parity). Extra kwargs (e.g.
    ``opt_state``) are forwarded to the evaluator only if its signature
    accepts them.

    Raises:
        BudgetExhausted: When the budget has been used up.
    """
    self._eval_semaphore.acquire()
    try:
        self.budget.check()

        try:
            if example is not None:
                score, info = self.eval_fn(candidate, example, **kwargs)
            else:
                score, info = self.eval_fn(candidate, **kwargs)
        except Exception:
            # check() passed above, so this record cannot itself raise.
            self.budget.record(0.0)
            raise

        self.budget.record(score)
        self._track(candidate, score, info)

        info = dict(info) if info else {}
        info["_budget"] = self.budget.status()
        return score, info
    finally:
        self._eval_semaphore.release()

evaluate_batch(pairs: list[tuple[str, Any]], opt_states: list[Any] | None = None) -> list[tuple[float, dict[str, Any]]]

Evaluate pairs in ONE call to the user's batch_evaluate function.

The grouped analogue of :meth:evaluate (e.g. to submit a provider batch job): the user function receives all pairs at once; each pair is then recorded against the budget and tracked individually. The budget is checked once up front, so a batch may overshoot the cap by at most len(pairs) - 1 — mirroring the launcher's iteration-boundary stops. Failed attempts consume budget (launcher parity): a failed grouped call records one tick per pair.

Raises:

Type	Description
BudgetExhausted	When the budget is already used up.
RuntimeError	When the server was constructed without batch_evaluate.

Source code in gepa/oa/eval_server.py

def evaluate_batch(
    self,
    pairs: list[tuple[str, Any]],
    opt_states: list[Any] | None = None,
) -> list[tuple[float, dict[str, Any]]]:
    """Evaluate ``pairs`` in ONE call to the user's ``batch_evaluate`` function.

    The grouped analogue of :meth:`evaluate` (e.g. to submit a provider
    batch job): the user function receives all pairs at once; each pair is
    then recorded against the budget and tracked individually. The budget
    is checked once up front, so a batch may overshoot the cap by at most
    ``len(pairs) - 1`` — mirroring the launcher's iteration-boundary stops.
    Failed attempts consume budget (launcher parity): a failed grouped
    call records one tick per pair.

    Raises:
        BudgetExhausted: When the budget is already used up.
        RuntimeError: When the server was constructed without ``batch_evaluate``.
    """
    if self.batch_fn is None:
        raise RuntimeError("evaluate_batch requires the server to be constructed with batch_evaluate")
    self._eval_semaphore.acquire()
    try:
        self.budget.check()
        try:
            results = self.batch_fn(pairs, opt_states=opt_states)
        except Exception:
            # Guard each record so crossing the cap while recording
            # failures never masks the user's original exception.
            for _ in pairs:
                try:
                    self.budget.record(0.0)
                except BudgetExhausted:
                    break
            raise
    finally:
        self._eval_semaphore.release()
    out: list[tuple[float, dict[str, Any]]] = []
    for (candidate, _example), (score, info) in zip(pairs, results, strict=True):
        self.budget.record(score)
        self._track(candidate, score, info)
        info = dict(info)
        info["_budget"] = self.budget.status()
        out.append((score, info))
    return out

evaluate_examples(candidate: str, example_ids: list[str] | None = None, split: str | None = None) -> tuple[float, dict[str, Any]]

Evaluate a candidate on specific examples or a whole split, in parallel.

Provide either example_ids or split ("train"/"val"/ "all"); example_ids takes precedence. The server holds no test split, so "all" means train+val. For single-task tasks, delegates to :meth:evaluate.

Returns:

Type	Description
tuple[float, dict[str, Any]]	(average_score, info_dict) with per-example scores and metadata.

Source code in gepa/oa/eval_server.py

def evaluate_examples(
    self,
    candidate: str,
    example_ids: list[str] | None = None,
    split: str | None = None,
) -> tuple[float, dict[str, Any]]:
    """Evaluate a candidate on specific examples or a whole split, in parallel.

    Provide either ``example_ids`` or ``split`` (``"train"``/``"val"``/
    ``"all"``); ``example_ids`` takes precedence. The server holds no
    ``test`` split, so ``"all"`` means train+val. For single-task tasks,
    delegates to :meth:`evaluate`.

    Returns:
        (average_score, info_dict) with per-example scores and metadata.
    """
    if not self.task.has_dataset:
        score, info = self.evaluate(candidate)
        return score, {
            "scores": {"_single": score},
            "infos": {"_single": info},
            "num_evaluated": 1,
            "num_total": 1,
            "partial": False,
            "_budget": self.budget.status(),
        }

    examples: list[tuple[str, Any]] = []
    if example_ids is not None:
        for eid in example_ids:
            if eid in self._examples:
                examples.append((eid, self._examples[eid]))
    elif split is not None:
        splits = ("train", "val") if split == "all" else (split,)
        for s in splits:
            examples.extend(self.iter_split(s))
    elif self.task.train_set:
        examples.extend(self.iter_split("train"))

    if not examples:
        score, info = self.evaluate(candidate)
        return score, {
            "scores": {"_single": score},
            "infos": {"_single": info},
            "num_evaluated": 1,
            "num_total": 1,
            "partial": False,
            "_budget": self.budget.status(),
        }

    if self.budget.remaining is not None and self.budget.remaining < len(examples):
        raise BudgetExhausted(
            f"Not enough budget to evaluate all examples: {self.budget.remaining} remaining, {len(examples)} needed"
        )

    scores: dict[str, float] = {}
    infos: dict[str, dict[str, Any]] = {}
    errors: dict[str, str] = {}

    grouped_done = False
    if self.batch_fn is not None:
        # Multi-pair evaluations prefer the grouped path (launcher parity):
        # the user's batch function sees all pairs in one call.
        try:
            results = self.evaluate_batch([(candidate, ex) for _eid, ex in examples])
        except BudgetExhausted:
            raise
        except Exception:
            # A failed grouped call must not zero the whole stage: fall
            # through to per-pair evaluation so one bad pair costs one 0.0,
            # matching the per-pair path's isolation. Both attempts consume
            # budget — each was a real eval call.
            pass
        else:
            grouped_done = True
            for (eid, _ex), (score, ex_info) in zip(examples, results, strict=True):
                scores[eid] = score
                infos[eid] = ex_info
    if not grouped_done:

        def _eval_one(eid: str, ex: Any) -> tuple[str, float, dict[str, Any] | None, str | None]:
            try:
                score, info = self.evaluate(candidate, ex)
                return (eid, score, info, None)
            except Exception as e:
                return (eid, 0.0, None, str(e))

        futures = {self._pool.submit(_eval_one, eid, ex): eid for eid, ex in examples}
        for future in as_completed(futures):
            eid, score, ex_info, err = future.result()
            if err is not None:
                errors[eid] = err
            else:
                scores[eid] = score
                if ex_info is not None:
                    infos[eid] = ex_info

    all_scores = {**scores, **dict.fromkeys(errors, 0.0)}
    avg = sum(all_scores.values()) / len(all_scores)
    info: dict[str, Any] = {
        "scores": all_scores,
        "infos": infos,
        "num_evaluated": len(examples),
        "_budget": self.budget.status(),
    }
    if errors:
        info["errors"] = errors
    return avg, info

validate(candidate: str) -> dict[str, Any]

Evaluate candidate on the hidden val_set and log progress.

Source code in gepa/oa/eval_server.py

def validate(self, candidate: str) -> dict[str, Any]:
    """Evaluate ``candidate`` on the hidden ``val_set`` and log progress."""
    if not self.task.val_set:
        raise ValueError("validate() requires a task with val_set")
    avg_score, _ = self.evaluate_examples(candidate, example_ids=list(self._split_ids["val"]))
    return self.log_progress(avg_score, candidate=candidate)

log_progress(val_score: float, candidate: str | None = None, reflection_cost: float = 0.0) -> dict[str, Any]

Record a progress checkpoint.

Source code in gepa/oa/eval_server.py

def log_progress(
    self, val_score: float, candidate: str | None = None, reflection_cost: float = 0.0
) -> dict[str, Any]:
    """Record a progress checkpoint."""
    candidate_id: int | None = None
    if candidate is not None:
        candidate_id = self._register_candidate(candidate)

    with self._lock:
        if val_score > self._best_val_score:
            self._best_val_score = val_score
        entry: dict[str, Any] = {
            "val_score": val_score,
            "best_val_score": self._best_val_score,
            "total_evals": self.budget.used,
            "wall_time": time.time() - self._start_time,
            "total_cost": self.total_cost,
            "reflection_cost": reflection_cost,
        }
        if candidate_id is not None:
            entry["candidate_id"] = candidate_id
        self._progress_log.append(entry)

    if self.output_dir is not None:
        with self._io_lock:
            with open(self.output_dir / "progress_log.jsonl", "a") as f:
                f.write(json.dumps(entry) + "\n")

    return {"val_score": val_score, "best_val_score": self._best_val_score}

start(port: int = 0) -> int

Start the HTTP server. Returns the bound port.

Source code in gepa/oa/eval_server.py

def start(self, port: int = 0) -> int:
    """Start the HTTP server. Returns the bound port."""
    server_ref = self

    class Handler(BaseHTTPRequestHandler):
        def do_POST(self):
            if self.path == "/evaluate":
                server_ref._handle_evaluate(self)
            elif self.path == "/evaluate_examples":
                server_ref._handle_evaluate_examples(self)
            elif self.path == "/validate":
                server_ref._handle_validate(self)
            else:
                self.send_error(404)

        def do_GET(self):
            if self.path == "/status":
                server_ref._handle_status(self)
            elif self.path == "/task":
                server_ref._handle_task_info(self)
            else:
                self.send_error(404)

        def log_message(self, format, *args):
            pass

    # threading.HTTPServer would be better for high concurrency, but the
    # underlying eval is already serialized through ``_eval_semaphore``;
    # we keep the simple single-threaded HTTPServer to avoid surprising
    # ordering, and rely on the thread pool inside evaluate_examples to
    # parallelize batches.
    from http.server import ThreadingHTTPServer

    self._server = ThreadingHTTPServer(("localhost", port), Handler)
    self._thread = threading.Thread(target=self._server.serve_forever, daemon=True)
    self._thread.start()
    return self.port

stop() -> None

Source code in gepa/oa/eval_server.py

def stop(self) -> None:
    if self._server:
        self._server.shutdown()
        self._server = None
    self._pool.shutdown(wait=False)