Reflective Evolution

# Traditional optimizer
candidate = "def solve(x): return x + 1"
score = evaluate(candidate)  # Returns: 0.3
# Now what? We know it's bad, but not why.

# GEPA evaluator
def evaluate(candidate, example):
    result = run_code(candidate, example["input"])
    score = 1.0 if result == example["expected"] else 0.0
    
    return score, {
        "Input": example["input"],
        "Output": result,
        "Expected": example["expected"],
        "Error": result.error if hasattr(result, 'error') else None
    }

# From candidate_selector.py
class ParetoCandidateSelector:
    def select_candidate_idx(self, state: GEPAState) -> int:
        # Get all candidates on the Pareto front
        pareto_programs = state.get_pareto_front_programs()
        
        # Randomly select one
        return self.rng.choice(list(pareto_programs))

# From reflective_mutation.py
subsample_ids = batch_sampler.next_minibatch_ids(trainset, state)
minibatch = trainset.fetch(subsample_ids)

# Evaluate with traces
eval_result = adapter.evaluate(
    batch=minibatch,
    candidate=current_candidate,
    capture_traces=True  # Crucial: captures execution details
)

# From adapter.py protocol
reflective_dataset = adapter.make_reflective_dataset(
    candidate=current_candidate,
    eval_batch=eval_result,
    components_to_update=["system_prompt"]  # Which params to evolve
)

# Returns a structured dataset, e.g.:
{
    "system_prompt": [
        {
            "Inputs": {
                "question": "What is 2+2?",
                "context": "Elementary arithmetic"
            },
            "Generated Outputs": "The answer is 5",
            "Feedback": "Incorrect. Expected: 4. Error: Basic arithmetic failure."
        },
        {
            "Inputs": {"question": "What is 10*10?"},
            "Generated Outputs": "100",
            "Feedback": "Correct!"
        },
        # ... more examples
    ]
}

# From instruction_proposal.py
def propose_new_texts(
    candidate: dict[str, str],
    reflective_dataset: dict[str, list[dict]],
    components_to_update: list[str]
) -> dict[str, str]:
    new_texts = {}
    
    for component_name in components_to_update:
        # Build prompt with current text + reflective dataset
        prompt = build_reflection_prompt(
            current_text=candidate[component_name],
            examples=reflective_dataset[component_name]
        )
        
        # LLM proposes improvement
        new_text = reflection_lm(prompt)
        new_texts[component_name] = new_text
    
    return new_texts

I am optimizing a parameter in my system. The current parameter value is:

Below is evaluation data showing how this parameter performed:

Your task is to propose a new, improved parameter value.

Carefully analyze all evaluation data. Look for patterns:
- Performance metrics and correlations
- Recurring issues or failure patterns
- Successful behaviors to preserve
- Domain-specific constraints

Based on your analysis, propose a new parameter value that addresses
the identified issues while maintaining what works well.

Provide the new parameter value within ``` blocks.

# From component_selector.py
class RoundRobinReflectionComponentSelector:
    def select_components(
        self, candidate: dict[str, str], state: GEPAState
    ) -> list[str]:
        # Cycle through components in order
        predictor_names = list(candidate.keys())
        next_id = state.named_predictor_id_to_update_next_for_program_candidate[
            candidate_idx
        ]
        selected = [predictor_names[next_id]]
        # Update for next iteration
        state.named_predictor_id_to_update_next_for_program_candidate[
            candidate_idx
        ] = (next_id + 1) % len(predictor_names)
        return selected

class AllReflectionComponentSelector:
    def select_components(
        self, candidate: dict[str, str], state: GEPAState
    ) -> list[str]:
        # Update all components together
        return list(candidate.keys())

from gepa.proposer.reflective_mutation.base import ReflectionComponentSelector

class CustomSelector(ReflectionComponentSelector):
    def select_components(
        self, candidate: dict[str, str], state: GEPAState
    ) -> list[str]:
        # Your custom logic
        # Example: prioritize components with lower scores
        worst_performing = find_worst_component(state)
        return [worst_performing]

config = GEPAConfig(
    reflection=ReflectionConfig(
        module_selector=CustomSelector()
    )
)

config = GEPAConfig(
    reflection=ReflectionConfig(
        skip_perfect_score=True,
        perfect_score=1.0  # What counts as "perfect"
    )
)

config = GEPAConfig(
    reflection=ReflectionConfig(
        reflection_prompt_template={
            "system_prompt": """
                You are optimizing a system prompt for math reasoning.
                Current prompt:
                <curr_param>
                
                Performance data:
                <side_info>
                
                Propose an improved prompt that fixes arithmetic errors
                while preserving chain-of-thought reasoning.
            """,
            "few_shot_examples": """
                You are curating few-shot examples.
                Current examples:
                <curr_param>
                
                Performance data:
                <side_info>
                
                Propose better examples that cover edge cases.
            """
        }
    )
)

from collections.abc import Mapping, Sequence

def my_proposer(
    candidate: dict[str, str],
    reflective_dataset: Mapping[str, Sequence[dict]],
    components_to_update: list[str]
) -> dict[str, str]:
    """
    Custom proposal logic.
    Could use: DSPy signatures, fine-tuned models,
    rule-based systems, genetic operators, etc.
    """
    new_texts = {}
    for component in components_to_update:
        # Your custom logic
        new_texts[component] = generate_improvement(
            candidate[component],
            reflective_dataset[component]
        )
    return new_texts

config = GEPAConfig(
    reflection=ReflectionConfig(
        custom_candidate_proposer=my_proposer
    )
)

Iteration 5: Reflection identifies:
- Failing on base conversion problems (3/10 correct)
- Success on polynomial problems (8/10 correct)
- Common error: not enforcing digit constraints in base-b

Proposed improvement:
"For base conversion: enforce digit ranges strictly.
In base b, digits ∈ {0,...,b-1}. Translate positional
notation correctly: (abc)_b = a·b² + b·b + c..."

Result: Base conversion accuracy improves from 30% → 80%

config = GEPAConfig(
    reflection=ReflectionConfig(
        reflection_minibatch_size=3  # 1-3 for most tasks
    )
)

result = optimize_anything(
    seed_candidate=None,  # No starting point
    evaluator=evaluate,
    objective="Generate a Python function that reverses a string",
    background="Use only built-in Python, no imports. Handle Unicode correctly.",
    dataset=test_cases
)