LLMOps & Continuous Improvement

1. Prompt Versioning

import json, hashlib
from datetime import datetime
from pathlib import Path

class PromptRegistry:
    """Version and manage prompts with content-addressable storage."""

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

    def register(self, name: str, template: str, metadata: dict = None):
        content_hash = hashlib.sha256(template.encode()).hexdigest()[:12]
        version = {
            "name": name,
            "hash": content_hash,
            "template": template,
            "metadata": metadata or {},
            "created_at": datetime.utcnow().isoformat(),
        }
        path = self.store / f"{name}_{content_hash}.json"
        path.write_text(json.dumps(version, indent=2))
        return content_hash

    def get(self, name: str, version_hash: str = None):
        if version_hash:
            path = self.store / f"{name}_{version_hash}.json"
            return json.loads(path.read_text())
        # Return latest version
        versions = sorted(self.store.glob(f"{name}_*.json"))
        return json.loads(versions[-1].read_text()) if versions else None

registry = PromptRegistry()
v1 = registry.register("summarizer", "Summarize: {text}")
v2 = registry.register("summarizer", "Provide a concise summary of: {text}")
print(f"v1={v1}, v2={v2}")

2. A/B Testing Framework

import hashlib, random
from dataclasses import dataclass

@dataclass
class ABExperiment:
    """Simple A/B test for prompt variants."""
    name: str
    variant_a: str
    variant_b: str
    traffic_split: float = 0.5   # fraction going to variant B

    def assign(self, user_id: str) -> str:
        """Deterministic assignment based on user ID hash."""
        h = hashlib.md5(f"{self.name}:{user_id}".encode()).hexdigest()
        bucket = int(h[:8], 16) / 0xFFFFFFFF
        if bucket < self.traffic_split:
            return "B"
        return "A"

    def get_prompt(self, user_id: str) -> str:
        variant = self.assign(user_id)
        return self.variant_a if variant == "A" else self.variant_b

exp = ABExperiment(
    name="summarizer_prompt",
    variant_a="Summarize the following text:\n{text}",
    variant_b="Write a 2-sentence summary:\n{text}",
)
for uid in ["user_101", "user_202", "user_303"]:
    print(f"{uid} -> variant {exp.assign(uid)}")

3. Online Evaluation and Feedback Loops

from dataclasses import dataclass, field
from datetime import datetime
import statistics

@dataclass
class FeedbackCollector:
    """Collect and aggregate user feedback for LLM outputs."""
    records: list = field(default_factory=list)

    def log(self, request_id: str, variant: str, rating: int,
           feedback_text: str = "", latency_ms: float = 0):
        self.records.append({
            "request_id": request_id, "variant": variant,
            "rating": rating, "feedback": feedback_text,
            "latency_ms": latency_ms,
            "timestamp": datetime.utcnow().isoformat(),
        })

    def summary(self):
        by_variant = {}
        for r in self.records:
            v = r["variant"]
            by_variant.setdefault(v, []).append(r["rating"])
        return {
            v: {"mean": statistics.mean(ratings), "n": len(ratings)}
            for v, ratings in by_variant.items()
        }

4. Model Registry

import mlflow

# Log a prompt experiment to MLflow
with mlflow.start_run(run_name="prompt_v2.1_test"):
    mlflow.log_param("prompt_version", "v2.1")
    mlflow.log_param("model", "gpt-4o-mini")
    mlflow.log_param("temperature", 0.7)

    # Log evaluation metrics
    mlflow.log_metric("mean_rating", 4.2)
    mlflow.log_metric("hallucination_rate", 0.03)
    mlflow.log_metric("p50_latency_ms", 820)
    mlflow.log_metric("cost_per_request", 0.0023)

    # Log the prompt template as an artifact
    mlflow.log_text(
        "Write a 2-sentence summary of:\n{text}",
        "prompt_template.txt"
    )