M9.1: Add prompt variation generation and optimization with cmdforge optimize command
This commit is contained in:
parent
4d22d532a3
commit
bf3686ae40
|
|
@ -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())
|
||||
|
|
|
|||
|
|
@ -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")
|
||||
|
|
@ -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
|
||||
Loading…
Reference in New Issue