mcpField
D
dspy-gepa-reflective

dspy-gepa-reflective

github.com/omidzamaniAgent workflows12442 installsUpdated 18w ago
Repository

About dspy-gepa-reflective

A DSPy skill for optimizing tool-using agents and other complex workflows. It applies LLM reflection to execution traces and Pareto-based evolutionary search, using textual feedback metrics to produce an optimized program.

Key features

  1. 01Reflect on full execution traces
  2. 02Use Pareto-based evolutionary search
  3. 03Optimize instructions for agentic systems
  4. 04Require actionable textual feedback

Use cases

  1. 01Refine a multi-tool agent
  2. 02Improve a complex multi-step workflow
  3. 03Optimize instructions from failure feedback

Install

Copy the command and run it in a terminal.

$npx skills add https://github.com/omidzamani/dspy-skills --skill dspy-gepa-reflective

After installing, type/dspy-gepa-reflectivein Claude Code to confirm it loaded.

Bundled filesexample.py

Details

From the project README

DSPy GEPA Optimizer

Goal

Optimize complex agentic systems using LLM reflection on full execution traces with Pareto-based evolutionary search.

When to Use

  • Agentic systems with tool use
  • When you have rich textual feedback on failures
  • Complex multi-step workflows
  • Instruction-only optimization needed

Related Skills

Inputs

InputTypeDescription
programdspy.ModuleAgent or complex program
trainsetlist[dspy.Example]Training examples
metriccallableAccepts five arguments and returns dspy.Prediction(score=..., feedback=...)
reflection_lmdspy.LMStrong LM for reflection (GPT-4)
autostr"light", "medium", "heavy"

Outputs

OutputTypeDescription
compiled_programdspy.ModuleReflectively optimized program

Workflow

Phase 1: Define Feedback Metric

GEPA requires metrics that return textual feedback:

def gepa_metric(example, pred, trace=None, pred_name=None, pred_trace=None):
    """Return score and actionable feedback for GEPA reflection."""
    is_correct = example.answer.lower() in pred.answer.lower()
    
    if is_correct:
        feedback = "Correct. The answer accurately addresses the question."
    else:
        feedback = f"Incorrect. Expected '{example.answer}' but got '{pred.answer}'. The model may have misunderstood the question or retrieved irrelevant information."
    
    return dspy.Prediction(score=float(is_correct), feedback=feedback)
Phase 2: Setup Agent
import dspy

def search(query: str) -> list[str]:
    """Search knowledge base for relevant information."""
    rm = dspy.ColBERTv2(url='http://20.102.90.50:2017/wiki17_abstracts')
    results = rm(query, k=3)
    return results if isinstance(results, list) else [results]

def calculate(expression: str) -> float:
    """Safely evaluate mathematical expressions."""
    with dspy.PythonInterpreter() as interp:
        return interp(expression)

agent = dspy.ReAct("question -> answer", tools=[search, calculate])
Phase 3: Optimize with GEPA
dspy.configure(lm=dspy.LM("openai/gpt-4o-mini"))

optimizer = dspy.GEPA(
    metric=gepa_metric,
    reflection_lm=dspy.LM("openai/gpt-4o"),  # Strong model for reflection
    auto="medium"
)

compiled_agent = optimizer.compile(agent, trainset=trainset)

Production Example

import dspy
from dspy.evaluate import Evaluate
import logging

logger = logging.getLogger(__name__)

class ResearchAgent(dspy.Module):
    def __init__(self):
        self.react = dspy.ReAct(
            "question -> answer",
            tools=[self.search, self.summarize]
        )
    
    def search(self, query: str) -> list[str]:
        """Search for relevant documents."""
        rm = dspy.ColBERTv2(url='http://20.102.90.50:2017/wiki17_abstracts')
        results = rm(query, k=5)
        return results if isinstance(results, list) else [results]
    
    def summarize(self, text: str) -> str:
        """Summarize long text into key points."""
        summarizer = dspy.Predict("text -> summary")
        return summarizer(text=text).summary
    
    def forward(self, question):
        return self.react(question=question)

def detailed_feedback_metric(example, pred, trace=None, pred_name=None, pred_trace=None):
    """Rich feedback for GEPA reflection."""
    expected = example.answer.lower().strip()
    actual = pred.answer.lower().strip() if pred.answer else ""
    
    # Exact match
    if expected == actual:
        return dspy.Prediction(score=1.0, feedback="Perfect match. Answer is correct and concise.")
    
    # Partial match
    if expected in actual or actual in expected:
        return dspy.Prediction(score=0.7, feedback=f"Partial match. Expected '{example.answer}', got '{pred.answer}'. Answer contains correct info but may be verbose or incomplete.")
    
    # Check for key terms
    expected_terms = set(expected.split())
    actual_terms = set(actual.split())
    overlap = len(expected_terms & actual_terms) / max(len(expected_terms), 1)
    
    if overlap > 0.5:
        return dspy.Prediction(score=0.5, feedback=f"Some overlap. Expected '{example.answer}', got '{pred.answer}'. Key terms present but answer structure differs.")
    
    return dspy.Prediction(score=0.0, feedback=f"Incorrect. Expected '{example.answer}', got '{pred.answer}'. The agent may need better search queries or reasoning.")

def optimize_research_agent(trainset, devset):
    """Full GEPA optimization pipeline."""
    
    dspy.configure(lm=dspy.LM("openai/gpt-4o-mini"))
    
    agent = ResearchAgent()
    
    # Convert metric for evaluation (just score)
    def eval_metric(example, pred, trace=None):
        return detailed_feedback_metric(example, pred, trace).score
    
    evaluator = Evaluate(devset=devset, num_threads=8, metric=eval_metric)
    baseline = evaluator(agent)
    logger.info(f"Baseline: {baseline:.2%}")
    
    # GEPA optimization
    optimizer = dspy.GEPA(
        metric=detailed_feedback_metric,
        reflection_lm=dspy.LM("openai/gpt-4o"),
        auto="medium"
    )
    
    compiled = optimizer.compile(agent, trainset=trainset)
    optimized = evaluator(compiled)
    logger.info(f"Optimized: {optimized:.2%}")
    
    compiled.save("research_agent_gepa.json")
    return compiled

Metric Contract

GEPA metrics must accept (gold, pred, trace, pred_name, pred_trace). Return dspy.Prediction(score=..., feedback=...) when textual feedback is available. Do not pass enable_tool_optimization; it is not a DSPy 3.2.1 GEPA constructor argument.

Best Practices

  1. Rich feedback - More detailed feedback = better reflection
  2. Strong reflection LM - Use GPT-4 or Claude for reflection
  3. Agentic focus - Best for ReAct and multi-tool systems
  4. Trace analysis - GEPA analyzes full execution trajectories

Limitations

  • Requires custom feedback metrics (not just scores)
  • Expensive: uses strong LM for reflection
  • Newer optimizer, less battle-tested than MIPROv2
  • Best for instruction optimization, less for demos

Official Documentation

Comments

Markdown supported