Building Reflective Prompt Optimization with GEPA: Multi-Component Prompts, Structured Feedback, and Held-Out Validation

!pip set up -q gepa litellm
import os, re, json, random, getpass, textwrap
import litellm
import gepa.optimize_anything as oa
from gepa.optimize_anything import (
   optimize_anything, GEPAConfig, EngineConfig, ReflectionConfig,
)
litellm.suppress_debug_info = True
if not os.environ.get("OPENAI_API_KEY"):
   os.environ["OPENAI_API_KEY"] = getpass.getpass("Enter your OpenAI API key: ")
TASK_LM        = "openai/gpt-4o-mini"
REFLECTION_LM  = "openai/gpt-4.1"
MAX_METRIC_CALLS = 100

def make_problems(n, seed=0):
   rng = random.Random(seed)
   out = []
   for _ in vary(n):
       t = rng.selection(["discount", "travel", "wallet", "chain"])
       if t == "low cost":
           unit  = rng.selection([40, 60, 80, 120])
           qty   = rng.selection([5, 6, 8, 10])
           disc  = rng.selection([10, 20, 25, 50])
           whole = unit * qty
           gold  = whole - whole * disc // 100
           q = (f"A store sells notebooks at {unit} rupees every. You purchase {qty} "
                f"notebooks and get a {disc}% low cost on the entire invoice. "
                f"How many rupees do you pay in whole?")
       elif t == "journey":
           s1, h1 = rng.selection([40, 50, 60]), rng.selection([2, 3])
           s2, h2 = rng.selection([30, 45, 70]), rng.selection([1, 2, 3])
           gold = s1 * h1 + s2 * h2
           q = (f"A automobile drives at {s1} km/h for {h1} hours, then at {s2} km/h "
                f"for {h2} hours. What is the entire distance travelled, in km?")
       elif t == "pockets":
           tens   = rng.selection([3, 5, 7, 9])
           fifties= rng.selection([2, 4, 6])
           spent  = rng.selection([50, 80, 110, 150])
           gold = tens * 10 + fifties * 50 - spent
           q = (f"You have {tens} ten-rupee notes and {fifties} fifty-rupee "
                f"notes. You spend {spent} rupees. How many rupees are left?")
       else:
           x = rng.selection([6, 9, 12, 15]); y = rng.selection([4, 7, 10]); z = rng.selection([3, 8, 11])
           gold = x * 2 - y + z
           q = (f"Start with the quantity {x}. Double it, then subtract {y}, "
                f"then add {z}. What quantity do you finish with?")
       out.append({"query": q, "reply": gold})
   return out
all_problems = make_problems(18, seed=42)
random.Random(1).shuffle(all_problems)
trainset = all_problems[:12]
valset   = all_problems[12:]
print(f"Dataset: {len(trainset)} prepare / {len(valset)} val problemsn")

def build_system_prompt(candidate: dict) -> str:
   return (f"{candidate['instructions']}nn"
           f"OUTPUT FORMAT RULES:n{candidate['format_rules']}")
def call_task_lm(system_prompt: str, query: str) -> str:
   for try in vary(3):
       strive:
           r = litellm.completion(
               mannequin=TASK_LM,
               messages=[{"role": "system", "content": system_prompt},
                         {"role": "user",   "content": question}],
               temperature=0, max_tokens=600, timeout=60,
           )
           return r["choices"][0]["message"]["content"] or ""
       besides Exception as e:
           if try == 2:
               return f"[LM_ERROR] {e}"
   return ""
def parse_answers(textual content: str):
   formatted = re.search(r"####s*(-?d+)", textual content)
   all_nums  = re.findall(r"-?d+", textual content)
   fmt_val   = int(formatted.group(1)) if formatted else None
   last_val  = int(all_nums[-1]) if all_nums else None
   return fmt_val, last_val
def consider(candidate: dict, instance: dict):
   system = build_system_prompt(candidate)
   uncooked    = call_task_lm(system, instance["question"])
   gold   = instance["answer"]
   fmt_val, last_val = parse_answers(uncooked)
   if fmt_val shouldn't be None and fmt_val == gold:
       rating, fb = 1.0, "Correct and accurately formatted."
   elif fmt_val shouldn't be None and fmt_val != gold:
       rating, fb = 0.0, (f"WRONG ANSWER. You output '#### {fmt_val}' however the "
                         f"appropriate reply is {gold}. Re-check the arithmetic and "
                         f"the order of the steps.")
   elif last_val == gold:
       rating, fb = 0.5, (f"Right quantity ({gold}) however FORMAT VIOLATION: the ultimate "
                         f"line was not precisely '#### {gold}'. Always finish with a "
                         f"line of the shape '#### <integer>' and nothing else.")
   else:
       rating, fb = 0.0, (f"WRONG. Correct reply is {gold}. The mannequin's remaining "
                         f"quantity was {last_val}. Likely a multi-step reasoning "
                         f"slip; present every step and confirm earlier than answering.")
   oa.log(f"rating={rating} gold={gold} parsed_fmt={fmt_val} parsed_last={last_val}")
   side_info = {
       "suggestions": fb,
       "drawback": instance["question"],
       "gold_answer": gold,
       "model_output": uncooked[:500],
   }
   return rating, side_info
def eval_set(candidate, dataset, label=""):
   scores, precise, formatted = [], 0, 0
   for ex in dataset:
       s, data = consider(candidate, ex)
       scores.append(s)
       if s == 1.0: precise += 1; formatted += 1
       elif s == 0.5: formatted += 0
   acc = precise / len(dataset)
   avg = sum(scores) / len(dataset)
   print(f"  [{label}] avg_score={avg:.3f}  exact_correct+formatted={precise}/{len(dataset)}")
   return avg, acc

seed_candidate = {
   "directions": "Solve the mathematics drawback.",
   "format_rules": "Give the reply.",
}
print("=== BASELINE (seed immediate) ===")
print("Train:"); base_train = eval_set(seed_candidate, trainset, "prepare")
print("Val:  "); base_val   = eval_set(seed_candidate, valset,   "val")
print()
goal = (
   "Evolve a system immediate (the 'directions' and 'format_rules' fields) so a "
   "small LLM reliably solves multi-step arithmetic phrase issues AND at all times "
   "ends with a line of precisely the shape '#### <integer>'. Maximize the rating."
)
background = (
   "Scoring: 1.0 = appropriate quantity within the precise '#### <int>' format; 0.5 = appropriate "
   "quantity however unsuitable/lacking format; 0.0 = unsuitable quantity. Common failures are (a) not "
   "emitting the '####' line, and (b) order-of-operations or multi-step slips. The "
   "profitable immediate ought to pressure specific step-by-step work, a verification step, and "
   "a strict final-answer line."
)
config = GEPAConfig(
   engine=EngineConfig(
       max_metric_calls=MAX_METRIC_CALLS,
       max_workers=4,
       parallel=True,
       display_progress_bar=True,
       seed=0,
   ),
   reflection=ReflectionConfig(
       reflection_lm=REFLECTION_LM,
   ),
)
print("=== RUNNING GEPA (this calls the LLMs; ~1-4 min) ===")
outcome = optimize_anything(
   seed_candidate=seed_candidate,
   evaluator=consider,
   dataset=trainset,
   valset=valset,
   goal=goal,
   background=background,
   config=config,
)

finest = outcome.best_candidate
print("n" + "=" * 78)
print("OPTIMIZED CANDIDATE")
print("=" * 78)
print("n--- directions ---n" + textwrap.fill(finest["instructions"], 96))
print("n--- format_rules ---n" + textwrap.fill(finest["format_rules"], 96))
print("n" + "=" * 78)
print("BEFORE vs AFTER  (held-out validation set)")
print("=" * 78)
print("Seed immediate:"); _ = eval_set(seed_candidate, valset, "val-seed")
print("GEPA  immediate:"); _ = eval_set(finest,          valset, "val-gepa")
print(f"nBaseline val avg_score : {base_val[0]:.3f}")
print("n" + "=" * 78)
print("EVOLUTION HISTORY (candidate index -> val rating, mother and father)")
print("=" * 78)
cands   = getattr(outcome, "candidates", [])
vscores = getattr(outcome, "val_aggregate_scores", [])
mother and father = getattr(outcome, "mother and father", [None] * len(cands))
for i, sc in enumerate(vscores):
   par = mother and father[i] if i < len(mother and father) else None
   tag = "  <-- BEST" if cands and cands[i] == finest else ""
   print(f"  cand {i:second}: val_score={sc:.3f}  mother and father={par}{tag}")
print(f"nTotal metric calls used : {getattr(outcome, 'total_metric_calls', 'n/a')}")
print(f"Full validation evals   : {getattr(outcome, 'num_full_val_evals', 'n/a')}")
print("nDone. Try elevating MAX_METRIC_CALLS or swapping REFLECTION_LM for a stronger mannequin.")