Module 8: Prompt Engineering in Production
5. Prompt Management Systems
Overview
Designing and building a system to manage prompts as a team: templates with variables, built-in versioning, automated testing, access control, and an audit trail. A comparison between building your own system (build) and using existing tools (buy).
Why You Need a Prompt Management System
Without a management system, prompts scatter and create problems:
Without a Prompt Management System:
──────────────────────────────
• The prompt is hardcoded in 5 different files
• The team doesn't know which version is in production
• Changing a prompt requires a full deploy
• There's no way to test a change before production
• Two people edit the same prompt concurrently
• Nobody knows who changed what or when
With a Prompt Management System:
──────────────────────────────
• Centralized prompts with versioning
• The active version is visible and switchable without a deploy
• Instant rollback if something fails
• Built-in testing before activating a new version
• An audit trail of every change
• Templating with reusable variables
System Components
A complete prompt management system has:
1. TEMPLATE ENGINE → Jinja2 for variables and logic
2. REGISTRY → Versioned prompt storage
3. VARIABLE STORE → Shared variable values
4. TESTING LAYER → Regression tests before activating
5. ACCESS CONTROL → Who can read/write/activate
6. AUDIT TRAIL → A log of every change
7. API → Interface for services
Implementation: Template Engine with Jinja2
from jinja2 import Environment, FileSystemLoader, Template, TemplateError
from pathlib import Path
import json
class PromptTemplateEngine:
"""
Template engine for prompts, powered by Jinja2.
Supports: variables, conditionals, loops, filters, includes.
"""
def __init__(self, templates_dir: str = "prompts/templates"):
self.templates_dir = Path(templates_dir)
self.templates_dir.mkdir(parents=True, exist_ok=True)
# Jinja2 environment with templates from a directory
self.env = Environment(
loader=FileSystemLoader(str(self.templates_dir)),
trim_blocks=True, # Drops the newline after blocks
lstrip_blocks=True # Drops the whitespace before blocks
)
# Register custom filters
self.env.filters["truncate"] = lambda s, n=200: s[:n] + "..." if len(s) > n else s
self.env.filters["uppercase"] = lambda s: s.upper()
self.env.filters["list_to_text"] = lambda items: ", ".join(items)
def render_string(self, template_str: str, variables: dict) -> str:
"""Renders a template string with variables."""
try:
template = Template(template_str)
return template.render(**variables)
except TemplateError as e:
raise ValueError(f"Template error: {e}")
def render_file(self, filename: str, variables: dict) -> str:
"""Renders a template from a file."""
try:
template = self.env.get_template(filename)
return template.render(**variables)
except TemplateError as e:
raise ValueError(f"Error in template {filename}: {e}")
def validate_variables(self, template_str: str, available_variables: dict) -> dict:
"""
Validates that every required variable is available.
Returns: {"valid": bool, "missing_variables": list, "extra_variables": list}
"""
from jinja2 import meta
env_temp = Environment()
ast = env_temp.parse(template_str)
required_variables = meta.find_undeclared_variables(ast)
missing = required_variables - set(available_variables.keys())
extras = set(available_variables.keys()) - required_variables
return {
"valid": len(missing) == 0,
"required_variables": list(required_variables),
"missing_variables": list(missing),
"extra_variables": list(extras),
"available_variables": list(available_variables.keys())
}
# Examples of complex templates:
ADVANCED_CLASSIFIER_TEMPLATE = """
You are a {{ domain }} classifier.
{% if special_instructions %}
SPECIAL INSTRUCTIONS:
{{ special_instructions }}
{% endif %}
Classify the following text into one of these categories:
{% for cat in categories %}
- {{ cat }}
{% endfor %}
{% if examples %}
EXAMPLES:
{% for ex in examples %}
- "{{ ex.input }}" → {{ ex.output }}
{% endfor %}
{% endif %}
IMPORTANT: Respond ONLY with the category. No explanations.
Text: {{ input }}
Category:
"""
# Render:
engine = PromptTemplateEngine()
prompt = engine.render_string(
ADVANCED_CLASSIFIER_TEMPLATE,
variables={
"domain": "review sentiment",
"categories": ["POSITIVE", "NEGATIVE", "NEUTRAL", "MIXED"],
"special_instructions": "Pay special attention to sarcasm",
"examples": [
{"input": "I love the product", "output": "POSITIVE"},
{"input": "Sure, 'fast' with a 3-week wait", "output": "NEGATIVE"}
],
"input": "Shipping took forever but the product is good"
}
)
print(prompt)
Complete System with Every Layer
import json
import sqlite3
from datetime import datetime
from pathlib import Path
from contextlib import contextmanager
from typing import Optional
from openai import OpenAI
client = OpenAI()
class PromptManagementSystem:
"""
A complete prompt management system.
Combines: template engine + registry + testing + audit trail + access control.
"""
def __init__(self, db_path: str = "prompt_management.db"):
self.db_path = db_path
self.template_engine = PromptTemplateEngine()
self._init_db()
def _init_db(self) -> None:
"""Initializes the database with every table."""
with self._conn() as conn:
# Main prompts table
conn.execute("""
CREATE TABLE IF NOT EXISTS prompts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
version TEXT NOT NULL,
template TEXT NOT NULL,
variables_schema TEXT DEFAULT '{}',
changelog TEXT DEFAULT '',
active INTEGER DEFAULT 0,
created_by TEXT DEFAULT 'system',
created_at TEXT NOT NULL,
UNIQUE(name, version)
)
""")
# Variable stores table (shared values)
conn.execute("""
CREATE TABLE IF NOT EXISTS variable_store (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
key TEXT NOT NULL,
value TEXT NOT NULL,
value_type TEXT DEFAULT 'string',
updated_at TEXT NOT NULL,
UNIQUE(name, key)
)
""")
# Audit trail
conn.execute("""
CREATE TABLE IF NOT EXISTS audit_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
prompt_name TEXT NOT NULL,
action TEXT NOT NULL,
previous_version TEXT,
new_version TEXT,
user TEXT NOT NULL,
timestamp TEXT NOT NULL,
details TEXT DEFAULT ''
)
""")
# Test results
conn.execute("""
CREATE TABLE IF NOT EXISTS test_results (
id INTEGER PRIMARY KEY AUTOINCREMENT,
prompt_name TEXT NOT NULL,
prompt_version TEXT NOT NULL,
accuracy REAL,
passed INTEGER,
failed INTEGER,
timestamp TEXT NOT NULL
)
""")
@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,
template: str,
variables_schema: dict | None = None,
changelog: str = "",
user: str = "system"
) -> None:
"""
Registers a new version of a prompt.
variables_schema: Dict defining the required variables.
{"input": {"value_type": "string", "required": True, "description": "..."},
"categories": {"value_type": "list", "required": True}}
"""
schema_json = json.dumps(variables_schema or {})
with self._conn() as conn:
conn.execute("""
INSERT INTO prompts (name, version, template, variables_schema, changelog, created_at, created_by)
VALUES (?, ?, ?, ?, ?, ?, ?)
""", (name, version, template, schema_json, changelog, datetime.now().isoformat(), user))
# If it's the first one, activate it automatically
count = conn.execute(
"SELECT COUNT(*) as n FROM prompts WHERE name = ? AND active = 1", (name,)
).fetchone()["n"]
if count == 0:
conn.execute(
"UPDATE prompts SET active = 1 WHERE name = ? AND version = ?",
(name, version)
)
# Audit
self._log(name, "REGISTERED", new_version=version, user=user, details=changelog)
print(f"✓ Registered: {name} {version}")
def activate(
self,
name: str,
version: str,
user: str = "system",
verify_tests: bool = True
) -> None:
"""
Activates a version of a prompt.
If verify_tests=True, it checks that the version passed the tests
before activating (protection against activating untested versions).
"""
if verify_tests:
with self._conn() as conn:
test_result = conn.execute("""
SELECT accuracy FROM test_results
WHERE prompt_name = ? AND prompt_version = ?
ORDER BY timestamp DESC LIMIT 1
""", (name, version)).fetchone()
if not test_result:
raise ValueError(
f"There are no test results for {name} {version}. "
"Run the tests before activating."
)
if test_result["accuracy"] < 0.80:
raise ValueError(
f"Version {version} did not pass the tests "
f"(accuracy: {test_result['accuracy']:.2%}). "
"Minimum required: 80%"
)
with self._conn() as conn:
# Get the current version for the audit trail
current = conn.execute(
"SELECT version FROM prompts WHERE name = ? AND active = 1",
(name,)
).fetchone()
previous_version = current["version"] if current else None
# Deactivate all, activate the new one
conn.execute("UPDATE prompts SET active = 0 WHERE name = ?", (name,))
conn.execute(
"UPDATE prompts SET active = 1 WHERE name = ? AND version = ?",
(name, version)
)
self._log(
name, "ACTIVATED",
previous_version=previous_version,
new_version=version,
user=user
)
print(f"✓ Activated: {name} {version} (previous: {previous_version})")
def rollback(
self,
name: str,
to_version: Optional[str] = None,
user: str = "system"
) -> None:
"""
Rollback to a previous version.
If to_version is None, it steps back one version.
"""
with self._conn() as conn:
versions = [
row["version"] for row in conn.execute(
"SELECT version FROM prompts WHERE name = ? ORDER BY created_at ASC",
(name,)
).fetchall()
]
current = conn.execute(
"SELECT version FROM prompts WHERE name = ? AND active = 1",
(name,)
).fetchone()
if not current:
raise ValueError(f"There is no active version for '{name}'")
current_version = current["version"]
if to_version is None:
idx = versions.index(current_version) if current_version in versions else len(versions)
if idx == 0:
raise ValueError("You're already on the first version")
to_version = versions[idx - 1]
# Activate without checking tests (this is an emergency rollback)
self.activate(name, to_version, user=user, verify_tests=False)
print(f"🔄 ROLLBACK: {name} {current_version} → {to_version}")
def render(
self,
name: str,
variables: dict,
version: Optional[str] = None
) -> str:
"""
Gets and renders the prompt with variables.
If version is None, uses the active one.
"""
with self._conn() as conn:
if version:
row = conn.execute(
"SELECT template, variables_schema FROM prompts WHERE name = ? AND version = ?",
(name, version)
).fetchone()
else:
row = conn.execute(
"SELECT template, variables_schema 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")
template = row["template"]
schema = json.loads(row["variables_schema"])
# Validate variables if there's a schema
if schema:
validation = self.template_engine.validate_variables(template, variables)
if not validation["valid"]:
raise ValueError(
f"Missing variables: {validation['missing_variables']}. "
f"Available variables: {validation['available_variables']}"
)
return self.template_engine.render_string(template, variables)
def run(
self,
name: str,
variables: dict,
model: str = "gpt-4o-mini",
version: Optional[str] = None
) -> dict:
"""Renders the prompt, runs it and returns the result."""
prompt = self.render(name, variables, version)
response = client.chat.completions.create(
model=model,
messages=[{"role": "user", "content": prompt}],
temperature=0
)
return {
"output": response.choices[0].message.content,
"prompt_used": name,
"version_used": version or self.active_version(name),
"tokens": response.usage.total_tokens
}
def active_version(self, name: str) -> Optional[str]:
"""Returns the active version of a prompt."""
with self._conn() as conn:
row = conn.execute(
"SELECT version FROM prompts WHERE name = ? AND active = 1",
(name,)
).fetchone()
return row["version"] if row else None
def save_test_result(
self,
name: str,
version: str,
accuracy: float,
passed: int,
failed: int
) -> None:
"""Saves a test result for the activation gate."""
with self._conn() as conn:
conn.execute("""
INSERT INTO test_results (prompt_name, prompt_version, accuracy, passed, failed, timestamp)
VALUES (?, ?, ?, ?, ?, ?)
""", (name, version, accuracy, passed, failed, datetime.now().isoformat()))
def _log(
self,
name: str,
action: str,
new_version: Optional[str] = None,
previous_version: Optional[str] = None,
user: str = "system",
details: str = ""
) -> None:
"""Records an entry in the audit trail."""
with self._conn() as conn:
conn.execute("""
INSERT INTO audit_log (prompt_name, action, previous_version, new_version, user, timestamp, details)
VALUES (?, ?, ?, ?, ?, ?, ?)
""", (name, action, previous_version, new_version, user, datetime.now().isoformat(), details))
def audit_trail(self, name: str, limit: int = 20) -> list[dict]:
"""Returns the change history of a prompt."""
with self._conn() as conn:
rows = conn.execute("""
SELECT * FROM audit_log WHERE prompt_name = ?
ORDER BY timestamp DESC LIMIT ?
""", (name, limit)).fetchall()
return [dict(row) for row in rows]
def set_variable(self, namespace: str, key: str, value, value_type: str = "string") -> None:
"""Stores a reusable variable in the variable store."""
with self._conn() as conn:
conn.execute("""
INSERT OR REPLACE INTO variable_store (name, key, value, value_type, updated_at)
VALUES (?, ?, ?, ?, ?)
""", (namespace, key, json.dumps(value), value_type, datetime.now().isoformat()))
def get_variable(self, namespace: str, key: str):
"""Gets a variable from the variable store."""
with self._conn() as conn:
row = conn.execute(
"SELECT value, value_type FROM variable_store WHERE name = ? AND key = ?",
(namespace, key)
).fetchone()
if not row:
raise KeyError(f"Variable '{namespace}.{key}' not found")
return json.loads(row["value"])
def get_variables_namespace(self, namespace: str) -> dict:
"""Gets every variable in a namespace."""
with self._conn() as conn:
rows = conn.execute(
"SELECT key, value, value_type FROM variable_store WHERE name = ?",
(namespace,)
).fetchall()
return {row["key"]: json.loads(row["value"]) for row in rows}
Variable Store: Shared Variables
# Full example of using the variable store:
pms = PromptManagementSystem()
# Store shared variables (configurable without touching the template)
pms.set_variable("classifier", "categories", ["POSITIVE", "NEGATIVE", "NEUTRAL", "MIXED"])
pms.set_variable("classifier", "examples", [
{"input": "Excellent product", "output": "POSITIVE"},
{"input": "Terrible service", "output": "NEGATIVE"},
])
pms.set_variable("classifier", "min_score", 0.8)
# Register a template that uses variables from the store
TEMPLATE = """
You are a review sentiment classifier.
Classify into: {{ categories | list_to_text }}
{% if examples %}
Examples:
{% for ex in examples %}
- "{{ ex.input }}" → {{ ex.output }}
{% endfor %}
{% endif %}
Review: {{ input }}
Category:
"""
pms.register(
name="reviews_classifier",
version="v1.0.0",
template=TEMPLATE,
variables_schema={
"input": {"value_type": "string", "required": True},
"categories": {"value_type": "list", "required": True},
"examples": {"value_type": "list", "required": False}
},
changelog="Initial version",
user="miguel"
)
# Run it combining variables from the store and from the request
store_variables = pms.get_variables_namespace("classifier")
result = pms.run(
name="reviews_classifier",
variables={
**store_variables, # categories, examples from the store
"input": "Shipping was slow but the product is excellent" # from the request
}
)
print(result["output"]) # POSITIVE
Testing Gate: Tests Before Activating
def test_and_activate(
pms: PromptManagementSystem,
name: str,
version: str,
golden_set: list[dict],
min_accuracy: float = 0.85,
user: str = "system"
) -> bool:
"""
Runs the tests and activates if they pass.
This pattern guarantees you never activate a prompt without green tests.
"""
print(f"🧪 Testing {name} {version}...")
# Render and test
correct = 0
common_variables = pms.get_variables_namespace(name.split("_")[0])
for ex in golden_set:
try:
output = pms.run(
name,
variables={**common_variables, "input": ex["input"]},
version=version
)["output"]
if output.strip().lower() == str(ex["expected_output"]).strip().lower():
correct += 1
except Exception as e:
print(f" Error on example {ex.get('id', '?')}: {e}")
n = len(golden_set)
accuracy = correct / n if n > 0 else 0.0
failed = n - correct
# Save the test result
pms.save_test_result(name, version, accuracy, correct, failed)
print(f" Accuracy: {accuracy:.2%} ({correct}/{n}) — Minimum: {min_accuracy:.2%}")
if accuracy >= min_accuracy:
# Activate if the tests pass
pms.activate(name, version, user=user)
print(f" ✅ Tests passed — {name} {version} activated")
return True
else:
print(f" ❌ Tests failed — {name} {version} NOT activated")
return False
Build vs Buy: When to Build vs Use Tools
| Criterion | Build | Buy (PromptLayer, LangSmith) |
|---|---|---|
| Upfront cost | High (days/weeks of dev) | Low (hours to integrate) |
| Recurring cost | Low (infra only) | $20-200/month |
| Control | Total (custom logic) | Bounded by the tool |
| Features | Only what you build | Analytics, UI, team, etc. |
| Maintenance | You | Them |
| Vendor lock-in | None | High |
| Development speed | Slow at first | Fast |
Recommendation:
- Startups and new projects: Start with Buy (PromptLayer or LangSmith)
- Teams with custom requirements: Build when the tools don't cover your needs
- Companies with compliance: Build for full control of data and audit trail
Integrating with PromptLayer (Managed Alternative)
import promptlayer
from openai import OpenAI
# PromptLayer intercepts the OpenAI calls and logs them automatically
promptlayer.api_key = "pl_xxx"
openai = promptlayer.openai
def use_prompt_from_promptlayer(
prompt_name: str,
variables: dict
) -> str:
"""
PromptLayer has its own templating and versioning system.
Returns the prompt rendered from their platform.
"""
# Get the template from PromptLayer
prompt_template = promptlayer.templates.get(prompt_name)
# Format it with variables
prompt = prompt_template["prompt"]["template"].format(**variables)
# Run it with automatic tracking
response = openai.ChatCompletion.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
temperature=0,
pl_tags=["production", prompt_name],
return_pl_id=True
)
return response.choices[0].message.content
Troubleshooting
Problem 1: Undefined variables in the template
Symptom: jinja2.UndefinedError: 'variable_name' is undefined
Cause: The template gets rendered without passing every required variable.
Solution:
# Validate before rendering
def safe_render(pms: PromptManagementSystem, name: str, variables: dict) -> str:
"""Renders with prior variable validation."""
# Get the prompt's schema
with pms._conn() as conn:
row = conn.execute(
"SELECT variables_schema FROM prompts WHERE name = ? AND active = 1",
(name,)
).fetchone()
if row:
schema = json.loads(row["variables_schema"])
required = {k for k, v in schema.items() if v.get("required", False)}
missing = required - set(variables.keys())
if missing:
raise ValueError(
f"Missing required variables for '{name}': {missing}\n"
f"Available variables: {set(variables.keys())}"
)
return pms.render(name, variables)
Problem 2: Name collisions between teams
Symptom: Team A has "classifier" and team B does too — they clobber each other.
Solution:
# Namespace by team/domain in the prompt name
# Format: {team}/{domain}/{name}
pms.register("team_a/support/urgency_classifier", "v1.0", template_a)
pms.register("team_b/sales/intent_classifier", "v1.0", template_b)
# They never clobber each other:
pms.render("team_a/support/urgency_classifier", {...})
pms.render("team_b/sales/intent_classifier", {...})
Problem 3: Testing is slow for large golden sets
Symptom: The test gate takes 10+ minutes.
Solution:
import asyncio
from openai import AsyncOpenAI
async_client = AsyncOpenAI()
async def test_prompt_async(
pms: PromptManagementSystem,
name: str,
version: str,
golden_set: list[dict],
max_concurrent: int = 10
) -> float:
"""Async version of the test gate for large golden sets."""
semaphore = asyncio.Semaphore(max_concurrent)
async def evaluate(ex):
async with semaphore:
template = pms.render(name, {"input": ex["input"]}, version)
response = await async_client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": template}],
temperature=0
)
output = response.choices[0].message.content.strip()
return output.lower() == str(ex["expected_output"]).lower()
results = await asyncio.gather(*[evaluate(ex) for ex in golden_set])
return sum(results) / len(results)
Exercises
Exercise 1: Create a template with conditional variables
Create a template for a summarization prompt that includes optional context:
See solution
from jinja2 import Template
SUMMARY_TEMPLATE = """
{% if audience %}
Write for an audience of {{ audience }}.
{% endif %}
{% if max_words %}
Summary in at most {{ max_words }} words.
{% else %}
Summary in at most 3 sentences.
{% endif %}
{% if style %}
Style: {{ style }}.
{% endif %}
Text to summarize:
{{ text }}
Summary:
"""
engine = PromptTemplateEngine()
# With every parameter
full_prompt = engine.render_string(SUMMARY_TEMPLATE, {
"text": "The AI market grew 30% this year...",
"audience": "non-technical executives",
"max_words": 50,
"style": "formal"
})
# With only the basics
basic_prompt = engine.render_string(SUMMARY_TEMPLATE, {
"text": "The AI market grew 30% this year..."
})
print("With parameters:")
print(full_prompt)
print("\nText only:")
print(basic_prompt)
Exercise 2: Implement a simple testing gate
Create a function that runs the tests and blocks activation if accuracy < 85%:
See solution
from openai import OpenAI
client = OpenAI()
def test_gate(
prompt_template: str,
golden_set: list[dict],
min_accuracy: float = 0.85
) -> dict:
"""
Testing gate: returns True if the prompt passes the tests.
Block the deploy if it returns False.
"""
correct = 0
failures = []
for ex in golden_set:
prompt = prompt_template.format(input=ex["input"])
r = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
temperature=0
)
output = r.choices[0].message.content.strip()
if output.lower() == str(ex["expected_output"]).lower():
correct += 1
else:
failures.append({
"id": ex.get("id", "?"),
"input": ex["input"][:50],
"expected": ex["expected_output"],
"actual": output
})
n = len(golden_set)
accuracy = correct / n
result = {
"passed": accuracy >= min_accuracy,
"accuracy": accuracy,
"correct": correct,
"failed": n - correct,
"failures": failures[:5],
"message": (
f"✅ PASS ({accuracy:.2%} >= {min_accuracy:.2%})"
if accuracy >= min_accuracy
else f"❌ FAIL ({accuracy:.2%} < {min_accuracy:.2%})"
)
}
print(result["message"])
if failures:
print(f"First {min(3, len(failures))} failures:")
for f in failures[:3]:
print(f" [{f['id']}] '{f['input']}' → expected: {f['expected']}, got: {f['actual']}")
return result
# Usage:
GOLDEN = [
{"id": "1", "input": "Excellent product", "expected_output": "POSITIVE"},
{"id": "2", "input": "Terrible", "expected_output": "NEGATIVE"},
]
result = test_gate(
"Classify as POSITIVE, NEGATIVE or NEUTRAL: {input}. Category only.",
GOLDEN,
min_accuracy=0.85
)
if result["passed"]:
print("✅ Deploy authorized")
else:
print("❌ Deploy blocked — fix the prompt")
Summary
- Template engine: Jinja2 for variables, conditionals and loops in prompts — more flexible than f-strings
- Registry: Versioned storage with explicit activation and rollback
- Variable store: Shared variables you can update without touching the template
- Testing gate: Automatic tests before activating — never activate without green tests
- Audit trail: Who changed what and when — indispensable for debugging in production
- Build vs buy: Buy (PromptLayer, LangSmith) to move fast; Build for total control
- Namespace: Use prefixes to avoid collisions between teams
Additional resources
- PromptLayer — Managed prompt management with a UI
- LangSmith — Observability + management for LangChain
- Jinja2 Documentation — Template engine
- Langfuse — Open source LLM observability
- Helicone — LLM proxy with analytics