GEPA

steerability.algorithms.input_control.gepa

args

Arguments for the GEPA input control.

control

GEPA, reflective prompt optimization (single-system-prompt variant).

logger = logging.getLogger(__name__) module-attribute

GEPA

Bases: InputControl

GEPA, reflective prompt optimization (single-system-prompt variant).

Optimizes a single system prompt via GEPA's reflective genetic search (Agrawal et al., 2025), restricted to the single-module (|M| = 1) case. It implements:

• Reflective prompt mutation: an LLM rewrites the current instruction using natural-language feedback drawn from rollouts.
• Pareto-based candidate selection: GEPA keeps an instance-wise Pareto frontier over a held-out set and samples parents by win-frequency (paper section 3.1, Algorithm 2).
• A genetic candidate pool that accumulates lessons across successive mutations.

Scope:

• This control optimizes ONE system prompt. It does NOT support compound, multi-module AI systems, and it does NOT implement the system-aware merge / crossover operator (GEPA+Merge). Merge is a multi-module operation that recombines complementary modules across lineages, and is vacuous with a single prompt.
• For compound, multi-module / multi-prompt optimization with module-level credit assignment and merge, use the reference implementation at https://github.com/gepa-ai/gepa (or DSPy's GEPA integration).

Scoring is supplied directly through a per-example row_scorer (the paper's metric µ at the instance level, required because Pareto selection needs per-instance scores), an optional feedback_fn returning the textual feedback GEPA reflects on (the paper's µf), and a format_query mapping a dataset row to the user query. GEPA runs generation, scoring, and feedback internally; there is no separate Evaluator object.

adapt_messages() / adapt() inject the optimized instruction as the system prompt at inference time. Memory: TextMemory(slots={"instruction": best_instruction}).

reflection_lm is optional and defaults to the task model. A separate, stronger reflection LM can help when the task model is small, but is not required.

Reference:

• "GEPA: Reflective Prompt Evolution Can Outperform Reinforcement Learning" Agrawal et al., 2025 https://arxiv.org/abs/2507.19457

Source code in steerability/algorithms/input_control/gepa/control.py

class GEPA(InputControl):
    """GEPA, reflective prompt optimization (single-system-prompt variant).

    Optimizes a single system prompt via GEPA's reflective genetic search (Agrawal et al.,
    2025), restricted to the single-module (|M| = 1) case. It implements:

      - Reflective prompt mutation: an LLM rewrites the current instruction using
        natural-language feedback drawn from rollouts.
      - Pareto-based candidate selection: GEPA keeps an instance-wise Pareto frontier over a
        held-out set and samples parents by win-frequency (paper section 3.1, Algorithm 2).
      - A genetic candidate pool that accumulates lessons across successive mutations.

    Scope:

      - This control optimizes ONE system prompt. It does NOT support compound, multi-module
        AI systems, and it does NOT implement the system-aware merge / crossover operator
        (GEPA+Merge). Merge is a multi-module operation that recombines complementary modules
        across lineages, and is vacuous with a single prompt.
      - For compound, multi-module / multi-prompt optimization with module-level credit
        assignment and merge, use the reference implementation at
        [https://github.com/gepa-ai/gepa](https://github.com/gepa-ai/gepa) (or DSPy's GEPA
        integration).

    Scoring is supplied directly through a per-example `row_scorer` (the paper's metric `µ` at
    the instance level, required because Pareto selection needs per-instance scores), an
    optional `feedback_fn` returning the textual feedback GEPA reflects on (the paper's `µf`),
    and a `format_query` mapping a dataset row to the user query. GEPA runs generation, scoring,
    and feedback internally; there is no separate Evaluator object.

    `adapt_messages()` / `adapt()` inject the optimized instruction as the system prompt at
    inference time. Memory: `TextMemory(slots={"instruction": best_instruction})`.

    `reflection_lm` is optional and defaults to the task model. A separate, stronger reflection
    LM can help when the task model is small, but is not required.

    Reference:

      - "GEPA: Reflective Prompt Evolution Can Outperform Reinforcement Learning"
        Agrawal et al., 2025
        [https://arxiv.org/abs/2507.19457](https://arxiv.org/abs/2507.19457)
    """

    Args = GEPAArgs
    supports_batching: bool = True

    # placeholders
    seed_instruction: str = ""
    train_set: list[dict] = []
    row_scorer: Any = None
    feedback_fn: Any = None
    format_query: Any = None
    gen_kwargs: dict | None = None
    reflection_lm: Any = None
    reflection_tokenizer: Any = None
    budget: int = 300
    minibatch_size: int = 3
    pareto_set_size: int = 30
    meta_prompt: str | None = None
    proposer_gen_kwargs: dict | None = None
    progress_callback: Any = None
    seed: int | None = None
    memory: TextMemory | None = None

    # state
    tokenizer: Any = None
    _formatter: SystemPromptFormatter | None = None

    def export_state(self) -> dict:
        """The optimized instruction memory under the `"memory"` key (after `steer()`)."""
        return {"memory": self.memory} if self.memory is not None else {}

    def frozen_form(self, state: dict) -> tuple[str, dict]:
        """A same-class frozen form: the recipe args with `memory=` set to the optimized
        memory (the search-only args stay inert)."""
        from dataclasses import fields

        kwargs = {f.name: getattr(self.args, f.name) for f in fields(self.args) if f.init}
        kwargs["memory"] = state["memory"]
        return "input_control/gepa", kwargs

    def fit_identity(self):
        """The optimizer-relevant args (everything except `memory` and `progress_callback`),
        or None when the recipe already carries a memory."""
        from dataclasses import fields

        if self.args.memory is not None:
            return None
        return {
            f.name: getattr(self.args, f.name)
            for f in fields(self.args) if f.init and f.name not in ("memory", "progress_callback")
        }

    def steer_access(self) -> ModelAccess:
        """`ModelAccess.ROLLOUTS`; task rollouts and reflection generate through the session,
        and adaptation is a formatter. With a precomputed `memory`, steering is model-free
        (`ModelAccess.FACTS`)."""
        if getattr(getattr(self, "args", None), "memory", None) is not None:
            return ModelAccess.FACTS
        return ModelAccess.ROLLOUTS

    def steer(
        self,
        model=None,
        tokenizer=None,
        session=None,
        **kwargs,
    ) -> None:
        self.tokenizer = tokenizer
        if self.args.memory is not None:
            self.memory = self.args.memory
            self._formatter = SystemPromptFormatter()
            return

        rng = random.Random(self.seed) if self.seed is not None else random.Random()
        task_lm = SessionLM(session) if session is not None else model

        if self.reflection_lm is None:
            logger.info("GEPA: no `reflection_lm` supplied; reflection falls back to the task model.")
        reflection_lm = self.reflection_lm if self.reflection_lm is not None else task_lm
        reflection_tok = self.reflection_tokenizer if self.reflection_tokenizer is not None else tokenizer
        proposer = LLMMetaPromptProposer(
            llm=reflection_lm,
            tokenizer=reflection_tok,
            meta_prompt_template=self.meta_prompt or reflective_meta_prompt.GEPA_DEFAULT,
            parse_fn=parse_fenced_or_whole,
            gen_kwargs=self.proposer_gen_kwargs or {
                "max_new_tokens": 1024,
                "do_sample": True,
                "temperature": 1.0,
                "top_p": 0.95,
            },
        )

        d_feedback, d_pareto = self._split_dataset(rng)
        budget = RolloutBudget(self.budget)

        pool = CandidatePool()
        _, seed_scores, _ = self._run(task_lm, self.seed_instruction, d_pareto, with_feedback=False)
        budget.charge(len(d_pareto))
        pool.add(self.seed_instruction, seed_scores)

        step = 0
        self._emit_progress(step=0, event="seed", parent_idx=None, parent_score=None,
                            candidate_score=None, accepted=True, pool_size=len(pool.candidates),
                            best_mean=float(pool.scores.mean(axis=1).max()),
                            proposed=self.seed_instruction)

        while budget:
            parent_idx = pareto_sampling.sample(pool.frontier(), len(pool.candidates), rng=rng)

            mb_size = min(self.minibatch_size, len(d_feedback))
            minibatch = rng.sample(d_feedback, mb_size) if mb_size <= len(d_feedback) else list(d_feedback)
            if not minibatch:
                break
            if budget.remaining < len(minibatch):
                break

            parent_outputs, parent_scores, parent_feedback = self._run(
                task_lm, pool.candidates[parent_idx], minibatch, with_feedback=True
            )
            budget.charge(len(minibatch))
            parent_mb_mean = mean(parent_scores)

            query_texts = [self._format_query(row) for row in minibatch]
            records = build_records(query_texts, parent_outputs, parent_feedback)
            new_candidates = proposer.propose(
                seed=pool.candidates[parent_idx],
                n=1,
                context={"records": render_records(records)},
            )
            new_text = (new_candidates[0].strip() if new_candidates else "")
            if not new_text:
                continue

            if budget.remaining < len(minibatch):
                break
            _, cand_scores, _ = self._run(task_lm, new_text, minibatch, with_feedback=False)
            budget.charge(len(minibatch))
            cand_mb_mean = mean(cand_scores)
            if cand_mb_mean <= parent_mb_mean:  # strict improvement
                step += 1
                self._emit_progress(step=step, event="reject", parent_idx=parent_idx,
                                    parent_score=parent_mb_mean, candidate_score=cand_mb_mean,
                                    accepted=False, pool_size=len(pool.candidates),
                                    best_mean=float(pool.scores.mean(axis=1).max()),
                                    proposed=new_text)
                continue

            if budget.remaining < len(d_pareto):
                break
            _, cand_full, _ = self._run(task_lm, new_text, d_pareto, with_feedback=False)
            budget.charge(len(d_pareto))
            pool.add(new_text, cand_full)
            step += 1
            self._emit_progress(step=step, event="accept", parent_idx=parent_idx,
                                parent_score=parent_mb_mean, candidate_score=cand_mb_mean,
                                accepted=True, pool_size=len(pool.candidates),
                                best_mean=float(pool.scores.mean(axis=1).max()),
                                proposed=new_text)

        best = pool.candidates[pool.best_index()]
        self.memory = TextMemory(slots={"instruction": best})
        self._formatter = SystemPromptFormatter()

    def _run(
        self,
        task_lm,
        instruction: str,
        batch: list[dict],
        *,
        with_feedback: bool,
    ) -> tuple[list[str], list[float], list[str] | None]:
        queries = [self._format_query(row) for row in batch]
        outputs = generate_with_system_prompt(
            task_lm, self.tokenizer, instruction, queries, gen_kwargs=self.gen_kwargs
        )
        scores = [float(self.row_scorer(out, row)) for out, row in zip(outputs, batch)]
        feedback: list[str] | None = None
        if with_feedback:
            if self.feedback_fn is not None:
                feedback = [
                    self.feedback_fn(out, row, score)
                    for out, row, score in zip(outputs, batch, scores)
                ]
            else:
                feedback = [f"score={score:.4f}" for score in scores]
        return outputs, scores, feedback

    def _emit_progress(self, **info) -> None:
        if self.progress_callback is not None:
            self.progress_callback(info)

    def _format_query(self, row: dict) -> str:
        if self.format_query is not None:
            return self.format_query(row)
        return row["input"]

    def _split_dataset(self, rng: random.Random) -> tuple[list[dict], list[dict]]:
        train = list(self.train_set)
        rng.shuffle(train)
        pareto_n = min(self.pareto_set_size, len(train))
        if pareto_n >= len(train):
            d_pareto = train[:]
            d_feedback = train[:]  # fall back to overlap when train_set is tiny
        else:
            d_pareto = train[:pareto_n]
            d_feedback = train[pareto_n:] or train[:]
        return d_feedback, d_pareto

    def adapt_messages(
        self,
        messages: list[list[dict]],
        runtime_kwargs: dict | None = None,
    ) -> list[list[dict]] | None:
        if self.memory is None or self._formatter is None:
            raise RuntimeError("GEPA.adapt_messages called before .steer().")
        slot_memory = TextMemory(slots={"instruction": self.memory["instruction"]})
        return self._formatter.apply_to_messages(messages, slot_memory, runtime_kwargs)

    def adapt(
        self,
        input_ids: list[int] | torch.Tensor,
        runtime_kwargs: dict | None = None,
    ) -> list[int] | torch.Tensor:
        if self.memory is None or self._formatter is None or self.tokenizer is None:
            raise RuntimeError("GEPA.adapt called before .steer().")
        slot_memory = TextMemory(slots={"instruction": self.memory["instruction"]})
        if not isinstance(input_ids, torch.Tensor):
            input_ids = torch.tensor(input_ids, dtype=torch.long)
        return self._formatter.apply_to_ids(input_ids, slot_memory, self.tokenizer, runtime_kwargs)

RUNTIME_KWARGS_SCHEMA = [] class-attribute instance-attribute

args = self.Args.validate(*args, **kwargs) instance-attribute

budget = 300 class-attribute instance-attribute

enabled = True class-attribute instance-attribute

feedback_fn = None class-attribute instance-attribute

format_query = None class-attribute instance-attribute

gen_kwargs = None class-attribute instance-attribute

memory = None class-attribute instance-attribute

meta_prompt = None class-attribute instance-attribute

minibatch_size = 3 class-attribute instance-attribute

pareto_set_size = 30 class-attribute instance-attribute

progress_callback = None class-attribute instance-attribute

proposer_gen_kwargs = None class-attribute instance-attribute

reflection_lm = None class-attribute instance-attribute

reflection_tokenizer = None class-attribute instance-attribute

row_scorer = None class-attribute instance-attribute

seed = None class-attribute instance-attribute

seed_instruction = '' class-attribute instance-attribute

supports_batching = True class-attribute instance-attribute

tokenizer = None class-attribute instance-attribute

train_set = [] class-attribute instance-attribute

adapt(input_ids, runtime_kwargs=None)

Transform input_ids into a steered prompt.

May read instance state (e.g. self.memory) that was populated by steer().

Parameters:

Name	Type	Description	Default
input_ids	list[int] | Tensor	The user's prompt token IDs.	required
runtime_kwargs	dict | None	Per-call parameters.	None

Returns:

Type	Description
list[int] | Tensor	The transformed token IDs.

Source code in steerability/algorithms/input_control/gepa/control.py

def adapt(
    self,
    input_ids: list[int] | torch.Tensor,
    runtime_kwargs: dict | None = None,
) -> list[int] | torch.Tensor:
    if self.memory is None or self._formatter is None or self.tokenizer is None:
        raise RuntimeError("GEPA.adapt called before .steer().")
    slot_memory = TextMemory(slots={"instruction": self.memory["instruction"]})
    if not isinstance(input_ids, torch.Tensor):
        input_ids = torch.tensor(input_ids, dtype=torch.long)
    return self._formatter.apply_to_ids(input_ids, slot_memory, self.tokenizer, runtime_kwargs)

adapt_messages(messages, runtime_kwargs=None)

Optional message-level adaptation, called BEFORE chat-template tokenization.

Default returns None (no message-level changes). Subclasses that modify chat structure (set/replace system prompt, insert example turns) override this. When this method returns a non-None result for a chat-input generation, the pipeline does NOT additionally call this control's adapt() for that call; a control may therefore implement both entry points (message-level for chat input, token-level for raw text/tensor input) without being applied twice. Other input controls in the pipeline still run per their own phase.

Parameters:

Name	Type	Description	Default
messages	list[list[dict]]	A batch of chats; outer list is the batch, inner list is the message sequence for one chat.	required
runtime_kwargs	dict | None	Per-call parameters.	None

Returns:

Type	Description
list[list[dict]] | None	The transformed messages, or None to indicate no change.

Source code in steerability/algorithms/input_control/gepa/control.py

def adapt_messages(
    self,
    messages: list[list[dict]],
    runtime_kwargs: dict | None = None,
) -> list[list[dict]] | None:
    if self.memory is None or self._formatter is None:
        raise RuntimeError("GEPA.adapt_messages called before .steer().")
    slot_memory = TextMemory(slots={"instruction": self.memory["instruction"]})
    return self._formatter.apply_to_messages(messages, slot_memory, runtime_kwargs)

cleanup()

Release resources allocated during steer().

Override in subclasses that allocate GPU memory or other resources during steering. Default no-op.

Source code in steerability/algorithms/core/base_control.py

def cleanup(self) -> None:
    """Release resources allocated during `steer()`.

    Override in subclasses that allocate GPU memory or other resources during steering. Default
    no-op.
    """
    pass

clone_for_call(seed=None)

A configuration-preserving shallow clone for one generation call.

The clone shares steer-time artifacts (memories, steering vectors, attached tokenizers) with the original but has its own attribute namespace, so per-call attribute mutation on the clone never races another call using the original. When seed is given and the control defines reseed(seed), the clone's client-side RNG is re-seeded.

Parameters:

Name	Type	Description	Default
seed	int | None	Optional seed forwarded to the clone's reseed().	None

Returns:

Type	Description
BaseControl	The clone.

Source code in steerability/algorithms/core/base_control.py

def clone_for_call(self, seed: int | None = None) -> "BaseControl":
    """A configuration-preserving shallow clone for one generation call.

    The clone shares steer-time artifacts (memories, steering vectors, attached tokenizers)
    with the original but has its own attribute namespace, so per-call attribute mutation on
    the clone never races another call using the original. When `seed` is given and the
    control defines `reseed(seed)`, the clone's client-side RNG is re-seeded.

    Args:
        seed: Optional seed forwarded to the clone's `reseed()`.

    Returns:
        The clone.
    """
    clone = copy.copy(self)
    if seed is not None:
        reseed = getattr(clone, "reseed", None)
        if callable(reseed):
            reseed(seed)
    return clone

export_state()

The optimized instruction memory under the "memory" key (after steer()).

Source code in steerability/algorithms/input_control/gepa/control.py

def export_state(self) -> dict:
    """The optimized instruction memory under the `"memory"` key (after `steer()`)."""
    return {"memory": self.memory} if self.memory is not None else {}

fit_identity()

The optimizer-relevant args (everything except memory and progress_callback), or None when the recipe already carries a memory.

Source code in steerability/algorithms/input_control/gepa/control.py

def fit_identity(self):
    """The optimizer-relevant args (everything except `memory` and `progress_callback`),
    or None when the recipe already carries a memory."""
    from dataclasses import fields

    if self.args.memory is not None:
        return None
    return {
        f.name: getattr(self.args, f.name)
        for f in fields(self.args) if f.init and f.name not in ("memory", "progress_callback")
    }

frozen_form(state)

A same-class frozen form: the recipe args with memory= set to the optimized memory (the search-only args stay inert).

Source code in steerability/algorithms/input_control/gepa/control.py

def frozen_form(self, state: dict) -> tuple[str, dict]:
    """A same-class frozen form: the recipe args with `memory=` set to the optimized
    memory (the search-only args stay inert)."""
    from dataclasses import fields

    kwargs = {f.name: getattr(self.args, f.name) for f in fields(self.args) if f.init}
    kwargs["memory"] = state["memory"]
    return "input_control/gepa", kwargs

requirements()

Backend requirements computed from this instance's configuration, per phase.

Input controls transform the prompt client-side, so the generate phase requires nothing beyond the session contract on any backend. A control whose steer() reads the live pipeline model (e.g. for rollouts or scoring) overrides this with a steer-phase Capability.IN_PROCESS_TORCH requirement.

Returns:

Type	Description
Requirements	The control's phase-keyed requirements.

Source code in steerability/algorithms/input_control/base.py

def requirements(self) -> Requirements:
    """Backend requirements computed from this instance's configuration, per phase.

    Input controls transform the prompt client-side, so the generate phase requires nothing
    beyond the session contract on any backend. A control whose `steer()` reads the live
    pipeline model (e.g. for rollouts or scoring) overrides this with a steer-phase
    `Capability.IN_PROCESS_TORCH` requirement.

    Returns:
        The control's phase-keyed requirements.
    """
    return Requirements()

steer(model=None, tokenizer=None, session=None, **kwargs)

Optional offline preparation. Default is no-op.

session is a SteeringSession on the steering backend, provided by the pipeline.

Source code in steerability/algorithms/input_control/gepa/control.py

def steer(
    self,
    model=None,
    tokenizer=None,
    session=None,
    **kwargs,
) -> None:
    self.tokenizer = tokenizer
    if self.args.memory is not None:
        self.memory = self.args.memory
        self._formatter = SystemPromptFormatter()
        return

    rng = random.Random(self.seed) if self.seed is not None else random.Random()
    task_lm = SessionLM(session) if session is not None else model

    if self.reflection_lm is None:
        logger.info("GEPA: no `reflection_lm` supplied; reflection falls back to the task model.")
    reflection_lm = self.reflection_lm if self.reflection_lm is not None else task_lm
    reflection_tok = self.reflection_tokenizer if self.reflection_tokenizer is not None else tokenizer
    proposer = LLMMetaPromptProposer(
        llm=reflection_lm,
        tokenizer=reflection_tok,
        meta_prompt_template=self.meta_prompt or reflective_meta_prompt.GEPA_DEFAULT,
        parse_fn=parse_fenced_or_whole,
        gen_kwargs=self.proposer_gen_kwargs or {
            "max_new_tokens": 1024,
            "do_sample": True,
            "temperature": 1.0,
            "top_p": 0.95,
        },
    )

    d_feedback, d_pareto = self._split_dataset(rng)
    budget = RolloutBudget(self.budget)

    pool = CandidatePool()
    _, seed_scores, _ = self._run(task_lm, self.seed_instruction, d_pareto, with_feedback=False)
    budget.charge(len(d_pareto))
    pool.add(self.seed_instruction, seed_scores)

    step = 0
    self._emit_progress(step=0, event="seed", parent_idx=None, parent_score=None,
                        candidate_score=None, accepted=True, pool_size=len(pool.candidates),
                        best_mean=float(pool.scores.mean(axis=1).max()),
                        proposed=self.seed_instruction)

    while budget:
        parent_idx = pareto_sampling.sample(pool.frontier(), len(pool.candidates), rng=rng)

        mb_size = min(self.minibatch_size, len(d_feedback))
        minibatch = rng.sample(d_feedback, mb_size) if mb_size <= len(d_feedback) else list(d_feedback)
        if not minibatch:
            break
        if budget.remaining < len(minibatch):
            break

        parent_outputs, parent_scores, parent_feedback = self._run(
            task_lm, pool.candidates[parent_idx], minibatch, with_feedback=True
        )
        budget.charge(len(minibatch))
        parent_mb_mean = mean(parent_scores)

        query_texts = [self._format_query(row) for row in minibatch]
        records = build_records(query_texts, parent_outputs, parent_feedback)
        new_candidates = proposer.propose(
            seed=pool.candidates[parent_idx],
            n=1,
            context={"records": render_records(records)},
        )
        new_text = (new_candidates[0].strip() if new_candidates else "")
        if not new_text:
            continue

        if budget.remaining < len(minibatch):
            break
        _, cand_scores, _ = self._run(task_lm, new_text, minibatch, with_feedback=False)
        budget.charge(len(minibatch))
        cand_mb_mean = mean(cand_scores)
        if cand_mb_mean <= parent_mb_mean:  # strict improvement
            step += 1
            self._emit_progress(step=step, event="reject", parent_idx=parent_idx,
                                parent_score=parent_mb_mean, candidate_score=cand_mb_mean,
                                accepted=False, pool_size=len(pool.candidates),
                                best_mean=float(pool.scores.mean(axis=1).max()),
                                proposed=new_text)
            continue

        if budget.remaining < len(d_pareto):
            break
        _, cand_full, _ = self._run(task_lm, new_text, d_pareto, with_feedback=False)
        budget.charge(len(d_pareto))
        pool.add(new_text, cand_full)
        step += 1
        self._emit_progress(step=step, event="accept", parent_idx=parent_idx,
                            parent_score=parent_mb_mean, candidate_score=cand_mb_mean,
                            accepted=True, pool_size=len(pool.candidates),
                            best_mean=float(pool.scores.mean(axis=1).max()),
                            proposed=new_text)

    best = pool.candidates[pool.best_index()]
    self.memory = TextMemory(slots={"instruction": best})
    self._formatter = SystemPromptFormatter()

steer_access()

ModelAccess.ROLLOUTS; task rollouts and reflection generate through the session, and adaptation is a formatter. With a precomputed memory, steering is model-free (ModelAccess.FACTS).

Source code in steerability/algorithms/input_control/gepa/control.py

def steer_access(self) -> ModelAccess:
    """`ModelAccess.ROLLOUTS`; task rollouts and reflection generate through the session,
    and adaptation is a formatter. With a precomputed `memory`, steering is model-free
    (`ModelAccess.FACTS`)."""
    if getattr(getattr(self, "args", None), "memory", None) is not None:
        return ModelAccess.FACTS
    return ModelAccess.ROLLOUTS

steer_fits()

The fit artifacts this instance's steer step will produce, for the steer plan.

Each entry is (artifact, artifact_class), where artifact is the fit source or recipe class name and artifact_class is "direction" or "calibrated". The default is an empty tuple (no fits).

Returns:

Type	Description
tuple[tuple[str, str], ...]	The declared fit artifacts, in declaration order.

Source code in steerability/algorithms/core/base_control.py

def steer_fits(self) -> tuple[tuple[str, str], ...]:
    """The fit artifacts this instance's steer step will produce, for the steer plan.

    Each entry is `(artifact, artifact_class)`, where `artifact` is the fit source or
    recipe class name and `artifact_class` is `"direction"` or `"calibrated"`. The default
    is an empty tuple (no fits).

    Returns:
        The declared fit artifacts, in declaration order.
    """
    return ()

utils

pareto_sampling

Frequency-weighted Pareto sampling for GEPA's parent selection (Agrawal et al., 2025, Algorithm 2).

Sample one candidate from the Pareto illumination set Ĉ, the per-instance winners pruned of dominated candidates, with probability proportional to win-frequency f[·]. Falls back to uniform over the pool only when the illumination set is empty (a degenerate score matrix).

sample(frontier, pool_size, rng=None)

Frequency-weighted draw over the Pareto illumination set (Algorithm 2, line 8).

Parameters:

Name	Type	Description	Default
frontier	ParetoFrontier	ParetoFrontier over the [n_candidates, n_instances] score matrix.	required
pool_size	int	Number of candidates in the pool.	required
rng	Random | None	Optional RNG.	None

Returns:

Type	Description
int	A candidate index in [0, pool_size).

Source code in steerability/algorithms/input_control/gepa/utils/pareto_sampling.py

def sample(
    frontier: ParetoFrontier,
    pool_size: int,
    rng: random.Random | None = None,
) -> int:
    """Frequency-weighted draw over the Pareto illumination set (Algorithm 2, line 8).

    Args:
        frontier: `ParetoFrontier` over the `[n_candidates, n_instances]` score matrix.
        pool_size: Number of candidates in the pool.
        rng: Optional RNG.

    Returns:
        A candidate index in `[0, pool_size)`.
    """
    rng = rng or random
    if pool_size <= 0:
        raise ValueError("pool_size must be positive.")
    if pool_size == 1:
        return 0

    illum = frontier.illumination_set  # [(idx, coverage), ...]
    if not illum:
        return rng.randrange(pool_size)

    indices, weights = zip(*illum)
    return rng.choices(list(indices), weights=list(weights), k=1)[0]

pool

CandidatePool, GEPA's genetic candidate pool with a Pareto score matrix.

Holds the candidate instructions and the [n_candidates, n_instances] score matrix over D_pareto.

CandidatePool dataclass

Genetic candidate pool with a Pareto score matrix.

candidates[k] and the rows of scores are parallel-indexed by candidate. scores is [n_candidates, n_instances] over D_pareto.

Attributes:

Name	Type	Description
candidates	list[str]	Candidate instruction strings.

Source code in steerability/algorithms/input_control/gepa/utils/pool.py

@dataclass
class CandidatePool:
    """Genetic candidate pool with a Pareto score matrix.

    `candidates[k]` and the rows of `scores` are parallel-indexed by candidate. `scores` is
    `[n_candidates, n_instances]` over D_pareto.

    Attributes:
        candidates: Candidate instruction strings.
    """

    candidates: list[str] = field(default_factory=list)
    _scores_rows: list[list[float]] = field(default_factory=list)

    @property
    def scores(self) -> np.ndarray:
        """`[n_candidates, n_instances]` Pareto score matrix."""
        return np.array(self._scores_rows, dtype=float)

    def add(self, candidate: str, score_row: list[float]) -> int:
        """Append a candidate and return its index."""
        index = len(self.candidates)
        self.candidates.append(candidate)
        self._scores_rows.append(list(score_row))
        return index

    def frontier(self) -> "ParetoFrontier":
        from steerability.algorithms.input_control.common.pareto import ParetoFrontier
        return ParetoFrontier(self.scores)

    def best_index(self) -> int:
        """Index of the candidate with the highest mean score over D_pareto."""
        return int(self.scores.mean(axis=1).argmax())

candidates = field(default_factory=list) class-attribute instance-attribute

scores property

[n_candidates, n_instances] Pareto score matrix.

add(candidate, score_row)

Append a candidate and return its index.

Source code in steerability/algorithms/input_control/gepa/utils/pool.py

def add(self, candidate: str, score_row: list[float]) -> int:
    """Append a candidate and return its index."""
    index = len(self.candidates)
    self.candidates.append(candidate)
    self._scores_rows.append(list(score_row))
    return index

best_index()

Index of the candidate with the highest mean score over D_pareto.

Source code in steerability/algorithms/input_control/gepa/utils/pool.py

def best_index(self) -> int:
    """Index of the candidate with the highest mean score over D_pareto."""
    return int(self.scores.mean(axis=1).argmax())

frontier()

Source code in steerability/algorithms/input_control/gepa/utils/pool.py

def frontier(self) -> "ParetoFrontier":
    from steerability.algorithms.input_control.common.pareto import ParetoFrontier
    return ParetoFrontier(self.scores)

reflective_dataset

Builder of reflective records from rollout outputs and feedback (single system prompt).

build_records(queries, outputs, feedback)

Build reflective records from rendered queries and their rollout outputs/feedback.

Produces one record per item with shape {"Inputs": <rendered query string>, "Generated Output": ..., "Feedback": ...}.

Inputs is the query the task model actually received (i.e. format_query(row)), not the raw training row -- the gold target must not appear here; it reaches reflection via Feedback when the user's feedback_fn includes it.

Parameters:

Name	Type	Description	Default
queries	list[str]	Per-instance rendered query strings shown to the task model.	required
outputs	list	Per-instance task-model outputs.	required
feedback	list	Per-instance textual feedback.	required

Returns:

Type	Description
list[dict]	A list of records, length equal to len(queries).

Source code in steerability/algorithms/input_control/gepa/utils/reflective_dataset.py

def build_records(queries: list[str], outputs: list, feedback: list) -> list[dict]:
    """Build reflective records from rendered queries and their rollout outputs/feedback.

    Produces one record per item with shape
    `{"Inputs": <rendered query string>, "Generated Output": ..., "Feedback": ...}`.

    `Inputs` is the query the task model actually received (i.e. `format_query(row)`), not
    the raw training row -- the gold target must not appear here; it reaches reflection via
    `Feedback` when the user's `feedback_fn` includes it.

    Args:
        queries: Per-instance rendered query strings shown to the task model.
        outputs: Per-instance task-model outputs.
        feedback: Per-instance textual feedback.

    Returns:
        A list of records, length equal to `len(queries)`.
    """
    return [
        {"Inputs": q, "Generated Output": out, "Feedback": fb}
        for q, out, fb in zip(queries, outputs, feedback)
    ]

reflective_meta_prompt

Reflective meta-prompt for GEPA.

GEPA_DEFAULT reproduces the reflective instruction-proposal prompt of Agrawal et al., 2025 (Appendix B; matching the open-source gepa-ai/gepa reference wording), adapted to the LLMMetaPromptProposer format-string convention. The placeholders are:

• {seed}: the current instruction for the module being optimized (the parent prompt).
• {records}: rendered per-example reflective records (inputs, outputs, feedback).

The matching parser (parse_fenced_or_whole) extracts the content of the first ``` fenced block. Per-module context (which module is being optimized) can be layered in by a caller-supplied template; the default omits it to match the reference.

GEPA_DEFAULT = "I provided an assistant with the following instructions to perform a task for me:\n```\n{seed}\n```\n\nThe following are examples of different task inputs provided to the assistant along with the assistant's response for each of them, and some feedback on how the assistant's response could be better:\n```\n{records}\n```\n\nYour task is to write a new instruction for the assistant.\n\nRead the inputs carefully and identify the input format and infer detailed task description about the task I wish to solve with the assistant.\n\nRead all the assistant responses and the corresponding feedback. Identify all niche and domain specific factual information about the task and include it in the instruction, as a lot of it may not be available to the assistant in the future. The assistant may have utilized a generalizable strategy to solve the task, if so, include that in the instruction as well.\n\nProvide the new instructions within ``` blocks." module-attribute

render_records(records)

Render a per-component list of reflective records into a readable block.

Each record may carry the keys Inputs, Generated Output, and Feedback. Missing fields render with a sensible placeholder. Returns (no records) for an empty list.

Source code in steerability/algorithms/input_control/gepa/utils/reflective_meta_prompt.py

def render_records(records: list[dict[str, Any]]) -> str:
    """Render a per-component list of reflective records into a readable block.

    Each record may carry the keys `Inputs`, `Generated Output`, and `Feedback`. Missing
    fields render with a sensible placeholder. Returns `(no records)` for an empty list.
    """
    if not records:
        return "(no records)"
    lines: list[str] = []
    for record in records:
        inputs = record.get("Inputs", "(no inputs)")
        output = record.get("Generated Output", "(no output)")
        feedback = record.get("Feedback", "(no feedback)")
        lines.append(f"- Inputs: {inputs}")
        lines.append(f"  Generated Output: {output}")
        lines.append(f"  Feedback: {feedback}")
    return "\n".join(lines)