dspy-gepa-reflective

DSPy GEPA オプティマイザー

目標

実行トレース全体に対する LLM の振り返りと、パレートベースの進化的探索を使用して、複雑なエージェント型システムを最適化します。

使用する場面

• ツールを使用するエージェント型システム
• 失敗に関する豊富なテキスト形式のフィードバックがある場合
• 複雑なマルチステップワークフロー
• 指示のみの最適化が必要な場合

関連スキル

• エージェント型ではないプログラムの場合：dspy-miprov2-optimizer、dspy-bootstrap-fewshot
• 改善を測定する場合：dspy-evaluation-suite

入力

入力	型	説明
program	dspy.Module	エージェントまたは複雑なプログラム
trainset	list[dspy.Example]	トレーニング用の例
metric	callable	5つの引数を受け取り、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：エージェントを設定する

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/
• エージェントガイド：https://dspy.ai/tutorials/agents/