Automatic Stateful Prompt Improver

Automatic Stateful Prompt Improver

MANDATORY AUTOMATIC BEHAVIOR

When this skill is active, I MUST follow these rules:

Auto-Optimization Triggers

I AUTOMATICALLY call mcp__prompt-learning__optimize_prompt BEFORE responding when:

1. Complex task (multi-step, requires reasoning) 2. Technical output (code, analysis, structured data) 3. Reusable content (system prompts, templates, instructions) 4. Explicit request ("improve", "better", "optimize") 5. Ambiguous requirements (underspecified, multiple interpretations) 6. Precision-critical (code, legal, medical, financial)

Auto-Optimization Process

1. INTERCEPT the user's request
2. CALL: mcp__prompt-learning__optimize_prompt
   - prompt: [user's original request]
   - domain: [inferred domain]
   - max_iterations: [3-20 based on complexity]
3. RECEIVE: optimized prompt + improvement details
4. INFORM user briefly: "I've refined your request for [reason]"
5. PROCEED with the OPTIMIZED version

Do NOT Optimize

• Simple questions ("what is X?")
• Direct commands ("run npm install")
• Conversational responses ("hello", "thanks")
• File operations without reasoning
• Already-optimized prompts

Learning Loop (Post-Response)

After completing ANY significant task:

1. ASSESS: Did the response achieve the goal?
2. CALL: mcp__prompt-learning__record_feedback
   - prompt_id: [from optimization response]
   - success: [true/false]
   - quality_score: [0.0-1.0]
3. This enables future retrievals to learn from outcomes

Quick Reference

Iteration Decision

Convergence (When to Stop)

• Improvement < 1% for 3 iterations
• User satisfied
• Token budget exhausted
• 20 iterations reached
• Validation score > 0.95

Performance Expectations

Anti-Patterns

Over-Optimization

Template Obsession

Iteration Without Measurement

Ignoring Model Capabilities

Reference Files

Load for detailed implementations:

---

Goal: Simplest prompt that achieves the outcome reliably. Optimize for clarity, specificity, and measurable improvement.

Changelog

[2.0.0] - 2024-12-XX

Changed

• SKILL.md restructured for progressive disclosure (481 → ~128 lines)
• Frontmatter fixed: tools: YAML list → allowed-tools: comma-separated

Added

• references/optimization-techniques.md - APE, OPRO, Chain-of-Thought, DSPy patterns
• references/learning-architecture.md - Warm start, embedding indexing, performance tracking
• references/iteration-strategy.md - Decision matrices, convergence criteria
• Clear convergence thresholds (≥0.85 success rate, ≤10% improvement)
• Detailed iteration budget management

Migration Guide

• Breaking: If you use this skill via configuration, update frontmatter format
• Old: tools:\n - tool1\n - tool2
• New: allowed-tools: tool1,tool2,tool3
• Reference files provide implementation details on demand

APE and OPRO Implementation Guide

This document details the implementation of Automatic Prompt Engineer (APE) and Optimization by Prompting (OPRO) algorithms for prompt improvement.

APE: Automatic Prompt Engineer

Source: Zhou et al., 2022 - "Large Language Models Are Human-Level Prompt Engineers"

Core Idea

Use LLMs to generate instruction candidates, then select the best based on evaluation.

Algorithm

APE Algorithm:
1. Given: Task description T, examples E = {(x_i, y_i)}
2. Generate instruction candidates:
   - Prompt LLM: "Given these examples, what instruction would produce this output?"
   - Generate N candidates (N=20 typical)
3. Evaluate each candidate:
   - For each instruction I_j:
     - Score = accuracy on held-out examples
     - Or: Score = log_prob(y | I_j, x)
4. Select best:
   - I* = argmax_j Score(I_j)
5. Optionally refine:
   - Generate variations of I*
   - Re-evaluate and select best variation

Implementation

class APEOptimizer:
    """
    Automatic Prompt Engineer implementation.
    """

    def __init__(self, llm_client, num_candidates: int = 20):
        self.llm = llm_client
        self.num_candidates = num_candidates

    async def generate_candidates(
        self,
        task_description: str,
        examples: list[tuple[str, str]],
        num_candidates: int = None
    ) -> list[str]:
        """
        Generate instruction candidates from examples.
        """
        n = num_candidates or self.num_candidates

        # Format examples
        examples_text = "\n".join([
            f"Input: {x}\nOutput: {y}"
            for x, y in examples[:5]  # Use first 5 examples
        ])

        generation_prompt = f"""
        I have a task where given certain inputs, I want specific outputs.

        Here are some examples:
        {examples_text}

        Your task: Generate {n} different instructions that would cause an AI to produce these outputs from these inputs.

        Requirements:
        - Each instruction should be clear and complete
        - Instructions should be diverse (different phrasings, approaches)
        - Instructions should be concise but complete

        Generate exactly {n} instructions, one per line, numbered 1-{n}:
        """

        response = await self.llm.generate(generation_prompt)
        candidates = self._parse_numbered_list(response)

        return candidates[:n]

    async def evaluate_candidate(
        self,
        instruction: str,
        eval_examples: list[tuple[str, str]]
    ) -> float:
        """
        Evaluate an instruction on held-out examples.
        """
        correct = 0
        total = len(eval_examples)

        for x, expected_y in eval_examples:
            prompt = f"{instruction}\n\nInput: {x}\n\nOutput:"
            response = await self.llm.generate(prompt)

            # Simple exact match (could use semantic similarity)
            if self._normalize(response) == self._normalize(expected_y):
                correct += 1

        return correct / total if total > 0 else 0

    async def optimize(
        self,
        task_description: str,
        examples: list[tuple[str, str]],
        refinement_iterations: int = 3
    ) -> tuple[str, float]:
        """
        Full APE optimization pipeline.

        Returns:
            (best_instruction, best_score)
        """
        # Split examples
        train_examples = examples[:len(examples)//2]
        eval_examples = examples[len(examples)//2:]

        # Generate initial candidates
        candidates = await self.generate_candidates(
            task_description,
            train_examples
        )

        # Evaluate each
        scores = []
        for candidate in candidates:
            score = await self.evaluate_candidate(candidate, eval_examples)
            scores.append((candidate, score))

        # Sort by score
        scores.sort(key=lambda x: x[1], reverse=True)
        best_instruction, best_score = scores[0]

        # Refinement iterations
        for i in range(refinement_iterations):
            # Generate variations of best
            variations = await self._generate_variations(best_instruction)

            # Evaluate variations
            for variation in variations:
                score = await self.evaluate_candidate(variation, eval_examples)
                if score > best_score:
                    best_instruction = variation
                    best_score = score

        return best_instruction, best_score

    async def _generate_variations(
        self,
        instruction: str,
        num_variations: int = 5
    ) -> list[str]:
        """
        Generate variations of an instruction.
        """
        variation_prompt = f"""
        Generate {num_variations} variations of the following instruction.
        Keep the semantic meaning but vary the phrasing, structure, or emphasis.

        Original instruction:
        {instruction}

        Variations (numbered 1-{num_variations}):
        """

        response = await self.llm.generate(variation_prompt)
        return self._parse_numbered_list(response)

    def _parse_numbered_list(self, text: str) -> list[str]:
        """Parse numbered list from LLM output."""
        lines = text.strip().split('\n')
        items = []
        for line in lines:
            # Remove numbering (1., 1), etc.)
            cleaned = re.sub(r'^[\d]+[\.\)\:]?\s*', '', line.strip())
            if cleaned:
                items.append(cleaned)
        return items

    def _normalize(self, text: str) -> str:
        """Normalize text for comparison."""
        return text.strip().lower()

APE for Prompt Improvement

async def improve_prompt_ape_style(
    original_prompt: str,
    examples: list[tuple[str, str]],
    llm_client
) -> str:
    """
    Improve a prompt using APE methodology.
    """
    optimizer = APEOptimizer(llm_client)

    # Use original prompt as task description
    task_description = f"Original prompt: {original_prompt}"

    # Optimize
    best_instruction, score = await optimizer.optimize(
        task_description,
        examples,
        refinement_iterations=3
    )

    return best_instruction

OPRO: Optimization by Prompting

Source: Yang et al., 2023 - "Large Language Models as Optimizers"

Core Idea

Treat optimization itself as a prompting task. The LLM sees previous solutions and their scores, then generates new (hopefully better) solutions.

Algorithm

OPRO Algorithm:
1. Initialize: meta_prompt = empty history
2. For i in range(max_iterations):
   a. Add instruction to meta_prompt:
      "[previous solutions + scores]
       Generate a new solution that scores higher."
   b. LLM generates new candidate solution
   c. Evaluate candidate on validation set
   d. Add (candidate, score) to history
   e. If converged: break
3. Return best solution from history

Implementation

class OPROOptimizer:
    """
    Optimization by Prompting implementation.
    """

    def __init__(
        self,
        llm_client,
        max_history: int = 20,
        temperature: float = 0.7
    ):
        self.llm = llm_client
        self.max_history = max_history
        self.temperature = temperature
        self.history = []

    def _build_meta_prompt(self, task_description: str) -> str:
        """
        Build the meta-prompt showing previous solutions and scores.
        """
        # Sort history by score (ascending) - show progression
        sorted_history = sorted(self.history, key=lambda x: x[1])

        # Take most recent/best entries
        recent = sorted_history[-self.max_history:]

        history_text = "\n".join([
            f"Instruction: {inst}\nScore: {score:.3f}"
            for inst, score in recent
        ])

        meta_prompt = f"""
        Your task is to generate an instruction that will achieve a high score on this task:

        {task_description}

        Here are some previous instructions and their scores (higher is better):

        {history_text}

        Based on the patterns in what works well:
        1. Analyze why high-scoring instructions perform better
        2. Identify elements that correlate with success
        3. Generate a NEW instruction that should score even higher

        Requirements:
        - The instruction should be different from previous ones
        - Build on what worked, avoid what didn't
        - Be specific and clear

        New instruction (just the instruction, nothing else):
        """

        return meta_prompt

    async def generate_candidate(self, task_description: str) -> str:
        """
        Generate a new candidate instruction based on history.
        """
        meta_prompt = self._build_meta_prompt(task_description)

        response = await self.llm.generate(
            meta_prompt,
            temperature=self.temperature
        )

        return response.strip()

    async def evaluate(
        self,
        instruction: str,
        eval_fn: callable
    ) -> float:
        """
        Evaluate an instruction using provided evaluation function.
        """
        return await eval_fn(instruction)

    async def optimize(
        self,
        task_description: str,
        eval_fn: callable,
        initial_candidates: list[str] = None,
        max_iterations: int = 20,
        convergence_threshold: float = 0.01,
        convergence_window: int = 5
    ) -> tuple[str, float, list]:
        """
        Run OPRO optimization.

        Args:
            task_description: Description of the task
            eval_fn: Async function that scores an instruction
            initial_candidates: Optional starting candidates
            max_iterations: Maximum optimization iterations
            convergence_threshold: Stop if improvement < this
            convergence_window: Check convergence over this many iterations

        Returns:
            (best_instruction, best_score, optimization_history)
        """
        # Initialize with any provided candidates
        if initial_candidates:
            for candidate in initial_candidates:
                score = await self.evaluate(candidate, eval_fn)
                self.history.append((candidate, score))

        # Track scores for convergence detection
        scores_over_time = [h[1] for h in self.history]

        for i in range(max_iterations):
            # Generate new candidate
            candidate = await self.generate_candidate(task_description)

            # Evaluate
            score = await self.evaluate(candidate, eval_fn)

            # Add to history
            self.history.append((candidate, score))
            scores_over_time.append(score)

            # Check convergence
            if len(scores_over_time) >= convergence_window:
                recent_best = max(scores_over_time[-convergence_window:])
                previous_best = max(scores_over_time[:-convergence_window])
                if recent_best - previous_best < convergence_threshold:
                    break

        # Find best
        best_instruction, best_score = max(self.history, key=lambda x: x[1])

        return best_instruction, best_score, self.history

OPRO for Prompt Improvement

async def improve_prompt_opro_style(
    original_prompt: str,
    eval_examples: list[tuple[str, str]],
    llm_client,
    max_iterations: int = 15
) -> tuple[str, float]:
    """
    Improve a prompt using OPRO methodology.
    """
    optimizer = OPROOptimizer(llm_client)

    # Create evaluation function
    async def eval_fn(instruction: str) -> float:
        correct = 0
        for x, expected_y in eval_examples:
            prompt = f"{instruction}\n\nInput: {x}"
            response = await llm_client.generate(prompt)
            if expected_y.lower() in response.lower():
                correct += 1
        return correct / len(eval_examples)

    # Task description
    task_description = f"""
    Original prompt: {original_prompt}

    The goal is to find an instruction that produces correct outputs for various inputs.
    The instruction should be clear, specific, and effective.
    """

    # Start with original as baseline
    initial_candidates = [original_prompt]

    # Optimize
    best, score, history = await optimizer.optimize(
        task_description=task_description,
        eval_fn=eval_fn,
        initial_candidates=initial_candidates,
        max_iterations=max_iterations
    )

    return best, score

Combining APE and OPRO

Hybrid Approach

class HybridPromptOptimizer:
    """
    Combines APE (generation) with OPRO (optimization) for best results.

    1. APE generates diverse initial candidates
    2. OPRO iteratively improves using history
    """

    def __init__(self, llm_client):
        self.llm = llm_client
        self.ape = APEOptimizer(llm_client, num_candidates=10)
        self.opro = OPROOptimizer(llm_client, max_history=20)

    async def optimize(
        self,
        original_prompt: str,
        examples: list[tuple[str, str]],
        max_iterations: int = 20
    ) -> tuple[str, float, dict]:
        """
        Hybrid optimization combining APE and OPRO.
        """
        # Split examples
        train_examples = examples[:len(examples)//2]
        eval_examples = examples[len(examples)//2:]

        # Phase 1: APE generates initial candidates
        candidates = await self.ape.generate_candidates(
            f"Task: {original_prompt}",
            train_examples,
            num_candidates=10
        )

        # Add original prompt as candidate
        candidates = [original_prompt] + candidates

        # Evaluate initial candidates
        initial_scores = []
        for candidate in candidates:
            score = await self._evaluate(candidate, eval_examples)
            initial_scores.append((candidate, score))

        # Sort and take top 5 as OPRO seed
        initial_scores.sort(key=lambda x: x[1], reverse=True)
        top_candidates = [c for c, s in initial_scores[:5]]

        # Phase 2: OPRO iterative optimization
        task_description = f"""
        Task: Generate instructions for this prompt optimization task.
        Original prompt: {original_prompt}

        The instruction should make the AI produce correct outputs for various inputs.
        """

        # Initialize OPRO with APE candidates
        for candidate, score in initial_scores[:5]:
            self.opro.history.append((candidate, score))

        # OPRO iterations
        async def eval_fn(instruction):
            return await self._evaluate(instruction, eval_examples)

        best, score, history = await self.opro.optimize(
            task_description=task_description,
            eval_fn=eval_fn,
            max_iterations=max_iterations - 5  # Already did 5 in APE
        )

        return best, score, {
            'ape_candidates': len(candidates),
            'opro_iterations': len(history) - 5,
            'final_history': history
        }

    async def _evaluate(
        self,
        instruction: str,
        examples: list[tuple[str, str]]
    ) -> float:
        """Simple evaluation function."""
        correct = 0
        for x, expected_y in examples:
            prompt = f"{instruction}\n\nInput: {x}"
            response = await self.llm.generate(prompt)
            if self._matches(response, expected_y):
                correct += 1
        return correct / len(examples)

    def _matches(self, response: str, expected: str) -> bool:
        """Check if response matches expected."""
        return expected.lower().strip() in response.lower()

Without LLM Calls: Pattern-Based Improvement

When you can't make LLM calls for optimization, use pattern-based rules:

class PatternBasedOptimizer:
    """
    Improve prompts using known-good patterns without LLM calls.
    """

    IMPROVEMENT_PATTERNS = [
        {
            "name": "add_structure",
            "check": lambda p: not any(x in p.lower() for x in ["1.", "2.", "step", "first"]),
            "apply": lambda p: f"{p}\n\nProvide your response in this format:\n1. [First point]\n2. [Second point]\n3. [Summary]",
            "expected_improvement": 0.15
        },
        {
            "name": "add_cot",
            "check": lambda p: "step by step" not in p.lower() and "think" not in p.lower(),
            "apply": lambda p: f"{p}\n\nThink through this step by step before providing your final answer.",
            "expected_improvement": 0.20
        },
        {
            "name": "add_constraints",
            "check": lambda p: len(p) < 100,
            "apply": lambda p: f"{p}\n\nRequirements:\n- Be specific and precise\n- Support claims with evidence\n- Keep response focused",
            "expected_improvement": 0.10
        },
        {
            "name": "add_role",
            "check": lambda p: not p.lower().startswith(("you are", "as a", "act as")),
            "apply": lambda p: f"You are an expert in this domain. {p}",
            "expected_improvement": 0.05
        },
        {
            "name": "add_output_format",
            "check": lambda p: "format" not in p.lower() and "json" not in p.lower(),
            "apply": lambda p: f"{p}\n\nFormat your response as a clear, structured answer.",
            "expected_improvement": 0.10
        }
    ]

    def improve(self, prompt: str) -> tuple[str, list[str]]:
        """
        Apply applicable improvement patterns.

        Returns:
            (improved_prompt, list of applied patterns)
        """
        improved = prompt
        applied = []

        for pattern in self.IMPROVEMENT_PATTERNS:
            if pattern["check"](improved):
                improved = pattern["apply"](improved)
                applied.append(pattern["name"])

        return improved, applied

    def estimate_improvement(self, applied_patterns: list[str]) -> float:
        """
        Estimate expected improvement based on applied patterns.
        """
        total = 0
        for pattern in self.IMPROVEMENT_PATTERNS:
            if pattern["name"] in applied_patterns:
                total += pattern["expected_improvement"]
        return min(total, 0.5)  # Cap at 50% improvement

Convergence and Stopping Criteria

When to Stop Optimization

def check_optimization_convergence(
    scores: list[float],
    iteration: int,
    config: dict = None
) -> tuple[bool, str]:
    """
    Check if optimization should stop.

    Config options:
    - max_iterations: Hard limit (default: 20)
    - min_iterations: Minimum before stopping (default: 5)
    - plateau_window: Iterations to check for plateau (default: 5)
    - plateau_threshold: Minimum improvement (default: 0.01)
    - target_score: Stop if reached (default: 0.95)
    """
    config = config or {}
    max_iter = config.get('max_iterations', 20)
    min_iter = config.get('min_iterations', 5)
    window = config.get('plateau_window', 5)
    threshold = config.get('plateau_threshold', 0.01)
    target = config.get('target_score', 0.95)

    # Not enough iterations
    if iteration < min_iter:
        return False, "below_minimum_iterations"

    # Max iterations reached
    if iteration >= max_iter:
        return True, "max_iterations_reached"

    # Target reached
    if scores[-1] >= target:
        return True, f"target_reached ({scores[-1]:.3f} >= {target})"

    # Plateau detection
    if len(scores) >= window:
        recent_improvement = max(scores[-window:]) - scores[-window]
        if recent_improvement < threshold:
            return True, f"plateau_detected (improvement: {recent_improvement:.4f})"

    # Decreasing performance (possible overfitting)
    if len(scores) >= 3 and all(scores[-i-1] < scores[-i-2] for i in range(2)):
        return True, "performance_decreasing"

    return False, "continue"

---

These implementations provide the foundation for systematic prompt optimization using research-backed algorithms.

DSPy Optimization Patterns for Prompt Learning

This document covers DSPy-style programmatic prompt optimization patterns.

Core Philosophy

DSPy treats prompts as trainable parameters rather than brittle strings. Instead of manually crafting prompts, you define: 1. What the module should do (signature) 2. How to measure success (metric) 3. Let the optimizer find the best prompt

Key Concepts

Signatures

Signatures declare input/output behavior:

# Simple signature
class QA(dspy.Signature):
    """Answer questions with brief responses."""
    question = dspy.InputField()
    answer = dspy.OutputField()

# Complex signature with constraints
class CodeReview(dspy.Signature):
    """Review code for bugs, style issues, and improvements."""
    code = dspy.InputField(desc="The code to review")
    language = dspy.InputField(desc="Programming language")
    severity_filter = dspy.InputField(desc="min, major, critical", default="min")

    bugs = dspy.OutputField(desc="List of potential bugs found")
    style_issues = dspy.OutputField(desc="Style violations")
    improvements = dspy.OutputField(desc="Suggested improvements")

Modules

Modules define HOW to accomplish the signature:

# Basic prediction
predictor = dspy.Predict(QA)

# Chain-of-thought reasoning
cot = dspy.ChainOfThought(QA)

# ReAct with tool use
react = dspy.ReAct(QA, tools=[search_tool, calculator])

# Multi-step pipeline
class RAGPipeline(dspy.Module):
    def __init__(self, num_passages=3):
        self.retrieve = dspy.Retrieve(k=num_passages)
        self.generate = dspy.ChainOfThought("context, question -> answer")

    def forward(self, question):
        context = self.retrieve(question).passages
        return self.generate(context=context, question=question)

Optimization Algorithms

BootstrapRS (Random Search)

Best for: Quick optimization, small datasets

# Generates few-shot examples from successful executions
optimizer = dspy.BootstrapRS(
    metric=my_metric,
    max_bootstrapped_demos=4,  # Examples to include
    max_labeled_demos=4,
    num_candidate_programs=16
)

optimized = optimizer.compile(
    student=my_module,
    trainset=train_examples
)

How it works: 1. Run the module on training examples 2. Identify successful executions 3. Use those as few-shot demonstrations 4. Select best combination randomly

MIPROv2 (Bayesian Optimization)

Best for: Production optimization, larger datasets

optimizer = dspy.MIPROv2(
    metric=my_metric,
    auto="medium",  # light, medium, heavy
    num_threads=4
)

optimized = optimizer.compile(
    student=my_module,
    trainset=train_examples,
    valset=val_examples,  # Important for evaluation
    max_bootstrapped_demos=4,
    max_labeled_demos=4
)

How it works: 1. Bootstrap: Collect successful execution traces 2. Propose: LLM generates instruction candidates 3. Search: Bayesian optimization over instruction space 4. Refine: Iteratively improve best candidates

Cost/Performance (from research):

COPRO (Coordinate Optimization)

Best for: Instruction-only optimization

optimizer = dspy.COPRO(
    metric=my_metric,
    depth=3,  # Optimization iterations
    breadth=10,  # Candidates per iteration
    init_temperature=1.0
)

# Only optimizes instructions, not examples
optimized = optimizer.compile(
    student=my_module,
    trainset=train_examples
)

Practical Patterns for Prompt Improvement

Pattern 1: Bootstrap → Optimize → Ensemble

def full_optimization_pipeline(module, trainset, valset):
    """
    Three-stage optimization for best results.
    """
    # Stage 1: Bootstrap few-shot examples
    bootstrap = dspy.BootstrapRS(
        metric=accuracy_metric,
        max_bootstrapped_demos=4
    )
    bootstrapped = bootstrap.compile(module, trainset=trainset)

    # Stage 2: Optimize instructions
    mipro = dspy.MIPROv2(metric=accuracy_metric, auto="medium")
    optimized = mipro.compile(
        bootstrapped,
        trainset=trainset,
        valset=valset
    )

    # Stage 3: Create ensemble (optional)
    # Run optimization 3 times, ensemble the results
    programs = []
    for _ in range(3):
        opt = dspy.MIPROv2(metric=accuracy_metric, auto="light")
        programs.append(opt.compile(module, trainset=trainset, valset=valset))

    ensemble = dspy.Ensemble(programs, strategy="majority")

    return optimized, ensemble

Pattern 2: Metric-Driven Optimization

def create_composite_metric():
    """
    Multi-objective metric combining accuracy, cost, and quality.
    """
    def metric(example, prediction, trace=None):
        # Accuracy (0-1)
        accuracy = 1.0 if prediction.answer == example.expected else 0.0

        # Token efficiency (penalize verbose responses)
        tokens_used = len(prediction.answer.split())
        efficiency = 1.0 / (1.0 + tokens_used / 100)

        # Quality heuristics
        has_reasoning = "because" in prediction.answer.lower()
        quality = 1.0 if has_reasoning else 0.5

        # Weighted combination
        return 0.6 * accuracy + 0.2 * efficiency + 0.2 * quality

    return metric

Pattern 3: Assertion-Based Validation

class ValidatedQA(dspy.Module):
    """
    Module with built-in validation via assertions.
    """
    def __init__(self):
        self.generate = dspy.ChainOfThought("question -> answer")

    def forward(self, question):
        response = self.generate(question=question)

        # Validate response
        dspy.Assert(
            len(response.answer) > 10,
            "Answer too short - provide more detail"
        )

        dspy.Assert(
            response.answer[-1] in ".!?",
            "Answer should end with punctuation"
        )

        dspy.Suggest(
            "step" in response.rationale.lower() or "because" in response.rationale.lower(),
            "Consider explaining your reasoning step by step"
        )

        return response

Pattern 4: Multi-Stage Pipelines

class OptimizedPipeline(dspy.Module):
    """
    Multi-stage pipeline where each stage can be optimized independently.
    """
    def __init__(self):
        # Stage 1: Extract key information
        self.extract = dspy.ChainOfThought(
            "document -> key_points, entities, summary"
        )

        # Stage 2: Reason about the information
        self.reason = dspy.ChainOfThought(
            "key_points, entities, question -> reasoning"
        )

        # Stage 3: Generate final answer
        self.answer = dspy.Predict(
            "summary, reasoning, question -> answer"
        )

    def forward(self, document, question):
        # Stage 1
        extracted = self.extract(document=document)

        # Stage 2
        reasoned = self.reason(
            key_points=extracted.key_points,
            entities=extracted.entities,
            question=question
        )

        # Stage 3
        result = self.answer(
            summary=extracted.summary,
            reasoning=reasoned.reasoning,
            question=question
        )

        return result

Adapting DSPy Patterns for Claude Code

Since we can't use DSPy's Python framework directly in Claude Code, we adapt the patterns:

Mental Bootstrapping

## Bootstrap Pattern (Manual)

When improving a prompt, I will:

1. **Generate variations**: Create 5-10 instruction candidates
2. **Mental evaluation**: For each, consider:
   - Clarity: Is it unambiguous?
   - Completeness: Does it cover all cases?
   - Constraint density: Right amount of guidance?
3. **Select best**: Choose highest-scoring candidate
4. **Extract patterns**: What made it better?

Example:
Original: "Summarize the text"

Candidates:
- "Extract the main points from this text" (clarity: 7)
- "Provide a 3-sentence summary" (constraint: 8)
- "Identify thesis, evidence, and conclusion" (structure: 9)

Best: Combine structure (9) + constraint (8)
Result: "Identify the thesis, key evidence, and conclusion, then summarize in 2-3 sentences"

Instruction Optimization Loop

## COPRO Pattern (Manual)

For instruction optimization:

1. **Baseline**: Evaluate current prompt
2. **Generate candidates**: Create variations
3. **Evaluate each**: Score on clarity/specificity/effectiveness
4. **Select best**: Choose highest scorer
5. **Iterate**: Use best as new baseline
6. **Stop when**: Improvement < threshold or max iterations

Track:
- Iteration 1: Score 0.65
- Iteration 2: Score 0.72 (+0.07)
- Iteration 3: Score 0.75 (+0.03)
- Iteration 4: Score 0.76 (+0.01) ← Converging
- STOP: Improvement < 0.02

Few-Shot Selection

## Example Selection Pattern

When adding few-shot examples:

1. **Diversity**: Cover different scenarios
2. **Relevance**: Similar to expected inputs
3. **Difficulty**: Progress from easy to hard
4. **Quality**: Only include high-quality examples

Anti-patterns:
- All examples from same category ← Low coverage
- Examples too different from test case ← Low relevance
- All easy or all hard ← Imbalanced

Convergence Criteria

When to Stop Optimizing

def should_stop_optimization(
    scores: list[float],
    iterations: int,
    max_iterations: int = 20,
    window: int = 3,
    threshold: float = 0.01
) -> tuple[bool, str]:
    """
    Determine if optimization should stop.

    Returns:
        (should_stop, reason)
    """
    # Max iterations reached
    if iterations >= max_iterations:
        return True, "max_iterations_reached"

    # Not enough data
    if len(scores) < window:
        return False, "insufficient_data"

    # Performance plateau
    recent_improvement = max(scores[-window:]) - scores[-window]
    if recent_improvement < threshold:
        return True, f"plateau_detected (improvement: {recent_improvement:.4f})"

    # High enough score
    if scores[-1] > 0.95:
        return True, f"target_reached (score: {scores[-1]:.4f})"

    return False, "continue"

Iteration Count Heuristics

Based on research and empirical testing:

Implementation Notes

Memory Efficiency

# Keep only top-k candidates to manage memory
class CandidatePool:
    def __init__(self, max_size: int = 10):
        self.candidates = []
        self.max_size = max_size

    def add(self, candidate: str, score: float):
        self.candidates.append((candidate, score))
        self.candidates.sort(key=lambda x: x[1], reverse=True)
        self.candidates = self.candidates[:self.max_size]

    def get_best(self) -> str:
        return self.candidates[0][0] if self.candidates else None

Reproducibility

# Track optimization history for reproducibility
optimization_history = {
    "initial_prompt": "...",
    "iterations": [
        {"candidate": "...", "score": 0.65, "selected": False},
        {"candidate": "...", "score": 0.72, "selected": True},
    ],
    "final_prompt": "...",
    "final_score": 0.76,
    "convergence_reason": "plateau_detected"
}

---

These patterns form the foundation for systematic, measurable prompt optimization that can be applied even without access to DSPy's infrastructure.

Embedding Architecture for Prompt Learning

This document details the embedding and retrieval architecture for stateful prompt learning.

Design Principles

1. Contextual Embeddings: Add domain context before embedding (49% error reduction) 2. Hybrid Search: Vector + BM25 for best recall 3. Recency Weighting: Balance historical stability with recent relevance 4. Drift Detection: Detect and adapt to distribution shifts

Embedding Strategy

Contextual Embedding Pipeline

Following Anthropic's contextual retrieval methodology:

class ContextualPromptEmbedder:
    """
    Create embeddings that include domain context for better retrieval.

    Standard embedding: embed(prompt_text)
    Contextual embedding: embed(context + prompt_text)

    This reduces retrieval errors by 49% (with BM25) to 67% (with reranking).
    """

    def __init__(self, embedding_model: str = "text-embedding-3-large"):
        self.model = embedding_model
        self.context_cache = {}  # Cache context generation

    async def create_embedding(
        self,
        prompt: str,
        domain: str,
        task_type: str
    ) -> dict:
        """
        Create contextual embedding for a prompt.

        Args:
            prompt: The prompt text to embed
            domain: Domain classification (e.g., 'code_review', 'summarization')
            task_type: Task type (e.g., 'generation', 'classification')

        Returns:
            {
                'embedding': [...],
                'contextualized_text': str,
                'metadata': dict
            }
        """
        # Generate contextualizing information
        context = await self._generate_context(prompt, domain, task_type)

        # Combine context with prompt
        contextualized = f"{context}\n\n{prompt}"

        # Generate embedding
        embedding = await self._embed(contextualized)

        return {
            'embedding': embedding,
            'contextualized_text': contextualized,
            'metadata': {
                'domain': domain,
                'task_type': task_type,
                'original_length': len(prompt),
                'context_length': len(context)
            }
        }

    async def _generate_context(
        self,
        prompt: str,
        domain: str,
        task_type: str
    ) -> str:
        """
        Generate 50-100 token context for a prompt.

        Uses LLM to create contextualizing information that improves retrieval.
        Results are cached to reduce API calls.
        """
        cache_key = f"{domain}:{task_type}:{hash(prompt[:100])}"

        if cache_key in self.context_cache:
            return self.context_cache[cache_key]

        context_prompt = f"""
        Generate a brief context (50-100 tokens) for this prompt to improve retrieval.

        Domain: {domain}
        Task Type: {task_type}
        Prompt: {prompt[:500]}

        Include:
        - Key domain terminology
        - Task intent
        - Important entities/concepts

        Context only, no explanation:
        """

        context = await self._call_llm(context_prompt)
        self.context_cache[cache_key] = context
        return context

Embedding Model Selection

Recommendation: Start with text-embedding-3-small, upgrade to large for production.

Retrieval Architecture

Hybrid Search Pipeline

class HybridPromptRetrieval:
    """
    Combine vector similarity with BM25 for optimal retrieval.

    Research shows hybrid search outperforms either alone:
    - Vector only: Good semantic matching
    - BM25 only: Good keyword matching
    - Hybrid: Best of both (5-10% improvement)
    """

    def __init__(
        self,
        vector_db: VectorDatabase,
        bm25_index: BM25Index,
        reranker: Optional[Reranker] = None
    ):
        self.vector_db = vector_db
        self.bm25 = bm25_index
        self.reranker = reranker

    async def search(
        self,
        query: str,
        query_embedding: list,
        top_k: int = 10,
        filters: dict = None
    ) -> list:
        """
        Hybrid retrieval with optional reranking.

        Pipeline:
        1. Vector search (retrieve 3x candidates)
        2. BM25 search (retrieve 3x candidates)
        3. Reciprocal Rank Fusion (combine results)
        4. Optional reranking (refine top candidates)
        """
        # Over-retrieve for fusion
        fetch_k = top_k * 3

        # Parallel retrieval
        vector_results, bm25_results = await asyncio.gather(
            self._vector_search(query_embedding, fetch_k, filters),
            self._bm25_search(query, fetch_k, filters)
        )

        # Reciprocal Rank Fusion
        fused = self._rrf_fusion(vector_results, bm25_results)

        # Optional reranking
        if self.reranker and len(fused) > 0:
            reranked = await self.reranker.rerank(
                query=query,
                documents=fused[:top_k * 2],
                top_k=top_k
            )
            return reranked

        return fused[:top_k]

    def _rrf_fusion(
        self,
        vector_results: list,
        bm25_results: list,
        k: int = 60
    ) -> list:
        """
        Reciprocal Rank Fusion: 1/(k + rank) scoring.

        Combines rankings from multiple retrieval methods.
        k=60 is standard from RRF paper.
        """
        scores = {}
        doc_map = {}

        for rank, doc in enumerate(vector_results):
            doc_id = doc['id']
            scores[doc_id] = scores.get(doc_id, 0) + 1 / (k + rank + 1)
            doc_map[doc_id] = doc

        for rank, doc in enumerate(bm25_results):
            doc_id = doc['id']
            scores[doc_id] = scores.get(doc_id, 0) + 1 / (k + rank + 1)
            doc_map[doc_id] = doc

        # Sort by fused score
        sorted_ids = sorted(scores.keys(), key=lambda x: scores[x], reverse=True)

        return [
            {**doc_map[doc_id], 'rrf_score': scores[doc_id]}
            for doc_id in sorted_ids
            if doc_id in doc_map
        ]

Reranking (Optional but Recommended)

class CohereReranker:
    """
    Use Cohere's reranker for final result refinement.

    Adds ~100ms latency but improves relevance significantly.
    """

    def __init__(self, api_key: str):
        self.client = cohere.Client(api_key)

    async def rerank(
        self,
        query: str,
        documents: list,
        top_k: int = 5
    ) -> list:
        """
        Rerank documents using Cohere's reranker model.
        """
        response = self.client.rerank(
            model="rerank-english-v2.0",
            query=query,
            documents=[d['prompt_text'] for d in documents],
            top_n=top_k
        )

        return [
            {
                **documents[r.index],
                'rerank_score': r.relevance_score
            }
            for r in response.results
        ]

Performance Tracking

Metric Structure

@dataclass
class PromptMetrics:
    """
    Performance metrics for a prompt.

    Updated using exponential moving average for recency weighting.
    """
    success_rate: float       # 0-1, task completion rate
    avg_latency_ms: float     # Response time
    token_efficiency: float   # quality_score / tokens_used
    coherence_score: float    # Logical consistency
    observation_count: int    # How many times evaluated
    last_updated: datetime    # For recency calculations

    def update(self, outcome: dict, alpha: float = 0.3):
        """
        Exponential moving average update.

        alpha = 0.3 means:
        - 30% weight to new observation
        - 70% weight to historical average
        """
        self.success_rate = alpha * float(outcome['success']) + (1 - alpha) * self.success_rate
        self.avg_latency_ms = alpha * outcome.get('latency_ms', self.avg_latency_ms) + (1 - alpha) * self.avg_latency_ms
        self.observation_count += 1
        self.last_updated = datetime.utcnow()

Temporal Weighting

class TemporalWeighting:
    """
    Apply time-decay to balance recency vs. historical performance.

    half_life: Time for weight to decay to 50%
    - 30 days: Standard for most use cases
    - 7 days: Fast-changing domains
    - 90 days: Stable domains
    """

    def __init__(self, half_life_days: int = 30):
        self.half_life = timedelta(days=half_life_days)

    def calculate_weight(self, observation_time: datetime) -> float:
        """
        Calculate time-decay weight.

        weight = 0.5^(time_diff / half_life)
        """
        time_diff = datetime.utcnow() - observation_time
        half_lives_elapsed = time_diff / self.half_life
        return math.pow(0.5, half_lives_elapsed)

    def weighted_score(
        self,
        observations: list[tuple[float, datetime]]
    ) -> float:
        """
        Calculate weighted average with time decay.
        """
        total_weight = 0
        weighted_sum = 0

        for value, timestamp in observations:
            weight = self.calculate_weight(timestamp)
            weighted_sum += value * weight
            total_weight += weight

        return weighted_sum / total_weight if total_weight > 0 else 0

Drift Detection

Distribution Shift Detection

class DriftDetector:
    """
    Detect when prompt distribution shifts (new domains, patterns).

    Triggers adaptive learning rate adjustment.
    """

    def __init__(
        self,
        window_size: int = 100,
        drift_threshold: float = 0.15
    ):
        self.window_size = window_size
        self.threshold = drift_threshold
        self.embedding_history = deque(maxlen=window_size)

    def check_drift(self, new_embedding: np.ndarray) -> bool:
        """
        Check if new embedding indicates distribution shift.

        Uses Maximum Mean Discrepancy (simplified).
        """
        if len(self.embedding_history) < 50:
            self.embedding_history.append(new_embedding)
            return False

        # Calculate centroid of historical embeddings
        historical = np.array(list(self.embedding_history))
        centroid = np.mean(historical, axis=0)

        # Measure distance from centroid
        distance = np.linalg.norm(new_embedding - centroid)

        # Normalize by average distance in history
        avg_distance = np.mean([
            np.linalg.norm(e - centroid)
            for e in historical
        ])

        normalized_distance = distance / avg_distance if avg_distance > 0 else 0

        self.embedding_history.append(new_embedding)

        return normalized_distance > (1 + self.threshold)

    def get_drift_metrics(self) -> dict:
        """
        Get drift detection metrics for monitoring.
        """
        if len(self.embedding_history) < 10:
            return {'status': 'insufficient_data'}

        historical = np.array(list(self.embedding_history))
        centroid = np.mean(historical, axis=0)

        distances = [np.linalg.norm(e - centroid) for e in historical]

        return {
            'avg_distance': np.mean(distances),
            'std_distance': np.std(distances),
            'max_distance': np.max(distances),
            'sample_size': len(self.embedding_history)
        }

Vector Database Selection

Comparison Matrix

Qdrant Setup (Recommended)

from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PayloadSchemaType

async def setup_qdrant():
    client = QdrantClient(url="http://localhost:6333")

    # Create collection
    await client.create_collection(
        collection_name="prompt_embeddings",
        vectors_config=VectorParams(
            size=3072,  # text-embedding-3-large
            distance=Distance.COSINE
        )
    )

    # Create indexes for filtering
    await client.create_payload_index(
        collection_name="prompt_embeddings",
        field_name="metrics.success_rate",
        field_schema=PayloadSchemaType.FLOAT
    )

    await client.create_payload_index(
        collection_name="prompt_embeddings",
        field_name="domain",
        field_schema=PayloadSchemaType.KEYWORD
    )

    await client.create_payload_index(
        collection_name="prompt_embeddings",
        field_name="created_at",
        field_schema=PayloadSchemaType.DATETIME
    )

    return client

Chroma Setup (Quick Start)

import chromadb
from chromadb.config import Settings

def setup_chroma():
    # Persistent storage
    client = chromadb.PersistentClient(
        path="./chroma_data",
        settings=Settings(anonymized_telemetry=False)
    )

    collection = client.get_or_create_collection(
        name="prompt_embeddings",
        metadata={"hnsw:space": "cosine"}
    )

    return collection

Caching Strategy

Embedding Cache

class EmbeddingCache:
    """
    Cache embeddings to reduce API calls.

    Cache hit rate typically 60-80% for repeated prompts.
    """

    def __init__(self, redis_client: Redis, ttl_hours: int = 24):
        self.redis = redis_client
        self.ttl = ttl_hours * 3600

    async def get_or_create(
        self,
        text: str,
        embed_fn: Callable
    ) -> list:
        """
        Get embedding from cache or create new.
        """
        cache_key = f"emb:{hashlib.sha256(text.encode()).hexdigest()}"

        # Try cache
        cached = await self.redis.get(cache_key)
        if cached:
            return json.loads(cached)

        # Generate new
        embedding = await embed_fn(text)

        # Cache with TTL
        await self.redis.setex(
            cache_key,
            self.ttl,
            json.dumps(embedding)
        )

        return embedding

Query Result Cache

class QueryCache:
    """
    Cache frequent queries for faster retrieval.

    Short TTL (5-15 min) since results may change.
    """

    def __init__(self, redis_client: Redis, ttl_minutes: int = 10):
        self.redis = redis_client
        self.ttl = ttl_minutes * 60

    async def get_or_query(
        self,
        query: str,
        filters: dict,
        query_fn: Callable
    ) -> list:
        """
        Get results from cache or execute query.
        """
        cache_key = f"query:{hashlib.sha256(f'{query}:{json.dumps(filters)}'.encode()).hexdigest()}"

        cached = await self.redis.get(cache_key)
        if cached:
            return json.loads(cached)

        results = await query_fn(query, filters)

        await self.redis.setex(
            cache_key,
            self.ttl,
            json.dumps(results)
        )

        return results

Performance Targets

---

This architecture provides a scalable foundation for stateful prompt learning that can grow from prototype to production.

Iteration Strategy

Decision framework for optimization depth.

Decision Matrix

Complexity Assessment

Complexity Score (0-1):
- Task decomposition depth (0.3): How many sub-tasks?
- Reasoning steps required (0.3): Chain length
- Domain specificity (0.2): Specialized knowledge?
- Output structure (0.2): Format complexity

Score < 0.3 → 3-5 iterations
Score 0.3-0.6 → 5-10 iterations
Score > 0.6 → 10-20 iterations

Ambiguity Assessment

Ambiguity Score (0-1):
- Requirement clarity (0.4): Are success criteria explicit?
- Interpretation variance (0.3): Multiple valid readings?
- Example availability (0.3): Concrete examples provided?

High ambiguity → +5 iterations (for exploration)

Decision Tree: When to Iterate More

START
│
├─ Is task critical? ──YES──→ +5 iterations
│
├─ Is domain novel? ──YES──→ +5 iterations
│
├─ Are requirements ambiguous? ──YES──→ +5 iterations
│
├─ Do I have similar prompts? ──YES──→ -3 iterations (start better)
│
└─ Base: 5 iterations

TOTAL = Base + adjustments (min 3, max 20)

Convergence Criteria

Stop iterating when: 1. Performance plateau: No improvement > 1% over last 3 iterations 2. Diminishing returns: Cost per improvement unit exceeds threshold 3. Statistical significance: Confidence interval < 2% 4. Budget exhausted: Max iterations or token limit reached

def check_convergence(scores, window=3, threshold=0.01):
    """Stop if improvement < threshold over window iterations"""
    if len(scores) < window:
        return False
    recent_improvement = max(scores[-window:]) - scores[-window]
    return recent_improvement < threshold

Stop Conditions

STOP if ANY:
- Improvement < 1% for 3 consecutive iterations
- User signals satisfaction
- Token budget exhausted
- 20 iterations reached
- Validation score > 0.95

Performance Expectations

Learning Architecture

Stateful learning through embedding-indexed history.

Retrieval Strategy (Warm Start)

1. Embed the current prompt with contextual metadata:
   - Domain classification
   - Task type
   - Complexity score

2. Hybrid search (vector similarity + BM25 keyword):
   - Retrieve top-20 candidates
   - Rerank to top-5

3. Filter by performance threshold:
   - Only prompts with success_rate > 0.7
   - Weight by recency (exponential decay, half-life: 30 days)

4. Generate improvements based on:
   - What made similar prompts succeed
   - Patterns in high-performing variants

Performance Metrics

Continuous Learning Loop

1. OBSERVE: Track prompt → outcome pairs
   - Store prompt embedding
   - Record performance metrics
   - Capture context (domain, task type, complexity)

2. INDEX: Build retrievable knowledge base
   - Vector database for semantic similarity
   - BM25 for keyword matching
   - Metadata for filtering

3. UPDATE: Exponential moving average
   - α = 0.3 (30% new, 70% historical)
   - Recency decay: half-life 30 days
   - Distribution drift detection

4. ADAPT: Adjust strategies based on patterns
   - Which techniques work for which domains?
   - What iteration counts converge fastest?
   - Where do certain patterns fail?

Drift Detection

When the distribution of prompts shifts (new domain, new patterns):

If embedding_drift > 0.15:
  - Increase learning rate (α → 0.5)
  - Flag for human review
  - Log potential new domain

MCP Server Setup

To enable persistent learning:

{
  "mcpServers": {
    "prompt-learning": {
      "command": "node",
      "args": ["path/to/prompt-learning-server/index.js"],
      "env": {
        "VECTOR_DB_URL": "your-vector-db-url",
        "REDIS_URL": "your-redis-url"
      }
    }
  }
}

Required Tools (exposed by MCP):

• retrieve_prompts: Get similar high-performing prompts
• record_feedback: Update prompt metrics
• suggest_improvements: RAG-based optimization
• get_analytics: Performance insights

Cold Start Mode

Without MCP, operate in cold-start mode only:

• Apply research-backed strategies
• No retrieval from history
• No persistent learning
• Still effective, just not stateful

MCP Server Specification: Prompt Learning Server

This document specifies the MCP server implementation for stateful prompt learning.

Overview

The prompt-learning MCP server provides persistent storage and retrieval of prompt performance data, enabling the skill to learn and improve over time.

Architecture

┌─────────────────────────────────────────────────────┐
│                 Claude Code                          │
│                                                      │
│  ┌──────────────────────────────────────────────┐   │
│  │     automatic-stateful-prompt-improver        │   │
│  └────────────────────┬─────────────────────────┘   │
│                       │ MCP Protocol                 │
└───────────────────────┼─────────────────────────────┘
                        │
┌───────────────────────┼─────────────────────────────┐
│                       ▼                              │
│            prompt-learning MCP Server                │
│                                                      │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐  │
│  │   Tools     │  │  Resources  │  │   State     │  │
│  │             │  │             │  │             │  │
│  │ retrieve    │  │ performance │  │   Redis     │  │
│  │ record      │  │ analytics   │  │   Cache     │  │
│  │ suggest     │  │ history     │  │             │  │
│  └──────┬──────┘  └──────┬──────┘  └──────┬──────┘  │
│         │                │                │          │
│         └────────────────┼────────────────┘          │
│                          │                           │
└──────────────────────────┼───────────────────────────┘
                           │
         ┌─────────────────┼─────────────────┐
         │                 ▼                 │
         │          Vector Database          │
         │     (Qdrant / Pinecone / Chroma)  │
         │                                   │
         │  ┌─────────────────────────────┐  │
         │  │     Prompt Embeddings       │  │
         │  │   + Performance Metrics     │  │
         │  │   + Metadata                │  │
         │  └─────────────────────────────┘  │
         │                                   │
         └───────────────────────────────────┘

Tools

1. retrieve_prompts

Retrieve similar high-performing prompts from the embedding database.

Input Schema:

{
  "type": "object",
  "properties": {
    "query": {
      "type": "string",
      "description": "The prompt to find similar examples for"
    },
    "domain": {
      "type": "string",
      "description": "Domain classification (e.g., 'code_review', 'summarization')"
    },
    "top_k": {
      "type": "integer",
      "default": 5,
      "description": "Number of results to return"
    },
    "min_performance": {
      "type": "number",
      "default": 0.7,
      "description": "Minimum success_rate threshold"
    }
  },
  "required": ["query"]
}

Output:

{
  "results": [
    {
      "prompt_id": "uuid",
      "prompt_text": "string",
      "similarity_score": 0.92,
      "metrics": {
        "success_rate": 0.85,
        "avg_latency_ms": 450,
        "token_efficiency": 0.78
      },
      "domain": "code_review",
      "created_at": "2024-01-15T10:30:00Z"
    }
  ]
}

2. record_feedback

Record the outcome of a prompt execution for learning.

Input Schema:

{
  "type": "object",
  "properties": {
    "prompt_id": {
      "type": "string",
      "description": "ID of the prompt (or 'new' to create)"
    },
    "prompt_text": {
      "type": "string",
      "description": "The prompt text (required if prompt_id is 'new')"
    },
    "domain": {
      "type": "string",
      "description": "Domain classification"
    },
    "outcome": {
      "type": "object",
      "properties": {
        "success": {"type": "boolean"},
        "latency_ms": {"type": "number"},
        "output_tokens": {"type": "integer"},
        "quality_score": {"type": "number", "minimum": 0, "maximum": 1}
      },
      "required": ["success"]
    },
    "user_feedback": {
      "type": "object",
      "properties": {
        "satisfaction": {"type": "number", "minimum": 0, "maximum": 1},
        "comments": {"type": "string"}
      }
    }
  },
  "required": ["prompt_id", "outcome"]
}

Output:

{
  "status": "recorded",
  "prompt_id": "uuid",
  "updated_metrics": {
    "success_rate": 0.82,
    "observation_count": 15
  }
}

3. suggest_improvements

Generate improvement suggestions based on similar high-performers.

Input Schema:

{
  "type": "object",
  "properties": {
    "prompt": {
      "type": "string",
      "description": "The prompt to improve"
    },
    "current_performance": {
      "type": "object",
      "properties": {
        "success_rate": {"type": "number"},
        "avg_latency_ms": {"type": "number"},
        "token_efficiency": {"type": "number"}
      }
    },
    "improvement_focus": {
      "type": "string",
      "enum": ["clarity", "specificity", "efficiency", "all"],
      "default": "all"
    }
  },
  "required": ["prompt"]
}

Output:

{
  "suggestions": [
    {
      "type": "add_structure",
      "description": "Add numbered output format",
      "example": "Respond with:\n1. Summary\n2. Key points\n3. Recommendations",
      "expected_improvement": 0.15
    }
  ],
  "based_on": {
    "similar_prompts_analyzed": 5,
    "avg_performance_of_similar": 0.87
  }
}

4. get_analytics

Retrieve performance analytics and trends.

Input Schema:

{
  "type": "object",
  "properties": {
    "domain": {
      "type": "string",
      "description": "Filter by domain (optional)"
    },
    "time_range": {
      "type": "string",
      "enum": ["7d", "30d", "90d", "all"],
      "default": "30d"
    },
    "metrics": {
      "type": "array",
      "items": {"type": "string"},
      "default": ["success_rate", "token_efficiency"]
    }
  }
}

Output:

{
  "summary": {
    "total_prompts": 150,
    "avg_success_rate": 0.78,
    "improvement_trend": 0.05
  },
  "by_domain": {
    "code_review": {"count": 45, "avg_success": 0.82},
    "summarization": {"count": 30, "avg_success": 0.75}
  },
  "top_patterns": [
    {"pattern": "structured_output", "success_rate": 0.89},
    {"pattern": "chain_of_thought", "success_rate": 0.85}
  ]
}

Implementation

TypeScript Server

import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { QdrantClient } from "@qdrant/js-client-rest";
import Redis from "ioredis";

// Configuration
const EMBEDDING_MODEL = "text-embedding-3-large";
const EMBEDDING_DIM = 3072;
const COLLECTION_NAME = "prompt_embeddings";

interface PromptMetrics {
  success_rate: number;
  avg_latency_ms: number;
  token_efficiency: number;
  observation_count: number;
}

class PromptLearningServer {
  private server: Server;
  private vectorDb: QdrantClient;
  private cache: Redis;
  private embeddingClient: any; // OpenAI or similar

  constructor() {
    this.server = new Server(
      { name: "prompt-learning", version: "1.0.0" },
      { capabilities: { tools: {}, resources: {} } }
    );

    this.vectorDb = new QdrantClient({
      url: process.env.VECTOR_DB_URL || "http://localhost:6333"
    });

    this.cache = new Redis(process.env.REDIS_URL || "redis://localhost:6379");

    this.registerTools();
  }

  private registerTools() {
    this.server.setRequestHandler("tools/list", async () => ({
      tools: [
        {
          name: "retrieve_prompts",
          description: "Find similar high-performing prompts",
          inputSchema: { /* schema above */ }
        },
        {
          name: "record_feedback",
          description: "Record prompt execution outcome",
          inputSchema: { /* schema above */ }
        },
        {
          name: "suggest_improvements",
          description: "Generate improvement suggestions",
          inputSchema: { /* schema above */ }
        },
        {
          name: "get_analytics",
          description: "Get performance analytics",
          inputSchema: { /* schema above */ }
        }
      ]
    }));

    this.server.setRequestHandler("tools/call", async (request) => {
      const { name, arguments: args } = request.params;

      switch (name) {
        case "retrieve_prompts":
          return await this.retrievePrompts(args);
        case "record_feedback":
          return await this.recordFeedback(args);
        case "suggest_improvements":
          return await this.suggestImprovements(args);
        case "get_analytics":
          return await this.getAnalytics(args);
        default:
          throw new Error(`Unknown tool: ${name}`);
      }
    });
  }

  private async retrievePrompts(args: any) {
    // 1. Generate embedding for query
    const queryEmbedding = await this.getEmbedding(args.query);

    // 2. Search vector DB
    const results = await this.vectorDb.search(COLLECTION_NAME, {
      vector: queryEmbedding,
      limit: args.top_k * 2, // Over-retrieve for filtering
      filter: {
        must: [
          { key: "metrics.success_rate", range: { gte: args.min_performance } }
        ],
        ...(args.domain && {
          must: [{ key: "domain", match: { value: args.domain } }]
        })
      }
    });

    // 3. Format and return
    return {
      content: [{
        type: "text",
        text: JSON.stringify({
          results: results.slice(0, args.top_k).map(r => ({
            prompt_id: r.id,
            prompt_text: r.payload?.prompt_text,
            similarity_score: r.score,
            metrics: r.payload?.metrics,
            domain: r.payload?.domain
          }))
        })
      }]
    };
  }

  private async recordFeedback(args: any) {
    const alpha = 0.3; // EMA weight for new observations

    // Get or create prompt record
    let metrics: PromptMetrics;
    if (args.prompt_id === "new") {
      // Create new prompt
      const embedding = await this.getEmbedding(args.prompt_text);
      const promptId = crypto.randomUUID();

      metrics = {
        success_rate: args.outcome.success ? 1.0 : 0.0,
        avg_latency_ms: args.outcome.latency_ms || 0,
        token_efficiency: args.outcome.quality_score || 0,
        observation_count: 1
      };

      await this.vectorDb.upsert(COLLECTION_NAME, {
        points: [{
          id: promptId,
          vector: embedding,
          payload: {
            prompt_text: args.prompt_text,
            domain: args.domain,
            metrics,
            created_at: new Date().toISOString()
          }
        }]
      });

      args.prompt_id = promptId;
    } else {
      // Update existing
      const existing = await this.vectorDb.retrieve(COLLECTION_NAME, {
        ids: [args.prompt_id],
        with_payload: true
      });

      if (existing.length === 0) {
        throw new Error("Prompt not found");
      }

      const oldMetrics = existing[0].payload?.metrics as PromptMetrics;

      // Exponential moving average update
      metrics = {
        success_rate: alpha * (args.outcome.success ? 1 : 0) + (1 - alpha) * oldMetrics.success_rate,
        avg_latency_ms: alpha * (args.outcome.latency_ms || 0) + (1 - alpha) * oldMetrics.avg_latency_ms,
        token_efficiency: alpha * (args.outcome.quality_score || oldMetrics.token_efficiency) + (1 - alpha) * oldMetrics.token_efficiency,
        observation_count: oldMetrics.observation_count + 1
      };

      await this.vectorDb.setPayload(COLLECTION_NAME, {
        points: [args.prompt_id],
        payload: { metrics }
      });
    }

    return {
      content: [{
        type: "text",
        text: JSON.stringify({
          status: "recorded",
          prompt_id: args.prompt_id,
          updated_metrics: metrics
        })
      }]
    };
  }

  private async suggestImprovements(args: any) {
    // Retrieve similar high-performers
    const similar = await this.retrievePrompts({
      query: args.prompt,
      top_k: 5,
      min_performance: 0.8
    });

    // Analyze patterns in high performers
    // (In production, this would use an LLM to generate suggestions)

    return {
      content: [{
        type: "text",
        text: JSON.stringify({
          suggestions: [
            // Pattern-based suggestions
          ],
          based_on: {
            similar_prompts_analyzed: 5,
            avg_performance_of_similar: 0.87
          }
        })
      }]
    };
  }

  private async getAnalytics(args: any) {
    // Aggregate metrics from vector DB
    // (Implementation depends on specific vector DB capabilities)

    return {
      content: [{
        type: "text",
        text: JSON.stringify({
          summary: {
            total_prompts: 0,
            avg_success_rate: 0,
            improvement_trend: 0
          }
        })
      }]
    };
  }

  private async getEmbedding(text: string): Promise<number[]> {
    // Call embedding API (OpenAI, Cohere, etc.)
    // Implementation depends on your embedding provider
    return [];
  }

  async start() {
    const transport = new StdioServerTransport();
    await this.server.connect(transport);
  }
}

const server = new PromptLearningServer();
server.start();

Setup Instructions

1. Prerequisites

# Install dependencies
npm install @modelcontextprotocol/sdk @qdrant/js-client-rest ioredis openai

# Start Qdrant (Docker)
docker run -p 6333:6333 qdrant/qdrant

# Start Redis (Docker)
docker run -p 6379:6379 redis

2. Initialize Vector Collection

const client = new QdrantClient({ url: "http://localhost:6333" });

await client.createCollection("prompt_embeddings", {
  vectors: {
    size: 3072, // text-embedding-3-large
    distance: "Cosine"
  }
});

// Create payload indexes for filtering
await client.createPayloadIndex("prompt_embeddings", {
  field_name: "metrics.success_rate",
  field_schema: "float"
});

await client.createPayloadIndex("prompt_embeddings", {
  field_name: "domain",
  field_schema: "keyword"
});

3. Configure Claude Code

Add to your claude_desktop_config.json or project config:

{
  "mcpServers": {
    "prompt-learning": {
      "command": "node",
      "args": ["/path/to/prompt-learning-server/dist/index.js"],
      "env": {
        "VECTOR_DB_URL": "http://localhost:6333",
        "REDIS_URL": "redis://localhost:6379",
        "OPENAI_API_KEY": "sk-..."
      }
    }
  }
}

Data Model

Prompt Embedding Record

interface PromptRecord {
  id: string;                    // UUID
  vector: number[];              // Embedding (3072 dim)
  payload: {
    prompt_text: string;         // Original prompt
    contextualized: string;      // With domain context
    domain: string;              // Classification
    task_type: string;           // e.g., "classification", "generation"
    metrics: {
      success_rate: number;      // 0-1, EMA
      avg_latency_ms: number;    // Response time
      token_efficiency: number;  // quality/tokens
      observation_count: number; // How many times evaluated
    };
    created_at: string;          // ISO timestamp
    updated_at: string;          // Last metric update
    tags: string[];              // User-defined tags
  };
}

Session State (Redis)

interface SessionState {
  session_id: string;
  user_id: string;
  current_prompt_id: string | null;
  iteration_count: number;
  best_score: number;
  history: Array<{
    prompt_id: string;
    score: number;
    timestamp: string;
  }>;
}

Security Considerations

1. Data Isolation: Use Redis key prefixes for multi-tenant isolation 2. API Keys: Store securely, rotate regularly 3. Rate Limiting: Implement per-user rate limits 4. Encryption: TLS for all connections, encryption at rest for sensitive prompts 5. GDPR: Implement user data deletion workflows

Monitoring

Track these metrics:

• Embedding latency (p50, p95, p99)
• Retrieval latency (p50, p95, p99)
• Success rate improvement over time
• Cache hit rate
• Error rates by tool

---

This MCP server enables the prompt improver skill to learn and adapt over time, making each optimization better informed by past successes.

Optimization Techniques

Research-backed strategies for prompt improvement.

APE-Style Instruction Generation

Source: Zhou et al., 2022 (outperformed human prompts on 19/24 NLP tasks)

Given your task and examples:
Task: [description]
Examples: Input → Output pairs

Generate 5-10 instruction candidates, then select best based on:
- Clarity score (0-1): How unambiguous is the instruction?
- Completeness (0-1): Does it cover all requirements?
- Constraint density: Appropriate constraints without over-specification

OPRO Meta-Prompting

Source: Yang et al., 2023 (up to 50% improvement on Big-Bench Hard)

Treat prompt optimization AS a prompting task:

Previous prompt attempts and their scores:
- "Step by step, solve..." | Score: 0.65
- "Carefully analyze each..." | Score: 0.72

Generate a new instruction that will achieve a higher score.

Chain-of-Thought Enhancement

Source: Wei et al., 2022 (best for complex reasoning)

Add reasoning scaffolds:

• "Let's think step by step" (zero-shot CoT)
• Structured reasoning templates
• Self-consistency through multiple paths

Task Decomposition

Break complex prompts into modular components: 1. Context setting - Domain and background 2. Instruction specification - What to do 3. Output format - How to respond 4. Constraints - What to avoid

Instruction Rewriting

Pattern: Generate variants, evaluate, select best

Original: "Summarize this text"

Generated Variants:
1. "Extract the key points from this text in bullet form"
2. "Provide a concise 2-3 sentence summary capturing the main argument"
3. "Identify the thesis and supporting evidence, then summarize"

Selection Criteria:
- Specificity: Variant 2 wins (format specified)
- Clarity: Variant 1 wins (clear structure)
- Completeness: Variant 3 wins (methodology included)

Best: Combine insights → "Extract the thesis and key supporting points,
then provide a concise 2-3 sentence summary in bullet form"

Few-Shot Optimization

Pattern: Select examples that maximize performance

Selection Methods:
1. Semantic similarity: Examples similar to test case
2. Diversity: Cover different scenarios
3. Difficulty progression: Easy → Hard examples
4. Contrastive: Include near-misses for boundary learning

Constraint Engineering

Pattern: Add/remove constraints to improve output

Under-constrained:
"Write code for a sorting function"

After Constraint Engineering:
"Write a Python function that sorts a list of integers in ascending order.
Requirements:
- Use O(n log n) time complexity
- Handle empty lists gracefully
- Include type hints
- Do not use built-in sort()"

Output Format Specification

Pattern: Explicit format reduces ambiguity

Before: "Analyze this data"

After: "Analyze this data and respond with:
1. **Summary** (2-3 sentences): Key findings
2. **Metrics** (bullet list): Quantitative observations
3. **Recommendations** (numbered): Actionable next steps"

Setup Guide: Automatic Stateful Prompt Improver

This guide walks you through setting up the full stateful learning infrastructure for the prompt improver skill.

One-Command Install

The fastest way to get started:

curl -fsSL https://someclaudeskills.com/install/prompt-learning.sh | bash

Or clone and install manually:

git clone https://github.com/erichowens/prompt-learning-mcp.git ~/mcp-servers/prompt-learning
cd ~/mcp-servers/prompt-learning
npm install && npm run build
npm run setup

Repository: github.com/erichowens/prompt-learning-mcp

Quick Start (Cold Start Mode)

No setup required! The skill works immediately in cold-start mode:

• Uses research-backed optimization strategies (APE, OPRO, DSPy patterns)
• No persistent learning, but still highly effective
• Great for testing before investing in infrastructure

Just invoke the skill:

"optimize this prompt: [your prompt]"

Full Setup (Stateful Learning Mode)

For persistent learning across conversations, you need: 1. Vector database (prompt embeddings) 2. Redis (session cache, metrics) 3. MCP server (protocol bridge)

Prerequisites

# Docker (recommended for quick setup)
docker --version  # Ensure Docker is installed

# Node.js 18+ (for MCP server)
node --version

# OpenAI API key (for embeddings and evaluation)
export OPENAI_API_KEY=sk-your-key-here

Step 1: Start Vector Database (Qdrant)

Qdrant is recommended for its hybrid search capabilities (vector + BM25).

# Quick start with Docker
docker run -d \
  --name qdrant \
  -p 6333:6333 \
  -p 6334:6334 \
  -v qdrant_storage:/qdrant/storage \
  qdrant/qdrant

# Verify it's running
curl http://localhost:6333/collections

Alternative: Chroma (simpler, smaller scale)

docker run -d \
  --name chroma \
  -p 8000:8000 \
  -v chroma_data:/chroma/chroma \
  chromadb/chroma

Step 2: Start Redis (Session Cache)

docker run -d \
  --name redis \
  -p 6379:6379 \
  -v redis_data:/data \
  redis:alpine redis-server --appendonly yes

# Verify
redis-cli ping  # Should return PONG

Step 3: Initialize Vector Collection

Create the prompt embeddings collection:

# Using curl
curl -X PUT http://localhost:6333/collections/prompt_embeddings \
  -H "Content-Type: application/json" \
  -d '{
    "vectors": {
      "size": 3072,
      "distance": "Cosine"
    }
  }'

# Create indexes for filtering
curl -X PUT http://localhost:6333/collections/prompt_embeddings/index \
  -H "Content-Type: application/json" \
  -d '{
    "field_name": "metrics.success_rate",
    "field_schema": "float"
  }'

curl -X PUT http://localhost:6333/collections/prompt_embeddings/index \
  -H "Content-Type: application/json" \
  -d '{
    "field_name": "domain",
    "field_schema": "keyword"
  }'

Step 4: Setup MCP Server

Create the MCP server project:

mkdir -p ~/mcp-servers/prompt-learning
cd ~/mcp-servers/prompt-learning
npm init -y
npm install @modelcontextprotocol/sdk @qdrant/js-client-rest ioredis openai

Create index.js:

// See /references/mcp-server-spec.md for full implementation
// This is a minimal working version

import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { QdrantClient } from "@qdrant/js-client-rest";
import Redis from "ioredis";
import OpenAI from "openai";

const vectorDb = new QdrantClient({ url: process.env.VECTOR_DB_URL || "http://localhost:6333" });
const redis = new Redis(process.env.REDIS_URL || "redis://localhost:6379");
const openai = new OpenAI({ apiKey: process.env.OPENAI_API_KEY });

const server = new Server(
  { name: "prompt-learning", version: "1.0.0" },
  { capabilities: { tools: {} } }
);

// Tool implementations...
// (See full spec in references/mcp-server-spec.md)

const transport = new StdioServerTransport();
await server.connect(transport);

Step 5: Configure Claude Code

Add to your Claude Code configuration (~/.claude.json or project config):

{
  "mcpServers": {
    "prompt-learning": {
      "command": "node",
      "args": ["/path/to/mcp-servers/prompt-learning/index.js"],
      "env": {
        "VECTOR_DB_URL": "http://localhost:6333",
        "REDIS_URL": "redis://localhost:6379",
        "OPENAI_API_KEY": "sk-your-key-here"
      }
    }
  }
}

Step 6: Verify Setup

Test the MCP server is working:

# In Claude Code, try:
"retrieve similar prompts for: summarize the document"

If you see results from the vector database, you're all set!

Architecture Overview

┌────────────────────────────────────────────┐
│           Claude Code                       │
│                                            │
│  ┌────────────────────────────────────┐   │
│  │  automatic-stateful-prompt-improver │   │
│  │                                     │   │
│  │  Cold Start Mode:                   │   │
│  │  - APE/OPRO patterns               │   │
│  │  - DSPy-style optimization         │   │
│  │  - Pattern-based improvement       │   │
│  │                                     │   │
│  │  Warm Start Mode (with MCP):       │   │
│  │  - Retrieve similar prompts        │   │
│  │  - Learn from outcomes             │   │
│  │  - Adaptive iteration count        │   │
│  └─────────────┬──────────────────────┘   │
│                │                           │
│                │ MCP Protocol              │
└────────────────┼───────────────────────────┘
                 │
┌────────────────┼───────────────────────────┐
│                ▼                            │
│     prompt-learning MCP Server              │
│                                            │
│  Tools:                                    │
│  - retrieve_prompts                        │
│  - record_feedback                         │
│  - suggest_improvements                    │
│  - get_analytics                           │
│                                            │
└─────────┬──────────────┬───────────────────┘
          │              │
          ▼              ▼
    ┌──────────┐   ┌──────────┐
    │  Qdrant  │   │  Redis   │
    │ (Vector) │   │ (Cache)  │
    └──────────┘   └──────────┘

Embedding Model Options

The system uses embeddings to find similar prompts. Choose based on your needs:

Recommendation: Start with text-embedding-3-small, upgrade if needed.

To change models, update the collection:

# For 1536-dim model
curl -X PUT http://localhost:6333/collections/prompt_embeddings \
  -d '{"vectors": {"size": 1536, "distance": "Cosine"}}'

Cost Considerations

Monthly Estimates (assuming moderate usage)

Total self-hosted: ~$15-70/month Total cloud-managed: ~$50-200/month

Cost Optimization Tips

1. Cache embeddings: Same prompts don't need re-embedding 2. Batch operations: Group embedding calls 3. Use smaller models: text-embedding-3-small is often sufficient 4. Rate limit: Prevent runaway API calls

Troubleshooting

MCP Server Not Connecting

# Check if server is running
ps aux | grep prompt-learning

# Test manually
cd ~/mcp-servers/prompt-learning
node index.js

# Check Claude Code logs
tail -f ~/.claude/logs/mcp.log

Vector Database Issues

# Check Qdrant health
curl http://localhost:6333/health

# Check collection exists
curl http://localhost:6333/collections/prompt_embeddings

# Recreate if corrupted
curl -X DELETE http://localhost:6333/collections/prompt_embeddings
# Then re-run Step 3

Redis Connection Issues

# Check Redis is running
docker ps | grep redis

# Test connection
redis-cli ping

# Check memory usage
redis-cli info memory

Scaling Considerations

When to Scale

• >100k prompts: Consider Qdrant cloud or cluster
• >10 QPS: Add Redis caching layer
• Multi-user: Add tenant isolation

Production Checklist

• [ ] TLS enabled for all connections
• [ ] API keys in secrets manager
• [ ] Backup strategy for vector DB
• [ ] Monitoring and alerting
• [ ] Rate limiting
• [ ] GDPR data deletion workflows

Next Steps

1. Start in cold-start mode: Test the skill without infrastructure 2. Add basic persistence: Set up Qdrant + Redis locally 3. Collect data: Use the skill, let it learn 4. Analyze patterns: Use get_analytics to see what works 5. Scale as needed: Move to managed services if usage grows

---

Questions? Issues? The skill's /references directory has detailed implementation specs.