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:
| Component | When to bump | Example |
|---|---|---|
| MAJOR | A change that affects the output format, or the prompt's main task | v1.0 → v2.0 |
| MINOR | Quality improvement, new instructions, more few-shot examples | v1.0 → v1.1 |
| PATCH | Typo fix, minor clarification with no impact on the output | v1.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
| Option | Upside | Downside | When to use |
|---|---|---|---|
| Files in Git | Simple, free, native diff | No runtime access | Small projects, technical teams |
| JSON registry on disk | Simple, no dependencies | No multi-service | 1-3 services, one team |
| SQLite registry | Queries, audit trail | No clustering | Single service, auditing needed |
| PostgreSQL registry | Multi-service, transactional | Infra dependency | Multi-service, mid-size team |
| PromptLayer | UI, analytics, managed | Monthly cost | Teams with no infra of their own |
| LangSmith | Built-in tracing | Vendor lock-in | You 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
- Semantic Versioning — Full semver specification
- PromptLayer — Managed versioning with analytics
- LangSmith — Tracing and prompt management for LangChain
- Feature Flags Best Practices — LaunchDarkly
- Git Flow — For a branch strategy with prompts