""" CLI — user entry point. Typer interface with -i (input) and -o (output) options. """ from __future__ import annotations import logging import os from dataclasses import asdict import dspy import typer from rich.console import Console from rich.panel import Panel from rich.table import Table from prometheus.application.bootstrap import SyntheticBootstrap from prometheus.application.dto import OptimizationConfig, OptimizationResult from prometheus.application.evaluator import PromptEvaluator from prometheus.application.use_cases import OptimizePromptUseCase from prometheus.infrastructure.file_io import YamlPersistence from prometheus.infrastructure.judge_adapter import DSPyJudgeAdapter from prometheus.infrastructure.llm_adapter import DSPyLLMAdapter from prometheus.infrastructure.proposer_adapter import DSPyProposerAdapter from prometheus.infrastructure.synth_adapter import DSPySyntheticAdapter app = typer.Typer( name="prometheus", help="PROMETHEUS — Prompt evolution without reference data.", no_args_is_help=True, ) console = Console() @app.command() def optimize( input: str = typer.Option( ..., "-i", "--input", help="Path to input YAML config file.", exists=True, readable=True, ), output: str = typer.Option( "output.yaml", "-o", "--output", help="Path to output YAML result file.", ), verbose: bool = typer.Option( False, "-v", "--verbose", help="Print detailed progress.", ), ) -> None: """Optimize a prompt without any reference data. Usage: prometheus optimize -i config.yaml -o result.yaml """ # Configure verbose logging if verbose: logging.basicConfig(level=logging.INFO, format="[PROMETHEUS] %(message)s") console.print( Panel.fit( "PROMETHEUS — Prompt Evolution Engine", subtitle="No reference data required", ) ) # 1. Load config persistence = YamlPersistence() raw_config = persistence.read_config(input) config = OptimizationConfig( seed_prompt=raw_config["seed_prompt"], task_description=raw_config["task_description"], task_model=raw_config.get("task_model", "openai/gpt-4o-mini"), judge_model=raw_config.get("judge_model", "openai/gpt-4o"), proposer_model=raw_config.get("proposer_model", "openai/gpt-4o"), synth_model=raw_config.get("synth_model", "openai/gpt-4o"), max_iterations=raw_config.get("max_iterations", 30), n_synthetic_inputs=raw_config.get("n_synthetic_inputs", 20), minibatch_size=raw_config.get("minibatch_size", 5), seed=raw_config.get("seed", 42), output_path=output, verbose=verbose, ) console.print(f"[dim]Task: {config.task_description[:80]}...[/dim]") console.print(f"[dim]Seed prompt: {config.seed_prompt[:80]}...[/dim]") # 2. Configure DSPy with optional api_base/api_key from config lm_kwargs: dict = {} api_base = raw_config.get("api_base") api_key_env = raw_config.get("api_key_env") if api_base: lm_kwargs["api_base"] = api_base if api_key_env: lm_kwargs["api_key"] = os.environ.get(api_key_env, "") task_lm = dspy.LM(config.task_model, **lm_kwargs) dspy.configure(lm=task_lm) # 3. Build adapters (Dependency Injection) synth_adapter = DSPySyntheticAdapter() llm_adapter = DSPyLLMAdapter(model=config.task_model) judge_adapter = DSPyJudgeAdapter() proposer_adapter = DSPyProposerAdapter() bootstrap = SyntheticBootstrap(generator=synth_adapter, seed=config.seed) evaluator = PromptEvaluator(executor=llm_adapter, judge=judge_adapter) use_case = OptimizePromptUseCase( evaluator=evaluator, proposer=proposer_adapter, bootstrap=bootstrap, ) # 4. Execute with console.status("[bold green]Evolving prompt..."): result = use_case.execute(config) # 5. Display results _display_result(result) # 6. Save _save_result(persistence, output, result) console.print(f"\n[green]Results saved to {output}[/green]") def _display_result(result: OptimizationResult) -> None: """Display a Rich summary in the terminal.""" console.print() console.print( Panel( f"[bold green]Optimized Prompt[/bold green]\n\n{result.optimized_prompt}", title="Result", ) ) table = Table(title="Metrics") table.add_column("Metric", style="cyan") table.add_column("Value", style="bold") table.add_row("Initial Score", f"{result.initial_score:.2f}") table.add_row("Final Score", f"{result.final_score:.2f}") table.add_row("Improvement", f"{result.improvement:+.2f}") table.add_row("Iterations", str(result.iterations_used)) table.add_row("LLM Calls", str(result.total_llm_calls)) console.print(table) def _save_result( persistence: YamlPersistence, path: str, result: OptimizationResult, ) -> None: """Save the result as YAML.""" persistence.write_result(path, asdict(result)) @app.command(hidden=True) def _help() -> None: """Internal placeholder to force multi-command Typer behavior.""" pass if __name__ == "__main__": app()