Managing Prompt Changes Safely: Testing, Versioning, and Monitoring in Production

# Version 1: Works great in testing
old_prompt = "Summarize this article: {text}"

# Version 2: Seems like a minor improvement
new_prompt = "Provide a concise summary of this article: {text}"

# Result: Summaries are now 50% longer, breaking your UI
# and increasing costs by 30%

prompts/
├── summarization/
│   ├── v1.txt
│   ├── v2.txt
│   └── v3.txt
├── translation/
│   ├── v1.txt
│   └── v2.txt
└── versions.json

{
  "summarization": {
    "current": "v3",
    "production": "v2",
    "history": [
      {
        "version": "v1",
        "date": "2026-04-01",
        "author": "alice",
        "changes": "Initial version"
      },
      {
        "version": "v2",
        "date": "2026-04-15",
        "author": "bob",
        "changes": "Added tone parameter"
      },
      {
        "version": "v3",
        "date": "2026-05-01",
        "author": "alice",
        "changes": "Improved clarity, reduced token count"
      }
    ]
  }
}

import json
from pathlib import Path
from langchain.prompts import PromptTemplate

class PromptManager:
    def __init__(self, prompts_dir="prompts"):
        self.prompts_dir = Path(prompts_dir)
        self.versions = self._load_versions()

    def _load_versions(self):
        with open(self.prompts_dir / "versions.json") as f:
            return json.load(f)

    def load_prompt(self, name, version=None, use_production=False):
        """Load a specific prompt version"""
        if use_production:
            version = self.versions[name]["production"]
        elif version is None:
            version = self.versions[name]["current"]

        prompt_path = self.prompts_dir / name / f"{version}.txt"

        with open(prompt_path) as f:
            template_text = f.read()

        return PromptTemplate.from_template(template_text)

    def get_version_info(self, name, version):
        """Get metadata about a prompt version"""
        history = self.versions[name]["history"]
        return next(v for v in history if v["version"] == version)

# Usage
manager = PromptManager()

# Development: use latest
dev_prompt = manager.load_prompt("summarization")

# Production: use stable version
prod_prompt = manager.load_prompt("summarization", use_production=True)

# Specific version for testing
test_prompt = manager.load_prompt("summarization", version="v2")

# Create a new prompt version
git checkout -b prompt/summarization-v4

# Edit the prompt
vim prompts/summarization/v4.txt

# Update versions.json
vim prompts/versions.json

# Commit with descriptive message
git commit -m "feat(prompts): Add summarization v4 - reduce token usage by 20%"

# Create PR for review
git push origin prompt/summarization-v4

import pytest
from langchain_openai import ChatOpenAI
from prompt_manager import PromptManager

@pytest.fixture
def llm():
    return ChatOpenAI(model="gpt-4", temperature=0)

@pytest.fixture
def manager():
    return PromptManager()

class TestSummarizationPrompt:
    def test_prompt_loads(self, manager):
        """Test that prompt loads without errors"""
        prompt = manager.load_prompt("summarization", version="v3")
        assert prompt is not None
        assert "text" in prompt.input_variables

    def test_prompt_format(self, manager):
        """Test that prompt formats correctly"""
        prompt = manager.load_prompt("summarization", version="v3")
        formatted = prompt.format(text="Sample article text")
        assert "Sample article text" in formatted

    def test_output_length(self, manager, llm):
        """Test that summaries are within expected length"""
        prompt = manager.load_prompt("summarization", version="v3")

        test_text = "This is a long article. " * 100
        formatted = prompt.format(text=test_text)
        response = llm.invoke(formatted)

        # Summary should be significantly shorter than input
        assert len(response.content) < len(test_text) * 0.3

    def test_output_format(self, manager, llm):
        """Test that output follows expected format"""
        prompt = manager.load_prompt("summarization", version="v3")

        formatted = prompt.format(text="AI is transforming industries.")
        response = llm.invoke(formatted)

        # Should be a single paragraph, no bullet points
        assert response.content.count('\n\n') <= 1
        assert '•' not in response.content
        assert '-' not in response.content[:10]  # No leading dash

class TestPromptRegression:
    """Ensure new versions don't break existing behavior"""

    @pytest.fixture
    def test_cases(self):
        """Known good input/output pairs"""
        return [
            {
                "input": "The quick brown fox jumps over the lazy dog.",
                "expected_keywords": ["fox", "dog", "jump"],
                "max_length": 50
            },
            {
                "input": "Artificial intelligence is revolutionizing healthcare...",
                "expected_keywords": ["AI", "healthcare"],
                "max_length": 100
            }
        ]

    def test_v3_maintains_quality(self, manager, llm, test_cases):
        """Test that v3 produces acceptable outputs"""
        prompt = manager.load_prompt("summarization", version="v3")

        for case in test_cases:
            formatted = prompt.format(text=case["input"])
            response = llm.invoke(formatted)

            # Check length constraint
            assert len(response.content) <= case["max_length"]

            # Check that key concepts are preserved
            content_lower = response.content.lower()
            keywords_found = sum(
                1 for kw in case["expected_keywords"]
                if kw.lower() in content_lower
            )
            assert keywords_found >= len(case["expected_keywords"]) * 0.5

def test_version_comparison(manager, llm):
    """Compare outputs between versions"""
    v2_prompt = manager.load_prompt("summarization", version="v2")
    v3_prompt = manager.load_prompt("summarization", version="v3")

    test_text = "Long article about AI..." * 50

    v2_response = llm.invoke(v2_prompt.format(text=test_text))
    v3_response = llm.invoke(v3_prompt.format(text=test_text))

    # v3 should be more concise
    assert len(v3_response.content) <= len(v2_response.content)

    # But should still cover main points (check similarity)
    # Use a simple word overlap metric
    v2_words = set(v2_response.content.lower().split())
    v3_words = set(v3_response.content.lower().split())
    overlap = len(v2_words & v3_words) / len(v2_words)

    assert overlap >= 0.5  # At least 50% word overlap

import os
from typing import Optional

class PromptSelector:
    def __init__(self, manager: PromptManager):
        self.manager = manager
        self.ab_test_enabled = os.getenv("AB_TEST_ENABLED", "false") == "true"
        self.ab_test_ratio = float(os.getenv("AB_TEST_RATIO", "0.1"))

    def get_prompt(self, name: str, user_id: Optional[str] = None):
        """Get prompt version based on A/B test configuration"""
        if not self.ab_test_enabled:
            return self.manager.load_prompt(name, use_production=True)

        # Use user_id for consistent assignment
        if user_id and self._should_use_test_version(user_id):
            return self.manager.load_prompt(name, version="v3")
        else:
            return self.manager.load_prompt(name, use_production=True)

    def _should_use_test_version(self, user_id: str) -> bool:
        """Deterministic assignment based on user_id"""
        hash_value = hash(user_id) % 100
        return hash_value < (self.ab_test_ratio * 100)

# Usage
selector = PromptSelector(manager)

def summarize_article(text: str, user_id: str):
    prompt = selector.get_prompt("summarization", user_id=user_id)
    formatted = prompt.format(text=text)
    return llm.invoke(formatted)

import time
from dataclasses import dataclass
from typing import Dict, List

@dataclass
class PromptMetrics:
    version: str
    latency_ms: float
    token_count: int
    success: bool
    user_feedback: Optional[float] = None

class MetricsCollector:
    def __init__(self):
        self.metrics: List[PromptMetrics] = []

    def record(self, version: str, latency_ms: float, 
               token_count: int, success: bool):
        self.metrics.append(PromptMetrics(
            version=version,
            latency_ms=latency_ms,
            token_count=token_count,
            success=success
        ))

    def get_summary(self, version: str) -> Dict:
        version_metrics = [m for m in self.metrics if m.version == version]

        if not version_metrics:
            return {}

        return {
            "count": len(version_metrics),
            "avg_latency": sum(m.latency_ms for m in version_metrics) / len(version_metrics),
            "avg_tokens": sum(m.token_count for m in version_metrics) / len(version_metrics),
            "success_rate": sum(m.success for m in version_metrics) / len(version_metrics)
        }

# Usage
collector = MetricsCollector()

def summarize_with_metrics(text: str, user_id: str):
    prompt = selector.get_prompt("summarization", user_id=user_id)
    version = "v3" if selector._should_use_test_version(user_id) else "v2"

    start = time.time()
    try:
        formatted = prompt.format(text=text)
        response = llm.invoke(formatted)
        latency = (time.time() - start) * 1000

        collector.record(
            version=version,
            latency_ms=latency,
            token_count=len(response.content.split()),
            success=True
        )
        return response.content
    except Exception as e:
        latency = (time.time() - start) * 1000
        collector.record(
            version=version,
            latency_ms=latency,
            token_count=0,
            success=False
        )
        raise

# Compare versions
print("v2 metrics:", collector.get_summary("v2"))
print("v3 metrics:", collector.get_summary("v3"))

class CanaryDeployment:
    def __init__(self, manager: PromptManager):
        self.manager = manager
        self.canary_percentage = 0  # Start at 0%

    def set_canary_percentage(self, percentage: int):
        """Gradually increase canary traffic"""
        assert 0 <= percentage <= 100
        self.canary_percentage = percentage
        print(f"Canary set to {percentage}%")

    def get_prompt(self, name: str, request_id: str):
        """Route requests based on canary percentage"""
        hash_value = hash(request_id) % 100

        if hash_value < self.canary_percentage:
            # Use new version
            return self.manager.load_prompt(name, version="v3"), "v3"
        else:
            # Use production version
            return self.manager.load_prompt(name, use_production=True), "v2"

# Rollout plan
canary = CanaryDeployment(manager)

# Day 1: 5% traffic
canary.set_canary_percentage(5)

# Day 2: If metrics look good, 25%
canary.set_canary_percentage(25)

# Day 3: 50%
canary.set_canary_percentage(50)

# Day 4: 100% - full rollout
canary.set_canary_percentage(100)

from dataclasses import dataclass
from datetime import datetime
from typing import List

@dataclass
class PromptAlert:
    timestamp: datetime
    version: str
    metric: str
    threshold: float
    actual: float
    message: str

class PromptMonitor:
    def __init__(self, collector: MetricsCollector):
        self.collector = collector
        self.alerts: List[PromptAlert] = []

    def check_health(self, version: str):
        """Check if prompt version is performing within acceptable bounds"""
        summary = self.collector.get_summary(version)

        if not summary:
            return

        # Check success rate
        if summary["success_rate"] < 0.95:
            self.alerts.append(PromptAlert(
                timestamp=datetime.now(),
                version=version,
                metric="success_rate",
                threshold=0.95,
                actual=summary["success_rate"],
                message=f"Success rate dropped to {summary['success_rate']:.2%}"
            ))

        # Check latency
        if summary["avg_latency"] > 2000:  # 2 seconds
            self.alerts.append(PromptAlert(
                timestamp=datetime.now(),
                version=version,
                metric="latency",
                threshold=2000,
                actual=summary["avg_latency"],
                message=f"Latency increased to {summary['avg_latency']:.0f}ms"
            ))

        # Check token usage (cost)
        if summary["avg_tokens"] > 500:
            self.alerts.append(PromptAlert(
                timestamp=datetime.now(),
                version=version,
                metric="tokens",
                threshold=500,
                actual=summary["avg_tokens"],
                message=f"Token usage increased to {summary['avg_tokens']:.0f}"
            ))

    def get_alerts(self) -> List[PromptAlert]:
        return self.alerts

# Usage
monitor = PromptMonitor(collector)
monitor.check_health("v3")

for alert in monitor.get_alerts():
    print(f"ALERT [{alert.version}]: {alert.message}")

class PromptRollback:
    def __init__(self, manager: PromptManager):
        self.manager = manager
        self.rollback_history = []

    def rollback(self, name: str, reason: str):
        """Immediately rollback to previous production version"""
        current = self.manager.versions[name]["production"]
        history = self.manager.versions[name]["history"]

        # Find previous version
        current_idx = next(
            i for i, v in enumerate(history)
            if v["version"] == current
        )

        if current_idx == 0:
            raise ValueError("No previous version to rollback to")

        previous = history[current_idx - 1]["version"]

        # Update production version
        self.manager.versions[name]["production"] = previous

        # Save versions.json
        with open(self.manager.prompts_dir / "versions.json", "w") as f:
            json.dump(self.manager.versions, f, indent=2)

        # Record rollback
        self.rollback_history.append({
            "timestamp": datetime.now().isoformat(),
            "prompt": name,
            "from": current,
            "to": previous,
            "reason": reason
        })

        print(f"Rolled back {name} from {current} to {previous}")
        print(f"Reason: {reason}")

# Usage
rollback = PromptRollback(manager)

# If something goes wrong
rollback.rollback(
    "summarization",
    reason="v3 causing 30% increase in latency"
)