Model Selection and Evaluation: Choosing the Right AI Model for Your Use Case

# Define your requirements
requirements = {
    "accuracy_target": 0.90,  # 90% success rate minimum
    "latency_target_ms": 2000,  # < 2 seconds response time
    "cost_budget_per_1k_requests": 0.50,  # $0.50 per 1K requests
    "volume_requests_per_day": 10000,
    "context_length_needed": 8000,  # tokens
    "languages": ["en", "es"],  # English and Spanish
    "special_capabilities": ["function_calling", "json_mode"]
}

def estimate_monthly_cost(
    requests_per_day,
    avg_input_tokens,
    avg_output_tokens,
    model_pricing
):
    """Estimate monthly API costs."""
    # Monthly requests
    monthly_requests = requests_per_day * 30

    # Total tokens
    total_input = monthly_requests * avg_input_tokens
    total_output = monthly_requests * avg_output_tokens

    # Calculate cost
    input_cost = (total_input / 1_000_000) * model_pricing["input"]
    output_cost = (total_output / 1_000_000) * model_pricing["output"]

    return {
        "total_cost": input_cost + output_cost,
        "input_cost": input_cost,
        "output_cost": output_cost,
        "cost_per_request": (input_cost + output_cost) / monthly_requests
    }

# Compare models
gpt4o_cost = estimate_monthly_cost(
    requests_per_day=10000,
    avg_input_tokens=1000,
    avg_output_tokens=500,
    model_pricing={"input": 5.00, "output": 15.00}
)

gpt4o_mini_cost = estimate_monthly_cost(
    requests_per_day=10000,
    avg_input_tokens=1000,
    avg_output_tokens=500,
    model_pricing={"input": 0.15, "output": 0.60}
)

print(f"GPT-4o monthly cost: ${gpt4o_cost['total_cost']:.2f}")
print(f"GPT-4o-mini monthly cost: ${gpt4o_mini_cost['total_cost']:.2f}")
print(f"Savings with mini: ${gpt4o_cost['total_cost'] - gpt4o_mini_cost['total_cost']:.2f} ({((1 - gpt4o_mini_cost['total_cost']/gpt4o_cost['total_cost']) * 100):.0f}% reduction)")

import json
from typing import List, Dict

class TestDataset:
    def __init__(self, name: str):
        self.name = name
        self.examples = []

    def add_example(self, input_text: str, expected_output: str, metadata: Dict = None):
        """Add a test example."""
        self.examples.append({
            "input": input_text,
            "expected": expected_output,
            "metadata": metadata or {},
            "id": len(self.examples)
        })

    def save(self, filepath: str):
        """Save dataset to file."""
        with open(filepath, 'w') as f:
            json.dump({
                "name": self.name,
                "examples": self.examples,
                "count": len(self.examples)
            }, f, indent=2)

    @classmethod
    def load(cls, filepath: str):
        """Load dataset from file."""
        with open(filepath, 'r') as f:
            data = json.load(f)
            dataset = cls(data["name"])
            dataset.examples = data["examples"]
            return dataset

# Create test dataset
dataset = TestDataset("customer_support_classification")

dataset.add_example(
    input_text="I can't log into my account",
    expected_output="technical_support",
    metadata={"difficulty": "easy", "priority": "high"}
)

dataset.add_example(
    input_text="Your app deleted all my data and I need it back NOW or I'm calling my lawyer",
    expected_output="critical_escalation",
    metadata={"difficulty": "hard", "priority": "critical", "requires_empathy": True}
)

# Add 50-100 more diverse examples...

dataset.save("test_dataset.json")

from openai import OpenAI
import anthropic
from datetime import datetime
import time

class ModelEvaluator:
    def __init__(self):
        self.openai_client = OpenAI()
        self.anthropic_client = anthropic.Anthropic()
        self.results = []

    def evaluate_model(self, model_name: str, provider: str, test_dataset: TestDataset, prompt_template: str):
        """Evaluate a model on a test dataset."""
        results = {
            "model": model_name,
            "provider": provider,
            "timestamp": datetime.now().isoformat(),
            "correct": 0,
            "total": len(test_dataset.examples),
            "examples": [],
            "latencies": [],
            "errors": 0
        }

        for example in test_dataset.examples:
            start_time = time.time()

            try:
                # Get model prediction
                prediction = self._get_prediction(
                    provider,
                    model_name,
                    prompt_template.format(input=example["input"])
                )

                latency = (time.time() - start_time) * 1000  # Convert to ms
                results["latencies"].append(latency)

                # Check correctness
                is_correct = self._check_correctness(prediction, example["expected"])

                if is_correct:
                    results["correct"] += 1

                results["examples"].append({
                    "id": example["id"],
                    "input": example["input"],
                    "expected": example["expected"],
                    "prediction": prediction,
                    "correct": is_correct,
                    "latency_ms": latency
                })

            except Exception as e:
                results["errors"] += 1
                print(f"Error on example {example['id']}: {e}")

        # Calculate metrics
        results["accuracy"] = results["correct"] / results["total"]
        results["avg_latency_ms"] = sum(results["latencies"]) / len(results["latencies"])
        results["p95_latency_ms"] = sorted(results["latencies"])[int(len(results["latencies"]) * 0.95)]

        return results

    def _get_prediction(self, provider: str, model: str, prompt: str):
        """Get prediction from model."""
        if provider == "openai":
            response = self.openai_client.chat.completions.create(
                model=model,
                messages=[{"role": "user", "content": prompt}],
                temperature=0.0  # Deterministic for evaluation
            )
            return response.choices[0].message.content.strip()

        elif provider == "anthropic":
            response = self.anthropic_client.messages.create(
                model=model,
                messages=[{"role": "user", "content": prompt}],
                max_tokens=500,
                temperature=0.0
            )
            return response.content[0].text.strip()

    def _check_correctness(self, prediction: str, expected: str):
        """Check if prediction matches expected output."""
        # For classification, exact match
        return prediction.lower() == expected.lower()

# Run evaluation
evaluator = ModelEvaluator()
dataset = TestDataset.load("test_dataset.json")

prompt_template = """Classify this customer message into one of these categories:
- technical_support
- billing_inquiry
- feature_request
- critical_escalation

Message: {input}

Category:"""

# Evaluate multiple models
models_to_test = [
    ("gpt-4o", "openai"),
    ("gpt-4o-mini", "openai"),
    ("claude-3-5-sonnet-20241022", "anthropic"),
    ("claude-3-5-haiku-20241022", "anthropic")
]

for model, provider in models_to_test:
    print(f"\nEvaluating {model}...")
    results = evaluator.evaluate_model(model, provider, dataset, prompt_template)

    print(f"Accuracy: {results['accuracy']:.1%}")
    print(f"Avg latency: {results['avg_latency_ms']:.0f}ms")
    print(f"P95 latency: {results['p95_latency_ms']:.0f}ms")
    print(f"Errors: {results['errors']}")

import pandas as pd
import matplotlib.pyplot as plt

def generate_comparison_report(all_results):
    """Create comparison report across models."""
    # Create comparison dataframe
    comparison = pd.DataFrame([{
        "Model": r["model"],
        "Accuracy": f"{r['accuracy']:.1%}",
        "Avg Latency (ms)": f"{r['avg_latency_ms']:.0f}",
        "P95 Latency (ms)": f"{r['p95_latency_ms']:.0f}",
        "Errors": r["errors"],
        "Cost (est)": calculate_cost(r)
    } for r in all_results])

    print("\n" + "="*80)
    print("MODEL COMPARISON REPORT")
    print("="*80)
    print(comparison.to_string(index=False))

    # Create visualizations
    fig, axes = plt.subplots(2, 2, figsize=(12, 10))

    # Accuracy comparison
    axes[0, 0].bar([r["model"] for r in all_results], [r["accuracy"] for r in all_results])
    axes[0, 0].set_title("Accuracy Comparison")
    axes[0, 0].set_ylabel("Accuracy")
    axes[0, 0].tick_params(axis='x', rotation=45)

    # Latency comparison
    axes[0, 1].bar([r["model"] for r in all_results], [r["avg_latency_ms"] for r in all_results])
    axes[0, 1].set_title("Average Latency")
    axes[0, 1].set_ylabel("Latency (ms)")
    axes[0, 1].tick_params(axis='x', rotation=45)

    plt.tight_layout()
    plt.savefig("model_comparison.png")

    return comparison

from rouge import Rouge
from bert_score import score as bert_score

def evaluate_generation(predictions, references):
    """Evaluate generated text quality."""
    # ROUGE scores (overlap-based)
    rouge = Rouge()
    rouge_scores = rouge.get_scores(predictions, references, avg=True)

    # BERTScore (semantic similarity)
    P, R, F1 = bert_score(predictions, references, lang="en")

    return {
        "rouge_1_f1": rouge_scores["rouge-1"]["f"],
        "rouge_2_f1": rouge_scores["rouge-2"]["f"],
        "rouge_l_f1": rouge_scores["rouge-l"]["f"],
        "bert_score_f1": F1.mean().item()
    }

# Use in evaluation
predictions = [model.generate(example["input"]) for example in test_data]
references = [example["expected_output"] for example in test_data]

scores = evaluate_generation(predictions, references)
print(f"ROUGE-L: {scores['rouge_l_f1']:.3f}")
print(f"BERTScore: {scores['bert_score_f1']:.3f}")

class HumanEvaluation:
    def __init__(self):
        self.ratings = []

    def collect_rating(self, example_id, model_output, criteria):
        """Collect human rating for a model output."""
        print(f"\nExample {example_id}:")
        print(f"Output: {model_output}\n")

        rating = {}
        for criterion, description in criteria.items():
            score = int(input(f"{criterion} ({description}) [1-5]: "))
            rating[criterion] = score

        self.ratings.append({
            "example_id": example_id,
            "ratings": rating,
            "avg_score": sum(rating.values()) / len(rating)
        })

    def get_summary(self):
        """Get summary statistics."""
        if not self.ratings:
            return {}

        criteria = self.ratings[0]["ratings"].keys()
        summary = {}

        for criterion in criteria:
            scores = [r["ratings"][criterion] for r in self.ratings]
            summary[criterion] = {
                "mean": sum(scores) / len(scores),
                "min": min(scores),
                "max": max(scores)
            }

        return summary

# Use for evaluation
criteria = {
    "accuracy": "Is the information correct?",
    "helpfulness": "Does it solve the user's problem?",
    "tone": "Is the tone appropriate?",
    "clarity": "Is it clear and well-written?"
}

evaluator = HumanEvaluation()

# Rate a sample of outputs
for example in random.sample(test_examples, 20):
    output = model.generate(example["input"])
    evaluator.collect_rating(example["id"], output, criteria)

summary = evaluator.get_summary()

import random
from datetime import datetime

class ABTest:
    def __init__(self, model_a, model_b, split_ratio=0.5):
        self.model_a = model_a
        self.model_b = model_b
        self.split_ratio = split_ratio
        self.results = {"A": [], "B": []}

    def route_request(self, user_id, request):
        """Route request to model A or B."""
        # Deterministic assignment based on user_id
        if hash(user_id) % 100 < self.split_ratio * 100:
            variant = "A"
            response = self.model_a.generate(request)
        else:
            variant = "B"
            response = self.model_b.generate(request)

        # Track assignment
        self.results[variant].append({
            "user_id": user_id,
            "request": request,
            "response": response,
            "timestamp": datetime.now()
        })

        return response, variant

    def collect_feedback(self, user_id, variant, feedback):
        """Collect user feedback (thumbs up/down, rating, etc.)."""
        # Find the interaction
        for interaction in self.results[variant]:
            if interaction["user_id"] == user_id:
                interaction["feedback"] = feedback
                break

    def analyze_results(self):
        """Analyze A/B test results."""
        results = {}

        for variant in ["A", "B"]:
            interactions = self.results[variant]
            feedback = [i.get("feedback") for i in interactions if "feedback" in i]

            if feedback:
                # Assuming binary feedback (1 = positive, 0 = negative)
                results[variant] = {
                    "total_interactions": len(interactions),
                    "feedback_count": len(feedback),
                    "positive_rate": sum(feedback) / len(feedback),
                    "response_rate": len(feedback) / len(interactions)
                }

        return results

# Run A/B test
ab_test = ABTest(
    model_a=GPT4oMini(),
    model_b=Claude35Haiku(),
    split_ratio=0.5
)

# In production, route requests
response, variant = ab_test.route_request(user_id="user123", request="How do I reset my password?")

# Later, collect feedback
ab_test.collect_feedback("user123", variant, feedback=1)  # Positive

# Analyze after 1000+ interactions
results = ab_test.analyze_results()
print(f"Model A positive rate: {results['A']['positive_rate']:.1%}")
print(f"Model B positive rate: {results['B']['positive_rate']:.1%}")

class ModelMigration:
    def __init__(self, old_model, new_model, test_dataset):
        self.old_model = old_model
        self.new_model = new_model
        self.test_dataset = test_dataset

    def validate_migration(self):
        """Ensure new model performs at least as well as old model."""
        print("Running migration validation...")

        old_results = self._evaluate_model(self.old_model)
        new_results = self._evaluate_model(self.new_model)

        comparison = {
            "accuracy_change": new_results["accuracy"] - old_results["accuracy"],
            "latency_change_pct": ((new_results["avg_latency"] - old_results["avg_latency"]) / old_results["avg_latency"]) * 100,
            "cost_change_pct": ((new_results["cost"] - old_results["cost"]) / old_results["cost"]) * 100
        }

        # Check if migration is safe
        is_safe = (
            comparison["accuracy_change"] >= -0.05 and  # No more than 5% accuracy drop
            comparison["latency_change_pct"] <= 50  # No more than 50% latency increase
        )

        print(f"\nMigration Safety Check: {'PASS' if is_safe else 'FAIL'}")
        print(f"Accuracy change: {comparison['accuracy_change']:+.1%}")
        print(f"Latency change: {comparison['latency_change_pct']:+.0f}%")
        print(f"Cost change: {comparison['cost_change_pct']:+.0f}%")

        return is_safe, comparison

    def gradual_rollout(self, traffic_percentage=10):
        """Gradually shift traffic to new model."""
        print(f"Starting gradual rollout: {traffic_percentage}% traffic to new model")

        # Monitor for issues
        # If metrics degrade, roll back
        # If stable, increase traffic_percentage incrementally

def optimize_prompt_for_model(base_prompt, model_name, test_dataset):
    """Test prompt variations to find best for model."""
    variations = [
        base_prompt,  # Original
        f"You are an expert assistant.\n\n{base_prompt}",  # Add role
        f"{base_prompt}\n\nThink step by step.",  # Add reasoning
        f"{base_prompt}\n\nProvide only the answer, no explanation."  # Constrain output
    ]

    best_accuracy = 0
    best_prompt = base_prompt

    for i, prompt in enumerate(variations):
        print(f"Testing variation {i+1}/{len(variations)}...")
        accuracy = evaluate_prompt(prompt, model_name, test_dataset)

        if accuracy > best_accuracy:
            best_accuracy = accuracy
            best_prompt = prompt

    print(f"\nBest prompt achieved {best_accuracy:.1%} accuracy")
    return best_prompt

class ModelRouter:
    """Route requests to the best model for the task."""

    def __init__(self):
        self.models = {
            "simple": {"model": "gpt-4o-mini", "cost_per_1k": 0.0003},
            "complex": {"model": "gpt-4o", "cost_per_1k": 0.01},
            "fast": {"model": "claude-3-5-haiku", "cost_per_1k": 0.0005}
        }

    def route(self, request, user_priority="balanced"):
        """Route request to appropriate model."""
        # Classify complexity
        complexity = self._estimate_complexity(request)

        if user_priority == "cost":
            # Always use cheapest model
            return self.models["simple"]

        elif user_priority == "quality":
            # Use best model for complex, good model for simple
            return self.models["complex"] if complexity > 0.7 else self.models["simple"]

        else:  # balanced
            # Use tiered approach
            if complexity > 0.8:
                return self.models["complex"]
            elif complexity < 0.3:
                return self.models["simple"]
            else:
                return self.models["fast"]

    def _estimate_complexity(self, request):
        """Estimate request complexity (0-1)."""
        # Use heuristics or small classifier model
        indicators = {
            "length": len(request.split()) > 100,
            "technical": any(word in request.lower() for word in ["code", "algorithm", "calculate"]),
            "multi_step": any(phrase in request.lower() for phrase in ["first", "then", "finally", "step by step"])
        }

        return sum(indicators.values()) / len(indicators)

# Use router
router = ModelRouter()
model_choice = router.route("What's the weather today?", user_priority="cost")
# → Returns simple model

model_choice = router.route("Analyze this complex algorithm and suggest optimizations...", user_priority="quality")
# → Returns complex model