1import json
2from pathlib import Path
3from datetime import datetime
4from typing import Optional
5import hashlib
6 
7# Level 1: Prompt version control
8# System prompts are versioned artifacts, not hardcoded strings
Level 1: Prompt versioning — treat system prompts as versioned artifacts, not hardcoded strings
9# Each version has metadata, eval scores, and deployment history
10 
11PROMPT_STORE_PATH = Path("prompts/")
12 
version_hash: Content-addressed storage — the hash is the version identifier
13class PromptVersion:
14    def __init__(self, name: str, content: str, author: str, notes: str = ""):
15        self.name = name
16        self.content = content
17        self.author = author
18        self.notes = notes
19        self.created_at = datetime.utcnow().isoformat()
20        self.version_hash = hashlib.sha256(content.encode()).hexdigest()[:8]
21        self.eval_score: Optional[float] = None
22 
23    def save(self):
24        version_dir = PROMPT_STORE_PATH / self.name
Prompt store on disk (or git): enables diffing, rollback, and deployment history
25        version_dir.mkdir(parents=True, exist_ok=True)
26        version_file = version_dir / f"{self.version_hash}.json"
27        version_file.write_text(json.dumps({
28            "name": self.name,
29            "content": self.content,
30            "author": self.author,
31            "notes": self.notes,
32            "created_at": self.created_at,
33            "version_hash": self.version_hash,
34            "eval_score": self.eval_score,
active.json: Separate record of which version is current — change active without code deploy
35        }, indent=2))
36        return self.version_hash
37 
38def get_active_prompt(name: str) -> str:
39    """Load the currently active prompt version."""
40    active_file = PROMPT_STORE_PATH / name / "active.json"
41    if not active_file.exists():
eval_score gate: Only promote prompts that pass quality threshold — no manual bypassing
42        raise ValueError(f"No active prompt for '{name}'")
43    active_data = json.loads(active_file.read_text())
44    version_file = PROMPT_STORE_PATH / name / f"{active_data['version_hash']}.json"
45    return json.loads(version_file.read_text())["content"]
46 
47def promote_prompt(name: str, version_hash: str, eval_score: float):
48    """Promote a prompt version to active, with eval score gate."""
49    if eval_score < 0.80:
50        raise ValueError(f"Eval score {eval_score:.2f} below minimum 0.80. Not promoting.")
51    active_file = PROMPT_STORE_PATH / name / "active.json"
52    active_file.write_text(json.dumps({
53        "version_hash": version_hash,
54        "promoted_at": datetime.utcnow().isoformat(),
55        "eval_score": eval_score,
56    }))

1import subprocess
2from dataclasses import dataclass
3from typing import Callable
4 
5# Level 2: Eval-gated deployment pipeline
6# Every model or prompt change runs through automated quality checks
7# Deployment blocked if any check fails
8 
Level 2: Deployment pipeline — automated checks gate every change before it reaches users
9@dataclass
10class DeploymentCheck:
11    name: str
12    threshold: float
13    run: Callable[[], float]
14 
15class LLMDeploymentPipeline:
16    def __init__(self, service_name: str):
17        self.service = service_name
18        self.checks: list[DeploymentCheck] = []
19 
20    def add_check(self, name: str, threshold: float, check_fn: Callable):
21        self.checks.append(DeploymentCheck(name, threshold, check_fn))
22 
23    def run(self, new_version: dict) -> bool:
24        print(f"Running deployment pipeline for {self.service} v{new_version['version']}")
25        all_passed = True
Each check has a score and threshold — structured to allow adding new checks without refactoring
26 
27        for check in self.checks:
28            print(f"  Running check: {check.name}...")
29            score = check.run()
30            status = "PASS" if score >= check.threshold else "FAIL"
31            print(f"  {status}: {check.name} = {score:.3f} (threshold: {check.threshold})")
32 
33            if score < check.threshold:
34                all_passed = False
Gradual rollout: 5% → 25% → 50% → 100% with quality monitoring at each stage
35 
36        if all_passed:
37            print("All checks passed. Proceeding with gradual rollout.")
38            self._gradual_rollout(new_version, stages=[5, 25, 50, 100])
39        else:
40            print("Deployment BLOCKED. Fix failing checks before deploying.")
error_threshold=0.02: Roll back automatically if error rate exceeds 2% at any rollout stage
41 
42        return all_passed
43 
Automatic rollback: the pipeline rolls back without human intervention if quality drops
44    def _gradual_rollout(self, version: dict, stages: list[int]):
45        for pct in stages:
46            print(f"  Rolling out to {pct}% of traffic...")
47            # Update load balancer weights
48            self._set_traffic_split(version["id"], pct)
49            # Monitor for 5 minutes at each stage
50            if not self._monitor_quality(duration_minutes=5, error_threshold=0.02):
51                print(f"  Quality degradation detected at {pct}%. Rolling back.")
52                self._rollback()
53                return
54            print(f"  Stable at {pct}%.")
55        print("Deployment complete: 100% traffic on new version.")

1from openai import OpenAI
2import boto3
3from datetime import datetime, timedelta
4import random
5import json
6 
7client = OpenAI()
8cloudwatch = boto3.client("cloudwatch", region_name="us-east-1")
9 
10# Level 3: Production quality monitoring
11# Continuously samples production traffic and measures quality
12# Alerts before quality regressions become user complaints
Level 3: Continuous quality monitoring — LLM-as-judge on 5% of production traffic
13 
14class ProductionQualityMonitor:
sample_rate=0.05: 5% sampling balances cost (judge calls) vs. signal quality for quality monitoring
15    def __init__(self, sample_rate: float = 0.05):
16        self.sample_rate = sample_rate  # Sample 5% of production traffic
17        self.judge_model = "gpt-4o"
18 
19    def should_sample(self) -> bool:
20        return random.random() < self.sample_rate
21 
22    def evaluate_response(self, query: str, response: str) -> dict:
23        """LLM-as-judge evaluation on a sampled production response."""
24        judge_prompt = f"""Rate this AI response on a scale 1-5 for each dimension.
25Query: {query}
26Response: {response}
27 
28Respond with JSON: {{"accuracy": 1-5, "format": 1-5, "helpfulness": 1-5}}"""
29 
30        scores = client.chat.completions.create(
31            model=self.judge_model,
32            messages=[{"role": "user", "content": judge_prompt}],
33            response_format={"type": "json_object"},
34        )
35        return json.loads(scores.choices[0].message.content)
36 
37    def log_metrics(self, scores: dict, model_version: str):
38        """Push quality metrics to CloudWatch for dashboards and alerts."""
39        timestamp = datetime.utcnow()
CloudWatch metrics: quality scores become time-series data for dashboards and anomaly detection
40        metrics = []
41        for dimension, score in scores.items():
42            metrics.append({
43                "MetricName": f"ai_quality_{dimension}",
44                "Dimensions": [{"Name": "ModelVersion", "Value": model_version}],
45                "Value": score,
46                "Unit": "None",
47                "Timestamp": timestamp,
48            })
49 
Alert threshold at 3.5/5: Alert proactively before users notice degradation
50        cloudwatch.put_metric_data(
51            Namespace="AI/Quality",
52            MetricData=metrics,
53        )
54 
55        # Alert if any dimension drops below threshold
56        for dimension, score in scores.items():
57            if score < 3.5:
58                self._send_alert(f"Quality alert: {dimension} score {score:.1f} below 3.5")
59 
60    def _send_alert(self, message: str):
61        print(f"ALERT: {message}")
62        # SNS, PagerDuty, Slack integration here