Production Prompt Engineering¶
Prerequisites¶
Before this lesson you should be comfortable with:
- The full prompt engineering toolkit — Lessons 1–9
- Prompt templates and versioning — Lesson 5
- Eval loops and A/B testing — Lesson 7
- Guardrails and fallback strategies — Lesson 8
This is the capstone lesson. It integrates everything into an end-to-end production workflow.
What You'll Learn¶
| Objective | Time | Difficulty |
|---|---|---|
| Design a production prompt lifecycle from development to monitoring | 10 min | Advanced |
| Implement prompt versioning, logging, and cost tracking | 10 min | Advanced |
| Run A/B tests on prompt changes in production traffic | 10 min | Advanced |
| Build a continuous improvement pipeline with eval loops and failure recovery | 10 min | Advanced |
Intuition First: Prompts Are Code¶
In production, a prompt is not a string you typed once in a notebook. It is a versioned artifact with the same lifecycle as any other critical code: development, testing, staging, deployment, monitoring, and iteration.
Teams that treat prompts as afterthoughts — hardcoded in application logic, changed without testing, deployed without logging — spend more time firefighting than building. Teams that treat prompts as first-class infrastructure ship faster, debug easier, and improve continuously.
This lesson gives you the end-to-end workflow.
The Production Prompt Lifecycle¶
Development → Evaluation → Staging → Production → Monitoring → Iteration
↑ │
└──────────────────────────────────────────────────────────────┘
Each stage has a gate:
- Development — write and version the prompt
- Evaluation — run against eval set; block if accuracy below threshold
- Staging — deploy to staging environment; run integration tests
- Production — canary deploy (10% traffic); monitor for 24–48 hours
- Monitoring — track accuracy, latency, cost, error rate, fallback rate
- Iteration — analyze failures, update eval set, refine prompt, repeat
Prompt Versioning in Production¶
import json
from pathlib import Path
from datetime import datetime, timezone
class PromptManager:
def __init__(self, prompts_dir: str = "./prompts"):
self.prompts_dir = Path(prompts_dir)
self.prompts_dir.mkdir(exist_ok=True)
def save(self, name: str, template: str, metadata: dict | None = None) -> int:
"""Save a new prompt version. Returns version number."""
versions_dir = self.prompts_dir / name
versions_dir.mkdir(exist_ok=True)
existing = sorted(versions_dir.glob("v*.json"))
version = len(existing) + 1
prompt_data = {
"template": template,
"version": version,
"created_at": datetime.now(timezone.utc).isoformat(),
"metadata": metadata or {},
}
filepath = versions_dir / f"v{version}.json"
filepath.write_text(json.dumps(prompt_data, indent=2))
return version
def load(self, name: str, version: int | None = None) -> dict:
"""Load a prompt by name and version (latest if version is None)."""
versions_dir = self.prompts_dir / name
if version:
filepath = versions_dir / f"v{version}.json"
else:
files = sorted(versions_dir.glob("v*.json"))
if not files:
raise FileNotFoundError(f"No versions found for prompt '{name}'")
filepath = files[-1]
return json.loads(filepath.read_text())
def list_versions(self, name: str) -> list[dict]:
versions_dir = self.prompts_dir / name
return [
json.loads(f.read_text())
for f in sorted(versions_dir.glob("v*.json"))
]
# Usage
pm = PromptManager()
pm.save("sentiment", "Classify sentiment: {text}", {"eval_accuracy": 0.72})
v2 = pm.save(
"sentiment",
"Classify as positive/negative/neutral:\n\n{text}",
{"eval_accuracy": 0.89, "author": "team", "change": "Added explicit labels"},
)
current = pm.load("sentiment") # latest (v2)
stable = pm.load("sentiment", version=1) # rollback target
Never overwrite a prompt in place. Every change gets a new version with metadata (eval score, author, change description).
Logging and Monitoring¶
Log every LLM call with enough context to debug and optimize:
import time
import json
import logging
logger = logging.getLogger("prompt_monitor")
class MonitoredLLMCall:
def __init__(self, client):
self.client = client
def call(
self,
prompt_name: str,
prompt_version: int,
messages: list[dict],
**kwargs,
) -> dict:
start = time.time()
model = kwargs.get("model", "gpt-4o-mini")
try:
response = self.client.chat.completions.create(
messages=messages, **kwargs,
)
latency = time.time() - start
output = response.choices[0].message.content
usage = response.usage
log_entry = {
"prompt_name": prompt_name,
"prompt_version": prompt_version,
"model": model,
"latency_ms": int(latency * 1000),
"input_tokens": usage.prompt_tokens,
"output_tokens": usage.completion_tokens,
"cost_usd": self._estimate_cost(usage, model),
"status": "success",
}
logger.info(json.dumps(log_entry))
return {
"output": output,
"latency": latency,
"usage": usage,
"log": log_entry,
}
except Exception as e:
logger.error(json.dumps({
"prompt_name": prompt_name,
"prompt_version": prompt_version,
"model": model,
"status": "error",
"error": str(e),
"latency_ms": int((time.time() - start) * 1000),
}))
raise
def _estimate_cost(self, usage, model: str) -> float:
rates = {
"gpt-4o": {"input": 2.50 / 1_000_000, "output": 10.00 / 1_000_000},
"gpt-4o-mini": {"input": 0.15 / 1_000_000, "output": 0.60 / 1_000_000},
}
r = rates.get(model, {"input": 0, "output": 0})
return (
usage.prompt_tokens * r["input"]
+ usage.completion_tokens * r["output"]
)
Key metrics to dashboard:
- Accuracy — automated eval score (daily)
- Latency — p50, p95, p99 per prompt
- Cost — USD per day per prompt, per model
- Error rate — API failures, timeouts, parse failures
- Fallback rate — how often degraded paths activate
Production A/B Testing¶
import random
import hashlib
class PromptABTest:
def __init__(
self,
name: str,
prompt_a: str,
prompt_b: str,
version_a: int,
version_b: int,
traffic_split: float = 0.1,
):
self.name = name
self.prompt_a = prompt_a
self.prompt_b = prompt_b
self.version_a = version_a
self.version_b = version_b
self.traffic_split = traffic_split
self.results: dict[str, list[float]] = {"A": [], "B": []}
def assign(self, user_id: str) -> tuple[str, str, int]:
"""Deterministic assignment based on user_id hash."""
hash_val = int(hashlib.md5(user_id.encode()).hexdigest(), 16)
if (hash_val % 100) < (self.traffic_split * 100):
return self.prompt_b, "B", self.version_b
return self.prompt_a, "A", self.version_a
def record(self, variant: str, score: float):
self.results[variant].append(score)
def stats(self) -> dict:
def avg(scores):
return sum(scores) / len(scores) if scores else 0.0
return {
"A": {"count": len(self.results["A"]), "avg_score": avg(self.results["A"])},
"B": {"count": len(self.results["B"]), "avg_score": avg(self.results["B"])},
}
def should_promote_b(self, min_samples: int = 100, min_lift: float = 0.02) -> bool:
s = self.stats()
if s["B"]["count"] < min_samples:
return False
return s["B"]["avg_score"] - s["A"]["avg_score"] >= min_lift
# Usage
ab = PromptABTest(
name="sentiment_v2_test",
prompt_a=pm.load("sentiment", version=1)["template"],
prompt_b=pm.load("sentiment", version=2)["template"],
version_a=1,
version_b=2,
traffic_split=0.1,
)
prompt, variant, version = ab.assign(user_id="user_12345")
# Render prompt, call LLM, record score
# ab.record(variant, score=1.0 if correct else 0.0)
# if ab.should_promote_b(): deploy v2 to 100%
Use deterministic assignment (hash of user ID) so the same user always gets the same variant during a test.
Continuous Improvement Pipeline¶
class PromptImprovementPipeline:
"""End-to-end workflow tying together versioning, eval, and deployment."""
def __init__(self, prompt_manager: PromptManager, eval_set: list[dict]):
self.pm = prompt_manager
self.eval_set = eval_set
self.accuracy_threshold = 0.85
def develop(self, name: str, template: str) -> int:
version = self.pm.save(name, template, {"status": "draft"})
return version
def evaluate(self, name: str, version: int, eval_fn) -> dict:
data = self.pm.load(name, version)
results = eval_fn(data["template"], self.eval_set)
data["metadata"]["eval_accuracy"] = results["accuracy"]
data["metadata"]["status"] = (
"approved" if results["accuracy"] >= self.accuracy_threshold else "rejected"
)
return results
def deploy(self, name: str, version: int, traffic_pct: float = 0.1):
data = self.pm.load(name, version)
if data["metadata"].get("status") != "approved":
raise ValueError(f"Prompt v{version} not approved for deployment")
data["metadata"]["deployed_at"] = datetime.now(timezone.utc).isoformat()
data["metadata"]["traffic_pct"] = traffic_pct
return data
def rollback(self, name: str, to_version: int):
"""Instant rollback to a known-good version."""
return self.pm.load(name, to_version)
| Step | Frequency | Action |
|---|---|---|
| Monitor accuracy | Daily | Check automated eval scores |
| Review failures | Weekly | Analyze low-scoring responses |
| Update eval set | Bi-weekly | Add new edge cases from production |
| Test new models | Monthly | Benchmark against newer models |
| Optimize cost | Monthly | Check if cheaper model achieves same quality |
| Update prompts | As needed | Iterate based on failure analysis |
Failure Recovery in Production¶
Production prompt systems need explicit recovery paths:
- Prompt render failure → fall back to previous template version
- LLM API error → retry with exponential backoff (max 2 retries)
- Structured output parse failure → retry with same schema, then fall back to JSON mode
- Output guardrail failure → retry with stricter format instruction
- All paths fail → return safe default, log for human review, alert if rate exceeds threshold
- Accuracy drop detected → automatic rollback to last known-good prompt version
def production_call(
prompt_manager: PromptManager,
name: str,
version: int | None,
render_kwargs: dict,
messages_builder,
fallback_version: int = 1,
) -> dict:
"""Production call with full failure recovery."""
try:
data = prompt_manager.load(name, version)
rendered = data["template"].format(**render_kwargs)
messages = messages_builder(rendered)
return monitored_call.call(name, data["version"], messages)
except (KeyError, FileNotFoundError) as e:
logger.error(f"Prompt render/load failed: {e}, rolling back to v{fallback_version}")
data = prompt_manager.load(name, fallback_version)
rendered = data["template"].format(**render_kwargs)
messages = messages_builder(rendered)
return monitored_call.call(name, fallback_version, messages)
Key Takeaways¶
- Treat prompts as versioned artifacts with a full lifecycle: develop, eval, stage, deploy, monitor, iterate.
- Never overwrite prompts in place — auto-increment versions with metadata (eval score, author, change description).
- Log every LLM call with prompt name, version, model, latency, tokens, and cost.
- A/B test prompt changes with deterministic user assignment before full rollout.
- Build a continuous improvement pipeline: daily monitoring, weekly failure review, bi-weekly eval set updates.
- Implement explicit failure recovery: render fallback → API retry → parse retry → safe default → rollback.
- Dashboard accuracy, latency, cost, error rate, and fallback rate as first-class metrics.
- Automatic rollback to the last known-good prompt version when accuracy drops below threshold.
- Cost monitoring is essential — small prompt changes can have large cost implications at scale.
Module Complete¶
You have completed the Prompt Engineering Mastery module. You can now:
- Design prompts with clear anatomy (role, context, task, format, constraints, examples)
- Apply few-shot and chain-of-thought techniques for complex reasoning
- Build system prompts and role designs that control behavior across conversations
- Get reliable structured output with JSON mode and Pydantic schemas
- Manage prompts as versioned, testable templates
- Chain prompts into production pipelines with error handling
- Optimize prompts systematically with eval sets and A/B testing
- Defend against edge cases and prompt injection with layered guardrails
- Adapt prompts across model providers
- Operate prompts in production with monitoring, versioning, and continuous improvement
Return to the module index or continue to the next module in the Build phase.