GEPA¶

GEPA (Agrawal et al., 2025) optimizes a system prompt by reflecting on where the current prompt failed. It proposes a revised instruction from (natural-language) feedback on outputs, keeps the revision only if it scores better, and repeats. The process maintains a diverse set of strong candidate prompt as opposed to trying to refine a single prompt.

In this notebook the model is instructed to answer a question while obeying an output rule that is not stated in the question (respond in all lowercase with no punctuation). We first run GEPA under default reflection (where the reflection model is the same as the task model) followed by second run with a stronger reflection model.

Reference: Agrawal et al., 2025, GEPA: Reflective Prompt Evolution Can Outperform Reinforcement Learning.

Setup¶

If running this from a Google Colab notebook, please uncomment the following cell to install the toolkit. The following block is not necessary if running this notebook from a virtual environment where the toolkit has already been installed.

In [1]:

# !git clone https://github.com/generative-computing/steerability.git
# %cd steerability
# !pip install -q -e .

# !git clone https://github.com/generative-computing/steerability.git # %cd steerability # !pip install -q -e .

The following authentication steps may be necessary to access any gated models (after being granted access by Hugging Face). Uncomment the following if you need to log in to the Hugging Face Hub using a HUGGINGFACE_TOKEN value stored in a .env file.

In [2]:

# !pip install -q python-dotenv
# from dotenv import load_dotenv
# import os

# load_dotenv()
# token = os.getenv("HUGGINGFACE_TOKEN")
# from huggingface_hub import login
# login(token=token)

# !pip install -q python-dotenv # from dotenv import load_dotenv # import os # load_dotenv() # token = os.getenv("HUGGINGFACE_TOKEN") # from huggingface_hub import login # login(token=token)

The task model is a small model (gemma-3-4b-it) that answers the factual questions correctly but does not follow the specific requirements. We investigate two reflection types in this notebook: default reflection (using the 4b model) and stronger reflection (using gemma-3-12b-it).

In [3]:

import string

import pandas as pd
import torch

from steerability.algorithms.core.steering_pipeline import SteeringPipeline
from steerability.algorithms.input_control.common.generation import generate_with_system_prompt
from steerability.algorithms.input_control.gepa import GEPA

TASK_MODEL = "google/gemma-3-4b-it"
REFLECTION_MODEL = "google/gemma-3-12b-it" 

import string import pandas as pd import torch from steerability.algorithms.core.steering_pipeline import SteeringPipeline from steerability.algorithms.input_control.common.generation import generate_with_system_prompt from steerability.algorithms.input_control.gepa import GEPA TASK_MODEL = "google/gemma-3-4b-it" REFLECTION_MODEL = "google/gemma-3-12b-it"

Task¶

The task is natural-language question-answering. The output rule (all lowercase, no punctuation) is enforced via the scorer and the feedback string. GEPA uses this feedback to update the system prompt in order to enforce the rule.

Scoring gives equal weight to answering correctly and following the rule. The small model answers questions correctly but with incorrect/invalid style, so the bare seed scores are 0.5 (right answer, wrong style). The feedback names the specific style violations so reflection can learn the rule from rollouts.

In [4]:

def follows_rule(output):
    t = output.strip()
    return t != "" and t == t.lower() and not any(c in string.punctuation for c in t)

def is_correct(output, row):
    return row["reference"].lower() in output.lower()

def score_row(output, row):
    return 0.5 * float(is_correct(output, row)) + 0.5 * float(follows_rule(output))

def feedback_row(output, row):
    t = output.strip()
    issues = []
    if t != t.lower():
        issues.append("contains uppercase letters")
    if any(c in string.punctuation for c in t):
        issues.append("contains punctuation")
    return "; ".join([
        f"output={t!r}",
        "answer present" if is_correct(output, row) else f"missing expected answer {row['reference']!r}",
        "required style: all lowercase, no punctuation",
        "style issues: " + (" and ".join(issues) if issues else "none"),
    ])

train_set = [
    {"question": "What is the capital of France?", "reference": "paris"},
    {"question": "What is the largest planet in the solar system?", "reference": "jupiter"},
    {"question": "Who wrote Romeo and Juliet?", "reference": "shakespeare"},
    {"question": "What planet is known as the Red Planet?", "reference": "mars"},
    {"question": "What is the capital of Japan?", "reference": "tokyo"},
    {"question": "What is the chemical symbol for gold?", "reference": "au"},
    {"question": "Who developed the theory of general relativity?", "reference": "einstein"},
    {"question": "What is the capital of Egypt?", "reference": "cairo"},
    {"question": "What is the closest planet to the sun?", "reference": "mercury"},
    {"question": "Who painted the Mona Lisa?", "reference": "vinci"},
    {"question": "What is the capital of Spain?", "reference": "madrid"},
    {"question": "What gas do plants absorb from the atmosphere?", "reference": "carbon dioxide"},
]

held_out = [
    {"question": "What is the capital of Italy?", "reference": "rome"},
    {"question": "What is the capital of Germany?", "reference": "berlin"},
    {"question": "Who wrote Hamlet?", "reference": "shakespeare"},
    {"question": "What is the smallest planet in the solar system?", "reference": "mercury"},
    {"question": "Who proposed the laws of motion?", "reference": "newton"},
    {"question": "What is the capital of Canada?", "reference": "ottawa"},
    {"question": "What is the chemical symbol for sodium?", "reference": "na"},
    {"question": "What is the sixth planet from the sun?", "reference": "saturn"},
    {"question": "What is the capital of Russia?", "reference": "moscow"},
    {"question": "Who developed the polio vaccine?", "reference": "salk"},
    {"question": "What is the capital of Australia?", "reference": "canberra"},
    {"question": "What gas do humans exhale?", "reference": "carbon dioxide"},
]

def follows_rule(output): t = output.strip() return t != "" and t == t.lower() and not any(c in string.punctuation for c in t) def is_correct(output, row): return row["reference"].lower() in output.lower() def score_row(output, row): return 0.5 * float(is_correct(output, row)) + 0.5 * float(follows_rule(output)) def feedback_row(output, row): t = output.strip() issues = [] if t != t.lower(): issues.append("contains uppercase letters") if any(c in string.punctuation for c in t): issues.append("contains punctuation") return "; ".join([ f"output={t!r}", "answer present" if is_correct(output, row) else f"missing expected answer {row['reference']!r}", "required style: all lowercase, no punctuation", "style issues: " + (" and ".join(issues) if issues else "none"), ]) train_set = [ {"question": "What is the capital of France?", "reference": "paris"}, {"question": "What is the largest planet in the solar system?", "reference": "jupiter"}, {"question": "Who wrote Romeo and Juliet?", "reference": "shakespeare"}, {"question": "What planet is known as the Red Planet?", "reference": "mars"}, {"question": "What is the capital of Japan?", "reference": "tokyo"}, {"question": "What is the chemical symbol for gold?", "reference": "au"}, {"question": "Who developed the theory of general relativity?", "reference": "einstein"}, {"question": "What is the capital of Egypt?", "reference": "cairo"}, {"question": "What is the closest planet to the sun?", "reference": "mercury"}, {"question": "Who painted the Mona Lisa?", "reference": "vinci"}, {"question": "What is the capital of Spain?", "reference": "madrid"}, {"question": "What gas do plants absorb from the atmosphere?", "reference": "carbon dioxide"}, ] held_out = [ {"question": "What is the capital of Italy?", "reference": "rome"}, {"question": "What is the capital of Germany?", "reference": "berlin"}, {"question": "Who wrote Hamlet?", "reference": "shakespeare"}, {"question": "What is the smallest planet in the solar system?", "reference": "mercury"}, {"question": "Who proposed the laws of motion?", "reference": "newton"}, {"question": "What is the capital of Canada?", "reference": "ottawa"}, {"question": "What is the chemical symbol for sodium?", "reference": "na"}, {"question": "What is the sixth planet from the sun?", "reference": "saturn"}, {"question": "What is the capital of Russia?", "reference": "moscow"}, {"question": "Who developed the polio vaccine?", "reference": "salk"}, {"question": "What is the capital of Australia?", "reference": "canberra"}, {"question": "What gas do humans exhale?", "reference": "carbon dioxide"}, ]

GEPA with default reflection¶

Under default reflection (reflection_lm=None) reflection is carried out by the task model. We build a steering pipeline around the task model and a GEPA control object. The format_query contains the question (rule is not visible to the model). A progress_callback records one entry per search step into trace_default to allow for inspection of how the method is working.

Also note that the minibatch and Pareto-set sizes are set larger than the defaults so the accept/reject signal is averaged over enough examples to be reliable.

In [5]:

trace_default = []

gepa_default = GEPA(
    seed_instruction="Answer the question.",
    train_set=train_set,
    row_scorer=score_row,
    feedback_fn=lambda output, row, score: feedback_row(output, row),
    format_query=lambda row: row["question"],
    gen_kwargs={"max_new_tokens": 32, "do_sample": False},
    budget=300,
    minibatch_size=6,
    pareto_set_size=8,
    reflection_lm=None,  # defaults to the task model
    progress_callback=trace_default.append,
    seed=0,
)

pipeline_default = SteeringPipeline(
    model_name_or_path=TASK_MODEL,
    controls=[gepa_default],
    hf_model_kwargs={"dtype": torch.bfloat16},
    device_map="auto",
)
pipeline_default.steer()

SEED_INSTRUCTION = "Answer the question."
print("Seed instruction:\n")
print(SEED_INSTRUCTION)
print("\nOptimized instruction (default reflection):\n")
print(gepa_default.memory["instruction"])

trace_default = [] gepa_default = GEPA( seed_instruction="Answer the question.", train_set=train_set, row_scorer=score_row, feedback_fn=lambda output, row, score: feedback_row(output, row), format_query=lambda row: row["question"], gen_kwargs={"max_new_tokens": 32, "do_sample": False}, budget=300, minibatch_size=6, pareto_set_size=8, reflection_lm=None, # defaults to the task model progress_callback=trace_default.append, seed=0, ) pipeline_default = SteeringPipeline( model_name_or_path=TASK_MODEL, controls=[gepa_default], hf_model_kwargs={"dtype": torch.bfloat16}, device_map="auto", ) pipeline_default.steer() SEED_INSTRUCTION = "Answer the question." print("Seed instruction:\n") print(SEED_INSTRUCTION) print("\nOptimized instruction (default reflection):\n") print(gepa_default.memory["instruction"])

Seed instruction:

Answer the question.

Optimized instruction (default reflection):

Answer the question concisely and directly. Provide only the factual answer to the question. Do not include follow-up questions, conversational elements, or extraneous information. Use sentence case, and avoid punctuation.

Specifically, the task is to provide factual answers to questions. The answers should be brief, direct statements of fact. 

Here's what I’ve observed from the examples:

*   **Format:** The input is a question. The output is a single sentence answering the question.
*   **Style:** The output should be entirely lowercase and contain no punctuation.
*   **Content:** Answers should consist of the specific fact requested in the question (e.g., a name, a place, a chemical compound). Do not elaborate beyond the core answer.
*   **Strategy:** The assistant appears to utilize a simple lookup and direct response strategy for factual questions.

Example:

Input: What is the highest mountain in the world?
Output: Mount Everest is the highest mountain in the world.

Inspecting the search¶

The trace contains one row per step where step 0 is the seed. Each subsequent row is a mutation attempt with the parent's minibatch mean, the candidate's minibatch mean, whether it was accepted, the pool size, and best_mean (over the Pareto set so far). The seed starts around 0.5 (correct answers, wrong style).

In [6]:

pd.DataFrame([
    {
        "step": r["step"],
        "event": r["event"],
        "parent_idx": r["parent_idx"],
        "parent_score": None if r["parent_score"] is None else round(r["parent_score"], 3),
        "candidate_score": None if r["candidate_score"] is None else round(r["candidate_score"], 3),
        "accepted": r["accepted"],
        "pool_size": r["pool_size"],
        "best_mean": round(r["best_mean"], 3),
    }
    for r in trace_default
])

pd.DataFrame([ { "step": r["step"], "event": r["event"], "parent_idx": r["parent_idx"], "parent_score": None if r["parent_score"] is None else round(r["parent_score"], 3), "candidate_score": None if r["candidate_score"] is None else round(r["candidate_score"], 3), "accepted": r["accepted"], "pool_size": r["pool_size"], "best_mean": round(r["best_mean"], 3), } for r in trace_default ])

Out[6]:

	step	event	parent_idx	parent_score	candidate_score	accepted	pool_size	best_mean
0	0	seed	NaN	NaN	NaN	True	1	0.5
1	1	reject	0.0	0.5	0.500	False	1	0.5
2	2	reject	0.0	0.5	0.500	False	1	0.5
3	3	reject	0.0	0.5	0.500	False	1	0.5
4	4	reject	0.0	0.5	0.500	False	1	0.5
5	5	reject	0.0	0.5	0.500	False	1	0.5
6	6	reject	0.0	0.5	0.500	False	1	0.5
7	7	accept	0.0	0.5	1.000	True	2	1.0
8	8	reject	1.0	1.0	1.000	False	2	1.0
9	9	reject	1.0	1.0	1.000	False	2	1.0
10	10	reject	1.0	1.0	1.000	False	2	1.0
11	11	reject	1.0	1.0	0.875	False	2	1.0
12	12	reject	1.0	1.0	1.000	False	2	1.0
13	13	reject	1.0	1.0	0.875	False	2	1.0
14	14	reject	1.0	1.0	0.875	False	2	1.0
15	15	reject	1.0	1.0	1.000	False	2	1.0
16	16	reject	1.0	1.0	1.000	False	2	1.0
17	17	reject	1.0	1.0	1.000	False	2	1.0
18	18	reject	1.0	1.0	1.000	False	2	1.0
19	19	reject	1.0	1.0	1.000	False	2	1.0
20	20	reject	1.0	1.0	1.000	False	2	1.0
21	21	reject	1.0	1.0	1.000	False	2	1.0
22	22	reject	1.0	1.0	0.500	False	2	1.0
23	23	reject	1.0	1.0	1.000	False	2	1.0
24	24	reject	1.0	1.0	1.000	False	2	1.0
25	25	reject	1.0	1.0	1.000	False	2	1.0
26	26	reject	1.0	1.0	1.000	False	2	1.0
27	27	reject	1.0	1.0	1.000	False	2	1.0
28	28	reject	1.0	1.0	0.500	False	2	1.0
29	29	reject	1.0	1.0	1.000	False	2	1.0
30	30	reject	1.0	1.0	0.875	False	2	1.0
31	31	reject	1.0	1.0	1.000	False	2	1.0
32	32	reject	1.0	1.0	1.000	False	2	1.0
33	33	reject	1.0	1.0	1.000	False	2	1.0
34	34	reject	1.0	1.0	1.000	False	2	1.0
35	35	reject	1.0	1.0	1.000	False	2	1.0

Filtering the trace to the seed and the accepted candidates shows the instruction evolving from the bare seed into an instruction that effectively enforces the lowercase + no-punctuation rule.

In [7]:

for r in trace_default:
    if r["event"] in {"seed", "accept"}:
        label = "seed" if r["event"] == "seed" else f"accepted (step {r['step']})"
        print(f"[{label}]")
        print(r["proposed"].strip())
        print()

for r in trace_default: if r["event"] in {"seed", "accept"}: label = "seed" if r["event"] == "seed" else f"accepted (step {r['step']})" print(f"[{label}]") print(r["proposed"].strip()) print()

[seed]
Answer the question.

[accepted (step 7)]
Answer the question concisely and directly. Provide only the factual answer to the question. Do not include follow-up questions, conversational elements, or extraneous information. Use sentence case, and avoid punctuation.

Specifically, the task is to provide factual answers to questions. The answers should be brief, direct statements of fact. 

Here's what I’ve observed from the examples:

*   **Format:** The input is a question. The output is a single sentence answering the question.
*   **Style:** The output should be entirely lowercase and contain no punctuation.
*   **Content:** Answers should consist of the specific fact requested in the question (e.g., a name, a place, a chemical compound). Do not elaborate beyond the core answer.
*   **Strategy:** The assistant appears to utilize a simple lookup and direct response strategy for factual questions.

Example:

Input: What is the highest mountain in the world?
Output: Mount Everest is the highest mountain in the world.

Held-out evaluation¶

We now evaluate how well the discovered system prompt performs on unseen questions. We generate under the seed and under the optimized instruction across the held-out set and send only the question (the rule stays hidden).

In [8]:

def evaluate(pipeline, instruction, items, max_new_tokens=32):
    outputs = generate_with_system_prompt(
        pipeline.model,
        pipeline.tokenizer,
        instruction,
        [item["question"] for item in items],
        gen_kwargs={"max_new_tokens": max_new_tokens, "do_sample": False},
    )
    rows = [
        {
            "question": item["question"],
            "output": out.strip(),
            "correct": is_correct(out, item),
            "follows_rule": follows_rule(out),
            "score": score_row(out, item),
        }
        for out, item in zip(outputs, items)
    ]
    return pd.DataFrame(rows)

eval_seed = evaluate(pipeline_default, SEED_INSTRUCTION, held_out)
eval_default = evaluate(pipeline_default, gepa_default.memory["instruction"], held_out)

def evaluate(pipeline, instruction, items, max_new_tokens=32): outputs = generate_with_system_prompt( pipeline.model, pipeline.tokenizer, instruction, [item["question"] for item in items], gen_kwargs={"max_new_tokens": max_new_tokens, "do_sample": False}, ) rows = [ { "question": item["question"], "output": out.strip(), "correct": is_correct(out, item), "follows_rule": follows_rule(out), "score": score_row(out, item), } for out, item in zip(outputs, items) ] return pd.DataFrame(rows) eval_seed = evaluate(pipeline_default, SEED_INSTRUCTION, held_out) eval_default = evaluate(pipeline_default, gepa_default.memory["instruction"], held_out)

The mean score and the rule following rate (across the held-out set) are computed as follows.

In [9]:

summary = pd.DataFrame({
    "mean score": [eval_seed["score"].mean(), eval_default["score"].mean()],
    "follows rule": [eval_seed["follows_rule"].mean(), eval_default["follows_rule"].mean()],
    "answer correct": [eval_seed["correct"].mean(), eval_default["correct"].mean()],
}, index=["seed", "optimized"]).round(3)
summary

summary = pd.DataFrame({ "mean score": [eval_seed["score"].mean(), eval_default["score"].mean()], "follows rule": [eval_seed["follows_rule"].mean(), eval_default["follows_rule"].mean()], "answer correct": [eval_seed["correct"].mean(), eval_default["correct"].mean()], }, index=["seed", "optimized"]).round(3) summary

Out[9]:

	mean score	follows rule	answer correct
seed	0.500	0.000	1.000
optimized	0.917	0.917	0.917

We also print some specific cases of held-out questions where the seed breaks the rule but the optimized instruction follows it (while keeping the answer correct).

In [10]:

comparison = eval_seed[["question", "output", "follows_rule"]].rename(
    columns={"output": "seed_output", "follows_rule": "seed_follows_rule"}
)
comparison["optimized_output"] = eval_default["output"]
comparison["optimized_follows_rule"] = eval_default["follows_rule"]

fixed = comparison[(~comparison["seed_follows_rule"]) & (comparison["optimized_follows_rule"])]
fixed[["question", "seed_output", "optimized_output"]].head(10).reset_index(drop=True)

comparison = eval_seed[["question", "output", "follows_rule"]].rename( columns={"output": "seed_output", "follows_rule": "seed_follows_rule"} ) comparison["optimized_output"] = eval_default["output"] comparison["optimized_follows_rule"] = eval_default["follows_rule"] fixed = comparison[(~comparison["seed_follows_rule"]) & (comparison["optimized_follows_rule"])] fixed[["question", "seed_output", "optimized_output"]].head(10).reset_index(drop=True)

Out[10]:

	question	seed_output	optimized_output
0	What is the capital of Italy?	The capital of Italy is **Rome**.	rome is the capital of italy
1	What is the capital of Germany?	The capital of Germany is **Berlin**.	berlin is the capital of germany
2	Who wrote Hamlet?	William Shakespeare wrote Hamlet. 😊 \n\nIt’s o...	william shakespeare wrote hamlet
3	Who proposed the laws of motion?	Sir Isaac Newton proposed the laws of motion. ...	sir isaac newton proposed the laws of motion
4	What is the capital of Canada?	The capital of Canada is **Ottawa**. \n\nIt’s ...	ottawa is the capital of canada
5	What is the chemical symbol for sodium?	Na	na
6	What is the sixth planet from the sun?	The sixth planet from the sun is **Saturn**. \...	uranus is the sixth planet from the sun
7	What is the capital of Russia?	The capital of Russia is **Moscow**.	moscow is the capital of russia
8	Who developed the polio vaccine?	The development of the polio vaccine is a comp...	jonas salk developed the polio vaccine
9	What is the capital of Australia?	The capital of Australia is **Canberra**. \n\n...	canberra is the capital of australia

GEPA under a stronger reflector¶

GEPA's reflection step can use a different model from the one being steered. Here we keep the same 4B task model and the same seed (so minibatch sampling is the same) but swap the reflector for gemma-3-12b-it. The question is whether a stronger reflector finds a cleaner or better-generalizing instruction.

In [11]:

from transformers import AutoModelForCausalLM, AutoTokenizer

del pipeline_default
if torch.cuda.is_available():
    torch.cuda.empty_cache()

reflection_model = AutoModelForCausalLM.from_pretrained(
    REFLECTION_MODEL, dtype=torch.bfloat16, device_map="auto"
)
reflection_tokenizer = AutoTokenizer.from_pretrained(REFLECTION_MODEL)

from transformers import AutoModelForCausalLM, AutoTokenizer del pipeline_default if torch.cuda.is_available(): torch.cuda.empty_cache() reflection_model = AutoModelForCausalLM.from_pretrained( REFLECTION_MODEL, dtype=torch.bfloat16, device_map="auto" ) reflection_tokenizer = AutoTokenizer.from_pretrained(REFLECTION_MODEL)

In [12]:

trace_strong = []

gepa_strong = GEPA(
    seed_instruction=SEED_INSTRUCTION,
    train_set=train_set,
    row_scorer=score_row,
    feedback_fn=lambda output, row, score: feedback_row(output, row),
    format_query=lambda row: row["question"],
    gen_kwargs={"max_new_tokens": 32, "do_sample": False},
    budget=300,
    minibatch_size=6,
    pareto_set_size=8,
    reflection_lm=reflection_model,
    reflection_tokenizer=reflection_tokenizer,
    progress_callback=trace_strong.append,
    seed=0,
)

pipeline_strong = SteeringPipeline(
    model_name_or_path=TASK_MODEL,
    controls=[gepa_strong],
    hf_model_kwargs={"dtype": torch.bfloat16},
    device_map="auto",
)
pipeline_strong.steer()

print("Optimized instruction (stronger reflection):\n")
print(gepa_strong.memory["instruction"])

trace_strong = [] gepa_strong = GEPA( seed_instruction=SEED_INSTRUCTION, train_set=train_set, row_scorer=score_row, feedback_fn=lambda output, row, score: feedback_row(output, row), format_query=lambda row: row["question"], gen_kwargs={"max_new_tokens": 32, "do_sample": False}, budget=300, minibatch_size=6, pareto_set_size=8, reflection_lm=reflection_model, reflection_tokenizer=reflection_tokenizer, progress_callback=trace_strong.append, seed=0, ) pipeline_strong = SteeringPipeline( model_name_or_path=TASK_MODEL, controls=[gepa_strong], hf_model_kwargs={"dtype": torch.bfloat16}, device_map="auto", ) pipeline_strong.steer() print("Optimized instruction (stronger reflection):\n") print(gepa_strong.memory["instruction"])

Optimized instruction (stronger reflection):

Answer the question concisely. Respond with only the answer, in all lowercase letters, and without any punctuation or extra text (e.g., no "Do you want to know anything else?", no emoticons, no conversational fillers). The assistant appears to be answering factual questions.

As with the default reflector, the trace records one row per step. The stronger reflector tends to reach best_mean in fewer accepts, often discovering the rule in a single revision rather than refining it over several.

In [13]:

pd.DataFrame([
    {
        "step": r["step"],
        "event": r["event"],
        "parent_idx": r["parent_idx"],
        "parent_score": None if r["parent_score"] is None else round(r["parent_score"], 3),
        "candidate_score": None if r["candidate_score"] is None else round(r["candidate_score"], 3),
        "accepted": r["accepted"],
        "pool_size": r["pool_size"],
        "best_mean": round(r["best_mean"], 3),
    }
    for r in trace_strong
])

pd.DataFrame([ { "step": r["step"], "event": r["event"], "parent_idx": r["parent_idx"], "parent_score": None if r["parent_score"] is None else round(r["parent_score"], 3), "candidate_score": None if r["candidate_score"] is None else round(r["candidate_score"], 3), "accepted": r["accepted"], "pool_size": r["pool_size"], "best_mean": round(r["best_mean"], 3), } for r in trace_strong ])

Out[13]:

	step	event	parent_idx	parent_score	candidate_score	accepted	pool_size	best_mean
0	0	seed	NaN	NaN	NaN	True	1	0.5
1	1	accept	0.0	0.5	1.00	True	2	1.0
2	2	reject	1.0	1.0	1.00	False	2	1.0
3	3	reject	1.0	1.0	1.00	False	2	1.0
4	4	reject	1.0	1.0	1.00	False	2	1.0
5	5	reject	1.0	1.0	1.00	False	2	1.0
6	6	reject	1.0	1.0	1.00	False	2	1.0
7	7	reject	1.0	1.0	1.00	False	2	1.0
8	8	reject	1.0	1.0	0.75	False	2	1.0
9	9	reject	1.0	1.0	1.00	False	2	1.0
10	10	reject	1.0	1.0	1.00	False	2	1.0
11	11	reject	1.0	1.0	1.00	False	2	1.0
12	12	reject	1.0	1.0	1.00	False	2	1.0
13	13	reject	1.0	1.0	1.00	False	2	1.0
14	14	reject	1.0	1.0	1.00	False	2	1.0
15	15	reject	1.0	1.0	1.00	False	2	1.0
16	16	reject	1.0	1.0	1.00	False	2	1.0
17	17	reject	1.0	1.0	1.00	False	2	1.0
18	18	reject	1.0	1.0	1.00	False	2	1.0
19	19	reject	1.0	1.0	1.00	False	2	1.0
20	20	reject	1.0	1.0	1.00	False	2	1.0
21	21	reject	1.0	1.0	1.00	False	2	1.0
22	22	reject	1.0	1.0	1.00	False	2	1.0
23	23	reject	1.0	1.0	1.00	False	2	1.0
24	24	reject	1.0	1.0	1.00	False	2	1.0
25	25	reject	1.0	1.0	1.00	False	2	1.0
26	26	reject	1.0	1.0	1.00	False	2	1.0
27	27	reject	1.0	1.0	1.00	False	2	1.0
28	28	reject	1.0	1.0	1.00	False	2	1.0
29	29	reject	1.0	1.0	1.00	False	2	1.0
30	30	reject	1.0	1.0	1.00	False	2	1.0
31	31	reject	1.0	1.0	1.00	False	2	1.0
32	32	reject	1.0	1.0	1.00	False	2	1.0
33	33	reject	1.0	1.0	1.00	False	2	1.0
34	34	reject	1.0	1.0	1.00	False	2	1.0
35	35	reject	1.0	1.0	1.00	False	2	1.0

We evaluate the stronger reflection on the same held-out set with the same task model, then compare the two reflection types.

In [14]:

eval_strong = evaluate(pipeline_strong, gepa_strong.memory["instruction"], held_out)

def search_stats(trace):
    accepts = [r for r in trace if r["event"] == "accept"]
    return {
        "first accept step": accepts[0]["step"] if accepts else None,
        "num accepts": len(accepts),
        "final best_mean": round(trace[-1]["best_mean"], 3),
    }

comparison = pd.DataFrame({
    "held-out mean score": [eval_default["score"].mean(), eval_strong["score"].mean()],
    "held-out follows rule": [eval_default["follows_rule"].mean(), eval_strong["follows_rule"].mean()],
    "first accept step": [search_stats(trace_default)["first accept step"], search_stats(trace_strong)["first accept step"]],
    "num accepts": [search_stats(trace_default)["num accepts"], search_stats(trace_strong)["num accepts"]],
    "final best_mean": [search_stats(trace_default)["final best_mean"], search_stats(trace_strong)["final best_mean"]],
}, index=["default reflector (4B)", "strong reflector (12B)"]).round(3)
comparison

eval_strong = evaluate(pipeline_strong, gepa_strong.memory["instruction"], held_out) def search_stats(trace): accepts = [r for r in trace if r["event"] == "accept"] return { "first accept step": accepts[0]["step"] if accepts else None, "num accepts": len(accepts), "final best_mean": round(trace[-1]["best_mean"], 3), } comparison = pd.DataFrame({ "held-out mean score": [eval_default["score"].mean(), eval_strong["score"].mean()], "held-out follows rule": [eval_default["follows_rule"].mean(), eval_strong["follows_rule"].mean()], "first accept step": [search_stats(trace_default)["first accept step"], search_stats(trace_strong)["first accept step"]], "num accepts": [search_stats(trace_default)["num accepts"], search_stats(trace_strong)["num accepts"]], "final best_mean": [search_stats(trace_default)["final best_mean"], search_stats(trace_strong)["final best_mean"]], }, index=["default reflector (4B)", "strong reflector (12B)"]).round(3) comparison

Out[14]:

	held-out mean score	held-out follows rule	first accept step	num accepts	final best_mean
default reflector (4B)	0.917	0.917	7	1	1.0
strong reflector (12B)	0.958	1.000	1	1	1.0

Generally, gepa_strong shows a marginal improvment over gepa_default althought the gains are small in this particular demo due to the simplicity of the constraints.

In [15]:

print("Default reflector (4B)\n")
print(gepa_default.memory["instruction"])
print("\n" + "-" * 80 + "\n")
print("Strong reflector (12B)\n")
print(gepa_strong.memory["instruction"])

print("Default reflector (4B)\n") print(gepa_default.memory["instruction"]) print("\n" + "-" * 80 + "\n") print("Strong reflector (12B)\n") print(gepa_strong.memory["instruction"])

Default reflector (4B)

Answer the question concisely and directly. Provide only the factual answer to the question. Do not include follow-up questions, conversational elements, or extraneous information. Use sentence case, and avoid punctuation.

Specifically, the task is to provide factual answers to questions. The answers should be brief, direct statements of fact. 

Here's what I’ve observed from the examples:

*   **Format:** The input is a question. The output is a single sentence answering the question.
*   **Style:** The output should be entirely lowercase and contain no punctuation.
*   **Content:** Answers should consist of the specific fact requested in the question (e.g., a name, a place, a chemical compound). Do not elaborate beyond the core answer.
*   **Strategy:** The assistant appears to utilize a simple lookup and direct response strategy for factual questions.

Example:

Input: What is the highest mountain in the world?
Output: Mount Everest is the highest mountain in the world.

--------------------------------------------------------------------------------

Strong reflector (12B)

Answer the question concisely. Respond with only the answer, in all lowercase letters, and without any punctuation or extra text (e.g., no "Do you want to know anything else?", no emoticons, no conversational fillers). The assistant appears to be answering factual questions.

Summary¶

This notebook demonstrated GEPA (Agrawal et al., 2025), an input control that optimizes a single system prompt by reflecting on natural-language feedback on outputs.

1. At steer time, GEPA runs a reflective search by rolling out the current candidate out on a minibatch of train_set rows. Each output is scored with row_scorer and feedback (on failures) is provided via the feedback_fn which asks the reflection model to propose a revised instruction. The revision is kept only if its minibatch mean beats the mean of the parent. GEPA maintains a Pareto set of strong candidates.
2. The output rule (all lowercase, no punctuation) was intentionally hidden from the model as part of the task in order to demonstrate how GEPA updates the system prompt. The search trace showed the instruction evolving from the bare seed to one that has satisfied the rule (best_mean increased as candidates were accepted). Evaluation on a held-out set showed generalization to unseen questions.
3. The above implementation of GEPA allows for specification of a stronger reflection_lm for more complex instructions; gains were marginal in this notebook due to the simplicity of the constraints.

GEPA fits tasks where a single system prompt can be improved from feedback and you can score each example. The failure should be fixable via an instruction (e.g., a rule, format, or strategy the prompt can encode). Additionally, GEPA requires a per-example row_scorer and ideally a textual feedback_fn that rewards the desired behavior. This toolkit's implementation of GEPA does not cover optimizing a multi-prompt system (with module-level credit assignment); please see the original implementation at gepa-ai/gepa (or DSPy's GEPA integration) for such functionality.