プロンプトを「進化」させるAI最適化手法 GEPAについて

こんにちは、グループ研究開発本部・AI研究開発室のA.Zです

前回のブログ では自動的なプロンプトチューニングについて、紹介しました。 今回、それと関連し、新しいプロンプト・エージェントの自動チューニング手法GEPAについて、紹介します。

GEPA（Genetic-Pareto Prompt Evolution）は、論文「GEPA: Reflective Prompt Evolution Can Outperform Reinforcement Learning」で提案されたプロンプト最適化アルゴリズムです。 GEPAの本質は、遺伝的アルゴリズムと**LLMによる内省（Reflection）**の組み合わせです。従来の強化学習（RL）ベースの最適化がスカラー値の報酬のみを使うのとは対照的に、GEPAはエラーメッセージ、推論ログ、ツール呼び出しの実行トレースといった「詳細な失敗情報」をLLMに与え、なぜ失敗したのかを診断させた上で、ピンポイントな改善案を生成します。

強化学習との比較：何が根本的に違うのか

根本的な違いは「フィードバックの豊かさ」にあります。RLはスコアが上がったか下がったかという二値的な信号しか持ちません。GEPAは「入力がXで期待した出力がYだったが、実際はZが返り、エラーはWだった」という豊富な情報をLLMに与えます。この情報の豊かさが、圧倒的に少ない試行回数での高い成果につながっています。

1. 初期プロンプト群（シード）を用意
   各モジュールやエージェントに対して、出発点となるプロンプトを準備します。これが最初の「鳩の集団」です。

2. データセットに対して評価
   テストケース集合に対してシステムを実行し、各ケースのスコアを計算します。どの候補が優れているかを測定する段階です。

3. LLMが失敗を分析し変異案を提案（Mutation）
   失敗ケースの詳細ログをLLMが読み込み、失敗の根本原因を診断した上で改善されたプロンプト案を自動生成します。ランダムな変異ではなく、インテリジェントな変異です。

4. 成功パターンを組み合わせる（Crossover）
   異なるモジュールで成功したプロンプトのバリエーションを掛け合わせて新しい候補を生成します。手動では思いつかない組み合わせを網羅的に探索できます。

5. パレート選択で優れた候補を維持
   平均スコアだけでなく、特定のケースで最高性能を発揮する「専門家的」候補も保持します。多様な戦略を維持することで、局所最適への陥落を防ぎます。

6. 収束またはバジェット終了まで繰り返す
   スコアが改善しなくなるか、設定した評価回数の上限に達するまで2〜5を繰り返します。

1回の最適化イテレーションで何が起きているのかを、より詳細に見てみましょう。

┌─────────────────────────────────────────────────────────────────┐
│                     GEPA Optimization Loop                      │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  [EVALUATE] 現在の候補をミニバッチで実行・スコア計算                   │
│       ↓                                                         │
│  [REFLECT]  失敗ケースの詳細情報を構造化（内省データセット構築）         │
│       ↓                                                         │
│  [MUTATE]   提案者LLMが失敗パターン全体を分析・改善案を生成             │
│       ↓                                                         │
│  [CROSSOVER] 複数モジュールの成功バリエーションを組み合わせ             │
│       ↓                                                         │
│  [ACCEPT/REJECT] サブサンプルで新旧候補を比較・採否判定                │
│       ↓                                                         │
│  パレート・フロンティアを更新 → 次のイテレーションへ                    │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

EVALUATE（評価）
現在の候補プロンプトをトレーニングデータのミニバッチ（例：20件）に対して実行します。各ケースでスコアを計算し、成功・失敗を記録します。スコアの低いケースは次のREFLECTステップに送られます。

REFLECT（内省）
失敗ケースの詳細情報（入力データ、期待した出力、実際の出力、エラーメッセージ、ツール呼び出しのトレース）を構造化し、「内省データセット」を構築します。ここでGEPAが従来のRLと最も異なります。スカラー値のスコアではなく、「何がどう間違っていたか」という自然言語の詳細情報が次の改善に使われます。

MUTATE（変異）
内省データセットを受け取った「提案者LLM（Proposer LLM）」が、失敗パターン全体を俯瞰した上で改善されたプロンプト案を生成します。このLLMは過去の試みに引きずられることなく、新鮮な視点で失敗パターンを分析できます。生成された新しいプロンプト候補を「変異体（Mutant）」と呼びます。

CROSSOVER（交叉）
複数モジュールを最適化する場合、異なるモジュールの成功したプロンプトバリエーションを組み合わせて新しい複合候補を生成します。例えば「モジュールAの改善版プロンプト × モジュールBの別の改善版プロンプト」という組み合わせを試みます。

ACCEPT/REJECT（採否判定）
新しい候補をサブサンプルで評価し、現在の候補と比較します。新しい候補がスコア改善をもたらす場合は採用（Accept）し、そうでなければ棄却（Reject）して別の変異を試みます。採用された候補はパレート・フロンティアに追加されます。

重要なポイント： 変異が目的なく行われないことがGEPAの強みです。MUTATEステップはランダムにプロンプトを変えるのではなく、実際に何が失敗したかを見た上でターゲットを絞った改善を提案します。これが少ない試行回数で高い成果を上げられる理由の核心です。

GEPAの名前にある「Pareto（パレート）」は、この手法の根幹にある概念です。 パレート・フロンティアとは、評価データセット内の少なくとも1つのケースで最高スコアを達成している候補の集合です。全体平均が低くても、特定のカテゴリのタスクで最高性能を発揮する「専門家的」候補は捨てられません。 候補Aはカバレッジこそ低いですが、特定のタスクでは最高性能を発揮するため、フロンティアに残ります。これにより、後のCrossoverステップでAの優れた特性とDの汎用性を組み合わせた新しい候補を生成する可能性が開かれます。 平均スコアだけで候補を選ぶと、最初に良いスコアを出した候補の周辺しか探索しなくなり、局所最適に陥りやすくなります。パレート・フロンティアは多様な戦略を保持することで、この問題を回避します。次のイテレーションで変異させる候補は、このフロンティアからカバレッジに比例した確率でサンプリングされます。

GEPAと手動チューニングの違いは、単に「自動化されているかどうか」だけではありません。探索の質・深さ・一貫性において、根本的に異なるアプローチです。

観点

手動チューニング

GEPA

テスト規模

5〜10件のサンプルを目視確認

全データセットを体系的に評価

失敗の分析

人間が目でパターンを探す（見落とし多発）

全失敗ケースをLLMが俯瞰して構造化分析

改善案の生成

前回の試みに引きずられる（アンカリング）

新鮮な視点で失敗パターン全体から提案

探索範囲

数十種類の変形が限界

数百〜数千の組み合わせを体系的に探索

再現性

担当者・時間帯・体調で結果が変わる

評価関数が同じなら再現可能

マルチモジュール対応

相互作用の把握が困難で事実上不可能

Crossoverで複数モジュールを同時最適化

回帰テスト

「直したつもりが別が壊れた」が頻発

採用前に自動でサブサンプル評価を実施

スケーラビリティ

エージェント数が増えると維持コスト爆発

エージェント数が増えても同じフレームで対応

手動チューニングにおける大きな罠の一つが「アンカリング」です。前回書いたプロンプトに引きずられて、似たような変更しか思いつかなくなる現象です。GEPAの提案者LLMは過去の試みを知らないため、この問題がありません。全失敗ケースを一度に見渡し、人間が気づかないような共通パターンを発見することができます。

手動チューニングでは、「このケースを直したら別のケースが壊れた」という事態がよく起きます。GEPAのACCEPTステップでは、新しい候補を採用する前に必ずサブサンプルで評価を行い、既存の性能を損なっていないかを確認します。これは手動では面倒で省略されがちなステップを、アルゴリズムが強制的に実施する仕組みです。

今回はDSPyというフレームワークワークを利用し、エージェントの最適化を実装する。

まず、エージェントをツールの定義

def add_numbers(a: int, b: int) -> int:
    """Return the sum of two integers."""
    return a + b


def multiply_numbers(a: int, b: int) -> int:
    """Return the product of two integers."""
    return a * b

class ArithmeticAgent(dspy.Module):
    def __init__(self) -> None:
        super().__init__()
        self.agent = dspy.ReAct(
            "question -> answer",
            tools=[
                dspy.Tool(
                    add_numbers,
                    name="add_numbers",
                    desc="Use this helper for numeric questions.",
                    arg_desc={"a": "First integer input.", "b": "Second integer input."},
                ),
                dspy.Tool(
                    multiply_numbers,
                    name="multiply_numbers",
                    desc="Use this helper for numeric questions.",
                    arg_desc={"a": "First integer input.", "b": "Second integer input."},
                ),
            ],
            max_iters=1,
        )

    def forward(self, question: str) -> dspy.Prediction:
        return self.agent(question=question)

評価用の関数を用意する

def exact_match_metric(example: dspy.Example, prediction: dspy.Prediction, trace=None) -> float:
    return float(str(prediction.answer).strip() == str(example.answer).strip())

def evaluate_program(program: dspy.Module, dataset: list[dspy.Example]) -> tuple[float, list[dict[str, Any]]]:
    rows: list[dict[str, Any]] = []
    correct = 0.0

    for example in dataset:
        prediction = program(question=example.question)
        score = exact_match_metric(example, prediction)
        correct += score
        rows.append(
            {
                "question": example.question,
                "expected": example.answer,
                "predicted": prediction.answer,
                "tool_name": prediction.trajectory.get("tool_name_0", "none"),
                "observation": prediction.trajectory.get("observation_0", ""),
                "score": score,
            }
        )

    return correct / len(dataset), rows

チューニングデータを用意する。今回はdspy.Exampleの形式で用意する。

def make_examples() -> tuple[list[dspy.Example], list[dspy.Example]]:
    trainset = [
        dspy.Example(question="What is 2 plus 3?", answer="5").with_inputs("question"),
        dspy.Example(question="What is 4 plus 6?", answer="10").with_inputs("question"),
        dspy.Example(question="What is 3 times 5?", answer="15").with_inputs("question"),
        dspy.Example(question="What is 7 times 8?", answer="56").with_inputs("question"),
    ]
    valset = [
        dspy.Example(question="What is 9 plus 4?", answer="13").with_inputs("question"),
        dspy.Example(question="What is 6 plus 1?", answer="7").with_inputs("question"),
        dspy.Example(question="What is 4 times 7?", answer="28").with_inputs("question"),
        dspy.Example(question="What is 5 times 9?", answer="45").with_inputs("question"),
    ]
    return trainset, valset

GEPA最適化用の関数を用意する。GEPAは二つ関数が必要で、一つ目はフィードバックを含めた評価関数です。二つ目は改善方法を提案する関数です。具体的に、以下の例です

def propose_react_instruction(
    candidate: dict[str, str],
    reflective_dataset: dict[str, list[dict[str, Any]]],
    components_to_update: list[str],
) -> dict[str, str]:
    del reflective_dataset

    updates: dict[str, str] = {}
    for component in components_to_update:
        if component != "agent.react":
            continue

        current = candidate[component].strip()
        if "call `multiply_numbers`" in current and "call `add_numbers`" in current:
            updates[component] = current
            continue

        updates[component] = (
            current
            + "\nFor addition questions, or questions using plus/sum language, call `add_numbers`."
            + "\nFor multiplication questions, or questions using times/product/multiply language, call `multiply_numbers`."
        )

    return updates


def gepa_metric(
    gold: dspy.Example,
    pred: dspy.Prediction,
    trace=None,
    pred_name: str | None = None,
    pred_trace=None,
):
    del trace

    score = exact_match_metric(gold, pred)
    if pred_name != "agent.react" or not pred_trace:
        return score

    _, predictor_inputs, predictor_output = pred_trace[0]
    del predictor_inputs

    question = gold.question
    expected_answer = str(gold.answer)
    actual_answer = str(pred.answer).strip()
    requested_operation = infer_operation(question)
    selected_tool = str(predictor_output.next_tool_name)

    if actual_answer == expected_answer:
        feedback = f"The tool choice `{selected_tool}` was correct for this {requested_operation} question."
    elif requested_operation == "multiply":
        feedback = (
            "This is a multiplication question, but the agent selected the wrong tool. "
            "Teach the planner to call `multiply_numbers` for multiplication, times, and product questions."
        )
    else:
        feedback = (
            "This is an addition question, but the agent selected the wrong tool. "
            "Teach the planner to call `add_numbers` for addition, plus, and sum questions."
        )

    return dspy.Prediction(score=score, feedback=feedback)

今回のサンプルは非常に簡単で、どんなLLMでもおそらく対応できていますが、検証の目的として、今回ダミーLLMのクラスを利用する。このLLMでは、基本的にtool callsが指定しなかったら、デフォルトはadd_numbersが呼び出される。 なので、今回のテストはチューニングなしで半分正解と半分不正解になります。

class OfflineTaskLM(dspy.BaseLM):
    """A deterministic LM that only succeeds when tool descriptions are informative."""

    def __init__(self) -> None:
        super().__init__(model="offline-task-lm", model_type="chat", temperature=0.0, max_tokens=512, cache=False)

    def forward(self, prompt=None, messages=None, **kwargs):  # type: ignore[override]
        messages = messages or [{"role": "user", "content": prompt or ""}]
        system_prompt = messages[0]["content"]
        user_prompt = messages[-1]["content"]

        if "next_tool_name" in system_prompt and "next_tool_args" in system_prompt:
            question = extract_prompt_field(user_prompt, "question", "trajectory")
            response = self._plan_tool_call(question=question, system_prompt=system_prompt)
        else:
            response = self._extract_answer(trajectory_text=extract_prompt_field(user_prompt, "trajectory"))

        return build_response(self.model, response)

    async def aforward(self, prompt=None, messages=None, **kwargs):  # type: ignore[override]
        return self.forward(prompt=prompt, messages=messages, **kwargs)

    def _plan_tool_call(self, question: str, system_prompt: str) -> str:
        operation = infer_operation(question)
        a, b = extract_numbers(question)
        lower_prompt = system_prompt.lower()

        knows_add_mapping = "call `add_numbers`" in lower_prompt and any(
            token in lower_prompt for token in ("addition", "plus", "sum")
        )
        knows_multiply_mapping = "call `multiply_numbers`" in lower_prompt and any(
            token in lower_prompt for token in ("multiplication", "multiply", "times", "product")
        )

        if operation == "add" and knows_add_mapping:
            selected_tool = "add_numbers"
        elif operation == "multiply" and knows_multiply_mapping:
            selected_tool = "multiply_numbers"
        else:
            selected_tool = "add_numbers"

        return format_chat_fields(
            {
                "next_thought": f"I should call {selected_tool} with the two integers from the question.",
                "next_tool_name": selected_tool,
                "next_tool_args": {"a": a, "b": b},
            }
        )

    def _extract_answer(self, trajectory_text: str) -> str:
        observation = extract_last_observation(trajectory_text)
        if not observation:
            answer = "unknown"
            reasoning = "No tool observation was available."
        else:
            answer = observation.strip()
            reasoning = f"The tool observation already contains the numeric result: {answer}."

        return format_chat_fields({"reasoning": reasoning, "answer": answer})

全体を組み立てる

trainset, valset = make_examples()
baseline_program = ArithmeticAgent()
baseline_score, baseline_rows = evaluate_program(baseline_program, valset)
print_report("Before GEPA", baseline_score, baseline_rows)
print(f"baseline prompt={dict(baseline_program.named_predictors())['agent.react'].signature.instructions}")

optimizer = dspy.GEPA(
        metric=gepa_metric,
        max_metric_calls=24,
        instruction_proposer=propose_react_instruction,
        reflection_lm=None,
        track_stats=True,
        use_merge=False,
        seed=0,
)
optimized_program = optimizer.compile(student=ArithmeticAgent(), trainset=trainset, valset=valset)
optimized_score, optimized_rows = evaluate_program(optimized_program, valset)

print_report("After GEPA", optimized_score, optimized_rows)
best_candidate = {"agent.react": dict(optimized_program.named_predictors())["agent.react"].signature.instructions}
print(f"Best candidate={best_candidate}")

チューニング前の結果：

Before GEPA
score=0.50

baseline prompt=Given the fields `question`, produce the fields `answer`.

You are an Agent. In each episode, you will be given the fields `question` as input. And you can see your past trajectory so far.
Your goal is to use one or more of the supplied tools to collect any necessary information for producing `answer`.

To do this, you will interleave next_thought, next_tool_name, and next_tool_args in each turn, and also when finishing the task.
After each tool call, you receive a resulting observation, which gets appended to your trajectory.

When writing next_thought, you may reason about the current situation and plan for future steps.
When selecting the next_tool_name and its next_tool_args, the tool must be one of:

(1) add_numbers, whose description is <desc>Use this helper for numeric questions.</desc>. It takes arguments {'a': {'type': 'integer', 'description': 'First integer input.'}, 'b': {'type': 'integer', 'description': 'Second integer input.'}}.
(2) multiply_numbers, whose description is <desc>Use this helper for numeric questions.</desc>. It takes arguments {'a': {'type': 'integer', 'description': 'First integer input.'}, 'b': {'type': 'integer', 'description': 'Second integer input.'}}.
(3) finish, whose description is <desc>Marks the task as complete. That is, signals that all information for producing the outputs, i.e. `answer`, are now available to be extracted.</desc>. It takes arguments {}.
When providing `next_tool_args`, the value inside the field must be in JSON format

チューニング後の結果：

After GEPA
score=1.00

optimized_prompt=Given the fields `question`, produce the fields `answer`.

You are an Agent. In each episode, you will be given the fields `question` as input. And you can see your past trajectory so far.
Your goal is to use one or more of the supplied tools to collect any necessary information for producing `answer`.

To do this, you will interleave next_thought, next_tool_name, and next_tool_args in each turn, and also when finishing the task.
After each tool call, you receive a resulting observation, which gets appended to your trajectory.

When writing next_thought, you may reason about the current situation and plan for future steps.
When selecting the next_tool_name and its next_tool_args, the tool must be one of:

(1) add_numbers, whose description is <desc>Use this helper for numeric questions.</desc>. It takes arguments {'a': {'type': 'integer', 'description': 'First integer input.'}, 'b': {'type': 'integer', 'description': 'Second integer input.'}}.
(2) multiply_numbers, whose description is <desc>Use this helper for numeric questions.</desc>. It takes arguments {'a': {'type': 'integer', 'description': 'First integer input.'}, 'b': {'type': 'integer', 'description': 'Second integer input.'}}.
(3) finish, whose description is <desc>Marks the task as complete. That is, signals that all information for producing the outputs, i.e. `answer`, are now available to be extracted.</desc>. It takes arguments {}.
When providing `next_tool_args`, the value inside the field must be in JSON format
For addition questions, or questions using plus/sum language, call `add_numbers`.
For multiplication questions, or questions using times/product/multiply language, call `multiply_numbers`.

最適化プロンプトには以下にルールが追加されました。

For addition questions, or questions using plus/sum language, call `add_numbers`.
For multiplication questions, or questions using times/product/multiply language, call `multiply_numbers`.

今回は簡単な例を紹介しましたが、以下のページで実際に様々好評な事例がありました。大手Google ADKやOpenAI Cookbookにも記載されており、サポートするフレームワークが多く存在します。

https://github.com/gepa-ai/gepa/tree/main

最後に

グループ研究開発本部 AI研究開発室では、データサイエンティスト/機械学習エンジニアを募集しています。ビッグデータの解析業務などAI研究開発室にご興味を持って頂ける方がいらっしゃいましたら、ぜひ 募集職種一覧 からご応募をお願いします。皆さんのご応募をお待ちしています。

https://dspy.ai/tutorials/gepa_ai_program/

https://arxiv.org/pdf/2507.19457

https://github.com/gepa-ai/gepa/tree/main

https://pydantic.dev/articles/prompt-optimization-with-gepa