Module 8: Prompt Engineering in Production

2. Prompt Versioning and Registries

Overview

Managing prompt versions like production code: semver, Git-based versioning, prompt registries with rollback. Changelog, audit trail and feature flags for gradual deployment. Open source and SaaS tooling.


The Problem with Unversioned Prompts

Without versioning, prompts live in a chaotic state:

Common situations without versioning:
─────────────────────────────────
"The prompt was in a text file on my Desktop"
"I changed something last week but I don't remember what"
"I need to go back to how it was yesterday, impossible"
"There are 3 versions of the prompt in 3 different files: which one is current?"
"I don't know if customer A uses the same prompt as customer B"

Prompt versioning has to be as rigorous as code versioning.


Semantic Versioning for Prompts

Semantic versioning (semver) is the standard: MAJOR.MINOR.PATCH

For prompts, we adapt the convention:

ComponentWhen to bumpExample
MAJORA change that affects the output format, or the prompt's main taskv1.0 → v2.0
MINORQuality improvement, new instructions, more few-shot examplesv1.0 → v1.1
PATCHTypo fix, minor clarification with no impact on the outputv1.0 → v1.0.1
Concrete examples:
v1.0   → v2.0: Going from "classify into 3 categories" to "classify into 5 categories"
v1.0   → v1.1: Adding 2 few-shot examples to improve edge cases
v1.1   → v1.1.1: Fixing "Classifiy" → "Classify" (typo)

Versioning with Git

The simplest solution: treat prompts as text files in Git.

File Structure

prompts/
├── classifier/
│   ├── v1.0.0.txt
│   ├── v1.1.0.txt
│   ├── v2.0.0.txt
│   └── CHANGELOG.md
├── extractor/
│   ├── v1.0.0.txt
│   └── CHANGELOG.md
├── summary/
│   └── v1.0.0.txt
└── registry.json

CHANGELOG.md per Prompt

# Changelog: classifier

## v2.0.0 (2025-03-01)
### Breaking Changes
- Output now includes 5 categories instead of 3: POSITIVE, NEGATIVE, NEUTRAL, MIXED, URGENT

### Added
- New URGENT category for texts with emergency language
- Explicit instruction for handling sarcasm

## v1.1.0 (2025-02-15)
### Added
- 3 extra few-shot examples for sarcasm edge cases
- Instruction: "Consider the overall tone, not just individual words"

## v1.0.0 (2025-01-10)
### Initial Release
- Basic classifier: POSITIVE, NEGATIVE, NEUTRAL

Git Operations

# See the history of a prompt
git log --oneline prompts/classifier/

# See what changed between versions
git diff HEAD~1 prompts/classifier/v1.1.0.txt

# Restore a previous version
git show HEAD~2:prompts/classifier/v1.0.0.txt > prompts/classifier/restored.txt

Prompt Registry: Python Implementation

A registry is a centralized system for accessing prompts by name and version:

import json
from pathlib import Path
from datetime import datetime
from typing import Optional


class PromptRegistry:
    """
    Prompt registry with versioning and rollback.
    
    Stores prompts in memory with optional persistence to disk.
    """
    
    def __init__(self, registry_path: str = "prompts/registry.json"):
        self.registry_path = Path(registry_path)
        self._data: dict = {}
        
        if self.registry_path.exists():
            self._load()
    
    def _load(self) -> None:
        """Loads the registry from disk."""
        with open(self.registry_path) as f:
            self._data = json.load(f)
        print(f"Registry loaded: {len(self._data)} prompts")
    
    def _save(self) -> None:
        """Persists the registry to disk."""
        self.registry_path.parent.mkdir(parents=True, exist_ok=True)
        with open(self.registry_path, "w", encoding="utf-8") as f:
            json.dump(self._data, f, indent=2, ensure_ascii=False)
    
    def register(
        self,
        name: str,
        version: str,
        prompt: str,
        changelog: str = "",
        metadata: dict | None = None
    ) -> None:
        """
        Registers a new version of a prompt.
        
        name: Prompt identifier (e.g. "sentiment_classifier")
        version: Semver (e.g. "v1.2.0")
        prompt: The prompt text (may include {variables})
        changelog: Description of the changes in this version
        """
        if name not in self._data:
            self._data[name] = {
                "versions": {},
                "active_version": None,
                "created_at": datetime.now().isoformat()
            }
        
        self._data[name]["versions"][version] = {
            "prompt": prompt,
            "changelog": changelog,
            "created_at": datetime.now().isoformat(),
            "metadata": metadata or {}
        }
        
        # Auto-activate if it's the first version
        if self._data[name]["active_version"] is None:
            self._data[name]["active_version"] = version
        
        self._save()
        print(f"Registered: {name} {version}")
    
    def get(
        self,
        name: str,
        version: Optional[str] = None
    ) -> str:
        """
        Gets a prompt by name and version.
        
        If version is None, returns the active version.
        """
        if name not in self._data:
            raise KeyError(f"Prompt '{name}' not found in the registry")
        
        if version is None:
            version = self._data[name]["active_version"]
        
        if version is None:
            raise ValueError(f"There is no active version for '{name}'")
        
        versions = self._data[name]["versions"]
        if version not in versions:
            available = list(versions.keys())
            raise KeyError(f"Version '{version}' does not exist. Available: {available}")
        
        return versions[version]["prompt"]
    
    def activate(self, name: str, version: str) -> None:
        """Activates a specific version (no rollback, just switching the active one)."""
        if name not in self._data:
            raise KeyError(f"Prompt '{name}' not found")
        
        if version not in self._data[name]["versions"]:
            raise KeyError(f"Version '{version}' does not exist")
        
        previous = self._data[name]["active_version"]
        self._data[name]["active_version"] = version
        self._save()
        
        print(f"Activated: {name}{version} (previous: {previous})")
    
    def rollback(self, name: str, to_version: Optional[str] = None) -> None:
        """
        Rollback to a previous version.
        
        If to_version is None, it goes to the version before the active one.
        """
        if name not in self._data:
            raise KeyError(f"Prompt '{name}' not found")
        
        versions = list(self._data[name]["versions"].keys())
        current = self._data[name]["active_version"]
        
        if to_version is None:
            # Step back one version
            if current in versions:
                current_idx = versions.index(current)
                if current_idx > 0:
                    to_version = versions[current_idx - 1]
                else:
                    raise ValueError(f"You're already on the first version ({current})")
            else:
                to_version = versions[-1]
        
        self.activate(name, to_version)
        print(f"🔄 Rollback: {name} {current}{to_version}")
    
    def list(self, name: str) -> dict:
        """Lists every version of a prompt with metadata."""
        if name not in self._data:
            raise KeyError(f"Prompt '{name}' not found")
        
        prompt_data = self._data[name]
        active = prompt_data["active_version"]
        
        return {
            "name": name,
            "active_version": active,
            "versions": {
                v: {
                    "active": v == active,
                    "changelog": data.get("changelog", ""),
                    "created_at": data.get("created_at", ""),
                    "preview": data["prompt"][:80] + "..."
                }
                for v, data in prompt_data["versions"].items()
            }
        }
    
    def active_version(self, name: str) -> str:
        """Returns the active version of a prompt."""
        if name not in self._data:
            raise KeyError(f"Prompt '{name}' not found")
        return self._data[name]["active_version"]
    
    def available_prompts(self) -> list[str]:
        """Lists every prompt in the registry."""
        return list(self._data.keys())


# Usage example:
registry = PromptRegistry()

# Register the first version
registry.register(
    name="sentiment_classifier",
    version="v1.0.0",
    prompt="Classify the text as POSITIVE, NEGATIVE or NEUTRAL. Category only.\n\nText: {input}\nCategory:",
    changelog="Initial version of the classifier"
)

# Register an improvement
registry.register(
    name="sentiment_classifier",
    version="v1.1.0",
    prompt="""Classify the sentiment of the text as POSITIVE, NEGATIVE or NEUTRAL.
Consider the overall tone, including sarcasm and irony.
Respond ONLY with the category.

Examples:
- "I love this product" → POSITIVE
- "Sure, 'fast' if waiting 3 weeks counts as fast" → NEGATIVE

Text: {input}
Category:""",
    changelog="Added 2 few-shot examples to improve sarcasm detection"
)

# Activate the new version
registry.activate("sentiment_classifier", "v1.1.0")

# Get the active prompt
prompt = registry.get("sentiment_classifier")

# If something goes wrong: rollback
registry.rollback("sentiment_classifier")  # Back to v1.0.0

Registry with Feature Flags

For gradual deployment, combine versioning with feature flags:

import random
from openai import OpenAI

client = OpenAI()

class PromptRegistryWithFlags:
    """Registry that supports A/B deployments and gradual rollouts."""
    
    def __init__(self, registry: PromptRegistry):
        self.registry = registry
        self._flags: dict = {}
    
    def configure_rollout(
        self,
        name: str,
        new_version: str,
        percentage: float = 0.05
    ) -> None:
        """
        Configures a gradual rollout.
        percentage: 0.05 = 5% of traffic goes to the new version
        """
        self._flags[name] = {
            "new_version": new_version,
            "percentage": percentage,
            "stable_version": self.registry.active_version(name)
        }
        print(f"Rollout configured: {name}{new_version} ({percentage:.0%} traffic)")
    
    def get_prompt(self, name: str, request_id: str = None) -> tuple[str, str]:
        """
        Gets the prompt with routing based on feature flags.
        
        Returns: (prompt, version_used)
        """
        if name in self._flags:
            flag = self._flags[name]
            
            # Consistency: the same request_id always uses the same version
            if request_id:
                import hashlib
                hash_val = int(hashlib.md5(f"{name}:{request_id}".encode()).hexdigest(), 16)
                use_new = (hash_val % 100) < (flag["percentage"] * 100)
            else:
                use_new = random.random() < flag["percentage"]
            
            version = flag["new_version"] if use_new else flag["stable_version"]
        else:
            version = self.registry.active_version(name)
        
        return self.registry.get(name, version), version
    
    def promote_rollout(self, name: str, new_percentage: float) -> None:
        """Increases the percentage of traffic going to the new version."""
        if name not in self._flags:
            raise ValueError(f"There is no active rollout for '{name}'")
        
        self._flags[name]["percentage"] = new_percentage
        print(f"Rollout promoted: {name}{new_percentage:.0%}")
    
    def complete_rollout(self, name: str) -> None:
        """Finishes the rollout: 100% traffic to the new version and clears the flag."""
        if name in self._flags:
            new_version = self._flags[name]["new_version"]
            self.registry.activate(name, new_version)
            del self._flags[name]
            print(f"Rollout completed: {name} is now {new_version}")
    
    def cancel_rollout(self, name: str) -> None:
        """Cancels the rollout: back to 100% on the stable version."""
        if name in self._flags:
            del self._flags[name]
            print(f"Rollout cancelled: {name} went back to the stable version")


# Gradual rollout example:
reg = PromptRegistry()
flags = PromptRegistryWithFlags(reg)

# Day 1: 5% to the new prompt
flags.configure_rollout("classifier", "v1.2.0", percentage=0.05)

# Monitor metrics... If they look good:
# Day 2: bump to 20%
flags.promote_rollout("classifier", new_percentage=0.20)

# Day 3: bump to 50%
flags.promote_rollout("classifier", new_percentage=0.50)

# If something fails at any point:
flags.cancel_rollout("classifier")  # 100% goes back to the stable version

# If all is well: complete
flags.complete_rollout("classifier")  # v1.2.0 becomes the active one

Registry with a Database

For systems with multiple services or teams, use a database:

import sqlite3
from contextlib import contextmanager


class PromptRegistryDB:
    """Registry with SQLite persistence (scales to PostgreSQL easily)."""
    
    def __init__(self, db_path: str = "prompts.db"):
        self.db_path = db_path
        self._init_db()
    
    def _init_db(self) -> None:
        """Creates the tables if they don't exist."""
        with self._conn() as conn:
            conn.execute("""
                CREATE TABLE IF NOT EXISTS prompts (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    name TEXT NOT NULL,
                    version TEXT NOT NULL,
                    prompt TEXT NOT NULL,
                    changelog TEXT DEFAULT '',
                    active INTEGER DEFAULT 0,
                    created_at TEXT NOT NULL,
                    UNIQUE(name, version)
                )
            """)
            
            conn.execute("""
                CREATE TABLE IF NOT EXISTS prompt_usage (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    name TEXT NOT NULL,
                    version TEXT NOT NULL,
                    timestamp TEXT NOT NULL,
                    request_id TEXT,
                    success INTEGER DEFAULT 1
                )
            """)
    
    @contextmanager
    def _conn(self):
        conn = sqlite3.connect(self.db_path)
        conn.row_factory = sqlite3.Row
        try:
            yield conn
            conn.commit()
        except Exception:
            conn.rollback()
            raise
        finally:
            conn.close()
    
    def register(self, name: str, version: str, prompt: str, changelog: str = "") -> None:
        with self._conn() as conn:
            conn.execute("""
                INSERT INTO prompts (name, version, prompt, changelog, created_at)
                VALUES (?, ?, ?, ?, ?)
            """, (name, version, prompt, changelog, datetime.now().isoformat()))
        print(f"Registered: {name} {version}")
    
    def activate(self, name: str, version: str) -> None:
        with self._conn() as conn:
            # Deactivate every current version
            conn.execute("UPDATE prompts SET active = 0 WHERE name = ?", (name,))
            # Activate the requested version
            conn.execute(
                "UPDATE prompts SET active = 1 WHERE name = ? AND version = ?",
                (name, version)
            )
        print(f"Activated: {name} {version}")
    
    def get(self, name: str, version: str | None = None) -> str:
        with self._conn() as conn:
            if version:
                row = conn.execute(
                    "SELECT prompt FROM prompts WHERE name = ? AND version = ?",
                    (name, version)
                ).fetchone()
            else:
                row = conn.execute(
                    "SELECT prompt FROM prompts WHERE name = ? AND active = 1",
                    (name,)
                ).fetchone()
        
        if not row:
            raise KeyError(f"Prompt '{name}' {f'v{version}' if version else '(active)'} not found")
        
        return row["prompt"]
    
    def log_usage(self, name: str, version: str, request_id: str, success: bool) -> None:
        """Records prompt usage for auditing."""
        with self._conn() as conn:
            conn.execute("""
                INSERT INTO prompt_usage (name, version, timestamp, request_id, success)
                VALUES (?, ?, ?, ?, ?)
            """, (name, version, datetime.now().isoformat(), request_id, int(success)))
    
    def usage_history(self, name: str, last_n: int = 100) -> list[dict]:
        """Returns the usage history of a prompt."""
        with self._conn() as conn:
            rows = conn.execute("""
                SELECT * FROM prompt_usage WHERE name = ?
                ORDER BY timestamp DESC LIMIT ?
            """, (name, last_n)).fetchall()
        
        return [dict(row) for row in rows]

Integrating into an API (FastAPI)

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()
registry = PromptRegistry()
client = OpenAI()

class ClassifyRequest(BaseModel):
    text: str
    prompt_version: str | None = None  # None = use the active version

class ClassifyResponse(BaseModel):
    result: str
    prompt_version: str

@app.post("/classify", response_model=ClassifyResponse)
async def classify(request: ClassifyRequest):
    try:
        # Get the prompt (specific version or active one)
        prompt_template = registry.get(
            "sentiment_classifier",
            version=request.prompt_version
        )
        version_used = (
            request.prompt_version or
            registry.active_version("sentiment_classifier")
        )
    except KeyError as e:
        raise HTTPException(status_code=404, detail=str(e))
    
    # Run the prompt
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": prompt_template.format(input=request.text)
        }],
        temperature=0
    )
    
    result = response.choices[0].message.content.strip()
    
    return ClassifyResponse(result=result, prompt_version=version_used)

@app.post("/prompts/{name}/rollback")
async def rollback_prompt(name: str, to_version: str | None = None):
    """Endpoint for emergency rollback."""
    try:
        registry.rollback(name, to_version)
        return {"status": "ok", "active_version": registry.active_version(name)}
    except (KeyError, ValueError) as e:
        raise HTTPException(status_code=400, detail=str(e))

@app.get("/prompts/{name}/versions")
async def list_versions(name: str):
    """Lists every version of a prompt."""
    try:
        return registry.list(name)
    except KeyError as e:
        raise HTTPException(status_code=404, detail=str(e))

Comparison: Versioning Options

OptionUpsideDownsideWhen to use
Files in GitSimple, free, native diffNo runtime accessSmall projects, technical teams
JSON registry on diskSimple, no dependenciesNo multi-service1-3 services, one team
SQLite registryQueries, audit trailNo clusteringSingle service, auditing needed
PostgreSQL registryMulti-service, transactionalInfra dependencyMulti-service, mid-size team
PromptLayerUI, analytics, managedMonthly costTeams with no infra of their own
LangSmithBuilt-in tracingVendor lock-inYou already use LangChain

Troubleshooting

Problem 1: Prompts hardcoded in the code

Symptom: The prompt sits directly in the Python code, with no versioning.

Cause: Fast development without thinking about maintainability.

Solution:

# BEFORE (problematic):
def classify(text: str) -> str:
    prompt = "Classify this: " + text  # Hardcoded
    ...

# AFTER (with a registry):
registry = PromptRegistry()

def classify(text: str) -> str:
    prompt_template = registry.get("classifier")
    prompt = prompt_template.format(input=text)
    ...

# Migration script:
# 1. Extract hardcoded prompts into .txt files
# 2. Register them in the registry with an initial version
# 3. Update the code to use registry.get()

Problem 2: Rollback without downtime

Symptom: You need to roll back but the service is taking traffic.

Cause: The new version has a serious bug.

Solution:

# The JSON-based registry allows rollback without restarting the service
# if you read it on every request (not cached in memory indefinitely)

# BAD: Load the prompt once at startup
PROMPT = registry.get("classifier")  # Fixed in memory, can't update without a restart

# GOOD: Load on each request (with a short 30s cache for performance)
from functools import lru_cache
from time import time

class RegistryWithCache:
    def __init__(self, registry: PromptRegistry, ttl_seconds: int = 30):
        self._registry = registry
        self._cache = {}
        self._ttl = ttl_seconds
    
    def get(self, name: str) -> str:
        now = time()
        if name in self._cache:
            cached_at, prompt = self._cache[name]
            if now - cached_at < self._ttl:
                return prompt
        
        prompt = self._registry.get(name)
        self._cache[name] = (now, prompt)
        return prompt

Problem 3: Conflicts between teams

Symptom: Two people edit the same prompt concurrently.

Solution:

# With a DB: use optimistic locking
def register_with_lock(
    name: str,
    version: str,
    prompt: str,
    based_on: str  # version the change was based on
):
    with db._conn() as conn:
        current_version = conn.execute(
            "SELECT version FROM prompts WHERE name = ? AND active = 1",
            (name,)
        ).fetchone()
        
        if current_version and current_version["version"] != based_on:
            raise ValueError(
                f"Conflict: the active version changed from {based_on} "
                f"to {current_version['version']} while you were editing. "
                "Please merge the changes."
            )
        
        # If there's no conflict, register normally
        db.register(name, version, prompt)

Exercises

Exercise 1: Build your first prompt registry

Implement a minimal PromptRegistry that supports: register, get, activate, and rollback:

See solution
import json
from pathlib import Path
from datetime import datetime

class MiniRegistry:
    """Minimal implementation of a prompt registry."""
    
    def __init__(self, path: str = "registry.json"):
        self.path = Path(path)
        self.data = {}
        if self.path.exists():
            with open(self.path) as f:
                self.data = json.load(f)
    
    def _save(self):
        with open(self.path, "w") as f:
            json.dump(self.data, f, indent=2, ensure_ascii=False)
    
    def register(self, name: str, version: str, prompt: str) -> None:
        if name not in self.data:
            self.data[name] = {"versions": {}, "active": None}
        self.data[name]["versions"][version] = {
            "prompt": prompt,
            "created": datetime.now().isoformat()
        }
        if self.data[name]["active"] is None:
            self.data[name]["active"] = version
        self._save()
        print(f"✓ Registered: {name} {version}")
    
    def get(self, name: str, version: str = None) -> str:
        if name not in self.data:
            raise KeyError(f"'{name}' not found")
        version = version or self.data[name]["active"]
        return self.data[name]["versions"][version]["prompt"]
    
    def activate(self, name: str, version: str) -> None:
        self.data[name]["active"] = version
        self._save()
        print(f"✓ Activated: {name}{version}")
    
    def rollback(self, name: str) -> None:
        versions = list(self.data[name]["versions"].keys())
        current = self.data[name]["active"]
        idx = versions.index(current) if current in versions else len(versions) - 1
        if idx > 0:
            self.activate(name, versions[idx - 1])
        else:
            print("You're already on the first version")

# Test:
r = MiniRegistry("/tmp/test_registry.json")
r.register("test", "v1.0", "Prompt v1: {input}")
r.register("test", "v1.1", "Improved prompt v1.1: {input}")
r.activate("test", "v1.1")
print(r.get("test"))  # v1.1
r.rollback("test")
print(r.get("test"))  # v1.0

Exercise 2: Gradual rollout with feature flags

Implement a simple system that sends 10% of traffic to a new prompt:

See solution
import hashlib
from openai import OpenAI

client = OpenAI()

PROMPT_V1 = "Classify as POSITIVE, NEGATIVE or NEUTRAL: {input}"
PROMPT_V2 = """Classify the sentiment as POSITIVE, NEGATIVE or NEUTRAL.
Consider the overall tone and sarcasm.
Category only.
Text: {input}
Category:"""

def get_prompt_for_request(request_id: str, pct_new: float = 0.10) -> tuple[str, str]:
    """Returns (prompt, version) based on request_id."""
    hash_val = int(hashlib.md5(request_id.encode()).hexdigest(), 16)
    use_new = (hash_val % 100) < (pct_new * 100)
    
    if use_new:
        return PROMPT_V2, "v2"
    return PROMPT_V1, "v1"

# Simulate 10 requests
results = {"v1": 0, "v2": 0}
for i in range(20):
    request_id = f"req_{i:04d}"
    _, version = get_prompt_for_request(request_id, pct_new=0.10)
    results[version] += 1

print(f"v1: {results['v1']} requests, v2: {results['v2']} requests")
print(f"% v2: {results['v2']/20*100:.0f}% (target: 10%)")

Summary

  • Semver for prompts: MAJOR (breaking), MINOR (improvement), PATCH (typo) — same logic as code
  • Git + files: The simplest option — prompts as versioned .txt files
  • Python registry: A dict with versions, an active version, and rollback in a persisted JSON file
  • Feature flags: Gradual deployment (5% → 20% → 50% → 100%) without downtime
  • DB-backed registry: For multi-service setups or when you need an audit trail
  • API integration: A rollback endpoint for emergencies without restarting the service

Additional resources

  1. Semantic Versioning — Full semver specification
  2. PromptLayer — Managed versioning with analytics
  3. LangSmith — Tracing and prompt management for LangChain
  4. Feature Flags Best Practices — LaunchDarkly
  5. Git Flow — For a branch strategy with prompts