Lab 12 · Ship Your First Production LLM Endpoint

This is capstone lab 1 of 4 — the service you build here is the exact system labs 13, 14, and 15 extend (eval gate, tracing/cost, injection defense). Build it well; you will live in it.

Time & tier

Advanced · ~90 minutes hands-on.

Setup

Self-contained. You need Python 3.11+, an API key with a few dollars of credit, and:

mkdir capstone && cd capstone && python -m venv .venv && source .venv/bin/activate
pip install fastapi "uvicorn[standard]" anthropic pydantic httpx
export ANTHROPIC_API_KEY="sk-ant-..."   # or set up `ant auth login`

The feature: a support-ticket triage endpoint. POST /triage takes raw ticket text and returns a validated JSON classification (category, urgency, a one-line summary, whether it needs a human). Small enough to finish, real enough to hold every production concern.

Cost control for the whole capstone: the default model below is claude-opus-4-8. For lab runs where you're iterating on plumbing, not answer quality, claude-haiku-4-5 costs a fraction and is more than capable for this classification task — set MODEL once and it flows through every lab. Choosing the cheaper model is a deliberate cost decision; make it consciously.

Guided steps

1. 1.

   Define the contract with Pydantic — the schema is the API. Create schema.py:

   Copy

   from enum import Enum
   from pydantic import BaseModel, Field
   
   class Category(str, Enum):
       billing = "billing"
       technical = "technical"
       account = "account"
       other = "other"
   
   class Triage(BaseModel):
       category: Category
       urgency: int = Field(ge=1, le=5, description="1=low, 5=drop-everything")
       summary: str = Field(max_length=200)
       needs_human: bool
   

2. 2.

   Add a redaction step — untrusted user text should never reach the model with raw PII. Create redact.py:

   Copy

   import re
   PATTERNS = [
       (re.compile(r"[\w.+-]+@[\w-]+\.[\w.-]+"), "[EMAIL]"),
       (re.compile(r"\b(?:\d[ -]?){13,16}\b"), "[CARD]"),
       (re.compile(r"\b\d{3}[ -]?\d{3}[ -]?\d{4}\b"), "[PHONE]"),
   ]
   def redact(text: str) -> str:
       for pat, repl in PATTERNS:
           text = pat.sub(repl, text)
       return text
   

3. 3.

   Write the LLM call with structured output, timeout, retry, and fallback. Create llm.py:

   Copy

   import os, anthropic
   from schema import Triage
   from redact import redact
   
   MODEL = os.getenv("MODEL", "claude-opus-4-8")
   FALLBACK = os.getenv("FALLBACK_MODEL", "claude-haiku-4-5")
   
   # SDK auto-retries 429/5xx/timeouts; timeout is in SECONDS in Python.
   client = anthropic.Anthropic(timeout=20.0, max_retries=2)
   
   SYSTEM = (
       "You triage support tickets. Classify the ticket the user provides. "
       "Return only the structured fields requested."
   )
   
   def _call(model: str, ticket: str) -> Triage:
       # messages.parse() validates the response against the Pydantic model.
       resp = client.messages.parse(
           model=model,
           max_tokens=512,
           system=SYSTEM,
           messages=[{"role": "user", "content": redact(ticket)}],
           output_format=Triage,
       )
       if resp.parsed_output is None:  # refusal or non-conforming
           raise ValueError(f"no parsed output (stop_reason={resp.stop_reason})")
       return resp.parsed_output
   
   def triage(ticket: str) -> tuple[Triage, str]:
       try:
           return _call(MODEL, ticket), MODEL
       except (anthropic.APIError, ValueError):
           # Primary failed/refused/timed-out after retries — fall back once.
           return _call(FALLBACK, ticket), FALLBACK
   

   Note what's happening: messages.parse() uses structured outputs (output_config.format under the hood) so the response is guaranteed to match Triage or you get parsed_output is None. The SDK's own retry handles transient 429/5xx; your try/except is the fallback-model layer for when retries are exhausted or the primary refuses.

4. 4.

   Wrap it in FastAPI. Create main.py:

   Copy

   from fastapi import FastAPI, HTTPException
   from pydantic import BaseModel
   from llm import triage
   
   app = FastAPI()
   
   class TriageIn(BaseModel):
       text: str
   
   @app.post("/triage")
   def do_triage(body: TriageIn):
       if not body.text.strip():
           raise HTTPException(422, "empty ticket")
       result, model_used = triage(body.text)
       return {"triage": result.model_dump(), "model": model_used}
   

5. 5.

   Run and smoke-test.

   Copy

   uvicorn main:app --port 8000 &
   curl -s localhost:8000/triage -H 'content-type: application/json' \
     -d '{"text":"URGENT my card 4111 1111 1111 1111 was charged twice, email me at a@b.com"}' | python -m json.tool
   

   Confirm the response validates as Triage and — importantly — add a temporary print(redact(ticket)) to verify the card/email never left your process in the clear.

6. 6.

   Load-test it. Create loadtest.py — fire N concurrent requests, record p50/p99 latency, success rate, and how often the fallback fired:

   Copy

   import asyncio, time, httpx
   TICKETS = ["can't log in", "billing question about invoice 42",
              "the export button does nothing", "please close my account"]
   async def one(client, i):
       t = time.perf_counter()
       r = await client.post("http://localhost:8000/triage",
                             json={"text": TICKETS[i % len(TICKETS)]}, timeout=30)
       return time.perf_counter() - t, r.status_code, r.json().get("model")
   async def main(n=30):
       async with httpx.AsyncClient() as c:
           res = await asyncio.gather(*(one(c, i) for i in range(n)))
       lat = sorted(x[0] for x in res)
       ok = sum(1 for x in res if x[1] == 200)
       fb = sum(1 for x in res if x[2] and "haiku" in x[2])
       print(f"ok={ok}/{n} p50={lat[len(lat)//2]:.2f}s "
             f"p99={lat[int(len(lat)*0.99)]:.2f}s fallback_used={fb}")
   asyncio.run(main())
   

   Run python loadtest.py. Record the numbers — they are part of your artifact and your labs-13-15 baseline.

7. 7.

   Force a failure to prove the fallback works. Temporarily set MODEL=does-not-exist and re-run one curl; confirm you still get a valid triage from the fallback model, then restore MODEL.

8. 8.

   (OpenAI path, if you chose it.) Swap the SDK: pip install openai; client = OpenAI(timeout=20, max_retries=2); use client.chat.completions.parse(model="gpt-4o-mini", response_format=Triage, messages=[...]) (the SDK's structured-output helper) — the return exposes .choices[0].message.parsed. Everything else (redaction, fallback try/except, FastAPI, load test) is unchanged. Verify the exact method name and parsed accessor against current OpenAI docs, as that helper's surface has moved between beta.chat.completions.parse and the Responses API.

Verification

• •
  POST /triage returns JSON that validates against Triage for all four sample tickets.
• •
  The redaction check confirms email/card/phone are masked before the model call (verified by the temporary print, not by trusting the code).
• •
  Killing the primary model (step 7) still yields a valid response via the fallback, and model in the response names the fallback.
• •
  loadtest.py prints a p99 and a success rate; 30/30 succeed under concurrency.
• •
  An empty-text request returns 422, not a 500.

Completion artifact

The running capstone/ service (schema.py, redact.py, llm.py, main.py, loadtest.py) plus a BASELINE.md recording model used, p50/p99, success rate, and fallback-fired count. Keep this folder — labs 13–15 build directly on it. Self-grade (5 criteria = pass at 4+):

1. 1.
   Structured output is schema-enforced (a malformed model response surfaces as a handled error, not a crash).
2. 2.
   PII is redacted before egress, verified by observation.
3. 3.
   Timeout + SDK retry + one fallback model are all present and the fallback is proven to fire.
4. 4.
   Load-test evidence exists with real p99 and success numbers.
5. 5.
   Input validation rejects empty/garbage without a 500.

Stuck?

• messages.parse / output_format not found: update the SDK (pip install -U anthropic); structured outputs need a recent version. As a fallback use messages.create(..., output_config={"format": {"type": "json_schema", "schema": Triage.model_json_schema()}}) and json.loads the first text block.
• parsed_output is None: check resp.stop_reason — refusal or max_tokens both produce this; bump max_tokens or inspect the input.
• Fallback never fires in step 7: a bad model ID raises NotFoundError (an APIError subclass) — make sure your except catches anthropic.APIError, not just ValueError.
• Load test all timeouts: the dev server is single-worker; run uvicorn main:app --workers 4 for the load test.

Client-safe variant

Client engagements that forbid sending code/data to public LLM APIs: point the SDK at a client-sanctioned endpoint — Amazon Bedrock (AnthropicBedrockMantle, model IDs prefixed anthropic.) or Azure OpenAI inside the client tenancy — with no code change beyond the client constructor and model string. The redaction step becomes more important, not less: make it the client's actual PII policy (add national-ID and account-number patterns per DPDP/GDPR), and treat "can this ticket text leave our network?" as a design decision recorded in BASELINE.md. If even a hosted enterprise endpoint is disallowed, the same service runs against a local model (Ollama + a Qwen-class model) via an OpenAI-compatible base URL — quality drops but the production scaffolding you're being graded on is identical.