Prompt Management & Evaluation

import hashlib
import json
from datetime import datetime
from dataclasses import dataclass, asdict
from pathlib import Path

@dataclass
class PromptVersion:
    name: str
    version: str
    template: str
    model: str
    temperature: float
    description: str
    author: str
    created_at: str = None
    content_hash: str = None

    def __post_init__(self):
        self.created_at = self.created_at or datetime.now().isoformat()
        self.content_hash = hashlib.sha256(self.template.encode()).hexdigest()[:12]

class PromptRegistry:
    """File-based prompt registry with versioning."""

    def __init__(self, base_dir: str = "./prompts"):
        self.base_dir = Path(base_dir)
        self.base_dir.mkdir(parents=True, exist_ok=True)

    def register(self, prompt: PromptVersion) -> str:
        prompt_dir = self.base_dir / prompt.name
        prompt_dir.mkdir(exist_ok=True)

        version_file = prompt_dir / f"v{prompt.version}.json"
        version_file.write_text(json.dumps(asdict(prompt), indent=2))

        # Update latest pointer
        latest_file = prompt_dir / "latest.json"
        latest_file.write_text(json.dumps({"version": prompt.version, "hash": prompt.content_hash}))

        return prompt.content_hash

    def get(self, name: str, version: str = None) -> PromptVersion:
        prompt_dir = self.base_dir / name
        if version is None:
            latest = json.loads((prompt_dir / "latest.json").read_text())
            version = latest["version"]
        data = json.loads((prompt_dir / f"v{version}.json").read_text())
        return PromptVersion(**data)

    def list_versions(self, name: str) -> list[str]:
        prompt_dir = self.base_dir / name
        return sorted([f.stem for f in prompt_dir.glob("v*.json")])

# Usage
registry = PromptRegistry()
registry.register(PromptVersion(
    name="sentiment_classifier",
    version="1.0",
    template="Classify the sentiment: {text}\nRespond: POSITIVE, NEGATIVE, or NEUTRAL",
    model="gpt-4o",
    temperature=0.0,
    description="Basic sentiment classification v1",
    author="alice"
))
registry.register(PromptVersion(
    name="sentiment_classifier",
    version="1.1",
    template="You are a sentiment analysis expert.\n\nClassify this text into exactly one category: POSITIVE, NEGATIVE, or NEUTRAL.\nConsider sarcasm and context.\n\nText: {text}\nSentiment:",
    model="gpt-4o",
    temperature=0.0,
    description="Added expert role and sarcasm handling",
    author="alice"
))

import openai
import json

client = openai.OpenAI()

JUDGE_PROMPT = """You are an expert evaluator. Score the following AI response on these criteria:

CRITERIA:
1. Accuracy (1-5): Are the facts correct? Is the information grounded in the provided context?
2. Completeness (1-5): Does it address all parts of the question?
3. Clarity (1-5): Is it well-organized and easy to understand?
4. Conciseness (1-5): Is it appropriately brief without being incomplete?

CONTEXT PROVIDED TO THE AI:
{context}

USER QUESTION:
{question}

AI RESPONSE TO EVALUATE:
{response}

REFERENCE ANSWER (gold standard):
{reference}

Score each criterion and provide a brief justification.
Respond in JSON:
{{
    "accuracy": {{"score": int, "reason": "..."}},
    "completeness": {{"score": int, "reason": "..."}},
    "clarity": {{"score": int, "reason": "..."}},
    "conciseness": {{"score": int, "reason": "..."}},
    "overall_score": float,
    "critical_issues": ["issue1", ...] or []
}}"""


def llm_judge(context: str, question: str, response: str, reference: str) -> dict:
    """Use GPT-4o as a judge to evaluate a response."""
    prompt = JUDGE_PROMPT.format(
        context=context,
        question=question,
        response=response,
        reference=reference
    )

    result = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "You are a fair, consistent evaluator. Be strict but reasonable."},
            {"role": "user", "content": prompt}
        ],
        response_format={"type": "json_object"},
        temperature=0.0
    )

    return json.loads(result.choices[0].message.content)


def evaluate_prompt_version(prompt_template: str, test_cases: list[dict], model: str = "gpt-4o") -> dict:
    """Evaluate a prompt version against a test suite using LLM-as-judge."""
    scores = []

    for case in test_cases:
        # Generate response using the prompt under test
        formatted_prompt = prompt_template.format(**case["input"])
        response = client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": formatted_prompt}],
            temperature=0.0
        )
        ai_response = response.choices[0].message.content

        # Judge the response
        judgment = llm_judge(
            context=case.get("context", ""),
            question=case["input"].get("question", formatted_prompt),
            response=ai_response,
            reference=case["expected"]
        )
        judgment["test_id"] = case.get("id", "unknown")
        judgment["ai_response"] = ai_response
        scores.append(judgment)

    # Aggregate
    avg_score = sum(s["overall_score"] for s in scores) / len(scores)
    return {
        "avg_overall_score": round(avg_score, 2),
        "pass_rate": sum(1 for s in scores if s["overall_score"] >= 3.5) / len(scores),
        "individual_scores": scores,
        "critical_failures": [s for s in scores if s["overall_score"] < 2.5]
    }

import random
import hashlib
from dataclasses import dataclass, field
from collections import defaultdict

@dataclass
class ABTest:
    name: str
    variant_a: str  # prompt template A (control)
    variant_b: str  # prompt template B (challenger)
    traffic_split: float = 0.5  # fraction going to B
    results: dict = field(default_factory=lambda: {"a": [], "b": []})

    def assign_variant(self, user_id: str) -> str:
        """Deterministic assignment based on user_id for consistency."""
        hash_val = int(hashlib.md5(f"{self.name}:{user_id}".encode()).hexdigest(), 16)
        return "b" if (hash_val % 100) < (self.traffic_split * 100) else "a"

    def get_prompt(self, user_id: str) -> tuple[str, str]:
        """Returns (variant_name, prompt_template)."""
        variant = self.assign_variant(user_id)
        template = self.variant_b if variant == "b" else self.variant_a
        return variant, template

    def record_result(self, variant: str, score: float):
        self.results[variant].append(score)

    def get_stats(self) -> dict:
        stats = {}
        for v in ["a", "b"]:
            scores = self.results[v]
            if scores:
                stats[v] = {
                    "n": len(scores),
                    "mean": sum(scores) / len(scores),
                    "min": min(scores),
                    "max": max(scores),
                }
        # Statistical significance (simplified)
        if stats.get("a") and stats.get("b"):
            diff = stats["b"]["mean"] - stats["a"]["mean"]
            stats["improvement"] = f"{diff:+.2f}"
            stats["winner"] = "b" if diff > 0.1 else "a" if diff < -0.1 else "tie"
        return stats

# Usage
test = ABTest(
    name="summarizer_v2_test",
    variant_a="Summarize this article in 3 bullet points:\n{text}",
    variant_b="You are an expert editor at The Economist. Summarize in exactly 3 bullets. Each bullet: bold key term, max 20 words, focus on implications.\n\nArticle:\n{text}",
    traffic_split=0.5
)

# In production request handler:
variant, prompt = test.get_prompt(user_id="user_123")
# ... get response, score it ...
test.record_result(variant, score=4.2)
print(test.get_stats())

from dataclasses import dataclass

@dataclass
class RegressionResult:
    version: str
    total_tests: int
    passed: int
    failed: int
    avg_score: float
    regressions: list[str]  # test IDs that got worse

def run_regression_suite(
    registry: PromptRegistry,
    prompt_name: str,
    new_version: str,
    test_cases: list[dict],
    threshold: float = 0.9
) -> RegressionResult:
    """Compare new prompt version against the current production version."""

    # Get current and new prompts
    try:
        current = registry.get(prompt_name)  # latest
    except FileNotFoundError:
        current = None
    new = registry.get(prompt_name, version=new_version)

    # Evaluate new version
    new_results = evaluate_prompt_version(new.template, test_cases, model=new.model)

    regressions = []
    if current:
        current_results = evaluate_prompt_version(current.template, test_cases, model=current.model)
        # Compare per-test
        for new_s, old_s in zip(new_results["individual_scores"], current_results["individual_scores"]):
            if new_s["overall_score"] < old_s["overall_score"] - 0.5:
                regressions.append(new_s["test_id"])

    passed = sum(1 for s in new_results["individual_scores"] if s["overall_score"] >= 3.5)
    result = RegressionResult(
        version=new_version,
        total_tests=len(test_cases),
        passed=passed,
        failed=len(test_cases) - passed,
        avg_score=new_results["avg_overall_score"],
        regressions=regressions
    )

    # Gate deployment
    pass_rate = passed / len(test_cases)
    if pass_rate < threshold:
        print(f"❌ BLOCKED: Pass rate {pass_rate:.0%} below threshold {threshold:.0%}")
    elif regressions:
        print(f"⚠️  WARNING: {len(regressions)} regressions detected: {regressions}")
    else:
        print(f"✅ PASSED: {pass_rate:.0%} pass rate, no regressions")

    return result