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

CriterionBuildBuy (PromptLayer, LangSmith)
Upfront costHigh (days/weeks of dev)Low (hours to integrate)
Recurring costLow (infra only)$20-200/month
ControlTotal (custom logic)Bounded by the tool
FeaturesOnly what you buildAnalytics, UI, team, etc.
MaintenanceYouThem
Vendor lock-inNoneHigh
Development speedSlow at firstFast

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

  1. PromptLayer — Managed prompt management with a UI
  2. LangSmith — Observability + management for LangChain
  3. Jinja2 Documentation — Template engine
  4. Langfuse — Open source LLM observability
  5. Helicone — LLM proxy with analytics