Skip to content

Optimization API

optimization

Optimization primitives — reflective prompt evolution via GEPA.

The cognitive substrate's evolutionary layer. Orqest already produces every signal a Pareto-optimizer needs (accuracy via output validation, confidence via :class:EnrichedOutput, latency via :class:Tracer.Span, cost via pydantic_ai.usage.Usage, robustness via :class:HealingRunner); this battery wires GEPA (Genetic-Pareto reflective evolution, Agrawal et al., ICLR 2026 Oral) into that signal stream so the prompts driving your agents evolve from real traces.

See :doc:/concepts/optimization for the full picture; this module hosts:

  • :class:OptimizationConfig — frozen dataclass.
  • :class:MetricBundle + :class:MetricWeights — per-example fitness.
  • :class:Genome + :class:PromptGene / :class:ScalarGene / :class:CategoricalGene — what's evolvable.
  • :class:GoldExample + :class:Evaluator — how to score a candidate.
  • :class:OrqestGEPAAdapter + :class:OrqestEvalBatch — the GEPA bridge.
  • :class:OptimizationRunner + :class:OptimizationResult — run the loop.
  • :class:OptimizationDiff + :func:apply_result — write the winner back.

GEPA is an optional dependency. Install with::

uv sync --group optimization

FrontierType module-attribute 

FrontierType = Literal[
    "instance", "objective", "hybrid", "cartesian"
]

GEPA's native Pareto-frontier shapes:

  • "instance" — best candidate per validation example (the original GEPA paper)
  • "objective" — best candidate per metric dimension (accuracy / cost / ...)
  • "hybrid" — both, jointly. The recommended default for Orqest because the battery feeds GEPA both per-example scores and per-objective scores.
  • "cartesian" — every (instance × objective) pair. Largest frontier; use only when search budget allows.

OrqestEvalBatch dataclass 

OrqestEvalBatch(bundles: list[MetricBundle] = list())

:class:EvaluationBatch extended with the typed :class:MetricBundles.

GEPA only reads outputs / scores / trajectories / objective_scores — bundles is a sidecar so reflective dataset construction (and downstream consumers like notebooks) can show per-dimension detail without re-evaluating.

OrqestGEPAAdapter 

OrqestGEPAAdapter(
    genome: Genome,
    evaluator: Evaluator[Any, Any],
    weights: MetricWeights,
    *,
    bus: EventBus | None = None,
    emit_per_example_events: bool = False,
)

:class:gepa.core.adapter.GEPAAdapter implementation backed by an Orqest :class:Evaluator + :class:Genome.

Constructed and held by :class:OptimizationRunner; never invoked directly by user code. The Protocol is structural in GEPA, so we declare conformance by shape rather than inheritance — keeps the optional-dependency story clean (no GEPA import needed at module load time for from orqest.optimization import OrqestGEPAAdapter).

Wire the adapter to an :class:Evaluator + an optional bus.

Source code in orqest/optimization/adapter.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def __init__(
    self,
    genome: Genome,
    evaluator: Evaluator[Any, Any],
    weights: MetricWeights,
    *,
    bus: EventBus | None = None,
    emit_per_example_events: bool = False,
) -> None:
    """Wire the adapter to an :class:`Evaluator` + an optional bus."""
    self._genome = genome
    self._evaluator = evaluator
    self._weights = weights
    self._bus = bus
    self._emit_per_example_events = emit_per_example_events
    self._iteration = 0
propose_new_texts class-attribute 
propose_new_texts: None = None

GEPA's reflective-mutation loop checks adapter.propose_new_texts is not None to decide whether to use a user-provided text proposer; when None it falls back to its built-in LLM-driven proposer (which is what we want — GEPA's default uses reflection_lm to evolve the prompt). Declared as a class attribute so the access site doesn't raise AttributeError; if we ever ship a custom proposer it becomes a method on this class.

evaluate 
evaluate(
    batch: list[GoldExample[Any, Any]],
    candidate: dict[str, str],
    capture_traces: bool = False,
) -> OrqestEvalBatch

Run candidate on batch and return an :class:OrqestEvalBatch.

Per the GEPA Protocol contract:

  • scores length == len(batch); higher is better.
  • trajectories populated only when capture_traces=True.
  • objective_scores populated unconditionally — Orqest's multi-signal nature makes them effectively free.
  • Per-example failures yield score=0.0 rather than raising.
Source code in orqest/optimization/adapter.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def evaluate(
    self,
    batch: list[GoldExample[Any, Any]],
    candidate: dict[str, str],
    capture_traces: bool = False,
) -> OrqestEvalBatch:
    """Run *candidate* on *batch* and return an :class:`OrqestEvalBatch`.

    Per the GEPA Protocol contract:

    * ``scores`` length == ``len(batch)``; higher is better.
    * ``trajectories`` populated only when ``capture_traces=True``.
    * ``objective_scores`` populated unconditionally — Orqest's
      multi-signal nature makes them effectively free.
    * Per-example failures yield ``score=0.0`` rather than raising.
    """
    decoded = self._genome.decode(candidate)
    bundles = self._run_async(self._evaluator.evaluate_batch(decoded, batch))

    scores: list[float] = [b.scalarize(self._weights) for b in bundles]
    objective_scores: list[dict[str, float]] = [
        b.to_per_instance_scores(self._weights) for b in bundles
    ]
    outputs: list[dict[str, Any]] = [
        {"accuracy": b.accuracy, "raw": b.raw} for b in bundles
    ]

    trajectories: list[dict[str, Any]] | None = None
    if capture_traces:
        trajectories = [
            self._build_trajectory(ex, b)
            for ex, b in zip(batch, bundles, strict=True)
        ]

    self._iteration += 1
    if self._bus is not None:
        self._emit_iteration_summary(scores, bundles)

    return OrqestEvalBatch(
        outputs=outputs,
        scores=scores,
        trajectories=trajectories,
        objective_scores=objective_scores,
        bundles=bundles,
    )
make_reflective_dataset 
make_reflective_dataset(
    candidate: dict[str, str],
    eval_batch: OrqestEvalBatch,
    components_to_update: list[str],
) -> Mapping[str, Sequence[Mapping[str, Any]]]

Build the per-component reflective dataset for the teacher LLM.

Schema follows GEPA's recommendation: each record carries Inputs / Generated Outputs / Feedback. The Feedback slot is where Orqest's typed signals shine — we surface the per-dimension :class:MetricBundle breakdown plus any uncertainty_targets the agent self-reported.

Source code in orqest/optimization/adapter.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
def make_reflective_dataset(
    self,
    candidate: dict[str, str],
    eval_batch: OrqestEvalBatch,
    components_to_update: list[str],
) -> Mapping[str, Sequence[Mapping[str, Any]]]:
    """Build the per-component reflective dataset for the teacher LLM.

    Schema follows GEPA's recommendation: each record carries
    ``Inputs`` / ``Generated Outputs`` / ``Feedback``. The Feedback
    slot is where Orqest's typed signals shine — we surface the
    per-dimension :class:`MetricBundle` breakdown plus any
    ``uncertainty_targets`` the agent self-reported.
    """
    if eval_batch.trajectories is None:
        return {name: [] for name in components_to_update}

    constraints_by_name = self._gene_constraints()

    dataset: dict[str, list[Mapping[str, Any]]] = {}
    for name in components_to_update:
        records: list[Mapping[str, Any]] = []
        for traj in eval_batch.trajectories:
            record: dict[str, Any] = {
                "Inputs": {"input": traj.get("input")},
                "Generated Outputs": {
                    "output": traj.get("output"),
                    "uncertainty_targets": traj.get("uncertainty_targets", []),
                },
                "Feedback": traj.get("feedback", ""),
                "score": traj.get("score"),
            }
            if name in constraints_by_name:
                record["Constraints"] = constraints_by_name[name]
            records.append(record)
        dataset[name] = records
    return dataset

OptimizationDiff dataclass 

OptimizationDiff(
    gene_name: str, before: str, after: str, unified: str
)

One slot's before/after diff.

unified instance-attribute 
unified: str

Unified diff (difflib.unified_diff output) — useful for printing or logging the change.

MetricBundle

One example's multi-dimensional fitness.

Accuracy / confidence / robustness are normalized to [0, 1] (higher is better). Cost (USD) and latency (ms) are raw, non-negative quantities (lower is better) — :class:MetricWeights carries negative weights so :meth:scalarize produces the right gradient.

raw class-attribute instance-attribute 
raw: dict[str, Any] = Field(default_factory=dict)

Extension point — consumer-defined extras (e.g., per-tool counters). Surfaced in events / notebooks but ignored by :meth:scalarize.

n_trials class-attribute instance-attribute 
n_trials: int = Field(default=1, ge=1)

Number of independent observations aggregated into this bundle. 1 (the default) means a single observation. Values > 1 indicate this bundle is the mean of n_trials underlying bundles, produced by :meth:MetricBundle.aggregate. Use this when you've re-run the same candidate / example multiple times to wash out LLM run-to-run variance — the discovery from the coding benchmark was that single-trial numbers swing ±10pp on weaker models, so multi-trial averaging is often the right default for evaluator pipelines.

stdev class-attribute instance-attribute 
stdev: dict[str, float] | None = Field(default=None)

Per-dimension standard deviation across the trials this bundle aggregates. Keys mirror the metric field names (accuracy, confidence, cost_usd, latency_ms, robustness). None when n_trials == 1 (no dispersion to report). When a dimension was None in some trials but present in others, the stdev is computed only over trials where it was present — and omitted entirely from stdev when fewer than 2 such trials exist (single observations have no defined standard deviation).

scalarize 
scalarize(w: MetricWeights) -> float

Single per-example fitness score for GEPA's acceptance test.

Optional dimensions (confidence, robustness) are skipped when None rather than treated as zero — a missing signal must not penalize a candidate that simply didn't surface it.

Source code in orqest/optimization/bundle.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def scalarize(self, w: MetricWeights) -> float:
    """Single per-example fitness score for GEPA's acceptance test.

    Optional dimensions (``confidence``, ``robustness``) are skipped
    when ``None`` rather than treated as zero — a missing signal must
    not penalize a candidate that simply didn't surface it.
    """
    score = self.accuracy * w.accuracy
    score += self.cost_usd * w.cost_usd
    score += self.latency_ms * w.latency_ms
    if self.confidence is not None:
        score += self.confidence * w.confidence
    if self.robustness is not None:
        score += self.robustness * w.robustness
    return score
to_per_instance_scores 
to_per_instance_scores(
    w: MetricWeights,
) -> dict[str, float]

Per-dimension scores for GEPA's objective_scores field.

Each key becomes an axis on GEPA's objective Pareto front when frontier_type is "objective" / "hybrid" / "cartesian". Optional dimensions that are None are omitted (not zero-filled) so the frontier doesn't conflate "absent" with "bad."

Source code in orqest/optimization/bundle.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
def to_per_instance_scores(self, w: MetricWeights) -> dict[str, float]:
    """Per-dimension scores for GEPA's ``objective_scores`` field.

    Each key becomes an axis on GEPA's objective Pareto front when
    ``frontier_type`` is ``"objective"`` / ``"hybrid"`` / ``"cartesian"``.
    Optional dimensions that are ``None`` are omitted (not zero-filled)
    so the frontier doesn't conflate "absent" with "bad."
    """
    scores: dict[str, float] = {
        "accuracy": self.accuracy,
        "cost_usd": self.cost_usd,
        "latency_ms": self.latency_ms,
    }
    if self.confidence is not None:
        scores["confidence"] = self.confidence
    if self.robustness is not None:
        scores["robustness"] = self.robustness
    return scores
aggregate classmethod 
aggregate(bundles: list[MetricBundle]) -> MetricBundle

Mean each metric across bundles; compute per-dimension stdev.

Use this to collapse N independent observations of the same candidate / example into a single representative bundle. The returned bundle has n_trials = len(bundles) and a populated stdev dict (when N >= 2) so consumers can see dispersion alongside the central tendency.

Optional dimensions (confidence, robustness) are handled gracefully:

  • If every bundle has it None, the aggregate also reports None for that dimension and omits it from stdev.
  • If some bundles have it set and others None, the aggregate reports the mean over present values. stdev includes the dimension only when 2 or more bundles supplied it.

The raw dict of the first bundle is preserved as a representative sample (it's a consumer-defined extension point; merging semantics belong to the consumer, not this aggregator).

Parameters:

Name Type Description Default
bundles list[MetricBundle]

At least one bundle. Empty list raises ValueError.

 required

Returns:

Type Description
MetricBundle

A new MetricBundle with mean metrics + populated stdev.

Raises:

Type Description
ValueError

When bundles is empty.
Source code in orqest/optimization/bundle.py 
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
@classmethod
def aggregate(cls, bundles: list[MetricBundle]) -> MetricBundle:
    """Mean each metric across ``bundles``; compute per-dimension stdev.

    Use this to collapse N independent observations of the same
    candidate / example into a single representative bundle. The
    returned bundle has ``n_trials = len(bundles)`` and a populated
    ``stdev`` dict (when N >= 2) so consumers can see dispersion
    alongside the central tendency.

    Optional dimensions (``confidence``, ``robustness``) are handled
    gracefully:

    * If *every* bundle has it ``None``, the aggregate also reports
      ``None`` for that dimension and omits it from ``stdev``.
    * If *some* bundles have it set and others ``None``, the aggregate
      reports the mean over present values. ``stdev`` includes the
      dimension only when 2 or more bundles supplied it.

    The ``raw`` dict of the first bundle is preserved as a
    representative sample (it's a consumer-defined extension point;
    merging semantics belong to the consumer, not this aggregator).

    Args:
        bundles: At least one bundle. Empty list raises ``ValueError``.

    Returns:
        A new ``MetricBundle`` with mean metrics + populated ``stdev``.

    Raises:
        ValueError: When ``bundles`` is empty.
    """
    if not bundles:
        raise ValueError(
            "MetricBundle.aggregate requires at least one bundle "
            "(got empty list); nothing to average."
        )
    if len(bundles) == 1:
        # Single observation: return unchanged (n_trials=1, stdev=None).
        # Callers can rely on aggregate([x]) == x for any x.
        return bundles[0]

    n = len(bundles)
    accuracy_mean = statistics.fmean(b.accuracy for b in bundles)
    cost_mean = statistics.fmean(b.cost_usd for b in bundles)
    latency_mean = statistics.fmean(b.latency_ms for b in bundles)

    # Optional dims: average over PRESENT values only
    confidence_vals = [b.confidence for b in bundles if b.confidence is not None]
    robustness_vals = [b.robustness for b in bundles if b.robustness is not None]
    confidence_mean = statistics.fmean(confidence_vals) if confidence_vals else None
    robustness_mean = statistics.fmean(robustness_vals) if robustness_vals else None

    # stdev requires >= 2 samples per dimension
    stdev: dict[str, float] = {
        "accuracy": statistics.stdev(b.accuracy for b in bundles),
        "cost_usd": statistics.stdev(b.cost_usd for b in bundles),
        "latency_ms": statistics.stdev(b.latency_ms for b in bundles),
    }
    if len(confidence_vals) >= 2:
        stdev["confidence"] = statistics.stdev(confidence_vals)
    if len(robustness_vals) >= 2:
        stdev["robustness"] = statistics.stdev(robustness_vals)

    return cls(
        accuracy=accuracy_mean,
        confidence=confidence_mean,
        cost_usd=cost_mean,
        latency_ms=latency_mean,
        robustness=robustness_mean,
        raw=dict(bundles[0].raw),  # representative; consumer owns merge semantics
        n_trials=n,
        stdev=stdev,
    )

MetricWeights

Weights for :meth:MetricBundle.scalarize.

Negative weights penalize (cost, latency); positive weights reward (accuracy, confidence, robustness). Defaults are calibrated for "accuracy is king, but don't ignore cost / latency / calibration" against realistic LLM call latencies (1–10 seconds = 1k–10k ms) and realistic per-call costs ($0.0001 – $0.05).

Magnitude rationale: a perfect-accuracy candidate at ~5 s / $0.005 scores ≈ 1.0 + 0 + 0 - 0.05 - 0.05 = 0.9. Latency and cost act as soft tiebreakers; accuracy stays the dominant axis. If you care more about cost, override; production users will tune these per app.

cost_usd class-attribute instance-attribute 
cost_usd: float = -10.0

-10.0 per USD: a $0.001 call is -0.01, $0.01 is -0.10, $0.10 is -1.0. Calibrated so per-call cost is a soft tiebreaker against accuracy (range 0–1), not a dominant signal.

latency_ms class-attribute instance-attribute 
latency_ms: float = -2e-05

-0.00002 per ms: a 1 s call is -0.02, a 5 s call is -0.10, a 30 s call is -0.60. Same logic as cost_usd — soft tiebreaker, not dominant.

OptimizationConfig dataclass 

OptimizationConfig(
    max_metric_calls: int = 150,
    reflection_model: str | None = None,
    minibatch_size: int = 3,
    valset_fraction: float = 0.3,
    weights: MetricWeights = MetricWeights(),
    seed: int | None = 42,
    dry_run_default: bool = True,
    enable_scalar_genes: bool = False,
    enable_categorical_genes: bool = False,
    cache_evaluations: bool = True,
    emit_per_example_events: bool = False,
    frontier_type: FrontierType = "hybrid",
)

Knobs for the optimization battery.

Pass explicitly to :class:OptimizationRunner.

max_metric_calls class-attribute instance-attribute 
max_metric_calls: int = 150

GEPA's rollout budget — total evaluate() calls across the search. 150 is a reasonable default for a 10–20 example gold set; scale linearly with eval-set size if you want more iterations.

reflection_model class-attribute instance-attribute 
reflection_model: str | None = None

provider:model_id for the LLM that proposes prompt mutations. None falls back to :attr:OrqestConfig.llm_model. Use the strongest model you can afford here — this is the optimizer's brain. The task model (whatever the user's agent_factory constructs the agent with) can stay cheap; the reflection model should not.

Note: there is no task_model field. The model the candidate prompt runs against is whatever the user's agent_factory wires into the :class:BaseAgent; GEPA's adapter API explicitly forbids passing a task_lm when an adapter is supplied (the adapter owns it).

minibatch_size class-attribute instance-attribute 
minibatch_size: int = 3

Examples sampled per acceptance test. GEPA uses sum(scores) over the minibatch to decide whether a mutation beats its parent.

valset_fraction class-attribute instance-attribute 
valset_fraction: float = 0.3

When :meth:OptimizationRunner.optimize receives no explicit valset, this fraction of trainset is held out (deterministically with :attr:seed) for Pareto tracking.

weights class-attribute instance-attribute 
weights: MetricWeights = field(
    default_factory=MetricWeights
)

Drives :meth:MetricBundle.scalarize — the per-example float that GEPA's acceptance test compares. The per-objective objective_scores GEPA also receives are unweighted; weights only matter for the scalar summary and for which candidate best_candidate resolves to.

seed class-attribute instance-attribute 
seed: int | None = 42

Random seed for trainset/valset splitting and any sampling. None disables seeding (every run differs).

dry_run_default class-attribute instance-attribute 
dry_run_default: bool = True

Default value for :func:apply_result(..., dry_run=...). When True, applying the result prints the diff but does not mutate the agent.

enable_scalar_genes class-attribute instance-attribute 
enable_scalar_genes: bool = False

Gate for :class:ScalarGene evolution (W1.1+). When False (default), :meth:OptimizationRunner.optimize raises NotImplementedError if a :class:ScalarGene appears in the genome.

enable_categorical_genes class-attribute instance-attribute 
enable_categorical_genes: bool = False

Gate for :class:CategoricalGene evolution (W1.1+). Same semantics as :attr:enable_scalar_genes.

cache_evaluations class-attribute instance-attribute 
cache_evaluations: bool = True

Pass-through to GEPA's built-in cache_evaluation knob — de-duplicates (candidate, example) evaluations during Pareto selection. 2–5× cost saving for free.

emit_per_example_events class-attribute instance-attribute 
emit_per_example_events: bool = False

Off by default — 150 metric_calls × N examples can be 1000+ events per run, flooding the bus. When False, :class:OrqestGEPAAdapter only emits optimization.iteration_completed summaries.

frontier_type class-attribute instance-attribute 
frontier_type: FrontierType = 'hybrid'

Pass-through to GEPA's optimize(frontier_type=...). Default "hybrid" because the battery feeds GEPA both scores (per-example scalar) and objective_scores (per-dimension); "hybrid" exploits both. See :data:FrontierType for alternatives.

Evaluator 

Evaluator(
    agent_factory: Callable[
        [dict[str, Any]], BaseAgent[Any, OutputT]
    ],
    score_fn: Callable[
        [OutputT, GoldExample[InputT, OutputT]],
        float | Awaitable[float],
    ],
    *,
    confidence_protocol: ConfidenceProtocol | None = None,
    cost_estimator: Callable[[Any], float] | None = None,
    timer: Callable[[], float] = time.monotonic,
    n_trials_per_example: int = 1,
)

Score one candidate against a gold set, producing :class:MetricBundles.

Construction takes the consumer-side wiring; per-call decoded is the decoded genome from :meth:Genome.decode.

Wire the per-evaluation building blocks.

Parameters:

Name Type Description Default
agent_factory Callable[[dict[str, Any]], BaseAgent[Any, OutputT]]

Callable taking a decoded genome and returning a fresh :class:BaseAgent. Must not mutate or share state across calls — one factory invocation per evaluation.

 required
score_fn Callable[[OutputT, GoldExample[InputT, OutputT]], float | Awaitable[float]]

(output, example) -> float in [0, 1]. May be sync or async.

 required
confidence_protocol ConfidenceProtocol | None

Optional :class:ConfidenceProtocol for self-rated calibration. When set, :attr:MetricBundle.confidence is filled via BaseAgent.run_enriched.

 None
cost_estimator Callable[[Any], float] | None

Optional (RunUsage) -> float translating token usage to USD. When omitted, :attr:MetricBundle.cost_usd stays 0 (token counts still surface via :attr:MetricBundle.raw).

 None
timer Callable[[], float]

Monotonic clock for latency measurement. Override only in tests.

 monotonic
n_trials_per_example int

When > 1, evaluate each example that many times and aggregate via :meth:MetricBundle.aggregate to wash out LLM run-to-run variance. The returned bundle carries n_trials and per-dimension stdev. 1 (default) preserves the legacy single-shot behavior at no extra cost. Each trial gets a fresh agent (agent_factory(decoded) called per trial) so trials are independent. Cost scales linearly; useful for weaker models where single trials swing ±10pp.

 1
Source code in orqest/optimization/evaluator.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
def __init__(
    self,
    agent_factory: Callable[[dict[str, Any]], BaseAgent[Any, OutputT]],
    score_fn: Callable[
        [OutputT, GoldExample[InputT, OutputT]],
        float | Awaitable[float],
    ],
    *,
    confidence_protocol: ConfidenceProtocol | None = None,
    cost_estimator: Callable[[Any], float] | None = None,
    timer: Callable[[], float] = time.monotonic,
    n_trials_per_example: int = 1,
) -> None:
    """Wire the per-evaluation building blocks.

    Args:
        agent_factory: Callable taking a decoded genome and returning a
            **fresh** :class:`BaseAgent`. Must not mutate or share state
            across calls — one factory invocation per evaluation.
        score_fn: ``(output, example) -> float in [0, 1]``. May be sync
            or async.
        confidence_protocol: Optional :class:`ConfidenceProtocol` for
            self-rated calibration. When set, :attr:`MetricBundle.confidence`
            is filled via ``BaseAgent.run_enriched``.
        cost_estimator: Optional ``(RunUsage) -> float`` translating
            token usage to USD. When omitted, :attr:`MetricBundle.cost_usd`
            stays 0 (token counts still surface via :attr:`MetricBundle.raw`).
        timer: Monotonic clock for latency measurement. Override only
            in tests.
        n_trials_per_example: When ``> 1``, evaluate each example that many
            times and aggregate via :meth:`MetricBundle.aggregate` to wash
            out LLM run-to-run variance. The returned bundle carries
            ``n_trials`` and per-dimension ``stdev``. ``1`` (default)
            preserves the legacy single-shot behavior at no extra cost.
            Each trial gets a fresh agent (``agent_factory(decoded)``
            called per trial) so trials are independent. Cost scales
            linearly; useful for weaker models where single trials swing
            ±10pp.

    """
    if n_trials_per_example < 1:
        raise ValueError(
            f"n_trials_per_example must be >= 1, got {n_trials_per_example}"
        )
    self._agent_factory = agent_factory
    self._score_fn = score_fn
    self._confidence_protocol = confidence_protocol
    self._cost_estimator = cost_estimator or _default_cost_estimator
    self._timer = timer
    self._n_trials_per_example = n_trials_per_example
evaluate_one async 
evaluate_one(
    decoded: dict[str, Any],
    example: GoldExample[InputT, OutputT],
) -> MetricBundle

Run one example, return the per-example :class:MetricBundle.

Failures (agent exceptions, score_fn exceptions) are swallowed and produce a zero-accuracy bundle with the error captured in :attr:MetricBundle.raw["error"]. GEPA's adapter Protocol explicitly requires evaluators to never raise for individual example failures — see :class:gepa.core.adapter.GEPAAdapter docstring.

When n_trials_per_example > 1 (configured at construction), this method runs _evaluate_single_trial N times and aggregates via :meth:MetricBundle.aggregate. Failed trials are folded into the average (counted as zero-accuracy). When every trial fails, the aggregate is the first failure's error bundle plus an aggregation_note in raw.

Source code in orqest/optimization/evaluator.py 
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
async def evaluate_one(
    self,
    decoded: dict[str, Any],
    example: GoldExample[InputT, OutputT],
) -> MetricBundle:
    """Run one example, return the per-example :class:`MetricBundle`.

    Failures (agent exceptions, score_fn exceptions) are swallowed and
    produce a zero-accuracy bundle with the error captured in
    :attr:`MetricBundle.raw["error"]`. GEPA's adapter Protocol explicitly
    requires evaluators to **never raise** for individual example failures
    — see :class:`gepa.core.adapter.GEPAAdapter` docstring.

    When ``n_trials_per_example > 1`` (configured at construction), this
    method runs ``_evaluate_single_trial`` N times and aggregates via
    :meth:`MetricBundle.aggregate`. Failed trials are folded into the
    average (counted as zero-accuracy). When every trial fails, the
    aggregate is the first failure's error bundle plus an
    ``aggregation_note`` in ``raw``.
    """
    if self._n_trials_per_example == 1:
        return await self._evaluate_single_trial(decoded, example)

    trials: list[MetricBundle] = []
    for _ in range(self._n_trials_per_example):
        trials.append(await self._evaluate_single_trial(decoded, example))

    # If literally every trial failed, the user almost certainly wants to
    # see the first failure's error, not a misleadingly-averaged 0.0
    # accuracy. Pass through but annotate as multi-trial.
    if all(t.raw.get("error") for t in trials):
        head = trials[0]
        return head.model_copy(
            update={
                "raw": {
                    **head.raw,
                    "aggregation_note": (
                        f"all {len(trials)} trials failed; representative "
                        f"error from trial 1"
                    ),
                    "n_trials_attempted": len(trials),
                }
            }
        )
    return MetricBundle.aggregate(trials)
evaluate_batch async 
evaluate_batch(
    decoded: dict[str, Any],
    batch: list[GoldExample[InputT, OutputT]],
) -> list[MetricBundle]

Evaluate a batch sequentially. Caller controls concurrency at a higher layer (parallel evaluation across candidates is GEPA's job).

Source code in orqest/optimization/evaluator.py 
225
226
227
228
229
230
231
232
233
async def evaluate_batch(
    self,
    decoded: dict[str, Any],
    batch: list[GoldExample[InputT, OutputT]],
) -> list[MetricBundle]:
    """Evaluate a batch sequentially. Caller controls concurrency at a
    higher layer (parallel evaluation across candidates is GEPA's job).
    """
    return [await self.evaluate_one(decoded, ex) for ex in batch]

GoldExample

One labeled example for the optimizer's eval set.

Generic over input/output so consumers can pass typed Pydantic shapes without losing type information at the score-function boundary.

input instance-attribute 
input: InputT

Whatever the agent accepts. Often a string prompt; can be a richer Pydantic model when the consumer wants to pass structured context.

expected class-attribute instance-attribute 
expected: OutputT | None = None

Ground-truth output, when one exists. None is fine for examples where success is judged by the rubric (LLM-judge) rather than equality.

rubric class-attribute instance-attribute 
rubric: str | None = None

Free-text scoring guidance for an LLM-judge score_fn. Ignored by purely-programmatic score_fns.

weight class-attribute instance-attribute 
weight: float = 1.0

Per-example weight surfaced to the score function — useful for upweighting hard / adversarial examples in the gold set.

id class-attribute instance-attribute 
id: str | None = None

Optional stable identifier. Defaults to id(example) when needed.

CategoricalGene

A fixed-set string choice. Decode validates against choices.

Genome

An ordered collection of genes — the mutable surface of one optimization run.

The optimizer encodes the genome to GEPA's seed format via :meth:to_seed, GEPA proposes candidate mutations as the same dict[str, str] shape, and :meth:decode round-trips them back to typed Python values consumed by the consumer's agent_factory.

to_seed 
to_seed() -> dict[str, str]

Encode the genome into GEPA's seed_candidate format.

Source code in orqest/optimization/genome.py 
150
151
152
def to_seed(self) -> dict[str, str]:
    """Encode the genome into GEPA's ``seed_candidate`` format."""
    return {g.name: g.encode() for g in self.genes}
decode 
decode(candidate: dict[str, str]) -> dict[str, Any]

Decode a GEPA-proposed candidate back into typed values.

Resilient by design: missing or malformed entries fall back to the gene's initial rather than raising.

Source code in orqest/optimization/genome.py 
154
155
156
157
158
159
160
def decode(self, candidate: dict[str, str]) -> dict[str, Any]:
    """Decode a GEPA-proposed candidate back into typed values.

    Resilient by design: missing or malformed entries fall back to
    the gene's ``initial`` rather than raising.
    """
    return {g.name: g.decode(candidate.get(g.name)) for g in self.genes}
gene_kinds 
gene_kinds() -> set[str]

Set of gene kinds present — useful for the runner's gating check (enable_scalar_genes / enable_categorical_genes).

Source code in orqest/optimization/genome.py 
162
163
164
165
166
def gene_kinds(self) -> set[str]:
    """Set of gene kinds present — useful for the runner's gating
    check (``enable_scalar_genes`` / ``enable_categorical_genes``).
    """
    return {g.kind for g in self.genes}

PromptGene

A prompt-string slot — the canonical GEPA target.

name instance-attribute 
name: str

Logical slot identifier, e.g. "researcher.system_prompt". Must match the key the consumer hands to :func:apply_result (or to a custom agent_factory).

initial instance-attribute 
initial: str

Seed prompt text. Becomes the W0 candidate GEPA mutates from.

constraints class-attribute instance-attribute 
constraints: str | None = None

Optional natural-language guardrail surfaced to the reflection LLM in :meth:OrqestGEPAAdapter.make_reflective_dataset. Use it to encode invariants the optimizer must not break (e.g. "Must abstain on ambiguous goals").

ScalarGene

A bounded float gene. Decode clamps + optional quantization.

quantize class-attribute instance-attribute 
quantize: float | None = None

When set, decode snaps to the nearest multiple of this grid step.

Archive 

Archive(strategy: ArchiveStrategy = 'top_k', size: int = 5)

Strategy-pluggable archive of evaluated candidates.

All entries are retained in storage (so :meth:pareto and :meth:best see the full history); serialize_for_prompt is the strategy-aware view shown to the meta agent on the next design step.

Source code in orqest/optimization/meta_agent.py 
169
170
171
172
173
174
175
176
177
178
def __init__(
    self, strategy: ArchiveStrategy = "top_k", size: int = 5
) -> None:
    if strategy not in ("cumulative", "top_k", "parallel"):
        raise ValueError(f"unknown strategy {strategy!r}")
    if size < 1:
        raise ValueError("size must be >= 1")
    self._strategy = strategy
    self._size = size
    self._entries: list[ArchiveEntry] = []
entries property 
entries: list[ArchiveEntry]

All entries in insertion order. Read-only view.

add 
add(entry: ArchiveEntry) -> int

Append entry and return its archive index.

Source code in orqest/optimization/meta_agent.py 
180
181
182
183
def add(self, entry: ArchiveEntry) -> int:
    """Append *entry* and return its archive index."""
    self._entries.append(entry)
    return len(self._entries) - 1
serialize_for_prompt 
serialize_for_prompt() -> str

The archive view passed to the meta agent for its next design.

Strategy-dependent:

  • "cumulative" — every entry, JSON-serialized.
  • "top_k" — top size entries by aggregate_score, JSON-serialized.
  • "parallel" — empty string. Meta agent designs without context.
Source code in orqest/optimization/meta_agent.py 
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
def serialize_for_prompt(self) -> str:
    """The archive view passed to the meta agent for its next design.

    Strategy-dependent:

    * ``"cumulative"`` — every entry, JSON-serialized.
    * ``"top_k"`` — top ``size`` entries by aggregate_score, JSON-serialized.
    * ``"parallel"`` — empty string. Meta agent designs without context.
    """
    if self._strategy == "parallel":
        return ""
    ranked = sorted(self._entries, key=lambda e: -e.aggregate_score)
    if self._strategy == "top_k":
        ranked = ranked[: self._size]
    return ",\n".join(
        json.dumps(
            {
                "generation": e.generation,
                "thought": e.thought[:500],
                "spec": json.loads(e.spec_json),
                "score": round(e.aggregate_score, 4),
            }
        )
        for e in ranked
    )
best 
best() -> ArchiveEntry

The highest-aggregate-score entry. Raises if empty.

Source code in orqest/optimization/meta_agent.py 
216
217
218
219
220
def best(self) -> ArchiveEntry:
    """The highest-aggregate-score entry. Raises if empty."""
    if not self._entries:
        raise ValueError("Archive is empty; call .add() first")
    return max(self._entries, key=lambda e: e.aggregate_score)
pareto 
pareto() -> list[ArchiveEntry]

Distinct entries on the accuracy / cost / latency Pareto front.

An entry is on the front if no other entry dominates it (strictly better on every axis). We negate cost and latency so "more is better" applies uniformly.

Source code in orqest/optimization/meta_agent.py 
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
def pareto(self) -> list[ArchiveEntry]:
    """Distinct entries on the accuracy / cost / latency Pareto front.

    An entry is on the front if no other entry dominates it (strictly
    better on every axis). We negate cost and latency so "more is
    better" applies uniformly.
    """
    if not self._entries:
        return []

    def axes(e: ArchiveEntry) -> tuple[float, float, float]:
        mean_acc = sum(b.accuracy for b in e.bundles) / max(1, len(e.bundles))
        mean_cost = sum(b.cost_usd for b in e.bundles) / max(1, len(e.bundles))
        mean_lat = sum(b.latency_ms for b in e.bundles) / max(1, len(e.bundles))
        return (mean_acc, -mean_cost, -mean_lat)

    front: list[ArchiveEntry] = []
    for e in self._entries:
        ea = axes(e)
        dominated = False
        for other in self._entries:
            if other is e:
                continue
            oa = axes(other)
            # other dominates e if it's >= on every axis and > on at least one
            if all(o >= a for o, a in zip(oa, ea, strict=False)) and any(
                o > a for o, a in zip(oa, ea, strict=False)
            ):
                dominated = True
                break
        if not dominated:
            front.append(e)
    return front

ArchiveEntry

One archived candidate.

generation instance-attribute 
generation: int

Generation index. -1 means the seed (pre-search baseline).

spec_json instance-attribute 
spec_json: str

The TopologySpec serialized to JSON — the GEPA wire format.

bundles instance-attribute 
bundles: list[MetricBundle]

Per-example MetricBundles from this entry's evaluation.

aggregate_score instance-attribute 
aggregate_score: float

Mean scalarized score across bundles — the archive ranking key.

thought class-attribute instance-attribute 
thought: str = ''

Meta-agent's design reasoning. Empty for the seed.

parent_idx class-attribute instance-attribute 
parent_idx: int | None = None

Index of the archive entry whose design inspired this one. None for the seed and for parallel-strategy entries (no parent context).

MetaAgentConfig dataclass 

MetaAgentConfig(
    n_generations: int = 10,
    archive_strategy: ArchiveStrategy = "top_k",
    archive_size: int = 5,
    reflexion_passes: int = 2,
    debug_max: int = 3,
    minibatch_size: int = 5,
    seed: int = 42,
    weights: MetricWeights = MetricWeights(),
)

Knobs for :class:MetaAgentSearch.

Distinct from :class:OptimizationConfig (which governs GEPA's prompt evolution loop) because the search-loop parameters are different in kind — generations of designed-and-evaluated topologies, not GEPA's sampled minibatches.

n_generations class-attribute instance-attribute 
n_generations: int = 10

Number of design-evaluate cycles after the seed evaluation.

archive_strategy class-attribute instance-attribute 
archive_strategy: ArchiveStrategy = 'top_k'

How to surface the archive to the next design step.

  • "top_k" (default) — show only the top archive_size by aggregate score. Per the 2510.06711 critique, the strongest signal-to- noise ratio for the meta agent.
  • "cumulative" — show every entry seen so far. ADAS-paper-faithful; tends to overflow context after ~20 entries and dilute strong signals.
  • "parallel" — show nothing. Meta agent designs from scratch each generation; selection happens at the end. Surprisingly competitive per the critique.
archive_size class-attribute instance-attribute 
archive_size: int = 5

Cap for top_k and the ring-buffer for cumulative if you want one (unlimited by default for cumulative). Ignored for parallel.

reflexion_passes class-attribute instance-attribute 
reflexion_passes: int = 2

Number of reflexion-revision passes after the design step. ADAS uses 2 by default; 0 disables reflexion entirely (raw design only).

debug_max class-attribute instance-attribute 
debug_max: int = 3

How many times to retry a generation that fails Pydantic validation or hydration. The ValidationError surface is fed back to the meta agent as feedback (analogue of ADAS's traceback retry).

minibatch_size class-attribute instance-attribute 
minibatch_size: int = 5

Validation subset size per candidate evaluation. Caps cost: each candidate evaluates against this many examples (sampled deterministically from the valset), not the full set. The seed and final winner are re-evaluated against the full set.

seed class-attribute instance-attribute 
seed: int = 42

Random seed for minibatch sampling and reproducibility.

weights class-attribute instance-attribute 
weights: MetricWeights = field(
    default_factory=MetricWeights
)

Aggregation weights for converting per-example MetricBundles to a scalar score for archive ranking. Defaults match GEPA's recommended weights so users see consistent ranking across optimizers.

MetaAgentSearch 

MetaAgentSearch(
    config: MetaAgentConfig,
    *,
    gene: TopologyGene,
    evaluator: TopologyEvaluator,
    meta_agent_model: str,
    api_key: str,
    bus: EventBus | None = None,
)

Run an ADAS-style search over :data:TopologySpec candidates.

Three-stage generation: design → reflexion (×N) → evaluate (with debug retry). Pydantic ValidationError + hydration KeyError are caught and fed back to the meta agent as debug feedback. Non-debuggable failures (after debug_max retries) skip the generation; the search continues.

Wire the search.

Parameters:

Name Type Description Default
config MetaAgentConfig

Frozen :class:MetaAgentConfig.

 required
gene TopologyGene

The :class:TopologyGene being evolved. Carries the seed, the constraints, and the allowed-step-kinds whitelist that shapes the meta-agent prompt.

 required
evaluator TopologyEvaluator

A :class:TopologyEvaluator configured with the user's callable + agent registries and score function.

 required
meta_agent_model str

provider:model_id string for the meta agent. Use the strongest model you can afford — design quality scales directly with this knob.

 required
api_key str

API key for the meta agent's model.

 required
bus EventBus | None

Optional :class:EventBus. When provided, the loop emits meta_agent.iteration_completed per generation and meta_agent.debug_retry on Pydantic-error retries.

 None
Source code in orqest/optimization/meta_agent.py 
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
def __init__(
    self,
    config: MetaAgentConfig,
    *,
    gene: TopologyGene,
    evaluator: TopologyEvaluator,
    meta_agent_model: str,
    api_key: str,
    bus: EventBus | None = None,
) -> None:
    """Wire the search.

    Args:
        config: Frozen :class:`MetaAgentConfig`.
        gene: The :class:`TopologyGene` being evolved. Carries the seed,
            the constraints, and the allowed-step-kinds whitelist that
            shapes the meta-agent prompt.
        evaluator: A :class:`TopologyEvaluator` configured with the
            user's callable + agent registries and score function.
        meta_agent_model: ``provider:model_id`` string for the meta agent.
            Use the strongest model you can afford — design quality
            scales directly with this knob.
        api_key: API key for the meta agent's model.
        bus: Optional :class:`EventBus`. When provided, the loop emits
            ``meta_agent.iteration_completed`` per generation and
            ``meta_agent.debug_retry`` on Pydantic-error retries.

    """
    self._config = config
    self._gene = gene
    self._evaluator = evaluator
    self._bus = bus
    self._rng = random.Random(config.seed)

    # Meta agent — a pydantic-ai Agent with structured output.
    # We don't go through orqest BaseAgent here because BaseAgent expects
    # StateT/OutputT to thread through call_model with state; we want a
    # one-shot structured-output call per design step.
    meta_model = resolve_model(meta_agent_model, api_key=api_key)
    self._meta_agent = PydanticAIAgent(
        model=meta_model,
        system_prompt=_DESIGN_SYSTEM,
        output_type=TopologyDesign,
    )
run async 
run(
    trainset: list[GoldExample[Any, Any]],
    valset: list[GoldExample[Any, Any]] | None = None,
) -> OptimizationResult

Run the search; return an :class:OptimizationResult.

trainset is currently used only to size the minibatch sampler; the actual evaluation runs against valset (or against trainset if no valset is provided). The asymmetry mirrors GEPA's setup but is less load-bearing here — topology evaluation doesn't have a train/val distinction in the gradient sense.

Source code in orqest/optimization/meta_agent.py 
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
async def run(
    self,
    trainset: list[GoldExample[Any, Any]],
    valset: list[GoldExample[Any, Any]] | None = None,
) -> OptimizationResult:
    """Run the search; return an :class:`OptimizationResult`.

    ``trainset`` is currently used only to size the minibatch sampler;
    the actual evaluation runs against ``valset`` (or against ``trainset``
    if no valset is provided). The asymmetry mirrors GEPA's setup but
    is less load-bearing here — topology evaluation doesn't have a
    train/val distinction in the gradient sense.
    """
    if not trainset:
        raise ValueError("trainset must contain at least 1 example")
    valset = valset if valset else trainset

    archive = Archive(self._config.archive_strategy, self._config.archive_size)

    # Seed evaluation: full valset, baseline reference for the user.
    seed_bundles = await self._evaluator.evaluate_batch(
        decoded={self._gene.name: self._gene.initial},
        batch=valset,
    )
    seed_score = _aggregate(seed_bundles, self._config.weights)
    archive.add(
        ArchiveEntry(
            generation=-1,
            spec_json=self._gene.encode(),
            bundles=seed_bundles,
            aggregate_score=seed_score,
            thought="(seed)",
            parent_idx=None,
        )
    )
    self._emit_iteration("seed", -1, seed_score, len(archive))

    history: list[dict[str, Any]] = []
    history.append(
        {
            "generation": -1,
            "score": seed_score,
            "skipped": False,
            "thought": "(seed)",
        }
    )

    for gen in range(self._config.n_generations):
        entry, hist_record = await self._one_generation(
            gen, valset, archive
        )
        history.append(hist_record)
        if entry is not None:
            idx = archive.add(entry)
            self._emit_iteration(
                "completed",
                gen,
                entry.aggregate_score,
                len(archive),
                archive_idx=idx,
            )

    return self._build_result(archive, history)

TopologyDesign

Structured output for the meta agent.

The meta agent's pydantic-ai :class:Agent returns this shape every design / reflexion / debug step.

thought instance-attribute 
thought: str

Free-text reasoning — quoted into the archive entry for inspection.

spec instance-attribute 
spec: TopologySpec

The proposed topology. Pydantic-validated; malformed candidates fail fast with a ValidationError that the debug-retry loop surfaces back to the meta agent.

OptimizationResult dataclass 

OptimizationResult(
    best_candidate: dict[str, str],
    best_decoded: dict[str, Any],
    best_score: float,
    pareto_candidates: list[dict[str, str]],
    history: list[dict[str, Any]] = list(),
    raw: Any = None,
)

Immutable summary of a completed optimization run.

best_candidate instance-attribute 
best_candidate: dict[str, str]

The winning candidate (per GEPA's best_idx aggregate score).

best_decoded instance-attribute 
best_decoded: dict[str, Any]

:meth:Genome.decode applied to best_candidate.

best_score instance-attribute 
best_score: float

The best candidate's aggregate validation score.

pareto_candidates instance-attribute 
pareto_candidates: list[dict[str, str]]

Distinct candidates appearing on GEPA's instance-level Pareto front (per_val_instance_best_candidates). Inspect for tradeoffs the aggregate winner doesn't surface.

history class-attribute instance-attribute 
history: list[dict[str, Any]] = field(default_factory=list)

Per-iteration summary (iteration index, mean score, mean accuracy) captured from the bus during the run, when a bus was provided.

raw class-attribute instance-attribute 
raw: Any = None

The original :class:gepa.core.result.GEPAResult — escape hatch for advanced inspection (lineage, val_subscores, per_objective_*).

OptimizationRunner 

OptimizationRunner(
    config: OptimizationConfig,
    *,
    genome: Genome,
    evaluator: Evaluator[Any, Any],
    bus: EventBus | None = None,
    api_key: str | None = None,
)

Drive gepa.optimize() against an Orqest :class:Evaluator.

Wire the optimizer.

Parameters:

Name Type Description Default
config OptimizationConfig

Frozen :class:OptimizationConfig (rollout budget, reflection model, weights, etc.).

 required
genome Genome

The mutable surface — list of typed genes.

 required
evaluator Evaluator[Any, Any]

Wraps the user's agent_factory + score_fn and produces :class:MetricBundles per example.

 required
bus EventBus | None

Optional :class:EventBus for optimization.iteration_completed events; when set, history is collected into the result.

 None
api_key str | None

Optional API key for the reflection model. Bridged to litellm's expected provider env var (OPENAI_API_KEY, ANTHROPIC_API_KEY, ...) via os.environ.setdefault — pre-existing env vars always win. Required when GEPA's default reflection path is used and no env var is set; ignored when irrelevant.

 None
Source code in orqest/optimization/runner.py 
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
def __init__(
    self,
    config: OptimizationConfig,
    *,
    genome: Genome,
    evaluator: Evaluator[Any, Any],
    bus: EventBus | None = None,
    api_key: str | None = None,
) -> None:
    """Wire the optimizer.

    Args:
        config: Frozen :class:`OptimizationConfig` (rollout budget,
            reflection model, weights, etc.).
        genome: The mutable surface — list of typed genes.
        evaluator: Wraps the user's ``agent_factory`` + ``score_fn``
            and produces :class:`MetricBundle`s per example.
        bus: Optional :class:`EventBus` for ``optimization.iteration_completed``
            events; when set, history is collected into the result.
        api_key: Optional API key for the *reflection* model. Bridged
            to litellm's expected provider env var (``OPENAI_API_KEY``,
            ``ANTHROPIC_API_KEY``, ...) via ``os.environ.setdefault``
            — pre-existing env vars always win. Required when GEPA's
            default reflection path is used and no env var is set;
            ignored when irrelevant.
    """
    self._config = config
    self._genome = genome
    self._evaluator = evaluator
    self._bus = bus
    self._api_key = api_key
    self._validate_genome()
optimize async 
optimize(
    trainset: list[GoldExample[Any, Any]],
    valset: list[GoldExample[Any, Any]] | None = None,
) -> OptimizationResult

Run GEPA against the given gold sets.

When valset is None, splits trainset deterministically using config.seed and config.valset_fraction.

Source code in orqest/optimization/runner.py 
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
async def optimize(
    self,
    trainset: list[GoldExample[Any, Any]],
    valset: list[GoldExample[Any, Any]] | None = None,
) -> OptimizationResult:
    """Run GEPA against the given gold sets.

    When ``valset`` is None, splits ``trainset`` deterministically using
    ``config.seed`` and ``config.valset_fraction``.
    """
    if valset is None:
        trainset, valset = self._split(trainset)

    adapter = OrqestGEPAAdapter(
        self._genome,
        self._evaluator,
        self._config.weights,
        bus=self._bus,
        emit_per_example_events=self._config.emit_per_example_events,
    )

    history = self._wire_history_collector()

    # Belt-and-suspenders env-var setdefault for any internal GEPA
    # path we don't directly intercept. Primary api_key delivery
    # happens via the explicit callable below.
    _ensure_litellm_api_key(self._config.reflection_model, self._api_key)

    # Build the reflection LM as an explicit callable when we have an
    # api_key — bypasses env vars entirely. Falls back to the
    # litellm-formatted string when no api_key is provided (GEPA then
    # builds its own internal callable that reads env vars).
    reflection_lm: Any = _make_reflection_lm(
        self._config.reflection_model, self._api_key
    )
    if reflection_lm is None:
        reflection_lm = _to_litellm_model_string(self._config.reflection_model)

    seed = self._genome.to_seed()
    gepa_result = _gepa_optimize(
        seed_candidate=seed,
        trainset=trainset,
        valset=valset,
        adapter=adapter,
        # No task_lm / evaluator here: the adapter owns both. GEPA's
        # api.py asserts task_lm is None when an adapter is provided,
        # because the user's agent_factory (inside the adapter's
        # Evaluator) is what actually calls the task model.
        reflection_lm=reflection_lm,
        max_metric_calls=self._config.max_metric_calls,
        frontier_type=self._config.frontier_type,
        cache_evaluation=self._config.cache_evaluations,
        seed=self._config.seed if self._config.seed is not None else 0,
        display_progress_bar=False,
        raise_on_exception=False,
    )

    best_idx = gepa_result.best_idx
    best_candidate: dict[str, str] = gepa_result.candidates[best_idx]
    best_decoded = self._genome.decode(best_candidate)
    best_score = float(gepa_result.val_aggregate_scores[best_idx])

    pareto_candidates = self._extract_pareto(gepa_result)

    return OptimizationResult(
        best_candidate=best_candidate,
        best_decoded=best_decoded,
        best_score=best_score,
        pareto_candidates=pareto_candidates,
        history=history,
        raw=gepa_result,
    )

TopologyEvaluator 

TopologyEvaluator(
    *,
    score_fn: Callable[
        [Any, GoldExample[Any, Any]],
        float | Awaitable[float],
    ],
    callable_registry: CallableRegistry,
    agent_registry: dict[
        str, Callable[[], BaseAgent[Any, Any]]
    ],
    topology_gene_name: str = "main",
    agent_factory: AgentFactory | None = None,
    confidence_protocol: ConfidenceProtocol | None = None,
    cost_estimator: Callable[[Any], float] | None = None,
    timer: Callable[[], float] = time.monotonic,
)

Score a topology candidate against gold examples.

Replaces :class:Evaluator's single-BaseAgent execution path with topology hydration. Adds n_agents and depth structural metrics to :attr:MetricBundle.raw so the optimizer's Pareto front can prefer smaller / shallower topologies on accuracy ties.

Cost handling: Unlike single-agent evaluation, a topology run does not have a single Usage object. We surface cost_usd=0.0 by default; consumers wanting cost-as-fitness should pass a cost_estimator callable that walks the topology and sums per-agent usage (see the concept doc for the recipe). This is a known limitation documented in the W3 risks section.

Wire the per-evaluation building blocks.

Parameters:

Name Type Description Default
score_fn Callable[[Any, GoldExample[Any, Any]], float | Awaitable[float]]

(output, example) -> float in [0, 1]. Identical contract to the parent :class:Evaluator.

 required
callable_registry CallableRegistry

Registry of named conditions / merges / state-updaters / function-steps. The meta agent's allowlist.

 required
agent_registry dict[str, Callable[[], BaseAgent[Any, Any]]]

Map from agent name to a factory producing a fresh :class:BaseAgent. Factories (not instances) avoid cached-_agent problems when the same agent appears in multiple candidates across a search.

 required
topology_gene_name str

Key under which the decoded genome carries the :data:TopologySpec. Default "main".

 'main'
agent_factory AgentFactory | None

Optional :class:AgentFactory for hydrating :class:AgentSpec inline-spawn references inside topology specs. Required only when topology candidates use inline_spec rather than agent_name.

 None
confidence_protocol ConfidenceProtocol | None

Reserved for future use — topologies don't produce confidence directly today (would require per-step EnrichedOutput plumbing); accepted for signature parity.

 None
cost_estimator Callable[[Any], float] | None

Optional callable for translating downstream usage into USD. See class docstring caveat.

 None
timer Callable[[], float]

Monotonic clock; override only in tests.

 monotonic
Source code in orqest/optimization/topology.py 
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
def __init__(
    self,
    *,
    score_fn: Callable[
        [Any, GoldExample[Any, Any]],
        float | Awaitable[float],
    ],
    callable_registry: CallableRegistry,
    agent_registry: dict[str, Callable[[], BaseAgent[Any, Any]]],
    topology_gene_name: str = "main",
    agent_factory: AgentFactory | None = None,
    confidence_protocol: ConfidenceProtocol | None = None,
    cost_estimator: Callable[[Any], float] | None = None,
    timer: Callable[[], float] = time.monotonic,
) -> None:
    """Wire the per-evaluation building blocks.

    Args:
        score_fn: ``(output, example) -> float in [0, 1]``. Identical
            contract to the parent :class:`Evaluator`.
        callable_registry: Registry of named conditions / merges /
            state-updaters / function-steps. The meta agent's allowlist.
        agent_registry: Map from agent name to a factory producing a
            fresh :class:`BaseAgent`. Factories (not instances) avoid
            cached-``_agent`` problems when the same agent appears in
            multiple candidates across a search.
        topology_gene_name: Key under which the decoded genome carries
            the :data:`TopologySpec`. Default ``"main"``.
        agent_factory: Optional :class:`AgentFactory` for hydrating
            :class:`AgentSpec` inline-spawn references inside topology
            specs. Required only when topology candidates use
            ``inline_spec`` rather than ``agent_name``.
        confidence_protocol: Reserved for future use — topologies don't
            produce confidence directly today (would require per-step
            EnrichedOutput plumbing); accepted for signature parity.
        cost_estimator: Optional callable for translating downstream
            usage into USD. See class docstring caveat.
        timer: Monotonic clock; override only in tests.

    """
    super().__init__(
        agent_factory=_no_op_factory,
        score_fn=score_fn,
        confidence_protocol=confidence_protocol,
        cost_estimator=cost_estimator,
        timer=timer,
    )
    self._callable_registry = callable_registry
    self._agent_registry = agent_registry
    self._spawn_factory = agent_factory
    self._topology_gene_name = topology_gene_name
evaluate_one async 
evaluate_one(
    decoded: dict[str, Any],
    example: GoldExample[InputT, OutputT],
) -> MetricBundle

Hydrate the topology, run on the example, return a MetricBundle.

Failures (hydration errors, runtime errors, score errors) yield a zero-accuracy bundle with the failure captured in :attr:MetricBundle.raw["error"]. This matches the parent's never-raise contract — GEPA's adapter Protocol forbids raising on per-example evaluation failures.

Source code in orqest/optimization/topology.py 
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
async def evaluate_one(
    self,
    decoded: dict[str, Any],
    example: GoldExample[InputT, OutputT],
) -> MetricBundle:
    """Hydrate the topology, run on the example, return a MetricBundle.

    Failures (hydration errors, runtime errors, score errors) yield a
    zero-accuracy bundle with the failure captured in
    :attr:`MetricBundle.raw["error"]`. This matches the parent's
    never-raise contract — GEPA's adapter Protocol forbids raising on
    per-example evaluation failures.
    """
    start = self._timer()
    spec_raw = decoded.get(self._topology_gene_name)

    # Track structural metrics even on failure paths so the optimizer
    # sees the topology's shape regardless of whether it ran.
    n_agents = (
        _count_agent_steps(spec_raw)
        if spec_raw is not None and not isinstance(spec_raw, str)
        else 0
    )
    depth = (
        _topology_depth(spec_raw)
        if spec_raw is not None and not isinstance(spec_raw, str)
        else 0
    )

    if spec_raw is None:
        elapsed_ms = (self._timer() - start) * 1000.0
        return MetricBundle(
            accuracy=0.0,
            latency_ms=elapsed_ms,
            raw={
                "error": (
                    f"decoded genome missing topology gene "
                    f"{self._topology_gene_name!r}"
                ),
                "error_type": "MissingGene",
                "n_agents": 0,
                "depth": 0,
            },
        )

    try:
        topology = topology_from_spec(
            spec_raw,
            callable_registry=self._callable_registry,
            agent_registry=self._agent_registry,
            agent_factory=self._spawn_factory,
        )
        run_result = await topology.run(example.input)
        output = unpack_topology_output(run_result)
        score_value = await self._await_if_needed(
            self._score_fn(output, example)
        )
        accuracy = max(0.0, min(1.0, float(score_value) * example.weight))
    except Exception as exc:  # noqa: BLE001
        elapsed_ms = (self._timer() - start) * 1000.0
        return MetricBundle(
            accuracy=0.0,
            latency_ms=elapsed_ms,
            raw={
                "error": str(exc),
                "error_type": type(exc).__name__,
                "n_agents": n_agents,
                "depth": depth,
            },
        )

    elapsed_ms = (self._timer() - start) * 1000.0
    cost_usd = 0.0  # see class docstring re: cost limitation

    return MetricBundle(
        accuracy=accuracy,
        cost_usd=cost_usd,
        latency_ms=elapsed_ms,
        raw={
            "n_agents": n_agents,
            "depth": depth,
        },
    )

TopologyGene

A topology-shaped gene whose initial value is a :data:TopologySpec.

Encodes to JSON (the GEPA wire format is dict[str, str]) and decodes back via Pydantic validation. Resilient on malformed reflection output: a candidate that fails JSON parsing or schema validation falls back to the gene's initial rather than raising — same defensive posture as :class:PromptGene (we'd rather lose one iteration than crash a search).

name instance-attribute 
name: str

Logical slot identifier — the key the meta agent emits and the :class:TopologyEvaluator reads from the decoded genome.

initial instance-attribute 
initial: TopologySpec

Seed topology. The meta-agent's W0 candidate; also the resilient fallback for malformed reflection output.

constraints class-attribute instance-attribute 
constraints: str | None = None

Optional natural-language guardrail surfaced to the meta agent in its design prompt (e.g. "must include the existing classifier as a Pipeline first step").

allowed_step_kinds class-attribute instance-attribute 
allowed_step_kinds: tuple[str, ...] = (
    "agent_step",
    "function_step",
)

Whitelisted leaf-step kinds. Surfaced to the meta agent in its prompt as the only legal atomic operations. Default permits both.

max_depth class-attribute instance-attribute 
max_depth: int = 4

Maximum nesting depth the meta agent is allowed to emit. Past this depth the evaluator rejects the candidate (max_depth_exceeded error) so the optimizer pivots away from runaway-recursion designs.

encode 
encode() -> str

Serialize the seed topology to JSON for GEPA's wire format.

Source code in orqest/optimization/topology.py 
91
92
93
def encode(self) -> str:
    """Serialize the seed topology to JSON for GEPA's wire format."""
    return self.initial.model_dump_json()
decode 
decode(candidate: str | None) -> TopologySpec

Decode a meta-agent-proposed JSON string back to a TopologySpec.

Three failure modes, all resilient:

  • None (key missing from the candidate dict) → initial
  • Malformed JSON → initial
  • Valid JSON but failing TopologySpec schema → initial
Source code in orqest/optimization/topology.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def decode(self, candidate: str | None) -> TopologySpec:
    """Decode a meta-agent-proposed JSON string back to a TopologySpec.

    Three failure modes, all resilient:

    * ``None`` (key missing from the candidate dict) → ``initial``
    * Malformed JSON → ``initial``
    * Valid JSON but failing TopologySpec schema → ``initial``
    """
    if candidate is None:
        return self.initial
    try:
        return _TOPOLOGY_ADAPTER.validate_json(candidate)
    except (ValidationError, json.JSONDecodeError, ValueError):
        return self.initial

apply_result 

apply_result(
    result: OptimizationResult,
    *,
    target: Any,
    dry_run: bool = True,
) -> list[OptimizationDiff]

Build per-gene diffs against target; write them when not dry-run.

Parameters:

Name Type Description Default
result OptimizationResult

The :class:OptimizationResult from :meth:OptimizationRunner.optimize.

 required
target Any

A :class:BaseAgent or a plain dict. For agents, the evolved values are written via setattr and the cached _agent is invalidated. For dicts, evolved values are written by key.

 required
dry_run bool

When True (default), no mutation happens — the diffs are returned for inspection only. Flip to False to commit.

 True

Returns:

Name Type Description
One list[OptimizationDiff]

class:OptimizationDiff per gene in result.best_decoded,
list[OptimizationDiff]

in iteration order. Diffs with before == after are still returned
list[OptimizationDiff]

(with unified="") so the caller can confirm "no change" explicitly
list[OptimizationDiff]

rather than inferring from absence.
Source code in orqest/optimization/apply.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
def apply_result(
    result: OptimizationResult,
    *,
    target: Any,
    dry_run: bool = True,
) -> list[OptimizationDiff]:
    """Build per-gene diffs against ``target``; write them when not dry-run.

    Args:
        result: The :class:`OptimizationResult` from
            :meth:`OptimizationRunner.optimize`.
        target: A :class:`BaseAgent` or a plain ``dict``. For agents, the
            evolved values are written via ``setattr`` and the cached
            ``_agent`` is invalidated. For dicts, evolved values are written
            by key.
        dry_run: When True (default), no mutation happens — the diffs are
            returned for inspection only. Flip to False to commit.

    Returns:
        One :class:`OptimizationDiff` per gene in ``result.best_decoded``,
        in iteration order. Diffs with ``before == after`` are still returned
        (with ``unified=""``) so the caller can confirm "no change" explicitly
        rather than inferring from absence.

    """
    diffs: list[OptimizationDiff] = []
    for name, after_value in result.best_decoded.items():
        before_value = _read_current(target, name)
        diff = _build_diff(name, before_value, after_value)
        diffs.append(diff)
        if not dry_run and diff.changed:
            _write_value(target, name, after_value)
    return diffs

unpack_topology_output 

unpack_topology_output(run_result: Any) -> Any

Extract the meaningful payload from a topology's .run() result.

  • :class:ParallelResult → .merged (the merge-strategy output)
  • :class:LoopResult → .output (the converged value)
  • :class:AgentRunResult (if a topology somehow returns one) → .output
  • Anything else → identity (Pipeline / Router return their final value)
Source code in orqest/optimization/topology.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
def unpack_topology_output(run_result: Any) -> Any:
    """Extract the meaningful payload from a topology's ``.run()`` result.

    * :class:`ParallelResult` → ``.merged`` (the merge-strategy output)
    * :class:`LoopResult` → ``.output`` (the converged value)
    * :class:`AgentRunResult` (if a topology somehow returns one) → ``.output``
    * Anything else → identity (Pipeline / Router return their final value)
    """
    if isinstance(run_result, ParallelResult):
        return run_result.merged
    if isinstance(run_result, LoopResult):
        return run_result.output
    if hasattr(run_result, "output"):
        return run_result.output
    return run_result