fix: added back the utils functions

Author: heyjustinai

src/prompt_ops/core/utils/strategy_utils.py

@@ -0,0 +1,25 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+# All rights reserved.
+#
+# This source code is licensed under the terms described in the LICENSE file in
+# the root directory of this source tree.
+"""
+Utility functions for working with optimization strategies.
+"""
+
+from typing import Literal, Optional
+
+
+def map_auto_mode_to_dspy(
+    auto_mode: Optional[Literal["basic", "intermediate", "advanced"]],
+) -> str:
+    """Map our naming convention to DSPy's expected values.
+
+    Args:
+        auto_mode: Our naming convention ('basic', 'intermediate', 'advanced')
+
+    Returns:
+        The corresponding DSPy auto mode ('light', 'medium', 'heavy')
+    """
+    mapping = {"basic": "light", "intermediate": "medium", "advanced": "heavy"}
+    return mapping.get(auto_mode, "light")  # Default to light if not found

src/prompt_ops/core/utils/summary_utils.py

@@ -0,0 +1,120 @@
+"""
+Summary utilities for creating and managing optimization summaries.
+
+This module provides utility functions for creating pre-optimization summaries
+in a clean, testable, and reusable way.
+"""
+
+import logging
+from typing import Any, Dict, Optional
+
+from .telemetry import PreOptimizationSummary
+
+
+def create_pre_optimization_summary(
+    strategy, prompt_data: Dict[str, Any]
+) -> PreOptimizationSummary:
+    """
+    Create a pre-optimization summary from strategy data.
+
+    This utility function extracts summary creation logic from strategy classes,
+    making it testable in isolation and reusable across different strategies.
+
+    Args:
+        strategy: The optimization strategy instance (BaseStrategy or subclass)
+        prompt_data: The prompt data being optimized
+
+    Returns:
+        PreOptimizationSummary instance ready for display/logging
+    """
+    # Import here to avoid circular imports
+    from ..prompt_strategies import map_auto_mode_to_dspy
+
+    # Collect guidance information
+    guidance = None
+    if (
+        hasattr(strategy, "proposer_kwargs")
+        and strategy.proposer_kwargs
+        and "tip" in strategy.proposer_kwargs
+    ):
+        guidance = strategy.proposer_kwargs["tip"]
+
+    # Compute baseline score if enabled
+    baseline_score = None
+    if getattr(strategy, "compute_baseline", False):
+        try:
+            if hasattr(strategy, "_compute_baseline_score"):
+                baseline_score = strategy._compute_baseline_score(prompt_data)
+        except Exception as e:
+            logging.warning(f"Failed to compute baseline score: {e}")
+
+    # Get model names using the strategy's method
+    task_model_name = "Unknown"
+    proposer_model_name = "Unknown"
+
+    if hasattr(strategy, "_get_model_name"):
+        if hasattr(strategy, "task_model"):
+            task_model_name = strategy._get_model_name(strategy.task_model)
+        if hasattr(strategy, "prompt_model"):
+            proposer_model_name = strategy._get_model_name(strategy.prompt_model)
+
+    # Get metric name
+    metric_name = "None"
+    if hasattr(strategy, "metric") and strategy.metric:
+        metric_name = getattr(strategy.metric, "__name__", str(strategy.metric))
+
+    # Collect MIPRO parameters with safe defaults
+    auto_mode = getattr(strategy, "auto", "basic")
+    mipro_params = {
+        "auto_user": auto_mode,
+        "auto_dspy": map_auto_mode_to_dspy(auto_mode),
+        "max_labeled_demos": getattr(strategy, "max_labeled_demos", 5),
+        "max_bootstrapped_demos": getattr(strategy, "max_bootstrapped_demos", 4),
+        "num_candidates": getattr(strategy, "num_candidates", 10),
+        "num_threads": getattr(strategy, "num_threads", 18),
+        "init_temperature": getattr(strategy, "init_temperature", 0.5),
+        "seed": getattr(strategy, "seed", 9),
+    }
+
+    return PreOptimizationSummary(
+        task_model=task_model_name,
+        proposer_model=proposer_model_name,
+        metric_name=metric_name,
+        train_size=len(getattr(strategy, "trainset", []) or []),
+        val_size=len(getattr(strategy, "valset", []) or []),
+        mipro_params=mipro_params,
+        guidance=guidance,
+        baseline_score=baseline_score,
+    )
+
+
+def create_and_display_summary(
+    strategy, prompt_data: Dict[str, Any]
+) -> PreOptimizationSummary:
+    """
+    Convenience function to create and display a pre-optimization summary.
+
+    Args:
+        strategy: The optimization strategy instance
+        prompt_data: The prompt data being optimized
+
+    Returns:
+        The created PreOptimizationSummary instance
+    """
+    try:
+        summary = create_pre_optimization_summary(strategy, prompt_data)
+        summary.log()
+        return summary
+    except Exception as e:
+        logging.warning(
+            f"Failed to create or display pre-optimization summary: {str(e)}"
+        )
+        # Return a minimal summary to avoid breaking the optimization flow
+        return PreOptimizationSummary(
+            task_model="Unknown",
+            proposer_model="Unknown",
+            metric_name="Unknown",
+            train_size=0,
+            val_size=0,
+            mipro_params={},
+        )

src/prompt_ops/core/utils/telemetry.py

@@ -0,0 +1,83 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+# All rights reserved.
+#
+# This source code is licensed under the terms described in the LICENSE file in
+# the root directory of this source tree.
+"""
+Telemetry module for tracking and displaying optimization process information.
+
+This module provides classes and utilities for collecting and displaying
+key information about the optimization process before it begins.
+"""
+
+import json
+from dataclasses import asdict, dataclass
+from typing import Any, Dict, Optional
+
+from .logging import get_logger
+
+
+@dataclass
+class PreOptimizationSummary:
+    """
+    Container for pre-optimization summary information.
+
+    This class collects and formats key information about the optimization
+    process that will be displayed to users before optimization begins.
+    """
+
+    task_model: str
+    proposer_model: str
+    metric_name: str
+    train_size: int
+    val_size: int
+    mipro_params: Dict[str, Any]
+    guidance: Optional[str] = None
+    baseline_score: Optional[float] = None
+
+    def to_pretty(self) -> str:
+        """
+        Format the summary as a human-readable string.
+
+        Returns:
+            A formatted string suitable for console display
+        """
+        pad = " " * 4
+        lines = [
+            "=== Pre-Optimization Summary ===",
+            f"{pad}Task Model       : {self.task_model}",
+            f"{pad}Proposer Model   : {self.proposer_model}",
+            f"{pad}Metric           : {self.metric_name}",
+            f"{pad}Train / Val size : {self.train_size} / {self.val_size}",
+            f"{pad}MIPRO Params     : {json.dumps(self.mipro_params, separators=(',', ':'))}",
+        ]
+
+        if self.guidance:
+            # Truncate guidance for readability
+            guidance_display = self.guidance[:120]
+            if len(self.guidance) > 120:
+                guidance_display += "..."
+            lines.append(f"{pad}Guidance         : {guidance_display}")
+
+        if self.baseline_score is not None:
+            lines.append(f"{pad}Baseline score   : {self.baseline_score:.4f}")
+
+        return "\n".join(lines)
+
+    def to_json(self) -> str:
+        """
+        Convert the summary to JSON format.
+
+        Returns:
+            JSON string representation of the summary
+        """
+        return json.dumps(asdict(self), indent=2)
+
+    def log(self) -> None:
+        """
+        Log the summary using the configured logger.
+
+        This method outputs the formatted summary at INFO level.
+        """
+        logger = get_logger()
+        logger.progress(self.to_pretty())