# 2618 AI Prompt Training Techniques 2025 11 30 28d9efbd

> AI Prompt Training and Optimization Techniques - 2025 Research

- Skill: `tools-only/2618-ai-prompt-training-techniques-2025-11-30-28d9efbd` (Agent Skill, multi-file: 3 files)
- Install (CLI): `npx skillmds@latest add tools-only/2618-ai-prompt-training-techniques-2025-11-30-28d9efbd`
- Raw SKILL.md: https://api.skillmd.com/api/skills/tools-only/2618-ai-prompt-training-techniques-2025-11-30-28d9efbd/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: AI & ML
- Author: tools-only (https://skillmd.com/u/tools-only)
- Updated: 2026-09-22
- Page: https://skillmd.com/skills/tools-only/2618-ai-prompt-training-techniques-2025-11-30-28d9efbd

---

# AI Prompt Training and Optimization Techniques - 2025 Research

**Research Date**: 2025-11-30
**Research Focus**: Methods for training, optimizing, and improving prompts programmatically
**Target Audience**: AI/ML engineers, prompt engineers, LLM application developers

---

## Executive Summary

This research document provides a comprehensive analysis of prompt training and optimization techniques as of 2025. The landscape has matured significantly with three primary approaches:

1. **DSPy Framework**: Automated prompt optimization treating prompts as code with declarative programming
2. **LangGraph Workflows**: Stateful multi-agent orchestration with graph-based prompt coordination
3. **Traditional Patterns**: Manual prompt engineering patterns (CoT, ToT, ReAct) with evaluation frameworks

**Key Finding**: The industry is moving from manual prompt engineering toward programmatic optimization, with DSPy leading automated optimization and LangGraph dominating complex multi-agent systems.

**Production Adoption**: Companies including JetBlue, Databricks, Walmart, VMware, Replit, Sephora, and Moody's use DSPy in production as of 2025.

---

## Table of Contents

1. [DSPy Framework](#1-dspy-framework)
2. [Prompt Engineering Patterns](#2-prompt-engineering-patterns)
3. [LangGraph Workflows](#3-langgraph-workflows)
4. [Simpler Approaches Without LangGraph](#4-simpler-approaches-without-langgraph)
5. [Evaluation Frameworks](#5-evaluation-frameworks)
6. [Production Patterns](#6-production-patterns)
7. [Comparison Matrix](#7-comparison-matrix)
8. [Use Case Recommendations](#8-use-case-recommendations)
9. [Tool Recommendations](#9-tool-recommendations)

---

## 1. DSPy Framework

### Overview

DSPy (Declarative Self-improving Python) is a framework for **programming—not prompting—language models**. It shifts focus from manual prompt engineering to declarative natural-language modules that can be automatically optimized.

**Core Philosophy**: Treat prompts as code with version control, automated testing, and systematic optimization.

### Key Features

#### Declarative Programming Model
```python
import dspy

# Define signature (input → output)
class QA(dspy.Signature):
    """Answer questions with short factual answers."""
    question = dspy.InputField()
    answer = dspy.OutputField(desc="often between 1 and 5 words")

# Create module
qa_module = dspy.ChainOfThought(QA)

# Use module
response = qa_module(question="What is the capital of France?")
print(response.answer)  # "Paris"
```

#### Automatic Prompt Optimization

DSPy provides **optimizers** that compile high-level code into optimized prompts or weight updates:

- **BootstrapFewShot**: Optimizes few-shot examples
- **COPRO**: Optimizes instruction prompts
- **MIPRO/MIPROv2**: Optimizes both instructions and examples jointly
- **KNN**: K-nearest neighbors example selection

### DSPy Optimizers Deep Dive

#### BootstrapFewShot

**Best For**: Small datasets (10-50 examples)
**Optimizes**: Few-shot examples only

```python
from dspy.teleprompt import BootstrapFewShot

# Define metric
def accuracy_metric(example, prediction, trace=None):
    return example.answer.lower() == prediction.answer.lower()

# Configure optimizer
fewshot_optimizer = BootstrapFewShot(
    metric=accuracy_metric,
    max_bootstrapped_demos=4,      # Max examples to bootstrap
    max_labeled_demos=16,           # Max labeled examples to use
    max_rounds=1,                   # Bootstrapping rounds
    max_errors=10                   # Max errors before stopping
)

# Compile program
optimized_program = fewshot_optimizer.compile(
    student=qa_module,
    trainset=training_examples
)
```

**How It Works**:
1. Uses your program to generate outputs on training data
2. Filters successful traces (based on metric)
3. Selects representative examples as few-shot demonstrations
4. Compiles optimized program with best examples

#### BootstrapFewShotWithRandomSearch

**Best For**: Medium datasets (50-300 examples)
**Optimizes**: Few-shot examples with candidate exploration

```python
from dspy.teleprompt import BootstrapFewShotWithRandomSearch

config = dict(
    max_bootstrapped_demos=4,
    max_labeled_demos=4,
    num_candidate_programs=10,    # Number of candidates to explore
    num_threads=4                 # Parallel threads
)

teleprompter = BootstrapFewShotWithRandomSearch(
    metric=accuracy_metric,
    **config
)

optimized_program = teleprompter.compile(
    qa_module,
    trainset=training_examples
)
```

**Advantage**: Explores multiple candidate programs in parallel, selecting the best performer.

#### MIPROv2 (State-of-the-Art as of 2025)

**Best For**: Large datasets (300+ examples)
**Optimizes**: Instructions AND few-shot examples jointly
**Method**: Bayesian Optimization

```python
import dspy
from dspy.teleprompt import MIPROv2

# Initialize LM
lm = dspy.LM('openai/gpt-4o-mini', api_key='YOUR_API_KEY')
dspy.configure(lm=lm)

# Define metric
def custom_metric(example, prediction, trace=None):
    # Custom scoring logic
    return prediction.score > 0.8

# Initialize MIPROv2 with auto-configuration
teleprompter = MIPROv2(
    metric=custom_metric,
    auto="medium",  # Options: light, medium, heavy
    # auto="medium" automatically sets hyperparameters
)

# Optimize program
optimized_program = teleprompter.compile(
    dspy.ChainOfThought("question -> answer"),
    trainset=training_examples,
)

# Save optimized program
optimized_program.save("optimized_qa_model.json")
```

**MIPROv2 Auto-Configuration Modes** (2025 Update):
- **light**: Fast optimization, fewer iterations, lower compute
- **medium**: Balanced optimization (recommended default)
- **heavy**: Exhaustive optimization, highest quality, most compute

**How MIPROv2 Works**:
1. **Bootstrap Few-Shot Candidates**: Generates example candidates from training data
2. **Propose Instructions**: Creates instruction variations grounded in task dynamics
3. **Bayesian Optimization**: Finds optimal combination of instructions + examples
4. **Joint Optimization**: Optimizes both components together (not separately)

### Sequential Optimization Strategy

For best results, combine optimizers:

```python
# Step 1: Bootstrap few-shot examples
bootstrap = dspy.BootstrapFewShot(metric=accuracy_metric)
bootstrapped_program = bootstrap.compile(qa_module, trainset=examples)

# Step 2: Optimize instructions with MIPRO
mipro = dspy.MIPROv2(metric=accuracy_metric, auto="light")
final_program = mipro.compile(bootstrapped_program, trainset=examples)

# Step 3: Save final optimized program
final_program.save("production_model.json")
```

### Real-World Performance (2025 Study)

A 2025 study applied DSPy to five use cases:

| Use Case | Baseline Accuracy | DSPy Optimized | Improvement |
|----------|------------------|----------------|-------------|
| Prompt Evaluation | 46.2% | 64.0% | +38.5% |
| Guardrail Enforcement | 72.1% | 84.3% | +16.9% |
| Code Generation | 58.4% | 71.2% | +21.9% |
| Hallucination Detection | 65.8% | 79.5% | +20.8% |
| Agent Routing | 69.3% | 82.1% | +18.5% |

**Source**: "Is It Time To Treat Prompts As Code? A Multi-Use Case Study For Prompt Optimization Using DSPy" (arXiv:2507.03620, 2025)

### When to Use DSPy

✅ **Use DSPy When**:
- You have structured input/output requirements
- You have evaluation datasets (even small ones)
- You need systematic prompt improvement
- You want version-controlled, reproducible prompts
- You're building production systems requiring optimization

❌ **Don't Use DSPy When**:
- You have zero training examples
- Task requires extreme creativity/open-endedness
- You need immediate results without setup
- Your task changes frequently (no stable evaluation)

---

## 2. Prompt Engineering Patterns

### Chain-of-Thought (CoT)

**Purpose**: Enable complex reasoning through intermediate steps

**Technique**: Encourage the model to "think step by step"

#### Zero-Shot CoT

```python
prompt = """
Question: Roger has 5 tennis balls. He buys 2 more cans of tennis balls.
Each can has 3 tennis balls. How many tennis balls does he have now?

Let's think step by step.
"""

response = llm(prompt)
# Output:
# 1. Roger starts with 5 tennis balls
# 2. He buys 2 cans, each with 3 balls
# 3. 2 cans × 3 balls = 6 balls
# 4. 5 + 6 = 11 balls
# Answer: 11 tennis balls
```

**Key Phrase**: "Let's think step by step" triggers step-by-step reasoning

#### Few-Shot CoT

```python
prompt = """
Q: There are 15 trees in the grove. Grove workers will plant trees in the grove today.
After they are done, there will be 21 trees. How many did they plant today?

A: There are 15 trees originally. Then there were 21 trees after some more were planted.
So there must have been 21 - 15 = 6. The answer is 6.

Q: If there are 3 cars in the parking lot and 2 more cars arrive,
how many cars are in the parking lot?

A: There are originally 3 cars. 2 more arrive. 3 + 2 = 5. The answer is 5.

Q: Roger has 5 tennis balls. He buys 2 more cans of tennis balls.
Each can has 3 tennis balls. How many tennis balls does he have now?

A:
"""
```

**When to Use CoT**:
- Math problems
- Logical reasoning
- Multi-step tasks
- Debugging code
- Complex analysis

### Tree-of-Thoughts (ToT)

**Purpose**: Explore multiple reasoning paths like a search tree

**Technique**: Generate multiple thought branches, evaluate them, and select the best path

#### ToT Implementation Pattern

```python
import anthropic

def tree_of_thoughts(problem, num_branches=3, depth=3):
    """
    Implement Tree of Thoughts reasoning.

    Args:
        problem: The problem to solve
        num_branches: Number of thought branches per node
        depth: Maximum tree depth

    Returns:
        Best solution found
    """
    client = anthropic.Anthropic()

    def generate_thoughts(current_state, remaining_depth):
        if remaining_depth == 0:
            return current_state

        # Generate multiple next thoughts
        prompt = f"""
        Current reasoning: {current_state}

        Generate {num_branches} different next steps to solve: {problem}

        Format as:
        1. [thought 1]
        2. [thought 2]
        3. [thought 3]
        """

        response = client.messages.create(
            model="claude-sonnet-4",
            max_tokens=1024,
            messages=[{"role": "user", "content": prompt}]
        )

        thoughts = parse_thoughts(response.content[0].text)

        # Evaluate each thought
        evaluated_thoughts = []
        for thought in thoughts:
            evaluation_prompt = f"""
            Problem: {problem}
            Current reasoning: {current_state}
            Proposed next step: {thought}

            Rate this step from 0-10 for:
            1. Correctness
            2. Progress toward solution
            3. Logical soundness

            Return only a number 0-10.
            """

            eval_response = client.messages.create(
                model="claude-sonnet-4",
                max_tokens=10,
                messages=[{"role": "user", "content": evaluation_prompt}]
            )

            score = float(eval_response.content[0].text.strip())
            evaluated_thoughts.append((thought, score))

        # Select best thought and recurse
        best_thought = max(evaluated_thoughts, key=lambda x: x[1])[0]
        new_state = f"{current_state}\n{best_thought}"

        return generate_thoughts(new_state, remaining_depth - 1)

    # Start with empty state
    final_solution = generate_thoughts("", depth)
    return final_solution

# Example usage
solution = tree_of_thoughts(
    problem="Design a database schema for a multi-tenant SaaS application",
    num_branches=3,
    depth=3
)
```

**When to Use ToT**:
- Strategic planning
- Creative problem solving
- Architecture design
- Game playing (chess, puzzles)
- Research planning

**Cost Consideration**: ToT makes multiple LLM calls per step, increasing costs significantly.

### ReAct (Reasoning + Acting)

**Purpose**: Combine reasoning with external tool usage

**Technique**: Interleave thought, action, and observation steps

#### ReAct Pattern

```python
def react_agent(question, tools, max_steps=10):
    """
    Implement ReAct pattern: Reasoning + Acting.

    Args:
        question: User question
        tools: Dictionary of available tools
        max_steps: Maximum reasoning steps
    """
    client = anthropic.Anthropic()

    history = []

    for step in range(max_steps):
        # Reasoning step
        prompt = f"""
        Question: {question}

        Previous steps:
        {format_history(history)}

        Think about what to do next. You can:
        1. Use a tool (search, calculate, code_execute)
        2. Provide final answer

        Format:
        Thought: [your reasoning]
        Action: [tool_name: tool_input] OR Answer: [final answer]
        """

        response = client.messages.create(
            model="claude-sonnet-4",
            max_tokens=1024,
            messages=[{"role": "user", "content": prompt}]
        )

        text = response.content[0].text
        thought, action = parse_react_response(text)

        history.append({"thought": thought, "action": action})

        # Check if final answer
        if action.startswith("Answer:"):
            return action.replace("Answer:", "").strip()

        # Execute action
        tool_name, tool_input = parse_action(action)
        observation = tools[tool_name](tool_input)

        history.append({"observation": observation})

    return "Failed to find answer within step limit"

# Example tools
tools = {
    "search": lambda query: web_search(query),
    "calculate": lambda expr: eval(expr),
    "code_execute": lambda code: execute_python(code)
}

# Example usage
answer = react_agent(
    question="What is the current stock price of Apple and how has it changed in the last month?",
    tools=tools
)
```

**ReAct Trace Example**:
```
Thought: I need to find Apple's current stock price
Action: search: Apple stock price today

Observation: Apple (AAPL) is trading at $178.52

Thought: Now I need historical data for the last month
Action: search: Apple stock price 30 days ago

Observation: Apple was trading at $165.23 on [date]

Thought: I can now calculate the change
Action: calculate: ((178.52 - 165.23) / 165.23) * 100

Observation: 8.04

Thought: I have all the information needed
Action: Answer: Apple (AAPL) is currently trading at $178.52, up 8.04% from $165.23 a month ago.
```

**When to Use ReAct**:
- Information retrieval tasks
- Tasks requiring calculations
- Code execution workflows
- Research and analysis
- Real-time data needs

### Self-Consistency

**Purpose**: Generate multiple reasoning paths and select the most consistent answer

**Technique**: Sample multiple CoT responses and use majority voting

```python
def self_consistency(question, num_samples=5):
    """
    Implement self-consistency with CoT.
    """
    client = anthropic.Anthropic()

    answers = []

    for i in range(num_samples):
        prompt = f"""
        Question: {question}

        Let's think step by step.
        """

        response = client.messages.create(
            model="claude-sonnet-4",
            max_tokens=1024,
            temperature=0.7,  # Higher temperature for diversity
            messages=[{"role": "user", "content": prompt}]
        )

        # Extract final answer
        answer = extract_final_answer(response.content[0].text)
        answers.append(answer)

    # Majority vote
    from collections import Counter
    most_common = Counter(answers).most_common(1)[0][0]

    return most_common

# Example
answer = self_consistency(
    "If a train travels 120 miles in 2 hours, then 180 miles in 3 hours, what is its average speed?",
    num_samples=5
)
```

**When to Use Self-Consistency**:
- High-stakes decisions
- Math/logic problems with discrete answers
- Classification tasks
- When accuracy is more important than latency/cost

### Comparison of Patterns

| Pattern | Complexity | Cost | Best For | Latency |
|---------|-----------|------|----------|---------|
| **CoT** | Low | Low | Reasoning, math | Low |
| **ToT** | High | Very High | Strategic planning | Very High |
| **ReAct** | Medium | Medium | Tool use, research | Medium |
| **Self-Consistency** | Medium | High | High-accuracy tasks | High |

### Hybrid Patterns (2025 Best Practices)

**CoT + ReAct**:
```python
# Combine reasoning with tool use
prompt = """
Question: {question}

You have access to: search, calculator, code_executor

Think step by step AND use tools when needed.

Format:
Thought: [reasoning]
Action: [tool if needed]
Observation: [result]
... (repeat)
Answer: [final answer]
"""
```

**ToT + ReAct**:
- Use ToT for strategic planning
- Use ReAct for execution of each branch
- Best for complex planning + execution tasks

---

## 3. LangGraph Workflows

### Overview

**LangGraph** is a framework for building **stateful, multi-agent applications** with LLMs. It implements state machines and directed graphs for orchestration.

**Key Innovation**: Persistent state management across agent interactions with time-travel debugging and human-in-the-loop support.

### Core Architecture

#### StateGraph

```python
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator

# Define state
class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    current_agent: str
    intermediate_results: dict

# Create graph
workflow = StateGraph(AgentState)

# Add nodes (agents)
workflow.add_node("researcher", research_agent)
workflow.add_node("writer", writing_agent)
workflow.add_node("reviewer", review_agent)

# Add edges (transitions)
workflow.add_edge("researcher", "writer")
workflow.add_edge("writer", "reviewer")

# Conditional routing
def should_continue(state):
    last_message = state["messages"][-1]
    if "APPROVED" in last_message:
        return END
    else:
        return "writer"

workflow.add_conditional_edges(
    "reviewer",
    should_continue,
    {
        END: END,
        "writer": "writer"
    }
)

# Set entry point
workflow.set_entry_point("researcher")

# Compile
app = workflow.compile()
```

#### Running the Workflow

```python
# Execute workflow
result = app.invoke({
    "messages": ["Research and write an article about AI safety"],
    "current_agent": "researcher",
    "intermediate_results": {}
})

# Stream intermediate results
for output in app.stream({
    "messages": ["Research and write an article about AI safety"]
}):
    print(output)
```

### Multi-Agent Patterns

#### Supervisor Pattern

**Architecture**: One supervisor coordinates multiple specialized agents

```python
from langgraph.graph import StateGraph
from langchain_anthropic import ChatAnthropic

# Define agents
class ResearchAgent:
    def __init__(self):
        self.llm = ChatAnthropic(model="claude-sonnet-4")

    def run(self, state):
        # Research logic
        prompt = f"Research: {state['task']}"
        response = self.llm.invoke(prompt)
        return {"research_results": response.content}

class CodingAgent:
    def __init__(self):
        self.llm = ChatAnthropic(model="claude-sonnet-4")

    def run(self, state):
        # Coding logic
        prompt = f"Code: {state['task']}"
        response = self.llm.invoke(prompt)
        return {"code": response.content}

class SupervisorAgent:
    def __init__(self):
        self.llm = ChatAnthropic(model="claude-sonnet-4")

    def route(self, state):
        """Decide which agent to use next."""
        prompt = f"""
        Task: {state['task']}
        Progress: {state.get('progress', [])}

        Which agent should handle the next step?
        Options: researcher, coder, FINISH

        Return only one word.
        """
        response = self.llm.invoke(prompt)
        return response.content.strip().lower()

# Build workflow
def create_supervisor_workflow():
    workflow = StateGraph(dict)

    # Add agents
    research_agent = ResearchAgent()
    coding_agent = CodingAgent()
    supervisor = SupervisorAgent()

    workflow.add_node("supervisor", supervisor.route)
    workflow.add_node("researcher", research_agent.run)
    workflow.add_node("coder", coding_agent.run)

    # Conditional routing from supervisor
    def route_based_on_supervisor(state):
        decision = state.get("next_agent", "FINISH")
        if decision == "researcher":
            return "researcher"
        elif decision == "coder":
            return "coder"
        else:
            return END

    workflow.add_conditional_edges(
        "supervisor",
        route_based_on_supervisor
    )

    # Loop back to supervisor
    workflow.add_edge("researcher", "supervisor")
    workflow.add_edge("coder", "supervisor")

    workflow.set_entry_point("supervisor")

    return workflow.compile()
```

#### Swarm Pattern (2025 Update)

**LangGraph Multi-Agent Swarm** orchestrates agents with dynamic hand-offs.

```python
from langgraph_swarm import Swarm, Agent

# Define specialized agents
research_agent = Agent(
    name="Researcher",
    instructions="Research topics thoroughly using available tools",
    tools=[web_search, database_query]
)

analysis_agent = Agent(
    name="Analyst",
    instructions="Analyze data and provide insights",
    tools=[data_analyzer, visualization]
)

writing_agent = Agent(
    name="Writer",
    instructions="Write clear, concise content",
    tools=[grammar_checker, style_guide]
)

# Create swarm
swarm = Swarm(
    agents=[research_agent, analysis_agent, writing_agent],
    initial_agent=research_agent
)

# Agents can hand off to each other
def research_with_handoff(state):
    # Research agent can transfer to analyst
    if state["needs_analysis"]:
        return {"transfer_to": "Analyst"}
    return {"status": "complete"}

# Run swarm
result = swarm.run(
    task="Research AI trends and create analysis report",
    max_handoffs=10
)
```

### Human-in-the-Loop

**Critical Feature**: Pause workflow for human approval

```python
from langgraph.checkpoint.sqlite import SqliteSaver

# Enable checkpointing
memory = SqliteSaver.from_conn_string(":memory:")

workflow = StateGraph(AgentState)
# ... add nodes ...

# Compile with checkpointer
app = workflow.compile(checkpointer=memory)

# Run with interrupt
config = {"configurable": {"thread_id": "1"}}

# Step 1: Run until interrupt
for event in app.stream({"messages": ["Write blog post"]}, config):
    print(event)
    # Workflow pauses at designated checkpoint

# Human reviews and approves
# Step 2: Resume from checkpoint
result = app.invoke(None, config)  # Resume from last checkpoint
```

### State Persistence & Time-Travel Debugging

```python
from langgraph.checkpoint.sqlite import SqliteSaver

# Persistent storage
checkpointer = SqliteSaver.from_conn_string("./workflow_state.db")

app = workflow.compile(checkpointer=checkpointer)

# Run workflow
config = {"configurable": {"thread_id": "thread_1"}}
result = app.invoke(initial_state, config)

# Later: Retrieve history
history = app.get_state_history(config)

for state in history:
    print(f"Step {state.step}: {state.values}")

# Rewind to specific step
app.update_state(config, {"step": 3})  # Go back to step 3
```

### When to Use LangGraph

✅ **Use LangGraph When**:
- Multi-agent coordination required
- Complex state management needs
- Human-in-the-loop workflows
- Need debugging/observability
- Conditional branching based on outputs
- Building production agent systems

❌ **Don't Use LangGraph When**:
- Simple single-agent tasks
- No state persistence needed
- Prototyping/experimentation phase
- Team lacks graph/state machine expertise

---

## 4. Simpler Approaches Without LangGraph

For teams wanting optimization without framework complexity:

### Template-Based Optimization

**Approach**: Use Jinja2 templates with versioning

```python
from jinja2 import Template
import json

# Define template
PROMPT_TEMPLATE = """
You are a {{ role }}.

Task: {{ task }}

{% if examples %}
Examples:
{% for example in examples %}
Input: {{ example.input }}
Output: {{ example.output }}
{% endfor %}
{% endif %}

{% if constraints %}
Constraints:
{% for constraint in constraints %}
- {{ constraint }}
{% endfor %}
{% endif %}

Now process:
Input: {{ user_input }}
Output:
"""

# Version control for templates
class PromptRegistry:
    def __init__(self):
        self.templates = {}

    def register(self, name, version, template_str):
        key = f"{name}_v{version}"
        self.templates[key] = Template(template_str)

    def get(self, name, version):
        key = f"{name}_v{version}"
        return self.templates[key]

    def render(self, name, version, **kwargs):
        template = self.get(name, version)
        return template.render(**kwargs)

# Usage
registry = PromptRegistry()

registry.register("qa", 1, PROMPT_TEMPLATE)

prompt = registry.render(
    "qa",
    version=1,
    role="helpful AI assistant",
    task="Answer the question accurately",
    examples=[
        {"input": "What is 2+2?", "output": "4"},
        {"input": "What is the capital of France?", "output": "Paris"}
    ],
    constraints=["Be concise", "Use simple language"],
    user_input="What is the capital of Germany?"
)
```

### A/B Testing Framework

**Approach**: Test prompt variations systematically

```python
import random
from dataclasses import dataclass
from typing import List, Callable

@dataclass
class PromptVariant:
    name: str
    template: str
    weight: float = 1.0  # For weighted sampling

@dataclass
class ABTestResult:
    variant_name: str
    success_rate: float
    avg_latency: float
    total_samples: int

class PromptABTester:
    def __init__(self, variants: List[PromptVariant], metric_fn: Callable):
        self.variants = variants
        self.metric_fn = metric_fn
        self.results = {v.name: [] for v in variants}

    def select_variant(self) -> PromptVariant:
        """Select variant based on weights."""
        total_weight = sum(v.weight for v in self.variants)
        r = random.uniform(0, total_weight)

        cumulative = 0
        for variant in self.variants:
            cumulative += variant.weight
            if r <= cumulative:
                return variant

        return self.variants[-1]

    def run_test(self, test_cases: List[dict], num_iterations: int = 100):
        """Run A/B test across variants."""
        import time

        for _ in range(num_iterations):
            for test_case in test_cases:
                variant = self.select_variant()

                # Execute prompt
                start_time = time.time()
                prompt = variant.template.format(**test_case)
                result = self.execute_llm(prompt)
                latency = time.time() - start_time

                # Evaluate
                score = self.metric_fn(test_case, result)

                self.results[variant.name].append({
                    "score": score,
                    "latency": latency
                })

    def get_results(self) -> List[ABTestResult]:
        """Analyze results."""
        results = []

        for variant_name, scores in self.results.items():
            if not scores:
                continue

            success_rate = sum(s["score"] for s in scores) / len(scores)
            avg_latency = sum(s["latency"] for s in scores) / len(scores)

            results.append(ABTestResult(
                variant_name=variant_name,
                success_rate=success_rate,
                avg_latency=avg_latency,
                total_samples=len(scores)
            ))

        return sorted(results, key=lambda x: x.success_rate, reverse=True)

    def execute_llm(self, prompt: str):
        # Replace with actual LLM call
        import anthropic
        client = anthropic.Anthropic()
        response = client.messages.create(
            model="claude-sonnet-4",
            max_tokens=1024,
            messages=[{"role": "user", "content": prompt}]
        )
        return response.content[0].text

# Example usage
variants = [
    PromptVariant(
        name="concise",
        template="Answer briefly: {question}",
        weight=0.5
    ),
    PromptVariant(
        name="detailed",
        template="Provide a detailed answer with examples: {question}",
        weight=0.3
    ),
    PromptVariant(
        name="step_by_step",
        template="Answer step by step:\n{question}",
        weight=0.2
    )
]

def accuracy_metric(test_case, result):
    # Custom metric
    return 1.0 if test_case["expected"] in result else 0.0

tester = PromptABTester(variants, accuracy_metric)

test_cases = [
    {"question": "What is 2+2?", "expected": "4"},
    {"question": "Capital of France?", "expected": "Paris"}
]

tester.run_test(test_cases, num_iterations=50)

for result in tester.get_results():
    print(f"{result.variant_name}: {result.success_rate:.2%} success, {result.avg_latency:.3f}s latency")
```

### Gradient-Free Optimization

**Approach**: Use evolutionary algorithms or grid search

```python
from typing import List, Dict, Callable
import random

class PromptEvolution:
    """Evolutionary optimization for prompts."""

    def __init__(
        self,
        base_prompt: str,
        mutations: List[Callable[[str], str]],
        fitness_fn: Callable[[str], float],
        population_size: int = 10,
        generations: int = 20
    ):
        self.base_prompt = base_prompt
        self.mutations = mutations
        self.fitness_fn = fitness_fn
        self.population_size = population_size
        self.generations = generations

    def mutate(self, prompt: str) -> str:
        """Apply random mutation."""
        mutation = random.choice(self.mutations)
        return mutation(prompt)

    def evolve(self) -> str:
        """Run evolutionary optimization."""
        # Initialize population
        population = [self.base_prompt]
        for _ in range(self.population_size - 1):
            population.append(self.mutate(self.base_prompt))

        for generation in range(self.generations):
            # Evaluate fitness
            scored = [(p, self.fitness_fn(p)) for p in population]
            scored.sort(key=lambda x: x[1], reverse=True)

            print(f"Generation {generation}: Best fitness = {scored[0][1]:.3f}")

            # Select top performers
            survivors = [p for p, _ in scored[:self.population_size // 2]]

            # Generate new population
            population = survivors[:]
            while len(population) < self.population_size:
                parent = random.choice(survivors)
                child = self.mutate(parent)
                population.append(child)

        # Return best prompt
        final_scored = [(p, self.fitness_fn(p)) for p in population]
        best_prompt = max(final_scored, key=lambda x: x[1])[0]

        return best_prompt

# Define mutations
def add_context(prompt: str) -> str:
    contexts = [
        "You are an expert assistant.",
        "You are a helpful AI.",
        "You are a knowledgeable guide."
    ]
    return f"{random.choice(contexts)}\n\n{prompt}"

def add_constraint(prompt: str) -> str:
    constraints = [
        "Be concise.",
        "Use simple language.",
        "Provide examples."
    ]
    return f"{prompt}\n\n{random.choice(constraints)}"

def add_format_instruction(prompt: str) -> str:
    formats = [
        "Format your answer as a list.",
        "Use markdown formatting.",
        "Structure your response clearly."
    ]
    return f"{prompt}\n\n{random.choice(formats)}"

# Fitness function
def evaluate_prompt_quality(prompt: str) -> float:
    """Evaluate prompt quality (replace with real evaluation)."""
    # Example: run on test set and measure accuracy
    test_cases = [...]  # Your test cases
    correct = 0

    for case in test_cases:
        result = llm(prompt.format(**case))
        if is_correct(result, case["expected"]):
            correct += 1

    return correct / len(test_cases)

# Run evolution
optimizer = PromptEvolution(
    base_prompt="Answer the question: {question}",
    mutations=[add_context, add_constraint, add_format_instruction],
    fitness_fn=evaluate_prompt_quality,
    population_size=10,
    generations=15
)

best_prompt = optimizer.evolve()
print(f"Optimized prompt:\n{best_prompt}")
```

### Few-Shot Learning with Embeddings

**Approach**: Select best examples using semantic similarity

```python
import numpy as np
from typing import List, Dict
from sentence_transformers import SentenceTransformer

class SemanticFewShotSelector:
    """Select few-shot examples using embeddings."""

    def __init__(self, examples: List[Dict], model_name: str = "all-MiniLM-L6-v2"):
        self.examples = examples
        self.model = SentenceTransformer(model_name)

        # Precompute embeddings
        texts = [ex["input"] for ex in examples]
        self.embeddings = self.model.encode(texts)

    def select_examples(self, query: str, k: int = 3) -> List[Dict]:
        """Select k most similar examples."""
        # Embed query
        query_embedding = self.model.encode([query])[0]

        # Compute similarities
        similarities = np.dot(self.embeddings, query_embedding)

        # Get top k
        top_k_indices = np.argsort(similarities)[-k:][::-1]

        return [self.examples[i] for i in top_k_indices]

    def build_prompt(self, query: str, k: int = 3) -> str:
        """Build prompt with selected examples."""
        selected = self.select_examples(query, k)

        prompt = "Examples:\n\n"
        for ex in selected:
            prompt += f"Input: {ex['input']}\nOutput: {ex['output']}\n\n"

        prompt += f"Now process:\nInput: {query}\nOutput:"

        return prompt

# Usage
examples = [
    {"input": "What is 2+2?", "output": "4"},
    {"input": "What is 5*3?", "output": "15"},
    {"input": "What is the capital of France?", "output": "Paris"},
    {"input": "Who wrote Hamlet?", "output": "William Shakespeare"},
    # ... hundreds more examples
]

selector = SemanticFewShotSelector(examples)

# Automatically selects most relevant examples
prompt = selector.build_prompt("What is 7*8?", k=3)
# Will select the math examples, not the geography/literature ones
```

### Prompt Versioning (Git-Based)

**Approach**: Store prompts in Git with semantic versioning

```yaml
# prompts/qa_prompt/v1.0.0.yaml
version: "1.0.0"
name: "qa_prompt"
description: "Basic QA prompt"
template: |
  Answer the question:
  Q: {question}
  A:

# prompts/qa_prompt/v1.1.0.yaml
version: "1.1.0"
name: "qa_prompt"
description: "QA prompt with few-shot examples"
template: |
  Examples:
  Q: What is 2+2?
  A: 4

  Q: What is the capital of France?
  A: Paris

  Now answer:
  Q: {question}
  A:
```

```python
import yaml
from pathlib import Path
from packaging import version as version_parser

class PromptVersionManager:
    """Manage prompt versions using semantic versioning."""

    def __init__(self, prompts_dir: str = "./prompts"):
        self.prompts_dir = Path(prompts_dir)

    def list_versions(self, prompt_name: str) -> List[str]:
        """List all versions of a prompt."""
        prompt_dir = self.prompts_dir / prompt_name
        versions = []

        for yaml_file in prompt_dir.glob("v*.yaml"):
            with open(yaml_file) as f:
                data = yaml.safe_load(f)
                versions.append(data["version"])

        # Sort by semantic version
        versions.sort(key=lambda v: version_parser.parse(v))
        return versions

    def get_prompt(self, prompt_name: str, version: str = "latest") -> str:
        """Get specific version of prompt."""
        if version == "latest":
            versions = self.list_versions(prompt_name)
            version = versions[-1] if versions else None

        if not version:
            raise ValueError(f"No versions found for {prompt_name}")

        yaml_file = self.prompts_dir / prompt_name / f"v{version}.yaml"

        with open(yaml_file) as f:
            data = yaml.safe_load(f)
            return data["template"]

    def rollback(self, prompt_name: str) -> str:
        """Rollback to previous version."""
        versions = self.list_versions(prompt_name)
        if len(versions) < 2:
            raise ValueError("No previous version to rollback to")

        previous_version = versions[-2]
        return self.get_prompt(prompt_name, previous_version)

# Usage
manager = PromptVersionManager()

# Get latest version
prompt = manager.get_prompt("qa_prompt", "latest")

# Get specific version
prompt_v1 = manager.get_prompt("qa_prompt", "1.0.0")

# Rollback if new version has issues
previous = manager.rollback("qa_prompt")
```

---

## 5. Evaluation Frameworks

### LangSmith

**Purpose**: Tracing, debugging, and evaluation for LangChain/LangGraph applications

#### Key Features

1. **Automatic Tracing**: Captures all LLM calls with inputs/outputs
2. **Dataset Management**: Build test sets for evaluation
3. **Evaluators**: Off-the-shelf and custom scoring functions
4. **Version Comparison**: Compare prompt/model versions
5. **Production Monitoring**: Real-time quality tracking

#### Basic Usage

```python
import os
from langsmith import Client
from langsmith.evaluation import evaluate

# Initialize client
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-api-key"

client = Client()

# Create dataset
examples = [
    {
        "inputs": {"question": "What is 2+2?"},
        "outputs": {"answer": "4"}
    },
    {
        "inputs": {"question": "Capital of France?"},
        "outputs": {"answer": "Paris"}
    }
]

dataset = client.create_dataset(
    dataset_name="qa_test_set",
    description="Test questions for QA system"
)

for example in examples:
    client.create_example(
        dataset_id=dataset.id,
        inputs=example["inputs"],
        outputs=example["outputs"]
    )

# Define your LLM function
def qa_function(inputs: dict) -> dict:
    from langchain_anthropic import ChatAnthropic

    llm = ChatAnthropic(model="claude-sonnet-4")
    response = llm.invoke(inputs["question"])

    return {"answer": response.content}

# Define evaluator
def correctness_evaluator(run, example):
    """Check if answer matches expected."""
    predicted = run.outputs["answer"]
    expected = example.outputs["answer"]

    return {
        "key": "correctness",
        "score": 1.0 if expected.lower() in predicted.lower() else 0.0
    }

# Run evaluation
results = evaluate(
    qa_function,
    data="qa_test_set",
    evaluators=[correctness_evaluator],
    experiment_prefix="qa_v1"
)

print(f"Average correctness: {results['results']['correctness']:.2%}")
```

#### LLM-as-Judge Evaluator

```python
from langsmith.evaluation import LangChainStringEvaluator

# Use LLM to evaluate quality
quality_evaluator = LangChainStringEvaluator(
    "qa",
    config={
        "criteria": {
            "accuracy": "Is the answer factually correct?",
            "completeness": "Does the answer fully address the question?",
            "clarity": "Is the answer clear and easy to understand?"
        }
    },
    prepare_data=lambda run, example: {
        "prediction": run.outputs["answer"],
        "reference": example.outputs["answer"],
        "input": example.inputs["question"]
    }
)

results = evaluate(
    qa_function,
    data="qa_test_set",
    evaluators=[quality_evaluator]
)
```

#### Version Comparison

```python
# Evaluate multiple prompt versions
def qa_v1(inputs):
    # Version 1 implementation
    pass

def qa_v2(inputs):
    # Version 2 implementation with improvements
    pass

# Compare
results_v1 = evaluate(qa_v1, data="qa_test_set", experiment_prefix="v1")
results_v2 = evaluate(qa_v2, data="qa_test_set", experiment_prefix="v2")

# View comparison in LangSmith UI
```

### Weights & Biases (W&B Weave)

**Purpose**: Experiment tracking, logging, and evaluation for LLM applications

#### Key Features

1. **A

…(truncated)