Prompt Engineering at Scale: Production Patterns & Optimization

from dataclasses import dataclass
from datetime import datetime
from typing import Optional
import json
import hashlib

@dataclass
class PromptVersion:
    """Represents a single prompt version"""
    id: str
    name: str
    version: int
    content: str
    template_vars: list[str]
    created_at: datetime
    created_by: str
    description: str
    status: str  # "draft", "testing", "production", "deprecated"
    metrics: dict  # accuracy, latency, cost
    tags: list[str]
    
    def get_hash(self) -> str:
        """Generate unique hash for content"""
        return hashlib.sha256(self.content.encode()).hexdigest()

class PromptRegistry:
    """Centralized prompt management system"""
    
    def __init__(self, db_connection):
        self.db = db_connection
    
    def create_prompt(self, name: str, content: str, 
                     template_vars: list[str], 
                     description: str) -> PromptVersion:
        """Create new prompt version"""
        version = PromptVersion(
            id=f"{name}-{datetime.now().timestamp()}",
            name=name,
            version=1,
            content=content,
            template_vars=template_vars,
            created_at=datetime.now(),
            created_by="system",
            description=description,
            status="draft",
            metrics={},
            tags=[]
        )
        self.db.insert("prompts", version.__dict__)
        return version
    
    def update_prompt(self, name: str, content: str, 
                     description: str) -> PromptVersion:
        """Create new version of existing prompt"""
        latest = self.db.query(
            "SELECT * FROM prompts WHERE name = ? ORDER BY version DESC LIMIT 1",
            (name,)
        )[0]
        
        new_version = PromptVersion(
            id=f"{name}-{datetime.now().timestamp()}",
            name=name,
            version=latest['version'] + 1,
            content=content,
            template_vars=latest['template_vars'],
            created_at=datetime.now(),
            created_by="system",
            description=description,
            status="draft",
            metrics={},
            tags=latest['tags']
        )
        self.db.insert("prompts", new_version.__dict__)
        return new_version
    
    def get_production_prompt(self, name: str) -> Optional[PromptVersion]:
        """Get current production prompt"""
        result = self.db.query(
            "SELECT * FROM prompts WHERE name = ? AND status = 'production' LIMIT 1",
            (name,)
        )
        return result[0] if result else None
    
    def promote_to_production(self, prompt_id: str) -> bool:
        """Promote prompt version to production"""
        # Demote current production version
        self.db.execute(
            "UPDATE prompts SET status = 'deprecated' WHERE status = 'production'"
        )
        # Promote new version
        self.db.execute(
            "UPDATE prompts SET status = 'production' WHERE id = ?",
            (prompt_id,)
        )
        return True
    
    def rollback_prompt(self, name: str, version: int) -> bool:
        """Rollback to previous prompt version"""
        target = self.db.query(
            "SELECT * FROM prompts WHERE name = ? AND version = ?",
            (name, version)
        )[0]
        
        self.promote_to_production(target['id'])
        return True

from sklearn.metrics.pairwise import cosine_similarity
import numpy as np

class FewShotSelector:
    """Intelligently select examples for few-shot prompting"""
    
    def __init__(self, embedding_model):
        self.embedding_model = embedding_model
        self.examples_db = []
    
    def add_example(self, input_text: str, output_text: str, 
                   category: str, quality_score: float):
        """Add example to database"""
        embedding = self.embedding_model.embed(input_text)
        self.examples_db.append({
            'input': input_text,
            'output': output_text,
            'embedding': embedding,
            'category': category,
            'quality_score': quality_score
        })
    
    def select_examples(self, user_input: str, num_examples: int = 3) -> list[dict]:
        """Select most relevant examples for user input"""
        user_embedding = self.embedding_model.embed(user_input)
        
        # Calculate similarity scores
        similarities = []
        for example in self.examples_db:
            similarity = cosine_similarity(
                [user_embedding], 
                [example['embedding']]
            )[0][0]
            
            # Weight by quality score
            weighted_score = similarity * example['quality_score']
            similarities.append((example, weighted_score))
        
        # Sort by weighted score and select top N
        top_examples = sorted(similarities, key=lambda x: x[1], reverse=True)[:num_examples]
        return [ex[0] for ex in top_examples]
    
    def build_few_shot_prompt(self, system_prompt: str, 
                             user_input: str, 
                             num_examples: int = 3) -> str:
        """Build complete prompt with few-shot examples"""
        examples = self.select_examples(user_input, num_examples)
        
        prompt = system_prompt + "\n\n"
        prompt += "Examples:\n"
        
        for i, example in enumerate(examples, 1):
            prompt += f"\nExample {i}:\n"
            prompt += f"Input: {example['input']}\n"
            prompt += f"Output: {example['output']}\n"
        
        prompt += f"\nNow process this input:\n"
        prompt += f"Input: {user_input}\n"
        prompt += f"Output:"
        
        return prompt

class PromptCompressor:
    """Reduce prompt token count while maintaining quality"""
    
    def __init__(self, tokenizer):
        self.tokenizer = tokenizer
    
    def count_tokens(self, text: str) -> int:
        """Count tokens in text"""
        return len(self.tokenizer.encode(text))
    
    def compress_prompt(self, prompt: str, target_reduction: float = 0.2) -> str:
        """Compress prompt by removing redundant content"""
        original_tokens = self.count_tokens(prompt)
        target_tokens = int(original_tokens * (1 - target_reduction))
        
        # Remove verbose explanations
        compressed = prompt
        
        # Remove redundant phrases
        redundant_phrases = [
            "Please note that",
            "It is important to",
            "As mentioned earlier",
            "In other words"
        ]
        
        for phrase in redundant_phrases:
            compressed = compressed.replace(phrase, "")
        
        # Shorten examples
        lines = compressed.split('\n')
        shortened_lines = []
        for line in lines:
            if len(line) > 100 and not line.startswith('Example'):
                # Truncate long lines
                shortened_lines.append(line[:100] + "...")
            else:
                shortened_lines.append(line)
        
        compressed = '\n'.join(shortened_lines)
        
        # Verify token reduction
        new_tokens = self.count_tokens(compressed)
        reduction_pct = (original_tokens - new_tokens) / original_tokens * 100
        
        print(f"Token reduction: {reduction_pct:.1f}% ({original_tokens} → {new_tokens})")
        
        return compressed
    
    def optimize_for_cost(self, prompt: str, max_tokens: int) -> str:
        """Optimize prompt to fit within token budget"""
        current_tokens = self.count_tokens(prompt)
        
        if current_tokens <= max_tokens:
            return prompt
        
        # Iteratively compress until within budget
        compression_rate = 0.1
        while self.count_tokens(prompt) > max_tokens and compression_rate < 0.9:
            prompt = self.compress_prompt(prompt, compression_rate)
            compression_rate += 0.1
        
        return prompt

from enum import Enum
import random

class TestStatus(Enum):
    RUNNING = "running"
    COMPLETED = "completed"
    WINNER_SELECTED = "winner_selected"

class PromptABTest:
    """A/B test framework for prompt optimization"""
    
    def __init__(self, db_connection):
        self.db = db_connection
    
    def create_test(self, name: str, prompt_a_id: str, 
                   prompt_b_id: str, sample_size: int = 100) -> str:
        """Create new A/B test"""
        test_id = f"test-{datetime.now().timestamp()}"
        
        self.db.insert("ab_tests", {
            'test_id': test_id,
            'name': name,
            'prompt_a_id': prompt_a_id,
            'prompt_b_id': prompt_b_id,
            'sample_size': sample_size,
            'status': TestStatus.RUNNING.value,
            'created_at': datetime.now(),
            'results_a': [],
            'results_b': []
        })
        
        return test_id
    
    def run_test_request(self, test_id: str, user_input: str, 
                        llm_client) -> dict:
        """Run single test request"""
        test = self.db.query("SELECT * FROM ab_tests WHERE test_id = ?", (test_id,))[0]
        
        # Randomly assign to A or B
        variant = random.choice(['A', 'B'])
        prompt_id = test['prompt_a_id'] if variant == 'A' else test['prompt_b_id']
        
        # Get prompt
        prompt = self.db.query("SELECT * FROM prompts WHERE id = ?", (prompt_id,))[0]
        
        # Execute request
        response = llm_client.complete(prompt['content'] + user_input)
        
        # Record result
        result = {
            'variant': variant,
            'input': user_input,
            'output': response['text'],
            'tokens_used': response['usage']['total_tokens'],
            'latency_ms': response['latency_ms'],
            'timestamp': datetime.now()
        }
        
        # Store result
        if variant == 'A':
            test['results_a'].append(result)
        else:
            test['results_b'].append(result)
        
        self.db.update("ab_tests", {'test_id': test_id}, test)
        
        return result
    
    def analyze_results(self, test_id: str) -> dict:
        """Analyze test results and determine winner"""
        test = self.db.query("SELECT * FROM ab_tests WHERE test_id = ?", (test_id,))[0]
        
        results_a = test['results_a']
        results_b = test['results_b']
        
        if len(results_a) < 10 or len(results_b) < 10:
            return {'status': 'insufficient_data'}
        
        # Calculate metrics
        avg_tokens_a = np.mean([r['tokens_used'] for r in results_a])
        avg_tokens_b = np.mean([r['tokens_used'] for r in results_b])
        
        avg_latency_a = np.mean([r['latency_ms'] for r in results_a])
        avg_latency_b = np.mean([r['latency_ms'] for r in results_b])
        
        # Determine winner (lower tokens + latency)
        score_a = avg_tokens_a + (avg_latency_a / 1000)
        score_b = avg_tokens_b + (avg_latency_b / 1000)
        
        winner = 'A' if score_a < score_b else 'B'
        
        return {
            'status': 'completed',
            'winner': winner,
            'metrics_a': {
                'avg_tokens': avg_tokens_a,
                'avg_latency_ms': avg_latency_a,
                'sample_size': len(results_a)
            },
            'metrics_b': {
                'avg_tokens': avg_tokens_b,
                'avg_latency_ms': avg_latency_b,
                'sample_size': len(results_b)
            },
            'improvement': f"{((score_b - score_a) / score_b * 100):.1f}%"
        }

class ChainOfThoughtBuilder:
    """Build optimized chain-of-thought prompts"""
    
    def __init__(self):
        self.reasoning_templates = {
            'step_by_step': """
Let's think through this step by step:
1. First, identify the key components
2. Then, analyze each component
3. Finally, synthesize the answer

Problem: {problem}
Solution:""",
            
            'decomposition': """
Break down the problem:
- What are we trying to solve?
- What information do we have?
- What are the constraints?
- What's the approach?

Problem: {problem}
Analysis:""",
            
            'verification': """
Solve this and verify your answer:
1. Solve the problem
2. Check your work
3. Verify the answer makes sense

Problem: {problem}
Solution:"""
        }
    
    def build_cot_prompt(self, problem: str, 
                        template: str = 'step_by_step') -> str:
        """Build chain-of-thought prompt"""
        if template not in self.reasoning_templates:
            template = 'step_by_step'
        
        return self.reasoning_templates[template].format(problem=problem)
    
    def optimize_cot_for_cost(self, problem: str) -> str:
        """Use minimal CoT for cost optimization"""
        # Use shorter template for simple problems
        if len(problem) < 100:
            return f"Solve: {problem}\nAnswer:"
        
        # Use full CoT for complex problems
        return self.build_cot_prompt(problem, 'step_by_step')

from dataclasses import dataclass
import statistics

@dataclass
class PromptMetrics:
    """Track prompt performance metrics"""
    prompt_id: str
    total_requests: int
    avg_tokens: float
    avg_latency_ms: float
    error_rate: float
    user_satisfaction: float  # 1-5 scale
    cost_per_request: float
    
    def get_efficiency_score(self) -> float:
        """Calculate overall efficiency score (0-100)"""
        # Lower tokens and latency = higher score
        token_score = max(0, 100 - (self.avg_tokens / 10))
        latency_score = max(0, 100 - (self.avg_latency_ms / 100))
        error_score = max(0, 100 - (self.error_rate * 100))
        
        return (token_score + latency_score + error_score) / 3

class PromptAnalytics:
    """Analyze prompt performance"""
    
    def __init__(self, db_connection):
        self.db = db_connection
    
    def calculate_metrics(self, prompt_id: str, 
                         time_window_hours: int = 24) -> PromptMetrics:
        """Calculate metrics for prompt"""
        requests = self.db.query(
            """SELECT * FROM prompt_requests 
               WHERE prompt_id = ? AND timestamp > datetime('now', '-' || ? || ' hours')""",
            (prompt_id, time_window_hours)
        )
        
        if not requests:
            return None
        
        tokens = [r['tokens_used'] for r in requests]
        latencies = [r['latency_ms'] for r in requests]
        errors = [1 if r['error'] else 0 for r in requests]
        satisfactions = [r['user_rating'] for r in requests if r['user_rating']]
        
        return PromptMetrics(
            prompt_id=prompt_id,
            total_requests=len(requests),
            avg_tokens=statistics.mean(tokens),
            avg_latency_ms=statistics.mean(latencies),
            error_rate=statistics.mean(errors),
            user_satisfaction=statistics.mean(satisfactions) if satisfactions else 0,
            cost_per_request=statistics.mean(tokens) * 0.0001  # Approximate cost
        )
    
    def compare_prompts(self, prompt_ids: list[str]) -> dict:
        """Compare multiple prompts"""
        comparison = {}
        for prompt_id in prompt_ids:
            metrics = self.calculate_metrics(prompt_id)
            comparison[prompt_id] = {
                'metrics': metrics,
                'efficiency_score': metrics.get_efficiency_score() if metrics else 0
            }
        
        return comparison
    
    def detect_degradation(self, prompt_id: str, 
                          threshold: float = 0.1) -> bool:
        """Detect if prompt performance has degraded"""
        current = self.calculate_metrics(prompt_id, time_window_hours=1)
        baseline = self.calculate_metrics(prompt_id, time_window_hours=24)
        
        if not current or not baseline:
            return False
        
        # Check if error rate increased significantly
        error_increase = current.error_rate - baseline.error_rate
        if error_increase > threshold:
            return True
        
        # Check if latency increased significantly
        latency_increase = (current.avg_latency_ms - baseline.avg_latency_ms) / baseline.avg_latency_ms
        if latency_increase > threshold:
            return True
        
        return False

class PromptVersionManager:
    """Manage and test prompt versions"""
    
    def __init__(self):
        self.versions = {}
        self.test_results = {}
    
    def create_version(self, name: str, prompt: str, description: str = None):
        """Create new prompt version"""
        
        self.versions[name] = {
            'prompt': prompt,
            'description': description,
            'created_at': datetime.now(),
            'performance': None
        }
    
    def ab_test(self, version_a: str, version_b: str, 
                test_queries: list, metric_fn) -> dict:
        """A/B test two prompt versions"""
        
        results_a = []
        results_b = []
        
        for query in test_queries:
            # Test version A
            response_a = self._call_llm(self.versions[version_a]['prompt'], query)
            score_a = metric_fn(response_a)
            results_a.append(score_a)
            
            # Test version B
            response_b = self._call_llm(self.versions[version_b]['prompt'], query)
            score_b = metric_fn(response_b)
            results_b.append(score_b)
        
        avg_a = sum(results_a) / len(results_a)
        avg_b = sum(results_b) / len(results_b)
        
        return {
            'version_a': {'avg_score': avg_a, 'scores': results_a},
            'version_b': {'avg_score': avg_b, 'scores': results_b},
            'winner': version_a if avg_a > avg_b else version_b
        }
    
    def _call_llm(self, prompt: str, query: str) -> str:
        """Call LLM with prompt"""
        # Implementation
        pass

class DynamicPromptGenerator:
    """Generate prompts dynamically based on context"""
    
    def __init__(self):
        self.templates = {
            'summarization': 'Summarize the following text in {length} sentences: {text}',
            'classification': 'Classify the following text as one of {categories}: {text}',
            'extraction': 'Extract {fields} from the following text: {text}'
        }
    
    def generate_prompt(self, task: str, **kwargs) -> str:
        """Generate prompt for task"""
        
        if task not in self.templates:
            return None
        
        template = self.templates[task]
        return template.format(**kwargs)
    
    def generate_with_examples(self, task: str, examples: list, **kwargs) -> str:
        """Generate prompt with examples"""
        
        base_prompt = self.generate_prompt(task, **kwargs)
        
        # Add examples
        examples_text = '
'.join([
            f"Example {i+1}: {example}" for i, example in enumerate(examples)
        ])
        
        return f"{base_prompt}

Examples:
{examples_text}"

class PromptDeploymentManager:
    """Manage prompt deployment to production"""
    
    def __init__(self):
        self.current_version = None
        self.deployment_history = []
    
    def deploy_prompt(self, version: str, rollout_percentage: int = 100):
        """Deploy prompt version"""
        
        deployment = {
            'version': version,
            'rollout_percentage': rollout_percentage,
            'deployed_at': datetime.now(),
            'status': 'active'
        }
        
        self.deployment_history.append(deployment)
        self.current_version = version
    
    def rollback_prompt(self, previous_version: str):
        """Rollback to previous version"""
        
        self.current_version = previous_version
        self.deployment_history[-1]['status'] = 'rolled_back'
    
    def canary_deploy(self, version: str, percentage: int = 10):
        """Deploy to small percentage of users"""
        
        self.deploy_prompt(version, rollout_percentage=percentage)