Production NLP APIs, Batching, Caching and Drift Monitoring

Design the service around failure

An API should validate input size, expose a stable response shape and reject overload clearly. Batch requests improve throughput when one model call can process several texts, but they need item and token limits. Cache only safe, repeatable results and include model version and normalized input in the key.

Monitor request count, latency, errors, input distribution, output distribution and resource use. Labels often arrive late, so drift is an early warning, not proof that accuracy fell. Link online alerts to a reviewed sample and later outcome metrics.

Rollback must restore a compatible bundle: model, tokenizer, preprocessing, label map and configuration. Keep the previous artifact available and rehearse the command before launch.

Run a measured FastAPI service

Save as service.py. The deterministic classifier keeps the lab small so the production controls are visible. Replace predict_one with a pinned model after the API tests pass.

python -m pip install fastapi "uvicorn[standard]" prometheus-client
$env:MODEL_VERSION="baseline-2"
uvicorn service:app --host 127.0.0.1 --port 8000

# service.py
import os
from collections import deque
from functools import lru_cache
from statistics import mean
from time import perf_counter
from fastapi import FastAPI, HTTPException, Response
from pydantic import BaseModel, Field
from prometheus_client import Counter, Histogram, generate_latest, CONTENT_TYPE_LATEST

ALLOWED_MODELS = {"baseline-1", "baseline-2"}
MODEL_VERSION = os.getenv("MODEL_VERSION", "baseline-2")
if MODEL_VERSION not in ALLOWED_MODELS:
    raise RuntimeError("unapproved model version")

BASELINE_MEAN_TOKENS = 8.0
recent_lengths = deque(maxlen=200)
requests_total = Counter("nlp_requests_total", "NLP requests", ["endpoint"])
latency = Histogram("nlp_request_seconds", "NLP request latency", ["endpoint"])
app = FastAPI(title="Versioned NLP API", version="1.0")

class TextItem(BaseModel):
    text: str = Field(min_length=1, max_length=1000)

class Batch(BaseModel):
    items: list[TextItem] = Field(min_length=1, max_length=32)

@lru_cache(maxsize=512)
def predict_one(model_version, normalized_text):
    refund_words = {"refund", "return", "wapas", "واپس"}
    words = set(normalized_text.lower().split())
    label = "refund" if words & refund_words else "other"
    return {"label": label, "model_version": model_version}

def run(items, endpoint):
    started = perf_counter()
    requests_total.labels(endpoint).inc()
    outputs = []
    for item in items:
        normalized = " ".join(item.text.split())
        length = len(normalized.split())
        recent_lengths.append(length)
        outputs.append(predict_one(MODEL_VERSION, normalized))
    latency.labels(endpoint).observe(perf_counter() - started)
    return outputs

@app.post("/predict")
def predict(item: TextItem):
    return run([item], "single")[0]

@app.post("/predict-batch")
def predict_batch(batch: Batch):
    total_tokens = sum(len(x.text.split()) for x in batch.items)
    if total_tokens > 4000:
        raise HTTPException(413, "batch token budget exceeded")
    return {"results": run(batch.items, "batch"), "token_proxy": total_tokens}

@app.get("/health")
def health():
    current = mean(recent_lengths) if recent_lengths else BASELINE_MEAN_TOKENS
    drift = abs(current - BASELINE_MEAN_TOKENS) / BASELINE_MEAN_TOKENS
    return {"model_version": MODEL_VERSION, "mean_tokens": round(current, 2),
            "input_length_drift": round(drift, 3), "review_required": drift > 0.30}

@app.get("/metrics")
def metrics():
    return Response(generate_latest(), media_type=CONTENT_TYPE_LATEST)

Exercise the endpoints

curl -X POST http://127.0.0.1:8000/predict -H "Content-Type: application/json" -d "{\"text\":\"payment wapas please\"}"
curl -X POST http://127.0.0.1:8000/predict-batch -H "Content-Type: application/json" -d "{\"items\":[{\"text\":\"refund please\"},{\"text\":\"where is delivery\"}]}"
curl http://127.0.0.1:8000/health
curl http://127.0.0.1:8000/metrics

Expected output and service checks

The responses include the model version. Repeated normalized input uses the cache. The health endpoint flags a large change in mean input length, while Prometheus exposes traffic and latency. The token proxy is an explainable workload signal, not a currency estimate.

Batching, cost and privacy decisions

Batch only requests that share a compatible model and privacy boundary. Do not place different customers’ raw text into a shared diagnostic record. Set a maximum wait time so low traffic does not create long delays. Measure throughput at batch sizes 1, 4, 8 and 16 on the target machine, then choose the smallest batch that meets both latency and capacity goals.

Track workload using input tokens or characters, model time and hardware time. Convert those measurements into money only from an actual hosting bill or provider price. A cache can reduce computation, but sensitive inputs may require a tenant-specific key, encryption and a short retention period, or no shared cache at all.

For drift, compare the current window with the approved baseline by language, length and label distribution. Alerting should open an investigation. It should not retrain or promote a model automatically.

Release and rollback procedure

1. Run offline quality, security and load gates against the candidate bundle.
2. Send a small canary share to the candidate and compare live signals.
3. Stop expansion when error, latency or reviewed-quality thresholds fail.
4. Set MODEL_VERSION=baseline-1 and restart from the previous immutable image.
5. Verify health, prediction shape and a known fixture after rollback.

Do not overwrite an artifact under the same version. A rollback that changes only model weights while leaving an incompatible tokenizer live is not a rollback.

Operational blind spots

Knowledge check

Operations references

• FastAPI documentation
• Prometheus instrumentation practices
• Python logging guide
• Model cards