Machine Learning Engineer Interview Questions — Python

The convention most teams follow is about meaning, not performance. A list holds many of the same thing and its length varies. A tuple holds a fixed set of fields where position carries meaning.

sensor_ids = ["T-01", "T-02", "T-03"]      # list: many of one thing
reading = ("T-01", 34.2, "2026-08-19")     # tuple: one record

The concrete capability immutability gives you is hashability. A tuple of immutable values can be a dictionary key or a set member; a list cannot.

daily_max = {}
daily_max[("T-01", "2026-08-19")] = 41.7   # fine

daily_max[["T-01", "2026-08-19"]] = 41.7
# TypeError: unhashable type: 'list'

That single property is why tuples turn up as composite keys in caches, lookup tables and grouping logic all over real codebases.

The performance argument is usually overstated. Tuples are marginally smaller and construct slightly faster, and unless you are creating crores of them in a hot loop, that difference will not be what makes your job slow. Do not lead with it in an interview; lead with intent and hashability.

Once a tuple has more than three fields, positional access becomes a liability. reading[2] tells a reader nothing. NamedTuple or a frozen dataclass keeps the immutability and adds names:

from typing import NamedTuple

class Reading(NamedTuple):
    sensor_id: str
    celsius: float
    taken_on: str

Still hashable, still immutable, and r.celsius survives someone reordering the fields.

settings = {"retries": 3, "timeout": None}

settings["region"]              # KeyError: 'region'
settings.get("region")          # None
settings.get("region", "apac")  # 'apac'

Choosing between them is a decision about whether absence is normal. Reading a required configuration key with .get() and a silent default means a typo in the config file produces a working program with wrong behaviour, which is far harder to debug than a KeyError on startup. Reading an optional field from a parsed API response with brackets means one missing attribute crashes the whole batch.

The trap in that example is timeout. Both a missing key and a key set to None make .get() return None, so you cannot tell them apart:

settings.get("timeout")        # None — but the key exists
settings.get("nonsense")       # None — key doesn't exist

"timeout" in settings          # True

When the difference matters, test with in rather than inferring from the returned value. This bites when None is a legitimate stored value meaning “no limit” and your code treats it as “not configured” and substitutes a default.

One more thing people get wrong: .get() never inserts. If you want to read-or-create in one step, that is .setdefault().

by_city = {}
by_city.setdefault("Pune", []).append("order-118")

For accumulating into a dict repeatedly, collections.defaultdict(list) reads better and avoids repeating the default on every line.

import copy

template = [["mon", 0], ["tue", 0]]

shallow = copy.copy(template)      # or template[:] or list(template)
shallow[0][1] = 9
print(template)     # [['mon', 9], ['tue', 0]]  ← changed

deep = copy.deepcopy(template)
deep[1][1] = 5
print(template)     # [['mon', 9], ['tue', 0]]  ← safe

The shallow copy did create a new outer list. What it did not do is create new inner lists, so shallow[0] and template[0] are the same object, and mutating through either label is visible through both.

This turns up most often with a default structure reused across iterations. A roster template copied per employee, where every employee ends up sharing one shift list, produces a bug where updating one person’s schedule silently updates everyone’s. It looks like a database problem until you find the copy.

Deep copy is not free. It walks the entire object graph and reconstructs it, which on a large nested structure inside a loop is genuinely slow. Two alternatives worth mentioning. Rebuild the structure explicitly with a comprehension when you know its shape:

fresh = [[day, count] for day, count in template]

Or avoid the problem by making the inner elements immutable, since a tuple cannot be mutated through a shared reference at all.

Two details for the follow-ups. deepcopy handles cyclic references correctly by keeping a memo of objects it has already copied, so it does not recurse forever. And a class can control what happens by defining __deepcopy__, which matters for objects holding a database connection or a file handle that should not be duplicated.

Same syntax, two directions. In the definition it packs; at the call it unpacks.

def log_event(event, *tags, **fields):
    print(event, tags, fields)

log_event("upload", "batch", "s3", size=940, retries=2)
# upload ('batch', 's3') {'size': 940, 'retries': 2}

payload = {"size": 940, "retries": 2}
log_event("upload", "batch", **payload)   # unpacking

The names are convention only; *a and **kw work identically. Stick with args and kwargs because everyone reading your code expects them.

Where these earn their place is a wrapper that must forward whatever it receives without knowing the signature it is wrapping. A retry helper, a timing wrapper, a decorator: all of them accept *args, **kwargs and pass them straight through, so they work with any function.

def with_retry(fn, *args, **kwargs):
    for attempt in range(3):
        try:
            return fn(*args, **kwargs)
        except TimeoutError:
            continue
    raise

Two things worth knowing beyond the basics.

A bare * forces everything after it to be keyword-only, which is a good way to stop callers passing a run of unlabelled booleans:

def export(path, *, overwrite=False, compress=True):
    ...
export("out.csv", overwrite=True)   # positional not allowed

And the cost of **kwargs is that it destroys the signature. Callers get no autocomplete, typos become silent extra dict entries rather than errors, and type checkers cannot help. Use it where forwarding is genuinely the point, not to avoid deciding what parameters a function takes.

def collect(row, batch=[]):
    batch.append(row)
    return batch

collect("r1")    # ['r1']
collect("r2")    # ['r1', 'r2']  ← not ['r2']
collect("r3")    # ['r1', 'r2', 'r3']

The list is created once, at def time, and stored on the function object itself. You can see it directly:

collect.__defaults__      # (['r1', 'r2', 'r3'],)

Every call that omits the argument gets that same list, so state leaks between calls that have nothing to do with each other.

The correct form is a sentinel:

def collect(row, batch=None):
    if batch is None:
        batch = []
    batch.append(row)
    return batch

is None rather than if not batch, because an empty list passed deliberately by the caller is falsy and would be silently replaced.

In production this rarely announces itself. A parsing helper with errors={} accumulates every error from every file processed since the service started, and memory grows steadily while each individual call looks correct. In a long-running worker that is a slow leak nobody attributes to the function.

The same evaluate-once rule catches other defaults. def report(as_of=datetime.now()) freezes the timestamp at import time, so a service running for a week reports the same date forever. Pass None and compute inside.

Immutable defaults are safe, since there is nothing to mutate: def f(n=0, label="") behaves as everyone expects.

There is one deliberate use of the behaviour, worth knowing but not worth writing: attaching a mutable default as a cheap per-function cache. functools.lru_cache does the same job without surprising the next reader.

staff = [
    {"name": "Riya", "dept": "ops", "joined": "2021-04-01"},
    {"name": "Arun", "dept": "eng", "joined": "2023-07-15"},
    {"name": "Meera", "dept": "ops", "joined": "2019-01-09"},
]

by_dept = sorted(staff, key=lambda r: (r["dept"], r["joined"]))

A tuple key sorts by the first element and uses the second to break ties, which handles most multi-field cases in one call.

Mixed directions are where stability earns its keep. You cannot put reverse=True on just one element of a tuple key, so you sort twice, least significant first:

rows = sorted(staff, key=lambda r: r["joined"], reverse=True)
rows = sorted(rows, key=lambda r: r["dept"])

Because Python’s sort is stable, the second sort preserves the date ordering within each department. Reverse the order of those two statements and the result is wrong. For numeric fields there is a shortcut: negate the value in the key, so key=lambda r: (r["dept"], -r["salary"]).

operator.itemgetter is often faster and clearer than a lambda for straightforward field access:

from operator import itemgetter
sorted(staff, key=itemgetter("dept", "joined"))

Two things to keep straight. sorted() returns a new list and works on any iterable; .sort() mutates a list in place and returns None. Writing rows = rows.sort() silently sets rows to None, and the failure surfaces later as a confusing TypeError.

And the key function is called exactly once per element, then the results are compared. So an expensive key such as a date parse costs n calls rather than n log n, which is the reason key= replaced the old comparison-function style.

books = [("Ghachar Ghochar", 2013), ("Em and the Big Hoom", 2012)]

titles  = [t for t, _ in books]                # list
by_year = {t: y for t, y in books}             # dict
years   = {y for _, y in books}                # set

Adding a condition filters, and a conditional expression before the for transforms:

recent = [t for t, y in books if y >= 2013]
tags = ["new" if y >= 2013 else "old" for _, y in books]

Those two ifs sit in different positions and do different jobs. Filtering goes after the for; choosing between two values goes before it. Getting them confused is a common syntax error.

Nesting reads in statement order, outer loop first:

shelves = [["a1", "a2"], ["b1"]]
flat = [code for shelf in shelves for code in shelf]

Read it as for shelf in shelves: then for code in shelf:, which is exactly the order you would write the loops. People frequently expect the reverse and get an error about an undefined name.

Where to stop is a judgement worth voicing. Two loops with a condition is around the limit before a comprehension stops being readable, and a reader having to decode it costs more than the two extra lines a loop would take. There is no rule here, only the next person reading it.

One memory point. Brackets build the entire list in memory before returning. Parentheses produce a generator expression instead, which yields lazily and is what you want when feeding a large sequence into sum or a loop:

total = sum(y for _, y in books)     # no intermediate list

Comprehensions have their own scope from Python 3 onwards, so the loop variable does not leak out.

def parse_log(path):
    with open(path) as f:
        for line in f:
            if line.startswith("ERROR"):
                yield line.strip()

for entry in parse_log("app.log"):   # one line in memory
    handle(entry)

The list version builds every matching line before the caller sees the first one. On a 40 GB log file that is the difference between a job that runs and one that gets killed by the memory limit.

Calling parse_log("app.log") executes nothing. It returns a generator object. The body runs only when something iterates it, advancing to the next yield and freezing there with all locals intact.

What you give up matters as much as what you gain.

gen = parse_log("app.log")
len(gen)          # TypeError
gen[0]            # TypeError
list(gen)         # works, but defeats the purpose

for x in gen: ... # consumes it
for x in gen: ... # yields nothing, silently

That last one is the failure that reaches production. A generator passed to two functions works for the first and gives the second an empty sequence, with no error and no warning. The downstream count comes out as zero and someone spends an afternoon on it. If two passes are needed, materialise once with list() and accept the memory cost, or call the generator function again.

One subtlety with the file example: the with block stays open while the generator is alive, and if the caller abandons it partway, the file is closed when the generator is garbage collected rather than at a predictable point.

yield from delegates to another iterable, which flattens nested generators without an explicit loop and correctly forwards values sent into the generator.

try:
    conn = connect(dsn)
except ConnectionError as exc:
    log.error("could not reach vendor API: %s", exc)
    raise
else:
    payload = conn.fetch_invoices()   # only if connect worked
finally:
    conn.close() if conn else None

The else block is the part most people never use, and it is what keeps the try narrow. Put fetch_invoices() inside the try and a ConnectionError raised deep inside the fetch gets caught by a handler written for the connection step, and your log says the wrong thing about the wrong operation.

finally runs whether or not an exception occurred, and whether or not the block returned. That makes it the right place for releasing a lock, closing a handle, or restoring a setting. In most cases a with block expresses the same intent more clearly, since the context manager owns the cleanup.

Three things that separate a working answer from a careful one.

Catch narrowly. A bare except: catches KeyboardInterrupt and SystemExit as well, so your job cannot be stopped cleanly. except Exception is better and still too broad as a default, since a typo raising NameError is swallowed and reported as a vendor failure.

Never swallow silently. except Exception: pass is how a nightly pipeline reports success while writing nothing. If you catch it, either handle it meaningfully or log it and re-raise.

Preserve the chain. When you wrap an error in your own exception type, raise LoadError("vendor feed failed") from exc keeps the original traceback attached. Without from, the underlying cause is harder to find in the logs.

One sharp edge worth knowing: a return inside finally discards any exception currently propagating, so an error disappears with no trace.

The guarantee is the point. A try/finally gives you the same thing, and with moves the cleanup logic into the object that owns the resource, so every caller gets it right without remembering.

class Stopwatch:
    def __init__(self, label):
        self.label = label

    def __enter__(self):
        self.start = time.perf_counter()
        return self          # bound to `as`

    def __exit__(self, exc_type, exc, tb):
        elapsed = time.perf_counter() - self.start
        log.info("%s took %.2fs", self.label, elapsed)
        return False         # do not suppress

with Stopwatch("model export") as sw:
    export_weights()

__exit__ receives three arguments describing any exception that escaped the block, all None if the block finished cleanly. Returning a truthy value from __exit__ suppresses that exception, which is occasionally what you want and much more often a mistake. Return False, or nothing at all, unless you deliberately mean to swallow it.

Whatever __enter__ returns is what as binds. Returning self is the common choice; returning None is fine when the block does not need a handle.

For most cases the generator form is shorter and reads better:

from contextlib import contextmanager

@contextmanager
def temp_setting(obj, **overrides):
    saved = {k: getattr(obj, k) for k in overrides}
    for k, v in overrides.items():
        setattr(obj, k, v)
    try:
        yield obj
    finally:
        for k, v in saved.items():
            setattr(obj, k, v)

The try/finally around the yield is essential. Without it, an exception in the body propagates out of the yield and the restoration never runs, which is exactly the guarantee you were trying to provide.

One gotcha: opening several resources with a single with and commas is fine, but if the second one fails during setup, the first is still cleaned up correctly. Nesting manually with plain assignments does not give you that.

class Turbine:
    site = "Jaisalmer"      # class attribute, shared
    log = []                # class attribute, shared and mutable

    def __init__(self, tag):
        self.tag = tag      # instance attribute

Lookup goes instance first, then class. So t.site finds nothing on the instance and falls back to the class, and every turbine reports the same site. That is useful for genuine constants and defaults.

Assignment does not follow the same path. Writing t.site = "Bhuj" creates an instance attribute that shadows the class one for that object only; the class attribute is untouched and every other instance still sees Jaisalmer. Reading falls back, writing does not.

The mutable class attribute is the real trap:

a, b = Turbine("T1"), Turbine("T2")
a.log.append("fault")
b.log              # ['fault']

a.log.append(...) never assigns, so it never creates an instance attribute. It reaches through to the shared list and mutates it. Two objects that should be independent now share state, and in a long-running service that list also grows without bound. Anything mutable belongs in __init__ as self.log = [].

self is not a keyword, just a convention. Python passes the instance as the first positional argument to any method called on it, so t.spin() is Turbine.spin(t). Rename it and the code works and every reviewer will object.

One detail for the follow-up: __init__ initialises an object that already exists and must return None. The object is created by __new__, which you rarely touch outside of immutable subclasses and singletons.

The problem is a single shared namespace. One project needs an older release of a plotting library that a colleague’s project cannot use, and there is exactly one place to install it.

python -m venv .venv
source .venv/bin/activate      # Windows: .venvScriptsactivate
pip install -r requirements.txt

venv is in the standard library from Python 3.3 onwards, so nothing needs installing first. Activating it puts that environment’s python and pip at the front of your PATH, so installs land inside the project directory rather than in the system interpreter.

Two failures make this concrete.

Installing into the system Python on Linux can break OS tooling that depends on specific package versions. Recent versions of Debian and Ubuntu now refuse a bare pip install outside a virtual environment for exactly this reason, and the error message confuses people who have never used one.

The subtler one is drift. Code runs on your machine and fails in the pipeline, because your machine picked up a library version installed eighteen months ago for something else. Nobody can reproduce it and nobody can explain it. A per-project environment plus a pinned requirements file removes the whole category.

Pin what you install. pip freeze > requirements.txt captures exact versions including transitive dependencies, which is what you want for reproducing a deployment. A hand-written file listing only your direct dependencies with ranges is better for a library others will install. Tools like Poetry and uv manage both sides with a lock file, and are worth naming if the interviewer asks what you use.

One practical note: keep .venv/ out of version control and commit the requirements file instead. The environment is derived; the specification is the artefact.

The syntax hides an assignment. These two are identical:

@audit
def issue_refund(ticket_id, amount):
    ...

def issue_refund(ticket_id, amount):
    ...
issue_refund = audit(issue_refund)

That second form is the whole concept. audit receives the original function object and returns something else, and the name now points at the replacement.

Writing one:

import functools

def audit(fn):
    @functools.wraps(fn)
    def wrapper(*args, **kwargs):
        log.info("calling %s", fn.__name__)
        result = fn(*args, **kwargs)
        log.info("%s returned", fn.__name__)
        return result
    return wrapper

Three details carry the weight. *args, **kwargs means the wrapper forwards anything, so one decorator works on functions with any signature. return result is easy to omit and turns every decorated function into one that returns None, which is a bug that passes import and fails at runtime. And functools.wraps copies the original’s __name__, __doc__ and __wrapped__ onto the wrapper, without which tracebacks, help output and anything doing introspection all report wrapper.

A decorator taking arguments needs one more layer, because @retry(times=3) calls retry(times=3) first and applies the result as the decorator:

def retry(times):
    def decorator(fn):
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            for _ in range(times - 1):
                try:
                    return fn(*args, **kwargs)
                except TimeoutError:
                    continue
            return fn(*args, **kwargs)
        return wrapper
    return decorator

Three nested functions is the standard shape and the part people fumble under pressure. Practise writing it.

Two things worth saying. Decorators run at import time, when the def is executed, so an expensive decorator body costs you on every import. And methods work unchanged: self arrives as the first element of args, since by then it is an ordinary positional argument.

from fastapi import FastAPI, HTTPException
import uuid, logging

app = FastAPI()
log = logging.getLogger(__name__)

@app.post("/v1/score")
def score(req: ScoreRequest) -> ScoreResponse:
    request_id = str(uuid.uuid4())
    try:
        features = build_features(req, FEATURE_ORDER)
        prob = float(MODEL.predict_proba([features])[0][1])
    except ValueError as exc:
        raise HTTPException(status_code=422, detail=str(exc))
    log.info("scored", extra={"rid": request_id, "p": prob})
    return ScoreResponse(request_id=request_id,
                         probability=prob,
                         model_version=MODEL_VERSION)

Flask does the same job and leaves validation, typing and documentation to you. Either is defensible; say why you picked one rather than asserting FastAPI is better.

Four things the endpoint needs beyond calling the model.

Feature order. Most estimators take a positional array, so assembling features in a different order than training produces confident nonsense with no error. Load the order from the model artefact rather than hardcoding it in the serving code.

Model version in the response. Without it, nobody can tell which model produced a prediction that is later disputed, and a rollback becomes guesswork.

A request identifier. Logged with the prediction and returned to the caller, so a complaint about one score can be traced to one log line.

Sensible error codes. A malformed payload is 422, a missing dependency or a model failure is 500. Returning 200 with an error message in the body is the pattern that causes downstream systems to silently record garbage.

Two operational points. /health should confirm the model actually loaded, not just that the process is alive, otherwise the orchestrator keeps routing traffic to a container serving 500s. And never call uvicorn.run with reload=True outside development; run behind Gunicorn with uvicorn workers in production.

The version that causes it:

@app.post("/v1/rank")
def rank(req: RankRequest):
    model = joblib.load("ranker.joblib")     # every request
    return model.predict(req.features)

Loading per request repeats disk read, deserialisation and any lazy initialisation on every call. It also allocates a fresh copy each time, so under concurrency you get several copies alive at once and memory spikes unpredictably.

Load once at import, or better, in a lifespan handler so failures surface at startup:

from contextlib import asynccontextmanager

STATE = {}

@asynccontextmanager
async def lifespan(app):
    STATE["model"] = joblib.load("ranker.joblib")
    STATE["model"].predict(WARMUP_BATCH)     # force lazy init
    yield
    STATE.clear()

app = FastAPI(lifespan=lifespan)

The warm-up call matters. Many frameworks defer graph construction, kernel selection or JIT compilation until the first real inference, so even a loaded model has a slow first call. Running one dummy prediction at startup pays that cost before traffic arrives.

Two things to raise unprompted.

Each worker process holds its own copy. Four Gunicorn workers with a 2 GB model is 8 GB, and that is frequently the binding constraint rather than CPU. If the model is large, fewer workers with threads, or a shared inference server such as Triton, is the alternative.

And with the spawn or fork behaviour of your server, loading at module import means the load happens in each worker after forking, which is what you want. Loading before the fork and relying on copy-on-write sharing sounds efficient and breaks in CPython, because reference count updates touch the pages and they get copied anyway.

Wire the readiness probe to the loaded state, so the container is not marked ready until the model is in memory and warm.

from pydantic import BaseModel, Field, field_validator

class ClaimRequest(BaseModel):
    policy_no: str = Field(min_length=8, max_length=20)
    claim_amount: float = Field(gt=0, le=5_000_000)
    hospital_tier: int = Field(ge=1, le=3)
    prior_claims: int = Field(default=0, ge=0)

    @field_validator("policy_no")
    @classmethod
    def upper(cls, v: str) -> str:
        return v.upper().strip()

With FastAPI, typing the handler parameter as ClaimRequest is the whole wiring. An invalid payload never reaches your code; the framework returns a 422 listing which field failed and why, which is far more useful to the caller than a generic error.

Hand-rolled if checks fail on three counts. They accumulate as a wall of guards at the top of every handler, they drift out of sync with what the model actually expects, and they produce inconsistent error messages that callers cannot parse.

The behaviour worth knowing precisely: pydantic v2 coerces where it is unambiguous. "42" for an int field is accepted and converted; "forty-two" raises. If you want strictness, Field(strict=True) or a strict model config refuses the coercion. Decide deliberately, because silent coercion of a string amount into a float is convenient at an API boundary and surprising inside a pipeline.

Two more points.

By default, extra fields are ignored. For an inference endpoint that is often wrong: a caller sending hospital_teir gets the default for the real field and no warning. model_config = ConfigDict(extra="forbid") rejects the payload instead.

And cross-field rules go in a @model_validator(mode="after"), which runs once every field has been validated, so you can check that a discharge date is after an admission date.

The validation model also documents the contract. FastAPI generates OpenAPI from it, so the schema the serving team publishes stays in step with the code.

Guessing wastes days. Time the stages and log them:

import time

def timed(stage: str, spans: dict):
    class _T:
        def __enter__(self): self.t = time.perf_counter()
        def __exit__(self, *a):
            spans[stage] = (time.perf_counter() - self.t) * 1000
    return _T()

spans = {}
with timed("features", spans):
    feats = assemble(req)
with timed("model", spans):
    out = MODEL.predict([feats])
log.info("latency", extra=spans)

Log the breakdown on every request and you can query the distribution per stage rather than reasoning about averages.

The usual culprits, in rough order of frequency.

A feature lookup against a warehouse rather than a key-value store. An analytical query returning one row still costs hundreds of milliseconds, which dwarfs the model.

Serialisation. A large JSON payload parsed and re-serialised per request is real CPU, and it grows with the number of features.

Queueing. With more concurrent requests than workers, time is spent waiting to be picked up, and that shows in the tail rather than the median.

Per-request work that should be at startup, such as reloading a lookup table or a tokeniser.

The p50-fine, p99-terrible pattern is characteristic and worth naming. Steady per-request cost raises the whole distribution; queueing, garbage collection pauses, connection pool exhaustion and cold caches hit a minority of requests hard. So a bad tail with a good median points at contention, not at the model.

Measure client-side latency too. If the endpoint reports 40 ms and the caller sees 400, the difference is network, TLS handshakes or a load balancer, and no amount of model optimisation will touch it.

For a running service, py-spy samples without a restart and adds little overhead, which is the only practical option when the slowness will not reproduce locally.

a = 256
b = 256
a == b      # True
a is b      # True

x = 1000
y = 1000
x == y      # True
x is y      # False

Nothing changed about the language between those two blocks. CPython pre-creates integer objects from -5 to 256 and reuses them, so both names point at the identical cached object. At 1000 there is no cache, so two separate objects exist with equal value.

Short strings behave similarly through interning, and the rules for when interning happens vary by how the string was created and by CPython version. Both behaviours are implementation details of CPython, not guarantees of Python, and code that depends on them can break on another interpreter or another release.

The rule that follows: use == for value, and is only where identity is genuinely the question. In practice that means the singletons.

if config is None:          # correct
if config == None:          # works, but non-idiomatic

is None is preferred for a real reason beyond style. == calls the left operand’s __eq__, and a class can define that method to return anything it likes, including True when compared with None. is cannot be overridden, so it always answers the actual question.

The failure this causes in production is a comparison that passes every test on small values and fails on real data. A validation checking status_code is 200 works throughout development and breaks when someone loads the code path where the value is parsed from JSON. It reads as a mysterious intermittent bug, and static analysis tools now flag it precisely because it is so common.

Training is a pipeline with two stages: prepare a batch on the CPU, compute on the GPU. If preparation takes 90 ms and computation takes 40, the GPU works for 40 and waits for 50, and utilisation sits around 30 to 45%.

Confirm before optimising. Time the loader alone by iterating it without any training step:

import time

start = time.perf_counter()
for i, batch in enumerate(loader):
    if i == 200:
        break
per_batch = (time.perf_counter() - start) / 200 * 1000
print(f"{per_batch:.1f} ms per batch, loader only")

Compare that against your measured step time. If the loader alone is slower, you have your answer and no profiler is needed.

Three levers, in order of impact.

Parallel workers. Loading in separate processes overlaps preparation with computation. Start around the number of physical cores and measure, because too many workers cause memory pressure and context switching that make things worse.

Prefetch. Prepare batch n+1 while the GPU computes on batch n. Most frameworks do this with a queue between the workers and the training loop; the queue depth is what buys the overlap.

Move work off the critical path. Decoding JPEGs, resizing and tokenising per epoch repeats identical work every time. Preprocess once into a compact binary format and the loader becomes a sequential read.

Two other things worth checking before blaming the loader. Random access to lakhs of small files on network storage is dominated by seek latency, and packing them into shards fixes it. And synchronous logging or metric computation inside the loop stalls the GPU just as effectively as slow loading.

Ordering matters too: augmentation applied to a batch on the GPU is often far cheaper than applying it per sample on the CPU.

Fixed row count is the wrong unit when inputs vary in size. A batch of 32 sequences averaging 60 tokens allocates 32 by 60. One outlier of 3,000 tokens in the same batch allocates 32 by 3,000, which is fifty times the memory, from a single row.

Two fixes.

Bucket by length. Sort or group similar lengths together so padding waste is small:

records.sort(key=len)
for i in range(0, len(records), batch_size):
    yield pad(records[i:i + batch_size])

Sorting destroys shuffling, so the usual compromise is to sort within a large pool, form batches, and shuffle the batches.

Cap on total elements rather than rows. Batch size then varies and memory does not:

def by_token_budget(records, budget=8000):
    batch, longest = [], 0
    for r in records:
        longest = max(longest, len(r))
        if longest * (len(batch) + 1) > budget and batch:
            yield batch
            batch, longest = [r], len(r)
        else:
            batch.append(r)
    if batch:
        yield batch

Note the final if batch: after the loop. Without it the last partial batch is silently dropped, and every run quietly loses the tail.

Two related points for training specifically.

Gradient accumulation lets you keep a large effective batch with small physical batches: run several forward and backward passes, accumulate gradients, then step once. Memory follows the physical batch; the optimiser sees the large one.

And peak memory is not just the batch. Activations, the optimiser state, and any cached intermediate all live at once, so the largest batch that fits is well below what the batch tensor alone suggests. Measure peak rather than reasoning about it, and leave headroom, because a job that OOMs at hour six of eight is expensive.

LEGB is the lookup order, and the part that catches people is not the lookup but the assignment rule.

tax_rate = 0.18

def price_with_tax(base):
    print(tax_rate)          # UnboundLocalError
    tax_rate = 0.05
    return base * (1 + tax_rate)

The error appears on a line that only reads the variable. Python scans the whole function body at compile time, sees an assignment to tax_rate, and marks it local for the entire function, including lines above the assignment. At the moment of the print it is local and not yet bound.

global says the name refers to the module-level variable, and nonlocal says it refers to the nearest enclosing function’s variable:

def make_counter():
    count = 0
    def record():
        nonlocal count       # without this: UnboundLocalError
        count += 1
        return count
    return record

count += 1 is an assignment, so the same rule applies inside the inner function. nonlocal points it back out one level. global would not work here, since count is not at module level.

Two things worth saying in an interview. Only functions, classes, modules and comprehensions create scopes. for, if and with do not, so a loop variable is still visible after the loop ends, which surprises people coming from C-family languages.

And global in production code is usually a smell. Module-level mutable state makes functions untestable and behaviour order-dependent, and in a multi-threaded worker it becomes a race condition. Pass the value in and return the result instead.

Seeding one generator and assuming the rest follow is the usual cause. A shuffle, a dropout mask and a weight initialisation can each draw from a different generator.

import os, random
import numpy as np
import torch

def seed_everything(seed: int) -> None:
    random.seed(seed)
    np.random.seed(seed)
    torch.manual_seed(seed)
    torch.cuda.manual_seed_all(seed)
    torch.use_deterministic_algorithms(True)
    torch.backends.cudnn.benchmark = False

PYTHONHASHSEED has to be set in the environment before the interpreter starts, not inside the script, because hash randomisation is fixed at launch. It matters wherever iteration over a set of strings feeds into an ordering.

Three things a seed alone does not fix.

Non-deterministic kernels. Several GPU reductions accumulate with atomics, so the order of addition varies between runs, and floating-point addition is not associative. use_deterministic_algorithms(True) forces deterministic implementations and raises if one does not exist, which is a useful signal in itself.

cuDNN autotuning. With benchmark = True, the library times several convolution algorithms on the first call and picks the fastest, and the winner can differ between runs. Setting it False keeps the choice stable at some throughput cost.

Worker processes. Data loader workers each need their own derived seed, or the augmentation order changes with worker scheduling. Seed from a base plus the worker id inside the worker init function.

Two honest caveats. Determinism has a real cost, sometimes tens of percent, so many teams enable it for debugging and disable it for production runs. And reproducibility does not survive a hardware or library change: the same seed on a different GPU generation or a different cuDNN version gives different last digits, which compound over thousands of steps.

Log the seed, the library versions and the device with every run, or you cannot reproduce even the deterministic ones.

multipliers = []
for factor in [2, 5, 10]:
    multipliers.append(lambda x: x * factor)

[m(3) for m in multipliers]     # [30, 30, 30] ← not [6, 15, 30]

Nothing is captured at the moment each lambda is created. Each one holds a reference to the enclosing variable factor, and looks it up when called. By the time you call them, the loop has finished and factor is 10.

This is late binding, and it is the correct behaviour for closures generally. A closure that captured values would be useless for the counter pattern, where the whole point is that the inner function sees updates to the outer variable.

Two ways to bind the value at definition time. A default argument, which is evaluated once when the lambda is defined:

for factor in [2, 5, 10]:
    multipliers.append(lambda x, f=factor: x * f)

Or functools.partial, which stores the argument explicitly:

from functools import partial
from operator import mul

multipliers = [partial(mul, f) for f in [2, 5, 10]]

Where this bites in real work is callbacks. Registering a handler per queue inside a loop, or building a list of retry functions each bound to a different endpoint, and every one of them ends up pointing at the last item. The symptom is that all your workers process the same partition, which looks like a configuration problem rather than a language one.

A comprehension has exactly the same issue, since it also builds the functions before any of them run. And each closure holds its variables in a cell object, visible as fn.__closure__, which is worth knowing if an interviewer pushes on the mechanism.

Split the code by what is actually deterministic.

Preprocessing, feature assembly, label encoding, metric computation and the collate function are pure functions. Test them with fixed inputs and exact expected outputs, exactly as you would any other code. This is most of the surface area and most of the bugs.

For the parts involving randomness, test properties rather than values:

def test_augment_preserves_shape_and_range():
    batch = torch.rand(8, 3, 64, 64)
    out = augment(batch)
    assert out.shape == batch.shape
    assert out.dtype == batch.dtype
    assert out.min() >= 0.0 and out.max() <= 1.0

def test_model_overfits_tiny_batch():
    torch.manual_seed(0)
    x, y = fixed_tiny_batch()
    model, opt = build()
    first = step(model, opt, x, y)
    for _ in range(50):
        last = step(model, opt, x, y)
    assert last < first * 0.5

That second test is the most valuable one in ML code. A model that cannot drive the loss down on eight examples it sees fifty times has a wiring bug: a detached gradient, a frozen layer, an optimiser not stepping, labels misaligned with inputs. It runs in seconds and catches a class of failure that costs days.

Three more worth having.

A shape and dtype contract test on the model’s forward pass, so a config change that breaks the input pipeline fails immediately rather than after twenty minutes of training.

A leakage test asserting that no identifier appears in both the train and validation splits.

A determinism test: seed, run twice, assert the outputs match. It catches an unseeded generator creeping in.

What not to write: an assertion that validation accuracy exceeds 0.87. It depends on data, hardware and library versions, so it fails for reasons unrelated to your change, and it gets disabled within a month. Track metrics in your experiment logging, and let tests check correctness.

from pydantic import BaseModel, Field
import tomllib, sys

class TrainConfig(BaseModel):
    model_config = {"extra": "forbid"}
    dataset: str
    lr: float = Field(gt=0, le=1)
    batch_size: int = Field(ge=1)
    epochs: int = Field(ge=1)
    seed: int = 20260819

def load_config(path: str) -> TrainConfig:
    with open(path, "rb") as f:
        return TrainConfig(**tomllib.load(f))

Two things that decoration buys you. extra="forbid" rejects an unrecognised key, so learning_rate where the field is lr fails at startup instead of silently using the default for four hours. And the field constraints catch a learning rate of 10 or a batch size of 0 before anything loads.

tomllib is standard library from Python 3.11. Hydra and OmegaConf are the common ML-specific alternatives and add composition and command-line override handling, which becomes worth it once you have a base config plus per-experiment fragments.

Three properties to aim for.

No hardcoded values below the entry point. A threshold typed into a function is invisible to the config, so two runs that differ only in that number cannot be told apart from their configs.

Command-line overrides for sweeps. --lr 0.003 applied over the file, so a sweep varies one field without generating a hundred files.

The resolved config saved with the run. Not the file path, the actual values after overrides, written into the output directory alongside the checkpoints. Six months later that file is the only reliable record of what produced a result.

Validate at startup rather than lazily. A config error discovered at epoch three, when the code first reads a field, has already cost you the GPU time.

Two layers, and both matter.

Once, at the start. The resolved config, the git commit SHA, whether the working tree was dirty, the dataset identifier and row count, the library and CUDA versions, the device, and the seed. That group answers “what was this run”, and it is the group people skip.

import subprocess, json

def run_manifest(cfg, dataset):
    sha = subprocess.check_output(
        ["git", "rev-parse", "HEAD"], text=True).strip()
    dirty = bool(subprocess.check_output(
        ["git", "status", "--porcelain"], text=True).strip())
    return {
        "commit": sha, "dirty": dirty,
        "config": cfg.model_dump(),
        "dataset": dataset.fingerprint,
        "n_rows": len(dataset),
    }

The dirty flag is worth the two lines. A run from an uncommitted working tree cannot be reproduced from the SHA, and knowing that immediately saves an hour of confusion.

Per step or per epoch. Training and validation loss, learning rate, gradient norm, throughput, and time per step. The learning rate catches a scheduler misconfigured; the gradient norm catches exploding or vanishing gradients before the loss curve makes it obvious.

Log at an interval, not every step, or a long run produces a file nobody can read. Every hundred steps for training loss, once per epoch for validation.

Two things that make the logs usable later.

A run identifier on every line, so the shared log can be filtered to one run. Structured output, JSON or a tracking tool such as MLflow, so the fields can be queried rather than parsed with regex.

And log data statistics, not only model metrics. Label distribution, null rates per feature, and batch shapes. When a run behaves strangely, the input having changed is at least as likely as the model, and the model logs cannot tell you that.

An ordinary Python project pins numpy==1.26.4 and the story ends. ML frameworks add a second axis.

The same version number ships as several builds: CPU-only, and one per CUDA toolkit version. Which one you get depends on the index you install from, not on the version string:

--extra-index-url https://download.pytorch.org/whl/cu121
torch==2.3.1+cu121

The local version suffix after the plus is what identifies the build. A requirements file saying only torch==2.3.1 resolves to whatever the default index serves, which is frequently the CPU build. The install succeeds, the training script runs, and it is thirty times slower with no error, which is how people lose a day.

Below that sits the driver. A build compiled against CUDA 12.1 needs a driver new enough to support it, and the driver is a system component you often cannot change on shared infrastructure. So the constraint chain runs from the driver upward: the driver fixes the maximum CUDA version, which fixes the framework build, which fixes what versions of everything else are compatible.

Three practical consequences.

Pin the build, not just the version, and record the index URL alongside it.

A lock file gets you the Python layer and stops at the system boundary. It cannot pin the driver, the CUDA runtime installed on the host, or the BLAS library, which is why a container image built on a matching base is the stronger answer for anything that must reproduce.

And check the environment at startup rather than trusting it. A line asserting that the GPU is visible and printing the framework and CUDA versions turns a silent CPU fallback into an immediate failure.

A for loop is doing three things behind the scenes: call iter() on the object, call next() repeatedly, and stop when StopIteration is raised.

racks = ["R1", "R2"]
it = iter(racks)          # list -> list_iterator
next(it)                  # 'R1'
next(it)                  # 'R2'
next(it)                  # StopIteration

The list itself has no position. Each iter() call creates a new iterator with its own cursor, which is why two nested loops over the same list work independently.

An iterator returns itself from __iter__, which is what lets you pass one to a for loop directly. It also means iterating it consumes it. A file object is its own iterator, so a second loop over an already-read file yields nothing at all, with no error. That is the same silent-empty failure a consumed generator produces.

Writing one by hand:

class Countdown:
    def __init__(self, start):
        self.n = start

    def __iter__(self):
        return self

    def __next__(self):
        if self.n <= 0:
            raise StopIteration
        self.n -= 1
        return self.n + 1

Making it self-consuming like this is a design choice with a cost: Countdown(3) can only be looped once. The better pattern for a reusable collection is to keep the state out of the container and return a fresh iterator, or simply make __iter__ a generator function, which gives you a new one per call for free.

Two details for follow-ups. StopIteration is caught by the for loop itself, so you never see it unless you call next() manually; pass a default to next(it, None) to avoid it. And iter() falls back to __getitem__ with integer indices from zero for classes predating the protocol, which is why some old sequence classes are iterable without defining __iter__.

while True:
    try:
        job = queue.pop()
        proces_job(job)          # typo
    except:
        log.warning("job failed, retrying")

Two things are broken here and neither is visible. The typo raises NameError on every single iteration, and the handler reports it as a job failure, so the log fills with retry messages about a queue that is working perfectly. And pressing Ctrl-C raises KeyboardInterrupt, which the bare except catches, so the loop keeps running and the only way out is to kill the process.

The exception hierarchy is why. KeyboardInterrupt, SystemExit and GeneratorExit inherit from BaseException directly, not from Exception, precisely so that ordinary handlers do not intercept them. A bare except: sits at the BaseException level and undoes that design.

except Exception:        # excludes Ctrl-C and sys.exit
except (TimeoutError, ConnectionError):   # better still

Catch what you can actually do something about. If the handler’s response is “retry the network call”, then catch the network errors, and let a KeyError from malformed data propagate to somewhere that knows what to do with it.

There are two legitimate uses of a broad catch. A top-level supervisor loop in a long-running worker, where the job must not die because one item failed, and a plugin boundary where you cannot know what user code raises. In both cases the handler must log the full traceback and re-raise or record the failure:

except Exception:
    log.exception("job %s failed", job.id)   # includes traceback
    failed.append(job.id)

log.exception rather than log.warning is the detail that matters. Without the traceback you have a message and no way to find the line.

Interviewers ask this because a swallowed exception is the hardest class of bug to debug: the program reports success while doing nothing.

# loads 30 GB into RAM, then dies
rows = open("exports/claims.csv").readlines()

# streams, one line at a time
with open("exports/claims.csv") as f:
    header = next(f)
    for line in f:
        process(line)

A file object is its own iterator, so the for loop reads a buffered chunk from disk and hands you one line per iteration. Memory stays flat regardless of file size. That next(f) for the header works because the same iterator is consumed by both.

For CSV specifically, csv.reader wraps the same streaming behaviour and handles quoted fields containing newlines, which naive line splitting gets wrong. Reach for it rather than line.split(",") on anything that came out of a spreadsheet.

Binary or fixed-size reading needs an explicit loop:

with open(path, "rb") as f:
    while chunk := f.read(1024 * 1024):
        digest.update(chunk)

The walrus operator needs Python 3.8 or later; before that it is a while True with a break.

Two things that decide whether streaming actually works for your problem.

Aggregates are fine if they are incremental. Running totals, counts, min and max all work in one pass with constant memory. A median or an exact distinct count needs the whole dataset, so either accept the memory, spill to a temporary file, or use an approximate algorithm.

The failure mode people hit anyway is accumulating inside the loop. Streaming the file and appending every parsed row to a list uses the same memory as reading it all at once, just more slowly. Write results out as you go, or aggregate into something bounded.

If the work per line is genuinely CPU-heavy, multiprocessing over line ranges helps; if it is I/O bound, it mostly will not, since one disk is the constraint.

class Route:
    def __init__(self, code, stops):
        self.code = code
        self.stops = stops

    def __repr__(self):
        return f"Route({self.code!r}, stops={self.stops})"

    def __str__(self):
        return f"{self.code} ({self.stops} stops)"

r = Route("KA-17", 12)
print(r)         # KA-17 (12 stops)
print([r])       # [Route('KA-17', stops=12)]
r                # Route('KA-17', stops=12)  in the REPL

The list case is the one people get wrong. Containers always call __repr__ on their elements, so a class with only __str__ defined prints beautifully on its own and shows <__main__.Route object at 0x7f...> inside a list, in a debugger, and in a log line that formats a collection. If you write only one of the two, write __repr__.

The !r in the f-string calls repr() on the field, which keeps quotes around strings so the output stays unambiguous.

__eq__ changes comparison from identity to whatever you define:

    def __eq__(self, other):
        if not isinstance(other, Route):
            return NotImplemented
        return self.code == other.code

Returning NotImplemented for unknown types lets Python try the reflected operation instead of forcing a wrong answer.

The consequence people forget: defining __eq__ sets __hash__ to None, so instances become unhashable and cannot go in a set or be used as dict keys. If you need both, define __hash__ over the same fields, and only when those fields are immutable. A hashable object whose hash changes after insertion becomes unfindable in the dict it is sitting in.

@dataclass generates __init__, __repr__ and __eq__ from annotations, and frozen=True adds a __hash__. For plain data-carrying classes it is usually the better starting point.

The distinction is what the handler spends its time doing.

@app.post("/v1/recommend")
async def recommend(req: RecRequest):
    profile = await feature_store.get(req.user_id)   # waits
    items = await catalogue.fetch(req.slot)          # waits
    scores = MODEL.predict(build(profile, items))    # burns CPU
    return top_k(scores)

The two awaits are where async pays. While one request waits on the feature store, the loop serves others, so a hundred concurrent requests do not need a hundred threads.

The third line is the problem. MODEL.predict holds the GIL and does not yield, so for however long it runs, the event loop is frozen and every other request waits. One 40 ms model call blocking the loop turns into 4 seconds of added latency for the hundredth request in the queue.

Two ways out. Push the blocking call off the loop:

scores = await asyncio.to_thread(MODEL.predict, batch)

That helps when the model releases the GIL internally, which most compiled inference backends do during the numeric work. If it does not release the GIL, use a process pool or a separate inference service.

Or define the handler with plain def. FastAPI runs non-async handlers in a threadpool automatically, so a synchronous CPU-bound endpoint does not block the loop at all. That is frequently the right answer, and it surprises people who assume async def is always the better choice.

Two things to state clearly in an interview. Async is a concurrency mechanism, not a speed-up: a single request is not faster, you serve more of them. And a single synchronous call anywhere in an async path, an ordinary requests.get or a blocking database driver, undoes the entire benefit with no error to tell you.

The duplication that causes trouble is not the model call, it is the preprocessing around it. Two implementations of the same feature drift, and the model then sees inputs at serving time that differ from training.

Separate the layers explicitly:

# shared: pure, no I/O
def build_features(raw: dict, spec: FeatureSpec) -> list[float]:
    return [spec.transform(k, raw.get(k)) for k in spec.order]

def predict(model, feature_rows: list[list[float]]):
    return model.predict_proba(feature_rows)[:, 1]

# batch path
rows = [build_features(r, SPEC) for r in warehouse_rows]
scores = predict(model, rows)          # large batch

# online path
row = build_features(request.dict(), SPEC)
score = predict(model, [row])[0]       # batch of one

build_features and predict are shared and pure. What differs is where the raw data comes from and how many rows arrive, and that difference is real rather than accidental.

The genuine asymmetry is the feature source. Batch can join against the warehouse; online cannot afford a warehouse query. The reconciliation is to precompute features into a store that both read, batch reading the historical partition for a given date and online reading the latest row per entity, with both populated by the same computation. That is the single-definition principle applied to features rather than to code.

Where a feature genuinely cannot be computed online, the honest answer is that it cannot be a model input. Training on a feature that serving cannot supply produces a model that works offline and not in production.

Two operational points. Keep the batch predictions and the online predictions comparable by scoring the same entities both ways periodically and alerting on divergence. Disagreement means a code path drifted, a version mismatch, or a feature computed differently, and finding it through monitoring beats finding it through a quarter of poor performance.

And ship one artefact containing the model, the feature spec and the version, so both paths load the same object rather than each holding their own copy of the configuration.

Fork duplicates the parent, so the child inherits open file handles, loaded modules and existing objects with no serialisation. Spawn starts a new interpreter, re-imports the main module, and pickles whatever it needs to send across.

Three consequences follow.

The main guard becomes mandatory. Under spawn, each child re-imports your script. Without if __name__ == "__main__":, the child runs the loader creation again, spawning more children, and the process count climbs until the machine gives up or the job hangs.

if __name__ == "__main__":
    loader = DataLoader(ds, batch_size=64, num_workers=8)
    train(loader)

Everything must pickle. A dataset holding a lambda transform, an open file handle or a database connection works under fork and fails under spawn with a pickling error. Open files inside __getitem__ or lazily on first access in the worker, not in __init__.

Fork has its own trap. Forking a process that has already initialised a CUDA context produces workers with a broken context, and the failure is confusing rather than clean. That is why the guidance is to create the loader before touching the GPU, or to use spawn when GPU state is involved.

Fork also copies memory lazily through copy-on-write, which sounds like it shares a large in-memory dataset for free. In CPython it does not, because reference count updates touch the object headers and the pages get copied anyway. Eight workers over a 6 GB in-memory dataset can reach 48 GB.

Two practical notes. Set the method explicitly with multiprocessing.set_start_method("spawn", force=True) if you need consistent behaviour across platforms, rather than relying on the default.

And seed each worker from a base seed plus its worker id, so augmentation differs between workers while the whole run stays reproducible.

import os, torch

def save_checkpoint(path, model, optim, sched, step, rng):
    tmp = f"{path}.tmp"
    torch.save({
        "model": model.state_dict(),
        "optim": optim.state_dict(),
        "sched": sched.state_dict(),
        "step": step,
        "rng": rng,
    }, tmp)
    os.replace(tmp, path)          # atomic on the same fs

os.replace is the detail that matters. Writing directly to the final path means a crash halfway through leaves a truncated file that looks like a valid checkpoint, and you discover it only when the resume fails. Write to a temporary name and rename, which is atomic within a filesystem.

Saving only the weights is the most common mistake, and the symptom is a loss that jumps sharply on resume. Adaptive optimisers carry per-parameter state such as momentum and second-moment estimates, and restarting without it effectively resets the optimiser. The learning rate scheduler is the same: resuming at step zero replays a warm-up that already happened.

Random state matters for reproducibility rather than for loss. Without it, the data order and any augmentation differ after the resume, so the run is no longer the run you started.

Frequency is a trade-off between the checkpoint’s write cost and the work you are willing to lose. Every N steps rather than every epoch, chosen so a crash costs at most fifteen or twenty minutes, is a reasonable starting point. Keep the last few plus the best-by-metric, and delete the rest, or a long run fills the disk and dies for a different reason.

Two more points. Save the config and the code version alongside the weights, because a checkpoint you cannot map to the architecture that produced it is close to useless. And make the resume path the default: a script that always looks for a checkpoint and continues from it, rather than one you have to remember to invoke with a flag, is what actually survives an unattended restart.

GPU operations are queued asynchronously. Your Python code returns immediately and the work happens later, which is what allows the CPU to prepare the next batch while the GPU computes. Anything that needs a value back breaks that overlap.

for batch in loader:
    x = batch.to(device, non_blocking=True)
    loss = model(x).mean()
    loss.backward()
    opt.step()
    running += loss.item()        # forces a sync every step

.item() cannot return until the loss has actually been computed, so the CPU blocks. Once per step it may be tolerable; called several times inside the loop for logging, it serialises the whole pipeline.

The fix is to accumulate on the device and read once per interval:

running += loss.detach()          # stays on GPU
if step % 100 == 0:
    log.info("loss %.4f", running.item() / 100)
    running = torch.zeros((), device=device)

detach() matters too. Accumulating the loss tensor without it keeps the whole computation graph alive across steps, which is a memory leak that looks like a batch-size problem.

Three transfer-side points.

Pinned memory lets the copy run asynchronously and go faster, since the pages cannot be swapped out. pin_memory=True on the loader plus non_blocking=True on the transfer is the pair; either alone gives little.

Many small transfers cost far more than one large one, because per-transfer overhead dominates. Move the whole batch at once rather than field by field.

And normalisation, augmentation and type casting done on the CPU send larger float tensors across the bus. Sending uint8 images and converting on the GPU can halve or quarter the bytes transferred.

Profile before assuming. A transfer bottleneck and a loader bottleneck look similar from the utilisation graph and need completely different fixes.

Filenames cannot carry the information you need, and by the third _final nobody knows which is deployed.

The minimum record, whether you write it yourself or use MLflow:

import hashlib, json, shutil
from pathlib import Path

def register(artifact: Path, manifest: dict, root: Path) -> str:
    digest = hashlib.sha256(artifact.read_bytes()).hexdigest()[:12]
    dest = root / digest
    dest.mkdir(parents=True, exist_ok=False)
    shutil.copy2(artifact, dest / "model.joblib")
    (dest / "manifest.json").write_text(json.dumps(manifest, indent=2))
    return digest

exist_ok=False enforces immutability. A registered version is never overwritten, so a deployed identifier always refers to the same bytes. Deriving the identifier from the content hash means two identical artifacts register as one and a changed artifact cannot reuse an id.

The manifest carries what the filename cannot: the config, the commit SHA, the dataset fingerprint, the metrics with the split they were measured on, the library versions, and who trained it.

Separate three things people conflate.

The artifact is bytes on object storage. The version is an immutable record pointing at those bytes plus the manifest. The stage is a mutable pointer, production or staging, moved between versions. Rollback is then repointing the stage at an earlier version, which is instant and reversible, rather than retraining or hunting for an old file.

Two points worth raising. Metrics alone do not distinguish runs; two configs can produce the same validation AUC and behave differently in production, which is why the config and data version belong in the record.

And the registry has to be the only path to deployment. If someone can copy a file onto a server, the registry is documentation rather than a control, and it will be wrong within a month.

The pattern that grows painful:

if kind == "gbdt":
    scores = booster.predict(dmatrix(features))
elif kind == "linear":
    scores = w @ features + b
elif kind == "neural":
    scores = net(torch.tensor(features)).numpy()

Every new model type edits this block, and it appears in the batch path, the online path and the evaluation script, so a change means finding all three.

One interface instead:

from typing import Protocol
import numpy as np

class Scorer(Protocol):
    version: str

    def score(self, features: np.ndarray) -> np.ndarray:
        """Return one probability per row."""

class GbdtScorer:
    def __init__(self, booster, version: str):
        self._b, self.version = booster, version

    def score(self, features):
        return self._b.predict(dmatrix(features))

Protocol gives structural typing checked statically with no inheritance, which suits wrapping third-party estimators you do not control. An abstract base class is the alternative when you want the failure at instantiation time and shared concrete helpers.

Two details that make this work in practice.

Fix the contract precisely, not just the name. What shape goes in, what comes out, and what the values mean. If one implementation returns log-odds and another returns probabilities, the interface has not actually unified anything and the bug will surface as miscalibrated scores rather than an error.

Extra parameters belong in the constructor, not the method. A model needing a temperature or a threshold takes it when built, so score keeps one signature across all types.

Selection happens once, in a factory that reads the model type from the registry manifest and returns the right implementation. Everything downstream sees a Scorer.

The honest limit: if two model types genuinely need different inputs, forcing one interface produces a signature full of optional arguments that most implementations ignore. That is worse than the conditionals.

Three causes, distinguishable by what the memory curve looks like.

Graph accumulation. The classic version:

losses = []
for batch in loader:
    loss = model(batch).mean()
    loss.backward()
    losses.append(loss)          # keeps the whole graph

Storing the loss tensor keeps every intermediate activation that produced it alive. Memory climbs steadily across the epoch and dies partway through. losses.append(loss.detach()), or better loss.item(), fixes it.

The same shape appears when evaluating without torch.no_grad(), since the graph is built even though no backward pass follows.

A larger batch. With variable-length inputs, memory depends on the longest item in the batch, so step 900 containing one very long sequence allocates far more than the average step. The curve is flat with a spike rather than a climb.

Fragmentation. The caching allocator keeps freed blocks rather than returning them to the driver, and after many differently sized allocations it may hold enough free memory in total but no single contiguous block large enough. torch.cuda.empty_cache() releases the cached blocks back to the driver, which helps here and does nothing at all for the first two causes. So if empty_cache does not help, you have a genuine leak, not fragmentation.

Diagnosis:

print(torch.cuda.memory_allocated() / 1e9,   # tensors alive
      torch.cuda.memory_reserved() / 1e9)    # held by allocator

Print both every hundred steps. Allocated climbing means accumulation. Allocated flat with reserved climbing means fragmentation.

Two more points. nvidia-smi shows the reserved figure, not the allocated one, so it looks alarming even for a healthy run. And torch.cuda.memory_summary() gives a detailed breakdown worth reading before guessing.

Reach for this only after profiling has identified one function, and only when the work is genuinely sequential so array operations cannot express it.

from numba import njit

@njit(cache=True)
def rolling_peak_count(signal, window, threshold):
    count = 0
    peak = 0.0
    for i in range(len(signal)):
        if signal[i] > peak:
            peak = signal[i]
        if i >= window:
            if peak - signal[i - window] > threshold:
                count += 1
            peak = 0.0
    return count

njit is nopython mode, which compiles the function to machine code with no interpreter involvement. Compilation happens on the first call, so the first invocation is slow and subsequent ones are fast; cache=True writes the compiled result to disk so later processes skip it.

The constraint is what nopython mode accepts: numeric scalars, numpy arrays and a subset of built-in types. Python objects, dicts of mixed types, string manipulation and most library calls are not supported, and the failure is a compilation error rather than a silent fallback in current versions.

Two reasons a Numba function comes out no faster. It fell back to object mode in an older version or under @jit without nopython=True, in which case it is the interpreter with extra steps. Or you are timing the first call and measuring compilation.

Cython suits a different case: a larger body of code, a need for C library interop, or a stable library you ship where a compiled extension is acceptable. The cost is a build step, a compiler dependency, and platform-specific wheels.

Before either, check the ordinary options. A better algorithm beats compilation. So does moving the loop into an existing compiled routine you had not thought of.

Keep the pure-Python version and test both against each other. A compiled function that is fast and subtly wrong is worse than the slow one it replaced.

The test is whether a caller would want to catch it separately. If the answer is no, ValueError with a clear message is enough and adding a class is noise.

class FeedError(Exception):
    """Base for everything this module raises."""

class SchemaMismatch(FeedError):
    def __init__(self, expected, found):
        self.expected = expected
        self.found = found
        super().__init__(
            f"expected {expected} columns, found {found}"
        )

class FeedUnavailable(FeedError):
    pass

The base class is the part that earns its keep. A caller who does not care which specific thing went wrong writes except FeedError: once, and keeps working when you add a fourth subclass next month. Without a shared base, every caller maintains a growing tuple of your exception names.

Carrying structured attributes matters more than the message. A retry layer needs to know whether the failure is transient, and parsing that out of a string is fragile:

except SchemaMismatch as exc:
    if exc.found > exc.expected:
        drop_extra_columns()

Two things worth mentioning in an interview. Inherit from a built-in when the semantics genuinely match and callers may already be catching it: a custom lookup failure inheriting from KeyError keeps existing code working. Do not inherit from BaseException; that is reserved for things that should escape ordinary handlers.

And when wrapping a lower-level error, preserve the chain:

except ConnectionError as exc:
    raise FeedUnavailable("vendor feed down") from exc

from exc attaches the original traceback. Without it, the log shows your exception and nothing about the socket error that caused it, which is exactly the detail you need at 2 AM.