The adapter system is GEPA’s extensibility layer. Adapters bridge the gap between your custom systems (prompt templates, DSPy programs, RAG pipelines, agent architectures) and GEPA’s optimization engine.
What is an Adapter? An adapter is a protocol-conforming object that tells GEPA:
How to evaluate a candidate on your system
How to extract feedback (Actionable Side Information) from evaluations
Optionally, how to propose improvements using domain-specific logic
Adapters are not required for simple use cases. Use optimize_anything directly for quick optimizations. Adapters shine when you need custom evaluation logic or want to optimize complex systems like DSPy programs or RAG pipelines.
The GEPAAdapter Protocol All adapters implement the GEPAAdapter protocol:
from gepa.core.adapter import GEPAAdapter class MyAdapter (GEPAAdapter[DataInst, Trajectory, RolloutOutput]): def evaluate ( self , batch : list[DataInst], candidate : dict[ str , str ], capture_traces : bool = False , ) -> EvaluationBatch[Trajectory, RolloutOutput]: """Run the candidate on a batch of data and return scores.""" ... def make_reflective_dataset ( self , candidate : dict[ str , str ], eval_batch : EvaluationBatch[Trajectory, RolloutOutput], components_to_update : list[ str ], ) -> dict[ str , list[dict[ str , Any]]]: """Extract feedback for each component from evaluation results.""" ... # Optional: custom proposal logic propose_new_texts: ProposalFn | None = None
Type Parameters
DataInst
Your task-specific input data format. # Simple format DataInst = dict # {"input": str, "expected": str} # Structured format class QADataInst ( TypedDict ): question: str context: str answer: str
Trajectory
Execution trace capturing intermediate states. Used for detailed feedback. class MyTrajectory ( TypedDict ): steps: list[ str ] reasoning: str error: str | None
RolloutOutput
Final output from running your system. class MyOutput ( TypedDict ): prediction: str confidence: float
Core Methods evaluate() Runs your system with the candidate parameters on a batch of examples.
def evaluate ( self , batch : list[DataInst], candidate : dict[ str , str ], capture_traces : bool = False , ) -> EvaluationBatch[Trajectory, RolloutOutput]: outputs = [] scores = [] trajectories = [] if capture_traces else None for example in batch: # Build your system with candidate parameters system = self .build_system(candidate) # Run it result = system.run(example) # Compute score (higher is better) score = self .compute_score(result, example) outputs.append(result) scores.append(score) if capture_traces: trajectory = self .extract_trajectory(result) trajectories.append(trajectory) return EvaluationBatch( outputs = outputs, scores = scores, trajectories = trajectories, )
Never raise exceptions for individual example failures. Return a valid EvaluationBatch with failure scores (e.g., 0.0). Reserve exceptions for systemic failures (missing model, misconfiguration).
make_reflective_dataset() Converts evaluation results into structured feedback for the reflection LLM.
def make_reflective_dataset ( self , candidate : dict[ str , str ], eval_batch : EvaluationBatch, components_to_update : list[ str ], ) -> dict[ str , list[dict[ str , Any]]]: reflective_data = {} for component in components_to_update: examples = [] for traj in eval_batch.trajectories: example = { "Inputs" : self .format_inputs(traj), "Generated Outputs" : self .format_outputs(traj), "Feedback" : self .generate_feedback(traj), } examples.append(example) reflective_data[component] = examples return reflective_data
Recommended structure: { "component_name" : [ { "Inputs" : { "question" : "What is ML?" }, "Generated Outputs" : { "answer" : "ML is..." }, "Feedback" : "Correct! Include more details about..." }, # More examples ] }
Built-in Adapters GEPA provides several ready-to-use adapters:
DefaultAdapter Simple adapter for optimizing system prompts.
from gepa.adapters.default_adapter import DefaultAdapter adapter = DefaultAdapter( model = "openai/gpt-4o-mini" , evaluator = my_evaluator, )
Use case: Single-prompt optimization with straightforward evaluation.DspyAdapter Optimizes DSPy program instructions.
from gepa.adapters.dspy_adapter import DspyAdapter adapter = DspyAdapter( student_module = my_dspy_program, metric_fn = my_metric, feedback_map = { "predictor_name" : feedback_fn}, )
Use case: DSPy programs with multiple predictors. See DSPy Integration .GenericRAGAdapter Optimizes RAG pipeline prompts.
from gepa.adapters.generic_rag_adapter import GenericRAGAdapter adapter = GenericRAGAdapter( vector_store = my_vector_store, llm_model = my_llm, embedding_model = "openai/text-embedding-3-small" , rag_config = { "top_k" : 5 , "retrieval_strategy" : "similarity" }, )
Use case: RAG systems with customizable retrieval and generation prompts.OptimizeAnythingAdapter Internal adapter powering optimize_anything(). Wraps simple evaluator functions.
# Used automatically by optimize_anything() result = optimize_anything( seed_candidate = "..." , evaluator = my_evaluator, ... )
Using Adapters with optimize() The gepa.optimize() function provides a streamlined API for adapter-based optimization:
from gepa import optimize result = optimize( seed_candidate = { "prompt" : "Initial prompt" }, trainset = train_data, valset = val_data, adapter = my_adapter, reflection_lm = "openai/gpt-4o" , max_metric_calls = 100 , )
API Comparison Feature optimize_anything()optimize()Use case Simple evaluator Custom adapter Candidate format str or dict dict[str, str] Evaluator Simple function Adapter protocol Flexibility Quick prototyping Full control
Adapter Best Practices
Type Your Data
Use TypedDict or dataclasses for clarity: from typing import TypedDict class MyDataInst ( TypedDict ): input : str expected_output: str metadata: dict
Handle Failures Gracefully
Return failure scores instead of raising: def evaluate ( self , batch , candidate , capture_traces = False ): scores = [] for example in batch: try : result = self .run_system(example, candidate) score = self .compute_score(result) except Exception as e: # Log error but don't raise score = 0.0 scores.append(score) return EvaluationBatch( outputs = ... , scores = scores, ... )
Provide Rich Feedback
The more context in your reflective dataset, the better: feedback = { "Inputs" : { "question" : example.question}, "Generated Outputs" : { "answer" : result.answer}, "Feedback" : ( f "Expected: { example.expected } \n " f "Got: { result.answer } \n " f "Error: { result.error if result.error else 'None' } \n " f "Suggestion: Check the reasoning in step 3." ), }
Use Multi-Objective Scores
Track multiple metrics: return EvaluationBatch( outputs = outputs, scores = scores, # Primary score objective_scores = [ # Additional metrics { "accuracy" : 0.9 , "latency" : 0.1 , "cost" : 0.8 } for _ in batch ], )
Example: Custom Adapter Here’s a minimal adapter for a custom system:
from gepa.core.adapter import GEPAAdapter, EvaluationBatch from typing import TypedDict class MyData ( TypedDict ): input : str expected: str class MyTrace ( TypedDict ): steps: list[ str ] final_output: str class MyOutput ( TypedDict ): result: str class MyAdapter (GEPAAdapter[MyData, MyTrace, MyOutput]): def __init__ ( self , my_system ): self .system = my_system def evaluate ( self , batch , candidate , capture_traces = False ): outputs = [] scores = [] traces = [] if capture_traces else None # Update system with candidate parameters self .system.update_config(candidate) for example in batch: # Run system result = self .system.process(example[ "input" ]) # Score result score = 1.0 if result == example[ "expected" ] else 0.0 outputs.append({ "result" : result}) scores.append(score) if capture_traces: traces.append({ "steps" : self .system.get_steps(), "final_output" : result, }) return EvaluationBatch( outputs = outputs, scores = scores, trajectories = traces, ) def make_reflective_dataset ( self , candidate , eval_batch , components ): data = {} for component in components: examples = [] for trace in eval_batch.trajectories: examples.append({ "Inputs" : trace[ "steps" ][ 0 ], "Generated Outputs" : trace[ "final_output" ], "Feedback" : "Analyze the steps and improve logic." , }) data[component] = examples return data # Use it adapter = MyAdapter(my_system) result = optimize( seed_candidate = { "config" : "initial config" }, trainset = train_data, valset = val_data, adapter = adapter, max_metric_calls = 50 , )
Advanced Features Custom Proposal Logic Override the default proposal mechanism:
class MyAdapter (GEPAAdapter[ ... ]): def propose_new_texts ( self , candidate : dict[ str , str ], reflective_dataset : dict[ str , list[ dict ]], components_to_update : list[ str ], ) -> dict[ str , str ]: # Your custom logic here new_texts = {} for comp in components_to_update: feedback = reflective_dataset[comp] new_texts[comp] = self .my_custom_proposer(feedback) return new_texts
Batched Evaluation Leverage parallelization in your adapter:
def evaluate ( self , batch , candidate , capture_traces = False ): # Batch inference for efficiency results = self .system.batch_process( [ex[ "input" ] for ex in batch], config = candidate, ) scores = [ self .score(r, ex[ "expected" ]) for r, ex in zip (results, batch)] return EvaluationBatch( ... )
Next Steps
Custom Adapters Step-by-step guide to building adapters
DSPy Integration Deep dive into the DSPy adapter
Evaluation Metrics Design effective scoring functions
optimize_anything Simple API without adapters