LLMOps & Continuous Improvement

# Define PromptRegistry; implement __init__, register, get
# Key operations: results display, data protection, prompt construction
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}")

# Define ABExperiment; implement assign, get_prompt
# Key operations: results display, experiment framework, data protection
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)}")

# Define FeedbackCollector; implement log, summary
# Key operations: structured logging
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()
 }

# Experiment tracking setup
# Key operations: monitoring and metrics, structured logging, cost tracking
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"
 )