Prompt Engineering Infrastructure: Versioning and A/B Testing

from dataclasses import dataclass
from datetime import datetime
from enum import Enum
from typing import Optional, Dict, Any
import json
from pathlib import Path
 
class Environment(Enum):
    DEV = "dev"
    STAGING = "staging"
    PROD = "prod"
 
@dataclass
class PromptVersion:
    """A versioned prompt artifact."""
    name: str  # e.g., "summarization"
    version: str  # semantic version: "1.2.3"
    text: str  # the actual prompt
    environment: Environment
    created_at: datetime
    created_by: str
    approved_by: Optional[str] = None
    approved_at: Optional[datetime] = None
    metadata: Dict[str, Any] = None
 
class PromptRegistry:
    """Local file-based prompt registry with version control."""
 
    def __init__(self, registry_path: str = "./prompts"):
        self.registry_path = Path(registry_path)
        self.registry_path.mkdir(exist_ok=True)
 
    def register_prompt(self, prompt: PromptVersion) -> str:
        """Register a new prompt version in dev environment."""
        prompt_dir = self.registry_path / prompt.name
        prompt_dir.mkdir(exist_ok=True)
 
        # Store as JSON
        prompt_file = prompt_dir / f"{prompt.version}_{prompt.environment.value}.json"
 
        payload = {
            "name": prompt.name,
            "version": prompt.version,
            "text": prompt.text,
            "environment": prompt.environment.value,
            "created_at": prompt.created_at.isoformat(),
            "created_by": prompt.created_by,
            "approved_by": prompt.approved_by,
            "approved_at": prompt.approved_at.isoformat() if prompt.approved_at else None,
            "metadata": prompt.metadata or {}
        }
 
        with open(prompt_file, "w") as f:
            json.dump(payload, f, indent=2)
 
        return str(prompt_file)
 
    def promote_prompt(self, name: str, version: str,
                      from_env: Environment, to_env: Environment,
                      approved_by: str) -> PromptVersion:
        """Promote a prompt from one environment to another."""
        # Load existing prompt
        prompt_file = self.registry_path / name / f"{version}_{from_env.value}.json"
        with open(prompt_file, "r") as f:
            data = json.load(f)
 
        # Create new version in target environment
        prompt = PromptVersion(
            name=data["name"],
            version=data["version"],
            text=data["text"],
            environment=to_env,
            created_at=datetime.fromisoformat(data["created_at"]),
            created_by=data["created_by"],
            approved_by=approved_by,
            approved_at=datetime.now(),
            metadata=data.get("metadata", {})
        )
 
        self.register_prompt(prompt)
        return prompt
 
    def get_prompt(self, name: str, version: str,
                   environment: Environment) -> Optional[PromptVersion]:
        """Fetch a prompt from the registry."""
        prompt_file = self.registry_path / name / f"{version}_{environment.value}.json"
 
        if not prompt_file.exists():
            return None
 
        with open(prompt_file, "r") as f:
            data = json.load(f)
 
        return PromptVersion(
            name=data["name"],
            version=data["version"],
            text=data["text"],
            environment=Environment(data["environment"]),
            created_at=datetime.fromisoformat(data["created_at"]),
            created_by=data["created_by"],
            approved_by=data.get("approved_by"),
            approved_at=datetime.fromisoformat(data["approved_at"])
                        if data.get("approved_at") else None,
            metadata=data.get("metadata", {})
        )
 
    def get_latest_prod(self, name: str) -> Optional[PromptVersion]:
        """Get the latest production version of a prompt."""
        prompt_dir = self.registry_path / name
        if not prompt_dir.exists():
            return None
 
        # Find all prod versions and sort by version
        prod_files = list(prompt_dir.glob("*_prod.json"))
        if not prod_files:
            return None
 
        # Simple version sorting (better: use packaging.version)
        prod_files.sort(reverse=True)
 
        with open(prod_files[0], "r") as f:
            data = json.load(f)
 
        return PromptVersion(
            name=data["name"],
            version=data["version"],
            text=data["text"],
            environment=Environment(data["environment"]),
            created_at=datetime.fromisoformat(data["created_at"]),
            created_by=data["created_by"],
            approved_by=data.get("approved_by"),
            approved_at=datetime.fromisoformat(data["approved_at"])
                        if data.get("approved_at") else None,
            metadata=data.get("metadata", {})
        )
 
# Usage example
if __name__ == "__main__":
    registry = PromptRegistry()
 
    # Create and register a new prompt
    prompt_v1 = PromptVersion(
        name="summarization",
        version="1.0.0",
        text="Summarize the following text in 2-3 sentences:\n\n{text}",
        environment=Environment.DEV,
        created_at=datetime.now(),
        created_by="alice@company.com",
        metadata={"model": "gpt-4", "temperature": 0.3}
    )
 
    print("Registering prompt v1.0.0...")
    registry.register_prompt(prompt_v1)
 
    # Promote to staging
    print("Promoting to staging...")
    registry.promote_prompt("summarization", "1.0.0",
                           Environment.DEV, Environment.STAGING,
                           "bob@company.com")
 
    # Promote to production
    print("Promoting to production...")
    registry.promote_prompt("summarization", "1.0.0",
                           Environment.STAGING, Environment.PROD,
                           "charlie@company.com")
 
    # Fetch from production
    prod_prompt = registry.get_latest_prod("summarization")
    print(f"\nProduction prompt:\n{prod_prompt.text}")
    print(f"Version: {prod_prompt.version}")
    print(f"Approved by: {prod_prompt.approved_by}")

import hashlib
import json
from dataclasses import dataclass
from enum import Enum
from typing import Tuple, Optional
from datetime import datetime
import random
 
class Variant(Enum):
    CONTROL = "A"
    TREATMENT = "B"
 
@dataclass
class ExperimentAssignment:
    """Track which variant a request was assigned to."""
    request_id: str
    variant: Variant
    experiment_id: str
    timestamp: datetime
    user_id: Optional[str] = None
 
class ExperimentGateway:
    """Route requests to prompt variants for A/B testing."""
 
    def __init__(self, control_prompt_version: str,
                 treatment_prompt_version: str):
        self.control_version = control_prompt_version
        self.treatment_version = treatment_prompt_version
        self.assignments = []  # In reality, use a database
 
    def assign_variant(self, request_id: str, user_id: Optional[str] = None,
                      traffic_split: float = 0.5) -> ExperimentAssignment:
        """
        Deterministically assign a request to A or B.
 
        Uses consistent hashing so the same user always sees the same variant.
        """
        # Use hash(user_id) for consistent assignment, or random for anonymous
        if user_id:
            hash_input = user_id
        else:
            hash_input = request_id
 
        # Hash to a float [0, 1]
        hash_value = int(hashlib.md5(hash_input.encode()).hexdigest(), 16)
        assignment_value = (hash_value % 1000) / 1000.0
 
        # Assign based on traffic split
        if assignment_value < traffic_split:
            variant = Variant.CONTROL
        else:
            variant = Variant.TREATMENT
 
        assignment = ExperimentAssignment(
            request_id=request_id,
            variant=variant,
            experiment_id="summarization_v1_0_1",
            timestamp=datetime.now(),
            user_id=user_id
        )
 
        self.assignments.append(assignment)
        return assignment
 
    def get_prompt_version(self, assignment: ExperimentAssignment) -> str:
        """Return the prompt version for this assignment."""
        if assignment.variant == Variant.CONTROL:
            return self.control_version
        else:
            return self.treatment_version
 
    def log_quality_metric(self, request_id: str, variant: Variant,
                          metric_name: str, metric_value: float):
        """Log quality metrics per variant for analysis."""
        # In reality, write to metrics backend (Prometheus, CloudWatch, etc.)
        print(f"METRIC: {metric_name}={metric_value} variant={variant.value} request={request_id}")
 
# Usage example
if __name__ == "__main__":
    gateway = ExperimentGateway(
        control_prompt_version="summarization:1.0.0",
        treatment_prompt_version="summarization:1.1.0"
    )
 
    # Simulate 10 requests
    for i in range(10):
        user_id = f"user_{i % 3}"  # 3 unique users
        assignment = gateway.assign_variant(
            request_id=f"req_{i}",
            user_id=user_id,
            traffic_split=0.5
        )
 
        prompt_version = gateway.get_prompt_version(assignment)
        print(f"Request {i}: User {user_id} → Variant {assignment.variant.value} (prompt {prompt_version})")
 
    # Count assignments
    control_count = sum(1 for a in gateway.assignments if a.variant == Variant.CONTROL)
    treatment_count = sum(1 for a in gateway.assignments if a.variant == Variant.TREATMENT)
 
    print(f"\nAssignments: Control={control_count}, Treatment={treatment_count}")
 
    # Consistent hashing check
    print("\nConsistent assignment check (same user → same variant):")
    user_1_assignments = [a for a in gateway.assignments if a.user_id == "user_1"]
    variant_set = set(a.variant for a in user_1_assignments)
    print(f"User 1 always sees: {list(variant_set)}")

from scipy import stats
from dataclasses import dataclass
from typing import List
import numpy as np
 
@dataclass
class MetricResult:
    """Results from an A/B test."""
    metric_name: str
    control_mean: float
    treatment_mean: float
    control_std: float
    treatment_std: float
    p_value: float
    is_significant: bool
    confidence_interval: Tuple[float, float]
 
class ABTestAnalyzer:
    """Analyze A/B test results for statistical significance."""
 
    def __init__(self, alpha: float = 0.05):
        """
        Initialize analyzer.
 
        alpha: significance level (0.05 = 95% confidence)
        """
        self.alpha = alpha
 
    def analyze(self, control_values: List[float],
                treatment_values: List[float],
                metric_name: str) -> MetricResult:
        """
        Compare two groups using t-test.
 
        Returns MetricResult with p-value and significance.
        """
        control = np.array(control_values)
        treatment = np.array(treatment_values)
 
        # Welch's t-test (doesn't assume equal variances)
        t_stat, p_value = stats.ttest_ind(treatment, control, equal_var=False)
 
        # 95% confidence interval for difference in means
        diff_mean = treatment.mean() - control.mean()
        se_diff = np.sqrt(
            (treatment.std()**2 / len(treatment)) +
            (control.std()**2 / len(control))
        )
        ci_lower = diff_mean - 1.96 * se_diff
        ci_upper = diff_mean + 1.96 * se_diff
 
        is_significant = p_value < self.alpha
 
        return MetricResult(
            metric_name=metric_name,
            control_mean=float(control.mean()),
            treatment_mean=float(treatment.mean()),
            control_std=float(control.std()),
            treatment_std=float(treatment.std()),
            p_value=float(p_value),
            is_significant=is_significant,
            confidence_interval=(float(ci_lower), float(ci_upper))
        )
 
# Usage example
if __name__ == "__main__":
    analyzer = ABTestAnalyzer(alpha=0.05)
 
    # Simulate quality scores from 100 requests per variant
    # Control: mean 0.82, std 0.12
    np.random.seed(42)
    control_scores = np.random.normal(0.82, 0.12, 100)
 
    # Treatment: mean 0.87, std 0.11 (better!)
    treatment_scores = np.random.normal(0.87, 0.11, 100)
 
    result = analyzer.analyze(
        control_values=control_scores.tolist(),
        treatment_values=treatment_scores.tolist(),
        metric_name="output_quality_score"
    )
 
    print(f"Metric: {result.metric_name}")
    print(f"Control:   mean={result.control_mean:.4f}, std={result.control_std:.4f}")
    print(f"Treatment: mean={result.treatment_mean:.4f}, std={result.treatment_std:.4f}")
    print(f"Difference: {result.treatment_mean - result.control_mean:.4f}")
    print(f"95% CI: [{result.confidence_interval[0]:.4f}, {result.confidence_interval[1]:.4f}]")
    print(f"P-value: {result.p_value:.6f}")
    print(f"Significant (p < 0.05): {result.is_significant}")

from enum import Enum
from dataclasses import dataclass
from typing import List, Dict, Any
import re
from abc import ABC, abstractmethod
 
class MetricType(Enum):
    ACCURACY = "accuracy"
    LENGTH = "length"
    FORMAT = "format"
    CLARITY = "clarity"
 
@dataclass
class EvaluationResult:
    """Result of evaluating a prompt output."""
    prompt_version: str
    test_case_id: str
    metric_type: MetricType
    score: float  # 0-1
    details: Dict[str, Any]
 
class EvaluationMetric(ABC):
    """Base class for evaluation metrics."""
 
    @abstractmethod
    def evaluate(self, expected: str, actual: str) -> EvaluationResult:
        pass
 
class LengthMetric(EvaluationMetric):
    """Check if output has correct number of sentences."""
 
    def __init__(self, target_sentences: int = 3, prompt_version: str = ""):
        self.target = target_sentences
        self.prompt_version = prompt_version
 
    def evaluate(self, expected: str, actual: str) -> EvaluationResult:
        # Count sentences (simple: ends with . ! or ?)
        sentences = [s.strip() for s in re.split(r'[.!?]+', actual) if s.strip()]
 
        # Score: 1.0 if exact, 0.5 if off by 1, 0.0 if off by >1
        diff = abs(len(sentences) - self.target)
        if diff == 0:
            score = 1.0
        elif diff == 1:
            score = 0.5
        else:
            score = max(0, 1.0 - (diff * 0.25))
 
        return EvaluationResult(
            prompt_version=self.prompt_version,
            test_case_id="length_check",
            metric_type=MetricType.LENGTH,
            score=score,
            details={
                "target_sentences": self.target,
                "actual_sentences": len(sentences),
                "difference": diff
            }
        )
 
class FormatMetric(EvaluationMetric):
    """Check if output matches required format."""
 
    def __init__(self, pattern: str, prompt_version: str = ""):
        self.pattern = pattern
        self.prompt_version = prompt_version
 
    def evaluate(self, expected: str, actual: str) -> EvaluationResult:
        # Check if actual matches the regex pattern
        matches = bool(re.match(self.pattern, actual))
 
        return EvaluationResult(
            prompt_version=self.prompt_version,
            test_case_id="format_check",
            metric_type=MetricType.FORMAT,
            score=1.0 if matches else 0.0,
            details={
                "pattern": self.pattern,
                "matches": matches
            }
        )
 
class SimulatedLLMJudgeMetric(EvaluationMetric):
    """
    Simulate LLM-as-judge evaluation for accuracy.
    In production, call Claude/GPT-4 with structured output.
    """
 
    def __init__(self, prompt_version: str = ""):
        self.prompt_version = prompt_version
 
    def evaluate(self, expected: str, actual: str) -> EvaluationResult:
        """
        In production, this would call:
        response = client.messages.create(
            model="claude-3-opus",
            messages=[{
                "role": "user",
                "content": f"Judge accuracy of this summary...
                    Original: {expected}
                    Summary: {actual}
                    Return JSON with score (0-100)."
            }]
        )
        score = json.loads(response.content[0].text)['accuracy_score']
 
        For this demo, we score based on content overlap.
        """
        # Simple heuristic: count overlapping words
        expected_words = set(expected.lower().split())
        actual_words = set(actual.lower().split())
 
        overlap = len(expected_words & actual_words)
        total = len(expected_words | actual_words)
 
        # Jaccard similarity
        score = overlap / total if total > 0 else 0.0
 
        return EvaluationResult(
            prompt_version=self.prompt_version,
            test_case_id="accuracy_judge",
            metric_type=MetricType.ACCURACY,
            score=score,
            details={
                "overlap_words": overlap,
                "total_words": total,
                "jaccard_similarity": score
            }
        )
 
class PromptEvaluator:
    """Run evaluation suite on prompt outputs."""
 
    def __init__(self, metrics: List[EvaluationMetric]):
        self.metrics = metrics
 
    def evaluate_batch(self, test_cases: List[Dict[str, str]]) -> Dict[str, Any]:
        """
        Run all metrics on all test cases.
 
        test_cases: [{"input": "...", "expected": "...", "actual": "..."}, ...]
        """
        all_results = []
 
        for test_case in test_cases:
            for metric in self.metrics:
                result = metric.evaluate(
                    expected=test_case["expected"],
                    actual=test_case["actual"]
                )
                all_results.append(result)
 
        # Aggregate scores by metric type
        summary = {}
        for result in all_results:
            key = result.metric_type.value
            if key not in summary:
                summary[key] = []
            summary[key].append(result.score)
 
        # Calculate mean scores
        aggregate = {}
        for metric_type, scores in summary.items():
            aggregate[metric_type] = {
                "mean": np.mean(scores),
                "std": np.std(scores),
                "min": np.min(scores),
                "max": np.max(scores),
                "count": len(scores)
            }
 
        return {
            "summary": aggregate,
            "detailed_results": all_results
        }
 
# Usage example
if __name__ == "__main__":
    # Test cases: original text, expected summary, actual output from LLM
    test_cases = [
        {
            "input": "The Arctic ice sheet is shrinking rapidly due to climate change.",
            "expected": "Arctic ice is shrinking due to climate change.",
            "actual": "The Arctic ice sheet shrinks rapidly from climate change."
        },
        {
            "input": "Machine learning has transformed industries from healthcare to finance.",
            "expected": "Machine learning changed healthcare and finance.",
            "actual": "ML transformed healthcare and finance industries significantly."
        },
        {
            "input": "Coffee plants grow in tropical regions around the equator.",
            "expected": "Coffee grows in tropical regions.",
            "actual": "Coffee plants are found in tropical equatorial zones."
        }
    ]
 
    # Create evaluator with multiple metrics
    evaluator = PromptEvaluator(metrics=[
        LengthMetric(target_sentences=1, prompt_version="summarization:1.1.0"),
        FormatMetric(
            pattern=r"^[A-Z].*\.$",  # Starts with capital, ends with period
            prompt_version="summarization:1.1.0"
        ),
        SimulatedLLMJudgeMetric(prompt_version="summarization:1.1.0")
    ])
 
    # Run evaluation
    import numpy as np
    results = evaluator.evaluate_batch(test_cases)
 
    print("Evaluation Results")
    print("=" * 50)
    for metric_type, stats in results["summary"].items():
        print(f"\n{metric_type.upper()}:")
        print(f"  Mean: {stats['mean']:.3f}")
        print(f"  Std:  {stats['std']:.3f}")
        print(f"  Min:  {stats['min']:.3f}")
        print(f"  Max:  {stats['max']:.3f}")

from jinja2 import Template, Environment
from dataclasses import dataclass
from typing import Dict, Any
 
@dataclass
class PromptTemplate:
    """A templated prompt with variables and logic."""
    name: str
    template_string: str
    version: str
 
    def render(self, **kwargs) -> str:
        """Fill in variables and return final prompt."""
        template = Template(self.template_string)
        return template.render(**kwargs)
 
# Define templates with variables and logic
 
SUMMARIZATION_TEMPLATE = PromptTemplate(
    name="summarization",
    version="1.1.0",
    template_string="""You are a professional summarizer. Summarize the following text in exactly {{ target_sentences }} sentence{{ 's' if target_sentences != 1 else '' }}.
 
Focus on:
{% for focus_point in focus_points %}
- {{ focus_point }}
{% endfor %}
 
Original text:
{{ text }}
 
Summary:"""
)
 
EXTRACTION_TEMPLATE = PromptTemplate(
    name="entity_extraction",
    version="1.0.0",
    template_string="""Extract {{ entity_types | join(', ') }} from this text.
 
Return as JSON:
{
{% for entity_type in entity_types %}
  "{{ entity_type }}": [],
{% endfor %}
}
 
Text:
{{ text }}
 
JSON:"""
)
 
# Usage example
if __name__ == "__main__":
    # Summarization with variables
    summary_prompt = SUMMARIZATION_TEMPLATE.render(
        target_sentences=2,
        focus_points=["main findings", "implications"],
        text="Machine learning models are increasingly used in clinical diagnosis. "
             "Recent studies show 95% accuracy in detecting early-stage cancer. "
             "However, bias in training data remains a concern."
    )
 
    print("Generated Summarization Prompt:")
    print("-" * 50)
    print(summary_prompt)
    print()
 
    # Entity extraction with dynamic types
    extraction_prompt = EXTRACTION_TEMPLATE.render(
        entity_types=["Person", "Organization", "Location"],
        text="Dr. Sarah Chen from Stanford University presented findings in New York."
    )
 
    print("Generated Extraction Prompt:")
    print("-" * 50)
    print(extraction_prompt)
 
class PromptChain:
    """Chain multiple prompts: output of one feeds into next."""
 
    def __init__(self, name: str):
        self.name = name
        self.steps = []
 
    def add_step(self, prompt_template: PromptTemplate,
                input_var: str = None):
        """Add a prompt to the chain."""
        self.steps.append({
            "template": prompt_template,
            "input_var": input_var or "input"
        })
 
    def execute(self, initial_input: str) -> Dict[str, str]:
        """Run all steps in sequence."""
        results = {"step_0": initial_input}
 
        for i, step in enumerate(self.steps):
            template = step["template"]
            input_var = step["input_var"]
            previous_output = results[f"step_{i}"]
 
            # Render the template with the previous output
            prompt = template.render(**{input_var: previous_output})
 
            # In reality, call LLM here
            # For demo, just echo
            results[f"step_{i+1}"] = f"[LLM Response to step {i+1}]"
 
        return results
 
# Prompt chaining example
if __name__ == "__main__":
    print("\n" + "=" * 50)
    print("Prompt Chaining Example")
    print("=" * 50 + "\n")
 
    chain = PromptChain("summarize_then_extract")
    chain.add_step(SUMMARIZATION_TEMPLATE, input_var="text")
    chain.add_step(EXTRACTION_TEMPLATE, input_var="text")
 
    long_text = """
    Dr. Jane Smith from MIT and Dr. Robert Chen from Stanford collaborated on
    groundbreaking AI research. They published findings in Nature AI last month,
    demonstrating 99.2% accuracy in medical image classification. The work was
    funded by the NIH and conducted in Boston and Palo Alto.
    """
 
    results = chain.execute(long_text)
 
    for step, output in results.items():
        print(f"{step}: {output}")

from dataclasses import dataclass
from datetime import datetime
from typing import Dict, List, Optional
import json
 
@dataclass
class PromptCostEvent:
    """Track cost and quality metrics per prompt version."""
    timestamp: datetime
    request_id: str
    prompt_version: str
    variant: str  # "A" or "B"
    input_tokens: int
    output_tokens: int
    cost_usd: float
    output_quality_score: Optional[float] = None
    latency_ms: Optional[float] = None
 
class PromptObservabilityStore:
    """In-memory store for prompt metrics (in prod: time-series DB)."""
 
    def __init__(self):
        self.events: List[PromptCostEvent] = []
 
    def record_event(self, event: PromptCostEvent):
        """Record a prompt execution event."""
        self.events.append(event)
 
    def aggregate_by_version(self) -> Dict[str, Dict]:
        """Aggregate metrics by prompt version."""
        aggregates = {}
 
        for event in self.events:
            version = event.prompt_version
            if version not in aggregates:
                aggregates[version] = {
                    "count": 0,
                    "total_cost": 0.0,
                    "total_tokens": 0,
                    "quality_scores": [],
                    "latencies": []
                }
 
            agg = aggregates[version]
            agg["count"] += 1
            agg["total_cost"] += event.cost_usd
            agg["total_tokens"] += event.input_tokens + event.output_tokens
 
            if event.output_quality_score is not None:
                agg["quality_scores"].append(event.output_quality_score)
            if event.latency_ms is not None:
                agg["latencies"].append(event.latency_ms)
 
        # Calculate means
        for version, agg in aggregates.items():
            if agg["quality_scores"]:
                agg["mean_quality"] = np.mean(agg["quality_scores"])
                agg["std_quality"] = np.std(agg["quality_scores"])
            else:
                agg["mean_quality"] = None
 
            if agg["latencies"]:
                agg["mean_latency_ms"] = np.mean(agg["latencies"])
            else:
                agg["mean_latency_ms"] = None
 
            agg["cost_per_request"] = agg["total_cost"] / agg["count"]
 
        return aggregates
 
    def detect_regressions(self, baseline_version: str,
                           current_version: str,
                           threshold: float = 0.05) -> Dict[str, bool]:
        """
        Check if current version regressed vs baseline.
 
        threshold: if quality drops >5%, flag as regression
        """
        agg = self.aggregate_by_version()
 
        baseline = agg.get(baseline_version)
        current = agg.get(current_version)
 
        if not baseline or not current:
            return {}
 
        regressions = {}
 
        # Quality regression check
        if (baseline.get("mean_quality") and current.get("mean_quality")):
            quality_drop = (
                (baseline["mean_quality"] - current["mean_quality"]) /
                baseline["mean_quality"]
            )
            regressions["quality_regression"] = quality_drop > threshold
 
        # Cost increase check
        cost_increase = (
            (current["cost_per_request"] - baseline["cost_per_request"]) /
            baseline["cost_per_request"]
        )
        regressions["cost_regression"] = cost_increase > threshold
 
        return regressions
 
# Usage example
if __name__ == "__main__":
    store = PromptObservabilityStore()
 
    # Simulate events from v1.0.0 (baseline)
    np.random.seed(42)
    for i in range(50):
        store.record_event(PromptCostEvent(
            timestamp=datetime.now(),
            request_id=f"req_{i}",
            prompt_version="summarization:1.0.0",
            variant="A",
            input_tokens=500 + np.random.randint(-50, 50),
            output_tokens=150 + np.random.randint(-20, 20),
            cost_usd=0.015,
            output_quality_score=0.82 + np.random.normal(0, 0.08),
            latency_ms=1200 + np.random.randint(-200, 200)
        ))
 
    # Simulate events from v1.1.0 (new version, slightly worse quality)
    for i in range(50):
        store.record_event(PromptCostEvent(
            timestamp=datetime.now(),
            request_id=f"req_new_{i}",
            prompt_version="summarization:1.1.0",
            variant="B",
            input_tokens=520 + np.random.randint(-50, 50),
            output_tokens=160 + np.random.randint(-20, 20),
            cost_usd=0.016,  # 7% more expensive
            output_quality_score=0.78 + np.random.normal(0, 0.08),  # 5% worse quality
            latency_ms=1300 + np.random.randint(-200, 200)
        ))
 
    # Aggregate and report
    agg = store.aggregate_by_version()
 
    print("Observability Report")
    print("=" * 60)
    for version, metrics in agg.items():
        print(f"\n{version}:")
        print(f"  Requests: {metrics['count']}")
        print(f"  Mean Quality: {metrics.get('mean_quality', 'N/A'):.3f}")
        print(f"  Total Cost: ${metrics['total_cost']:.2f}")
        print(f"  Cost/Request: ${metrics['cost_per_request']:.4f}")
        print(f"  Mean Latency: {metrics.get('mean_latency_ms', 'N/A'):.0f}ms")
 
    # Check for regressions
    print("\n" + "=" * 60)
    print("Regression Detection")
    print("=" * 60)
    regressions = store.detect_regressions(
        baseline_version="summarization:1.0.0",
        current_version="summarization:1.1.0",
        threshold=0.05
    )
 
    for regression_type, detected in regressions.items():
        status = "ALERT!" if detected else "OK"
        print(f"{regression_type}: {status}")

# Observability schema for production prompt usage
from dataclasses import dataclass
from datetime import datetime
from typing import Optional, Dict, Any
import json
 
@dataclass
class PromptRequest:
    """Logged observation of a prompt in production."""
    request_id: str
    timestamp: datetime
    prompt_name: str
    prompt_version: str
    model_name: str
    model_version: str
    input_text: str  # User's input
    output_text: str  # Model's output
 
    # Quality signals
    output_length: int
    latency_ms: float
    cost_cents: float
 
    # Custom metrics (domain-specific)
    custom_metrics: Dict[str, Any]  # e.g., {"factuality": 0.92, "readability": 0.78}
 
    # User feedback (logged later, after user interaction)
    user_feedback: Optional[str] = None  # "thumbs_up", "thumbs_down", null
    user_correction: Optional[str] = None  # If user corrected the output
 
    # Metadata for debugging
    user_id: Optional[str] = None
    session_id: Optional[str] = None
    ip_address: Optional[str] = None
 
def log_prompt_request(request: PromptRequest):
    """Log to observability backend (DataDog, Honeycomb, etc.)"""
    event = {
        "request_id": request.request_id,
        "timestamp": request.timestamp.isoformat(),
        "prompt": f"{request.prompt_name}:{request.prompt_version}",
        "model": f"{request.model_name}:{request.model_version}",
        "metrics": {
            "output_length": request.output_length,
            "latency_ms": request.latency_ms,
            "cost_cents": request.cost_cents,
            **request.custom_metrics,
        },
        "feedback": request.user_feedback,
        "corrected": request.user_correction is not None,
    }
 
    # Send to observability platform
    send_to_datadog(event)
    # Or Honeycomb, NewRelic, etc.