GEPA

https://gepa-ai.github.io/gepa
Admin
GEPA is a reflective prompt and artifact optimization framework that automatically improves prompts,...
Tokens:203,368
Snippets:1,529
Trust Score:9.7
Update:1 month ago
Show doc for...
Tokens:
Raw
### Setup environment with uv

Source: https://gepa-ai.github.io/gepa/guides/contributing

Recommended installation method using the uv package manager to clone, sync, and verify the project.

```bash
# Install uv
curl -LsSf https://astral.sh/uv/install.sh | sh

# Clone and setup
git clone https://github.com/gepa-ai/gepa
cd gepa
uv sync --extra dev --python 3.11

# Verify installation
uv run pytest tests/
```

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

### Setup environment with conda and pip

Source: https://gepa-ai.github.io/gepa/guides/contributing

Alternative installation method using conda for environment management and pip for package installation.

```bash
conda create -n gepa-dev python=3.11
conda activate gepa-dev
pip install -e ".[dev]"

# Verify installation
pytest tests/
```

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

### Install Confidence-Aware Optimization

Source: https://gepa-ai.github.io/gepa/guides/confidence-adapter

Install the necessary dependencies to use the ConfidenceAdapter.

```bash
pip install "gepa[confidence]"
```

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

### Install and Run GEPA Tutorials Locally

Source: https://gepa-ai.github.io/gepa/tutorials

Commands to install the full GEPA package and Jupyter, then launch the notebook server.

```bash
# Install GEPA with full dependencies
pip install gepa[full]

# Install Jupyter
pip install jupyter

# Start Jupyter
jupyter notebook
```

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

### Install GEPA with all dependencies

Source: https://gepa-ai.github.io/gepa/guides/quickstart

Installation command for GEPA including all optional dependencies.

```bash
pip install gepa[full]
```

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

### Install DSPy Dependency

Source: https://gepa-ai.github.io/gepa/tutorials

Optional installation step for tutorials that specifically utilize DSPy.

```bash
pip install dspy
```

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

### Install gskill and dependencies

Source: https://gepa-ai.github.io/gepa/guides/gskill

Install the necessary packages to run gskill and its associated tools.

```bash
pip install gepa[gskill]
pip install mini-swe-agent swebench
```

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

### Build documentation locally

Source: https://gepa-ai.github.io/gepa/guides/contributing

Commands to install documentation dependencies and serve the site locally.

```bash
cd docs
pip install -r requirements.txt
python scripts/generate_api_docs.py
mkdocs serve
```

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

### Install GEPA via pip

Source: https://gepa-ai.github.io/gepa/guides/quickstart

Standard installation command for the GEPA library from PyPI.

```bash
pip install gepa
```

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

### Install Tracking Dependencies

Source: https://gepa-ai.github.io/gepa/guides/experiment-tracking

Install optional dependencies for WandB and MLflow support.

```bash
pip install "gepa[full]"         # includes wandb + mlflow
pip install wandb mlflow         # or install individually
```

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

### POST /evaluate_examples

Source: https://gepa-ai.github.io/gepa/api/optimize_anything/EvalServer

Evaluates a set of examples.

```APIDOC
## POST /evaluate_examples

### Description
Evaluates a batch of examples.

### Method
POST

### Endpoint
/evaluate_examples
```

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

### Initialize MCPAdapter for Local MCP Server

Source: https://gepa-ai.github.io/gepa/api/adapters/MCPAdapter

Example of configuring the adapter to connect to a local MCP server using StdioServerParameters.

```python
>>> from mcp import StdioServerParameters >>> adapter = MCPAdapter( ... tool_names=["read_file", "write_file"], ... task_model="gpt-4o-mini", ... metric_fn=lambda item, output: 1.0 if item["reference_answer"] in output else 0.0, ... server_params=StdioServerParameters( ... command="python", ... args=["server.py"], ... ), ... )
```

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

### Install GEPA development version

Source: https://gepa-ai.github.io/gepa/guides/quickstart

Installation command for the latest development version directly from the GitHub repository.

```bash
pip install git+https://github.com/gepa-ai/gepa.git
```

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

### Context Manager Entry

Source: https://gepa-ai.github.io/gepa/api/logging/ExperimentTracker

Initializes and starts the experiment run when entering the context.

```python
def __enter__(self):
    """Context manager entry."""
    self.initialize()
    self.start_run()
    return self
```

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

### Implement a Complete Fitness Function

Source: https://gepa-ai.github.io/gepa/guides/gskill

A full example demonstrating pool management, task execution, verification, and diagnostic reporting.

```python
from gepa.optimize_anything import optimize_anything, GEPAConfig, EngineConfig

def create_fitness_fn(model_name: str, n_workers: int):
    """Create a fitness function with a pool of Docker harnesses."""
    harness_pool = [SWEHarness() for _ in range(n_workers)]
    # ... pool management code ...

    def fitness_fn(candidate: dict[str, str], example: dict[str, Any]) -> tuple[float, dict]:
        skills = candidate["skills"]
        harness = get_available_harness()

        try:
            # 1. Setup Docker container with the task
            harness.setup_task(task_instance=example)

            # 2. Run the agent with current skills
            patch, trace, metrics = harness.run_agent(
                example["problem_statement"], skills, model_name=model_name
            )

            # 3. Verify the patch with tests
            passed, test_output = harness.verify_with_patch(patch)
            score = 1.0 if passed else 0.0

            # 4. Return score + diagnostic info for reflection
            side_info = {
                "Input": {"Problem": example["problem_statement"][:200]},
                "Generated Outputs": {"Patch": patch[:500], "Agent Trace": trace},
                "Feedback": {"Status": "passed" if passed else "failed", "Test Output": test_output},
                "scores": {"correctness": score},
            }
            return score, side_info

        finally:
            harness.cleanup()
            release_harness(harness)

    return fitness_fn

# Use it
fitness_fn = create_fitness_fn("gpt-5-mini", n_workers=6)

result = optimize_anything(
    seed_candidate={"skills": ""},
    evaluator=fitness_fn,
    dataset=train_data,
    valset=val_data,
    config=GEPAConfig(
        engine=EngineConfig(parallel=True, max_workers=6, max_metric_calls=600),
    ),
)
```

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

### Full End-to-End Optimization Script

Source: https://gepa-ai.github.io/gepa/guides/claude-cli-as-proposer

A complete example script using Claude CLI for both task and reflection language models.

```python
import subprocess
import gepa


def claude_cli_lm(prompt: str) -> str:
    result = subprocess.run(
        ["claude", "-p", prompt],
        capture_output=True, text=True, timeout=120,
    )
    if result.returncode != 0:
        raise RuntimeError(f"claude -p failed: {result.stderr}")
    return result.stdout


def claude_cli_chat(messages: list) -> str:
    parts = []
    for m in messages:
        if m["role"] == "system":
            parts.append(f"[System]\n{m['content']}")
        elif m["role"] == "user":
            parts.append(f"[User]\n{m['content']}")
    return claude_cli_lm("\n\n".join(parts))


dataset = [
    {"input": "What is 2+2?", "answer": "4", "additional_context": {}},
    {"input": "What is the capital of France?", "answer": "Paris", "additional_context": {}},
    {"input": "What color is the sky on a clear day?", "answer": "blue", "additional_context": {}},
]

result = gepa.optimize(
    seed_candidate={"instructions": "Answer the question correctly."},
    trainset=dataset,
    task_lm=claude_cli_chat,
    reflection_lm=claude_cli_lm,
    max_metric_calls=20,
    reflection_minibatch_size=1,
    display_progress_bar=True,
)

print(f"Best instructions: {result.best_candidate}")
print(f"Best score: {result.val_aggregate_scores[result.best_idx]:.3f}")
```

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

### Optimize prompts with DefaultAdapter

Source: https://gepa-ai.github.io/gepa/guides/quickstart

Use the DefaultAdapter for single-turn LLM tasks by providing a seed prompt and a list of examples.

```python
import gepa

# The DefaultAdapter expects each example as a TypedDict with three fields:
#   input               — the user question
#   additional_context  — dict of extra context (can be empty {})
#   answer              — the expected answer (used for substring match)
trainset = [
    {"input": "What is 2+2?", "additional_context": {}, "answer": "4"},
    {"input": "What is the capital of France?", "additional_context": {}, "answer": "Paris"},
    # ... more examples
]

# Define your seed prompt
seed_prompt = {
    "system_prompt": "You are a helpful assistant. Answer questions concisely."
}

# Run optimization
result = gepa.optimize(
    seed_candidate=seed_prompt,
    trainset=trainset,
    task_lm="openai/gpt-4o-mini",      # Model to optimize
    reflection_lm="openai/gpt-4o",      # Model for reflection
    max_metric_calls=50,                # Budget
)

# Get the optimized prompt and best score
print("Best prompt:", result.best_candidate['system_prompt'])
print("Best score:", result.val_aggregate_scores[result.best_idx])
```

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

### Implement dynamic training data updates

Source: https://gepa-ai.github.io/gepa/guides/callbacks

Adds hard examples from the validation set to the training set during the start of an iteration.

```python
class ActiveLearner:
    """Add hard examples from the valset to the trainset when we find a new best."""

    def __init__(self, hard_threshold: float = 0.3):
        self.hard_threshold = hard_threshold
        self._last_best = None

    def on_valset_evaluated(self, event):
        if event["is_best_program"] and event["outputs_by_val_id"]:
            self._last_best = event

    def on_iteration_start(self, event):
        if self._last_best is None:
            return
        scores = self._last_best["scores_by_val_id"]
        hard_ids = [vid for vid, s in scores.items() if s < self.hard_threshold]
        if hard_ids:
            # Add the raw inputs for the hard examples to the training set
            hard_examples = [{"id": vid} for vid in hard_ids]
            event["trainset_loader"].add_items(hard_examples)
            print(f"Added {len(hard_ids)} hard examples to trainset")
        self._last_best = None
```

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

### Initialize and Optimize with GenericRAGAdapter

Source: https://gepa-ai.github.io/gepa/api/adapters/RAGAdapter

Demonstrates setting up a ChromaDB-backed adapter and running the GEPA optimization process.

```python
from gepa.adapters.generic_rag_adapter import GenericRAGAdapter, ChromaVectorStore import gepa
vector_store = ChromaVectorStore.create_local("./kb", "docs") adapter = GenericRAGAdapter(vector_store=vector_store, llm_model="gpt-4")
result = gepa.optimize( seed_candidate={"answer_generation": "Answer based on context:"}, trainset=train_data, valset=val_data, adapter=adapter, max_metric_calls=50 ) print(result.best_candidate) # Optimized prompts
```

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

### get(candidate: dict[str, str], example_id: DataId) -> CachedEvaluation[RolloutOutput] | None

Source: https://gepa-ai.github.io/gepa/api/core/EvaluationCache

Retrieves a cached evaluation result for a specific candidate and example ID if it exists.

```APIDOC
## get(candidate, example_id)

### Description
Retrieve cached evaluation result if it exists.

### Parameters
- **candidate** (dict[str, str]) - The candidate dictionary to hash.
- **example_id** (DataId) - The identifier for the example.

### Returns
- **CachedEvaluation[RolloutOutput] | None** - The cached evaluation result or None if not found.
```

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

### DefaultAdapter.__init__

Source: https://gepa-ai.github.io/gepa/api/adapters/DefaultAdapter

Initializes a new instance of the DefaultAdapter.

```APIDOC
## Constructor: DefaultAdapter(model, evaluator=None, max_litellm_workers=10, litellm_batch_completion_kwargs=None)

### Description
Initializes the adapter with a model, an optional evaluator, and configuration for batch processing.

### Parameters
- **model** (str | ChatCompletionCallable) - Required - The model name or a callable function for chat completion.
- **evaluator** (Evaluator | None) - Optional - The evaluator instance to use. Defaults to ContainsAnswerEvaluator.
- **max_litellm_workers** (int) - Optional - Maximum number of workers for LiteLLM batch processing. Defaults to 10.
- **litellm_batch_completion_kwargs** (dict[str, Any] | None) - Optional - Additional keyword arguments for LiteLLM batch completion.
```

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

### Basic optimize_anything usage

Source: https://gepa-ai.github.io/gepa/blog/introducing-optimize-anything

Demonstrates initializing the optimization process using either a seed candidate or a descriptive objective.

```python
import gepa.optimize_anything as oa

def evaluate(candidate: str) -> float:
    score, diagnostic = run_my_system(candidate)
    oa.log(f"Error: {diagnostic}")  # captured as ASI
    return score

# Start from an existing artifact…
result = oa.optimize_anything(
    seed_candidate="<your initial artifact>",
    evaluator=evaluate,
)

# … or just describe what you need.
result = oa.optimize_anything(
    evaluator=evaluate,
    objective="Generate a Python function `reverse()` that reverses a string.",
)

print(result.best_candidate)
```

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

### Initialize GEPAConfig

Source: https://gepa-ai.github.io/gepa/api/optimize_anything/GEPAConfig

Configure the optimization engine, reflection model, and refiner settings.

```python
config = GEPAConfig(
    engine=EngineConfig(max_metric_calls=200),
    reflection=ReflectionConfig(reflection_lm="openai/gpt-5.1"),
    refiner=RefinerConfig(max_refinements=2),
)
```

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

### MCPAdapter.__init__

Source: https://gepa-ai.github.io/gepa/api/adapters/MCPAdapter

Initializes the MCPAdapter with tool definitions, task model, and metric functions for optimization.

```APIDOC
## MCPAdapter(tool_names, task_model, metric_fn, ...)

### Description
Initializes the MCPAdapter to optimize tool usage. It supports local stdio-based servers or remote servers via SSE or StreamableHTTP.

### Parameters
- **tool_names** (str | list[str]) - Required - Name(s) of tool(s) to optimize.
- **task_model** (str | Callable) - Required - Model for task execution (litellm string or callable).
- **metric_fn** (Callable[[MCPDataInst, str], float]) - Required - Scoring function: (data_inst, output) -> float.
- **server_params** (StdioServerParameters | None) - Optional - Local MCP server configuration (stdio).
- **remote_url** (str | None) - Optional - Remote MCP server URL.
- **remote_transport** (str) - Optional - "sse" or "streamable_http". Default: 'sse'.
- **remote_headers** (dict[str, str] | None) - Optional - HTTP headers for remote connections.
- **remote_timeout** (float) - Optional - Timeout for remote HTTP operations. Default: 30.
- **base_system_prompt** (str) - Optional - Base system prompt template. Default: 'You are a helpful assistant with access to tools.'
- **enable_two_pass** (bool) - Optional - Use two-pass workflow (tool + answer). Default: True.
- **failure_score** (float) - Optional - Score assigned when execution fails. Default: 0.0.
```

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

### Verify Claude CLI Installation

Source: https://gepa-ai.github.io/gepa/guides/claude-cli-as-proposer

Ensure the Claude CLI is installed and authenticated before use.

```bash
claude  # opens interactive mode; complete login if needed
claude -p "say hello"  # verify non-interactive mode works
```

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

### DspyAdapter.__init__

Source: https://gepa-ai.github.io/gepa/api/adapters/DSPyFullProgramAdapter

Initializes the DspyAdapter with the necessary language models and configuration for full-program evolution.

```APIDOC
## DspyAdapter.__init__

### Description
Initializes a new instance of the DspyAdapter class.

### Parameters
- **task_lm** (dspy.LM) - The language model used for the task.
- **metric_fn** (Callable) - The function used to evaluate program performance.
- **reflection_lm** (LanguageModel) - The language model used for reflection.
- **failure_score** (float) - The score assigned to failed programs (default: 0.0).
- **num_threads** (int | None) - Number of threads for execution (default: None).
- **add_format_failure_as_feedback** (bool) - Whether to include format failures as feedback (default: False).
- **rng** (random.Random | None) - Random number generator instance (default: None).
```

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

### Install GEPA for Parallel Scaling

Source: https://gepa-ai.github.io/gepa/blog/2026/04/09/gepa-at-scale-with-combee

Use this command to install or upgrade the GEPA package to access Combee-integrated features.

```bash
pip install -U gepa
```

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

### NoImprovementStopper.__init__

Source: https://gepa-ai.github.io/gepa/api/stop_conditions/NoImprovementStopper

Initializes the NoImprovementStopper with the maximum number of iterations allowed without improvement.

```APIDOC
## Constructor: NoImprovementStopper(max_iterations_without_improvement: int)

### Description
Initializes the stopper instance with the threshold for iterations without score improvement.

### Parameters
- **max_iterations_without_improvement** (int) - Required - The number of iterations to wait for an improvement before triggering the stop condition.
```

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

### Start HTTP Server

Source: https://gepa-ai.github.io/gepa/api/optimize_anything/EvalServer

Initializes and starts a ThreadingHTTPServer to handle evaluation requests. Returns the port the server is bound to.

```python
def start(self, port: int = 0) -> int:
    """Start the HTTP server. Returns the bound port."""
    server_ref = self

    class Handler(BaseHTTPRequestHandler):
        def do_POST(self):
            if self.path == "/evaluate":
                server_ref._handle_evaluate(self)
            elif self.path == "/evaluate_examples":
                server_ref._handle_evaluate_examples(self)
            elif self.path == "/validate":
                server_ref._handle_validate(self)
            else:
                self.send_error(404)

        def do_GET(self):
            if self.path == "/status":
                server_ref._handle_status(self)
            elif self.path == "/task":
                server_ref._handle_task_info(self)
            else:
                self.send_error(404)

        def log_message(self, format, *args):
            pass

    # threading.HTTPServer would be better for high concurrency, but the
    # underlying eval is already serialized through ``_eval_semaphore``;
    # we keep the simple single-threaded HTTPServer to avoid surprising
    # ordering, and rely on the thread pool inside evaluate_examples to
    # parallelize batches.
    from http.server import ThreadingHTTPServer

    self._server = ThreadingHTTPServer(("localhost", port), Handler)
    self._thread = threading.Thread(target=self._server.serve_forever, daemon=True)
    self._thread.start()
    return self.port
```

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

### ExperimentTracker.initialize

Source: https://gepa-ai.github.io/gepa/api/logging/ExperimentTracker

Initializes the configured logging backends.

```APIDOC
## ExperimentTracker.initialize()

### Description
Initializes the logging backends (wandb and/or mlflow) based on the configuration provided during initialization.
```

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

### Install GEPA Agent Skill via Claude Code

Source: https://gepa-ai.github.io/gepa/guides/agent-skill

Commands to add and install the GEPA optimization skill as a plugin in a project.

```bash
/plugin marketplace add gepa-ai/gepa
/plugin install gepa-optimize-anything@gepa
```

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

### Implement a Custom Batch Sampler

Source: https://gepa-ai.github.io/gepa/guides/batch-sampling

Define a class that satisfies the BatchSampler protocol to control example selection, such as prioritizing harder examples.

```python
from gepa.strategies.batch_sampler import BatchSampler
from gepa.core.data_loader import DataLoader
from gepa.core.state import GEPAState


class MyBatchSampler:
    """Custom batch sampler that always picks the hardest examples."""

    def __init__(self, minibatch_size: int):
        self.minibatch_size = minibatch_size

    def next_minibatch_ids(
        self, loader: DataLoader, state: GEPAState
    ) -> list:
        # Access the Pareto front to find the hardest examples
        hardest = sorted(
            state.pareto_front_valset.items(),
            key=lambda x: x[1],  # sort by best score (ascending)
        )
        # Return the IDs with the lowest best scores
        ids = [val_id for val_id, _score in hardest[: self.minibatch_size]]

        # Fall back to all IDs if not enough in the Pareto front
        if len(ids) < self.minibatch_size:
            all_ids = list(loader.all_ids())
            ids = all_ids[: self.minibatch_size]

        return ids
```

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

### TerminusAdapter.__init__

Source: https://gepa-ai.github.io/gepa/api/adapters/TerminalBenchAdapter

Initializes the TerminusAdapter with concurrency settings and prompt template paths.

```APIDOC
## TerminusAdapter(n_concurrent: int = 6, instruction_prompt_path: str = 'prompt-templates/instruction_prompt.txt')

### Description
Initializes a new instance of the TerminusAdapter.

### Parameters
- **n_concurrent** (int) - Optional - Number of concurrent tasks to run. Defaults to 6.
- **instruction_prompt_path** (str) - Optional - Path to the instruction prompt template file.
```

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

### Evaluate candidate examples in parallel

Source: https://gepa-ai.github.io/gepa/api/optimize_anything/EvalServer

This method evaluates a candidate string against a list of example IDs or a dataset split. It automatically handles budget checks and falls back to individual evaluation if batch processing fails.

```python
def evaluate_examples(
    self,
    candidate: str,
    example_ids: list[str] | None = None,
    split: str | None = None,
) -> tuple[float, dict[str, Any]]:
    """Evaluate a candidate on specific examples or a whole split, in parallel.

    Provide either ``example_ids`` or ``split`` (``"train"``/``"val"``/
    ``"all"``); ``example_ids`` takes precedence. The server holds no
    ``test`` split, so ``"all"`` means train+val. For single-task tasks,
    delegates to :meth:`evaluate`.

    Returns:
        (average_score, info_dict) with per-example scores and metadata.
    """
    if not self.task.has_dataset:
        score, info = self.evaluate(candidate)
        return score, {
            "scores": {"_single": score},
            "infos": {"_single": info},
            "num_evaluated": 1,
            "num_total": 1,
            "partial": False,
            "_budget": self.budget.status(),
        }

    examples: list[tuple[str, Any]] = []
    if example_ids is not None:
        for eid in example_ids:
            if eid in self._examples:
                examples.append((eid, self._examples[eid]))
    elif split is not None:
        splits = ("train", "val") if split == "all" else (split,)
        for s in splits:
            examples.extend(self.iter_split(s))
    elif self.task.train_set:
        examples.extend(self.iter_split("train"))

    if not examples:
        score, info = self.evaluate(candidate)
        return score, {
            "scores": {"_single": score},
            "infos": {"_single": info},
            "num_evaluated": 1,
            "num_total": 1,
            "partial": False,
            "_budget": self.budget.status(),
        }

    if self.budget.remaining is not None and self.budget.remaining < len(examples):
        raise BudgetExhausted(
            f"Not enough budget to evaluate all examples: {self.budget.remaining} remaining, {len(examples)} needed"
        )

    scores: dict[str, float] = {}
    infos: dict[str, dict[str, Any]] = {}
    errors: dict[str, str] = {}

    grouped_done = False
    if self.batch_fn is not None:
        # Multi-pair evaluations prefer the grouped path (launcher parity):
        # the user's batch function sees all pairs in one call.
        try:
            results = self.evaluate_batch([(candidate, ex) for _eid, ex in examples])
        except BudgetExhausted:
            raise
        except Exception:
            # A failed grouped call must not zero the whole stage: fall
            # through to per-pair evaluation so one bad pair costs one 0.0,
            # matching the per-pair path's isolation. Both attempts consume
            # budget — each was a real eval call.
            pass
        else:
            grouped_done = True
            for (eid, _ex), (score, ex_info) in zip(examples, results, strict=True):
                scores[eid] = score
                infos[eid] = ex_info
    if not grouped_done:

        def _eval_one(eid: str, ex: Any) -> tuple[str, float, dict[str, Any] | None, str | None]:
            try:
                score, info = self.evaluate(candidate, ex)
                return (eid, score, info, None)
            except Exception as e:
                return (eid, 0.0, None, str(e))

        futures = {self._pool.submit(_eval_one, eid, ex): eid for eid, ex in examples}
        for future in as_completed(futures):
            eid, score, ex_info, err = future.result()
            if err is not None:
                errors[eid] = err
            else:
                scores[eid] = score
                if ex_info is not None:
                    infos[eid] = ex_info
```

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

### GET /task

Source: https://gepa-ai.github.io/gepa/api/optimize_anything/EvalServer

Retrieves information about the current task.

```APIDOC
## GET /task

### Description
Returns information about the task being evaluated.

### Method
GET

### Endpoint
/task
```

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

### GET /status

Source: https://gepa-ai.github.io/gepa/api/optimize_anything/EvalServer

Retrieves the current status of the evaluation server.

```APIDOC
## GET /status

### Description
Returns the current status of the evaluation server.

### Method
GET

### Endpoint
/status
```

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

### Initialize Engine with Configuration

Source: https://gepa-ai.github.io/gepa/api/optimize_anything/Engine

The constructor reads cross-cutting fields from OptimizeAnythingConfig and handles engine-specific keys. Engines must enforce the proposer-cost cap defined in config.max_token_cost.

```python
def __init__(self, config: OptimizeAnythingConfig) -> None:
    """Configure the engine from a single :class:`OptimizeAnythingConfig`.

    Read cross-cutting fields directly (``config.run_dir``,
    ``config.stop_at_score``, …) and pop engine-specific keys from
    ``config.engine_config`` onto ``self``. Warn (or raise) on unknown keys
    in ``config.engine_config`` to surface typos.

    Budget contract: the engine MUST honor ``config.max_token_cost`` (the
    proposer-cost cap — USD on the engine's own optimizer LLM tokens). The
    eval server enforces only the eval-call budget and cannot enforce the
    proposer cap: it never observes the engine's out-of-band LLM spend (and
    a subprocess's cost isn't knowable until it exits). Read the cap here
    and translate it into your native mechanism (``max_reflection_cost``,
    ``--max-budget-usd``, …).
    """
    ...
```

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

### Initialize TimeoutStopCondition

Source: https://gepa-ai.github.io/gepa/api/stop_conditions/TimeoutStopCondition

Constructor for setting the timeout duration and capturing the start time.

```python
def __init__(self, timeout_seconds: float):
    self.timeout_seconds = timeout_seconds
    self.start_time = time.time()
```

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

### Configuring and Running Different Optimization Engines

Source: https://gepa-ai.github.io/gepa/blog/2026/07/22/optimize-anything-omni

Demonstrates how to swap between 'gepa', 'autoresearch', and 'meta_harness' engines while maintaining the same evaluation function and task configuration.

```python
from gepa.optimize_anything import optimize_anything, OptimizeAnythingConfig

def evaluate(candidate: str) -> tuple[float, dict]:
    score, feedback = run_judge(candidate)
    return score, {"Feedback": feedback}

seed = open("seed_solution.py").read()
task = dict(
    evaluator=evaluate,
    objective="Maximize the score for this competitive-programming problem.",
    background="A sandboxed judge runs hidden tests and returns a 0–100 score.",
)

# Same task, same evaluator, swap only the engine:
result_gepa = optimize_anything(seed, **task, 
                          config=OptimizeAnythingConfig(engine="gepa",        ))
result_autoresearch = optimize_anything(seed, **task, 
                          config=OptimizeAnythingConfig(engine="autoresearch",))
result_meta_harness = optimize_anything(seed, **task, 
                          config=OptimizeAnythingConfig(engine="meta_harness",))
```

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

### Initialize MCPAdapter for Remote MCP Server

Source: https://gepa-ai.github.io/gepa/api/adapters/MCPAdapter

Example of configuring the adapter to connect to a remote MCP server via SSE transport.

```python
>>> adapter = MCPAdapter( ... tool_names="search_web", ... task_model="gpt-4o-mini", ... metric_fn=accuracy_metric, ... remote_url="https://mcp-server.com/sse", ... remote_transport="sse", ... )
```

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

### Iteration Lifecycle Callbacks

Source: https://gepa-ai.github.io/gepa/api/callbacks/GEPACallback

Methods triggered at the start and end of each optimization iteration.

```python
def on_iteration_start(self, event: IterationStartEvent) -> None:
    """Called at the start of each iteration."""
    ...
```

```python
def on_iteration_end(self, event: IterationEndEvent) -> None:
    """Called at the end of each iteration."""
    ...
```

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

### FileStopper.__init__

Source: https://gepa-ai.github.io/gepa/api/stop_conditions/FileStopper

Initializes the FileStopper with the path to the file that triggers the stop condition.

```APIDOC
## FileStopper(stop_file_path: str)

### Description
Initializes a new instance of the FileStopper class.

### Parameters
- **stop_file_path** (str) - Required - The filesystem path to the file that, when present, triggers the stop condition.
```

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

### Configure and run the optimizer

Source: https://gepa-ai.github.io/gepa/blog/introducing-optimize-anything

Initializes the GEPA optimization engine with training datasets and reflection configurations for strategy evolution.

```python
from gepa.optimize_anything import optimize_anything, GEPAConfig, EngineConfig, ReflectionConfig
from utils.dataset import load_trace_dataset
from utils.lm import make_reflection_lm

# Load spot-availability traces split into train/val/test
splits = load_trace_dataset()
train_set, val_set = splits["train"], splits["val"]

result = optimize_anything(
    seed_candidate={"program": SEED_PROGRAM},
    evaluator=evaluate,
    dataset=train_set,
    valset=val_set,
    objective="Optimize a cloud scheduling strategy for the 'Can't Be Late' problem. "
              "Minimize cost while ensuring task completion before deadline.",
    background="ClusterType.SPOT: ~$0.3/hr, cheap but preemptible at any time. "
               "ClusterType.ON_DEMAND: ~$1/hr, guaranteed availability. "
               "ClusterType.NONE: wait with no cost or progress. restart_overhead: "
               "time penalty when switching instance types. The strategy MUST "
               "ensure deadline completion (hard constraint).",
    config=GEPAConfig(
        engine=EngineConfig(max_metric_calls=100, parallel=True, max_workers=128),
        reflection=ReflectionConfig(reflection_lm=make_reflection_lm("anthropic/claude-opus-4-5-20251101")),
    ),
)
```

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

### Optimization Lifecycle Callbacks

Source: https://gepa-ai.github.io/gepa/api/callbacks/GEPACallback

Methods triggered at the start and end of the optimization process.

```python
def on_optimization_start(self, event: OptimizationStartEvent) -> None:
    """Called when optimization begins."""
    ...
```

```python
def on_optimization_end(self, event: OptimizationEndEvent) -> None:
    """Called when optimization completes."""
    ...
```

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

### Implement Evaluator Function

Source: https://gepa-ai.github.io/gepa/blog/introducing-optimize-anything

Evaluates a candidate prompt against a specific math problem example.

```python
from examples.aime_math.eval import run_llm, math_metric

def evaluate(candidate: str, example) -> tuple[float, SideInfo]:
    """Evaluate a candidate on a single example."""
    prediction = run_llm(example, candidate)
    score, feedback = math_metric(example, prediction)

    side_info = {
        "score": score,
        "input": example.input,
        "output": prediction.answer,
        "reasoning": getattr(prediction, "reasoning", ""),
        "execution_feedback": feedback,
    }

    return score, side_info
```

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

### Configure environment for gskill

Source: https://gepa-ai.github.io/gepa/guides/gskill

Set up required API keys and verify Docker connectivity before running training.

```bash
export OPENAI_API_KEY=<your-key>

# Docker must be running
docker ps

# Download SWE-smith images for your target repo
python -m swesmith.build_repo.download_images
```

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

### Manage pre-commit hooks

Source: https://gepa-ai.github.io/gepa/guides/contributing

Commands to install and execute pre-commit hooks for maintaining code quality.

```bash
# Install hooks (one-time setup)
uv run pre-commit install

# Hooks run automatically on commit
git add .
git commit -m "your message"

# Or run manually
uv run pre-commit run --all-files
```

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

### start_run()

Source: https://gepa-ai.github.io/gepa/api/logging/ExperimentTracker

Starts a new experiment run. Depending on configuration, it can either initialize a new run or attach to an existing active run in W&B or MLflow.

```APIDOC
## start_run()

### Description
Starts a new run. When `wandb_attach_existing=True` the tracker skips `wandb.init()` and logs into whatever run is already active in the process. When `mlflow_attach_existing=True` the tracker skips `mlflow.start_run()` and logs into the already-active MLflow run. In both cases `end_run()` will not terminate the run.
```

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

### Initialize and Run ConfidenceAdapter

Source: https://gepa-ai.github.io/gepa/guides/confidence-adapter

Configures the adapter with a specific JSON schema and runs the optimization process using a provided training set.

```python
import gepa
from gepa.adapters.confidence_adapter import ConfidenceAdapter

adapter = ConfidenceAdapter(
    model="openai/gpt-4.1-mini",
    field_path="category_name",
    response_format={
        "type": "json_schema",
        "json_schema": {
            "name": "classification",
            "strict": True,
            "schema": {
                "type": "object",
                "properties": {
                    "category_name": {
                        "type": "string",
                        "enum": [
                            "Bills/Electricity",
                            "Bills/Gas & Oil",
                            "Food & Drinks/Restaurants",
                            "Shopping/Electronics",
                            "Shopping/Video Games",
                        ],
                    }
                },
                "required": ["category_name"],
                "additionalProperties": False,
            },
        },
    },
)

result = gepa.optimize(
    seed_candidate={"system_prompt": "Classify the following transaction."},
    trainset=[
        {"input": "UBER EATS payment", "answer": "Food & Drinks/Restaurants", "additional_context": {}},
        {"input": "LIGHT electricity bill", "answer": "Bills/Electricity", "additional_context": {}},
        {"input": "Steam purchase", "answer": "Shopping/Video Games", "additional_context": {}},
    ],
    adapter=adapter,
    reflection_lm="openai/gpt-4.1",
    max_metric_calls=500,
)
```
Tokens:
Raw