diff --git a/coolprompt/assistant.py b/coolprompt/assistant.py index 6c86b101..7e7ad90e 100644 --- a/coolprompt/assistant.py +++ b/coolprompt/assistant.py @@ -17,6 +17,7 @@ from coolprompt.utils.enums import PD_Method, Task from coolprompt.utils.correction.corrector import correct from coolprompt.utils.correction.rule import LanguageRule +from coolprompt.utils.utils import get_dataset_split from coolprompt.optimizer.autoprompting_method import AutoPromptingMethod @@ -91,33 +92,6 @@ def reset_stats(self): if hasattr(self._target_model, "reset_stats"): self._target_model.reset_stats() - def _get_dataset_split( - self, - dataset: Iterable[str], - target: Iterable[str], - validation_size: float, - train_as_test: bool, - ) -> Tuple[Iterable[str], Iterable[str], Iterable[str], Iterable[str]]: - """Split the dataset into training and validation sets. - - Args: - dataset (Iterable[str]): Input texts. - target (Iterable[str]): Corresponding labels/targets. - validation_size (float): Fraction of data to use for validation. - train_as_test (bool): If True, use the full dataset as both - train and validation (ignoring `validation_size`). - - Returns: - Tuple[Iterable[str], Iterable[str], Iterable[str], Iterable[str]]: - A tuple (train_data, val_data, train_targets, val_targets). - """ - if train_as_test: - return (dataset, dataset, target, target) - train_data, val_data, train_targets, val_targets = train_test_split( - dataset, target, test_size=validation_size - ) - return (train_data, val_data, train_targets, val_targets) - def run( self, start_prompt: str, @@ -144,6 +118,7 @@ def run( return_final_prompt: bool = True, hyper_meta_info: dict = None, system_model_as_optimizer: bool = False, + use_structured_output: bool = True, **kwargs, ) -> Optional[str]: """Run prompt optimization using the selected method. @@ -205,6 +180,8 @@ def run( merged into the meta-info block for ``hyper`` and ``hyper_light``. system_model_as_optimizer (bool): If True, use the system model for optimizing processes, while target model will be used for inference. + use_structured_output (bool): Either to use structured output or not. + Defaults to True. **kwargs: Additional arguments passed to the optimization method. Returns: @@ -220,7 +197,10 @@ def run( validate_verbose(verbose) set_verbose(verbose) - task_detector = TaskDetector(self._system_model) + task_detector = TaskDetector( + model=self._system_model, + use_structured_output=use_structured_output + ) if task is None: task = task_detector.generate(start_prompt) @@ -255,10 +235,17 @@ def run( ) metric_name = base_metric._get_name() evaluator = Evaluator( - self._target_model, task_value, base_metric, batch_size=batch_size + model=self._target_model, + task=task_value, + metric=base_metric, + batch_size=batch_size, + use_structured_output=use_structured_output ) final_prompt = "" - generator = SyntheticDataGenerator(self._system_model) + generator = SyntheticDataGenerator( + model=self._system_model, + use_structured_output=use_structured_output + ) if dataset is None: dataset, target, problem_description = generator.generate( @@ -271,7 +258,7 @@ def run( self.synthetic_dataset = dataset self.synthetic_target = target - dataset_split = self._get_dataset_split( + dataset_split = get_dataset_split( dataset=dataset, target=target, validation_size=validation_size, @@ -320,13 +307,14 @@ def run( dataset_split=dataset_split, evaluator=evaluator, problem_description=problem_description, + use_structured_output=use_structured_output, **kwargs, ) logger.info("Running the prompt format checking...") final_prompt = correct( prompt=final_prompt, - rule=LanguageRule(self._system_model), + rule=LanguageRule(self._system_model, use_structured_output=use_structured_output), start_prompt=start_prompt, ) @@ -368,6 +356,7 @@ def test( metric: Optional[str] = None, batch_size: int = 25, return_raw_outputs: bool = True, + use_structured_output: bool = True ) -> List[str] | Tuple[List[str], float]: """ Generate model predictions for a test dataset and optionally compute a metric. @@ -390,6 +379,8 @@ def test( batch_size (int, default=25): Number of samples per inference batch. return_raw_outputs (bool, default=True): If True, return raw model outputs; if False, return parsed outputs via metric.parse_output(). + use_structured_output: a boolean variable. + Either to use structured output or nor. Returns: If targets is None: List[str] of raw/parsed outputs. @@ -404,32 +395,33 @@ def test( "No prompt provided and self.final_prompt is not set. " "Either call .run() first or pass prompt explicitly." ) - + if task is None: task_detector = TaskDetector(self._system_model) task = task_detector.generate(use_prompt) - + task_str = task.lower() if task_str not in ("classification", "generation"): raise ValueError("task must be 'classification' or 'generation'.") - + task_enum = Task.CLASSIFICATION if task_str == "classification" else Task.GENERATION - + if metric is None: metric = "accuracy" if task_enum == Task.CLASSIFICATION else "meteor" - + metric_impl = validate_and_create_metric(task_enum, metric) - + evaluator = Evaluator( model=self._target_model, task=task_enum, metric=metric_impl, batch_size=batch_size, + use_structured_output=use_structured_output ) - + dataset_list = list(dataset) use_targets = list(targets) if targets is not None else [""] * len(dataset_list) - + result = evaluator.evaluate( prompt=use_prompt, dataset=dataset_list, @@ -437,11 +429,11 @@ def test( template=None, return_detailed=True, ) - + outputs = result.raw_outputs if return_raw_outputs else [ metric_impl.parse_output(a) for a in result.raw_outputs ] - + if targets is not None: return outputs, result.aggregate_score return outputs diff --git a/coolprompt/data_generator/generator.py b/coolprompt/data_generator/generator.py index f62fc6b7..6f8295b1 100644 --- a/coolprompt/data_generator/generator.py +++ b/coolprompt/data_generator/generator.py @@ -1,15 +1,14 @@ from typing import Optional, List, Tuple, Any from langchain_core.language_models.base import BaseLanguageModel -from langchain_core.language_models.chat_models import BaseChatModel from langchain_core.messages.ai import AIMessage from pydantic import BaseModel -from coolprompt.data_generator.pydantic_formatters import ( - ProblemDescriptionStructuredOutputSchema, - ClassificationTaskStructuredOutputSchema, +from coolprompt.utils.structured_schemas.data_generator import ( ClassificationTaskExample, + ClassificationTaskResponse, GenerationTaskExample, - GenerationTaskStructuredOutputSchema, + GenerationTaskResponse, + ProblemDescriptionResponse, ) from coolprompt.utils.prompt_templates.data_generator_templates import ( PROBLEM_DESCRIPTION_TEMPLATE, @@ -25,39 +24,49 @@ class SyntheticDataGenerator: - """Synthetic Data Generator - Generates synthetic dataset for prompt optimization - based on given initial prompt and optional problem description + """Synthetic Data Generator. + + Generates synthetic datasets for prompt optimization based on a + given initial prompt and an optional problem description. Attributes: - model: langchain.BaseLanguageModel class of model to use. + model: ``langchain.BaseLanguageModel`` instance to use for LLM calls. + use_structured_output: When ``True``, every LLM call routes through + ``model.with_structured_output(schema, method="json_schema")`` + using the Pydantic schemas defined in + :mod:`coolprompt.utils.structured_schemas.data_generator`. + When ``False`` (default), the generator falls back to a plain + ``model.invoke`` call followed by JSON extraction from the + raw model response — the same convention used by + :mod:`coolprompt.optimizer` submodules. """ - def __init__(self, model: BaseLanguageModel) -> None: + def __init__( + self, + model: BaseLanguageModel, + use_structured_output: bool = False, + ) -> None: self.model = model + self.use_structured_output = use_structured_output def _generate( - self, request: str, schema: BaseModel, field_name: str + self, request: str, schema: type[BaseModel], field_name: str ) -> Any: - """Generates model output - either using structured output from langchain - or just strict json output format for LLM + """Generates model output using either structured-output via + LangChain or a raw JSON parsing fallback. Args: - request (str): request to LLM - when langchain structured output is used - schema (BaseModel): Pydantic output format - field_name (str): field name to select from output + request (str): request to send to the LLM. + schema (type[BaseModel]): Pydantic schema describing the + expected structured output. Only used when + ``self.use_structured_output`` is ``True``. + field_name (str): top-level field name to extract from the + model output. Returns: - Any: generated data + Any: extracted value of ``field_name`` from the model output. """ - if hasattr(self.model, "model"): - wrapped_model = self.model.model - else: - wrapped_model = self.model - - if not isinstance(wrapped_model, BaseChatModel): + if not self.use_structured_output: output = self.model.invoke(request) if isinstance(output, AIMessage): output = output.content @@ -92,13 +101,16 @@ def _examples_to_str(self, examples: List[Tuple[str, str]]) -> str: def _generate_problem_description( self, prompt: str, examples: Optional[List[Tuple[str, str]]] = None ) -> str: - """Generates problem description based on given user prompt + """Generates problem description based on given user prompt. Args: - prompt (str): initial user prompt + prompt (str): initial user prompt. + examples (Optional[List[Tuple[str, str]]]): optional list of + ``(input, output)`` examples drawn from the task dataset + to ground the description. Returns: - str: generated problem description + str: generated problem description. """ if examples: request = PROBLEM_DESCRIPTION_BASED_ON_EXAMPLES_TEMPLATE.format( @@ -109,7 +121,7 @@ def _generate_problem_description( return self._generate( request, - ProblemDescriptionStructuredOutputSchema, + ProblemDescriptionResponse, "problem_description", ) @@ -119,7 +131,7 @@ def _convert_dataset( dict | ClassificationTaskExample | GenerationTaskExample ], ) -> Tuple[List[str], List[str]]: - """Converts outputs to the dataset format + """Converts outputs to the dataset format. Args: examples ( @@ -128,11 +140,11 @@ def _convert_dataset( ClassificationTaskExample | GenerationTaskExample ] - ): outputs of the model + ): outputs of the model. Returns: Tuple[List[str], List[str]]: - converted dataset and target + converted dataset and target. """ dataset = [] targets = [] @@ -156,12 +168,11 @@ def generate( num_samples: int = 8, corner_ratio: float = 0.4, ) -> Tuple[List[str], List[str], str]: - """Generates synthetic dataset - based on given user prompt, optimization task - and optionally provided problem description + """Generates synthetic dataset based on the given user prompt, + optimization task and optionally provided problem description. - If problem description isn't provided - - it will be generated automatically + If problem description isn't provided it will be generated + automatically. Args: prompt (str): initial user prompt @@ -184,7 +195,7 @@ def generate( Returns: Tuple[List[str], List[str], str]: - generated dataset, target and problem description + generated dataset, target and problem description. """ if not 1 <= num_samples <= 100: @@ -204,11 +215,11 @@ def generate( if task == Task.CLASSIFICATION: regular_template = CLASSIFICATION_DATA_GENERATING_TEMPLATE corner_template = CLASSIFICATION_CORNER_CASE_GENERATING_TEMPLATE - schema = ClassificationTaskStructuredOutputSchema + schema = ClassificationTaskResponse else: regular_template = GENERATION_DATA_GENERATING_TEMPLATE corner_template = GENERATION_CORNER_CASE_GENERATING_TEMPLATE - schema = GenerationTaskStructuredOutputSchema + schema = GenerationTaskResponse n_corner = int(num_samples * corner_ratio) n_regular = num_samples - n_corner diff --git a/coolprompt/data_generator/pydantic_formatters.py b/coolprompt/data_generator/pydantic_formatters.py deleted file mode 100644 index ce4cb6a9..00000000 --- a/coolprompt/data_generator/pydantic_formatters.py +++ /dev/null @@ -1,41 +0,0 @@ -from pydantic import BaseModel, Field -from typing import List - - -class ProblemDescriptionStructuredOutputSchema(BaseModel): - """Structured response containing a generated problem description.""" - - problem_description: str = Field( - description="Determined problem description" - ) - - -class ClassificationTaskExample(BaseModel): - """Single synthetic classification sample.""" - - input: str = Field(description="Input request") - output: str = Field(description="Output label") - - -class ClassificationTaskStructuredOutputSchema(BaseModel): - """Structured response containing classification examples.""" - - examples: List[ClassificationTaskExample] = Field( - description="List of examples like " - + '{"input": "...", "output": "ground-truth label"}' - ) - - -class GenerationTaskExample(BaseModel): - """Single synthetic generation sample.""" - - input: str = Field(description="Input request") - output: str = Field(description="LLM answer") - - -class GenerationTaskStructuredOutputSchema(BaseModel): - """Structured response containing generation examples.""" - - examples: List[GenerationTaskExample] = Field( - description='List of examples like {"input": "...", "output": "..."}' - ) diff --git a/coolprompt/evaluator/evaluator.py b/coolprompt/evaluator/evaluator.py index f6f059dc..a81dbdc5 100644 --- a/coolprompt/evaluator/evaluator.py +++ b/coolprompt/evaluator/evaluator.py @@ -12,7 +12,13 @@ from coolprompt.utils.enums import Task from coolprompt.utils.prompt_templates.default_templates import ( CLASSIFICATION_TASK_TEMPLATE, + CLASSIFICATION_TASK_TEMPLATE_STRUCTURED, GENERATION_TASK_TEMPLATE, + GENERATION_TASK_TEMPLATE_STRUCTURED, +) +from coolprompt.utils.structured_schemas.evaluator import ( + ClassificationAnswerResponse, + GenerationAnswerResponse, ) @@ -52,13 +58,34 @@ def __init__( task: Task, metric: BaseMetric, batch_size: int = 25, + use_structured_output: bool = False, ) -> None: - """Initialize the evaluator with a model, task type, metric, and batch size.""" + """Initialize the evaluator with a model, task type, metric, and batch size. + + Args: + model (BaseLanguageModel): LangChain model used to generate + answers on dataset samples. + task (Task): Task type (classification / generation). + metric (BaseMetric): Metric instance used to score answers. + batch_size (int): Batch size for the model. + use_structured_output (bool): If ``True``, the target model + is invoked via ``model.with_structured_output(schema, + method="json_schema")`` with a task-specific pydantic + schema (see ``coolprompt.utils.structured_schemas.evaluator``). + The extracted ``answer`` field is then wrapped back into + ``...`` so the downstream metric parsing in + ``BaseMetric.compute`` is preserved unchanged. Defaults + to ``False``. + """ self.model = model self.task = task self.metric = metric self.batch_size = batch_size - logger.info(f"Evaluator successfully initialized with {metric} metric") + self.use_structured_output = use_structured_output + logger.info( + f"Evaluator successfully initialized with {metric} metric " + f"(use_structured_output={use_structured_output})" + ) def evaluate( self, @@ -152,12 +179,41 @@ def evaluate( raw_outputs=answers, ) + def _get_response_schema(self): + """Returns the pydantic response schema for the current task.""" + match self.task: + case Task.CLASSIFICATION: + return ClassificationAnswerResponse + case Task.GENERATION: + return GenerationAnswerResponse + raise ValueError(f"Unsupported task for structured output: {self.task}") + + def _wrap_in_ans_tags(self, answer: str) -> str: + """Wraps a raw answer string in ... tags so that the + downstream metric parsing (``extract_answer``) treats it the same + way as a free-text answer.""" + start, end = self.metric.ANS_TAGS + return f"{start}{answer}{end}" + def _run_batches(self, full_prompts: list[str]) -> list[str]: - """Run the model on preformatted prompts in batches with progress tracking.""" + """Run the model on preformatted prompts in batches with progress tracking. + + When ``self.use_structured_output`` is ``True`` the target model is + invoked through ``with_structured_output`` with the task-specific + schema; the resulting ``answer`` field is wrapped in ```` + tags to preserve the standard metric-parsing contract. + """ answers: list[str] = [] total = len(full_prompts) total_batches = (total + self.batch_size - 1) // self.batch_size + runner = self.model + if self.use_structured_output: + schema = self._get_response_schema() + runner = self.model.with_structured_output( + schema, method="json_schema" + ) + with tqdm( total=total, desc="Evaluating", @@ -171,7 +227,7 @@ def _run_batches(self, full_prompts: list[str]) -> list[str]: batch_answers = None for attempt in range(5): try: - batch_answers = self.model.batch(batch) + batch_answers = runner.batch(batch) break except Exception as exception: logger.warning( @@ -187,10 +243,18 @@ def _run_batches(self, full_prompts: list[str]) -> list[str]: start // self.batch_size + 1}/{total_batches} failed after 5 attempts" ) from exception - normalized_answers = [ - a.content if isinstance(a, AIMessage) else str(a) - for a in batch_answers - ] + if self.use_structured_output: + normalized_answers = [ + self._wrap_in_ans_tags( + a.answer if hasattr(a, "answer") else str(a) + ) + for a in batch_answers + ] + else: + normalized_answers = [ + a.content if isinstance(a, AIMessage) else str(a) + for a in batch_answers + ] answers.extend(normalized_answers) pbar.update(len(batch)) logger.debug( @@ -233,9 +297,19 @@ def _get_full_prompt( return template.format(PROMPT=prompt, INPUT=sample) def _get_default_template(self) -> str: - """Returns the default template for the task type.""" + """Returns the default template for the task type. + + When ``self.use_structured_output`` is ``True`` a structured-output + variant of the template is returned — it omits the instruction to + wrap the answer in ```` tags, since the schema field already + defines the answer location. + """ match self.task: case Task.CLASSIFICATION: + if self.use_structured_output: + return CLASSIFICATION_TASK_TEMPLATE_STRUCTURED return CLASSIFICATION_TASK_TEMPLATE case Task.GENERATION: + if self.use_structured_output: + return GENERATION_TASK_TEMPLATE_STRUCTURED return GENERATION_TASK_TEMPLATE diff --git a/coolprompt/evaluator/metrics.py b/coolprompt/evaluator/metrics.py index e1271341..3768880e 100644 --- a/coolprompt/evaluator/metrics.py +++ b/coolprompt/evaluator/metrics.py @@ -26,6 +26,7 @@ FLUENCY_TEMPLATE, RELEVANCE_TEMPLATE, ) +from coolprompt.utils.structured_schemas.evaluator import JudgeScoreResponse class HFEvaluateMetric(ABC): @@ -410,12 +411,30 @@ def __init__( prompt_template: Optional[str] = None, custom_templates: Optional[dict[str, str]] = None, metric_ceil: int = 10, + use_structured_output: bool = False, ): - """Initialize judge prompts and scoring scale.""" + """Initialize judge prompts and scoring scale. + + Args: + model (BaseLanguageModel): LangChain judge model. + criteria (str | list[str]): Criterion name(s) to score on. + prompt_template (Optional[str]): Unused legacy argument kept + for backwards compatibility. + custom_templates (Optional[dict[str, str]]): Optional mapping + ``criterion -> template`` that overrides/extends the + built-in templates. + metric_ceil (int): Maximum integer score the judge can give. + use_structured_output (bool): If ``True``, the judge is + invoked via ``model.with_structured_output(JudgeScoreResponse, + method="json_schema")`` so the score is returned as a + validated integer instead of being regex-parsed from a + free-text response. Defaults to ``False``. + """ super().__init__() self.model = model self.prompt_template = prompt_template self.metric_ceil = metric_ceil + self.use_structured_output = use_structured_output self.prompt_templates = { "accuracy": ACCURACY_QA_TEMPLATE, @@ -437,6 +456,11 @@ def __init__( def _compute_raw(self, outputs, targets, dataset): scores = [] + runner = self.model + if self.use_structured_output: + runner = self.model.with_structured_output( + JudgeScoreResponse, method="json_schema" + ) for _, template in self.templates.items(): requests = [ template.format( @@ -446,20 +470,27 @@ def _compute_raw(self, outputs, targets, dataset): ) for request, response in zip(dataset, outputs) ] - answers = self.model.batch(requests) + answers = runner.batch(requests) parsed = [] - for a in answers: - if isinstance(a, AIMessage): - content = ( - a.content - if isinstance(a.content, str) - else str(a.content) - ) - match = re.search(r"\d+", content) - parsed.append(int(match.group()) if match else 0) - else: - parsed.append(0) + if self.use_structured_output: + for a in answers: + try: + parsed.append(int(a.score)) + except (AttributeError, TypeError, ValueError): + parsed.append(0) + else: + for a in answers: + if isinstance(a, AIMessage): + content = ( + a.content + if isinstance(a.content, str) + else str(a.content) + ) + match = re.search(r"\d+", content) + parsed.append(int(match.group()) if match else 0) + else: + parsed.append(0) normalized = [ clip(ans, 0, self.metric_ceil) / self.metric_ceil @@ -484,10 +515,21 @@ def __init__( evaluation_steps: Optional[list[str]] = None, evaluation_params: Optional[list[LLMTestCaseParams]] = None, strict_mode: bool = False, + use_structured_output: bool = False, ) -> None: """Configure a GEval metric around a LangChain model.""" super().__init__() - wrapped_model = DeepEvalLangChainModel(model) + self.use_structured_output = use_structured_output + # The flag is forwarded to the DeepEval wrapper so that every + # LLM call DeepEval issues for this metric goes through + # ``with_structured_output(schema, method="json_schema")`` when + # DeepEval provides a pydantic ``schema`` (per the + # ``DeepEvalBaseLLM`` contract). When DeepEval does not supply a + # schema, the wrapper falls back to a plain ``invoke`` that + # returns the model's textual answer (legacy behaviour). + wrapped_model = DeepEvalLangChainModel( + model, use_structured_output=use_structured_output + ) if criteria is not None and evaluation_steps is not None: raise ValueError( @@ -638,6 +680,9 @@ def validate_and_create_metric( "llm_as_judge_custom_templates" ), metric_ceil=kwargs.get("llm_as_judge_metric_ceil", 10), + use_structured_output=kwargs.get( + "use_structured_output", False + ), ) if metric == "geval": if model is None: @@ -650,6 +695,9 @@ def validate_and_create_metric( evaluation_steps=kwargs.get("geval_evaluation_steps"), evaluation_params=kwargs.get("geval_evaluation_params"), strict_mode=kwargs.get("geval_strict_mode", False), + use_structured_output=kwargs.get( + "use_structured_output", False + ), ) if metric in GENERATION_METRIC_NAME_MAPPING.keys(): return GENERATION_METRIC_NAME_MAPPING[metric]() diff --git a/coolprompt/language_model/deepeval_model.py b/coolprompt/language_model/deepeval_model.py index 94281e44..fa9f832f 100644 --- a/coolprompt/language_model/deepeval_model.py +++ b/coolprompt/language_model/deepeval_model.py @@ -1,21 +1,45 @@ +from typing import Optional, Type, Union + from deepeval.models.base_model import DeepEvalBaseLLM from langchain_core.language_models import BaseLanguageModel from langchain_core.messages import AIMessage +from pydantic import BaseModel class DeepEvalLangChainModel(DeepEvalBaseLLM): - """DeepEval LLM wrapper for LangChain BaseLanguageModel.""" + """DeepEval LLM wrapper for a LangChain ``BaseLanguageModel``. + + The wrapper exposes the ``DeepEvalBaseLLM`` interface (``generate`` / + ``a_generate``) on top of an arbitrary LangChain chat model so that + DeepEval metrics (e.g. ``GEval``) can drive the same model that the + rest of CoolPrompt uses. + + Args: + model: The underlying LangChain language model to delegate to. + use_structured_output: If ``True`` **and** DeepEval supplies a + pydantic ``schema`` to ``generate`` / ``a_generate`` (per + the ``DeepEvalBaseLLM`` contract), the call is routed + through ``model.with_structured_output(schema, + method="json_schema")``. Otherwise the wrapper falls back + to the legacy plain ``invoke`` + ``AIMessage`` → ``str`` + extraction. Defaults to ``False`` (always legacy + behaviour). + """ - def __init__(self, model: BaseLanguageModel): + def __init__( + self, + model: BaseLanguageModel, + use_structured_output: bool = False, + ): self.model = model + self.use_structured_output = use_structured_output def load_model(self) -> BaseLanguageModel: return self.model - def generate(self, prompt: str) -> str: - """Generate a synchronous text response for DeepEval.""" - chat_model = self.load_model() - result = chat_model.invoke(prompt) + @staticmethod + def _extract_text(result) -> str: + """Coerce a LangChain ``invoke`` result into a plain string.""" if isinstance(result, AIMessage): return ( result.content @@ -24,17 +48,52 @@ def generate(self, prompt: str) -> str: ) return str(result) - async def a_generate(self, prompt: str) -> str: + def _structured_runner(self, schema: Type[BaseModel]): + """Return the LangChain runnable with structured output bound.""" + return self.model.with_structured_output( + schema, method="json_schema" + ) + + def generate( + self, + prompt: str, + schema: Optional[Type[BaseModel]] = None, + ) -> Union[str, BaseModel]: + """Generate a synchronous text response for DeepEval. + + Args: + prompt: Prompt text to send to the underlying model. + schema: Optional pydantic schema passed by DeepEval to + request a structured response. Honoured only when + :attr:`use_structured_output` is ``True``. + + Returns: + * When :attr:`use_structured_output` is ``True`` **and** a + ``schema`` is supplied: a populated pydantic instance of + that schema. + * Otherwise: the raw text response (legacy behaviour). + """ + chat_model = self.load_model() + if self.use_structured_output and schema is not None: + runner = self._structured_runner(schema) + return runner.invoke(prompt) + + result = chat_model.invoke(prompt) + return self._extract_text(result) + + async def a_generate( + self, + prompt: str, + schema: Optional[Type[BaseModel]] = None, + ) -> Union[str, BaseModel]: """Generate an asynchronous text response for DeepEval.""" chat_model = self.load_model() + if self.use_structured_output and schema is not None: + runner = self._structured_runner(schema) + return await runner.ainvoke(prompt) + result = await chat_model.ainvoke(prompt) - if isinstance(result, AIMessage): - return ( - result.content - if isinstance(result.content, str) - else str(result.content) - ) - return str(result) + return self._extract_text(result) def get_model_name(self) -> str: return "CoolPrompt DeepEval LangChain Model" diff --git a/coolprompt/optimizer/autoprompting_method.py b/coolprompt/optimizer/autoprompting_method.py index 2c50a1e5..9dfee353 100644 --- a/coolprompt/optimizer/autoprompting_method.py +++ b/coolprompt/optimizer/autoprompting_method.py @@ -69,7 +69,12 @@ def build_benchmark_context( task = validate_task(config["task"]) metric = validate_and_create_metric(task, config["metric"]) - evaluator = Evaluator(model, task, metric) + evaluator = Evaluator( + model, + task, + metric, + use_structured_output=config.get("use_structured_output", False) + ) return BenchmarkContext( model=model, @@ -98,9 +103,26 @@ def optimize( dataset_split: Tuple[List[str], List[str], List[str], List[str]] | None, evaluator: Evaluator | None, problem_description: str | None, + *, + use_structured_output: bool = False, **kwargs, ) -> str: - """Run the prompt optimization process.""" + """Run the prompt optimization process. + + Args: + model: Language model used by the optimizer. + initial_prompt: Starting prompt to optimize. + dataset_split: Optional ``(train, val, train_targets, val_targets)`` + split for data-driven methods. + evaluator: Optional evaluator used to score prompts. + problem_description: Optional natural-language task description. + use_structured_output: If True, the underlying optimizer should use + LangChain ``with_structured_output`` calls (JSON schema) for + LLM interactions instead of free-form text parsing. Concrete + methods that do not support structured output may raise + ``NotImplementedError`` when this is True. Defaults to False. + **kwargs: Method-specific extra parameters. + """ pass @abstractmethod @@ -114,9 +136,9 @@ def name(self) -> str: """Short method id (e.g. ``hyper_light``, ``reflective``).""" pass - def get_template(self, task: Task) -> str: - """Return the default prompt-formatting template for a task.""" - match task: + def get_template(self, task: Task) -> str: + """Return the default prompt-formatting template for a task.""" + match task: case Task.CLASSIFICATION: return CLASSIFICATION_TASK_TEMPLATE case Task.GENERATION: @@ -126,8 +148,18 @@ def run_configured_benchmark( self, ctx: BenchmarkContext, start_prompt: str, + *, + use_structured_output: bool = False, ) -> str: - """Optimization step for YAML benchmarks; override where supported.""" + """Optimization step for YAML benchmarks; override where supported. + + Args: + ctx: Pre-built :class:`BenchmarkContext` (datasets + evaluator). + start_prompt: Prompt to start optimization from. + use_structured_output: Centralized structured-output flag forwarded + from :meth:`run`. Subclasses should propagate it to their + underlying optimizer. Defaults to False. + """ raise NotImplementedError( f"{type(self).__name__} does not support method_evaluation benchmarks" ) @@ -146,7 +178,14 @@ def run( dict with keys ``final_prompt``, ``val_score``, ``test_score``. """ ctx = build_benchmark_context(model, config) - final_prompt = self.run_configured_benchmark(ctx, start_prompt) + use_structured_output = bool( + config.get("method", {}).get("use_structured_output", False) + ) + final_prompt = self.run_configured_benchmark( + ctx, + start_prompt, + use_structured_output=use_structured_output, + ) val_score = ctx.evaluator.evaluate( prompt=final_prompt, dataset=ctx.dataset_split[1], diff --git a/coolprompt/optimizer/distill_prompt/run.py b/coolprompt/optimizer/distill_prompt/run.py index 08b43933..9e1153ae 100644 --- a/coolprompt/optimizer/distill_prompt/run.py +++ b/coolprompt/optimizer/distill_prompt/run.py @@ -22,6 +22,7 @@ def distillprompt( num_epochs: int = 5, output_path: str = "./distillprompt_outputs", use_cache: bool = True, + use_structured_output: bool = False, ) -> str: """Runs the full DistillPrompt optimization process. @@ -43,12 +44,23 @@ def distillprompt( cached results. Defaults to './distillprompt_outputs'. use_cache (bool, optional): If True, caches intermediate results to the output path. Defaults to True. + use_structured_output (bool, optional): Kept for interface parity + with other optimizers. DistillPrompt is deprecated and does + not support structured output, so passing ``True`` raises + ``NotImplementedError``. Defaults to ``False``. Returns: str: The best prompt found after the optimization process. + + Raises: + NotImplementedError: If ``use_structured_output`` is ``True``. """ warn_deprecated("DistillPrompt") + if use_structured_output: + raise NotImplementedError( + "The method is deprecated and does not support structured output" + ) ( train_dataset, validation_dataset, @@ -82,6 +94,8 @@ def optimize( dataset_split, evaluator, problem_description=None, + *, + use_structured_output: bool = False, **kwargs, ): """Run DistillPrompt through the shared method interface.""" @@ -90,6 +104,7 @@ def optimize( dataset_split=dataset_split, evaluator=evaluator, initial_prompt=initial_prompt, + use_structured_output=use_structured_output, **kwargs, ) @@ -97,6 +112,8 @@ def run_configured_benchmark( self, ctx: BenchmarkContext, start_prompt: str, + *, + use_structured_output: bool = False, ) -> str: """Run DistillPrompt from a benchmark context.""" mc = ctx.config.get("method", {}) @@ -105,6 +122,7 @@ def run_configured_benchmark( start_prompt, dataset_split=ctx.dataset_split, evaluator=ctx.evaluator, + use_structured_output=use_structured_output, num_epochs=mc.get("num_epochs", 5), output_path=mc.get("output_path", "./distillprompt_outputs"), use_cache=mc.get("use_cache", True), diff --git a/coolprompt/optimizer/hyper/feedback_module.py b/coolprompt/optimizer/hyper/feedback_module.py index 5f65ffb0..32e5b257 100644 --- a/coolprompt/optimizer/hyper/feedback_module.py +++ b/coolprompt/optimizer/hyper/feedback_module.py @@ -18,6 +18,12 @@ Recommendation, SECTION_GROUPS_FILTER_PROMPT, ) +from coolprompt.utils.structured_schemas.optimizer.hyper import ( + InstanceLeakAuditResponse, + RecommendationGroupsResponse, + SectionRecommendationResponse, + SynthesizedRecommendationsResponse, +) logger = logging.getLogger(__name__) @@ -50,6 +56,7 @@ def __init__( contrastive_max_answer_chars: int = 500, feedback_answer_head_chars: int = 500, feedback_answer_tail_chars: int = 500, + use_structured_output: bool = False, **kwargs: Any, ) -> None: """Configure the feedback LLM client and truncation budgets. @@ -92,6 +99,7 @@ def __init__( self.contrastive_max_answer_chars = contrastive_max_answer_chars self.feedback_answer_head_chars = feedback_answer_head_chars self.feedback_answer_tail_chars = feedback_answer_tail_chars + self.use_structured_output = use_structured_output self.last_audit_trace: List[Dict[str, Any]] = [] def _build_section_descriptions(self) -> str: @@ -161,9 +169,38 @@ def generate_recommendation( ground_truth=ground_truth, section_descriptions=self._build_section_descriptions(), ) + if self.use_structured_output: + return self._invoke_structured_recommendation(formatted_prompt) result = get_model_answer_extracted(self.model, formatted_prompt) return self._parse_recommendation(result) + def _invoke_structured_recommendation(self, formatted_prompt: str) -> Recommendation: + """Run a structured LLM call returning a SectionRecommendationResponse. + + Validates the returned section against ``self._valid_sections`` and + maps unknown / blank sections to the ``general`` section, mirroring + the behavior of :meth:`_try_parse` for the text-mode path. + """ + structured = self.model.with_structured_output( + SectionRecommendationResponse, method="json_schema" + ) + try: + response = structured.invoke(formatted_prompt) + except Exception as exc: + logger.debug(f"[Feedback] structured recommendation call failed: {exc}") + return Recommendation(section=GENERAL_SECTION, text="") + section = (response.section or "").strip() + text = (response.text or "").strip() + if not text: + return Recommendation(section=GENERAL_SECTION, text="") + if section not in self._valid_sections: + if section: + logger.debug( + f"[Feedback] Unknown section '{section}' from model -> general" + ) + section = GENERAL_SECTION + return Recommendation(section=section, text=text) + @staticmethod def _pick_best_contrastive( candidates: List[ContrastiveCandidate], failing_score: float | int @@ -266,6 +303,8 @@ def _generate_contrastive( ground_truth=ground_truth, section_descriptions=self._build_section_descriptions(), ) + if self.use_structured_output: + return self._invoke_structured_recommendation(formatted_prompt) result = get_model_answer_extracted(self.model, formatted_prompt) return self._parse_recommendation(result) @@ -277,7 +316,7 @@ def _try_parse(self, raw_str: str) -> Tuple[str, str, Optional[str]]: Returns: Tuple ``(section, text, error_kind)`` where ``error_kind`` is ``None`` on - success, ``"json_error"`` for malformed JSON, or ``"invalid_section"`` when + success, ``"json_error"`` for malformed JSON, or ``"invalid_section"`` when the section is not whitelisted (text still returned, mapped to ``general`` upstream by callers). """ try: @@ -402,15 +441,32 @@ def drop_instance_leaks( problem_description=problem_description, recommendations_json=json.dumps(payload, ensure_ascii=False, indent=2), ) - raw = get_model_answer_extracted(self.model, prompt) - raw_str = raw if isinstance(raw, str) else str(raw) + + raw_str = "" try: - data = extract_json(raw_str) - if not isinstance(data, dict) or "verdicts" not in data: - raise ValueError("missing 'verdicts' key") - verdicts = data["verdicts"] - if not isinstance(verdicts, list) or len(verdicts) != len(recs): - raise ValueError("verdicts count mismatch") + if self.use_structured_output: + structured = self.model.with_structured_output( + InstanceLeakAuditResponse, method="json_schema" + ) + response = structured.invoke(prompt) + verdicts: List[Any] = [ + { + "verdict": (v.verdict or "").strip(), + "text": (v.text or "").strip(), + } + for v in (response.verdicts or []) + ] + if len(verdicts) != len(recs): + raise ValueError("verdicts count mismatch") + else: + raw = get_model_answer_extracted(self.model, prompt) + raw_str = raw if isinstance(raw, str) else str(raw) + data = extract_json(raw_str) + if not isinstance(data, dict) or "verdicts" not in data: + raise ValueError("missing 'verdicts' key") + verdicts = data["verdicts"] + if not isinstance(verdicts, list) or len(verdicts) != len(recs): + raise ValueError("verdicts count mismatch") kept: List[Recommendation] = [] trace: List[Dict[str, Any]] = [] for r, v in zip(recs, verdicts): @@ -501,9 +557,33 @@ def _filter_section( section_name=section_name, groups_json=json.dumps(group_payload, ensure_ascii=False, indent=2), ) - raw = get_model_answer_extracted(self.model, prompt) - synthesized = self._parse_synthesized_filter_response(raw) + synthesized: Optional[List[Tuple[str, int]]] = None + if self.use_structured_output: + try: + structured = self.model.with_structured_output( + SynthesizedRecommendationsResponse, method="json_schema" + ) + response = structured.invoke(prompt) + parsed: List[Tuple[str, int]] = [] + for item in response.synthesized or []: + text = (item.text or "").strip() + try: + weight = max(1, int(item.weight)) + except (TypeError, ValueError): + weight = 1 + if text: + parsed.append((text, weight)) + synthesized = parsed or None + except Exception as exc: + logger.debug( + f"[Feedback] structured section filter failed: {exc}" + ) + synthesized = None + else: + raw = get_model_answer_extracted(self.model, prompt) + synthesized = self._parse_synthesized_filter_response(raw) + if synthesized is None: logger.warning( f"[Feedback] Section '{section_name}': group filter parse " @@ -547,6 +627,34 @@ def _llm_partition_into_groups(self, texts: List[str]) -> List[List[int]]: prompt = RECOMMENDATIONS_GROUP_PROMPT.format( items_json=json.dumps(payload, ensure_ascii=False, indent=2) ) + + if self.use_structured_output: + try: + structured = self.model.with_structured_output( + RecommendationGroupsResponse, method="json_schema" + ) + response = structured.invoke(prompt) + raw_groups = response.groups or [] + groups: List[List[int]] = [] + for g in raw_groups: + if not isinstance(g, list): + continue + ids = [int(x) for x in g if isinstance(x, (int, float, str))] + ids = [i for i in ids if 0 <= i < len(texts)] + if ids: + groups.append(ids) + seen = {i for grp in groups for i in grp} + for i in range(len(texts)): + if i not in seen: + groups.append([i]) + if groups: + return groups + except Exception as exc: + logger.debug( + f"[Feedback] structured group partition failed: {exc}" + ) + return [[i] for i in range(len(texts))] + raw = get_model_answer_extracted(self.model, prompt) try: data = extract_json(raw if isinstance(raw, str) else str(raw)) diff --git a/coolprompt/optimizer/hyper/hyper.py b/coolprompt/optimizer/hyper/hyper.py index 4f00af5c..9df23948 100644 --- a/coolprompt/optimizer/hyper/hyper.py +++ b/coolprompt/optimizer/hyper/hyper.py @@ -29,6 +29,9 @@ PARAPHRASE_PROMPT, Recommendation, ) +from coolprompt.utils.structured_schemas.optimizer.hyper import ( + ParaphrasedVariantResponse, +) _BERTSCORE_MODEL_TYPE = "microsoft/deberta-large-mnli" _bertscore_evaluate = None @@ -229,6 +232,7 @@ def __init__( feedback_answer_tail_chars: int = 500, enable_instance_leak_audit: bool = True, random_seed: Optional[int] = None, + use_structured_output: bool = False, **kwargs ) -> None: """Configure HyPER hyperparameters and construct submodules. @@ -249,9 +253,14 @@ def __init__( enable_instance_leak_audit: If True, run ``drop_instance_leaks`` when ``meta_info`` contains a non-empty ``problem_description``. Defaults to True. random_seed: Base seed for mini-batch sampling (per-iteration offset applied). + use_structured_output: a boolean variable. + Either to use structured output or nor. """ super().__init__(model) - self.meta_prompt_module = MetaPromptOptimizer(model) + self.use_structured_output = use_structured_output + self.meta_prompt_module = MetaPromptOptimizer( + model, use_structured_output=use_structured_output + ) self.evaluator = evaluator self.contrastive_probability = contrastive_probability self.contrastive_max_answer_chars = contrastive_max_answer_chars @@ -265,6 +274,7 @@ def __init__( contrastive_max_answer_chars=contrastive_max_answer_chars, feedback_answer_head_chars=feedback_answer_head_chars, feedback_answer_tail_chars=feedback_answer_tail_chars, + use_structured_output=use_structured_output, ) self.n_iterations = n_iterations or kwargs.get("num_epochs", 5) self.patience = patience @@ -284,8 +294,18 @@ def _get_variants_from_best(self, best_prompt: str, n_candidates: int) -> List[s Returns: List whose first element is ``best_prompt`` followed by paraphrases. """ + query = PARAPHRASE_PROMPT.format(prompt=best_prompt) + if self.use_structured_output: + structured = self.model.bind(temperature=0.9).with_structured_output( + ParaphrasedVariantResponse, method="json_schema" + ) + raw_outputs = [ + r.paraphrased_prompt for r in structured.batch([query] * n_candidates) + ] + raw_outputs = list(dict.fromkeys(raw_outputs)) + return [best_prompt] + raw_outputs raw_result = get_model_answer_extracted( - self.model, PARAPHRASE_PROMPT.format(prompt=best_prompt), n=n_candidates, temperature=0.9 + self.model, query, n=n_candidates, temperature=0.9 ) return [best_prompt] + [self._process_model_output(r) for r in raw_result] @@ -682,6 +702,8 @@ def optimize( dataset_split=None, evaluator=None, problem_description=None, + *, + use_structured_output: bool = False, **kwargs, ): """Run iterative HyPER optimization through the PromptTuner method API.""" @@ -717,6 +739,7 @@ def optimize( feedback_answer_tail_chars=feedback_answer_tail_chars, enable_instance_leak_audit=enable_instance_leak_audit, random_seed=random_seed, + use_structured_output=use_structured_output, ) meta_info = meta_info.copy() if meta_info else {} @@ -734,6 +757,8 @@ def run_configured_benchmark( self, ctx: BenchmarkContext, start_prompt: str, + *, + use_structured_output: bool = False, ) -> str: """Run HyPER from a benchmark context and method config.""" meta = dict(ctx.config.get("meta_info", {})) @@ -748,6 +773,7 @@ def run_configured_benchmark( dataset_split=ctx.dataset_split, evaluator=ctx.evaluator, problem_description=ctx.config.get("problem_description"), + use_structured_output=use_structured_output, meta_info=meta if meta else None, n_iterations=mc.get("n_iterations", 5), patience=mc.get("patience", None), diff --git a/coolprompt/optimizer/hyper/meta_prompt.py b/coolprompt/optimizer/hyper/meta_prompt.py index 10e28dd8..c1d7aab2 100644 --- a/coolprompt/optimizer/hyper/meta_prompt.py +++ b/coolprompt/optimizer/hyper/meta_prompt.py @@ -16,6 +16,9 @@ MetaPromptConfig, Recommendation, ) +from coolprompt.utils.structured_schemas.optimizer.hyper import ( + ResultPromptResponse, +) def _build_full_meta_prompt_template(builder: MetaPromptBuilder) -> str: @@ -28,11 +31,11 @@ def _build_full_meta_prompt_template(builder: MetaPromptBuilder) -> str: ) -class Optimizer(ABC): - """Abstract base for optimizers that consume a LangChain-compatible ``model``.""" - - def __init__(self, model: Any) -> None: - self.model = model +class Optimizer(ABC): + """Abstract base for optimizers that consume a LangChain-compatible ``model``.""" + + def __init__(self, model: Any) -> None: + self.model = model @abstractmethod def optimize(self, *args: Any, **kwargs: Any) -> Any: @@ -43,17 +46,17 @@ def optimize(self, *args: Any, **kwargs: Any) -> Any: class MetaPromptOptimizer(Optimizer): """Single-shot meta-prompt optimizer: one structured LLM call per ``optimize``.""" - def __init__( - self, - model: Any, - config: Optional[MetaPromptConfig] = None, - meta_prompt: Optional[str] = None, - use_structured_output: bool = False, - ) -> None: - """Initialize the meta-prompt builder and full prompt template.""" - super().__init__(model) - self.use_structured_output = use_structured_output - self.builder = MetaPromptBuilder(config) + def __init__( + self, + model: Any, + config: Optional[MetaPromptConfig] = None, + meta_prompt: Optional[str] = None, + use_structured_output: bool = False, + ) -> None: + """Initialize the meta-prompt builder and full prompt template.""" + super().__init__(model) + self.use_structured_output = use_structured_output + self.builder = MetaPromptBuilder(config) if meta_prompt is not None: self.meta_prompt = meta_prompt else: @@ -87,6 +90,16 @@ def optimize( ) -> Union[str, List[str]]: """Generate improved prompt(s) via the meta-prompt + LLM path.""" query = self._format_meta_prompt(prompt, **(meta_info or {})) + if self.use_structured_output: + structured = self.model.with_structured_output( + ResultPromptResponse, method="json_schema" + ) + if n_prompts == 1: + return structured.invoke(query).result_prompt + return [ + r.result_prompt + for r in structured.batch([query] * n_prompts) + ] raw_result = get_model_answer_extracted(self.model, query, n=n_prompts) if n_prompts == 1: return self._process_model_output(raw_result) @@ -117,49 +130,57 @@ def _process_model_output(self, output: Any) -> str: class HyPERLightMethod(AutoPromptingMethod): """Benchmark wrapper for :class:`MetaPromptOptimizer` (single LLM meta-prompt step).""" - def optimize( - self, - model, + def optimize( + self, + model, initial_prompt, dataset_split=None, evaluator=None, - problem_description=None, - **kwargs, - ): - """Run a single HyPER Light meta-prompt optimization call.""" - meta_info = kwargs.pop( - "meta_info", - kwargs.pop("hyper_meta_info", None), - ) - kwargs.setdefault("use_structured_output", False) - optimizer = MetaPromptOptimizer(model=model, **kwargs) - meta_info = meta_info.copy() if meta_info else {} - if "problem_description" not in meta_info: - meta_info["problem_description"] = problem_description + problem_description=None, + *, + use_structured_output: bool = False, + **kwargs, + ): + """Run a single HyPER Light meta-prompt optimization call.""" + meta_info = kwargs.pop( + "meta_info", + kwargs.pop("hyper_meta_info", None), + ) + optimizer = MetaPromptOptimizer( + model=model, + use_structured_output=use_structured_output, + **kwargs, + ) + meta_info = meta_info.copy() if meta_info else {} + if "problem_description" not in meta_info: + meta_info["problem_description"] = problem_description return optimizer.optimize( prompt=initial_prompt, meta_info=meta_info if meta_info else None, n_prompts=1, ) - def run_configured_benchmark( - self, - ctx: BenchmarkContext, - start_prompt: str, - ) -> str: - """Run HyPER Light from a benchmark context.""" - meta = dict(ctx.config.get("meta_info", {})) - return self.optimize( - ctx.model, - start_prompt, - problem_description=ctx.config.get("problem_description"), - meta_info=meta if meta else None, - ) - - def is_data_driven(self) -> bool: - return False - - @property - @override - def name(self) -> str: - return "hyper_light" + def run_configured_benchmark( + self, + ctx: BenchmarkContext, + start_prompt: str, + *, + use_structured_output: bool = False, + ) -> str: + """Run HyPER Light from a benchmark context.""" + meta = dict(ctx.config.get("meta_info", {})) + return self.optimize( + ctx.model, + start_prompt, + problem_description=ctx.config.get("problem_description"), + meta_info=meta if meta else None, + use_structured_output=use_structured_output, + ) + + def is_data_driven(self) -> bool: + return False + + @property + @override + def name(self) -> str: + return "hyper_light" diff --git a/coolprompt/optimizer/prompt_compressor/compressor.py b/coolprompt/optimizer/prompt_compressor/compressor.py index 76970b18..4bf09f63 100644 --- a/coolprompt/optimizer/prompt_compressor/compressor.py +++ b/coolprompt/optimizer/prompt_compressor/compressor.py @@ -103,9 +103,29 @@ def optimize( dataset_split=None, evaluator=None, problem_description=None, + *, + use_structured_output: bool = False, **kwargs, ): - """Compress ``initial_prompt`` through the shared method interface.""" + """Compress ``initial_prompt`` through the shared method interface. + + Note: + :class:`PromptCompressor` is intrinsically built on top of + ``with_structured_output`` and cannot operate without it. + The ``use_structured_output`` flag is accepted here for + interface uniformity with other methods, but passing ``False`` + raises ``NotImplementedError`` because the compressor does + not support a non-structured execution path. + + Raises: + NotImplementedError: If ``use_structured_output`` is ``False``. + """ + if not use_structured_output: + raise NotImplementedError( + "PromptCompressor is built on top of structured output " + "and cannot run with use_structured_output=False" + ) + compressor = PromptCompressor( model=model, system_prompt=self.system_prompt, @@ -127,6 +147,8 @@ def run_configured_benchmark( self, ctx: BenchmarkContext, start_prompt: str, + *, + use_structured_output: bool = False, ) -> str: """Run prompt compression from a benchmark context.""" mc = ctx.config.get("method", {}) @@ -135,7 +157,11 @@ def run_configured_benchmark( user_prompt=mc.get("user_prompt", self.user_prompt), return_metadata=mc.get("return_metadata", False), ) - return method.optimize(ctx.model, start_prompt) + return method.optimize( + ctx.model, + start_prompt, + use_structured_output=use_structured_output, + ) def is_data_driven(self) -> bool: return False diff --git a/coolprompt/optimizer/reflective_prompt/evoluter.py b/coolprompt/optimizer/reflective_prompt/evoluter.py index 46cb38a7..0f5e1aca 100644 --- a/coolprompt/optimizer/reflective_prompt/evoluter.py +++ b/coolprompt/optimizer/reflective_prompt/evoluter.py @@ -21,6 +21,14 @@ REFLECTIVEPROMPT_PROMPT_BY_DESCRIPTION_TEMPLATE, ) from coolprompt.utils.parsing import extract_answer, extract_json +from coolprompt.utils.structured_schemas.optimizer.reflective_prompt import ( + InitialPromptResponse, + ParaphrasedPromptsResponse, + ShortTermHintResponse, + LongTermHintResponse, + CrossoverPromptResponse, + MutatedPromptResponse, +) class ReflectiveEvoluter: @@ -45,6 +53,8 @@ class ReflectiveEvoluter: Defaults to 10. use_cache: a boolean variable. Either to use caching files or not. + use_structured_output: a boolean variable. + Either to use structured output or nor. output_path: a path to store logs of evolution. elitist: a prompt with highest score in population. best_score_overall: best evaluation score during evolution. @@ -72,6 +82,7 @@ def __init__( output_path: str = "./reflectiveprompt_outputs", checkpoint_path: Optional[str] = None, use_cache: bool = True, + use_structured_output: bool = False, ) -> None: """Initialize ReflectivePrompt state and search configuration.""" self.model = model @@ -87,6 +98,7 @@ def __init__( self.output_path = output_path self.initial_prompt = initial_prompt self.checkpoint_path = checkpoint_path + self.use_structured_output = use_structured_output self.elitist = None self._long_term_reflection_str = "" @@ -151,6 +163,11 @@ def _create_initial_prompt(self) -> str: request = REFLECTIVEPROMPT_PROMPT_BY_DESCRIPTION_TEMPLATE.format( PROBLEM_DESCRIPTION=self.problem_description ) + if self.use_structured_output: + structured = self.model.with_structured_output( + InitialPromptResponse, method="json_schema" + ) + return structured.invoke(request).prompt answer = self._llm_query([request])[0] return extract_answer( answer, self.PROMPT_TAGS, format_mismatch_label="" @@ -183,8 +200,14 @@ def _init_pop(self) -> List[Prompt]: request = REFLECTIVEPROMPT_PARAPHRASING_TEMPLATE.format( PROMPT=self.initial_prompt, NUM_PROMPTS=self.population_size ) - answer = self._llm_query([request])[0] - prompts = extract_json(answer)["prompts"] + if self.use_structured_output: + structured = self.model.with_structured_output( + ParaphrasedPromptsResponse, method="json_schema" + ) + prompts = structured.invoke(request).prompts + else: + answer = self._llm_query([request])[0] + prompts = extract_json(answer)["prompts"] initial_population = [ Prompt(prompt, origin=PromptOrigin.APE) for prompt in prompts ] @@ -352,11 +375,19 @@ def _short_term_reflection( worse_prompts.append(worse_prompt) better_prompts.append(better_prompt) - responses = self._llm_query(requests) - responses = [ - extract_answer(response, self.HINT_TAGS, format_mismatch_label="") - for response in responses - ] + if self.use_structured_output: + structured = self.model.with_structured_output( + ShortTermHintResponse, method="json_schema" + ) + responses = [r.hint for r in structured.batch(requests)] + else: + responses = self._llm_query(requests) + responses = [ + extract_answer( + response, self.HINT_TAGS, format_mismatch_label="" + ) + for response in responses + ] return responses, worse_prompts, better_prompts def _crossover( @@ -388,11 +419,19 @@ def _crossover( ) requests.append(request) - responses = self._llm_query(requests) - responses = [ - extract_answer(response, self.PROMPT_TAGS, format_mismatch_label="") - for response in responses - ] + if self.use_structured_output: + structured = self.model.with_structured_output( + CrossoverPromptResponse, method="json_schema" + ) + responses = [r.prompt for r in structured.batch(requests)] + else: + responses = self._llm_query(requests) + responses = [ + extract_answer( + response, self.PROMPT_TAGS, format_mismatch_label="" + ) + for response in responses + ] crossed_population = [Prompt(response) for response in responses] assert len(crossed_population) == self.population_size @@ -444,11 +483,16 @@ def _long_term_reflection(self, short_term_reflections: List[str]) -> None: NEW_SHORT_TERM_REFLECTIONS="\n".join(short_term_reflections), ) - response = self._llm_query([request])[0] - - self._long_term_reflection_str = extract_answer( - response, self.HINT_TAGS, format_mismatch_label="" - ) + if self.use_structured_output: + structured = self.model.with_structured_output( + LongTermHintResponse, method="json_schema" + ) + self._long_term_reflection_str = structured.invoke(request).hint + else: + response = self._llm_query([request])[0] + self._long_term_reflection_str = extract_answer( + response, self.HINT_TAGS, format_mismatch_label="" + ) def _llm_query(self, requests: List[str]) -> List[str]: """Provides api to query requests to the model. @@ -480,11 +524,22 @@ def _mutate(self) -> List[Prompt]: LONG_TERM_REFLECTION=self._long_term_reflection_str, ELITIST_PROMPT=self.elitist.text, ) - responses = self._llm_query([request] * self.population_size) - responses = [ - extract_answer(response, self.PROMPT_TAGS, format_mismatch_label="") - for response in responses - ] + if self.use_structured_output: + structured = self.model.with_structured_output( + MutatedPromptResponse, method="json_schema" + ) + responses = [ + r.prompt + for r in structured.batch([request] * self.population_size) + ] + else: + responses = self._llm_query([request] * self.population_size) + responses = [ + extract_answer( + response, self.PROMPT_TAGS, format_mismatch_label="" + ) + for response in responses + ] population = [ Prompt(response, origin=PromptOrigin.MUTATED) for response in responses diff --git a/coolprompt/optimizer/reflective_prompt/run.py b/coolprompt/optimizer/reflective_prompt/run.py index 33adac8b..855bf9eb 100644 --- a/coolprompt/optimizer/reflective_prompt/run.py +++ b/coolprompt/optimizer/reflective_prompt/run.py @@ -84,6 +84,8 @@ def optimize( dataset_split, evaluator, problem_description, + *, + use_structured_output: bool = False, **kwargs, ): """Run ReflectivePrompt through the shared method interface.""" @@ -93,6 +95,7 @@ def optimize( evaluator=evaluator, problem_description=problem_description, initial_prompt=initial_prompt, + use_structured_output=use_structured_output, **kwargs, ) @@ -100,21 +103,27 @@ def run_configured_benchmark( self, ctx: BenchmarkContext, start_prompt: str, + *, + use_structured_output: bool = False, ) -> str: """Run ReflectivePrompt from a benchmark context.""" problem_description = ctx.config.get("problem_description") + mc = ctx.config["method"] if problem_description is None: - generator = SyntheticDataGenerator(ctx._system_model) + generator = SyntheticDataGenerator( + ctx._system_model, + use_structured_output=use_structured_output, + ) problem_description = generator._generate_problem_description( prompt=start_prompt ) - mc = ctx.config["method"] return self.optimize( ctx.model, start_prompt, dataset_split=ctx.dataset_split, evaluator=ctx.evaluator, problem_description=problem_description, + use_structured_output=use_structured_output, population_size=mc.get("population_size", 10), num_epochs=mc.get("num_epochs", 5), output_path=mc.get("output_path", "./reflectiveprompt_outputs"), diff --git a/coolprompt/optimizer/regps/evoluter.py b/coolprompt/optimizer/regps/evoluter.py index a98b225f..03da230f 100644 --- a/coolprompt/optimizer/regps/evoluter.py +++ b/coolprompt/optimizer/regps/evoluter.py @@ -15,6 +15,11 @@ REGPS_TEXTUAL_GRADIENT_TEMPLATE, MUTATION_TEXTGRAD_TEMPLATE, ) +from coolprompt.utils.structured_schemas.optimizer.regps import ( + TextualGradientResponse, + ShortTermHintResponse, + MutatedPromptResponse, +) class ReGPSEvoluter(ReflectiveEvoluter): @@ -39,6 +44,8 @@ class ReGPSEvoluter(ReflectiveEvoluter): Defaults to 5. use_cache: a boolean variable. Either to use caching files or not. + use_structured_output: a boolean variable. + Either to use structured output or nor. output_path: a path to store logs of evolution. elitist: a prompt with highest score in population. best_score_overall: best evaluation score during evolution. @@ -66,6 +73,7 @@ def __init__( use_cache: bool = True, bad_examples_number: int = 5, checkpoint_path: Optional[str] = None, + use_structured_output: bool = False, ) -> None: """Initialize Re-GPS state and feedback-generation settings.""" super().__init__( @@ -82,6 +90,7 @@ def __init__( output_path, checkpoint_path, use_cache, + use_structured_output=use_structured_output, ) self.bad_examples_num = bad_examples_number @@ -146,6 +155,11 @@ def _gen_textual_gradient(self, prompt: Prompt) -> str: PROMPT=prompt.text, EXAMPLES=self._make_bad_examples(prompt.bad_examples), ) + if self.use_structured_output: + structured = self.model.with_structured_output( + TextualGradientResponse, method="json_schema" + ) + return structured.invoke(request).feedback return extract_answer( self._llm_query([request])[0], self.FEEDBACK_TAGS, @@ -230,11 +244,19 @@ def _short_term_reflection( feedbacks, self._make_output_path("feedbacks"), ) - responses = self._llm_query(requests) - responses = [ - extract_answer(response, self.HINT_TAGS, format_mismatch_label="") - for response in responses - ] + if self.use_structured_output: + structured = self.model.with_structured_output( + ShortTermHintResponse, method="json_schema" + ) + responses = [r.hint for r in structured.batch(requests)] + else: + responses = self._llm_query(requests) + responses = [ + extract_answer( + response, self.HINT_TAGS, format_mismatch_label="" + ) + for response in responses + ] return responses, worse_prompts, better_prompts def _mutate(self) -> List[Prompt]: @@ -254,11 +276,22 @@ def _mutate(self) -> List[Prompt]: ELITIST_PROMPT=self.elitist.text, FEEDBACK=feedback, ) - responses = self._llm_query([request] * self.population_size) - responses = [ - extract_answer(response, self.PROMPT_TAGS, format_mismatch_label="") - for response in responses - ] + if self.use_structured_output: + structured = self.model.with_structured_output( + MutatedPromptResponse, method="json_schema" + ) + responses = [ + r.prompt + for r in structured.batch([request] * self.population_size) + ] + else: + responses = self._llm_query([request] * self.population_size) + responses = [ + extract_answer( + response, self.PROMPT_TAGS, format_mismatch_label="" + ) + for response in responses + ] population = [ Prompt(response, origin=PromptOrigin.MUTATED) for response in responses diff --git a/coolprompt/optimizer/regps/run.py b/coolprompt/optimizer/regps/run.py index e5e8f3a5..08591b79 100644 --- a/coolprompt/optimizer/regps/run.py +++ b/coolprompt/optimizer/regps/run.py @@ -19,6 +19,7 @@ def regps( evaluator: Evaluator, problem_description: str, initial_prompt: Optional[str] = None, + use_structured_output: bool = False, **kwargs, ) -> str: """Runs Re-GPS evolution. @@ -33,6 +34,7 @@ def regps( short description of problem to optimize. initial_prompt (str, optional): initial prompt to start evolution from. Defaults to None. + use_structured_output (bool): either use structured output or not. **kwargs (dict[str, Any]): other parameters (such as population_size, num_epochs, output_path, use_cache). @@ -59,6 +61,7 @@ def regps( validation_targets=validation_targets, problem_description=problem_description, initial_prompt=initial_prompt, + use_structured_output=use_structured_output, population_size=args["population_size"], num_epochs=args["num_epochs"], output_path=args["output_path"], @@ -84,6 +87,8 @@ def optimize( dataset_split, evaluator, problem_description, + *, + use_structured_output: bool = False, **kwargs, ): """Run Re-GPS through the shared method interface.""" @@ -93,6 +98,7 @@ def optimize( evaluator=evaluator, problem_description=problem_description, initial_prompt=initial_prompt, + use_structured_output=use_structured_output, **kwargs, ) @@ -100,11 +106,17 @@ def run_configured_benchmark( self, ctx: BenchmarkContext, start_prompt: str, + *, + use_structured_output: bool = False, ) -> str: """Run Re-GPS from a benchmark context.""" problem_description = ctx.config.get("problem_description") + mc = ctx.config["method"] if problem_description is None: - generator = SyntheticDataGenerator(ctx._system_model) + generator = SyntheticDataGenerator( + ctx._system_model, + use_structured_output=use_structured_output, + ) indices = sample(range(0, len(ctx.dataset_split[0])), 5) examples = [ (ctx.dataset_split[0][ind], ctx.dataset_split[2][ind]) @@ -113,13 +125,13 @@ def run_configured_benchmark( problem_description = generator._generate_problem_description( prompt=start_prompt, examples=examples ) - mc = ctx.config["method"] return self.optimize( ctx.model, start_prompt, dataset_split=ctx.dataset_split, evaluator=ctx.evaluator, problem_description=problem_description, + use_structured_output=use_structured_output, population_size=mc.get("population_size", 10), num_epochs=mc.get("num_epochs", 5), output_path=mc.get("output_path", "./regps_outputs"), diff --git a/coolprompt/task_detector/detector.py b/coolprompt/task_detector/detector.py index 43a29870..287634fa 100644 --- a/coolprompt/task_detector/detector.py +++ b/coolprompt/task_detector/detector.py @@ -1,12 +1,11 @@ from typing import Any from langchain_core.language_models.base import BaseLanguageModel -from langchain_core.language_models.chat_models import BaseChatModel from langchain_core.messages.ai import AIMessage from pydantic import BaseModel -from coolprompt.task_detector.pydantic_formatters import ( - TaskDetectionStructuredOutputSchema, +from coolprompt.utils.structured_schemas.task_detector import ( + TaskDetectionResponse, ) from coolprompt.utils.prompt_templates.task_detector_templates import ( TASK_DETECTOR_TEMPLATE, @@ -21,50 +20,53 @@ class TaskDetector: Attributes: model: langchain.BaseLanguageModel class of model to use. - """ - - def __init__(self, model: BaseLanguageModel) -> None: - self.model = model + use_structured_output: if True, the LLM is queried via + ``model.with_structured_output(..., method="json_schema")`` + using the dedicated pydantic schema; otherwise a plain + ``invoke()`` is performed and the JSON payload is parsed + from the raw text response. + """ + + def __init__( + self, + model: BaseLanguageModel, + use_structured_output: bool = False, + ) -> None: + self.model = model + self.use_structured_output = use_structured_output def _generate( self, request: str, schema: BaseModel, field_name: str ) -> Any: - """Generates model output - either using structured output from langchain - or just strict json output format for LLM + """Generates model output either using structured output from + langchain (when ``self.use_structured_output`` is True) or a + plain ``invoke()`` call combined with JSON extraction from text. Args: request (str): request to LLM - when langchain structured output is used - schema (BaseModel): Pydantic output format + schema (BaseModel): Pydantic output format (only used when + structured output is enabled) field_name (str): field name to select from output Returns: Any: generated data """ - if hasattr(self.model, "model"): - wrapped_model = self.model.model - else: - wrapped_model = self.model - - if not isinstance(wrapped_model, BaseChatModel): - output = self.model.invoke(request) + if self.use_structured_output: + structured_model = self.model.with_structured_output( + schema=schema, method="json_schema" + ) + output = structured_model.invoke(request) if isinstance(output, AIMessage): output = output.content - return extract_json(output)[field_name] + try: + return getattr(output, field_name) + except Exception: + return output[field_name] - structured_model = self.model.with_structured_output( - schema=schema, method="json_schema" - ) - output = structured_model.invoke(request) + output = self.model.invoke(request) if isinstance(output, AIMessage): output = output.content - - try: - output = getattr(output, field_name) - except Exception: - output = output[field_name] - return output + return extract_json(output)[field_name] def generate( self, @@ -78,10 +80,8 @@ def generate( Returns: str: task class """ - schema = TaskDetectionStructuredOutputSchema - request = TASK_DETECTOR_TEMPLATE - - request = request.format(query=prompt) + schema = TaskDetectionResponse + request = TASK_DETECTOR_TEMPLATE.format(query=prompt) logger.info("Detecting the task by query") diff --git a/coolprompt/task_detector/pydantic_formatters.py b/coolprompt/task_detector/pydantic_formatters.py index b2575f81..e2171160 100644 --- a/coolprompt/task_detector/pydantic_formatters.py +++ b/coolprompt/task_detector/pydantic_formatters.py @@ -1,7 +1,5 @@ -from pydantic import BaseModel, Field +from coolprompt.utils.structured_schemas.task_detector import ( + TaskDetectionResponse as TaskDetectionStructuredOutputSchema, +) - -class TaskDetectionStructuredOutputSchema(BaseModel): - """Structured response containing the detected CoolPrompt task type.""" - - task: str = Field(description="Determined task classification") +__all__ = ["TaskDetectionStructuredOutputSchema"] diff --git a/coolprompt/utils/correction/rule.py b/coolprompt/utils/correction/rule.py index c23443fc..af0db9bb 100644 --- a/coolprompt/utils/correction/rule.py +++ b/coolprompt/utils/correction/rule.py @@ -1,6 +1,7 @@ from abc import ABC from typing import Any from langchain_core.language_models.base import BaseLanguageModel +from langchain_core.messages.ai import AIMessage from coolprompt.utils.prompt_templates.correction_templates import ( TRANSLATION_TEMPLATE, ) @@ -10,6 +11,7 @@ get_model_answer_extracted, safe_template, ) +from coolprompt.utils.structured_schemas.correction import TranslationResponse class Rule(ABC): @@ -54,9 +56,21 @@ class LanguageRule(Rule): """The rule which checks if the final prompt and the start prompt are in the same languages.""" - def __init__(self, llm: BaseLanguageModel) -> None: - """Initializes with LangChain language model.""" + def __init__( + self, llm: BaseLanguageModel, use_structured_output: bool = False + ) -> None: + """Initializes with LangChain language model. + + Args: + llm (BaseLanguageModel): LangChain language model. + use_structured_output (bool): if True, both language detection and + translation are performed via + ``llm.with_structured_output(...)`` using the dedicated + Pydantic schemas; otherwise plain ``invoke()`` calls with + JSON extraction from raw text are used. + """ self.llm = llm + self.use_structured_output = use_structured_output @property def is_guaranteed_after_first_fix(self): @@ -76,8 +90,12 @@ def check( and meta data with the target language. """ - start_prompt_lang = detect_language(start_prompt, self.llm) - final_prompt_lang = detect_language(final_prompt, self.llm) + start_prompt_lang = detect_language( + start_prompt, self.llm, self.use_structured_output + ) + final_prompt_lang = detect_language( + final_prompt, self.llm, self.use_structured_output + ) if start_prompt_lang != final_prompt_lang: return False, { @@ -104,6 +122,18 @@ def fix(self, final_prompt: str, meta: dict[str, Any]) -> str: to_lang=meta["to_lang"], ) + if self.use_structured_output: + structured_model = self.llm.with_structured_output( + schema=TranslationResponse, method="json_schema" + ) + output = structured_model.invoke(prompt) + if isinstance(output, AIMessage): + output = output.content + try: + return output.translated_text + except Exception: + return output["translated_text"] + answer = get_model_answer_extracted(self.llm, prompt) result = extract_json(answer) diff --git a/coolprompt/utils/language_detection.py b/coolprompt/utils/language_detection.py index 3982e5db..325e207b 100644 --- a/coolprompt/utils/language_detection.py +++ b/coolprompt/utils/language_detection.py @@ -7,22 +7,43 @@ get_model_answer_extracted, safe_template, ) +from coolprompt.utils.structured_schemas.correction import LanguageDetectionResponse from langchain_core.language_models.base import BaseLanguageModel +from langchain_core.messages.ai import AIMessage -def detect_language(text: str, llm: BaseLanguageModel) -> str: +def detect_language( + text: str, llm: BaseLanguageModel, use_structured_output: bool = False +) -> str: """Detects the provided text's language using the LangChain language model. Args: text (str): text for language detection. llm (BaseLanguageModel): LangChain language model. + use_structured_output (bool): if True, the LLM is queried via + ``llm.with_structured_output(...)`` using + :class:`~coolprompt.utils.structured_schemas.correction.LanguageDetectionResponse`; + otherwise a plain ``invoke()`` is performed and the JSON payload + is parsed from the raw text response. Returns: str: `text`'s language code in ISO 639-1 format. """ prompt = safe_template(LANGUAGE_DETECTION_TEMPLATE, text=text) + if use_structured_output: + structured_model = llm.with_structured_output( + schema=LanguageDetectionResponse, method="json_schema" + ) + output = structured_model.invoke(prompt) + if isinstance(output, AIMessage): + output = output.content + try: + return output.language_code + except Exception: + return output["language_code"] + answer = get_model_answer_extracted(llm, prompt) result = extract_json(answer) diff --git a/coolprompt/utils/prompt_templates/default_templates.py b/coolprompt/utils/prompt_templates/default_templates.py index 680348c2..ae8bcfea 100644 --- a/coolprompt/utils/prompt_templates/default_templates.py +++ b/coolprompt/utils/prompt_templates/default_templates.py @@ -18,3 +18,24 @@ RESPONSE: """ + +CLASSIFICATION_TASK_TEMPLATE_STRUCTURED = """{PROMPT} + +Answer using the label from [{LABELS}]. +Return the chosen label in the `answer` field of the structured response. + +Input: +{INPUT} + +Response: +""" + +GENERATION_TASK_TEMPLATE_STRUCTURED = """{PROMPT} + +Return the final answer in the `answer` field of the structured response. + +INPUT: +{INPUT} + +RESPONSE: +""" diff --git a/coolprompt/utils/structured_schemas/__init__.py b/coolprompt/utils/structured_schemas/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/coolprompt/utils/structured_schemas/correction/__init__.py b/coolprompt/utils/structured_schemas/correction/__init__.py new file mode 100644 index 00000000..95adf0d5 --- /dev/null +++ b/coolprompt/utils/structured_schemas/correction/__init__.py @@ -0,0 +1,6 @@ +from coolprompt.utils.structured_schemas.correction.schemas import ( + LanguageDetectionResponse, + TranslationResponse, +) + +__all__ = ["LanguageDetectionResponse", "TranslationResponse"] diff --git a/coolprompt/utils/structured_schemas/correction/schemas.py b/coolprompt/utils/structured_schemas/correction/schemas.py new file mode 100644 index 00000000..c68bf89f --- /dev/null +++ b/coolprompt/utils/structured_schemas/correction/schemas.py @@ -0,0 +1,44 @@ +from pydantic import BaseModel, Field + + +class LanguageDetectionResponse(BaseModel): + """Response schema for language detection. + + Used by :func:`coolprompt.utils.language_detection.detect_language` + to obtain a strict, single-field structured output from the LLM + when ``use_structured_output=True``. + + Mirrors the JSON contract from ``LANGUAGE_DETECTION_TEMPLATE``: + ``{"language_code": "XX"}`` or ``{"language_code": "XX-YY"}``. + """ + + language_code: str = Field( + description=( + "ISO 639-1 language code of the detected text " + "(e.g. 'en', 'ru', 'zh-CN', 'pt-BR'). " + "Use 5-character regional codes when the region is clearly " + "specified or culturally important; otherwise use 2-character codes." + ) + ) + + +class TranslationResponse(BaseModel): + """Response schema for prompt translation. + + Used by :class:`coolprompt.utils.correction.rule.LanguageRule` + to obtain a strict, single-field structured output from the LLM + when ``use_structured_output=True``. + + Mirrors the JSON contract from ``TRANSLATION_TEMPLATE``: + ``{"translated_text": ""}``. + """ + + translated_text: str = Field( + description=( + "The full translated text in the target language. " + "All original formatting, spacing, punctuation, and line breaks " + "must be preserved. Code blocks, variables, function names, URLs, " + "technical terms, proper names, and any text already in the target " + "language must not be translated." + ) + ) diff --git a/coolprompt/utils/structured_schemas/data_generator/__init__.py b/coolprompt/utils/structured_schemas/data_generator/__init__.py new file mode 100644 index 00000000..0889db5f --- /dev/null +++ b/coolprompt/utils/structured_schemas/data_generator/__init__.py @@ -0,0 +1,15 @@ +from coolprompt.utils.structured_schemas.data_generator.schemas import ( + ClassificationTaskExample, + ClassificationTaskResponse, + GenerationTaskExample, + GenerationTaskResponse, + ProblemDescriptionResponse, +) + +__all__ = [ + "ClassificationTaskExample", + "ClassificationTaskResponse", + "GenerationTaskExample", + "GenerationTaskResponse", + "ProblemDescriptionResponse", +] diff --git a/coolprompt/utils/structured_schemas/data_generator/schemas.py b/coolprompt/utils/structured_schemas/data_generator/schemas.py new file mode 100644 index 00000000..833e9061 --- /dev/null +++ b/coolprompt/utils/structured_schemas/data_generator/schemas.py @@ -0,0 +1,75 @@ +from typing import List + +from pydantic import BaseModel, Field + + +class ProblemDescriptionResponse(BaseModel): + """Response schema for synthetic problem description generation. + + Used by :meth:`coolprompt.data_generator.generator. + SyntheticDataGenerator._generate_problem_description` to obtain a + textual description of the task that the user's initial prompt was + created to solve. + """ + + problem_description: str = Field( + description=( + "Detailed textual problem description for which the user's " + "prompt was created." + ) + ) + + +class ClassificationTaskExample(BaseModel): + """A single (input, output) example for a classification task.""" + + input: str = Field( + description=( + "Textual input for the classification task. Must contain all " + "data required to predict the label; if answer choices are " + "part of the task, concatenate them into the input string." + ) + ) + output: str = Field( + description=( + "Textual ground-truth label corresponding to the input." + ) + ) + + +class ClassificationTaskResponse(BaseModel): + """Response schema for classification dataset synthesis.""" + + examples: List[ClassificationTaskExample] = Field( + description=( + "List of synthetic classification examples. Try to make the " + "answer distribution as random as possible." + ) + ) + + +class GenerationTaskExample(BaseModel): + """A single (input, output) example for a generation task.""" + + input: str = Field( + description=( + "Textual input for the generation task. Must contain all " + "data required to produce the expected output." + ) + ) + output: str = Field( + description=( + "Textual correct model output corresponding to the input." + ) + ) + + +class GenerationTaskResponse(BaseModel): + """Response schema for generation dataset synthesis.""" + + examples: List[GenerationTaskExample] = Field( + description=( + "List of synthetic input-output examples for the generation " + "task." + ) + ) diff --git a/coolprompt/utils/structured_schemas/evaluator/__init__.py b/coolprompt/utils/structured_schemas/evaluator/__init__.py new file mode 100644 index 00000000..89089110 --- /dev/null +++ b/coolprompt/utils/structured_schemas/evaluator/__init__.py @@ -0,0 +1,11 @@ +from coolprompt.utils.structured_schemas.evaluator.schemas import ( + ClassificationAnswerResponse, + GenerationAnswerResponse, + JudgeScoreResponse, +) + +__all__ = [ + "ClassificationAnswerResponse", + "GenerationAnswerResponse", + "JudgeScoreResponse", +] diff --git a/coolprompt/utils/structured_schemas/evaluator/schemas.py b/coolprompt/utils/structured_schemas/evaluator/schemas.py new file mode 100644 index 00000000..3709f86f --- /dev/null +++ b/coolprompt/utils/structured_schemas/evaluator/schemas.py @@ -0,0 +1,19 @@ +from pydantic import BaseModel, Field + + +class ClassificationAnswerResponse(BaseModel): + """Response schema for classification-task answers.""" + + answer: str = Field(description="The chosen label for the given input.") + + +class GenerationAnswerResponse(BaseModel): + """Response schema for free-form generation-task answers.""" + + answer: str = Field(description="The final answer to the task.") + + +class JudgeScoreResponse(BaseModel): + """Response schema for the LLM-as-a-judge metric score.""" + + score: int = Field(description="Integer score for the requested criterion.") diff --git a/coolprompt/utils/structured_schemas/optimizer/hyper/__init__.py b/coolprompt/utils/structured_schemas/optimizer/hyper/__init__.py new file mode 100644 index 00000000..4b3e748e --- /dev/null +++ b/coolprompt/utils/structured_schemas/optimizer/hyper/__init__.py @@ -0,0 +1,23 @@ +from coolprompt.utils.structured_schemas.optimizer.hyper.meta_prompt_schemas import ( + ResultPromptResponse, + ParaphrasedVariantResponse, +) +from coolprompt.utils.structured_schemas.optimizer.hyper.feedback_schemas import ( + SectionRecommendationResponse, + RecommendationGroupsResponse, + SynthesizedRecommendationItem, + SynthesizedRecommendationsResponse, + InstanceLeakVerdict, + InstanceLeakAuditResponse, +) + +__all__ = [ + "ResultPromptResponse", + "ParaphrasedVariantResponse", + "SectionRecommendationResponse", + "RecommendationGroupsResponse", + "SynthesizedRecommendationItem", + "SynthesizedRecommendationsResponse", + "InstanceLeakVerdict", + "InstanceLeakAuditResponse", +] diff --git a/coolprompt/utils/structured_schemas/optimizer/hyper/feedback_schemas.py b/coolprompt/utils/structured_schemas/optimizer/hyper/feedback_schemas.py new file mode 100644 index 00000000..ed56209e --- /dev/null +++ b/coolprompt/utils/structured_schemas/optimizer/hyper/feedback_schemas.py @@ -0,0 +1,70 @@ +from typing import List, Literal + +from pydantic import BaseModel, Field + + +class SectionRecommendationResponse(BaseModel): + """Response schema for a single section-targeted recommendation + (used for both regular and contrastive feedback).""" + + section: str = Field( + description="Target section name for the recommendation, or 'general'." + ) + text: str = Field( + description="The recommendation text." + ) + + +class RecommendationGroupsResponse(BaseModel): + """Response schema for grouping recommendations by semantic similarity.""" + + groups: List[List[int]] = Field( + description=( + "Partition of input recommendation ids into groups. " + "Each inner list contains the zero-based ids belonging to one group." + ) + ) + + +class SynthesizedRecommendationItem(BaseModel): + """A single synthesized recommendation derived from a cluster.""" + + text: str = Field( + description="The synthesized recommendation text for the group." + ) + weight: int = Field( + ge=1, + description=( + "Number of original recommendations represented by this synthesized item." + ), + ) + + +class SynthesizedRecommendationsResponse(BaseModel): + """Response schema for the per-section synthesis/filter step.""" + + synthesized: List[SynthesizedRecommendationItem] = Field( + description="Synthesized recommendations for the section." + ) + + +class InstanceLeakVerdict(BaseModel): + """A single audit verdict for one recommendation.""" + + verdict: Literal["KEEP", "REWRITE", "DROP"] = Field( + description="Audit verdict for the recommendation." + ) + text: str = Field( + default="", + description="Rewritten recommendation; used only when verdict is 'REWRITE'.", + ) + + +class InstanceLeakAuditResponse(BaseModel): + """Response schema for the instance-leak audit pass over recommendations.""" + + verdicts: List[InstanceLeakVerdict] = Field( + description=( + "One verdict per input recommendation, in the same order as the input." + ) + ) diff --git a/coolprompt/utils/structured_schemas/optimizer/hyper/meta_prompt_schemas.py b/coolprompt/utils/structured_schemas/optimizer/hyper/meta_prompt_schemas.py new file mode 100644 index 00000000..b9e49235 --- /dev/null +++ b/coolprompt/utils/structured_schemas/optimizer/hyper/meta_prompt_schemas.py @@ -0,0 +1,17 @@ +from pydantic import BaseModel, Field + + +class ResultPromptResponse(BaseModel): + """Response schema for HyPER meta-prompt single-step optimization.""" + + result_prompt: str = Field( + description="The optimized prompt produced by the meta-prompt." + ) + + +class ParaphrasedVariantResponse(BaseModel): + """Response schema for paraphrasing the current best HyPER prompt.""" + + paraphrased_prompt: str = Field( + description="A paraphrased variant of the input prompt." + ) diff --git a/coolprompt/utils/structured_schemas/optimizer/reflective_prompt/__init__.py b/coolprompt/utils/structured_schemas/optimizer/reflective_prompt/__init__.py new file mode 100644 index 00000000..592677c4 --- /dev/null +++ b/coolprompt/utils/structured_schemas/optimizer/reflective_prompt/__init__.py @@ -0,0 +1,17 @@ +from coolprompt.utils.structured_schemas.optimizer.reflective_prompt.schemas import ( + InitialPromptResponse, + ParaphrasedPromptsResponse, + ShortTermHintResponse, + LongTermHintResponse, + CrossoverPromptResponse, + MutatedPromptResponse, +) + +__all__ = [ + "InitialPromptResponse", + "ParaphrasedPromptsResponse", + "ShortTermHintResponse", + "LongTermHintResponse", + "CrossoverPromptResponse", + "MutatedPromptResponse", +] diff --git a/coolprompt/utils/structured_schemas/optimizer/reflective_prompt/schemas.py b/coolprompt/utils/structured_schemas/optimizer/reflective_prompt/schemas.py new file mode 100644 index 00000000..117a62b5 --- /dev/null +++ b/coolprompt/utils/structured_schemas/optimizer/reflective_prompt/schemas.py @@ -0,0 +1,62 @@ +from typing import List +from pydantic import BaseModel, Field + + +class InitialPromptResponse(BaseModel): + """Response schema for initial-prompt generation from a problem description.""" + + prompt: str = Field( + description="A prompt that effectively solves the described task." + ) + + +class ParaphrasedPromptsResponse(BaseModel): + """Response schema for paraphrasing the initial prompt into a population.""" + + prompts: List[str] = Field( + description=( + "New variations of the original prompt keeping its initial meaning." + ) + ) + + +class ShortTermHintResponse(BaseModel): + """Response schema for short-term reflection hints.""" + + hint: str = Field( + description=( + "One small hint for designing better prompts, based on the two " + "prompt versions, using less than 20 words." + ) + ) + + +class LongTermHintResponse(BaseModel): + """Response schema for long-term reflection hints.""" + + hint: str = Field( + description=( + "One constructive hint for designing better prompts, based on " + "prior reflections and new insights, using less than 50 words." + ) + ) + + +class CrossoverPromptResponse(BaseModel): + """Response schema for crossover-stage improved prompts.""" + + prompt: str = Field( + description=( + "An improved prompt for the task, written according to the reflection." + ) + ) + + +class MutatedPromptResponse(BaseModel): + """Response schema for mutation-stage prompts.""" + + prompt: str = Field( + description=( + "A mutated prompt for the task, written according to the prior reflection." + ) + ) diff --git a/coolprompt/utils/structured_schemas/optimizer/regps/__init__.py b/coolprompt/utils/structured_schemas/optimizer/regps/__init__.py new file mode 100644 index 00000000..bf257d25 --- /dev/null +++ b/coolprompt/utils/structured_schemas/optimizer/regps/__init__.py @@ -0,0 +1,11 @@ +from coolprompt.utils.structured_schemas.optimizer.regps.schemas import ( + TextualGradientResponse, + ShortTermHintResponse, + MutatedPromptResponse, +) + +__all__ = [ + "TextualGradientResponse", + "ShortTermHintResponse", + "MutatedPromptResponse", +] diff --git a/coolprompt/utils/structured_schemas/optimizer/regps/schemas.py b/coolprompt/utils/structured_schemas/optimizer/regps/schemas.py new file mode 100644 index 00000000..fe8205ca --- /dev/null +++ b/coolprompt/utils/structured_schemas/optimizer/regps/schemas.py @@ -0,0 +1,43 @@ +from pydantic import BaseModel, Field + + +class TextualGradientResponse(BaseModel): + """Response schema for textual-gradient feedback generation.""" + + feedback: str = Field( + description=( + "Detailed feedback on how the given prompt can be improved " + "to achieve the best quality answer on the given problem " + "description and not to repeat the same mistakes observed " + "in the provided failed examples." + ) + ) + + +class ShortTermHintResponse(BaseModel): + """Response schema for RE-GPS short-term reflection hints.""" + + hint: str = Field( + description=( + "One new hint for designing a better prompt, derived from " + "the two provided prompt versions (worse and better) and " + "their respective improvement feedbacks. For example, the " + "hint can recommend a word replacement, an active/positive " + "voice conversion, adding a word or deleting a word." + ) + ) + + +class MutatedPromptResponse(BaseModel): + """Response schema for RE-GPS elitist mutation prompts.""" + + prompt: str = Field( + description=( + "A mutated prompt derived from the elitist prompt. The main " + "priority is the provided prior (long-term) reflection, which " + "accumulates essential information about correct prompt " + "structure and other prompt features; the improvement " + "feedback generated by the expert for the elitist prompt is " + "also taken into account." + ) + ) diff --git a/coolprompt/utils/structured_schemas/task_detector/__init__.py b/coolprompt/utils/structured_schemas/task_detector/__init__.py new file mode 100644 index 00000000..e528a1bd --- /dev/null +++ b/coolprompt/utils/structured_schemas/task_detector/__init__.py @@ -0,0 +1,5 @@ +from coolprompt.utils.structured_schemas.task_detector.schemas import ( + TaskDetectionResponse, +) + +__all__ = ["TaskDetectionResponse"] diff --git a/coolprompt/utils/structured_schemas/task_detector/schemas.py b/coolprompt/utils/structured_schemas/task_detector/schemas.py new file mode 100644 index 00000000..0971bf3a --- /dev/null +++ b/coolprompt/utils/structured_schemas/task_detector/schemas.py @@ -0,0 +1,12 @@ +from pydantic import BaseModel, Field + + +class TaskDetectionResponse(BaseModel): + """Response schema for task-type classification. + + Used by :class:`coolprompt.task_detector.detector.TaskDetector` + to obtain a strict, single-field structured output from the LLM + when ``use_structured_output=True``. + """ + + task: str = Field(description="The name of the detected task.") diff --git a/docs/API.md b/docs/API.md index e00eccdb..b1534c9c 100644 --- a/docs/API.md +++ b/docs/API.md @@ -68,3 +68,4 @@ Foundational utilities. Can be useful if you want to dive deeper in our project. --- +