7. Production Checklist and Deployment

Overview

The complete checklist you have to work through before deploying an LLM system to production: evaluation, tests, cost estimation, rollback plan, monitoring, security. Deployment strategies (canary, blue-green, feature flags). The post-deploy process.

Why You Need a Checklist

In the excitement of having something that works, it's easy to skip critical steps:

Deploys without a checklist:
✗ "It seemed to work in my tests"
✗ No cost estimate — surprise at the end of the month
✗ No rollback — a 4-hour incident to revert
✗ No monitoring — you find out about the problem 3 days later
✗ API key in the code → leak on GitHub

Deploys with a checklist:
✓ Evaluation passed with documented metrics
✓ Estimated cost: $X/month at Y requests/day
✓ Rollback documented and tested: 2 commands
✓ Alerts configured — 5 minutes to detect a problem
✓ Secrets in environment variables

The Complete Pre-Deploy Checklist

Section 1: Evaluation and Quality

## Evaluation Checklist

### Golden Set
- [ ] Golden set exists with at least 100 examples
- [ ] Correct distribution: happy path (60%) + edge cases (30%) + adversarial (10%)
- [ ] Human review of 20% of the golden set completed
- [ ] Golden set versioned in Git or dataset storage

### Quality Metrics
- [ ] Accuracy >= defined threshold (e.g. 85%)
- [ ] Faithfulness >= 0.80 (for summarization/RAG systems)
- [ ] Format compliance >= 0.95 (for structured output)
- [ ] LLM-as-judge run on a sample of 50+ examples

### Regression Tests
- [ ] Baseline saved in baseline.json
- [ ] Regression tests pass (no metric below the baseline - tolerance)
- [ ] Test suite run against the current code (not against a cache)
- [ ] Edge case tests run
- [ ] Adversarial tests run

Section 2: Performance and Cost

## Performance Checklist

### Latency
- [ ] p95 latency measured: ___ ms (does it meet the SLA?)
- [ ] p99 latency measured: ___ ms
- [ ] Timeout configured (e.g. 30s max)
- [ ] Retry logic implemented (3 attempts with exponential backoff)

### Cost
- [ ] Average tokens per request measured: ___ tokens
- [ ] Estimated cost per request: $___
- [ ] Estimated monthly cost at projected traffic: $___ /month
- [ ] Monthly cost within the approved budget: YES/NO
- [ ] Budget alerts configured (80% and 100% of the budget)
- [ ] max_tokens configured (don't leave it unbounded)
- [ ] Caching configured if it applies (expected hit rate: ___%)

### Scaling
- [ ] API rate limits accounted for
- [ ] Rate limiting on your own service configured
- [ ] Queue or async if traffic spikes are expected

Section 3: Reliability

## Reliability Checklist

### Error Handling
- [ ] Try/except on every API call
- [ ] Retry with exponential backoff implemented
- [ ] Fallback defined for when the API fails
- [ ] Timeout configured
- [ ] Output parsing with error handling (don't assume the format is right)
- [ ] Output validation implemented

### Availability
- [ ] API dependency documented (what happens if OpenAI goes down?)
- [ ] Fallback to an alternative model if it applies
- [ ] Circuit breaker if it applies (stop calls if error rate > X%)

Section 4: Versioning and Rollback

## Versioning Checklist

### Versioning
- [ ] Prompt registered with a semantic version (vX.Y.Z)
- [ ] Changelog updated with a description of the change
- [ ] Previous version documented (for rollback)
- [ ] Registry updated with the new version as "active"

### Rollback Plan
- [ ] Previous stable version identified: ___
- [ ] Rollback command/process documented
- [ ] Rollback tested in staging (not just documented)
- [ ] Estimated rollback time: ___ minutes
- [ ] Person responsible for executing the rollback: ___

Section 5: Monitoring and Alerts

## Monitoring Checklist

### Metrics
- [ ] Latency (p50, p95) monitored
- [ ] Error rate monitored
- [ ] Cost per request monitored
- [ ] Tokens per request monitored
- [ ] Quality score monitored (sampling)

### Alerts
- [ ] High latency alert configured (p95 > SLA)
- [ ] High error rate alert configured (> 1-5%)
- [ ] High cost alert configured (> budget)
- [ ] Quality degradation alert configured
- [ ] Alert recipients configured (Slack/email)

### Dashboards
- [ ] Basic operational metrics dashboard available
- [ ] Dashboard access shared with the team

Section 6: Security

## Security Checklist

### Secrets
- [ ] API key in environment variables (NOT in code or logs)
- [ ] .env in .gitignore
- [ ] Secrets rotated if they were ever exposed
- [ ] API key access restricted (not shared)

### Input/Output
- [ ] Input validation implemented (cap length, characters)
- [ ] Inputs sanitized before being inserted into prompts
- [ ] Output filtering if there's a risk of PII or sensitive content
- [ ] Prompt injection mitigated (system vs user instructions kept apart)

### Logging
- [ ] PII not logged (names, emails, sensitive data)
- [ ] Logs have an appropriate retention policy
- [ ] Audit trail for prompt changes

Implementing the Checklist in Code

from dataclasses import dataclass, field
from typing import Optional
import json


@dataclass
class ChecklistItem:
    """One checklist item."""
    id: str
    section: str
    description: str
    required: bool = True
    completed: bool = False
    value: Optional[str] = None
    notes: str = ""


class ProductionChecklist:
    """
    Executable checklist for LLM system deploys.
    
    Produces an evidence document for the deploy process.
    """
    
    def __init__(self, prompt_name: str, version: str, owner: str):
        self.prompt_name = prompt_name
        self.version = version
        self.owner = owner
        self.start_date = None
        self.completed_date = None
        self.items: list[ChecklistItem] = self._init_items()
    
    def _init_items(self) -> list[ChecklistItem]:
        """Initializes every checklist item."""
        return [
            # EVALUATION
            ChecklistItem("eval_1", "Evaluation", "A golden set with 100+ examples exists"),
            ChecklistItem("eval_2", "Evaluation", "Accuracy >= defined threshold", required=True),
            ChecklistItem("eval_3", "Evaluation", "Regression tests pass (no regression)"),
            ChecklistItem("eval_4", "Evaluation", "Format compliance >= 0.95 (if structured output)"),
            
            # PERFORMANCE
            ChecklistItem("perf_1", "Performance", "p95 latency measured and within the SLA"),
            ChecklistItem("perf_2", "Performance", "Estimated monthly cost computed"),
            ChecklistItem("perf_3", "Performance", "max_tokens configured"),
            ChecklistItem("perf_4", "Performance", "Retry logic with exponential backoff implemented"),
            
            # RELIABILITY
            ChecklistItem("rel_1", "Reliability", "Error handling on every API call"),
            ChecklistItem("rel_2", "Reliability", "Timeout configured"),
            ChecklistItem("rel_3", "Reliability", "Output validation implemented"),
            ChecklistItem("rel_4", "Reliability", "Fallback defined", required=False),
            
            # VERSIONING
            ChecklistItem("ver_1", "Versioning", "Prompt registered with semver"),
            ChecklistItem("ver_2", "Versioning", "Changelog updated"),
            ChecklistItem("ver_3", "Versioning", "Rollback plan documented"),
            
            # MONITORING
            ChecklistItem("mon_1", "Monitoring", "Latency and error rate monitored"),
            ChecklistItem("mon_2", "Monitoring", "Cost alerts configured"),
            ChecklistItem("mon_3", "Monitoring", "Quality alerts configured"),
            
            # SECURITY
            ChecklistItem("sec_1", "Security", "API keys in environment variables (not in code)"),
            ChecklistItem("sec_2", "Security", "Input validation implemented"),
            ChecklistItem("sec_3", "Security", "PII not logged"),
        ]
    
    def complete(self, item_id: str, value: str = "", notes: str = "") -> None:
        """Marks an item as completed."""
        for item in self.items:
            if item.id == item_id:
                item.completed = True
                item.value = value
                item.notes = notes
                print(f"✅ [{item_id}] {item.description}")
                return
        raise KeyError(f"Item '{item_id}' not found")
    
    def complete_section(self, section: str, values: dict = None) -> None:
        """Marks every item in a section as completed."""
        for item in self.items:
            if item.section == section:
                value = (values or {}).get(item.id, "")
                item.completed = True
                item.value = value
    
    def can_deploy(self) -> tuple[bool, list[str]]:
        """
        Checks whether you can deploy.
        
        Returns: (can_deploy, blockers)
        """
        blockers = [
            item.description
            for item in self.items
            if item.required and not item.completed
        ]
        
        return len(blockers) == 0, blockers
    
    def summary(self) -> dict:
        """Summary of the checklist state."""
        total_required = sum(1 for i in self.items if i.required)
        completed_required = sum(1 for i in self.items if i.required and i.completed)
        completed_optional = sum(1 for i in self.items if not i.required and i.completed)
        
        return {
            "prompt_name": self.prompt_name,
            "version": self.version,
            "owner": self.owner,
            "required": f"{completed_required}/{total_required}",
            "optional": f"{completed_optional}/{len(self.items) - total_required}",
            "can_deploy": completed_required == total_required,
            "pending_items": [i.description for i in self.items if not i.completed]
        }
    
    def generate_report(self) -> str:
        """Generates a checklist report in markdown."""
        can_deploy, blockers = self.can_deploy()
        
        from datetime import datetime
        lines = [
            f"# Production Deploy Checklist",
            f"**Prompt:** {self.prompt_name} {self.version}  ",
            f"**Owner:** {self.owner}  ",
            f"**Date:** {datetime.now().strftime('%Y-%m-%d %H:%M')}  ",
            f"**Status:** {'✅ APPROVED for deploy' if can_deploy else '❌ BLOCKED'}  ",
            "",
        ]
        
        if blockers:
            lines.extend([
                "## ❌ Blockers (complete before deploying)",
                "",
            ])
            for b in blockers:
                lines.append(f"- {b}")
            lines.append("")
        
        # Sections
        sections = {}
        for item in self.items:
            if item.section not in sections:
                sections[item.section] = []
            sections[item.section].append(item)
        
        for section, items in sections.items():
            completed = sum(1 for i in items if i.completed)
            total = len(items)
            section_status = "✅" if completed == total else "⚠️" if completed > 0 else "❌"
            
            lines.append(f"## {section_status} {section} ({completed}/{total})")
            lines.append("")
            
            for item in items:
                status = "✅" if item.completed else ("⚡" if not item.required else "❌")
                value_str = f" — {item.value}" if item.value else ""
                notes_str = f" *(Note: {item.notes})*" if item.notes else ""
                lines.append(f"- {status} {item.description}{value_str}{notes_str}")
            
            lines.append("")
        
        return "\n".join(lines)
    
    def save(self, path: str = None) -> str:
        """Saves the report to disk."""
        from pathlib import Path
        from datetime import datetime
        
        if not path:
            ts = datetime.now().strftime("%Y%m%d_%H%M%S")
            path = f"checklists/{self.prompt_name}_{self.version}_{ts}.md"
        
        Path(path).parent.mkdir(parents=True, exist_ok=True)
        with open(path, "w") as f:
            f.write(self.generate_report())
        
        return path

Deployment Strategies

Strategy Comparison

Recommended Strategy: Feature Flag + Canary

from enum import Enum
import hashlib

class DeployStrategy(Enum):
    BIG_BANG = "big_bang"
    CANARY = "canary"
    FEATURE_FLAG = "feature_flag"


class GradualDeployer:
    """
    Implements gradual deployment with feature flags and canary releases.
    """
    
    def __init__(self, registry, alert_manager):
        self.registry = registry
        self.alerts = alert_manager
        self._canary_configs: dict = {}
    
    def start_canary(
        self,
        prompt_name: str,
        new_version: str,
        initial_percentage: float = 0.05,
        target_metrics: dict = None
    ) -> dict:
        """
        Starts a canary deployment.
        
        Traffic splits: initial_percentage → new version
                        the rest → current version (stable)
        """
        stable_version = self.registry.active_version(prompt_name)
        
        self._canary_configs[prompt_name] = {
            "new_version": new_version,
            "stable_version": stable_version,
            "percentage": initial_percentage,
            "target_metrics": target_metrics or {"error_rate": 0.01, "latency_p95": 3000},
            "start": time.time(),
            "status": "ACTIVE"
        }
        
        print(f"🚀 Canary started: {prompt_name}")
        print(f"   New version: {new_version} ({initial_percentage:.0%} traffic)")
        print(f"   Stable version: {stable_version} ({1-initial_percentage:.0%} traffic)")
        
        return self._canary_configs[prompt_name]
    
    def get_version(self, prompt_name: str, request_id: str) -> str:
        """
        Determines which version to use for this request.
        
        Uses hashing of the request_id for consistency
        (the same request always lands in the same bucket).
        """
        if prompt_name not in self._canary_configs:
            return self.registry.active_version(prompt_name)
        
        config = self._canary_configs[prompt_name]
        
        if config["status"] != "ACTIVE":
            return self.registry.active_version(prompt_name)
        
        # Deterministic hash for consistency
        hash_val = int(hashlib.md5(f"{prompt_name}:{request_id}".encode()).hexdigest(), 16)
        use_new = (hash_val % 100) < (config["percentage"] * 100)
        
        return config["new_version"] if use_new else config["stable_version"]
    
    def promote_canary(
        self,
        prompt_name: str,
        new_percentage: float
    ) -> None:
        """Increases the percentage of traffic going to the canary."""
        if prompt_name not in self._canary_configs:
            raise ValueError(f"There is no active canary for '{prompt_name}'")
        
        config = self._canary_configs[prompt_name]
        previous = config["percentage"]
        config["percentage"] = new_percentage
        
        print(f"📈 Canary promoted: {prompt_name} {previous:.0%} → {new_percentage:.0%}")
    
    def complete_canary(self, prompt_name: str) -> None:
        """Finishes the canary: 100% traffic to the new version."""
        if prompt_name not in self._canary_configs:
            raise ValueError(f"There is no active canary for '{prompt_name}'")
        
        new_version = self._canary_configs[prompt_name]["new_version"]
        
        # Activate the new version in the registry
        self.registry.activate(prompt_name, new_version)
        
        # Clear the canary config
        del self._canary_configs[prompt_name]
        
        print(f"✅ Canary completed: {prompt_name} is now {new_version}")
    
    def cancel_canary(self, prompt_name: str, reason: str = "manual") -> None:
        """
        Cancels the canary (automatic rollback).
        100% of traffic goes back to the stable version.
        """
        if prompt_name not in self._canary_configs:
            return
        
        config = self._canary_configs[prompt_name]
        stable_version = config["stable_version"]
        
        # Make sure the stable version is active
        self.registry.activate(prompt_name, stable_version)
        config["status"] = "CANCELLED"
        
        print(f"🔄 Canary cancelled: {prompt_name} → {stable_version} (reason: {reason})")
        del self._canary_configs[prompt_name]


# Progressive rollout example:
"""
Day 1, Hour 0:  start_canary(5%)   — monitor
Day 1, Hour 4:  promote_canary(10%)  — if the metrics are OK
Day 2:          promote_canary(25%)  — if the metrics are OK
Day 3:          promote_canary(50%)  — if the metrics are OK
Day 4:          promote_canary(100%) — if the metrics are OK
Day 5:          complete_canary()    — confirm in the registry
"""

The Deploy Process Step by Step

import asyncio
from openai import OpenAI

client = OpenAI()

async def full_deploy_process(
    prompt_name: str,
    new_version: str,
    new_prompt: str,
    golden_set: list[dict],
    owner: str,
    use_canary: bool = True
) -> dict:
    """
    The complete deploy process with checklist, evaluation and gradual deployment.
    
    Returns: dict with the result of the process.
    """
    print(f"\n{'='*60}")
    print(f"🚀 STARTING THE DEPLOY PROCESS")
    print(f"   {prompt_name} → {new_version}")
    print(f"   Owner: {owner}")
    print(f"{'='*60}\n")
    
    # 1. INITIALIZE THE CHECKLIST
    checklist = ProductionChecklist(prompt_name, new_version, owner)
    
    # 2. EVALUATION (automatic)
    print("📊 Step 1: Quality evaluation...")
    
    correct = 0
    for ex in golden_set:
        r = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": new_prompt.format(input=ex["input"])}],
            temperature=0
        )
        if r.choices[0].message.content.strip().lower() == str(ex["expected_output"]).lower():
            correct += 1
    
    accuracy = correct / len(golden_set)
    print(f"   Accuracy: {accuracy:.2%}")
    
    if accuracy >= 0.85:
        checklist.complete("eval_1", f"100 examples in the golden set")
        checklist.complete("eval_2", f"Accuracy: {accuracy:.2%}")
        checklist.complete("eval_3", "Regression tests: PASS")
    else:
        print(f"❌ BLOCKED: Accuracy {accuracy:.2%} < 85%")
        return {"success": False, "reason": "insufficient_accuracy", "accuracy": accuracy}
    
    # 3. CHECK WHETHER WE CAN DEPLOY
    can_deploy, blockers = checklist.can_deploy()
    
    if not can_deploy:
        print(f"❌ BLOCKED by {len(blockers)} items:")
        for b in blockers:
            print(f"   - {b}")
        
        report = checklist.save()
        return {"success": False, "reason": "incomplete_checklist", "blockers": blockers}
    
    # 4. DEPLOY
    print("\n🚢 Step 2: Deploy...")
    
    if use_canary:
        # Register it in the registry
        # registry.register(prompt_name, new_version, new_prompt)
        
        # Start the canary at 5%
        # deployer.start_canary(prompt_name, new_version, 0.05)
        
        print(f"   Canary started at 5% → monitor for 24h")
        print(f"   Command to promote: deployer.promote_canary('{prompt_name}', 0.25)")
        print(f"   Command to cancel: deployer.cancel_canary('{prompt_name}')")
    else:
        # Big bang deploy
        # registry.activate(prompt_name, new_version)
        print(f"   Direct deploy: {prompt_name} → {new_version}")
    
    # 5. POST-DEPLOY
    print("\n📋 Post-deploy tasks:")
    print("   [ ] Monitor metrics for the first 2 hours")
    print("   [ ] Check that the alerts are live")
    print("   [ ] Document it in the changelog")
    
    # 6. SAVE THE REPORT
    report_path = checklist.save()
    print(f"\n📄 Report saved: {report_path}")
    
    return {
        "success": True,
        "accuracy": accuracy,
        "strategy": "canary" if use_canary else "big_bang",
        "report": report_path
    }

Emergency Rollback

def emergency_rollback(
    prompt_name: str,
    registry,
    alert_manager,
    reason: str = "manual"
) -> None:
    """
    Executes an emergency rollback.
    
    Designed to be fast and simple — in an emergency there's no time for complexity.
    """
    print(f"\n🚨 EXECUTING EMERGENCY ROLLBACK")
    print(f"   Prompt: {prompt_name}")
    print(f"   Reason: {reason}")
    print(f"   Started: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    
    # Step 1: Get the current version
    current_version = registry.active_version(prompt_name)
    print(f"   Current version: {current_version}")
    
    # Step 2: Rollback
    try:
        registry.rollback(prompt_name)
        new_version = registry.active_version(prompt_name)
        print(f"\n✅ Rollback successful: {current_version} → {new_version}")
        print(f"   Time: {(time.time() - time.time()):.1f}s")  # Almost instant
    except Exception as e:
        print(f"❌ Rollback error: {e}")
        print("   MANUAL ACTION REQUIRED")
        return
    
    # Step 3: Check that it's working
    print("\n🔍 Checking health post-rollback...")
    
    test_response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": "test"}],
        temperature=0
    )
    
    if test_response:
        print("   ✅ System responding correctly")
    
    # Step 4: Notify
    print("\n📢 Notifications sent to the team")
    
    # Post-rollback checklist
    print("\n📋 Post-rollback tasks:")
    print("   [ ] Investigate the cause of the rollback")
    print("   [ ] Document the incident")
    print("   [ ] Analyze the logs from the affected window")
    print("   [ ] Fix the problem before the next deploy attempt")

Post-Deploy: The First 24 Hours

class PostDeployMonitor:
    """
    Intensive monitoring during the first 24 hours post-deploy.
    
    More frequent than normal monitoring, to catch problems early.
    """
    
    def __init__(
        self,
        metrics_collector,
        alert_manager,
        baseline_metrics: dict,
        verification_interval: int = 300  # 5 minutes
    ):
        self.metrics = metrics_collector
        self.alerts = alert_manager
        self.baseline = baseline_metrics
        self.interval = verification_interval
        self._start = time.time()
        self._monitoring_hours = 24
    
    def verify(self, prompt_name: str) -> dict:
        """Post-deploy check."""
        current_metrics = self.metrics.current_metrics(prompt_name)
        
        issues = []
        
        # Compare against the baseline
        for metric in ["error_rate", "latency_p95"]:
            current_value = current_metrics.get(metric, 0)
            baseline_value = self.baseline.get(metric, 0)
            
            if baseline_value == 0:
                continue
            
            degradation = (current_value - baseline_value) / baseline_value
            
            if degradation > 0.20:  # 20% degradation
                issues.append({
                    "metric": metric,
                    "baseline": baseline_value,
                    "current": current_value,
                    "degradation": f"{degradation:.0%}"
                })
        
        elapsed_hours = (time.time() - self._start) / 3600
        
        return {
            "hours_post_deploy": elapsed_hours,
            "status": "ISSUES_DETECTED" if issues else "OK",
            "issues": issues,
            "metrics": current_metrics,
            "recommendation": (
                "Consider a rollback if the problems persist"
                if issues else
                "System stable"
            )
        }
    
    def report_24h(self, prompt_name: str) -> str:
        """Final report for the first 24 hours."""
        status = self.verify(prompt_name)
        
        return f"""
## Post-Deploy 24h Report: {prompt_name}

**Status:** {status['status']}

### Final Metrics vs Baseline
| Metric | Baseline | Current | Delta |
|---------|---------|--------|-------|
| Error rate | {self.baseline.get('error_rate', 0):.2%} | {status['metrics'].get('error_rate', 0):.2%} | ... |
| Latency p95 | {self.baseline.get('latency_p95', 0):.0f}ms | {status['metrics'].get('latency_p95', 0):.0f}ms | ... |

### Conclusion
{'✅ Successful deploy — promote to 100%' if status['status'] == 'OK' else '⚠️ Issues detected — review before promoting'}
"""

Troubleshooting

Problem 1: The deploy breaks production unexpectedly

Symptom: Shortly after the deploy, the error rate jumps to 20%.

Immediate action:

# 1. IMMEDIATE ROLLBACK (don't debug in production)
registry.rollback("classifier")  # 1 line

# 2. CHECK that the rollback succeeded
print(f"Active version now: {registry.active_version('classifier')}")

# 3. INVESTIGATE in staging (not in production)
# Review the logs from the problem window
# Identify which change caused the problem

Golden rule: If something fails in production, roll back first. Debug afterwards.

Problem 2: The checklist gets skipped "because we're in a hurry"

Symptom: A fast deploy that ends in an incident.

Structural solution:

# Automate the checklist so it's faster than skipping it
def deploy_gate(prompt_name: str, version: str) -> bool:
    """
    Automatic gate that blocks the deploy if the checklist isn't complete.
    Runs in CI/CD.
    """
    # Check the metrics from the last test
    with db_conn() as conn:
        test = conn.execute(
            "SELECT accuracy FROM test_results WHERE prompt_name = ? AND prompt_version = ?",
            (prompt_name, version)
        ).fetchone()
    
    if not test:
        print(f"❌ DEPLOY BLOCKED: There are no tests for {prompt_name} {version}")
        return False
    
    if test["accuracy"] < 0.85:
        print(f"❌ DEPLOY BLOCKED: Accuracy {test['accuracy']:.2%} < 85%")
        return False
    
    print(f"✅ Deploy gate passed for {prompt_name} {version}")
    return True

# In the CI/CD pipeline:
# if not deploy_gate(PROMPT_NAME, VERSION):
#     sys.exit(1)  # Blocks the pipeline

Exercises

Exercise 1: Complete the checklist for a real deploy

Take a prompt you have in production or in development and work through the checklist:

# Initialize the checklist
checklist = ProductionChecklist(
    prompt_name="my_classifier",
    version="v1.2.0",
    owner="your_name"
)

# Complete the items you already have
checklist.complete("eval_1", "150 examples in golden_set.json")
checklist.complete("eval_2", "Accuracy: 91.3%")
checklist.complete("sec_1", "API key in .env, .env in .gitignore")

# See what's missing
summary = checklist.summary()
print(f"Pending items: {summary['pending_items']}")

# Generate the report
report = checklist.generate_report()
print(report)

Exercise 2: Implement a simple canary deployment

Implement a router that sends 10% of traffic to a new version based on the user_id:

import hashlib
from openai import OpenAI

client = OpenAI()

PROMPT_V1 = "Classify as POSITIVE, NEGATIVE or NEUTRAL: {input}. Category only."
PROMPT_V2 = """Classify the sentiment. Consider sarcasm and irony.
Respond ONLY with: POSITIVE, NEGATIVE, or NEUTRAL

Text: {input}
Category:"""

def canary_router(user_id: str, pct_new: float = 0.10) -> tuple[str, str]:
    """Returns (prompt, version) for a given user_id."""
    hash_val = int(hashlib.md5(f"classifier:{user_id}".encode()).hexdigest(), 16)
    use_new = (hash_val % 100) < (pct_new * 100)
    
    if use_new:
        return PROMPT_V2, "v2"
    return PROMPT_V1, "v1"

# Simulate routing
distribution = {"v1": 0, "v2": 0}
for user_num in range(200):
    user_id = f"user_{user_num:04d}"
    _, version = canary_router(user_id, pct_new=0.10)
    distribution[version] += 1

print(f"v1: {distribution['v1']} users ({distribution['v1']/200:.0%})")
print(f"v2: {distribution['v2']} users ({distribution['v2']/200:.0%})")
# v1: ~180 users (90%), v2: ~20 users (10%) ✓

Summary

• Checklist: 6 sections — Evaluation, Performance, Reliability, Versioning, Monitoring, Security
• Deploy gate: Automate the checklist as part of the CI/CD pipeline
• Big bang: Simple but risky — only for urgent hotfixes
• Canary: 5% → 25% → 50% → 100% — with monitoring windows at each stage
• Feature flags: Flexible, they let you cancel without a code rollback
• Rollback: It has to be automatic and < 5 minutes — if it's more complex than that, something is wrong
• Post-deploy: Intensive monitoring for the first 24 hours — catch problems before they escalate

Additional resources

1. Google SRE Book — Reliability engineering best practices
2. Feature Flags Guide — LaunchDarkly feature flags
3. Martin Fowler: Canary Release — The canary pattern
4. OpenAI Production Best Practices — Official guide
5. Deployment Checklist Template — Reference template