From d9ff25387012cabec75c77b71fe24fcfffe0611a Mon Sep 17 00:00:00 2001 From: Gregory Pevnev Date: Mon, 1 Jun 2026 12:11:19 +0200 Subject: [PATCH] feat: PE2 + SGR prompt optimization methods Add PE2 beam-search optimizer, PE2+SGR (Schema-Guided Reasoning) variant, APE, and OPRO methods. Includes IFEval metric, structured Pydantic diagnosis schemas, adaptive mode switching, benchmark infrastructure with local model ladder, and multi-seed evaluation runners. --- .gitignore | 7 +- coolprompt/evaluator/ifeval_checkers.py | 186 + coolprompt/evaluator/metrics.py | 79 + coolprompt/evaluator/portfolio.py | 18 + coolprompt/optimizer/ape/__init__.py | 6 + coolprompt/optimizer/ape/proposer.py | 66 + coolprompt/optimizer/ape/run.py | 101 + coolprompt/optimizer/ape/trainer.py | 185 + coolprompt/optimizer/hyper/__init__.py | 1 - coolprompt/optimizer/opro/__init__.py | 6 + coolprompt/optimizer/opro/proposer.py | 139 + coolprompt/optimizer/opro/run.py | 108 + coolprompt/optimizer/opro/trainer.py | 169 + coolprompt/optimizer/pe2/__init__.py | 6 + coolprompt/optimizer/pe2/node.py | 34 + coolprompt/optimizer/pe2/proposer.py | 76 + coolprompt/optimizer/pe2/run.py | 102 + coolprompt/optimizer/pe2/trainer.py | 415 ++ coolprompt/optimizer/pe2_sgr/__init__.py | 9 + coolprompt/optimizer/pe2_sgr/proposer.py | 437 ++ coolprompt/optimizer/pe2_sgr/run.py | 115 + coolprompt/optimizer/pe2_sgr/schemas.py | 164 + coolprompt/utils/enums.py | 18 + .../utils/prompt_templates/ape_templates.py | 12 + .../utils/prompt_templates/opro_templates.py | 24 + .../prompt_templates/pe2_sgr_templates.py | 206 + .../utils/prompt_templates/pe2_templates.py | 85 + coolprompt/utils/var_validation.py | 8 + pyproject.toml | 136 +- src/solutions/PE2/__init__.py | 0 src/solutions/PE2/extra_benchmarks_config.py | 214 + src/solutions/PE2/extra_benchmarks_test.py | 163 + src/solutions/PE2/local_models.py | 157 + src/solutions/PE2/pe2_baseline_test.py | 101 + src/solutions/PE2/pe2_paper_config.py | 256 ++ src/solutions/PE2/pe2_paper_test.py | 283 ++ src/solutions/PE2/pe2_sgr_test.py | 131 + src/solutions/PE2/pe2_test.py | 112 + src/solutions/PE2/portfolio_runner.py | 249 ++ src/solutions/PE2/sgr_advantage_config.py | 339 ++ src/solutions/PE2/sgr_advantage_test.py | 286 ++ src/utils/load_dataset_anli.py | 53 + src/utils/load_dataset_banking77.py | 46 + src/utils/load_dataset_classification.py | 48 + src/utils/load_dataset_ifeval.py | 53 + src/utils/load_dataset_math.py | 41 + src/utils/load_dataset_pe2_paper.py | 115 + src/utils/load_dataset_russian.py | 66 + src/utils/load_dataset_summarization.py | 19 + src/utils/parallel_runner.py | 236 ++ test/coolprompt/evaluator/__init__.py | 0 test/coolprompt/evaluator/test_em_metric.py | 35 + .../evaluator/test_ifeval_checkers.py | 291 ++ test/coolprompt/evaluator/test_portfolio.py | 22 + test/utils/__init__.py | 0 test/utils/test_new_loaders.py | 39 + uv.lock | 3575 +++++++++++++++++ 57 files changed, 9777 insertions(+), 71 deletions(-) create mode 100644 coolprompt/evaluator/ifeval_checkers.py create mode 100644 coolprompt/evaluator/portfolio.py create mode 100644 coolprompt/optimizer/ape/__init__.py create mode 100644 coolprompt/optimizer/ape/proposer.py create mode 100644 coolprompt/optimizer/ape/run.py create mode 100644 coolprompt/optimizer/ape/trainer.py create mode 100644 coolprompt/optimizer/opro/__init__.py create mode 100644 coolprompt/optimizer/opro/proposer.py create mode 100644 coolprompt/optimizer/opro/run.py create mode 100644 coolprompt/optimizer/opro/trainer.py create mode 100644 coolprompt/optimizer/pe2/__init__.py create mode 100644 coolprompt/optimizer/pe2/node.py create mode 100644 coolprompt/optimizer/pe2/proposer.py create mode 100644 coolprompt/optimizer/pe2/run.py create mode 100644 coolprompt/optimizer/pe2/trainer.py create mode 100644 coolprompt/optimizer/pe2_sgr/__init__.py create mode 100644 coolprompt/optimizer/pe2_sgr/proposer.py create mode 100644 coolprompt/optimizer/pe2_sgr/run.py create mode 100644 coolprompt/optimizer/pe2_sgr/schemas.py create mode 100644 coolprompt/utils/prompt_templates/ape_templates.py create mode 100644 coolprompt/utils/prompt_templates/opro_templates.py create mode 100644 coolprompt/utils/prompt_templates/pe2_sgr_templates.py create mode 100644 coolprompt/utils/prompt_templates/pe2_templates.py create mode 100644 src/solutions/PE2/__init__.py create mode 100644 src/solutions/PE2/extra_benchmarks_config.py create mode 100644 src/solutions/PE2/extra_benchmarks_test.py create mode 100644 src/solutions/PE2/local_models.py create mode 100644 src/solutions/PE2/pe2_baseline_test.py create mode 100644 src/solutions/PE2/pe2_paper_config.py create mode 100644 src/solutions/PE2/pe2_paper_test.py create mode 100644 src/solutions/PE2/pe2_sgr_test.py create mode 100644 src/solutions/PE2/pe2_test.py create mode 100644 src/solutions/PE2/portfolio_runner.py create mode 100644 src/solutions/PE2/sgr_advantage_config.py create mode 100644 src/solutions/PE2/sgr_advantage_test.py create mode 100644 src/utils/load_dataset_anli.py create mode 100644 src/utils/load_dataset_banking77.py create mode 100644 src/utils/load_dataset_classification.py create mode 100644 src/utils/load_dataset_ifeval.py create mode 100644 src/utils/load_dataset_math.py create mode 100644 src/utils/load_dataset_pe2_paper.py create mode 100644 src/utils/load_dataset_russian.py create mode 100644 src/utils/load_dataset_summarization.py create mode 100644 src/utils/parallel_runner.py create mode 100644 test/coolprompt/evaluator/__init__.py create mode 100644 test/coolprompt/evaluator/test_em_metric.py create mode 100644 test/coolprompt/evaluator/test_ifeval_checkers.py create mode 100644 test/coolprompt/evaluator/test_portfolio.py create mode 100644 test/utils/__init__.py create mode 100644 test/utils/test_new_loaders.py create mode 100644 uv.lock diff --git a/.gitignore b/.gitignore index 9f33e022..c514748b 100644 --- a/.gitignore +++ b/.gitignore @@ -123,6 +123,7 @@ celerybeat.pid # Environments .env +.envrc .venv env/ venv/ @@ -175,7 +176,7 @@ src/solutions/SPELL/outputs/* src/solutions/evo/self_evo/data/* src/solutions/SPELL/data/* -# Let's fix some directory for run logs +.DS_Store +.worktrees/ run_logs/* - -coolprompt_test/* \ No newline at end of file +coolprompt_test/* diff --git a/coolprompt/evaluator/ifeval_checkers.py b/coolprompt/evaluator/ifeval_checkers.py new file mode 100644 index 00000000..9d1feb4e --- /dev/null +++ b/coolprompt/evaluator/ifeval_checkers.py @@ -0,0 +1,186 @@ +"""Verifiable checkers for a subset of IFEval instructions. + +Each checker returns True iff the model `response` satisfies the +instruction described by its `kwargs`. Implements a focused +subset of the IFEval (Zhou et al., 2023) instruction registry; +the loader filters the dataset to these ids so every evaluated +prompt is checkable. +""" + +import json +import re + + +def _word_count(text: str) -> int: + return len(re.findall(r"\b\w+\b", text)) + + +def _sentence_count(text: str) -> int: + parts = re.split(r"[.!?]+", text) + return len([p for p in parts if p.strip()]) + + +def _check_relation( + value: int, threshold: int, relation: str +) -> bool: + if relation == "at least": + return value >= threshold + if relation == "at most": + return value <= threshold + if relation == "less than": + return value < threshold + if relation == "exactly": + return value == threshold + return False + + +def _number_words(resp, kw): + return _check_relation( + _word_count(resp), + int(kw.get("num_words", 0)), + kw.get("relation", "at least"), + ) + + +def _number_sentences(resp, kw): + return _check_relation( + _sentence_count(resp), + int(kw.get("num_sentences", 0)), + kw.get("relation", "at least"), + ) + + +def _keyword_existence(resp, kw): + # Substring (not word-boundary) match is intentional, matching + # the IFEval reference impl; differs from _keyword_frequency. + low = resp.lower() + return all(k.lower() in low for k in kw.get("keywords", [])) + + +def _keyword_frequency(resp, kw): + word = str(kw.get("keyword", "")).lower() + count = len( + re.findall( + r"\b" + re.escape(word) + r"\b", resp.lower() + ) + ) + return _check_relation( + count, + int(kw.get("frequency", 0)), + kw.get("relation", "at least"), + ) + + +def _forbidden_words(resp, kw): + low = resp.lower() + return all( + re.search( + r"\b" + re.escape(w.lower()) + r"\b", low + ) is None + for w in kw.get("forbidden_words", []) + ) + + +def _all_lowercase(resp, kw): + letters = [c for c in resp if c.isalpha()] + return bool(letters) and all(c.islower() for c in letters) + + +def _all_uppercase(resp, kw): + letters = [c for c in resp if c.isalpha()] + return bool(letters) and all(c.isupper() for c in letters) + + +def _number_bullets(resp, kw): + bullets = re.findall( + r"^\s*[\*\-]\s+", resp, re.MULTILINE + ) + return len(bullets) == int(kw.get("num_bullets", 0)) + + +def _json_format(resp, kw): + text = resp.strip() + text = re.sub(r"^```(json)?|```$", "", text).strip() + try: + json.loads(text) + return True + except (ValueError, TypeError): + return False + + +def _number_highlighted(resp, kw): + highlights = re.findall( + r"(?= int( + kw.get("num_highlights", 0) + ) + + +def _title(resp, kw): + return re.search(r"<<[^>\n]+>>", resp) is not None + + +def _end_checker(resp, kw): + phrase = str(kw.get("end_phrase", "")).strip().lower() + return resp.strip().lower().endswith(phrase) + + +def _quotation(resp, kw): + t = resp.strip() + return len(t) >= 2 and t.startswith('"') and t.endswith('"') + + +def _postscript(resp, kw): + marker = str( + kw.get("postscript_marker", "P.S.") + ).lower() + return marker in resp.lower() + + +def _placeholders(resp, kw): + found = re.findall(r"\[[^\]\n]+\](?!\()", resp) + return len(found) >= int( + kw.get("num_placeholders", 0) + ) + + +# instruction id -> checker function +_CHECKERS = { + "length_constraints:number_words": _number_words, + "length_constraints:number_sentences": _number_sentences, + "keywords:existence": _keyword_existence, + "keywords:frequency": _keyword_frequency, + "keywords:forbidden_words": _forbidden_words, + "change_case:english_lowercase": _all_lowercase, + "change_case:english_capital": _all_uppercase, + "detectable_format:number_bullet_lists": _number_bullets, + "detectable_format:json_format": _json_format, + "detectable_format:number_highlighted_sections": ( + _number_highlighted + ), + "detectable_format:title": _title, + "startend:end_checker": _end_checker, + "startend:quotation": _quotation, + "detectable_content:postscript": _postscript, + "detectable_content:number_placeholders": _placeholders, +} + +SUPPORTED_INSTRUCTIONS = frozenset(_CHECKERS.keys()) + + +def check_instruction( + instruction_id: str, response: str, kwargs: dict +) -> bool: + """Return True iff `response` satisfies the instruction. + + Unknown instruction ids return False (treated as a failed + constraint), so unsupported rows must be filtered upstream. + """ + checker = _CHECKERS.get(instruction_id) + if checker is None: + return False + try: + return bool(checker(response, kwargs or {})) + except (ValueError, TypeError, re.error): + return False diff --git a/coolprompt/evaluator/metrics.py b/coolprompt/evaluator/metrics.py index 310291c3..b85fda0f 100644 --- a/coolprompt/evaluator/metrics.py +++ b/coolprompt/evaluator/metrics.py @@ -1,3 +1,5 @@ +import json +import re from abc import ABC, abstractmethod import re from typing import Optional, Dict, List, Tuple @@ -26,6 +28,7 @@ FLUENCY_TEMPLATE, RELEVANCE_TEMPLATE, ) +from coolprompt.evaluator.ifeval_checkers import check_instruction class HFEvaluateMetric(ABC): @@ -570,6 +573,82 @@ def parse_output(self, output: str) -> str: return extract_number_from_text(extracted) +class IFEvalMetric(GenerationMetric): + """IFEval prompt-level strict accuracy. + + `targets` are JSON specs: + {"instruction_id_list": [...], "kwargs": [{...}, ...]} + A prompt scores 1.0 iff every listed instruction is + satisfied by the raw model output; the metric returns the + mean over prompts. Unlike other generation metrics, this + scores the raw output (no tag extraction). + """ + + @staticmethod + def _get_name(): + return "ifeval" + + def __init__(self): + super().__init__() + + def compute(self, outputs, targets, dataset=None): + return self._compute_raw( + list(map(str, outputs)), + list(map(str, targets)), + dataset, + ) + + def _compute_raw(self, outputs, targets, dataset): + per_prompt = [] + for output, spec_json in zip(outputs, targets): + try: + spec = json.loads(spec_json) + except (ValueError, TypeError): + per_prompt.append(0.0) + continue + ids = spec.get("instruction_id_list", []) + kw_list = spec.get("kwargs", [{}] * len(ids)) + satisfied = all( + check_instruction(iid, output, kw or {}) + for iid, kw in zip(ids, kw_list) + ) + per_prompt.append( + 1.0 if satisfied and ids else 0.0 + ) + return float(mean(per_prompt)) if per_prompt else 0.0 + + def failure_breakdown(self, outputs, targets): + """Per-instruction failure summary for the diagnosis. + + Returns a string like + 'change_case:english_lowercase failed 3/5; ...' listing + only the constraints that failed at least once, or None + if every constraint passed / targets are unparseable. + """ + fails = {} + totals = {} + for output, spec_json in zip( + map(str, outputs), map(str, targets) + ): + try: + spec = json.loads(spec_json) + except (ValueError, TypeError): + continue + ids = spec.get("instruction_id_list", []) + kw_list = spec.get("kwargs", [{}] * len(ids)) + for iid, kw in zip(ids, kw_list): + totals[iid] = totals.get(iid, 0) + 1 + if not check_instruction(iid, output, kw or {}): + fails[iid] = fails.get(iid, 0) + 1 + if not fails: + return None + parts = [ + f"{iid} failed {fails[iid]}/{totals[iid]}" + for iid in sorted(fails, key=lambda k: -fails[k]) + ] + return "; ".join(parts) + + def define_lang(outputs, targets): langs = [detect_language(target) for target in targets] return max(set(langs), key=langs.count) diff --git a/coolprompt/evaluator/portfolio.py b/coolprompt/evaluator/portfolio.py new file mode 100644 index 00000000..213c326d --- /dev/null +++ b/coolprompt/evaluator/portfolio.py @@ -0,0 +1,18 @@ +"""Best-of selection across optimizer methods (portfolio).""" + + +def select_portfolio(results): + """Pick the method whose prompt scored highest on val. + + Args: + results: dict mapping method name -> (prompt, val_score). + Insertion order breaks ties (first listed wins). + + Returns: + (method, prompt, val_score) of the winner. + """ + best = None + for method, (prompt, score) in results.items(): + if best is None or score > best[2]: + best = (method, prompt, score) + return best diff --git a/coolprompt/optimizer/ape/__init__.py b/coolprompt/optimizer/ape/__init__.py new file mode 100644 index 00000000..d3d0cda2 --- /dev/null +++ b/coolprompt/optimizer/ape/__init__.py @@ -0,0 +1,6 @@ +from coolprompt.optimizer.ape.run import ape_optimizer, APEMethod + +__all__ = [ + 'ape_optimizer', + 'APEMethod', +] diff --git a/coolprompt/optimizer/ape/proposer.py b/coolprompt/optimizer/ape/proposer.py new file mode 100644 index 00000000..d6f78b5b --- /dev/null +++ b/coolprompt/optimizer/ape/proposer.py @@ -0,0 +1,66 @@ +"""APE proposer: generates prompt variations via paraphrasing.""" + +from langchain_core.language_models.base import BaseLanguageModel + +from coolprompt.optimizer.pe2.node import Node +from coolprompt.utils.parsing import ( + extract_answer, + get_model_answer_extracted, +) +from coolprompt.utils.prompt_templates.ape_templates import ( + APE_PARAPHRASE_TEMPLATE, +) + + +class APEProposer: + """Generates prompt variations by paraphrasing. + + Unlike PE2's proposer, APE ignores failure examples + and simply paraphrases the current prompt. + + Args: + model (BaseLanguageModel): LLM for paraphrasing. + prompt_max_tokens (int): Max token budget hint + for the new prompt. + """ + + def __init__( + self, + model: BaseLanguageModel, + prompt_max_tokens: int = 300, + ) -> None: + self.model = model + self.prompt_max_tokens = prompt_max_tokens + + def propose( + self, + node: Node, + examples_str: str, + full_template: str, + batch_size: int, + ) -> tuple[str, str]: + """Proposes a paraphrased prompt variation. + + Args: + node (Node): Current beam node whose prompt + is being varied. + examples_str (str): Ignored by APE. + full_template (str): Ignored by APE. + batch_size (int): Ignored by APE. + + Returns: + tuple[str, str]: (new_prompt, "paraphrase"). + """ + prompt = APE_PARAPHRASE_TEMPLATE.format( + prompt=node.prompt, + max_tokens=self.prompt_max_tokens, + ) + answer = get_model_answer_extracted(self.model, prompt) + + new_prompt = extract_answer( + answer, + ("", ""), + format_mismatch_label=node.prompt, + ) + + return new_prompt.strip(), "paraphrase" diff --git a/coolprompt/optimizer/ape/run.py b/coolprompt/optimizer/ape/run.py new file mode 100644 index 00000000..0c0e05f5 --- /dev/null +++ b/coolprompt/optimizer/ape/run.py @@ -0,0 +1,101 @@ +"""High-level entry point for APE optimization.""" + +from typing import List, Tuple + +from langchain_core.language_models.base import BaseLanguageModel + +from coolprompt.evaluator import Evaluator +from coolprompt.optimizer.autoprompting_method import ( + AutoPromptingMethod, +) +from coolprompt.optimizer.ape.proposer import APEProposer +from coolprompt.optimizer.ape.trainer import APETrainer +from coolprompt.utils.logging_config import logger + + +def ape_optimizer( + model: BaseLanguageModel, + dataset_split: Tuple[ + List[str], List[str], List[str], List[str] + ], + evaluator: Evaluator, + initial_prompt: str, + **kwargs, +) -> str: + """Runs APE evaluate-select-paraphrase optimization. + + Uses paraphrase-only proposals (no failure analysis). + Evaluates on validation data, selects top-k, paraphrases, + and repeats. + + Args: + model (BaseLanguageModel): The language model to use. + dataset_split: A tuple of (train_dataset, + val_dataset, train_targets, val_targets). + evaluator (Evaluator): Evaluator for scoring prompts. + initial_prompt (str): The starting prompt to optimize. + **kwargs: Optional overrides for train_steps, n_beam, + n_expand, backtrack, prompt_max_tokens. + + Returns: + str: The best prompt found after optimization. + """ + ( + train_dataset, + val_dataset, + train_targets, + val_targets, + ) = dataset_split + + args = { + "train_steps": 3, + "n_beam": 3, + "n_expand": 4, + "prompt_max_tokens": 300, + } + args.update(kwargs) + + proposer = APEProposer( + model=model, + prompt_max_tokens=args["prompt_max_tokens"], + ) + + template = evaluator._get_default_template() + + trainer = APETrainer( + model=model, + evaluator=evaluator, + proposer=proposer, + val_dataset=val_dataset, + val_targets=val_targets, + template=template, + train_steps=args["train_steps"], + n_beam=args["n_beam"], + n_expand=args["n_expand"], + ) + + logger.info("Starting APE optimization...") + logger.debug(f"Start prompt:\n{initial_prompt}") + final_prompt = trainer.train(initial_prompt) + logger.info("APE optimization completed") + return final_prompt + + +class APEMethod(AutoPromptingMethod): + + def optimize( + self, model, initial_prompt, + dataset_split=None, evaluator=None, + problem_description=None, **kwargs, + ) -> str: + return ape_optimizer( + model, dataset_split, evaluator, + initial_prompt, **kwargs, + ) + + def is_data_driven(self): + return True + + @property + def name(self): + return "ape" diff --git a/coolprompt/optimizer/ape/trainer.py b/coolprompt/optimizer/ape/trainer.py new file mode 100644 index 00000000..283daeb0 --- /dev/null +++ b/coolprompt/optimizer/ape/trainer.py @@ -0,0 +1,185 @@ +"""APE trainer: evaluate-select-paraphrase loop.""" + +from concurrent.futures import ThreadPoolExecutor, as_completed +from typing import List + +from langchain_core.language_models.base import BaseLanguageModel + +from coolprompt.evaluator import Evaluator +from coolprompt.optimizer.ape.proposer import APEProposer +from coolprompt.optimizer.pe2.node import Node +from coolprompt.utils.logging_config import logger + + +class APETrainer: + """APE trainer with its own evaluate-select-paraphrase loop. + + Matches the original APE paper: evaluates candidates on + validation data, selects top-k, paraphrases each to + generate new candidates, and repeats. + + No failure mining or training data needed — the proposer + simply paraphrases the current prompt. + + Args: + model (BaseLanguageModel): LLM used for inference. + evaluator (Evaluator): Evaluator for scoring prompts. + proposer (APEProposer): Proposer for paraphrasing. + val_dataset (List[str]): Validation input samples. + val_targets (List[str]): Validation ground-truth + targets. + template (str): Prompt template for the task. + train_steps (int): Number of optimization iterations. + n_beam (int): Top-k candidates to keep per step. + n_expand (int): Paraphrases per selected candidate. + """ + + def __init__( + self, + model: BaseLanguageModel, + evaluator: Evaluator, + proposer: APEProposer, + val_dataset: List[str], + val_targets: List[str], + template: str, + train_steps: int = 3, + n_beam: int = 3, + n_expand: int = 4, + ) -> None: + self.model = model + self.evaluator = evaluator + self.proposer = proposer + self.val_dataset = val_dataset + self.val_targets = val_targets + self.template = template + self.train_steps = train_steps + self.n_beam = n_beam + self.n_expand = n_expand + self._node_counter = 0 + + def _next_id(self) -> int: + """Returns the next unique node ID.""" + nid = self._node_counter + self._node_counter += 1 + return nid + + def train(self, initial_prompt: str) -> str: + """Runs the APE evaluate-select-paraphrase loop. + + Args: + initial_prompt (str): The starting prompt. + + Returns: + str: The best prompt found during optimization. + """ + initial_node = Node( + timestamp=0, + id=self._next_id(), + prompt=initial_prompt, + ) + all_nodes: List[Node] = [initial_node] + unevaluated: List[Node] = [initial_node] + + for t in range(self.train_steps): + logger.info( + f"APE step {t + 1}/{self.train_steps}" + ) + + for node in unevaluated: + score = self.evaluator.evaluate( + prompt=node.prompt, + dataset=self.val_dataset, + targets=self.val_targets, + template=self.template, + ) + node.register_score(score, "val") + logger.info( + f" Node {node.id} val score: " + f"{score:.4f}" + ) + + scored = [ + n for n in all_nodes + if "val" in n.scores + ] + scored.sort( + key=lambda n: n.scores["val"], reverse=True + ) + selected = scored[: self.n_beam] + best_score = selected[0].scores["val"] + logger.info( + f" Best val score at step {t + 1}: " + f"{best_score:.4f}" + ) + + if t == self.train_steps - 1: + break + + proposal_jobs = [ + node + for node in selected + for _ in range(self.n_expand) + ] + + def _do_propose(n): + prompt, _ = self.proposer.propose( + node=n, + examples_str="", + full_template=self.template, + batch_size=0, + ) + return n, prompt + + new_nodes: List[Node] = [] + with ThreadPoolExecutor( + max_workers=min(len(proposal_jobs), 12) + ) as pool: + futures = [ + pool.submit(_do_propose, j) + for j in proposal_jobs + ] + for fut in as_completed(futures): + parent, new_prompt = fut.result() + child = Node( + timestamp=t + 1, + id=self._next_id(), + prompt=new_prompt, + parent=parent.id, + ) + parent.n_child += 1 + new_nodes.append(child) + + existing = { + n.prompt.strip().lower() for n in all_nodes + } + unique = [] + for node in new_nodes: + key = node.prompt.strip().lower() + if key not in existing: + existing.add(key) + unique.append(node) + + if not unique: + logger.info( + " No new unique nodes generated, " + "stopping" + ) + break + + all_nodes.extend(unique) + unevaluated = unique + logger.info( + f" Generated {len(unique)} new candidates" + ) + + scored_all = [ + n for n in all_nodes if "val" in n.scores + ] + best = max( + scored_all, key=lambda n: n.scores["val"] + ) + logger.info( + f"APE best prompt (node {best.id}, " + f"val={best.scores['val']:.4f})" + ) + return best.prompt diff --git a/coolprompt/optimizer/hyper/__init__.py b/coolprompt/optimizer/hyper/__init__.py index 7091d4e3..c9cc3497 100644 --- a/coolprompt/optimizer/hyper/__init__.py +++ b/coolprompt/optimizer/hyper/__init__.py @@ -4,7 +4,6 @@ MetaPromptOptimizer, Optimizer, ) - __all__ = [ "Optimizer", "MetaPromptOptimizer", diff --git a/coolprompt/optimizer/opro/__init__.py b/coolprompt/optimizer/opro/__init__.py new file mode 100644 index 00000000..b8becbb0 --- /dev/null +++ b/coolprompt/optimizer/opro/__init__.py @@ -0,0 +1,6 @@ +from coolprompt.optimizer.opro.run import opro_optimizer, OPROMethod + +__all__ = [ + 'opro_optimizer', + 'OPROMethod', +] diff --git a/coolprompt/optimizer/opro/proposer.py b/coolprompt/optimizer/opro/proposer.py new file mode 100644 index 00000000..75992b59 --- /dev/null +++ b/coolprompt/optimizer/opro/proposer.py @@ -0,0 +1,139 @@ +"""OPRO proposer: trajectory-based prompt optimization.""" + +import random +from typing import List, Optional + +from langchain_core.language_models.base import BaseLanguageModel + +from coolprompt.utils.parsing import ( + extract_answer, + get_model_answer_extracted, +) +from coolprompt.utils.prompt_templates.opro_templates import ( + OPRO_META_TEMPLATE, +) + + +class OPROProposer: + """Generates prompts based on a trajectory of past attempts. + + Maintains a sorted history of (prompt, score) pairs and + asks the LLM to propose a prompt that scores higher than + all previous attempts. Includes task demonstrations from + training data in the meta-prompt, matching the original + OPRO paper. + + Args: + model (BaseLanguageModel): LLM for meta-optimization. + train_dataset (List[str]): Training input samples. + train_targets (List[str]): Training ground-truth + targets. + prompt_max_tokens (int): Max token budget hint + for the new prompt. + max_trajectory (int): Maximum number of past attempts + to keep in the trajectory. + n_demonstrations (int): Number of task demonstrations + to include in the meta-prompt. + """ + + def __init__( + self, + model: BaseLanguageModel, + train_dataset: List[str], + train_targets: List[str], + prompt_max_tokens: int = 300, + max_trajectory: int = 20, + n_demonstrations: int = 5, + ) -> None: + self.model = model + self.train_dataset = train_dataset + self.train_targets = train_targets + self.prompt_max_tokens = prompt_max_tokens + self.max_trajectory = max_trajectory + self.n_demonstrations = n_demonstrations + self.trajectory: list[tuple[str, float]] = [] + + def update_trajectory( + self, prompt: str, score: float + ) -> None: + """Adds a (prompt, score) pair to the trajectory. + + Keeps only the top max_trajectory entries by score. + + Args: + prompt (str): The prompt text. + score (float): The evaluation score. + """ + self.trajectory.append((prompt, score)) + self.trajectory.sort(key=lambda x: x[1]) + if len(self.trajectory) > self.max_trajectory: + self.trajectory = self.trajectory[ + -self.max_trajectory: + ] + + def _format_trajectory(self) -> str: + """Formats trajectory as a numbered list, worst to best. + + Returns: + str: Formatted trajectory string. + """ + parts = [] + for i, (prompt, score) in enumerate( + self.trajectory, 1 + ): + parts.append( + f"{i}. Score: {score:.4f}\n" + f" Instruction: {prompt}" + ) + return "\n\n".join(parts) + + def _format_task_demonstrations(self) -> str: + """Samples and formats task input-output pairs. + + Returns: + str: Formatted task demonstrations string. + """ + n = min( + self.n_demonstrations, len(self.train_dataset) + ) + indices = random.sample( + range(len(self.train_dataset)), n + ) + parts = [] + for i, idx in enumerate(indices, 1): + parts.append( + f"{i}. Input: {self.train_dataset[idx]}\n" + f" Output: {self.train_targets[idx]}" + ) + return "\n\n".join(parts) + + def propose(self) -> tuple[str, str]: + """Proposes a prompt based on the trajectory. + + Returns: + tuple[str, str]: (new_prompt, "trajectory"). + """ + trajectory_str = self._format_trajectory() + demos_str = self._format_task_demonstrations() + + prompt = OPRO_META_TEMPLATE.format( + task_demonstrations=demos_str, + trajectory=trajectory_str, + max_tokens=self.prompt_max_tokens, + ) + answer = get_model_answer_extracted( + self.model, prompt + ) + + fallback = ( + self.trajectory[-1][0] + if self.trajectory + else "" + ) + new_prompt = extract_answer( + answer, + ("", ""), + format_mismatch_label=fallback, + ) + + return new_prompt.strip(), "trajectory" diff --git a/coolprompt/optimizer/opro/run.py b/coolprompt/optimizer/opro/run.py new file mode 100644 index 00000000..42053964 --- /dev/null +++ b/coolprompt/optimizer/opro/run.py @@ -0,0 +1,108 @@ +"""High-level entry point for OPRO optimization.""" + +from typing import List, Tuple + +from langchain_core.language_models.base import BaseLanguageModel + +from coolprompt.evaluator import Evaluator +from coolprompt.optimizer.autoprompting_method import ( + AutoPromptingMethod, +) +from coolprompt.optimizer.opro.proposer import OPROProposer +from coolprompt.optimizer.opro.trainer import OPROTrainer +from coolprompt.utils.logging_config import logger + + +def opro_optimizer( + model: BaseLanguageModel, + dataset_split: Tuple[ + List[str], List[str], List[str], List[str] + ], + evaluator: Evaluator, + initial_prompt: str, + **kwargs, +) -> str: + """Runs OPRO trajectory-based prompt optimization. + + Uses a meta-prompt showing past (prompt, score) pairs + sorted worst-to-best plus task demonstrations, asking + the LLM to propose a higher-scoring prompt. + + Args: + model (BaseLanguageModel): The language model to use. + dataset_split: A tuple of (train_dataset, + val_dataset, train_targets, val_targets). + evaluator (Evaluator): Evaluator for scoring prompts. + initial_prompt (str): The starting prompt to optimize. + **kwargs: Optional overrides for train_steps, + n_candidates, prompt_max_tokens, + max_trajectory, n_demonstrations. + + Returns: + str: The best prompt found after optimization. + """ + ( + train_dataset, + val_dataset, + train_targets, + val_targets, + ) = dataset_split + + args = { + "train_steps": 3, + "n_candidates": 8, + "prompt_max_tokens": 300, + "max_trajectory": 20, + "n_demonstrations": 5, + } + args.update(kwargs) + + proposer = OPROProposer( + model=model, + train_dataset=train_dataset, + train_targets=train_targets, + prompt_max_tokens=args["prompt_max_tokens"], + max_trajectory=args["max_trajectory"], + n_demonstrations=args["n_demonstrations"], + ) + + template = evaluator._get_default_template() + + trainer = OPROTrainer( + model=model, + evaluator=evaluator, + proposer=proposer, + train_dataset=train_dataset, + train_targets=train_targets, + val_dataset=val_dataset, + val_targets=val_targets, + template=template, + train_steps=args["train_steps"], + n_candidates=args["n_candidates"], + ) + + logger.info("Starting OPRO optimization...") + logger.debug(f"Start prompt:\n{initial_prompt}") + final_prompt = trainer.train(initial_prompt) + logger.info("OPRO optimization completed") + return final_prompt + + +class OPROMethod(AutoPromptingMethod): + + def optimize( + self, model, initial_prompt, + dataset_split=None, evaluator=None, + problem_description=None, **kwargs, + ) -> str: + return opro_optimizer( + model, dataset_split, evaluator, + initial_prompt, **kwargs, + ) + + def is_data_driven(self): + return True + + @property + def name(self): + return "opro" diff --git a/coolprompt/optimizer/opro/trainer.py b/coolprompt/optimizer/opro/trainer.py new file mode 100644 index 00000000..9e192d64 --- /dev/null +++ b/coolprompt/optimizer/opro/trainer.py @@ -0,0 +1,169 @@ +"""OPRO trainer: flat iterative loop for prompt optimization.""" + +from concurrent.futures import ThreadPoolExecutor, as_completed +from typing import List + +from langchain_core.language_models.base import BaseLanguageModel + +from coolprompt.evaluator import Evaluator +from coolprompt.optimizer.opro.proposer import OPROProposer +from coolprompt.utils.logging_config import logger + + +class OPROTrainer: + """Standalone OPRO trainer using a flat iterative loop. + + Matches the original OPRO paper: maintains a trajectory + of (prompt, score) pairs, proposes new candidates each + step, evaluates on training data, and returns the best. + + Args: + model (BaseLanguageModel): LLM used for inference. + evaluator (Evaluator): Evaluator for scoring prompts. + proposer (OPROProposer): Proposer for generating + new prompts from the trajectory. + train_dataset (List[str]): Training input samples. + train_targets (List[str]): Training ground-truth + targets. + val_dataset (List[str]): Validation input samples. + val_targets (List[str]): Validation ground-truth + targets. + template (str): Prompt template for the task. + train_steps (int): Number of optimization iterations. + n_candidates (int): Candidates generated per step. + """ + + def __init__( + self, + model: BaseLanguageModel, + evaluator: Evaluator, + proposer: OPROProposer, + train_dataset: List[str], + train_targets: List[str], + val_dataset: List[str], + val_targets: List[str], + template: str, + train_steps: int = 3, + n_candidates: int = 8, + ) -> None: + self.model = model + self.evaluator = evaluator + self.proposer = proposer + self.train_dataset = train_dataset + self.train_targets = train_targets + self.val_dataset = val_dataset + self.val_targets = val_targets + self.template = template + self.train_steps = train_steps + self.n_candidates = n_candidates + + def _evaluate_on_train(self, prompt: str) -> float: + """Evaluates a prompt on training data. + + Args: + prompt (str): The prompt to evaluate. + + Returns: + float: The evaluation score. + """ + return self.evaluator.evaluate( + prompt=prompt, + dataset=self.train_dataset, + targets=self.train_targets, + template=self.template, + ) + + def _evaluate_on_val(self, prompt: str) -> float: + """Evaluates a prompt on validation data. + + Args: + prompt (str): The prompt to evaluate. + + Returns: + float: The evaluation score. + """ + return self.evaluator.evaluate( + prompt=prompt, + dataset=self.val_dataset, + targets=self.val_targets, + template=self.template, + ) + + def train(self, initial_prompt: str) -> str: + """Runs the OPRO iterative optimization. + + Args: + initial_prompt (str): The starting prompt. + + Returns: + str: The best prompt found during optimization. + """ + score = self._evaluate_on_train(initial_prompt) + self.proposer.update_trajectory(initial_prompt, score) + logger.info( + f"OPRO initial train score: {score:.4f}" + ) + + seen_prompts = {initial_prompt.strip().lower()} + best_prompt = initial_prompt + best_val_score = -1.0 + + for t in range(self.train_steps): + logger.info( + f"OPRO step {t + 1}/{self.train_steps}" + ) + + def _do_propose(_): + prompt, _ = self.proposer.propose() + return prompt + + new_prompts = [] + with ThreadPoolExecutor( + max_workers=min(self.n_candidates, 12) + ) as pool: + futures = [ + pool.submit(_do_propose, i) + for i in range(self.n_candidates) + ] + for fut in as_completed(futures): + new_prompts.append(fut.result()) + + unique_prompts = [] + for p in new_prompts: + key = p.strip().lower() + if key not in seen_prompts: + seen_prompts.add(key) + unique_prompts.append(p) + + if not unique_prompts: + logger.info( + " No new unique prompts, stopping" + ) + break + + for p in unique_prompts: + s = self._evaluate_on_train(p) + self.proposer.update_trajectory(p, s) + logger.info( + f" Candidate train score: {s:.4f}" + ) + + logger.info( + f" Added {len(unique_prompts)} new " + f"prompts to trajectory" + ) + + all_prompts = [ + p for p, _ in self.proposer.trajectory + ] + for p in all_prompts: + val_score = self._evaluate_on_val(p) + if val_score > best_val_score: + best_val_score = val_score + best_prompt = p + + logger.info( + f"OPRO best prompt " + f"(val={best_val_score:.4f})" + ) + return best_prompt diff --git a/coolprompt/optimizer/pe2/__init__.py b/coolprompt/optimizer/pe2/__init__.py new file mode 100644 index 00000000..5fa23d4e --- /dev/null +++ b/coolprompt/optimizer/pe2/__init__.py @@ -0,0 +1,6 @@ +from coolprompt.optimizer.pe2.run import pe2_optimizer, PE2Method + +__all__ = [ + 'pe2_optimizer', + 'PE2Method', +] diff --git a/coolprompt/optimizer/pe2/node.py b/coolprompt/optimizer/pe2/node.py new file mode 100644 index 00000000..f5fc0ea7 --- /dev/null +++ b/coolprompt/optimizer/pe2/node.py @@ -0,0 +1,34 @@ +"""Lightweight node for PE2 beam search.""" + +from dataclasses import dataclass, field +from typing import Optional + + +@dataclass +class Node: + """A node in the PE2 beam search tree. + + Args: + timestamp (int): The training step at which this node was created. + id (int): Unique identifier for this node. + prompt (str): The prompt text held by this node. + parent (Optional[int]): ID of the parent node, or None for root. + scores (dict): Mapping of split name to evaluation score. + n_child (int): Number of child nodes spawned from this node. + """ + + timestamp: int + id: int + prompt: str + parent: Optional[int] = None + scores: dict = field(default_factory=dict) + n_child: int = 0 + + def register_score(self, score: float, split_name: str) -> None: + """Records an evaluation score for a given split. + + Args: + score (float): The metric score. + split_name (str): Name of the data split (e.g. "val"). + """ + self.scores[split_name] = score diff --git a/coolprompt/optimizer/pe2/proposer.py b/coolprompt/optimizer/pe2/proposer.py new file mode 100644 index 00000000..f0ef4088 --- /dev/null +++ b/coolprompt/optimizer/pe2/proposer.py @@ -0,0 +1,76 @@ +"""PE2 proposer: generates refined prompts from failure analysis.""" + +from langchain_core.language_models.base import BaseLanguageModel + +from coolprompt.optimizer.pe2.node import Node +from coolprompt.utils.parsing import ( + extract_answer, + get_model_answer_extracted, +) +from coolprompt.utils.prompt_templates.pe2_templates import ( + PE2_REASONING_TEMPLATE, + PE2_REFINEMENT_TEMPLATE, +) + + +class Proposer: + """Wraps LLM calls for PE2 prompt refinement. + + Args: + model (BaseLanguageModel): LLM for reasoning and refinement. + prompt_max_tokens (int): Max token budget hint for the new prompt. + """ + + def __init__( + self, + model: BaseLanguageModel, + prompt_max_tokens: int = 300, + ) -> None: + self.model = model + self.prompt_max_tokens = prompt_max_tokens + + def propose( + self, + node: Node, + examples_str: str, + full_template: str, + batch_size: int, + best_val_score: float = 0.0, + constraint_feedback=None, + ) -> tuple[str, str]: + """Proposes a refined prompt by analyzing failures. + + Args: + node (Node): Current beam node whose prompt is being refined. + examples_str (str): Formatted failure examples string. + full_template (str): The full prompt template in use. + batch_size (int): Number of failure examples shown. + + Returns: + tuple[str, str]: (new_prompt, reasoning) where new_prompt is + the refined prompt text and reasoning is the analysis. + """ + reasoning_prompt = PE2_REASONING_TEMPLATE.format( + batch_size=batch_size, + prompt=node.prompt, + full_template=full_template, + examples=examples_str, + ) + reasoning = get_model_answer_extracted(self.model, reasoning_prompt) + + refinement_prompt = PE2_REFINEMENT_TEMPLATE.format( + reasoning=reasoning, + prompt=node.prompt, + max_tokens=self.prompt_max_tokens, + ) + refinement_answer = get_model_answer_extracted( + self.model, refinement_prompt + ) + + new_prompt = extract_answer( + refinement_answer, + ("", ""), + format_mismatch_label=node.prompt, + ) + + return new_prompt.strip(), reasoning diff --git a/coolprompt/optimizer/pe2/run.py b/coolprompt/optimizer/pe2/run.py new file mode 100644 index 00000000..9d7c797f --- /dev/null +++ b/coolprompt/optimizer/pe2/run.py @@ -0,0 +1,102 @@ +"""High-level entry point for the PE2 optimization process.""" + +from typing import List, Tuple + +from langchain_core.language_models.base import BaseLanguageModel + +from coolprompt.evaluator import Evaluator +from coolprompt.optimizer.autoprompting_method import ( + AutoPromptingMethod, +) +from coolprompt.optimizer.pe2.proposer import Proposer +from coolprompt.optimizer.pe2.trainer import PE2Trainer +from coolprompt.utils.logging_config import logger + + +def pe2_optimizer( + model: BaseLanguageModel, + dataset_split: Tuple[List[str], List[str], List[str], List[str]], + evaluator: Evaluator, + initial_prompt: str, + **kwargs, +) -> str: + """Runs PE2 beam-search prompt optimization. + + Args: + model (BaseLanguageModel): The language model to use. + dataset_split (Tuple[List[str], List[str], List[str], List[str]]): + A tuple of (train_dataset, val_dataset, + train_targets, val_targets). + evaluator (Evaluator): Evaluator for scoring prompts. + initial_prompt (str): The starting prompt to optimize. + **kwargs: Optional overrides for train_steps, n_beam, n_expand, + batch_size, backtrack, prompt_max_tokens. + + Returns: + str: The best prompt found after optimization. + """ + ( + train_dataset, + val_dataset, + train_targets, + val_targets, + ) = dataset_split + + args = { + "train_steps": 3, + "n_beam": 3, + "n_expand": 4, + "batch_size": 4, + "backtrack": True, + "prompt_max_tokens": 300, + } + args.update(kwargs) + + proposer = Proposer( + model=model, + prompt_max_tokens=args["prompt_max_tokens"], + ) + + template = evaluator._get_default_template() + + trainer = PE2Trainer( + model=model, + evaluator=evaluator, + proposer=proposer, + train_dataset=train_dataset, + train_targets=train_targets, + val_dataset=val_dataset, + val_targets=val_targets, + template=template, + train_steps=args["train_steps"], + n_beam=args["n_beam"], + n_expand=args["n_expand"], + batch_size=args["batch_size"], + backtrack=args["backtrack"], + ) + + logger.info("Starting PE2 optimization...") + logger.debug(f"Start prompt:\n{initial_prompt}") + final_prompt = trainer.train(initial_prompt) + logger.info("PE2 optimization completed") + return final_prompt + + +class PE2Method(AutoPromptingMethod): + + def optimize( + self, model, initial_prompt, + dataset_split=None, evaluator=None, + problem_description=None, **kwargs, + ) -> str: + return pe2_optimizer( + model, dataset_split, evaluator, + initial_prompt, **kwargs, + ) + + def is_data_driven(self): + return True + + @property + def name(self): + return "pe2" diff --git a/coolprompt/optimizer/pe2/trainer.py b/coolprompt/optimizer/pe2/trainer.py new file mode 100644 index 00000000..56d52e4f --- /dev/null +++ b/coolprompt/optimizer/pe2/trainer.py @@ -0,0 +1,415 @@ +"""PE2 trainer: beam-search loop for prompt optimization.""" + +import random +from concurrent.futures import ThreadPoolExecutor, as_completed +from typing import List + +from langchain_core.language_models.base import BaseLanguageModel +from langchain_core.messages.ai import AIMessage + +from coolprompt.evaluator import Evaluator +from coolprompt.evaluator.metrics import ClassificationMetric +from coolprompt.optimizer.pe2.node import Node +from coolprompt.optimizer.pe2.proposer import Proposer +from coolprompt.utils.enums import Task +from coolprompt.utils.logging_config import logger +from coolprompt.utils.parsing import extract_answer + + +class PE2Trainer: + """Core beam-search trainer for PE2 prompt optimization. + + Args: + model (BaseLanguageModel): LLM used for inference. + evaluator (Evaluator): Evaluator for scoring prompts. + proposer (Proposer): Proposer for generating refined prompts. + train_dataset (List[str]): Training input samples. + train_targets (List[str]): Training ground-truth targets. + val_dataset (List[str]): Validation input samples. + val_targets (List[str]): Validation ground-truth targets. + template (str): Prompt template for the task. + train_steps (int): Number of beam-search iterations. + n_beam (int): Beam width (top-k nodes kept per step). + n_expand (int): Number of children per selected node. + batch_size (int): Number of failure examples shown to proposer. + backtrack (bool): If True, select best across all timesteps. + """ + + ANS_TAGS = ("", "") + + def __init__( + self, + model: BaseLanguageModel, + evaluator: Evaluator, + proposer: Proposer, + train_dataset: List[str], + train_targets: List[str], + val_dataset: List[str], + val_targets: List[str], + template: str, + train_steps: int = 3, + n_beam: int = 3, + n_expand: int = 4, + batch_size: int = 4, + backtrack: bool = True, + feedback_mode: str = "auto", + ) -> None: + self.model = model + self.evaluator = evaluator + self.proposer = proposer + self.train_dataset = train_dataset + self.train_targets = train_targets + self.val_dataset = val_dataset + self.val_targets = val_targets + self.template = template + self.train_steps = train_steps + self.n_beam = n_beam + self.n_expand = n_expand + self.batch_size = batch_size + self.backtrack = backtrack + self.feedback_mode = feedback_mode + self._node_counter = 0 + + def _next_id(self) -> int: + """Returns the next unique node ID.""" + nid = self._node_counter + self._node_counter += 1 + return nid + + def train(self, initial_prompt: str) -> str: + """Runs the PE2 beam-search optimization. + + Args: + initial_prompt (str): The starting prompt to optimize. + + Returns: + str: The best prompt found during optimization. + """ + initial_node = Node( + timestamp=0, + id=self._next_id(), + prompt=initial_prompt, + ) + states: List[List[Node]] = [[initial_node]] + + for t in range(self.train_steps): + logger.info(f"PE2 step {t + 1}/{self.train_steps}") + + for node in states[-1]: + if "val" not in node.scores: + score = self.evaluator.evaluate( + prompt=node.prompt, + dataset=self.val_dataset, + targets=self.val_targets, + template=self.template, + ) + node.register_score(score, "val") + logger.info( + f" Node {node.id} val score: {score:.4f}" + ) + + candidates = self._select_candidates(states) + best_score = max(n.scores["val"] for n in candidates) + logger.info( + f" Best val score at step {t + 1}: {best_score:.4f}" + ) + + if t == self.train_steps - 1: + break + + new_nodes: List[Node] = [] + proposal_jobs = [] + for node in candidates: + results = self._get_per_example_results( + node.prompt + ) + failures = [ + r for r in results if not r["correct"] + ] + + if not failures: + logger.info( + f" Node {node.id}: no failures " + "on train set, skipping expansion" + ) + continue + + cfb = None + if self.feedback_mode != "off": + metric = getattr( + self.evaluator, "metric", None + ) + if metric is not None and hasattr( + metric, "failure_breakdown" + ): + cfb = metric.failure_breakdown( + [ + f["raw_output"] + for f in failures + ], + [f["target"] for f in failures], + ) + + full_template = self._instantiate_template( + node.prompt + ) + + for _ in range(self.n_expand): + sampled = self._sample_failures( + failures, self.batch_size + ) + examples_str = self._pack_examples( + sampled + ) + proposal_jobs.append( + (node, examples_str, + full_template, len(sampled), + best_score, cfb) + ) + + if not proposal_jobs: + logger.info( + " No proposal jobs, stopping" + ) + break + + def _do_propose(job): + n, ex_str, ft, bs, bvs, cf = job + prompt, _ = self.proposer.propose( + node=n, + examples_str=ex_str, + full_template=ft, + batch_size=bs, + best_val_score=bvs, + constraint_feedback=cf, + ) + return n, prompt + + with ThreadPoolExecutor( + max_workers=min(len(proposal_jobs), 12) + ) as pool: + futures = [ + pool.submit(_do_propose, j) + for j in proposal_jobs + ] + for fut in as_completed(futures): + node, new_prompt = fut.result() + child = Node( + timestamp=t + 1, + id=self._next_id(), + prompt=new_prompt, + parent=node.id, + ) + node.n_child += 1 + new_nodes.append(child) + + all_existing = [n for state in states for n in state] + new_nodes = self._deduplicate(new_nodes, all_existing) + + if not new_nodes: + logger.info(" No new unique nodes generated, stopping") + break + + states.append(new_nodes) + logger.info(f" Generated {len(new_nodes)} new nodes") + + all_nodes = [n for state in states for n in state] + scored = [n for n in all_nodes if "val" in n.scores] + best = max(scored, key=lambda n: n.scores["val"]) + logger.info( + f"PE2 best prompt (node {best.id}, " + f"val={best.scores['val']:.4f})" + ) + return best.prompt + + def _select_candidates( + self, states: List[List[Node]] + ) -> List[Node]: + """Selects top n_beam nodes by validation score. + + Args: + states (List[List[Node]]): All beam states so far. + + Returns: + List[Node]: Top-k nodes by val score. + """ + if self.backtrack: + pool = [n for state in states for n in state] + else: + pool = list(states[-1]) + + scored = [n for n in pool if "val" in n.scores] + scored.sort(key=lambda n: n.scores["val"], reverse=True) + return scored[: self.n_beam] + + def _instantiate_template(self, prompt: str) -> str: + """Instantiates the task template with the prompt. + + Replaces the prompt placeholder with the actual prompt + and the input placeholder with a literal ````, + matching the official PE2 behavior where the proposer + sees the concrete full prompt structure. + + Args: + prompt (str): The instruction prompt. + + Returns: + str: The instantiated full template string. + """ + safe_prompt = prompt.replace("{", "{{").replace( + "}", "}}" + ) + if self.evaluator.task == Task.CLASSIFICATION: + if isinstance( + self.evaluator.metric, ClassificationMetric + ) and self.evaluator.metric.label_to_id: + labels = ", ".join( + map( + str, + self.evaluator.metric.label_to_id + .keys(), + ) + ) + else: + labels = "" + return self.template.format( + PROMPT=safe_prompt, + LABELS=labels, + INPUT="", + ) + else: + return self.template.format( + PROMPT=safe_prompt, + INPUT="", + ) + + def _get_per_example_results( + self, prompt: str + ) -> List[dict]: + """Runs the model on training data and checks + correctness via exact match. + + Uses exact string comparison on extracted answers + for failure detection, matching official PE2 behavior + (``score == 0.0`` where score is EM). The actual + evaluation metric is only used for overall prompt + scoring on the validation set. + + Args: + prompt (str): The prompt to evaluate. + + Returns: + List[dict]: Per-example results with keys: input, + target, answer, raw_output, correct. + """ + if self.evaluator.task == Task.CLASSIFICATION: + if isinstance( + self.evaluator.metric, ClassificationMetric + ): + self.evaluator.metric.extract_labels( + self.train_targets + ) + + full_prompts = [ + self.evaluator._get_full_prompt( + prompt, sample, self.template + ) + for sample in self.train_dataset + ] + raw_answers = self.model.batch(full_prompts) + raw_answers = [ + a.content if isinstance(a, AIMessage) else a + for a in raw_answers + ] + + results = [] + for inp, target, raw in zip( + self.train_dataset, + self.train_targets, + raw_answers, + ): + raw_str = str(raw).strip() + extracted = extract_answer( + raw_str, self.ANS_TAGS, raw_str + ) + extracted_str = str(extracted).strip() + target_str = str(target).strip() + + correct = ( + extracted_str.lower() == target_str.lower() + ) + results.append({ + "input": inp, + "target": target_str, + "answer": extracted_str, + "raw_output": raw_str, + "correct": correct, + }) + return results + + def _pack_examples(self, examples: List[dict]) -> str: + """Formats failure examples matching official PE2 format. + + Includes a Reasoning field when the raw model output + contains more than the extracted answer (e.g. chain-of- + thought), matching official PE2's inclusion of reasoning + traces for math/BBH tasks. + + Args: + examples (List[dict]): List of failure example dicts. + + Returns: + str: Formatted string of numbered examples. + """ + parts = [] + for i, ex in enumerate(examples, 1): + lines = [ + f"### Example {i}", + f"Input: {ex['input']}", + ] + # Include reasoning when raw output differs from + # the extracted answer (indicates CoT / reasoning) + raw = ex.get("raw_output", "") + if raw and raw != ex["answer"]: + lines.append(f"Reasoning: {raw}") + lines.append(f"Output: {ex['answer']}") + lines.append(f"Label: {ex['target']}") + parts.append("\n".join(lines)) + return "\n\n".join(parts) + + def _deduplicate( + self, + new_nodes: List[Node], + all_nodes: List[Node], + ) -> List[Node]: + """Removes nodes whose prompts already exist. + + Args: + new_nodes (List[Node]): Candidate new nodes. + all_nodes (List[Node]): All existing nodes. + + Returns: + List[Node]: Deduplicated new nodes. + """ + existing = {n.prompt.strip().lower() for n in all_nodes} + unique = [] + for node in new_nodes: + key = node.prompt.strip().lower() + if key not in existing: + existing.add(key) + unique.append(node) + return unique + + def _sample_failures( + self, results: List[dict], k: int + ) -> List[dict]: + """Randomly samples k failure examples. + + Args: + results (List[dict]): List of failure example dicts. + k (int): Number of examples to sample. + + Returns: + List[dict]: Sampled failure examples. + """ + return random.sample(results, min(k, len(results))) diff --git a/coolprompt/optimizer/pe2_sgr/__init__.py b/coolprompt/optimizer/pe2_sgr/__init__.py new file mode 100644 index 00000000..c4f7ea63 --- /dev/null +++ b/coolprompt/optimizer/pe2_sgr/__init__.py @@ -0,0 +1,9 @@ +from coolprompt.optimizer.pe2_sgr.run import ( + pe2_sgr_optimizer, + PE2SGRMethod, +) + +__all__ = [ + 'pe2_sgr_optimizer', + 'PE2SGRMethod', +] diff --git a/coolprompt/optimizer/pe2_sgr/proposer.py b/coolprompt/optimizer/pe2_sgr/proposer.py new file mode 100644 index 00000000..19a3d6e3 --- /dev/null +++ b/coolprompt/optimizer/pe2_sgr/proposer.py @@ -0,0 +1,437 @@ +"""PE2+SGR v4 proposer: adaptive free-form → structured. + +Always starts with free-form reasoning for radical +exploration. Tracks improvement between beam search +steps. When improvement slows below ``min_improvement``, +locks into structured SGR for precision refinement. +One-way switch per task. +""" + +import json +from pathlib import Path +from typing import Optional + +from langchain_core.language_models.base import ( + BaseLanguageModel, +) + +from coolprompt.optimizer.pe2.node import Node +from coolprompt.optimizer.pe2_sgr.schemas import ( + FullDiagnosis, +) +from coolprompt.utils.logging_config import logger +from coolprompt.utils.parsing import ( + extract_answer, + get_model_answer_extracted, +) +from coolprompt.utils.prompt_templates.pe2_sgr_templates import ( + PE2_SGR_FREEFORM_DIAGNOSIS_TEMPLATE, + PE2_SGR_FULL_DIAGNOSIS_TEMPLATE, + PE2_SGR_GEN_INCREMENTAL, + PE2_SGR_GEN_REIMAGINE, + PE2_SGR_GEN_STRUCTURAL, +) + +_GEN_TEMPLATES = { + "incremental_edit": PE2_SGR_GEN_INCREMENTAL, + "structural_rewrite": PE2_SGR_GEN_STRUCTURAL, + "complete_reimagine": PE2_SGR_GEN_REIMAGINE, +} + + +class SGRProposer: + """Adaptive proposer: starts free-form, switches to + structured SGR when improvement slows. + + Args: + model (BaseLanguageModel): LLM that supports + structured output (tool use / function calling). + prompt_max_tokens (int): Max token budget hint + for the new prompt. + log_path (str | Path | None): Optional path to a + JSONL file for persisting diagnoses. + min_improvement (float): Minimum improvement in + best_val_score between beam steps to stay + in free-form mode. Below this threshold + the proposer locks into structured SGR. + """ + + def __init__( + self, + model: BaseLanguageModel, + prompt_max_tokens: int = 300, + log_path: Optional[str | Path] = None, + min_improvement: float = 0.02, + ) -> None: + self.model = model + self.prompt_max_tokens = prompt_max_tokens + self.log_path = ( + Path(log_path) if log_path else None + ) + self.min_improvement = min_improvement + self._prev_best: Optional[float] = None + self._locked_structured: bool = False + + def propose( + self, + node: Node, + examples_str: str, + full_template: str, + batch_size: int, + best_val_score: float = 0.0, + constraint_feedback: Optional[str] = None, + ) -> tuple[str, str]: + """Proposes a refined prompt via dual-path reasoning. + + Args: + node (Node): Current beam node being refined. + examples_str (str): Formatted failure examples. + full_template (str): The full prompt template. + batch_size (int): Number of failure examples. + best_val_score (float): Best validation score + among current beam candidates. + constraint_feedback (Optional[str]): Per- + constraint failure rates from the trainer. + When None the diagnosis is unchanged. + + Returns: + tuple[str, str]: (new_prompt, reasoning). + """ + if self._prev_best is not None: + if best_val_score != self._prev_best: + delta = best_val_score - self._prev_best + if ( + delta < self.min_improvement + and not self._locked_structured + ): + logger.debug( + "SGR switching to structured " + f"(improvement {delta:.4f} " + f"< {self.min_improvement})" + ) + self._locked_structured = True + + if ( + self._prev_best is None + or best_val_score != self._prev_best + ): + self._prev_best = best_val_score + + cf_block = ( + f"Per-constraint failure rates:\n" + f"{constraint_feedback}" + if constraint_feedback else "" + ) + + if self._locked_structured: + return self._structured_path( + node, examples_str, + full_template, batch_size, + best_val_score, + constraint_feedback=cf_block, + ) + else: + return self._freeform_path( + node, examples_str, + full_template, batch_size, + best_val_score, + constraint_feedback=cf_block, + ) + + def _freeform_path( + self, + node: Node, + examples_str: str, + full_template: str, + batch_size: int, + best_val_score: float, + constraint_feedback: str = "", + ) -> tuple[str, str]: + """Free-form reasoning → always reimagine.""" + logger.debug( + "SGR free-form path " + f"(best_val={best_val_score:.4f} " + f"< {self.min_improvement})" + ) + + reasoning = self._freeform_diagnose( + node, examples_str, + full_template, batch_size, + constraint_feedback=constraint_feedback, + ) + + logger.debug( + "SGR Phase 2 strategy: complete_reimagine " + "(free-form path)" + ) + gen_prompt = PE2_SGR_GEN_REIMAGINE.format( + prompt=node.prompt, + formatted_diagnosis=reasoning, + max_tokens=self.prompt_max_tokens, + ) + result = get_model_answer_extracted( + self.model, gen_prompt + ) + new_prompt = self._extract_prompt( + result, node.prompt + ) + + return new_prompt.strip(), reasoning[:200] + + def _freeform_diagnose( + self, + node: Node, + examples_str: str, + full_template: str, + batch_size: int, + constraint_feedback: str = "", + ) -> str: + """Free-form Phase 1: enhanced PE2-style reasoning.""" + prompt = PE2_SGR_FREEFORM_DIAGNOSIS_TEMPLATE.format( + prompt=node.prompt, + full_template=full_template, + batch_size=batch_size, + examples=examples_str, + constraint_feedback=constraint_feedback, + ) + return get_model_answer_extracted( + self.model, prompt + ) + + def _structured_path( + self, + node: Node, + examples_str: str, + full_template: str, + batch_size: int, + best_val_score: float, + constraint_feedback: str = "", + ) -> tuple[str, str]: + """Structured diagnosis → strategy override → gen.""" + logger.debug( + "SGR structured path " + f"(best_val={best_val_score:.4f} " + f">= {self.min_improvement})" + ) + + diagnosis = self._structured_diagnose( + node, examples_str, + full_template, batch_size, + constraint_feedback=constraint_feedback, + ) + + strategy = self._override_strategy(diagnosis) + formatted = self._format_diagnosis(diagnosis) + + logger.debug( + f"SGR Phase 2 strategy: {strategy}" + ) + gen_template = _GEN_TEMPLATES[strategy] + gen_prompt = gen_template.format( + prompt=node.prompt, + formatted_diagnosis=formatted, + max_tokens=self.prompt_max_tokens, + ) + result = get_model_answer_extracted( + self.model, gen_prompt + ) + new_prompt = self._extract_prompt( + result, node.prompt + ) + + self._log_result(diagnosis, strategy) + self._save_jsonl(diagnosis) + + reasoning = ( + diagnosis.rewrite_strategy.justification + ) + return new_prompt.strip(), reasoning + + def _structured_diagnose( + self, + node: Node, + examples_str: str, + full_template: str, + batch_size: int, + constraint_feedback: str = "", + ) -> FullDiagnosis: + """Structured Phase 1: FullDiagnosis.""" + prompt = PE2_SGR_FULL_DIAGNOSIS_TEMPLATE.format( + prompt=node.prompt, + full_template=full_template, + batch_size=batch_size, + examples=examples_str, + constraint_feedback=constraint_feedback, + ) + structured_model = self.model.with_structured_output( + FullDiagnosis, + ) + return structured_model.invoke(prompt) + + @staticmethod + def _extract_prompt( + result: str, fallback: str + ) -> str: + """Extracts prompt from tags with + robust fallback for empty original prompts.""" + extracted = extract_answer( + result, + ("", ""), + format_mismatch_label=fallback, + ) + extracted_str = str(extracted).strip() + if not extracted_str and result.strip(): + logger.debug( + "SGR extraction empty, using raw " + "result as prompt" + ) + return result.strip() + return extracted_str + + @staticmethod + def _override_strategy( + diagnosis: FullDiagnosis, + ) -> str: + """Overrides model's strategy choice based on its + own severity + homogeneity signals. + + Fixes the conservatism bias where the model + correctly diagnoses fundamental problems but + conservatively picks incremental strategies. + """ + sev = diagnosis.pattern_synthesis.pattern_severity + hom = diagnosis.pattern_synthesis.error_homogeneity + original = diagnosis.rewrite_strategy.approach + + if sev == "fundamental" and hom == "high": + forced = "complete_reimagine" + elif sev == "fundamental" and hom == "medium": + forced = "structural_rewrite" + elif sev == "structural" and hom == "high": + forced = "structural_rewrite" + else: + forced = original + + if forced != original: + logger.debug( + f"SGR strategy override: " + f"{original} -> {forced} " + f"(severity={sev}, " + f"homogeneity={hom})" + ) + + return forced + + @staticmethod + def _format_diagnosis( + diagnosis: FullDiagnosis, + ) -> str: + """Formats a FullDiagnosis into readable text + for Phase 2 input.""" + parts = [] + + parts.append("### Per-Example Analysis") + for i, ea in enumerate( + diagnosis.error_analyses, 1 + ): + parts.append( + f"Example {i}: " + f"{ea.root_cause.value} — " + f"{ea.root_cause_explanation}" + ) + parts.append("") + + ed = diagnosis.edit_decision + parts.append( + f"### Edit Decision\n" + f"Necessary: {ed.editing_necessary} " + f"(confidence: {ed.confidence})\n" + f"Justification: {ed.justification}" + ) + parts.append("") + + ps = diagnosis.pattern_synthesis + parts.append( + f"### Cross-Example Pattern\n" + f"Common failure pattern: " + f"{ps.common_failure_pattern}\n" + f"Severity: {ps.pattern_severity}\n" + f"Error homogeneity: {ps.error_homogeneity}" + ) + parts.append("") + + pa = diagnosis.prompt_analysis + parts.append( + f"### Prompt Analysis\n" + f"Describes task correctly: " + f"{pa.describes_task_correctly}" + ) + if pa.missing_elements: + parts.append( + f"Missing: " + f"{', '.join(pa.missing_elements)}" + ) + if pa.misleading_elements: + parts.append( + f"Misleading: " + f"{', '.join(pa.misleading_elements)}" + ) + parts.append("") + + rs = diagnosis.rewrite_strategy + parts.append( + f"### Strategy: {rs.approach}\n" + f"Key insight: {rs.key_insight}\n" + f"Justification: {rs.justification}" + ) + + return "\n".join(parts) + + def _log_result( + self, + diagnosis: FullDiagnosis, + final_strategy: str, + ) -> None: + """Logs diagnosis at DEBUG level.""" + ps = diagnosis.pattern_synthesis + logger.debug( + f"SGR pattern: " + f"severity={ps.pattern_severity}, " + f"homogeneity={ps.error_homogeneity}, " + f"pattern={ps.common_failure_pattern!r}" + ) + + rs = diagnosis.rewrite_strategy + logger.debug( + f"SGR strategy: " + f"model={rs.approach}, " + f"final={final_strategy}, " + f"key_insight={rs.key_insight!r}" + ) + + for i, ea in enumerate( + diagnosis.error_analyses + ): + logger.debug( + f"SGR error[{i}]: " + f"cause={ea.root_cause.value}, " + f"explanation=" + f"{ea.root_cause_explanation!r}" + ) + + def _save_jsonl( + self, + diagnosis: FullDiagnosis, + ) -> None: + """Appends diagnosis as JSON line to log_path.""" + if self.log_path is None: + return + self.log_path.parent.mkdir( + parents=True, exist_ok=True + ) + with open( + self.log_path, "a", encoding="utf-8" + ) as f: + f.write( + json.dumps(diagnosis.model_dump()) + "\n" + ) diff --git a/coolprompt/optimizer/pe2_sgr/run.py b/coolprompt/optimizer/pe2_sgr/run.py new file mode 100644 index 00000000..5ae3af15 --- /dev/null +++ b/coolprompt/optimizer/pe2_sgr/run.py @@ -0,0 +1,115 @@ +"""High-level entry point for PE2+SGR optimization.""" + +from typing import List, Tuple + +from langchain_core.language_models.base import BaseLanguageModel + +from coolprompt.evaluator import Evaluator +from coolprompt.optimizer.autoprompting_method import ( + AutoPromptingMethod, +) +from coolprompt.optimizer.pe2.trainer import PE2Trainer +from coolprompt.optimizer.pe2_sgr.proposer import SGRProposer +from coolprompt.utils.logging_config import logger + + +def pe2_sgr_optimizer( + model: BaseLanguageModel, + dataset_split: Tuple[ + List[str], List[str], List[str], List[str] + ], + evaluator: Evaluator, + initial_prompt: str, + **kwargs, +) -> str: + """Runs PE2+SGR beam-search prompt optimization. + + Uses structured output (Pydantic schemas) for the + proposer instead of free-text reasoning. Reuses the + PE2 beam-search trainer. + + Args: + model (BaseLanguageModel): The language model to use. + Must support with_structured_output(). + dataset_split: A tuple of (train_dataset, + val_dataset, train_targets, val_targets). + evaluator (Evaluator): Evaluator for scoring prompts. + initial_prompt (str): The starting prompt to optimize. + **kwargs: Optional overrides for train_steps, n_beam, + n_expand, batch_size, backtrack, prompt_max_tokens. + + Returns: + str: The best prompt found after optimization. + """ + ( + train_dataset, + val_dataset, + train_targets, + val_targets, + ) = dataset_split + + args = { + "train_steps": 3, + "n_beam": 3, + "n_expand": 4, + "batch_size": 4, + "backtrack": True, + "prompt_max_tokens": 300, + } + args.update(kwargs) + + proposer = SGRProposer( + model=model, + prompt_max_tokens=args["prompt_max_tokens"], + log_path=args.get("log_path"), + min_improvement=args.get( + "min_improvement", 0.02 + ), + ) + + template = evaluator._get_default_template() + + trainer = PE2Trainer( + model=model, + evaluator=evaluator, + proposer=proposer, + train_dataset=train_dataset, + train_targets=train_targets, + val_dataset=val_dataset, + val_targets=val_targets, + template=template, + train_steps=args["train_steps"], + n_beam=args["n_beam"], + n_expand=args["n_expand"], + batch_size=args["batch_size"], + backtrack=args["backtrack"], + feedback_mode=args.get( + "sgr_constraint_feedback", "auto" + ), + ) + + logger.info("Starting PE2+SGR optimization...") + logger.debug(f"Start prompt:\n{initial_prompt}") + final_prompt = trainer.train(initial_prompt) + logger.info("PE2+SGR optimization completed") + return final_prompt + + +class PE2SGRMethod(AutoPromptingMethod): + + def optimize( + self, model, initial_prompt, + dataset_split=None, evaluator=None, + problem_description=None, **kwargs, + ) -> str: + return pe2_sgr_optimizer( + model, dataset_split, evaluator, + initial_prompt, **kwargs, + ) + + def is_data_driven(self): + return True + + @property + def name(self): + return "pe2_sgr" diff --git a/coolprompt/optimizer/pe2_sgr/schemas.py b/coolprompt/optimizer/pe2_sgr/schemas.py new file mode 100644 index 00000000..b22c2227 --- /dev/null +++ b/coolprompt/optimizer/pe2_sgr/schemas.py @@ -0,0 +1,164 @@ +"""Pydantic schemas for PE2+SGR v3 structured reasoning. + +Above the score threshold, Phase 1 produces a structured +FullDiagnosis with strategy override. Below the threshold, +Phase 1 uses free-form reasoning (no structured output). +Phase 2 always generates the improved prompt via free-form +text. +""" + +from enum import Enum +from typing import List, Literal + +from pydantic import BaseModel, Field + + +class ErrorType(str, Enum): + """Categories of prompt errors.""" + + TASK_UNCLEAR = "task_description_unclear" + MISSING_CONSTRAINTS = "missing_constraints" + WRONG_FORMAT = "wrong_output_format" + INCOMPLETE_GUIDANCE = "incomplete_guidance" + OVERSPECIFICATION = "overspecification" + OTHER = "other" + + +class ErrorAnalysis(BaseModel): + """Analysis of a single failure example.""" + + input_summary: str = Field( + description=( + "Brief description of the input that " + "caused the error" + ) + ) + expected_vs_actual: str = Field( + description=( + "Comparison of expected and actual output" + ) + ) + root_cause: ErrorType = Field( + description=( + "Category of the root cause of the error" + ) + ) + root_cause_explanation: str = Field( + description=( + "Detailed explanation of why the error occurred" + ) + ) + + +class PromptAnalysis(BaseModel): + """Analysis of task description correctness.""" + + describes_task_correctly: bool = Field( + description=( + "Does the prompt correctly describe the task?" + ) + ) + missing_elements: List[str] = Field( + default_factory=list, + description=( + "List of missing elements in the task " + "description" + ), + ) + misleading_elements: List[str] = Field( + default_factory=list, + description=( + "List of potentially misleading instructions" + ), + ) + + +class EditDecision(BaseModel): + """Decision on whether editing is necessary.""" + + editing_necessary: bool = Field( + description=( + "Is editing of the prompt necessary?" + ) + ) + confidence: Literal["low", "medium", "high"] = Field( + description=( + "Confidence level in the editing decision" + ) + ) + justification: str = Field( + description=( + "Justification for the editing decision" + ) + ) + + +class PatternSynthesis(BaseModel): + """Cross-example failure pattern analysis.""" + + common_failure_pattern: str = Field( + description="What unifies all failure examples" + ) + pattern_severity: Literal[ + "surface", "structural", "fundamental" + ] = Field( + description=( + "surface=cosmetic/format issues, " + "structural=missing key instructions, " + "fundamental=entire approach is wrong" + ) + ) + error_homogeneity: Literal[ + "low", "medium", "high" + ] = Field( + description=( + "How similar are the root causes across " + "examples. low=diverse errors, " + "high=all same root cause" + ) + ) + + +class RewriteStrategy(BaseModel): + """Explicit strategy selection for Phase 2.""" + + approach: Literal[ + "incremental_edit", + "structural_rewrite", + "complete_reimagine", + ] = Field( + description=( + "incremental_edit=fix specific issues, " + "structural_rewrite=reorganize significantly, " + "complete_reimagine=design from scratch" + ) + ) + justification: str = Field( + description="Why this strategy was chosen" + ) + key_insight: str = Field( + description=( + "The single most important thing to change" + ) + ) + + +class FullDiagnosis(BaseModel): + """Phase 1 schema when best_val_score >= threshold. + + Full per-example analysis plus global patterns. + Used when task is partially solved and needs + surgical precision. + """ + + error_analyses: List[ErrorAnalysis] = Field( + min_length=1, + description=( + "Analysis of each failure example " + "from the batch" + ), + ) + pattern_synthesis: PatternSynthesis + prompt_analysis: PromptAnalysis + edit_decision: EditDecision + rewrite_strategy: RewriteStrategy diff --git a/coolprompt/utils/enums.py b/coolprompt/utils/enums.py index bd7b12c5..426e6f56 100644 --- a/coolprompt/utils/enums.py +++ b/coolprompt/utils/enums.py @@ -1,6 +1,24 @@ from enum import Enum +class Method(Enum): + HYPE = "hype" + REFLECTIVE = "reflective" + DISTILL = "distill" + PE2 = "pe2" + PE2_SGR = "pe2_sgr" + APE = "ape" + OPRO = "opro" + + def is_data_driven(self) -> bool: + if self is Method.HYPE: + return False + return True + + def __str__(self): + return self.value + + class Task(Enum): CLASSIFICATION = "classification" GENERATION = "generation" diff --git a/coolprompt/utils/prompt_templates/ape_templates.py b/coolprompt/utils/prompt_templates/ape_templates.py new file mode 100644 index 00000000..f17f0714 --- /dev/null +++ b/coolprompt/utils/prompt_templates/ape_templates.py @@ -0,0 +1,12 @@ +APE_PARAPHRASE_TEMPLATE = """You are a prompt engineer. \ +Generate a variation of the instruction below that preserves \ +its semantic meaning but uses different wording, structure, \ +or phrasing. The new instruction should be at most \ +{max_tokens} tokens long. + +Current instruction: +{prompt} + +Output the new instruction between and tags. + +your paraphrased instruction here""" diff --git a/coolprompt/utils/prompt_templates/opro_templates.py b/coolprompt/utils/prompt_templates/opro_templates.py new file mode 100644 index 00000000..0088e1f4 --- /dev/null +++ b/coolprompt/utils/prompt_templates/opro_templates.py @@ -0,0 +1,24 @@ +OPRO_META_TEMPLATE = """\ +You are a prompt engineer. Your task is to generate an \ +instruction that achieves a high score on a given task. + +## Task Demonstrations +Below are some input-output pairs that demonstrate the task: + +{task_demonstrations} + +## Trajectory +Below is a history of instructions that were tried for this \ +task, along with their scores. A higher score is better. \ +The entries are sorted from worst to best. + +{trajectory} + +## Instructions +Generate a new instruction that will achieve a higher score \ +than all previous attempts. The new instruction should be at \ +most {max_tokens} tokens long. + +Output the new instruction between and tags. + +your improved instruction here""" diff --git a/coolprompt/utils/prompt_templates/pe2_sgr_templates.py b/coolprompt/utils/prompt_templates/pe2_sgr_templates.py new file mode 100644 index 00000000..819e32e3 --- /dev/null +++ b/coolprompt/utils/prompt_templates/pe2_sgr_templates.py @@ -0,0 +1,206 @@ +"""PE2+SGR v3 prompt templates. + +Phase 1 — free-form diagnosis (below threshold) or + structured diagnosis (above threshold). +Phase 2 — generation templates (incremental / structural / reimagine). +""" + +# ------------------------------------------------------------------ +# Phase 1: Diagnosis templates +# ------------------------------------------------------------------ + +PE2_SGR_FREEFORM_DIAGNOSIS_TEMPLATE = """\ +A prompt is a text paragraph that outlines the expected actions \ +and instructs the model to generate a specific output. This \ +prompt is concatenated with the input text, and the model then \ +creates the required output. + +In our collaboration, we'll work together to refine a prompt. \ +The process consists of two main steps: + +## Step 1 +I will provide you with the current prompt, how the prompt is \ +concatenated with the input text (i.e., "full template"), along \ +with {batch_size} example(s) that are associated with this \ +prompt. Each example contains the input, the final answer \ +produced by the model, and the ground-truth label to the input. \ +Your task is to analyze the examples, determining whether the \ +existing prompt is describing the task reflected by these \ +examples precisely, and suggest changes to the prompt. + +## Step 2 +Next, you will carefully review your reasoning in step 1, \ +integrate the insights to craft a new, optimized prompt. + +## Prompt +{prompt} + +## Full Template +This describes how the prompt of interest is concatenated with \ +the input text. The prompt may appear before the input text, or \ +after the input text. Optionally the full template may contain \ +other template information. +``` +{full_template} +``` + +## Examples +{examples} +{constraint_feedback} + +## Instructions +For some of these examples, the output does not match with the \ +label. This may be due to the prompt being misleading or not \ +describing the task precisely. + +Please examine the example(s) carefully. Note that the \ +ground-truth labels are __absolutely correct__, but the prompts \ +(task descriptions) may be incorrect and need modification. + +As you analyze, pay special attention to: +1. What common pattern explains ALL these failures? +2. Is the problem surface-level (formatting), structural \ +(missing key instructions), or fundamental (entire approach \ +is wrong)? +3. Are the errors homogeneous (all same root cause) or diverse? +4. If the entire prompt approach seems wrong, consider what a \ +completely different prompt would look like. + +For each example, provide reasoning according to the following \ +template: + +### Example +Input: +Output: +Label: