dspy-gepa-reflective

DSPy GEPA 优化器

目标

使用 LLM 基于完整执行轨迹进行反思，并通过基于 Pareto 的进化搜索优化复杂的智能体系统。

适用场景

• 使用工具的智能体系统
• 失败时有丰富的文本反馈
• 复杂的多步骤工作流
• 需要仅优化指令

相关技能

• 对于非智能体程序：dspy-miprov2-optimizer、dspy-bootstrap-fewshot
• 衡量改进效果：dspy-evaluation-suite

输入

输入	类型	描述
program	dspy.Module	Agent 或复杂程序
trainset	list[dspy.Example]	训练示例
metric	callable	接受五个参数，并返回 dspy.Prediction(score=..., feedback=...)
reflection_lm	dspy.LM	用于反思的强大 LM（GPT-4）
auto	str	"light"、"medium"、"heavy"

输出

输出	类型	描述
compiled_program	dspy.Module	经过反思优化的程序

工作流

阶段 1：定义反馈指标

GEPA 要求指标返回文本反馈：

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)

阶段 2：设置 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])

阶段 3：使用 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)

生产示例

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

指标约定

GEPA 指标必须接受 (gold, pred, trace, pred_name, pred_trace)。有文本反馈时，返回 dspy.Prediction(score=..., feedback=...)。不要传入 enable_tool_optimization；它不是 DSPy 3.2.1 中 GEPA 的构造函数参数。

最佳实践

1. 丰富的反馈——反馈越详细，反思效果越好
2. 强大的反思 LM——使用 GPT-4 或 Claude 进行反思
3. 侧重智能体系统——最适用于 ReAct 和多工具系统
4. 轨迹分析——GEPA 会分析完整的执行轨迹

限制

• 需要自定义反馈指标（不能只有分数）
• 成本较高：使用强大的 LM 进行反思
• 这是较新的优化器，经过实战验证的程度不及 MIPROv2
• 最适用于指令优化，不太适用于演示示例

官方文档

• DSPy 文档：https://dspy.ai/
• DSPy GitHub：https://github.com/stanfordnlp/dspy
• GEPA 优化器：https://dspy.ai/api/optimizers/GEPA/
• Agent 指南：https://dspy.ai/tutorials/agents/