diff --git a/src/cmdforge/cli/__init__.py b/src/cmdforge/cli/__init__.py index 74864a0..8c01db8 100644 --- a/src/cmdforge/cli/__init__.py +++ b/src/cmdforge/cli/__init__.py @@ -103,6 +103,12 @@ def main(): ) p_inspect.set_defaults(func=cmd_inspect) + # 'optimize' command + p_optimize = subparsers.add_parser("optimize", help="Generate and test prompt variations for a tool") + p_optimize.add_argument("name", help="Tool name") + p_optimize.add_argument("-n", "--count", type=int, default=5, help="Number of variations") + p_optimize.set_defaults(func=cmd_optimize) + # 'check' command p_check = subparsers.add_parser("check", help="Check dependencies for a tool (meta-tools)") p_check.add_argument("name", help="Tool name") @@ -674,5 +680,38 @@ def cmd_inspect(args): return 0 if not report.errors else 1 +def cmd_optimize(args): + """Generate and test prompt variations for a tool.""" + from ..prompt_optimizer import optimize_tool + from ..tool import load_tool + + tool = load_tool(args.name) + if not tool: + print(f"Error: Tool '{args.name}' not found.", file=sys.stderr) + return 1 + + print(f"Optimizing '{tool.name}' — generating {args.count} prompt variations...") + print() + + result = optimize_tool(tool, count=args.count) + + print(f"Baseline score: {result.baseline_score}") + print(f"Variations tested: {len(result.variations)}") + print() + + if result.best: + print(f"Best variation (index {result.best_index + 1}, strategy: {result.best.strategy}):") + print(f" Score: {result.scores.get(result.best_index, 0)}") + print(f" Improvement: {result.improvement:.0f}%") + print(f" Original: {result.best.original[:100]}{'...' if len(result.best.original) > 100 else ''}") + print(f" Variation: {result.best.variation[:100]}{'...' if len(result.best.variation) > 100 else ''}") + print() + print("To apply this variation, edit the tool's config.yaml and update the prompt step.") + else: + print("No improvements found. The original prompt may already be optimal.") + + return 0 + + if __name__ == "__main__": sys.exit(main()) diff --git a/src/cmdforge/prompt_optimizer.py b/src/cmdforge/prompt_optimizer.py new file mode 100644 index 0000000..a3f0b18 --- /dev/null +++ b/src/cmdforge/prompt_optimizer.py @@ -0,0 +1,216 @@ +"""Prompt variation generation and optimization (M9.1). + +Generates N variations of a tool's prompt steps, runs each through M8.V1 +contract conformance tests, and surfaces the highest-scoring variation. +""" + +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional + +from .tool import Tool, PromptStep + + +@dataclass +class PromptVariation: + """One generated variation of a tool's prompt.""" + + step_index: int + original: str + variation: str + strategy: str # how this variation was created + + def to_dict(self) -> dict: + return { + "step_index": self.step_index, + "original": self.original, + "variation": self.variation, + "strategy": self.strategy, + } + + +@dataclass +class OptimizationResult: + """Results of running all variations through contract tests.""" + + tool_name: str + variations: List[PromptVariation] = field(default_factory=list) + scores: Dict[int, int] = field(default_factory=dict) # variation_index -> score + best_index: Optional[int] = None + baseline_score: int = 0 + + @property + def best(self) -> Optional[PromptVariation]: + if self.best_index is not None and self.best_index < len(self.variations): + return self.variations[self.best_index] + return None + + @property + def improvement(self) -> float: + if self.best is None or self.baseline_score == 0: + return 0.0 + best = self.scores.get(self.best_index, 0) + return (best - self.baseline_score) / max(self.baseline_score, 1) * 100 + + def to_dict(self) -> dict: + return { + "tool": self.tool_name, + "baseline_score": self.baseline_score, + "best_index": self.best_index, + "improvement_pct": round(self.improvement, 1), + "best": self.best.to_dict() if self.best else None, + "variations": [v.to_dict() for v in self.variations], + "scores": {str(k): v for k, v in self.scores.items()}, + } + + +def generate_variations( + tool: Tool, + count: int = 5, + provider: str = "mock", +) -> List[PromptVariation]: + """Generate prompt variations for each prompt step in the tool. + + Strategies: + - rephrase: ask the AI to rephrase the prompt + - structure: convert to bullet points + - concise: make it shorter + - detailed: add more context + - focused: emphasize the key task + + Args: + tool: The tool to optimize. + count: Number of variations per prompt step. + provider: Provider to use for variation generation. + + Returns: + List of PromptVariation objects. + """ + variations: List[PromptVariation] = [] + + for i, step in enumerate(tool.steps): + if not isinstance(step, PromptStep): + continue + + original = step.prompt + + strategies = _get_strategies(original, count) + + for strategy, variation_text in strategies: + if variation_text and variation_text != original: + variations.append(PromptVariation( + step_index=i, + original=original, + variation=variation_text, + strategy=strategy, + )) + + return variations + + +def _get_strategies(original: str, count: int) -> List[tuple]: + """Generate variations using different strategies.""" + strategies: List[tuple] = [] + + # Attempt AI-driven rephrase if a real provider is available + rephrased = _call_provider_for_variation( + f"Rephrase this instruction while keeping the same meaning. Output only the rephrased text:\n\n{original}" + ) + if rephrased and rephrased != original: + strategies.append(("rephrase", rephrased)) + + # Deterministic mock variations + mock_strategies = [ + ("focused", f"Complete this specific task — {original}"), + ("concise", _truncate(original, 120)), + ("detailed", f"Carefully perform the following: {original}. Ensure correctness and completeness."), + ("rephrase", f"I need you to do this: {original}"), + ("structure", f"Step through this task:\n\n{original}\n\nRespond step by step."), + ] + for strat, text in mock_strategies: + if len(strategies) >= count: + break + if text and text != original: + strategies.append((strat, text)) + + return strategies[:count] + + +def _truncate(text: str, max_len: int) -> str: + if len(text) <= max_len: + return text + return text[:max_len - 3].rsplit(" ", 1)[0] + "..." + + +def _call_provider_for_variation(prompt: str) -> Optional[str]: + """Try to call a provider for variation generation.""" + try: + from .providers import call_provider + result = call_provider("opencode-pickle", prompt, timeout=15) + if result.success and result.text: + text = result.text.strip() + if text and len(text) >= 10: + return text + except Exception: + pass + return None + + +def optimize_tool( + tool: Tool, + count: int = 5, + provider: str = "mock", +) -> OptimizationResult: + """Run prompt optimization for a tool. + + Generates variations, runs each through contract tests, and returns + the best-performing variation. + + Args: + tool: The tool to optimize. + count: Number of variations per prompt step. + provider: Provider for variation generation. + + Returns: + OptimizationResult with scores and best variation. + """ + from .contract_testing import run_contract_tests + + result = OptimizationResult(tool_name=tool.name) + + # Baseline score + baseline = run_contract_tests(tool) + result.baseline_score = _count_passed(baseline) + + # Generate variations + result.variations = generate_variations(tool, count=count, provider=provider) + + if not result.variations: + return result + + # Test each variation + for idx, variation in enumerate(result.variations): + # Create a copy with the modified prompt + import copy + tool_copy = copy.deepcopy(tool) + if 0 <= variation.step_index < len(tool_copy.steps): + step = tool_copy.steps[variation.step_index] + if isinstance(step, PromptStep): + step.prompt = variation.variation + + try: + test_result = run_contract_tests(tool_copy) + result.scores[idx] = _count_passed(test_result) + except Exception: + result.scores[idx] = 0 + + # Find best + if result.scores: + result.best_index = max(result.scores, key=lambda k: result.scores[k]) + + return result + + +def _count_passed(test_result) -> int: + """Count passed tests in a ConformanceReport.""" + return sum(1 for r in getattr(test_result, "results", []) + if getattr(r, "state", "") == "passed") diff --git a/tests/test_prompt_optimizer.py b/tests/test_prompt_optimizer.py new file mode 100644 index 0000000..0972925 --- /dev/null +++ b/tests/test_prompt_optimizer.py @@ -0,0 +1,148 @@ +"""Tests for prompt optimization (M9.1).""" + +from unittest.mock import patch + +from cmdforge.prompt_optimizer import ( + PromptVariation, + OptimizationResult, + generate_variations, + optimize_tool, + _get_strategies, + _truncate, +) + + +class TestTruncate: + def test_short_text_unchanged(self): + assert _truncate("hello", 10) == "hello" + + def test_long_text_truncated(self): + result = _truncate("this is a very long sentence that needs truncation", 20) + assert len(result) <= 20 + assert result.endswith("...") + + +class TestGenerateVariations: + def test_generates_for_prompt_steps(self, tmp_path): + from cmdforge.tool import Tool, PromptStep + + with patch("cmdforge.tool.TOOLS_DIR", tmp_path / ".cmdforge"): + tool = Tool( + name="test-tool", + steps=[ + PromptStep( + prompt="Analyze the input and produce a summary", + provider="mock", + output_var="summary", + ), + ], + output="{summary}", + ) + variations = generate_variations(tool, count=3) + assert len(variations) >= 1 + for v in variations: + assert v.step_index == 0 + assert v.original == "Analyze the input and produce a summary" + assert v.variation != v.original + + def test_skips_non_prompt_steps(self, tmp_path): + from cmdforge.tool import Tool, CodeStep + + with patch("cmdforge.tool.TOOLS_DIR", tmp_path / ".cmdforge"): + tool = Tool( + name="test-tool", + steps=[CodeStep(code="print('hello')", output_var="out")], + output="{out}", + ) + variations = generate_variations(tool) + assert len(variations) == 0 + + +class TestGetStrategies: + def test_produces_variations(self): + strategies = _get_strategies("Summarize the text", 3) + assert len(strategies) == 3 + for strat, text in strategies: + assert len(text) > 0 + assert strat in ("focused", "concise", "detailed", "rephrase", "structure") + + def test_truncates_concise(self): + original = "A" * 200 + strategies = _get_strategies(original, 3) + concise = [t for s, t in strategies if s == "concise"] + if concise: + assert len(concise[0]) <= 125 # 120 + "..." + + +class TestOptimizeTool: + def test_baseline_score_recorded(self, tmp_path): + from cmdforge.tool import Tool, PromptStep + + with patch("cmdforge.tool.TOOLS_DIR", tmp_path / ".cmdforge"): + tool = Tool( + name="opt-test", + steps=[ + PromptStep( + prompt="Summarize: {input}", + provider="mock", + output_var="result", + ), + ], + output="{result}", + ) + result = optimize_tool(tool, count=2) + assert result.tool_name == "opt-test" + assert result.baseline_score >= 0 + assert len(result.variations) >= 1 + assert len(result.scores) == len(result.variations) + + def test_no_prompt_steps_returns_empty(self, tmp_path): + from cmdforge.tool import Tool, CodeStep + + with patch("cmdforge.tool.TOOLS_DIR", tmp_path / ".cmdforge"): + tool = Tool( + name="code-only", + steps=[CodeStep(code="pass", output_var="x")], + output="{x}", + ) + result = optimize_tool(tool) + assert len(result.variations) == 0 + assert result.best is None + + +class TestPromptVariation: + def test_to_dict(self): + v = PromptVariation( + step_index=0, + original="hello", + variation="hi there", + strategy="rephrase", + ) + d = v.to_dict() + assert d["original"] == "hello" + assert d["strategy"] == "rephrase" + + +class TestOptimizationResult: + def test_improvement_calculation(self): + result = OptimizationResult( + tool_name="test", + variations=[ + PromptVariation(0, "a", "b", "rephrase"), + ], + scores={0: 5}, + best_index=0, + baseline_score=3, + ) + assert result.best is not None + assert result.improvement > 0 + + def test_no_improvement_when_no_baseline(self): + result = OptimizationResult( + tool_name="test", + variations=[], + scores={}, + baseline_score=0, + ) + assert result.best is None + assert result.improvement == 0.0