The demo works. Someone types a question, the model answers, everyone in the room nods. Then it goes in front of real users and the questions start: what happens when the provider returns a 529? Why did last month's bill triple? Can customer A's search surface customer B's documents? Did the answer quality drop when we changed that prompt, or does it just feel that way?
None of those are model problems. They are the layers a demo does not have, and every one of them is ordinary software engineering applied to an unusually failure-prone dependency. This walks through six of them, in the order they tend to become urgent.
What a demo is missing
A working prototype typically has: one model call, no timeout, no retry policy, no cost accounting, credentials in a notebook, no tenant boundary in the retrieval query, no evaluation set, no structured logs, and no defined behaviour when the provider is unavailable.
That list is the work. The model is a dependency like any other, except it is slower, more expensive, less reliable and non-deterministic — which means every discipline you already apply to a third-party API applies harder.
Layer 1: a boundary around every model call
No model call should be made directly from feature code. Wrap it once, and put the timeout, the retry policy and the error taxonomy in that wrapper.
import asyncio
import httpx
from dataclasses import dataclass
class ModelUnavailable(Exception):
"""Retryable: the provider is overloaded or unreachable."""
class ModelRefused(Exception):
"""Not retryable: the request itself is unacceptable."""
class ModelBudgetExceeded(Exception):
"""Not retryable: this request would exceed its cost ceiling."""
RETRYABLE_STATUS = {408, 429, 500, 502, 503, 504, 529}
@dataclass
class Completion:
text: str
input_tokens: int
output_tokens: int
stop_reason: str
async def complete(
client: httpx.AsyncClient,
payload: dict,
*,
timeout_s: float = 45.0,
attempts: int = 3,
) -> Completion:
delay = 0.5
last_error: Exception | None = None
for attempt in range(1, attempts + 1):
try:
response = await client.post("/v1/messages", json=payload, timeout=timeout_s)
except (httpx.TimeoutException, httpx.TransportError) as exc:
last_error = ModelUnavailable(str(exc))
else:
if response.status_code == 200:
return _parse(response.json())
if response.status_code in RETRYABLE_STATUS:
last_error = ModelUnavailable(f"status {response.status_code}")
else:
raise ModelRefused(f"status {response.status_code}")
if attempt < attempts:
await asyncio.sleep(delay)
delay *= 2
raise last_error or ModelUnavailable("exhausted attempts")Line 13 is the taxonomy that everything upstream branches on. "The provider is briefly overloaded" and "your request is malformed" demand opposite responses, and collapsing both into a generic exception guarantees you will retry something that will never succeed.
Line 29 is doubling backoff — retrying a rate-limited provider immediately makes the rate limiting worse for everyone sharing your key.
Line 40 raises immediately on non-retryable statuses. A 400 will still be a 400 in two seconds.
Two things this deliberately omits: jitter, which you should add if you have many concurrent workers so retries do not synchronise, and a circuit breaker, which becomes worth it once a sustained outage would otherwise queue thousands of doomed requests.
Layer 2: budgets, not just metering
Cost surprises in AI applications almost never come from the average request. They come from the tail: the user who pastes a novel, the agent loop that runs to its step cap on every retry, the background job that reprocesses the corpus after a bug.
Budgets have to be enforced before the call, not measured after it.
MAX_INPUT_TOKENS = 60_000
MAX_OUTPUT_TOKENS = 4_000
def enforce_request_budget(payload: dict, estimate_tokens) -> None:
estimated = estimate_tokens(payload)
if estimated > MAX_INPUT_TOKENS:
raise ModelBudgetExceeded(
f"request would send ~{estimated} input tokens "
f"(ceiling {MAX_INPUT_TOKENS}); truncate context first"
)
if payload.get("max_tokens", 0) > MAX_OUTPUT_TOKENS:
raise ModelBudgetExceeded("max_tokens above ceiling")
async def enforce_account_budget(redis, account_id: str, period_cap_units: int) -> None:
key = f"spend:{account_id}"
spent = int(await redis.get(key) or 0)
if spent >= period_cap_units:
raise ModelBudgetExceeded(f"account {account_id} reached its period cap")Three levels are worth having: per request (this one call cannot be enormous), per account (one customer cannot consume the month), and per feature (an experimental endpoint cannot outspend the product). The account-level check is the one that turns a runaway loop from an incident into a rejected request.
Record usage per request against a request id, an account id and a feature name. Without those three dimensions, a bill increase is unattributable and you will spend a week guessing.
Layer 3: tenant isolation belongs in the query
The most damaging bug class in retrieval-backed products is cross-tenant leakage, and it has an unusually simple cause: the filter was applied after the search, or it was expressed as an instruction in the prompt.
-- Correct: tenant is a predicate the index enforces before ranking.
SELECT id, document_title, text, url
FROM doc_chunks
WHERE tenant_id = $1
AND deleted_at IS NULL
ORDER BY embedding <=> $2::vector
LIMIT $3;Two properties matter here. The tenant predicate is in the WHERE clause, so rows from
other tenants are never candidates. And soft-deleted rows are excluded in the query
rather than filtered in application code, because "we filter it in Python" is exactly
the code path someone will bypass in a hurry.
Where the database supports it, row-level security is stronger still, because it makes the isolation a property of the schema rather than of every query anyone writes:
ALTER TABLE doc_chunks ENABLE ROW LEVEL SECURITY;
CREATE POLICY tenant_isolation ON doc_chunks
USING (tenant_id = current_setting('app.tenant_id')::uuid);Then test it adversarially. A test that seeds two tenants, searches as one, and asserts that zero rows belonging to the other are returned is worth more than any amount of review. The wider architecture of retrieval systems is covered in RAG vs AI agents; this is the part of it that is a security control rather than a quality one.
Layer 4: evaluation as a build step
Without an evaluation set, every prompt change is a vibe. You will "improve" the system into a regression and find out from a customer.
The minimum viable harness is a JSONL file of cases and a script that runs them.
import json, asyncio
from pathlib import Path
async def run_case(case: dict, pipeline) -> dict:
result = await pipeline(case["question"], tenant_id=case["tenant_id"])
retrieved_ids = {source["id"] for source in result["sources"]}
expected = set(case["expected_source_ids"])
return {
"id": case["id"],
"recall": len(retrieved_ids & expected) / len(expected) if expected else None,
"refused": result["answer"].strip().lower().startswith("i don't"),
"should_refuse": case["should_refuse"],
"answer": result["answer"],
}
async def main(pipeline):
cases = [json.loads(line) for line in Path("eval/cases.jsonl").read_text().splitlines()]
results = await asyncio.gather(*(run_case(c, pipeline) for c in cases))
recallable = [r for r in results if r["recall"] is not None]
honesty = [r for r in results if r["should_refuse"]]
print(f"retrieval recall: {sum(r['recall'] for r in recallable) / len(recallable):.2f}")
print(f"correct refusals: {sum(r['refused'] for r in honesty)}/{len(honesty)}")
Path("eval/last-run.json").write_text(json.dumps(results, indent=2))The should_refuse cases carry most of the signal. Questions whose answer is genuinely
absent from the corpus test the one behaviour that separates a trustworthy assistant
from a confident one, and they are the first thing a prompt "improvement" quietly
breaks.
Run this in CI on every change to a prompt, a retrieval parameter or a model version. Store the output. A number without a previous number is not information.
Layer 5: observability, and what never to log
Trace every request end to end: the request id, the account, the feature, retrieval latency and hit count, model latency, token counts in and out, stop reason, retries, and the final outcome. Span-based tracing works well here because a single user action fans out into several dependent calls with very different latency characteristics.
What to log carefully: prompts and completions may contain customer data. The workable default is to log identifiers, metrics and stop reasons always, and content only under an explicit, time-limited debug flag scoped to a single account — with retention shorter than your general log retention.
What to alert on is different from what to graph. Useful alerts: retryable error rate crossing a threshold, p95 latency past the point where the UI has already given up, refusal rate moving sharply in either direction, and spend rate against the period budget. A refusal-rate jump usually means retrieval broke, not that the model changed.
Layer 6: define what happens when it fails
Every AI feature needs a specified behaviour for provider unavailability, timeouts, and budget exhaustion. "Spinner forever" is the default if you do not choose, and it is the worst option.
Streaming makes this concrete, because the connection can fail mid-answer. Two things are non-negotiable: the client must be able to cancel, and cancellation must actually stop the upstream work rather than leaving it running and billing.
async def stream_answer(request, pipeline):
async def generate():
try:
async for chunk in pipeline.stream(request.question):
if await request.is_disconnected():
break
yield chunk
except ModelUnavailable:
yield "\n\n[The assistant is temporarily unavailable. Your question was not lost.]"
except ModelBudgetExceeded:
yield "\n\n[This request was too large to process. Try narrowing the question.]"
return StreamingResponse(generate(), media_type="text/plain; charset=utf-8")The disconnect check inside the loop is what turns an abandoned tab into a stopped
generation. On the browser side, an AbortController on the fetch is the other half;
without it, closing a stream client-side leaves the server generating into a socket
nobody reads. The rendering side of this — streaming into a small interactive leaf
without blocking the rest of the page — is covered in
Server Components and streaming AI interfaces.
For non-streaming paths, decide between a cached previous answer, a degraded non-AI experience, and an honest error. All three are defensible. Silence is not.
The launch checklist
Before an AI feature goes in front of users:
- Every model call goes through one wrapper with a timeout, bounded retries and a typed error taxonomy.
- Per-request, per-account and per-feature budgets are enforced before the call.
- Tenant isolation is a database predicate, and an adversarial test proves it.
- An evaluation set with refusal cases runs in CI, and its output is stored per commit.
- Traces carry request id, account, tokens, latency and stop reason; content logging is flag-gated and short-retention.
- Every failure mode has a defined user-visible behaviour.
- Cancellation propagates from the browser to the provider.
- Untrusted content reaching the model cannot reach a tool with side effects — the reasoning behind that is in why prompt injection is an architecture problem.
None of this is novel engineering. It is the ordinary discipline of running a dependency you do not control, applied to one that is slower, pricier and less predictable than the ones you are used to. Teams that skip it do not avoid the work; they do it during an incident instead.