Topology Optimization API

spec

Serializable IR for Orqest's orchestration primitives.

Closes the gap that an LLM cannot emit a composition topology at runtime because Pipeline / Parallel / Router / RefinementLoop accept Python callables (conditions, merges, state-updaters) that have no JSON representation.

This module provides Pydantic v2 models that round-trip cleanly to JSON and hydrate back to live runtime objects via :mod:orqest.orchestration.hydrate.

Design:

• One discriminated union (:data:OperationSpec) covers every shape that can appear in a step position — atomic (AgentStepSpec / FunctionStepSpec) and composite (PipelineSpec / ParallelSpec / RouterSpec / RefinementLoopSpec). The single union keeps recursion uniform: a Pipeline's step list, a Route's step, and a RefinementLoop's step all accept the same type.
• All callables go through name registries. Conditions, merges, state updaters, and FunctionStep functions are referenced by string name; the hydrator resolves them against a :class:CallableRegistry. Agents are referenced either by registry name (string) or inlined as an :class:AgentSpec for spawn-fresh-per-execution. This is the load-bearing safety property: the meta agent's emission surface is names from an allowlist, never code.
• TopologySpec is the narrower top-level union (composites only) — used by :class:TopologyGene to type the genome's seed.

OperationSpec module-attribute

OperationSpec = Annotated[
    "AgentStepSpec | FunctionStepSpec | PipelineSpec | ParallelSpec | RouterSpec | RefinementLoopSpec",
    Field(discriminator="kind"),
]

Any composable operation. Both atomic (AgentStep / FunctionStep) and composite (Pipeline / Parallel / Router / RefinementLoop) shapes share this union so recursion is uniform: a Pipeline's step entry, a Route's step, and a RefinementLoop's step all accept OperationSpec.

The string-form annotation defers Pydantic's discriminator resolution until all four composite classes are defined (forward references).

TopologySpec module-attribute

TopologySpec = Annotated[
    PipelineSpec
    | ParallelSpec
    | RouterSpec
    | RefinementLoopSpec,
    Field(discriminator="kind"),
]

The narrower top-level union — composite topologies only. Used by :class:TopologyGene to type the genome seed: a meta-agent-emitted candidate must be one of the four composite shapes (an atomic AgentStep alone is not a "topology" worth evolving).

StepConfigSpec

Per-step error-handling config used inside :class:PipelineSpec.

Mirrors :class:orqest.orchestration.types.StepConfig but with a string on_error literal instead of the runtime ErrorStrategy Enum (so it serializes cleanly to JSON).

AgentStepSpec

An atomic step that wraps a single :class:BaseAgent.

The agent is resolved by agent_name against the hydrator's agent registry, OR constructed fresh from inline_spec (an :class:AgentSpec). Exactly one of the two must be present — the validator enforces this.

agent_name class-attribute instance-attribute

agent_name: str | None = None

Lookup key in the agent_registry. Mutually exclusive with inline_spec.

inline_spec class-attribute instance-attribute

inline_spec: AgentSpec | None = None

When set, the agent is spawned fresh from this spec via :class:AgentFactory. Mutually exclusive with agent_name.

FunctionStepSpec

An atomic step that wraps a registered async callable.

callable_name instance-attribute

callable_name: str

Lookup key in the callable_registry. The hydrator resolves this name to a callable; unknown names raise :class:KeyError at hydration time.

name class-attribute instance-attribute

name: str | None = None

Optional display name. Falls back to callable_name.

PipelineStepEntry

One entry inside a :class:PipelineSpec's steps list.

Wraps an :data:OperationSpec with optional per-step :class:StepConfigSpec. Keeping the entry shape uniform makes JSON emission predictable for the meta agent (no mixed-list-shape footgun).

PipelineSpec

Sequential composition: each entry's output feeds the next entry's input.

Hydrates to :class:orqest.orchestration.pipeline.Pipeline.

steps class-attribute instance-attribute

steps: list[PipelineStepEntry] = Field(min_length=1)

At least one step is required. Empty pipelines raise at hydration (mirroring :class:Pipeline's own ValueError).

ParallelSpec

Concurrent composition: all entries run on input_data, results merge.

Hydrates to :class:orqest.orchestration.parallel.Parallel.

merge class-attribute instance-attribute

merge: str = 'collect_all'

Merge strategy name. "collect_all" and "first_wins" resolve to :class:orqest.orchestration.parallel.MergeStrategy built-ins; any other value is looked up in the hydrator's callable_registry.

timeout class-attribute instance-attribute

timeout: float | None = None

Wall-clock timeout in seconds. None means no limit.

RouteSpec

One named branch in a :class:RouterSpec.

The step may be an atomic (AgentStep / FunctionStep) or another nested topology — the hydrator wraps composite operations in a Step adapter so they conform to the :class:Step protocol.

condition_name class-attribute instance-attribute

condition_name: str | None = None

Registered predicate name; None makes this route classifier-only.

The :class:RouterSpec itself enforces: at least one route must have a condition_name OR a classifier must be set.

RouterSpec

Dispatch input to one of N routes via classifier or rule conditions.

Hydrates to :class:orqest.orchestration.router.Router.

classifier class-attribute instance-attribute

classifier: AgentSpec | str | None = None

Optional classifier. str is an agent_registry lookup; an :class:AgentSpec spawns fresh. The :class:Router constructor requires EITHER a classifier OR at least one route with a condition — the hydrator surfaces the underlying ValueError if neither is present.

fallback_step class-attribute instance-attribute

fallback_step: OperationSpec | None = None

Step to run when no route matches. None raises RouterError on no-match.

RefinementLoopSpec

Iterate a step with evaluator feedback until convergence or limits.

Hydrates to :class:orqest.orchestration.loop.RefinementLoop.

evaluator instance-attribute

evaluator: AgentSpec | str

Required evaluator. str is an agent_registry name (the agent must return an :class:EvalResult-shaped output); an :class:AgentSpec spawns fresh. A pure-callable evaluator is also accepted via the registry — pass its name as the str and have it resolve via the agent_registry's callable-fallback (or use a tiny BaseAgent wrapper around it).

state_updater_name instance-attribute

state_updater_name: str

Required callable_registry name for the (current_input, output, eval_result) -> next_input updater.

hydrate

Hydrate :mod:orqest.orchestration.spec Pydantic models into runtime objects.

The spec layer is the canonical serialization surface; this module owns the deserialization side. Two responsibilities:

1. :class:CallableRegistry — a name→callable allowlist. The meta-agent emits names from this registry's keys; the hydrator looks them up here. No eval / exec ever runs — the security perimeter is "names from a user-controlled allowlist," nothing more.
2. :func:topology_from_spec — dispatches on spec.kind and recursively builds a live Pipeline / Parallel / Router / RefinementLoop. Nested composites (a Pipeline inside a Route, a Parallel inside a RefinementLoop) are wrapped in :class:_TopologyAsStep so they conform to the :class:Step protocol the runtime classes expect.

Agents are resolved two ways:

• By name against agent_registry — a dict[str, Callable[[], BaseAgent]] of factories (not instances). Factories match the existing pattern used by :class:Evaluator.agent_factory and avoid cached-_agent problems when the same logical agent is instantiated multiple times across a search.
• By inline :class:AgentSpec — spawned fresh via :class:AgentFactory. Requires a factory to be provided to the hydrator.

CallableRegistry

CallableRegistry()

Name → callable map. Explicit, no introspection magic.

The meta agent's design step receives registry.names() as its allowed vocabulary; any name it emits MUST be in here, or hydration raises :class:KeyError. This is the load-bearing security property of the W3.A layer: there is no eval, no exec, no name forgery — the registry holds the actual callables, the meta agent can only refer to them by name.

Source code in orqest/orchestration/hydrate.py

def __init__(self) -> None:
    self._fns: dict[str, Callable[..., Any]] = {}

register

register(name: str, fn: Callable[..., Any]) -> None

Register fn under name. Re-registering replaces silently.

Source code in orqest/orchestration/hydrate.py

def register(self, name: str, fn: Callable[..., Any]) -> None:
    """Register *fn* under *name*. Re-registering replaces silently."""
    if not isinstance(name, str) or not name:
        raise ValueError("CallableRegistry name must be a non-empty string")
    if not callable(fn):
        raise TypeError(f"CallableRegistry value for {name!r} must be callable")
    self._fns[name] = fn

get

get(name: str) -> Callable[..., Any]

Look up name. Raises :class:KeyError with all known names on miss.

Source code in orqest/orchestration/hydrate.py

def get(self, name: str) -> Callable[..., Any]:
    """Look up *name*. Raises :class:`KeyError` with all known names on miss."""
    try:
        return self._fns[name]
    except KeyError as exc:
        raise KeyError(
            f"CallableRegistry has no entry for {name!r}; "
            f"known names: {sorted(self._fns.keys())}"
        ) from exc

names

names() -> list[str]

All registered names, sorted — surface for the meta-agent prompt.

Source code in orqest/orchestration/hydrate.py

def names(self) -> list[str]:
    """All registered names, sorted — surface for the meta-agent prompt."""
    return sorted(self._fns.keys())

pipeline_from_spec

pipeline_from_spec(
    spec: PipelineSpec,
    *,
    callable_registry: CallableRegistry,
    agent_registry: dict[str, Callable[[], BaseAgent]],
    agent_factory: AgentFactory | None = None,
) -> Pipeline

Hydrate a :class:PipelineSpec into a live :class:Pipeline.

Source code in orqest/orchestration/hydrate.py

def pipeline_from_spec(
    spec: PipelineSpec,
    *,
    callable_registry: CallableRegistry,
    agent_registry: dict[str, Callable[[], BaseAgent]],
    agent_factory: AgentFactory | None = None,
) -> Pipeline:
    """Hydrate a :class:`PipelineSpec` into a live :class:`Pipeline`."""
    entries: list[Any] = []
    for entry in spec.steps:
        step = _hydrate_operation_as_step(
            entry.operation,
            callable_registry=callable_registry,
            agent_registry=agent_registry,
            agent_factory=agent_factory,
        )
        cfg = _to_step_config(entry.config)
        if cfg is None:
            entries.append(step)
        else:
            entries.append((step, cfg))
    return Pipeline(entries, name=spec.name)

parallel_from_spec

parallel_from_spec(
    spec: ParallelSpec,
    *,
    callable_registry: CallableRegistry,
    agent_registry: dict[str, Callable[[], BaseAgent]],
    agent_factory: AgentFactory | None = None,
) -> Parallel

Hydrate a :class:ParallelSpec into a live :class:Parallel.

Source code in orqest/orchestration/hydrate.py

def parallel_from_spec(
    spec: ParallelSpec,
    *,
    callable_registry: CallableRegistry,
    agent_registry: dict[str, Callable[[], BaseAgent]],
    agent_factory: AgentFactory | None = None,
) -> Parallel:
    """Hydrate a :class:`ParallelSpec` into a live :class:`Parallel`."""
    steps = [
        _hydrate_operation_as_step(
            op,
            callable_registry=callable_registry,
            agent_registry=agent_registry,
            agent_factory=agent_factory,
        )
        for op in spec.steps
    ]
    merge_fn = _BUILTIN_MERGES.get(spec.merge)
    if merge_fn is None:
        merge_fn = callable_registry.get(spec.merge)
    return Parallel(
        steps,
        merge=merge_fn,
        timeout=spec.timeout,
        name=spec.name,
    )

router_from_spec

router_from_spec(
    spec: RouterSpec,
    *,
    callable_registry: CallableRegistry,
    agent_registry: dict[str, Callable[[], BaseAgent]],
    agent_factory: AgentFactory | None = None,
) -> Router

Hydrate a :class:RouterSpec into a live :class:Router.

Source code in orqest/orchestration/hydrate.py

def router_from_spec(
    spec: RouterSpec,
    *,
    callable_registry: CallableRegistry,
    agent_registry: dict[str, Callable[[], BaseAgent]],
    agent_factory: AgentFactory | None = None,
) -> Router:
    """Hydrate a :class:`RouterSpec` into a live :class:`Router`."""
    routes: list[Route] = []
    for r in spec.routes:
        step = _hydrate_operation_as_step(
            r.step,
            callable_registry=callable_registry,
            agent_registry=agent_registry,
            agent_factory=agent_factory,
        )
        condition = (
            callable_registry.get(r.condition_name)
            if r.condition_name is not None
            else None
        )
        routes.append(Route(name=r.name, step=step, condition=condition))

    classifier: BaseAgent | Callable[..., Any] | None = None
    if isinstance(spec.classifier, str):
        # str = agent_registry name OR callable_registry name (try agent first)
        if spec.classifier in agent_registry:
            classifier = agent_registry[spec.classifier]()
        else:
            classifier = callable_registry.get(spec.classifier)
    elif spec.classifier is not None:
        # Inline AgentSpec
        classifier = _resolve_agent(
            agent_name=None,
            inline_spec=spec.classifier,
            agent_registry=agent_registry,
            agent_factory=agent_factory,
        )

    fallback: Step | None = None
    if spec.fallback_step is not None:
        fallback = _hydrate_operation_as_step(
            spec.fallback_step,
            callable_registry=callable_registry,
            agent_registry=agent_registry,
            agent_factory=agent_factory,
        )

    return Router(
        routes,
        classifier=classifier,
        fallback=fallback,
        name=spec.name,
    )

refinement_loop_from_spec

refinement_loop_from_spec(
    spec: RefinementLoopSpec,
    *,
    callable_registry: CallableRegistry,
    agent_registry: dict[str, Callable[[], BaseAgent]],
    agent_factory: AgentFactory | None = None,
) -> RefinementLoop

Hydrate a :class:RefinementLoopSpec into a live :class:RefinementLoop.

Source code in orqest/orchestration/hydrate.py

def refinement_loop_from_spec(
    spec: RefinementLoopSpec,
    *,
    callable_registry: CallableRegistry,
    agent_registry: dict[str, Callable[[], BaseAgent]],
    agent_factory: AgentFactory | None = None,
) -> RefinementLoop:
    """Hydrate a :class:`RefinementLoopSpec` into a live :class:`RefinementLoop`."""
    step = _hydrate_operation_as_step(
        spec.step,
        callable_registry=callable_registry,
        agent_registry=agent_registry,
        agent_factory=agent_factory,
    )

    if isinstance(spec.evaluator, str):
        if spec.evaluator in agent_registry:
            evaluator: Any = agent_registry[spec.evaluator]()
        else:
            evaluator = callable_registry.get(spec.evaluator)
    else:
        evaluator = _resolve_agent(
            agent_name=None,
            inline_spec=spec.evaluator,
            agent_registry=agent_registry,
            agent_factory=agent_factory,
        )

    state_updater = callable_registry.get(spec.state_updater_name)

    return RefinementLoop(
        step,
        evaluator,
        state_updater=state_updater,
        max_iterations=spec.max_iterations,
        timeout=spec.timeout,
        convergence_window=spec.convergence_window,
        convergence_threshold=spec.convergence_threshold,
        confidence_threshold=spec.confidence_threshold,
    )

topology_from_spec

topology_from_spec(
    spec: TopologySpec,
    *,
    callable_registry: CallableRegistry,
    agent_registry: dict[str, Callable[[], BaseAgent]],
    agent_factory: AgentFactory | None = None,
) -> Pipeline | Parallel | Router | RefinementLoop

Dispatch on spec.kind and hydrate to the right runtime class.

Recursive: nested composites inside step positions are hydrated and wrapped via :class:_TopologyAsStep so the parent runtime sees a Step.

Source code in orqest/orchestration/hydrate.py

def topology_from_spec(
    spec: TopologySpec,
    *,
    callable_registry: CallableRegistry,
    agent_registry: dict[str, Callable[[], BaseAgent]],
    agent_factory: AgentFactory | None = None,
) -> Pipeline | Parallel | Router | RefinementLoop:
    """Dispatch on ``spec.kind`` and hydrate to the right runtime class.

    Recursive: nested composites inside step positions are hydrated and
    wrapped via :class:`_TopologyAsStep` so the parent runtime sees a Step.
    """
    hydrator = _HYDRATORS.get(spec.kind)
    if hydrator is None:
        raise ValueError(
            f"topology_from_spec: unknown topology kind {spec.kind!r}; "
            f"expected one of {sorted(_HYDRATORS.keys())}"
        )
    return hydrator(  # type: ignore[no-any-return]
        spec,
        callable_registry=callable_registry,
        agent_registry=agent_registry,
        agent_factory=agent_factory,
    )

topology

Topology evolution: gene + evaluator atop the orchestration spec layer.

Two pieces:

• :class:TopologyGene — a Pydantic gene whose value is a serialized :data:TopologySpec JSON string. Mirrors the encode/decode contract of :class:PromptGene, so it slots into existing :class:Genome machinery (when the consumer wants to combine prompt + topology evolution in one GEPA run via :class:TopologyGEPAAdapter).
• :class:TopologyEvaluator — an :class:Evaluator subclass that hydrates the genome's topology spec into a live runtime, runs it on each example, and unpacks composite results (ParallelResult.merged / LoopResult.output) before scoring. Adds n_agents and depth to :attr:MetricBundle.raw as proxies for cost / latency ceilings — these flow through to the optimizer's Pareto frontier without any further wiring.

YAGNI note: :class:TopologyGene is not added to the existing :data:Gene discriminated union in :mod:orqest.optimization.genome. The recommended path for combining ADAS+GEPA is the two-phase one (ADAS first, GEPA on the winner), which uses two distinct genome shapes — there's no need for a unified Gene union today. If users prove the combined-single-genome case matters, extending the union is a one-line change.

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

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

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

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

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

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,
        },
    )

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

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

meta_agent

ADAS-style Meta Agent Search loop for topology evolution.

Faithful to ADAS in shape (design → reflexion → evaluate → archive, arXiv 2408.08435) but not in surface: the meta agent emits typed :data:TopologySpec JSON, never raw Python. This sidesteps ADAS's exec()-in-process gap (paper claimed containerization; public repo ships in-process exec) — there is no code to sandbox because there is no code, only Pydantic-validated structure.

The loop owes its archive-strategy menu to the 2510.06711 critique: top_k (default) and parallel outperform ADAS's original cumulative on most benchmarks. Users who want to reproduce the ADAS paper verbatim flip MetaAgentConfig.archive_strategy="cumulative".

Three pieces in this module:

• :class:MetaAgentConfig — frozen dataclass, search-loop knobs.
• :class:Archive + :class:ArchiveEntry — manages the candidate population with strategy-pluggable retention and prompt-serialization.
• :class:MetaAgentSearch — the loop. Returns an :class:OptimizationResult shaped identically to GEPA's so downstream consumers (apply_result, notebook visualizations) work without dispatch.

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.

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).

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.

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

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

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

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

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

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

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

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

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)