Project tutorials · 8 steps

Ship a production AI service in Python

API keys, rate limits, validation, caching, model fallbacks, structured logs and an eval, around a service that classifies support tickets.

What you'll build

A model call is one line. The rest of a production AI service is everything around it: who's allowed in, what happens when the model returns nonsense or the provider is down, and how you know it still works after a change. This tutorial builds a support ticket classifier with all of that, using only the standard library.

What you'll learn

Combine authentication, rate limiting, validation, caching, fallbacks and logging in one service
Treat model output as untrusted: parse it, check it against allowed values, and fall back
Log every request in a structured way, without keys or customer text
Measure the service with an eval set before and after each change

What you need

Python classes and what an API is. It uses only the standard library, and the demo uses a fake model, so it costs nothing and gives the same result every time. To go live, pass in functions that call your real models.

Step 1: Set the allowed answers and a last resort

The allowed categories and priorities are checked in code, because a model's output can't be trusted to stay within them. rules is a simple keyword fallback for when every model fails. It isn't clever, but a rough answer beats a 500 error when a provider is down. PROMPT_VERSION will matter for the cache.

import hashlib, json, math

CATEGORIES = {"billing", "bug", "account", "feature", "other"}
PRIORITIES = {"low", "normal", "high", "urgent"}
PROMPT_VERSION = "triage-v3"

def rules(text):
    t = text.lower()
    category = "billing" if any(w in t for w in ("refund", "charge", "invoice")) else "other"
    priority = "urgent" if any(w in t for w in ("down", "outage", "can't log in")) else "normal"
    return {"category": category, "priority": priority}

Step 2: Set up the service and its log

Models, keys and even the clock are passed in, so tests can use a fake model and a fixed time. Every response goes through _respond, so every request gets exactly one log entry. Only the first six characters of a key are logged, and no customer text, because logs get copied around and read by many people.

class TriageService:
    def __init__(self, models, keys, clock, window=60):
        self.models = models
        self.keys = keys
        self.clock = clock
        self.window = window
        self.cache = {}
        self.seen = {}
        self.log = []

    def _respond(self, request, status, body, source=None, cached=False):
        key = request.get("api_key") or ""
        self.log.append({"request_id": request.get("request_id"), "key": key[:6] + "...", "status": status,
                         "source": source, "cached": cached})
        return status, body

Step 3: Limit each key's requests

A sliding window: keep each key's recent request times and drop any older than the window. When a key is over its limit, the 429 says when to try again, so a well-behaved client knows how long to wait. Each key has its own limit, and one busy customer can't use up everyone's capacity. A rejected request isn't counted against the window.

    def _rate_limited(self, key):
        now = self.clock()
        recent = [t for t in self.seen.get(key, []) if now - t < self.window]
        if len(recent) >= self.keys[key]:
            self.seen[key] = recent
            return math.ceil(recent[0] + self.window - now)
        self.seen[key] = recent + [now]
        return None

Rather write it yourself?

In the lesson you write this from a starter, in your browser with nothing to install, and checks tell you what's wrong as you go. A free account opens it and all 244 lessons.

Create a free account

Step 4: Validate before you spend money

Check the request before any model sees it. The 5,000-character limit protects your bill, since someone could otherwise send a huge document on every call.

    def _problem(self, body):
        if not isinstance(body, dict) or not isinstance(body.get("ticket_id"), str) or not body["ticket_id"]:
            return "ticket_id is required"
        text = body.get("text")
        if not isinstance(text, str) or not text.strip() or len(text) > 5000:
            return "text must be 1 to 5000 characters"
        return None

Step 5: Treat the model's output as untrusted

This is the core of a reliable AI service. Try each model in order. Parse the reply as JSON, and check it against the allowed values. Any exception or wrong value means moving on to the next model, and if they all fail, the rules answer. The response records which source answered, so you can see in the logs when you're falling back.

    def _classify(self, text):
        for name, fn in self.models:
            try:
                out = json.loads(fn(text))
            except Exception:
                continue
            if isinstance(out, dict) and out.get("category") in CATEGORIES and out.get("priority") in PRIORITIES:
                return {"category": out["category"], "priority": out["priority"], "source": name}
        return {**rules(text), "source": "rules"}

Step 6: Put the request flow together

The order matters: authenticate, then rate limit, then validate, then spend money. The cache key includes the prompt version, so changing the prompt can't serve old answers. Rule-based answers aren't cached, so the service goes back to using a real model as soon as one recovers.

    def handle(self, request):
        key = request.get("api_key")
        if key not in self.keys:
            return self._respond(request, 401, {"error": "invalid_api_key"})
        wait = self._rate_limited(key)
        if wait is not None:
            return self._respond(request, 429, {"error": "rate_limited", "retry_after": wait})
        body = request.get("body")
        problem = self._problem(body)
        if problem:
            return self._respond(request, 400, {"error": "invalid_request", "detail": problem})
        cache_key = hashlib.sha256((PROMPT_VERSION + "\n" + body["text"]).encode()).hexdigest()
        cached = cache_key in self.cache
        result = self.cache[cache_key] if cached else self._classify(body["text"])
        if not cached and result["source"] != "rules":
            self.cache[cache_key] = result
        return self._respond(request, 200, {"ticket_id": body["ticket_id"], **result, "cached": cached},
                             source=result["source"], cached=cached)

    def health(self):
        return {"status": "ok", "cache_size": len(self.cache)}

Step 7: Measure it with an eval

Run labeled tickets through the real service, including its fallbacks, and report accuracy for each label along with the ids that failed. Run this before every change to a prompt or a model. Without it, you only find out from customers.

def evaluate(service, cases, api_key):
    cat = pri = 0
    failures = []
    for case in cases:
        status, body = service.handle({"api_key": api_key, "request_id": f"eval-{case['id']}",
                                        "body": {"ticket_id": case["id"], "text": case["text"]}})
        ok_cat = status == 200 and body["category"] == case["category"]
        ok_pri = status == 200 and body["priority"] == case["priority"]
        cat += ok_cat
        pri += ok_pri
        if not (ok_cat and ok_pri):
            failures.append(case["id"])
    n = len(cases)
    return {"category_accuracy": round(cat / n, 3), "priority_accuracy": round(pri / n, 3), "failures": sorted(failures)}

Step 8: Try it

The same ticket twice: the second answer comes from the cache and costs nothing. The log shows both requests, with the key shortened and no customer text.

def fake_model(text):
    return json.dumps({"category": "bug", "priority": "high"})

service = TriageService([("fast-model", fake_model)], {"sk-demo-key-1": 10}, clock=lambda: 0)
request = {"api_key": "sk-demo-key-1", "request_id": "req-1",
           "body": {"ticket_id": "T-100", "text": "The export button does nothing"}}
print(service.handle(request))
print(service.handle({**request, "request_id": "req-2"}))
print(service.log)

What it prints

(200, {'ticket_id': 'T-100', 'category': 'bug', 'priority': 'high', 'source': 'fast-model', 'cached': False})
(200, {'ticket_id': 'T-100', 'category': 'bug', 'priority': 'high', 'source': 'fast-model', 'cached': True})
[{'request_id': 'req-1', 'key': 'sk-dem...', 'status': 200, 'source': 'fast-model', 'cached': False}, {'request_id': 'req-2', 'key': 'sk-dem...', 'status': 200, 'source': 'fast-model', 'cached': True}]

Where to take it next

Put it behind a web framework, and keep the cache and rate limits in something shared like Redis once you run more than one process.
Add a timeout to every model call, so a slow provider falls through to the next.
Alert on the share of answers coming from rules. That's your early warning that a model is failing.

Mistakes to watch for

Trusting that a model will always return valid JSON.
Caching fallback answers as if they were model answers.
Logging full API keys or customer messages for debugging.
Validating input before checking who the caller is.

The complete program

import hashlib, json, math

CATEGORIES = {"billing", "bug", "account", "feature", "other"}
PRIORITIES = {"low", "normal", "high", "urgent"}
PROMPT_VERSION = "triage-v3"

def rules(text):
    t = text.lower()
    category = "billing" if any(w in t for w in ("refund", "charge", "invoice")) else "other"
    priority = "urgent" if any(w in t for w in ("down", "outage", "can't log in")) else "normal"
    return {"category": category, "priority": priority}

class TriageService:
    def __init__(self, models, keys, clock, window=60):
        self.models = models
        self.keys = keys
        self.clock = clock
        self.window = window
        self.cache = {}
        self.seen = {}
        self.log = []

    def _respond(self, request, status, body, source=None, cached=False):
        key = request.get("api_key") or ""
        self.log.append({"request_id": request.get("request_id"), "key": key[:6] + "...", "status": status,
                         "source": source, "cached": cached})
        return status, body

    def _rate_limited(self, key):
        now = self.clock()
        recent = [t for t in self.seen.get(key, []) if now - t < self.window]
        if len(recent) >= self.keys[key]:
            self.seen[key] = recent
            return math.ceil(recent[0] + self.window - now)
        self.seen[key] = recent + [now]
        return None

    def _problem(self, body):
        if not isinstance(body, dict) or not isinstance(body.get("ticket_id"), str) or not body["ticket_id"]:
            return "ticket_id is required"
        text = body.get("text")
        if not isinstance(text, str) or not text.strip() or len(text) > 5000:
            return "text must be 1 to 5000 characters"
        return None

    def _classify(self, text):
        for name, fn in self.models:
            try:
                out = json.loads(fn(text))
            except Exception:
                continue
            if isinstance(out, dict) and out.get("category") in CATEGORIES and out.get("priority") in PRIORITIES:
                return {"category": out["category"], "priority": out["priority"], "source": name}
        return {**rules(text), "source": "rules"}

    def handle(self, request):
        key = request.get("api_key")
        if key not in self.keys:
            return self._respond(request, 401, {"error": "invalid_api_key"})
        wait = self._rate_limited(key)
        if wait is not None:
            return self._respond(request, 429, {"error": "rate_limited", "retry_after": wait})
        body = request.get("body")
        problem = self._problem(body)
        if problem:
            return self._respond(request, 400, {"error": "invalid_request", "detail": problem})
        cache_key = hashlib.sha256((PROMPT_VERSION + "\n" + body["text"]).encode()).hexdigest()
        cached = cache_key in self.cache
        result = self.cache[cache_key] if cached else self._classify(body["text"])
        if not cached and result["source"] != "rules":
            self.cache[cache_key] = result
        return self._respond(request, 200, {"ticket_id": body["ticket_id"], **result, "cached": cached},
                             source=result["source"], cached=cached)

    def health(self):
        return {"status": "ok", "cache_size": len(self.cache)}

def evaluate(service, cases, api_key):
    cat = pri = 0
    failures = []
    for case in cases:
        status, body = service.handle({"api_key": api_key, "request_id": f"eval-{case['id']}",
                                        "body": {"ticket_id": case["id"], "text": case["text"]}})
        ok_cat = status == 200 and body["category"] == case["category"]
        ok_pri = status == 200 and body["priority"] == case["priority"]
        cat += ok_cat
        pri += ok_pri
        if not (ok_cat and ok_pri):
            failures.append(case["id"])
    n = len(cases)
    return {"category_accuracy": round(cat / n, 3), "priority_accuracy": round(pri / n, 3), "failures": sorted(failures)}


def fake_model(text):
    return json.dumps({"category": "bug", "priority": "high"})

service = TriageService([("fast-model", fake_model)], {"sk-demo-key-1": 10}, clock=lambda: 0)
request = {"api_key": "sk-demo-key-1", "request_id": "req-1",
           "body": {"ticket_id": "T-100", "text": "The export button does nothing"}}
print(service.handle(request))
print(service.handle({**request, "request_id": "req-2"}))
print(service.log)

Write it yourself

In the lesson you write this from a starter, in your browser with nothing to install, and checks tell you what's wrong as you go. A free account opens it and all 244 lessons.

Create a free account