Production architecture: monitoring, alerting, scaling
On this page
Your application in production: the three pillars
Deploying an LLM application to production isn’t just about the code. Three pillars guarantee reliability: monitoring to observe, alerting to react, and scaling to absorb load. They form a chain, and a chain is only as strong as its weakest link: metrics no one looks at are useless, alerts without a response procedure are noise, and scaling that kicks in too late is just an outage. This final lesson brings together every concept from the course into a complete production architecture.
Observe: what to measure
Four families of metrics cover the essentials: call volume, latency, errors, and cost, broken down by model. Breaking down by model is what makes the dashboard actionable—a global error rate of 3% tells you nothing, while that same figure concentrated on a single model immediately points to the culprit.
The class below accumulates these metrics in memory, using a sliding window of 10,000 latencies to prevent the process from growing indefinitely. Note the use of percentiles rather than averages: the average latency of an LLM service is almost always misleading because a handful of very slow calls disappear into the average, yet those are the ones your users complain about.
import time
from dataclasses import dataclass, field
from collections import defaultdict
@dataclass
class ProductionMetrics:
"""Collects essential production metrics."""
counters: dict = field(default_factory=lambda: defaultdict(int))
latencies: dict = field(default_factory=lambda: defaultdict(list))
errors: dict = field(default_factory=lambda: defaultdict(int))
costs: dict = field(default_factory=lambda: defaultdict(float))
def record_call(
self,
model: str,
latency: float,
input_tokens: int,
output_tokens: int,
cost: float,
success: bool,
):
self.counters[model] += 1
self.latencies[model].append(latency)
self.costs[model] += cost
if not success:
self.errors[model] += 1
# Keep only the last 10,000 latencies
if len(self.latencies[model]) > 10_000:
self.latencies[model] = self.latencies[model][-10_000:]
def percentile(self, model: str, p: float) -> float:
"""Calculates a latency percentile."""
values = sorted(self.latencies.get(model, []))
if not values:
return 0.0
idx = int(len(values) * p / 100)
return values[min(idx, len(values) - 1)]
def error_rate(self, model: str) -> float:
total = self.counters.get(model, 0)
if total == 0:
return 0.0
return self.errors.get(model, 0) / total
def dashboard(self) -> dict:
"""Generates a complete dashboard."""
dashboard = {}
for model in self.counters:
dashboard[model] = {
"calls": self.counters[model],
"latency_p50": f"{self.percentile(model, 50):.2f}s",
"latency_p99": f"{self.percentile(model, 99):.2f}s",
"error_rate": f"{self.error_rate(model):.2%}",
"total_cost": f"${self.costs[model]:.2f}",
}
return dashboardOf course, collection must be systematic. Manually instrumenting every call guarantees you’ll forget one—and probably the one that causes trouble. The decorator below moves measurement to the call function level: the finally block records latency and success no matter what, even when an exception propagates.
import openai
import functools
metrics = ProductionMetrics()
def monitor(func):
"""Decorator that monitors every API call."""
@functools.wraps(func)
def wrapper(*args, **kwargs):
model = kwargs.get("model", args[0] if args else "unknown")
start = time.perf_counter()
success = True
try:
result = func(*args, **kwargs)
return result
except Exception as e:
success = False
raise
finally:
latency = time.perf_counter() - start
metrics.record_call(
model=model,
latency=latency,
input_tokens=0,
output_tokens=0,
cost=0,
success=success,
)
return wrapperToken counters and cost remain zero in this version because the response object isn’t available at recording time. Complete them by reading the usage fields returned by the API: without this information, your dashboard will faithfully display a column of zero costs.
React: graduated alerts
An alert is only valuable if it calls for action. That’s why the system distinguishes two thresholds per metric: a warning threshold, signaling a drift to examine during the day, and a critical threshold, justifying waking someone up. Confusing the two produces the worst possible outcome—a team that has learned to ignore notifications.
from enum import Enum
class AlertLevel(Enum):
INFO = "info"
WARNING = "warning"
CRITICAL = "critical"
class AlertSystem:
"""Triggers alerts based on metrics."""
def __init__(self):
self.thresholds = {
"error_rate": {"warning": 0.05, "critical": 0.15},
"latency_p99": {"warning": 10.0, "critical": 30.0},
"hourly_cost": {"warning": 50.0, "critical": 200.0},
}
self.sent_alerts: list[dict] = []
def check(self, metrics: ProductionMetrics) -> list[dict]:
"""Checks metrics and generates alerts."""
alerts = []
for model in metrics.counters:
# Error rate
rate = metrics.error_rate(model)
if rate >= self.thresholds["error_rate"]["critical"]:
alerts.append({
"level": AlertLevel.CRITICAL,
"message": (
f"CRITICAL error rate for {model}: "
f"{rate:.1%}"
),
"model": model,
})
elif rate >= self.thresholds["error_rate"]["warning"]:
alerts.append({
"level": AlertLevel.WARNING,
"message": (
f"High error rate for {model}: "
f"{rate:.1%}"
),
"model": model,
})
# Latency
p99 = metrics.percentile(model, 99)
if p99 >= self.thresholds["latency_p99"]["critical"]:
alerts.append({
"level": AlertLevel.CRITICAL,
"message": (
f"CRITICAL p99 latency for {model}: "
f"{p99:.1f}s"
),
"model": model,
})
self.sent_alerts.extend(alerts)
return alerts
async def notify(self, alerts: list[dict]):
"""Sends notifications (webhook, email, Slack)."""
import httpx
for alert in alerts:
if alert["level"] == AlertLevel.CRITICAL:
# Immediate notification
async with httpx.AsyncClient() as http:
await http.post(
"https://your-webhook.example.com/alerts",
json={
"level": alert["level"].value,
"message": alert["message"],
},
)Hourly cost alerts are the ones most often forgotten, and that’s a shame: an unbounded retry loop or a batch job launched twice results in a bill, not an error. It’s the only safety net that detects this type of incident while it’s happening.
Absorb: three resilience mechanisms
When the API returns an error, retrying makes things worse: you add load to a service already in trouble and make your own users wait for nothing. The circuit breaker reverses this logic. After a defined number of consecutive failures, it opens the circuit and immediately rejects subsequent calls, then retries after a reset delay.
class CircuitBreaker:
"""Cuts off calls when too many errors occur."""
def __init__(
self,
open_threshold: int = 5,
reset_delay: float = 60.0,
):
self.threshold = open_threshold
self.reset_delay = reset_delay
self.consecutive_failures = 0
self.opened_at: float | None = None
@property
def is_open(self) -> bool:
if self.opened_at is None:
return False
if time.time() - self.opened_at > self.reset_delay:
# Attempt to close the circuit
self.opened_at = None
self.consecutive_failures = 0
return False
return True
def record_success(self):
self.consecutive_failures = 0
self.opened_at = None
def record_failure(self):
self.consecutive_failures += 1
if self.consecutive_failures >= self.threshold:
self.opened_at = time.time()
def call(self, func, *args, **kwargs):
"""Executes the function if the circuit is closed."""
if self.is_open:
raise RuntimeError(
"Circuit open — API unavailable. "
f"Retry in {self.reset_delay}s."
)
try:
result = func(*args, **kwargs)
self.record_success()
return result
except Exception as e:
self.record_failure()
raiseThe circuit breaker protects your application; it doesn’t make it work, though. That’s the role of model fallback, which tries alternatives in order, skipping those whose circuits are already open. One design point to remember: the order of the list is a product decision, not a technical one. Put first the model that delivers the expected service, followed by those that deliver a degraded but acceptable service—a slightly less sharp response is better than an unavailable message.
class ModelFallback:
"""Switches to a backup model in case of failure."""
def __init__(self):
self.models = ["gpt-5.6-terra", "gpt-5.6-sol", "gpt-5.6-luna"]
self.circuits = {
m: CircuitBreaker() for m in self.models
}
def call(self, prompt: str) -> str:
"""Tries each model until a response is obtained."""
client = openai.OpenAI()
for model in self.models:
circuit = self.circuits[model]
if circuit.is_open:
continue
try:
response = circuit.call(
client.responses.create,
model=model,
input=prompt,
)
return response.output_text
except Exception:
continue
raise RuntimeError("All models are unavailable.")The third mechanism concerns sizing. Driving the number of workers based on queue size rather than CPU load is the right reflex for a service that spends most of its time waiting for network responses: the CPU sits idle while the queue grows. Note the asymmetry in the applied rule—we double on scale-up and halve on scale-down, with widely separated thresholds (100 vs. 10). This gap prevents oscillation: without it, the system would spend its time creating and destroying workers around the tipping point.
class AutoScaler:
"""Adjusts the number of workers based on load."""
def __init__(
self,
min_workers: int = 2,
max_workers: int = 50,
scale_up_threshold: int = 100,
scale_down_threshold: int = 10,
):
self.min_workers = min_workers
self.max_workers = max_workers
self.up_threshold = scale_up_threshold
self.down_threshold = scale_down_threshold
self.active_workers = min_workers
def adjust(self, queue_size: int) -> int:
"""Returns the recommended number of workers."""
if queue_size > self.up_threshold:
self.active_workers = min(
self.active_workers * 2,
self.max_workers,
)
elif queue_size < self.down_threshold:
self.active_workers = max(
self.active_workers // 2,
self.min_workers,
)
return self.active_workersProduction deployment checklist
This course ends with the checklist to run through before opening your application to the public. Each item refers to a lesson you’ve covered; if any seems unclear, reread that lesson before deploying.
- Latency monitoring (p50, p95, p99) and error rates
- Alerts configured on critical thresholds
- Circuit breaker to protect against API failures
- Model fallback configured
- Input and output moderation enabled
- Client-side rate limiting implemented
- Structured logs for debugging
- Automated evals in CI/CD
- Cost budget with alerts
- Versioned prompt documentation
Key takeaways
- Monitoring covers latency, errors, costs, and response quality
- Track latency percentiles, never just the average
- Alerts must be graduated: warning then critical
- Circuit breaker protects against failure cascades; fallback ensures availability
- A production checklist prevents critical oversights
Test your knowledge
Latency, costs, evals: production reflexes in five questions.
1. What are the main levers for reducing latency?
Answer: Streaming (perceived latency), prompt caching (stable prefixes), Predicted Outputs when the response is largely predictable (Chat Completions, on models retired January 14, 2027), and optimizing the prompt itself—measured end to end.
2. When should you use the Batch API?
Answer: For massive processing without real-time constraints: asynchronous batches at reduced cost—ranking, enrichment, or volume generation have no place on the synchronous path.
3. What does conversation compaction do?
Answer: It compresses long history (summaries, pruning) to fit within context and control costs—essential for agents and long sessions.
4. What is the purpose of the Evals framework in CI/CD?
Answer: To automatically run your evaluations on every prompt or model change: regression is detected at commit, not in production.
5. What does a complete production architecture include?
Answer: Monitoring (latency, costs, quality), alerting, asynchronous queues, moderation at scale, and a scaling plan—AI in production is a system, not an API call.
Measure, cache, batch, evaluate continuously: four habits that make applications fast, lean, and reliable.