EvaluationCache

gepa.core.state.EvaluationCache(_cache: dict[CacheKey, CachedEvaluation[RolloutOutput]] = dict()) dataclass

Bases: Generic[RolloutOutput, DataId]

Cache for (candidate, split, example) evaluation results.

Data loaders identify examples by position, so a trainset and a separate valset both number their examples from zero. Every lookup therefore takes a required keyword-only split of :data:TRAINSET_CACHE_SPLIT or :data:VALSET_CACHE_SPLIT. A valset that is a distinct loader from the trainset uses VALSET_CACHE_SPLIT; when the two are the same loader the ids mean the same thing and both sides share TRAINSET_CACHE_SPLIT so minibatch rollouts are reused on valset evaluation.

Keys persisted before splits existed are (candidate_hash, example_id). Those cannot be told apart from contaminated distinct-valset entries, so :meth:GEPAState.load drops them rather than serving them. There is no default split: omitting it is a TypeError, not a silent trainset hit.

Methods:

drop_unsplit_entries() -> int

Discard entries written before splits existed. Returns the number dropped.

Those keys are (candidate_hash, example_id) with no namespace, so a valset rollout and the trainset rollout at the same position were stored under the same key. They can no longer be served (every lookup is namespaced now), and keeping them would grow the persisted state on every resume, so GEPAState.load drops them.

Source code in gepa/core/state.py

def drop_unsplit_entries(self) -> int:
    """Discard entries written before splits existed. Returns the number dropped.

    Those keys are ``(candidate_hash, example_id)`` with no namespace, so a valset rollout and
    the trainset rollout at the same position were stored under the same key. They can no longer
    be served (every lookup is namespaced now), and keeping them would grow the persisted state
    on every resume, so ``GEPAState.load`` drops them.
    """
    stale = [k for k in self._cache if not self._is_split_key(k)]
    for k in stale:
        del self._cache[k]
    return len(stale)

get(candidate: dict[str, str], example_id: DataId, *, split: str) -> CachedEvaluation[RolloutOutput] | None

Retrieve cached evaluation result if it exists. split is required.

Source code in gepa/core/state.py

def get(
    self, candidate: dict[str, str], example_id: DataId, *, split: str
) -> CachedEvaluation[RolloutOutput] | None:
    """Retrieve cached evaluation result if it exists. ``split`` is required."""
    return self._cache.get(self._key(_candidate_hash(candidate), example_id, split))

put(candidate: dict[str, str], example_id: DataId, output: RolloutOutput, score: float, objective_scores: ObjectiveScores | None = None, *, split: str) -> None

Store an evaluation result in the cache.

Source code in gepa/core/state.py

def put(
    self,
    candidate: dict[str, str],
    example_id: DataId,
    output: RolloutOutput,
    score: float,
    objective_scores: ObjectiveScores | None = None,
    *,
    split: str,
) -> None:
    """Store an evaluation result in the cache."""
    key = self._key(_candidate_hash(candidate), example_id, split)
    self._cache[key] = CachedEvaluation(output, score, objective_scores)

get_batch(candidate: dict[str, str], example_ids: list[DataId], *, split: str) -> tuple[dict[DataId, CachedEvaluation[RolloutOutput]], list[DataId]]

Look up cached results for a batch. Returns (cached_results, uncached_ids).

Source code in gepa/core/state.py

def get_batch(
    self, candidate: dict[str, str], example_ids: list[DataId], *, split: str
) -> tuple[dict[DataId, CachedEvaluation[RolloutOutput]], list[DataId]]:
    """Look up cached results for a batch. Returns (cached_results, uncached_ids)."""
    h = _candidate_hash(candidate)
    cached, uncached = {}, []
    for eid in example_ids:
        if entry := self._cache.get(self._key(h, eid, split)):
            cached[eid] = entry
        else:
            uncached.append(eid)
    return cached, uncached

put_batch(candidate: dict[str, str], example_ids: list[DataId], outputs: list[RolloutOutput], scores: list[float], objective_scores_list: Sequence[ObjectiveScores] | None = None, *, split: str) -> None

Store evaluation results for a batch of examples.

Source code in gepa/core/state.py

def put_batch(
    self,
    candidate: dict[str, str],
    example_ids: list[DataId],
    outputs: list[RolloutOutput],
    scores: list[float],
    objective_scores_list: Sequence[ObjectiveScores] | None = None,
    *,
    split: str,
) -> None:
    """Store evaluation results for a batch of examples."""
    h = _candidate_hash(candidate)
    for i, eid in enumerate(example_ids):
        self._cache[self._key(h, eid, split)] = CachedEvaluation(
            outputs[i], scores[i], objective_scores_list[i] if objective_scores_list else None
        )

evaluate_with_cache_full(candidate: dict[str, str], example_ids: list[DataId], fetcher: Callable[[list[DataId]], Any], evaluator: Callable[[Any, dict[str, str]], tuple[Any, list[float], Sequence[ObjectiveScores] | None]], *, split: str) -> tuple[dict[DataId, RolloutOutput], dict[DataId, float], dict[DataId, ObjectiveScores] | None, int]

Evaluate using cache, returning full results.

Returns (outputs_by_id, scores_by_id, objective_scores_by_id, num_actual_evals).

Source code in gepa/core/state.py

def evaluate_with_cache_full(
    self,
    candidate: dict[str, str],
    example_ids: list[DataId],
    fetcher: Callable[[list[DataId]], Any],
    evaluator: Callable[[Any, dict[str, str]], tuple[Any, list[float], Sequence[ObjectiveScores] | None]],
    *,
    split: str,
) -> tuple[dict[DataId, RolloutOutput], dict[DataId, float], dict[DataId, ObjectiveScores] | None, int]:
    """
    Evaluate using cache, returning full results.

    Returns (outputs_by_id, scores_by_id, objective_scores_by_id, num_actual_evals).
    """
    cached, uncached_ids = self.get_batch(candidate, example_ids, split=split)

    outputs_by_id: dict[DataId, RolloutOutput] = {eid: c.output for eid, c in cached.items()}
    scores_by_id: dict[DataId, float] = {eid: c.score for eid, c in cached.items()}
    objective_by_id: dict[DataId, ObjectiveScores] | None = None

    # Populate objective scores from cache
    for eid, c in cached.items():
        if c.objective_scores is not None:
            objective_by_id = objective_by_id or {}
            objective_by_id[eid] = c.objective_scores

    # Evaluate uncached examples
    if uncached_ids:
        batch = fetcher(uncached_ids)
        outputs, scores, obj_scores = evaluator(batch, candidate)
        for idx, eid in enumerate(uncached_ids):
            outputs_by_id[eid] = outputs[idx]
            scores_by_id[eid] = scores[idx]
            if obj_scores is not None:
                objective_by_id = objective_by_id or {}
                objective_by_id[eid] = obj_scores[idx]
        self.put_batch(candidate, uncached_ids, outputs, scores, obj_scores, split=split)

    return outputs_by_id, scores_by_id, objective_by_id, len(uncached_ids)