Add a Batch Path and Measure What It Costs

The endpoint from the last three lessons validates one record at a time. The LoanRecord schema rejects a malformed annual_inc, the leakage-safe ServingRequest schema excludes every post-outcome column, and the endpoint returns a structured PredictionResponse the caller can build on. That is the right shape for a checkout service scoring one loan application the instant a borrower submits it. It is the wrong shape for the caller who arrives with a million records at once.

I learned that on a nightly churn job. The real-time scoring service handled live loan scoring without complaint, so when a batch job needed to score the entire subscriber base every night, it did the obvious thing and called /predict once per row in a loop. The run took most of a workday and hammered the service the whole time. The same model, scored in batched calls, would have finished in minutes. Nothing was wrong with the model. The access pattern was the bottleneck, and the endpoint had no batch path to offer the caller a better one. This lesson builds that path: a route that scores a list in one request, keeps its output index-aligned with its input, and reports a latency number measured at the boundary instead of guessed from the model call.

Real-time and batch optimize opposite quantities

A reasonable assumption, holding the one-record endpoint, is that a batch is the same endpoint called more times: scoring a thousand records is a thousand /predict requests, and the only question is whether they go out in a loop or in parallel. That assumption is what produced the workday-long churn run. It treats the two access patterns as the same operation at different volumes, when they optimize opposite quantities.

Real-time and batch are different operations, not the same one at different scale

Real-time (online) inference and batch inference optimize opposite quantities, so a workload built for one is mis-served by the other. Online inference processes a request the moment it arrives and minimizes the latency of that single request under a tight deadline, because a borrower is waiting. Batch inference processes many records in bulk and maximizes throughput across the whole set; no single record is waiting, so latency per record can be loose as long as the total finishes in the window. Designing ML Systems draws the same line between batch prediction that precomputes in bulk and online prediction that generates one result per input as it arrives. The cost axis is the giveaway: as a general rule, shorter latency equals higher cost, so the real-time path buys low latency by paying more per record, and the batch path trades latency away to drive cost per record down.

The reason a loop of single calls is slow is not the model. Every single-record call carries fixed overhead that is independent of the model: the HTTP framing, the JSON parse of the request body, the pydantic validation at the boundary, and the Python-level dispatch into the model. That overhead is paid once per request. For a model whose predict is vectorized (scikit-learn’s is), predicting one row and predicting a thousand rows cost almost the same inside the model, because it is one matrix operation either way. NumPy is built for efficiency on large arrays of data, and a vectorized array operation runs far faster than the equivalent element-wise Python loop. So a loop of single calls multiplies the fixed per-request overhead by the row count while the model work stays flat; a batched call pays that overhead once and lets the vectorized predict absorb the rows for free. That is the entire gap between a run that takes hours and a run that takes minutes: the model is identical, only the count of times the fixed overhead got paid changed.

The number that matters here is the multiplier, not the absolute time, so read it as overhead × rows. The wrong mental model is “the loop is slow because there are a lot of rows.” The rows are nearly free inside a vectorized predict. The loop is slow because each iteration re-pays the framing, parse, and validation that a batch pays once. The block below scores the same records two ways, a Python loop of single calls versus one vectorized call on the whole array, and prints the wall-clock time for each. Watch the gap: it is the per-call overhead, multiplied.

import time

import numpy as np

rng = np.random.default_rng(0)
# 2,000 member records, 12 features each -- the KKBox churn shape.
records: list[list[float]] = rng.normal(size=(2_000, 12)).tolist()


def predict_one(record: list[float]) -> float:
    """Stand-in for the per-request path: build a 1-row array, score it.

    The np.asarray + reshape is the per-call cost a real handler also pays
    as JSON parse + pydantic validation + framework dispatch.
    """
    x = np.asarray(record, dtype=float).reshape(1, -1)
    return float((1.0 / (1.0 + np.exp(-x.sum(axis=1))))[0])


def predict_batch(batch: list[list[float]]) -> list[float]:
    """One vectorized pass over the whole 2-D array -- overhead paid once."""
    x = np.asarray(batch, dtype=float)
    return (1.0 / (1.0 + np.exp(-x.sum(axis=1)))).tolist()


start = time.perf_counter()
loop_scores = [predict_one(r) for r in records]
loop_s = time.perf_counter() - start

start = time.perf_counter()
batch_scores = predict_batch(records)
batch_s = time.perf_counter() - start

print("rows scored:        " + str(len(records)))
print("loop of single:     " + format(loop_s * 1000, ".1f") + " ms")
print("one batched call:   " + format(batch_s * 1000, ".1f") + " ms")
print("speedup:            " + format(loop_s / batch_s, ".0f") + "x")
print("same scores?        " + str(np.allclose(loop_scores, batch_scores)))

The two paths compute identical scores (np.allclose confirms it) yet the batched call is many times faster, and the multiplier grows with the row count because the loop re-pays its per-row overhead every iteration while the batch does not. Why does the speedup climb as the row count rises rather than staying fixed? Because the batch’s single fixed cost is amortized across more rows, so the per-row overhead in the batch path falls while the loop’s per-row overhead is constant. That is the crossover: it lives in the access pattern, not the row count. Looping single calls over even a few hundred thousand rows costs hours; one batched call over the same set costs minutes, the same model and same arithmetic, with a different number of times the fixed overhead was paid.

This is a decision the API forces on the designer, and the wrong default is to offer only the real-time route. A serving app with a single /predict endpoint silently tells every caller “score one record per request,” so the batch caller has no path except the loop. The fix is not a faster model; it is a second route that accepts the whole list, scores it vectorized, and pays the overhead once.

import time

import numpy as np

rng = np.random.default_rng(1)
frame: list[list[float]] = rng.normal(size=(3_000, 8)).tolist()


def score_one(row: list[float]) -> float:
    x = np.asarray(row, dtype=float).reshape(1, -1)
    return float(x.mean())


# Per-row loop -- the slow path.
start = time.perf_counter()
loop_out = [score_one(r) for r in frame]
loop_s = time.perf_counter() - start


# TODO: write score_batch(frame) -> list[float] as ONE vectorized call,
#       time it the same way, and print the speedup.
def score_batch(batch: list[list[float]]) -> list[float]:
    return [0.0]  # placeholder -- replace with one vectorized pass


batch_out = score_batch(frame)
print("loop ms: " + format(loop_s * 1000, ".1f"))
print("TODO: time score_batch and print the speedup")

import time

import numpy as np

rng = np.random.default_rng(1)
frame: list[list[float]] = rng.normal(size=(3_000, 8)).tolist()


def score_one(row: list[float]) -> float:
    x = np.asarray(row, dtype=float).reshape(1, -1)
    return float(x.mean())


def score_batch(batch: list[list[float]]) -> list[float]:
    x = np.asarray(batch, dtype=float)
    return x.mean(axis=1).tolist()


start = time.perf_counter()
loop_out = [score_one(r) for r in frame]
loop_s = time.perf_counter() - start

start = time.perf_counter()
batch_out = score_batch(frame)
batch_s = time.perf_counter() - start

print("loop ms:    " + format(loop_s * 1000, ".1f"))
print("batch ms:   " + format(batch_s * 1000, ".1f"))
print("speedup:    " + format(loop_s / batch_s, ".0f") + "x")
print("same out?   " + str(np.allclose(loop_out, batch_out)))

A batch route must keep output aligned with input

With a batch route in hand, the next assumption a careful engineer makes is the dangerous one: that the response is “a list of predictions” and the work is done. A batch endpoint takes a list of records and returns a list of predictions, but which prediction belongs to which record is carried by nothing except position. Get that wrong and every test still passes, because the failure only appears on data the tests did not use.

Output position is the only thing tying a prediction to its record

A batch response is a positional list: the only thing tying output i to input i is that the two lists share length and order, and nothing in JSON enforces that, so alignment is a correctness invariant maintained by construction, not a property that comes for free. A batch response carries no keys back to the inputs. ML Design Patterns names the canonical fix for this, Keyed Predictions (carrying a key through the prediction), precisely because batch and distributed serving do not otherwise guarantee the output order matches the input order. The vectorized model supplies in-order alignment as a starting point: a model.predict(X) over a 2-D array returns one output row per input row, in input order, the way Hands-On ML describes the output matrix having one row per instance matching the input matrix’s one row per instance. Alignment holds by construction as long as X is built by stacking the validated records in arrival order and no row is dropped, reordered, or deduplicated.

Every “helpful” transformation breaks that invariant, and each breaks it silently. The one that feels most responsible is the worst: validate each record and skip the bad ones. Skipping an invalid record shortens the output by one and shifts every later index by one, so the caller’s positional zip(inputs, outputs) attributes every prediction after the dropped row to the wrong member. The block below does exactly that, dropping the one record that fails validation and returning a shorter list, and then shows what the caller sees when they line the two lists up by position.

def score_batch_buggy(records: list[dict]) -> list[float]:
    out = []
    for r in records:
        if r["annual_inc"] < 0:        # "helpfully" skip the bad row
            continue                   # <-- shortens output, shifts every later index
        out.append(score(r))
    return out                          # len(out) < len(records) when any row is bad


records = [
    {"id": "A", "annual_inc": 50_000},
    {"id": "B", "annual_inc": -1},      # invalid — gets dropped
    {"id": "C", "annual_inc": 80_000},
]
scores = score_batch_buggy(records)     # length 2, not 3
for rec, s in zip(records, scores):     # A->scoreA, B->scoreC (!), C->nothing
    print(rec["id"], s)                 # B is now wearing C's score

The output has length two, the input has length three, and zip stops at the shorter one, so member B is reported with member C’s score and member C drops off the end entirely. This passed every test the author had, because the tests used all-valid inputs: the lengths always matched, and the misattribution only appeared once real data carried a malformed row. That is the silent index-shift failure, and it is the same mis-attribution-on-reorder bug the data-wrangling module (M2) drilled, now reopened at the API request/response contract.

The correctness-preserving design keeps the list dense, one output slot per input slot always, and represents a per-record failure as an error object in that record’s slot rather than as an absent element. Architecture Patterns with Python frames the intent: a clean error boundary keeps the system from ending in an inconsistent state, and a dropped element is exactly the inconsistency here. The block below scores each record, and when one fails it writes an error object at that position instead of omitting it, so length and order stay invariant no matter how many records are bad. Watch the output length: it equals the input length regardless of how many rows failed.

import numpy as np


def score(record: dict) -> float:
    x = np.array([record["annual_inc"], record["loan_amnt"]], dtype=float)
    return float(1.0 / (1.0 + np.exp(-(x[0] / 1e5 - x[1] / 1e5))))


def score_batch_aligned(records: list[dict]) -> list[dict]:
    """One result slot per input slot -- a failure becomes an error object in place."""
    out: list[dict] = []
    for r in records:
        try:
            if r["annual_inc"] < 0:
                raise ValueError("annual_inc must be non-negative")
            out.append({"probability": score(r), "error": None})
        except (KeyError, ValueError) as exc:
            out.append({"probability": None, "error": str(exc)})  # in the SAME slot
    return out


records = [
    {"id": "A", "annual_inc": 50_000, "loan_amnt": 10_000},
    {"id": "B", "annual_inc": -1, "loan_amnt": 5_000},  # invalid
    {"id": "C", "annual_inc": 80_000, "loan_amnt": 20_000},
]
results = score_batch_aligned(records)

print("input len:  " + str(len(records)))
print("output len: " + str(len(results)))
print("aligned?    " + str(len(records) == len(results)))
for rec, res in zip(records, results):
    print(rec["id"] + " -> " + str(res))

The output length equals the input length, B carries its own error in its own slot, and C still lines up with C: the positional contract holds even though a record failed. Why is an error object in the slot the only representation that preserves the invariant, when returning a shorter list or a separate “errors” list both seem reasonable? Because any representation that changes the length or order forces the caller to re-derive the mapping, and the caller has nothing to re-derive it from: the slot is the key. This is also the seam where the deterministic-transform discipline from the data-wrangling module (M2) attaches: the same feature transform must produce the identical row whether that row arrives alone or inside a batch. A transform that re-derives one-hot columns per request will produce a different row shape single-versus-batch; a frozen encoder produces the same row either way. The invariant is therefore not only length and order; it is that the per-row computation is independent of batch membership, which is why the serving-contract test must exercise both the single-row path and the batch path, not batches alone.

The full FastAPI batch handler puts those pieces together: it accepts list[MemberRecord] (the analogue of the ServingRequest/LoanRecord schema from the prior lessons, here scoring subscribers from the KKBox churn dataset — the larger, where-scale-is-the-lesson dataset the track introduces alongside the loan book — because batch scoring is where a member base, not one loan, arrives at once), scores them, and returns list[PredictionResponse] of the same length, with the length invariant asserted in the response so a desync fails loudly instead of silently. The handler below is declared with plain def, which matters: FastAPI runs a plain-def path operation in an external threadpool that is then awaited, instead of calling it directly as it would for an async def, so the blocking vectorized predict inside it does not stall the event loop. The scaling consequence of that choice is the next lesson; here it is enough that the batch handler does not freeze the server.

import numpy as np
from pydantic import BaseModel


class MemberRecord(BaseModel):
    member_id: str
    annual_inc: float
    loan_amnt: float


class PredictionResponse(BaseModel):
    member_id: str
    probability: float | None
    error: str | None = None


def _score(rec: MemberRecord) -> float:
    x = np.array([rec.annual_inc / 1e5, rec.loan_amnt / 1e5])
    return float(1.0 / (1.0 + np.exp(-(x[0] - x[1]))))


def predict_batch(records: list[MemberRecord]) -> list[PredictionResponse]:
    """Dense list: out[i] corresponds to records[i], same length, same order."""
    out: list[PredictionResponse] = []
    for rec in records:
        try:
            if rec.annual_inc < 0:
                raise ValueError("annual_inc must be non-negative")
            out.append(
                PredictionResponse(member_id=rec.member_id, probability=_score(rec))
            )
        except ValueError as exc:
            out.append(
                PredictionResponse(
                    member_id=rec.member_id, probability=None, error=str(exc)
                )
            )
    assert len(out) == len(records), "batch desync: output length != input length"
    return out


batch = [
    MemberRecord(member_id="m1", annual_inc=60_000, loan_amnt=12_000),
    MemberRecord(member_id="m2", annual_inc=-1, loan_amnt=8_000),
    MemberRecord(member_id="m3", annual_inc=95_000, loan_amnt=30_000),
]
for resp in predict_batch(batch):
    print(resp.model_dump())
print("length invariant held: " + str(len(predict_batch(batch)) == len(batch)))

Every input id appears in the output at the same position, the failed record carries its error in place, and the assert makes a desync a loud failure instead of a misattribution the caller discovers in production. The member_id echoed in each PredictionResponse is belt-and-suspenders: it lets a caller cross-check the positional alignment, which is the keyed-prediction defense for the case where a future code path does reorder.

import numpy as np


def transform_buggy(rows: list[dict]) -> np.ndarray:
    # Re-derives one-hot columns from THIS call's grades only -- batch-dependent.
    grades = sorted({r["grade"] for r in rows})
    out = []
    for r in rows:
        onehot = [1.0 if r["grade"] == g else 0.0 for g in grades]
        out.append([r["annual_inc"] / 1e5] + onehot)
    return np.array(out, dtype=float)


one = [{"grade": "B", "annual_inc": 50_000}]
batch = [
    {"grade": "A", "annual_inc": 40_000},
    {"grade": "B", "annual_inc": 50_000},
    {"grade": "C", "annual_inc": 60_000},
]

# TODO: pull row "B" out of the single-row transform and out of the batch transform,
#       then assert the two feature rows are equal. It should FAIL on transform_buggy.
single_row = transform_buggy(one)[0]
batch_row = None  # TODO: pull the matching "B" row out of transform_buggy(batch)
print("single:", single_row)
print("batch:", batch_row)

import numpy as np

GRADES = ["A", "B", "C", "D", "E", "F", "G"]  # frozen, known ahead of time


def transform(rows: list[dict]) -> np.ndarray:
    out = []
    for r in rows:
        onehot = [1.0 if r["grade"] == g else 0.0 for g in GRADES]
        out.append([r["annual_inc"] / 1e5] + onehot)
    return np.array(out, dtype=float)


one = [{"grade": "B", "annual_inc": 50_000}]
batch = [
    {"grade": "A", "annual_inc": 40_000},
    {"grade": "B", "annual_inc": 50_000},
    {"grade": "C", "annual_inc": 60_000},
]

single_row = transform(one)[0]
batch_row = transform(batch)[1]  # "B" is at index 1 in the batch

assert np.array_equal(single_row, batch_row), "row B differs single vs batch"
print("single B:", single_row.tolist())
print("batch  B:", batch_row.tolist())
print("identical single vs batch: " + str(np.array_equal(single_row, batch_row)))

The timeline below makes the alignment invariant concrete: the same five records flowing into the batch handler, one of them failing validation, and the output list staying dense so position i always points back to input i.

The dense output is the whole defense: out[i] points back to input i whether or not record i failed, so the caller’s positional zip never silently misattributes a score. With alignment guaranteed by construction, the one number the route still owes the caller is its cost, which is where timing it correctly comes in.

Measure latency at the boundary, not at model.predict()

With a batch path that stays aligned, the last question is the one the rubric demands an answer to: what does a request actually cost? The intuitive instinct is to time model.predict(), since that is the model and that is the work, so that is the latency. Predict the result of that instinct before reading on: an engineer instruments the model call on the small Lending Club scorer, sees a number well under a millisecond, and reports the endpoint as sub-millisecond. The caller then measures the round trip and sees something many times larger. The reported number was honest and useless.

The model call is usually the smallest part of a small model’s latency

Latency is the request’s total response time as the caller experiences it, request in to response out, so it must be measured at the endpoint boundary, not at model.predict(), because for a fast model the matrix multiply is often the smallest part of the response time. Reliable ML defines prediction latency as the time between making a request and getting an answer back, and locates it at the caller boundary, due to network effects and overall system load, not at one internal call. The work around the model call is real and on the critical path: JSON deserialization of the request, pydantic validation at the boundary, JSON serialization of the response, and ASGI framework dispatch. Each of these can rival or exceed the compute of a small model. Observability Engineering gives the diagnostic instinct for the same reason (on a latency spike, start at the edge, group by endpoint, compute the average and the 90th and 99th percentiles, and trace one slow request) because the edge is where the caller’s real latency lives.

The mechanism behind the surprise is that the layers around the model do not shrink when the model is fast; they are fixed costs of crossing the boundary. AI Systems Performance states the rule directly: speeding up model compute does not help if the data pipeline cuts throughput in half, so always profile end-to-end, not the model alone. The gap between the model-only number and the end-to-end number is exactly where production surprises live, because an engineer who optimizes the model has optimized the part that was already small. The block below times two layers on the same prediction: model.predict() by itself, and the full validate-predict-serialize path a real handler runs. The clock is time.perf_counter(), which the Python docs specify as the highest-resolution performance counter, where only the difference between two calls is meaningful. Watch the ratio between the two numbers.

import json
import time

import numpy as np
from pydantic import BaseModel


class LoanRecord(BaseModel):
    annual_inc: float
    loan_amnt: float
    dti: float


def model_predict(x: np.ndarray) -> float:
    # Stand-in for a small sklearn model: one vectorized dot product.
    w = np.array([0.3, -0.4, -0.2])
    return float(1.0 / (1.0 + np.exp(-(x @ w))))


raw_body = json.dumps({"annual_inc": 60_000, "loan_amnt": 12_000, "dti": 18.5})

N = 5_000

# Layer 1: time ONLY the model call.
features = np.array([0.6, 0.12, 0.185])
start = time.perf_counter()
for _ in range(N):
    model_predict(features)
model_only_us = (time.perf_counter() - start) / N * 1e6

# Layer 2: time the FULL boundary path -- parse, validate, predict, serialize.
start = time.perf_counter()
for _ in range(N):
    parsed = json.loads(raw_body)  # JSON deserialize
    rec = LoanRecord(**parsed)  # pydantic validation
    x = np.array([rec.annual_inc / 1e5, rec.loan_amnt / 1e5, rec.dti / 100])
    prob = model_predict(x)  # the model call
    json.dumps({"probability": prob, "label": prob > 0.5})  # JSON serialize
end_to_end_us = (time.perf_counter() - start) / N * 1e6

print("model.predict() only:   " + format(model_only_us, ".1f") + " us")
print("full boundary path:     " + format(end_to_end_us, ".1f") + " us")
print("model is this fraction:  " + format(model_only_us / end_to_end_us, ".2f"))
print(
    "overhead the model call hides: "
    + format(end_to_end_us - model_only_us, ".1f")
    + " us"
)

The model call is a small fraction of the full boundary path: the parse, validation, and serialization together dominate, and they are exactly what the model-only timer never sees. An engineer who reported the model-only number would be off by the multiple printed on the last line, which is the latency the caller actually pays. Why does optimizing the model produce almost no improvement in the number a caller sees here? Because the model is already the smallest term; shaving it shrinks a small fraction of the total, while the fixed boundary costs (the ones the model-only timer never measured) stay exactly where they were.

The rule for the rubric follows directly: the documented latency budget is the end-to-end number, measured at the boundary, in p50 and p99 terms. The percentile framing is the next lesson, where p50 describes the typical request and p99 the latency the worst one percent experience, and the mean hides the tail that a caller’s timeout depends on. For this lesson the point is where the clock starts and stops: wrap the full request-handling path, not the model call, or the published number will be the one part of the system that was never the problem.

import json
import time

import numpy as np
from pydantic import BaseModel


class Profile(BaseModel):
    age: int
    income: float


def model_predict(x: np.ndarray) -> float:
    return float(np.tanh(x.sum()))


raw = json.dumps({"age": 41, "income": 72_000})
N = 5_000

# Current code times ONLY the model. Fix it to time the whole path.
x = np.array([0.41, 0.72])
start = time.perf_counter()
for _ in range(N):
    model_predict(x)
model_us = (time.perf_counter() - start) / N * 1e6
print("model only us:", round(model_us, 1))

# TODO: time the full path (json.loads -> Profile(**...) -> features -> predict
#       -> json.dumps) over N iterations and print the model's fraction of it.
boundary_us = model_us  # placeholder -- replace with the measured full-path time
print("TODO: full boundary path and the model's fraction of it")

import json
import time

import numpy as np
from pydantic import BaseModel


class Profile(BaseModel):
    age: int
    income: float


def model_predict(x: np.ndarray) -> float:
    return float(np.tanh(x.sum()))


raw = json.dumps({"age": 41, "income": 72_000})
N = 5_000

x = np.array([0.41, 0.72])
start = time.perf_counter()
for _ in range(N):
    model_predict(x)
model_us = (time.perf_counter() - start) / N * 1e6

start = time.perf_counter()
for _ in range(N):
    parsed = json.loads(raw)
    p = Profile(**parsed)
    feats = np.array([p.age / 100, p.income / 1e5])
    prob = model_predict(feats)
    json.dumps({"probability": prob})
boundary_us = (time.perf_counter() - start) / N * 1e6

print("model only us:    " + format(model_us, ".1f"))
print("full boundary us: " + format(boundary_us, ".1f"))
print("model fraction:   " + format(model_us / boundary_us, ".2f"))

Summary

• Real-time and batch inference optimize opposite quantities: online minimizes per-request latency under a deadline, batch maximizes throughput across many records, so a loop of single calls re-pays the fixed per-request overhead (HTTP, parse, validation, dispatch) once per row, while a batched call pays it once and lets the vectorized predict absorb the rows.
• The crossover is in the access pattern, not the row count: looping single calls over a large set costs hours where one batched call over the same set costs minutes, with an identical model.
• A batch response is a positional list: out[i] belongs to in[i] only because the lists share length and order. Dropping or reordering a record silently misattributes every later prediction; represent a per-record failure as an error object in that record’s slot so length and order stay invariant.
• The serving-contract shape test must exercise the single-row path and the batch path, because a transform that re-derives columns per call produces a different row alone than in a batch.
• Measure latency at the endpoint boundary (parse, validate, predict, serialize), not at model.predict(). For a small model the model call is the smallest term, so a model-only number understates what the caller pays by a large multiple.

Check your understanding:

• A nightly job must score a million members. Which serving pattern fits, and what specifically does the wrong choice multiply?
• A batch endpoint validates each record and skips the bad ones, returning a shorter list. The tests pass. What breaks the first time real data carries a malformed row, and why did the tests miss it?
• Without looking back: why does timing only model.predict() on a small sklearn model understate the latency a caller actually experiences, and what must the clock wrap instead?