Python Interview Questions for AI Engineers: With Executable Code

Introduction

AI engineer interviews test Python differently from classic software interviews: alongside language fundamentals, you are expected to stream large datasets, call model APIs concurrently and reliably, vectorize math with NumPy, and turn messy model output into validated data. This article collects 21 questions that come up repeatedly, each with a concise answer and code you can run.

The questions are grouped into ten areas, moving from core language behavior to production concerns:

Core language semantics

Data structures and complexity

Iterators and generators

Decorators and context managers

Classes, typing and validation

Concurrency and async

NumPy for machine learning

Machine learning from scratch

LLM and RAG engineering

Testing and production quality

How to use this article. Every snippet is self-contained and runs on Python 3.10 or later. The NumPy sections need numpy, and the validation section needs pydantic (v2); everything else uses only the standard library. Expected output is shown in comments. In an interview, the code is only half the answer: interviewers listen for why you chose an approach, its time and memory cost, and how it fails. Each answer ends with the follow-up points worth mentioning.

1. Core language semantics

These questions check whether you understand how Python actually treats objects and names. Bugs from these behaviors are common in data and ML code, where configs, datasets and callbacks get passed around constantly.

Q1. What is wrong with using a mutable default argument?

Answer. Default values are evaluated once, when the function is defined, not on each call. A mutable default such as a list is therefore shared across every call that does not pass the argument.

def add_example(example, dataset=[]):
    dataset.append(example)
    return dataset

print(add_example("a"))   # ['a']
print(add_example("b"))   # ['a', 'b']  <- state leaked from the first call

def add_example_fixed(example, dataset=None):
    if dataset is None:
        dataset = []
    dataset.append(example)
    return dataset

print(add_example_fixed("a"))   # ['a']
print(add_example_fixed("b"))   # ['b']

Follow-up points. Use None as a sentinel for lists, dicts and sets. The same trap appears in dataclasses, which is why dataclasses.field(default_factory=list) exists.

Q2. What is the difference between a shallow copy and a deep copy?

Answer. A shallow copy creates a new outer container but shares the nested objects. A deep copy recursively copies everything. With nested configs, such as model parameters, a shallow copy lets one experiment silently change another.

import copy

config = {"model": "gpt", "params": {"temperature": 0.7, "stop": ["\n"]}}
shallow = copy.copy(config)
deep = copy.deepcopy(config)

shallow["params"]["temperature"] = 0.0
print(config["params"]["temperature"])   # 0.0  <- original changed
print(deep["params"]["temperature"])     # 0.7  <- independent

Follow-up points. dict(config), config.copy() and slicing lst[:] are all shallow. Deep copies cost time and memory on large objects; immutable structures (tuples, frozen dataclasses) avoid the problem entirely.

Q3. Why do these lambdas all return the same value?

Answer. Closures capture variables, not values. Each lambda looks up i when it is called, after the loop has finished, so all see the final value. Binding the current value as a default argument fixes it.

callbacks = [lambda: i for i in range(3)]
print([f() for f in callbacks])   # [2, 2, 2]

callbacks = [lambda i=i: i for i in range(3)]
print([f() for f in callbacks])   # [0, 1, 2]

Follow-up points. This is called late binding. It bites when registering callbacks or building per-epoch learning-rate schedules in a loop; functools.partial is a clean alternative to the default-argument trick.

2. Data structures and complexity

AI systems constantly count tokens, rank search results and cache expensive calls. Interviewers want to see that you reach for the right built-in structure and know its cost.

Q4. How would you count word frequencies across a corpus?

Answer. collections.Counter counts any iterable in O(n) and gives the most common items directly. Feeding it a generator expression avoids building an intermediate list.

from collections import Counter

docs = ["the cat sat", "the dog sat", "a cat ran"]
counts = Counter(word for doc in docs for word in doc.split())
print(counts.most_common(2))   # [('the', 2), ('cat', 2)]

Follow-up points. Ties are returned in first-seen order. Counters support arithmetic (c1 + c2, c1 - c2), which is handy for merging counts from parallel workers. For vocabulary building at scale, mention that real tokenizers use subword units, not whitespace splits.

Q5. Return the top-k documents by similarity score. What is the complexity?

Answer. Sorting everything is O(n log n). A min-heap of size k is O(n log k) time and O(k) memory, and it works on a stream that does not fit in memory, which is exactly the situation when scoring millions of chunks in a retrieval system.

import heapq

scores = [("doc_a", 0.82), ("doc_b", 0.91), ("doc_c", 0.45),
          ("doc_d", 0.88), ("doc_e", 0.67)]

# Built-in helper
print(heapq.nlargest(3, scores, key=lambda x: x[1]))
# [('doc_b', 0.91), ('doc_d', 0.88), ('doc_a', 0.82)]

# Streaming version: keep only the k best seen so far
def top_k_stream(stream, k):
    heap = []                                    # min-heap of (score, doc_id)
    for doc_id, score in stream:
        if len(heap) < k:
            heapq.heappush(heap, (score, doc_id))
        elif score > heap[0][0]:                 # beats the current worst
            heapq.heapreplace(heap, (score, doc_id))
    return sorted(heap, reverse=True)

print(top_k_stream(scores, 3))
# [(0.91, 'doc_b'), (0.88, 'doc_d'), (0.82, 'doc_a')]

Follow-up points. The heap root is always the weakest of the current top k, so each new item needs only one comparison to reject. With NumPy arrays, np.argpartition gives O(n) top-k selection, shown in Section 7.

Q6. Implement an LRU cache for embeddings.

Answer. An LRU cache needs O(1) lookup and O(1) reordering. OrderedDict provides both: move_to_end marks an item as recently used, and popitem(last=False) evicts the oldest.

from collections import OrderedDict

class LRUCache:
    def __init__(self, capacity: int):
        self.capacity = capacity
        self.data = OrderedDict()

    def get(self, key):
        if key not in self.data:
            return None
        self.data.move_to_end(key)            # mark as recently used
        return self.data[key]

    def put(self, key, value):
        self.data[key] = value
        self.data.move_to_end(key)
        if len(self.data) > self.capacity:
            self.data.popitem(last=False)     # evict least recently used

cache = LRUCache(2)
cache.put("q1", [0.1, 0.2])
cache.put("q2", [0.3, 0.4])
cache.get("q1")                  # q1 is now most recent
cache.put("q3", [0.5, 0.6])      # evicts q2
print(list(cache.data.keys()))   # ['q1', 'q3']
print(cache.get("q2"))           # None

Follow-up points. For pure functions, functools.lru_cache does this in one line (Section 4). Interviewers may ask for the classic implementation with a hash map plus a doubly linked list; OrderedDict is exactly that internally. In production, embedding caches usually live in Redis so all workers share them.

3. Iterators and generators

Training data, evaluation sets and document corpora are often far larger than memory. Generators let you process them one record at a time, and interviewers use them to test whether you can write memory-safe data pipelines.

Q7. What is the difference between a list comprehension and a generator expression?

Answer. A list comprehension builds every element immediately and holds them all in memory. A generator expression produces elements lazily, one at a time, using constant memory. It can only be iterated once.

import sys

squares_list = [x * x for x in range(1_000_000)]
squares_gen = (x * x for x in range(1_000_000))

print(sys.getsizeof(squares_list) > 8_000_000)   # True: ~8 MB of pointers alone
print(sys.getsizeof(squares_gen) < 300)          # True: ~200 bytes, regardless of n

Follow-up points. Use a list when you need random access, len(), or multiple passes. Use a generator when you stream once through the data, for example sum(len(t) for t in texts). A generator that has been exhausted silently yields nothing on a second loop, which is a frequent bug.

Q8. Write a pipeline that streams a large JSONL file and yields records in batches.

Answer. Chain two generators: one parses lines lazily and skips bad records, the other groups items into fixed-size batches. Only one batch is in memory at a time, which is the shape you need for sending documents to an embedding API.

import io
import json

def read_jsonl(f):
    for line_no, line in enumerate(f, start=1):
        line = line.strip()
        if not line:
            continue
        try:
            yield json.loads(line)
        except json.JSONDecodeError:
            print(f"skipping bad line {line_no}")

def batched(iterable, size):
    batch = []
    for item in iterable:
        batch.append(item)
        if len(batch) == size:
            yield batch
            batch = []
    if batch:                       # don't drop the final partial batch
        yield batch

# io.StringIO stands in for open("data.jsonl")
data = io.StringIO('{"text": "a"}\n{"text": "b"}\nnot json\n'
                   '{"text": "c"}\n\n{"text": "d"}\n{"text": "e"}\n')

for batch in batched(read_jsonl(data), 2):
    print([r["text"] for r in batch])
# ['a', 'b']
# skipping bad line 3
# ['c', 'd']
# ['e']

Follow-up points. Python 3.12 added itertools.batched, which yields tuples. Mention the trailing partial batch explicitly; forgetting it silently drops data. In production, log bad lines with their line numbers to a dead-letter file rather than printing them.

4. Decorators and context managers

Model APIs fail intermittently, calls are slow and expensive, and GPU or network resources must be released reliably. Decorators and context managers are how Python engineers package retry, caching, timing and cleanup logic once and reuse it everywhere.

Q9. Write a retry decorator with exponential backoff for flaky API calls.

Answer. A decorator with arguments is three nested functions: the outer one takes the configuration, the middle one takes the function, and the inner one wraps each call. Backoff doubles the delay each attempt, and jitter randomizes it so many clients do not retry in lockstep.

import functools
import random
import time

class TransientError(Exception):
    pass

def retry(max_attempts=3, base_delay=0.1, exceptions=(TransientError,)):
    def decorator(func):
        @functools.wraps(func)                  # keep name and docstring
        def wrapper(*args, **kwargs):
            for attempt in range(1, max_attempts + 1):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    if attempt == max_attempts:
                        raise                   # give up, re-raise original
                    delay = base_delay * 2 ** (attempt - 1) * random.uniform(0.5, 1.5)
                    print(f"attempt {attempt} failed ({e}); retrying in {delay:.2f}s")
                    time.sleep(delay)
        return wrapper
    return decorator

calls = {"n": 0}

@retry(max_attempts=4, base_delay=0.05)
def call_model(prompt):
    calls["n"] += 1
    if calls["n"] < 3:                          # fail twice, then succeed
        raise TransientError("503 Service Unavailable")
    return f"response to {prompt!r}"

print(call_model("hello"))
# attempt 1 failed (503 Service Unavailable); retrying in 0.05s
# attempt 2 failed (503 Service Unavailable); retrying in 0.10s
# response to 'hello'
print(call_model.__name__)   # call_model (thanks to functools.wraps)

Follow-up points. Retry only transient errors (timeouts, HTTP 429 and 5xx), never 400-level validation errors. Honor a Retry-After header when the API sends one. Only retry idempotent operations, or you may charge a customer twice. The tenacity library offers a production-grade version.

Q10. Write a context manager that times a block of code.

Answer. contextlib.contextmanager turns a generator into a context manager: code before yield runs on entry, code after it runs on exit. Putting the exit code in finally guarantees it runs even when the block raises.

import time
from contextlib import contextmanager

@contextmanager
def timer(label):
    start = time.perf_counter()
    try:
        yield
    finally:
        print(f"{label}: {time.perf_counter() - start:.3f}s")

with timer("embedding batch"):
    sum(i * i for i in range(1_000_000))
# embedding batch: 0.075s  (varies by machine)

Follow-up points. The class-based equivalent defines __enter__ and __exit__; returning True from __exit__ suppresses the exception. Use time.perf_counter() for durations, not time.time(), which can jump when the system clock changes. Real uses include holding a GPU lock, opening a database transaction, or temporarily setting torch.no_grad().

Q11. How do you cache results of an expensive pure function?

Answer. functools.lru_cache memoizes by arguments. The first call computes; repeated calls with the same arguments return instantly.

import functools
import time

@functools.lru_cache(maxsize=1024)
def embed(text: str) -> tuple[float, ...]:
    time.sleep(0.2)                          # simulate a slow model call
    return tuple(float(ord(c)) for c in text[:3])

t = time.perf_counter(); embed("hello"); first = time.perf_counter() - t
t = time.perf_counter(); embed("hello"); second = time.perf_counter() - t
print(f"first: {first:.2f}s, second: {second:.4f}s")   # first: 0.20s, second: 0.0000s
print(embed.cache_info())
# CacheInfo(hits=1, misses=1, maxsize=1024, currsize=1)

Follow-up points. Arguments must be hashable, so pass tuples, not lists. Return immutable values too: a cached list can be mutated by one caller and corrupt every later result. The cache is per process and is lost on restart; caching non-deterministic LLM outputs (temperature above 0) changes behavior, so be deliberate about it.

5. Classes, typing and validation

AI codebases pass around configs, prompts and model outputs that must have the right shape. Interviewers look for structured types instead of loose dictionaries, and for validation at the boundary where untrusted data, including LLM output, enters your system.

Q12. When would you use a dataclass, and how do you make one immutable and validated?

Answer. Dataclasses generate __init__, __repr__ and __eq__ from type-annotated fields, which is ideal for configuration objects. frozen=True makes instances immutable and hashable, and __post_init__ is the hook for validation.

from dataclasses import dataclass, replace

@dataclass(frozen=True)
class ModelConfig:
    name: str
    temperature: float = 0.7
    max_tokens: int = 512
    stop: tuple[str, ...] = ()               # tuple, not list: immutable default

    def __post_init__(self):
        if not 0.0 <= self.temperature <= 2.0:
            raise ValueError(f"temperature must be in [0, 2], got {self.temperature}")

cfg = ModelConfig("my-model", temperature=0.2)
print(cfg)   # ModelConfig(name='my-model', temperature=0.2, max_tokens=512, stop=())

try:
    cfg.temperature = 1.0
except Exception as e:
    print(type(e).__name__)                  # FrozenInstanceError

try:
    ModelConfig("x", temperature=3.0)
except ValueError as e:
    print(e)                                 # temperature must be in [0, 2], got 3.0

print(replace(cfg, max_tokens=1024).max_tokens)   # 1024: a modified copy

Follow-up points. Immutable configs make experiments reproducible and safe to share across threads. replace() creates variants without mutation. Type hints are not enforced at runtime by dataclasses; that is what the next question solves.

Q13. How do you validate structured output from an LLM?

Answer. Define the expected shape as a Pydantic model and parse the raw JSON through it. Pydantic enforces types and constraints at runtime, coerces compatible values (such as the string "0.93" to a float), and reports every error with its location, which you can feed back to the model in a repair prompt.

from typing import Literal
from pydantic import BaseModel, Field, ValidationError

class Sentiment(BaseModel):
    label: Literal["positive", "negative", "neutral"]
    confidence: float = Field(ge=0.0, le=1.0)
    reasons: list[str] = []

raw = '{"label": "positive", "confidence": "0.93", "reasons": ["praises battery"]}'
print(Sentiment.model_validate_json(raw))
# label='positive' confidence=0.93 reasons=['praises battery']

try:
    Sentiment.model_validate_json('{"label": "very positive", "confidence": 1.4}')
except ValidationError as e:
    print(e.error_count(), "validation errors")
    for err in e.errors():
        print(err["loc"], err["type"])
# 2 validation errors
# ('label',) literal_error
# ('confidence',) less_than_equal

print(Sentiment.model_json_schema()["required"])   # ['label', 'confidence']

Follow-up points. model_json_schema() produces the JSON Schema you can pass to a model's structured-output or tool-calling feature, so one class defines both the request and the validation. Pydantic's mutable defaults like [] are safe because it copies them per instance, unlike plain function defaults in Q1. Use strict=True when you do not want coercion.

6. Concurrency and async

An AI application often spends most of its time waiting on model APIs, vector databases and HTTP calls, while preprocessing and feature computation burn CPU. Choosing the right concurrency model for each is one of the most reliable signals of seniority in these interviews.

Q14. What is the GIL, and when do you use threads versus processes?

Answer. In standard CPython, the Global Interpreter Lock lets only one thread execute Python bytecode at a time. Threads still help with I/O-bound work, because a thread waiting on the network releases the GIL. CPU-bound pure-Python work needs separate processes, each with its own interpreter and GIL.

import time
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor

def cpu_task(n):
    return sum(i * i for i in range(n))

def io_task(_):
    time.sleep(0.5)          # stands in for a network call
    return "done"

if __name__ == "__main__":   # required for processes on Windows and macOS
    t = time.perf_counter()
    with ThreadPoolExecutor(max_workers=8) as ex:
        list(ex.map(io_task, range(8)))
    print(f"8 I/O calls, threads: {time.perf_counter() - t:.1f}s")   # 0.5s, not 4.0s

    for name, Pool in [("threads", ThreadPoolExecutor), ("processes", ProcessPoolExecutor)]:
        t = time.perf_counter()
        with Pool(max_workers=4) as ex:
            list(ex.map(cpu_task, [3_000_000] * 4))
        print(f"CPU work, {name}: {time.perf_counter() - t:.2f}s")
    # Threads take about as long as running sequentially;
    # processes scale with the number of CPU cores.

Follow-up points. NumPy, PyTorch and most C extensions release the GIL inside heavy operations, so threads can speed those up. Processes pay for pickling arguments and results, so send small inputs and large work. Python 3.13 introduced an experimental free-threaded build without the GIL, and 3.14 made it officially supported; mention it, but note that most production deployments still run the standard build.

Q15. Make 20 LLM calls concurrently without exceeding 5 in flight, and survive individual failures.

Answer. Use asyncio: a single thread can wait on many network calls at once. asyncio.Semaphore caps concurrency, asyncio.wait_for adds a per-call timeout, and gather(..., return_exceptions=True) collects failures instead of letting one error cancel the whole batch.

import asyncio
import random
import time

async def call_llm(prompt, semaphore):
    async with semaphore:                    # at most 5 requests in flight
        await asyncio.sleep(0.5)             # stands in for an async HTTP call
        if random.random() < 0.1:
            raise TimeoutError(f"timeout on {prompt}")
        return f"answer to {prompt}"

async def main():
    prompts = [f"q{i}" for i in range(20)]
    semaphore = asyncio.Semaphore(5)
    start = time.perf_counter()

    tasks = [asyncio.wait_for(call_llm(p, semaphore), timeout=5) for p in prompts]
    results = await asyncio.gather(*tasks, return_exceptions=True)

    ok = [r for r in results if not isinstance(r, Exception)]
    failed = [r for r in results if isinstance(r, Exception)]
    print(f"{len(ok)} succeeded, {len(failed)} failed "
          f"in {time.perf_counter() - start:.1f}s")

random.seed(7)
asyncio.run(main())
# 15 succeeded, 5 failed in 2.0s   (20 calls / 5 at a time x 0.5s)

Sequentially, the 20 calls would take 10 seconds; with five in flight they take 2. gather returns results in the same order as the inputs, so you can zip them back to their prompts and retry only the failures.

Follow-up points. A semaphore limits concurrency, not rate. If the provider enforces requests or tokens per minute, add a token-bucket limiter as well. Never call blocking code such as time.sleep or requests.get inside a coroutine, because it freezes the entire event loop; use httpx.AsyncClient or the provider's async SDK, or offload with asyncio.to_thread. Python 3.11's asyncio.TaskGroup is the structured alternative when one failure should cancel the rest.

7. NumPy for machine learning

NumPy questions test whether you think in arrays instead of loops. Vectorized code is typically hundreds of times faster, and numerical stability matters as soon as you work with exponentials and probabilities.

Q16. Implement a numerically stable softmax.

Answer. The naive formula overflows: exp(1000) is infinity, and infinity divided by infinity is NaN. Subtracting the maximum before exponentiating gives a mathematically identical result, because softmax is unchanged when every input shifts by the same constant, and keeps every exponent at most zero.

import numpy as np

x = np.array([1000.0, 1001.0, 1002.0])
print(np.exp(x) / np.exp(x).sum())       # [nan nan nan] + overflow warning

def softmax(z, axis=-1):
    z = z - z.max(axis=axis, keepdims=True)
    e = np.exp(z)
    return e / e.sum(axis=axis, keepdims=True)

print(softmax(x).round(4))               # [0.09   0.2447 0.6652]

logits = np.array([[2.0, 1.0, 0.1],
                   [1.0, 3.0, 0.2]])     # a batch of 2 rows
print(softmax(logits).round(3))
# [[0.659 0.242 0.099]
#  [0.113 0.836 0.051]]
print(softmax(logits).sum(axis=1))       # [1. 1.]

Follow-up points. keepdims=True keeps the reduced axis with size 1, so the result broadcasts correctly against the original batch. Broadcasting compares shapes from the right; dimensions match when they are equal or one of them is 1, so a (3, 1) array plus a (4,) array gives (3, 4). For cross-entropy loss, compute log-softmax directly (the log-sum-exp trick) rather than taking log(softmax(x)), which loses precision for tiny probabilities.

Q17. Given 10,000 document embeddings, find the 5 most similar to a query.

Answer. Cosine similarity is the dot product of unit vectors. Normalize once, compute all similarities with one matrix-vector product, then use np.argpartition to select the top k in O(n) before sorting just those k. This is the core of every vector search engine.

import numpy as np

rng = np.random.default_rng(0)
docs = rng.normal(size=(10_000, 384)).astype(np.float32)        # 384-dim embeddings
query = docs[42] + 0.1 * rng.normal(size=384).astype(np.float32)  # near doc 42

def normalize(v):
    return v / np.linalg.norm(v, axis=-1, keepdims=True)

docs_n = normalize(docs)          # do this once, at index time
q_n = normalize(query)

sims = docs_n @ q_n               # (10000, 384) @ (384,) -> (10000,)

k = 5
top = np.argpartition(-sims, k)[:k]      # unordered top 5, O(n)
top = top[np.argsort(-sims[top])]        # sort only those 5
print(top[0], sims[top[0]].round(3))     # 42 0.995

On a typical laptop, the equivalent pure-Python loop over 10,000 vectors takes over a second, while the matrix product takes about 2 milliseconds.

Follow-up points. Normalizing at index time makes each query a single dot product. Use float32 (or float16) to halve memory, since the extra precision of float64 is irrelevant for similarity ranking. Exact search is O(n·d) per query; beyond a few million vectors, approximate nearest-neighbor indexes such as HNSW (in FAISS or a vector database) trade a little recall for orders-of-magnitude speed.

8. Machine learning from scratch

You will rarely implement a model from scratch at work, but interviewers ask for it because it reveals whether you understand what libraries do: the forward pass, the loss, the gradient and the update.

Q18. Implement logistic regression with gradient descent and evaluate it.

Answer. Logistic regression passes a linear score through a sigmoid to get a probability and minimizes binary cross-entropy. Its gradient has a clean form: for weights, X.T @ (p - y) / n; for the bias, the mean of p - y. Each step moves the parameters against that gradient.

import numpy as np

def sigmoid(z):
    return 1.0 / (1.0 + np.exp(-np.clip(z, -500, 500)))   # clip avoids overflow

class LogisticRegressionGD:
    def __init__(self, lr=0.1, epochs=1000, l2=0.0):
        self.lr, self.epochs, self.l2 = lr, epochs, l2

    def fit(self, X, y):
        n, d = X.shape
        self.w = np.zeros(d)
        self.b = 0.0
        self.losses = []
        for _ in range(self.epochs):
            p = sigmoid(X @ self.w + self.b)                # forward pass
            err = p - y
            self.w -= self.lr * (X.T @ err / n + self.l2 * self.w)   # gradient step
            self.b -= self.lr * err.mean()
            eps = 1e-12                                     # avoid log(0)
            self.losses.append(-np.mean(y * np.log(p + eps)
                                        + (1 - y) * np.log(1 - p + eps)))
        return self

    def predict_proba(self, X):
        return sigmoid(X @ self.w + self.b)

    def predict(self, X, threshold=0.5):
        return (self.predict_proba(X) >= threshold).astype(int)

def classification_metrics(y_true, y_pred):
    tp = np.sum((y_pred == 1) & (y_true == 1))
    fp = np.sum((y_pred == 1) & (y_true == 0))
    fn = np.sum((y_pred == 0) & (y_true == 1))
    tn = np.sum((y_pred == 0) & (y_true == 0))
    precision = tp / (tp + fp) if tp + fp else 0.0
    recall = tp / (tp + fn) if tp + fn else 0.0
    f1 = 2 * precision * recall / (precision + recall) if precision + recall else 0.0
    return {"accuracy": round(float((tp + tn) / len(y_true)), 3),
            "precision": round(float(precision), 3),
            "recall": round(float(recall), 3),
            "f1": round(float(f1), 3)}

# Synthetic data from a known model: true weights [2, -3], bias 0.5
rng = np.random.default_rng(1)
X = rng.normal(size=(1000, 2))
y = (rng.random(1000) < sigmoid(X @ np.array([2.0, -3.0]) + 0.5)).astype(int)

model = LogisticRegressionGD(lr=0.5, epochs=500).fit(X[:800], y[:800])
print("weights:", model.w.round(2), "bias:", round(model.b, 2))
# weights: [ 1.69 -2.46] bias: 0.56
print(f"loss: {model.losses[0]:.3f} -> {model.losses[-1]:.3f}")
# loss: 0.693 -> 0.355
print(classification_metrics(y[800:], model.predict(X[800:])))
# {'accuracy': 0.87, 'precision': 0.839, 'recall': 0.934, 'f1': 0.884}

The starting loss of 0.693 is ln 2, exactly what a model that predicts 0.5 for everything should score, which is a useful sanity check. The learned weights have the right signs and rough magnitudes; the gap from the true values comes from sampling noise in 800 labeled points, not from the optimizer, since the loss has converged.

Follow-up points. Explain each metric's meaning: precision is the share of predicted positives that are real, recall the share of real positives found. Mention feature scaling (gradient descent converges much faster on standardized inputs), L2 regularization (the l2 term), and mini-batch or stochastic gradient descent for data that does not fit in memory. If asked to extend it, multiclass logistic regression replaces the sigmoid with the softmax from Q16.

9. LLM and RAG engineering

These are the questions most specific to AI engineering roles. They test whether you can prepare documents for retrieval and safely consume what a model returns.

Q19. Write a function that splits a document into overlapping chunks for RAG.

Answer. Split on sentence boundaries so chunks stay readable, pack sentences up to a size limit, and carry the tail of each chunk into the next. The overlap ensures an idea that straddles a boundary is fully present in at least one chunk.

import re

def chunk_text(text, chunk_size=200, overlap=40):
    if overlap >= chunk_size:
        raise ValueError("overlap must be smaller than chunk_size")
    sentences = re.split(r"(?<=[.!?])\s+", text.strip())
    chunks, current = [], ""
    for sentence in sentences:
        if current and len(current) + len(sentence) + 1 > chunk_size:
            chunks.append(current)
            tail = current[-overlap:]
            tail = tail[tail.find(" ") + 1:] if " " in tail else tail  # whole words
            current = f"{tail} {sentence}"
        else:
            current = f"{current} {sentence}".strip()
    if current:
        chunks.append(current)
    return chunks

text = ("Retrieval-augmented generation grounds a model in external documents. "
        "Documents are split into chunks and embedded. "
        "At query time, the most similar chunks are retrieved. "
        "They are inserted into the prompt as context. "
        "Chunk size trades precision against context. "
        "Overlap keeps ideas that span a boundary retrievable.")

for i, chunk in enumerate(chunk_text(text, chunk_size=120, overlap=30)):
    print(i, len(chunk), chunk)
# 0 115 Retrieval-augmented generation grounds ... split into chunks and embedded.
# 1 79 into chunks and embedded. At query time, the most similar chunks are retrieved.
# 2 120 similar chunks are retrieved. They are inserted ... precision against context.
# 3 80 precision against context. Overlap keeps ideas that span a boundary retrievable.

Follow-up points. This version measures characters for clarity; production systems measure tokens with the model's own tokenizer (for example tiktoken), because context limits and pricing are in tokens. A single sentence longer than the limit still becomes one oversized chunk; a complete answer splits it further. Discuss trade-offs: small chunks retrieve precisely but lose context, large ones dilute the embedding. Store metadata (source, page, offset) with each chunk so answers can cite their sources.

Q20. An LLM was asked for JSON but sometimes wraps it in prose or code fences. Parse it robustly.

Answer. Prefer the provider's structured-output or tool-calling mode, which constrains the format. When you must parse free text, look for a fenced block first, then fall back to the outermost braces, and always validate the result (Q13) before trusting it.

import json
import re

def extract_json(raw: str) -> dict:
    fenced = re.search(r"```(?:json)?\s*(\{.*?\})\s*```", raw, re.DOTALL)
    candidate = fenced.group(1) if fenced else raw
    start, end = candidate.find("{"), candidate.rfind("}")
    if start == -1 or end == -1:
        raise ValueError("no JSON object found")
    return json.loads(candidate[start:end + 1])

outputs = [
    '{"label": "positive", "confidence": 0.9}',
    'Sure! Here is the result:\n```json\n{"label": "negative", "confidence": 0.8}\n```',
    'The answer is {"label": "neutral", "confidence": 0.6}. Hope this helps!',
]
for raw in outputs:
    print(extract_json(raw))
# {'label': 'positive', 'confidence': 0.9}
# {'label': 'negative', 'confidence': 0.8}
# {'label': 'neutral', 'confidence': 0.6}

try:
    extract_json("I could not classify this.")
except ValueError as e:
    print("error:", e)          # error: no JSON object found

Follow-up points. Combine this with Q9 and Q13 into a repair loop: extract, validate with Pydantic, and on failure re-prompt the model with the validation errors, up to a small retry limit. Never pass model output to eval(), and treat it as untrusted input everywhere: it can contain prompt-injected instructions from retrieved documents. Log every parse failure, because a rising failure rate is often the first sign of a model or prompt regression.

10. Testing and production quality

Senior interviews increasingly end with "how would you test this?" LLM-backed code is non-deterministic, slow and costs money per call, so the expected answer is to isolate the model behind an interface and test your own logic with a fake.

Q21. How do you unit-test code that calls an LLM?

Answer. Inject the client as a dependency instead of creating it inside the class. In tests, pass a Mock whose return value you control. The tests then run in milliseconds, cost nothing, behave identically every time, and can simulate failures that are hard to trigger against a real API.

# test_classifier.py  -- run with:  pytest -q test_classifier.py
import json
from unittest.mock import Mock

import pytest

class SentimentClassifier:
    def __init__(self, client):
        self.client = client                 # injected, not created here

    def classify(self, text: str) -> str:
        if not text.strip():
            raise ValueError("text must not be empty")
        raw = self.client.complete(f"Classify the sentiment as JSON: {text}")
        label = json.loads(raw)["label"]
        if label not in {"positive", "negative", "neutral"}:
            return "neutral"                 # safe fallback for unexpected labels
        return label

@pytest.fixture
def fake_client():
    return Mock()

@pytest.mark.parametrize("model_output, expected", [
    ('{"label": "positive"}', "positive"),
    ('{"label": "negative"}', "negative"),
    ('{"label": "ecstatic"}', "neutral"),     # unexpected label
])
def test_classify_maps_labels(fake_client, model_output, expected):
    fake_client.complete.return_value = model_output
    assert SentimentClassifier(fake_client).classify("great phone") == expected

def test_prompt_contains_input(fake_client):
    fake_client.complete.return_value = '{"label": "positive"}'
    SentimentClassifier(fake_client).classify("great phone")
    prompt = fake_client.complete.call_args.args[0]
    assert "great phone" in prompt
    fake_client.complete.assert_called_once()

def test_empty_input_never_calls_model(fake_client):
    with pytest.raises(ValueError):
        SentimentClassifier(fake_client).classify("   ")
    fake_client.complete.assert_not_called()  # no wasted API spend

def test_api_error_propagates(fake_client):
    fake_client.complete.side_effect = TimeoutError("upstream timeout")
    with pytest.raises(TimeoutError):
        SentimentClassifier(fake_client).classify("great phone")

# ......                                                    [100%]
# 6 passed in 0.04s

Follow-up points. Distinguish three layers of testing. Unit tests with mocks check your logic (parsing, fallbacks, prompt construction). A small set of integration tests hits the real API on a schedule to catch contract changes. Evaluations ("evals") score model quality on a labeled dataset and track it over time, since a passing unit test says nothing about whether the answers are good. Mention Mock(spec=RealClient) to catch calls to methods that do not exist, and recording tools such as vcrpy for replaying real responses.

Quick reference and preparation tips

Use this table for a last review before an interview: each row is the one idea the interviewer is listening for.

How to prepare

Run every snippet yourself, then change one thing and predict the result before running it again. Being able to predict output is exactly what interviewers probe.

Narrate trade-offs. For each answer, practice saying the complexity, the memory cost and one way it fails in production. That is what separates a senior answer from a correct one.

Connect the pieces. Strong candidates combine answers: a batching generator (Q8) feeding an async client (Q15) with retries (Q9), whose output is extracted (Q20) and validated (Q13). Practice describing that end-to-end pipeline in two minutes.

Know your versions. Mention modern features where they fit: itertools.batched (3.12), asyncio.TaskGroup (3.11), the free-threaded build (3.13 and later) and Pydantic v2's model_validate_json.

Write tests in the interview. Even one assert after your solution shows production habits, and it often catches your own bug before the interviewer does.

Python fluency in AI engineering is ultimately about reliability: code that handles data larger than memory, APIs that fail, and models that do not always follow instructions. If you can explain these 21 answers and few more similar questions and the reasoning behind them, you are on the right path to confidently face an AI Engineer interview. Understanding the reason and practice together are the key.

Back to all blogs