dspy.GEPA - Advanced Features¶
Custom Instruction Proposers¶
What is instruction_proposer?¶
The instruction_proposer is the component responsible for invoking the reflection_lm and proposing new prompts during GEPA optimization. When GEPA identifies underperforming components in your DSPy program, the instruction proposer analyzes execution traces, feedback, and failures to generate improved instructions tailored to the observed issues.
Default Implementation¶
By default, GEPA uses the built-in instruction proposer from the GEPA library, which implements the ProposalFn. The default proposer uses this prompt template:
I provided an assistant with the following instructions to perform a task for me:
```
<curr_instructions>
```
The following are examples of different task inputs provided to the assistant along with the assistant's response for each of them, and some feedback on how the assistant's response could be better:
```
<inputs_outputs_feedback>
```
Your task is to write a new instruction for the assistant.
Read the inputs carefully and identify the input format and infer detailed task description about the task I wish to solve with the assistant.
Read all the assistant responses and the corresponding feedback. Identify all niche and domain specific factual information about the task and include it in the instruction, as a lot of it may not be available to the assistant in the future. The assistant may have utilized a generalizable strategy to solve the task, if so, include that in the instruction as well.
Provide the new instructions within ``` blocks.
This template is automatically filled with:
<curr_instructions>: The current instruction being optimized<inputs_outputs_feedback>: Structured markdown containing predictor inputs, generated outputs, and evaluation feedback
Example of default behavior:
# Default instruction proposer is used automatically
gepa = dspy.GEPA(
metric=my_metric,
reflection_lm=dspy.LM(model="gpt-5", temperature=1.0, max_tokens=32000, api_key=api_key),
auto="medium"
)
optimized_program = gepa.compile(student, trainset=examples)
When to Use Custom instruction_proposer¶
Note: Custom instruction proposers are an advanced feature. Most users should start with the default proposer, which works well for most text-based optimization tasks.
Consider implementing a custom instruction proposer when you need:
- Multi-modal handling: Process images (dspy.Image) alongside textual information in your inputs
- Nuanced control on limits and length constraints: Have more fine-grained control over instruction length, format, and structural requirements
- Domain-specific information: Inject specialized knowledge, terminology, or context that the default proposer lacks and cannot be provided via feedback_func. This is an advanced feature, and most users should not need to use this.
- Provider-specific prompting guides: Optimize instructions for specific LLM providers (OpenAI, Anthropic, etc.) with their unique formatting preferences
- Coupled component updates: Handle situations where 2 or more components need to be updated together in a coordinated manner, rather than optimizing each component independently (refer to component_selector parameter, in Custom Component Selection section, for related functionality)
- External knowledge integration: Connect to databases, APIs, or knowledge bases during instruction generation
Available Options¶
Built-in Options:
- Default Proposer: The standard GEPA instruction proposer (used when
instruction_proposer=None). The default instruction proposer IS an instruction proposer as well! It is the most general one, that was used for the diverse experiments reported in the GEPA paper and tutorials. - MultiModalInstructionProposer: Handles
dspy.Imageinputs and structured multimodal content.
from dspy.teleprompt.gepa.instruction_proposal import MultiModalInstructionProposer
# For tasks involving images or multimodal inputs
gepa = dspy.GEPA(
metric=my_metric,
reflection_lm=dspy.LM(model="gpt-5", temperature=1.0, max_tokens=32000, api_key=api_key),
instruction_proposer=MultiModalInstructionProposer(),
auto="medium"
)
We invite community contributions of new instruction proposers for specialized domains as the GEPA library continues to grow.
How to Implement Custom Instruction Proposers¶
Custom instruction proposers must implement the ProposalFn protocol by defining a callable class or function. GEPA will call your proposer during optimization:
from dspy.teleprompt.gepa.gepa_utils import ReflectiveExample
class CustomInstructionProposer:
def __call__(
self,
candidate: dict[str, str], # Candidate component name -> instruction mapping to be updated in this round
reflective_dataset: dict[str, list[ReflectiveExample]], # Component -> examples with structure: {"Inputs": ..., "Generated Outputs": ..., "Feedback": ...}
components_to_update: list[str] # Which components to improve
) -> dict[str, str]: # Return new instruction mapping only for components being updated
# Your custom instruction generation logic here
return updated_instructions
# Or as a function:
def custom_instruction_proposer(candidate, reflective_dataset, components_to_update):
# Your custom instruction generation logic here
return updated_instructions
Reflective Dataset Structure:
dict[str, list[ReflectiveExample]]- Maps component names to lists of examplesReflectiveExampleTypedDict contains:Inputs: dict[str, Any]- Predictor inputs (may include dspy.Image objects)Generated_Outputs: dict[str, Any] | str- Success: output fields dict, Failure: error messageFeedback: str- Always a string from metric function or auto-generated by GEPA
Basic Example: Word Limit Proposer¶
import dspy
from gepa.core.adapter import ProposalFn
from dspy.teleprompt.gepa.gepa_utils import ReflectiveExample
class GenerateWordLimitedInstruction(dspy.Signature):
"""Given a current instruction and feedback examples, generate an improved instruction with word limit constraints."""
current_instruction = dspy.InputField(desc="The current instruction that needs improvement")
feedback_summary = dspy.InputField(desc="Feedback from examples that might include both positive and negative cases")
max_words = dspy.InputField(desc="Maximum number of words allowed in the new instruction")
improved_instruction = dspy.OutputField(desc="A new instruction that fixes the issues while staying under the max_words limit")
class WordLimitProposer(ProposalFn):
def __init__(self, max_words: int = 1000):
self.max_words = max_words
self.instruction_improver = dspy.ChainOfThought(GenerateWordLimitedInstruction)
def __call__(self, candidate: dict[str, str], reflective_dataset: dict[str, list[ReflectiveExample]], components_to_update: list[str]) -> dict[str, str]:
updated_components = {}
for component_name in components_to_update:
if component_name not in candidate or component_name not in reflective_dataset:
continue
current_instruction = candidate[component_name]
component_examples = reflective_dataset[component_name]
# Create feedback summary
feedback_text = "\n".join([
f"Example {i+1}: {ex.get('Feedback', 'No feedback')}"
for i, ex in enumerate(component_examples) # Limit examples to prevent context overflow
])
# Use the module to improve the instruction
result = self.instruction_improver(
current_instruction=current_instruction,
feedback_summary=feedback_text,
max_words=self.max_words
)
updated_components[component_name] = result.improved_instruction
return updated_components
# Usage
gepa = dspy.GEPA(
metric=my_metric,
reflection_lm=dspy.LM(model="gpt-5", temperature=1.0, max_tokens=32000, api_key=api_key),
instruction_proposer=WordLimitProposer(max_words=700),
auto="medium"
)
Advanced Example: RAG-Enhanced Instruction Proposer¶
import dspy
from gepa.core.adapter import ProposalFn
from dspy.teleprompt.gepa.gepa_utils import ReflectiveExample
class GenerateDocumentationQuery(dspy.Signature):
"""Analyze examples with feedback to identify common issue patterns and generate targeted database queries for retrieving relevant documentation.
Your goal is to search a document database for guidelines that address the problematic patterns found in the examples. Look for recurring issues, error types, or failure modes in the feedback, then craft specific search queries that will find documentation to help resolve these patterns."""
current_instruction = dspy.InputField(desc="The current instruction that needs improvement")
examples_with_feedback = dspy.InputField(desc="Examples with their feedback showing what issues occurred and any recurring patterns")
failure_patterns: str = dspy.OutputField(desc="Summarize the common failure patterns identified in the examples")
retrieval_queries: list[str] = dspy.OutputField(desc="Specific search queries to find relevant documentation in the database that addresses the common issue patterns identified in the problematic examples")
class GenerateRAGEnhancedInstruction(dspy.Signature):
"""Generate improved instructions using retrieved documentation and examples analysis."""
current_instruction = dspy.InputField(desc="The current instruction that needs improvement")
relevant_documentation = dspy.InputField(desc="Retrieved guidelines and best practices from specialized documentation")
examples_with_feedback = dspy.InputField(desc="Examples showing what issues occurred with the current instruction")
improved_instruction: str = dspy.OutputField(desc="Enhanced instruction that incorporates retrieved guidelines and addresses the issues shown in the examples")
class RAGInstructionImprover(dspy.Module):
"""Module that uses RAG to improve instructions with specialized documentation."""
def __init__(self, retrieval_model):
super().__init__()
self.retrieve = retrieval_model # Could be dspy.Retrieve or custom retriever
self.query_generator = dspy.ChainOfThought(GenerateDocumentationQuery)
self.generate_answer = dspy.ChainOfThought(GenerateRAGEnhancedInstruction)
def forward(self, current_instruction: str, component_examples: list):
"""Improve instruction using retrieved documentation."""
# Let LM analyze examples and generate targeted retrieval queries
query_result = self.query_generator(
current_instruction=current_instruction,
examples_with_feedback=component_examples
)
results = self.retrieve.query(
query_texts=query_result.retrieval_queries,
n_results=3
)
relevant_docs_parts = []
for i, (query, query_docs) in enumerate(zip(query_result.retrieval_queries, results['documents'])):
if query_docs:
docs_formatted = "\n".join([f" - {doc}" for doc in query_docs])
relevant_docs_parts.append(
f"**Search Query #{i+1}**: {query}\n"
f"**Retrieved Guidelines**:\n{docs_formatted}"
)
relevant_docs = "\n\n" + "="*60 + "\n\n".join(relevant_docs_parts) + "\n" + "="*60
# Generate improved instruction with retrieved context
result = self.generate_answer(
current_instruction=current_instruction,
relevant_documentation=relevant_docs,
examples_with_feedback=component_examples
)
return result
class DocumentationEnhancedProposer(ProposalFn):
"""Instruction proposer that accesses specialized documentation via RAG."""
def __init__(self, documentation_retriever):
"""
Args:
documentation_retriever: A retrieval model that can search your specialized docs
Could be dspy.Retrieve, ChromadbRM, or custom retriever
"""
self.instruction_improver = RAGInstructionImprover(documentation_retriever)
def __call__(self, candidate: dict[str, str], reflective_dataset: dict[str, list[ReflectiveExample]], components_to_update: list[str]) -> dict[str, str]:
updated_components = {}
for component_name in components_to_update:
if component_name not in candidate or component_name not in reflective_dataset:
continue
current_instruction = candidate[component_name]
component_examples = reflective_dataset[component_name]
result = self.instruction_improver(
current_instruction=current_instruction,
component_examples=component_examples
)
updated_components[component_name] = result.improved_instruction
return updated_components
import chromadb
client = chromadb.Client()
collection = client.get_collection("instruction_guidelines")
gepa = dspy.GEPA(
metric=task_specific_metric,
reflection_lm=dspy.LM(model="gpt-5", temperature=1.0, max_tokens=32000, api_key=api_key),
instruction_proposer=DocumentationEnhancedProposer(collection),
auto="medium"
)
Integration Patterns¶
Using Custom Proposer with External LM:
class ExternalLMProposer(ProposalFn):
def __init__(self):
# Manage your own LM instance
self.external_lm = dspy.LM('gemini/gemini-2.5-pro')
def __call__(self, candidate, reflective_dataset, components_to_update):
updated_components = {}
with dspy.context(lm=self.external_lm):
# Your custom logic here using self.external_lm
for component_name in components_to_update:
# ... implementation
pass
return updated_components
gepa = dspy.GEPA(
metric=my_metric,
reflection_lm=None, # Optional when using custom proposer
instruction_proposer=ExternalLMProposer(),
auto="medium"
)
Best Practices:
- Use the full power of DSPy: Leverage DSPy components like
dspy.Module,dspy.Signature, anddspy.Predictto create your instruction proposer rather than direct LM calls. Considerdspy.Refinefor constraint satisfaction,dspy.ChainOfThoughtfor complex reasoning tasks, and compose multiple modules for sophisticated instruction improvement workflows - Enable holistic feedback analysis: While dspy.GEPA's
GEPAFeedbackMetricprocesses one (gold, prediction) pair at a time, instruction proposers receive all examples for a component in batch, enabling cross-example pattern detection and systematic issue identification. - Mind data serialization: Serializing everything to strings might not be ideal - handle complex input types (like
dspy.Image) by maintaining their structure for better LM processing - Test thoroughly: Test your custom proposer with representative failure cases
Custom Component Selection¶
What is component_selector?¶
The component_selector parameter controls which components (predictors) in your DSPy program are selected for optimization at each GEPA iteration. Instead of the default round-robin approach that updates one component at a time, you can implement custom selection strategies that choose single or multiple components based on optimization state, performance trajectories, and other contextual information.
Default Behavior¶
By default, GEPA uses a round-robin strategy (RoundRobinReflectionComponentSelector) that cycles through components sequentially, optimizing one component per iteration:
# Default round-robin component selection
gepa = dspy.GEPA(
metric=my_metric,
reflection_lm=dspy.LM(model="gpt-5", temperature=1.0, max_tokens=32000, api_key=api_key),
# component_selector="round_robin" # This is the default
auto="medium"
)
Built-in Selection Strategies¶
String-based selectors:
"round_robin"(default): Cycles through components one at a time"all": Selects all components for simultaneous optimization
# Optimize all components simultaneously
gepa = dspy.GEPA(
metric=my_metric,
reflection_lm=reflection_lm,
component_selector="all", # Update all components together
auto="medium"
)
# Explicit round-robin selection
gepa = dspy.GEPA(
metric=my_metric,
reflection_lm=reflection_lm,
component_selector="round_robin", # One component per iteration
auto="medium"
)
When to Use Custom Component Selection¶
Consider implementing custom component selection when you need:
- Dependency-aware optimization: Update related components together (e.g., a classifier and its input formatter)
- LLM-driven selection: Let an LLM analyze trajectories and decide which components need attention
- Resource-conscious optimization: Balance optimization thoroughness with computational budget
Custom Component Selector Protocol¶
Custom component selectors must implement the ReflectionComponentSelector protocol by defining a callable class or function. GEPA will call your selector during optimization:
from dspy.teleprompt.gepa.gepa_utils import GEPAState, Trajectory
class CustomComponentSelector:
def __call__(
self,
state: GEPAState, # Complete optimization state with history
trajectories: list[Trajectory], # Execution traces from the current minibatch
subsample_scores: list[float], # Scores for each example in the current minibatch
candidate_idx: int, # Index of the current program candidate being optimized
candidate: dict[str, str], # Component name -> instruction mapping
) -> list[str]: # Return list of component names to optimize
# Your custom component selection logic here
return selected_components
# Or as a function:
def custom_component_selector(state, trajectories, subsample_scores, candidate_idx, candidate):
# Your custom component selection logic here
return selected_components
Custom Implementation Example¶
Here's a simple function that alternates between optimizing different halves of your components:
def alternating_half_selector(state, trajectories, subsample_scores, candidate_idx, candidate):
"""Optimize half the components on even iterations, half on odd iterations."""
components = list(candidate.keys())
# If there's only one component, always optimize it
if len(components) <= 1:
return components
mid_point = len(components) // 2
# Use state.i (iteration counter) to alternate between halves
if state.i % 2 == 0:
# Even iteration: optimize first half
return components[:mid_point]
else:
# Odd iteration: optimize second half
return components[mid_point:]
# Usage
gepa = dspy.GEPA(
metric=my_metric,
reflection_lm=reflection_lm,
component_selector=alternating_half_selector,
auto="medium"
)
Integration with Custom Instruction Proposers¶
Component selectors work seamlessly with custom instruction proposers. The selector determines which components to update, then the instruction proposer generates new instructions for those components: