dspy.GEPA: 反射式提示优化器

def compile(
    self,
    student: Module,
    *,
    trainset: list[Example],
    teacher: Module | None = None,
    valset: list[Example] | None = None,
) -> Module:
    """
    GEPA uses the trainset to perform reflective updates to the prompt, but uses the valset for tracking Pareto scores.
    If no valset is provided, GEPA will use the trainset for both.

    Parameters:
    - student: The student module to optimize.
    - trainset: The training set to use for reflective updates.
    - valset: The validation set to use for tracking Pareto scores. If not provided, GEPA will use the trainset for both.
    """
    from gepa import GEPAResult, optimize

    from dspy.teleprompt.gepa.gepa_utils import DspyAdapter, LoggerAdapter

    assert trainset is not None and len(trainset) > 0, "Trainset must be provided and non-empty"
    assert teacher is None, "Teacher is not supported in DspyGEPA yet."

    if self.auto is not None:
        self.max_metric_calls = self.auto_budget(
            num_preds=len(student.predictors()),
            num_candidates=AUTO_RUN_SETTINGS[self.auto]["n"],
            valset_size=len(valset) if valset is not None else len(trainset),
        )
    elif self.max_full_evals is not None:
        self.max_metric_calls = self.max_full_evals * (len(trainset) + (len(valset) if valset is not None else 0))
    else:
        assert self.max_metric_calls is not None, "Either auto, max_full_evals, or max_metric_calls must be set."

    logger.info(f"Running GEPA for approx {self.max_metric_calls} metric calls of the program. This amounts to {self.max_metric_calls / len(trainset) if valset is None else self.max_metric_calls / (len(trainset) + len(valset)):.2f} full evals on the {'train' if valset is None else 'train+val'} set.")

    valset = valset or trainset
    logger.info(f"Using {len(valset)} examples for tracking Pareto scores. You can consider using a smaller sample of the valset to allow GEPA to explore more diverse solutions within the same budget.")

    rng = random.Random(self.seed)

    def feedback_fn_creator(pred_name: str, predictor) -> "PredictorFeedbackFn":
        def feedback_fn(
            predictor_output: dict[str, Any],
            predictor_inputs: dict[str, Any],
            module_inputs: Example,
            module_outputs: Prediction,
            captured_trace: "DSPyTrace",
        ) -> "ScoreWithFeedback":
            trace_for_pred = [(predictor, predictor_inputs, predictor_output)]
            o = self.metric_fn(
                module_inputs,
                module_outputs,
                captured_trace,
                pred_name,
                trace_for_pred,
            )
            if hasattr(o, "feedback"):
                if o["feedback"] is None:
                    o["feedback"] = f"This trajectory got a score of {o['score']}."
                return o
            else:
                return dict(score=o, feedback=f"This trajectory got a score of {o}.")
        return feedback_fn

    feedback_map = {
        k: feedback_fn_creator(k, v)
        for k, v in student.named_predictors()
    }

    # Build the DSPy adapter that encapsulates evaluation, trace capture, feedback extraction, and instruction proposal
    adapter = DspyAdapter(
        student_module=student,
        metric_fn=self.metric_fn,
        feedback_map=feedback_map,
        failure_score=self.failure_score,
        num_threads=self.num_threads,
        add_format_failure_as_feedback=self.add_format_failure_as_feedback,
        rng=rng,
    )

    reflection_lm = self.reflection_lm

    # Instantiate GEPA with the simpler adapter-based API
    base_program = {name: pred.signature.instructions for name, pred in student.named_predictors()}
    gepa_result: GEPAResult = optimize(
        seed_candidate=base_program,
        trainset=trainset,
        valset=valset,
        adapter=adapter,

        # Reflection-based configuration
        reflection_lm=reflection_lm,
        candidate_selection_strategy=self.candidate_selection_strategy,
        skip_perfect_score=self.skip_perfect_score,
        reflection_minibatch_size=self.reflection_minibatch_size,

        perfect_score=self.perfect_score,

        # Merge-based configuration
        use_merge=self.use_merge,
        max_merge_invocations=self.max_merge_invocations,

        # Budget
        max_metric_calls=self.max_metric_calls,

        # Logging
        logger=LoggerAdapter(logger),
        run_dir=self.log_dir,
        use_wandb=self.use_wandb,
        wandb_api_key=self.wandb_api_key,
        wandb_init_kwargs=self.wandb_init_kwargs,
        track_best_outputs=self.track_best_outputs,
        display_progress_bar=True,
        raise_on_exception=True,

        # Reproducibility
        seed=self.seed,
    )

    new_prog = adapter.build_program(gepa_result.best_candidate)

    if self.track_stats:
        dspy_gepa_result = DspyGEPAResult.from_gepa_result(gepa_result, adapter)
        new_prog.detailed_results = dspy_gepa_result

    return new_prog

def __call__(
    gold: Example,
    pred: Prediction,
    trace: Optional["DSPyTrace"],
    pred_name: str | None,
    pred_trace: Optional["DSPyTrace"],
) -> Union[float, "ScoreWithFeedback"]:
    """
    This function is called with the following arguments:
    - gold: The gold example.
    - pred: The predicted output.
    - trace: Optional. The trace of the program's execution.
    - pred_name: Optional. The name of the target predictor currently being optimized by GEPA, for which 
        the feedback is being requested.
    - pred_trace: Optional. The trace of the target predictor's execution GEPA is seeking feedback for.

    Note the `pred_name` and `pred_trace` arguments. During optimization, GEPA will call the metric to obtain
    feedback for individual predictors being optimized. GEPA provides the name of the predictor in `pred_name`
    and the sub-trace (of the trace) corresponding to the predictor in `pred_trace`.
    If available at the predictor level, the metric should return dspy.Prediction(score: float, feedback: str) corresponding 
    to the predictor.
    If not available at the predictor level, the metric can also return a text feedback at the program level
    (using just the gold, pred and trace).
    If no feedback is returned, GEPA will use a simple text feedback consisting of just the score: 
    f"This trajectory got a score of {score}."
    """
    ...

def to_dict(self) -> dict[str, Any]:
    cands = [
        {k: v for k, v in cand.items()}
        for cand in self.candidates
    ]

    return dict(
        candidates=cands,
        parents=self.parents,
        val_aggregate_scores=self.val_aggregate_scores,
        best_outputs_valset=self.best_outputs_valset,
        val_subscores=self.val_subscores,
        per_val_instance_best_candidates=[list(s) for s in self.per_val_instance_best_candidates],
        discovery_eval_counts=self.discovery_eval_counts,
        total_metric_calls=self.total_metric_calls,
        num_full_val_evals=self.num_full_val_evals,
        log_dir=self.log_dir,
        seed=self.seed,
        best_idx=self.best_idx,
    )

@staticmethod
def from_gepa_result(gepa_result: "GEPAResult", adapter: "DspyAdapter") -> "DspyGEPAResult":
    return DspyGEPAResult(
        candidates=[adapter.build_program(c) for c in gepa_result.candidates],
        parents=gepa_result.parents,
        val_aggregate_scores=gepa_result.val_aggregate_scores,
        best_outputs_valset=gepa_result.best_outputs_valset,
        val_subscores=gepa_result.val_subscores,
        per_val_instance_best_candidates=gepa_result.per_val_instance_best_candidates,
        discovery_eval_counts=gepa_result.discovery_eval_counts,
        total_metric_calls=gepa_result.total_metric_calls,
        num_full_val_evals=gepa_result.num_full_val_evals,
        log_dir=gepa_result.run_dir,
        seed=gepa_result.seed,
    )