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:
- DSPy Framework: Automated prompt optimization treating prompts as code with declarative programming
- LangGraph Workflows: Stateful multi-agent orchestration with graph-based prompt coordination
- 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
- DSPy Framework
- Prompt Engineering Patterns
- LangGraph Workflows
- Simpler Approaches Without LangGraph
- Evaluation Frameworks
- Production Patterns
- Comparison Matrix
- Use Case Recommendations
- 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
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
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:
- Uses your program to generate outputs on training data
- Filters successful traces (based on metric)
- Selects representative examples as few-shot demonstrations
- Compiles optimized program with best examples
BootstrapFewShotWithRandomSearch
Best For: Medium datasets (50-300 examples) Optimizes: Few-shot examples with candidate exploration
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
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:
- Bootstrap Few-Shot Candidates: Generates example candidates from training data
- Propose Instructions: Creates instruction variations grounded in task dynamics
- Bayesian Optimization: Finds optimal combination of instructions + examples
- Joint Optimization: Optimizes both components together (not separately)
Sequential Optimization Strategy
For best results, combine optimizers:
# 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
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
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
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
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
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:
# 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
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
# 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
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.
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
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
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
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
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
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
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
# 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:
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
- Automatic Tracing: Captures all LLM calls with inputs/outputs
- Dataset Management: Build test sets for evaluation
- Evaluators: Off-the-shelf and custom scoring functions
- Version Comparison: Compare prompt/model versions
- Production Monitoring: Real-time quality tracking
Basic Usage
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
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
# 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
- **A
…(truncated)