GEPA: Optimize Anything

# Basic installation
pip install gepa

# With all optional dependencies
pip install gepa[all]

# With specific LLM backends
pip install gepa[openai,anthropic,google]

# Development installation
git clone https://github.com/gepa-ai/gepa.git
cd gepa
pip install -e ".[dev]"

from gepa import optimize, Artifact, Metric, EvalResult

# Define the artifact to optimize
artifact = Artifact(
    name="sort_function",
    template="""
def sort(arr: list[int]) -> list[int]:
    # Your sorting implementation here
    {{IMPLEMENTATION}}
""",
    language="python"
)

# Define the evaluation metric
def evaluate_sort(candidate: str, task: dict) -> EvalResult:
    """Evaluate sorting correctness and performance."""
    import time
    exec_globals = {}
    exec(candidate, exec_globals)
    sort_fn = exec_globals['sort']

    test_cases = task['test_cases']
    correct = 0
    total_time = 0.0

    for tc in test_cases:
        start = time.perf_counter()
        result = sort_fn(tc['input'].copy())
        elapsed = time.perf_counter() - start
        total_time += elapsed
        if result == tc['expected']:
            correct += 1

    accuracy = correct / len(test_cases)
    avg_time = total_time / len(test_cases)

    return EvalResult(
        scores={"accuracy": accuracy, "speed": 1.0 / (avg_time + 1e-9)},
        side_info={
            "failed_cases": [tc for tc, r in zip(test_cases, results) if r != tc['expected']],
            "avg_time_ms": avg_time * 1000,
        }
    )

# Run optimization
result = optimize(
    artifact=artifact,
    evaluate=evaluate_sort,
    metrics=[
        Metric("accuracy", direction="maximize", weight=0.8),
        Metric("speed", direction="maximize", weight=0.2),
    ],
    max_iterations=50,
    llm="claude-sonnet-4-20250514",
)

print(f"Best solution: {result.best.score}")
print(result.best.code)

+-----------------------------------------------------------------------------------+
|                              GEPA System Architecture                              |
+-----------------------------------------------------------------------------------+
|                                                                                    |
|  +------------------+     +-------------------+     +-------------------------+    |
|  |   User API       |     |   GEPAConfig      |     |   Artifact Registry     |    |
|  |                  |---->|                   |---->|                         |    |
|  |  optimize()      |     |  EngineConfig     |     |  Templates, Validators  |    |
|  |  Artifact()      |     |  ReflectionConfig |     |  Language Specs         |    |
|  |  Metric()        |     |  MergeConfig      |     |  Seed Solutions         |    |
|  +------------------+     |  RefinerConfig    |     +-------------------------+    |
|                           +-------------------+                                    |
|                                    |                                               |
|                                    v                                               |
|  +------------------------------------------------------------------------+        |
|  |                        Evolution Engine                                 |        |
|  |                                                                         |        |
|  |  +-------------------+    +--------------------+    +----------------+  |        |
|  |  | Population Manager|    | Pareto Frontier    |    | History Store  |  |        |
|  |  |                   |    |                    |    |                |  |        |
|  |  | - Candidates[]    |    | - Non-dominated    |    | - All evals   |  |        |
|  |  | - Islands[]       |    |   solutions        |    | - ASI records |  |        |
|  |  | - Selection       |    | - Dominance checks |    | - Genealogy   |  |        |
|  |  +-------------------+    +--------------------+    +----------------+  |        |
|  |           |                        |                       |            |        |
|  |           v                        v                       v            |        |
|  |  +------------------------------------------------------------------+  |        |
|  |  |                    Reflection Engine                               |  |        |
|  |  |                                                                    |  |        |
|  |  |  1. Select parent from Pareto frontier                            |  |        |
|  |  |  2. Sample minibatch of 2-3 evaluation examples                   |  |        |
|  |  |  3. Attach ASI (traces, errors, images) to reflection prompt      |  |        |
|  |  |  4. LLM proposes targeted mutation                                |  |        |
|  |  |  5. Validate & insert into population                            |  |        |
|  |  +------------------------------------------------------------------+  |        |
|  |           |                                                             |        |
|  |           v                                                             |        |
|  |  +------------------------------------------------------------------+  |        |
|  |  |                    Evaluation Pipeline                             |  |        |
|  |  |                                                                    |  |        |
|  |  |  [Cache Check] --> [Sandbox Exec] --> [Metric Compute] --> [ASI]  |  |        |
|  |  |       |                  |                   |               |     |  |        |
|  |  |  Content Hash      Docker/Process        Multi-metric     Traces  |  |        |
|  |  |  Dedup              Isolation             Aggregation      Errors  |  |        |
|  |  |                     max_workers           Pareto update    Images  |  |        |
|  |  +------------------------------------------------------------------+  |        |
|  +------------------------------------------------------------------------+        |
|                                    |                                               |
|                                    v                                               |
|  +------------------------------------------------------------------------+        |
|  |                     Stopping Controller                                 |        |
|  |                                                                         |        |
|  |  MaxMetricCalls | Timeout | NoImprovement | ScoreThreshold | Composite  |        |
|  +------------------------------------------------------------------------+        |
|                                    |                                               |
|                                    v                                               |
|  +------------------------------------------------------------------------+        |
|  |                     OptimizationResult                                  |        |
|  |                                                                         |        |
|  |  .best         - Best solution per primary metric                       |        |
|  |  .pareto_front - All non-dominated solutions                            |        |
|  |  .history      - Full evaluation history with ASI                       |        |
|  |  .stats        - Timing, cost, convergence metrics                      |        |
|  +------------------------------------------------------------------------+        |
+-----------------------------------------------------------------------------------+

from gepa import optimize, Artifact, Metric, SingleTaskConfig

result = optimize(
    artifact=Artifact(
        name="heuristic",
        template="def heuristic(state: dict) -> float:\n    {{BODY}}",
    ),
    evaluate=my_eval_fn,
    metrics=[Metric("score", direction="maximize")],
    task={"problem_instance": load_instance("tsp_100")},
    config=SingleTaskConfig(
        population_size=20,
        max_iterations=100,
        reflection_minibatch_size=3,
    ),
)

from gepa import optimize, Artifact, Metric, MultiTaskConfig

tasks = [
    {"name": "tsp_50", "instance": load_instance("tsp_50")},
    {"name": "tsp_100", "instance": load_instance("tsp_100")},
    {"name": "tsp_200", "instance": load_instance("tsp_200")},
]

result = optimize(
    artifact=Artifact(name="tsp_heuristic", template="..."),
    evaluate=eval_tsp,
    metrics=[Metric("tour_length", direction="minimize")],
    tasks=tasks,
    config=MultiTaskConfig(
        population_size=30,
        cross_transfer=True,          # Enable cross-task solution transfer
        transfer_frequency=5,          # Transfer every 5 iterations
        min_improvement_for_transfer=0.01,
    ),
)

# Access per-task results
for task_name, task_result in result.per_task.items():
    print(f"{task_name}: best={task_result.best.score:.4f}")

from gepa import optimize, Artifact, Metric, GeneralizationConfig

all_tasks = load_benchmark_tasks("arc-agi-v1")
train_tasks = all_tasks[:80]
val_tasks = all_tasks[80:]

result = optimize(
    artifact=Artifact(name="arc_solver", template="..."),
    evaluate=eval_arc,
    metrics=[Metric("accuracy", direction="maximize")],
    train_tasks=train_tasks,
    val_tasks=val_tasks,
    config=GeneralizationConfig(
        population_size=50,
        val_frequency=10,            # Validate every 10 iterations
        early_stopping_patience=20,  # Stop if val doesn't improve for 20 iters
        overfitting_threshold=0.15,  # Alert if train-val gap exceeds 15%
    ),
)

print(f"Train accuracy: {result.train_score:.3f}")
print(f"Val accuracy:   {result.val_score:.3f}")
print(f"Generalization gap: {result.generalization_gap:.3f}")

from gepa import EvalResult, TraceASI, ErrorASI, MetricASI

def evaluate_with_asi(candidate: str, task: dict) -> EvalResult:
    """Evaluation function that produces rich ASI feedback."""
    try:
        # Execute candidate
        exec_globals = {}
        exec(candidate, exec_globals)
        solve_fn = exec_globals['solve']

        # Collect execution trace
        trace = []
        original_print = print

        def traced_print(*args, **kwargs):
            trace.append(" ".join(str(a) for a in args))
            original_print(*args, **kwargs)

        import builtins
        builtins.print = traced_print

        result = solve_fn(task['input'])
        builtins.print = original_print

        # Compute metrics
        accuracy = compute_accuracy(result, task['expected'])
        efficiency = compute_efficiency(result)

        # Build ASI
        side_info = [
            TraceASI(trace=trace, label="execution_trace"),
            MetricASI(
                metrics={
                    "accuracy": accuracy,
                    "efficiency": efficiency,
                    "output_length": len(str(result)),
                },
                label="detailed_metrics"
            ),
        ]

        if accuracy < 1.0:
            mismatches = find_mismatches(result, task['expected'])
            side_info.append(
                ComparisonASI(
                    expected=str(task['expected']),
                    actual=str(result),
                    diff=mismatches,
                    label="output_comparison"
                )
            )

        return EvalResult(
            scores={"accuracy": accuracy, "efficiency": efficiency},
            side_info=side_info,
        )

    except Exception as e:
        import traceback
        return EvalResult(
            scores={"accuracy": 0.0, "efficiency": 0.0},
            side_info=[
                ErrorASI(
                    error_type=type(e).__name__,
                    message=str(e),
                    traceback=traceback.format_exc(),
                    label="runtime_error",
                )
            ],
        )

# Internal reflection prompt template (simplified)
REFLECTION_PROMPT = """
You are optimizing a {artifact_type} to maximize {objectives}.

## Current Candidate (Score: {score})

## Diagnostic Feedback (ASI)
{formatted_asi}

## Optimization History
- Iteration {iter}: {score_history}
- Recent improvements: {recent_improvements}
- Stagnation detector: {stagnation_status}

## Reflection Minibatch ({minibatch_size} examples)
{minibatch_details}

## Task
Analyze the diagnostic feedback and propose a TARGETED improvement.
Focus on the specific failure modes revealed by the ASI.
Output the complete improved candidate.
"""

from gepa.engine import ParetoFrontier

class ParetoFrontier:
    def __init__(self, objectives: list[str], directions: list[str]):
        self.objectives = objectives
        self.directions = directions  # "maximize" or "minimize"
        self.frontier: list[Solution] = []

    def dominates(self, a: Solution, b: Solution) -> bool:
        """Check if solution a dominates solution b."""
        at_least_one_better = False
        for obj, direction in zip(self.objectives, self.directions):
            a_val = a.scores[obj]
            b_val = b.scores[obj]
            if direction == "minimize":
                a_val, b_val = -a_val, -b_val
            if a_val < b_val:
                return False
            if a_val > b_val:
                at_least_one_better = True
        return at_least_one_better

    def update(self, candidate: Solution) -> bool:
        """Add candidate to frontier if non-dominated. Returns True if added."""
        # Check if any frontier member dominates candidate
        for member in self.frontier:
            if self.dominates(member, candidate):
                return False  # Dominated, discard

        # Remove frontier members dominated by candidate
        self.frontier = [
            m for m in self.frontier
            if not self.dominates(candidate, m)
        ]
        self.frontier.append(candidate)
        return True

    def select_parent(self, strategy: str = "crowding") -> Solution:
        """Select a parent from the frontier for mutation."""
        if strategy == "crowding":
            # Prefer solutions in sparse regions of the frontier
            distances = self._crowding_distances()
            probs = distances / distances.sum()
            return np.random.choice(self.frontier, p=probs)
        elif strategy == "random":
            return random.choice(self.frontier)
        elif strategy == "tournament":
            a, b = random.sample(self.frontier, 2)
            return a if self._hypervolume_contribution(a) > self._hypervolume_contribution(b) else b

from gepa.reflection import ReflectionEngine, ReflectionConfig

config = ReflectionConfig(
    minibatch_size=3,               # Sample 2-3 examples per reflection
    failure_bias=0.7,               # 70% chance to sample failure cases
    include_trace=True,             # Include execution traces in prompt
    include_error=True,             # Include error details in prompt
    include_comparison=True,        # Include output comparisons
    max_history_length=10,          # Include last 10 mutations in context
    temperature=0.8,                # LLM temperature for diversity
    reflection_prompt_version="v3", # Latest reflection prompt template
)

engine = ReflectionEngine(
    config=config,
    llm=LLMClient("claude-sonnet-4-20250514"),
    artifact_spec=artifact,
)

# Generate a mutation
mutation = await engine.reflect(
    parent=best_candidate,
    eval_results=parent_eval_results,
    history=optimization_history,
)

# Reflection prompt structure
"""
[ROLE] You are an expert optimizer improving a {artifact_type}.

[OBJECTIVE] Maximize: {objectives_description}

[CURRENT SOLUTION] (Score: {score_summary})
{candidate_code}

[DIAGNOSTIC ANALYSIS]
--- Example 1/{minibatch_size} ---
Input: {example_input}
Expected: {expected_output}
Actual: {actual_output}
Trace: {execution_trace}
Error: {error_if_any}
Score: {per_example_score}

--- Example 2/{minibatch_size} ---
...

[MUTATION HISTORY]
Iteration {n-2}: Changed X to Y -> score improved by +0.05
Iteration {n-1}: Changed A to B -> score decreased by -0.02 (reverted)

[INSTRUCTIONS]
1. Analyze the diagnostic feedback above
2. Identify the ROOT CAUSE of failures
3. Propose a SPECIFIC, TARGETED fix (not a rewrite)
4. Explain your reasoning before the code
5. Output the complete improved solution

[OUTPUT FORMAT]
## Analysis
(your analysis here)

## Improved Solution

"""

from gepa import optimize, Artifact, Metric

# No seed solution provided - GEPA generates initial candidates
result = optimize(
    artifact=Artifact(
        name="bin_packing",
        description="A function that packs items into bins to minimize wasted space.",
        signature="def pack(items: list[tuple[float, float]], bin_capacity: float) -> list[list[int]]",
        language="python",
        # No template or seed - GEPA bootstraps from the description + signature
    ),
    evaluate=eval_bin_packing,
    metrics=[Metric("utilization", direction="maximize")],
    config=SeedlessConfig(
        initial_population_size=10,     # Generate 10 diverse starting points
        diversity_prompt=True,          # Instruct LLM to generate diverse solutions
        bootstrap_strategies=[
            "greedy_first_fit",
            "best_fit_decreasing",
            "random_search",
            "dynamic_programming",
            "genetic_algorithm",
        ],  # Hint different algorithmic approaches
    ),
)

from gepa.stopping import MaxMetricCalls, Timeout, NoImprovement, ScoreThreshold, Composite

# Stop when ANY of these conditions is met (OR logic)
stopping = Composite(
    criteria=[
        MaxMetricCalls(200),                     # Hard limit: 200 evaluations
        Timeout(7200),                           # Hard limit: 2 hours
        ScoreThreshold("accuracy", 1.0),         # Perfect score reached
        Composite(                               # AND: both must be true
            criteria=[
                NoImprovement(patience=30, min_delta=0.001),
                MaxMetricCalls(50),               # At least 50 evals before early stop
            ],
            mode="AND",
        ),
    ],
    mode="OR",
)

result = optimize(
    artifact=artifact,
    evaluate=evaluate_fn,
    metrics=metrics,
    stopping=stopping,
)

print(f"Stopped because: {result.stop_reason}")

from gepa.config import (
    GEPAConfig, EngineConfig, ReflectionConfig,
    MergeConfig, RefinerConfig
)

config = GEPAConfig(
    # Engine configuration
    engine=EngineConfig(
        population_size=30,           # Max population size
        elite_size=5,                 # Protected elite solutions
        tournament_size=3,            # Tournament selection size
        mutation_rate=1.0,            # Always mutate (vs crossover)
        crossover_rate=0.0,           # No crossover by default
        island_count=1,               # Number of islands (for island model)
        migration_rate=0.1,           # Migration rate between islands
        migration_interval=10,        # Migrate every N iterations
    ),

    # Reflection configuration
    reflection=ReflectionConfig(
        minibatch_size=3,             # Examples per reflection
        failure_bias=0.7,             # Bias toward failure cases
        temperature=0.8,              # LLM temperature
        max_tokens=4096,              # Max response tokens
        include_trace=True,           # Include execution traces
        include_error=True,           # Include error details
        include_comparison=True,      # Include output comparisons
        include_image=False,          # Include image ASI (multimodal)
        max_history_length=10,        # Mutation history context
        reflection_model="claude-sonnet-4-20250514",  # Model for reflection
    ),

    # Merge configuration (for crossover-like operations)
    merge=MergeConfig(
        strategy="llm_guided",        # LLM picks best parts from 2 parents
        merge_model="claude-sonnet-4-20250514",
        max_parents=2,                # Number of parents to merge
    ),

    # Refiner configuration (post-mutation polish)
    refiner=RefinerConfig(
        enabled=True,                 # Enable post-mutation refinement
        refiner_model="claude-sonnet-4-20250514",
        max_refinement_steps=2,       # Max refinement iterations
        refinement_threshold=0.9,     # Only refine if score > threshold
    ),
)

result = optimize(
    artifact=artifact,
    evaluate=evaluate_fn,
    metrics=metrics,
    config=config,
)

# gepa_config.yaml
engine:
  population_size: 30
  elite_size: 5
  tournament_size: 3

reflection:
  minibatch_size: 3
  failure_bias: 0.7
  temperature: 0.8
  include_trace: true
  include_error: true

merge:
  strategy: llm_guided
  max_parents: 2

refiner:
  enabled: true
  max_refinement_steps: 2

stopping:
  type: composite
  mode: OR
  criteria:
    - type: max_metric_calls
      max_calls: 200
    - type: timeout
      seconds: 7200
    - type: score_threshold
      metric: accuracy
      threshold: 1.0

from gepa import optimize_from_config

result = optimize_from_config("gepa_config.yaml")

from gepa.evaluation import EvaluationPipeline, EvalConfig

pipeline = EvaluationPipeline(
    eval_fn=evaluate,
    config=EvalConfig(
        max_workers=8,              # Parallel evaluation workers
        timeout_per_eval=60,        # Timeout per evaluation (seconds)
        retry_on_error=True,        # Retry failed evaluations once
        cache_enabled=True,         # Enable content-hash caching
        cache_backend="sqlite",     # Cache backend: sqlite, redis, memory
        sandbox="docker",           # Sandbox: docker, subprocess, none
    ),
)

# Evaluate a batch of candidates
results = await pipeline.evaluate_batch(candidates, task)

import hashlib

def content_hash(candidate: str, task: dict) -> str:
    """Compute a deterministic hash for a (candidate, task) pair."""
    content = candidate + json.dumps(task, sort_keys=True)
    return hashlib.sha256(content.encode()).hexdigest()

# Cache lookup before evaluation
cache_key = content_hash(candidate_code, task)
if cache_key in cache:
    return cache[cache_key]  # Skip evaluation

# Evaluate and cache
result = evaluate(candidate_code, task)
cache[cache_key] = result

from gepa.llm import LLMProvider, LLMResponse

class MyCustomLLM(LLMProvider):
    """Custom LLM provider for GEPA."""

    def __init__(self, model_name: str, api_key: str):
        self.model_name = model_name
        self.api_key = api_key

    async def generate(
        self,
        prompt: str,
        temperature: float = 0.8,
        max_tokens: int = 4096,
        stop: list[str] | None = None,
    ) -> LLMResponse:
        # Call your LLM API here
        response = await my_api_call(
            model=self.model_name,
            prompt=prompt,
            temperature=temperature,
            max_tokens=max_tokens,
        )
        return LLMResponse(
            text=response.text,
            usage={"input_tokens": response.input_tokens, "output_tokens": response.output_tokens},
            model=self.model_name,
        )

# Use custom LLM with GEPA
result = optimize(
    artifact=artifact,
    evaluate=evaluate_fn,
    metrics=metrics,
    llm=MyCustomLLM("my-model", "my-api-key"),
)

from gepa import co_optimize, Artifact, Metric

# Co-evolve a prompt and a post-processor together
prompt_artifact = Artifact(
    name="system_prompt",
    template="You are a helpful assistant. {{INSTRUCTIONS}}",
    language="text",
)

processor_artifact = Artifact(
    name="output_processor",
    template="def process(raw_output: str) -> str:\n    {{BODY}}",
    language="python",
)

result = co_optimize(
    artifacts=[prompt_artifact, processor_artifact],
    evaluate=eval_pipeline,        # Evaluation uses both artifacts
    metrics=[Metric("quality", direction="maximize")],
    dependencies={
        "output_processor": ["system_prompt"],  # Processor depends on prompt
    },
    max_iterations=50,
)

print(f"Best prompt: {result.artifacts['system_prompt'].best.code}")
print(f"Best processor: {result.artifacts['output_processor'].best.code}")

# Artifact: YAML routing configuration
artifact = Artifact(
    name="routing_config",
    language="yaml",
    template="""
routing:
  default_region: {{DEFAULT_REGION}}
  rules:
    {{ROUTING_RULES}}
  fallback:
    strategy: {{FALLBACK_STRATEGY}}
    timeout_ms: {{TIMEOUT}}
""",
)

# Multi-objective optimization: minimize cost, maximize latency SLA compliance
result = optimize(
    artifact=artifact,
    evaluate=eval_routing,
    metrics=[
        Metric("cost", direction="minimize", weight=0.6),
        Metric("sla_compliance", direction="maximize", weight=0.4),
    ],
    config=GEPAConfig(
        engine=EngineConfig(population_size=40),
        reflection=ReflectionConfig(minibatch_size=2),
    ),
)