dspy.MIPROv2

MIPROv2（Multiprompt Instruction PRoposal Optimizer Version 2，多提示指令建议优化器版本2）是一种提示优化器，能够联合优化指令和少样本示例。它通过自举（bootstrapping）少样本示例候选项，提出基于任务不同动态的指令，并利用贝叶斯优化找到这些选项的最佳组合来实现这一点。它可以用于联合优化少样本示例和指令，也可以仅用于零样本优化的指令。

dspy.MIPROv2(metric: Callable, prompt_model: Optional[Any] = None, task_model: Optional[Any] = None, teacher_settings: Optional[dict] = None, max_bootstrapped_demos: int = 4, max_labeled_demos: int = 4, auto: Optional[Literal['light', 'medium', 'heavy']] = 'light', num_candidates: Optional[int] = None, num_threads: Optional[int] = None, max_errors: int = 10, seed: int = 9, init_temperature: float = 0.5, verbose: bool = False, track_stats: bool = True, log_dir: Optional[str] = None, metric_threshold: Optional[float] = None)

基类: Teleprompter

源代码位于 dspy/teleprompt/mipro_optimizer_v2.py

def __init__(
    self,
    metric: Callable,
    prompt_model: Optional[Any] = None,
    task_model: Optional[Any] = None,
    teacher_settings: Optional[dict] = None,
    max_bootstrapped_demos: int = 4,
    max_labeled_demos: int = 4,
    auto: Optional[Literal["light", "medium", "heavy"]] = "light",
    num_candidates: Optional[int] = None,
    num_threads: Optional[int] = None,
    max_errors: int = 10,
    seed: int = 9,
    init_temperature: float = 0.5,
    verbose: bool = False,
    track_stats: bool = True,
    log_dir: Optional[str] = None,
    metric_threshold: Optional[float] = None,
):
    # Validate 'auto' parameter
    allowed_modes = {None, "light", "medium", "heavy"}
    if auto not in allowed_modes:
        raise ValueError(f"Invalid value for auto: {auto}. Must be one of {allowed_modes}.")
    self.auto = auto
    self.num_fewshot_candidates = num_candidates
    self.num_instruct_candidates = num_candidates
    self.num_candidates = num_candidates
    self.metric = metric
    self.init_temperature = init_temperature
    self.task_model = task_model if task_model else dspy.settings.lm
    self.prompt_model = prompt_model if prompt_model else dspy.settings.lm
    self.max_bootstrapped_demos = max_bootstrapped_demos
    self.max_labeled_demos = max_labeled_demos
    self.verbose = verbose
    self.track_stats = track_stats
    self.log_dir = log_dir
    self.teacher_settings = teacher_settings or {}
    self.prompt_model_total_calls = 0
    self.total_calls = 0
    self.num_threads = num_threads
    self.max_errors = max_errors
    self.metric_threshold = metric_threshold
    self.seed = seed
    self.rng = None

函数

compile(student: Any, *, trainset: List, teacher: Any = None, valset: Optional[List] = None, num_trials: Optional[int] = None, max_bootstrapped_demos: Optional[int] = None, max_labeled_demos: Optional[int] = None, seed: Optional[int] = None, minibatch: bool = True, minibatch_size: int = 35, minibatch_full_eval_steps: int = 5, program_aware_proposer: bool = True, data_aware_proposer: bool = True, view_data_batch_size: int = 10, tip_aware_proposer: bool = True, fewshot_aware_proposer: bool = True, requires_permission_to_run: bool = True, provide_traceback: Optional[bool] = None) -> Any

源代码位于 dspy/teleprompt/mipro_optimizer_v2.py

def compile(
    self,
    student: Any,
    *,
    trainset: List,
    teacher: Any = None,
    valset: Optional[List] = None,
    num_trials: Optional[int] = None,
    max_bootstrapped_demos: Optional[int] = None,
    max_labeled_demos: Optional[int] = None,
    seed: Optional[int] = None,
    minibatch: bool = True,
    minibatch_size: int = 35,
    minibatch_full_eval_steps: int = 5,
    program_aware_proposer: bool = True,
    data_aware_proposer: bool = True,
    view_data_batch_size: int = 10,
    tip_aware_proposer: bool = True,
    fewshot_aware_proposer: bool = True,
    requires_permission_to_run: bool = True,
    provide_traceback: Optional[bool] = None,
) -> Any:

    zeroshot_opt = (self.max_bootstrapped_demos == 0) and (self.max_labeled_demos == 0)

    # If auto is None, and num_trials is not provided (but num_candidates is), raise an error that suggests a good num_trials value
    if self.auto is None and (self.num_candidates is not None and num_trials is None):
        raise ValueError(f"If auto is None, num_trials must also be provided. Given num_candidates={self.num_candidates}, we'd recommend setting num_trials to ~{self._set_num_trials_from_num_candidates(student, zeroshot_opt, self.num_candidates)}.")

    # If auto is None, and num_candidates or num_trials is None, raise an error
    if self.auto is None and (self.num_candidates is None or num_trials is None):
        raise ValueError("If auto is None, num_candidates must also be provided.")

    # If auto is provided, and either num_candidates or num_trials is not None, raise an error
    if self.auto is not None and (self.num_candidates is not None or num_trials is not None):
        raise ValueError("If auto is not None, num_candidates and num_trials cannot be set, since they would be overrided by the auto settings. Please either set auto to None, or do not specify num_candidates and num_trials.")

    # Set random seeds
    seed = seed or self.seed
    self._set_random_seeds(seed)

    # Update max demos if specified
    if max_bootstrapped_demos is not None:
        self.max_bootstrapped_demos = max_bootstrapped_demos
    if max_labeled_demos is not None:
        self.max_labeled_demos = max_labeled_demos

    # Set training & validation sets
    trainset, valset = self._set_and_validate_datasets(trainset, valset)

    # Set hyperparameters based on run mode (if set)
    num_trials, valset, minibatch = self._set_hyperparams_from_run_mode(
        student, num_trials, minibatch, zeroshot_opt, valset
    )

    if self.auto:
        self._print_auto_run_settings(num_trials, minibatch, valset)

    if minibatch and minibatch_size > len(valset):
        raise ValueError(f"Minibatch size cannot exceed the size of the valset. Valset size: {len(valset)}.")

    # Estimate LM calls and get user confirmation
    if requires_permission_to_run:
        if not self._get_user_confirmation(
            student,
            num_trials,
            minibatch,
            minibatch_size,
            minibatch_full_eval_steps,
            valset,
            program_aware_proposer,
        ):
            logger.info("Compilation aborted by the user.")
            return student  # Return the original student program

    # Initialize program and evaluator
    program = student.deepcopy()
    evaluate = Evaluate(
        devset=valset,
        metric=self.metric,
        num_threads=self.num_threads,
        max_errors=self.max_errors,
        display_table=False,
        display_progress=True,
        provide_traceback=provide_traceback,
    )

    # Step 1: Bootstrap few-shot examples
    demo_candidates = self._bootstrap_fewshot_examples(program, trainset, seed, teacher)

    # Step 2: Propose instruction candidates
    instruction_candidates = self._propose_instructions(
        program,
        trainset,
        demo_candidates,
        view_data_batch_size,
        program_aware_proposer,
        data_aware_proposer,
        tip_aware_proposer,
        fewshot_aware_proposer,
    )

    # If zero-shot, discard demos
    if zeroshot_opt:
        demo_candidates = None

    # Step 3: Find optimal prompt parameters
    best_program = self._optimize_prompt_parameters(
        program,
        instruction_candidates,
        demo_candidates,
        evaluate,
        valset,
        num_trials,
        minibatch,
        minibatch_size,
        minibatch_full_eval_steps,
        seed,
    )

    return best_program

get_params() -> dict[str, Any]

获取 teleprompter 的参数。

返回

类型	描述
dict[str, Any]	teleprompter 的参数。

源代码位于 dspy/teleprompt/teleprompt.py

def get_params(self) -> dict[str, Any]:
    """
    Get the parameters of the teleprompter.

    Returns:
        The parameters of the teleprompter.
    """
    return self.__dict__

示例用法

下面的程序展示了如何使用 MIPROv2 优化数学程序

import dspy
from dspy.datasets.gsm8k import GSM8K, gsm8k_metric

# Import the optimizer
from dspy.teleprompt import MIPROv2

# Initialize the LM
lm = dspy.LM('openai/gpt-4o-mini', api_key='YOUR_OPENAI_API_KEY')
dspy.configure(lm=lm)

# Initialize optimizer
teleprompter = MIPROv2(
    metric=gsm8k_metric,
    auto="medium", # Can choose between light, medium, and heavy optimization runs
)

# Optimize program
print(f"Optimizing program with MIPROv2...")
optimized_program = teleprompter.compile(
    dspy.ChainOfThought("question -> answer"),
    trainset=gsm8k.train,
    requires_permission_to_run=False,
)

# Save optimize program for future use
optimized_program.save(f"optimized.json")

MIPROv2 工作原理

从高层次上看，MIPROv2 通过为您的 LM 程序中的每个预测器创建少样本示例和新指令，然后使用贝叶斯优化搜索这些选项的最佳组合来工作。如果您想要一个可视化解释，请查看这个 Twitter 帖子。

以下是更详细的步骤分解

1) 自举少样本示例：从您的训练集中随机采样示例，并在您的 LM 程序中运行它们。如果程序对此示例的输出是正确的，则将其保留为有效的少样本示例候选项。否则，我们将尝试另一个示例，直到我们收集到指定数量的少样本示例候选项。此步骤创建 num_candidates 组，每组包含 max_bootstrapped_demos 个自举示例和 max_labeled_demos 个从训练集中采样的基本示例。

2) 提出指令候选项。指令提出器包括 (1) 训练数据集属性的生成摘要，(2) LM 程序代码和正在生成指令的特定预测器的生成摘要，(3) 先前自举的少样本示例，用于显示给定预测器的参考输入/输出，以及 (4) 随机采样的生成提示（例如，“要有创意”、“要简洁”等），以帮助探索潜在指令的特征空间。此上下文提供给 prompt_model，由其编写高质量的指令候选项。

3) 找到少样本示例与指令的优化组合。最后，我们使用贝叶斯优化来选择指令和演示的哪些组合最适合我们程序中的每个预测器。这通过运行一系列 num_trials 次试验来实现，在每次试验中，都会在我们的验证集上评估一组新的提示。每次试验中（当 minibatch=True 时），新提示集仅在大小为 minibatch_size 的小批量上进行评估。然后，每隔 minibatch_full_eval_steps 步，在完整的验证集上评估平均表现最好的提示集。在优化过程结束时，返回在完整验证集上表现最佳的提示集所对应的 LM 程序。

对于那些对更多细节感兴趣的人，可以在这篇论文中找到有关 MIPROv2 的更多信息，以及与 MIPROv2 相比其他 DSPy 优化器的研究。