Module 8: Prompt Engineering in Production
8. Final Project: Production Prompt System
Overview
The guide's final project: you build a production-ready LLM system that brings together every technique you learned in the previous modules. The result is a working system with a prompt registry, an evaluation pipeline, cost tracking, caching, model routing and monitoring.
This isn't an academic exercise. It's the system you'd use to deploy a classifier, an answer generator, or any LLM application to real production.
Why This Project
The previous modules covered each technique in isolation:
- Module 2: Few-shot prompting
- Module 3: Structured output
- Module 4: Chain-of-thought
- Module 5: Model routing
- Module 7: Evaluation
- Module 8: Versioning, caching, monitoring
The problem with learning techniques in isolation: In production, everything interacts. The cache affects costs. Routing affects quality. Versioning affects evaluation. This project brings it all together into one coherent system.
The System You're Going to Build
A Support Ticket Classification System with the following capabilities:
| Capability | Related Modules | Complexity |
|---|---|---|
| Classify tickets (Urgent/Normal/Low) | M2, M3, M4 | Base |
| Prompt versioning | M8 | Intermediate |
| Automatic evaluation | M7 | Intermediate |
| Cache for repeated queries | M8 | Intermediate |
| Routing by complexity | M5, M8 | Intermediate |
| Cost tracking | M8 | Basic |
| Monitoring and alerts | M8 | Advanced |
| Deploy checklist | M8 | Process |
System Architecture
production-prompt-system/
├── prompts/
│ ├── registry.py # Prompt version management
│ ├── templates.py # Templates with Jinja2
│ └── techniques.py # Few-shot bank, CoT
├── evaluation/
│ ├── metrics.py # Accuracy, faithfulness, format
│ ├── judge.py # LLM-as-judge
│ └── pipeline.py # Complete evaluation pipeline
├── production/
│ ├── cache.py # InMemory + Semantic cache
│ ├── cost_tracker.py # Token counting and cost tracking
│ ├── router.py # Model routing by complexity
│ └── monitor.py # Metrics collection and alerts
├── datasets/
│ └── golden_set.json # 100 annotated ticket examples
├── config/
│ └── settings.py # Centralized configuration
├── api.py # FastAPI endpoints
├── cli.py # CLI for operations
└── main.py # Entry point and demo
Step 1: Centralized Configuration
# config/settings.py
from dataclasses import dataclass, field
from typing import Optional
import os
@dataclass
class Settings:
"""Centralized system configuration."""
# OpenAI
openai_api_key: str = field(default_factory=lambda: os.getenv("OPENAI_API_KEY", ""))
cheap_model: str = "gpt-4o-mini"
premium_model: str = "gpt-4o"
# Prompt Registry
registry_path: str = "data/prompt_registry.json"
# Evaluation
golden_set_path: str = "datasets/golden_set.json"
min_accuracy: float = 0.85
min_faithfulness: float = 0.80
# Cache
cache_max_size: int = 1000
cache_ttl_seconds: int = 3600 # 1 hour
cache_semantic_threshold: float = 0.92
# Cost
daily_budget_usd: float = 10.0
monthly_budget_usd: float = 200.0
# Monitoring
max_latency_ms: int = 5000
error_rate_max: float = 0.05
# Canary
canary_initial_percentage: float = 0.05
@property
def tokens_per_dollar_mini(self) -> float:
return 1_000_000 / 0.15 # gpt-4o-mini: $0.15 / 1M input tokens
@property
def tokens_per_dollar_premium(self) -> float:
return 1_000_000 / 2.50 # gpt-4o: $2.50 / 1M input tokens
settings = Settings()Step 2: The Prompt Registry
# prompts/registry.py
import json
from pathlib import Path
from datetime import datetime
from typing import Optional
class PromptRegistry:
"""Registry of prompt versions with rollback."""
def __init__(self, path: str = "data/prompt_registry.json"):
self.path = Path(path)
self.path.parent.mkdir(parents=True, exist_ok=True)
self._data = self._load()
def _load(self) -> dict:
if self.path.exists():
with open(self.path) as f:
return json.load(f)
return {}
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, template: str, metadata: dict = None) -> dict:
"""Registers a new version of a prompt."""
if name not in self._data:
self._data[name] = {"active": None, "versions": {}}
entry = {
"template": template,
"metadata": metadata or {},
"created_at": datetime.now().isoformat(),
"status": "inactive"
}
self._data[name]["versions"][version] = entry
self._save()
print(f"📝 Registered: {name} {version}")
return entry
def activate(self, name: str, version: str) -> None:
"""Activates a specific version."""
if name not in self._data:
raise KeyError(f"Prompt '{name}' does not exist")
versions = self._data[name]["versions"]
if version not in versions:
raise KeyError(f"Version '{version}' does not exist in '{name}'")
# Deactivate the previous version
previous_version = self._data[name]["active"]
if previous_version and previous_version in versions:
versions[previous_version]["status"] = "inactive"
# Activate the new version
versions[version]["status"] = "active"
self._data[name]["active"] = version
self._save()
print(f"✅ Activated: {name} → {version}")
def get(self, name: str, version: str = None) -> str:
"""Gets the template of a prompt."""
if name not in self._data:
raise KeyError(f"Prompt '{name}' does not exist")
v = version or self._data[name]["active"]
if not v:
raise ValueError(f"'{name}' has no active version")
return self._data[name]["versions"][v]["template"]
def rollback(self, name: str) -> str:
"""Rolls back to the previous stable version."""
versions = self._data.get(name, {}).get("versions", {})
active = self._data.get(name, {}).get("active")
# Find the previous version by creation date
candidates = [
(v, info["created_at"])
for v, info in versions.items()
if v != active
]
if not candidates:
raise ValueError(f"There is no previous version for '{name}'")
candidates.sort(key=lambda x: x[1], reverse=True)
rollback_version = candidates[0][0]
self.activate(name, rollback_version)
return rollback_version
def list_versions(self, name: str) -> list[dict]:
"""Lists every version of a prompt."""
versions = self._data.get(name, {}).get("versions", {})
active = self._data.get(name, {}).get("active")
return [
{
"version": v,
"status": info["status"],
"is_active": v == active,
"created_at": info["created_at"],
"metadata": info.get("metadata", {})
}
for v, info in versions.items()
]Step 3: Templates with Few-Shot and CoT
# prompts/templates.py
from jinja2 import Template
from openai import OpenAI
client = OpenAI()
# Few-shot examples for ticket classification
FEW_SHOT_TICKETS = [
{
"ticket": "The app crashed and I lost 3 hours of unsaved work",
"urgency": "URGENT",
"reason": "Loss of the user's work"
},
{
"ticket": "How can I change my password?",
"urgency": "LOW",
"reason": "Standard information request"
},
{
"ticket": "I can't process payments since this morning, it affects my whole sales team",
"urgency": "URGENT",
"reason": "Business impact and multiple users"
},
{
"ticket": "The export to PDF button doesn't work properly in Firefox",
"urgency": "NORMAL",
"reason": "Bug in a specific feature, has a workaround"
},
{
"ticket": "I'd like to suggest a new feature",
"urgency": "LOW",
"reason": "Feature request, not a current problem"
}
]
CLASSIFICATION_TEMPLATE = Template("""Classify the urgency level of this support ticket.
SCALE:
- URGENT: System down, data loss, impact on multiple users or business-critical
- NORMAL: Bug affecting an important feature but with a workaround
- LOW: Questions, suggestions, minor bugs
{% if few_shot %}REFERENCE EXAMPLES:
{% for ex in examples %}
Ticket: "{{ ex.ticket }}"
Urgency: {{ ex.urgency }}
Reason: {{ ex.reason }}
{% endfor %}
{% endif %}
{% if use_cot %}INSTRUCTIONS:
1. Identify the impact on the user
2. Evaluate whether it affects the business or data
3. Determine whether a workaround is available
4. Assign the urgency based on the criteria
Step-by-step analysis:
{% endif %}
TICKET: {{ ticket }}
Respond in JSON: {"urgency": "URGENT|NORMAL|LOW", "confidence": 0.0-1.0, "reason": "..."}""")
def render_prompt(ticket: str, few_shot: bool = True, use_cot: bool = False) -> str:
"""Renders the prompt with configurable options."""
return CLASSIFICATION_TEMPLATE.render(
ticket=ticket,
few_shot=few_shot,
examples=FEW_SHOT_TICKETS[:3] if few_shot else [],
use_cot=use_cot
)Step 4: Evaluation System
# evaluation/metrics.py
import json
from openai import OpenAI
client = OpenAI()
def evaluate_accuracy(predictions: list[str], ground_truth: list[str]) -> float:
"""Normalized accuracy for classification."""
if not predictions:
return 0.0
def normalize(s: str) -> str:
return s.strip().upper()
correct = sum(
1 for p, g in zip(predictions, ground_truth)
if normalize(p) == normalize(g)
)
return correct / len(predictions)
def evaluate_format_compliance(outputs: list[str]) -> float:
"""Checks that the outputs are valid JSON with the required fields."""
valid = 0
required_fields = {"urgency", "confidence", "reason"}
for output in outputs:
try:
data = json.loads(output)
if required_fields.issubset(set(data.keys())):
if data["urgency"] in ["URGENT", "NORMAL", "LOW"]:
valid += 1
except json.JSONDecodeError:
pass
return valid / len(outputs) if outputs else 0.0
def evaluate_average_confidence(outputs: list[str]) -> float:
"""Average of the reported confidence scores."""
confidences = []
for output in outputs:
try:
data = json.loads(output)
confidences.append(float(data.get("confidence", 0)))
except (json.JSONDecodeError, ValueError):
confidences.append(0.0)
return sum(confidences) / len(confidences) if confidences else 0.0
def llm_judge_batch(
tickets: list[str],
classifications: list[str],
ground_truth: list[str],
sample_size: int = 20
) -> float:
"""
LLM-as-judge to evaluate the quality of the reasoning.
Only evaluates a sample, to keep costs down.
"""
import random
indices = random.sample(range(len(tickets)), min(sample_size, len(tickets)))
scores = []
for i in indices:
try:
data = json.loads(classifications[i])
reason = data.get("reason", "")
except (json.JSONDecodeError, KeyError):
reason = classifications[i]
prompt = f"""Evaluate the quality of the reasoning for this ticket classification.
Ticket: {tickets[i]}
Classification: {ground_truth[i]}
Reasoning given: {reason}
Score ONLY the reasoning from 1-5 (not whether the classification is correct):
5 = Clear, specific and justified reasoning
3 = Valid but generic reasoning
1 = No reasoning, or incorrect
Respond ONLY with a number from 1 to 5."""
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
temperature=0,
max_tokens=10
)
try:
score = float(response.choices[0].message.content.strip()) / 5.0
scores.append(score)
except ValueError:
scores.append(0.6) # Default if it doesn't parse
return sum(scores) / len(scores) if scores else 0.0# evaluation/pipeline.py
import time
from openai import OpenAI
from evaluation.metrics import (
evaluate_accuracy, evaluate_format_compliance,
evaluate_average_confidence, llm_judge_batch
)
client = OpenAI()
class EvaluationPipeline:
"""Complete evaluation pipeline for the classification system."""
def __init__(self, settings):
self.settings = settings
def run(
self,
prompt_template: str,
golden_set: list[dict],
name: str = "evaluation",
verbose: bool = True
) -> dict:
"""
Runs the full evaluation.
Returns: dict with every metric and a report.
"""
start = time.time()
tickets = [ex["input"] for ex in golden_set]
ground_truth = [ex["expected_output"] for ex in golden_set]
if verbose:
print(f"🔍 Evaluating '{name}' with {len(golden_set)} examples...")
# 1. Run the prompt on every example
outputs = []
for ticket in tickets:
try:
response = client.chat.completions.create(
model=self.settings.cheap_model,
messages=[{
"role": "user",
"content": prompt_template.replace("{ticket}", ticket)
}],
temperature=0,
max_tokens=200,
response_format={"type": "json_object"}
)
outputs.append(response.choices[0].message.content)
except Exception as e:
outputs.append(f'{{"urgency": "LOW", "confidence": 0, "reason": "ERROR: {str(e)}"}}')
# 2. Extract the predictions from the JSON
import json
predictions = []
for output in outputs:
try:
data = json.loads(output)
predictions.append(data.get("urgency", "UNKNOWN"))
except json.JSONDecodeError:
predictions.append("PARSE_ERROR")
# 3. Compute the metrics
accuracy = evaluate_accuracy(predictions, ground_truth)
format_compliance = evaluate_format_compliance(outputs)
avg_confidence = evaluate_average_confidence(outputs)
# LLM judge on a sample (more expensive)
reasoning_quality = llm_judge_batch(tickets, outputs, ground_truth, sample_size=10)
# 4. Identify the failures
failures = [
{
"ticket": tickets[i],
"prediction": predictions[i],
"ground_truth": ground_truth[i]
}
for i in range(len(tickets))
if predictions[i] != ground_truth[i]
]
duration = time.time() - start
return {
"name": name,
"n_examples": len(golden_set),
"duration_s": round(duration, 2),
"metrics": {
"accuracy": round(accuracy, 4),
"format_compliance": round(format_compliance, 4),
"avg_confidence": round(avg_confidence, 4),
"reasoning_quality": round(reasoning_quality, 4)
},
"failures": failures[:5], # Top 5 failures for diagnosis
"approved": (
accuracy >= self.settings.min_accuracy and
format_compliance >= 0.95
)
}
def compare_versions(
self,
prompt_a: str,
prompt_b: str,
golden_set: list[dict]
) -> dict:
"""Compares two versions of a prompt."""
print("⚔️ Comparing versions A vs B...")
result_a = self.run(prompt_a, golden_set, "version_a", verbose=False)
result_b = self.run(prompt_b, golden_set, "version_b", verbose=False)
metrics_a = result_a["metrics"]
metrics_b = result_b["metrics"]
deltas = {
metric: round(metrics_b[metric] - metrics_a[metric], 4)
for metric in metrics_a
}
winner = "B" if deltas["accuracy"] > 0 else ("A" if deltas["accuracy"] < 0 else "TIE")
return {
"version_a": metrics_a,
"version_b": metrics_b,
"deltas": deltas,
"winner": winner,
"accuracy_improvement": f"{deltas['accuracy']:+.2%}"
}Step 5: Production - Cache, Router and Cost Tracker
# production/cache.py
import hashlib
import time
from typing import Optional
class InMemoryCache:
"""In-memory cache with TTL for LLM queries."""
def __init__(self, max_size: int = 1000, ttl_seconds: int = 3600):
self.max_size = max_size
self.ttl = ttl_seconds
self._cache: dict[str, dict] = {}
self._hits = 0
self._misses = 0
def _key(self, prompt: str, model: str) -> str:
content = f"{model}:{prompt}"
return hashlib.sha256(content.encode()).hexdigest()[:16]
def get(self, prompt: str, model: str) -> Optional[str]:
key = self._key(prompt, model)
entry = self._cache.get(key)
if not entry:
self._misses += 1
return None
# Check the TTL
if time.time() - entry["timestamp"] > self.ttl:
del self._cache[key]
self._misses += 1
return None
self._hits += 1
return entry["value"]
def set(self, prompt: str, model: str, value: str) -> None:
# Simple eviction if it's full
if len(self._cache) >= self.max_size:
oldest_key = min(self._cache.items(), key=lambda x: x[1]["timestamp"])[0]
del self._cache[oldest_key]
key = self._key(prompt, model)
self._cache[key] = {"value": value, "timestamp": time.time()}
@property
def hit_rate(self) -> float:
total = self._hits + self._misses
return self._hits / total if total > 0 else 0.0
def stats(self) -> dict:
return {
"size": len(self._cache),
"max_size": self.max_size,
"hits": self._hits,
"misses": self._misses,
"hit_rate": round(self.hit_rate, 3)
}# production/router.py
import tiktoken
from openai import OpenAI
client = OpenAI()
class ModelRouter:
"""
Routes requests to the appropriate model based on complexity.
Simple/fast → gpt-4o-mini
Complex/ambiguous → gpt-4o
"""
SIMPLE_TICKETS = [
"password", "login", "access", "how", "where", "when",
"tutorial", "guide", "suggestion", "question"
]
COMPLEX_TICKETS = [
"lost", "down", "critical", "urgent", "emergency", "data",
"not working", "error", "failure", "bug", "production"
]
def __init__(self, settings):
self.settings = settings
self._model_decisions = []
def select_model(self, ticket: str) -> tuple[str, str]:
"""
Selects the model and returns (model, reason).
"""
ticket_lower = ticket.lower()
# Fast heuristic by keywords
has_simple = any(k in ticket_lower for k in self.SIMPLE_TICKETS)
has_complex = any(k in ticket_lower for k in self.COMPLEX_TICKETS)
if has_complex:
model = self.settings.premium_model
reason = "high-urgency keywords detected"
elif has_simple and not has_complex:
model = self.settings.cheap_model
reason = "low-complexity ticket"
else:
# Tokens as a complexity signal
enc = tiktoken.encoding_for_model("gpt-4o")
n_tokens = len(enc.encode(ticket))
if n_tokens > 100:
model = self.settings.premium_model
reason = f"long ticket ({n_tokens} tokens)"
else:
model = self.settings.cheap_model
reason = "short, direct ticket"
self._model_decisions.append({
"ticket_preview": ticket[:50],
"model": model,
"reason": reason
})
return model, reason
def stats(self) -> dict:
"""Routing statistics."""
total = len(self._model_decisions)
if not total:
return {}
cheap = sum(1 for d in self._model_decisions if d["model"] == self.settings.cheap_model)
return {
"total_requests": total,
"cheap": cheap,
"premium": total - cheap,
"pct_cheap": round(cheap / total, 3)
}# production/cost_tracker.py
from dataclasses import dataclass, field
from datetime import datetime
@dataclass
class RequestCost:
"""Cost of an individual request."""
timestamp: str
prompt_tokens: int
completion_tokens: int
model: str
cost_usd: float
cached: bool = False
class CostTracker:
"""Real-time cost tracking."""
PRICES = {
"gpt-4o-mini": {"input": 0.15 / 1_000_000, "output": 0.60 / 1_000_000},
"gpt-4o": {"input": 2.50 / 1_000_000, "output": 10.00 / 1_000_000}
}
def __init__(self, settings):
self.settings = settings
self._requests: list[RequestCost] = []
def record(self, response, model: str, cached: bool = False) -> float:
"""Records the cost of an API response."""
usage = response.usage
prices = self.PRICES.get(model, self.PRICES["gpt-4o-mini"])
cost = (
usage.prompt_tokens * prices["input"] +
usage.completion_tokens * prices["output"]
)
if cached:
cost = 0.0
self._requests.append(RequestCost(
timestamp=datetime.now().isoformat(),
prompt_tokens=usage.prompt_tokens,
completion_tokens=usage.completion_tokens,
model=model,
cost_usd=cost,
cached=cached
))
return cost
def total_cost(self) -> float:
return sum(r.cost_usd for r in self._requests)
def cost_by_model(self) -> dict[str, float]:
costs = {}
for r in self._requests:
costs[r.model] = costs.get(r.model, 0.0) + r.cost_usd
return costs
def report(self) -> str:
total = self.total_cost()
by_model = self.cost_by_model()
n_cached = sum(1 for r in self._requests if r.cached)
n_total = len(self._requests)
lines = [
"## Cost Report",
f"Total requests: {n_total}",
f"Cached (no cost): {n_cached} ({n_cached/n_total:.0%} if n_total>0)",
f"",
"Cost by model:",
]
for model, cost in by_model.items():
lines.append(f" {model}: ${cost:.4f}")
lines.append(f"\nTotal: ${total:.4f}")
# Monthly projection
if n_total > 0:
cost_per_request = total / n_total
projection_1000_req = cost_per_request * 1000
lines.append(f"\nProjection per 1,000 requests: ${projection_1000_req:.2f}")
return "\n".join(lines)Step 6: Monitoring
# production/monitor.py
import time
import threading
from collections import deque
from datetime import datetime
class ProductionMonitor:
"""
Production monitor with a sliding window and alerts.
"""
def __init__(self, settings, window_seconds: int = 300):
self.settings = settings
self.window = window_seconds
self._lock = threading.Lock()
self._latencies = deque()
self._errors = deque()
self._requests = deque()
self._alerts_sent: set = set()
def record(self, latency_ms: float, error: bool = False) -> None:
"""Records a request."""
now = time.time()
with self._lock:
self._latencies.append((now, latency_ms))
self._requests.append(now)
if error:
self._errors.append(now)
# Clean up anything outside the window
cutoff = now - self.window
while self._latencies and self._latencies[0][0] < cutoff:
self._latencies.popleft()
while self._requests and self._requests[0] < cutoff:
self._requests.popleft()
while self._errors and self._errors[0] < cutoff:
self._errors.popleft()
def current_metrics(self) -> dict:
"""Computes the current metrics for the window."""
with self._lock:
latencies = [l for _, l in self._latencies]
n_requests = len(self._requests)
n_errors = len(self._errors)
if not latencies:
return {"status": "no_data"}
sorted_latencies = sorted(latencies)
n = len(sorted_latencies)
return {
"n_requests": n_requests,
"error_rate": n_errors / n_requests if n_requests > 0 else 0,
"latency_p50": sorted_latencies[int(n * 0.50)],
"latency_p95": sorted_latencies[int(n * 0.95)],
"latency_p99": sorted_latencies[int(n * 0.99)] if n >= 100 else sorted_latencies[-1],
"window_seconds": self.window
}
def verify_alerts(self) -> list[dict]:
"""Checks whether there are any alert conditions."""
metrics = self.current_metrics()
alerts = []
if metrics.get("status") == "no_data":
return alerts
# High latency alert
if metrics["latency_p95"] > self.settings.max_latency_ms:
alert_id = "high_latency"
if alert_id not in self._alerts_sent:
alerts.append({
"type": alert_id,
"message": f"High p95 latency: {metrics['latency_p95']:.0f}ms (max: {self.settings.max_latency_ms}ms)",
"severity": "WARNING"
})
self._alerts_sent.add(alert_id)
else:
self._alerts_sent.discard("high_latency")
# High error rate alert
if metrics["error_rate"] > self.settings.error_rate_max:
alert_id = "high_error_rate"
if alert_id not in self._alerts_sent:
alerts.append({
"type": alert_id,
"message": f"High error rate: {metrics['error_rate']:.1%} (max: {self.settings.error_rate_max:.1%})",
"severity": "CRITICAL"
})
self._alerts_sent.add(alert_id)
else:
self._alerts_sent.discard("high_error_rate")
return alertsStep 7: The Integrated System
# main.py — the complete integrated system
import json
import time
from pathlib import Path
from openai import OpenAI
from config.settings import Settings
from prompts.registry import PromptRegistry
from prompts.templates import render_prompt
from evaluation.pipeline import EvaluationPipeline
from production.cache import InMemoryCache
from production.router import ModelRouter
from production.cost_tracker import CostTracker
from production.monitor import ProductionMonitor
settings = Settings()
client = OpenAI()
# Initialize the components
registry = PromptRegistry(settings.registry_path)
eval_pipeline = EvaluationPipeline(settings)
cache = InMemoryCache(settings.cache_max_size, settings.cache_ttl_seconds)
router = ModelRouter(settings)
cost_tracker = CostTracker(settings)
monitor = ProductionMonitor(settings)
def classify_ticket(ticket: str, request_id: str = None) -> dict:
"""
Classifies a ticket, wiring together every component of the system.
"""
start = time.time()
cached = False
# 1. Get the active prompt from the registry
prompt_template = registry.get("ticket_classifier")
prompt = prompt_template.replace("{ticket}", ticket)
# 2. Check the cache
selected_model, routing_reason = router.select_model(ticket)
cached_result = cache.get(prompt, selected_model)
if cached_result:
result = json.loads(cached_result)
latency_ms = (time.time() - start) * 1000
monitor.record(latency_ms, error=False)
return {
**result,
"cached": True,
"model": selected_model,
"latency_ms": round(latency_ms, 1),
"cost_usd": 0.0
}
# 3. Call the API
try:
response = client.chat.completions.create(
model=selected_model,
messages=[{"role": "user", "content": prompt}],
temperature=0,
max_tokens=200,
response_format={"type": "json_object"}
)
output = response.choices[0].message.content
result = json.loads(output)
# 4. Record the cost
cost = cost_tracker.record(response, selected_model, cached=False)
# 5. Store it in the cache
cache.set(prompt, selected_model, output)
latency_ms = (time.time() - start) * 1000
monitor.record(latency_ms, error=False)
return {
**result,
"cached": False,
"model": selected_model,
"routing_reason": routing_reason,
"latency_ms": round(latency_ms, 1),
"cost_usd": round(cost, 6)
}
except Exception as e:
latency_ms = (time.time() - start) * 1000
monitor.record(latency_ms, error=True)
return {
"urgency": "NORMAL",
"confidence": 0.0,
"reason": f"Error while classifying: {str(e)}",
"error": True,
"latency_ms": round(latency_ms, 1)
}
def demo_system():
"""Demo of the complete system."""
# Register the initial version of the prompt
registry.register(
name="ticket_classifier",
version="v1.0.0",
template=render_prompt("{ticket}", few_shot=True, use_cot=False),
metadata={"description": "Classifier with few-shot"}
)
registry.activate("ticket_classifier", "v1.0.0")
# Example tickets
tickets = [
"The app won't load and I have an important demo in 30 minutes",
"How do I export my data to Excel?",
"The payment system failed and we're losing sales right now",
"Do you have a dark mode option?",
"I found a small bug in a button's hover state",
]
print("🚀 DEMO: Ticket Classification System")
print("=" * 60)
for ticket in tickets:
result = classify_ticket(ticket)
emoji = {"URGENT": "🔴", "NORMAL": "🟡", "LOW": "🟢"}.get(result["urgency"], "⚪")
cached_str = " [CACHE]" if result.get("cached") else ""
print(f"\n{emoji} {result['urgency']}{cached_str}")
print(f" Ticket: {ticket[:60]}...")
print(f" Reason: {result.get('reason', '')[:80]}")
print(f" Confidence: {result.get('confidence', 0):.0%}")
print(f" Model: {result.get('model', '')} | {result['latency_ms']:.0f}ms | ${result.get('cost_usd', 0):.5f}")
print("\n" + "=" * 60)
print("📊 SYSTEM STATISTICS")
print("\n🗄️ Cache:")
stats = cache.stats()
print(f" Hit rate: {stats['hit_rate']:.0%} | Size: {stats['size']}/{stats['max_size']}")
print("\n🔀 Model Routing:")
routing_stats = router.stats()
print(f" Cheap: {routing_stats.get('pct_cheap', 0):.0%}")
print(f" Premium: {1 - routing_stats.get('pct_cheap', 0):.0%}")
print("\n💰 Costs:")
print(cost_tracker.report())
print("\n📡 Monitoring:")
metrics = monitor.current_metrics()
if metrics.get("status") != "no_data":
print(f" Latency p95: {metrics.get('latency_p95', 0):.0f}ms")
print(f" Error rate: {metrics.get('error_rate', 0):.1%}")
alerts = monitor.verify_alerts()
if alerts:
print(f"\n⚠️ {len(alerts)} active alert(s)")
else:
print("\n✅ No active alerts")
if __name__ == "__main__":
demo_system()Success Criteria
## Final Project Checklist
### Core Functionality
- [ ] Working prompt registry (register, activate, rollback)
- [ ] Ticket classification works with accuracy ≥ 85%
- [ ] Few-shot examples wired into the template
- [ ] Output is valid JSON (format_compliance ≥ 95%)
### Evaluation
- [ ] Golden set with ≥ 50 examples in golden_set.json
- [ ] Evaluation pipeline runs and produces a report
- [ ] Metrics include: accuracy, format_compliance, reasoning_quality
- [ ] A/B comparison between two versions of the prompt
### Production
- [ ] Cache implemented (hit rate ≥ 20% with repeated queries)
- [ ] Model routing works (≥ 60% of requests go to the cheap model)
- [ ] Cost tracker live (cost report by model)
- [ ] Monitor live (p95 latency and error rate)
### Deploy
- [ ] ProductionChecklist completed
- [ ] Rollback documented and tested
- [ ] Alerts configured
### Code Quality
- [ ] Centralized configuration in settings.py
- [ ] No hardcoded API keys
- [ ] Error handling on every API call
- [ ] Runnable code (not just a pseudocode demo)Optional Extensions
Once the base project is done, consider these extensions to go deeper:
| Extension | Technique | Difficulty |
|---|---|---|
| REST API with FastAPI | FastAPI + Pydantic | Intermediate |
| Semantic cache with embeddings | Embeddings + cosine similarity | Intermediate |
| Dashboard in Streamlit | Streamlit | Low |
| Tracing with LangSmith | LangSmith SDK | Low |
| Automated A/B testing | Stats + pytest | High |
| Prometheus + Grafana | prometheus_client | High |
# api.py — optional extension: FastAPI endpoint
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import uvicorn
app = FastAPI(title="Ticket Classifier API", version="1.0.0")
class TicketRequest(BaseModel):
ticket: str
request_id: str = None
class TicketResponse(BaseModel):
urgency: str
confidence: float
reason: str
model: str
latency_ms: float
cached: bool
cost_usd: float
@app.post("/classify", response_model=TicketResponse)
async def classify(request: TicketRequest):
result = classify_ticket(request.ticket, request.request_id)
if result.get("error"):
raise HTTPException(status_code=500, detail=result["reason"])
return result
@app.get("/health")
async def health():
metrics = monitor.current_metrics()
return {"status": "ok", "metrics": metrics}
@app.get("/stats")
async def stats():
return {
"cache": cache.stats(),
"routing": router.stats(),
"costs": cost_tracker.cost_by_model()
}
# To run it: uvicorn api:app --reloadTroubleshooting
Problem 1: Low accuracy (< 85%)
# Diagnose which examples fail and why
eval_result = eval_pipeline.run(
prompt_template=registry.get("ticket_classifier"),
golden_set=golden_set,
name="diagnosis"
)
print("Most common failures:")
for failure in eval_result["failures"]:
print(f" Ticket: {failure['ticket'][:60]}")
print(f" Prediction: {failure['prediction']} | Ground truth: {failure['ground_truth']}")
print()
# Solutions:
# 1. Check whether the failures share a pattern (are they all NORMAL→URGENT?)
# 2. Add examples of the failing type to the few-shot bank
# 3. Tune the prompt instructions for that specific caseProblem 2: Low cache hit rate
# Analyze why the cache isn't getting hits
stats = cache.stats()
print(f"Hit rate: {stats['hit_rate']:.0%}")
# If hit rate < 5% with similar queries → check whether the exact prompt varies
# Solution: normalize the ticket before generating the cache key
def normalize_ticket(ticket: str) -> str:
"""Normalizes to improve the hit rate."""
import re
# Lowercase, collapse double punctuation, trim whitespace
ticket = ticket.lower().strip()
ticket = re.sub(r'\s+', ' ', ticket)
ticket = re.sub(r'[!?]{2,}', '!', ticket)
return ticketSummary
- Modular architecture: every component is independent and testable
- Prompt registry: semantic versioning with rollback in 1 line
- Evaluation pipeline: accuracy + format_compliance + automatic LLM judge
- Cache: cuts costs by up to 40-60% on repeated queries
- Model routing: 60-80% of traffic to the cheap model
- Cost tracking: full visibility of spend by model and by period
- Monitoring: p95 latency, error rate, automatic alerts
- Deploy checklist: the complete process before going to production
Additional resources
- OpenAI Production Best Practices — Official guide
- LangSmith Docs — Observability for LLMs
- FastAPI Tutorial — Building APIs with FastAPI
- Jinja2 Template Designer — Advanced templates
- Prometheus Python Client — Production metrics
- The Twelve-Factor App — Principles for cloud-native apps